Implement transport/protocol pair for unix pipes

8 files changed, 861 insertions, 72 deletions

diff --git a/tests/events_test.py b/tests/events_test.py
index 7964834..a71f10c 100644
--- a/tests/events_test.py
+++ b/tests/events_test.py
@@ -3,6 +3,7 @@
 import concurrent.futures
 import contextlib
 import errno
+import fcntl
 import gc
 import io
 import os
@@ -78,6 +79,48 @@ class MyDatagramProto(protocols.DatagramProtocol):
         self.state = 'CLOSED'
 
 
+class MyReadPipeProto(protocols.Protocol):
+
+    def __init__(self):
+        self.state = ['INITIAL']
+        self.nbytes = 0
+        self.transport = None
+
+    def connection_made(self, transport):
+        self.transport = transport
+        assert self.state == ['INITIAL'], self.state
+        self.state.append('CONNECTED')
+
+    def data_received(self, data):
+        assert self.state == ['INITIAL', 'CONNECTED'], self.state
+        self.nbytes += len(data)
+
+    def eof_received(self):
+        assert self.state == ['INITIAL', 'CONNECTED'], self.state
+        self.state.append('EOF')
+        self.transport.close()
+
+    def connection_lost(self, exc):
+        assert self.state == ['INITIAL', 'CONNECTED', 'EOF'], self.state
+        self.state.append('CLOSED')
+
+
+class MyWritePipeProto(protocols.Protocol):
+
+    def __init__(self):
+        self.state = 'INITIAL'
+        self.transport = None
+
+    def connection_made(self, transport):
+        self.transport = transport
+        assert self.state == 'INITIAL', self.state
+        self.state = 'CONNECTED'
+
+    def connection_lost(self, exc):
+        assert self.state == 'CONNECTED', self.state
+        self.state = 'CLOSED'
+
+
 class EventLoopTestsMixin:
 
     def setUp(self):
@@ -968,6 +1011,92 @@ class EventLoopTestsMixin:
         self.assertIsNone(event_loop._csock)
         self.assertIsNone(event_loop._ssock)
 
+    @unittest.skipUnless(sys.platform != 'win32',
+                         "Don't support pipes for Windows")
+    def test_read_pipe(self):
+        proto = None
+
+        def factory():
+            nonlocal proto
+            proto = MyReadPipeProto()
+            return proto
+
+        rpipe, wpipe = os.pipe()
+        pipeobj = io.open(rpipe, 'rb', 1024)
+
+        @tasks.task
+        def connect():
+            t, p = yield from self.event_loop.connect_read_pipe(factory,
+                                                                pipeobj)
+            self.assertIs(p, proto)
+            self.assertIs(t, proto.transport)
+            self.assertEqual(['INITIAL', 'CONNECTED'], proto.state)
+            self.assertEqual(0, proto.nbytes)
+
+        self.event_loop.run_until_complete(connect())
+
+        os.write(wpipe, b'1')
+        self.event_loop.run_once()
+        self.assertEqual(1, proto.nbytes)
+
+        os.write(wpipe, b'2345')
+        self.event_loop.run_once()
+        self.assertEqual(['INITIAL', 'CONNECTED'], proto.state)
+        self.assertEqual(5, proto.nbytes)
+
+        os.close(wpipe)
+        self.event_loop.run_once()
+        self.assertEqual(['INITIAL', 'CONNECTED', 'EOF', 'CLOSED'], proto.state)
+        # extra info is available
+        self.assertIsNotNone(proto.transport.get_extra_info('pipe'))
+
+    @unittest.skipUnless(sys.platform != 'win32',
+                         "Don't support pipes for Windows")
+    def test_write_pipe(self):
+        proto = None
+        transport = None
+
+        def factory():
+            nonlocal proto
+            proto = MyWritePipeProto()
+            return proto
+
+        rpipe, wpipe = os.pipe()
+        pipeobj = io.open(wpipe, 'wb', 1024)
+
+        @tasks.task
+        def connect():
+            nonlocal transport
+            t, p = yield from self.event_loop.connect_write_pipe(factory,
+                                                                 pipeobj)
+            self.assertIs(p, proto)
+            self.assertIs(t, proto.transport)
+            self.assertEqual('CONNECTED', proto.state)
+            transport = t
+
+        self.event_loop.run_until_complete(connect())
+
+        transport.write(b'1')
+        self.event_loop.run_once()
+        data = os.read(rpipe, 1024)
+        self.assertEqual(b'1', data)
+
+        transport.write(b'2345')
+        self.event_loop.run_once()
+        data = os.read(rpipe, 1024)
+        self.assertEqual(b'2345', data)
+        self.assertEqual('CONNECTED', proto.state)
+
+        os.close(rpipe)
+
+        # extra info is available
+        self.assertIsNotNone(proto.transport.get_extra_info('pipe'))
+
+        # close connection
+        proto.transport.close()
+        self.event_loop.run_once()
+        self.assertEqual('CLOSED', proto.state)
+
 
 if sys.platform == 'win32':
     from tulip import windows_events
