fix(dtls): make reader-thread death visible and fail fast
The reader loop exited silently on any socket error, leaving conn/sock
set so the session still looked open. Every later get()/post()/ping()
then waited out its full request timeout on a session nobody was
reading, raising SessionTimeoutError on repeat, forever.
This started biting in v0.1.3 (d677c72), which moved to connected UDP
sockets: a connected socket surfaces ICMP errors on recv, so one
ECONNREFUSED from a rebooting appliance now killed the reader.
- Advisory ICMP errnos (ECONNREFUSED/EHOSTUNREACH/...) no longer kill
the reader; the next datagram usually works.
- Real reader exits log at WARNING; close()-driven exits stay quiet.
- A _reader_running Event lets get/post/ping/subscribe/refresh_observes
fail fast via _check_live() with SessionClosedError instead of waiting
out a timeout. Callers that never start a reader are unaffected.
Refs QuiteYellow/SmartThings-Local#37
This commit is contained in:
Jack Nagy
@@ -18,6 +18,7 @@ Reader thread owns the UDP socket. Callers issue get()/post() and block
|
||||
on a per-token Event the reader signals. OBSERVE notifications are
|
||||
delivered via the on_notification callback.
|
||||
"""
|
||||
import errno
|
||||
import os
|
||||
import socket
|
||||
import threading
|
||||
@@ -72,6 +73,21 @@ _BLOCK_ACK_TIMEOUT = 4.0
|
||||
# once the ceiling is measured empirically.
|
||||
_DEFAULT_RATE_LIMIT_RPS = 5.0
|
||||
|
||||
# ICMP errors a connected UDP socket surfaces on the next recv. On these
|
||||
# appliances they show up while the device is rebooting, while it holds an
|
||||
# orphaned association, or across a router blip, and the next datagram
|
||||
# usually works. UDP delivery was never guaranteed, so treat them as
|
||||
# advisory and keep reading. Unconnected sockets never see any of this,
|
||||
# which is why the reader survived them before the connected-socket change
|
||||
# in d677c72 (v0.1.3).
|
||||
_ADVISORY_ERRNOS = frozenset(
|
||||
value for value in (
|
||||
getattr(errno, name, None)
|
||||
for name in ('ECONNREFUSED', 'EHOSTUNREACH', 'ENETUNREACH',
|
||||
'EHOSTDOWN', 'ENETDOWN')
|
||||
) if value is not None
|
||||
)
|
||||
|
||||
class DtlsCoapSession:
|
||||
"""Single sustained DTLS-CoAP session.
|
||||
|
||||
@@ -168,6 +184,10 @@ class DtlsCoapSession:
|
||||
|
||||
self._stop = threading.Event()
|
||||
self._reader_thread = None
|
||||
# Set while the reader owns the socket. Cleared when it exits for
|
||||
# any reason, so callers fail fast through _check_live() instead of
|
||||
# waiting out a request timeout against a session nobody is reading.
|
||||
self._reader_running = threading.Event()
|
||||
self._last_send_ts = 0.0
|
||||
|
||||
def pace(self) -> None:
|
||||
@@ -253,11 +273,27 @@ class DtlsCoapSession:
|
||||
"""Spawn the reader thread. Must be called after connect()."""
|
||||
if self.sock is None:
|
||||
raise RuntimeError("connect() before start_reader()")
|
||||
self._reader_running.set()
|
||||
t = threading.Thread(target=self._reader_loop,
|
||||
daemon=True, name='dtls-reader')
|
||||
t.start()
|
||||
self._reader_thread = t
|
||||
|
||||
def _check_live(self):
|
||||
"""Raise if the session cannot carry a request. A dead reader is
|
||||
as fatal as a closed connection: the socket may still accept
|
||||
sends, but no response will ever be dispatched, so waiting out the
|
||||
request timeout only delays the inevitable SessionClosedError.
|
||||
|
||||
Callers that never start a reader (config-flow style) keep the old
|
||||
behaviour — only the conn check applies while _reader_thread is
|
||||
None."""
|
||||
if self.conn is None:
|
||||
raise SessionClosedError()
|
||||
if self._reader_thread is not None and \
|
||||
not self._reader_running.is_set():
|
||||
raise SessionClosedError()
|
||||
|
||||
def join(self):
|
||||
"""Block until the reader thread exits (i.e. socket dies)."""
|
||||
if self._reader_thread is not None:
|
||||
@@ -374,7 +410,23 @@ class DtlsCoapSession:
|
||||
d = sock.recv(65535)
|
||||
except socket.timeout:
|
||||
continue
|
||||
except (OSError, ValueError):
|
||||
except OSError as e:
|
||||
if self._stop.is_set():
|
||||
return # close() got here first
|
||||
if e.errno in _ADVISORY_ERRNOS:
|
||||
logger.debug("reader: advisory %s from %s, continuing",
|
||||
errno.errorcode.get(e.errno, e.errno),
|
||||
self.host)
|
||||
continue
|
||||
logger.warning("reader exiting: socket error %s from %s",
|
||||
errno.errorcode.get(e.errno, e.errno),
|
||||
self.host)
|
||||
return
|
||||
except ValueError:
|
||||
# recv on a socket closed underneath the reader.
|
||||
if not self._stop.is_set():
|
||||
logger.warning("reader exiting: socket closed "
|
||||
"underneath it")
|
||||
return
|
||||
if not d:
|
||||
continue
|
||||
@@ -419,6 +471,8 @@ class DtlsCoapSession:
|
||||
if exit_reader:
|
||||
return
|
||||
finally:
|
||||
# Reader no longer owns the socket — callers must fail fast.
|
||||
self._reader_running.clear()
|
||||
# Make sure pending waiters don't hang if the reader dies.
|
||||
for tok, (ev, container) in list(self._pending.items()):
|
||||
container.setdefault('err', SessionClosedError())
|
||||
@@ -490,8 +544,7 @@ class DtlsCoapSession:
|
||||
response — Samsung's server keys per-transfer state on the
|
||||
token, and dropping a fresh token on block 1+ silently drops
|
||||
the request."""
|
||||
if self.conn is None:
|
||||
raise SessionClosedError()
|
||||
self._check_live()
|
||||
tok = self._next_tok()
|
||||
blob = b''
|
||||
num = 0
|
||||
@@ -566,8 +619,7 @@ class DtlsCoapSession:
|
||||
def post(self, path_segs, body_cbor, timeout=8.0):
|
||||
"""Single-frame POST with a CBOR-encoded body. Returns
|
||||
(code, payload_bytes). body_cbor must already be encoded."""
|
||||
if self.conn is None:
|
||||
raise SessionClosedError()
|
||||
self._check_live()
|
||||
tok = self._next_tok()
|
||||
mid = self._next_mid()
|
||||
opts = [(URI_PATH, s.encode()) for s in path_segs]
|
||||
@@ -600,8 +652,7 @@ class DtlsCoapSession:
|
||||
Real half-open-session detection lives in PollScheduler's
|
||||
`last_success_ts`, surfaced through KeepaliveTask's
|
||||
`liveness_fn`."""
|
||||
if self.conn is None:
|
||||
raise SessionClosedError()
|
||||
self._check_live()
|
||||
mid = self._next_mid()
|
||||
self._send_dgram(build_coap(TYPE_CON, 0, mid, b'', []))
|
||||
return mid
|
||||
@@ -618,8 +669,7 @@ class DtlsCoapSession:
|
||||
tokens via subscribe. Brief race window where a notify on the
|
||||
old token gets dropped as 'stale' — acceptable for a 6h-scale
|
||||
safety net."""
|
||||
if self.conn is None:
|
||||
raise SessionClosedError()
|
||||
self._check_live()
|
||||
for tok, href in list(self._observe_tokens.items()):
|
||||
segs = [s for s in href.split('/') if s]
|
||||
try:
|
||||
@@ -642,8 +692,7 @@ class DtlsCoapSession:
|
||||
|
||||
Returns the token used (in case the caller wants to deregister
|
||||
later)."""
|
||||
if self.conn is None:
|
||||
raise SessionClosedError()
|
||||
self._check_live()
|
||||
tok = self._next_observe_tok()
|
||||
href = '/' + '/'.join(path_segs)
|
||||
# Register the token BEFORE sending — otherwise the device
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
"""Reader-thread death visibility (QuiteYellow/SmartThings-Local#37).
|
||||
|
||||
A connected UDP socket surfaces ICMP errors on recv; before this the
|
||||
reader exited silently on the first one and every later request waited
|
||||
out its full timeout against a session nobody was reading. These tests
|
||||
pin the three behaviours that fixed it: advisory ICMP errnos keep the
|
||||
reader alive, a real socket error exits with a WARNING and clears
|
||||
_reader_running, and callers then fail fast with SessionClosedError.
|
||||
"""
|
||||
import errno
|
||||
import logging
|
||||
import socket
|
||||
import threading
|
||||
import time
|
||||
|
||||
import pytest
|
||||
from OpenSSL import SSL
|
||||
|
||||
from smartthings_local.errors import SessionClosedError
|
||||
from smartthings_local.protocol.dtls_session import DtlsCoapSession
|
||||
|
||||
_LOGGER_NAME = "smartthings_local.protocol.dtls_session"
|
||||
|
||||
|
||||
class _NullAuth:
|
||||
"""Structural AuthenticationProvider — never configured, we skip connect()."""
|
||||
|
||||
def configure_context(self, _context):
|
||||
return None
|
||||
|
||||
|
||||
class _FakeConn:
|
||||
"""Minimal stand-in for SSL.Connection: each datagram written to the
|
||||
BIO surfaces as one decrypted packet on the next recv(), then
|
||||
WantReadError like a drained DTLS record buffer."""
|
||||
|
||||
def __init__(self):
|
||||
self._decrypted = []
|
||||
|
||||
def bio_write(self, datagram):
|
||||
self._decrypted.append(datagram)
|
||||
|
||||
def recv(self, _n):
|
||||
if self._decrypted:
|
||||
return self._decrypted.pop(0)
|
||||
raise SSL.WantReadError()
|
||||
|
||||
def bio_read(self, _n):
|
||||
return b""
|
||||
|
||||
def send(self, _datagram):
|
||||
return None
|
||||
|
||||
def shutdown(self):
|
||||
return None
|
||||
|
||||
|
||||
class _FakeSock:
|
||||
"""Scripted UDP socket. Each step is bytes to return, an exception to
|
||||
raise, or a callable to run (then a timeout, so the loop re-checks
|
||||
_stop). An exhausted script blocks like a real recv timeout; once
|
||||
close()d it raises EBADF the way a closed fd does."""
|
||||
|
||||
def __init__(self, steps=()):
|
||||
self._steps = list(steps)
|
||||
self.closed = False
|
||||
self.timeout = None
|
||||
|
||||
def settimeout(self, value):
|
||||
self.timeout = value
|
||||
|
||||
def recv(self, _n):
|
||||
if self.closed:
|
||||
raise OSError(errno.EBADF, "bad file descriptor")
|
||||
if self._steps:
|
||||
step = self._steps.pop(0)
|
||||
if callable(step):
|
||||
step()
|
||||
raise socket.timeout()
|
||||
if isinstance(step, BaseException):
|
||||
raise step
|
||||
return step
|
||||
time.sleep(0.01)
|
||||
raise socket.timeout()
|
||||
|
||||
def send(self, data):
|
||||
return len(data)
|
||||
|
||||
def close(self):
|
||||
self.closed = True
|
||||
|
||||
|
||||
def _make_session():
|
||||
sess = DtlsCoapSession("host", 1234, auth=_NullAuth())
|
||||
sess.conn = _FakeConn()
|
||||
return sess
|
||||
|
||||
|
||||
def _run_reader(sess, steps, timeout=2.0):
|
||||
sess.sock = _FakeSock(steps)
|
||||
sess.start_reader()
|
||||
sess._reader_thread.join(timeout)
|
||||
assert not sess._reader_thread.is_alive(), "reader thread did not exit"
|
||||
|
||||
|
||||
def test_advisory_icmp_error_does_not_kill_reader(caplog):
|
||||
sess = _make_session()
|
||||
dispatched = []
|
||||
sess._dispatch_coap = dispatched.append
|
||||
|
||||
with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME):
|
||||
_run_reader(sess, [
|
||||
OSError(errno.ECONNREFUSED, "connection refused"),
|
||||
b"\x60\x00\x00\x00", # survives, gets dispatched
|
||||
lambda: sess._stop.set(), # end the loop cleanly
|
||||
])
|
||||
|
||||
assert dispatched == [b"\x60\x00\x00\x00"]
|
||||
assert not sess._reader_running.is_set()
|
||||
assert any(r.levelno == logging.DEBUG and "advisory" in r.getMessage()
|
||||
for r in caplog.records)
|
||||
# An advisory errno is not a real exit — no WARNING.
|
||||
assert not any(r.levelno >= logging.WARNING for r in caplog.records)
|
||||
|
||||
|
||||
def test_fatal_socket_error_exits_with_warning(caplog):
|
||||
sess = _make_session()
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME):
|
||||
_run_reader(sess, [OSError(errno.EBADF, "bad file descriptor")])
|
||||
|
||||
assert not sess._reader_running.is_set()
|
||||
warnings = [r for r in caplog.records if r.levelno == logging.WARNING]
|
||||
assert len(warnings) == 1
|
||||
assert "reader exiting" in warnings[0].getMessage()
|
||||
|
||||
|
||||
def test_request_fails_fast_after_reader_death():
|
||||
sess = _make_session()
|
||||
_run_reader(sess, [OSError(errno.EBADF, "bad file descriptor")])
|
||||
assert not sess._reader_running.is_set()
|
||||
|
||||
start = time.monotonic()
|
||||
with pytest.raises(SessionClosedError):
|
||||
sess.get(["oic", "d"], timeout=10.0)
|
||||
elapsed = time.monotonic() - start
|
||||
# The whole point: no waiting out the request timeout.
|
||||
assert elapsed < 1.0, f"get() waited {elapsed:.2f}s instead of failing fast"
|
||||
|
||||
|
||||
def test_close_does_not_log_warning_on_teardown(caplog):
|
||||
sess = _make_session()
|
||||
sess.sock = _FakeSock() # empty script: blocks on recv
|
||||
sess.start_reader()
|
||||
time.sleep(0.05) # let the reader reach recv
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME):
|
||||
sess.close()
|
||||
sess._reader_thread.join(2.0)
|
||||
|
||||
assert not sess._reader_thread.is_alive()
|
||||
assert not sess._reader_running.is_set()
|
||||
assert not any(r.levelno >= logging.WARNING for r in caplog.records)
|
||||
|
||||
|
||||
def test_check_live_without_reader_matches_old_conn_guard():
|
||||
sess = _make_session() # conn set, reader never started
|
||||
assert sess._reader_thread is None
|
||||
sess._check_live() # must not raise — config-flow behaviour
|
||||
|
||||
sess.conn = None
|
||||
with pytest.raises(SessionClosedError):
|
||||
sess._check_live()
|
||||
Reference in New Issue
Block a user