Added keep-alive messages from load balancers to workers to detect wo…

…rkers that have gone away uncleanly.

src/zc/resumelb/README.txt

@@ -242,6 +242,9 @@ maintaining ZooKeeper trees.
 Change History
 ==============
 
+- Added keep-alive messages from load balancers to workers to detect
+  workers that have gone away uncleanly.
+
 0.7.1 (2012-10-17)
 ------------------
 

src/zc/resumelb/lb.py

@@ -437,9 +437,16 @@ def content():
             del self.requests[rno]
 
     def disconnected(self):
+        # keep-alive messages send after disconnecting can cause
+        # disconnected to be called a second time.  To avoid this, we
+        # replace the method on the instance with a noop.
+        self.disconnected = already_disconnected
         self.pool.remove(self)
         zc.resumelb.util.Worker.disconnected(self)
 
+def already_disconnected():
+    pass
+
 def parse_addr(addr):
     host, port = addr.split(':')
     return host, int(port)

src/zc/resumelb/lb.test

@@ -35,7 +35,7 @@ to these addresses. Let's wait for it to do so:
 When the workers get connections, they send the lb their resumes:
 
     >>> worker1, worker2 = [w.socket for w in workers]
-    >>> from zc.resumelb.util import read_message, write_message
+    >>> from zc.resumelb.util import write_message
 
     >>> write_message(worker1, 0, {'h1.com': 10.0})
     >>> write_message(worker2, 0, {'h2.com': 10.0})
@@ -50,6 +50,12 @@ supposed to go.
     >>> app1 = webtest.TestApp(lb.handle_wsgi)
     >>> g1 = spawn(app1.get, '/hi.html', {}, [('Host', 'h1.com')])
 
+    >>> import zc.resumelb.util
+    >>> def read_message(sock): # filter out keep alives
+    ...     while 1:            # we'll look at them later
+    ...         rno, data = zc.resumelb.util.read_message(sock)
+    ...         if rno != 0 or data is not None:
+    ...             return rno, data
     >>> rno, env1 = read_message(worker1)
 
     >>> rno
@@ -194,6 +200,17 @@ We'll interleave data for the 2 responses on worker1
     >>> g2.value.status, g2.value.body == '2'*10000
     ('200 OK', True)
 
+Keep-alive messages
+===================
+
+Periodically, the lb sends keep-alive messages to workers:
+
+    >>> gevent.sleep(.15)
+    >>> zc.resumelb.util.read_message(worker1)
+    (0, None)
+
+workers ignore these.
+
 Disconnecting clients
 =====================
 

src/zc/resumelb/protocol.txt

@@ -8,7 +8,12 @@ consist of:
 - 4-byte request number
 - Marshalled data
 
-The marshalled data can be:
+If request number is 0, the marshalled data:
+
+- resume dict from worker to lb
+- None from lb->worker, ignored keep alive
+
+For request number > 0, the marshalled data can be:
 
 - environment dict from LB->worker
 - string part of HTTP message body

src/zc/resumelb/tests.py

@@ -423,6 +423,11 @@ def setUp(test):
     test.globs['print_response'] = print_response
     test.globs['spawn'] = spawn
     test.globs['Worker'] = FauxWorker
+    old_write_keepalive_interval = zc.resumelb.util.LBWorker.write_keepalive_interval
+    zc.resumelb.util.LBWorker.write_keepalive_interval = 0.1
+    zope.testing.setupstack.register(
+        test, setattr, zc.resumelb.util.LBWorker,
+        'write_keepalive_interval', old_write_keepalive_interval)
 
 def zkSetUp(test):
     setUp(test)

src/zc/resumelb/util.py

@@ -55,7 +55,7 @@ def write_message(sock, rno, *a):
         try:
             sent = sock.send(data)
         except socket.error, err:
-            if err.args[0] in disconnected_errors:
+            if err.args[0] in disconnected_errors or sock.closed:
                 logger.debug("write_message disconnected %s", sock)
                 raise Disconnected()
             else:
@@ -65,8 +65,13 @@ def write_message(sock, rno, *a):
 def writer(writeq, sock, multiplexer):
     get = writeq.get
     write_message_ = write_message
+    timeout = multiplexer.write_keepalive_interval
     while 1:
-        rno, data = get()
+        try:
+            rno, data = get(True, timeout)
+        except gevent.queue.Empty:
+            rno = 0
+            data = None
         try:
             write_message_(sock, rno, data)
         except Disconnected:
@@ -191,6 +196,8 @@ class Worker:
 
     ReadQueue = gevent.queue.Queue
 
+    write_keepalive_interval = None
+
     def connected(self, socket, addr=None):
         if addr is None:
             addr = socket.getpeername()
@@ -230,6 +237,8 @@ def disconnected(self):
 
 class LBWorker(Worker):
 
+    write_keepalive_interval = 0.1
+
     ReadQueue = BufferedQueue
 
     def connected(self, socket, addr=None):

src/zc/resumelb/worker.test

@@ -194,6 +194,17 @@ In this example, we've also requested a very large output.
     <BLANKLINE>
     1200000
 
+Keep-alive messages
+===================
+
+Occassionally, the server will send keep-alive messages to the worker.
+These are ignored. They're used by the server to detect that a worker
+has been disconnected uncleanly.
+
+    >>> write_message(worker_socket, 0, None)
+
+These messages are ignored by the worker.
+
 Multiple connections (multiple load balancers)
 ==============================================