python thread продолжается после завершения

Question

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

Прилагается прикрепленный фрагмент кода и фактический вывод по сравнению с тем, что я ожидал. Я удалил свой фактический рабочий код и заменил его на сон, чтобы легко продемонстрировать это и установить рабочий поток в течение 10 секунд, а защита времени для завершения потока - 5 секунд.

Благодаря

from threading import Thread, Timer 
import Queue 
from time import strftime, gmtime, sleep 

class test(object): 

    worker_thread = None 
    q = Queue.Queue() 

    def terminate_test(self): 

     if self.worker_thread.isAlive(): 
      print 'stopping', self.worker_thread.getName(), strftime("%Y-%m-%d %H:%M:%S", gmtime()) 
      self.worker_thread._Thread__stop() 
      self.worker_thread.join() 
      print "Thread alive?", self.worker_thread.getName(), self.worker_thread.isAlive() 
      del self.worker_thread 
      print 'stopped' 
    #   self._testq.task_done() 

     else: 
      print 'worker finished before stopping', strftime("%Y-%m-%d %H:%M:%S", gmtime())  

    def worker(self, item): 
     k = item.keys()[0] 
     v = item.values()[0] 
     print 'starting', k, strftime("%Y-%m-%d %H:%M:%S", gmtime()) 
     print 'sleeping', v, strftime("%Y-%m-%d %H:%M:%S", gmtime()) 
     sleep(v) 
     print "waking up", k,v, strftime("%Y-%m-%d %H:%M:%S", gmtime()) 
     self.q.task_done() 
     self.test_terminate_thread.cancel() 

    def start(self): 
     source = [{'test a':10},{'test b':10},{'test c':10},{'test d':10}] 
     for item in source: 
      self.q.put(item) 

     while not self.q.empty(): 
      item = self.q.get() 
      self.worker_thread = Thread(target=self.worker, args=(item,)) 
      self.worker_thread.setName('test_thread %s' % item.keys()[0]) 

      self.test_terminate_thread = Timer(5, self.terminate_test) 
      self.test_terminate_thread.setName("Test_terminate_thread") 

      self.worker_thread.start() 
      self.test_terminate_thread.start() 

      self.worker_thread.join()  
      self.test_terminate_thread.join() 

     print "joining Q", strftime("%Y-%m-%d %H:%M:%S", gmtime())   
     self.q.join()  

if __name__ == "__main__": 
    test().start() 

Фактический выход: (Я не ожидая, просыпаясь линии)

starting test a 2015-01-16 15:07:07 
sleeping 10 2015-01-16 15:07:07 
stopping test_thread test a 2015-01-16 15:07:12 
Thread alive? test_thread test a False 
stopped 
starting test b 2015-01-16 15:07:12 
sleeping 10 2015-01-16 15:07:12 
waking up test a 10 2015-01-16 15:07:17 
waking up test b 10 2015-01-16 15:07:22 
starting test c 2015-01-16 15:07:22 
sleeping 10 2015-01-16 15:07:22 
stopping test_thread test c 2015-01-16 15:07:27 
Thread alive? test_thread test c False 
stopped 
starting test d 2015-01-16 15:07:27 
sleeping 10 2015-01-16 15:07:27 
stopping test_thread test d 2015-01-16 15:07:32 
Thread alive? test_thread test d False 
stopped 
joining Q 2015-01-16 15:07:32 
waking up test c 10 2015-01-16 15:07:32 
waking up test d 10 2015-01-16 15:07:37 

Я Ожидал

starting test a 2015-01-16 15:07:07 
sleeping 10 2015-01-16 15:07:07 
stopping test_thread test a 2015-01-16 15:07:12 
Thread alive? test_thread test a False 
stopped 
starting test b 2015-01-16 15:07:12 
sleeping 10 2015-01-16 15:07:12 
starting test c 2015-01-16 15:07:22 
sleeping 10 2015-01-16 15:07:22 
stopping test_thread test c 2015-01-16 15:07:27 
Thread alive? test_thread test c False 
stopped 
starting test d 2015-01-16 15:07:27 
sleeping 10 2015-01-16 15:07:27 
stopping test_thread test d 2015-01-16 15:07:32 
Thread alive? test_thread test d False 
stopped 
joining Q 2015-01-16 15:07:32 

Answer 1

_Thread__stop() функции вы звоните фактически не заканчивает нить. Это внутренний метод, который вызывается Python, когда метод потока run заканчивается - это означает, что он вызван только после потока, на самом деле имеет. Все это на самом деле установлен атрибут __stopped в True и уведомить ничего ожидания на резьбе, чтобы остановить, что они могут продолжать:

def __stop(self): 
    # DummyThreads delete self.__block, but they have no waiters to 
    # notify anyway (join() is forbidden on them). 
    if not hasattr(self, '_Thread__block'): 
     return 
    self.__block.acquire() 
    self.__stopped = True 
    self.__block.notify_all() 
    self.__block.release() 

Кроме того, метод is_alive()/isAlive() просто проверяет __stopped флаг, он фактически не убедитесь, что метод run закончен, так как предполагается, что __stopped будет установлен только в том случае, если это правда.

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

Вот как вы могли бы изменить свой пример, чтобы использовать Event:

def terminate_test(self, event): 

    if self.worker_thread.isAlive(): 
     print 'stopping', self.worker_thread.getName(), strftime("%Y-%m-%d %H:%M:%S", gmtime()) 
     event.set() 
     self.worker_thread.join() 
     print "Thread alive?", self.worker_thread.getName(), self.worker_thread.isAlive() 
     del self.worker_thread 
     print 'stopped' 

    else: 
     print 'worker finished before stopping', strftime("%Y-%m-%d %H:%M:%S", gmtime())  

def worker(self, item, event): 
    k = item.keys()[0] 
    v = item.values()[0] 
    print 'starting', k, strftime("%Y-%m-%d %H:%M:%S", gmtime()) 
    print 'sleeping', v, strftime("%Y-%m-%d %H:%M:%S", gmtime()) 
    if event.wait(v): 
     return 
    print "waking up", k,v, strftime("%Y-%m-%d %H:%M:%S", gmtime()) 
    self.q.task_done() 
    self.test_terminate_thread.cancel() 

def start(self): 
    source = [{'test a':10},{'test b':10},{'test c':10},{'test d':10}] 
    for item in source: 
     self.q.put(item) 

    while not self.q.empty(): 
     item = self.q.get() 
     e = Event() 
     self.worker_thread = Thread(target=self.worker, args=(item, e)) 
     self.worker_thread.setName('test_thread %s' % item.keys()[0]) 

     self.test_terminate_thread = Timer(5, self.terminate_test, args=(e,)) 
     self.test_terminate_thread.setName("Test_terminate_thread") 

Выход:

starting test a 2015-01-16 15:42:57 
sleeping 10 2015-01-16 15:42:57 
stopping test_thread test a 2015-01-16 15:43:02 
Thread alive? test_thread test a False 
stopped 
starting test b 2015-01-16 15:43:02 
sleeping 10 2015-01-16 15:43:02 
stopping test_thread test b 2015-01-16 15:43:07 
Thread alive? test_thread test b False 
stopped 
starting test c 2015-01-16 15:43:07 
sleeping 10 2015-01-16 15:43:07 
stopping test_thread test c 2015-01-16 15:43:12 
Thread alive? test_thread test c False 
stopped 
starting test d 2015-01-16 15:43:12 
sleeping 10 2015-01-16 15:43:12 
stopping test_thread test d 2015-01-16 15:43:17 
Thread alive? test_thread test d False 
stopped 
joining Q 2015-01-16 15:43:17