batch modifications

2 files changed, 174 insertions, 36 deletions

diff --git a/src/exchangedb/pg_batch_reserves_in_insert.c b/src/exchangedb/pg_batch_reserves_in_insert.c
index 455f080dd..f40641edd 100644
--- a/src/exchangedb/pg_batch_reserves_in_insert.c
+++ b/src/exchangedb/pg_batch_reserves_in_insert.c
@@ -16,7 +16,7 @@
 /**
  * @file exchangedb/pg_batch_reserves_in_insert.c
  * @brief Implementation of the reserves_in_insert function for Postgres
- * @author JOSEPHxu
+ * @author Joseph XU
  */
 #include "platform.h"
 #include "taler_error_codes.h"
@@ -35,15 +35,12 @@
 
 
 /**
- * Generate event notification for the reserve
- * change.
+ * Generate event notification for the reserve change.
  *
- * @param pg plugin state
  * @param reserve_pub reserve to notfiy on
  */
-static void
-notify_on_reserve (struct PostgresClosure *pg,
-                   const struct TALER_ReservePublicKeyP *reserve_pub)
+static char *
+compute_notify_on_reserve (const struct TALER_ReservePublicKeyP *reserve_pub)
 {
   struct TALER_ReserveEventP rep = {
     .header.size = htons (sizeof (rep)),
@@ -51,12 +48,7 @@ notify_on_reserve (struct PostgresClosure *pg,
     .reserve_pub = *reserve_pub
   };
 
-  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
-              "Notifying on reserve!\n");
-  TEH_PG_event_notify (pg,
-                       &rep.header,
-                       NULL,
-                       0);
+  return GNUNET_PG_get_event_notify_channel (&rep.header);
 }
 
 
@@ -75,8 +67,11 @@ TEH_PG_batch_reserves_in_insert (void *cls,
   uint64_t reserve_uuid;
   bool conflicted;
   bool transaction_duplicate;
+  bool need_update = false;
   struct GNUNET_TIME_Timestamp reserve_expiration
     = GNUNET_TIME_relative_to_timestamp (pg->idle_reserve_expiration_time);
+  bool conflicts[reserves_length];
+  char *notify_s[reserves_length];
 
   PREPARE (pg,
            "reserve_create",
@@ -84,8 +79,8 @@ TEH_PG_batch_reserves_in_insert (void *cls,
            "out_reserve_found AS conflicted"
            ",transaction_duplicate"
            ",ruuid AS reserve_uuid"
-           " FROM exchange_do_batch_reserves_in"
-           " ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11);");
+           " FROM batch_reserves_insert"
+           " ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12);");
   expiry = GNUNET_TIME_absolute_to_timestamp (
     GNUNET_TIME_absolute_add (reserves->execution_time.abs_time,
                               pg->idle_reserve_expiration_time));
@@ -98,6 +93,16 @@ TEH_PG_batch_reserves_in_insert (void *cls,
               GNUNET_STRINGS_relative_time_to_string (
                 pg->idle_reserve_expiration_time,
                 GNUNET_NO));
+
+  {
+    if (GNUNET_OK !=
+        TEH_PG_start_read_committed(pg,
+                                   "READ_COMMITED"))
+    {
+      GNUNET_break (0);
+      return GNUNET_DB_STATUS_HARD_ERROR;
+    }
+  }
   /* Optimistically assume this is a new reserve, create balance for the first
      time; we do this before adding the actual transaction to "reserves_in",
      as for a new reserve it can't be a duplicate 'add' operation, and as
@@ -105,27 +110,34 @@ TEH_PG_batch_reserves_in_insert (void *cls,
   for (unsigned int i = 0; i<reserves_length; i++)
   {
     const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
+    notify_s[i] = compute_notify_on_reserve (&reserve->reserve_pub);
+  }
+
+  for (unsigned int i=0;i<reserves_length;i++)
+  {
+    const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
     struct GNUNET_PQ_QueryParam params[] = {
-      GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub), /*$1*/
-      GNUNET_PQ_query_param_timestamp (&expiry),  /*$4*/
-      GNUNET_PQ_query_param_timestamp (&gc),  /*$5*/
-      GNUNET_PQ_query_param_uint64 (&reserve->wire_reference), /*6*/
-      TALER_PQ_query_param_amount (&reserve->balance), /*7+8*/
-      GNUNET_PQ_query_param_string (reserve->exchange_account_name), /*9*/
-      GNUNET_PQ_query_param_timestamp (&reserve->execution_time), /*10*/
-      GNUNET_PQ_query_param_auto_from_type (&h_payto), /*11*/
-      GNUNET_PQ_query_param_string (reserve->sender_account_details),/*12*/
-      GNUNET_PQ_query_param_timestamp (&reserve_expiration),/*13*/
+      GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub),
+      GNUNET_PQ_query_param_timestamp (&expiry),
+      GNUNET_PQ_query_param_timestamp (&gc),
+      GNUNET_PQ_query_param_uint64 (&reserve->wire_reference),
+      TALER_PQ_query_param_amount (&reserve->balance),
+      GNUNET_PQ_query_param_string (reserve->exchange_account_name),
+      GNUNET_PQ_query_param_timestamp (&reserve->execution_time),
+      GNUNET_PQ_query_param_auto_from_type (&h_payto),
+      GNUNET_PQ_query_param_string (reserve->sender_account_details),
+      GNUNET_PQ_query_param_timestamp (&reserve_expiration),
+      GNUNET_PQ_query_param_string (notify_s[i]),
       GNUNET_PQ_query_param_end
     };
