From 973e428caa04096b5ce98448a5ea346fd94434d9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?C=C3=A9dric=20Krier?= <ced@b2ck.com>
Date: Sat, 28 Dec 2019 18:05:25 +0100
Subject: [PATCH] Replace advisory lock by SKIP LOCKED in queue pulling

It is simpler and closer to standard SQL.

issue8910
review288701002
---
 CHANGELOG                              |  1 +
 trytond/backend/database.py            |  5 +++++
 trytond/backend/postgresql/database.py | 23 ++++++++++++++++++-
 trytond/ir/queue.py                    | 31 +++++++++++++++-----------
 4 files changed, 46 insertions(+), 14 deletions(-)

diff --git a/CHANGELOG b/CHANGELOG
index b54aa3d23d..44092915c0 100644
--- a/CHANGELOG
+++ b/CHANGELOG
@@ -1,3 +1,4 @@
+* Replace advisory lock by SKIP LOCKED in queue pulling
 * Load WSGI middleware from configuration
 * Add help text to Dict keys
 * Update button_action with returned values
diff --git a/trytond/backend/database.py b/trytond/backend/database.py
index 592439e5e5..0e859f227c 100644
--- a/trytond/backend/database.py
+++ b/trytond/backend/database.py
@@ -1,6 +1,7 @@
 # This file is part of Tryton.  The COPYRIGHT file at the top level of
 # this repository contains the full copyright notices and license terms.
 from collections import namedtuple
+from sql import For
 
 DatabaseIntegrityError = None
 DatabaseOperationalError = None
@@ -164,6 +165,10 @@ class DatabaseInterface(object):
         "Return if database supports FOR UPDATE/SHARE clause in SELECT."
         return False
 
+    def get_select_for_skip_locked(self):
+        "Return For class with skip locked"
+        return For
+
     def has_window_functions(self):
         "Return if database supports window functions."
         return False
diff --git a/trytond/backend/postgresql/database.py b/trytond/backend/postgresql/database.py
index 8bf3d9dd8b..1a0f86df34 100644
--- a/trytond/backend/postgresql/database.py
+++ b/trytond/backend/postgresql/database.py
@@ -30,7 +30,7 @@ from psycopg2 import IntegrityError as DatabaseIntegrityError
 from psycopg2 import OperationalError as DatabaseOperationalError
 from psycopg2.extras import register_default_json, register_default_jsonb
 
-from sql import Flavor, Cast
+from sql import Flavor, Cast, For
 from sql.functions import Function
 from sql.operators import BinaryOperator
 
@@ -67,6 +67,12 @@ class LoggingCursor(cursor):
         cursor.execute(self, sql, args)
 
 
+class ForSkipLocked(For):
+    def __str__(self):
+        assert not self.nowait, "Can not use both NO WAIT and SKIP LOCKED"
+        return super().__str__() + (' SKIP LOCKED' if not self.nowait else '')
+
+
 class Unaccent(Function):
     __slots__ = ()
     _function = 'unaccent'
@@ -132,6 +138,7 @@ class Database(DatabaseInterface):
     _search_path = None
     _current_user = None
     _has_returning = None
+    _has_select_for_skip_locked = None
     _has_unaccent = {}
     flavor = Flavor(ilike=True)
 
@@ -413,6 +420,20 @@ class Database(DatabaseInterface):
     def has_select_for(self):
         return True
 
+    def get_select_for_skip_locked(self):
+        if self._has_select_for_skip_locked is None:
+            connection = self.get_connection()
+            try:
+                # SKIP LOCKED clause is available since PostgreSQL 9.5
+                self._has_select_for_skip_locked = (
+                    self.get_version(connection) >= (9, 5))
+            finally:
+                self.put_connection(connection)
+        if self._has_select_for_skip_locked:
+            return ForSkipLocked
+        else:
+            return For
+
     def has_window_functions(self):
         return True
 
diff --git a/trytond/ir/queue.py b/trytond/ir/queue.py
index 2f42a1a965..e294dc067a 100644
--- a/trytond/ir/queue.py
+++ b/trytond/ir/queue.py
@@ -93,15 +93,20 @@ class Queue(ModelSQL):
                 order_by=[
                     queue.scheduled_at.nulls_first,
                     queue.expected_at.nulls_first]))
-        selected = With('id', query=candidates.select(
-                candidates.id,
-                where=((candidates.scheduled_at <= CurrentTimestamp())
-                    | (candidates.scheduled_at == Null))
-                & database.lock_id(candidates.id),
-                order_by=[
-                    candidates.scheduled_at.nulls_first,
-                    candidates.expected_at.nulls_first],
-                limit=1))
+        selected = queue.select(
+            queue.id,
+            where=((queue.name == name) if name else Literal(True))
+            & (queue.dequeued_at == Null)
+            & ((queue.scheduled_at <= CurrentTimestamp())
+                | (queue.scheduled_at == Null)),
+            order_by=[
+                queue.scheduled_at.nulls_first,
+                queue.expected_at.nulls_first],
+            limit=1)
+        if database.has_select_for():
+            For = database.get_select_for_skip_locked()
+            selected.for_ = For('UPDATE')
+
         next_timeout = With('seconds', query=candidates.select(
                 Min(Extract('second',
                         candidates.scheduled_at - CurrentTimestamp())
@@ -111,8 +116,8 @@ class Queue(ModelSQL):
         task_id, seconds = None, None
         if database.has_returning():
             query = queue.update([queue.dequeued_at], [CurrentTimestamp()],
-                where=queue.id == selected.select(selected.id),
-                with_=[candidates, selected, next_timeout],
+                where=queue.id.in_(selected),
+                with_=[candidates, next_timeout],
                 returning=[
                     queue.id, next_timeout.select(next_timeout.seconds)])
             cursor.execute(*query)
@@ -121,8 +126,8 @@ class Queue(ModelSQL):
                 task_id, seconds = row
         else:
             query = queue.select(queue.id,
-                where=queue.id == selected.select(selected.id),
-                with_=[candidates, selected])
+                where=queue.id.in_(selected),
+                with_=[candidates])
             cursor.execute(*query)
             row = cursor.fetchone()
             if row: