wip

Author: codingjoe

galahad/celery.py

@@ -12,19 +12,19 @@
 
 def jitter():
     """Return a random number between 0 and 1."""
-    return random.randrange(1)
+    return random.randrange(2)
 
 
 def backoff(retries):
     """Return an exponentially growing number limited to 600 plus a random jitter."""
     return min(600, 2 ** retries) + jitter()
 
 
-@shared_task(bind=True, ignore_results=True)
-def task_wrapper(self, task_pk, process_pk, retries=0):
+@shared_task(bind=True, ignore_results=True, max_retries=None)
+def task_wrapper(self, task_pk, process_pk):
     with locking.lock(process_pk) as lock_result:
+        countdown = backoff(self.request.retries)
         if not lock_result:
-            countdown = backoff(self.request.retries)
             logger.info("Process is locked, retrying in %s seconds", countdown)
             self.retry(countdown=countdown)
         Task = apps.get_model('galahad', 'Task')
@@ -43,9 +43,8 @@ def task_wrapper(self, task_pk, process_pk, retries=0):
             logger.exception("Execution of %r failed", task)
         else:
             if result is False:
-                countdown = backoff(retries)
                 logger.info("Task returned False, retrying in %s seconds", countdown)
-                transaction.on_commit(lambda: task.enqueue(countdown=countdown, retires=retries+1))
+                transaction.on_commit(lambda: self.retry(countdown=countdown))
                 return
             elif result is True:
                 result = None

galahad/conf.py

@@ -30,8 +30,3 @@ class GalahadAppConfig(AppConf):
     """
     Queue name in which all machine tasks will be queued.
     """
-
-    GALAHAD_CELERY_QUEUE_NAME = 'celery'
-    """
-    Queue name in which all machine tasks will be queued.
-    """

galahad/migrations/0001_initial.py

@@ -23,9 +23,6 @@ class Migration(migrations.Migration):
                 ('started', models.DateTimeField(auto_now_add=True, db_index=True)),
                 ('completed', models.DateTimeField(blank=True, db_index=True, editable=False, null=True)),
             ],
-            options={
-                'abstract': False,
-            },
         ),
         migrations.CreateModel(
             name='Task',

galahad/models.py

@@ -59,18 +59,7 @@ def __new__(cls, name, bases, attrs, **kwargs):
         return klass
 
 
-class AbstractProcessState(models.Model, metaclass=BaseProcess):
-    id = models.BigAutoField(primary_key=True, editable=False)
-    started = models.DateTimeField(auto_now_add=True, db_index=True)
-    completed = models.DateTimeField(blank=True, null=True, editable=False, db_index=True)
-
-    task_set = GenericRelation('galahad.Task', object_id_field='_process_id')
-
-    class Meta:
-        abstract = True
-
-
-class Process(AbstractProcessState):
+class Process(models.Model, metaclass=BaseProcess):
     """
     The `Process` object holds the state of a workflow instances.
 
@@ -80,6 +69,11 @@ class Process(AbstractProcessState):
     Processes are also the vehicle for the other two components tasks and
     :attr:`.edges`.
     """
+    id = models.BigAutoField(primary_key=True, editable=False)
+    started = models.DateTimeField(auto_now_add=True, db_index=True)
+    completed = models.DateTimeField(blank=True, null=True, editable=False, db_index=True)
+
+    task_set = GenericRelation('galahad.Task', object_id_field='_process_id')
 
     edges = None
     """
@@ -363,7 +357,7 @@ def fail(self):
         self.stacktrace = "".join(tb)
         self.save(update_fields=['failed', 'exception', 'stacktrace'])
 
-    def enqueue(self, countdown=None, eta=None, retries=0):
+    def enqueue(self, countdown=None, eta=None):
         """
         Schedule the tasks for execution.
 
@@ -379,7 +373,7 @@ def enqueue(self, countdown=None, eta=None, retries=0):
 
         """
         return celery.task_wrapper.apply_async(
-            args=(self.pk, self._process_id, retries),
+            args=(self.pk, self._process_id),
             countdown=countdown,
             eta=eta,
             queue=settings.GALAHAD_CELERY_QUEUE_NAME,

tests/testapp/migrations/0001_initial.py

@@ -0,0 +1,67 @@
+# Generated by Django 2.1.3 on 2018-11-15 10:51
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = [
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+        ('galahad', '0001_initial'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='GatewayProcess',
+            fields=[
+                ('process_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, serialize=False, to='galahad.Process')),
+            ],
+            bases=('galahad.process',),
+        ),
+        migrations.CreateModel(
+            name='LoopProcess',
+            fields=[
+                ('process_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, serialize=False, to='galahad.Process')),
+                ('counter', models.PositiveIntegerField(default=0)),
+            ],
+            bases=('galahad.process',),
+        ),
+        migrations.CreateModel(
+            name='SimpleProcess',
+            fields=[
+                ('process_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, serialize=False, to='galahad.Process')),
+            ],
+            bases=('galahad.process',),
+        ),
+        migrations.CreateModel(
+            name='SplitJoinProcess',
+            fields=[
+                ('process_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, serialize=False, to='galahad.Process')),
+                ('parallel_task_value', models.PositiveIntegerField(default=0)),
+            ],
+            bases=('galahad.process',),
+        ),
+        migrations.CreateModel(
+            name='WaitProcess',
+            fields=[
+                ('process_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, serialize=False, to='galahad.Process')),
+                ('parallel_task_value', models.PositiveIntegerField(default=0)),
+            ],
+            bases=('galahad.process',),
+        ),
+        migrations.CreateModel(
+            name='WelcomeProcess',
+            fields=[
+                ('process_ptr', models.OneToOneField(auto_created=True, on_delete=django.db.models.deletion.CASCADE, parent_link=True, primary_key=True, serialize=False, to='galahad.Process')),
+                ('user', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'abstract': False,
+            },
+            bases=('galahad.process', models.Model),
+        ),
+    ]