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 12 : *end_row = GNUNET_MIN (shard_limit,
108 : *start_row + shard_size - 1);
109 12 : now = GNUNET_TIME_absolute_get ();
110 12 : GNUNET_log (GNUNET_ERROR_TYPE_INFO,
111 : "Trying to claim shard %llu-%llu\n",
112 : (unsigned long long) *start_row,
113 : (unsigned long long) *end_row);
114 :
115 : /* Used in #postgres_claim_revolving_shard() */
116 12 : PREPARE (pg,
117 : "begin_revolving_shard_create_revolving_shard",
118 : "INSERT INTO revolving_work_shards"
119 : "(job_name"
120 : ",last_attempt"
121 : ",start_row"
122 : ",end_row"
123 : ",active"
124 : ") VALUES "
125 : "($1, $2, $3, $4, TRUE);");
126 12 : qs = GNUNET_PQ_eval_prepared_non_select (pg->conn,
127 : "begin_revolving_shard_create_revolving_shard",
128 : params);
129 12 : switch (qs)
130 : {
131 0 : case GNUNET_DB_STATUS_HARD_ERROR:
132 0 : GNUNET_break (0);
133 0 : TALER_EXCHANGEDB_rollback (pg);
134 0 : return qs;
135 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
136 0 : TALER_EXCHANGEDB_rollback (pg);
137 0 : continue;
138 0 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
139 : /* someone else got this shard already,
140 : try again */
141 0 : TALER_EXCHANGEDB_rollback (pg);
142 0 : continue;
143 12 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
144 : /* continued below (with commit) */
145 12 : break;
146 : }
147 : } /* end create fresh reovlving shard */
148 : else
149 : {
150 : /* claim oldest existing shard */
151 : enum GNUNET_DB_QueryStatus qs;
152 93 : struct GNUNET_PQ_QueryParam params[] = {
153 93 : GNUNET_PQ_query_param_string (job_name),
154 : GNUNET_PQ_query_param_end
155 : };
156 93 : struct GNUNET_PQ_ResultSpec rs[] = {
157 93 : GNUNET_PQ_result_spec_uint32 ("start_row",
158 : start_row),
159 93 : GNUNET_PQ_result_spec_uint32 ("end_row",
160 : end_row),
161 : GNUNET_PQ_result_spec_end
162 : };
163 :
164 93 : PREPARE (pg,
165 : "begin_revolving_shard_open_revolving_shard",
166 : "SELECT"
167 : " start_row"
168 : ",end_row"
169 : " FROM revolving_work_shards"
170 : " WHERE job_name=$1"
171 : " AND active=FALSE"
172 : " ORDER BY last_attempt ASC"
173 : " LIMIT 1;");
174 93 : qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn,
175 : "begin_revolving_shard_open_revolving_shard",
176 : params,
177 : rs);
178 93 : switch (qs)
179 : {
180 0 : case GNUNET_DB_STATUS_HARD_ERROR:
181 0 : GNUNET_break (0);
182 0 : TALER_EXCHANGEDB_rollback (pg);
183 0 : return qs;
184 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
185 0 : TALER_EXCHANGEDB_rollback (pg);
186 0 : continue;
187 2 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
188 : /* no open shards available */
189 2 : TALER_EXCHANGEDB_rollback (pg);
190 2 : return qs;
191 91 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
192 : {
193 : enum GNUNET_DB_QueryStatus qsz;
194 : struct GNUNET_TIME_Timestamp now;
195 91 : struct GNUNET_PQ_QueryParam iparams[] = {
196 91 : GNUNET_PQ_query_param_string (job_name),
197 91 : GNUNET_PQ_query_param_timestamp (&now),
198 91 : GNUNET_PQ_query_param_uint32 (start_row),
199 91 : GNUNET_PQ_query_param_uint32 (end_row),
200 : GNUNET_PQ_query_param_end
201 : };
202 :
203 91 : now = GNUNET_TIME_timestamp_get ();
204 91 : PREPARE (pg,
205 : "begin_revolving_shard_reclaim_revolving_shard",
206 : "UPDATE revolving_work_shards"
207 : " SET last_attempt=$2"
208 : " ,active=TRUE"
209 : " WHERE job_name=$1"
210 : " AND start_row=$3"
211 : " AND end_row=$4");
212 91 : qsz = GNUNET_PQ_eval_prepared_non_select (pg->conn,
213 : "begin_revolving_shard_reclaim_revolving_shard",
214 : iparams);
215 91 : switch (qsz)
216 : {
217 0 : case GNUNET_DB_STATUS_HARD_ERROR:
218 0 : GNUNET_break (0);
219 0 : TALER_EXCHANGEDB_rollback (pg);
220 0 : return qsz;
221 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
222 0 : TALER_EXCHANGEDB_rollback (pg);
223 0 : continue;
224 91 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
225 91 : break; /* continue with commit */
226 0 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
227 0 : GNUNET_break (0); /* logic error, should be impossible */
228 0 : TALER_EXCHANGEDB_rollback (pg);
229 0 : return GNUNET_DB_STATUS_HARD_ERROR;
230 : }
231 : }
232 91 : break; /* continue with commit */
233 : }
234 : } /* end claim oldest existing shard */
235 :
236 : /* commit */
237 : {
238 : enum GNUNET_DB_QueryStatus qs;
239 :
240 103 : qs = TALER_EXCHANGEDB_commit (pg);
241 103 : switch (qs)
242 : {
243 0 : case GNUNET_DB_STATUS_HARD_ERROR:
244 0 : GNUNET_break (0);
245 0 : TALER_EXCHANGEDB_rollback (pg);
246 0 : return qs;
247 0 : case GNUNET_DB_STATUS_SOFT_ERROR:
248 0 : TALER_EXCHANGEDB_rollback (pg);
249 0 : continue;
250 103 : case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS:
251 : case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT:
252 103 : return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT;
253 : }
254 : }
255 : } /* retry 'for' loop */
256 0 : return GNUNET_DB_STATUS_SOFT_ERROR;
257 : }
|
