[Zodb-checkins] CVS: ZODB3/ZEO/zrpc - client.py:1.24 connection.py:1.41 trigger.py:1.14

Jeremy Hylton jeremy@zope.com
Fri, 3 Jan 2003 17:08:17 -0500


Update of /cvs-repository/ZODB3/ZEO/zrpc
In directory cvs.zope.org:/tmp/cvs-serv21352/ZEO/zrpc

Modified Files:
	client.py connection.py trigger.py 
Log Message:
Merge ZODB3-fast-restart-branch to the trunk


=== ZODB3/ZEO/zrpc/client.py 1.23 => 1.24 ===
--- ZODB3/ZEO/zrpc/client.py:1.23	Fri Dec 13 17:03:11 2002
+++ ZODB3/ZEO/zrpc/client.py	Fri Jan  3 17:07:41 2003
@@ -119,7 +119,7 @@
 
         # XXX need each connection started with async==0 to have a
         # callback
-        log("CM.set_async(%s)" % repr(map))
+        log("CM.set_async(%s)" % repr(map), level=zLOG.DEBUG)
         if not self.closed and self.trigger is None:
             log("CM.set_async(): first call")
             self.trigger = trigger()
@@ -294,6 +294,9 @@
             if success > 0:
                 break
             time.sleep(delay)
+            if self.mgr.is_connected():
+                log("CT: still trying to replace fallback connection",
+                    level=zLOG.INFO)
             delay = min(delay*2, self.tmax)
         log("CT: exiting thread: %s" % self.getName())
 


=== ZODB3/ZEO/zrpc/connection.py 1.40 => 1.41 ===
--- ZODB3/ZEO/zrpc/connection.py:1.40	Fri Dec 13 17:03:28 2002
+++ ZODB3/ZEO/zrpc/connection.py	Fri Jan  3 17:07:41 2003
@@ -21,7 +21,7 @@
 import ThreadedAsync
 from ZEO.zrpc import smac
 from ZEO.zrpc.error import ZRPCError, DisconnectedError
-from ZEO.zrpc.log import log, short_repr
+from ZEO.zrpc.log import short_repr, log
 from ZEO.zrpc.marshal import Marshaller
 from ZEO.zrpc.trigger import trigger
 import zLOG
@@ -115,13 +115,32 @@
     __super_init = smac.SizedMessageAsyncConnection.__init__
     __super_close = smac.SizedMessageAsyncConnection.close
 
-    protocol_version = "Z200"
+    # Protocol variables:
+    #
+    # oldest_protocol_version -- the oldest protocol version we support
+    # protocol_version -- the newest protocol version we support; preferred
+
+    oldest_protocol_version = "Z200"
+    protocol_version = "Z201"
+
+    # Protocol history:
+    #
+    # Z200 -- original ZEO 2.0 protocol
+    #
+    # Z201 -- added invalidateTransaction() to client;
+    #         renamed several client methods;
+    #         added lastTransaction() to server
 
     def __init__(self, sock, addr, obj=None):
         self.obj = None
         self.marshal = Marshaller()
         self.closed = 0
         self.msgid = 0
+        self.peer_protocol_version = None # Set in recv_handshake()
+        if isinstance(addr, types.TupleType):
+            self.log_label = "zrpc-conn:%s:%d" % addr
+        else:
+            self.log_label = "zrpc-conn:%s" % addr
         self.__super_init(sock, addr)
         # A Connection either uses asyncore directly or relies on an
         # asyncore mainloop running in a separate thread.  If
@@ -147,6 +166,9 @@
 
     __str__ = __repr__ # Defeat asyncore's dreaded __getattr__
 
+    def log(self, message, level=zLOG.BLATHER, error=None):
+        zLOG.LOG(self.log_label, level, message, error=error)
+
     def close(self):
         if self.closed:
             return
@@ -156,7 +178,7 @@
         self.__super_close()
 
     def close_trigger(self):
-        # overridden by ManagedConnection
+        # Overridden by ManagedConnection
         if self.trigger is not None:
             self.trigger.close()
 
@@ -164,7 +186,9 @@
         """Register obj as the true object to invoke methods on"""
         self.obj = obj
 
-    def handshake(self):
+    def handshake(self, proto=None):
+        # Overridden by ManagedConnection
+
         # When a connection is created the first message sent is a
         # 4-byte protocol version.  This mechanism should allow the
         # protocol to evolve over time, and let servers handle clients
@@ -174,17 +198,18 @@
         # first message received.
 
         # The client sends the protocol version it is using.
-        self._message_input = self.message_input
         self.message_input = self.recv_handshake
-        self.message_output(self.protocol_version)
+        self.message_output(proto or self.protocol_version)
 
