Есть ли способ убить поток?

Никогда не следует принудительно убивать поток, не сотрудничая с ним.

Удаление потока удаляет любые гарантии, которые блокируют настройку try / finally, поэтому вы можете оставить блокировки заблокированными, файлы открытыми и т. д.

Единственный раз, когда вы можете утверждать, что принудительное уничтожение потоков - это хорошая идея, - это быстрое уничтожение программы, но никогда не одиночные потоки.

Для этого нет официального API, нет.

Вам нужно использовать API платформы, чтобы убить поток, например pthread_kill или TerminateThread. Вы можете получить доступ к такому API, например через pythonwin или через ctypes.

Обратите внимание, что это небезопасно по своей сути. Это, вероятно, приведет к необратимому мусору (из локальных переменных фреймов стека, которые становятся мусором) и может привести к тупикам, если убиваемый поток имеет GIL в момент, когда он убит.

Лучше не убивать нить. Одним из способов может быть введение блока «try» в цикл потока и выдача исключения, когда вы хотите остановить поток (например, break / return / ..., который останавливает ваш for / while / ...). Я использовал это в своем приложении, и он работает ...

Вы можете убить поток, установив трассировку в поток, который будет выходить из потока. См. Прикрепленную ссылку для одной из возможных реализаций.

Убить поток в Python

Как правило, резкое завершение потока в Python и на любом языке - плохой шаблон. Подумайте о следующих случаях:

• поток содержит критический ресурс, который должен быть закрыт должным образом
• поток создал несколько других потоков, которые также необходимо убить.

Хороший способ справиться с этим, если вы можете себе это позволить (если вы управляете своими собственными потоками), - это иметь флаг exit_request, который каждый поток проверяет через регулярный интервал, чтобы увидеть, не пора ли ему выйти.

Например:

import threading

class StoppableThread(threading.Thread):
    """Thread class with a stop() method. The thread itself has to check
    regularly for the stopped() condition."""

    def __init__(self,  *args, **kwargs):
        super(StoppableThread, self).__init__(*args, **kwargs)
        self._stop_event = threading.Event()

    def stop(self):
        self._stop_event.set()

    def stopped(self):
        return self._stop_event.is_set()

В этом коде вы должны вызвать stop() в потоке, когда вы хотите, чтобы он завершился, и подождать, пока поток не завершится должным образом, используя join(). Поток должен регулярно проверять флаг остановки.

Однако бывают случаи, когда вам действительно нужно убить поток. Примером может служить упаковка внешней библиотеки, которая занята долгими вызовами, и вы хотите ее прервать.

Следующий код позволяет (с некоторыми ограничениями) вызывать исключение в потоке Python:

def _async_raise(tid, exctype):
    '''Raises an exception in the threads with id tid'''
    if not inspect.isclass(exctype):
        raise TypeError("Only types can be raised (not instances)")
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid),
                                                     ctypes.py_object(exctype))
    if res == 0:
        raise ValueError("invalid thread id")
    elif res != 1:
        # "if it returns a number greater than one, you're in trouble,
        # and you should call it again with exc=NULL to revert the effect"
        ctypes.pythonapi.PyThreadState_SetAsyncExc(ctypes.c_long(tid), None)
        raise SystemError("PyThreadState_SetAsyncExc failed")

class ThreadWithExc(threading.Thread):
    '''A thread class that supports raising an exception in the thread from
       another thread.
    '''
    def _get_my_tid(self):
        """determines this (self's) thread id

        CAREFUL: this function is executed in the context of the caller
        thread, to get the identity of the thread represented by this
        instance.
        """
        if not self.isAlive():
            raise threading.ThreadError("the thread is not active")

        # do we have it cached?
        if hasattr(self, "_thread_id"):
            return self._thread_id

        # no, look for it in the _active dict
        for tid, tobj in threading._active.items():
            if tobj is self:
                self._thread_id = tid
                return tid

        # TODO: in python 2.6, there's a simpler way to do: self.ident

        raise AssertionError("could not determine the thread's id")

    def raiseExc(self, exctype):
        """Raises the given exception type in the context of this thread.

        If the thread is busy in a system call (time.sleep(),
        socket.accept(), ...), the exception is simply ignored.

        If you are sure that your exception should terminate the thread,
        one way to ensure that it works is:

            t = ThreadWithExc( ... )
            ...
            t.raiseExc( SomeException )
            while t.isAlive():
                time.sleep( 0.1 )
                t.raiseExc( SomeException )

        If the exception is to be caught by the thread, you need a way to
        check that your thread has caught it.

        CAREFUL: this function is executed in the context of the
        caller thread, to raise an exception in the context of the
        thread represented by this instance.
        """
        _async_raise( self._get_my_tid(), exctype )

