protocols/pipe_ipc: fix read transport resource leak

This caused the "Exception ignored in:" message on exit.
m-labs · Feb 11, 2016 · 4a03335 · 4a03335
1 parent 29d4755
commit 4a03335
Showing 1 changed file with 12 additions and 12 deletions.
diff --git a/artiq/protocols/pipe_ipc.py b/artiq/protocols/pipe_ipc.py
@@ -24,17 +24,14 @@ async def read(self, n):
     async def _fds_to_asyncio(rfd, wfd, loop):
         reader = asyncio.StreamReader(loop=loop)
         reader_protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
+        rf = open(rfd, "rb", 0)
+        rt, _ = await loop.connect_read_pipe(lambda: reader_protocol, rf)
 
         wf = open(wfd, "wb", 0)
-        transport, _ = await loop.connect_write_pipe(
-            FlowControlMixin, wf)
-        writer = asyncio.StreamWriter(transport, reader_protocol,
-                                      None, loop)
-
-        rf = open(rfd, "rb", 0)
-        await loop.connect_read_pipe(lambda: reader_protocol, rf)
+        wt, _ = await loop.connect_write_pipe(FlowControlMixin, wf)
+        writer = asyncio.StreamWriter(wt, reader_protocol, None, loop)
 
-        return reader, writer
+        return rt, reader, writer
 
 
     class AsyncioParentComm(_BaseIO):
@@ -47,6 +44,7 @@ def get_address(self):
 
         async def _autoclose(self):
             await self.process.wait()
+            self.reader_transport.close()
             self.writer.close()
 
         async def create_subprocess(self, *args, **kwargs):
@@ -56,8 +54,8 @@ async def create_subprocess(self, *args, **kwargs):
             os.close(self.c_rfd)
             os.close(self.c_wfd)
 
-            self.reader, self.writer = await _fds_to_asyncio(
-                self.p_rfd, self.p_wfd, loop)
+            self.reader_transport, self.reader, self.writer = \
+                await _fds_to_asyncio(self.p_rfd, self.p_wfd, loop)
             asyncio.ensure_future(self._autoclose())
 
 
@@ -67,10 +65,12 @@ def __init__(self, address):
 
         async def connect(self):
             rfd, wfd = self.address.split(",", maxsplit=1)
-            self.reader, self.writer = await _fds_to_asyncio(
-                int(rfd), int(wfd), asyncio.get_event_loop())
+            self.reader_transport, self.reader, self.writer = \
+                await _fds_to_asyncio(int(rfd), int(wfd),
+                                      asyncio.get_event_loop())
 
         def close(self):
+            self.reader_transport.close()
             self.writer.close()