-    def recv_handshake(self, message):
-        if message == self.protocol_version:
-            self.message_input = self._message_input
+    def recv_handshake(self, proto):
+        # Extended by ManagedConnection
+        del self.message_input
+        self.peer_protocol_version = proto
+        if self.oldest_protocol_version <= proto <= self.protocol_version:
+            self.log("received handshake %r" % proto, level=zLOG.INFO)
         else:
-            log("recv_handshake: bad handshake %s" % short_repr(message),
-                level=zLOG.ERROR)
-        # otherwise do something else...
+            self.log("bad handshake %s" % short_repr(proto), level=zLOG.ERROR)
+            raise ZRPCError("bad handshake %r" % proto)
 
     def message_input(self, message):
         """Decoding an incoming message and dispatch it"""
@@ -195,9 +220,9 @@
         msgid, flags, name, args = self.marshal.decode(message)
 
         if __debug__:
-            log("recv msg: %s, %s, %s, %s" % (msgid, flags, name,
-                                              short_repr(args)),
-                level=zLOG.TRACE)
+            self.log("recv msg: %s, %s, %s, %s" % (msgid, flags, name,
+                                                   short_repr(args)),
+                     level=zLOG.TRACE)
         if name == REPLY:
             self.handle_reply(msgid, flags, args)
         else:
@@ -205,8 +230,8 @@
 
     def handle_reply(self, msgid, flags, args):
         if __debug__:
-            log("recv reply: %s, %s, %s" % (msgid, flags, short_repr(args)),
-                level=zLOG.DEBUG)
+            self.log("recv reply: %s, %s, %s"
+                     % (msgid, flags, short_repr(args)), level=zLOG.DEBUG)
         self.replies_cond.acquire()
         try:
             self.replies[msgid] = flags, args
@@ -219,7 +244,8 @@
             msg = "Invalid method name: %s on %s" % (name, repr(self.obj))
             raise ZRPCError(msg)
         if __debug__:
-            log("calling %s%s" % (name, short_repr(args)), level=zLOG.BLATHER)
+            self.log("calling %s%s" % (name, short_repr(args)),
+                     level=zLOG.BLATHER)
 
         meth = getattr(self.obj, name)
         try:
@@ -228,8 +254,8 @@
             raise
         except Exception, msg:
             error = sys.exc_info()
-            log("%s() raised exception: %s" % (name, msg), zLOG.INFO,
-                error=error)
+            self.log("%s() raised exception: %s" % (name, msg), zLOG.INFO,
+                     error=error)
             error = error[:2]
             return self.return_error(msgid, flags, *error)
 
@@ -239,7 +265,7 @@
                                 (name, short_repr(ret)))
         else:
             if __debug__:
-                log("%s returns %s" % (name, short_repr(ret)), zLOG.DEBUG)
+                self.log("%s returns %s" % (name, short_repr(ret)), zLOG.DEBUG)
             if isinstance(ret, Delay):
                 ret.set_sender(msgid, self.send_reply, self.return_error)
             else:
@@ -252,7 +278,7 @@
         self.close()
 
     def log_error(self, msg="No error message supplied"):
-        log(msg, zLOG.ERROR, error=sys.exc_info())
+        self.log(msg, zLOG.ERROR, error=sys.exc_info())
 
     def check_method(self, name):
         # XXX Is this sufficient "security" for now?
@@ -304,8 +330,8 @@
         finally:
             self.msgid_lock.release()
         if __debug__:
-            log("send msg: %d, %d, %s, ..." % (msgid, flags, method),
-                zLOG.TRACE)
+            self.log("send msg: %d, %d, %s, ..." % (msgid, flags, method),
+                     zLOG.TRACE)
         buf = self.marshal.encode(msgid, flags, method, args)
         self.message_output(buf)
         return msgid
@@ -342,7 +368,7 @@
         self.thr_async = 1
 
     def is_async(self):
-        # overridden for ManagedConnection
+        # Overridden by ManagedConnection
         if self.thr_async:
             return 1
         else:
@@ -360,8 +386,8 @@
     def wait(self, msgid):
         """Invoke asyncore mainloop and wait for reply."""
         if __debug__:
-            log("wait(%d), async=%d" % (msgid, self.is_async()),
-                level=zLOG.TRACE)
+            self.log("wait(%d), async=%d" % (msgid, self.is_async()),
+                     level=zLOG.TRACE)
         if self.is_async():
             self._pull_trigger()
 
@@ -378,8 +404,8 @@
                 if reply is not None:
                     del self.replies[msgid]
                     if __debug__:
