Re: Serialization, Locking...implement processing Queue with a table

On Mon, 12 May 2003 14:28:18 -0500, "D. Dante Lorenso"
<dante@lorenso.com> wrote:
>BLOCKING [is] more desireable than the Fail/Retry
>approach, eh?
>
>Can you confirm that this solution will perform as I expect
>while keeping the transaction isolation level at a
>READ COMMITTED mode instead of SERIALIZABLE?

Starting with Tom's suggestion I played around with READ COMMITTED.

>>2. When idle, you try to reserve a job like so:
>>
>>    BEGIN;
    -- TRANSACTION ISOLATION LEVEL is READ COMMITTED by default
>>    SELECT job_id, ... FROM job_table
>>        WHERE processor_id = 0 LIMIT 1 FOR UPDATE;

The SELECT may be blocked for a while and has two possible outcomes:

>>2a: One row is returned.  You do
>>
>>    UPDATE job_table SET processor_id = $me
>>        WHERE job_id = $jobid;
>>    COMMIT;
>>
>>and then go about executing the job.
>>
>>2b: No row is returned:
This can have one of two reasons

(i)  no jobs are pending.  Commit your transaction,
>>sleep for an appropriate delay period, and try again.

(ii)  The row has been reserved by another transaction running at the
same time.  In this case, restart at SELECT FOR UPDATE.  You can stay
in the same transaction.  And you don't need to sleep (this has
already happened while SELECT FOR UPDATE was blocked), unless there
are lots of job processors in which case it might be advisible to
sleep for a short random time.

How do you distinguish between (i) and (ii)?  Just do

    SELECT job_id FROM job_table
        WHERE processor_id = 0 LIMIT 1;

If this returns 0 rows, you have (i).
If it returns one row, you have (ii).

I didn't try, but you should be able to do this in a function.  Your
function looks like a good start, you just have to add a retry loop
and a test for (i) vs. (ii)

>CREATE OR REPLACE FUNCTION reserve_next_import (bigint)
>RETURNS bigint AS'
>DECLARE
>    processor_id ALIAS FOR $1;
>    my_import_id BIGINT;
>BEGIN
>    -- initialize the id
>    my_import_id := -1;
>
-- start of loop here!
>    -- Find the import ID we wish to reserve and get a lock on that row
>    SELECT import_id INTO my_import_id
>    FROM import
>    WHERE import_state = ''Q''
>    AND import_processor_id IS NULL
>    ORDER BY import_id
>    LIMIT 1
>    FOR UPDATE;
>
>    -- abort if there are no queued rows
>    IF NOT FOUND THEN
SELECT ...;  -- without FOR UPDATE
IF NOT FOUND THEN
>        RETURN (-1);
ELSE
    continue at top of loop
END IF;
>    END IF;
>
>    -- now go reserve the record with our processor id
>    UPDATE import SET
>        import_processor_id = processor_id,
>        import_prc_start = NULL,
>        import_prc_end = NULL,
>        import_state = ''R''
>    WHERE import_id = my_import_id;
>
>    -- this is the row we reserved...
>    RETURN (my_import_id);
>END;
>'LANGUAGE 'plpgsql';

Servus
 Manfred