[Python-ideas] Tulip / PEP 3156 (original) (raw)

[Python-ideas] Tulip / PEP 3156 - subprocess events

Paul Moore p.f.moore at gmail.com
Tue Jan 22 17:10:18 CET 2013

On 22 January 2013 15:43, Paul Moore <p.f.moore at gmail.com> wrote:

I'll try implementing the pipe transport approach and see how it looks in contrast.

Here's a quick proof of concept (for a read pipe):

class UnixEventLoop(events.EventLoop): ... @tasks.task def connect_read_pipe(self, protocol_factory, rpipe): protocol = protocol_factory() waiter = futures.Future() transport = _UnixReadPipeTransport(self, rpipe, protocol, waiter) yield from waiter return transport, protocol

class _UnixReadPipeTransport(transports.Transport):

def __init__(self, event_loop, rpipe, protocol, waiter=None):
    self._event_loop = event_loop
    self._pipe = rpipe.fileno()
    self._protocol = protocol
    self._buffer = []
    self._event_loop.add_reader(self._pipe.fileno(), self._read_ready)
    self._event_loop.call_soon(self._protocol.connection_made, self)
    if waiter is not None:
        self._event_loop.call_soon(waiter.set_result, None)

def _read_ready(self):
    try:
        data = os.read(self._pipe, 16*1024)
    except BlockingIOError:
        return
    if data:
        self._event_loop.call_soon(self._protocol.data_received, data)
    else:
        self._event_loop.remove_reader(self._pipe)
        self._event_loop.call_soon(self._protocol.eof_received)

Using this to re-implement the subprocess test looks something like this (the protocol is unchanged from the existing test):

def testUnixSubprocessWithPipe(self): proc = subprocess.Popen(['/bin/ls', '-lR'], stdout=subprocess.PIPE) t, p = yield from self.event_loop.connect_read_pipe(MyProto, proc.stdout) self.event_loop.run()

To be honest, this looks sufficiently straightforward that I don't see the benefit in a less-general high-level transport type...

Paul

More information about the Python-ideas mailing list