diff --git a/artiq/frontend/artiq_gui.py b/artiq/frontend/artiq_gui.py index ffe3b7cd3..2e05ec91a 100755 --- a/artiq/frontend/artiq_gui.py +++ b/artiq/frontend/artiq_gui.py @@ -141,10 +141,9 @@ def main(): atexit_register_coroutine(d_applets.stop) smgr.register(d_applets) - if os.name != "nt": - d_ttl_dds = moninj.MonInj() - loop.run_until_complete(d_ttl_dds.start(args.server, args.port_notify)) - atexit_register_coroutine(d_ttl_dds.stop) + d_ttl_dds = moninj.MonInj() + loop.run_until_complete(d_ttl_dds.start(args.server, args.port_notify)) + atexit_register_coroutine(d_ttl_dds.stop) d_schedule = schedule.ScheduleDock( status_bar, rpc_clients["schedule"], sub_clients["schedule"]) @@ -154,10 +153,11 @@ def main(): smgr.register(logmgr) # lay out docks - right_docks = [d_explorer, d_shortcuts] - if os.name != "nt": - right_docks += [d_ttl_dds.ttl_dock, d_ttl_dds.dds_dock] - right_docks += [d_datasets, d_applets] + right_docks = [ + d_explorer, d_shortcuts, + d_ttl_dds.ttl_dock, d_ttl_dds.dds_dock, + d_datasets, d_applets + ] main_window.addDockWidget(QtCore.Qt.RightDockWidgetArea, right_docks[0]) for d1, d2 in zip(right_docks, right_docks[1:]): main_window.tabifyDockWidget(d1, d2) diff --git a/artiq/gui/moninj.py b/artiq/gui/moninj.py index 35ebee1a0..62cf6de20 100644 --- a/artiq/gui/moninj.py +++ b/artiq/gui/moninj.py @@ -1,4 +1,5 @@ import asyncio +import threading import logging import socket import struct @@ -261,34 +262,43 @@ class MonInj(TaskObject): self.subscriber = Subscriber("devices", self.init_devices) self.dm = None - self.transport = None + self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + # Never ceasing to disappoint, asyncio has an issue about UDP + # not being supported on Windows (ProactorEventLoop) open since 2014. + self.loop = asyncio.get_event_loop() + self.thread = threading.Thread(target=self.receiver_thread, + daemon=True) + self.thread.start() async def start(self, server, port): - loop = asyncio.get_event_loop() - await loop.create_datagram_endpoint(lambda: self, - family=socket.AF_INET) + await self.subscriber.connect(server, port) try: - await self.subscriber.connect(server, port) - try: - TaskObject.start(self) - except: - await self.subscriber.close() - raise + TaskObject.start(self) except: - self.transport.close() + await self.subscriber.close() raise async def stop(self): await TaskObject.stop(self) await self.subscriber.close() - if self.transport is not None: - self.transport.close() - self.transport = None + try: + # This is required to make recvfrom terminate in the thread. + # On Linux, this raises "OSError: Transport endpoint is not + # connected", but still has the intended effect. + self.socket.shutdown(socket.SHUT_RDWR) + except OSError: + pass + self.socket.close() + self.thread.join() - def connection_made(self, transport): - self.transport = transport + def receiver_thread(self): + while True: + data, addr = self.socket.recvfrom(2048) + if addr is None: + break + self.loop.call_soon_threadsafe(self.datagram_received, data) - def datagram_received(self, data, addr): + def datagram_received(self, data): if self.dm is None: logger.debug("received datagram, but device manager " "is not present yet") @@ -318,12 +328,6 @@ class MonInj(TaskObject): except: logger.warning("failed to process datagram", exc_info=True) - def error_received(self, exc): - logger.warning("datagram endpoint error") - - def connection_lost(self, exc): - self.transport = None - def send_to_device(self, data): if self.dm is None: logger.debug("cannot sent to device yet, no device manager") @@ -332,10 +336,8 @@ class MonInj(TaskObject): logger.debug("core device address: %s", ca) if ca is None: logger.warning("could not find core device address") - elif self.transport is None: - logger.warning("datagram endpoint not available") else: - self.transport.sendto(data, (ca, 3250)) + self.socket.sendto(data, (ca, 3250)) async def _do(self): while True: