Skip to content

Commit 5cdaa09

Browse files
authored
Several fixes to worker logic (#57)
* Fix worker stop, timestamp handling, and imports Preserve explicit timestamp values and strengthen worker shutdown behavior. Key changes: - recording_manager: use timestamp if timestamp is not None else time.time() so falsy timestamps (e.g. 0) are preserved instead of being overridden. - services/dlc_processor: import DLCLive at module top and enable profiling; make _stop_worker return a boolean and have reset() check that return value to avoid resetting the DLCLive instance while the worker thread is still alive (with warnings/logging). Adjusted comments about thread termination. - gui/main_window: use a truthiness check for dlc_cam_id to avoid treating empty values as valid IDs. These updates prevent spurious timestamp overrides, avoid unsafe DLCLive resets when threads are still running, and clarify import/profiling intent. * Add WorkerState enum for worker lifecycle Introduce a WorkerState enum and import Enum/auto in dlc_processor.py to represent worker lifecycle states (STOPPED, STARTING, RUNNING, STOPPING, FAULTED). This centralizes state representation for processing workers and improves clarity for future state management. * Improve DLCLive processor lifecycle and safety Add explicit worker lifecycle management and synchronization to DLCLive processor: introduce WorkerState, a lifecycle Lock, and stop/start state transitions. Make _start/_stop behavior safer by snapshotting the queue, returning boolean on stop, joining the thread with fault handling, and avoiding races by using a local queue reference in the worker loop. Harden enqueue_frame to respect lifecycle/state/stop flags and reduce copies by reusing a single frame copy and enqueuing enqueue timestamp. Improve DLCLive initialization error handling (set FAULTED on failure) and ensure caller-visible errors when DLCLive is not initialized. Prevent configuring the processor while running and add logging/warnings for shutdown/termination issues. Minor cleanup: adjust where initialized/state flags are set and remove redundant checks. * Restart worker if dead and clear stop event When queuing work, ensure the background worker is started if the worker thread is None or not alive by calling _start_worker_locked(frame_c, timestamp). Also clear the _stop_event in both the early-return (no thread) and final cleanup paths when stopping so a stale stop flag doesn't prevent future restarts. * Mark worker FAULTED on queue errors Improve error handling in the worker loop when the queue is missing or queue.get raises unexpected exceptions. When q is None the code now logs a clearer message, sets the worker state to WorkerState.FAULTED under the lifecycle lock, and emits an error signal. Also adds a broad except around queue.get to log the exception, set FAULTED state, emit the error message, and break the loop to avoid silent failures. * Force-terminate frozen camera threads on stop Make multi-camera shutdown more robust by iterating threads with their camera IDs, skipping non-running threads, and quitting each thread. If a thread fails to stop within 5000ms, log an error and call terminate() then wait an additional 1000ms to ensure it stops. Also simplify the start() docstring. * Count unique cameras when starting multi-camera Previously _expected_cameras was set to len(active_settings), which could overcount when multiple CameraSettings referred to the same physical camera. This change computes a set of unique camera IDs (via get_camera_id) and sets _expected_cameras to its length, ensuring duplicates aren't double-counted. * Improve VideoRecorder shutdown and writer loop Add robust shutdown and error handling for the video recorder. Prevent writes after stop by checking the stop event early; record and raise encoding errors as before. Refactor stop() to use local vars, signal the writer via a sentinel, join the writer thread safely, and avoid saving timestamps prematurely. Rework _writer_loop to handle a missing queue, broaden exception handling when pulling/writing items, ensure q.task_done() is always called, properly handle the sentinel, update encoding stats and logging, and perform final cleanup (save timestamps, clear queue and thread) when exiting. * Refactor writer thread error handling and stats Consolidate the writer-thread loop and tighten up exception handling. Removed the conditional _save_timestamps path during teardown. In the writer loop sentinel handling and frame processing are unified; on write errors the exception is stored under _stats_lock, logged, and stop_event/stop_now are set. The stats updates (frames_written, total_latency, last_latency, written_times, frame_timestamps and periodic FPS computation) were moved into the same control flow to ensure consistent accounting and termination behavior. * Revert DLCLive imports due to unguarded torch * Improve camera thread cleanup and recorder finalization Add cooperative shutdown timeouts and robust cleanup for camera threads: introduce QUIT_WAIT_MS and TERMINATE_WAIT_MS, connect thread.finished to a new _cleanup_camera that removes frame data and deletes worker/thread QObjects, and enhance stop() to wait, terminate, and log/retain references if threads refuse to terminate (avoids use-after-free/segfaults). Adjust per-camera shutdown checks and ensure frames/timestamps are cleared only after safe termination. In video_recorder, remove premature clearing of the queue and _writer_thread during finalization to avoid unsafe cleanup while writer thread may still be exiting. * Update dlc_processor.py * Add lifecycle management & robust writer shutdown Introduce lifecycle locking and an "abandoned" state to VideoRecorder to prevent unsafe restarts and improve shutdown handling. start() now acquires a lifecycle lock, refuses to restart when a leftover thread is marked abandoned, and initializes/reset writer state inside the lock. stop() now marks the recorder as abandoned and records an encode error if the writer thread fails to terminate within the timeout, and ensures timestamps are saved. The writer loop was simplified to break directly on sentinel or encode errors and now closes WriteGear during finalization with exception handling. Miscellaneous reordering and small bookkeeping fixes were made to improve safety and logging during shutdown. * Improve lifecycle handling and stale-writer cleanup dlclivegui/services/dlc_processor.py: Move the frame copy and enqueue timestamp into the lifecycle lock to avoid unnecessary copying when the processor is stopped; on initialization errors set WorkerState.FAULTED under the lifecycle lock before emitting the error. dlclivegui/services/multi_camera_controller.py: Reformat the running-cameras shutdown check into a multi-line condition (keeps the same intent of detecting normal shutdown when no cameras are started and all threads are stopped). dlclivegui/services/video_recorder.py: Add best-effort cleanup for a stale video writer in start() by attempting to call its close() and logging any errors before clearing writer/queue/thread references. Wrap stop() logic with the lifecycle lock to avoid races when stopping, preserve timeout/abandon behavior for a non-terminating writer thread, and ensure writer closure and timestamp saving are handled with proper error logging. * Cleanup stale camera workers on start/shutdown Detect and remove stale camera worker/thread state when starting cameras and during shutdown. _start_camera now checks the existing thread for a cam_id and calls _cleanup_camera if the worker exists but the thread is missing or not running, and updates the log message to indicate the camera is already running. The shutdown loop likewise invokes _cleanup_camera for None or not-running threads and after successful quit/terminate waits to ensure worker/thread entries are cleared. These changes prevent stale entries from blocking restarts and ensure cleaner shutdown behavior. * Make stop join timeout configurable and add test Introduce STOP_JOIN_TIMEOUT (5s) and use it for the writer thread join timeout instead of a hardcoded value. Reorder lifecycle checks in VideoRecorder.start to raise if the recorder is marked _abandoned before short-circuiting on is_running, preventing accidental restarts of abandoned recorders. Add a BlockingWriteGear fake, fixture, and a test (test_stop_timeout_marks_abandoned_and_prevents_restart) that forces a stop timeout, verifies the recorder is marked abandoned, prevents restart, and performs cleanup of the blocked writer thread. * Fix locking, camera check, and writer finalization Minimize time holding the DLCLiveProcessor lifecycle lock by moving expensive operations (frame.copy and timestamp handling) outside the lock and re-checking state before starting the worker to avoid races. Correct a duplicated condition in MultiCameraController so normal shutdown is detected using self._running. Move writer cleanup into a dedicated _finalize_writer(), add a warning when Queue.task_done() is called too many times, and centralize writer close/exception handling. * Add background reaper for stalled DLC worker Introduce STOP_WORKER_TIMEOUT and a background reaper to handle worker threads that don't terminate within the stop timeout. Increase join timeout to STOP_WORKER_TIMEOUT, set _pending_reset when a reset is requested while the worker is alive, and schedule _schedule_reap to join the thread in the background instead of immediately marking the processor as FAULTED. The reaper performs final cleanup (clearing thread, queue, state, stop event) and applies a pending DLCLive reset once the thread is reaped, with guards to ensure only one reaper runs at a time. * Cache _queue to local variable in VideoRecorder Cache self._queue to a local variable (q) and use it for subsequent operations to avoid races and repeated attribute access. Remove the redundant assert and simplify qsize lookup by relying on the cached reference. This makes queue checks and put/qsize calls consistent in a multithreaded context and prevents potential None/attribute changes between checks and use. * Prevent configure while processor running Add a lifecycle lock and guard in DLCLiveProcessor.configure to ensure the processor is stopped before reconfiguration. The method now acquires self._lifecycle_lock and raises a RuntimeError if the processor state is not WorkerState.STOPPED, preventing concurrent/race-condition updates to settings or processor while running. * Cleanup abandoned recorder and set pending reset Ensure proper cleanup and restartability when writer threads time out: VideoRecorder.stop now clears abandoned state and associated resources if the writer thread has already exited, allowing the recorder to be fully stopped and restarted. DLCLiveProcessor: moved settings/processor assignment under the lifecycle lock, and mark _pending_reset when shutdown cannot stop the worker thread so the instance can be reaped later. Minor comment tweak for STOP_WORKER_TIMEOUT semantics. Tests updated to assert abandoned is cleared and that recorder can be restarted after the writer exits. * Refactor VideoRecorder.stop lifecycle handling Detect and early-return when recorder is already stopped or recovered from an abandoned state, and reduce the time holding the lifecycle lock by moving queue/thread/writer shutdown work outside the lock. Ensure sentinel is enqueued, writer thread is joined with timeout (marking recorder as abandoned and recording an encode_error if it remains alive), close the writer cleanly, save timestamps, and finally clear lifecycle fields under the lock. Improves concurrency, cleanup correctness, and prevents restarting a recorder that was marked abandoned. * Disable profiling for now * Log late DLC cleanup; save recorder timestamps Add a warning when the DLC worker thread is cleaned up after a timeout to make late reaping visible in logs. When VideoRecorder fails to stop within the timeout, ensure the recorder is marked abandoned under the lifecycle lock and save timestamps before returning; expand the critical log to note that timestamps were saved (but may be incomplete). These changes improve observability and help preserve timing data for debugging. * Handle force-stop with leftover camera threads When a forced stop finds camera threads still running, also clear failed camera state, buffered frames and timestamps (under the frame lock), and reset expected cameras to 0. Add a critical log warning that background camera threads may still be active and emit all_stopped so the GUI can leave the running state; warn that an application restart may be required. This prevents stale state from blocking future starts while making the risk visible in logs. * Remove direct writer.close() call Stop capturing and closing the WriteGear writer from the shutdown path. The change removes the local `writer` assignment and the try/except block that called `writer.close()`, avoiding potential double-close/race-condition exceptions during shutdown. The recorder closure is handled by the finally block calling _finalize_writer * Clarify timeout comment and camera log message Tweak wording for clarity in two places: update STOP_WORKER_TIMEOUT comment to better describe waiting in the STOPPING state before scheduling background reaping, and reflow the MultiCameraController critical log message into a single, spaced string to avoid awkward concatenation. These are documentation/logging refinements only and do not change runtime behavior. Files modified: - dlclivegui/services/dlc_processor.py: adjust timeout comment - dlclivegui/services/multi_camera_controller.py: reformat and clarify critical log message
1 parent 5b5d04c commit 5cdaa09

6 files changed

Lines changed: 508 additions & 139 deletions

File tree

‎dlclivegui/gui/main_window.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1376,7 +1376,7 @@ def _on_multi_frame_ready(self, frame_data: MultiFrameData) -> None:
13761376
dlc_cam_id = selected_id
13771377
else:
13781378
dlc_cam_id = available_ids[0] if available_ids else ""
1379-
if dlc_cam_id is not None:
1379+
if dlc_cam_id:
13801380
self._inference_camera_id = dlc_cam_id
13811381
self._set_dlc_combo_to_id(dlc_cam_id)
13821382
self.statusBar().showMessage(

‎dlclivegui/gui/recording_manager.py‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -139,7 +139,7 @@ def write_frame(self, cam_id: str, frame: np.ndarray, timestamp: float | None =
139139
if not rec or not rec.is_running:
140140
return
141141
try:
142-
rec.write(frame, timestamp=timestamp or time.time())
142+
rec.write(frame, timestamp=timestamp if timestamp is not None else time.time())
143143
except Exception as exc:
144144
log.warning("Failed to write frame for %s: %s", cam_id, exc)
145145
try:

‎dlclivegui/services/dlc_processor.py‎

Lines changed: 172 additions & 37 deletions
Original file line numberDiff line numberDiff line change
@@ -21,9 +21,7 @@
2121
from dlclivegui.temp import Engine # type: ignore # TODO use main package enum when released
2222

2323
logger = logging.getLogger(__name__)
24-
25-
# Enable profiling
26-
ENABLE_PROFILING = True
24+
STOP_WORKER_TIMEOUT = 10.0 # # seconds to wait in STOPPING state before scheduling background reaping
2725

2826
try: # pragma: no cover - optional dependency
2927
from dlclive import (
@@ -34,10 +32,22 @@
3432
DLCLive = None # type: ignore[assignment]
3533

3634

35+
# Enable profiling to get more detailed timing metrics for debugging and optimization.
36+
ENABLE_PROFILING = False
37+
38+
3739
class PoseBackends(Enum):
3840
DLC_LIVE = auto()
3941

4042

43+
class WorkerState(Enum):
44+
STOPPED = auto()
45+
STARTING = auto()
46+
RUNNING = auto()
47+
STOPPING = auto()
48+
FAULTED = auto()
49+
50+
4151
@dataclass
4252
class PoseResult:
4353
pose: np.ndarray | None
@@ -140,13 +150,20 @@ class DLCLiveProcessor(QObject):
140150

141151
def __init__(self) -> None:
142152
super().__init__()
153+
# DLCLive instance and config
143154
self._settings = DLCProcessorSettings()
144155
self._dlc: Any | None = None
145156
self._processor: Any | None = None
157+
# Worker thread and queue
146158
self._queue: queue.Queue[Any] | None = None
147159
self._worker_thread: threading.Thread | None = None
160+
self._state = WorkerState.STOPPED
161+
self._lifecycle_lock = threading.Lock()
148162
self._stop_event = threading.Event()
149163
self._initialized = False
164+
## Worker cleanup
165+
self._reaping = False
166+
self._pending_reset = False
150167

151168
# Statistics tracking
152169
self._frames_enqueued = 0
@@ -169,12 +186,22 @@ def get_model_backend(model_path: str) -> Engine:
169186
return Engine.from_model_path(model_path)
170187

171188
def configure(self, settings: DLCProcessorSettings, processor: Any | None = None) -> None:
172-
self._settings = settings
173-
self._processor = processor
189+
with self._lifecycle_lock:
190+
if self._state != WorkerState.STOPPED:
191+
raise RuntimeError("Cannot configure DLCLiveProcessor while it is running. Please stop it first.")
192+
self._settings = settings
193+
self._processor = processor
174194

175195
def reset(self) -> None:
176196
"""Stop the worker thread and drop the current DLCLive instance."""
177-
self._stop_worker()
197+
stopped = self._stop_worker()
198+
if not stopped:
199+
with self._lifecycle_lock:
200+
self._pending_reset = True
201+
logger.warning(
202+
"Reset requested but worker thread is still alive; skipping DLCLive reset to avoid potential issues."
203+
)
204+
return
178205
self._dlc = None
179206
self._initialized = False
180207
with self._stats_lock:
@@ -191,22 +218,48 @@ def reset(self) -> None:
191218
self._processor_overhead_times.clear()
192219

193220
def shutdown(self) -> None:
194-
self._stop_worker()
221+
stopped = self._stop_worker()
222+
if not stopped:
223+
with self._lifecycle_lock:
224+
self._pending_reset = True
225+
logger.warning(
226+
"Shutdown requested but worker thread is still alive; DLCLive instance may not be fully released."
227+
)
228+
return
195229
self._dlc = None
196230
self._initialized = False
197231

198232
def enqueue_frame(self, frame: np.ndarray, timestamp: float) -> None:
199-
# Start worker on first frame
200-
if self._worker_thread is None:
201-
self._start_worker(frame.copy(), timestamp)
202-
return
203-
204-
# As long as worker and queue are ready, ALWAYS enqueue
205-
if self._queue is None:
233+
# Keep lifecycle lock held only for quick state checks and snapshots.
234+
with self._lifecycle_lock:
235+
if self._state in (WorkerState.STOPPING, WorkerState.FAULTED) or self._stop_event.is_set():
236+
return
237+
t = self._worker_thread
238+
q = self._queue
239+
should_start = t is None or not t.is_alive()
240+
241+
frame_c = frame.copy()
242+
enq_time = time.perf_counter()
243+
244+
if should_start:
245+
# Re-acquire the lifecycle lock to safely (re)start the worker if needed.
246+
with self._lifecycle_lock:
247+
# Re-check state in case it changed while we were copying the frame.
248+
if self._state in (WorkerState.STOPPING, WorkerState.FAULTED) or self._stop_event.is_set():
249+
return
250+
t = self._worker_thread
251+
if t is None or not t.is_alive():
252+
# _start_worker_locked expects the lifecycle lock to be held.
253+
self._start_worker_locked(frame_c, timestamp)
254+
return
255+
# Worker is now running; refresh queue snapshot.
256+
q = self._queue
257+
258+
if q is None:
206259
return
207260

208261
try:
209-
self._queue.put_nowait((frame.copy(), timestamp, time.perf_counter()))
262+
q.put_nowait((frame_c, timestamp, enq_time))
210263
with self._stats_lock:
211264
self._frames_enqueued += 1
212265
except queue.Full:
@@ -264,12 +317,13 @@ def get_stats(self) -> ProcessorStats:
264317
avg_processor_overhead=avg_proc_overhead,
265318
)
266319

267-
def _start_worker(self, init_frame: np.ndarray, init_timestamp: float) -> None:
320+
def _start_worker_locked(self, init_frame: np.ndarray, init_timestamp: float) -> None:
321+
# lifecycle_lock must already be held
268322
if self._worker_thread is not None and self._worker_thread.is_alive():
269323
return
270-
271324
self._queue = queue.Queue(maxsize=1)
272325
self._stop_event.clear()
326+
self._state = WorkerState.STARTING
273327
self._worker_thread = threading.Thread(
274328
target=self._worker_loop,
275329
args=(init_frame, init_timestamp),
@@ -278,19 +332,67 @@ def _start_worker(self, init_frame: np.ndarray, init_timestamp: float) -> None:
278332
)
279333
self._worker_thread.start()
280334

281-
def _stop_worker(self) -> None:
282-
if self._worker_thread is None:
283-
return
335+
def _start_worker(self, init_frame: np.ndarray, init_timestamp: float) -> None:
336+
with self._lifecycle_lock:
337+
self._start_worker_locked(init_frame, init_timestamp)
338+
339+
def _stop_worker(self) -> bool:
340+
with self._lifecycle_lock:
341+
t = self._worker_thread
342+
if t is None:
343+
self._state = WorkerState.STOPPED
344+
self._stop_event.clear()
345+
return True
346+
self._state = WorkerState.STOPPING
347+
self._stop_event.set()
348+
349+
t.join(timeout=STOP_WORKER_TIMEOUT)
350+
if t.is_alive():
351+
qsize = self._queue.qsize() if self._queue is not None else -1
352+
logger.warning("DLC worker thread did not terminate cleanly (qsize=%s)", qsize)
353+
self._schedule_reap(t)
354+
return False
355+
356+
# Normal cleanup
357+
with self._lifecycle_lock:
358+
self._worker_thread = None
359+
self._queue = None
360+
self._state = WorkerState.STOPPED
361+
self._stop_event.clear()
362+
return True
284363

285-
self._stop_event.set()
364+
def _schedule_reap(self, t: threading.Thread) -> None:
365+
with self._lifecycle_lock:
366+
if self._reaping:
367+
return
368+
self._reaping = True
286369

287-
# Just wait for the timed get() loop to observe the flag and drain
288-
self._worker_thread.join(timeout=2.0)
289-
if self._worker_thread.is_alive():
290-
logger.warning("DLC worker thread did not terminate cleanly")
370+
# ensure only one reaper
371+
def reap():
372+
try:
373+
t.join() # wait without timeout in background
374+
with self._lifecycle_lock:
375+
# only clean if we're still stopping this thread
376+
if self._worker_thread is t:
377+
self._worker_thread = None
378+
self._queue = None
379+
self._state = WorkerState.STOPPED
380+
self._stop_event.clear()
381+
382+
if self._pending_reset:
383+
self._dlc = None
384+
self._initialized = False
385+
self._pending_reset = False
386+
387+
logger.warning("DLC worker thread stopped after the timeout and was cleaned up late.")
291388

292-
self._worker_thread = None
293-
self._queue = None
389+
finally:
390+
with self._lifecycle_lock:
391+
self._reaping = False
392+
393+
logger.debug("[Stop worker] DLC worker thread reaped; processor is STOPPED again")
394+
395+
threading.Thread(target=reap, name="DLCLiveReaper", daemon=True).start()
294396

295397
@contextmanager
296398
def _timed_processor(self):
@@ -333,6 +435,8 @@ def _process_frame(
333435
Single source of truth for: inference -> (optional) processor timing -> signal emit -> stats.
334436
Updates: frames_processed, latency, processing timeline, profiling metrics.
335437
"""
438+
if self._dlc is None:
439+
raise RuntimeError("DLCLive instance is not initialized.")
336440
# Time GPU inference (and processor overhead when present)
337441
with self._timed_processor() as proc_holder:
338442
inference_start = time.perf_counter()
@@ -382,8 +486,6 @@ def _process_frame(
382486
def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None:
383487
try:
384488
# -------- Initialization (unchanged) --------
385-
if DLCLive is None:
386-
raise RuntimeError("The 'dlclive' package is required for pose estimation.")
387489
if not self._settings.model_path:
388490
raise RuntimeError("No DLCLive model path configured.")
389491

@@ -408,7 +510,18 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None:
408510
if self._settings.device is not None:
409511
options["device"] = self._settings.device
410512

411-
self._dlc = DLCLive(**options)
513+
try:
514+
if DLCLive is None:
515+
raise RuntimeError(
516+
"DLCLive class is not available. Ensure the dlclive package is installed and can be imported."
517+
)
518+
self._dlc = DLCLive(**options)
519+
except Exception as exc:
520+
with self._lifecycle_lock:
521+
self._state = WorkerState.FAULTED
522+
raise RuntimeError(
523+
f"Failed to initialize DLCLive with model '{self._settings.model_path}': {exc}"
524+
) from exc
412525

413526
# First inference to initialize
414527
init_inference_start = time.perf_counter()
@@ -421,6 +534,8 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None:
421534

422535
self._initialized = True
423536
self.initialized.emit(True)
537+
with self._lifecycle_lock:
538+
self._state = WorkerState.RUNNING
424539

425540
total_init_time = time.perf_counter() - init_start
426541
logger.info(
@@ -436,18 +551,30 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None:
436551

437552
except Exception as exc:
438553
logger.exception("Failed to initialize DLCLive", exc_info=exc)
554+
with self._lifecycle_lock:
555+
self._state = WorkerState.FAULTED
439556
self.error.emit(str(exc))
440557
self.initialized.emit(False)
441558
return
442559

560+
q = (
561+
self._queue
562+
) # Assign to local to avoid issues if self._queue is set to None during shutdown while loop is still running.
563+
if q is None:
564+
logger.warning("Worker started without a queue; exiting")
565+
with self._lifecycle_lock:
566+
self._state = WorkerState.FAULTED
567+
self.error.emit("Worker started without a queue")
568+
return
569+
443570
# -------- Main processing loop: stop-flag + timed get + drain --------
444571
# NOTE: We never exit early unless _stop_event is set.
445572
while True:
446573
# If stop requested, only exit when queue is empty
447574
if self._stop_event.is_set():
448-
if self._queue is not None:
575+
if q is not None:
449576
try:
450-
frame, ts, enq = self._queue.get_nowait()
577+
frame, ts, enq = q.get_nowait()
451578
except queue.Empty:
452579
# NOW it is safe to exit
453580
break
@@ -460,18 +587,24 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None:
460587
self.error.emit(str(exc))
461588
finally:
462589
try:
463-
self._queue.task_done()
590+
q.task_done()
464591
except ValueError:
465592
pass
466593
continue # check stop_event again WITHOUT breaking
467594

468595
# Normal operation: timed get
469596
try:
470597
wait_start = time.perf_counter()
471-
item = self._queue.get(timeout=0.05)
598+
item = q.get(timeout=0.05)
472599
queue_wait_time = time.perf_counter() - wait_start
473600
except queue.Empty:
474601
continue
602+
except Exception as exc:
603+
logger.exception("Error getting item from queue", exc_info=exc)
604+
with self._lifecycle_lock:
605+
self._state = WorkerState.FAULTED
606+
self.error.emit(str(exc))
607+
break
475608

476609
try:
477610
frame, ts, enq = item
@@ -481,7 +614,7 @@ def _worker_loop(self, init_frame: np.ndarray, init_timestamp: float) -> None:
481614
self.error.emit(str(exc))
482615
finally:
483616
try:
484-
self._queue.task_done()
617+
q.task_done()
485618
except ValueError:
486619
pass
487620

@@ -518,6 +651,10 @@ def enqueue(self, frame, ts):
518651
self._proc.enqueue_frame(frame, ts)
519652

520653
def configure(self, settings: DLCProcessorSettings, scanned_processors: dict, selected_key) -> bool:
654+
with self._proc._lifecycle_lock:
655+
if self._proc._state != WorkerState.STOPPED:
656+
raise RuntimeError("Cannot configure DLCLiveProcessor while it is running. Please stop it first.")
657+
521658
processor = None
522659
if selected_key is not None and scanned_processors:
523660
try:
@@ -531,11 +668,9 @@ def configure(self, settings: DLCProcessorSettings, scanned_processors: dict, se
531668
def start(self):
532669
self._proc.reset()
533670
self.active = True
534-
self.initialized = False
535671

536672
def stop(self):
537673
self.active = False
538-
self.initialized = False
539674
self._proc.reset()
540675
self._last_pose = None
541676

0 commit comments

Comments
 (0)