The finalizer's lock probe couldn't see its own thread
The cursor finalizer runs at GC time on whatever thread happened to allocate, so it must not touch the wire while a statement owns it. It checked with _wire_lock.acquire(blocking=False), meaning to ask "is anyone using the wire?" -- but an RLock grants a reentrant acquire to its own owner. When GC fired on the thread that was mid-statement, the probe returned True, the finalizer concluded it had exclusive access, and it sent CLOSE/RELEASE into the middle of the statement it had just interrupted. The victim got -208 on valid SQL. Refcounting hid it. A dropped cursor is freed at the drop, before the next statement runs. A cursor caught in a reference cycle waits for a collection instead, and cycles are ordinary -- any traceback that holds a cursor makes one. Reproduced by putting an abandoned scrollable cursor in a cycle and collecting mid-statement. _wire_lock is now a small wrapper that tracks owner and depth, so held_by_current_thread answers the question the finalizer was actually asking. Same-thread GC defers to the cleanup queue exactly as another thread's would. This also retires the _is_owned() call in _ensure_transaction, which was reaching into CPython private API for the same information. The finalizer's error handling had the stale-cleanup defect too: a leftover CLOSE draws -267, an OperationalError, which is in WIRE_ERRORS, so it force-closed a healthy connection over a no-op. Server-reported errors are told apart from wire failures by their sqlcode, matching _drain_pending_cleanup.
This commit is contained in:
parent
ac00a25cb7
commit
2eb5ac8f0f
@ -206,6 +206,62 @@ _DEFAULT_ENV: dict[str, str] = {
|
|||||||
}
|
}
|
||||||
|
|
||||||
|
|
||||||
|
class _WireLock:
|
||||||
|
"""A reentrant lock that can say whether *this* thread already holds it.
|
||||||
|
|
||||||
|
``threading.RLock`` cannot answer that question in public API, and
|
||||||
|
the difference is not academic. The cursor finalizer runs at GC time
|
||||||
|
on whatever thread happened to allocate — including a thread that is
|
||||||
|
at that moment mid-statement holding this lock. It probes with
|
||||||
|
``acquire(blocking=False)`` intending to mean "is anyone using the
|
||||||
|
wire?", but an RLock grants a reentrant acquire to its own owner, so
|
||||||
|
the probe returns True and the finalizer sends CLOSE/RELEASE into the
|
||||||
|
middle of the statement it interrupted. The victim gets ``-208``.
|
||||||
|
|
||||||
|
Reachability is not exotic. Refcounting frees a dropped cursor
|
||||||
|
immediately, before the next statement, which is why this went
|
||||||
|
unnoticed — but a cursor caught in a reference cycle waits for a
|
||||||
|
collection instead, and cycles are ordinary in Python. Any traceback
|
||||||
|
that holds a cursor makes one.
|
||||||
|
|
||||||
|
``held_by_current_thread`` reads ``_depth``/``_owner`` without the
|
||||||
|
lock, which is safe: both are mutated only under it, and the only
|
||||||
|
values another thread can leave behind are a depth of zero or an
|
||||||
|
owner that isn't us. Either way the answer is False, correctly.
|
||||||
|
"""
|
||||||
|
|
||||||
|
__slots__ = ("_depth", "_lock", "_owner")
|
||||||
|
|
||||||
|
def __init__(self) -> None:
|
||||||
|
self._lock = threading.RLock()
|
||||||
|
self._owner: int | None = None
|
||||||
|
self._depth = 0
|
||||||
|
|
||||||
|
def acquire(self, blocking: bool = True, timeout: float = -1) -> bool:
|
||||||
|
acquired = self._lock.acquire(blocking, timeout)
|
||||||
|
if acquired:
|
||||||
|
self._owner = threading.get_ident()
|
||||||
|
self._depth += 1
|
||||||
|
return acquired
|
||||||
|
|
||||||
|
def release(self) -> None:
|
||||||
|
self._depth -= 1
|
||||||
|
if self._depth == 0:
|
||||||
|
self._owner = None
|
||||||
|
self._lock.release()
|
||||||
|
|
||||||
|
def __enter__(self) -> _WireLock:
|
||||||
|
self.acquire()
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(self, *exc_info: object) -> None:
|
||||||
|
self.release()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def held_by_current_thread(self) -> bool:
|
||||||
|
return self._depth > 0 and self._owner == threading.get_ident()
|
||||||
|
|
||||||
|
|
||||||
class Connection:
|
class Connection:
|
||||||
"""A SQLI session. Owns one TCP socket and the post-login state.
|
"""A SQLI session. Owns one TCP socket and the post-login state.
|
||||||
|
|
||||||
@ -250,7 +306,7 @@ class Connection:
|
|||||||
# this lock with a timeout, then calls ``conn.rollback()`` —
|
# this lock with a timeout, then calls ``conn.rollback()`` —
|
||||||
# which itself acquires the lock. Same thread, two acquires.
|
# which itself acquires the lock. Same thread, two acquires.
|
||||||
# Reentrance must be cheap and correct.
|
# Reentrance must be cheap and correct.
|
||||||
self._wire_lock = threading.RLock()
|
self._wire_lock = _WireLock()
|
||||||
# Phase 29: deferred-cleanup queue for cursor finalizers that
|
# Phase 29: deferred-cleanup queue for cursor finalizers that
|
||||||
# couldn't acquire the wire lock at GC time. Each entry is a
|
# couldn't acquire the wire lock at GC time. Each entry is a
|
||||||
# PDU's worth of bytes (typically a CLOSE or RELEASE) that
|
# PDU's worth of bytes (typically a CLOSE or RELEASE) that
|
||||||
@ -696,7 +752,7 @@ class Connection:
|
|||||||
# method but stable across versions; cheap (~50ns) and only
|
# method but stable across versions; cheap (~50ns) and only
|
||||||
# checks the current thread. If it ever changes shape, drop
|
# checks the current thread. If it ever changes shape, drop
|
||||||
# this assert — the doc still names the precondition.
|
# this assert — the doc still names the precondition.
|
||||||
assert self._wire_lock._is_owned(), (
|
assert self._wire_lock.held_by_current_thread, (
|
||||||
"_ensure_transaction called without _wire_lock held; "
|
"_ensure_transaction called without _wire_lock held; "
|
||||||
"the cursor method that called it must wrap its body in "
|
"the cursor method that called it must wrap its body in "
|
||||||
"`with self._conn._wire_lock:`"
|
"`with self._conn._wire_lock:`"
|
||||||
|
|||||||
@ -131,6 +131,24 @@ def _finalize_cursor(
|
|||||||
conn = conn_ref()
|
conn = conn_ref()
|
||||||
if conn is None or conn.closed:
|
if conn is None or conn.closed:
|
||||||
return
|
return
|
||||||
|
if conn._wire_lock.held_by_current_thread:
|
||||||
|
# GC fired on the very thread that is mid-statement. The
|
||||||
|
# non-blocking acquire below would *succeed* here — an RLock
|
||||||
|
# grants a reentrant acquire to its own owner — and we would
|
||||||
|
# send CLOSE/RELEASE into the middle of that statement, killing
|
||||||
|
# it with -208. Defer instead, exactly as for another thread.
|
||||||
|
#
|
||||||
|
# Refcounting hides this: a dropped cursor is freed at the drop,
|
||||||
|
# before the next statement. A cursor caught in a reference cycle
|
||||||
|
# waits for a collection instead, and cycles are ordinary — any
|
||||||
|
# traceback holding a cursor makes one.
|
||||||
|
conn._enqueue_cleanup([_CLOSE_PDU, _RELEASE_PDU])
|
||||||
|
_log.debug(
|
||||||
|
"cursor finalizer: GC ran on the thread holding the wire lock; "
|
||||||
|
"enqueued CLOSE+RELEASE for deferred cleanup on conn %s",
|
||||||
|
id(conn),
|
||||||
|
)
|
||||||
|
return
|
||||||
if not conn._wire_lock.acquire(blocking=False):
|
if not conn._wire_lock.acquire(blocking=False):
|
||||||
# Another thread is mid-operation on this connection. Don't
|
# Another thread is mid-operation on this connection. Don't
|
||||||
# deadlock; instead, hand the cleanup bytes to the connection's
|
# deadlock; instead, hand the cleanup bytes to the connection's
|
||||||
@ -152,18 +170,32 @@ def _finalize_cursor(
|
|||||||
conn._send_pdu(_RELEASE_PDU)
|
conn._send_pdu(_RELEASE_PDU)
|
||||||
conn._drain_to_eot()
|
conn._drain_to_eot()
|
||||||
except WIRE_ERRORS as exc:
|
except WIRE_ERRORS as exc:
|
||||||
# Wire desync during cleanup — same doctrine as
|
if getattr(exc, "sqlcode", None) is not None:
|
||||||
# ``_raise_sq_err``: the wire is unrecoverable, force-close
|
# The *server* rejected the cleanup — typically a stale
|
||||||
# the connection. Asymmetric handling of the same failure
|
# CLOSE for a cursor it no longer has, which answers
|
||||||
# mode would be a Hamilton smell.
|
# -267 "the transaction has been rolled back". That is
|
||||||
_log.warning(
|
# an OperationalError, which is in WIRE_ERRORS, so this
|
||||||
"cursor finalizer: wire desync during cleanup; "
|
# branch used to force-close a perfectly healthy
|
||||||
"force-closing connection: %r",
|
# connection over a no-op. ``_raise_sq_err`` self-drains
|
||||||
exc,
|
# the trailing SQ_EOT, so the wire is still aligned.
|
||||||
)
|
# Same distinction as _drain_pending_cleanup.
|
||||||
conn._closed = True
|
_log.debug(
|
||||||
with contextlib.suppress(Exception):
|
"cursor finalizer: server rejected cleanup (stale): %r",
|
||||||
conn._sock.close()
|
exc,
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
# No sqlcode: the socket died or framing desynced and we
|
||||||
|
# can no longer say where a response ends. Force-close —
|
||||||
|
# same doctrine as ``_raise_sq_err``. Asymmetric handling
|
||||||
|
# of the same failure mode would be a Hamilton smell.
|
||||||
|
_log.warning(
|
||||||
|
"cursor finalizer: wire desync during cleanup; "
|
||||||
|
"force-closing connection: %r",
|
||||||
|
exc,
|
||||||
|
)
|
||||||
|
conn._closed = True
|
||||||
|
with contextlib.suppress(Exception):
|
||||||
|
conn._sock.close()
|
||||||
except InterfaceError:
|
except InterfaceError:
|
||||||
# Connection was closed by another thread between our
|
# Connection was closed by another thread between our
|
||||||
# ``conn.closed`` check above and the actual write. No-op:
|
# ``conn.closed`` check above and the actual write. No-op:
|
||||||
|
|||||||
@ -260,3 +260,96 @@ def test_connection_closes_cleanly_with_a_scroll_cursor_open(
|
|||||||
with contextlib.suppress(Exception):
|
with contextlib.suppress(Exception):
|
||||||
conn.close()
|
conn.close()
|
||||||
assert conn.closed
|
assert conn.closed
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# The wire lock's blind spot
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
def test_wire_lock_reports_its_own_thread() -> None:
|
||||||
|
"""The whole point. ``RLock.acquire(blocking=False)`` returns True for
|
||||||
|
the owning thread, so the finalizer's probe could not tell "nobody is
|
||||||
|
using the wire" from "I am, right now, mid-statement"."""
|
||||||
|
from informix_db.connections import _WireLock
|
||||||
|
|
||||||
|
lock = _WireLock()
|
||||||
|
assert not lock.held_by_current_thread
|
||||||
|
with lock:
|
||||||
|
assert lock.held_by_current_thread
|
||||||
|
assert lock.acquire(blocking=False), "must still be reentrant"
|
||||||
|
lock.release()
|
||||||
|
assert lock.held_by_current_thread, "still held at depth 1"
|
||||||
|
assert not lock.held_by_current_thread
|
||||||
|
|
||||||
|
|
||||||
|
def test_wire_lock_is_not_held_by_other_threads() -> None:
|
||||||
|
import threading
|
||||||
|
|
||||||
|
from informix_db.connections import _WireLock
|
||||||
|
|
||||||
|
lock = _WireLock()
|
||||||
|
seen: list[bool] = []
|
||||||
|
entered = threading.Event()
|
||||||
|
done = threading.Event()
|
||||||
|
|
||||||
|
def holder() -> None:
|
||||||
|
with lock:
|
||||||
|
entered.set()
|
||||||
|
done.wait(5)
|
||||||
|
|
||||||
|
t = threading.Thread(target=holder)
|
||||||
|
t.start()
|
||||||
|
entered.wait(5)
|
||||||
|
seen.append(lock.held_by_current_thread)
|
||||||
|
done.set()
|
||||||
|
t.join(5)
|
||||||
|
assert seen == [False], "another thread's hold must not read as ours"
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.integration
|
||||||
|
def test_finalizer_defers_when_gc_runs_on_the_locking_thread(
|
||||||
|
conn_params: ConnParams,
|
||||||
|
) -> None:
|
||||||
|
"""GC fires on whatever thread allocated. When that is the thread
|
||||||
|
holding the wire lock, the finalizer must write nothing — it used to
|
||||||
|
acquire the RLock reentrantly and send CLOSE/RELEASE into the running
|
||||||
|
statement, killing it with -208."""
|
||||||
|
import gc
|
||||||
|
|
||||||
|
with _connect(conn_params) as conn:
|
||||||
|
gc.disable()
|
||||||
|
try:
|
||||||
|
victim = conn.cursor(scrollable=True)
|
||||||
|
victim.execute("SELECT tabid FROM systables ORDER BY tabid")
|
||||||
|
victim.fetch_first()
|
||||||
|
# A reference cycle, so collection waits for gc rather than
|
||||||
|
# happening at the drop. Cycles are ordinary in Python.
|
||||||
|
cycle = [victim]
|
||||||
|
cycle.append(cycle)
|
||||||
|
del victim, cycle
|
||||||
|
|
||||||
|
writes: list[bytes] = []
|
||||||
|
original_write = conn._sock.write_all
|
||||||
|
conn._sock.write_all = lambda b: (
|
||||||
|
writes.append(b),
|
||||||
|
original_write(b),
|
||||||
|
)[1]
|
||||||
|
try:
|
||||||
|
with conn._wire_lock: # stand in for "mid-statement"
|
||||||
|
gc.collect()
|
||||||
|
assert writes == [], (
|
||||||
|
"finalizer wrote to the wire while a statement "
|
||||||
|
"owned it"
|
||||||
|
)
|
||||||
|
assert conn._pending_cleanup, (
|
||||||
|
"cleanup should have been deferred, not dropped"
|
||||||
|
)
|
||||||
|
finally:
|
||||||
|
conn._sock.write_all = original_write
|
||||||
|
finally:
|
||||||
|
gc.enable()
|
||||||
|
|
||||||
|
cur = conn.cursor()
|
||||||
|
cur.execute("SELECT COUNT(*) FROM systables")
|
||||||
|
assert cur.fetchone() is not None
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user