(На основе Убиваемые потоки Томера Филиба. Цитата о возвращаемом значении PyThreadState_SetAsyncExc, похоже, взята из старая версия Python.)

Как отмечено в документации, это не волшебная палочка, потому что, если поток занят вне интерпретатора Python, он не поймает прерывание.

Хороший образец использования этого кода - заставить поток перехватить определенное исключение и выполнить очистку. Таким образом, вы можете прервать выполнение задачи и при этом произвести надлежащую очистку.

Если вы пытаетесь завершить всю программу, вы можете установить поток как «демон». видеть Thread.daemon

This is a bad answer, see the comments

Вот как это сделать:

from threading import *

...

for thread in enumerate():
    if thread.isAlive():
        try:
            thread._Thread__stop()
        except:
            print(str(thread.getName()) + ' could not be terminated'))

Подождите несколько секунд, и ваш поток должен быть остановлен. Также проверьте метод thread._Thread__delete().

Я бы порекомендовал для удобства метод thread.quit(). Например, если у вас есть сокет в вашем потоке, я бы рекомендовал создать метод quit() в вашем классе дескриптора сокета, завершить сокет, а затем запустить thread._Thread__stop() внутри вашего quit().

multiprocessing.Process может p.terminate()

В тех случаях, когда я хочу убить поток, но не хочу использовать флаги / блокировки / сигналы / семафоры / события / что угодно, я продвигаю потоки в полноценные процессы. Для кода, который использует всего несколько потоков, накладные расходы не так уж и плохи.

Например. это удобно, чтобы легко завершить вспомогательные "потоки", которые выполняют блокирующий ввод-вывод.

Преобразование тривиально: в соответствующем коде замените все threading.Thread на multiprocessing.Process и все queue.Queue на multiprocessing.Queue и добавьте необходимые вызовы p.terminate() в родительский процесс, который хочет убить свой дочерний p.

Смотрите Документация Python для multiprocessing.

Пример:

import multiprocessing
proc = multiprocessing.Process(target=your_proc_function, args=())
proc.start()
# Terminate the process
proc.terminate()  # sends a SIGTERM

В Python вы просто не можете убить поток напрямую.

Если вам на самом деле НЕ нужен поток (!), Вы можете вместо использования Пакет заправка использовать Пакет многопроцессорность. Здесь, чтобы убить процесс, вы можете просто вызвать метод:

yourProcess.terminate()  # kill the process!

Python убьет ваш процесс (в Unix через сигнал SIGTERM, а в Windows через вызов TerminateProcess()). Обратите внимание на то, чтобы использовать его при использовании Queue или Pipe! (это может привести к повреждению данных в очереди / конвейере)

Обратите внимание, что multiprocessing.Event и multiprocessing.Semaphore работают точно так же, как threading.Event и threading.Semaphore соответственно. По сути, первые - клоны последних.

Если вам ДЕЙСТВИТЕЛЬНО нужно использовать поток, нет способа убить его напрямую. Однако вы можете использовать "поток демона". Фактически, в Python поток может быть помечен как демон:

yourThread.daemon = True  # set the Thread as a "daemon thread"

Основная программа завершится, когда не останется живых потоков, не являющихся демонами. Другими словами, когда ваш основной поток (который, конечно, не является потоком демона) завершит свои операции, программа завершится, даже если некоторые потоки демона все еще работают.

Обратите внимание, что перед вызовом метода daemon необходимо установить поток как start()!

Конечно, вы можете и должны использовать daemon даже с multiprocessing. Здесь, когда основной процесс завершается, он пытается завершить все свои демонические дочерние процессы.

