Merge branch 'sunbirddcim-Use_skip_locked'

Author: ukd1

CHANGELOG.md

@@ -2,7 +2,7 @@
 
 All notable changes to this project will be documented in this file.
 
-## Unreleased, 3.3.0
+## [MASTER] - WIP - 2019-07-18
 - Fixed a bug in the offset calculation of `.enqueue_at`.
 - Use the jsonb type for the args column from now on. If not available, fall back to json or text.
 - `enqueue`, `enqueue_at`, `enqueue_in` return job hash with id.
@@ -13,6 +13,8 @@ All notable changes to this project will be documented in this file.
 - Change to only support >= Postgres 9.6. We will be bringing in newer changes and testing on only 9.6+ going forward.
 - Change to only support currently supported Ruby versions: 2.4, 2.5 and 2.6.
 - Add tests for installing fresh on rails 5.2.3 + running migrations
+- Bump to version 4, breaking changes (requirements of ruby, postgres)
+- Use skip-locked see - https://github.com/QueueClassic/queue_classic/pull/303
 
 ## [3.0.0rc] - 2014-01-07
 

lib/queue_classic/queue.rb

@@ -90,8 +90,12 @@ def enqueue_in(seconds, method, *args)
 
     def lock
       QC.log_yield(:measure => 'queue.lock') do
-        s = "SELECT * FROM lock_head($1, $2)"
-        if r = conn_adapter.execute(s, name, top_bound)
+        s = "UPDATE queue_classic_jobs SET locked_at = now(), locked_by = pg_backend_pid()
+             WHERE id IN ( SELECT id FROM queue_classic_jobs WHERE locked_at IS NULL AND
+             q_name = $1 AND scheduled_at <= now() LIMIT 1 FOR NO KEY UPDATE SKIP LOCKED )
+             RETURNING *"
+
+        if r = conn_adapter.execute(s, name)
           {}.tap do |job|
             job[:id] = r["id"]
             job[:q_name] = r["q_name"]

lib/queue_classic/version.rb

@@ -1,5 +1,5 @@
 # frozen_string_literal: true
 
 module QC
-  VERSION = "3.3.0.ALPHA"
+  VERSION = "4.0.0-alpha1"
 end

sql/ddl.sql

@@ -1,73 +1,3 @@
--- We are declaring the return type to be queue_classic_jobs.
--- This is ok since I am assuming that all of the users added queues will
--- have identical columns to queue_classic_jobs.
--- When QC supports queues with columns other than the default, we will have to change this.
-
-CREATE OR REPLACE FUNCTION lock_head(q_name varchar, top_boundary integer)
-RETURNS SETOF queue_classic_jobs AS $$
-DECLARE
-  unlocked bigint;
-  relative_top integer;
-  job_count integer;
-BEGIN
-  -- The purpose is to release contention for the first spot in the table.
-  -- The select count(*) is going to slow down dequeue performance but allow
-  -- for more workers. Would love to see some optimization here...
-
-  EXECUTE 'SELECT count(*) FROM '
-    || '(SELECT * FROM queue_classic_jobs '
-    || ' WHERE locked_at IS NULL'
-    || ' AND q_name = '
-    || quote_literal(q_name)
-    || ' AND scheduled_at <= '
-    || quote_literal(now())
-    || ' LIMIT '
-    || quote_literal(top_boundary)
-    || ') limited'
-  INTO job_count;
-
-  SELECT TRUNC(random() * (top_boundary - 1))
-  INTO relative_top;
-
-  IF job_count < top_boundary THEN
-    relative_top = 0;
-  END IF;
-
-  LOOP
-    BEGIN
-      EXECUTE 'SELECT id FROM queue_classic_jobs '
-        || ' WHERE locked_at IS NULL'
-        || ' AND q_name = '
-        || quote_literal(q_name)
-        || ' AND scheduled_at <= '
-        || quote_literal(now())
-        || ' ORDER BY id ASC'
-        || ' LIMIT 1'
-        || ' OFFSET ' || quote_literal(relative_top)
-        || ' FOR UPDATE NOWAIT'
-      INTO unlocked;
-      EXIT;
-    EXCEPTION
-      WHEN lock_not_available THEN
-        -- do nothing. loop again and hope we get a lock
-    END;
-  END LOOP;
-
-  RETURN QUERY EXECUTE 'UPDATE queue_classic_jobs '
-    || ' SET locked_at = (CURRENT_TIMESTAMP),'
-    || ' locked_by = (select pg_backend_pid())'
-    || ' WHERE id = $1'
-    || ' AND locked_at is NULL'
-    || ' RETURNING *'
-  USING unlocked;
-
-  RETURN;
-END $$ LANGUAGE plpgsql;
-
-CREATE OR REPLACE FUNCTION lock_head(tname varchar) RETURNS SETOF queue_classic_jobs AS $$ BEGIN
-  RETURN QUERY EXECUTE 'SELECT * FROM lock_head($1,10)' USING tname;
-END $$ LANGUAGE plpgsql;
-
 -- queue_classic_notify function and trigger
 CREATE FUNCTION queue_classic_notify() RETURNS TRIGGER AS $$ BEGIN
   perform pg_notify(new.q_name, ''); RETURN NULL;