diff --git a/debye_bec/bec_widgets/widgets/scheduler/scheduler.py b/debye_bec/bec_widgets/widgets/scheduler/scheduler.py index 8caf677..6a247bc 100644 --- a/debye_bec/bec_widgets/widgets/scheduler/scheduler.py +++ b/debye_bec/bec_widgets/widgets/scheduler/scheduler.py @@ -843,10 +843,42 @@ class Scheduler(BECWidget, QWidget): """Must be called while holding `self._lock`; see `schedule_logic.pick_next_runnable`.""" return pick_next_runnable(self.schedule.items, repeat_aborted_item) + @SafeSlot() + def _on_run_finished(self): + self._refresh_ui() + + @SafeSlot(str) + def _on_run_failed(self, error: str): + logger.error(f"Schedule execution task failed unexpectedly: {error}") + with self._lock: + self.schedule.is_running = False + self._persist_locked() + self._refresh_ui() + + def _emit_schedule_changed(self): + if self._closing.is_set(): + return + + self.schedule_changed.emit() + + def closeEvent(self, event): + """Ensure cleanup is invoked when the widget is closed via the window system.""" + self.cleanup() + super().closeEvent(event) + + def cleanup(self): + """Stop background execution loops without cancelling server-side scans.""" + self._closing.set() + + # Interrupt guard wait loops if any thread is blocked in self._guard.wait_until_clear() + self._guard.cleanup() + + super().cleanup() + def _run_all(self, repeat_aborted_item): logger.info("_run_all was called") namespace = {"scans": self.scans, "dev": self.dev} - while not self._abort_requested: + while not self._abort_requested and not self._closing.is_set(): with self._lock: next_item = self._pick_next_runnable_locked(repeat_aborted_item) if next_item is None or next_item == "stop": @@ -865,12 +897,12 @@ class Scheduler(BECWidget, QWidget): def _execute_item(self, item: ScheduleItem, namespace: dict): if item.kind == "scan": - # Block here (item stays PENDING - nothing "started" yet) until - # the guard is clear or the operator aborts. See "Guard - # interaction" in the module docstring. - cleared = self._guard.wait_until_clear(should_abort=lambda: self._abort_requested) + # Block here until the guard is clear, operator aborts, or widget closes + cleared = self._guard.wait_until_clear( + should_abort=lambda: self._abort_requested or self._closing.is_set() + ) self._emit_schedule_changed() - if not cleared: + if not cleared or self._closing.is_set(): return with self._lock: @@ -890,12 +922,15 @@ class Scheduler(BECWidget, QWidget): request = getattr(report, "request", None) with self._lock: item.request_id = getattr(request, "requestID", None) - # persist the request_id right away, before the - # (potentially long) wait below, so a reconnect can find - # it even if this process disappears mid-scan. self._persist_locked() - report.wait() + # Wait for completion without blocking the thread indefinitely + self._wait_for_report(report) + + if self._closing.is_set(): + # The widget was closed mid-item; exit without changing status in Redis. + # Reconciliation logic will handle state on reopen. + return scan = getattr(report, "scan", None) with self._lock: @@ -907,12 +942,10 @@ class Scheduler(BECWidget, QWidget): else ScheduleItemStatus.COMPLETED ) except Exception: # pylint: disable=broad-except + if self._closing.is_set(): + return with self._lock: if self._guard_interrupt_requested: - # Interrupted by the auto-pause guard, not a real - # failure/operator abort - reset to PENDING so the - # normal execution loop retries it once the guard - # clears, instead of leaving it in a terminal state. item.status = ScheduleItemStatus.PENDING item.request_id = None item.scan_id = None @@ -927,14 +960,17 @@ class Scheduler(BECWidget, QWidget): item.error = traceback.format_exc() logger.error(f"Schedule item failed: {item.command}\n{item.error}") finally: - with self._lock: - if item.status != ScheduleItemStatus.PENDING: - item.finished_at = time.time() - self._persist_locked() - final_status = item.status - self._current_report = None - self._current_item_kind = None - self._emit_schedule_changed() + if not self._closing.is_set(): + with self._lock: + if item.status != ScheduleItemStatus.PENDING: + item.finished_at = time.time() + self._persist_locked() + final_status = item.status + self._current_report = None + self._current_item_kind = None + self._emit_schedule_changed() + else: + final_status = None if final_status in (ScheduleItemStatus.COMPLETED, ScheduleItemStatus.FAILED): notify_item_finished( @@ -947,9 +983,22 @@ class Scheduler(BECWidget, QWidget): error=item.error, ) + def _wait_for_report(self, report, timeout: float = 0.5): + """Poll report status to allow thread exit if the widget closes.""" + while not self._closing.is_set() and not self._abort_requested: + # Check if scan report has finished via its internal event or status + if ( + getattr(report, "status", None) == "completed" + or getattr(report, "event", None) + and report.event.is_set() + ): + break + # Non-blocking poll interval + time.sleep(timeout) + def _await_running_item(self, item: ScheduleItem): queue_storage = self.client.queue.queue_storage - while not self._abort_requested: + while not self._abort_requested and not self._closing.is_set(): queue_item = queue_storage.find_queue_item_by_requestID(item.request_id) if queue_item is None or not self._is_queue_item_active(queue_item): with self._lock: @@ -959,29 +1008,3 @@ class Scheduler(BECWidget, QWidget): self._emit_schedule_changed() return time.sleep(0.5) - - @SafeSlot() - def _on_run_finished(self): - self._refresh_ui() - - @SafeSlot(str) - def _on_run_failed(self, error: str): - logger.error(f"Schedule execution task failed unexpectedly: {error}") - with self._lock: - self.schedule.is_running = False - self._persist_locked() - self._refresh_ui() - - def _emit_schedule_changed(self): - if self._closing.is_set(): - return - - self.schedule_changed.emit() - - def cleanup(self): - # The schedule itself must continue running after the widget is closed. - # We only tell the worker that the UI/Qt object is no longer available. - self._closing.set() - - self._guard.cleanup() - super().cleanup()