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

fix double continuation scheduling

Browse Source
This commit is contained in:
Christian Grothoff 2020-02-04 00:04:29 +01:00
parent 232606fb57
commit d47241e0e1
No known key found for this signature in database
GPG Key ID: 939E6BE1E29FC3CC
1 changed files with 26 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1183,6 +1183,8 @@ expired_reserve_cb (void *cls,
GNUNET_log (GNUNET_ERROR_TYPE_INFO, GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Reserve was virtually empty, moving on\n"); "Reserve was virtually empty, moving on\n");
(void) commit_or_warn (session); (void) commit_or_warn (session);
erc->async_cont = GNUNET_YES;
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_transfers, task = GNUNET_SCHEDULER_add_now (&run_transfers,
NULL); NULL);
return qs; return qs;
@ -1223,6 +1225,7 @@ expired_reserve_cb (void *cls,
return GNUNET_DB_STATUS_SUCCESS_NO_RESULTS; return GNUNET_DB_STATUS_SUCCESS_NO_RESULTS;
} }
erc->async_cont = GNUNET_YES; erc->async_cont = GNUNET_YES;
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_transfers, task = GNUNET_SCHEDULER_add_now (&run_transfers,
NULL); NULL);
GNUNET_free (ctc->method); GNUNET_free (ctc->method);
@ -1299,6 +1302,7 @@ run_reserve_closures (void *cls)
case GNUNET_DB_STATUS_SOFT_ERROR: case GNUNET_DB_STATUS_SOFT_ERROR:
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
session); session);
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_reserve_closures, task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
NULL); NULL);
return; return;
@ -1308,6 +1312,7 @@ run_reserve_closures (void *cls)
reserves_idle = GNUNET_YES; reserves_idle = GNUNET_YES;
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
session); session);
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
return; return;
@ -1315,6 +1320,7 @@ run_reserve_closures (void *cls)
(void) commit_or_warn (session); (void) commit_or_warn (session);
if (GNUNET_YES == erc.async_cont) if (GNUNET_YES == erc.async_cont)
break; break;
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_reserve_closures, task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
NULL); NULL);
return; return;
@ -1345,6 +1351,7 @@ run_aggregation (void *cls)
return; return;
if (0 == (++swap % 2)) if (0 == (++swap % 2))
{ {
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_reserve_closures, task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
NULL); NULL);
return; return;
@ -1392,6 +1399,7 @@ run_aggregation (void *cls)
{ {
/* should re-try immediately */ /* should re-try immediately */
swap--; /* do not count failed attempts */ swap--; /* do not count failed attempts */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
return; return;
@ -1408,15 +1416,21 @@ run_aggregation (void *cls)
{ {
if ( (GNUNET_NO == reserves_idle) || if ( (GNUNET_NO == reserves_idle) ||
(GNUNET_YES == test_mode) ) (GNUNET_YES == test_mode) )
{
/* Possibly more to on reserves, go for it immediately */ /* Possibly more to on reserves, go for it immediately */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_reserve_closures, task = GNUNET_SCHEDULER_add_now (&run_reserve_closures,
NULL); NULL);
}
else else
{
/* FIXME(dold): We might want to read the duration to sleep from the config */ /* FIXME(dold): We might want to read the duration to sleep from the config */
/* nothing to do, sleep for a minute and try again */ /* nothing to do, sleep for a minute and try again */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES, task = GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_MINUTES,
&run_aggregation, &run_aggregation,
NULL); NULL);
}
} }
return; return;
} }
@ -1451,6 +1465,7 @@ run_aggregation (void *cls)
"Serialization issue, trying again later!\n"); "Serialization issue, trying again later!\n");
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
session); session);
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
return; return;
@ -1515,6 +1530,7 @@ run_aggregation (void *cls)
session); session);
cleanup_au (); cleanup_au ();
/* start again */ /* start again */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
return; return;
@ -1531,6 +1547,7 @@ run_aggregation (void *cls)
(void) commit_or_warn (session); (void) commit_or_warn (session);
cleanup_au (); cleanup_au ();
/* start again */ /* start again */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
return; return;
@ -1587,6 +1604,7 @@ run_aggregation (void *cls)
db_plugin->rollback (db_plugin->cls, db_plugin->rollback (db_plugin->cls,
session); session);
/* start again */ /* start again */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
return; return;
@ -1613,6 +1631,7 @@ run_aggregation (void *cls)
/* try again */ /* try again */
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Commit issue for prepared wire data; trying again later!\n"); "Commit issue for prepared wire data; trying again later!\n");
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
return; return;
@ -1625,6 +1644,7 @@ run_aggregation (void *cls)
GNUNET_log (GNUNET_ERROR_TYPE_INFO, GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"Preparation complete, switching to transfer mode\n"); "Preparation complete, switching to transfer mode\n");
/* run alternative task: actually do wire transfer! */ /* run alternative task: actually do wire transfer! */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_transfers, task = GNUNET_SCHEDULER_add_now (&run_transfers,
NULL); NULL);
return; return;
@ -1685,6 +1705,7 @@ wire_confirm_cb (void *cls,
if (GNUNET_DB_STATUS_SOFT_ERROR == qs) if (GNUNET_DB_STATUS_SOFT_ERROR == qs)
{ {
/* try again */ /* try again */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
} }
@ -1703,6 +1724,7 @@ wire_confirm_cb (void *cls,
{ {
case GNUNET_DB_STATUS_SOFT_ERROR: case GNUNET_DB_STATUS_SOFT_ERROR:
/* try again */ /* try again */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
return; return;
@ -1716,6 +1738,7 @@ wire_confirm_cb (void *cls,
"Wire transfer complete\n"); "Wire transfer complete\n");
/* 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. */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_transfers, task = GNUNET_SCHEDULER_add_now (&run_transfers,
NULL); NULL);
return; return;
@ -1846,6 +1869,7 @@ run_transfers (void *cls)
return; return;
case GNUNET_DB_STATUS_SOFT_ERROR: case GNUNET_DB_STATUS_SOFT_ERROR:
/* try again */ /* try again */
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_transfers, task = GNUNET_SCHEDULER_add_now (&run_transfers,
NULL); NULL);
return; return;
@ -1853,6 +1877,7 @@ run_transfers (void *cls)
/* no more prepared wire transfers, go back to aggregation! */ /* no more prepared wire transfers, go back to aggregation! */
GNUNET_log (GNUNET_ERROR_TYPE_INFO, GNUNET_log (GNUNET_ERROR_TYPE_INFO,
"No more pending wire transfers, starting aggregation\n"); "No more pending wire transfers, starting aggregation\n");
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_aggregation, task = GNUNET_SCHEDULER_add_now (&run_aggregation,
NULL); NULL);
return; return;
@ -1898,6 +1923,7 @@ run (void *cls,
return; return;
} }
GNUNET_assert (NULL == task);
task = GNUNET_SCHEDULER_add_now (&run_transfers, task = GNUNET_SCHEDULER_add_now (&run_transfers,
NULL); NULL);
GNUNET_SCHEDULER_add_shutdown (&shutdown_task, GNUNET_SCHEDULER_add_shutdown (&shutdown_task,