Наконец, обратите внимание, что sys.exit() и os.kill() не подходят.

Это основано на thread2 - убиваемые потоки (рецепт Python)

Вам нужно вызвать PyThreadState_SetasyncExc (), который доступен только через ctypes.

Это было протестировано только на Python 2.7.3, но, вероятно, будет работать с другими недавними выпусками 2.x.

import ctypes

def terminate_thread(thread):
    """Terminates a python thread from another thread.

    :param thread: a threading.Thread instance
    """
    if not thread.isAlive():
        return

    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:
        # """if it returns a number greater than one, you're in trouble,
        # and you should call it again with exc=NULL to revert the effect"""
        ctypes.pythonapi.PyThreadState_SetAsyncExc(thread.ident, None)
        raise SystemError("PyThreadState_SetAsyncExc failed")

from ctypes import *
pthread = cdll.LoadLibrary("libpthread-2.15.so")
pthread.pthread_cancel(c_ulong(t.ident))

ттrong> - ваш объект Thread.

Прочтите исходный код Python (Modules/threadmodule.c и Python/thread_pthread.h), вы увидите, что Thread.ident относится к типу pthread_t, поэтому вы можете делать все, что pthread может делать в Python, используя libpthread.

Я хочу добавить, что если вы читаете официальную документацию в библиотека потоковой передачи Python, рекомендуется избегать использования «демонических» потоков, когда вы не хотите, чтобы потоки завершались резко, с флагом Paolo Rovelli упомянул.

Из официальной документации:

Daemon threads are abruptly stopped at shutdown. Their resources (such as open files, database transactions, etc.) may not be released properly. If you want your threads to stop gracefully, make them non-daemonic and use a suitable signaling mechanism such as an Event.

Я думаю, что создание демонических потоков зависит от вашего приложения, но в целом (и на мой взгляд) лучше избегать их уничтожения или превращения в демонические. В многопроцессорной обработке вы можете использовать is_alive() для проверки статуса процесса и «завершить» для их завершения (также вы избегаете проблем с GIL). Но иногда вы можете обнаружить больше проблем, когда выполняете свой код в Windows.

И всегда помните, что если у вас есть «живые потоки», интерпретатор Python будет ждать их. (Благодаря этому демон может помочь вам, если неважно резко закончится).

Если вам действительно нужна возможность убить подзадачу, используйте альтернативную реализацию. multiprocessing и gevent оба поддерживают неизбирательное уничтожение «нити».

Потоковая обработка Python не поддерживает отмену. Даже не пытайся. Ваш код, скорее всего, зайдет в тупик, повредит память или вызовет утечку памяти, или будет иметь другие непреднамеренные "интересные" трудно поддающиеся отладке эффекты, которые происходят редко и недетерминированно.

Определенно возможно реализовать метод Thread.stop, как показано в следующем примере кода:

import sys
import threading
import time


class StopThread(StopIteration):
    pass

threading.SystemExit = SystemExit, StopThread


class Thread2(threading.Thread):

    def stop(self):
        self.__stop = True

    def _bootstrap(self):
        if threading._trace_hook is not None:
            raise ValueError('Cannot run thread with tracing!')
        self.__stop = False
        sys.settrace(self.__trace)
        super()._bootstrap()

    def __trace(self, frame, event, arg):
        if self.__stop:
            raise StopThread()
        return self.__trace


class Thread3(threading.Thread):

    def _bootstrap(self, stop_thread=False):
        def stop():
            nonlocal stop_thread
            stop_thread = True
        self.stop = stop

        def tracer(*_):
            if stop_thread:
                raise StopThread()
            return tracer
        sys.settrace(tracer)
        super()._bootstrap()

###############################################################################


def main():
    test1 = Thread2(target=printer)
    test1.start()
    time.sleep(1)
    test1.stop()
    test1.join()
    test2 = Thread2(target=speed_test)
    test2.start()
    time.sleep(1)
    test2.stop()
    test2.join()
    test3 = Thread3(target=speed_test)
    test3.start()
    time.sleep(1)
    test3.stop()
    test3.join()


