难倒我遇到的这个线程同步。基本上,我正在写入输出缓冲区,并等待条件变量,直到读取缓冲区填充来自套接字的响应。这是一个非常简单的线程同步。
def write_wait_response(self, buffer, timeout=30):
'''
Write and wait for response
Params:
Buffer BYTE encoded data
Timeout timeout to wait for response
Returns:
response str if successful
'''
self.buffer = buffer
if self.waitLock(timeout):
# condition var was signaled, we can return a response
readbuf = bytes(self.readbuffer)
self.readbuffer = b''
return readbuf
else:
print("AsyncClientSocket: No response recieved from {} in {} seconds, dumping buffer".format(
self.sa, timeout))
self.buffer = ''
raise TimeoutError("AsyncClientSocket Timed Out")
def handle_read(self):
self.readbuffer, address = self.recvfrom(2048)
print(self.readbuffer)
print("notifying")
self.cond.notifyAll()
看起来够简单了吧?有 1 个线程在等待条件变量,还有 1 个线程(asyncore 异步回调循环)将填充 self.readbuffer 并通知条件变量。更奇怪的是:如果我执行 time.sleep() 而不是使用条件变量,我会在 write_wait_response() 的调用线程上得到一个完美填充的 self.readbuffer。显然,这不是我可以接受的解决方案。
这是我期望发生的事情:
控制台输出:
waiting <- thread 1 waiting on CV
AsyncClientSocket: writing 5 bytes <- thread 2: handle_write
b'200,2' <- thread 2: that's the server response
notifying <- thread 2: that's handle_read attempting to notify the held CV
error: uncaptured python exception, closing channel <my_socket_stuff.AsyncClientSocket connected 127.0.0.1:50000 at 0x1051bf438> (<class 'RuntimeError'>:cannot notify on un-acquired lock
注意:在此日志的末尾,线程 1 仍在等待 self.cond。这是怎么回事?
全班:
class AsyncClientSocket(asyncore.dispatcher):
def __init__(self, socketargs):
asyncore.dispatcher.__init__(self)
family, type, proto, canonname, sa = socketargs
self.sa = sa
self.create_socket(family, type)
if type == socket.SOCK_STREAM:
self.connect( sa )
elif type == socket.SOCK_DGRAM:
pass
self.buffer = b''
self.lock = threading.Lock()
self.cond = threading.Condition(self.lock)
self.readbuffer = b''
def write_wait_response(self, buffer, timeout=30):
'''
Write and wait for response
Params:
Buffer BYTE encoded data
Timeout timeout to wait for response
Returns:
response str if successful
'''
self.buffer = buffer
if self.waitLock(timeout):
# condition var was signaled, we can return a response
readbuf = bytes(self.readbuffer)
self.readbuffer = b''
return readbuf
else:
print("AsyncClientSocket: No response recieved from {} in {} seconds, dumping buffer".format(
self.sa, timeout))
self.buffer = ''
raise TimeoutError("AsyncClientSocket Timed Out")
def waitLock(self, timeout):
'''
Wait for timeout seconds on CV
'''
try:
self.cond.acquire()
print("waiting")
return self.cond.wait(timeout)
finally:
self.cond.release()
def handle_connect(self):
pass
def handle_close(self):
self.close()
def handle_read(self):
self.readbuffer, address = self.recvfrom(2048)
print(self.readbuffer)
print("notifying")
self.cond.notifyAll()
def writable(self):
return (len(self.buffer) > 0)
def handle_write(self):
print("AsyncClientSocket: writing {} bytes".format(len(self.buffer)))
self.readbuffer = b''
sent = self.sendto(self.buffer, self.sa)
self.buffer = self.buffer[sent:]
弄清楚了。这与异步无关。我只是错误地向条件变量发出信号。python3 threading api doc说notify()的调用线程必须获取底层锁,这是有道理的,不希望两个生产者通知同一个条件变量。希望一个阻塞在临界区,而另一个执行其任务。
def handle_read(self):
try:
self.cond.acquire()
self.readbuffer, address = self.recvfrom(2048)
print(self.readbuffer)
self.cond.notify()
finally:
self.cond.release()
本文收集自互联网,转载请注明来源。
如有侵权,请联系[email protected] 删除。
我来说两句