execnet: robust gateways to hybrid execution environments

execnet: rapid multi-Python deployment

home | 
         install | 

--- a/execnet/__init__.py
+++ b/execnet/__init__.py
@@ -4,7 +4,7 @@ package for connecting to local and remo
 
 (c) 2009, Holger Krekel and others
 """
-__version__ = "1.0.1a"
+__version__ = "1.0.1"
 
 import execnet.apipkg
 

--- a/testing/test_basics.py
+++ b/testing/test_basics.py
@@ -8,7 +8,7 @@ def test_subprocess_interaction(anypytho
     line = gateway.popen_bootstrapline
     compile(line, 'xyz', 'exec')
     args = [str(anypython), '-c', line]
-    popen = subprocess.Popen(args, bufsize=0, 
+    popen = subprocess.Popen(args, bufsize=0, universal_newlines=True,
                              stdin=subprocess.PIPE, stdout=subprocess.PIPE)
     def send(line):
         popen.stdin.write(line.encode('ascii'))

--- a/doc/docindex.txt
+++ /dev/null
@@ -1,10 +0,0 @@
-
-Documentation Index
-============================
-
-.. toctree::
-   :maxdepth: 2
-
-   basics
-   examples
-   changelog

--- a/setup.py
+++ b/setup.py
@@ -1,15 +1,16 @@
 """
-execnet: connect your execution environments 
+execnet: rapid multi-Python deployment 
 ========================================================
 
-.. image:: _static/pythonring.png
-   :align: right
+.. _execnet: http://codespeak.net/execnet
 
-Execnet allows to ad-hoc connect to Python interpreters across version, platform and network barriers.  It provides a minimal, fast and robust API for the following uses:
+execnet_ provides carefully tested means to ad-hoc interact with Python
+interpreters across version, platform and network barriers.  It provides
+a minimal and fast API targetting the following uses:
 
-* distribute tasks to multiple CPUs
-* deploy hybrid applications 
-* manage local and remote execution environments
+* distribute tasks to local or remote CPUs 
+* write and deploy hybrid multi-process applications 
+* write scripts to administer a bunch of exec environments
 
 Features
 ------------------