def printer():
    while True:
        print(time.time() % 1)
        time.sleep(0.1)


def speed_test(count=0):
    try:
        while True:
            count += 1
    except StopThread:
        print('Count =', count)

if __name__ == '__main__':
    main()

Класс Thread3, похоже, выполняет код примерно на 33% быстрее, чем класс Thread2.

Как уже упоминали другие, нормой является установка флажка остановки. Для чего-то легкого (без подкласса Thread, без глобальной переменной) возможен лямбда-обратный вызов. (Обратите внимание на скобки в if stop().)

import threading
import time

def do_work(id, stop):
    print("I am thread", id)
    while True:
        print("I am thread {} doing something".format(id))
        if stop():
            print("  Exiting loop.")
            break
    print("Thread {}, signing off".format(id))


def main():
    stop_threads = False
    workers = []
    for id in range(0,3):
        tmp = threading.Thread(target=do_work, args=(id, lambda: stop_threads))
        workers.append(tmp)
        tmp.start()
    time.sleep(3)
    print('main: done sleeping; time to stop the threads.')
    stop_threads = True
    for worker in workers:
        worker.join()
    print('Finis.')

if __name__ == '__main__':
    main()

Замена print() на функцию pr(), которая всегда очищает (sys.stdout.flush()), может улучшить точность вывода оболочки.

(Проверено только в Windows / Eclipse / Python3.3)

Кажется, это работает с pywin32 в Windows 7

my_thread = threading.Thread()
my_thread.start()
my_thread._Thread__stop()

Вы можете выполнить свою команду в процессе, а затем убить его, используя идентификатор процесса. Мне нужно было синхронизировать два потока, один из которых не возвращается сам по себе.

processIds = []

def executeRecord(command):
    print(command)

    process = subprocess.Popen(command, stdout=subprocess.PIPE)
    processIds.append(process.pid)
    print(processIds[0])

    #Command that doesn't return by itself
    process.stdout.read().decode("utf-8")
    return;


def recordThread(command, timeOut):

    thread = Thread(target=executeRecord, args=(command,))
    thread.start()
    thread.join(timeOut)

    os.kill(processIds.pop(), signal.SIGINT)

    return;

Запустите дополнительный поток с помощью setDaemon (True).

def bootstrap(_filename):
    mb = ModelBootstrap(filename=_filename) # Has many Daemon threads. All get stopped automatically when main thread is stopped.

t = threading.Thread(target=bootstrap,args=('models.conf',))
t.setDaemon(False)

while True:
    t.start()
    time.sleep(10) # I am just allowing the sub-thread to run for 10 sec. You can listen on an event to stop execution.
    print('Thread stopped')
    break

Хотя он довольно старый, это может быть удобным решением для некоторых:

A little module that extends the threading's module functionality -- allows one thread to raise exceptions in the context of another thread. By raising SystemExit, you can finally kill python threads.

import threading
import ctypes     

def _async_raise(tid, excobj):
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, ctypes.py_object(excobj))
    if res == 0:
        raise ValueError("nonexistent thread id")
    elif res > 1:
        # """if it returns a number greater than one, you're in trouble, 
        # and you should call it again with exc=NULL to revert the effect"""
        ctypes.pythonapi.PyThreadState_SetAsyncExc(tid, 0)
        raise SystemError("PyThreadState_SetAsyncExc failed")

class Thread(threading.Thread):
    def raise_exc(self, excobj):
        assert self.isAlive(), "thread must be started"
        for tid, tobj in threading._active.items():
            if tobj is self:
                _async_raise(tid, excobj)
                return

        # the thread was alive when we entered the loop, but was not found 
        # in the dict, hence it must have been already terminated. should we raise
        # an exception here? silently ignore?

    def terminate(self):
        # must raise the SystemExit type, instead of a SystemExit() instance
        # due to a bug in PyThreadState_SetAsyncExc
        self.raise_exc(SystemExit)

Таким образом, он позволяет «потоку вызывать исключения в контексте другого потока», и, таким образом, завершенный поток может обрабатывать завершение без регулярной проверки флага прерывания.

Однако, согласно его первоисточник, с этим кодом есть некоторые проблемы.

