begin_revolving_shard.c (9080B)
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 *end_row = GNUNET_MIN (shard_limit, 108 *start_row + shard_size - 1); 109 now = GNUNET_TIME_absolute_get (); 110 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 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 qs = GNUNET_PQ_eval_prepared_non_select (pg->conn, 127 "begin_revolving_shard_create_revolving_shard", 128 params); 129 switch (qs) 130 { 131 case GNUNET_DB_STATUS_HARD_ERROR: 132 GNUNET_break (0); 133 TALER_EXCHANGEDB_rollback (pg); 134 return qs; 135 case GNUNET_DB_STATUS_SOFT_ERROR: 136 TALER_EXCHANGEDB_rollback (pg); 137 continue; 138 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 139 /* someone else got this shard already, 140 try again */ 141 TALER_EXCHANGEDB_rollback (pg); 142 continue; 143 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 144 /* continued below (with commit) */ 145 break; 146 } 147 } /* end create fresh reovlving shard */ 148 else 149 { 150 /* claim oldest existing shard */ 151 enum GNUNET_DB_QueryStatus qs; 152 struct GNUNET_PQ_QueryParam params[] = { 153 GNUNET_PQ_query_param_string (job_name), 154 GNUNET_PQ_query_param_end 155 }; 156 struct GNUNET_PQ_ResultSpec rs[] = { 157 GNUNET_PQ_result_spec_uint32 ("start_row", 158 start_row), 159 GNUNET_PQ_result_spec_uint32 ("end_row", 160 end_row), 161 GNUNET_PQ_result_spec_end 162 }; 163 164 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 qs = GNUNET_PQ_eval_prepared_singleton_select (pg->conn, 175 "begin_revolving_shard_open_revolving_shard", 176 params, 177 rs); 178 switch (qs) 179 { 180 case GNUNET_DB_STATUS_HARD_ERROR: 181 GNUNET_break (0); 182 TALER_EXCHANGEDB_rollback (pg); 183 return qs; 184 case GNUNET_DB_STATUS_SOFT_ERROR: 185 TALER_EXCHANGEDB_rollback (pg); 186 continue; 187 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 188 /* no open shards available */ 189 TALER_EXCHANGEDB_rollback (pg); 190 return qs; 191 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 192 { 193 enum GNUNET_DB_QueryStatus qsz; 194 struct GNUNET_TIME_Timestamp now; 195 struct GNUNET_PQ_QueryParam iparams[] = { 196 GNUNET_PQ_query_param_string (job_name), 197 GNUNET_PQ_query_param_timestamp (&now), 198 GNUNET_PQ_query_param_uint32 (start_row), 199 GNUNET_PQ_query_param_uint32 (end_row), 200 GNUNET_PQ_query_param_end 201 }; 202 203 now = GNUNET_TIME_timestamp_get (); 204 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 qsz = GNUNET_PQ_eval_prepared_non_select (pg->conn, 213 "begin_revolving_shard_reclaim_revolving_shard", 214 iparams); 215 switch (qsz) 216 { 217 case GNUNET_DB_STATUS_HARD_ERROR: 218 GNUNET_break (0); 219 TALER_EXCHANGEDB_rollback (pg); 220 return qsz; 221 case GNUNET_DB_STATUS_SOFT_ERROR: 222 TALER_EXCHANGEDB_rollback (pg); 223 continue; 224 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 225 break; /* continue with commit */ 226 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 227 GNUNET_break (0); /* logic error, should be impossible */ 228 TALER_EXCHANGEDB_rollback (pg); 229 return GNUNET_DB_STATUS_HARD_ERROR; 230 } 231 } 232 break; /* continue with commit */ 233 } 234 } /* end claim oldest existing shard */ 235 236 /* commit */ 237 { 238 enum GNUNET_DB_QueryStatus qs; 239 240 qs = TALER_EXCHANGEDB_commit (pg); 241 switch (qs) 242 { 243 case GNUNET_DB_STATUS_HARD_ERROR: 244 GNUNET_break (0); 245 TALER_EXCHANGEDB_rollback (pg); 246 return qs; 247 case GNUNET_DB_STATUS_SOFT_ERROR: 248 TALER_EXCHANGEDB_rollback (pg); 249 continue; 250 case GNUNET_DB_STATUS_SUCCESS_NO_RESULTS: 251 case GNUNET_DB_STATUS_SUCCESS_ONE_RESULT: 252 return GNUNET_DB_STATUS_SUCCESS_ONE_RESULT; 253 } 254 } 255 } /* retry 'for' loop */ 256 return GNUNET_DB_STATUS_SOFT_ERROR; 257 }