-    /* We should get all our results into results[]*/
+
     struct GNUNET_PQ_ResultSpec rs[] = {
-      GNUNET_PQ_result_spec_uint64 ("reserve_uuid",
-                                    &reserve_uuid),
       GNUNET_PQ_result_spec_bool ("conflicted",
                                   &conflicted),
       GNUNET_PQ_result_spec_bool ("transaction_duplicate",
                                   &transaction_duplicate),
+      GNUNET_PQ_result_spec_uint64 ("reserve_uuid",
+                                    &reserve_uuid),
       GNUNET_PQ_result_spec_end
     };
 
@@ -137,15 +149,92 @@ TEH_PG_batch_reserves_in_insert (void *cls,
                                                     params,
                                                     rs);
     if (qs1 < 0)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                  "Failed to create reserves (%d)\n",
+                  qs1);
       return qs1;
-    notify_on_reserve (pg,
-                       &reserve->reserve_pub);
-    GNUNET_assert (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != qs1);
-    results[i] = (transaction_duplicate)
+    }
+   GNUNET_assert (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS != qs1);
+   results[i] = (transaction_duplicate)
       ? GNUNET_DB_STATUS_SUCCESS_NO_RESULTS
       : GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
-    if ( (! conflicted) && transaction_duplicate)
-      TEH_PG_rollback (pg);
+   conflicts[i] = conflicted;
+   if (!conflicts[i] && transaction_duplicate)
+   {
+     GNUNET_break (0);
+     TEH_PG_rollback (pg);
+     return GNUNET_DB_STATUS_HARD_ERROR;
+   }
+   need_update |= conflicted;
   }
+  // commit
+  {
+    enum GNUNET_DB_QueryStatus cs;
+
+    cs = TEH_PG_commit (pg);
+    if (cs < 0)
+      return cs;
+  }
+
+  if (!need_update)
+    goto exit;
+  // begin serializable
+  {
+    if (GNUNET_OK !=
+        TEH_PG_start(pg,
+                     "reserve-insert-continued"))
+    {
+      GNUNET_break (0);
+      return GNUNET_DB_STATUS_HARD_ERROR;
+    }
+  }
+
+  enum GNUNET_DB_QueryStatus qs2;
+  PREPARE (pg,
+           "reserves_in_add_transaction",
+           "SELECT batch_reserves_update"
+           " ($1,$2,$3,$4,$5,$6,$7,$8,$9);");
+  for (unsigned int i=0;i<reserves_length;i++)
+  {
+    if (! conflicts[i])
+      continue;
+    {
+      const struct TALER_EXCHANGEDB_ReserveInInfo *reserve = &reserves[i];
+      struct GNUNET_PQ_QueryParam params[] = {
+        GNUNET_PQ_query_param_auto_from_type (&reserve->reserve_pub),
+        GNUNET_PQ_query_param_timestamp (&expiry),
+        GNUNET_PQ_query_param_uint64 (&reserve->wire_reference),
+        TALER_PQ_query_param_amount (&reserve->balance),
+        GNUNET_PQ_query_param_string (reserve->exchange_account_name),
+        GNUNET_PQ_query_param_bool (conflicted),
+        GNUNET_PQ_query_param_auto_from_type (&h_payto),
+        GNUNET_PQ_query_param_string (notify_s[i]),
+        GNUNET_PQ_query_param_end
+      };
+      qs2 = GNUNET_PQ_eval_prepared_non_select (pg->conn,
+                                                "reserves_in_add_transaction",
+                                                params);
+      if (qs2<0)
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                    "Failed to update reserves (%d)\n",
+                    qs2);
+        return qs2;
+      }
+    }
+  }
+  {
+    enum GNUNET_DB_QueryStatus cs;
+
+    cs = TEH_PG_commit (pg);
+    if (cs < 0)
+      return cs;
+  }
+
+ exit:
+  for (unsigned int i=0;i<reserves_length;i++)
+    GNUNET_free (notify_s[i]);
+
   return reserves_length;
 }