• The exception will be raised only when executing python bytecode. If your thread calls a native/built-in blocking function, the exception will be raised only when execution returns to the python code.
  • There is also an issue if the built-in function internally calls PyErr_Clear(), which would effectively cancel your pending exception. You can try to raise it again.
• Only exception types can be raised safely. Exception instances are likely to cause unexpected behavior, and are thus restricted.
  • For example: t1.raise_exc(TypeError) and not t1.raise_exc(TypeError("blah")).
  • IMHO it's a bug, and I reported it as one. For more info, http://mail.python.org/pipermail/python-dev/2006-August/068158.html
• I asked to expose this function in the built-in thread module, but since ctypes has become a standard library (as of 2.5), and this
  feature is not likely to be implementation-agnostic, it may be kept
  unexposed.

Питер Хинтьенс - один из основателей проекта ØMQ - говорит, что использование ØMQ и отказ от примитивов синхронизации, таких как блокировки, мьютексы, события и т. д., Является самым разумным и безопасным способом написания многопоточных программ:

http://zguide.zeromq.org/py:all#Multithreading-with-ZeroMQ

Это включает в себя сообщение дочернему потоку о том, что он должен отменить свою работу. Это можно сделать, оснастив поток ØMQ-сокетом и опросив этот сокет на предмет сообщения о том, что он должен отменить.

Ссылка также предоставляет пример многопоточного кода Python с ØMQ.

Чтобы убить поток, можно использовать следующий обходной путь:

kill_threads = False

def doSomething():
    global kill_threads
    while True:
        if kill_threads:
            thread.exit()
        ......
        ......

thread.start_new_thread(doSomething, ())

Это можно использовать даже для завершения потоков, код которых написан в другом модуле, из основного потока. Мы можем объявить глобальную переменную в этом модуле и использовать ее для завершения потоков, порожденных в этом модуле.

Я обычно использую это для завершения всех потоков при выходе из программы. Это может быть не лучший способ завершить поток / ы, но может помочь.

Если вы явно вызываете time.sleep() как часть вашего потока (например, опрашиваете какую-то внешнюю службу), улучшением метода Филиппа является использование тайм-аута в методе eventwait() везде, где вы sleep()

Например:

import threading

class KillableThread(threading.Thread):
    def __init__(self, sleep_interval=1):
        super().__init__()
        self._kill = threading.Event()
        self._interval = sleep_interval

    def run(self):
        while True:
            print("Do Something")

            # If no kill signal is set, sleep for the interval,
            # If kill signal comes in while sleeping, immediately
            #  wake up and handle
            is_killed = self._kill.wait(self._interval)
            if is_killed:
                break

        print("Killing Thread")

    def kill(self):
        self._kill.set()

Затем запустить его

t = KillableThread(sleep_interval=5)
t.start()
# Every 5 seconds it prints:
#: Do Something
t.kill()
#: Killing Thread

Преимущество использования wait() вместо sleep()ing и регулярной проверки события заключается в том, что вы можете программировать в более длительных интервалах сна, поток останавливается почти сразу (когда в противном случае вы были бы sleep()ing) и, на мой взгляд, код для обработки выхода значительно проще.

Я очень опаздываю в эту игру, но я боролся с аналогичный вопрос, и следующее, похоже, отлично решает проблему для меня И позволяет мне выполнить базовую проверку и очистку состояния потока при выходе из демонизированного подпотока:

import threading
import time
import atexit

def do_work():

  i = 0
  @atexit.register
  def goodbye():
    print ("'CLEANLY' kill sub-thread with value: %s [THREAD: %s]" %
           (i, threading.currentThread().ident))

  while True:
    print i
    i += 1
    time.sleep(1)

t = threading.Thread(target=do_work)
t.daemon = True
t.start()

def after_timeout():
  print "KILL MAIN THREAD: %s" % threading.currentThread().ident
  raise SystemExit

threading.Timer(2, after_timeout).start()

Урожайность:

0
1
KILL MAIN THREAD: 140013208254208
'CLEANLY' kill sub-thread with value: 2 [THREAD: 140013674317568]

