Add context manager (__enter__ / __exit__) support to InferencePipeline

Aperta Adatta ai principianti
  • Qualcosa non va?
#2,744 1 commento 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@KushagraKanaujia ci sta già lavorando.

Dal 7/9/2026.

  • #2939 di @KushagraKanaujia — aperta

Valutazione

Difficoltà
2/5
Tempo stimato
1-3 ore
Idoneità per principianti
76/100
Tipo di issue
Funzionalità
Chiarezza
Specificata chiaramente
Stato di attività
Tranquilla
Stack tecnologico
python
Ambito
computer-vision

Direzione di ricerca

Individua la classe InferencePipeline e analizza i metodi esistenti del ciclo di vita start(), terminate() e join(), quindi esamina il precedente del context manager in inference/core/interfaces/stream/sinks.py. Aggiungi test mirati per la pulizia nei percorsi normale e di eccezione, verificando che terminate() preceda join() e che le eccezioni vengano rilanciate.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

Title

Add context manager (__enter__ / __exit__) support to InferencePipeline

Description

Search before asking
  • I searched the existing issues and found no similar request for context manager support on InferencePipeline.
Feature Description

Add the Python context manager protocol (__enter__ / __exit__) to InferencePipeline so it can be used with with statements, guaranteeing resource cleanup even when exceptions occur.

Motivation — the resource leak problem

Currently, the only way to run an InferencePipeline is the manual lifecycle pattern:

pipeline = InferencePipeline.init(...)
pipeline.start()
pipeline.join()

If any exception is raised between start() and join() — whether from user callback code, an inference error, or a sink failure — join() is never reached. This leaks resources because join() is responsible for:

  1. Joining the inference thread (self._inference_thread.join()) — line 938
  2. Joining the dispatching thread (self._dispatching_thread.join()) — line 941
  3. Calling on_pipeline_end (line 943), which:
    • Shuts down the internal ThreadPoolExecutor (thread_pool_executor.shutdown()) — inference/core/interfaces/stream/utils.py:128
    • Saves workflow profiling traces to disk — utils.py:117

Without join(), orphan threads and thread pool workers keep running in the background, and profiling data is silently lost.

terminate() alone is not sufficient — it signals threads to stop (self._stop = True) and purges video source buffers, but it does not join threads or invoke on_pipeline_end. Both terminate() and join() must be called for a clean shutdown.

Precedent in the same codebase

VideoFileSink in inference/core/interfaces/stream/sinks.py (lines 537–541) already implements this pattern:

def __enter__(self) -> "VideoFileSink":
    return self

def __exit__(self, exc_type, exc_val, exc_tb) -> None:
    self.release()

Adding the same protocol to InferencePipeline would make the API consistent across the streaming subsystem.

Proposed Implementation

Add two methods to InferencePipeline:

def __enter__(self) -> "InferencePipeline":
    return self

def __exit__(self, exc_type, exc_val, exc_tb) -> None:
    self.terminate()
    self.join()

The __exit__ method calls terminate() first to signal all threads to stop (sets _stop = True, purges video source buffers), then join() to wait for thread completion and run the on_pipeline_end cleanup (thread pool shutdown, profiling save). The exception parameters are intentionally not consumed — they are re-raised automatically by the with statement after cleanup.

Example Usage

Before (current — resource leak on exception):

pipeline = InferencePipeline.init(
    video_reference="./video.mp4",
    model_id="my-model/1",
    on_prediction=my_sink,
)
try:
    pipeline.start()
except Exception:
    # If start() partially succeeded (inference thread launched),
    # we leak that thread. Even if we catch the error, join() is
    # not called unless we explicitly handle it here.
    pipeline.terminate()
    pipeline.join()
    raise
pipeline.join()

After (with context manager — guaranteed cleanup):

with InferencePipeline.init(
    video_reference="./video.mp4",
    model_id="my-model/1",
    on_prediction=my_sink,
) as pipeline:
    pipeline.start()
# terminate() + join() are called automatically, even on exceptions
Alternatives Considered

  1. try/finally wrapper — works but is boilerplate-heavy and error-prone; every user must remember to call both terminate() and join() in the correct order. The context manager encapsulates this once.

  2. Making start() return a context manager — more invasive; changes the return type of an existing public API and would break callers that use start() without with.

  3. Adding cleanup to terminate() — would change the semantics of an existing method. terminate() is designed as a lightweight signal (set _stop, purge buffers); folding thread-joining and profiler-saving into it would be a breaking change for users who call terminate() from a different thread and join() separately.

Additional context
  • The change is backward-compatible — existing code using init() → start() → join() continues to work unchanged.
  • The implementation is ~8 lines of code with no new dependencies.
  • Tests can be added to verify that __exit__ calls terminate() and join() (including the exception path).
Lingua principale
Python
Stelle
2.5k
Fork
340
Merge medio
2g 11h
PR unite (30g)
92

Preparare l'ambiente

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di roboflow/inference

Tutte le issue di roboflow/inference

Issue simili

Altre issue su Python

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.