Merge pull request #25 from dabapps/priority-queueing

Priority queueing

Author: jordaneb

django_dbq/migrations/0003_auto_20180713_1000.py

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+# Generated by Django 1.11 on 2018-07-13 10:00
+from __future__ import unicode_literals
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('django_dbq', '0002_auto_20151016_1027'),
+    ]
+
+    operations = [
+        migrations.AlterModelOptions(
+            name='job',
+            options={'ordering': ['-priority', 'created']},
+        ),
+        migrations.AddField(
+            model_name='job',
+            name='priority',
+            field=models.SmallIntegerField(db_index=True, default=0),
+        ),
+    ]

django_dbq/models.py

@@ -20,7 +20,6 @@
 
 
 class JobManager(models.Manager):
-
     def get_ready_or_none(self, queue_name, max_retries=3):
         """
         Get a job in state READY or NEW for a given queue. Supports retrying in case of database deadlock
@@ -40,7 +39,7 @@ def get_ready_or_none(self, queue_name, max_retries=3):
         retries_left = max_retries
         while True:
             try:
-                return self.select_for_update().filter(queue_name=queue_name, state__in=(Job.STATES.READY, Job.STATES.NEW)).first()
+                return self.to_process(queue_name).first()
             except Exception as e:
                 if retries_left == 0:
                     raise
@@ -56,6 +55,9 @@ def delete_old(self):
         logger.info("Deleting all job in states %s created before %s", ", ".join(delete_jobs_in_states), delete_jobs_created_before.isoformat())
         Job.objects.filter(state__in=delete_jobs_in_states, created__lte=delete_jobs_created_before).delete()
 
+    def to_process(self, queue_name):
+        return self.select_for_update().filter(queue_name=queue_name, state__in=(Job.STATES.READY, Job.STATES.NEW))
+
 
 class Job(models.Model):
 
@@ -69,9 +71,10 @@ class Job(models.Model):
     next_task = models.CharField(max_length=100, blank=True)
     workspace = JSONField(null=True)
     queue_name = models.CharField(max_length=20, default='default', db_index=True)
+    priority = models.SmallIntegerField(default=0, db_index=True)
 
     class Meta:
-        ordering = ['created']
+        ordering = ['-priority', 'created']
 
     objects = JobManager()
 

django_dbq/tests.py

@@ -138,6 +138,45 @@ def test_get_next_ready_job(self):
 
         self.assertEqual(Job.objects.get_ready_or_none('default'), expected)
 
+    def test_gets_jobs_in_priority_order(self):
+        job_1 = Job.objects.create(name='testjob')
+        job_2 = Job.objects.create(name='testjob', state=Job.STATES.PROCESSING)
+        job_3 = Job.objects.create(name='testjob', priority=3)
+        job_4 = Job.objects.create(name='testjob', priority=2)
+        self.assertEqual({
+            job for job in Job.objects.to_process('default')
+        }, {
+            job_3, job_4, job_1
+        })
+        self.assertEqual(Job.objects.get_ready_or_none('default'), job_3)
+        self.assertFalse(Job.objects.to_process('default').filter(id=job_2.id).exists())
+
+    def test_gets_jobs_in_negative_priority_order(self):
+        job_1 = Job.objects.create(name='testjob')
+        job_2 = Job.objects.create(name='testjob', state=Job.STATES.PROCESSING)
+        job_3 = Job.objects.create(name='testjob', priority=-2)
+        job_4 = Job.objects.create(name='testjob', priority=1)
+        self.assertEqual({
+            job for job in Job.objects.to_process('default')
+        }, {
+            job_4, job_3, job_1
+        })
+        self.assertEqual(Job.objects.get_ready_or_none('default'), job_4)
+        self.assertFalse(Job.objects.to_process('default').filter(id=job_2.id).exists())
+
+    def test_gets_jobs_in_priority_and_date_order(self):
+        job_1 = Job.objects.create(name='testjob', priority=3)
+        job_2 = Job.objects.create(name='testjob', state=Job.STATES.PROCESSING, priority=3)
+        job_3 = Job.objects.create(name='testjob', priority=3)
+        job_4 = Job.objects.create(name='testjob', priority=3)
+        self.assertEqual({
+            job for job in Job.objects.to_process('default')
+        }, {
+            job_1, job_3, job_4
+        })
+        self.assertEqual(Job.objects.get_ready_or_none('default'), job_1)
+        self.assertFalse(Job.objects.to_process('default').filter(id=job_2.id).exists())
+
     def test_get_next_ready_job_created(self):
         """
         Created jobs should be picked too.