Open macosforgebot opened 10 years ago
@wsanchez originally submitted this as comment:1:ticket:839
@cyrusdaboo originally submitted this as comment:2:ticket:839
Experimented with the following.
Use “select … for update” and find a way to skip rows already locked and return the first unlocked row.
Postgres:
create or replace function next_job() returns integer as $$
declare
result integer;
begin
select ID into result from JOB where pg_try_advisory_xact_lock(ID) limit 1 for update;
return result;
end
$$ LANGUAGE plpgsql;
To get the next locked ID value:
select * from next_job();
Oracle:
create or replace function next_job return integer is
cursor c1 is select ID from JOB for update skip locked;
result integer;
begin
open c1;
fetch c1 into result;
select ID from JOB where ID = result for update;
return result;
end;
/
To get the next locked ID value:
var r number;
call next_job() into :r;
select :r from dual;
With each of the above, multiple sessions each get there next available ID without being blocked on existing sessions. For our purposes we need to have a single JOB table that lists all outstanding jobs (work tables still exist to describe each type of work):
create table JOB (
ID integer primary key default nextval('JOB_SEQ') not null, --implicit index
CLS varchar(255) not null,
PRIORITY integer default 0,
WEIGHT integer default 0,
NOT_BEFORE timestamp default null,
NOT_AFTER timestamp default null
);
When a job is enqueued an entry in the JOB table and class/type specific work queue table is made (the later references the ID column in the JOB table - or not in the case where there is a bunch of work to be done by a smaller pool of “workers”). Each job includes priority/weight and timing details (notBefore holds the job until the specified time, notAfter will cause the job priority to bump up if after that time).
To dequeue the next job, the next_job() stored procedure is used. That will need to define the logic for sorting the JOB rows so as to pick the next job taking priority/weight/notBefore/notAfter into account.
@cyrusdaboo originally submitted this as ticket:839
txext.enterprise.queue implements a database-based distributed work queue. The basic behavior is this:
There are several issues with the current implementation that we need to address: