occheung
/
artiq
forked from M-Labs/artiq
1
0
Fork 0
Code Pull Requests Activity

coreanalyzer_proxy: cleanups/renames

This commit is contained in:
Sebastien Bourdeauducq 2023-12-13 13:07:35 +08:00
parent be08862606
commit a26cee6ca7
1 changed files with 8 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
artiq/frontend/aqctl_coreanalyzer_proxy.py
View File

@ -15,7 +15,7 @@ logger = logging.getLogger(__name__)
# simplified version of sipyco Broadcaster # simplified version of sipyco Broadcaster
class ProxyServer(AsyncioServer): class ProxyServer(AsyncioServer):
def __init__(self, queue_limit=1024): def __init__(self, queue_limit=8):
AsyncioServer.__init__(self) AsyncioServer.__init__(self)
self._recipients = set() self._recipients = set()
self._queue_limit = queue_limit self._queue_limit = queue_limit
@ -38,26 +38,26 @@ class ProxyServer(AsyncioServer):
finally: finally:
writer.close() writer.close()
def request_dump_cb(self, dump): def distribute(self, dump):
for recipient in self._recipients: for recipient in self._recipients:
recipient.put_nowait(dump) recipient.put_nowait(dump)
class ProxyControl: class ProxyControl:
def __init__(self, request_dump_cb, core_addr, core_port=1382): def __init__(self, distribute_cb, core_addr, core_port=1382):
self.request_dump_cb = request_dump_cb self.distribute_cb = distribute_cb
self.core_addr = core_addr self.core_addr = core_addr
self.core_port = core_port self.core_port = core_port
def ping(self): def ping(self):
return True return True
def request_dump(self): def trigger(self):
try: try:
dump = get_analyzer_dump(self.core_addr, self.core_port) dump = get_analyzer_dump(self.core_addr, self.core_port)
self.request_dump_cb(dump) self.distribute_cb(dump)
except: except:
logger.warning("Failed to get analyzer dump:", exc_info=1) logger.warning("Trigger failed:", exc_info=True)
return False return False
else: else:
return True return True
@ -94,7 +94,7 @@ def main():
loop.run_until_complete(proxy_server.start(bind_address, args.port_proxy)) loop.run_until_complete(proxy_server.start(bind_address, args.port_proxy))
atexit_register_coroutine(proxy_server.stop, loop=loop) atexit_register_coroutine(proxy_server.stop, loop=loop)
controller = ProxyControl(proxy_server.request_dump_cb, args.core_addr) controller = ProxyControl(proxy_server.distribute, args.core_addr)
server = Server({"coreanalyzer_proxy_control": controller}, None, True) server = Server({"coreanalyzer_proxy_control": controller}, None, True)
loop.run_until_complete(server.start(bind_address, args.port_control)) loop.run_until_complete(server.start(bind_address, args.port_control))
atexit_register_coroutine(server.stop, loop=loop) atexit_register_coroutine(server.stop, loop=loop)