fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-09-15 13:37:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

Change task timeout logic to have now() as execution time (#58)

Browse Source
This commit is contained in:
msabatier
2023-01-11 01:48:57 +01:00
committed by GitHub
parent 8cd1028391
commit 865b1a5ba8
1 changed files with 6 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
django_q/brokers/orm.py
View File

@@ -11,7 +11,7 @@ from django_q.models import OrmQ
def _timeout(): def _timeout():
return timezone.now() - timedelta(seconds=Conf.RETRY) return timezone.now() + timedelta(seconds=Conf.RETRY)
class ORM(Broker): class ORM(Broker):
@@ -31,13 +31,13 @@ class ORM(Broker):
def queue_size(self) -> int: def queue_size(self) -> int:
return ( return (
self.get_connection() self.get_connection()
.filter(key=self.list_key, lock__lte=_timeout()) .filter(key=self.list_key, lock__lte=timezone.now())
.count() .count()
) )
def lock_size(self) -> int: def lock_size(self) -> int:
return ( return (
self.get_connection().filter(key=self.list_key, lock__gt=_timeout()).count() self.get_connection().filter(key=self.list_key, lock__gt=timezone.now()).count()
) )
def purge_queue(self): def purge_queue(self):
@@ -56,12 +56,12 @@ class ORM(Broker):
def enqueue(self, task): def enqueue(self, task):
package = self.get_connection().create( package = self.get_connection().create(
key=self.list_key, payload=task, lock=_timeout() key=self.list_key, payload=task, lock=timezone.now()
) )
return package.pk return package.pk
def dequeue(self): def dequeue(self):
tasks = self.get_connection().filter(key=self.list_key, lock__lt=_timeout())[ tasks = self.get_connection().filter(key=self.list_key, lock__lt=timezone.now())[
0 : Conf.BULK # noqa: E203 0 : Conf.BULK # noqa: E203
] ]
if tasks: if tasks:
@@ -70,7 +70,7 @@ class ORM(Broker):
if ( if (
self.get_connection() self.get_connection()
.filter(id=task.id, lock=task.lock) .filter(id=task.id, lock=task.lock)
.update(lock=timezone.now()) .update(lock=_timeout())
): ):
task_list.append((task.pk, task.payload)) task_list.append((task.pk, task.payload))
# else don't process, as another cluster has been faster than us on # else don't process, as another cluster has been faster than us on