From ca4b56b2a723c3b04d4e7121b4a720be7a005176 Mon Sep 17 00:00:00 2001 From: Kurt Biery Date: Thu, 16 Jul 2026 10:09:12 -0500 Subject: [PATCH 1/5] Adding AI-assisted script for running run control process manager and unified shell as separate processes. --- scripts/multiprocess_runcontrol_driver.py | 99 +++++++++++++++++++++++ 1 file changed, 99 insertions(+) create mode 100644 scripts/multiprocess_runcontrol_driver.py diff --git a/scripts/multiprocess_runcontrol_driver.py b/scripts/multiprocess_runcontrol_driver.py new file mode 100644 index 0000000..28505a4 --- /dev/null +++ b/scripts/multiprocess_runcontrol_driver.py @@ -0,0 +1,99 @@ +import asyncio +import sys +import time + +async def read_stream(stream, process_name): + """Asynchronously reads lines from a stream and prints them immediately.""" + while True: + line = await stream.readline() + if not line: + break + # Decode and strip line endings + print(f"[{process_name}] {line.decode().rstrip()}", flush=True) + +async def interactive_manager(commands): + processes = {} + tasks = [] + + # 1. Start all interactive subprocesses + for i, cmd in enumerate(commands): + #name = f"Proc-{i+1}" + if i == 0: + name = "pm" + else: + name = "shell" + proc = await asyncio.create_subprocess_exec( + *cmd, + stdin=asyncio.subprocess.PIPE, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.STDOUT + ) + processes[name] = proc + + # 2. Schedule output reading tasks concurrently + tasks.append(asyncio.create_task(read_stream(proc.stdout, name))) + time.sleep(3) + + print(f"Started {len(processes)} processes. Type: ':' (e.g., Proc-1:help)") + print("Type 'exit' to quit everything.") + + # 3. Handle interactive user input from the main terminal + loop = asyncio.get_running_loop() + reader = asyncio.StreamReader() + protocol = asyncio.StreamReaderProtocol(reader) + await loop.connect_read_pipe(lambda: protocol, sys.stdin) + + try: + while True: + user_line = await reader.readline() + if not user_line: + break + + command_text = user_line.decode().strip() + if command_text.lower() == 'exit': + break + + # Parse target process and message (Format: Proc-1:your_command) + if ":" in command_text: + target, msg = command_text.split(":", 1) + target = target.strip() + + if target in processes: + proc = processes[target] + if proc.returncode is None: # Check if still running + proc.stdin.write((msg + "\n").encode()) + await proc.stdin.drain() + print(f"[System] Sent to {target}: {msg}") + else: + print(f"[System] Error: {target} has already exited.") + else: + print(f"[System] Error: Process '{target}' not found.") + else: + print("[System] Invalid format. Use: :") + + except asyncio.CancelledError: + pass + finally: + # 4. Cleanup and terminate remaining processes + print("\n[System] Shutting down processes...") + for name, proc in reversed(processes.items()): + if proc.returncode is None: + proc.terminate() + await proc.wait() + + # Cancel background reading tasks + for task in tasks: + task.cancel() + +if __name__ == "__main__": + # Example using interactive Python shells as subprocesses + # On Windows, replace with appropriate interactive executables (like ['cmd.exe']) + interactive_cmds = [ + ["drunc-process-manager", "ssh-standalone", "50520"], # Launch Interactive Python Instance 1 + ["drunc-unified-shell", "grpc://localhost:50520", "config/daqsystemtest/example-configs.data.xml", "local-1x1-config", "biery-local-test"] # Launch Interactive Python Instance 2 + ] + + try: + asyncio.run(interactive_manager(interactive_cmds)) + except KeyboardInterrupt: + print("\n[System] Exited via Ctrl+C.") From 92356ce99eb179406faf3ba8aa338dc9b0ff5a58 Mon Sep 17 00:00:00 2001 From: Kurt Biery Date: Thu, 16 Jul 2026 10:27:16 -0500 Subject: [PATCH 2/5] added messages for the user to multiprocess_runcontrol_driver.py --- scripts/multiprocess_runcontrol_driver.py | 13 ++++++++++--- 1 file changed, 10 insertions(+), 3 deletions(-) mode change 100644 => 100755 scripts/multiprocess_runcontrol_driver.py diff --git a/scripts/multiprocess_runcontrol_driver.py b/scripts/multiprocess_runcontrol_driver.py old mode 100644 new mode 100755 index 28505a4..6409a95 --- a/scripts/multiprocess_runcontrol_driver.py +++ b/scripts/multiprocess_runcontrol_driver.py @@ -1,3 +1,5 @@ +#!/bin/env python3 + import asyncio import sys import time @@ -22,6 +24,10 @@ async def interactive_manager(commands): name = "pm" else: name = "shell" + time.sleep(2) + print() + print(f"*** Starting \"{cmd[0]}\" with local process name \"{name}\"...") + print() proc = await asyncio.create_subprocess_exec( *cmd, stdin=asyncio.subprocess.PIPE, @@ -32,10 +38,11 @@ async def interactive_manager(commands): # 2. Schedule output reading tasks concurrently tasks.append(asyncio.create_task(read_stream(proc.stdout, name))) - time.sleep(3) - print(f"Started {len(processes)} processes. Type: ':' (e.g., Proc-1:help)") - print("Type 'exit' to quit everything.") + print() + print(f"*** Started {len(processes)} processes. Type: ':' (e.g., shell:help)") + print("*** Type 'exit' to quit everything.") + print() # 3. Handle interactive user input from the main terminal loop = asyncio.get_running_loop() From a065491e75660f22ca4d47d0f78aaeb4b90cd924 Mon Sep 17 00:00:00 2001 From: Kurt Biery Date: Thu, 16 Jul 2026 21:15:30 -0500 Subject: [PATCH 3/5] added pm shell to multiprocess_runcontrol_driver.py --- scripts/multiprocess_runcontrol_driver.py | 21 ++++++++++----------- 1 file changed, 10 insertions(+), 11 deletions(-) diff --git a/scripts/multiprocess_runcontrol_driver.py b/scripts/multiprocess_runcontrol_driver.py index 6409a95..0c8c836 100755 --- a/scripts/multiprocess_runcontrol_driver.py +++ b/scripts/multiprocess_runcontrol_driver.py @@ -3,6 +3,7 @@ import asyncio import sys import time +from daqconf.utils import find_free_port async def read_stream(stream, process_name): """Asynchronously reads lines from a stream and prints them immediately.""" @@ -18,13 +19,8 @@ async def interactive_manager(commands): tasks = [] # 1. Start all interactive subprocesses - for i, cmd in enumerate(commands): - #name = f"Proc-{i+1}" - if i == 0: - name = "pm" - else: - name = "shell" - time.sleep(2) + for cmd in commands: + name = cmd.pop(0) print() print(f"*** Starting \"{cmd[0]}\" with local process name \"{name}\"...") print() @@ -39,6 +35,8 @@ async def interactive_manager(commands): # 2. Schedule output reading tasks concurrently tasks.append(asyncio.create_task(read_stream(proc.stdout, name))) + time.sleep(2) + print() print(f"*** Started {len(processes)} processes. Type: ':' (e.g., shell:help)") print("*** Type 'exit' to quit everything.") @@ -93,11 +91,12 @@ async def interactive_manager(commands): task.cancel() if __name__ == "__main__": - # Example using interactive Python shells as subprocesses - # On Windows, replace with appropriate interactive executables (like ['cmd.exe']) + pm_port = find_free_port(50001, 52000) + interactive_cmds = [ - ["drunc-process-manager", "ssh-standalone", "50520"], # Launch Interactive Python Instance 1 - ["drunc-unified-shell", "grpc://localhost:50520", "config/daqsystemtest/example-configs.data.xml", "local-1x1-config", "biery-local-test"] # Launch Interactive Python Instance 2 + ["pm", "drunc-process-manager", "ssh-standalone", str(pm_port)], # Launch Interactive Python Instance 1 + ["pmshell", "drunc-process-manager-shell", f"grpc://localhost:{pm_port}"], # Launch Interactive Python Instance 2 + ["drunc", "drunc-unified-shell", f"grpc://localhost:{pm_port}", "config/daqsystemtest/example-configs.data.xml", "local-1x1-config", "biery-local-test"] # Launch Interactive Python Instance 3 ] try: From 4c9bf7ca9a248aad03c2ffd9794b8c0768ff2e6a Mon Sep 17 00:00:00 2001 From: Kurt Biery Date: Sat, 18 Jul 2026 15:25:36 -0500 Subject: [PATCH 4/5] Added logic to wait for command completion in multiprocess_runcontrol_driver.py --- scripts/multiprocess_runcontrol_driver.py | 39 +++++++++++++++++++++-- 1 file changed, 37 insertions(+), 2 deletions(-) diff --git a/scripts/multiprocess_runcontrol_driver.py b/scripts/multiprocess_runcontrol_driver.py index 0c8c836..cc09199 100755 --- a/scripts/multiprocess_runcontrol_driver.py +++ b/scripts/multiprocess_runcontrol_driver.py @@ -5,18 +5,31 @@ import time from daqconf.utils import find_free_port -async def read_stream(stream, process_name): +last_msg_time = 0 + +async def read_stream(stream, process_name, completion_event): """Asynchronously reads lines from a stream and prints them immediately.""" + global last_msg_time while True: line = await stream.readline() if not line: break + decoded_line = line.decode().rstrip() + + if "*** COMMAND HAS COMPLETED ***" in decoded_line: + #print("=== Setting the completion event ===", flush=True) + completion_event.set() + continue + # Decode and strip line endings print(f"[{process_name}] {line.decode().rstrip()}", flush=True) + last_msg_time = time.time() async def interactive_manager(commands): processes = {} tasks = [] + command_completion_event = asyncio.Event() + global last_msg_time # 1. Start all interactive subprocesses for cmd in commands: @@ -33,7 +46,7 @@ async def interactive_manager(commands): processes[name] = proc # 2. Schedule output reading tasks concurrently - tasks.append(asyncio.create_task(read_stream(proc.stdout, name))) + tasks.append(asyncio.create_task(read_stream(proc.stdout, name, command_completion_event))) time.sleep(2) @@ -50,6 +63,7 @@ async def interactive_manager(commands): try: while True: + print("\nmprc_drvr> ", end="", flush=True) user_line = await reader.readline() if not user_line: break @@ -66,9 +80,30 @@ async def interactive_manager(commands): if target in processes: proc = processes[target] if proc.returncode is None: # Check if still running + cmd_start_time = time.time() proc.stdin.write((msg + "\n").encode()) await proc.stdin.drain() print(f"[System] Sent to {target}: {msg}") + if "drunc" in target: + proc.stdin.write(("echo '*** COMMAND HAS COMPLETED ***'\n").encode()) + await proc.stdin.drain() + #print(f"[System] Sent to {target}: echo '*** COMMAND HAS COMPLETED ***'") + await command_completion_event.wait() + command_completion_event.clear() + else: + now = time.time() + #print(f"{cmd_start_time} {last_msg_time} {now}", flush=True) + while True: + if last_msg_time <= cmd_start_time: + if now - cmd_start_time > 5: + break + else: + if now - last_msg_time >= 5: + break + #print(f"{cmd_start_time} {last_msg_time} {now}", flush=True) + await asyncio.sleep(0.25) + now = time.time() + #print(f"{cmd_start_time} {last_msg_time} {now}", flush=True) else: print(f"[System] Error: {target} has already exited.") else: From 9b38fcf27c77000fb9ac51c795880959b3d4fc59 Mon Sep 17 00:00:00 2001 From: Kurt Biery Date: Thu, 20 Aug 2026 14:05:07 -0500 Subject: [PATCH 5/5] Added comments to multiprocess_runcontrol_driver.py --- scripts/multiprocess_runcontrol_driver.py | 37 ++++++++++++++++++----- 1 file changed, 29 insertions(+), 8 deletions(-) diff --git a/scripts/multiprocess_runcontrol_driver.py b/scripts/multiprocess_runcontrol_driver.py index cc09199..8c7117f 100755 --- a/scripts/multiprocess_runcontrol_driver.py +++ b/scripts/multiprocess_runcontrol_driver.py @@ -1,5 +1,25 @@ #!/bin/env python3 +# August 2026, KAB: As part of developing functionality to support multiple user-specified +# applications running in our integration tests, I ran some web searches to learn more about +# how to start and receive output from multiple processes in Python. The Python code that +# was suggested in the response was very helpful. +# I modified that sample code to start drunc-unified-shell, drunc-process-manager, and +# drunc-process-manager-shell processes. The modified script was so helpful that I wanted +# to capture it for later use, and this is that script. +# I can imagine this script being used by non-run-control experts to learn about how the +# different RC apps interact, and maybe the script could be used to provide an easy way +# to start up a different set of applications. +# This code is far from production-ready. So, if we ever decide to make it a general-purpose +# tool, we should improve various aspects of it. The integrationtest/async_proc_mgmt +# code (which used this script as a starting point) might be useful when thinking about any +# possible improvements. +# +# The script can by run by typing 'multiprocess_runcontrol_driver.py' with no arguments. +# Once the script has been started, commands can be sent to one of the three processes by +# pre-pending the process nickname to the command (with a colon separator). For example, 'drunc:ps'. +# Typing 'exit' (with no process prefix) will exit the script. + import asyncio import sys import time @@ -17,7 +37,6 @@ async def read_stream(stream, process_name, completion_event): decoded_line = line.decode().rstrip() if "*** COMMAND HAS COMPLETED ***" in decoded_line: - #print("=== Setting the completion event ===", flush=True) completion_event.set() continue @@ -87,12 +106,11 @@ async def interactive_manager(commands): if "drunc" in target: proc.stdin.write(("echo '*** COMMAND HAS COMPLETED ***'\n").encode()) await proc.stdin.drain() - #print(f"[System] Sent to {target}: echo '*** COMMAND HAS COMPLETED ***'") + print(f"[System] Sent to {target}: echo '*** COMMAND HAS COMPLETED ***'") await command_completion_event.wait() command_completion_event.clear() else: now = time.time() - #print(f"{cmd_start_time} {last_msg_time} {now}", flush=True) while True: if last_msg_time <= cmd_start_time: if now - cmd_start_time > 5: @@ -100,10 +118,8 @@ async def interactive_manager(commands): else: if now - last_msg_time >= 5: break - #print(f"{cmd_start_time} {last_msg_time} {now}", flush=True) await asyncio.sleep(0.25) now = time.time() - #print(f"{cmd_start_time} {last_msg_time} {now}", flush=True) else: print(f"[System] Error: {target} has already exited.") else: @@ -128,10 +144,15 @@ async def interactive_manager(commands): if __name__ == "__main__": pm_port = find_free_port(50001, 52000) + # This set of DUNE-DAQ control applications is the first useful one that came to mind. + # It allows testing of process-manager-as-a-service and gives us a way to see how these + # three run control programs interact. Of course, there may be different sets of apps + # that will be useful in the future. At that time, we may want to simply edit the following + # list to have different apps, or we may consider something more dynamic - to be decided. interactive_cmds = [ - ["pm", "drunc-process-manager", "ssh-standalone", str(pm_port)], # Launch Interactive Python Instance 1 - ["pmshell", "drunc-process-manager-shell", f"grpc://localhost:{pm_port}"], # Launch Interactive Python Instance 2 - ["drunc", "drunc-unified-shell", f"grpc://localhost:{pm_port}", "config/daqsystemtest/example-configs.data.xml", "local-1x1-config", "biery-local-test"] # Launch Interactive Python Instance 3 + ["pm", "drunc-process-manager", "ssh-standalone", str(pm_port)], + ["pmshell", "drunc-process-manager-shell", f"grpc://localhost:{pm_port}"], + ["drunc", "drunc-unified-shell", f"grpc://localhost:{pm_port}", "config/daqsystemtest/example-configs.data.xml", "local-1x1-config", "biery-local-test"] ] try: