reorganize core device communication code

RELEASE_NOTES.rst

@@ -24,6 +24,8 @@ Release notes
   the device database.
 * ``int(a, width=b)`` has been removed. Use ``int32(a)`` and ``int64(a)``.
 * The kc705 gateware target has been renamed kc705_dds.
+* ``artiq.coredevice.comm_tcp`` has been renamed ``artiq.coredevice.comm_kernel``,
+  and ``Comm`` has been renamed ``CommKernel``.
 
 
 2.2

artiq/coredevice/comm_analyzer.py

@@ -2,15 +2,49 @@ from operator import itemgetter
 from collections import namedtuple
 from itertools import count
 from contextlib import contextmanager
+from enum import Enum
 import struct
 import logging
-
+import socket
-from artiq.protocols.analyzer import MessageType, ExceptionType
 
 
 logger = logging.getLogger(__name__)
 
 
+class MessageType(Enum):
+    output = 0b00
+    input = 0b01
+    exception = 0b10
+    stopped = 0b11
+
+
+class ExceptionType(Enum):
+    legacy_reset = 0b000000
+    legacy_reset_falling = 0b000001
+    legacy_reset_phy = 0b000010
+    legacy_reset_phy_falling = 0b000011
+
+    o_underflow_reset = 0b010000
+    o_sequence_error_reset = 0b010001
+    o_collision_reset = 0b010010
+
+    i_overflow_reset = 0b100000
+
+
+def get_analyzer_dump(host, port=1382):
+    sock = socket.create_connection((host, port))
+    try:
+        r = bytes()
+        while True:
+            buf = sock.recv(8192)
+            if not buf:
+                break
+            r += buf
+    finally:
+        sock.close()
+    return r
+
+
 OutputMessage = namedtuple(
     "OutputMessage", "channel timestamp rtio_counter address data")
 

artiq/coredevice/comm_kernel.py

@@ -1,5 +1,7 @@
 import struct
 import logging
+import socket
+import sys
 import traceback
 import numpy
 from enum import Enum
@@ -69,29 +71,60 @@ class RPCReturnValueError(ValueError):
 RPCKeyword = namedtuple('RPCKeyword', ['name', 'value'])
 
 
-class CommGeneric:
+def set_keepalive(sock, after_idle, interval, max_fails):
-    def __init__(self):
+    if sys.platform.startswith("linux"):
+        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
+        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, after_idle)
+        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, interval)
+        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, max_fails)
+    elif sys.platform.startswith("win") or sys.platform.startswith("cygwin"):
+        # setting max_fails is not supported, typically ends up being 5 or 10
+        # depending on Windows version
+        sock.ioctl(socket.SIO_KEEPALIVE_VALS,
+                   (1, after_idle*1000, interval*1000))
+    else:
+        logger.warning("TCP keepalive not supported on platform '%s', ignored",
+                       sys.platform)
+
+
+def initialize_connection(host, port):
+    sock = socket.create_connection((host, port), 5.0)
+    sock.settimeout(None)
+    set_keepalive(sock, 3, 2, 3)
+    logger.debug("connected to host %s on port %d", host, port)
+    return sock
+
+
+class CommKernel:
+    def __init__(self, dmgr, host, port=1381):
         self._read_type = None
+        self.host = host
+        self.port = port
 
     def open(self):
-        """Opens the communication channel.
+        if hasattr(self, "socket"):
-        Must do nothing if already opened."""
+            return
-        raise NotImplementedError
+        self.socket = initialize_connection(self.host, self.port)
+        self.socket.sendall(b"ARTIQ coredev\n")
 
     def close(self):
-        """Closes the communication channel.
+        if not hasattr(self, "socket"):
-        Must do nothing if already closed."""
+            return
-        raise NotImplementedError
+        self.socket.close()
+        del self.socket
+        logger.debug("disconnected")
 
     def read(self, length):
-        """Reads exactly length bytes from the communication channel.
+        r = bytes()
-        The channel is assumed to be opened."""
+        while len(r) < length:
-        raise NotImplementedError
+            rn = self.socket.recv(min(8192, length - len(r)))
+            if not rn:
+                raise ConnectionResetError("Connection closed")
+            r += rn
+        return r
 
     def write(self, data):
-        """Writes exactly length bytes to the communication channel.
+        self.socket.sendall(data)
-        The channel is assumed to be opened."""
-        raise NotImplementedError
 
     #
     # Reader interface

artiq/coredevice/comm_kernel_dummy.py

@@ -1,7 +1,7 @@
 from operator import itemgetter
 
 
-class Comm:
+class CommKernel:
     def __init__(self, dmgr):
         super().__init__()
 

artiq/coredevice/comm_moninj.py

@@ -4,7 +4,7 @@ import struct
 from enum import Enum
 
 
-__all__ = ["TTLProbe", "TTLOverride", "MonInjComm"]
+__all__ = ["TTLProbe", "TTLOverride", "CommMonInj"]
 
 
 logger = logging.getLogger(__name__)