@@ -1210,6 +1339,14 @@ class AbstractEventLoopTests(unittest.TestCase):
             NotImplementedError, ev_loop.add_signal_handler, 1, f)
         self.assertRaises(
             NotImplementedError, ev_loop.remove_signal_handler, 1)
+        self.assertRaises(
+            NotImplementedError, ev_loop.remove_signal_handler, 1)
+        self.assertRaises(
+            NotImplementedError, ev_loop.connect_read_pipe, f,
+            unittest.mock.sentinel.pipe)
+        self.assertRaises(
+            NotImplementedError, ev_loop.connect_write_pipe, f,
+            unittest.mock.sentinel.pipe)
 
 
 class ProtocolsAbsTests(unittest.TestCase):
diff --git a/tests/unix_events_test.py b/tests/unix_events_test.py
index 24ea494..a30be2c 100644
--- a/tests/unix_events_test.py
+++ b/tests/unix_events_test.py
@@ -1,6 +1,7 @@
 """Tests for unix_events.py."""
 
 import errno
+import io
 import unittest
 import unittest.mock
 
@@ -10,6 +11,8 @@ except ImportError:
     signal = None
 
 from tulip import events
+from tulip import futures
+from tulip import protocols
 from tulip import unix_events
 
 
@@ -166,3 +169,406 @@ class SelectorEventLoopTests(unittest.TestCase):
 
         self.assertRaises(
             RuntimeError, self.event_loop.remove_signal_handler, signal.SIGHUP)
