Taakwachtrij als SQL-tabel met meerdere consumenten (PostgreSQL)

Ik gebruik postgres ook voor een FIFO-wachtrij. Ik heb oorspronkelijk ACCESS EXCLUSIVE gebruikt, wat correcte resultaten oplevert bij hoge gelijktijdigheid, maar het ongelukkige effect heeft dat het elkaar uitsluit met pg_dump, dat een ACCESS SHARE-vergrendeling krijgt tijdens de uitvoering ervan. Dit zorgt ervoor dat mijn next()-functie voor een zeer lange tijd wordt vergrendeld (de duur van de pg_dump). Dit was niet acceptabel omdat we een 24x7-winkel zijn en klanten niet van de dode tijd in de rij midden in de nacht hielden.

Ik dacht dat er een minder beperkende vergrendeling moest zijn die nog steeds gelijktijdig veilig zou zijn en niet zou vergrendelen terwijl pg_dump actief is. Mijn zoektocht leidde me naar dit SO-bericht.

Daarna heb ik wat onderzoek gedaan.

De volgende modi zijn voldoende voor een FIFO-wachtrij NEXT()-functie die de status van een taak van wachtrij bijwerkt tot rennen zonder enige gelijktijdigheid mislukken, en ook niet blokkeren tegen pg_dump:

SHARE UPDATE EXCLUSIVE
SHARE ROW EXCLUSIVE
EXCLUSIVE

Vraag:

begin;
lock table tx_test_queue in exclusive mode;
update 
    tx_test_queue
set 
    status='running'
where
    job_id in (
        select
            job_id
        from
            tx_test_queue
        where
            status='queued'
        order by 
            job_id asc
        limit 1
    )
returning job_id;
commit;

Resultaat ziet er als volgt uit:

UPDATE 1
 job_id
--------
     98
(1 row)

Hier is een shellscript dat alle verschillende vergrendelingsmodi test met hoge gelijktijdigheid (30).

#!/bin/bash
# RESULTS, feel free to repro yourself
#
# noLock                    FAIL
# accessShare               FAIL
# rowShare                  FAIL
# rowExclusive              FAIL
# shareUpdateExclusive      SUCCESS
# share                     FAIL+DEADLOCKS
# shareRowExclusive         SUCCESS
# exclusive                 SUCCESS
# accessExclusive           SUCCESS, but LOCKS against pg_dump

#config
strategy="exclusive"

db=postgres
dbuser=postgres
queuecount=100
concurrency=30

# code
psql84 -t -U $dbuser $db -c "create table tx_test_queue (job_id serial, status text);"
# empty queue
psql84 -t -U $dbuser $db -c "truncate tx_test_queue;";
echo "Simulating 10 second pg_dump with ACCESS SHARE"
psql84 -t -U $dbuser $db -c "lock table tx_test_queue in ACCESS SHARE mode; select pg_sleep(10); select 'pg_dump finished...'" &

echo "Starting workers..."
# queue $queuecount items
seq $queuecount | xargs -n 1 -P $concurrency -I {} psql84 -q -U $dbuser $db -c "insert into tx_test_queue (status) values ('queued');"
#psql84 -t -U $dbuser $db -c "select * from tx_test_queue order by job_id;"
# process $queuecount w/concurrency of $concurrency
case $strategy in
    "noLock")               strategySql="update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "accessShare")          strategySql="lock table tx_test_queue in ACCESS SHARE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "rowShare")             strategySql="lock table tx_test_queue in ROW SHARE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "rowExclusive")         strategySql="lock table tx_test_queue in ROW EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "shareUpdateExclusive") strategySql="lock table tx_test_queue in SHARE UPDATE EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "share")                strategySql="lock table tx_test_queue in SHARE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "shareRowExclusive")    strategySql="lock table tx_test_queue in SHARE ROW EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "exclusive")            strategySql="lock table tx_test_queue in EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    "accessExclusive")      strategySql="lock table tx_test_queue in ACCESS EXCLUSIVE mode; update tx_test_queue set status='running{}' where job_id in (select job_id from tx_test_queue where status='queued' order by job_id asc limit 1);";;
    *) echo "Unknown strategy $strategy";;
esac
echo $strategySql
seq $queuecount | xargs -n 1 -P $concurrency -I {} psql84 -U $dbuser $db -c "$strategySql"
#psql84 -t -U $dbuser $db -c "select * from tx_test_queue order by job_id;"
psql84 -U $dbuser $db -c "select count(distinct(status)) as should_output_100 from tx_test_queue;"
psql84 -t -U $dbuser $db -c "drop table tx_test_queue;";

Code is hier ook als je wilt bewerken:https://gist.github.com/1083936

Ik ben mijn applicatie aan het updaten om de EXCLUSIEVE modus te gebruiken, omdat dit de meest beperkende modus is die a) correct is en b) niet in strijd is met pg_dump. Ik koos de meest beperkende omdat dit het minst riskant lijkt in termen van het wijzigen van de app van ACCESS EXCLUSIVE zonder een uber-expert te zijn in postgres-vergrendeling.

Ik voel me redelijk op mijn gemak met mijn testopstelling en met de algemene ideeën achter het antwoord. Ik hoop dat door dit te delen dit probleem voor anderen kan worden opgelost.