@@ -21,7 +21,7 @@ class TTLOverride(Enum):
     oe = 2
 
 
-class MonInjComm:
+class CommMonInj:
     def __init__(self, monitor_cb, injection_status_cb, disconnect_cb=None):
         self.monitor_cb = monitor_cb
         self.injection_status_cb = injection_status_cb

artiq/coredevice/comm_tcp.py

@@ -1,76 +0,0 @@
-import logging
-import socket
-import sys
-
-from artiq.coredevice.comm_generic import CommGeneric
-
-
-logger = logging.getLogger(__name__)
-
-
-def set_keepalive(sock, after_idle, interval, max_fails):
-    if sys.platform.startswith("linux"):
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
-        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPIDLE, after_idle)
-        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPINTVL, interval)
-        sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_KEEPCNT, max_fails)
-    elif sys.platform.startswith("win") or sys.platform.startswith("cygwin"):
-        # setting max_fails is not supported, typically ends up being 5 or 10
-        # depending on Windows version
-        sock.ioctl(socket.SIO_KEEPALIVE_VALS,
-                   (1, after_idle*1000, interval*1000))
-    else:
-        logger.warning("TCP keepalive not supported on platform '%s', ignored",
-                       sys.platform)
-
-
-def initialize_connection(host, port):
-    sock = socket.create_connection((host, port), 5.0)
-    sock.settimeout(None)
-    set_keepalive(sock, 3, 2, 3)
-    logger.debug("connected to host %s on port %d", host, port)
-    return sock
-
-
-class Comm(CommGeneric):
-    def __init__(self, dmgr, host, port=1381, port_analyzer=1382):
-        super().__init__()
-        self.host = host
-        self.port = port
-        self.port_analyzer = port_analyzer
-
-    def open(self):
-        if hasattr(self, "socket"):
-            return
-        self.socket = initialize_connection(self.host, self.port)
-        self.socket.sendall(b"ARTIQ coredev\n")
-
-    def close(self):
-        if not hasattr(self, "socket"):
-            return
-        self.socket.close()
-        del self.socket
-        logger.debug("disconnected")
-
-    def read(self, length):
-        r = bytes()
-        while len(r) < length:
-            rn = self.socket.recv(min(8192, length - len(r)))
-            if not rn:
-                raise ConnectionResetError("Connection closed")
-            r += rn
-        return r
-
-    def write(self, data):
-        self.socket.sendall(data)
-
-    def get_analyzer_dump(self):
-        sock = initialize_connection(self.host, self.port_analyzer)
-        r = bytes()
-        while True:
-            buf = sock.recv(8192)
-            if not buf:
-                break
-            r += buf
-        sock.close()
-        return r

artiq/dashboard/moninj.py

@@ -4,7 +4,7 @@ import logging
 from PyQt5 import QtCore, QtWidgets, QtGui
 
 from artiq.protocols.sync_struct import Subscriber
-from artiq.coredevice.moninj import *
+from artiq.coredevice.comm_moninj import *
 from artiq.gui.tools import LayoutWidget
 from artiq.gui.flowlayout import FlowLayout
 
@@ -197,7 +197,7 @@ class _DeviceManager:
         try:
             if v["type"] == "local":
                 widget = None
-                if v["module"] == "artiq.coredevice.comm_tcp":
+                if k == "comm":
                     self.core_addr = v["arguments"]["host"]
                     self.new_core_addr.set()
                 elif v["module"] == "artiq.coredevice.ttl":
@@ -300,7 +300,7 @@ class _DeviceManager:
             if self.core_connection is not None:
                 await self.core_connection.close()
                 self.core_connection = None
