How to stop a running pipeline on a separate thread the right way?

[Wwilcz2]

I have Pipeline.from_pretrained("pyannote/speaker-diarization-3.1",use_auth_token) running on a separate thread and want somehow to stop it with a function so that I can start a new diarization process. Is there a way to do that?

[hbredin]

Kill the thread?

[Wwilcz2]

Maybe I'm doing something wrong, but I get an AssertionError when I try to thread._stop() from the main thread. And of course the thread is not stopped, but the main throws and exception:

File "/usr/lib/python3.10/threading.py", line 1047, in _stop
    assert not lock.locked()
AssertionError
The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/dist-packages/gradio/queueing.py", line 501, in process_events
    response = await self.call_prediction(awake_events, batch)
  File "/usr/local/lib/python3.10/dist-packages/gradio/queueing.py", line 465, in call_prediction
    raise Exception(str(error) if show_error else None) from error
Exception: None

[hbredin]

Looks like a problem with Gradio, not pyannote.

[Wwilcz2]

Not really, even if I try to _stop() it with a function without even running Gradio, I get the assertion error:

[Wwilcz2]

Worked if "force feeding" the thread with an exception. Here is my example code, but I wouldn't recommend using this. It's not very elegant and managing the leftover resources are a thing to keep an eye on.

import ctypes

pipeline_threads = [] #a list that keeps all threads
def clear_pipeline_threads():
  count = 0
  for thread in pipeline_threads:
    count += 1
    if thread.is_alive():
      print(f"Killing thread: {thread}")
      exc = ctypes.py_object(SystemExit)
      res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(thread.ident), exc)
      if res == 0:
          raise ValueError("Nonexistent thread ID")
      elif res > 1:
          ctypes.pythonapi.PyThreadState_SetAsyncExc(thread.ident, None)
          raise SystemError("PyThreadState_SetAsyncExc failed")
    else:
      print(f"Thread dead: {thread}")
  #pipeline_threads.clear()
  if count == 0:
    print("No threads alive")


clear_pipeline_threads()