Stop the pipeline in __del__ to avoid Python hangs at exit (#122)

src/spdl/dataloader/_pipeline.py

@@ -272,6 +272,21 @@ def __init__(
     def __str__(self) -> str:
         return self._str
 
+    def __del__(self) -> None:
+        """Stop the pipeline if running."""
+        if _EventLoopState.STARTED <= self._event_loop_state < _EventLoopState.STOPPED:
+            warnings.warn(
+                f"Pipeline ({repr(self)}) is running in the background, but "
+                "there is no valid reference pointing the foreground object. "
+                "Stopping the background thread. "
+                "It is strongly advised to stop the pipeline explicity, "
+                "using the `auto_stop` context manager. "
+                "If you are using a framework and you cannot use the "
+                "context manager, try calling `stop` in done callback and "
+                "error callback."
+            )
+            self.stop()
+
     def start(self, *, timeout: float | None = None, **kwargs) -> None:
         """Start the pipeline in background thread.
 

tests/spdl_unittest/dataloader/pipeline_test.py

@@ -2,6 +2,7 @@
 import random
 import time
 from contextlib import asynccontextmanager
+from multiprocessing import Process
 
 import pytest
 
@@ -1312,3 +1313,20 @@ def test_pipeline_stop_multiple_times():
     pipeline.stop()
     pipeline.stop()
     pipeline.stop()
+
+
+def _run_pipeline_without_closing():
+    pipeline = PipelineBuilder().add_source(range(10)).add_sink(1).build()
+    pipeline.start()
+
+
+def test_pipeline_no_close():
+    """Python interpreter can terminate even when Pipeline is not explicitly closed."""
+
+    p = Process(target=_run_pipeline_without_closing)
+    p.start()
+    p.join(timeout=10)
+
+    if p.exitcode is None:
+        p.kill()
+        raise RuntimeError("Process did not self-terminate.")