m-labs · Oct 19, 2015 · Oct 19, 2015
Showing with 11 additions and 10 deletions.

+2 −9 artiq/protocols/logging.py

+3 −1 artiq/protocols/sync_struct.py

+6 −0 artiq/tools.py
diff --git a/artiq/protocols/logging.py b/artiq/protocols/logging.py
@@ -2,7 +2,7 @@
 import logging
 
 from artiq.protocols.asyncio_server import AsyncioServer
-from artiq.tools import TaskObject
+from artiq.tools import TaskObject, workaround_asyncio263
 
 
 logger = logging.getLogger(__name__)
@@ -119,19 +119,12 @@ async def _do(self):
             try:
                 reader, writer = await asyncio.open_connection(self.host,
                                                                self.port)
-                detect_close = asyncio.ensure_future(reader.read(1))
                 writer.write(_init_string)
                 while True:
                     message = await self._queue.get() + "\n"
                     writer.write(message.encode())
+                    await workaround_asyncio263()
                     await writer.drain()
-                    # HACK: detect connection termination through the completion
-                    # of a read operation. For some reason, write/drain operations
-                    # on a closed socket do not raise exceptions, but print
-                    # "asyncio:socket.send() raised exception."
-                    if detect_close.done():
-                        await asyncio.sleep(self.reconnect_timer)
-                        break
             except asyncio.CancelledError:
                 return
             except:

diff --git a/artiq/protocols/sync_struct.py b/artiq/protocols/sync_struct.py
@@ -16,6 +16,7 @@
 
 from artiq.protocols import pyon
 from artiq.protocols.asyncio_server import AsyncioServer
+from artiq.tools import workaround_asyncio263
 
 
 _init_string = b"ARTIQ sync_struct\n"
@@ -233,10 +234,11 @@ async def _handle_connection_cr(self, reader, writer):
                     line = await queue.get()
                     writer.write(line)
                     # raise exception on connection error
+                    await workaround_asyncio263()
                     await writer.drain()
             finally:
                 self._recipients[notifier_name].remove(queue)
-        except ConnectionResetError:
+        except (ConnectionResetError, BrokenPipeError):
             # subscribers disconnecting are a normal occurence
             pass
         finally:

diff --git a/artiq/tools.py b/artiq/tools.py
@@ -175,3 +175,9 @@ def notify(self):
         for fut in self._waiters:
             if not fut.done():
                 fut.set_result(False)
+
+
+# See: https://github.com/python/asyncio/issues/263
+@asyncio.coroutine
+def workaround_asyncio263():
+    yield