我有两个线程(生产者和消费者),我与Queue
. 问题是,当我强行中止生产者时,消费者有时会锁定。
我在文档中读到取消带有队列的线程可能会损坏队列并导致死锁。我没有明确地获得任何锁,但阅读 Queue.py 的源代码put
并get
正在这样做。
请问,有没有人知道当我中止线程时,它可能在get
/的中间put
,即使用锁然后不释放它?我能做些什么呢?我有时需要提前终止生产者。使用进程而不是线程会有什么不同吗?
我有两个线程(生产者和消费者),我与Queue
. 问题是,当我强行中止生产者时,消费者有时会锁定。
我在文档中读到取消带有队列的线程可能会损坏队列并导致死锁。我没有明确地获得任何锁,但阅读 Queue.py 的源代码put
并get
正在这样做。
请问,有没有人知道当我中止线程时,它可能在get
/的中间put
,即使用锁然后不释放它?我能做些什么呢?我有时需要提前终止生产者。使用进程而不是线程会有什么不同吗?
很可能您的死锁是由于未完成的线程造成的。如果你有 linux,你可以使用pyrasite的注入器来打印回溯(你会知道你的程序挂在哪里)
如果您在信号处理程序中使用任何锁 - 那么这可能是您的死锁(这有点复杂,请询问您是否需要解释)
创建进程而不是线程肯定会改变情况,但请记住,任何数据交换和同步都非常复杂。
也许这会有所帮助:
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()