Для этого создана библиотека стопить. Хотя некоторые из перечисленных здесь предостережений все еще применимы, по крайней мере, эта библиотека представляет собой регулярный, повторяемый метод для достижения заявленной цели.

Предполагая, что вы хотите иметь несколько потоков одной и той же функции, это ИМХО самая простая реализация, чтобы остановить один по идентификатору:

import time
from threading import Thread

def doit(id=0):
    doit.stop=0
    print("start id:%d"%id)
    while 1:
        time.sleep(1)
        print(".")
        if doit.stop==id:
            doit.stop=0
            break
    print("end thread %d"%id)

t5=Thread(target=doit, args=(5,))
t6=Thread(target=doit, args=(6,))

t5.start() ; t6.start()
time.sleep(2)
doit.stop =5  #kill t5
time.sleep(2)
doit.stop =6  #kill t6

Здесь хорошо то, что вы можете иметь несколько одинаковых и разных функций и останавливать их все с помощью functionname.stop.

Если вы хотите иметь только один поток функции, вам не нужно запоминать идентификатор. Просто остановитесь, если doit.stop> 0.

Просто чтобы развить идею @ SCB (это именно то, что мне нужно) создать подкласс KillableThread с настраиваемой функцией:

from threading import Thread, Event

class KillableThread(Thread):
    def __init__(self, sleep_interval=1, target=None, name=None, args=(), kwargs = {}):
        super().__init__(None, target, name, args, kwargs)
        self._kill = Event()
        self._interval = sleep_interval
        print(self._target)

    def run(self):
        while True:
            # Call custom function with arguments
            self._target(*self._args)

            # If no kill signal is set, sleep for the interval,
            # If kill signal comes in while sleeping, immediately
            #  wake up and handle
            is_killed = self._kill.wait(self._interval)
            if is_killed:
                break

        print("Killing Thread")

    def kill(self):
        self._kill.set()

if __name__ == '__main__':

    def print_msg(msg):
        print(msg)

    t = KillableThread(10, print_msg, args=("hello world"))
    t.start()
    time.sleep(6)
    print("About to kill thread")
    t.kill()

Естественно, как и в случае с @SBC, поток не дожидается запуска нового цикла до остановки. В этом примере вы увидите сообщение «Killing Thread», напечатанное сразу после «About to kill thread», вместо того, чтобы ждать еще 4 секунды для завершения потока (поскольку мы уже спали 6 секунд).

Второй аргумент в конструкторе KillableThread - это ваша пользовательская функция (здесь print_msg). Аргумент args - это аргументы, которые будут использоваться здесь при вызове функции (("hello world")).

Как упоминалось в отвечать @ Kozyarchuk, установка трассировки работает. Поскольку этот ответ не содержал кода, вот рабочий, готовый к использованию пример:

import sys, threading, time 

class TraceThread(threading.Thread): 
    def __init__(self, *args, **keywords): 
        threading.Thread.__init__(self, *args, **keywords) 
        self.killed = False
    def start(self): 
        self._run = self.run 
        self.run = self.settrace_and_run
        threading.Thread.start(self) 
    def settrace_and_run(self): 
        sys.settrace(self.globaltrace) 
        self._run()
    def globaltrace(self, frame, event, arg): 
        return self.localtrace if event == 'call' else None
    def localtrace(self, frame, event, arg): 
        if self.killed and event == 'line': 
            raise SystemExit() 
        return self.localtrace 

def f(): 
    while True: 
        print('1') 
        time.sleep(2)
        print('2') 
        time.sleep(2)
        print('3') 
        time.sleep(2)

t = TraceThread(target=f) 
t.start() 
time.sleep(2.5) 
t.killed = True

Он останавливается после того, как напечатал 1 и 2. 3 не печатается.

Вот еще один способ сделать это, но с чрезвычайно чистым и простым кодом, который будет работать в Python 3.7 в 2021 году:

import ctypes 

def kill_thread(thread):
    """
    thread: a threading.Thread object
    """
    thread_id = thread.ident
    res = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, ctypes.py_object(SystemExit))
    if res > 1:
        ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, 0)
        print('Exception raise failure')

