[backend] Make process launch an DB atomic transaction

beat/web/backend/models.py

@@ -424,29 +424,31 @@ class Worker(models.Model):
         if settings.DEBUG: cmdline += ['-vv']
 
         # start newly assigned job splits
-        for split in JobSplit.objects.select_for_update().filter(worker=self,
-            status=Job.QUEUED, start_date__isnull=True,
-            process_id__isnull=True):
-            execute = pick_execute(split, environments)
-            if execute is None:
-                message = "Environment `%s' is not available for split " \
-                    "%d/%d running at worker `%s', for block `%s' of " \
-                    "experiment `%s': %s" % \
-                    (split.job.block.environment,
-                        split.split_index+1,
-                        split.job.block.required_slots,
-                        self,
-                        split.job.block.name,
-                        split.job.block.experiment.fullname(),
-                        "Available environments are `%s'" % \
-                            '|'.join(environments.keys()),
-                            )
-                split.end(Result(status=1, usrerr=settings.DEFAULT_USER_ERROR,
-                  syserr=message))
-                continue
-
-            # if we get to this point, then we launch the user process
-            subprocess.Popen(cmdline + [execute, str(split.pk)])
+        with transaction.atomic():
+            splits = JobSplit.objects.select_for_update().filter(worker=self,
+                status=Job.QUEUED, start_date__isnull=True,
+                process_id__isnull=True)
+            for split in splits:
+                execute = pick_execute(split, environments)
+                if execute is None:
+                    message = "Environment `%s' is not available for split " \
+                        "%d/%d running at worker `%s', for block `%s' of " \
+                        "experiment `%s': %s" % \
+                        (split.job.block.environment,
+                            split.split_index+1,
+                            split.job.block.required_slots,
+                            self,
+                            split.job.block.name,
+                            split.job.block.experiment.fullname(),
+                            "Available environments are `%s'" % \
+                                '|'.join(environments.keys()),
+                                )
+                    split.end(Result(status=1,
+                      usrerr=settings.DEFAULT_USER_ERROR, syserr=message))
+                    continue
+
+                # if we get to this point, then we launch the user process
+                subprocess.Popen(cmdline + [execute, str(split.pk)])
 
         # cleans-up zombie processes that may linger
         _cleanup_zombies()