Merge pull request #187 from openqj/feat-bridge-perf
perf: 桥变化才写、stdin 优先、stream 50ms 轮询
This commit is contained in:
4 files changed
+93
-21
No files matched your search
+10
-1
@@ -430,14 +430,23 @@ def main():
|
||||
if _get_bridge(browser) is not None:
|
||||
browser.set_frame_sink(_frame_queue())
|
||||
reader = _LineReader(sys.stdin.fileno())
|
||||
last_drain = 0.0
|
||||
while True:
|
||||
_drain_remote(browser)
|
||||
lines, drained = reader.poll()
|
||||
for line in lines:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
_handle_line(line, browser)
|
||||
# Pending requests go first: a tool call used to pay a
|
||||
# full drain round of file I/O before executing. Drain
|
||||
# on idle iterations as before, plus a 200ms floor while
|
||||
# busy so a request burst can never starve remote input
|
||||
# (takeover/click) or the state heartbeat for long.
|
||||
now = time.time()
|
||||
if not lines or now - last_drain >= 0.2:
|
||||
_drain_remote(browser)
|
||||
last_drain = now
|
||||
if drained:
|
||||
break
|
||||
if not lines:
|
||||
|
||||
+46
-18
@@ -14,12 +14,12 @@ is four regular files in the worktree, all written atomically
|
||||
(temp file + os.replace) except the append-only input log:
|
||||
|
||||
- ``.browser-remote-state.json``: {url, title, started, holder,
|
||||
streaming, updated_at} -- rewritten by the MCP drain loop on
|
||||
every iteration, so its mtime is the session-liveness signal.
|
||||
The server treats a state file older than STATE_STALE_SECONDS
|
||||
as a dead session (503).
|
||||
streaming, updated_at} -- rewritten when a meaningful field
|
||||
changes or at the heartbeat interval, so its updated_at is the
|
||||
session-liveness signal. The server treats a state file older
|
||||
than STATE_STALE_SECONDS as a dead session (503).
|
||||
- ``.browser-remote-frame.jpg``: latest screencast frame, base64
|
||||
text, rewritten only when a new frame arrives.
|
||||
text, rewritten only when its bytes change.
|
||||
- ``.browser-remote-input.jsonl``: the server appends one JSON
|
||||
object per line. The MCP tracks a byte offset and executes
|
||||
complete lines only (a trailing partial line waits for the
|
||||
@@ -34,10 +34,10 @@ Threading contract (unchanged from the socket version, measured
|
||||
2026-10-30: Playwright sync calls from any thread but the
|
||||
creating one raise "Cannot switch to a different thread"):
|
||||
only the MCP main thread touches the browser. It calls
|
||||
RemoteBridge.poll() on every select() iteration (~50ms), so
|
||||
control/input/frames flow even while the agent is idle, and CDP
|
||||
frame acks keep flowing (no ack = CDP stops after the first
|
||||
frame).
|
||||
RemoteBridge.poll() on every idle main-loop iteration (~50ms),
|
||||
so control/input/frames flow even while the agent is idle, and
|
||||
CDP frame acks keep flowing (no ack = CDP stops after the
|
||||
first frame).
|
||||
|
||||
Control model (unchanged): holder "agent" by default; holder
|
||||
"user" while the console user has taken over, during which
|
||||
@@ -55,6 +55,10 @@ FRAME_NAME = ".browser-remote-frame.jpg"
|
||||
INPUT_NAME = ".browser-remote-input.jsonl"
|
||||
CONTROL_NAME = ".browser-remote-control.json"
|
||||
STATE_STALE_SECONDS = 30
|
||||
# Heartbeat interval for state writes. The server treats state
|
||||
# older than STATE_STALE_SECONDS as dead, so 0.5s is plenty;
|
||||
# writing every ~50ms iteration was pure gofer round-trips.
|
||||
STATE_HEARTBEAT_SECONDS = 0.5
|
||||
|
||||
|
||||
class SimpleQueue:
|
||||
@@ -114,6 +118,9 @@ class RemoteBridge:
|
||||
self._input_offset = 0
|
||||
self._control_seq = None
|
||||
self._frame_written = False
|
||||
self._last_frame = None
|
||||
self._last_state_key = None
|
||||
self._last_state_write = 0.0
|
||||
|
||||
def _path(self, name):
|
||||
return os.path.join(self.workdir, name)
|
||||
@@ -180,7 +187,18 @@ class RemoteBridge:
|
||||
print("playwright_remote: input error: %s" % exc)
|
||||
|
||||
def poll(self, browser=None):
|
||||
"""One drain-loop iteration: control, inputs, frame, state."""
|
||||
"""One drain-loop iteration: control, inputs, frame, state.
|
||||
|
||||
Writes are change-gated. The old code rewrote both files
|
||||
on every ~50ms iteration: ~20 atomic temp+replace
|
||||
round-trips per second through the runsc gofer mount for
|
||||
identical bytes (pump returns the latest frame
|
||||
persistently, and updated_at changes every call). Now the
|
||||
frame file is rewritten only when its bytes change, and
|
||||
the state file only when a meaningful field changes or
|
||||
the heartbeat interval elapsed -- a static page costs
|
||||
~2 writes per second instead of ~40.
|
||||
"""
|
||||
target = browser if browser is not None else self.browser
|
||||
self._poll_control()
|
||||
self._poll_inputs()
|
||||
@@ -188,20 +206,30 @@ class RemoteBridge:
|
||||
frame = target.pump_screencast()
|
||||
except Exception: # noqa: BLE001 - screencast best-effort
|
||||
frame = ""
|
||||
if frame:
|
||||
if frame and frame != self._last_frame:
|
||||
try:
|
||||
_atomic_write(self._path(FRAME_NAME),
|
||||
frame.encode("utf-8"))
|
||||
self._last_frame = frame
|
||||
self._frame_written = True
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
_atomic_write(
|
||||
self._path(STATE_NAME),
|
||||
json.dumps(self.state_dict(),
|
||||
ensure_ascii=False).encode("utf-8"))
|
||||
except OSError:
|
||||
pass
|
||||
state = self.state_dict()
|
||||
key = (state.get("url"), state.get("title"),
|
||||
state.get("started"), state.get("holder"),
|
||||
state.get("streaming"))
|
||||
now = state["updated_at"]
|
||||
if key != self._last_state_key or \
|
||||
now - self._last_state_write >= STATE_HEARTBEAT_SECONDS:
|
||||
try:
|
||||
_atomic_write(
|
||||
self._path(STATE_NAME),
|
||||
json.dumps(state,
|
||||
ensure_ascii=False).encode("utf-8"))
|
||||
self._last_state_key = key
|
||||
self._last_state_write = now
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def bridge_paths(workdir):
|
||||
|
||||
+2
-2
@@ -5318,7 +5318,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
|
||||
def _browser_stream(self, user_id, task_id):
|
||||
"""GET /tasks/{id}/browser/stream: SSE from bridge files.
|
||||
|
||||
Polls the frame file every 200ms and emits a data line
|
||||
Polls the frame file every 50ms and emits a data line
|
||||
whenever its bytes change (plus a heartbeat with the
|
||||
current holder at least every 5s). Ends the stream when
|
||||
the state file goes stale, so a dead session reads as a
|
||||
@@ -5364,7 +5364,7 @@ class CloudControlPlaneHandler(BaseHTTPRequestHandler):
|
||||
self.wfile.write(
|
||||
("data: %s\n\n" % payload).encode("utf-8"))
|
||||
self.wfile.flush()
|
||||
time.sleep(0.2)
|
||||
time.sleep(0.05)
|
||||
except (BrokenPipeError, ConnectionResetError, OSError):
|
||||
pass
|
||||
|
||||
|
||||
@@ -116,6 +116,41 @@ class BridgeTestCase(unittest.TestCase):
|
||||
"rb") as fh:
|
||||
self.assertEqual(fh.read(), b"FRAMEBYTES")
|
||||
|
||||
def test_writes_are_change_gated(self):
|
||||
"""Unchanged polls must not rewrite the bridge files.
|
||||
|
||||
The old code rewrote both files every ~50ms iteration
|
||||
(~40 atomic temp+replace round-trips per second through
|
||||
the runsc gofer mount for identical bytes). Now only
|
||||
changed bytes and the heartbeat get written.
|
||||
"""
|
||||
state_path = os.path.join(
|
||||
self.tmp.name, ".browser-remote-state.json")
|
||||
frame_path = os.path.join(
|
||||
self.tmp.name, ".browser-remote-frame.jpg")
|
||||
self.browser.frame = "FRAMEBYTES"
|
||||
self.bridge.poll()
|
||||
old = time.time() - 100
|
||||
os.utime(state_path, (old, old))
|
||||
os.utime(frame_path, (old, old))
|
||||
# Unchanged second poll: neither file rewritten.
|
||||
self.bridge.poll()
|
||||
self.assertEqual(os.stat(state_path).st_mtime, old)
|
||||
self.assertEqual(os.stat(frame_path).st_mtime, old)
|
||||
# Changed frame: rewritten at once.
|
||||
self.browser.frame = "NEWBYTES"
|
||||
self.bridge.poll()
|
||||
with open(frame_path, "rb") as fh:
|
||||
self.assertEqual(fh.read(), b"NEWBYTES")
|
||||
# Changed holder: state rewritten at once, bypassing
|
||||
# the heartbeat throttle.
|
||||
os.utime(state_path, (old, old))
|
||||
write_control(self.tmp.name, "user")
|
||||
self.bridge.poll()
|
||||
state, err = read_state(self.tmp.name)
|
||||
self.assertIsNone(err)
|
||||
self.assertEqual(state["holder"], "user")
|
||||
|
||||
def test_read_state_missing_and_stale(self):
|
||||
state, err = read_state(self.tmp.name)
|
||||
self.assertEqual(err[0], 503)
|
||||
|
||||
Reference in new issue
Block a user