Line data Source code
1 : /*
2 : This file is part of TALER
3 : Copyright (C) 2022 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_revolving_shard.c
18 : * @brief Implementation of the begin_revolving_shard function for Postgres
19 : * @author Christian Grothoff
20 : */
21 : #include "taler/taler_pq_lib.h"
22 : #include "exchange-database/begin_revolving_shard.h"
23 : #include "exchange-database/commit.h"
24 : #include "helper.h"
25 : #include "exchange-database/start.h"
26 : #include "exchange-database/rollback.h"
27 :
28 : enum GNUNET_DB_QueryStatus
29 105 : TALER_EXCHANGEDB_begin_revolving_shard (
30 : struct TALER_EXCHANGEDB_PostgresContext *pg,
31 : const char *job_name,
32 : uint32_t shard_size,
33 : uint32_t shard_limit,
34 : uint32_t *start_row,
35 : uint32_t *end_row)
36 : {
37 105 : GNUNET_assert (shard_limit <= 1U + (uint32_t) INT_MAX);
38 105 : GNUNET_assert (shard_limit > 0);
39 105 : GNUNET_assert (shard_size > 0);
40 105 : for (unsigned int retries = 0; retries<3; retries++)
41 : {
42 105 : if (GNUNET_OK !=
43 105 : TALER_EXCHANGEDB_start (pg,
44 : "begin_revolving_shard"))
45 : {
46 0 : GNUNET_break (0);
47 0 : return GNUNET_DB_STATUS_HARD_ERROR;
48 : }
49 :
50 : /* First, find last 'end_row' */
51 : {
52 : enum GNUNET_DB_QueryStatus qs;
53 : uint32_t last_end;
54 105 : struct GNUNET_PQ_QueryParam params[] = {
55 105 : GNUNET_PQ_query_param_string (job_name),
56 : GNUNET_PQ_query_param_end
57 : };
58 105 : struct GNUNET_PQ_ResultSpec rs[] = {
59 105 : GNUNET_PQ_result_spec_uint32 ("end_row",
60 : &last_end),
61 : GNUNET_PQ_result_spec_end
62 : };
63 :
64 105 : PREPARE (pg,
65 : "begin_revolving_shard_last_revolving_shard",
66 : "SELECT"
67 : " end_row"
68 : " FROM revolving_work_shards"
69 : " WHERE job_name=$1"
70 : " ORDER BY end_row DESC"
71 : " LIMIT 1;");
72 105 : qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
73 : "begin_revolving_shard_last_revolving_shard",
74 : params,
75 : rs);
76 105 : switch (qs)
77 : {
78 0 : case GNUNET_DB_STATUS_HARD_ERROR:
79 0 : GNUNET_break (0);
80 0 : TALER_EXCHANGEDB_rollback (pg);
81 0 : return qs;
82 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
83 0 : TALER_EXCHANGEDB_rollback (pg);
84 0 : continue;
85 10 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
86 10 : *start_row = 0; /* base-case: no shards yet */
87 10 : break; /* continued below */
88 95 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
89 95 : *start_row = 1U + last_end;
90 95 : break;
91 : }
92 : } /* get_last_shard */
93 :
94 105 : if (*start_row < shard_limit)
95 : {
96 : /* Claim fresh shard */
97 : enum GNUNET_DB_QueryStatus qs;
98 : struct GNUNET_TIME_Absolute now;
99 12 : struct GNUNET_PQ_QueryParam params[] = {
100 12 : GNUNET_PQ_query_param_string (job_name),
101 12 : GNUNET_PQ_query_param_absolute_time (&now),
102 12 : GNUNET_PQ_query_param_uint32 (start_row),
103 12 : GNUNET_PQ_query_param_uint32 (end_row),
104 : GNUNET_PQ_query_param_end
105 : };
106 :
107 : /* @a shard_limit is exclusive, @a end_row inclusive; the aggregator
108 : passes 2^31, which would not even fit the INT4 column. */
109 12 : *end_row = GNUNET_MIN (shard_limit - 1,
110 : *start_row + shard_size - 1);
111 12 : now = GNUNET_TIME_absolute_get ();
112 12 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
113 : "Trying to claim shard %llu-%llu\n",
114 : (unsigned long long) *start_row,
115 : (unsigned long long) *end_row);
116 :
117 : /* Used in #postgres_claim_revolving_shard() */
118 12 : PREPARE (pg,
119 : "begin_revolving_shard_create_revolving_shard",
120 : "INSERT INTO revolving_work_shards"
121 : "(job_name"
122 : ",last_attempt"
123 : ",start_row"
124 : ",end_row"
125 : ",active"
126 : ") VALUES "
127 : "($1, $2, $3, $4, TRUE);");
128 12 : qs = GNUNET_PQ_eval_prepared_non_select (pg->conn,
129 : "begin_revolving_shard_create_revolving_shard",
130 : params);
131 12 : switch (qs)
132 : {
133 0 : case GNUNET_DB_STATUS_HARD_ERROR:
134 0 : GNUNET_break (0);
135 0 : TALER_EXCHANGEDB_rollback (pg);
136 0 : return qs;
137 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
138 0 : TALER_EXCHANGEDB_rollback (pg);
139 0 : continue;
140 0 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
141 : /* someone else got this shard already,
142 : try again */
143 0 : TALER_EXCHANGEDB_rollback (pg);
144 0 : continue;
145 12 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
146 : /* continued below (with commit) */
147 12 : break;
148 : }
149 : } /* end create fresh reovlving shard */
150 : else
151 : {
152 : /* claim oldest existing shard */
153 : enum GNUNET_DB_QueryStatus qs;
154 93 : struct GNUNET_PQ_QueryParam params[] = {
155 93 : GNUNET_PQ_query_param_string (job_name),
156 : GNUNET_PQ_query_param_end
157 : };
158 93 : struct GNUNET_PQ_ResultSpec rs[] = {
159 93 : GNUNET_PQ_result_spec_uint32 ("start_row",
160 : start_row),
161 93 : GNUNET_PQ_result_spec_uint32 ("end_row",
162 : end_row),
163 : GNUNET_PQ_result_spec_end
164 : };
165 :
166 93 : PREPARE (pg,
167 : "begin_revolving_shard_open_revolving_shard",
168 : "SELECT"
169 : " start_row"
170 : ",end_row"
171 : " FROM revolving_work_shards"
172 : " WHERE job_name=$1"
173 : " AND active=FALSE"
174 : " ORDER BY last_attempt ASC"
175 : " LIMIT 1;");
176 93 : qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
177 : "begin_revolving_shard_open_revolving_shard",
178 : params,
179 : rs);
180 93 : switch (qs)
181 : {
182 0 : case GNUNET_DB_STATUS_HARD_ERROR:
183 0 : GNUNET_break (0);
184 0 : TALER_EXCHANGEDB_rollback (pg);
185 0 : return qs;
186 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
187 0 : TALER_EXCHANGEDB_rollback (pg);
188 0 : continue;
189 2 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
190 : /* no open shards available */
191 2 : TALER_EXCHANGEDB_rollback (pg);
192 2 : return qs;
193 91 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
194 : {
195 : enum GNUNET_DB_QueryStatus qsz;
196 : struct GNUNET_TIME_Timestamp now;
197 91 : struct GNUNET_PQ_QueryParam iparams[] = {
198 91 : GNUNET_PQ_query_param_string (job_name),
199 91 : GNUNET_PQ_query_param_timestamp (&now),
200 91 : GNUNET_PQ_query_param_uint32 (start_row),
201 91 : GNUNET_PQ_query_param_uint32 (end_row),
202 : GNUNET_PQ_query_param_end
203 : };
204 :
205 91 : now = GNUNET_TIME_timestamp_get ();
206 91 : PREPARE (pg,
207 : "begin_revolving_shard_reclaim_revolving_shard",
208 : "UPDATE revolving_work_shards"
209 : " SET last_attempt=$2"
210 : " ,active=TRUE"
211 : " WHERE job_name=$1"
212 : " AND start_row=$3"
213 : " AND end_row=$4");
214 91 : qsz = GNUNET_PQ_eval_prepared_non_select (pg->conn,
215 : "begin_revolving_shard_reclaim_revolving_shard",
216 : iparams);
217 91 : switch (qsz)
218 : {
219 0 : case GNUNET_DB_STATUS_HARD_ERROR:
220 0 : GNUNET_break (0);
221 0 : TALER_EXCHANGEDB_rollback (pg);
222 0 : return qsz;
223 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
224 0 : TALER_EXCHANGEDB_rollback (pg);
225 0 : continue;
226 91 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
227 91 : break; /* continue with commit */
228 0 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
229 0 : GNUNET_break (0); /* logic error, should be impossible */
230 0 : TALER_EXCHANGEDB_rollback (pg);
231 0 : return GNUNET_DB_STATUS_HARD_ERROR;
232 : }
233 : }
234 91 : break; /* continue with commit */
235 : }
236 : } /* end claim oldest existing shard */
237 :
238 : /* commit */
239 : {
240 : enum GNUNET_DB_QueryStatus qs;
241 :
242 103 : qs = TALER_EXCHANGEDB_commit (pg);
243 103 : switch (qs)
244 : {
245 0 : case GNUNET_DB_STATUS_HARD_ERROR:
246 0 : GNUNET_break (0);
247 0 : TALER_EXCHANGEDB_rollback (pg);
248 0 : return qs;
249 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
250 0 : TALER_EXCHANGEDB_rollback (pg);
251 0 : continue;
252 103 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
253 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
254 103 : return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
255 : }
256 : }
257 : } /* retry 'for' loop */
258 0 : return GNUNET_DB_STATUS_SOFT_ERROR;
259 : }
|
