begin_revolving_shard.c (9226B)
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 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 GNUNET_assert (shard_limit <= 1U + (uint32_t) INT_MAX); 38 GNUNET_assert (shard_limit > 0); 39 GNUNET_assert (shard_size > 0); 40 for (unsigned int retries = 0; retries<3; retries++) 41 { 42 if (GNUNET_OK != 43 TALER_EXCHANGEDB_start (pg, 44 "begin_revolving_shard")) 45 { 46 GNUNET_break (0); 47 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 struct GNUNET_PQ_QueryParam params[] = { 55 GNUNET_PQ_query_param_string (job_name), 56 GNUNET_PQ_query_param_end 57 }; 58 struct GNUNET_PQ_ResultSpec rs[] = { 59 GNUNET_PQ_result_spec_uint32 ("end_row", 60 &last_end), 61 GNUNET_PQ_result_spec_end 62 }; 63 64 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 qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn, 73 "begin_revolving_shard_last_revolving_shard", 74 params, 75 rs); 76 switch (qs) 77 { 78 case GNUNET_DB_STATUS_HARD_ERROR: 79 GNUNET_break (0); 80 TALER_EXCHANGEDB_rollback (pg); 81 return qs; 82 case GNUNET_DB_STATUS_SOFT_ERROR: 83 TALER_EXCHANGEDB_rollback (pg); 84 continue; 85 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 86 *start_row = 0; /* base-case: no shards yet */ 87 break; /* continued below */ 88 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 89 *start_row = 1U + last_end; 90 break; 91 } 92 } /* get_last_shard */ 93 94 if (*start_row < shard_limit) 95 { 96 /* Claim fresh shard */ 97 enum GNUNET_DB_QueryStatus qs; 98 struct GNUNET_TIME_Absolute now; 99 struct GNUNET_PQ_QueryParam params[] = { 100 GNUNET_PQ_query_param_string (job_name), 101 GNUNET_PQ_query_param_absolute_time (&now), 102 GNUNET_PQ_query_param_uint32 (start_row), 103 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 *end_row = GNUNET_MIN (shard_limit - 1, 110 *start_row + shard_size - 1); 111 now = GNUNET_TIME_absolute_get (); 112 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 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 qs = GNUNET_PQ_eval_prepared_non_select (pg->conn, 129 "begin_revolving_shard_create_revolving_shard", 130 params); 131 switch (qs) 132 { 133 case GNUNET_DB_STATUS_HARD_ERROR: 134 GNUNET_break (0); 135 TALER_EXCHANGEDB_rollback (pg); 136 return qs; 137 case GNUNET_DB_STATUS_SOFT_ERROR: 138 TALER_EXCHANGEDB_rollback (pg); 139 continue; 140 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 141 /* someone else got this shard already, 142 try again */ 143 TALER_EXCHANGEDB_rollback (pg); 144 continue; 145 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 146 /* continued below (with commit) */ 147 break; 148 } 149 } /* end create fresh reovlving shard */ 150 else 151 { 152 /* claim oldest existing shard */ 153 enum GNUNET_DB_QueryStatus qs; 154 struct GNUNET_PQ_QueryParam params[] = { 155 GNUNET_PQ_query_param_string (job_name), 156 GNUNET_PQ_query_param_end 157 }; 158 struct GNUNET_PQ_ResultSpec rs[] = { 159 GNUNET_PQ_result_spec_uint32 ("start_row", 160 start_row), 161 GNUNET_PQ_result_spec_uint32 ("end_row", 162 end_row), 163 GNUNET_PQ_result_spec_end 164 }; 165 166 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 qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn, 177 "begin_revolving_shard_open_revolving_shard", 178 params, 179 rs); 180 switch (qs) 181 { 182 case GNUNET_DB_STATUS_HARD_ERROR: 183 GNUNET_break (0); 184 TALER_EXCHANGEDB_rollback (pg); 185 return qs; 186 case GNUNET_DB_STATUS_SOFT_ERROR: 187 TALER_EXCHANGEDB_rollback (pg); 188 continue; 189 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 190 /* no open shards available */ 191 TALER_EXCHANGEDB_rollback (pg); 192 return qs; 193 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 194 { 195 enum GNUNET_DB_QueryStatus qsz; 196 struct GNUNET_TIME_Timestamp now; 197 struct GNUNET_PQ_QueryParam iparams[] = { 198 GNUNET_PQ_query_param_string (job_name), 199 GNUNET_PQ_query_param_timestamp (&now), 200 GNUNET_PQ_query_param_uint32 (start_row), 201 GNUNET_PQ_query_param_uint32 (end_row), 202 GNUNET_PQ_query_param_end 203 }; 204 205 now = GNUNET_TIME_timestamp_get (); 206 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 qsz = GNUNET_PQ_eval_prepared_non_select (pg->conn, 215 "begin_revolving_shard_reclaim_revolving_shard", 216 iparams); 217 switch (qsz) 218 { 219 case GNUNET_DB_STATUS_HARD_ERROR: 220 GNUNET_break (0); 221 TALER_EXCHANGEDB_rollback (pg); 222 return qsz; 223 case GNUNET_DB_STATUS_SOFT_ERROR: 224 TALER_EXCHANGEDB_rollback (pg); 225 continue; 226 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 227 break; /* continue with commit */ 228 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 229 GNUNET_break (0); /* logic error, should be impossible */ 230 TALER_EXCHANGEDB_rollback (pg); 231 return GNUNET_DB_STATUS_HARD_ERROR; 232 } 233 } 234 break; /* continue with commit */ 235 } 236 } /* end claim oldest existing shard */ 237 238 /* commit */ 239 { 240 enum GNUNET_DB_QueryStatus qs; 241 242 qs = TALER_EXCHANGEDB_commit (pg); 243 switch (qs) 244 { 245 case GNUNET_DB_STATUS_HARD_ERROR: 246 GNUNET_break (0); 247 TALER_EXCHANGEDB_rollback (pg); 248 return qs; 249 case GNUNET_DB_STATUS_SOFT_ERROR: 250 TALER_EXCHANGEDB_rollback (pg); 251 continue; 252 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 253 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 254 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 255 } 256 } 257 } /* retry 'for' loop */ 258 return GNUNET_DB_STATUS_SOFT_ERROR; 259 }