-                        log("wait(%d): reply=%s" % (msgid, short_repr(reply)),
-                            level=zLOG.DEBUG)
+                        self.log("wait(%d): reply=%s" %
+                                 (msgid, short_repr(reply)), level=zLOG.DEBUG)
                     return reply
                 if self.is_async():
                     self.replies_cond.wait(10.0)
@@ -388,14 +414,14 @@
                     try:
                         try:
                             if __debug__:
-                                log("wait(%d): asyncore.poll(%s)" %
-                                    (msgid, delay), level=zLOG.TRACE)
+                                self.log("wait(%d): asyncore.poll(%s)" %
+                                         (msgid, delay), level=zLOG.TRACE)
                             asyncore.poll(delay, self._map)
                             if delay < 1.0:
                                 delay += delay
                         except select.error, err:
-                            log("Closing.  asyncore.poll() raised %s." % err,
-                                level=zLOG.BLATHER)
+                            self.log("Closing.  asyncore.poll() raised %s."
+                                     % err, level=zLOG.BLATHER)
                             self.close()
                     finally:
                         self.replies_cond.acquire()
@@ -405,7 +431,7 @@
     def poll(self):
         """Invoke asyncore mainloop to get pending message out."""
         if __debug__:
-            log("poll(), async=%d" % self.is_async(), level=zLOG.TRACE)
+            self.log("poll(), async=%d" % self.is_async(), level=zLOG.TRACE)
         if self.is_async():
             self._pull_trigger()
         else:
@@ -414,7 +440,7 @@
     def pending(self):
         """Invoke mainloop until any pending messages are handled."""
         if __debug__:
-            log("pending(), async=%d" % self.is_async(), level=zLOG.TRACE)
+            self.log("pending(), async=%d" % self.is_async(), level=zLOG.TRACE)
         if self.is_async():
             return
         # Inline the asyncore poll() function to know whether any input
@@ -464,6 +490,64 @@
         self.mgr = mgr
         self.__super_init(sock, addr, obj)
         self.check_mgr_async()
+
+    # PROTOCOL NEGOTIATION:
+    #
+    # The code implementing protocol version 2.0.0 (which is deployed
+    # in the field and cannot be changed) *only* talks to peers that
+    # send a handshake indicating protocol version 2.0.0.  In that
+    # version, both the client and the server immediately send out
+    # their protocol handshake when a connection is established,
+    # without waiting for their peer, and disconnect when a different
+    # handshake is receive.
+    #
+    # The new protocol uses this to enable new clients to talk to
+    # 2.0.0 servers: in the new protocol, the client waits until it
+    # receives the server's protocol handshake before sending its own
+    # handshake.  The client sends the lower of its own protocol
+    # version and the server protocol version, allowing it to talk to
+    # servers using later protocol versions (2.0.2 and higher) as
+    # well: the effective protocol used will be the lower of the
+    # client and server protocol.
+    #
+    # The ZEO modules ClientStorage and ServerStub have backwards
+    # compatibility code for dealing with the previous version of the
+    # protocol.  The client accept the old version of some messages,
+    # and will not send new messages when talking to an old server.
+    #
+    # As long as the client hasn't sent its handshake, it can't send
+    # anything else; output messages are queued during this time.
+    # (Output can happen because the connection testing machinery can
+    # start sending requests before the handshake is received.)
+    #
+    # UPGRADING FROM ZEO 2.0.0 TO NEWER VERSIONS:
+    #
+    # Because a new client can talk to an old server, but not vice
+    # versa, all clients should be upgraded before upgrading any
+    # servers.  Protocol upgrades beyond 2.0.1 will not have this
+    # restriction, because clients using protocol 2.0.1 or later can
+    # talk to both older and newer servers.
+    #
+    # No compatibility with protocol version 1 is provided.
+
+    def handshake(self):
+        self.message_input = self.recv_handshake
+        self.message_output = self.queue_output
+        self.output_queue = []
+        # The handshake is sent by recv_handshake() below
+
+    def queue_output(self, message):
+        self.output_queue.append(message)
+
+    def recv_handshake(self, proto):
+        del self.message_output
+        proto = min(proto, self.protocol_version)
+        Connection.recv_handshake(self, proto) # Raise error if wrong proto
+        self.message_output(proto)
+        queue = self.output_queue
+        del self.output_queue
+        for message in queue:
+            self.message_output(message)
 
     # Defer the ThreadedAsync work to the manager.
 


=== ZODB3/ZEO/zrpc/trigger.py 1.13 => 1.14 ===
--- ZODB3/ZEO/zrpc/trigger.py:1.13	Fri Dec 13 17:03:41 2002
+++ ZODB3/ZEO/zrpc/trigger.py	Fri Jan  3 17:07:41 2003
@@ -16,6 +16,7 @@
 import os
 import socket
 import thread
+import errno
 
 if os.name == 'posix':