pool: un-batch payouts

Send payments one at a time

src/pool.c

@@ -1917,8 +1917,8 @@ rpc_on_wallet_transferred(const char* data, rpc_callback_t *callback)
 
     int rc;
     char *err;
-    MDB_txn *txn;
-    MDB_cursor *cursor;
+    MDB_txn *txn = NULL;
+    MDB_cursor *cursor = NULL;
 
     /* First, updated balance(s) */
     if ((rc = mdb_txn_begin(env, NULL, 0, &txn)) != 0)
@@ -1935,22 +1935,23 @@ rpc_on_wallet_transferred(const char* data, rpc_callback_t *callback)
         goto cleanup;
     }
     payment_t *payment = (payment_t*) callback->data;
-    for (; payment->amount; payment++)
+    if (payment)
     {
         MDB_cursor_op op = MDB_SET;
         MDB_val key = {ADDRESS_MAX, (void*)payment->address};
         MDB_val val;
+
         rc = mdb_cursor_get(cursor, &key, &val, op);
         if (rc == MDB_NOTFOUND)
         {
             log_error("Payment made to non-existent address");
-            continue;
+            goto skip_balance;
         }
         else if (rc != 0 && rc != MDB_NOTFOUND)
         {
             err = mdb_strerror(rc);
             log_error("%s", err);
-            continue;
+            goto skip_balance;
         }
         uint64_t current_amount = *(uint64_t*)val.mv_data;
 
@@ -1978,6 +1979,8 @@ rpc_on_wallet_transferred(const char* data, rpc_callback_t *callback)
             log_error("%s", err);
         }
     }
+
+skip_balance:
     if ((rc = mdb_txn_commit(txn)) != 0)
     {
         err = mdb_strerror(rc);
@@ -2001,8 +2004,8 @@ rpc_on_wallet_transferred(const char* data, rpc_callback_t *callback)
         goto cleanup;
     }
     time_t now = time(NULL);
-    payment = (payment_t*) callback->data;
-    for (; payment->amount; payment++)
+
+    if (payment)
     {
         payment->timestamp = now;
         MDB_val key = {ADDRESS_MAX, (void*)payment->address};
@@ -2011,7 +2014,6 @@ rpc_on_wallet_transferred(const char* data, rpc_callback_t *callback)
         {
             err = mdb_strerror(rc);
             log_error("Error putting payment: %s", err);
-            continue;
         }
     }
     if ((rc = mdb_txn_commit(txn)) != 0)
@@ -2050,13 +2052,6 @@ send_payments(void)
         return rc;
     }
 
-    size_t payments_count = 0;
-    size_t payments_max_count = 25;
-    size_t payments_size = payments_max_count * sizeof(payment_t);
-    payment_t *payments = (payment_t*) calloc(1, payments_size);
-    payment_t *payment = payments;
-    payment_t *end_payment = payment + payments_max_count;
-
     MDB_cursor_op op = MDB_FIRST;
     while (1)
     {
@@ -2075,52 +2070,33 @@ send_payments(void)
 
         log_info("Sending payment: %"PRIu64", %.8s", amount, address);
 
-        strncpy(payment->address, address, ADDRESS_MAX-1);
-        payment->amount = amount;
-        payments_count++;
+        // free'd in rpc_callback_free
+        payment_t *payment = (payment_t*) calloc(1, sizeof (payment_t));
 
-        if (++payment == end_payment)
-        {
-            payments_size <<= 1;
-            payments = (payment_t*) realloc(payments, payments_size);
-            payment = payments + payments_max_count;
-            memset(payment, 0, sizeof(payment_t) * payments_max_count);
-            payments_max_count <<= 1;
-            end_payment = payments + payments_max_count;
+        if (!payment) {
+            log_error("send_payments: OOM: alloc payments");
+            return -1;
         }
-    }
-    mdb_cursor_close(cursor);
-    mdb_txn_abort(txn);
 
-    if (payments_count)
-    {
-        size_t body_size = 160 * payments_count + 128;
-        char body[body_size];
-        char *start = body;
-        char *end = body + body_size;
-        start = stecpy(start, "{\"id\":\"0\",\"jsonrpc\":\"2.0\",\"method\":"
+        strncpy(payment->address, address, ADDRESS_MAX-1);
+        payment->amount = amount;
+
+        char body[160+128];
+        const size_t body_size = sizeof(body);
+        snprintf(body, body_size,
+                "{\"id\":\"0\",\"jsonrpc\":\"2.0\",\"method\":"
                 "\"transfer_split\",\"params\":{"
-                "\"ring_size\":11,\"destinations\":[", end);
-        for (size_t i=0; i<payments_count; i++)
-        {
-            payment_t *p = &payments[i];
-            start = stecpy(start, "{\"address\":\"", end);
-            start = stecpy(start, p->address, end);
-            start = stecpy(start, "\",\"amount\":", end);
-            sprintf(start, "%"PRIu64"}", p->amount);
-            start = body + strlen(body);
-            if (i != payments_count -1)
-                start = stecpy(start, ",", end);
-            else
-                start = stecpy(start, "]}}", end);
-        }
+                "\"destinations\":["
+                "{\"address\":\"%s\",\"amount\":""%"PRIu64"}"
+                "]}}",
+                payment->address, payment->amount);
         log_trace(body);
         rpc_callback_t *cb = rpc_callback_new(
-                rpc_on_wallet_transferred, payments);
+                rpc_on_wallet_transferred, payment);
         rpc_wallet_request(pool_base, body, cb);
     }
-    else
-        free(payments);
+    mdb_cursor_close(cursor);
+    mdb_txn_abort(txn);
 
     return 0;
 }