receiver: serialize requests per handle so that threads do not receive response for a different request

lib/logitech_receiver/base.py

@@ -22,8 +22,12 @@
 
 from __future__ import absolute_import, division, print_function, unicode_literals
 
+import threading as _threading
+
 from collections import namedtuple
+from contextlib import contextmanager
 from logging import DEBUG as _DEBUG
+from logging import INFO as _INFO
 from logging import getLogger
 from random import getrandbits as _random_bits
 from time import time as _timestamp
@@ -350,6 +354,31 @@ del namedtuple
 #
 #
 
+request_lock = _threading.Lock()  # serialize all requests
+handles_lock = {}
+
+
+def handle_lock(handle):
+    with request_lock:
+        if handles_lock.get(int(handle)) is None:
+            if _log.isEnabledFor(_INFO):
+                _log.info('New lock %s', int(handle))
+            handles_lock[int(handle)] = _threading.Lock()  # Serialize requests on the handle
+    return handles_lock[int(handle)]
+
+
+# context manager for locks with a timeout
+@contextmanager
+def acquire_timeout(lock, handle, timeout):
+    result = lock.acquire(timeout=timeout)
+    try:
+        if not result:
+            _log.error('lock on handle %d not acquired, probably due to timeout', int(handle))
+        yield result
+    finally:
+        if result:
+            lock.release()
+
 
 # a very few requests (e.g., host switching) do not expect a reply, but use no_reply=True with extreme caution
 def request(handle, devnumber, request_id, *params, no_reply=False, return_error=False, long_message=False):
@@ -364,6 +393,7 @@ def request(handle, devnumber, request_id, *params, no_reply=False, return_error
     # import inspect as _inspect
     # print ('\n  '.join(str(s) for s in _inspect.stack()))
 
+    with acquire_timeout(handle_lock(handle), handle, 10.):
         assert isinstance(request_id, int)
         if devnumber != 0xFF and request_id < 0x8000:
             # For HID++ 2.0 feature requests, randomize the SoftwareId to make it
@@ -408,17 +438,10 @@ def request(handle, devnumber, request_id, *params, no_reply=False, return_error
             if reply:
                 report_id, reply_devnumber, reply_data = reply
                 if reply_devnumber == devnumber:
-                if report_id == HIDPP_SHORT_MESSAGE_ID and reply_data[:1] == b'\x8F' and reply_data[1:3] == request_data[:2]:
+                    if report_id == HIDPP_SHORT_MESSAGE_ID and reply_data[:1] == b'\x8F' and reply_data[1:3] == request_data[:2
+                                                                                                                             ]:
                         error = ord(reply_data[3:4])
 
-                    # if error == _hidpp10.ERROR.resource_error: # device unreachable
-                    #     _log.warn("(%s) device %d error on request {%04X}: unknown device", handle, devnumber, request_id)
-                    #     raise DeviceUnreachable(number=devnumber, request=request_id)
-
-                    # if error == _hidpp10.ERROR.unknown_device: # unknown device
-                    #     _log.error("(%s) device %d error on request {%04X}: unknown device", handle, devnumber, request_id)
-                    #     raise NoSuchDevice(number=devnumber, request=request_id)
-
                         if _log.isEnabledFor(_DEBUG):
                             _log.debug(
                                 '(%s) device 0x%02X error on request {%04X}: %d = %s', handle, devnumber, request_id, error,
@@ -494,6 +517,8 @@ def ping(handle, devnumber, long_message=False):
     assert devnumber >= 0x00
     assert devnumber < 0x0F
 
+    with acquire_timeout(handle_lock(handle), handle, 10.):
+
         # randomize the SoftwareId and mark byte to be able to identify the ping
         # reply, and set most significant (0x8) bit in SoftwareId so that the reply
         # is always distinguishable from notifications
@@ -524,7 +549,8 @@ def ping(handle, devnumber, long_message=False):
                         # HID++ 2.0+ device, currently connected
                         return ord(reply_data[2:3]) + ord(reply_data[3:4]) / 10.0
 
-                if report_id == HIDPP_SHORT_MESSAGE_ID and reply_data[:1] == b'\x8F' and reply_data[1:3] == request_data[:2]:
+                    if report_id == HIDPP_SHORT_MESSAGE_ID and reply_data[:1] == b'\x8F' and reply_data[1:3] == request_data[:2
+                                                                                                                             ]:
                         assert reply_data[-1:] == b'\x00'
                         error = ord(reply_data[3:4])
 

lib/logitech_receiver/listener.py

@@ -129,7 +129,7 @@ class _ThreadedHandle(object):
 # a while for it to acknowledge it.
 # Forcibly closing the file handle on another thread does _not_ interrupt the
 # read on Linux systems.
-_EVENT_READ_TIMEOUT = 0.4  # in seconds
+_EVENT_READ_TIMEOUT = 1.  # in seconds
 
 # After this many reads that did not produce a packet, call the tick() method.
 # This only happens if tick_period is enabled (>0) for the Listener instance.
@@ -174,7 +174,7 @@ class EventsListener(_threading.Thread):
             if self._queued_notifications.empty():
                 try:
                     # _log.debug("read next notification")
-                    n = _base.read(ihandle, _EVENT_READ_TIMEOUT)
+                    n = _base.read(self.receiver.handle, _EVENT_READ_TIMEOUT)
                 except _base.NoReceiver:
                     _log.warning('receiver disconnected')
                     self.receiver.close()