Line data Source code
1 : /*
2 : This file is part of TALER
3 : Copyright (C) 2022, 2026 Taler Systems SA
4 :
5 : TALER is free software; you can redistribute it and/or modify it under the
6 : terms of the GNU 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 General Public License for more details.
12 :
13 : You should have received a copy of the GNU General Public License along with
14 : TALER; see the file COPYING. If not, see <http://www.gnu.org/licenses/>
15 : */
16 : /**
17 : * @file exchangedb/begin_shard.c
18 : * @brief Implementation of the begin_shard function for Postgres
19 : * @author Christian Grothoff
20 : */
21 : #include "taler/taler_pq_lib.h"
22 : #include "exchange-database/begin_shard.h"
23 : #include "helper.h"
24 :
25 :
26 : /**
27 : * How often do we retry when another worker opened the very shard we were
28 : * about to open? Each round is one round-trip and the loser of the race can
29 : * simply take the shard after it, so this converges quickly.
30 : */
31 : #define MAX_RETRIES 10
32 :
33 :
34 : enum GNUNET_DB_QueryStatus
35 300 : TALER_EXCHANGEDB_begin_shard (struct TALER_EXCHANGEDB_PostgresContext *pg,
36 : const char *job_name,
37 : struct GNUNET_TIME_Relative delay,
38 : uint64_t shard_size,
39 : uint64_t *start_row,
40 : uint64_t *end_row,
41 : uint64_t *progress_row)
42 : {
43 300 : PREPARE (pg,
44 : "begin_shard",
45 : "SELECT"
46 : " out_start_row"
47 : ",out_end_row"
48 : ",out_progress_row"
49 : " FROM exchange_do_begin_shard"
50 : " ($1,$2,$3,$4);");
51 300 : for (unsigned int retries = 0; retries<MAX_RETRIES; retries++)
52 : {
53 : struct GNUNET_TIME_Absolute now
54 300 : = GNUNET_TIME_absolute_get ();
55 : struct GNUNET_TIME_Absolute lease_until
56 300 : = GNUNET_TIME_relative_to_absolute (delay);
57 300 : struct GNUNET_PQ_QueryParam params[] = {
58 300 : GNUNET_PQ_query_param_string (job_name),
59 300 : GNUNET_PQ_query_param_absolute_time (&now),
60 300 : GNUNET_PQ_query_param_absolute_time (&lease_until),
61 300 : GNUNET_PQ_query_param_uint64 (&shard_size),
62 : GNUNET_PQ_query_param_end
63 : };
64 300 : bool lost_race = false;
65 300 : struct GNUNET_PQ_ResultSpec rs[] = {
66 300 : GNUNET_PQ_result_spec_allow_null (
67 : GNUNET_PQ_result_spec_uint64 ("out_start_row",
68 : start_row),
69 : &lost_race),
70 300 : GNUNET_PQ_result_spec_allow_null (
71 : GNUNET_PQ_result_spec_uint64 ("out_end_row",
72 : end_row),
73 : NULL),
74 300 : GNUNET_PQ_result_spec_allow_null (
75 : GNUNET_PQ_result_spec_uint64 ("out_progress_row",
76 : progress_row),
77 : NULL),
78 : GNUNET_PQ_result_spec_end
79 : };
80 : enum GNUNET_DB_QueryStatus qs;
81 :
82 300 : qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
83 : "begin_shard",
84 : params,
85 : rs);
86 300 : if (0 > qs)
87 300 : return qs;
88 300 : if (GNUNET_DB_STATUS_SUCCESS_NO_RESULTS == qs)
89 : {
90 0 : GNUNET_break (0); /* the function always returns exactly one row */
91 0 : return GNUNET_DB_STATUS_HARD_ERROR;
92 : }
93 300 : if (lost_race)
94 : {
95 : /* Someone else opened the shard we picked. Go around: either take over
96 : an even older one, or open the one after theirs. */
97 0 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
98 : "Lost the race to open a shard of %s, trying again\n",
99 : job_name);
100 0 : continue;
101 : }
102 300 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
103 : "Acquired shard %s (%llu,%llu], resuming at %llu\n",
104 : job_name,
105 : (unsigned long long) *start_row,
106 : (unsigned long long) *end_row,
107 : (unsigned long long) *progress_row);
108 300 : return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
109 : }
110 : /* Persistent contention. Not an error, but the caller should back off. */
111 0 : return GNUNET_DB_STATUS_SOFT_ERROR;
112 : }
|
