diff --git a/confluent_client/bin/nodeconsole b/confluent_client/bin/nodeconsole index da274319..ab360d84 100755 --- a/confluent_client/bin/nodeconsole +++ b/confluent_client/bin/nodeconsole @@ -254,14 +254,23 @@ class InputHandler: currloop = asyncio.get_running_loop() self.fd = sys.stdin.fileno() self.seqtimeout = None + self.pendinginput = asyncio.Queue() + self.inputconsumer = asyncio.create_task(self.consume_input()) currloop.add_reader(self.fd, self.handle_input) return self - + def handle_input(self): data = os.read(self.fd, 1) if not data: return - asyncio.create_task(self.process_input(data)) + self.pendinginput.put_nowait(data) + + async def consume_input(self): + # One consumer, so a keystroke that blocks relaying cannot be overtaken + # by the next one, and the sequence buffer is only touched by one task + while True: + data = await self.pendinginput.get() + await self.process_input(data) async def timeout_sequence(self, timeout=3, flushbuffer=False): await asyncio.sleep(timeout) @@ -814,15 +823,17 @@ def redraw(): sys.stdout.write('\n') sys.stdout.flush() resized = False +inputwatcher = None async def do_screenshot(): global streaming global resized global numrows global firstnodename + global inputwatcher sess = client.Command() if streaming: - asyncio.create_task(watch_input()) + inputwatcher = asyncio.create_task(watch_input()) if options.tile: imageformat = os.environ.get('CONFLUENT_IMAGE_PROTOCOL', 'kitty') if imageformat not in ('kitty', 'iterm'): diff --git a/confluent_server/confluent/asynchttp.py b/confluent_server/confluent/asynchttp.py index 76458f8d..b24b5505 100644 --- a/confluent_server/confluent/asynchttp.py +++ b/confluent_server/confluent/asynchttp.py @@ -66,6 +66,7 @@ class AsyncSession(object): self.wshandler = wshandler self.termrelations = [] self.consoles = set([]) + self.handlertasks = set() async def add(self, requestid, rsp): await self.wshandler(messages.AsyncMessage((requestid, rsp))) @@ -107,7 +108,11 @@ async def run_handler(hdlr, req): raise exc.InvalidArgumentException( 'Invalid Session ID or missing request id') cloop = asyncio.get_running_loop() - cloop.create_task(asyncsession.run_handler(hdlr, requestid)) + # The loop only keeps a weak reference, so the session holds the task until + # it finishes, otherwise the request can be collected while still pending + hdlrtask = cloop.create_task(asyncsession.run_handler(hdlr, requestid)) + asyncsession.handlertasks.add(hdlrtask) + hdlrtask.add_done_callback(asyncsession.handlertasks.discard) return requestid diff --git a/ruff.toml b/ruff.toml index bf49679c..a051d317 100644 --- a/ruff.toml +++ b/ruff.toml @@ -76,6 +76,9 @@ select = [ "B023", # closure captures a loop variable, so every closure sees the # last value rather than the one from its iteration "B035", # dict comprehension with a static key + "RUF006", # asyncio.create_task result discarded. The loop keeps only a + # weak reference, so an unreferenced task can be collected + # while it is still pending, and the work is silently dropped ] # Deliberately not selected, though currently at zero: B905 (zip without an