mirror of
https://github.com/daijro/camoufox.git
synced 2026-09-11 08:00:51 +00:00
fix(virtdisplay): atomically claim X11 display via Xvfb -displayfd (#597)
Replace the userspace lock-file scan + random-jitter retry loop with Xvfb's own -displayfd mechanism. Xvfb scans up from :0 and atomically binds the first free X11 socket (kernel-mediated, no userspace race), then writes the chosen display number back through an inherited pipe. This eliminates the duplicate-display race that occurred when many camoufox processes started concurrently and all observed the same set of free display numbers before any of them bound. Adds a 10s read timeout on the displayfd pipe so a hung Xvfb fails fast instead of blocking forever, and adds tests covering single launch, idempotent get(), 50 concurrent reservations with uniqueness, and post-kill display reuse.
This commit is contained in:
@@ -0,0 +1,131 @@
|
||||
"""
|
||||
Tests for camoufox.virtdisplay.
|
||||
|
||||
Mirrors camoufox-js/test/virtdisplay.test.ts.
|
||||
|
||||
Run with:
|
||||
cd pythonlib && python -m pytest tests/test_virtdisplay.py -v
|
||||
|
||||
VIRTDISPLAY_TEST_N controls the concurrent-launch count. Default is kept
|
||||
low so the test passes on any developer box; set to e.g. 1000 to exercise
|
||||
real scaling. At high N you will need `ulimit -n` headroom -- each Xvfb
|
||||
takes one X11 socket plus our -displayfd pipe.
|
||||
"""
|
||||
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
import time
|
||||
from typing import List, Set
|
||||
|
||||
import pytest
|
||||
|
||||
# Make `import camoufox` resolve to the in-tree pythonlib without an install.
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
|
||||
|
||||
from camoufox.virtdisplay import VirtualDisplay # noqa: E402
|
||||
|
||||
DISPLAY_RE = re.compile(r"^:\d+$")
|
||||
N = int(os.environ.get("VIRTDISPLAY_TEST_N", "50"))
|
||||
|
||||
pytestmark = pytest.mark.skipif(
|
||||
sys.platform != "linux", reason="VirtualDisplay is Linux-only"
|
||||
)
|
||||
|
||||
|
||||
# Track every VirtualDisplay we spawn so the fixture can guarantee
|
||||
# cleanup even if an assertion fails mid-test.
|
||||
@pytest.fixture
|
||||
def tracked() -> List[VirtualDisplay]:
|
||||
items: List[VirtualDisplay] = []
|
||||
yield items
|
||||
for vd in items:
|
||||
try:
|
||||
vd.kill()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _track(items: List[VirtualDisplay], vd: VirtualDisplay) -> VirtualDisplay:
|
||||
items.append(vd)
|
||||
return vd
|
||||
|
||||
|
||||
def _wait_for_exit(vd: VirtualDisplay, timeout_s: float = 5.0) -> None:
|
||||
deadline = time.monotonic() + timeout_s
|
||||
while time.monotonic() < deadline:
|
||||
if vd.proc is None or vd.proc.poll() is not None:
|
||||
return
|
||||
time.sleep(0.025)
|
||||
|
||||
|
||||
def test_single_launch_returns_valid_display_and_kill_terminates_xvfb(tracked):
|
||||
vd = _track(tracked, VirtualDisplay())
|
||||
display = vd.get()
|
||||
assert DISPLAY_RE.match(display), display
|
||||
assert vd.proc is not None
|
||||
assert vd.proc.poll() is None # alive
|
||||
|
||||
vd.kill()
|
||||
_wait_for_exit(vd)
|
||||
assert vd.proc.poll() is not None # exited
|
||||
|
||||
|
||||
def test_get_is_idempotent_within_one_virtual_display(tracked):
|
||||
vd = _track(tracked, VirtualDisplay())
|
||||
a = vd.get()
|
||||
b = vd.get()
|
||||
assert a == b
|
||||
|
||||
|
||||
def test_concurrent_reservations_all_get_unique_displays(tracked):
|
||||
# Every VirtualDisplay spawns its own Xvfb. Each Xvfb scans up from :0
|
||||
# and atomically claims the first free X11 socket (kernel-mediated
|
||||
# bind, no userspace race). -displayfd reports the chosen number back
|
||||
# to us. A duplicate here would mean we mis-parsed or mis-routed the
|
||||
# displayfd output, or two Xvfbs somehow bound the same socket.
|
||||
vds = [_track(tracked, VirtualDisplay()) for _ in range(N)]
|
||||
|
||||
displays = [vd.get() for vd in vds]
|
||||
|
||||
for d in displays:
|
||||
assert DISPLAY_RE.match(d), d
|
||||
|
||||
unique: Set[str] = set(displays)
|
||||
assert len(unique) == len(displays), (
|
||||
f"duplicate displays: {len(displays) - len(unique)} of {len(displays)}"
|
||||
)
|
||||
|
||||
# Every Xvfb is alive.
|
||||
for vd in vds:
|
||||
assert vd.proc is not None and vd.proc.poll() is None
|
||||
|
||||
# Tear them all down and confirm every Xvfb actually exited -- no
|
||||
# leaked processes.
|
||||
for vd in vds:
|
||||
vd.kill()
|
||||
for vd in vds:
|
||||
_wait_for_exit(vd)
|
||||
for vd in vds:
|
||||
assert vd.proc.poll() is not None
|
||||
|
||||
|
||||
def test_released_display_numbers_can_be_reused_on_the_next_launch(tracked):
|
||||
a = _track(tracked, VirtualDisplay())
|
||||
a_display = a.get()
|
||||
|
||||
a.kill()
|
||||
_wait_for_exit(a)
|
||||
|
||||
# Spawning a new Xvfb after release must succeed. The new display
|
||||
# number may or may not equal a_display -- Xvfb's allocation order is
|
||||
# its concern -- but we must get *some* display.
|
||||
b = _track(tracked, VirtualDisplay())
|
||||
b_display = b.get()
|
||||
assert DISPLAY_RE.match(b_display), b_display
|
||||
|
||||
b.kill()
|
||||
_wait_for_exit(b)
|
||||
|
||||
# Sanity: a_display was a valid form too.
|
||||
assert DISPLAY_RE.match(a_display), a_display
|
||||
Reference in New Issue
Block a user