diff --git a/src/exchangedb/test_exchangedb_by_j.c b/src/exchangedb/test_exchangedb_by_j.c
index 43f471671..0f252a097 100644
--- a/src/exchangedb/test_exchangedb_by_j.c
+++ b/src/exchangedb/test_exchangedb_by_j.c
@@ -91,9 +91,22 @@ run (void *cls)
     result = 77;
     goto cleanup;
   }
+<<<<<<< HEAD
   for (unsigned int i = 0; i< 7; i++)
+=======
+  if (GNUNET_OK !=
+      plugin->setup_partitions (plugin->cls,
+                                num_partitions))
   {
-    static unsigned int batches[] = {1, 1, 2, 4, 16, 64, 256};
+    GNUNET_break (0);
+    result = 77;
+    goto cleanup;
+  }
+
+  for (unsigned int i = 0; i< 8; i++)
+>>>>>>> 26922c6d (batch modifications)
+  {
+    static unsigned int batches[] = {1, 1,0, 2, 4, 16, 64, 256};
     const char *sndr = "payto://x-taler-bank/localhost:8080/1";
     struct TALER_Amount value;
     unsigned int batch_size = batches[i];
@@ -101,6 +114,7 @@ run (void *cls)
     struct GNUNET_TIME_Timestamp ts;
     struct GNUNET_TIME_Relative duration;
     struct TALER_EXCHANGEDB_ReserveInInfo reserves[batch_size];
+    /*   struct TALER_EXCHANGEDB_ReserveInInfo reserves2[batch_size];*/
     enum GNUNET_DB_QueryStatus results[batch_size];
     GNUNET_assert (GNUNET_OK ==
                    TALER_string_to_amount (CURRENCY ":1.000010",
@@ -109,9 +123,14 @@ run (void *cls)
     ts = GNUNET_TIME_timestamp_get ();
     for (unsigned int r = 0; r<10; r++)
     {
+<<<<<<< HEAD
       plugin->start_read_committed (plugin->cls,
                                     "test_by_j");
 
+=======
+       plugin->start (plugin->cls,
+         "test_by_exchange_j");
+>>>>>>> 26922c6d (batch modifications)
       for (unsigned int k = 0; k<batch_size; k++)
       {
         RND_BLK (&reserves[k].reserve_pub);
@@ -120,6 +139,7 @@ run (void *cls)
         reserves[k].sender_account_details = sndr;
         reserves[k].exchange_account_name = "name";
         reserves[k].wire_reference = k;
+<<<<<<< HEAD
 
       }
       FAILIF (batch_size !=
@@ -129,13 +149,42 @@ run (void *cls)
                                                 results));
 
       plugin->commit (plugin->cls);
+=======
+      }
+      FAILIF (batch_size !=
+            plugin->batch_reserves_in_insert (plugin->cls,
+                                              reserves,
+                                              batch_size,
+                                              results));
+      /*plugin->commit (plugin->cls);*/
+>>>>>>> 26922c6d (batch modifications)
     }
+    /*
+    for (unsigned int s=0;s<10;s++)
+    {
+      for (unsigned int k = 0; k<batch_size; k++)
+      {
+        RND_BLK (&reserves2[k].reserve_pub);
+        reserves2[k].balance = value;
+        reserves2[k].execution_time = ts;
+        reserves2[k].sender_account_details = sndr;
+        reserves2[k].exchange_account_name = "name";
+        reserves2[k].wire_reference = k;
+      }
+      FAILIF (batch_size !=
+            plugin->batch_reserves_in_insert (plugin->cls,
+                                              reserves2,
+                                              batch_size,
+                                              results));
+                                              }*/
+
     duration = GNUNET_TIME_absolute_get_duration (now);
     fprintf (stdout,
              "for a batchsize equal to %d it took %s\n",
              batch_size,
              GNUNET_STRINGS_relative_time_to_string (duration,
                                                      GNUNET_NO) );
+
   }
   result = 0;
 drop:
@@ -155,7 +204,6 @@ main (int argc,
   char *config_filename;
   char *testname;
   struct GNUNET_CONFIGURATION_Handle *cfg;
-
   (void) argc;
   result = -1;
   if (NULL == (plugin_name = strrchr (argv[0], (int) '-')))
@@ -163,6 +211,7 @@ main (int argc,
     GNUNET_break (0);
     return -1;
   }
+
   GNUNET_log_setup (argv[0],
                     "WARNING",
                     NULL);