12 hour inactivity shutdown

mephisto/data_model/task_run.py

@@ -82,6 +82,15 @@ class TaskRunArgs:
         default=30 * 60,
         metadata={"help": "Time that workers have to work on your task once accepted."},
     )
+    no_submission_patience: int = field(
+        default=60 * 60 * 12,
+        metadata={
+            "help": (
+                "How long to wait between task submissions before shutting the run down "
+                "for a presumed issue. Value in seconds, default 12 hours. "
+            )
+        },
+    )
     allowed_concurrent: int = field(
         default=0,
         metadata={

mephisto/operations/client_io_handler.py

@@ -96,6 +96,7 @@ def __init__(self, db: "MephistoDB"):
         self.request_id_to_packet: Dict[str, Packet] = {}  # For metrics purposes
 
         self.is_shutdown = False
+        self.last_submission_time = time.time()  # For patience tracking
 
         # Deferred initializiation
         self._live_run: Optional["LiveTaskRun"] = None
@@ -383,6 +384,7 @@ def _on_message(self, packet: Packet, channel_id: str):
             elif packet.type == PACKET_TYPE_SUBMIT_UNIT:
                 self._on_submit_unit(packet, channel_id)
                 self.log_metrics_for_packet(packet)
+                self.last_submission_time = time.time()
             elif packet.type == PACKET_TYPE_SUBMIT_METADATA:
                 self._on_submit_metadata(packet)
             elif packet.type == PACKET_TYPE_MEPHISTO_BOUND_LIVE_UPDATE:

mephisto/operations/operator.py

@@ -329,6 +329,13 @@ async def _track_and_kill_runs(self):
             runs_to_check = list(self._task_runs_tracked.values())
             for tracked_run in runs_to_check:
                 await asyncio.sleep(0.01)  # Low pri, allow to be interrupted
+                patience = tracked_run.task_run.get_task_args().no_submission_patience
+                if patience < time.time() - tracked_run.client_io.last_submission_time:
+                    logger.warn(
+                        f"It has been greater than the set no_submission_patience of {patience} "
+                        f"for {tracked_run.task_run} since the last submission, shutting this run down."
+                    )
+                    tracked_run.force_shutdown = True
                 if not tracked_run.force_shutdown:
                     task_run = tracked_run.task_run
                     if tracked_run.task_launcher.finished_generators is False:
@@ -344,6 +351,7 @@ async def _track_and_kill_runs(self):
                 tracked_run.client_io.shutdown()
                 tracked_run.worker_pool.shutdown()
                 tracked_run.task_launcher.shutdown()
+                tracked_run.task_launcher.expire_units()
                 tracked_run.architect.shutdown()
                 del self._task_runs_tracked[task_run.db_id]
             await asyncio.sleep(RUN_STATUS_POLL_TIME)

test/core/test_operator.py

@@ -261,6 +261,50 @@ def test_run_job_not_concurrent(self):
         assignment = task_run.get_assignments()[0]
         self.assertEqual(assignment.get_status(), AssignmentState.COMPLETED)
 
+    @patch("mephisto.operations.operator.RUN_STATUS_POLL_TIME", 1.5)
+    def test_patience_shutdown(self):
+        """Ensure that a job shuts down if patience is exceeded"""
+        self.operator = Operator(self.db)
+        config = MephistoConfig(
+            blueprint=MockBlueprintArgs(num_assignments=1, is_concurrent=False),
+            provider=MockProviderArgs(requester_name=self.requester_name),
+            architect=MockArchitectArgs(should_run_server=True),
+            task=TaskRunArgs(
+                task_title="title",
+                task_description="This is a description",
+                task_reward=0.3,
+                task_tags="1,2,3",
+                submission_timeout=5,
+                no_submission_patience=1,  # Expire in a second
+            ),
+        )
+
+        self.operator.launch_task_run(OmegaConf.structured(config))
+        tracked_runs = self.operator.get_running_task_runs()
+        self.assertEqual(len(tracked_runs), 1, "Run not launched")
+        task_run_id, tracked_run = list(tracked_runs.items())[0]
+
+        self.assertIsNotNone(tracked_run)
+        self.assertIsNotNone(tracked_run.task_launcher)
+        self.assertIsNotNone(tracked_run.task_runner)
+        self.assertIsNotNone(tracked_run.architect)
+        self.assertIsNotNone(tracked_run.task_run)
+        self.assertEqual(tracked_run.task_run.db_id, task_run_id)
+
+        # Give a few seconds for the operator to shutdown
+        start_time = time.time()
+        self.operator._wait_for_runs_in_testing(TIMEOUT_TIME)
+        self.assertLess(
+            time.time() - start_time, TIMEOUT_TIME, "Task shutdown not enacted in time"
+        )
+
+        # Ensure the task run was forced to shut down
+        task_run = tracked_run.task_run
+        self.assertTrue(tracked_run.force_shutdown)
+        assignment = task_run.get_assignments()[0]
+        unit = assignment.get_units()[0]
+        self.assertEqual(unit.get_status(), AssignmentState.EXPIRED)
+
     @patch("mephisto.operations.operator.RUN_STATUS_POLL_TIME", 1.5)
     def test_run_jobs_with_restrictions(self):
         """Ensure allowed_concurrent and maximum_units_per_worker work"""