taler/exchange
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

address 'global_ret' simplification TODO

Browse Source
This commit is contained in:
Christian Grothoff 2016-04-06 10:34:42 +02:00
parent ad8351c912
commit 4977a3eb8f
1 changed files with 69 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

116
src/exchange/taler-exchange-aggregator.c
View File

@ -20,7 +20,6 @@
* @author Christian Grothoff * @author Christian Grothoff
* *
* TODO: * TODO:
* - simplify global_ret: make it a global!
* - handle shutdown more nicely (call 'cancel' method on wire transfers) * - handle shutdown more nicely (call 'cancel' method on wire transfers)
*/ */
#include "platform.h" #include "platform.h"
@ -67,6 +66,12 @@ static struct TALER_WIRE_Plugin *wire_plugin;
*/ */
static struct GNUNET_SCHEDULER_Task *task; static struct GNUNET_SCHEDULER_Task *task;
/**
* Value to return from main(). #GNUNET_OK on success, #GNUNET_SYSERR
* on serious errors.
*/
static int global_ret;
/** /**
* #GNUNET_YES if we are in test mode and are using temporary tables. * #GNUNET_YES if we are in test mode and are using temporary tables.
*/ */
@ -84,6 +89,25 @@ static int test_mode;
static unsigned int aggregation_limit = TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT; static unsigned int aggregation_limit = TALER_EXCHANGEDB_MATCHING_DEPOSITS_LIMIT;
/**
* We're being aborted with CTRL-C (or SIGTERM). Shut down.
*
* @param cls closure
* @param tc scheduler context
*/
static void
shutdown_task (void *cls,
const struct GNUNET_SCHEDULER_TaskContext *tc)
{
if (NULL != task)
{
GNUNET_SCHEDULER_cancel (task);
task = NULL;
}
/* FIXME: other shutdown stuff here! */
}
/** /**
* Load configuration parameters for the exchange * Load configuration parameters for the exchange
* server into the corresponding global variables. * server into the corresponding global variables.
@ -216,11 +240,6 @@ struct AggregationUnit
*/ */
unsigned long long *additional_rows; unsigned long long *additional_rows;
/**
* Pointer to global return value. Closure for #run().
*/
int *global_ret;
/** /**
* Offset specifying how many #additional_rows are in use. * Offset specifying how many #additional_rows are in use.
*/ */
@ -314,7 +333,6 @@ deposit_cb (void *cls,
} }
/** /**
* Function called with details about another deposit we * Function called with details about another deposit we
* can aggregate into an existing aggregation unit. * can aggregate into an existing aggregation unit.
@ -433,14 +451,13 @@ prepare_cb (void *cls,
* Main work function that queries the DB and aggregates transactions * Main work function that queries the DB and aggregates transactions
* into larger wire transfers. * into larger wire transfers.
* *
* @param cls pointer to an `int` which we will return from main() * @param cls NULL
* @param tc scheduler context * @param tc scheduler context
*/ */
static void static void
run_aggregation (void *cls, run_aggregation (void *cls,
const struct GNUNET_SCHEDULER_TaskContext *tc) const struct GNUNET_SCHEDULER_TaskContext *tc)
{ {
int *global_ret = cls;
struct TALER_EXCHANGEDB_Session *session; struct TALER_EXCHANGEDB_Session *session;
struct AggregationUnit *au; struct AggregationUnit *au;
unsigned int i; unsigned int i;
@ -453,7 +470,7 @@ run_aggregation (void *cls,
{ {
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to obtain database session!\n"); "Failed to obtain database session!\n");
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
return; return;
} }
if (GNUNET_OK != if (GNUNET_OK !=
@ -462,7 +479,7 @@ run_aggregation (void *cls,
{ {
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to start database transaction!\n"); "Failed to start database transaction!\n");
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
return; return;
} }
au = GNUNET_new (struct AggregationUnit); au = GNUNET_new (struct AggregationUnit);
@ -482,7 +499,7 @@ run_aggregation (void *cls,
{ {
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to execute deposit iteration!\n"); "Failed to execute deposit iteration!\n");
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
return; return;
} }
if (GNUNET_YES == test_mode) if (GNUNET_YES == test_mode)
@ -495,7 +512,7 @@ run_aggregation (void *cls,
/* nothing to do, sleep for a minute and try again */ /* nothing to do, sleep for a minute and try again */
task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES, task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
&run_aggregation, &run_aggregation,
global_ret); NULL);
} }
return; return;
} }
@ -518,7 +535,7 @@ run_aggregation (void *cls,
GNUNET_free (au); GNUNET_free (au);
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
session); session);
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
return; return;
} }
/* Round to the unit supported by the wire transfer method */ /* Round to the unit supported by the wire transfer method */
@ -541,7 +558,7 @@ run_aggregation (void *cls,
{ {
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to start database transaction!\n"); "Failed to start database transaction!\n");
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
GNUNET_free_non_null (au->additional_rows); GNUNET_free_non_null (au->additional_rows);
if (NULL != au->wire) if (NULL != au->wire)
json_decref (au->wire); json_decref (au->wire);
@ -576,10 +593,9 @@ run_aggregation (void *cls,
GNUNET_free (au); GNUNET_free (au);
/* start again */ /* start again */
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
global_ret); NULL);
return; return;
} }
au->global_ret = global_ret;
au->ph = wire_plugin->prepare_wire_transfer (wire_plugin->cls, au->ph = wire_plugin->prepare_wire_transfer (wire_plugin->cls,
au->wire, au->wire,
&au->total_amount, &au->total_amount,
@ -600,7 +616,7 @@ run_aggregation (void *cls,
GNUNET_free (au); GNUNET_free (au);
/* start again */ /* start again */
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
global_ret); NULL);
return; return;
} }
/* otherwise we continue with #prepare_cb(), see below */ /* otherwise we continue with #prepare_cb(), see below */
@ -632,7 +648,6 @@ prepare_cb (void *cls,
size_t buf_size) size_t buf_size)
{ {
struct AggregationUnit *au = cls; struct AggregationUnit *au = cls;
int *global_ret = au->global_ret;
struct TALER_EXCHANGEDB_Session *session = au->session; struct TALER_EXCHANGEDB_Session *session = au->session;
GNUNET_free_non_null (au->additional_rows); GNUNET_free_non_null (au->additional_rows);
@ -646,7 +661,7 @@ prepare_cb (void *cls,
session); session);
/* start again */ /* start again */
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
global_ret); NULL);
return; return;
} }
@ -663,7 +678,7 @@ prepare_cb (void *cls,
session); session);
/* start again */ /* start again */
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
global_ret); NULL);
return; return;
} }
@ -677,13 +692,13 @@ prepare_cb (void *cls,
"Failed to commit database transaction!\n"); "Failed to commit database transaction!\n");
/* try again */ /* try again */
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
global_ret); NULL);
return; return;
} }
/* run alternative task: actually do wire transfer! */ /* run alternative task: actually do wire transfer! */
task = GNUNET_SCHEDULER_add_now (&run_transfers, task = GNUNET_SCHEDULER_add_now (&run_transfers,
&global_ret); NULL);
} }
@ -703,12 +718,6 @@ struct WirePrepareData
*/ */
struct TALER_WIRE_ExecuteHandle *eh; struct TALER_WIRE_ExecuteHandle *eh;
/**
* Pointer to global return value. Closure for #run().
*/
int *global_ret;
/** /**
* Row ID of the transfer. * Row ID of the transfer.
*/ */
@ -730,7 +739,6 @@ wire_confirm_cb (void *cls,
const char *emsg) const char *emsg)
{ {
struct WirePrepareData *wpd = cls; struct WirePrepareData *wpd = cls;
int *global_ret = wpd->global_ret;
struct TALER_EXCHANGEDB_Session *session = wpd->session; struct TALER_EXCHANGEDB_Session *session = wpd->session;
wpd->eh = NULL; wpd->eh = NULL;
@ -741,7 +749,7 @@ wire_confirm_cb (void *cls,
emsg); emsg);
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
session); session);
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
GNUNET_free (wpd); GNUNET_free (wpd);
return; return;
} }
@ -753,7 +761,7 @@ wire_confirm_cb (void *cls,
GNUNET_break (0); /* why!? */ GNUNET_break (0); /* why!? */
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
session); session);
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
GNUNET_free (wpd); GNUNET_free (wpd);
return; return;
} }
@ -766,13 +774,13 @@ wire_confirm_cb (void *cls,
"Failed to commit database transaction!\n"); "Failed to commit database transaction!\n");
/* try again */ /* try again */
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
global_ret); NULL);
return; return;
} }
/* continue with #run_transfers(), just to guard /* continue with #run_transfers(), just to guard
against the unlikely case that there are more. */ against the unlikely case that there are more. */
task = GNUNET_SCHEDULER_add_now (&run_transfers, task = GNUNET_SCHEDULER_add_now (&run_transfers,
&global_ret); NULL);
} }
@ -792,7 +800,6 @@ wire_prepare_cb (void *cls,
size_t buf_size) size_t buf_size)
{ {
struct WirePrepareData *wpd = cls; struct WirePrepareData *wpd = cls;
int *global_ret = wpd->global_ret;
wpd->row_id = rowid; wpd->row_id = rowid;
wpd->eh = wire_plugin->execute_wire_transfer (wire_plugin->cls, wpd->eh = wire_plugin->execute_wire_transfer (wire_plugin->cls,
@ -808,7 +815,7 @@ wire_prepare_cb (void *cls,
GNUNET_break (0); /* why? how to best recover? */ GNUNET_break (0); /* why? how to best recover? */
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
wpd->session); wpd->session);
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
GNUNET_free (wpd); GNUNET_free (wpd);
return; return;
} }
@ -819,14 +826,13 @@ wire_prepare_cb (void *cls,
* Execute the wire transfers that we have committed to * Execute the wire transfers that we have committed to
* do. * do.
* *
* @param cls pointer to an `int` which we will return from main() * @param cls NULL
* @param tc scheduler context * @param tc scheduler context
*/ */
static void static void
run_transfers (void *cls, run_transfers (void *cls,
const struct GNUNET_SCHEDULER_TaskContext *tc) const struct GNUNET_SCHEDULER_TaskContext *tc)
{ {
int *global_ret = cls;
int ret; int ret;
struct WirePrepareData *wpd; struct WirePrepareData *wpd;
struct TALER_EXCHANGEDB_Session *session; struct TALER_EXCHANGEDB_Session *session;
@ -838,7 +844,7 @@ run_transfers (void *cls,
{ {
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to obtain database session!\n"); "Failed to obtain database session!\n");
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
return; return;
} }
if (GNUNET_OK != if (GNUNET_OK !=
@ -847,12 +853,11 @@ run_transfers (void *cls,
{ {
GNUNET_log (GNUNET_ERROR_TYPE_ERROR, GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
"Failed to start database transaction!\n"); "Failed to start database transaction!\n");
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
return; return;
} }
wpd = GNUNET_new (struct WirePrepareData); wpd = GNUNET_new (struct WirePrepareData);
wpd->session = session; wpd->session = session;
wpd->global_ret = global_ret;
ret = db_plugin->wire_prepare_data_get (db_plugin->cls, ret = db_plugin->wire_prepare_data_get (db_plugin->cls,
session, session,
exchange_wireformat, exchange_wireformat,
@ -863,7 +868,7 @@ run_transfers (void *cls,
GNUNET_break (0); /* why? how to best recover? */ GNUNET_break (0); /* why? how to best recover? */
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
session); session);
*global_ret = GNUNET_SYSERR; global_ret = GNUNET_SYSERR;
GNUNET_free (wpd); GNUNET_free (wpd);
return; return;
} }
@ -873,7 +878,7 @@ run_transfers (void *cls,
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
session); session);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
global_ret); NULL);
GNUNET_free (wpd); GNUNET_free (wpd);
return; return;
} }
@ -881,6 +886,24 @@ run_transfers (void *cls,
} }
/**
* First task.
*
* @param cls closure, NULL
* @param tc scheduler context
*/
static void
run (void *cls,
const struct GNUNET_SCHEDULER_TaskContext *tc)
{
task = GNUNET_SCHEDULER_add_now (&run_transfers,
cls);
GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL,
&shutdown_task,
cls);
}
/** /**
* The main function of the taler-exchange-httpd server ("the exchange"). * The main function of the taler-exchange-httpd server ("the exchange").
* *
@ -907,7 +930,6 @@ main (int argc,
GNUNET_GETOPT_OPTION_VERSION (VERSION "-" VCS_VERSION), GNUNET_GETOPT_OPTION_VERSION (VERSION "-" VCS_VERSION),
GNUNET_GETOPT_OPTION_END GNUNET_GETOPT_OPTION_END
}; };
int ret = GNUNET_OK;
GNUNET_assert (GNUNET_OK == GNUNET_assert (GNUNET_OK ==
GNUNET_log_setup ("taler-exchange-aggregator", GNUNET_log_setup ("taler-exchange-aggregator",
@ -929,12 +951,12 @@ main (int argc,
{ {
return 1; return 1;
} }
global_ret = GNUNET_OK;
GNUNET_SCHEDULER_run (&run_transfers, &ret); GNUNET_SCHEDULER_run (&run, NULL);
TALER_EXCHANGEDB_plugin_unload (db_plugin); TALER_EXCHANGEDB_plugin_unload (db_plugin);
TALER_WIRE_plugin_unload (wire_plugin); TALER_WIRE_plugin_unload (wire_plugin);
return (GNUNET_SYSERR == ret) ? 1 : 0; return (GNUNET_SYSERR == global_ret) ? 1 : 0;
} }
/* end of taler-exchange-aggregator.c */ /* end of taler-exchange-aggregator.c */