xavierk/voculinux
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

fix(evdev): leave held keyboards ungrabbed at startup (#941)

Browse Source 
Solid fix, this was leaving keys stuck on the desktop whenever someone was typing while the app started.
This commit is contained in:
AlexThunder1989
2026-10-11 23:50:43 -07:00
committed by GitHub
parent 05ae041220
commit b0fefd6ec1
2 changed files with 392 additions and 44 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/vocalinux/ui/keyboard_backends/evdev_backend.py
+178 -44
View File
@@ -137,6 +137,65 @@ def _clone_device_name(device: InputDevice) -> str:
return f"{base}{_CLONE_NAME_SUFFIX}"
def _drain_queued_events(device: InputDevice) -> list[InputEvent]:
"""Unread events sitting on the device's fd, drained without blocking.
The queue holds everything delivered since the fd was opened —
including events the compositor already saw — so the caller must split
it at the grab instant rather than treat every queued event alike. The
bound just prevents a pathological backlog from delaying startup; the
kernel's per-client queue is far smaller.
"""
events: list[InputEvent] = []
try:
for _ in range(512):
event = device.read_one()
if event is None:
break
events.append(event)
except OSError:
pass
return events
def _is_post_grab(event: InputEvent, grab_ts: float) -> bool:
"""True when the event was timestamped after the grab instant.
Kernel timestamps share ``time.time()``'s realtime domain. Anything
unreadable is treated as pre-grab — refusing the grab is the safe side.
"""
try:
return event.timestamp() >= grab_ts
except (AttributeError, TypeError):
return False
def _replay_events_on_clone(forwarder: UInput, events: list[InputEvent]) -> None:
"""Emit drained events on a clone that is about to be closed.
Events delivered after the grab reached only our fd — dropping them
loses real input — so on the refuse-grab path they are replayed through
the clone before it disappears. A press still held at drain time is
skipped: replaying it would leave the key logically down forever on a
clone that is going away, and the real release reaches the compositor
naturally once the device is ungrabbed.
"""
held: set[int] = set()
for event in events:
if event.type == ecodes.EV_KEY:
if event.value == 0:
held.discard(event.code)
else:
held.add(event.code)
for event in events:
if event.type == ecodes.EV_KEY and event.code in held:
continue
try:
forwarder.write_event(event)
except (OSError, IOError):
return
if EVDEV_AVAILABLE:
class _WriteOnlyUInput(UInput):
@@ -634,9 +693,11 @@ _HUB_OWNED_ATTRIBUTES = frozenset(
"_release_failed_forwarder",
"_forward_event",
"_dispatch_events",
"_dispatch_event",
"_detach_devices",
"_close_detached",
"_resync_clone_key_state",
"_pending_events",
}
)
@@ -655,6 +716,7 @@ _HUB_METHOD_ATTRIBUTES = frozenset(
"_release_failed_forwarder",
"_forward_event",
"_dispatch_events",
"_dispatch_event",
"_detach_devices",
"_close_detached",
"_resync_clone_key_state",
@@ -719,6 +781,10 @@ class EvdevDeviceHub:
# Grabbed fd -> key codes the clone currently believes are held.
# Needed to reconcile after SYN_DROPPED eats release events.
self._forwarded_held: dict[int, set[int]] = {}
# fd -> events drained at open time that arrived after the grab.
# They reached only our fd, so dispatch runs them through the same
# shortcut/forward path as live events, ahead of the next read().
self._pending_events: dict[int, list[InputEvent]] = {}
self._uinput_warned = False
def reset(self) -> None:
@@ -857,56 +923,78 @@ class EvdevDeviceHub:
if engines is None:
engines = self._engine_snapshot()
for event in device.read():
for event in self._pending_events.pop(fd, []):
self._dispatch_event(fd, device, event, engines, generation)
try:
events = list(device.read())
except BlockingIOError:
# The fd was marked readable only because stored startup events
# were pending — nothing new arrived. EAGAIN is not a
# disconnect; let the monitor keep the device.
events = []
for event in events:
if generation is not None and self._generation != generation:
# The generation advanced mid-read: this monitor's teardown
# is done or in flight, so drop the rest of the buffer
# rather than act on a newer generation's state.
return
if event.type == ecodes.EV_SYN:
if event.code == ecodes.SYN_DROPPED:
# Kernel buffer overflowed — discard until SYN_REPORT
self._dropped_devices.add(fd)
logger.warning(f"SYN_DROPPED on {device.name} (fd={fd}), resetting key state")
elif event.code == ecodes.SYN_REPORT:
if fd in self._dropped_devices:
# End of dropped sequence — clear stale state
self._dropped_devices.discard(fd)
for engine in engines:
engine.key_pressed_devices.discard(id(device))
# Release keys the clone still thinks are held before
# the SYN_REPORT reaches it, atomically.
self._resync_clone_key_state(fd, device, engines, generation)
# A dropped modifier release must not leave any
# engine's combo logically held.
for engine in engines:
engine._reset_combo_state()
self._forward_event(fd, event, generation)
continue
if fd in self._dropped_devices:
# Handling shortcut state mid-drop is unsafe, but surviving
# non-shortcut events still pass through so the app sees
# what the kernel kept.
consumed = False
for engine in engines:
if engine._event_is_shortcut(fd, event):
consumed = True
if not consumed:
self._forward_event(fd, event, generation)
continue
if event.type == ecodes.EV_KEY:
# Every key event updates each engine's shortcut state (combo
# modifiers must be tracked AND forwarded); only the
# forwarding decision differs.
consumed = False
for engine in engines:
if engine._event_is_shortcut(fd, event):
consumed = True
engine._handle_key_event(event, device)
if not consumed:
self._forward_event(fd, event, generation)
else:
self._dispatch_event(fd, device, event, engines, generation)
def _dispatch_event(
self,
fd: int,
device: InputDevice,
event: InputEvent,
engines: Sequence["EvdevKeyboardBackend"],
generation: Optional[int] = None,
) -> None:
"""Shortcut-check and forward one event (per-event body of dispatch)."""
if event.type == ecodes.EV_SYN:
if event.code == ecodes.SYN_DROPPED:
# Kernel buffer overflowed — discard until SYN_REPORT
self._dropped_devices.add(fd)
logger.warning(f"SYN_DROPPED on {device.name} (fd={fd}), resetting key state")
elif event.code == ecodes.SYN_REPORT:
if fd in self._dropped_devices:
# End of dropped sequence — clear stale state
self._dropped_devices.discard(fd)
for engine in engines:
engine.key_pressed_devices.discard(id(device))
# Release keys the clone still thinks are held before
# the SYN_REPORT reaches it, atomically.
self._resync_clone_key_state(fd, device, engines, generation)
# A dropped modifier release must not leave any
# engine's combo logically held.
for engine in engines:
engine._reset_combo_state()
self._forward_event(fd, event, generation)
return
if fd in self._dropped_devices:
# Handling shortcut state mid-drop is unsafe, but surviving
# non-shortcut events still pass through so the app sees
# what the kernel kept.
consumed = False
for engine in engines:
if engine._event_is_shortcut(fd, event):
consumed = True
if not consumed:
self._forward_event(fd, event, generation)
return
if event.type == ecodes.EV_KEY:
# Every key event updates each engine's shortcut state (combo
# modifiers must be tracked AND forwarded); only the
# forwarding decision differs.
consumed = False
for engine in engines:
if engine._event_is_shortcut(fd, event):
consumed = True
engine._handle_key_event(event, device)
if not consumed:
self._forward_event(fd, event, generation)
else:
self._forward_event(fd, event, generation)
def _monitor_devices(
self,
@@ -936,6 +1024,11 @@ class EvdevDeviceHub:
continue
readable, _, _ = select.select(device_fds, [], [], 1.0) # 1 second timeout
# Events drained past the grab at open time count as
# pending input even when nothing sits on the fd.
for pending_fd in self._pending_events:
if pending_fd not in readable:
readable.append(pending_fd)
for fd in readable:
if self._generation != generation:
@@ -1040,6 +1133,7 @@ class EvdevDeviceHub:
self._forwarder_paths = {}
self._clone_paths = set()
self._forwarded_held = {}
self._pending_events = {}
return devices, forwarders
def _close_detached(self, devices: list[InputDevice], forwarders: list[UInput]) -> None:
@@ -1096,6 +1190,7 @@ class EvdevDeviceHub:
return False
forwarder: Optional[UInput] = None
post_grab_events: list[InputEvent] = []
if _is_virtual_keyboard(device_path, device_name):
# A virtual keyboard is a remapper's output. Grabbing it
# starves the compositor of every event the remapper emits,
@@ -1113,12 +1208,45 @@ class EvdevDeviceHub:
forwarder = self._create_forwarder(device)
if forwarder is not None:
try:
# A key the compositor already saw go down must also come
# up on this device; grabbed, its release would only reach
# the clone and the key would stay held across the desktop.
if device.active_keys():
raise OSError("a key is held")
# Events already queued on this fd were delivered to the
# compositor's fd too. A key transition among them means
# input is in flight: grabbing now would route that
# key's release through the clone for a press the desktop
# saw on the physical device, which cannot reliably
# clear it. Refuse and let the rescan retry once quiet.
pre_grab_queued = _drain_queued_events(device)
if any(e.type == ecodes.EV_KEY for e in pre_grab_queued):
raise OSError("a key is held")
grab_ts = time.time()
device.grab()
# The drained queue straddles the grab. Events timestamped
# before it were delivered to the compositor's fd as well:
# a key transition there is the handover case — a press the
# compositor saw go down whose release arrived only to us —
# so the grab is refused. Events after it reached only this
# fd and are real input the compositor never saw; they are
# replayed on either path so a tap racing the grab is not
# lost. A key still held here was pressed after the grab —
# the compositor never saw it go down, so keeping the grab
# (and replaying the press) is consistent.
queued = _drain_queued_events(device)
post_grab_events = [e for e in queued if _is_post_grab(e, grab_ts)]
if any(
e.type == ecodes.EV_KEY and not _is_post_grab(e, grab_ts) for e in queued
):
device.ungrab()
raise OSError("a key is held")
except (OSError, IOError) as e:
logger.warning(
f"Cannot grab {device_path} ({e}); "
"shortcut keys will also reach the focused app"
)
_replay_events_on_clone(forwarder, post_grab_events)
try:
forwarder.close()
except (OSError, IOError, RuntimeError) as e:
@@ -1157,6 +1285,11 @@ class EvdevDeviceHub:
self._forwarder_paths[fd] = clone_path
self._clone_paths.add(clone_path)
self._forwarded_held[fd] = set()
if post_grab_events:
# Drained events that arrived after the grab reached only
# this fd; dispatch replays them before the next read so
# engines (and the clone) see the input in kernel order.
self._pending_events[fd] = post_grab_events
logger.debug(
f"Opened keyboard device: {device_path} ({device.name})"
@@ -1347,6 +1480,7 @@ class EvdevDeviceHub:
self.device_paths.discard(device_path)
self._fd_generation.pop(fd, None)
self._dropped_devices.discard(fd)
self._pending_events.pop(fd, None)
forwarder = self._forwarders.pop(fd, None)
clone_path = self._forwarder_paths.pop(fd, None)
if clone_path is not None:
tests/test_evdev_backend_ext.py
+214
View File
@@ -1348,6 +1348,7 @@ class TestEvdevGrabAndForwarding:
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03 # BUS_USB
device.active_keys.return_value = []
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
@@ -1446,6 +1447,7 @@ class TestEvdevGrabAndForwarding:
device.fileno.return_value = 10
device.info.bustype = 0x03
device.name = "keyd virtual keyboard"
device.active_keys.return_value = []
mock_input_device.return_value = device
forwarder = MagicMock()
forwarder.device.path = "/dev/input/event20"
@@ -1536,6 +1538,7 @@ class TestEvdevGrabAndForwarding:
device = MagicMock()
device.fileno.return_value = 10
device.info.bustype = 0x03
device.active_keys.return_value = []
mock_input_device.return_value = device
mock_uinput.side_effect = OSError("could not open uinput device in write mode")
forwarder = MagicMock()
@@ -1572,6 +1575,192 @@ class TestEvdevGrabAndForwarding:
assert backend._forwarders == {}
assert device in backend.devices
def _queued_key_event(self, code: int, value: int, ts: float) -> MagicMock:
"""A queued EV_KEY event timestamped pre- or post-grab."""
event = MagicMock()
event.type = ecodes.EV_KEY
event.code = code
event.value = value
event.timestamp.return_value = ts
return event
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_leaves_keyboard_with_held_key_ungrabbed(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""A key held before the grab keeps the device ungrabbed."""
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = [54]
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.grab.assert_not_called()
device.ungrab.assert_not_called()
forwarder.close.assert_called_once()
assert backend._forwarders == {}
assert device in backend.devices
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_refuses_grab_when_key_transition_already_queued(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""A key transition queued at open refuses the grab entirely.
Everything drained before the grab was delivered to the
compositor's fd too, so a key event there means input is in
flight: grabbing would route a release through the clone for a
press the desktop saw on the physical device. The device stays
monitored ungrabbed and the rescan retries once quiet.
"""
press = self._queued_key_event(54, 1, time.time())
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
device.read_one.side_effect = [press, None]
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.grab.assert_not_called()
device.ungrab.assert_not_called()
forwarder.write_event.assert_not_called()
forwarder.close.assert_called_once()
assert backend._forwarders == {}
assert backend._pending_events == {}
assert device in backend.devices
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_ungrabs_on_pre_grab_key_transition(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""A press arriving between the pre-grab drain and the grab refuses it.
The event slipped past the empty pre-drain but is timestamped
before the grab, so it belongs to the compositor's view too — the
handover case. Its post-grab release still reaches the desktop
through the clone before the clone is closed, and engines get it
via pending events.
"""
pre = time.time() - 10
post = time.time() + 10
press = self._queued_key_event(54, 1, pre)
release = self._queued_key_event(54, 0, post)
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
# The pre-grab drain comes up empty; the straddle drain sees both.
device.read_one.side_effect = [None, press, release, None]
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.grab.assert_called_once()
device.ungrab.assert_called_once()
forwarder.write_event.assert_called_once_with(release)
forwarder.close.assert_called_once()
assert backend._forwarders == {}
assert backend._pending_events == {10: [release]}
assert device in backend.devices
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_skips_replaying_post_grab_held_press(
self, mock_input_device: Mock, mock_uinput: Mock
) -> None:
"""A post-grab press still held is not replayed onto a dying clone.
Refusing the grab over a pre-grab transition, the held press would
leave the key logically down forever if replayed before close, so
only events for released keys are emitted.
"""
pre = time.time() - 10
post = time.time() + 10
stale_press = self._queued_key_event(54, 1, pre)
held_press = self._queued_key_event(100, 1, post)
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
device.read_one.side_effect = [None, stale_press, held_press, None]
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.ungrab.assert_called_once()
forwarder.write_event.assert_not_called()
assert backend._pending_events == {10: [held_press]}
@pytest.mark.parametrize(
"queued, expected_pending",
[
("none", "none"), # quiet handover: grab stays, nothing pending
("post_tap", "post_tap"), # post-grab tap replays, grab stays
("post_press", "post_press"), # held post-grab press keeps the grab
],
)
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_keeps_grab_when_only_post_grab_events(
self,
mock_input_device: Mock,
mock_uinput: Mock,
queued: str,
expected_pending: str,
) -> None:
"""Events entirely after the grab are input, not handover evidence.
A tap delivered after the grab reached only our fd, so the grab is
kept and the events are queued for dispatch — dropping them would
lose the keystroke the compositor never saw.
"""
post = time.time() + 10
events = {
"post_tap": [
self._queued_key_event(54, 1, post),
self._queued_key_event(54, 0, post),
],
"post_press": [self._queued_key_event(100, 1, post)],
}
queue = events.get(queued, [])
device = MagicMock()
device.fileno.return_value = 10
device.active_keys.return_value = []
# First None empties the pre-grab drain; the straddle drain follows.
device.read_one.side_effect = [None, *queue, None]
mock_input_device.return_value = device
forwarder = MagicMock()
mock_uinput.return_value = forwarder
backend = EvdevKeyboardBackend()
assert backend._open_keyboard_device("/dev/input/event0") is True
device.grab.assert_called_once()
device.ungrab.assert_not_called()
forwarder.close.assert_not_called()
forwarder.write_event.assert_not_called()
assert backend._forwarders == {10: forwarder}
expected = events.get(expected_pending, [])
if expected:
assert backend._pending_events == {10: expected}
else:
assert backend._pending_events == {}
@patch("vocalinux.ui.keyboard_backends.evdev_backend.UInput")
@patch("vocalinux.ui.keyboard_backends.evdev_backend.InputDevice")
def test_open_falls_back_when_grab_fails(self, mock_input_device, mock_uinput):
@@ -2318,6 +2507,7 @@ class TestSharedEvdevDeviceLayer:
device.name = "test-kbd"
device.fileno.return_value = 10
device.capabilities.return_value = {}
device.active_keys.return_value = []
mock_input_device.return_value = device
forwarder = MagicMock()
forwarder.device.path = "/dev/input/event9"
@@ -2425,6 +2615,30 @@ class TestSharedEvdevDeviceLayer:
active._handle_key_event.assert_called_once()
idle._handle_key_event.assert_not_called()
def test_dispatch_pending_events_tolerates_empty_read(self) -> None:
"""A pending-only fd replays stored events without disconnecting.
Regression test for "replay drops a working keyboard": when stored
startup events mark a device readable but no new input arrived,
device.read() raises BlockingIOError. That error must not escape to
the monitor, which treats OSError as a disconnect and would close a
healthy keyboard and its clone.
"""
engine = EvdevKeyboardBackend(shortcut="ctrl+ctrl")
hub = self._registered(engine)
engine._handle_key_event = MagicMock()
release = self._key_event(29, 0)
hub._pending_events[10] = [release]
device = MagicMock()
device.read.side_effect = BlockingIOError(11, "Resource temporarily unavailable")
hub._dispatch_events(10, device)
engine._handle_key_event.assert_called_once_with(release, device)
assert hub._pending_events == {}
def test_register_waits_for_in_flight_teardown(self) -> None:
"""Devices opened after teardown starts are never closed by it.