begin_shard.sql (3101B)
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 DROP FUNCTION IF EXISTS exchange_do_begin_shard; 18 CREATE FUNCTION exchange_do_begin_shard( 19 IN in_job_name TEXT, 20 IN in_now INT8, 21 IN in_lease_until INT8, 22 IN in_shard_size INT8, 23 OUT out_start_row INT8, 24 OUT out_end_row INT8, 25 OUT out_progress_row INT8) 26 LANGUAGE plpgsql 27 AS $$ 28 BEGIN 29 -- Prefer an existing shard nobody is working on any more. FOR UPDATE SKIP 30 -- LOCKED hands concurrent workers *different* shards instead of letting 31 -- them collide and retry, which is what the explicit serializable 32 -- transaction this function replaced used to do. 33 SELECT start_row 34 ,end_row 35 ,progress_row 36 INTO out_start_row 37 ,out_end_row 38 ,out_progress_row 39 FROM work_shards 40 WHERE job_name=in_job_name 41 AND completed=FALSE 42 AND last_attempt<in_now 43 ORDER BY last_attempt ASC 44 LIMIT 1 45 FOR UPDATE SKIP LOCKED; 46 47 IF FOUND 48 THEN 49 -- Take over the lease. Whatever progress the previous worker committed 50 -- stands, and we return it so the caller resumes instead of restarting. 51 UPDATE work_shards 52 SET last_attempt=in_lease_until 53 WHERE job_name=in_job_name 54 AND start_row=out_start_row 55 AND end_row=out_end_row; 56 RETURN; 57 END IF; 58 59 -- No shard to take over, so open the next one past the last one known. 60 SELECT end_row 61 INTO out_start_row 62 FROM work_shards 63 WHERE job_name=in_job_name 64 ORDER BY end_row DESC 65 LIMIT 1; 66 IF NOT FOUND 67 THEN 68 out_start_row = 0; -- base case: no shards for this job yet 69 END IF; 70 out_end_row = out_start_row + in_shard_size; 71 out_progress_row = out_start_row; 72 73 INSERT INTO work_shards 74 (job_name 75 ,last_attempt 76 ,start_row 77 ,end_row 78 ,progress_row 79 ) VALUES ( 80 in_job_name 81 ,in_lease_until 82 ,out_start_row 83 ,out_end_row 84 ,out_start_row 85 ) 86 ON CONFLICT DO NOTHING; 87 88 IF NOT FOUND 89 THEN 90 -- Someone else opened this very shard between our two statements. Report 91 -- "nothing acquired" and let the caller come back around. 92 out_start_row = NULL; 93 out_end_row = NULL; 94 out_progress_row = NULL; 95 END IF; 96 END $$; 97 98 COMMENT ON FUNCTION exchange_do_begin_shard 99 IS 'Acquires a lease on a work shard for in_job_name, resuming an abandoned shard if there is one and opening a fresh shard of in_shard_size rows otherwise. Returns NULL columns if another worker won the race to open the same shard.';