-            new_core_connection = MonInjComm(self.monitor_cb, self.injection_status_cb,
+            new_core_connection = CommMonInj(self.monitor_cb, self.injection_status_cb,
                     lambda: logger.error("lost connection to core device moninj"))
             try:
                 await new_core_connection.connect(self.core_addr, 1383)

artiq/examples/drtio/device_db.pyon

@@ -5,8 +5,8 @@
 {
     "comm": {
         "type": "local",
-        "module": "artiq.coredevice.comm_tcp",
+        "module": "artiq.coredevice.comm_kernel",
-        "class": "Comm",
+        "class": "CommKernel",
         "arguments": {"host": "kc705.lab.m-labs.hk"}
     },
     "core": {

artiq/examples/master/device_db.pyon

@@ -5,8 +5,8 @@
 {
     "comm": {
         "type": "local",
-        "module": "artiq.coredevice.comm_tcp",
+        "module": "artiq.coredevice.comm_kernel",
-        "class": "Comm",
+        "class": "CommKernel",
         "arguments": {"host": "kc705.lab.m-labs.hk"}
     },
     "core": {

artiq/examples/phaser/device_db.pyon

@@ -3,8 +3,8 @@
 {
     "comm": {
         "type": "local",
-        "module": "artiq.coredevice.comm_tcp",
+        "module": "artiq.coredevice.comm_kernel",
-        "class": "Comm",
+        "class": "CommKernel",
         "arguments": {"host": "kc705aux.lab.m-labs.hk"}
     },
     "core": {

artiq/frontend/artiq_coreanalyzer.py

@@ -6,7 +6,8 @@ import sys
 from artiq.tools import verbosity_args, init_logger
 from artiq.master.databases import DeviceDB
 from artiq.master.worker_db import DeviceManager
-from artiq.coredevice.analyzer import decode_dump, decoded_dump_to_vcd
+from artiq.coredevice.comm_analyzer import (get_analyzer_dump,
+                                            decode_dump, decoded_dump_to_vcd)
 
 
 def get_argparser():
@@ -38,13 +39,12 @@ def main():
         sys.exit(1)
 
     device_mgr = DeviceManager(DeviceDB(args.device_db))
-    try:
     if args.read_dump:
         with open(args.read_dump, "rb") as f:
             dump = f.read()
     else:
-            comm = device_mgr.get("comm")
+        core_addr = device_mgr.get_desc("comm")["arguments"]["host"]
-            dump = comm.get_analyzer_dump()
+        dump = get_analyzer_dump(core_addr)
     decoded_dump = decode_dump(dump)
     if args.print_decoded:
         print("Log channel:", decoded_dump.log_channel)
@@ -58,8 +58,6 @@ def main():
     if args.write_dump:
         with open(args.write_dump, "wb") as f:
             f.write(dump)
-    finally:
-        device_mgr.close_devices()        
 
 
 if __name__ == "__main__":

artiq/gateware/rtio/analyzer.py

@@ -3,7 +3,7 @@ from migen.genlib.record import Record, layout_len
 from misoc.interconnect.csr import *
 from misoc.interconnect import stream
 
-from artiq.protocols.analyzer import MessageType, ExceptionType
+from artiq.coredevice.comm_analyzer import MessageType, ExceptionType
 
 
 __all__ = ["Analyzer"]

artiq/protocols/analyzer.py

@@ -1,21 +0,0 @@
-from enum import Enum
-
-
-class MessageType(Enum):
-    output = 0b00
-    input = 0b01
-    exception = 0b10
-    stopped = 0b11
-
-
-class ExceptionType(Enum):
-    legacy_reset = 0b000000
-    legacy_reset_falling = 0b000001
-    legacy_reset_phy = 0b000010
-    legacy_reset_phy_falling = 0b000011
-
-    o_underflow_reset = 0b010000
-    o_sequence_error_reset = 0b010001
-    o_collision_reset = 0b010010
-
-    i_overflow_reset = 0b100000

artiq/test/coredevice/test_analyzer.py

@@ -1,5 +1,5 @@
 from artiq.experiment import *
-from artiq.coredevice.analyzer import (decode_dump, StoppedMessage,
+from artiq.coredevice.comm_analyzer import (decode_dump, StoppedMessage,
                                             OutputMessage, InputMessage,
                                            _extract_log_chars)
 from artiq.test.hardware_testbench import ExperimentCase

artiq/test/coredevice/test_moninj.py

@@ -21,7 +21,7 @@ class MonInjTest(ExperimentCase):
 
         loop = asyncio.get_event_loop()
         try:
-            moninj_comm = MonInjComm(monitor_cb, injection_status_cb)
+            moninj_comm = CommMonInj(monitor_cb, injection_status_cb)
             loop.run_until_complete(moninj_comm.connect(core_host))
             try:
                 moninj_comm.get_injection_status(loop_out_channel, TTLOverride.en.value)

artiq/test/lit/devirtualization/device_db.pyon

@@ -1,8 +1,8 @@
 {
     "comm": {
         "type": "local",
-        "module": "artiq.coredevice.comm_dummy",
+        "module": "artiq.coredevice.comm_kernel_dummy",
-        "class": "Comm",
+        "class": "CommKernel",
         "arguments": {}
     },
     "core": {

artiq/test/lit/embedding/device_db.pyon

@@ -1,8 +1,8 @@
 {
     "comm": {
         "type": "local",
-        "module": "artiq.coredevice.comm_dummy",
+        "module": "artiq.coredevice.comm_kernel_dummy",
-        "class": "Comm",
+        "class": "CommKernel",
         "arguments": {}
     },
     "core": {

artiq/test/lit/escape/device_db.pyon

@@ -1,8 +1,8 @@
 {
     "comm": {
         "type": "local",
-        "module": "artiq.coredevice.comm_dummy",
+        "module": "artiq.coredevice.comm_kernel_dummy",
-        "class": "Comm",
+        "class": "CommKernel",
         "arguments": {}
     },
     "core": {

artiq/test/lit/regression/device_db.pyon

@@ -1,8 +1,8 @@
 {
     "comm": {
         "type": "local",
-        "module": "artiq.coredevice.comm_dummy",
+        "module": "artiq.coredevice.comm_kernel_dummy",
-        "class": "Comm",
+        "class": "CommKernel",
         "arguments": {}
     },
     "core": {