Адаптировано отсюда: https://www.geeksforgeeks.org/python-different-ways-to-kill-a-thread/

Версия Python: 3.8

Используя поток демона для выполнения того, что мы хотели, если мы хотим, чтобы поток демона был завершен, все, что нам нужно, это выполнить выход из родительского потока, тогда система завершит поток демона, созданный родительским потоком.

Также поддерживает сопрограммы и функции сопрограмм.

def main():
    start_time = time.perf_counter()
    t1 = ExitThread(time.sleep, (10,), debug=False)
    t1.start()
    time.sleep(0.5)
    t1.exit()
    try:
        print(t1.result_future.result())
    except concurrent.futures.CancelledError:
        pass
    end_time = time.perf_counter()
    print(f"time cost {end_time - start_time:0.2f}")

ниже исходный код ExitThread

import concurrent.futures
import threading
import typing
import asyncio


class _WorkItem(object):
    """ concurrent\futures\thread.py

    """

    def __init__(self, future, fn, args, kwargs, *, debug=None):
        self._debug = debug
        self.future = future
        self.fn = fn
        self.args = args
        self.kwargs = kwargs

    def run(self):
        if self._debug:
            print("ExitThread._WorkItem run")
        if not self.future.set_running_or_notify_cancel():
            return

        try:
            coroutine = None
            if asyncio.iscoroutinefunction(self.fn):
                coroutine = self.fn(*self.args, **self.kwargs)
            elif asyncio.iscoroutine(self.fn):
                coroutine = self.fn
            if coroutine is None:
                result = self.fn(*self.args, **self.kwargs)
            else:
                result = asyncio.run(coroutine)
            if self._debug:
                print("_WorkItem done")
        except BaseException as exc:
            self.future.set_exception(exc)
            # Break a reference cycle with the exception 'exc'
            self = None
        else:
            self.future.set_result(result)


class ExitThread:
    """ Like a stoppable thread

    Using coroutine for target then exit before running may cause RuntimeWarning.

    """

    def __init__(self, target: typing.Union[typing.Coroutine, typing.Callable] = None
                 , args=(), kwargs = {}, *, daemon=None, debug=None):
        #
        self._debug = debug
        self._parent_thread = threading.Thread(target=self._parent_thread_run, name = "ExitThread_parent_thread"
                                               , daemon=daemon)
        self._child_daemon_thread = None
        self.result_future = concurrent.futures.Future()
        self._workItem = _WorkItem(self.result_future, target, args, kwargs, debug=debug)
        self._parent_thread_exit_lock = threading.Lock()
        self._parent_thread_exit_lock.acquire()
        self._parent_thread_exit_lock_released = False  # When done it will be True
        self._started = False
        self._exited = False
        self.result_future.add_done_callback(self._release_parent_thread_exit_lock)

    def _parent_thread_run(self):
        self._child_daemon_thread = threading.Thread(target=self._child_daemon_thread_run
                                                     , name = "ExitThread_child_daemon_thread"
                                                     , daemon=True)
        self._child_daemon_thread.start()
        # Block manager thread
        self._parent_thread_exit_lock.acquire()
        self._parent_thread_exit_lock.release()
        if self._debug:
            print("ExitThread._parent_thread_run exit")

    def _release_parent_thread_exit_lock(self, _future):
        if self._debug:
            print(f"ExitThread._release_parent_thread_exit_lock {self._parent_thread_exit_lock_released} {_future}")
        if not self._parent_thread_exit_lock_released:
            self._parent_thread_exit_lock_released = True
            self._parent_thread_exit_lock.release()

    def _child_daemon_thread_run(self):
        self._workItem.run()

    def start(self):
        if self._debug:
            print(f"ExitThread.start {self._started}")
        if not self._started:
            self._started = True
            self._parent_thread.start()

    def exit(self):
        if self._debug:
            print(f"ExitThread.exit exited: {self._exited} lock_released: {self._parent_thread_exit_lock_released}")
        if self._parent_thread_exit_lock_released:
            return
        if not self._exited:
            self._exited = True
            if not self.result_future.cancel():
                if self.result_future.running():
                    self.result_future.set_exception(concurrent.futures.CancelledError())