[svn r93] "nonblocking" support in tpool, to be exploited by dbmgr, so that

nonblocking DB calls don't need to context-switch to a native thread.

eventlet/db_pool.py

@@ -157,7 +157,7 @@ class GenericConnectionWrapper(object):
     def character_set_name(self,*args, **kwargs): return self._base.character_set_name(*args, **kwargs)
     def close(self,*args, **kwargs): return self._base.close(*args, **kwargs)
     def commit(self,*args, **kwargs): return self._base.commit(*args, **kwargs)
-    def cursor(self, cursorclass=None): return self._base.cursor(cursorclass)
+    def cursor(self, cursorclass=None, **kwargs): return self._base.cursor(cursorclass, **kwargs)
     def dump_debug_info(self,*args, **kwargs): return self._base.dump_debug_info(*args, **kwargs)
     def errno(self,*args, **kwargs): return self._base.errno(*args, **kwargs)
     def error(self,*args, **kwargs): return self._base.error(*args, **kwargs)

eventlet/httpd.py

@@ -495,6 +495,10 @@ class HttpProtocol(BaseHTTPServer.BaseHTTPRequestHandler):
                     try:
                         self.server.site.handle_request(request)
                     except ErrorResponse, err:
+                        import sys, traceback
+                        (a,b,tb) = sys.exc_info()
+                        traceback.print_exception(ErrorResponse,err,tb)
+
                         request.response(code=err.code,
                                          reason_phrase=err.reason,
                                          headers=err.headers,
@@ -528,6 +532,11 @@ class HttpProtocol(BaseHTTPServer.BaseHTTPRequestHandler):
                 raise
         self.socket.close()
 
+def tid():
+    n = long(id(api.getcurrent()))
+    if n < 0:
+        n = -n
+    return hex(n)
 
 class Server(BaseHTTPServer.HTTPServer):
     def __init__(self, socket, address, site, log, max_http_version=DEFAULT_MAX_HTTP_VERSION):
@@ -543,7 +552,7 @@ class Server(BaseHTTPServer.HTTPServer):
             self.log = self
 
     def write(self, something):
-        sys.stdout.write('%s' % (something, ))
+        sys.stdout.write('%s: %s' % (tid(), something, )); sys.stdout.flush()
 
     def log_message(self, message):
         self.log.write(message)

eventlet/tpool.py

@@ -24,6 +24,7 @@ from sys import stdout
 from Queue import Empty, Queue
 
 from eventlet import api, coros, httpc, httpd, util, wrappedfd
+from eventlet.api import trampoline, get_hub
 
 _rpipe, _wpipe = os.pipe()
 _rfile = os.fdopen(_rpipe,"r",0)
@@ -36,10 +37,10 @@ def _signal_t2e():
 _reqq = Queue(maxsize=-1)
 _rspq = Queue(maxsize=-1)
 
-def trampoline():
+def tpool_trampoline():
     global _reqq, _rspq
     while(True):
-        _c = _wrap_rfile.read(1)
+        _c = _wrap_rfile.recv(1)
         assert(_c != "")
         while not _rspq.empty():
             try:
@@ -62,14 +63,20 @@ def tworker():
         try:
             rv = meth(*args,**kwargs)
         except Exception,exn:
-            rv = exn
+            import sys, traceback
+            (a,b,tb) = sys.exc_info()
+            rv = (exn,a,b,tb)
         _rspq.put((e,rv))
         _signal_t2e()
 
 def erecv(e):
     rv = e.wait()
-    if isinstance(rv,Exception):
-        raise rv
+    if isinstance(rv,tuple) and len(rv) == 4 and isinstance(rv[0],Exception):
+        import sys, traceback
+        (e,a,b,tb) = rv
+        traceback.print_exception(Exception,e,tb)
+        traceback.print_stack()
+        raise e
     return rv
 
 def erpc(meth,*args, **kwargs):
@@ -92,7 +99,10 @@ class Proxy(object):
         if not callable(f):
             return f
         def doit(*args, **kwargs):
-            rv = erpc(f,*args,**kwargs)
+            if kwargs.pop('nonblocking',False):
+                rv = f(*args, **kwargs)
+            else:
+                rv = erpc(f,*args,**kwargs)
             if type(rv) in self._autowrap:
                 return Proxy(rv)
             else:
@@ -108,6 +118,6 @@ def setup():
         _threads[i].setDaemon(True)
         _threads[i].start()
 
-    api.spawn(trampoline)
+    api.spawn(tpool_trampoline)
 
 setup()