wip

re6st/compat.py

+try:
+    from socket import socketpair
+except ImportError
+    import errno, socket
+
+    def socketpair():
+        # Originally written by Tim Peters for ZEO.zrpc.trigger
+        w = socket.socket()
+        failed = 0
+        while 1:
+            # Bind to a local port; for efficiency, let the OS pick
+            # a free port for us.
+            # Unfortunately, stress tests showed that we may not
+            # be able to connect to that port ("Address already in
+            # use") despite that the OS picked it.  This appears
+            # to be a race bug in the Windows socket implementation.
+            # So we loop until a connect() succeeds (almost always
+            # on the first try).  See the long thread at
+            # http://mail.zope.org/pipermail/zope/2005-July/160433.html
+            # for hideous details.
+            a = socket.socket()
+            try:
+                a.bind(("127.0.0.1", 0))
+                a.listen(1)
+                w.connect(a.getsockname())
+                return w, a.accept()[0] # success
+            except socket.error, detail:
+                if detail[0] != errno.WSAEADDRINUSE or failed >= 9:
+                    # "Address already in use" is the only error
+                    # I've seen on two WinXP Pro SP2 boxes, under
+                    # Pythons 2.3.5 and 2.4.1.
+                    w.close()
+                    raise
+                # assert failed < 2 # never triggered in Tim's tests
+                failed += 1
+            finally:
+                # Close `a` and try again.  Note:  I originally put a short
+                # sleep() here, but it didn't appear to help or hurt.
+                a.close()

re6st/utils.py

-import argparse, errno, logging, os, shlex, signal, socket
-import struct, subprocess, textwrap, threading, time
+import argparse, errno, logging, os, select, shlex, signal, socket
+import struct, subprocess, sys, textwrap, thread, threading, time
+from .compat import socketpair
 
 logging_levels = logging.WARNING, logging.INFO, logging.DEBUG, 5
 
@@ -105,6 +106,58 @@ class Popen(subprocess.Popen):
         return r
 
 
+class SelectableJob(threading.Thread):
+
+    def __init__(self, func=None):
+        super(SelectableJob, self).__init__()
+        self.daemon = True
+        self.__master, self.__slave = socketpair()
+        if func is not None:
+            self.release(func)
+        self.start()
+
+    @property
+    def fileno(self):
+        return self.__master.fileno
+
+    def run(self):
+        try:
+            while self.__slave.recv(1):
+                try:
+                    self.__result = self.__func()
+                except:
+                    self.__result = sys.exc_info()
+                    self.__slave.send('\1')
+                else:
+                    self.__slave.send('\0')
+        finally:
+            self.__slave.close()
+
+    def read(self):
+        t = self.__master.recv(1)
+        r = self.__result
+        del self.__result
+        if t == '\0':
+            return r
+        raise r[0], r[1], r[2]
+
+    def release(self, func=None):
+        if func is None:
+            self.__master.close()
+        else:
+            self.__func = func
+            self.__master.send('\0')
+
+
+def select_no_eintr(rlist, wlist=[], elist=[], wakeup=None):
+    while True:
+        try:
+            return select.select(rlist, wlist, elist, wakeup and
+                                 max(0, wakeup - time.time()))
+        except select.error as e:
+            if e.args[0] != errno.EINTR:
+                raise
+
 def makedirs(path):
     try:
         os.makedirs(path)

re6stnet

 #!/usr/bin/python
-import atexit, errno, logging, os, select, signal
-import sqlite3, subprocess, sys, time, traceback
+import atexit, logging, os, signal, sqlite3
+import subprocess, sys, threading, time, traceback
 from collections import deque
 from OpenSSL import crypto
 from re6st import db, plib, tunnel, utils
@@ -114,9 +114,6 @@ def getConfig():
 
     return parser.parse_args()
 
-def renew(*args):
-    raise ReexecException("Restart to renew certificate")
-
 def maybe_renew(path, cert, info, renew):
     while True:
         next_renew = utils.notAfter(cert) - RENEW_PERIOD
@@ -348,33 +345,29 @@ def main():
 
             # main loop
             if tunnel_manager is None:
-                signal.signal(signal.SIGALRM, renew)
-                signal.alarm(int(next_renew - time.time()))
-                try:
-                    sys.exit(os.WEXITSTATUS(os.wait()[1]))
-                finally:
-                    signal.alarm(0)
+                wait = utils.SelectableJob(os.wait)
+                cleanup.append(wait.release)
+                if utils.select_no_eintr([wait], wakeup=next_renew)[0]:
+                    sys.exit(os.WEXITSTATUS(wait.read()[1]))
+                raise ReexecException("Restart to renew certificate")
             cleanup += tunnel_manager.delInterfaces, tunnel_manager.killAll
             while True:
-                next = tunnel_manager.next_refresh
+                wakeup = tunnel_manager.next_refresh
                 if forwarder:
-                    next = min(next, forwarder.next_refresh)
-                r = [read_pipe, tunnel_manager.sock]
-                try:
-                    r = select.select(r, [], [], max(0, next - time.time()))[0]
-                except select.error as e:
-                    if e.args[0] != errno.EINTR:
-                        raise
-                    continue
+                    wakeup = min(wakeup, forwarder.next_refresh)
+                r = utils.select_no_eintr([read_pipe, tunnel_manager.sock],
+                                          wakeup=wakeup)[0]
                 if read_pipe in r:
                     tunnel_manager.handleTunnelEvent(read_pipe.readline())
                 if tunnel_manager.sock in r:
                     tunnel_manager.handlePeerEvent()
+                # IDEA: Use SelectableJob(os.wait) to wake up as soon as a
+                #       tunnel is closed, thus avoiding to wake up for nothing.
                 t = time.time()
                 if t >= tunnel_manager.next_refresh:
                     tunnel_manager.refresh()
                     if t >= next_renew:
-                        renew()
+                        raise ReexecException("Restart to renew certificate")
                 if forwarder and t >= forwarder.next_refresh:
                     forwarder.refresh()
         finally: