Integrate cdp_queue into CDP path
All CDP sessions now go through per-node queue (max 2 concurrent, priority levels). DM sends use high priority. Graceful fallback if module unavailable. Session: sidechat/chromebox-ops
This commit is contained in:
@@ -27,6 +27,7 @@ Usage:
|
||||
dm.py thread --from opm --to 646 --target main "message"
|
||||
"""
|
||||
import argparse
|
||||
import os
|
||||
import re
|
||||
import subprocess
|
||||
import time
|
||||
@@ -47,8 +48,12 @@ def log_event(event):
|
||||
with open(LOG_FILE, "a") as f:
|
||||
f.write(json.dumps(event) + "\n")
|
||||
|
||||
def run(cmd, timeout=60):
|
||||
result = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=timeout)
|
||||
def run(cmd, timeout=60, priority=None):
|
||||
# priority: CDP queue priority for the subprocess (high/normal/low).
|
||||
# DM sends are user-facing -> high. Reads stay at default (normal).
|
||||
env = dict(os.environ, CDP_PRIORITY=priority) if priority else None
|
||||
result = subprocess.run(cmd, shell=True, capture_output=True, text=True,
|
||||
timeout=timeout, env=env)
|
||||
return result.stdout.strip()
|
||||
|
||||
def run_full(cmd, timeout=60):
|
||||
@@ -105,7 +110,8 @@ def dm_send(agent, target, message, verify=True, raw=False, to_agent=None):
|
||||
max_retries = 3
|
||||
delivered = False
|
||||
for attempt in range(max_retries):
|
||||
run(f'{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} send "{safe}"')
|
||||
run(f'{NETVM_EXEC} {recipient} -- python3 {API} --account {recipient} send "{safe}"',
|
||||
priority="high")
|
||||
time.sleep(3) # Wait for message to propagate
|
||||
|
||||
# Verify by reading recipient's chat (independent check, not local echo)
|
||||
|
||||
@@ -25,6 +25,11 @@ try:
|
||||
HAS_SIDECHAT_MANAGER = True
|
||||
except ImportError:
|
||||
HAS_SIDECHAT_MANAGER = False
|
||||
try:
|
||||
from cdp_queue import cdp_slot, PRIORITY_HIGH, PRIORITY_NORMAL, PRIORITY_LOW
|
||||
HAS_CDP_QUEUE = True
|
||||
except ImportError:
|
||||
HAS_CDP_QUEUE = False
|
||||
|
||||
|
||||
def _load_accounts():
|
||||
@@ -493,6 +498,19 @@ def main():
|
||||
|
||||
node, cdp_url = ACCOUNTS[args.account]
|
||||
page = get_page(node, cdp_url)
|
||||
|
||||
# CDP operation queue: serialize browser access across processes.
|
||||
# Priority from CDP_PRIORITY env (high/normal/low); default normal.
|
||||
# Falls back to unqueued if cdp_queue is unavailable.
|
||||
import contextlib
|
||||
if HAS_CDP_QUEUE:
|
||||
_prio_name = os.environ.get("CDP_PRIORITY", "normal").lower()
|
||||
_prio = {"high": PRIORITY_HIGH, "low": PRIORITY_LOW}.get(
|
||||
_prio_name, PRIORITY_NORMAL)
|
||||
_slot = cdp_slot(node, priority=_prio)
|
||||
else:
|
||||
_slot = contextlib.nullcontext()
|
||||
with _slot:
|
||||
ws = websocket.create_connection(page['webSocketDebuggerUrl'], timeout=15)
|
||||
|
||||
try:
|
||||
|
||||
Reference in New Issue
Block a user