[py-svn] r6917 - in py/dist: doc py/execnet

[hpk at codespeak.net]

Author: hpk
Date: Wed Oct 13 18:44:29 2004
New Revision: 6917

Modified:
   py/dist/doc/execnet.txt
   py/dist/py/execnet/gateway.py
   py/dist/py/execnet/gateway_test.py
Log:
- introduced reraising of RemoteError in 
  Channel.waitclose() and  Channel.receive(). 

- there still is a minor problem in the shutdown procedure
  of the SocketGateway
  


Modified: py/dist/doc/execnet.txt
==============================================================================
--- py/dist/doc/execnet.txt	(original)
+++ py/dist/doc/execnet.txt	Wed Oct 13 18:44:29 2004
@@ -81,14 +81,14 @@
         receives an item that was sent from the other side, 
         possibly blocking if there is none. 
         Note that exceptions from the other side will be 
-        reraised as gateway.ExecutionFailed exceptions containing 
+        reraised as gateway.RemoteError exceptions containing 
         a textual representation of the remote traceback. 
 
     channel.waitclose(timeout=None): 
         wait until this channel is closed.  Note that a closed
         channel may still hold items that will be received or 
         send. Note that exceptions from the other side will be 
-        reraised as gateway.ExecutionFailed exceptions containing 
+        reraised as gateway.RemoteError exceptions containing 
         a textual representation of the remote traceback. 
 
     #

Modified: py/dist/py/execnet/gateway.py
==============================================================================
--- py/dist/py/execnet/gateway.py	(original)
+++ py/dist/py/execnet/gateway.py	Wed Oct 13 18:44:29 2004
@@ -7,8 +7,21 @@
 debug = 0
 sysex = (KeyboardInterrupt, SystemExit) 
 
+class RemoteError(Exception):
+    """ Contains an Exceptions from the other side. """
+    def __init__(self, formatted): 
+        Exception.__init__(self) 
+        self.formatted = formatted 
+
+    def __str__(self):
+        return self.formatted 
+
+    def __repr__(self):
+        return "%s: %s" %(self.__class__.__name__, self.formatted) 
+
 class Gateway(object):
     num_worker_threads = 2
+    RemoteError = RemoteError
 
     def __init__(self, io, startcount=2): 
         self.io = io
@@ -166,7 +179,11 @@
         self._closeevent = threading.Event()
 
     def _close(self, error=None):
-        self._error = error
+        if error is not None:
+            self._error = RemoteError(error)
+            self._items.put(self._error) 
+        else:
+            self._error = None
         self._closeevent.set()
 
     def __repr__(self):
@@ -183,7 +200,8 @@
         self._closeevent.wait(timeout=timeout) 
         if not self._closeevent.isSet():
             raise IOError, "Timeout"
-        return self._error 
+        if self._error:
+            raise self._error 
         
     def send(self, item): 
         """sends the given item to the other side of the channel, 
@@ -200,7 +218,10 @@
         reraised as gateway.ExecutionFailed exceptions containing 
         a textual representation of the remote traceback. 
         """
-        return self._items.get() 
+        x = self._items.get() 
+        if isinstance(x, RemoteError):
+            raise x 
+        return x
 
 # 
 # helpers 
@@ -286,6 +307,7 @@
             return "<Message.%s channelid=%d %r>" %(self.__class__.__name__, 
                         self.channelid, self.data)
 
+
 def _setupmessages():
     class EXIT_GATEWAY(Message):
         def handle(self, gateway):
@@ -333,7 +355,7 @@
     try:
         return cache.setdefault(name, {})[id(gw)]
     except KeyError:
-        cache[name][id(gw)] = x = "%s.%d" %(gw.__class__.__name__, len(cache[name]))
+        cache[name][id(gw)] = x = "%s:%s.%d" %(os.getpid(), gw.__class__.__name__, len(cache[name]))
         return x
 
 _gateways = []

Modified: py/dist/py/execnet/gateway_test.py
==============================================================================
--- py/dist/py/execnet/gateway_test.py	(original)
+++ py/dist/py/execnet/gateway_test.py	Wed Oct 13 18:44:29 2004
@@ -48,7 +48,7 @@
 
     def test_channel_timeouterror(self):
         channel = self.fac.new() 
-        py.test.raises(IOError, channel.waitclose, timeout=0.1)
+        py.test.raises(IOError, channel.waitclose, timeout=0.01)
 
     def test_channel_close(self):
         channel = self.fac.new()
@@ -58,9 +58,8 @@
     def test_channel_close_error(self):
         channel = self.fac.new()
         channel._close("error") 
-        err = channel.waitclose(0.1)
-        assert err == "error"
-    
+        py.test.raises(gateway.RemoteError, channel.waitclose, 0.1) 
+
 class PopenGatewayTestSetup: 
     disabled = True 
     def setup_class(cls):
@@ -69,7 +68,6 @@
     def teardown_class(cls):
         cls.gw.exit()  
 
-
 class BasicRemoteExecution: 
     disabled = True 
 
@@ -89,6 +87,18 @@
         result = channel.receive()
         assert result == 42
 
+    def test_channel_close_and_then_receive_error(self):
+        channel = self.gw.remote_exec_async('raise ValueError')
+        py.test.raises(gateway.RemoteError, channel.receive) 
+
+    def test_channel_close_and_then_receive_error_multiple(self):
+        channel = self.gw.remote_exec_async('channel.send(42) ; raise ValueError')
+        import time
+        time.sleep(0.1)
+        x = channel.receive()
+        assert x == 42 
+        py.test.raises(gateway.RemoteError, channel.receive) 
+
 class TestBasicPopenGateway(PopenGatewayTestSetup, BasicRemoteExecution): 
     disabled = False 
     def test_many_popen(self):
@@ -134,7 +144,6 @@
                 else:
                     channel.send(i) 
                     startserver(sock)
-                    print "started server with socket"
                     break
             else:
                 channel.send(None)