From 4197bd911881689827962d684638e04119512213 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 01:50:24 +0200 Subject: [PATCH 1/9] Fix nodediscover register and subscribe in the async port register_endpoint and subscribe_discovery were left as plain functions iterating the async client generators, so nodediscover register, subscribe and unsubscribe all failed immediately with TypeError: 'async_generator' object is not iterable --- confluent_client/bin/nodediscover | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/confluent_client/bin/nodediscover b/confluent_client/bin/nodediscover index f4e2691a..78a45163 100755 --- a/confluent_client/bin/nodediscover +++ b/confluent_client/bin/nodediscover @@ -51,10 +51,10 @@ columnmapping = { } #TODO: add chassis uuid -def register_endpoint(options, session, addr): +async def register_endpoint(options, session, addr): neednewline = False current = 0 - for rsp in session.update('/discovery/register', {'addresses': addr}): + async for rsp in session.update('/discovery/register', {'addresses': addr}): if 'count' in rsp: total = rsp['count'] elif total > 1: @@ -70,15 +70,15 @@ def register_endpoint(options, session, addr): if neednewline: print('') -def subscribe_discovery(options, session, subscribe, targ): +async def subscribe_discovery(options, session, subscribe, targ): keyn = 'subscribe' if subscribe else 'unsubscribe' payload = {keyn: targ} if subscribe: - for rsp in session.update('/discovery/subscriptions/{0}'.format(targ), payload): + async for rsp in session.update('/discovery/subscriptions/{0}'.format(targ), payload): if 'status' in rsp: print(rsp['status']) else: - for rsp in session.delete('/discovery/subscriptions/{0}'.format(targ)): + async for rsp in session.delete('/discovery/subscriptions/{0}'.format(targ)): if 'status' in rsp: print(rsp['status']) @@ -427,11 +427,11 @@ async def main(): if args[0] == 'reassign': await assign_discovery(options, session, False) if args[0] == 'register': - register_endpoint(options, session, args[1]) + await register_endpoint(options, session, args[1]) if args[0] == 'subscribe': - subscribe_discovery(options, session, True, args[1]) + await subscribe_discovery(options, session, True, args[1]) if args[0] == 'unsubscribe': - subscribe_discovery(options, session, False, args[1]) + await subscribe_discovery(options, session, False, args[1]) if args[0] == 'rescan': await blocking_scan(session) print("Rescan complete") From 7ebc1dc616bebe632ff34e1383176716ce9a6b62 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 01:51:16 +0200 Subject: [PATCH 2/9] Fix nodediscover CSV import in the async port import_csv was left with several synchronous idioms: - search_record is a coroutine function, but was called without await. The returned coroutine is always truthy, so the rescan on incomplete discovery data never happened, and iterating the result raised TypeError: 'coroutine' object is not iterable - the node creation loop iterated an async generator with plain for - the per-node discovery assignment was forked off with os.fork() while the event loop was running, and the child then built a fresh session on the inherited selector Assign discovery entries with asyncio.gather instead of a forked child, which keeps the assignments concurrent and lets their exit codes propagate. The forked child always ended in sys.exit(0), so its accumulated errorcode was discarded. --- confluent_client/bin/nodediscover | 50 +++++++++++++++---------------- 1 file changed, 25 insertions(+), 25 deletions(-) diff --git a/confluent_client/bin/nodediscover b/confluent_client/bin/nodediscover index 78a45163..36fc6b95 100755 --- a/confluent_client/bin/nodediscover +++ b/confluent_client/bin/nodediscover @@ -211,12 +211,12 @@ async def import_csv(options, session): alldata.append(nodedatum) allthere = True for nodedatum in alldata: - if not search_record(nodedatum, options, session) and not broken: + if not await search_record(nodedatum, options, session) and not broken: allthere = False await blocking_scan(session) break for nodedatum in alldata: - if not allthere and not search_record(nodedatum, options, session): + if not allthere and not await search_record(nodedatum, options, session): sys.stderr.write( "Could not match the following data: " + repr(nodedatum) + '\n') @@ -224,11 +224,12 @@ async def import_csv(options, session): nodedata.append(nodedatum) if broken: sys.exit(1) + assignments = [] for datum in nodedata: - maclist = search_record(datum, options, session) + maclist = await search_record(datum, options, session) datum = datum_to_attrib(datum) nodename = datum['name'] - for res in session.create('/nodes/', datum): + async for res in session.create('/nodes/', datum): if 'error' in res: sys.stderr.write(res['error'] + '\n') exitcode |= res.get('errorcode', 1) @@ -237,31 +238,30 @@ async def import_csv(options, session): print('Defined ' + res['created']) else: print(repr(res)) - child = os.fork() - if child: - continue - for mac in maclist: - mysess = client.Command() - for res in mysess.update('/discovery/by-mac/{0}'.format(mac), - {'node': nodename}): - if 'error' in res: - sys.stderr.write(res['error'] + '\n') - exitcode |= res.get('errorcode', 1) - continue - elif 'assigned' in res: - print('Discovered ' + res['assigned']) - else: - print(repr(res)) - sys.exit(0) - while True: - try: - os.wait() - except ChildProcessError: - break + assignments.append(assign_macs(maclist, nodename)) + for rcode in await asyncio.gather(*assignments): + exitcode |= rcode if exitcode: sys.exit(exitcode) +async def assign_macs(maclist, nodename): + exitcode = 0 + for mac in maclist: + mysess = client.Command() + async for res in mysess.update('/discovery/by-mac/{0}'.format(mac), + {'node': nodename}): + if 'error' in res: + sys.stderr.write(res['error'] + '\n') + exitcode |= res.get('errorcode', 1) + continue + elif 'assigned' in res: + print('Discovered ' + res['assigned']) + else: + print(repr(res)) + return exitcode + + async def list_discovery(options, session): orderby = None if options.fields: From 550751d0ff8b24351a6440ab583f649ac5d56c5c Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 01:51:53 +0200 Subject: [PATCH 3/9] Fix the file descriptor send retry in asynctlvdata When sendmsg() reports EAGAIN, _sendmsg rescheduled itself with loop.add_reader(fd, _sendmsg, loop, fut, sock, fd) which waits for the socket to become readable rather than writable, and passes four of the six required arguments, so the callback raised TypeError once it did fire. Wait for writability and pass the message and descriptors through. Also skip the work in _recvmsg if the future was cancelled while waiting for data, as _sendmsg already does, so a cancelled read does not end in InvalidStateError from set_result. This module is imported by the server as well, so both paths are reached by the daemon whenever a descriptor is passed over the local socket. --- confluent_client/confluent/asynctlvdata.py | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/confluent_client/confluent/asynctlvdata.py b/confluent_client/confluent/asynctlvdata.py index 809880b5..e27b20d4 100644 --- a/confluent_client/confluent/asynctlvdata.py +++ b/confluent_client/confluent/asynctlvdata.py @@ -99,9 +99,9 @@ class ClientFile(object): -def _sendmsg(loop, fut, sock, msg, fds, rfd): - if rfd is not None: - loop.remove_reader(rfd) +def _sendmsg(loop, fut, sock, msg, fds, wfd): + if wfd is not None: + loop.remove_writer(wfd) if fut.cancelled(): return try: @@ -110,7 +110,7 @@ def _sendmsg(loop, fut, sock, msg, fds, rfd): [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", fds))]) except (BlockingIOError, InterruptedError): fd = sock.fileno() - loop.add_reader(fd, _sendmsg, loop, fut, sock, fd) + loop.add_writer(fd, _sendmsg, loop, fut, sock, msg, fds, fd) except Exception as exc: fut.set_exception(exc) else: @@ -127,6 +127,8 @@ def send_fds(sock, msg, fds): def _recvmsg(loop, fut, sock, msglen, maxfds, rfd): if rfd is not None: loop.remove_reader(rfd) + if fut.cancelled(): + return fds = array.array("i") # Array of ints try: msg, ancdata, flags, addr = sock.recvmsg( From 64cfad09affc2bc9d947e7c08f82026075d54d6b Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 01:52:09 +0200 Subject: [PATCH 4/9] Complete the async port of the nodegroup attribute paths simple_nodegroups_command awaited the async generators returned by read and update, which raises TypeError: 'async_generator' object can't be awaited and printgroupattributes was left synchronous, iterating one of those generators with plain for. Neither is reachable yet, since nodeattrib only ever passes a noderange and nodegroupattrib still uses the traditional client, but they are the paths nodegroupattrib will use once it is ported. --- confluent_client/confluent/asynclient.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/confluent_client/confluent/asynclient.py b/confluent_client/confluent/asynclient.py index 79b4308f..1064a2aa 100644 --- a/confluent_client/confluent/asynclient.py +++ b/confluent_client/confluent/asynclient.py @@ -347,12 +347,12 @@ class Command(object): else: ikey = key if input is None: - for res in await self.read('/nodegroups/{0}/{1}'.format( + async for res in self.read('/nodegroups/{0}/{1}'.format( noderange, resource)): rc = self.handle_results(ikey, rc, res) else: kwargs[ikey] = input - for res in await self.update('/nodegroups/{0}/{1}'.format( + async for res in self.update('/nodegroups/{0}/{1}'.format( noderange, resource), kwargs): rc = self.handle_results(ikey, rc, res) return rc @@ -667,10 +667,10 @@ def show_attr(attr, requestargs, seenattributes, options, node): return processattr -def printgroupattributes(session, requestargs, showtype, nodetype, noderange, options): +async def printgroupattributes(session, requestargs, showtype, nodetype, noderange, options): exitcode = 0 seenattributes = set([]) - for res in session.read('/{0}/{1}/attributes/{2}'.format(nodetype, noderange, showtype)): + async for res in session.read('/{0}/{1}/attributes/{2}'.format(nodetype, noderange, showtype)): if 'error' in res: sys.stderr.write(res['error'] + '\n') exitcode = 1 From 15670f0ab1af34e4c31e4b601e158610ab13dbea Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 01:54:01 +0200 Subject: [PATCH 5/9] Put the local socket in non-blocking mode in the async client _connect_unix left the socket blocking, while _connect_tls sets a zero timeout, so every loop.sock_recv and sock_sendall against the local socket ran the blocking call inline and stalled the whole event loop. nodeconsole --video showed this most clearly: a power action opens its own session, so the tiles stopped refreshing and keystrokes went unhandled until the BMC finished. asyncio only enforces this in debug mode, where the client failed outright with ValueError: the socket must be non-blocking. The descriptor passing retries in asynctlvdata also assume a non-blocking socket, since they wait for BlockingIOError. --- confluent_client/confluent/asynclient.py | 1 + 1 file changed, 1 insertion(+) diff --git a/confluent_client/confluent/asynclient.py b/confluent_client/confluent/asynclient.py index 1064a2aa..2f29e2e6 100644 --- a/confluent_client/confluent/asynclient.py +++ b/confluent_client/confluent/asynclient.py @@ -396,6 +396,7 @@ class Command(object): self.connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) self.connection.setsockopt(socket.SOL_SOCKET, SO_PASSCRED, 1) self.connection.connect(self.serverloc) + self.connection.setblocking(False) async def _connect_tls(self): server, port = _parseserver(self.serverloc) From 53f1d4a7c27464ac9af3697e8548804860daf2bb Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 01:54:24 +0200 Subject: [PATCH 6/9] Do not block the event loop with time.sleep in the async clients nodediscover's rescan poll and nodeconsole's screenshot refresh both slept with time.sleep inside a coroutine. In nodeconsole --video that stops the input handler and the VNC streaming tasks for the whole interval. --- confluent_client/bin/nodeconsole | 2 +- confluent_client/bin/nodediscover | 3 +-- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/confluent_client/bin/nodeconsole b/confluent_client/bin/nodeconsole index 29e9f204..fe6decc9 100755 --- a/confluent_client/bin/nodeconsole +++ b/confluent_client/bin/nodeconsole @@ -910,7 +910,7 @@ async def do_screenshot(): dorefresh = False else: dorefresh = True - time.sleep(options.interval) + await asyncio.sleep(options.interval) sys.exit(0) async def grab_vncs(urlbynode): diff --git a/confluent_client/bin/nodediscover b/confluent_client/bin/nodediscover index 36fc6b95..6838b911 100755 --- a/confluent_client/bin/nodediscover +++ b/confluent_client/bin/nodediscover @@ -20,7 +20,6 @@ import csv import optparse import os import sys -import time path = os.path.dirname(os.path.realpath(__file__)) path = os.path.realpath(os.path.join(path, '..', 'lib', 'python')) @@ -364,7 +363,7 @@ async def assign_discovery(options, session, needid=True): async def blocking_scan(session): list([x async for x in session.update('/discovery/rescan', {'rescan': 'start'})]) while(list([x async for x in session.read('/discovery/rescan')])[0].get('scanning', False)): - time.sleep(0.5) + await asyncio.sleep(0.5) list([x async for x in session.update('/networking/macs/rescan', {'rescan': 'start'})]) From 18c24effc6a77056534cb6569c17b61838542526 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 06:15:08 +0200 Subject: [PATCH 7/9] Initialize the scan total in nodediscover register register_endpoint primes current but not total, so a first response without a count field goes straight to UnboundLocalError: local variable 'total' referenced before assignment on the elif. Start at zero, which skips the progress line until the server does report a count. --- confluent_client/bin/nodediscover | 1 + 1 file changed, 1 insertion(+) diff --git a/confluent_client/bin/nodediscover b/confluent_client/bin/nodediscover index 6838b911..ea93f298 100755 --- a/confluent_client/bin/nodediscover +++ b/confluent_client/bin/nodediscover @@ -53,6 +53,7 @@ columnmapping = { async def register_endpoint(options, session, addr): neednewline = False current = 0 + total = 0 async for rsp in session.update('/discovery/register', {'addresses': addr}): if 'count' in rsp: total = rsp['count'] From a619b6ed6f4e4c43bd7de1c9e2d4cf5c27f97205 Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 06:15:32 +0200 Subject: [PATCH 8/9] Bound the sessions the nodediscover CSV import opens Replacing the forked children with a gather kept their fan-out: every row of the import file gets a session of its own and they all start at once, so a large file opens a local socket and a server side session task per node simultaneously. Hold a semaphore for the duration of each node's assignment instead, so a finished node's session is dropped before the next one starts. Also build that session once per node rather than once per MAC, and say why the caller's session is not reused, which was self evident while this ran in a forked child. --- confluent_client/bin/nodediscover | 33 +++++++++++++++++++------------ 1 file changed, 20 insertions(+), 13 deletions(-) diff --git a/confluent_client/bin/nodediscover b/confluent_client/bin/nodediscover index ea93f298..533a983e 100755 --- a/confluent_client/bin/nodediscover +++ b/confluent_client/bin/nodediscover @@ -180,6 +180,9 @@ def datum_to_attrib(datum): unique_fields = frozenset(['serial', 'mac', 'uuid']) +# Cap how many nodes hold a discovery session at once while importing +maxconcurrentassign = 128 + async def import_csv(options, session): nodedata = [] unique_data = {} @@ -225,6 +228,7 @@ async def import_csv(options, session): if broken: sys.exit(1) assignments = [] + assignlimit = asyncio.Semaphore(maxconcurrentassign) for datum in nodedata: maclist = await search_record(datum, options, session) datum = datum_to_attrib(datum) @@ -238,27 +242,30 @@ async def import_csv(options, session): print('Defined ' + res['created']) else: print(repr(res)) - assignments.append(assign_macs(maclist, nodename)) + assignments.append(assign_macs(maclist, nodename, assignlimit)) for rcode in await asyncio.gather(*assignments): exitcode |= rcode if exitcode: sys.exit(exitcode) -async def assign_macs(maclist, nodename): +async def assign_macs(maclist, nodename, assignlimit): exitcode = 0 - for mac in maclist: + async with assignlimit: + # A session of our own, since the connection carries one request at a + # time and the caller's is busy defining the remaining nodes mysess = client.Command() - async for res in mysess.update('/discovery/by-mac/{0}'.format(mac), - {'node': nodename}): - if 'error' in res: - sys.stderr.write(res['error'] + '\n') - exitcode |= res.get('errorcode', 1) - continue - elif 'assigned' in res: - print('Discovered ' + res['assigned']) - else: - print(repr(res)) + for mac in maclist: + async for res in mysess.update('/discovery/by-mac/{0}'.format(mac), + {'node': nodename}): + if 'error' in res: + sys.stderr.write(res['error'] + '\n') + exitcode |= res.get('errorcode', 1) + continue + elif 'assigned' in res: + print('Discovered ' + res['assigned']) + else: + print(repr(res)) return exitcode From f8ea1adec7ad4c4d7b5acf9e9d728ba1dca44c3a Mon Sep 17 00:00:00 2001 From: Markus Hilger Date: Mon, 27 Jul 2026 06:15:45 +0200 Subject: [PATCH 9/9] Keep the nodediscover CSV import going past a failed assignment gather propagates the first exception and leaves its siblings running, so a transport level failure against one node ends the import with a traceback while the rest of the batch is cancelled at loop shutdown. The forked children used to contain such a failure to their own node. assign_macs already reports an error response itself, so this is the connection dropping rather than the server refusing the assignment. Collect the exceptions instead, report each one and count it towards the exit code. Schedule the assignments as tasks while doing so, since the plain coroutines are left unawaited if defining a later node raises before the gather is reached. --- confluent_client/bin/nodediscover | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/confluent_client/bin/nodediscover b/confluent_client/bin/nodediscover index 533a983e..4c45a933 100755 --- a/confluent_client/bin/nodediscover +++ b/confluent_client/bin/nodediscover @@ -242,8 +242,12 @@ async def import_csv(options, session): print('Defined ' + res['created']) else: print(repr(res)) - assignments.append(assign_macs(maclist, nodename, assignlimit)) - for rcode in await asyncio.gather(*assignments): + assignments.append( + asyncio.create_task(assign_macs(maclist, nodename, assignlimit))) + for rcode in await asyncio.gather(*assignments, return_exceptions=True): + if isinstance(rcode, BaseException): + sys.stderr.write('Error assigning discovery data: {0}\n'.format(rcode)) + rcode = 1 exitcode |= rcode if exitcode: sys.exit(exitcode)