+
+
+class UnixReadPipeTransportTests(unittest.TestCase):
+
+    def setUp(self):
+        self.event_loop = unittest.mock.Mock(spec_set=events.AbstractEventLoop)
+        self.pipe = unittest.mock.Mock(spec_set=io.RawIOBase)
+        self.pipe.fileno.return_value = 5
+        self.protocol = unittest.mock.Mock(spec_set=protocols.Protocol)
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_ctor(self, m_fcntl):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+        self.event_loop.add_reader.assert_called_with(5, tr._read_ready)
+        self.event_loop.call_soon.assert_called_with(
+            self.protocol.connection_made, tr)
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_ctor_with_waiter(self, m_fcntl):
+        fut = futures.Future()
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol, fut)
+        self.event_loop.call_soon.assert_called_with(fut.set_result, None)
+
+    @unittest.mock.patch('os.read')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__read_ready(self, m_fcntl, m_read):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+        m_read.return_value = b'data'
+        tr._read_ready()
+
+        m_read.assert_called_with(5, tr.max_size)
+        self.protocol.data_received.assert_called_with(b'data')
+
+    @unittest.mock.patch('os.read')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__read_ready_eof(self, m_fcntl, m_read):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+        m_read.return_value = b''
+        tr._read_ready()
+
+        m_read.assert_called_with(5, tr.max_size)
+        self.protocol.eof_received.assert_called_with()
+        self.event_loop.remove_reader.assert_called_with(5)
+
+    @unittest.mock.patch('os.read')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__read_ready_blocked(self, m_fcntl, m_read):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+        self.event_loop.reset_mock()
+        m_read.side_effect = BlockingIOError
+        tr._read_ready()
+
+        m_read.assert_called_with(5, tr.max_size)
+        self.assertFalse(self.protocol.data_received.called)
+
+    @unittest.mock.patch('logging.exception')
+    @unittest.mock.patch('os.read')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__read_ready_error(self, m_fcntl, m_read, m_logexc):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+        err = OSError()
+        m_read.side_effect = err
+        tr._close = unittest.mock.Mock()
+        tr._read_ready()
+
+        m_read.assert_called_with(5, tr.max_size)
+        tr._close.assert_called_with(err)
+        m_logexc.assert_called_with('Fatal error for %s', tr)
+
+    @unittest.mock.patch('os.read')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_pause(self, m_fcntl, m_read):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+
+        tr.pause()
+        self.event_loop.remove_reader.assert_called_with(5)
+
+    @unittest.mock.patch('os.read')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_resume(self, m_fcntl, m_read):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+
+        tr.resume()
+        self.event_loop.add_reader.assert_called_with(5, tr._read_ready)
+
+    @unittest.mock.patch('os.read')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_close(self, m_fcntl, m_read):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+
+        tr._close = unittest.mock.Mock()
+        tr.close()
+        tr._close.assert_called_with(None)
+
+    @unittest.mock.patch('os.read')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_close_already_closing(self, m_fcntl, m_read):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+
+        tr._closing = True
+        tr._close = unittest.mock.Mock()
+        tr.close()
+        self.assertFalse(tr._close.called)
+
+
+    @unittest.mock.patch('os.read')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__close(self, m_fcntl, m_read):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+
+        err = object()
+        tr._close(err)
+        self.assertTrue(tr._closing)
+        self.event_loop.remove_reader.assert_called_with(5)
+        self.protocol.connection_lost.assert_called_with(err)
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__call_connection_lost(self, m_fcntl):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+
+        err = None
+        tr._call_connection_lost(err)
+        self.protocol.connection_lost.assert_called_with(err)
+        self.pipe.close.assert_called_with()
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__call_connection_lost_with_err(self, m_fcntl):
+        tr = unix_events._UnixReadPipeTransport(self.event_loop, self.pipe,
+                                                self.protocol)
+
+        err = OSError()
+        tr._call_connection_lost(err)
+        self.protocol.connection_lost.assert_called_with(err)
+        self.pipe.close.assert_called_with()
+
+
+class UnixWritePipeTransportTests(unittest.TestCase):
+
+    def setUp(self):
+        self.event_loop = unittest.mock.Mock(spec_set=events.AbstractEventLoop)
+        self.pipe = unittest.mock.Mock(spec_set=io.RawIOBase)
+        self.pipe.fileno.return_value = 5
+        self.protocol = unittest.mock.Mock(spec_set=protocols.Protocol)
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_ctor(self, m_fcntl):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+        self.event_loop.call_soon.assert_called_with(
+            self.protocol.connection_made, tr)
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_ctor_with_waiter(self, m_fcntl):
+        fut = futures.Future()
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol, fut)
+        self.event_loop.call_soon.assert_called_with(fut.set_result, None)
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_can_write_eof(self, m_fcntl):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+        self.assertTrue(tr.can_write_eof())
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_write(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        m_write.return_value = 4
+        tr.write(b'data')
+        m_write.assert_called_with(5, b'data')
+        self.assertFalse(self.event_loop.add_writer.called)
+        self.assertEqual([], tr._buffer)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_write_no_data(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr.write(b'')
+        self.assertFalse(m_write.called)
+        self.assertFalse(self.event_loop.add_writer.called)
+        self.assertEqual([], tr._buffer)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_write_partial(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        m_write.return_value = 2
+        tr.write(b'data')
+        m_write.assert_called_with(5, b'data')
+        self.event_loop.add_writer.assert_called_with(5, tr._write_ready)
+        self.assertEqual([b'ta'], tr._buffer)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_write_buffer(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr._buffer = [b'previous']
+        tr.write(b'data')
+        self.assertFalse(m_write.called)
+        self.assertFalse(self.event_loop.add_writer.called)
+        self.assertEqual([b'previous', b'data'], tr._buffer)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_write_again(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        m_write.side_effect = BlockingIOError()
+        tr.write(b'data')
+        m_write.assert_called_with(5, b'data')
+        self.event_loop.add_writer.assert_called_with(5, tr._write_ready)
+        self.assertEqual([b'data'], tr._buffer)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_write_err(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        err = OSError()
+        m_write.side_effect = err
+        tr._fatal_error = unittest.mock.Mock()
+        tr.write(b'data')
+        m_write.assert_called_with(5, b'data')
+        self.assertFalse(self.event_loop.called)
+        self.assertEqual([], tr._buffer)
+        tr._fatal_error.assert_called_with(err)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__write_ready(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+        tr._buffer = [b'da', b'ta']
+        m_write.return_value = 4
+        tr._write_ready()
+        m_write.assert_called_with(5, b'data')
+        self.event_loop.remove_writer.assert_called_with(5)
+        self.assertEqual([], tr._buffer)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__write_ready_partial(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr._buffer = [b'da', b'ta']
+        m_write.return_value = 3
+        tr._write_ready()
+        m_write.assert_called_with(5, b'data')
+        self.assertFalse(self.event_loop.remove_writer.called)
+        self.assertEqual([b'a'], tr._buffer)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__write_ready_again(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr._buffer = [b'da', b'ta']
+        m_write.side_effect = BlockingIOError()
+        tr._write_ready()
+        m_write.assert_called_with(5, b'data')
+        self.assertFalse(self.event_loop.remove_writer.called)
+        self.assertEqual([b'data'], tr._buffer)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__write_ready_empty(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr._buffer = [b'da', b'ta']
+        m_write.return_value = 0
+        tr._write_ready()
+        m_write.assert_called_with(5, b'data')
+        self.assertFalse(self.event_loop.remove_writer.called)
+        self.assertEqual([b'data'], tr._buffer)
+
+    @unittest.mock.patch('logging.exception')
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__write_ready_err(self, m_fcntl, m_write, m_logexc):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr._buffer = [b'da', b'ta']
+        m_write.side_effect = err = OSError()
+        tr._write_ready()
+        m_write.assert_called_with(5, b'data')
+        self.event_loop.remove_writer.assert_called_with(5)
+        self.assertEqual([], tr._buffer)
+        self.assertTrue(tr._closing)
+        self.protocol.connection_lost.assert_called_with(err)
+        m_logexc.assert_called_with('Fatal error for %s', tr)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__write_ready_closing(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr._closing = True
+        tr._buffer = [b'da', b'ta']
+        m_write.return_value = 4
+        tr._write_ready()
+        m_write.assert_called_with(5, b'data')
+        self.event_loop.remove_writer.assert_called_with(5)
+        self.assertEqual([], tr._buffer)
+        self.protocol.connection_lost.assert_called_with(None)
+
+    @unittest.mock.patch('os.write')
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_abort(self, m_fcntl, m_write):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr._buffer = [b'da', b'ta']
+        tr.abort()
+        self.assertFalse(m_write.called)
+        self.event_loop.remove_writer.assert_called_with(5)
+        self.assertEqual([], tr._buffer)
+        self.assertTrue(tr._closing)
+        self.protocol.connection_lost.assert_called_with(None)
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__call_connection_lost(self, m_fcntl):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        err = None
+        tr._call_connection_lost(err)
+        self.protocol.connection_lost.assert_called_with(err)
+        self.pipe.close.assert_called_with()
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test__call_connection_lost_with_err(self, m_fcntl):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        err = OSError()
+        tr._call_connection_lost(err)
+        self.protocol.connection_lost.assert_called_with(err)
+        self.pipe.close.assert_called_with()
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_close(self, m_fcntl):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr.write_eof = unittest.mock.Mock()
+        tr.close()
+        tr.write_eof.assert_called_with()
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_close_closing(self, m_fcntl):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr.write_eof = unittest.mock.Mock()
+        tr._closing = True
+        tr.close()
+        self.assertFalse(tr.write_eof.called)
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_write_eof(self, m_fcntl):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+
+        tr.write_eof()
+        self.assertTrue(tr._closing)
+        self.protocol.connection_lost.assert_called_with(None)
+
+    @unittest.mock.patch('fcntl.fcntl')
+    def test_write_eof_pending(self, m_fcntl):
+        tr = unix_events._UnixWritePipeTransport(self.event_loop, self.pipe,
+                                                 self.protocol)
+        tr._buffer = [b'data']
+        tr.write_eof()
+        self.assertTrue(tr._closing)
+        self.assertFalse(self.protocol.connection_lost.called)
diff --git a/tulip/base_events.py b/tulip/base_events.py
index 75cb9dd..1dc0b52 100644
--- a/tulip/base_events.py
+++ b/tulip/base_events.py
@@ -64,6 +64,16 @@ class BaseEventLoop(events.AbstractEventLoop):
         """Create datagram transport."""
         raise NotImplementedError
 
+    def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
+                                  extra=None):
+        """Create read pipe transport."""
+        raise NotImplementedError
+
+    def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
+                                  extra=None):
+        """Create write pipe transport."""
+        raise NotImplementedError
+
     def _read_from_self(self):
         """XXX"""
         raise NotImplementedError
@@ -441,6 +451,24 @@ class BaseEventLoop(events.AbstractEventLoop):
 
         return sock
 
+    @tasks.coroutine
+    def connect_read_pipe(self, protocol_factory, pipe):
+        protocol = protocol_factory()
+        waiter = futures.Future()
+        transport = self._make_read_pipe_transport(pipe, protocol, waiter,
+                                                   extra={})
+        yield from waiter
+        return transport, protocol
+
+    @tasks.coroutine
+    def connect_write_pipe(self, protocol_factory, pipe):
+        protocol = protocol_factory()
+        waiter = futures.Future()
+        transport = self._make_write_pipe_transport(pipe, protocol, waiter,
+                                                    extra={})
+        yield from waiter
+        return transport, protocol
+
     def _add_callback(self, handle):
         """Add a Handle to ready or scheduled."""
         if handle.cancelled:
diff --git a/tulip/events.py b/tulip/events.py
index 9bad35f..da89299 100644
--- a/tulip/events.py
+++ b/tulip/events.py
@@ -193,6 +193,35 @@ class AbstractEventLoop:
                                family=0, proto=0, flags=0):
         raise NotImplementedError
 
+    def connect_read_pipe(self, protocol_factory, pipe):
+        """Register read pipe in eventloop.
+
+        protocol_factory should instantiate object with Protocol interface.
+        pipe is file-like object already switched to nonblocking.
+        Return pair (transport, protocol), where transport support
+        ReadTransport ABC"""
+        # The reason to accept file-like object instead of just file descriptor
+        # is: we need to own pipe and close it at transport finishing
+        # Can got complicated errors if pass f.fileno(),
+        # close fd in pipe transport then close f and vise versa.
+        raise NotImplementedError
+
+    def connect_write_pipe(self, protocol_factory, pipe):
+        """Register write pipe in eventloop.
+
+        protocol_factory should instantiate object with BaseProtocol interface.
+        Pipe is file-like object already switched to nonblocking.
+        Return pair (transport, protocol), where transport support
+        WriteTransport ABC"""
+        # The reason to accept file-like object instead of just file descriptor
+        # is: we need to own pipe and close it at transport finishing
+        # Can got complicated errors if pass f.fileno(),
+        # close fd in pipe transport then close f and vise versa.
+        raise NotImplementedError
+
+    #def spawn_subprocess(self, protocol_factory, pipe):
+    #    raise NotImplementedError
+
     # Ready-based callback registration methods.
     # The add_*() methods return a Handle.
     # The remove_*() methods return True if something was removed,
diff --git a/tulip/protocols.py b/tulip/protocols.py
index f01e2fd..593ee74 100644
--- a/tulip/protocols.py
+++ b/tulip/protocols.py
@@ -3,7 +3,34 @@
 __all__ = ['Protocol', 'DatagramProtocol']
 
 
-class Protocol:
+class BaseProtocol:
+    """ABC for base protocol class.
+
+    Usually user implements protocols that derived from BaseProtocol
+    like Protocol or ProcessProtocol.
+
+    The only case when BaseProtocol should be implemented directly is
+    write-only transport like write pipe
+    """
+
+    def connection_made(self, transport):
+        """Called when a connection is made.
+
+        The argument is the transport representing the pipe connection.
+        To receive data, wait for data_received() calls.
+        When the connection is closed, connection_lost() is called.
+        """
+
+    def connection_lost(self, exc):
+        """Called when the connection is lost or closed.
+
+        The argument is an exception object or None (the latter
+        meaning a regular EOF is received or the connection was
+        aborted or closed).
+        """
+
+
+class Protocol(BaseProtocol):
     """ABC representing a protocol.
 
     The user should implement this interface.  They can inherit from
@@ -16,7 +43,7 @@ class Protocol:
     When the connection is made successfully, connection_made() is
     called with a suitable transport object.  Then data_received()
     will be called 0 or more times with data (bytes) received from the
-    transport; finally, connection_list() will be called exactly once
+    transport; finally, connection_lost() will be called exactly once
     with either an exception object or None as an argument.
 
     State machine of calls:
@@ -24,15 +51,6 @@ class Protocol:
       start -> CM [-> DR*] [-> ER?] -> CL -> end
     """
 
-    def connection_made(self, transport):
-        """Called when a connection is made.
-
-        The argument is the transport representing the connection.
-        To send data, call its write() or writelines() method.
-        To receive data, wait for data_received() calls.
-        When the connection is closed, connection_lost() is called.
-        """
-
     def data_received(self, data):
         """Called when some data is received.
 
@@ -49,26 +67,12 @@ class Protocol:
         set it).
         """
 
-    def connection_lost(self, exc):
-        """Called when the connection is lost or closed.
 
-        The argument is an exception object or None (the latter
-        meaning a regular EOF is received or the connection was
-        aborted or closed).
-        """
-
-
-class DatagramProtocol:
+class DatagramProtocol(BaseProtocol):
     """ABC representing a datagram protocol."""
 
-    def connection_made(self, transport):
-        """Called when a datagram transport is ready."""
-
     def datagram_received(self, data, addr):
         """Called when some datagram is received."""
 
     def connection_refused(self, exc):
         """Connection is refused."""
-
-    def connection_lost(self, exc):
-        """Called when the connection is lost or closed."""
diff --git a/tulip/selector_events.py b/tulip/selector_events.py
index 9bc9c23..5999846 100644
--- a/tulip/selector_events.py
+++ b/tulip/selector_events.py
@@ -659,3 +659,5 @@ class _SelectorDatagramTransport(transports.Transport):
             self._protocol.connection_lost(exc)
         finally:
             self._sock.close()
+
+
diff --git a/tulip/transports.py b/tulip/transports.py
index 6eb1c55..984e7e8 100644
--- a/tulip/transports.py
+++ b/tulip/transports.py
@@ -1,28 +1,10 @@
 """Abstract Transport class."""
 
-__all__ = ['Transport']
+__all__ = ['ReadTransport', 'WriteTransport', 'Transport']
 
 
-class Transport:
-    """ABC representing a transport.
-
-    There may be several implementations, but typically, the user does
-    not implement new transports; rather, the platform provides some
-    useful transports that are implemented using the platform's best
-    practices.
-
-    The user never instantiates a transport directly; they call a
-    utility function, passing it a protocol factory and other
-    information necessary to create the transport and protocol.  (E.g.
-    EventLoop.create_connection() or EventLoop.start_serving().)
-
-    The utility function will asynchronously create a transport and a
-    protocol and hook them up by calling the protocol's
-    connection_made() method, passing it the transport.
-
-    The implementation here raises NotImplemented for every method
-    except writelines(), which calls write() in a loop.
-    """
+class BaseTransport:
+    """Base ABC for transports."""
 
     def __init__(self, extra=None):
         if extra is None:
@@ -33,6 +15,40 @@ class Transport:
         """Get optional transport information."""
         return self._extra.get(name, default)
 
+    def close(self):
+        """Closes the transport.
+
+        Buffered data will be flushed asynchronously.  No more data
+        will be received.  After all buffered data is flushed, the
+        protocol's connection_lost() method will (eventually) called
+        with None as its argument.
+        """
+        raise NotImplementedError
+
+
+class ReadTransport(BaseTransport):
+    """ABC for read-only transports."""
+
+    def pause(self):
+        """Pause the receiving end.
+
+        No data will be passed to the protocol's data_received()
+        method until resume() is called.
+        """
+        raise NotImplementedError
+
+    def resume(self):
+        """Resume the receiving end.
+
+        Data received will once again be passed to the protocol's
+        data_received() method.
+        """
+        raise NotImplementedError
+
+
+class WriteTransport(BaseTransport):
+    """ABC for write-only transports."""
+
     def write(self, data):
         """Write some data bytes to the transport.
 
@@ -63,37 +79,33 @@ class Transport:
         """Return True if this protocol supports write_eof(), False if not."""
         raise NotImplementedError
 
-    def pause(self):
-        """Pause the receiving end.
+    def abort(self):
+        """Closes the transport immediately.
 
-        No data will be passed to the protocol's data_received()
-        method until resume() is called.
+        Buffered data will be lost.  No more data will be received.
+        The protocol's connection_lost() method will (eventually) be
+        called with None as its argument.
         """
         raise NotImplementedError
 
-    def resume(self):
-        """Resume the receiving end.
 
-        Data received will once again be passed to the protocol's
-        data_received() method.
-        """
-        raise NotImplementedError
+class Transport(ReadTransport, WriteTransport):
+    """ABC representing a bidirectional transport.
 
-    def close(self):
-        """Closes the transport.
+    There may be several implementations, but typically, the user does
+    not implement new transports; rather, the platform provides some
+    useful transports that are implemented using the platform's best
+    practices.
 
-        Buffered data will be flushed asynchronously.  No more data
-        will be received.  After all buffered data is flushed, the
-        protocol's connection_lost() method will (eventually) called
-        with None as its argument.
-        """
-        raise NotImplementedError
+    The user never instantiates a transport directly; they call a
+    utility function, passing it a protocol factory and other
+    information necessary to create the transport and protocol.  (E.g.
+    EventLoop.create_connection() or EventLoop.start_serving().)
 
-    def abort(self):
-        """Closes the transport immediately.
+    The utility function will asynchronously create a transport and a
+    protocol and hook them up by calling the protocol's
+    connection_made() method, passing it the transport.
 
-        Buffered data will be lost.  No more data will be received.
-        The protocol's connection_lost() method will (eventually) be
-        called with None as its argument.
-        """
-        raise NotImplementedError
+    The implementation here raises NotImplemented for every method
+    except writelines(), which calls write() in a loop.
+    """
diff --git a/tulip/unix_events.py b/tulip/unix_events.py
index 41f8e0b..833c661 100644
--- a/tulip/unix_events.py
+++ b/tulip/unix_events.py
@@ -1,7 +1,9 @@
 """Selector eventloop for Unix with signal handling."""
 
 import errno
+import fcntl
 import logging
+import os
 import socket
 import sys
 
@@ -12,6 +14,7 @@ except ImportError:  # pragma: no cover
 
 from . import events
 from . import selector_events
+from . import transports
 
 
 __all__ = ['SelectorEventLoop']
@@ -126,3 +129,171 @@ class SelectorEventLoop(selector_events.BaseSelectorEventLoop):
         if not (1 <= sig < signal.NSIG):
             raise ValueError(
                 'sig {} out of range(1, {})'.format(sig, signal.NSIG))
+
+    def _make_read_pipe_transport(self, pipe, protocol, waiter=None,
+                                  extra=None):
+        return _UnixReadPipeTransport(self, pipe, protocol, waiter, extra)
+
+    def _make_write_pipe_transport(self, pipe, protocol, waiter=None,
+                                   extra=None):
+        return _UnixWritePipeTransport(self, pipe, protocol, waiter, extra)
+
+
+def _set_nonblocking(fd):
+    flags = fcntl.fcntl(fd, fcntl.F_GETFL)
+    flags = flags | os.O_NONBLOCK
+    fcntl.fcntl(fd, fcntl.F_SETFL, flags)
+
+
+class _UnixReadPipeTransport(transports.ReadTransport):
+
+    max_size = 256 * 1024  # max bytes we read in one eventloop iteration
+
+    def __init__(self, event_loop, pipe, protocol, waiter=None, extra=None):
+        super().__init__(extra)
+        self._extra['pipe'] = pipe
+        self._event_loop = event_loop
+        self._pipe = pipe
+        self._fileno = pipe.fileno()
+        _set_nonblocking(self._fileno)
+        self._protocol = protocol
+        self._closing = False
+        self._event_loop.add_reader(self._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._fileno, self.max_size)
+        except BlockingIOError:
+            pass
+        except OSError as exc:
+            self._fatal_error(exc)
+        else:
+            if data:
+                self._protocol.data_received(data)
+            else:
+                self._event_loop.remove_reader(self._fileno)
+                self._protocol.eof_received()
+
+    def pause(self):
+        self._event_loop.remove_reader(self._fileno)
+
+    def resume(self):
+        self._event_loop.add_reader(self._fileno, self._read_ready)
+
+    def close(self):
+        if not self._closing:
+            self._close(None)
+
+    def _fatal_error(self, exc):
+        # should be called by exception handler only
+        logging.exception('Fatal error for %s', self)
+        self._close(exc)
+
+    def _close(self, exc):
+        self._closing = True
+        self._event_loop.remove_reader(self._fileno)
+        self._call_connection_lost(exc)
+
+    def _call_connection_lost(self, exc):
+        try:
+            self._protocol.connection_lost(exc)
+        finally:
+            self._pipe.close()
+
+
+class _UnixWritePipeTransport(transports.WriteTransport):
+
+    def __init__(self, event_loop, pipe, protocol, waiter=None, extra=None):
+        super().__init__(extra)
+        self._extra['pipe'] = pipe
+        self._event_loop = event_loop
+        self._pipe = pipe
+        self._fileno = pipe.fileno()
+        _set_nonblocking(self._fileno)
+        self._protocol = protocol
+        self._buffer = []
+        self._closing = False  # Set when close() or write_eof() called.
+        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 write(self, data):
+        assert isinstance(data, (bytes, bytearray)), repr(data)
+        assert not self._closing
+        if not data:
+            return
+        if not self._buffer:
+            # Attempt to send it right away first.
+            try:
+                n = os.write(self._fileno, data)
+            except BlockingIOError:
+                n = 0
+            except Exception as exc:
+                self._fatal_error(exc)
+                return
+            if n == len(data):
+                return
+            elif n > 0:
+                data = data[n:]
+            self._event_loop.add_writer(self._fileno, self._write_ready)
+        assert data, "Data shold not be empty"
+        self._buffer.append(data)
+
+    def _write_ready(self):
+        data = b''.join(self._buffer)
+        assert data, "Data shold not be empty"
+        try:
+            n = os.write(self._fileno, data)
+        except BlockingIOError:
+            self._buffer = [data]
+            return
+        except Exception as exc:
+            self._fatal_error(exc)
+            return
+        if n == len(data):
+            self._buffer = []
+            self._event_loop.remove_writer(self._fileno)
+            if self._closing:
+                self._call_connection_lost(None)
+            return
+        elif n > 0:
+            data = data[n:]
+        self._buffer = [data]  # Try again later.
+
+    def can_write_eof(self):
+        return True
+
+    def write_eof(self):
+        assert not self._closing
+        assert self._pipe
+        self._closing = True
+        if not self._buffer:
+            self._call_connection_lost(None)
+
+    def close(self):
+        if not self._closing:
+            # write_eof is all what we needed to close the write pipe
+            self.write_eof()
+
+    def abort(self):
+        self._close(None)
+
+    def _fatal_error(self, exc):
+        # should be called by exception handler only
+        logging.exception('Fatal error for %s', self)
+        self._close(exc)
+
+    def _close(self, exc=None):
+        self._closing = True
+        self._buffer = []
+        self._event_loop.remove_writer(self._fileno)
+        self._call_connection_lost(exc)
+
+    def _call_connection_lost(self, exc):
+        try:
+            self._protocol.connection_lost(exc)
+        finally:
+            self._pipe.close()