Commit 153fb529 authored by Guido van Rossum's avatar Guido van Rossum

Tim saw Win2k crashes that made me realize that the input and output

buffer variables here are accessed from multiple threads without any
locking@  Add such locking: a separate lock for input and one for
output.  XXX Note: handle_read() keeps the lock for a potentially long
time.  But this is required to serialize incoming calls anyway.

Unrelated nicety: use short_repr() when logging message output, for
consistency with other places.
parent b3c871b2
...@@ -13,15 +13,16 @@ ...@@ -13,15 +13,16 @@
############################################################################## ##############################################################################
"""Sized message async connections """Sized message async connections
$Id: smac.py,v 1.30 2002/09/29 02:46:58 gvanrossum Exp $ $Id: smac.py,v 1.31 2002/09/29 03:22:28 gvanrossum Exp $
""" """
import asyncore, struct import asyncore, struct
import threading
from ZEO.Exceptions import Disconnected from ZEO.Exceptions import Disconnected
import zLOG import zLOG
from types import StringType from types import StringType
from ZEO.zrpc.log import log from ZEO.zrpc.log import log, short_repr
import socket, errno import socket, errno
...@@ -63,6 +64,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher): ...@@ -63,6 +64,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher):
self._debug = debug self._debug = debug
elif not hasattr(self, '_debug'): elif not hasattr(self, '_debug'):
self._debug = __debug__ self._debug = __debug__
# __input_lock protects __inp, __input_len, __state, __msg_size
self.__input_lock = threading.Lock()
self.__inp = None # None, a single String, or a list self.__inp = None # None, a single String, or a list
self.__input_len = 0 self.__input_len = 0
# Instance variables __state and __msg_size work together: # Instance variables __state and __msg_size work together:
...@@ -74,6 +77,7 @@ class SizedMessageAsyncConnection(asyncore.dispatcher): ...@@ -74,6 +77,7 @@ class SizedMessageAsyncConnection(asyncore.dispatcher):
# The state alternates between 0 and 1. # The state alternates between 0 and 1.
self.__state = 0 self.__state = 0
self.__msg_size = 4 self.__msg_size = 4
self.__output_lock = threading.Lock() # Protects __output
self.__output = [] self.__output = []
self.__closed = 0 self.__closed = 0
self.__super_init(sock, map) self.__super_init(sock, map)
...@@ -95,6 +99,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher): ...@@ -95,6 +99,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher):
if not d: if not d:
return return
self.__input_lock.acquire()
try:
input_len = self.__input_len + len(d) input_len = self.__input_len + len(d)
msg_size = self.__msg_size msg_size = self.__msg_size
state = self.__state state = self.__state
...@@ -130,12 +136,24 @@ class SizedMessageAsyncConnection(asyncore.dispatcher): ...@@ -130,12 +136,24 @@ class SizedMessageAsyncConnection(asyncore.dispatcher):
else: else:
msg_size = 4 msg_size = 4
state = 0 state = 0
# XXX We call message_input() with __input_lock
# held!!! And message_input() may end up calling
# message_output(), which has its own lock. But
# message_output() cannot call message_input(), so
# the locking order is always consistent, which
# prevents deadlock. Also, message_input() may
# take a long time, because it can cause an
# incoming call to be handled. During all this
# time, the __input_lock is held. That's a good
# thing, because it serializes incoming calls.
self.message_input(msg) self.message_input(msg)
self.__state = state self.__state = state
self.__msg_size = msg_size self.__msg_size = msg_size
self.__inp = inp[offset:] self.__inp = inp[offset:]
self.__input_len = input_len - offset self.__input_len = input_len - offset
finally:
self.__input_lock.release()
def readable(self): def readable(self):
return 1 return 1
...@@ -147,6 +165,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher): ...@@ -147,6 +165,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher):
return 1 return 1
def handle_write(self): def handle_write(self):
self.__output_lock.acquire()
try:
output = self.__output output = self.__output
while output: while output:
# Accumulate output into a single string so that we avoid # Accumulate output into a single string so that we avoid
...@@ -177,6 +197,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher): ...@@ -177,6 +197,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher):
if n < len(v): if n < len(v):
output.insert(0, v[n:]) output.insert(0, v[n:])
break # we can't write any more break # we can't write any more
finally:
self.__output_lock.release()
def handle_close(self): def handle_close(self):
self.close() self.close()
...@@ -184,11 +206,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher): ...@@ -184,11 +206,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher):
def message_output(self, message): def message_output(self, message):
if __debug__: if __debug__:
if self._debug: if self._debug:
if len(message) > 40: log('message_output %d bytes: %s' %
m = message[:40]+' ...' (len(message), short_repr(message)),
else:
m = message
log('message_output %d bytes: %s' % (len(message), `m`),
level=zLOG.TRACE) level=zLOG.TRACE)
if self.__closed: if self.__closed:
...@@ -196,6 +215,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher): ...@@ -196,6 +215,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher):
"This action is temporarily unavailable." "This action is temporarily unavailable."
"<p>" "<p>"
) )
self.__output_lock.acquire()
try:
# do two separate appends to avoid copying the message string # do two separate appends to avoid copying the message string
self.__output.append(struct.pack(">i", len(message))) self.__output.append(struct.pack(">i", len(message)))
if len(message) <= SEND_SIZE: if len(message) <= SEND_SIZE:
...@@ -203,6 +224,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher): ...@@ -203,6 +224,8 @@ class SizedMessageAsyncConnection(asyncore.dispatcher):
else: else:
for i in range(0, len(message), SEND_SIZE): for i in range(0, len(message), SEND_SIZE):
self.__output.append(message[i:i+SEND_SIZE]) self.__output.append(message[i:i+SEND_SIZE])
finally:
self.__output_lock.release()
def close(self): def close(self):
if not self.__closed: if not self.__closed:
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment