multiprocess_runcontrol_driver Namespace Reference

Functions
	read_stream (stream, process_name, completion_event)
	interactive_manager (commands)

Variables
int	last_msg_time = 0
	pm_port = find_free_port(50001, 52000)
list	interactive_cmds

Function Documentation

interactive_manager()

interactive_manager

(

commands

)

Definition at line 47 of file multiprocess_runcontrol_driver.py.

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:
        name = cmd.pop(0)
        print()
        print(f"*** Starting \"{cmd[0]}\" with local process name \"{name}\"...")
        print()
        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, command_completion_event)))
 
        time.sleep(2)
 
    print()
    print(f"*** Started {len(processes)} processes. Type: '<process_name>:<input>' (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()
    reader = asyncio.StreamReader()
    protocol = asyncio.StreamReaderProtocol(reader)
    await loop.connect_read_pipe(lambda: protocol, sys.stdin)
 
    try:
        while True:
            print("\nmprc_drvr> ", end="", flush=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
                        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()
                            while True:
                                if last_msg_time <= cmd_start_time:
                                    if now - cmd_start_time > 5:
                                        break
                                else:
                                    if now - last_msg_time >= 5:
                                        break
                                await asyncio.sleep(0.25)
                                now = time.time()
                    else:
                        print(f"[System] Error: {target} has already exited.")
                else:
                    print(f"[System] Error: Process '{target}' not found.")
            else:
                print("[System] Invalid format. Use: <process_name>:<command>")
 
    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()
 

read_stream()

read_stream	(		stream,
			process_name,
			completion_event )

Asynchronously reads lines from a stream and prints them immediately.

Definition at line 30 of file multiprocess_runcontrol_driver.py.

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:
            completion_event.set()
            continue
 
        # Decode and strip line endings
        print(f"[{process_name}] {line.decode().rstrip()}", flush=True)
        last_msg_time = time.time()
 

Variable Documentation

interactive_cmds

list multiprocess_runcontrol_driver.interactive_cmds

Initial value:

=  [
        ["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"]
    ]

Definition at line 152 of file multiprocess_runcontrol_driver.py.

last_msg_time

int multiprocess_runcontrol_driver.last_msg_time = 0

Definition at line 28 of file multiprocess_runcontrol_driver.py.

pm_port

multiprocess_runcontrol_driver.pm_port = find_free_port(50001, 52000)

Definition at line 145 of file multiprocess_runcontrol_driver.py.