Existe alguma maneira de matar um Thread?

É possível finalizar um thread em execução sem configurar / verificar nenhum sinalizador / semáforo / etc.?

Geralmente, é um padrão ruim matar um thread abruptamente, em Python e em qualquer idioma. Pense nos seguintes casos:

• o encadeamento está mantendo um recurso crítico que deve ser fechado corretamente
• o encadeamento criou vários outros encadeamentos que também devem ser eliminados.

A boa maneira de lidar com isso, se você puder pagar (se estiver gerenciando seus próprios encadeamentos), é ter um sinalizador exit_request que cada encadeamento verifica em intervalos regulares para ver se é hora de sair.

Por exemplo:

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()

Nesse código, você deve chamar stop()o encadeamento quando quiser que ele saia e aguardar o encadeamento sair corretamente usando join(). A linha deve verificar o sinalizador de parada em intervalos regulares.

No entanto, existem casos em que você realmente precisa matar um thread. Um exemplo é quando você está agrupando uma biblioteca externa ocupada para chamadas longas e deseja interrompê-la.

O código a seguir permite (com algumas restrições) gerar uma exceção em um thread 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 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 excpetion in the context of the
        thread represented by this instance.
        """
        _async_raise( self._get_my_tid(), exctype )

(Baseado em tópicos matáveis ​​de Tomer Filiba. A citação sobre o valor de retorno de PyThreadState_SetAsyncExcparece ser de uma versão antiga do Python .)

Conforme observado na documentação, isso não é um item mágico, porque se o encadeamento estiver ocupado fora do interpretador Python, ele não detectará a interrupção.

Um bom padrão de uso desse código é fazer com que o encadeamento capture uma exceção específica e execute a limpeza. Dessa forma, você pode interromper uma tarefa e ainda ter uma limpeza adequada.

Não existe uma API oficial para fazer isso, não.

Você precisa usar a API da plataforma para eliminar o encadeamento, por exemplo, pthread_kill ou TerminateThread. Você pode acessar essa API, por exemplo, através de pythonwin ou ctypes.

Observe que isso é inerentemente inseguro. Provavelmente, isso resultará em lixo incobrável (das variáveis ​​locais dos quadros de pilha que se tornam lixo) e poderá levar a conflitos, se o encadeamento que está sendo morto tiver o GIL no ponto em que é morto.

Uma multiprocessing.Processlatap.terminate()

Nos casos em que quero matar um encadeamento, mas não quero usar sinalizadores / bloqueios / sinais / semáforos / eventos / o que for, eu promovo os encadeamentos para processos completos. Para código que utiliza apenas alguns threads, a sobrecarga não é tão ruim assim.

Por exemplo, isso é útil para encerrar facilmente "threads" auxiliares que executam E / S de bloqueio

A conversão é trivial: no código relacionado, substitua all threading.Threadpor multiprocessing.Processe all queue.Queuewith multiprocessing.Queuee adicione as chamadas necessárias de p.terminate()seu processo pai, que deseja matar seu filho.p

Veja a documentaçãomultiprocessing do Python para .

Se você está tentando finalizar o programa inteiro, pode definir o encadeamento como um "daemon". consulte Thread.daemon

Como outros já mencionaram, a norma é definir um sinalizador de parada. Para algo leve (sem subclassificação de Thread, sem variável global), um retorno de chamada lambda é uma opção. (Observe os parênteses em 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()

Substituir print()por uma pr()função que sempre libera ( sys.stdout.flush()) pode melhorar a precisão da saída do shell.

(Testado apenas no Windows / Eclipse / Python3.3)

Isso é baseado no thread2 - threads matáveis ​​(receita Python)

Você precisa chamar PyThreadState_SetasyncExc (), que está disponível apenas através de ctypes.

Isso foi testado apenas no Python 2.7.3, mas é provável que funcione com outras versões 2.x recentes.

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")

Você nunca deve matar um fio à força sem cooperar com ele.

Matar um encadeamento remove todas as garantias que os blocos try / finalmente configuram para que você possa deixar bloqueios bloqueados, arquivos abertos etc.

O único momento em que você pode argumentar que matar forçosamente threads é uma boa ideia é matar um programa rapidamente, mas nunca threads únicos.

No Python, você simplesmente não pode matar um Thread diretamente.

Se você NÃO precisa realmente de um Thread (!), O que você pode fazer, em vez de usar o pacote de threading , é usar o pacote de multiprocessamento . Aqui, para matar um processo, você pode simplesmente chamar o método:

yourProcess.terminate()  # kill the process!

Python matará seu processo (no Unix, através do sinal SIGTERM, enquanto no Windows, através da TerminateProcess()chamada). Preste atenção para usá-lo enquanto estiver usando uma fila ou um cano! (pode corromper os dados na fila / canal)

Observe que o multiprocessing.Evente o multiprocessing.Semaphoretrabalho funcionam exatamente da mesma maneira que o threading.Evente o threading.Semaphorerespectivamente. De fato, os primeiros são clones dos últimos.

Se você REALMENTE precisar usar um Thread, não há como matá-lo diretamente. O que você pode fazer, no entanto, é usar um "thread daemon" . De fato, em Python, um Thread pode ser sinalizado como daemon :

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

O programa principal será encerrado quando nenhum encadeamento ativo não-daemon for deixado. Em outras palavras, quando seu encadeamento principal (que é, obviamente, um encadeamento que não é daemon) termina suas operações, o programa será encerrado mesmo se ainda houver alguns encadeamentos daemon funcionando.

Observe que é necessário definir um Thread como daemonantes do start()método ser chamado!

Claro que você pode e deve usar daemonmesmo com multiprocessing. Aqui, quando o processo principal termina, ele tenta finalizar todos os seus processos filhos daemônicos.

Finalmente, por favor, note que sys.exit()e os.kill()não são escolhas.

Você pode eliminar um encadeamento instalando o rastreio no encadeamento que sairá do encadeamento. Veja o link em anexo para uma possível implementação.

Mate um tópico em Python

Se você está chamando explicitamente time.sleep()como parte de seu segmento (polling digamos algum serviço externo), uma melhoria sobre o método de Phillipe é usar o tempo limite na event's wait()método onde quer que vocêsleep()

Por exemplo:

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()

Então, para executá-lo

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

A vantagem de usar em wait()vez de sleep()ing e verificar regularmente o evento é que você pode programar em intervalos mais longos de suspensão, o encadeamento é interrompido quase imediatamente (quando você estaria sleep()ining) e, na minha opinião, o código para lidar com a saída é significativamente mais simples .

É melhor se você não matar um fio. Uma maneira seria introduzir um bloco "try" no ciclo do encadeamento e lançar uma exceção quando você quiser interromper o encadeamento (por exemplo, uma quebra / retorno / ... que interrompa seu processo por / enquanto / ...). Eu usei isso no meu aplicativo e funciona ...

Definitivamente, é possível implementar um Thread.stopmétodo conforme mostrado no seguinte código de exemplo:

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()

A Thread3classe parece executar código aproximadamente 33% mais rápido que a Thread2classe.

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

Esse é o seu Threadobjeto.

Leia a fonte python ( Modules/threadmodule.ce Python/thread_pthread.h) você pode ver que Thread.identé um pthread_ttipo, para que você possa fazer o que pthreadpuder no uso de python libpthread.

A solução alternativa a seguir pode ser usada para eliminar um encadeamento:

kill_threads = False

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

thread.start_new_thread(doSomething, ())

Isso pode ser usado mesmo para encerrar threads, cujo código está escrito em outro módulo, a partir do thread principal. Podemos declarar uma variável global nesse módulo e usá-la para finalizar threads gerados nesse módulo.

Eu costumo usar isso para finalizar todos os threads na saída do programa. Essa pode não ser a maneira perfeita de encerrar os threads, mas pode ajudar.

Estou muito atrasado para este jogo, mas tenho lutado com uma pergunta semelhante e o seguinte parece resolver o problema perfeitamente para mim E me permite fazer uma verificação e limpeza básicas do estado do encadeamento quando o sub-encadeamento daemonizado terminar:

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()

Rendimentos:

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

Uma coisa que quero acrescentar é que, se você ler a documentação oficial no threading lib Python , é recomendável evitar o uso de threads "demoníacos", quando você não quiser que os threads terminem abruptamente, com a bandeira que Paolo Rovelli mencionou .

Da documentação oficial:

Os encadeamentos do daemon são interrompidos abruptamente no desligamento. Seus recursos (como arquivos abertos, transações de banco de dados etc.) podem não ser liberados corretamente. Se você deseja que seus encadeamentos parem normalmente, torne-os não daemônicos e use um mecanismo de sinalização adequado, como um Evento.

Penso que a criação de threads daemônicos depende da sua aplicação, mas em geral (e na minha opinião) é melhor evitar matá-los ou torná-los daemônicos. No multiprocessamento, você pode usar is_alive()para verificar o status do processo e "encerrar" para finalizá-los (também evita problemas de GIL). Mas, às vezes, você pode encontrar mais problemas ao executar seu código no Windows.

E lembre-se sempre de que, se você tiver "threads ativos", o interpretador Python estará executando para aguardá-los. (Devido a esse daemônico, você pode ajudá-lo se não importa abruptamente).

Existe uma biblioteca criada para esse fim, stopit . Embora algumas das mesmas precauções listadas aqui ainda se apliquem, pelo menos essa biblioteca apresenta uma técnica regular e repetível para atingir o objetivo declarado.

Embora seja bastante antiga, essa pode ser uma solução útil para alguns:

Um pequeno módulo que estende a funcionalidade do módulo do encadeamento - permite que um encadeamento crie exceções no contexto de outro encadeamento. Ao aumentar SystemExit, você pode finalmente matar threads python.

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)

Portanto, ele permite que um "encadeamento crie exceções no contexto de outro encadeamento" e, dessa maneira, o encadeamento finalizado pode lidar com o encerramento sem verificar regularmente um sinalizador de cancelamento.

No entanto, de acordo com sua fonte original , há alguns problemas com esse código.

• A exceção será gerada apenas ao executar o bytecode do python. Se o seu thread chamar uma função de bloqueio nativa / interna, a exceção será gerada apenas quando a execução retornar ao código python.
  • Também há um problema se a função interna chamar internamente PyErr_Clear (), o que efetivamente cancelaria sua exceção pendente. Você pode tentar aumentá-lo novamente.
• Somente tipos de exceção podem ser gerados com segurança. Instâncias de exceção provavelmente causam comportamento inesperado e, portanto, são restritas.
  • Por exemplo: t1.raise_exc (TypeError) e não t1.raise_exc (TypeError ("blah")).
  • IMHO é um bug, e eu relatei como um. Para mais informações, http://mail.python.org/pipermail/python-dev/2006-August/068158.html
• Pedi para expor essa função no módulo de encadeamento interno, mas como o ctypes se tornou uma biblioteca padrão (a partir do 2.5), e esse
  recurso provavelmente não é independente de implementação, pode ser mantido sem
  exposição.

Isso parece funcionar com o pywin32 no windows 7

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

Pieter Hintjens - um dos fundadores do projeto ØMQ - diz que usar o ØMQ e evitar primitivas de sincronização como bloqueios, mutexes, eventos etc. é a maneira mais segura e segura de escrever programas multithread:

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

Isso inclui dizer a um thread filho que ele deve cancelar seu trabalho. Isso seria feito equipando o thread com um soquete ØMQ e pesquisando nesse soquete para obter uma mensagem dizendo que ele deveria ser cancelado.

O link também fornece um exemplo no código python multiencadeado com ØMQ.

Assumindo que você deseja ter vários threads da mesma função, esta é IMHO a implementação mais fácil de parar um por id:

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

O legal é que aqui você pode ter várias funções iguais e diferentes e interromper todas elas functionname.stop

Se você deseja ter apenas um segmento da função, não precisa se lembrar do id. Apenas pare, se doit.stop> 0.

Apenas para desenvolver a idéia do @ SCB (que era exatamente o que eu precisava) para criar uma subclasse KillableThread com uma função personalizada:

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()

Naturalmente, como no @SBC, o thread não espera para executar um novo loop para parar. Neste exemplo, você veria a mensagem "Killing Thread" impressa logo após a mensagem "About to kill thread", em vez de esperar mais 4 segundos para que o thread fosse concluído (já que dormimos por 6 segundos).

O segundo argumento no construtor KillableThread é sua função personalizada (print_msg aqui). O argumento Args são os argumentos que serão usados ​​ao chamar a função (("hello world")) aqui.

Como mencionado na resposta de @ Kozyarchuk , a instalação de rastreamento funciona. Como esta resposta não continha código, eis um exemplo pronto para uso:

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

Para depois de imprimir 1e 2. 3não é impresso.

Você pode executar seu comando em um processo e depois matá-lo usando a identificação do processo. Eu precisava sincronizar entre dois threads, um dos quais não retorna por si só.

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;

Inicie o subencadeamento com 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

Esta é uma resposta ruim, veja os comentários

Veja como fazê-lo:

from threading import *

...

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

Aguarde alguns segundos para que seu tópico seja interrompido. Verifique também o thread._Thread__delete()método.

Eu recomendaria um thread.quit()método por conveniência. Por exemplo, se você tiver um soquete em seu encadeamento, recomendo criar um quit()método em sua classe de identificador de soquete, encerrar o soquete e executar um thread._Thread__stop()interior do seu quit().

Se você realmente precisa eliminar uma subtarefa, use uma implementação alternativa. multiprocessinge geventambos apóiam indiscriminadamente matar um "fio".

A segmentação do Python não suporta cancelamento. Nem tente. É muito provável que seu código entre em conflito, corrompa ou vaze memória, ou tenha outros efeitos "interessantes" difíceis de depurar não intencionais que ocorrem raramente e sem determinação.