Streaming sockets from ProcessPoolExecutor
Question: I'm trying to create a Python application in which one process (process 'A')
receives a request and puts it into a ProcessPool (from concurrent.futures).
In handling this request, a message may need to passed to a second process
(process 'B'). I'm using tornado's iostream module to help wrap the
connections and get responses.

Process A is failing to successfully connect to process B from within the
ProcessPool execution. Where am I going wrong?

The client, which makes the initial request to process A:

    
    
    #!/usr/bin/env python
    
    import socket
    import tornado.iostream
    import tornado.ioloop
    
    def print_message ( data ):
        print 'client received', data
    
    s = socket.socket(socket.AF_INET,socket.SOCK_STREAM, 0)
    stream = tornado.iostream.IOStream(s)
    stream.connect(('localhost',2001))
    stream.read_until('\0',print_message)
    stream.write('test message\0')
    tornado.ioloop.IOLoop().instance().start()
    

Process A, that received the initial request:

    
    
    #!/usr/bin/env python
    
    import tornado.ioloop
    import tornado.tcpserver
    import tornado.iostream
    import socket
    import concurrent.futures
    import functools
    
    def handle_request ( data ):
        s = socket.socket(socket.AF_INET,socket.SOCK_STREAM,0)
        out_stream = tornado.iostream.IOStream(s)
        out_stream.connect(('localhost',2002))
        future = out_stream.read_until('\0')
        out_stream.write(data+'\0')
        return future.result()
    
    class server_a (tornado.tcpserver.TCPServer):
    
       def return_response ( self, in_stream, future ):
           in_stream.write(future.result()+'\0')
    
       def handle_read ( self, in_stream, data ):
           future = self.executor.submit(handle_request,data)
           future.add_done_callback(functools.partial(self.return_response,in_stream))
    
       def handle_stream ( self, in_stream, address ):
           in_stream.read_until('\0',functools.partial(self.handle_read,in_stream))
    
       def __init__ ( self ):
           self.executor = concurrent.futures.ProcessPoolExecutor()
           tornado.tcpserver.TCPServer.__init__(self)
    
    server = server_a()
    server.bind(2001)
    server.start(0)
    tornado.ioloop.IOLoop().instance().start()
    

Process B, that should receive the relayed request from Process A:

    
    
    #!/usr/bin/env python
    
    import tornado.ioloop
    import tornado.tcpserver
    import functools
    
    class server_b (tornado.tcpserver.TCPServer):
    
        def handle_read ( self, in_stream, data ):
            in_stream.write('server B read'+data+'\0')
    
        def handle_stream ( self, in_stream, address ):
           in_stream.read_until('\0',functools.partial(self.handle_read,in_stream))
    
    server = server_b()
    server.bind(2002)
    server.start(0)
    tornado.ioloop.IOLoop().instance().start()
    

And finally, the error returned by Process A, which is raised during the
'read_until' method:

    
    
    ERROR:concurrent.futures:exception calling callback for <Future at 0x10654b890 state=finished raised OSError>
    Traceback (most recent call last):
      File "/usr/local/lib/python2.7/site-packages/concurrent/futures/_base.py", line 299, in _invoke_callbacks
        callback(self)
      File "./a.py", line 26, in return_response
        in_stream.write(future.result()+'\0')
      File "/usr/local/lib/python2.7/site-packages/concurrent/futures/_base.py", line 397, in result
        return self.__get_result()
      File "/usr/local/lib/python2.7/site-packages/concurrent/futures/_base.py", line 356, in __get_result
        raise self._exception
    OSError: [Errno 9] Bad file descriptor
    


Answer: I'm not 100% sure why you're getting this "Bad file descriptor" error
(concurrent.futures unfortunately lost backtrace information when it was
backported to 2.7), but there is no IOLoop running in the
ProcessPoolExecutor's worker processes, so you won't be able to use Tornado
constructs like IOStream in this context (unless you spin up a new IOLoop for
each task, but that may not make much sense unless you need compatibility with
other asynchronous libraries).

I'm also not sure if it works to mix tornado's multi-process mode and
ProcessPoolExecutor in this way. I think you may need to move the
initialization of the ProcessPoolExecutor until after the start(0) call.

