[Checkins] SVN: zc.resumelb/trunk/src/zc/resumelb/ Added graceful worker shutdown on SIGTERM.

[jim]

Log message for revision 124661:
  Added graceful worker shutdown on SIGTERM.
  

Changed:
  U   zc.resumelb/trunk/src/zc/resumelb/README.txt
  U   zc.resumelb/trunk/src/zc/resumelb/tests.py
  U   zc.resumelb/trunk/src/zc/resumelb/worker.py
  U   zc.resumelb/trunk/src/zc/resumelb/worker.test
  U   zc.resumelb/trunk/src/zc/resumelb/zk.py
  U   zc.resumelb/trunk/src/zc/resumelb/zk.test

-=-
Modified: zc.resumelb/trunk/src/zc/resumelb/README.txt
===================================================================
--- zc.resumelb/trunk/src/zc/resumelb/README.txt	2012-03-21 19:55:46 UTC (rev 124660)
+++ zc.resumelb/trunk/src/zc/resumelb/README.txt	2012-03-21 19:56:48 UTC (rev 124661)
@@ -243,6 +243,8 @@
 0.1.1 (2012-03-??)
 ------------------
 
+- Added graceful worker shutdown on SIGTERM.
+
 - Updated the API for application trace logging to match that of
   zc.zservertracelog, mainly to get database logging for ZTK
   applications.

Modified: zc.resumelb/trunk/src/zc/resumelb/tests.py
===================================================================
--- zc.resumelb/trunk/src/zc/resumelb/tests.py	2012-03-21 19:55:46 UTC (rev 124660)
+++ zc.resumelb/trunk/src/zc/resumelb/tests.py	2012-03-21 19:56:48 UTC (rev 124661)
@@ -18,6 +18,7 @@
 import manuel.capture
 import manuel.doctest
 import manuel.testing
+import mock
 import os
 import re
 import time
@@ -77,6 +78,7 @@
 
 def setUp(test):
     zope.testing.setupstack.setUpDirectory(test)
+    zope.testing.setupstack.context_manager(test, mock.patch('gevent.signal'))
     global pid
     pid = 6115
     test.globs['wait'] = zope.testing.wait.Wait(getsleep=lambda : gevent.sleep)

Modified: zc.resumelb/trunk/src/zc/resumelb/worker.py
===================================================================
--- zc.resumelb/trunk/src/zc/resumelb/worker.py	2012-03-21 19:55:46 UTC (rev 124660)
+++ zc.resumelb/trunk/src/zc/resumelb/worker.py	2012-03-21 19:56:48 UTC (rev 124661)
@@ -136,6 +136,15 @@
         if hasattr(self, 'threadpool'):
             self.threadpool.kill()
 
+    def shutdown(self):
+        self.server.close()
+        while 1:
+            if [conn for conn in self.connections if conn.readers]:
+                gevent.sleep(.01)
+            else:
+                break
+        self.stop()
+
     def handle_connection(self, sock, addr):
         try:
             conn = zc.resumelb.util.Worker()

Modified: zc.resumelb/trunk/src/zc/resumelb/worker.test
===================================================================
--- zc.resumelb/trunk/src/zc/resumelb/worker.test	2012-03-21 19:55:46 UTC (rev 124660)
+++ zc.resumelb/trunk/src/zc/resumelb/worker.test	2012-03-21 19:56:48 UTC (rev 124661)
@@ -341,6 +341,30 @@
     True
 
 
+Gracefull shutdown
+==================
+
+Workers have a shutdown method that stops accepting and waits for
+outstanding requests to be completed.  We have some outstating
+requests. Let's shutdown:
+
+    >>> shutdown_greenlet = gevent.spawn(worker.shutdown)
+
+We used a greenlet, because we don't want to block.
+
+    >>> shutdown_greenlet.join(.01)
+    >>> shutdown_greenlet.ready()
+    False
+
+We can't connect to the worker anymore:
+
+    >>> gevent.socket.create_connection(worker.addr)
+    Traceback (most recent call last):
+    ...
+    error: [Errno 111] Connection refused
+
+If we finish the outstanding requests, the worker will finish shutting down:
+
     >>> write_message(worker_socket, 3, '')
     >>> print_response(worker_socket, 3) # doctest: +NORMALIZE_WHITESPACE
     3 200 OK
@@ -365,10 +389,10 @@
     6115 10000 44bd0dbf8208fea52dc6180376d14798b86847bd
     <BLANKLINE>
 
-Cleanup:
+    >>> shutdown_greenlet.join(.1)
+    >>> shutdown_greenlet.ready()
+    True
 
-    >>> worker.stop()
-
 Saving and loading resumes
 --------------------------
 

Modified: zc.resumelb/trunk/src/zc/resumelb/zk.py
===================================================================
--- zc.resumelb/trunk/src/zc/resumelb/zk.py	2012-03-21 19:55:46 UTC (rev 124660)
+++ zc.resumelb/trunk/src/zc/resumelb/zk.py	2012-03-21 19:56:48 UTC (rev 124661)
@@ -18,6 +18,7 @@
 import logging
 import os
 import re
+import signal
 import sys
 import zc.parse_addr
 import zc.zk
@@ -60,6 +61,12 @@
     worker.zk = zk
     worker.__zksettings = settings
 
+    def shutdown():
+        zk.close()
+        worker.shutdown()
+
+    gevent.signal(signal.SIGTERM, shutdown)
+
     if run:
         try:
             worker.server.serve_forever()

Modified: zc.resumelb/trunk/src/zc/resumelb/zk.test
===================================================================
--- zc.resumelb/trunk/src/zc/resumelb/zk.test	2012-03-21 19:55:46 UTC (rev 124660)
+++ zc.resumelb/trunk/src/zc/resumelb/zk.test	2012-03-21 19:56:48 UTC (rev 124661)
@@ -101,9 +101,32 @@
     >>> bool(meta['ephemeralOwner'])
     True
 
+Shutdown
+--------
+
+A shutdown signal handler is registered.  We can call it to shut the
+worker down:
+
+    >>> import gevent, signal
+    >>> gevent.signal.call_args[0][0] == signal.SIGTERM
+    True
+
+    >>> gevent.signal.call_args[0][1]()
+    >>> gevent.sleep(.01)
+    >>> zk.get_children('/test/lb/workers/providers')
+    []
+
+    >>> gevent.socket.create_connection(worker.addr)
+    Traceback (most recent call last):
+    ...
+    error: [Errno 111] Connection refused
+
+
+Variations
+----------
+
 Let's try again, but this time, don't set up logging:
 
-    >>> worker.stop(); worker.zk.close()
     >>> with mock.patch('ZConfig.configureLoggers') as configureLoggers:
     ...   with mock.patch('logging.basicConfig') as basicConfig:
     ...     worker = zc.resumelb.zk.worker(