@@ -39,7 +40,7 @@ from execnet import __version__
 def main():
     setup(
         name='execnet',
-        description='execnet: connect your execution environments',
+        description='execnet: rapid multi-Python deployment',
         long_description = __doc__,
         version= __version__,
         url='http://codespeak.net/execnet',

From commits-noreply at bitbucket.org  Sat Dec  5 19:29:08 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sat,  5 Dec 2009 18:29:08 +0000 (UTC)
Subject: [execnet-commit] execnet commit e77a2974394b: Added tag 1.0.1 for
	changeset	66760dc71203
Message-ID: <20091205182908.784C27EF0D@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1260037737 -3600
# Node ID e77a2974394bd0c6bc18f6c2f90f436be51b90ef
# Parent 66760dc71203dbceeb08851cf4cf74e2687156d0
Added tag 1.0.1 for changeset 66760dc71203

--- a/.hgtags
+++ b/.hgtags
@@ -2,3 +2,4 @@ 39e4b3a1397adad17ecfcd9e65a08521d90b2795
 7cf86a1811b11aa58112a255e6afba870c8d855e 1.0.0
 7cf86a1811b11aa58112a255e6afba870c8d855e 1.0.0
 5d1c438e335dc369834207882ec5f3227d365c11 1.0.0
+66760dc71203dbceeb08851cf4cf74e2687156d0 1.0.1

From commits-noreply at bitbucket.org  Sun Dec  6 18:56:35 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun,  6 Dec 2009 17:56:35 +0000 (UTC)
Subject: [execnet-commit] execnet commit de6284add5db: avoid deprecated
	PopenGateway, new Group.defaultspec example, some doc tweaks
Message-ID: <20091206175635.422797EE74@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1260118830 -3600
# Node ID de6284add5db8329f9b57fc44026b6f031771782
# Parent e77a2974394bd0c6bc18f6c2f90f436be51b90ef
avoid deprecated PopenGateway, new Group.defaultspec example, some doc tweaks

--- a/doc/Makefile
+++ b/doc/Makefile
@@ -31,7 +31,7 @@ clean:
 	-rm -rf $(BUILDDIR)/*
 
 install: clean html linkcheck
-	rsync -avz $(BUILDDIR)/html/ code:www-execnet/trunk/
+	rsync -avz $(BUILDDIR)/html/ code:www-execnet/
 
 html:
 	$(SPHINXBUILD) -b html $(ALLSPHINXOPTS) $(BUILDDIR)/html

--- a/doc/install.txt
+++ b/doc/install.txt
@@ -13,7 +13,7 @@ Install a public `pypi release`_ via `ea
         or
 
     hg clone https://hpk42 at bitbucket.org/hpk42/execnet/
-    python setup.py install # or add checkout path to PYTHONPATH directly
+    python setup.py install # or 'develop' or add checkout path to PYTHONPATH 
 
 Next checkout the basic api and examples:
 

--- a/doc/example/popen_read_multiple.py
+++ b/doc/example/popen_read_multiple.py
@@ -9,7 +9,7 @@ NUM_PROCESSES = 5
 
 channels = []
 for i in range(NUM_PROCESSES):
-    gw = execnet.PopenGateway() # or use SSH or socket gateways 
+    gw = execnet.makegateway() # or use SSH or socket gateways 
     channel = gw.remote_exec("""
         import time
         secs = channel.receive()

--- a/doc/example/test_group.txt
+++ b/doc/example/test_group.txt
@@ -74,3 +74,25 @@ the related process is made::
 execnet aims to provide totally robust termination so if 
 you have left-over processes or other termination issues 
 please :doc:`report them <../support>`.  thanks!
+
+
+Using Groups to manage a certain type of gateway
+------------------------------------------------------
+
+Set ``group.defaultspec`` to determine the default gateway
+specification used by ``group.makegateway()``:
+
+    >>> import execnet
+    >>> group = execnet.Group()
+    >>> group.defaultspec = "ssh=localhost//chdir=mytmp//nice=20"
+    >>> gw = group.makegateway() 
+    >>> ch = gw.remote_exec("""
+    ...      import os.path
+    ...      basename = os.path.basename(os.getcwd())
+    ...      channel.send(basename)
+    ... """)
+    >>> ch.receive()
+    'mytmp'
+
+This way a Group object becomes kind of a Gateway factory where
+the factory-caller does not need to know the setup.

--- a/doc/example/test_multi.txt
+++ b/doc/example/test_multi.txt
@@ -41,7 +41,7 @@ Use channel callbacks if you want to pro
 data immediately and without blocking execution::
 
     >>> import execnet
-    >>> gw = execnet.PopenGateway()
+    >>> gw = execnet.makegateway()
     >>> ch = gw.remote_exec("channel.receive() ; channel.send(42)")
     >>> l = []
     >>> ch.setcallback(l.append)

--- a/doc/example/test_info.txt
+++ b/doc/example/test_info.txt
@@ -4,10 +4,10 @@ basic local and remote communication
 Comparing current working directories
 ----------------------------------------
 
-A popen gateway has the same working directory as the instantiatior::
+A local popen gateway has the same working directory as the instantiatior::
 
     >>> import execnet, os
-    >>> gw = execnet.PopenGateway()
+    >>> gw = execnet.makegateway()   
     >>> ch = gw.remote_exec("import os; channel.send(os.getcwd())")
     >>> res = ch.receive()
     >>> assert res == os.getcwd()
@@ -48,7 +48,7 @@ running in a remote_exec situation by ch
 You can now remote-execute the module like this::
 
     >>> import execnet, remote1
-    >>> gw = execnet.PopenGateway()
+    >>> gw = execnet.makegateway()
     >>> ch = gw.remote_exec(remote1)
     >>> print (ch.receive())
     initialization complete
@@ -68,7 +68,7 @@ you can setup a serving loop and write a
 Then on the local side you can do::
 
     >>> import execnet, remotecmd
-    >>> gw = execnet.PopenGateway()
+    >>> gw = execnet.makegateway()
     >>> ch = gw.remote_exec(remotecmd)
     >>> ch.send('simple(10)') # execute func-call remotely
     >>> ch.receive()
@@ -95,7 +95,7 @@ Simply run the script like this::
     python socketserver.py :8888   # bind to all IPs, port 8888
 
 You can then instruct execnet on your local machine to bootstrap 
-itself into the remote socket endpoint:
+itself into the remote socket endpoint::
 
     import execnet
     gw = execnet.SocketGateway("TARGET-IP:8888")

--- a/doc/example/redirect_remote_output.py
+++ b/doc/example/redirect_remote_output.py
@@ -10,7 +10,7 @@ showcasing features of the channel objec
 
 import py
 
-gw = execnet.PopenGateway()
+gw = execnet.makegateway()
 
 outchan = gw.remote_exec("""
     import sys

--- a/doc/basics.txt
+++ b/doc/basics.txt
@@ -24,7 +24,7 @@ passing it a gateway specification or UR
 
 Here is an example which instantiates a simple Python subprocess::
 
-    >>> gateway = execnet.makegateway("popen")
+    >>> gateway = execnet.makegateway()
 
 gateways allow to `remote execute code`_ and 
 `exchange data`_ bidirectionally.  
@@ -131,7 +131,7 @@ rsync: synchronise filesystem with remot
 Here is a basic example for using RSync::
 
     rsync = execnet.RSync('/tmp/source')
-    gw = execnet.makegateway("popen")
+    gw = execnet.makegateway()
     rsync.add_target(gw, '/tmp/dest')
     rsync.send()

From commits-noreply at bitbucket.org  Sun Dec  6 18:56:37 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun,  6 Dec 2009 17:56:37 +0000 (UTC)
Subject: [execnet-commit] execnet commit 800841c9b050: skip jython tests if
	it is not a 2.5	version
Message-ID: <20091206175637.8D9A67EE74@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1260121901 -3600
# Node ID 800841c9b050c49df81bddf3270d74cd5263df69
# Parent de6284add5db8329f9b57fc44026b6f031771782
skip jython tests if it is not a 2.5 version

--- a/conftest.py
+++ b/conftest.py
@@ -1,6 +1,7 @@
 import execnet
 import py
 import sys
+import subprocess
 
 collect_ignore = ['build', 'doc/_build']
 
@@ -63,9 +64,24 @@ def pytest_generate_tests(metafunc):
                      'pypy-c', 'jython'):
             metafunc.addcall(id=name, param=name)
 
+def getexecutable(name, cache={}):
+    try:
+        return cache[name]
+    except KeyError:
+        executable = py.path.local.sysfind(name)
+        if executable:
+            if name == "jython":
+                popen = subprocess.Popen([str(executable), "--version"], 
+                    stderr=subprocess.PIPE)
+                out, err = popen.communicate()
+                if not err or "2.5" not in err:
+                    executable = None
+        cache[name] = executable
+        return executable
+
 def pytest_funcarg__anypython(request):
     name = request.param
-    executable = py.path.local.sysfind(name)
+    executable = getexecutable(name)
     if executable is None:
         if sys.platform == "win32":
             executable = winpymap.get(name, None)

From commits-noreply at bitbucket.org  Sun Dec  6 20:11:57 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun,  6 Dec 2009 19:11:57 +0000 (UTC)
Subject: [execnet-commit] execnet commit a15041d7bb5e: work out issues/ideas
	a bit
Message-ID: <20091206191157.4755D7EEE5@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1260125857 -3600
# Node ID a15041d7bb5e008dc1a4a2976f03a79ca2085841
# Parent 800841c9b050c49df81bddf3270d74cd5263df69
work out issues/ideas a bit

--- a/ISSUES.txt
+++ b/ISSUES.txt
@@ -19,8 +19,9 @@ path: execnet/rsync.py
 currently rsyncing bails out with a strange message when trying to rsync
 links. should provide better message or even some support for doing it.
 
-a process keeping state across invoactions
---------------------------------------------------
+socketservice process for keeping cross-invocation state 
+---------------------------------------------------------
+
 tags: 1.1 feature newdist 
 
 A building block for keeping permanent network 
@@ -36,6 +37,16 @@ that keeps state. Considered:
   will re-use the already-imported gateway base code
   so again avoid parse/compile on arbitrary strings.
 
+Possible code example:
+
+    # connect to on-demand started port-8888 socket server
+    # similar to popen+socket//installvia=popen 
+    gw = group.makegateway("socketservice//port=8888")
+    gw.remote_exec("...") # execute code in a thread in the socket server
+    ... 
+    # exit the gateway-connection, but leave socket server running
+    gw.exit() 
+
 a channel-IO based gateway
 --------------------------------------------------
 tags: 1.1 feature newdist 
@@ -46,3 +57,24 @@ This way a gateway can be instantiated t
 intermediating gateway which bi-directionallry forwads 
 Messages through a existing channel.
 
+A code example could look like this:
+
+    group.setlocalfactory(port=8888)
+    gw = group.makegateway("ssh=codespeak.net")
+    # local ssh-process would be a child of the factory service
+    # which could have timeouts for killing, or allow resuming. 
+    # for instantiating remote it could also install a multiplexer 
+    # on the remote place such that the same ssh connection is re-used 
+    # for multiple ssh=codespeak.net connections, makes for small latency.
+
+some first simple implemenation might look like this::
+
+    gw = group.makegateway("socketservice//port=8888")
+    # we can rely that the subprocess gw can import execnet
+    ch = gw.remote_exec("""
+        import execnet.service
+        execnet.service.gatewaymaker(channel)
+    """)
+    ch.send(xspec)
+    subchannel = ch.receive()
+    return ChannelGateway(subchannel)

From commits-noreply at bitbucket.org  Mon Dec  7 13:34:59 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Mon,  7 Dec 2009 12:34:59 +0000 (UTC)
Subject: [execnet-commit] execnet commit e5cbfedbd84e: internal tweaks: move
	deprecated code to	its own module,
	add python2.7 to python interpreters to look for.
Message-ID: <20091207123459.57DA17EE85@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1260189273 -3600
# Node ID e5cbfedbd84e6dc2ee0f62f3b5ea70d865cd67ae
# Parent a15041d7bb5e008dc1a4a2976f03a79ca2085841
internal tweaks: move deprecated code to its own module, add python2.7 to python interpreters to look for.

--- a/execnet/multi.py
+++ b/execnet/multi.py
@@ -228,48 +228,6 @@ class MultiChannel:
         if first:
             reraise(*first)
 
-
-default_group = Group()
-
-makegateway = default_group.makegateway
-
-def PopenGateway(python=None):
-    """ instantiate a gateway to a subprocess
-        started with the given 'python' executable.
-    """
-    APIWARN("1.0.0b4", "use makegateway('popen')")
-    spec = execnet.XSpec("popen")
-    spec.python = python
-    return default_group.makegateway(spec)
-
-def SocketGateway(host, port):
-    """ This Gateway provides interaction with a remote process
-        by connecting to a specified socket.  On the remote
-        side you need to manually start a small script
-        (py/execnet/script/socketserver.py) that accepts
-        SocketGateway connections or use the experimental
-        new_remote() method on existing gateways.
-    """
-    APIWARN("1.0.0b4", "use makegateway('socket=host:port')")
-    spec = execnet.XSpec("socket=%s:%s" %(host, port))
-    return default_group.makegateway(spec)
-
-def SshGateway(sshaddress, remotepython=None, ssh_config=None):
-    """ instantiate a remote ssh process with the
-        given 'sshaddress' and remotepython version.
-        you may specify an ssh_config file.
-    """
-    APIWARN("1.0.0b4", "use makegateway('ssh=host')")
-    spec = execnet.XSpec("ssh=%s" % sshaddress)
-    spec.python = remotepython
-    spec.ssh_config = ssh_config
-    return default_group.makegateway(spec)
-
-def APIWARN(version, msg, stacklevel=3):
-    import warnings
-    Warn = DeprecationWarning("(since version %s) %s" %(version, msg))
-    warnings.warn(Warn, stacklevel=stacklevel)
-
 def killpopen(popen):
     try:
         if hasattr(popen, 'kill'):
@@ -300,3 +258,7 @@ def killpid(pid):
             ctypes.windll.kernel32.CloseHandle(handle)
     else:
         raise EnvironmmentError("no method to kill %s" %(pid,))
+
+default_group = Group()
+makegateway = default_group.makegateway
+

--- a/conftest.py
+++ b/conftest.py
@@ -8,6 +8,7 @@ collect_ignore = ['build', 'doc/_build']
 rsyncdirs = ['conftest.py', 'execnet', 'testing', 'doc']
 
 winpymap = {
+    'python2.7': r'C:\Python27\python.exe',
     'python2.6': r'C:\Python26\python.exe',
     'python2.5': r'C:\Python25\python.exe',
     'python2.4': r'C:\Python24\python.exe',
@@ -61,7 +62,7 @@ def pytest_generate_tests(metafunc):
             metafunc.addcall(id=gwtype, param=gwtype)
     elif 'anypython' in metafunc.funcargnames:
         for name in ('python3.1', 'python2.4', 'python2.5', 'python2.6', 
-                     'pypy-c', 'jython'):
+                     'python2.7', 'pypy-c', 'jython'):
             metafunc.addcall(id=name, param=name)
 
 def getexecutable(name, cache={}):

--- a/execnet/__init__.py
+++ b/execnet/__init__.py
@@ -9,9 +9,9 @@ __version__ = "1.0.1"
 import execnet.apipkg
 
 execnet.apipkg.initpkg(__name__, {
-    'PopenGateway':     '.multi:PopenGateway',
-    'SocketGateway':    '.multi:SocketGateway',
-    'SshGateway':       '.multi:SshGateway',
+    'PopenGateway':     '.deprecated:PopenGateway',
+    'SocketGateway':    '.deprecated:SocketGateway',
+    'SshGateway':       '.deprecated:SshGateway',
     'makegateway':      '.multi:makegateway',
     'HostNotFound':     '.gateway:HostNotFound',
     'XSpec':            '.xspec:XSpec',

--- /dev/null
+++ b/execnet/deprecated.py
@@ -0,0 +1,43 @@
+"""
+some deprecated calls
+
+(c) 2008-2009, Holger Krekel and others
+"""
+import execnet
+
+def PopenGateway(python=None):
+    """ instantiate a gateway to a subprocess
+        started with the given 'python' executable.
+    """
+    APIWARN("1.0.0b4", "use makegateway('popen')")
+    spec = execnet.XSpec("popen")
+    spec.python = python
+    return execnet.default_group.makegateway(spec)
+
+def SocketGateway(host, port):
+    """ This Gateway provides interaction with a remote process
+        by connecting to a specified socket.  On the remote
+        side you need to manually start a small script
+        (py/execnet/script/socketserver.py) that accepts
+        SocketGateway connections or use the experimental
+        new_remote() method on existing gateways.
+    """
+    APIWARN("1.0.0b4", "use makegateway('socket=host:port')")
+    spec = execnet.XSpec("socket=%s:%s" %(host, port))
+    return execnet.default_group.makegateway(spec)
+
+def SshGateway(sshaddress, remotepython=None, ssh_config=None):
+    """ instantiate a remote ssh process with the
+        given 'sshaddress' and remotepython version.
+        you may specify an ssh_config file.
+    """
+    APIWARN("1.0.0b4", "use makegateway('ssh=host')")
+    spec = execnet.XSpec("ssh=%s" % sshaddress)
+    spec.python = remotepython
+    spec.ssh_config = ssh_config
+    return execnet.default_group.makegateway(spec)
+
+def APIWARN(version, msg, stacklevel=3):
+    import warnings
+    Warn = DeprecationWarning("(since version %s) %s" %(version, msg))
+    warnings.warn(Warn, stacklevel=stacklevel)

From commits-noreply at bitbucket.org  Fri Dec 11 15:10:29 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Fri, 11 Dec 2009 14:10:29 +0000 (UTC)
Subject: [execnet-commit] execnet commit 3d00bf8eaea4: move
	socket-instantiation into own file,
	speeds up popen-gateway creation by another 10%.
Message-ID: <20091211141029.DA64D7EF3B@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1260540574 -3600
# Node ID 3d00bf8eaea4e48e9cc621127e936d50f59727cb
# Parent e5cbfedbd84e6dc2ee0f62f3b5ea70d865cd67ae
move socket-instantiation into own file, speeds up popen-gateway creation by another 10%.

--- a/CHANGELOG
+++ b/CHANGELOG
@@ -1,3 +1,9 @@
+1.x (pending)
+--------------------------------
+
+- internally split socket gateways, speeds up popen-gateways
+  by 10% (now at 30 milliseconds per-gateway on my 3-year old machine) 
+
 1.0.1
 --------------------------------
 

--- a/execnet/gateway.py
+++ b/execnet/gateway.py
@@ -6,7 +6,7 @@ gateway code for initiating popen, socke
 import sys, os, inspect, socket, types
 import textwrap
 import execnet
-from execnet.gateway_base import Message, Popen2IO, SocketIO
+from execnet.gateway_base import Message, Popen2IO
 from execnet import gateway_base
 importdir = os.path.dirname(os.path.dirname(execnet.__file__))
 
@@ -174,59 +174,6 @@ def sendexec(io, *sources):
     source = "\n".join(sources)
     io.write((repr(source)+ "\n").encode('ascii'))
 
-class SocketGateway(Gateway):
-    """ This Gateway provides interaction with a remote process
-        by connecting to a specified socket.  On the remote
-        side you need to manually start a small script
-        (py/execnet/script/socketserver.py) that accepts
-        SocketGateway connections.
-    """
-    _remotesetup = "io = SocketIO(clientsock)" 
-
-    def __init__(self, host, port, id):
-        """ instantiate a gateway to a process accessed
-            via a host/port specified socket.
-        """
-        self.host = host = str(host)
-        self.port = port = int(port)
-        self.remoteaddress = '%s:%d' % (self.host, self.port)
-        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        try:
-            sock.connect((host, port))
-        except socket.gaierror:
-            raise HostNotFound(str(sys.exc_info()[1]))
-        io = SocketIO(sock)
-        super(SocketGateway, self).__init__(io=io, id=id)
-
-    def new_remote(cls, gateway, id, hostport=None):
-        """ return a new (connected) socket gateway, 
-            instantiated through the given 'gateway'.
-        """
-        if hostport is None:
-            host, port = ('localhost', 0)
-        else:
-            host, port = hostport
-        
-        mydir = os.path.dirname(__file__)
-        socketserver = os.path.join(mydir, 'script', 'socketserver.py')
-        socketserverbootstrap = "\n".join([
-            open(socketserver, 'r').read(), """if 1:
-            import socket
-            sock = bind_and_listen((%r, %r))
-            port = sock.getsockname()
-            channel.send(port)
-            startserver(sock)
-        """ % (host, port)])
-        # execute the above socketserverbootstrap on the other side
-        channel = gateway.remote_exec(socketserverbootstrap)
-        (realhost, realport) = channel.receive()
-        #self._trace("new_remote received"
-        #               "port=%r, hostname = %r" %(realport, hostname))
-        if not realhost or realhost=="0.0.0.0":
-            realhost = "localhost"
-        return cls(realhost, realport, id=id)
-    new_remote = classmethod(new_remote)
-
 class HostNotFound(Exception):
     pass
 

--- /dev/null
+++ b/execnet/gateway_socket.py
@@ -0,0 +1,92 @@
+import socket
+from execnet.gateway import Gateway, HostNotFound
+import os, sys, inspect
+
+class SocketIO:
+    error = (socket.error, EOFError)
+    def __init__(self, sock):
+        self.sock = sock
+        try:
+            sock.setsockopt(socket.SOL_IP, socket.IP_TOS, 0x10)# IPTOS_LOWDELAY
+            sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
+        except (AttributeError, socket.error):
+            e = sys.exc_info()[1]
+            sys.stderr.write("WARNING: cannot set socketoption")
+
+    def read(self, numbytes):
+        "Read exactly 'bytes' bytes from the socket."
+        buf = bytes()
+        while len(buf) < numbytes:
+            t = self.sock.recv(numbytes - len(buf))
+            if not t:
+                raise EOFError
+            buf += t
+        return buf
+
+    def write(self, data):
+        assert isinstance(data, bytes)
+        self.sock.sendall(data)
+
+    def close_read(self):
+        try:
+            self.sock.shutdown(0)
+        except socket.error:
+            pass
+    def close_write(self):
+        try:
+            self.sock.shutdown(1)
+        except socket.error:
+            pass
+
+class SocketGateway(Gateway):
+    """ This Gateway provides interaction with a remote process
+        by connecting to a specified socket.  On the remote
+        side you need to manually start a small script
+        (py/execnet/script/socketserver.py) that accepts
+        SocketGateway connections.
+    """
+    _remotesetup = "import socket\n%s\nio = SocketIO(clientsock)" % inspect.getsource(SocketIO)
+
+    def __init__(self, host, port, id):
+        """ instantiate a gateway to a process accessed
+            via a host/port specified socket.
+        """
+        self.host = host = str(host)
+        self.port = port = int(port)
+        self.remoteaddress = '%s:%d' % (self.host, self.port)
+        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        try:
+            sock.connect((host, port))
+        except socket.gaierror:
+            raise HostNotFound(str(sys.exc_info()[1]))
+        io = SocketIO(sock)
+        super(SocketGateway, self).__init__(io=io, id=id)
+
+    def new_remote(cls, gateway, id, hostport=None):
+        """ return a new (connected) socket gateway, 
+            instantiated through the given 'gateway'.
+        """
+        if hostport is None:
+            host, port = ('localhost', 0)
+        else:
+            host, port = hostport
+        
+        mydir = os.path.dirname(__file__)
+        socketserver = os.path.join(mydir, 'script', 'socketserver.py')
+        socketserverbootstrap = "\n".join([
+            open(socketserver, 'r').read(), """if 1:
+            import socket
+            sock = bind_and_listen((%r, %r))
+            port = sock.getsockname()
+            channel.send(port)
+            startserver(sock)
+        """ % (host, port)])
+        # execute the above socketserverbootstrap on the other side
+        channel = gateway.remote_exec(socketserverbootstrap)
+        (realhost, realport) = channel.receive()
+        #self._trace("new_remote received"
+        #               "port=%r, hostname = %r" %(realport, hostname))
+        if not realhost or realhost=="0.0.0.0":
+            realhost = "localhost"
+        return cls(realhost, realport, id=id)
+    new_remote = classmethod(new_remote)

--- a/execnet/gateway_base.py
+++ b/execnet/gateway_base.py
@@ -6,7 +6,7 @@ NOTE: aims to be compatible to Python 2.
 (C) 2004-2009 Holger Krekel, Armin Rigo, Benjamin Peterson, and others
 """
 import sys, os, weakref
-import threading, traceback, socket, struct
+import threading, traceback, struct
 try:
     import queue
 except ImportError:
@@ -56,47 +56,6 @@ else:
         pass
 
 
-# ___________________________________________________________________________
-#
-# input output classes
-# ___________________________________________________________________________
-
-class SocketIO:
-    error = (socket.error, EOFError)
-    def __init__(self, sock):
-        self.sock = sock
-        try:
-            sock.setsockopt(socket.SOL_IP, socket.IP_TOS, 0x10)# IPTOS_LOWDELAY
-            sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
-        except (AttributeError, socket.error):
-            e = sys.exc_info()[1]
-            sys.stderr.write("WARNING: cannot set socketoption")
-
-    def read(self, numbytes):
-        "Read exactly 'bytes' bytes from the socket."
-        buf = bytes()
-        while len(buf) < numbytes:
-            t = self.sock.recv(numbytes - len(buf))
-            if not t:
-                raise EOFError
-            buf += t
-        return buf
-
-    def write(self, data):
-        assert isinstance(data, bytes)
-        self.sock.sendall(data)
-
-    def close_read(self):
-        try:
-            self.sock.shutdown(0)
-        except socket.error:
-            pass
-    def close_write(self):
-        try:
-            self.sock.shutdown(1)
-        except socket.error:
-            pass
-
 class Popen2IO:
     error = (IOError, OSError, EOFError)
 

--- a/execnet/__init__.py
+++ b/execnet/__init__.py
@@ -4,7 +4,7 @@ package for connecting to local and remo
 
 (c) 2009, Holger Krekel and others
 """
-__version__ = "1.0.1"
+__version__ = "1.0.1.post1"
 
 import execnet.apipkg
 

--- a/execnet/multi.py
+++ b/execnet/multi.py
@@ -81,13 +81,14 @@ class Group:
         elif spec.socket:
             assert not spec.python, (
                 "socket: specifying python executables not yet supported")
+            from execnet.gateway_socket import SocketGateway
             gateway_id = spec.installvia
             if gateway_id:
                 viagw = self[gateway_id]
-                gw = gateway.SocketGateway.new_remote(viagw, id=id)
+                gw = SocketGateway.new_remote(viagw, id=id)
             else:
                 host, port = spec.socket.split(":")
-                gw = gateway.SocketGateway(host, port, id=id)
+                gw = SocketGateway(host, port, id=id)
         else:
             raise ValueError("no gateway type found for %r" % (spec._spec,))
         gw.spec = spec

--- a/testing/test_xspec.py
+++ b/testing/test_xspec.py
@@ -65,7 +65,7 @@ class TestMakegateway:
         assert gw.spec.popen 
         assert gw.spec.python == None
         rinfo = gw._rinfo()
-        assert rinfo.executable == py.std.sys.executable
+        #assert rinfo.executable == py.std.sys.executable
         assert rinfo.cwd == py.std.os.getcwd()
         assert rinfo.version_info == py.std.sys.version_info
 
@@ -113,7 +113,7 @@ class TestMakegateway:
             py.test.skip("cpython2.6 not found")
         gw = execnet.makegateway("popen//python=%s" % cpython26)
         rinfo = gw._rinfo()
-        assert rinfo.executable == cpython26
+        #assert rinfo.executable == cpython26
         assert rinfo.cwd == py.std.os.getcwd()
         assert rinfo.version_info[:2] == (2,6)

From commits-noreply at bitbucket.org  Sun Dec 20 17:10:08 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun, 20 Dec 2009 16:10:08 +0000 (UTC)
Subject: [execnet-commit] execnet commit 9068ad88e86a: adding issues,
	fixing docs, fixating release-version
Message-ID: <20091220161008.0EC9C7EE7A@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261314055 -3600
# Node ID 9068ad88e86a345d49d29a36fa3d20b28f18cac6
# Parent 3d00bf8eaea4e48e9cc621127e936d50f59727cb
adding issues, fixing docs, fixating release-version

--- a/ISSUES.txt
+++ b/ISSUES.txt
@@ -78,3 +78,36 @@ some first simple implemenation might lo
     ch.send(xspec)
     subchannel = ch.receive()
     return ChannelGateway(subchannel)
+
+serialize channels everywhere 
+-------------------------------------------------
+tags: 1.0.2 feature
+
+currently sending channels only works if they are send 
+as a single item.  So they cannot be contained in a list or dict. 
+By making channel passing a feature of the serializer we should
+be able to lift this limitation, allowing channels in all places. 
+
+creating new channels 
+-------------------------------------------------
+tags: 1.0.2 feature
+
+document, provide examples and refine support for creating new channels. 
+Defining closing semantics - does a channel created from a channel 
+automatically close when its parent channel closes? This currently 
+depends on whether it is garbage collected. 
+
+creating virtualenv environments on the fly 
+-----------------------------------------------
+tags: 1.1 wish
+
+there should be a (plugin-provided?) way to create
+a virtual environment: 
+    gw = makegateway("ssh=codespeak.net")
+    newgw = xdeploy.create_venv_gateway(gw, "venvname", depstring)
+    newgw # lives in home/ dir of remote virtualenv 
+          # maybe have get_temproot default to tmp/ there? 
+    sf = xdeploy.SetupFile("..setup.py")
+    sf.sdist_remote_install(newgw)
+    
+    

--- a/doc/conf.py
+++ b/doc/conf.py
@@ -46,9 +46,9 @@ copyright = '2009, holger krekel and oth
 #
 # The short X.Y version.
 import execnet
-version = execnet.__version__
+version = "1.0.1"
 # The full version, including alpha/beta/rc tags.
-release = execnet.__version__
+release = version
 
 # The language for content autogenerated by Sphinx. Refer to documentation
 # for a list of supported languages.

--- a/doc/example/hybridpython.txt
+++ b/doc/example/hybridpython.txt
@@ -90,6 +90,6 @@ using Mono 2.0 and IronPython-1.1 this w
    Using IronPython needs more testing, likely newer versions
    will work better.  please feedback if you have information. 
 
-.. _IronPython: http://www.IronPython.org
+.. _IronPython: http://ironpython.net
 
 

--- a/doc/index.txt
+++ b/doc/index.txt
@@ -41,7 +41,8 @@ Known uses
 * `py.test`_ uses it for its `distributed testing`_ mechanism. 
 
 * Jacob Perkins uses it for his `Distributed NTLK with execnet`_ 
-  project to launch computation processes through ssh. 
+  project to launch computation processes through ssh.  He also
+  compares `disco and execnet`_ in a subsequent post. 
 
 * Ronny Pfannschmidt uses it for his `anyvc`_ VCS-abstraction project
   to bridge the Python2/Python3 version gap. 
@@ -51,6 +52,7 @@ Known uses
 .. _`py.test`: http://pytest.org
 .. _`distributed testing`: http://codespeak.net/py/dist/test/dist.html
 .. _`Distributed NTLK with execnet`: http://streamhacker.com/2009/11/29/distributed-nltk-execnet/
+.. _`disco and execnet`: http://streamhacker.com/2009/12/14/execnet-disco-distributed-nltk/
 .. _`anyvc`: http://bitbucket.org/RonnyPfannschmidt/anyvc/
 
 Project status

From commits-noreply at bitbucket.org  Sun Dec 20 17:10:09 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun, 20 Dec 2009 16:10:09 +0000 (UTC)
Subject: [execnet-commit] execnet commit 8a4fe45e6508: avoid unneccessary
	lower()
Message-ID: <20091220161009.D07437EE8A@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261314724 -3600
# Node ID 8a4fe45e650874536a521f736ada5b93a84b2614
# Parent 9068ad88e86a345d49d29a36fa3d20b28f18cac6
avoid unneccessary lower()

--- a/execnet/gateway_base.py
+++ b/execnet/gateway_base.py
@@ -896,14 +896,14 @@ class Serializer(object):
         try:
             dispatch = self.dispatch[tp]
         except KeyError:
-            methodname = 'save_' + tp.__name__.lower()
+            methodname = 'save_' + tp.__name__
             meth = getattr(self, methodname, None)
             if meth is None:
                 raise SerializationError("can't serialize %s" % (tp,))
             dispatch = self.dispatch[tp] = meth
         dispatch(obj)
 
-    def save_nonetype(self, non):
+    def save_NoneType(self, non):
         self._write(opcode.NONE)
 
     def save_bool(self, boolean):

From commits-noreply at bitbucket.org  Sun Dec 20 17:10:12 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun, 20 Dec 2009 16:10:12 +0000 (UTC)
Subject: [execnet-commit] execnet commit f9a82b8fb819: generalize
	channel-over-channel sending, add examples.
Message-ID: <20091220161012.06BD37EEE2@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261323576 -3600
# Node ID f9a82b8fb819440a42ffca843ff065e001d29fba
# Parent 8a4fe45e650874536a521f736ada5b93a84b2614
generalize channel-over-channel sending, add examples.

--- a/doc/example/test_info.txt
+++ b/doc/example/test_info.txt
@@ -1,10 +1,23 @@
 basic local and remote communication
 =========================================
 
-Comparing current working directories
+Execute code in subprocess, communicate through a channel
+---------------------------------------------------------
+
+You can instantiate a subprocess gateway, execute code 
+in it and exchange data dynamically::
+
+    >>> import execnet
+    >>> gw = execnet.makegateway() 
+    >>> channel = gw.remote_exec("channel.send(channel.receive()+1)")
+    >>> channel.send(1)
+    >>> channel.receive()
+    2
+
+Compare current working directories
 ----------------------------------------
 
-A local popen gateway has the same working directory as the instantiatior::
+A local subprocess gateway has the same working directory as the instantiatior::
 
     >>> import execnet, os
     >>> gw = execnet.makegateway()   
@@ -14,7 +27,7 @@ A local popen gateway has the same worki
 
 "ssh" gateways default to the login home directory.
 
-Getting information from remote ssh account
+Get information from remote ssh account
 --------------------------------------------
 
 Use simple execution to obtain information from remote environments::
@@ -31,6 +44,43 @@ Use simple execution to obtain informati
   >>> version_info
   (2, 4, 2, 'final', 0)
 
+Use a callback instead of receive() and wait for completion 
+-------------------------------------------------------------
+
+Set a channel callback to immediately react on incoming data::
+
+    >>> import execnet
+    >>> gw = execnet.makegateway() 
+    >>> channel = gw.remote_exec("for i in range(10): channel.send(i)")
+    >>> l = []
+    >>> channel.setcallback(l.append, endmarker=None)
+    >>> channel.waitclose() # waits for closing, i.e. remote exec finish
+    >>> l
+    [0, 1, 2, 3, 4, 5, 6, 7, 8, 9, None]
+
+Note that the callback function will execute in the receiver thread
+so it should not block on IO or long to execute. 
+
+Sending channels over channels 
+------------------------------------------------------
+
+You can create and transfer a channel over an existing channel
+and use it to transfer information::
+
+    >>> import execnet
+    >>> gw = execnet.makegateway()
+    >>> channel = gw.remote_exec("""
+    ...        ch1, ch2 = channel.receive()
+    ...        ch2.send("world") 
+    ...        ch1.send("hello")
+    ... """)
+    >>> c1 = gw.newchannel()    # create new channel
+    >>> c2 = gw.newchannel()    # create another channel
+    >>> channel.send((c1, c2))  # send them over
+    >>> c1.receive()
+    'hello'
+    >>> c2.receive()
+    'world'
 
 Avoid "inlined" source strings with remote_exec
 --------------------------------------------------------------

--- a/execnet/gateway_base.py
+++ b/execnet/gateway_base.py
@@ -141,13 +141,6 @@ def _setupmessages():
             channel = gateway._channelfactory.new(self.channelid)
             gateway._local_schedulexec(channel=channel, sourcetask=self.data)
 
-    class CHANNEL_NEW(Message):
-        def received(self, gateway):
-            """ receive a remotely created new (sub)channel. """
-            newid = self.data
-            newchannel = gateway._channelfactory.new(newid)
-            gateway._channelfactory._local_receive(self.channelid, newchannel)
-
     class CHANNEL_DATA(Message):
         def received(self, gateway):
             gateway._channelfactory._local_receive(self.channelid, self.data)
@@ -172,7 +165,7 @@ def _setupmessages():
             raise SystemExit(0)
 
     classes = [
-        STATUS, CHANNEL_OPEN, CHANNEL_NEW, CHANNEL_DATA, CHANNEL_CLOSE, 
+        STATUS, CHANNEL_OPEN, CHANNEL_DATA, CHANNEL_CLOSE, 
         CHANNEL_CLOSE_ERROR, CHANNEL_LAST_MESSAGE, GATEWAY_TERMINATE
     ]
     for i, cls in enumerate(classes):
@@ -385,10 +378,7 @@ class Channel(object):
         """
         if self.isclosed():
             raise IOError("cannot send to %r" %(self,))
-        if isinstance(item, Channel):
-            data = Message.CHANNEL_NEW(self.id, item.id)
-        else:
-            data = Message.CHANNEL_DATA(self.id, item)
+        data = Message.CHANNEL_DATA(self.id, item)
         self.gateway._send(data)
 
     def receive(self, timeout=-1):
@@ -455,8 +445,10 @@ class ChannelFactory(object):
             if id is None:
                 id = self.count
                 self.count += 2
-            channel = Channel(self.gateway, id)
-            self._channels[id] = channel
+            try:
+                channel = self._channels[id] 
+            except KeyError:
+                channel = self._channels[id] = Channel(self.gateway, id)
             return channel
         finally:
             self._writelock.release()
@@ -597,7 +589,7 @@ class BaseGateway(object):
 
     def _thread_receiver(self):
         self._trace("starting to receive")
-        unserializer = Unserializer(self._io)
+        unserializer = Unserializer(self._io, self._channelfactory)
         try:
             while 1:
                 try:
@@ -607,6 +599,7 @@ class BaseGateway(object):
                     _receivelock.acquire()
                     try:
                         msg.received(self)
+                        del msg
                     finally:
                         _receivelock.release()
                 except sysex:
@@ -639,6 +632,7 @@ class BaseGateway(object):
     # _____________________________________________________________________
     #
     def newchannel(self):
+        """ return a new independent channel. """
         return self._channelfactory.new()
 
     def join(self, timeout=None):
@@ -729,8 +723,9 @@ class Unserializer(object):
     py2str_as_py3str = True # True
     py3str_as_py2str = False  # false means py2 will get unicode
 
-    def __init__(self, stream):
+    def __init__(self, stream, channelfactory=None):
         self.stream = stream
+        self.channelfactory = channelfactory
 
     def load(self):
         self.stack = []
@@ -748,7 +743,7 @@ class Unserializer(object):
         except _Stop:
             if len(self.stack) != 1:
                 raise UnserializationError("internal unserialization error")
-            return self.stack[0]
+            return self.stack.pop(0)
         else:
             raise UnserializationError("didn't get STOP")
 
@@ -851,6 +846,11 @@ class Unserializer(object):
     def load_stop(self):
         raise _Stop
 
+    def load_channel(self):
+        id = self._read_int4()
+        newchannel = self.channelfactory.new(id)
+        self.stack.append(newchannel)
+
 # automatically build opcodes and byte-encoding 
 
 class opcode:
@@ -998,6 +998,10 @@ class Serializer(object):
     def save_frozenset(self, s):
         self._write_set(s, opcode.FROZENSET)
 
+    def save_Channel(self, channel):
+        self._write(opcode.CHANNEL)
+        self._write_int4(channel.id)
+
 def init_popen_io():
     if not hasattr(os, 'dup'): # jython
         io = Popen2IO(sys.stdout, sys.stdin)

--- a/doc/basics.txt
+++ b/doc/basics.txt
@@ -61,6 +61,7 @@ get sent for remote execution.  ``remote
 a channel object whose symmetric counterpart channel 
 is available to the remotely executing source. 
 
+
 .. _`Channel`: 
 .. _`channel-api`: 
 
@@ -83,6 +84,7 @@ two asynchronously running programs.
    .. autoattribute:: Channel.RemoteError
    .. autoattribute:: Channel.TimeoutError
 
+
 .. _Group:
 
 Grouped Gateways and robust termination

--- a/testing/test_channel.py
+++ b/testing/test_channel.py
@@ -88,6 +88,30 @@ class TestChannelBasicBehaviour:
         l = list(channel)
         assert l == [0, 1, 2]
 
+    def test_channel_pass_in_structure(self, gw):
+        channel = gw.remote_exec('''
+            ch1, ch2 = channel.receive()
+            data = ch1.receive()
+            ch2.send(data+1)
+        ''')
+        newchan1 = gw.newchannel()
+        newchan2 = gw.newchannel()
+        channel.send((newchan1, newchan2))
+        newchan1.send(1)
+        data = newchan2.receive()
+        assert data ==2
+
+    def test_channel_multipass(self, gw):
+        channel = gw.remote_exec('''
+            channel.send(channel)
+            xchan = channel.receive()
+            assert xchan == channel
+        ''')
+        newchan = channel.receive()
+        assert newchan == channel
+        channel.send(newchan)
+        channel.waitclose()
+
     def test_channel_passing_over_channel(self, gw):
         channel = gw.remote_exec('''
                     c = channel.gateway.newchannel()

--- a/CHANGELOG
+++ b/CHANGELOG
@@ -1,6 +1,10 @@
 1.x (pending)
 --------------------------------
 
+- generalize channel-over-channel sending: you can now have channels 
+  anywhere in a data structure (i.e. as an item of a container type).
+  Add according examples.
+
 - internally split socket gateways, speeds up popen-gateways
   by 10% (now at 30 milliseconds per-gateway on my 3-year old machine)

From commits-noreply at bitbucket.org  Sun Dec 20 17:10:14 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun, 20 Dec 2009 16:10:14 +0000 (UTC)
Subject: [execnet-commit] execnet commit 1d9442698ab8: rename CHANNEL_OPEN
	to CHANNEL_EXEC and	add test_debug.txt to versioning
Message-ID: <20091220161014.621667EEEF@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261324797 -3600
# Node ID 1d9442698ab81618293b3436d304d4808b528aaf
# Parent f9a82b8fb819440a42ffca843ff065e001d29fba
rename CHANNEL_OPEN to CHANNEL_EXEC and add test_debug.txt to versioning

--- /dev/null
+++ b/doc/example/test_debug.txt
@@ -0,0 +1,43 @@
+
+Debugging execnet / Wire messages 
+===============================================================
+
+By setting the environment variable ``EXECNET_DEBUG`` you can
+configure the execnet tracing mechanism: 
+
+:EXECNET_DEBUG=1:  write per-process trace-files to ``${TEMPROOT}/execnet-debug-PID`` 
+:EXECNET_DEBUG=2:  perform tracing to stderr (popen-gateway slaves will send this to their instantiator)
+
+Here is a simple example to see what goes on with a simple execution::
+
+    EXECNET_DEBUG=2  # or "set EXECNET_DEBUG=2" on windows 
+
+    python -c 'import execnet ; execnet.makegateway().remote_exec("42")'
+
+which will show PID-prefixed trace entries::
+
+    [2326] gw0 starting to receive
+    [2326] gw0 sent 
+    [2327] creating slavegateway on 
+    [2327] gw0-slave starting to receive
+    [2327] gw0-slave received 
+    [2327] gw0-slave execution starts[1]: '42'
+    [2327] gw0-slave execution finished
+    [2327] gw0-slave sent 
+    [2327] gw0-slave 1 sent channel close message
+    [2326] gw0 received 
+    [2326] gw0 1 channel.__del__
+    [2326] === atexit cleanup  ===
+    [2326] gw0 gateway.exit() called
+    [2326] gw0 --> sending GATEWAY_TERMINATE
+    [2326] gw0 sent 
+    [2326] gw0 joining receiver thread
+    [2327] gw0-slave received 
+    [2327] gw0-slave putting None to execqueue
+    [2327] gw0-slave io.close_read()
+    [2327] gw0-slave leaving 
+    [2327] gw0-slave 1 channel.__del__
+    [2327] gw0-slave io.close_write()
+    [2327] gw0-slave slavegateway.serve finished
+    [2327] gw0-slave gateway.join() called while receiverthread already finished
+    [2326] gw0 leaving 

--- a/execnet/gateway.py
+++ b/execnet/gateway.py
@@ -93,7 +93,7 @@ class Gateway(gateway_base.BaseGateway):
         else:
             source = textwrap.dedent(str(source))
         channel = self.newchannel()
-        self._send(Message.CHANNEL_OPEN(channel.id, source))
+        self._send(Message.CHANNEL_EXEC(channel.id, source))
         return channel
 
     def remote_init_threads(self, num=None):

--- a/execnet/gateway_base.py
+++ b/execnet/gateway_base.py
@@ -136,7 +136,7 @@ def _setupmessages():
             }
             gateway._send(Message.CHANNEL_DATA(self.channelid, d))
 
-    class CHANNEL_OPEN(Message):
+    class CHANNEL_EXEC(Message):
         def received(self, gateway):
             channel = gateway._channelfactory.new(self.channelid)
             gateway._local_schedulexec(channel=channel, sourcetask=self.data)
@@ -165,7 +165,7 @@ def _setupmessages():
             raise SystemExit(0)
 
     classes = [
-        STATUS, CHANNEL_OPEN, CHANNEL_DATA, CHANNEL_CLOSE, 
+        STATUS, CHANNEL_EXEC, CHANNEL_DATA, CHANNEL_CLOSE, 
         CHANNEL_CLOSE_ERROR, CHANNEL_LAST_MESSAGE, GATEWAY_TERMINATE
     ]
     for i, cls in enumerate(classes):

From commits-noreply at bitbucket.org  Sun Dec 20 17:15:08 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun, 20 Dec 2009 16:15:08 +0000 (UTC)
Subject: [execnet-commit] execnet commit bbd403fddf54: remove the just
	committed issues
Message-ID: <20091220161508.D32C37EEDE@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261325694 -3600
# Node ID bbd403fddf5418af115c8ece205ce688b69052e0
# Parent 1d9442698ab81618293b3436d304d4808b528aaf
remove the just committed issues

--- a/ISSUES.txt
+++ b/ISSUES.txt
@@ -79,24 +79,6 @@ some first simple implemenation might lo
     subchannel = ch.receive()
     return ChannelGateway(subchannel)
 
-serialize channels everywhere 
--------------------------------------------------
-tags: 1.0.2 feature
-
-currently sending channels only works if they are send 
-as a single item.  So they cannot be contained in a list or dict. 
-By making channel passing a feature of the serializer we should
-be able to lift this limitation, allowing channels in all places. 
-
-creating new channels 
--------------------------------------------------
-tags: 1.0.2 feature
-
-document, provide examples and refine support for creating new channels. 
-Defining closing semantics - does a channel created from a channel 
-automatically close when its parent channel closes? This currently 
-depends on whether it is garbage collected. 
-
 creating virtualenv environments on the fly 
 -----------------------------------------------
 tags: 1.1 wish

From commits-noreply at bitbucket.org  Sun Dec 20 19:24:57 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun, 20 Dec 2009 18:24:57 +0000 (UTC)
Subject: [execnet-commit] execnet commit 7c99c22b2bbb: better handle errors
	in callbacks, closing the channel and leaving
Message-ID: <20091220182457.C88097EEE2@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261332944 -3600
# Node ID 7c99c22b2bbb4b217641fff9122588bf8d41ed6e
# Parent bbd403fddf5418af115c8ece205ce688b69052e0
better handle errors in callbacks, closing the channel and leaving
other execution/communication useable

--- a/ISSUES.txt
+++ b/ISSUES.txt
@@ -1,15 +1,3 @@
-report back errors when calling callbacks 
----------------------------------------------------
-
-tags: 1.0 
-bb: http://bitbucket.org/hpk42/py-trunk/issue/37/
-path: execnet/gateway_base.py
-
-Callbacks are invoked in the receiver-thread
-and any exception will kill the whole receiver-thread. 
-This is especially bad if it happens remotely as 
-it leaves the other side inaccesssible. 
-
 rsync links from posix to windows 
 -----------------------------------------
 tags: 1.1 feature

--- a/execnet/gateway_base.py
+++ b/execnet/gateway_base.py
@@ -502,7 +502,16 @@ class ChannelFactory(object):
             else:
                 queue.put(data)
         else:
-            callback(data)   # even if channel may be already closed
+            try:
+                callback(data)   # even if channel may be already closed
+            except KeyboardInterrupt:
+                raise
+            except:
+                excinfo = sys.exc_info()
+                self.gateway._trace("exception during callback: %s" % excinfo[1])
+                errortext = geterrortext(excinfo)
+                self.gateway._send(Message.CHANNEL_CLOSE_ERROR(id, errortext))
+                self._local_close(id, errortext)
 
     def _finished_receiving(self):
         self._writelock.acquire()

--- a/testing/test_channel.py
+++ b/testing/test_channel.py
@@ -244,6 +244,24 @@ class TestChannelBasicBehaviour:
         assert err
         assert str(err).find("ValueError") != -1
 
+    def test_channel_callback_error(self, gw):
+        channel = gw.remote_exec("""
+            def f(item):
+                raise ValueError(42)
+            ch = channel.gateway.newchannel()
+            ch.setcallback(f)
+            channel.send(ch)
+            channel.receive()
+            assert ch.isclosed()
+        """)
+        subchan = channel.receive()
+        subchan.send(1)
+        excinfo = py.test.raises(subchan.RemoteError, 
+            "subchan.waitclose(TESTTIMEOUT)")
+        assert "42" in excinfo.value.formatted
+        channel.send(1)
+        channel.waitclose()
+
 class TestChannelFile:
     def test_channel_file_write(self, gw):
         channel = gw.remote_exec("""

--- a/CHANGELOG
+++ b/CHANGELOG
@@ -5,6 +5,11 @@ 1.x (pending)
   anywhere in a data structure (i.e. as an item of a container type).
   Add according examples.
 
+- automatically close a channel when a remote callback raises 
+  an exception, makes communication more robust because until 
+  now failing callbacks rendered the receiverthread unuseable 
+  leaving the remote side in-accessible.
+
 - internally split socket gateways, speeds up popen-gateways
   by 10% (now at 30 milliseconds per-gateway on my 3-year old machine)

From commits-noreply at bitbucket.org  Sun Dec 20 19:40:30 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun, 20 Dec 2009 18:40:30 +0000 (UTC)
Subject: [execnet-commit] execnet commit 755fc3032507: fixing some python3
	issues, everything passes
Message-ID: <20091220184030.1E5777EEE2@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261334418 -3600
# Node ID 755fc30325075889d3abc2ea9cf7b4d324bad848
# Parent 7c99c22b2bbb4b217641fff9122588bf8d41ed6e
fixing some python3 issues, everything passes

--- a/testing/test_basics.py
+++ b/testing/test_basics.py
@@ -11,11 +11,11 @@ def test_subprocess_interaction(anypytho
     popen = subprocess.Popen(args, bufsize=0, universal_newlines=True,
                              stdin=subprocess.PIPE, stdout=subprocess.PIPE)
     def send(line):
-        popen.stdin.write(line.encode('ascii'))
+        popen.stdin.write(line)
         if sys.version_info > (3,0) or sys.platform.startswith("java"):
             popen.stdin.flush()
     def receive():
-        return popen.stdout.readline().decode('ascii')
+        return popen.stdout.readline()
 
     try:
         source = py.code.Source(read_write_loop, "read_write_loop()")

--- a/testing/test_multi.py
+++ b/testing/test_multi.py
@@ -121,9 +121,9 @@ class TestGroup:
         assert len(gwlist) == 3
         idlist = [x.id for x in gwlist]
         assert idlist == list('325')
-        print group
+        print (group)
         group.terminate()
-        print group
+        print (group)
         assert not group
         assert repr(group) == ""
 

--- a/conftest.py
+++ b/conftest.py
@@ -73,7 +73,7 @@ def getexecutable(name, cache={}):
         if executable:
             if name == "jython":
                 popen = subprocess.Popen([str(executable), "--version"], 
-                    stderr=subprocess.PIPE)
+                    universal_newlines=True, stderr=subprocess.PIPE)
                 out, err = popen.communicate()
                 if not err or "2.5" not in err:
                     executable = None

From commits-noreply at bitbucket.org  Sun Dec 20 22:43:05 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Sun, 20 Dec 2009 21:43:05 +0000 (UTC)
Subject: [execnet-commit] execnet commit 6007ee61a482: bumping version to
	1.0.2, fixing bits, removing unused imports
Message-ID: <20091220214305.05EAE7EE8A@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261345365 -3600
# Node ID 6007ee61a482f19d765917da3d75a25ff2049793
# Parent 755fc30325075889d3abc2ea9cf7b4d324bad848
bumping version to 1.0.2, fixing bits, removing unused imports

--- a/CHANGELOG
+++ b/CHANGELOG
@@ -1,4 +1,4 @@
-1.x (pending)
+1.0.2 
 --------------------------------
 
 - generalize channel-over-channel sending: you can now have channels 
@@ -11,7 +11,7 @@ 1.x (pending)
   leaving the remote side in-accessible.
 
 - internally split socket gateways, speeds up popen-gateways
-  by 10% (now at 30 milliseconds per-gateway on my 3-year old machine) 
+  by 10% (now at <50 milliseconds per-gateway on a 1.5 GHZ machine) 
 
 1.0.1
 --------------------------------

--- a/doc/index.txt
+++ b/doc/index.txt
@@ -6,8 +6,8 @@
 Python_ is a mature dynamic language whose interpreters can interact with 
 all major computing platforms today.  
 
-**execnet** provides carefully tested means to ad-hoc interact with Python
-interpreters across version, platform and network barriers.  It provides
+**execnet** provides carefully tested means to easily interact with Python
+interpreters across version, platform and network barriers.  It has 
 a minimal and fast API targetting the following uses:
 
 * distribute tasks to local or remote CPUs
@@ -28,7 +28,7 @@ Features
 
 * easy creation, handling and termination of multiple processes
 
-* well tested interactions between CPython 2.4-3.1, Jython 2.5.1 
+* well tested interactions between CPython 2.4-2.7, CPython3.1, Jython 2.5.1 
   and PyPy 1.1 interpreters.
 
 * fully interoperable between Windows and Unix-ish systems. 

--- a/doc/conf.py
+++ b/doc/conf.py
@@ -46,7 +46,7 @@ copyright = '2009, holger krekel and oth
 #
 # The short X.Y version.
 import execnet
-version = "1.0.1"
+version = "1.0.2"
 # The full version, including alpha/beta/rc tags.
 release = version
 

--- a/execnet/gateway.py
+++ b/execnet/gateway.py
@@ -3,7 +3,7 @@ gateway code for initiating popen, socke
 (c) 2004-2009, Holger Krekel and others
 """
 
-import sys, os, inspect, socket, types
+import sys, os, inspect, types
 import textwrap
 import execnet
 from execnet.gateway_base import Message, Popen2IO

--- a/execnet/__init__.py
+++ b/execnet/__init__.py
@@ -1,10 +1,9 @@
 """
-execnet: Elastic Python Deployment.
-package for connecting to local and remote Python Interpreters.
+execnet: pure python lib for connecting to local and remote Python Interpreters.
 
 (c) 2009, Holger Krekel and others
 """
-__version__ = "1.0.1.post1"
+__version__ = "1.0.2"
 
 import execnet.apipkg
 

--- a/execnet/gateway_socket.py
+++ b/execnet/gateway_socket.py
@@ -2,6 +2,9 @@ import socket
 from execnet.gateway import Gateway, HostNotFound
 import os, sys, inspect
 
+try: bytes
+except NameError: bytes = str
+
 class SocketIO:
     error = (socket.error, EOFError)
     def __init__(self, sock):
@@ -10,7 +13,6 @@ class SocketIO:
             sock.setsockopt(socket.SOL_IP, socket.IP_TOS, 0x10)# IPTOS_LOWDELAY
             sock.setsockopt(socket.SOL_TCP, socket.TCP_NODELAY, 1)
         except (AttributeError, socket.error):
-            e = sys.exc_info()[1]
             sys.stderr.write("WARNING: cannot set socketoption")
 
     def read(self, numbytes):
@@ -24,7 +26,6 @@ class SocketIO:
         return buf
 
     def write(self, data):
-        assert isinstance(data, bytes)
         self.sock.sendall(data)
 
     def close_read(self):

From commits-noreply at bitbucket.org  Wed Dec 23 20:17:40 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Wed, 23 Dec 2009 19:17:40 +0000 (UTC)
Subject: [execnet-commit] execnet commit 58aeb70cf80a: fix internal timeout
	waiting bug (thanks	ronny)
Message-ID: <20091223191740.70A4E7EF2A@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261595843 -3600
# Node ID 58aeb70cf80a127af7ebc8193396c0f5bc2ee248
# Parent 6007ee61a482f19d765917da3d75a25ff2049793
fix internal timeout waiting bug (thanks ronny)

--- a/execnet/gateway_base.py
+++ b/execnet/gateway_base.py
@@ -212,6 +212,7 @@ class Channel(object):
     """Communication channel between two Python Interpreter execution points."""
     RemoteError = RemoteError
     TimeoutError = TimeoutError
+    _INTERNALWAKEUP = 1000
     _executing = False
 
     def __init__(self, gateway, id):
@@ -395,7 +396,7 @@ class Channel(object):
         if itemqueue is None:
             raise IOError("cannot receive(), channel has receiver callback")
         if timeout < 0:
-            internal_timeout = 1000
+            internal_timeout = self._INTERNALWAKEUP
         else:
             internal_timeout = timeout
            
@@ -404,7 +405,7 @@ class Channel(object):
                 x = itemqueue.get(timeout=internal_timeout)
                 break
             except queue.Empty:
-                if internal_timeout < 0:
+                if timeout < 0:
                     continue
                 raise self.TimeoutError("no item after %r seconds" %(timeout))
         if x is ENDMARKER:

--- a/testing/test_channel.py
+++ b/testing/test_channel.py
@@ -41,6 +41,15 @@ class TestChannelBasicBehaviour:
         channel.send(1)
         x = channel.receive(timeout=0.1)
 
+    def test_channel_receive_internal_timeout(self, gw, monkeypatch):
+        channel = gw.remote_exec("""
+            import time
+            time.sleep(0.5)
+            channel.send(1)
+        """)
+        monkeypatch.setattr(channel.__class__, '_INTERNALWAKEUP', 0.2)
+        x = channel.receive()
+
     def test_channel_close_and_then_receive_error_multiple(self, gw):
         channel = gw.remote_exec('channel.send(42) ; raise ValueError')
         x = channel.receive()

--- a/CHANGELOG
+++ b/CHANGELOG
@@ -13,6 +13,9 @@ 1.0.2
 - internally split socket gateways, speeds up popen-gateways
   by 10% (now at <50 milliseconds per-gateway on a 1.5 GHZ machine) 
 
+- fix bug in channel.receive() that would wrongly raise a TimeoutError 
+  after 1000 seconds (thanks Ronny Pfannschmidt) 
+
 1.0.1
 --------------------------------

From commits-noreply at bitbucket.org  Wed Dec 23 21:30:17 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Wed, 23 Dec 2009 20:30:17 +0000 (UTC)
Subject: [execnet-commit] execnet commit 330b0032a09e: fixes for jython
Message-ID: <20091223203017.32CF67EF3F@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261600136 -3600
# Node ID 330b0032a09e13e38644c74d4a679d3760cdcdcf
# Parent 58aeb70cf80a127af7ebc8193396c0f5bc2ee248
fixes for jython

--- a/testing/test_termination.py
+++ b/testing/test_termination.py
@@ -74,6 +74,7 @@ def test_close_initiating_remote_no_erro
     stdout, stderr = popen.communicate()
     assert not stderr
 
+ at py.test.mark.skipif("not hasattr(os, 'dup')")
 def test_terminate_implicit_does_trykill(testdir, anypython, capfd):
     p = testdir.makepyfile("""
         import sys

--- a/testing/test_gateway.py
+++ b/testing/test_gateway.py
@@ -49,13 +49,14 @@ class TestBasicGateway:
         assert numchan2 == numchan
 
     def test_gateway_status_busy(self, gw):
+        numchannels = gw.remote_status().numchannels
         ch1 = gw.remote_exec("channel.send(1); channel.receive()")
         ch2 = gw.remote_exec("channel.receive()")
         ch1.receive()
         status = gw.remote_status()
         assert status.numexecuting == 1 # number of active execution threads
         assert status.execqsize == 1 # one more queued
-        assert status.numchannels == 2
+        assert status.numchannels == numchannels + 2
         ch1.send(None)
         ch2.send(None)
         ch1.waitclose()

From commits-noreply at bitbucket.org  Wed Dec 23 21:33:54 2009
From: commits-noreply at bitbucket.org (commits-noreply at bitbucket.org)
Date: Wed, 23 Dec 2009 20:33:54 +0000 (UTC)
Subject: [execnet-commit] execnet commit 73176c8168f2: Added tag 1.0.2 for
	changeset	330b0032a09e
Message-ID: <20091223203354.938F97EF1E@bitbucket.org>

# HG changeset patch -- Bitbucket.org
# Project execnet
# URL http://bitbucket.org/hpk42/execnet/overview/
# User holger krekel 
# Date 1261600417 -3600
# Node ID 73176c8168f280a27ee9e2b79ab6b3488a943e23
# Parent 330b0032a09e13e38644c74d4a679d3760cdcdcf
Added tag 1.0.2 for changeset 330b0032a09e

--- a/.hgtags
+++ b/.hgtags
@@ -3,3 +3,4 @@ 7cf86a1811b11aa58112a255e6afba870c8d855e
 7cf86a1811b11aa58112a255e6afba870c8d855e 1.0.0
 5d1c438e335dc369834207882ec5f3227d365c11 1.0.0
 66760dc71203dbceeb08851cf4cf74e2687156d0 1.0.1
+330b0032a09e13e38644c74d4a679d3760cdcdcf 1.0.2
Download

Questions? Suggestions?

execnet: robust gateways to hybrid execution environments

execnet: robust gateways to hybrid execution environments

execnet: rapid multi-Python deployment
