mirror of
https://github.com/X11Libre/xserver.git
synced 2026-11-15 05:23:40 +00:00
Each test starts its own X server via -displayfd, and pytest-xdist runs
several workers (meson several suites) in parallel. Servers scanning for a
free display number at the same time can both end up claiming the same one:
the losing test's client then connects to the winner's server, and when the
winner is torn down at the end of its own test the loser's connection dies
mid-test. It surfaces as an opaque
BrokenPipeError: [Errno 32] Broken pipe
test/pyxtest/xclient.py:270: in send_request / self.sock.sendall(data)
in an unrelated test, which looks like a flake and hides the cause.
Hold a cross-process lock for the short display-allocation window (Popen
plus reading the display number). Once a server has announced its display
it owns the socket and the lock file, so later servers skip that number and
that window is the only place the race can happen.
As a safety net, record the server PID that owns the socket (SO_PEERCRED)
and fail the xclient fixture with a clear message when it is not the server
the fixture started, instead of leaving a confusing BrokenPipeError.
Measured with a --buildtype=debug build and 4 pytest-xdist workers, 40
parallel pytest sessions of test_xi.py: before, 7 sessions failed with
"connected to PID X instead of the test server" and 5 with
BrokenPipeError; after, 0 of 40, plus 0 errors over 8 full test/pyxtest
runs.
Signed-off-by: Enrico Weigelt, metux IT consult <info@metux.net>
601 lines
18 KiB
Python
601 lines
18 KiB
Python
# SPDX-License-Identifier: MIT
|
|
#
|
|
# X11 client connection utilities
|
|
|
|
import select
|
|
import socket
|
|
import struct
|
|
import time
|
|
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
from typing import Protocol, Self, runtime_checkable
|
|
|
|
from proto import xkb
|
|
from proto.bigrequests import BigRequestsEnableRequest
|
|
from proto.x11 import (
|
|
ChangeKeyboardMappingRequest,
|
|
CreatePixmapRequest,
|
|
CreateWindowRequest,
|
|
InternAtomRequest,
|
|
QueryExtensionRequest,
|
|
)
|
|
|
|
|
|
class X11ConnectionError(Exception):
|
|
"""Raised when the X11 connection fails."""
|
|
|
|
|
|
@dataclass
|
|
class XExtensionData:
|
|
"""Cached result of a QueryExtension reply."""
|
|
|
|
opcode: int
|
|
first_event: int
|
|
first_error: int
|
|
|
|
|
|
@runtime_checkable
|
|
class X11Request(Protocol):
|
|
"""Protocol for X11 request objects that can be serialized to wire format.
|
|
|
|
All ``@dataclass`` request builders in ``proto/`` satisfy this protocol.
|
|
"""
|
|
|
|
def to_bytes(self, byte_order: str = "<") -> bytes: ...
|
|
|
|
|
|
class Extension(StrEnum):
|
|
"""X11 extension wire names as used in QueryExtension requests."""
|
|
|
|
BIG_REQUESTS = "BIG-REQUESTS"
|
|
COMPOSITE = "Composite"
|
|
DAMAGE = "DAMAGE"
|
|
DBE = "DOUBLE-BUFFER"
|
|
DPMS = "DPMS"
|
|
DRI2 = "DRI2"
|
|
DRI3 = "DRI3"
|
|
GENERIC_EVENT = "Generic Event Extension"
|
|
GLX = "GLX"
|
|
MIT_SCREEN_SAVER = "MIT-SCREEN-SAVER"
|
|
MIT_SHM = "MIT-SHM"
|
|
PRESENT = "Present"
|
|
RANDR = "RANDR"
|
|
RECORD = "RECORD"
|
|
RENDER = "RENDER"
|
|
SECURITY = "SECURITY"
|
|
SHAPE = "SHAPE"
|
|
SYNC = "SYNC"
|
|
XC_MISC = "XC-MISC"
|
|
XF86DGA = "XFree86-DGA"
|
|
XF86VIDMODE = "XFree86-VidModeExtension"
|
|
XFIXES = "XFIXES"
|
|
XI = "XInputExtension"
|
|
XRES = "X-Resource"
|
|
XINERAMA = "XINERAMA"
|
|
XKB = "XKEYBOARD"
|
|
XTEST = "XTEST"
|
|
XVIDEO = "XVideo"
|
|
XVIDEO_MC = "XVideo-MotionCompensation"
|
|
|
|
|
|
# X11 core protocol error codes -- canonical definitions are in proto/x11.py,
|
|
# re-exported here for backward compatibility.
|
|
from proto.x11 import ( # noqa: F401
|
|
BadAccess,
|
|
BadAlloc,
|
|
BadAtom,
|
|
BadColor,
|
|
BadCursor,
|
|
BadDrawable,
|
|
BadFont,
|
|
BadGC,
|
|
BadIDChoice,
|
|
BadImplementation,
|
|
BadLength,
|
|
BadMatch,
|
|
BadName,
|
|
BadPixmap,
|
|
BadRequest,
|
|
BadValue,
|
|
BadWindow,
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class X11Error:
|
|
"""An X11 error reply from the server."""
|
|
|
|
response_type: int
|
|
error_code: int
|
|
sequence: int
|
|
resource_id: int
|
|
minor_code: int
|
|
major_code: int
|
|
|
|
@classmethod
|
|
def from_data(cls, data: bytes, byte_order: str = "<") -> "X11Error":
|
|
if len(data) < 32:
|
|
data = data + b"\x00" * (32 - len(data))
|
|
response_type, error_code, sequence, resource_id, minor_code, major_code = (
|
|
struct.unpack_from(f"{byte_order}BBHIHB", data)
|
|
)
|
|
return cls(
|
|
response_type, error_code, sequence, resource_id, minor_code, major_code
|
|
)
|
|
|
|
def __repr__(self):
|
|
return (
|
|
f"<X11Error code={self.error_code} seq={self.sequence} "
|
|
f"major={self.major_code} minor={self.minor_code}>"
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class X11Reply:
|
|
"""An X11 reply from the server."""
|
|
|
|
data: bytes
|
|
response_type: int
|
|
sequence: int
|
|
length: int
|
|
|
|
@classmethod
|
|
def from_data(cls, data: bytes, byte_order: str = "<") -> "X11Reply":
|
|
if len(data) >= 8:
|
|
response_type = data[0]
|
|
sequence = struct.unpack_from(f"{byte_order}H", data, 2)[0]
|
|
length = struct.unpack_from(f"{byte_order}I", data, 4)[0]
|
|
else:
|
|
response_type = sequence = length = 0
|
|
return cls(data, response_type, sequence, length)
|
|
|
|
def __repr__(self):
|
|
return (
|
|
f"<X11Reply seq={self.sequence} "
|
|
f"extra_len={self.length} total={len(self.data)}>"
|
|
)
|
|
|
|
|
|
class RawX11Connection:
|
|
"""
|
|
Minimal X11 connection for sending raw (possibly malformed) requests.
|
|
|
|
Set swapped=True for testing byte-swap code paths (SProcXxx).
|
|
"""
|
|
|
|
def __init__(self, display_num, swapped=False):
|
|
self.display_num = display_num
|
|
self.swapped = swapped
|
|
self._byte_order = ">" if swapped else "<"
|
|
self.sock = None
|
|
self.seq = 0
|
|
self.root_window = 0
|
|
self.root_visual = 0
|
|
self.root_depth = 0
|
|
self._resource_id_base = 0
|
|
self._resource_id_mask = 0
|
|
self._next_resource_id = 0
|
|
self._extensions = {}
|
|
self._connect()
|
|
|
|
def _connect(self):
|
|
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self._peer_pid = None
|
|
path = f"/tmp/.X11-unix/X{self.display_num}"
|
|
try:
|
|
self.sock.connect(path)
|
|
except (ConnectionRefusedError, FileNotFoundError) as e:
|
|
self.sock.close()
|
|
self.sock = None
|
|
raise X11ConnectionError(f"Cannot connect to {path}: {e}")
|
|
# Record who actually owns this socket. Under concurrent tests the
|
|
# display socket can be taken over by another server, in which case the
|
|
# client ends up talking to the wrong server and sees its connection
|
|
# die mid-test (BrokenPipeError).
|
|
try:
|
|
import struct as _struct
|
|
cred = self.sock.getsockopt(
|
|
socket.SOL_SOCKET, socket.SO_PEERCRED, _struct.calcsize("iii")
|
|
)
|
|
self._peer_pid, _, _ = _struct.unpack("iii", cred)
|
|
except OSError:
|
|
self._peer_pid = None
|
|
try:
|
|
self._handshake()
|
|
except Exception:
|
|
self.sock.close()
|
|
self.sock = None
|
|
raise
|
|
|
|
def _handshake(self):
|
|
byte_order = 0x42 if self.swapped else 0x6C
|
|
bo = self._byte_order
|
|
setup = struct.pack(f"{bo}BxHHHHxx", byte_order, 11, 0, 0, 0)
|
|
assert self.sock
|
|
self.sock.sendall(setup)
|
|
|
|
header = self._recv_exact(8)
|
|
status = header[0]
|
|
|
|
if status == 0:
|
|
reason_len = header[1]
|
|
extra_len = struct.unpack_from(f"{bo}H", header, 6)[0]
|
|
extra = self._recv_exact(extra_len * 4)
|
|
reason = extra[:reason_len].decode("ascii", errors="replace")
|
|
raise X11ConnectionError(f"X server refused connection: {reason}")
|
|
|
|
if status == 2:
|
|
raise X11ConnectionError("X server requires authentication")
|
|
if status != 1:
|
|
raise X11ConnectionError(f"Unexpected setup status: {status}")
|
|
|
|
extra_len = struct.unpack_from(f"{bo}H", header, 6)[0]
|
|
data = self._recv_exact(extra_len * 4)
|
|
|
|
(_, res_base, res_mask, _, vendor_len, _, num_screens, num_formats) = (
|
|
struct.unpack_from(f"{bo}IIIIHH BB", data, 0)
|
|
)
|
|
|
|
self._resource_id_base = res_base
|
|
self._resource_id_mask = res_mask
|
|
self._next_resource_id = 1
|
|
|
|
offset = 32 + ((vendor_len + 3) & ~3) + num_formats * 8
|
|
|
|
if num_screens > 0 and offset + 40 <= len(data):
|
|
(
|
|
self.root_window,
|
|
_,
|
|
_,
|
|
_,
|
|
_,
|
|
_,
|
|
_,
|
|
_,
|
|
_,
|
|
_,
|
|
_,
|
|
self.root_visual,
|
|
_,
|
|
_,
|
|
self.root_depth,
|
|
) = struct.unpack_from(f"{bo}IIIIIHHHHHHI BBB", data, offset)
|
|
|
|
# --- Core protocol helpers ---
|
|
|
|
def alloc_id(self) -> int:
|
|
xid = self._resource_id_base | self._next_resource_id
|
|
self._next_resource_id += 1
|
|
return xid
|
|
|
|
def send_request(self, data: bytes | X11Request) -> int:
|
|
"""Send a raw or structured X11 request.
|
|
|
|
*data* may be a ``bytes`` object (sent as-is) or any object that
|
|
implements :class:`X11Request` (i.e. has a ``to_bytes(byte_order)``
|
|
method). In the latter case ``to_bytes`` is called automatically
|
|
with the connection's byte order.
|
|
"""
|
|
if isinstance(data, X11Request):
|
|
data = data.to_bytes(self._byte_order)
|
|
self.seq += 1
|
|
assert self.sock
|
|
self.sock.sendall(data)
|
|
return self.seq
|
|
|
|
def recv_response(self, timeout: float = 5.0) -> X11Error | X11Reply | None:
|
|
ready = select.select([self.sock], [], [], timeout)
|
|
if not ready[0]:
|
|
return None
|
|
try:
|
|
header = self._recv_exact(32, timeout=timeout)
|
|
except (ConnectionResetError, BrokenPipeError, OSError, X11ConnectionError):
|
|
return None
|
|
if not header:
|
|
return None
|
|
|
|
rtype = header[0]
|
|
bo = self._byte_order
|
|
if rtype == 0:
|
|
return X11Error.from_data(header, bo)
|
|
elif rtype == 1:
|
|
extra_len = struct.unpack_from(f"{bo}I", header, 4)[0]
|
|
if extra_len > 0:
|
|
try:
|
|
extra = self._recv_exact(extra_len * 4, timeout=timeout)
|
|
return X11Reply.from_data(header + extra, bo)
|
|
except (
|
|
ConnectionResetError,
|
|
BrokenPipeError,
|
|
OSError,
|
|
X11ConnectionError,
|
|
):
|
|
return X11Reply.from_data(header, bo)
|
|
return X11Reply.from_data(header, bo)
|
|
else:
|
|
return X11Reply.from_data(header, bo)
|
|
|
|
def flush_responses(self, timeout: float = 0.5) -> list[X11Error | X11Reply]:
|
|
responses: list[X11Error | X11Reply] = []
|
|
while True:
|
|
resp = self.recv_response(timeout=timeout)
|
|
if resp is None:
|
|
break
|
|
responses.append(resp)
|
|
return responses
|
|
|
|
def is_connected(self) -> bool:
|
|
assert self.sock
|
|
try:
|
|
ready = select.select([self.sock], [], [], 0)
|
|
if ready[0]:
|
|
data = self.sock.recv(1, socket.MSG_PEEK)
|
|
return len(data) > 0
|
|
return True
|
|
except (ConnectionResetError, BrokenPipeError, OSError):
|
|
return False
|
|
|
|
def wait_for_disconnect(self, timeout: float = 5.0) -> bool:
|
|
assert self.sock
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
break
|
|
ready = select.select([self.sock], [], [], min(remaining, 0.5))
|
|
if ready[0]:
|
|
try:
|
|
data = self.sock.recv(4096)
|
|
if not data:
|
|
return True
|
|
except (ConnectionResetError, BrokenPipeError, OSError):
|
|
return True
|
|
return False
|
|
|
|
# --- Extension negotiation ---
|
|
|
|
def query_extension(self, name: str) -> XExtensionData | None:
|
|
if name in self._extensions:
|
|
return self._extensions[name]
|
|
|
|
req = QueryExtensionRequest(name=name)
|
|
self.send_request(req)
|
|
resp = self.recv_response(timeout=5.0)
|
|
if resp is None or isinstance(resp, X11Error):
|
|
return None
|
|
|
|
if len(resp.data) >= 12:
|
|
present, major, first_event, first_error = struct.unpack_from(
|
|
"BBBB", resp.data, 8
|
|
)
|
|
if present:
|
|
result = XExtensionData(major, first_event, first_error)
|
|
self._extensions[name] = result
|
|
return result
|
|
return None
|
|
|
|
def xkb_use_extension(self, major_version: int = 1, minor_version: int = 0) -> int:
|
|
ext = self.query_extension(Extension.XKB)
|
|
if not ext:
|
|
raise X11ConnectionError("XKB extension not available")
|
|
|
|
req = xkb.UseExtensionRequest(
|
|
opcode=ext.opcode,
|
|
major=major_version,
|
|
minor=minor_version,
|
|
)
|
|
self.send_request(req)
|
|
|
|
resp = self.recv_response(timeout=5.0)
|
|
if isinstance(resp, X11Error):
|
|
raise X11ConnectionError(f"XkbUseExtension failed: error {resp.error_code}")
|
|
return ext.opcode
|
|
|
|
def enable_big_requests(self) -> int:
|
|
ext = self.query_extension(Extension.BIG_REQUESTS)
|
|
if not ext:
|
|
raise X11ConnectionError("BIG-REQUESTS not available")
|
|
|
|
req = BigRequestsEnableRequest(opcode=ext.opcode)
|
|
self.send_request(req)
|
|
|
|
resp = self.recv_response(timeout=5.0)
|
|
if isinstance(resp, X11Error):
|
|
raise X11ConnectionError("BigRequestsEnable failed")
|
|
if resp and len(resp.data) >= 12:
|
|
return struct.unpack_from(f"{self._byte_order}I", resp.data, 8)[0]
|
|
return 0
|
|
|
|
# --- Resource creation helpers ---
|
|
|
|
def create_window(
|
|
self,
|
|
width: int = 100,
|
|
height: int = 100,
|
|
depth: int | None = None,
|
|
parent: int | None = None,
|
|
x: int = 0,
|
|
y: int = 0,
|
|
) -> int:
|
|
wid = self.alloc_id()
|
|
if parent is None:
|
|
parent = self.root_window
|
|
if depth is None:
|
|
depth = self.root_depth
|
|
|
|
assert parent is not None
|
|
assert depth is not None
|
|
|
|
req = CreateWindowRequest(
|
|
wid=wid,
|
|
parent=parent,
|
|
x=x,
|
|
y=y,
|
|
width=width,
|
|
height=height,
|
|
depth=depth,
|
|
)
|
|
self.send_request(req)
|
|
self.flush_responses(timeout=0.2)
|
|
return wid
|
|
|
|
def create_pixmap(
|
|
self,
|
|
width: int = 100,
|
|
height: int = 100,
|
|
depth: int | None = None,
|
|
drawable: int | None = None,
|
|
) -> int:
|
|
pid = self.alloc_id()
|
|
if drawable is None:
|
|
drawable = self.root_window
|
|
if depth is None:
|
|
depth = self.root_depth
|
|
|
|
assert drawable is not None
|
|
assert depth is not None
|
|
|
|
req = CreatePixmapRequest(
|
|
pid=pid,
|
|
drawable=drawable,
|
|
width=width,
|
|
height=height,
|
|
depth=depth,
|
|
)
|
|
self.send_request(req)
|
|
self.flush_responses(timeout=0.2)
|
|
return pid
|
|
|
|
def intern_atom(self, name: str, only_if_exists: bool = False) -> int:
|
|
req = InternAtomRequest(name=name, only_if_exists=only_if_exists)
|
|
self.send_request(req)
|
|
|
|
resp = self.recv_response(timeout=5.0)
|
|
if isinstance(resp, X11Error) or resp is None:
|
|
return 0
|
|
if len(resp.data) >= 12:
|
|
return struct.unpack_from(f"{self._byte_order}I", resp.data, 8)[0]
|
|
return 0
|
|
|
|
def xkb_get_map(
|
|
self, opcode: int, full: int = 0, partial: int = 0, **kwargs
|
|
) -> xkb.GetMapReply | None:
|
|
"""Send XkbGetMap and return the parsed reply."""
|
|
req = xkb.GetMapRequest(opcode=opcode, full=full, partial=partial, **kwargs)
|
|
self.send_request(req.to_bytes(self._byte_order))
|
|
resp = self.recv_response(timeout=5.0)
|
|
if isinstance(resp, X11Error) or resp is None:
|
|
return None
|
|
# resp.data is the full reply: 32 standard header + extra
|
|
header_data = resp.data[:32]
|
|
extra_data = resp.data[32:]
|
|
return xkb.GetMapReply.from_bytes(header_data, extra_data)
|
|
|
|
def change_keyboard_mapping(
|
|
self,
|
|
first_keycode: int,
|
|
keysyms_per_keycode: int,
|
|
keycodes: int = 1,
|
|
keysyms: list[int] | None = None,
|
|
) -> None:
|
|
"""Send ChangeKeyboardMapping request."""
|
|
req = ChangeKeyboardMappingRequest(
|
|
first_keycode=first_keycode,
|
|
keysyms_per_keycode=keysyms_per_keycode,
|
|
keycodes=keycodes,
|
|
keysyms=keysyms,
|
|
)
|
|
self.send_request(req.to_bytes(self._byte_order))
|
|
|
|
def get_fd(self) -> int:
|
|
assert self.sock
|
|
return self.sock.fileno()
|
|
|
|
def close(self) -> None:
|
|
if self.sock:
|
|
try:
|
|
self.sock.close()
|
|
except OSError:
|
|
pass
|
|
self.sock = None
|
|
|
|
def _recv_exact(self, nbytes: int, timeout: float = 10.0) -> bytes:
|
|
assert self.sock
|
|
data = b""
|
|
deadline = time.monotonic() + timeout
|
|
while len(data) < nbytes:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise X11ConnectionError(
|
|
f"Timeout reading ({len(data)}/{nbytes} bytes)"
|
|
)
|
|
ready = select.select([self.sock], [], [], min(remaining, 1.0))
|
|
if ready[0]:
|
|
chunk = self.sock.recv(nbytes - len(data))
|
|
if not chunk:
|
|
raise X11ConnectionError(
|
|
f"Connection closed ({len(data)}/{nbytes} bytes)"
|
|
)
|
|
data += chunk
|
|
return data
|
|
|
|
def __enter__(self) -> Self:
|
|
return self
|
|
|
|
def __exit__(self, *args) -> bool:
|
|
self.close()
|
|
return False
|
|
|
|
def __del__(self) -> None:
|
|
self.close()
|
|
|
|
|
|
class XlibConnection:
|
|
"""python-xlib based connection for higher-level X operations."""
|
|
|
|
def __init__(self, display_num):
|
|
from Xlib import display as xlib_display
|
|
|
|
self.display_num = display_num
|
|
self.display = xlib_display.Display(f":{display_num}")
|
|
self.screen = self.display.screen()
|
|
self.root = self.screen.root
|
|
|
|
def get_fd(self):
|
|
return self.display.fileno()
|
|
|
|
def create_window(self, width=100, height=100, x=0, y=0):
|
|
from Xlib import X
|
|
|
|
window = self.root.create_window(
|
|
x,
|
|
y,
|
|
width,
|
|
height,
|
|
0,
|
|
self.screen.root_depth,
|
|
X.InputOutput,
|
|
X.CopyFromParent,
|
|
)
|
|
self.display.sync()
|
|
return window
|
|
|
|
def flush(self):
|
|
self.display.flush()
|
|
|
|
def sync(self):
|
|
self.display.sync()
|
|
|
|
def close(self):
|
|
try:
|
|
self.display.close()
|
|
except OSError:
|
|
# Display may already be closed (e.g. server gone).
|
|
pass
|
|
|
|
def __enter__(self) -> Self:
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
self.close()
|
|
return False
|