taler-exchange-transfer.c (26578B)
1 /* 2 This file is part of TALER 3 Copyright (C) 2016-2021 Taler Systems SA 4 5 TALER is free software; you can redistribute it and/or modify it under the 6 terms of the GNU Affero General Public License as published by the Free Software 7 Foundation; either version 3, or (at your option) any later version. 8 9 TALER is distributed in the hope that it will be useful, but WITHOUT ANY 10 WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR 11 A PARTICULAR PURPOSE. See the GNU Affero General Public License for more details. 12 13 You should have received a copy of the GNU Affero General Public License along with 14 TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/> 15 */ 16 /** 17 * @file taler-exchange-transfer.c 18 * @brief Process that actually finalizes outgoing transfers with the wire gateway / bank 19 * @author Christian Grothoff 20 */ 21 #include "platform.h" 22 #include <gnunet/gnunet_util_lib.h> 23 #include <jansson.h> 24 #include <pthread.h> 25 #include "exchangedb_lib.h" 26 #include "taler/taler_json_lib.h" 27 #include "taler/taler_bank_service.h" 28 #include "exchange-database/begin_shard.h" 29 #include "exchange-database/update_shard_progress.h" 30 #include "exchange-database/commit.h" 31 #include "exchange-database/preflight.h" 32 #include "exchange-database/insert_sanction_list_hit.h" 33 #include "exchange-database/start_read_committed.h" 34 #include "exchange-database/rollback.h" 35 #include "exchange-database/update_to_prewire_finished.h" 36 #include "exchange-database/update_to_prewire_failed.h" 37 #include "exchange-database/iterate_prewires.h" 38 #include "exchange-database/event_listen.h" 39 #include "exchange-database/event_listen_cancel.h" 40 #include "exchange-database/update_to_prewire_finished.h" 41 42 43 /** 44 * What is the default batch size we use for credit history 45 * requests with the bank. See `batch_size` below. 46 */ 47 #define DEFAULT_BATCH_SIZE 32 48 49 /** 50 * How often will we retry a request (given certain 51 * HTTP status codes) before giving up? 52 */ 53 #define MAX_RETRIES 3 54 55 /** 56 * Information about our work shard. 57 */ 58 struct Shard 59 { 60 61 /** 62 * Time when we started to work on this shard. 63 */ 64 struct GNUNET_TIME_Absolute shard_start_time; 65 66 /** 67 * Offset the shard begins at. 68 */ 69 uint64_t shard_start; 70 71 /** 72 * Exclusive offset where the shard ends. 73 */ 74 uint64_t shard_end; 75 76 /** 77 * Offset where our current batch begins. 78 */ 79 uint64_t batch_start; 80 81 /** 82 * Highest row processed in the current batch. 83 */ 84 uint64_t batch_end; 85 86 /** 87 * For how long we hold this shard. Renewed whenever we record progress, 88 * so that a worker that is slowly but steadily working through a shard 89 * keeps it. 90 */ 91 struct GNUNET_TIME_Relative lease; 92 93 }; 94 95 96 /** 97 * Data we keep to #run_transfers(). There is at most 98 * one of these around at any given point in time. 99 * Note that this limits parallelism, and we might want 100 * to revise this decision at a later point. 101 */ 102 struct WirePrepareData 103 { 104 105 /** 106 * All transfers done in the same transaction 107 * are kept in a DLL. 108 */ 109 struct WirePrepareData *next; 110 111 /** 112 * All transfers done in the same transaction 113 * are kept in a DLL. 114 */ 115 struct WirePrepareData *prev; 116 117 /** 118 * Wire execution handle. 119 */ 120 struct TALER_BANK_TransferHandle *eh; 121 122 /** 123 * Wire account used for this preparation. 124 */ 125 const struct TALER_EXCHANGEDB_AccountInfo *wa; 126 127 /** 128 * Row ID of the transfer. 129 */ 130 unsigned long long row_id; 131 132 /** 133 * Number of bytes allocated after this struct 134 * with the prewire data. 135 */ 136 size_t buf_size; 137 138 /** 139 * How often did we retry so far? 140 */ 141 unsigned int retries; 142 143 }; 144 145 146 /** 147 * The exchange's configuration. 148 */ 149 static const struct GNUNET_CONFIGURATION_Handle *cfg; 150 151 /** 152 * Our database plugin. 153 */ 154 static struct TALER_EXCHANGEDB_PostgresContext *pg; 155 156 /** 157 * Next task to run, if any. 158 */ 159 static struct GNUNET_SCHEDULER_Task *task; 160 161 /** 162 * If we are currently executing transfers, information about 163 * the active transfers is here. Otherwise, this variable is NULL. 164 */ 165 static struct WirePrepareData *wpd_head; 166 167 /** 168 * If we are currently executing transfers, information about 169 * the active transfers is here. Otherwise, this variable is NULL. 170 */ 171 static struct WirePrepareData *wpd_tail; 172 173 /** 174 * Information about our work shard. 175 */ 176 static struct Shard *shard; 177 178 /** 179 * Handle to the context for interacting with the bank / wire gateway. 180 */ 181 static struct GNUNET_CURL_Context *ctx; 182 183 /** 184 * Randomized back-off we use on serialization errors. 185 */ 186 static struct GNUNET_TIME_Relative serialization_delay; 187 188 /** 189 * Scheduler context for running the @e ctx. 190 */ 191 static struct GNUNET_CURL_RescheduleContext *rc; 192 193 /** 194 * Value to return from main(). 0 on success, non-zero on errors. 195 */ 196 static int global_ret; 197 198 /** 199 * #GNUNET_YES if we are in test mode and should exit when idle. 200 */ 201 static int test_mode; 202 203 /** 204 * How long should we sleep when idle before trying to find more work? 205 * Also used for how long we wait to grab a shard before trying it again. 206 * The value should be set to a bit above the average time it takes to 207 * process a shard. 208 */ 209 static struct GNUNET_TIME_Relative transfer_idle_sleep_interval; 210 211 /** 212 * How long did we take to finish the last shard? 213 */ 214 static struct GNUNET_TIME_Relative shard_delay; 215 216 /** 217 * Size of the shards. 218 */ 219 static unsigned int shard_size = DEFAULT_BATCH_SIZE; 220 221 /** 222 * How many workers should we plan our scheduling with? 223 */ 224 static unsigned int max_workers = 0; 225 226 227 /** 228 * Clean up all active bank interactions. 229 */ 230 static void 231 cleanup_wpd (void) 232 { 233 struct WirePrepareData *wpd; 234 235 while (NULL != (wpd = wpd_head)) 236 { 237 GNUNET_CONTAINER_DLL_remove (wpd_head, 238 wpd_tail, 239 wpd); 240 if (NULL != wpd->eh) 241 { 242 TALER_BANK_transfer_cancel (wpd->eh); 243 wpd->eh = NULL; 244 } 245 GNUNET_free (wpd); 246 } 247 } 248 249 250 /** 251 * We're being aborted with CTRL-C (or SIGTERM). Shut down. 252 * 253 * @param cls closure 254 */ 255 static void 256 shutdown_task (void *cls) 257 { 258 (void) cls; 259 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 260 "Running shutdown\n"); 261 if (NULL != task) 262 { 263 GNUNET_SCHEDULER_cancel (task); 264 task = NULL; 265 } 266 cleanup_wpd (); 267 GNUNET_free (shard); 268 TALER_EXCHANGEDB_rollback (pg); /* just in case */ 269 TALER_EXCHANGEDB_disconnect (pg); 270 pg = NULL; 271 TALER_EXCHANGEDB_unload_accounts (); 272 cfg = NULL; 273 if (NULL != ctx) 274 { 275 GNUNET_CURL_fini (ctx); 276 ctx = NULL; 277 } 278 if (NULL != rc) 279 { 280 GNUNET_CURL_gnunet_rc_destroy (rc); 281 rc = NULL; 282 } 283 } 284 285 286 /** 287 * Parse the configuration for taler-exchange-transfer. 288 * 289 * @return #GNUNET_OK on success 290 */ 291 static enum GNUNET_GenericReturnValue 292 parse_transfer_config (void) 293 { 294 if (GNUNET_OK != 295 GNUNET_CONFIGURATION_get_value_time (cfg, 296 "exchange", 297 "TRANSFER_IDLE_SLEEP_INTERVAL", 298 &transfer_idle_sleep_interval)) 299 { 300 GNUNET_log_config_missing (GNUNET_ERROR_TYPE_ERROR, 301 "exchange", 302 "TRANSFER_IDLE_SLEEP_INTERVAL"); 303 return GNUNET_SYSERR; 304 } 305 if (NULL == 306 (pg = TALER_EXCHANGEDB_connect (cfg))) 307 { 308 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 309 "Failed to initialize DB subsystem\n"); 310 return GNUNET_SYSERR; 311 } 312 if (GNUNET_OK != 313 TALER_EXCHANGEDB_load_accounts (cfg, 314 TALER_EXCHANGEDB_ALO_DEBIT 315 | TALER_EXCHANGEDB_ALO_AUTHDATA)) 316 { 317 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 318 "No wire accounts configured for debit!\n"); 319 TALER_EXCHANGEDB_disconnect (pg); 320 pg = NULL; 321 return GNUNET_SYSERR; 322 } 323 return GNUNET_OK; 324 } 325 326 327 /** 328 * Perform a database commit. If it fails, print a warning. 329 * 330 * @return status of commit 331 */ 332 static enum GNUNET_DB_QueryStatus 333 commit_or_warn (void) 334 { 335 enum GNUNET_DB_QueryStatus qs; 336 337 qs = TALER_EXCHANGEDB_commit (pg); 338 if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs) 339 { 340 serialization_delay = GNUNET_TIME_UNIT_ZERO; 341 return qs; 342 } 343 GNUNET_log ((GNUNET_DB_STATUS_SOFT_ERROR == qs) 344 ? GNUNET_ERROR_TYPE_INFO 345 : GNUNET_ERROR_TYPE_ERROR, 346 "Failed to commit database transaction!\n"); 347 return qs; 348 } 349 350 351 /** 352 * Execute the wire transfers that we have committed to 353 * do. 354 * 355 * @param cls NULL 356 */ 357 static void 358 run_transfers (void *cls); 359 360 361 static void 362 run_transfers_delayed (void *cls) 363 { 364 (void) cls; 365 shard->shard_start_time = GNUNET_TIME_absolute_get (); 366 run_transfers (NULL); 367 } 368 369 370 /** 371 * Select shard to process. 372 * 373 * @param cls NULL 374 */ 375 static void 376 select_shard (void *cls); 377 378 379 /** 380 * We are done with the current batch. Commit 381 * and move on. 382 */ 383 static void 384 batch_done (void) 385 { 386 enum GNUNET_DB_QueryStatus qs; 387 388 /* batch done */ 389 GNUNET_assert (NULL == wpd_head); 390 /* Rides along with the transaction that persists this batch, so the 391 transfers and the record of how far the shard has come become visible 392 together. Once batch_end+1 reaches the end of the shard this same 393 statement marks it completed, which is why there is no separate 394 "shard done" write any more. */ 395 qs = TALER_EXCHANGEDB_update_shard_progress (pg, 396 "transfer", 397 shard->shard_start, 398 shard->shard_end, 399 shard->batch_end + 1, 400 shard->lease); 401 if (0 > qs) 402 { 403 TALER_EXCHANGEDB_rollback (pg); 404 if (GNUNET_DB_STATUS_SOFT_ERROR == qs) 405 { 406 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 407 "Serialization failure recording shard progress, trying again immediately!\n"); 408 GNUNET_assert (NULL == task); 409 task = GNUNET_SCHEDULER_add_now (&run_transfers, 410 NULL); 411 return; 412 } 413 GNUNET_break (0); 414 global_ret = EXIT_FAILURE; 415 GNUNET_SCHEDULER_shutdown (); 416 return; 417 } 418 switch (commit_or_warn ()) 419 { 420 case GNUNET_DB_STATUS_SOFT_ERROR: 421 /* try again */ 422 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 423 "Serialization failure, trying again immediately!\n"); 424 GNUNET_assert (NULL == task); 425 task = GNUNET_SCHEDULER_add_now (&run_transfers, 426 NULL); 427 return; 428 case GNUNET_DB_STATUS_HARD_ERROR: 429 GNUNET_break (0); 430 global_ret = EXIT_FAILURE; 431 GNUNET_SCHEDULER_shutdown (); 432 return; 433 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 434 shard->batch_start = shard->batch_end + 1; 435 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 436 "Batch complete\n"); 437 /* continue with #run_transfers(), just to guard 438 against the unlikely case that there are more. */ 439 GNUNET_assert (NULL == task); 440 task = GNUNET_SCHEDULER_add_now (&run_transfers, 441 NULL); 442 return; 443 default: 444 GNUNET_break (0); 445 global_ret = EXIT_FAILURE; 446 GNUNET_SCHEDULER_shutdown (); 447 return; 448 } 449 } 450 451 452 /** 453 * Function called with the result from the execute step. 454 * On success, we mark the respective wire transfer as finished, 455 * and in general we afterwards continue to #run_transfers(), 456 * except for irrecoverable errors. 457 * 458 * @param cls `struct WirePrepareData` we are working on 459 * @param tr transfer response 460 */ 461 static void 462 wire_confirm_cb (void *cls, 463 const struct TALER_BANK_TransferResponse *tr) 464 { 465 struct WirePrepareData *wpd = cls; 466 enum GNUNET_DB_QueryStatus qs; 467 468 wpd->eh = NULL; 469 switch (tr->http_status) 470 { 471 case MHD_HTTP_OK: 472 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 473 "Wire transfer %llu completed successfully\n", 474 (unsigned long long) wpd->row_id); 475 qs = TALER_EXCHANGEDB_update_to_prewire_finished (pg, 476 wpd->row_id); 477 /* continued below */ 478 break; 479 case MHD_HTTP_NOT_FOUND: 480 case MHD_HTTP_CONFLICT: 481 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 482 "Wire transaction %llu failed: %u/%d\n", 483 (unsigned long long) wpd->row_id, 484 tr->http_status, 485 tr->ec); 486 qs = TALER_EXCHANGEDB_update_to_prewire_failed (pg, 487 wpd->row_id); 488 /* continued below */ 489 break; 490 case 0: 491 case MHD_HTTP_TOO_MANY_REQUESTS: 492 case MHD_HTTP_INTERNAL_SERVER_ERROR: 493 case MHD_HTTP_BAD_GATEWAY: 494 case MHD_HTTP_SERVICE_UNAVAILABLE: 495 case MHD_HTTP_GATEWAY_TIMEOUT: 496 wpd->retries++; 497 if (wpd->retries < MAX_RETRIES) 498 { 499 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 500 "Wire transfer %llu failed (%u), trying again\n", 501 (unsigned long long) wpd->row_id, 502 tr->http_status); 503 wpd->eh = TALER_BANK_transfer (ctx, 504 wpd->wa->auth, 505 &wpd[1], 506 wpd->buf_size, 507 &wire_confirm_cb, 508 wpd); 509 return; 510 } 511 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 512 "Wire transaction %llu failed: %u/%d\n", 513 (unsigned long long) wpd->row_id, 514 tr->http_status, 515 tr->ec); 516 cleanup_wpd (); 517 TALER_EXCHANGEDB_rollback (pg); 518 global_ret = EXIT_FAILURE; 519 GNUNET_SCHEDULER_shutdown (); 520 return; 521 default: 522 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 523 "Wire transfer %llu failed: %u/%d\n", 524 (unsigned long long) wpd->row_id, 525 tr->http_status, 526 tr->ec); 527 TALER_EXCHANGEDB_rollback (pg); 528 cleanup_wpd (); 529 global_ret = EXIT_FAILURE; 530 GNUNET_SCHEDULER_shutdown (); 531 return; 532 } 533 shard->batch_end = GNUNET_MAX (wpd->row_id, 534 shard->batch_end); 535 switch (qs) 536 { 537 case GNUNET_DB_STATUS_SOFT_ERROR: 538 TALER_EXCHANGEDB_rollback (pg); 539 cleanup_wpd (); 540 GNUNET_assert (NULL == task); 541 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 542 "Serialization failure, trying again immediately!\n"); 543 task = GNUNET_SCHEDULER_add_now (&run_transfers, 544 NULL); 545 return; 546 case GNUNET_DB_STATUS_HARD_ERROR: 547 TALER_EXCHANGEDB_rollback (pg); 548 cleanup_wpd (); 549 global_ret = EXIT_FAILURE; 550 GNUNET_SCHEDULER_shutdown (); 551 return; 552 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 553 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 554 GNUNET_CONTAINER_DLL_remove (wpd_head, 555 wpd_tail, 556 wpd); 557 GNUNET_free (wpd); 558 break; 559 } 560 if (NULL != wpd_head) 561 return; /* wait for other queries to complete */ 562 batch_done (); 563 } 564 565 566 /** 567 * Callback with data about a prepared transaction. Triggers the respective 568 * wire transfer using the prepared transaction data. 569 * 570 * @param cls NULL 571 * @param rowid row identifier used to mark prepared transaction as done 572 * @param wire_method wire method the preparation was done for 573 * @param buf transaction data that was persisted, NULL on error 574 * @param buf_size number of bytes in @a buf, 0 on error 575 */ 576 static void 577 wire_prepare_cb (void *cls, 578 uint64_t rowid, 579 const char *wire_method, 580 const char *buf, 581 size_t buf_size) 582 { 583 struct WirePrepareData *wpd; 584 585 (void) cls; 586 if ( (NULL != task) || 587 (EXIT_SUCCESS != global_ret) ) 588 return; /* current transaction was aborted */ 589 if (rowid >= shard->shard_end) 590 { 591 /* skip */ 592 shard->batch_end = shard->shard_end - 1; 593 if (NULL != wpd_head) 594 return; 595 batch_done (); 596 return; 597 } 598 if ( (NULL == wire_method) || 599 (NULL == buf) ) 600 { 601 GNUNET_break (0); 602 TALER_EXCHANGEDB_rollback (pg); 603 global_ret = EXIT_FAILURE; 604 GNUNET_SCHEDULER_shutdown (); 605 return; 606 } 607 wpd = GNUNET_malloc (sizeof (struct WirePrepareData) 608 + buf_size); 609 GNUNET_memcpy (&wpd[1], 610 buf, 611 buf_size); 612 wpd->buf_size = buf_size; 613 wpd->row_id = rowid; 614 GNUNET_CONTAINER_DLL_insert (wpd_head, 615 wpd_tail, 616 wpd); 617 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 618 "Starting wire transfer %llu\n", 619 (unsigned long long) rowid); 620 wpd->wa = TALER_EXCHANGEDB_find_account_by_method (wire_method); 621 if (NULL == wpd->wa) 622 { 623 /* Should really never happen here, as when we get 624 here the wire account should be in the cache. */ 625 GNUNET_break (0); 626 cleanup_wpd (); 627 TALER_EXCHANGEDB_rollback (pg); 628 global_ret = EXIT_NO_RESTART; 629 GNUNET_SCHEDULER_shutdown (); 630 return; 631 } 632 wpd->eh = TALER_BANK_transfer (ctx, 633 wpd->wa->auth, 634 buf, 635 buf_size, 636 &wire_confirm_cb, 637 wpd); 638 if (NULL == wpd->eh) 639 { 640 GNUNET_break (0); /* Irrecoverable */ 641 cleanup_wpd (); 642 TALER_EXCHANGEDB_rollback (pg); 643 global_ret = EXIT_FAILURE; 644 GNUNET_SCHEDULER_shutdown (); 645 return; 646 } 647 } 648 649 650 /** 651 * Execute the wire transfers that we have committed to 652 * do. 653 * 654 * @param cls NULL 655 */ 656 static void 657 run_transfers (void *cls) 658 { 659 enum GNUNET_DB_QueryStatus qs; 660 int64_t limit; 661 662 (void) cls; 663 task = NULL; 664 limit = shard->shard_end - shard->batch_start; 665 if (0 >= limit) 666 { 667 /* The last batch we committed already carried this shard's progress past 668 its end, and was therefore also what marked it completed. Nothing left 669 to write here. */ 670 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 671 "Shard [%llu,%llu) completed\n", 672 (unsigned long long) shard->shard_start, 673 (unsigned long long) shard->batch_end); 674 shard_delay = GNUNET_TIME_absolute_get_duration ( 675 shard->shard_start_time); 676 GNUNET_free (shard); 677 GNUNET_assert (NULL == task); 678 task = GNUNET_SCHEDULER_add_now (&select_shard, 679 NULL); 680 return; 681 } 682 /* cap number of parallel connections to a reasonable 683 limit for concurrent requests to the bank */ 684 limit = GNUNET_MIN (limit, 685 256); 686 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 687 "Checking for %lld pending wire transfers [%llu-...)\n", 688 (long long) limit, 689 (unsigned long long) shard->batch_start); 690 if (GNUNET_OK != 691 TALER_EXCHANGEDB_start_read_committed (pg, 692 "aggregator run transfer")) 693 { 694 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 695 "Failed to start database transaction!\n"); 696 global_ret = EXIT_FAILURE; 697 GNUNET_SCHEDULER_shutdown (); 698 return; 699 } 700 GNUNET_assert (NULL == task); 701 qs = TALER_EXCHANGEDB_iterate_prewires (pg, 702 shard->batch_start, 703 limit, 704 &wire_prepare_cb, 705 NULL); 706 switch (qs) 707 { 708 case GNUNET_DB_STATUS_HARD_ERROR: 709 cleanup_wpd (); 710 TALER_EXCHANGEDB_rollback (pg); 711 GNUNET_break (0); 712 global_ret = EXIT_FAILURE; 713 GNUNET_SCHEDULER_shutdown (); 714 return; 715 case GNUNET_DB_STATUS_SOFT_ERROR: 716 /* try again */ 717 TALER_EXCHANGEDB_rollback (pg); 718 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 719 "Serialization failure, trying again immediately!\n"); 720 cleanup_wpd (); 721 GNUNET_assert (NULL == task); 722 task = GNUNET_SCHEDULER_add_now (&run_transfers, 723 NULL); 724 return; 725 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 726 /* no more prepared wire transfers, go sleep a bit! */ 727 TALER_EXCHANGEDB_rollback (pg); 728 GNUNET_assert (NULL == wpd_head); 729 GNUNET_assert (NULL == task); 730 if (GNUNET_YES == test_mode) 731 { 732 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 733 "No more pending wire transfers, shutting down (because we are in test mode)\n"); 734 GNUNET_SCHEDULER_shutdown (); 735 } 736 else 737 { 738 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 739 "No more pending wire transfers, going idle\n"); 740 GNUNET_assert (NULL == task); 741 task = GNUNET_SCHEDULER_add_delayed (transfer_idle_sleep_interval, 742 &run_transfers_delayed, 743 NULL); 744 } 745 return; 746 default: 747 /* continued in wire_prepare_cb() */ 748 return; 749 } 750 } 751 752 753 /** 754 * Select shard to process. 755 * 756 * @param cls NULL 757 */ 758 static void 759 select_shard (void *cls) 760 { 761 enum GNUNET_DB_QueryStatus qs; 762 struct GNUNET_TIME_Relative delay; 763 uint64_t start; 764 uint64_t end; 765 uint64_t progress; 766 767 (void) cls; 768 task = NULL; 769 GNUNET_assert (NULL == wpd_head); 770 if (GNUNET_SYSERR == 771 TALER_EXCHANGEDB_preflight (pg)) 772 { 773 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 774 "Failed to obtain database connection!\n"); 775 global_ret = EXIT_FAILURE; 776 GNUNET_SCHEDULER_shutdown (); 777 return; 778 } 779 if (0 == max_workers) 780 delay = GNUNET_TIME_UNIT_ZERO; 781 else 782 delay.rel_value_us = GNUNET_CRYPTO_random_u64 ( 783 4 * GNUNET_TIME_relative_max ( 784 transfer_idle_sleep_interval, 785 GNUNET_TIME_relative_multiply (shard_delay, 786 max_workers)).rel_value_us); 787 qs = TALER_EXCHANGEDB_begin_shard (pg, 788 "transfer", 789 delay, 790 shard_size, 791 &start, 792 &end, 793 &progress); 794 switch (qs) 795 { 796 case GNUNET_DB_STATUS_HARD_ERROR: 797 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 798 "Failed to obtain starting point for monitoring from database!\n"); 799 global_ret = EXIT_FAILURE; 800 GNUNET_SCHEDULER_shutdown (); 801 return; 802 case GNUNET_DB_STATUS_SOFT_ERROR: 803 /* try again */ 804 { 805 serialization_delay = GNUNET_TIME_randomized_backoff (serialization_delay, 806 GNUNET_TIME_UNIT_SECONDS); 807 GNUNET_assert (NULL == task); 808 GNUNET_log (GNUNET_ERROR_TYPE_WARNING, 809 "Serialization failure, trying again in %s!\n", 810 GNUNET_TIME_relative2s (serialization_delay, 811 true)); 812 task = GNUNET_SCHEDULER_add_delayed (serialization_delay, 813 &select_shard, 814 NULL); 815 } 816 return; 817 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 818 GNUNET_break (0); 819 GNUNET_assert (NULL == task); 820 task = GNUNET_SCHEDULER_add_delayed (transfer_idle_sleep_interval, 821 &select_shard, 822 NULL); 823 return; 824 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 825 /* continued below */ 826 break; 827 } 828 GNUNET_log (GNUNET_ERROR_TYPE_INFO, 829 "Starting with shard [%llu,%llu)\n", 830 (unsigned long long) start, 831 (unsigned long long) end); 832 shard = GNUNET_new (struct Shard); 833 shard->shard_start_time = GNUNET_TIME_absolute_get (); 834 shard->shard_start = start; 835 shard->shard_end = end; 836 shard->lease = delay; 837 /* Resume where the shard says it got to, which is @a start for a fresh 838 shard and further along for one abandoned mid-way. */ 839 shard->batch_start = progress; 840 shard->batch_end = (0 == progress) ? 0 : progress - 1; 841 GNUNET_assert (NULL == task); 842 task = GNUNET_SCHEDULER_add_now (&run_transfers, 843 NULL); 844 } 845 846 847 /** 848 * First task. 849 * 850 * @param cls closure, NULL 851 * @param args remaining command-line arguments 852 * @param cfgfile name of the configuration file used (for saving, can be NULL!) 853 * @param c configuration 854 */ 855 static void 856 run (void *cls, 857 char *const *args, 858 const char *cfgfile, 859 const struct GNUNET_CONFIGURATION_Handle *c) 860 { 861 (void) cls; 862 (void) args; 863 (void) cfgfile; 864 865 cfg = c; 866 if (GNUNET_OK != parse_transfer_config ()) 867 { 868 cfg = NULL; 869 global_ret = EXIT_NOTCONFIGURED; 870 return; 871 } 872 ctx = GNUNET_CURL_init (&GNUNET_CURL_gnunet_scheduler_reschedule, 873 &rc); 874 rc = GNUNET_CURL_gnunet_rc_create (ctx); 875 if (NULL == ctx) 876 { 877 GNUNET_break (0); 878 return; 879 } 880 if (GNUNET_SYSERR == 881 TALER_EXCHANGEDB_preflight (pg)) 882 { 883 GNUNET_log (GNUNET_ERROR_TYPE_ERROR, 884 "Failed to obtain database connection!\n"); 885 global_ret = EXIT_FAILURE; 886 GNUNET_SCHEDULER_shutdown (); 887 return; 888 } 889 GNUNET_assert (NULL == task); 890 task = GNUNET_SCHEDULER_add_now (&select_shard, 891 NULL); 892 GNUNET_SCHEDULER_add_shutdown (&shutdown_task, 893 cls); 894 } 895 896 897 /** 898 * The main function of the taler-exchange-transfer. 899 * 900 * @param argc number of arguments from the command line 901 * @param argv command line arguments 902 * @return 0 ok, 1 on error 903 */ 904 int 905 main (int argc, 906 char *const *argv) 907 { 908 struct GNUNET_GETOPT_CommandLineOption options[] = { 909 GNUNET_GETOPT_option_uint ('S', 910 "size", 911 "SIZE", 912 "Size to process per shard (default: 1024)", 913 &shard_size), 914 GNUNET_GETOPT_option_timetravel ('T', 915 "timetravel"), 916 GNUNET_GETOPT_option_flag ('t', 917 "test", 918 "run in test mode and exit when idle", 919 &test_mode), 920 GNUNET_GETOPT_option_uint ('w', 921 "workers", 922 "COUNT", 923 "Plan work load with up to COUNT worker processes (default: 16)", 924 &max_workers), 925 GNUNET_GETOPT_option_version (VERSION), 926 GNUNET_GETOPT_OPTION_END 927 }; 928 enum GNUNET_GenericReturnValue ret; 929 930 ret = GNUNET_PROGRAM_run ( 931 TALER_EXCHANGE_project_data (), 932 argc, argv, 933 "taler-exchange-transfer", 934 gettext_noop ( 935 "background process that executes outgoing wire transfers"), 936 options, 937 &run, NULL); 938 if (GNUNET_SYSERR == ret) 939 return EXIT_INVALIDARGUMENT; 940 if (GNUNET_NO == ret) 941 return EXIT_SUCCESS; 942 return global_ret; 943 } 944 945 946 /* end of taler-exchange-transfer.c */