Python threading deadlock

Question

У меня есть два потока (производитель и потребитель), и я делюсь данными с Queue. Проблема в том, что, когда я решительно прерываю производителя, потребитель иногда блокирует.

В документах я прочитал, что отмена потока с очередью может привести к повреждению очереди и вызвать тупик. Я не получаю никаких блокировок явно, но чтение источника Queue.py говорит, что put и get делают это.

Возможно, кто-нибудь знает, что это может быть так, что когда я прерываю нить, это может быть в середине get/put, то есть с помощью блокировки, а затем не отпускать его? Что я могу сделать с этим? Иногда мне нужно преждевременно прекратить работу продюсера. Используют ли процессы, а не потоки, какие-либо различия?

Answer 1

Скорее всего, ваш тупик из-за не законченных потоков. Если у вас есть Linux вы можете использовать форсунку от pyrasite для печати трассировки (вы знаете, где вы программа повесили)

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

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

Answer 2

Может быть, это поможет:

import threading 

class MyQueue: 
    def __init__(self): 
     self.tasks = [] 
     self.tlock = threading.Semaphore(0) 
     self.dlock = threading.Lock() 
     self.aborted = False 

    def put(self, arg): 
     try: 
      self.dlock.acquire() 
      self.tasks.append(arg) 
     finally: 
      self.dlock.release() 
      self.tlock.release() 

    def get(self): 
     if self.aborted: 
      return None 
     self.tlock.acquire() 
     if self.aborted: 
      self.tlock.release() 
      return None 
     try: 
      self.dlock.acquire() 
      if self.tasks: 
       return self.tasks.pop() 
      else: # executed abort 
       return None 
     finally: 
      self.dlock.release() 

    def abort(self): 
     self.aborted = True 
     self.tlock.release() 

# TESTING 

mq = MyQueue() 
import sys 

def tlog(line): 
    sys.stdout.write("[ %s ] %s\n" % (threading.currentThread().name, line)) 
    sys.stdout.flush() 

def reader(): 
    arg = 1 
    while arg is not None: 
     tlog("start reading") 
     arg = mq.get() 
     tlog("read: %s" % arg) 
    tlog("END") 

import time, random 
def writer(): 
    try: 
     pos = 1 
     while not mq.aborted: 
      x = random.random() * 5 
      tlog("writer sleep (%s)" % x) 
      pending = x 
      while pending > 0: 
       tosleep = min(0.5, pending) 
       if mq.aborted: 
        return 
       time.sleep(tosleep) 
       pending -= tosleep 

      tlog("write: %s" % x) 
      mq.put("POS %s val=%s" % (pos, x)) 
      pos += 1 
    finally: 
     tlog("writer END") 

def testStart(): 
    try: 
     for i in xrange(9): 
      th = threading.Thread(None, reader, "reader %s" % i,(), {}, None) 
      th.start() 
     for i in xrange(3): 
      th = threading.Thread(None, writer, "writer %s" % i,(), {}, None) 
      th.start() 
     time.sleep(30) # seconds for testing 
    finally: 
     print "main thread: abort()" 
     mq.abort() 

if __name__ == "__main__": 
    testStart()