diff --git a/kafka/net/backend/abstract.py b/kafka/net/backend/abstract.py index 633775c50..6d39f8b7f 100644 --- a/kafka/net/backend/abstract.py +++ b/kafka/net/backend/abstract.py @@ -38,8 +38,9 @@ Method families: * **Lifecycle** -- ``start`` / ``stop`` / ``close`` / ``on_io_thread``. -* **Scheduling** -- ``call_soon`` / ``call_soon_threadsafe`` / - ``call_soon_with_future`` / ``call_at`` / ``call_later`` / ``cancel``. +* **Scheduling** -- ``call_soon`` (thread-safe; wakes the loop only on a + cross-thread schedule) / ``call_soon_with_future`` / ``call_at`` / + ``call_later`` / ``cancel``. * **Timing** -- ``sleep`` (backend-specific awaitable; core coroutines await it). * **Connection** -- ``create_connection`` (returns a :class:`Transport`). * **Cross-thread bridge** -- ``run`` (schedule on the loop, block the caller). @@ -181,10 +182,13 @@ def on_io_thread(self) -> bool: # --- scheduling ------------------------------------------------------- def call_soon(self, task: Any) -> Any: - """Enqueue a coroutine/callable to run on the next loop iteration.""" + """Enqueue a coroutine/callable to run on the next loop iteration. - def call_soon_threadsafe(self, callback: Any) -> Any: - """``call_soon`` from another thread; wakes the loop.""" + Thread-safe: a cross-thread schedule (a running IO thread that is not + the caller) wakes the loop; an on-thread schedule skips the pointless + wakeup. Returns a cancelable handle (the selector's ``Task``, asyncio's + deferred-handle box). + """ def call_soon_with_future(self, coro: Any, *args: Any) -> NetBackendFuture: """Schedule ``coro`` and return a future that resolves with its result.""" diff --git a/kafka/net/backend/asyncio_backend.py b/kafka/net/backend/asyncio_backend.py index 6bfd62dcc..f34d21868 100644 --- a/kafka/net/backend/asyncio_backend.py +++ b/kafka/net/backend/asyncio_backend.py @@ -61,8 +61,8 @@ class _DeferredHandle: """Cancelable handle for a timer/callback armed cross-thread. ``call_later``/``call_soon`` invoked off the loop thread schedule the real - handle via ``call_soon_threadsafe``; this box lets the caller ``cancel()`` - synchronously whether or not the real handle has been armed yet. + handle via the loop's own ``call_soon_threadsafe``; this box lets the caller + ``cancel()`` synchronously whether or not the real handle has been armed yet. """ __slots__ = ('_handle', '_cancelled') @@ -232,19 +232,17 @@ def _call(): def call_soon(self, task): # On the loop thread: schedule directly. Off it (or before start()): - # route through call_soon_threadsafe so create_task/call_soon run on - # the loop thread as asyncio requires. + # route through the loop's own call_soon_threadsafe so create_task/ + # call_soon run on the loop thread as asyncio requires. That threadsafe + # hop inherently wakes the loop -- asyncio has no separate wakeup, and + # no way to enqueue cross-thread *without* waking, so the selector's + # call_soon/call_soon_threadsafe split has nothing to mirror here. if self.on_io_thread(): return self._schedule(task) - box = _DeferredHandle() - self._loop.call_soon_threadsafe(lambda: box._arm(self._schedule(task))) - return box - - def call_soon_threadsafe(self, callback): if self._closed: raise RuntimeError('AsyncioBackend closed!') box = _DeferredHandle() - self._loop.call_soon_threadsafe(lambda: box._arm(self._schedule(callback))) + self._loop.call_soon_threadsafe(lambda: box._arm(self._schedule(task))) return box def _as_callback(self, task): diff --git a/kafka/net/backend/selector.py b/kafka/net/backend/selector.py index c7afff710..6d8efff98 100644 --- a/kafka/net/backend/selector.py +++ b/kafka/net/backend/selector.py @@ -265,8 +265,8 @@ def __str__(self): def run_forever(self): """Run the event loop until stop() is called. Intended to be driven by - a dedicated IO thread. Wake-ups from other threads must go through - call_soon_threadsafe() so the select() loop returns promptly.""" + a dedicated IO thread. Cross-thread schedules go through call_soon(), + which wakes the select() loop so it returns promptly.""" self._stop = False log.info('IO loop starting (client_id=%s)', self.config['client_id']) try: @@ -424,7 +424,7 @@ async def waiter(): event.set() with self._pending_waiters_lock: self._pending_waiters[event] = state - self.call_soon_threadsafe(waiter) + self.call_soon(waiter) if not event.wait(timeout=deadline_secs): # Loop never ran the coroutine to completion within the deadline. # Leave the waiter registered: if the coroutine later finishes, its @@ -469,19 +469,35 @@ def _task_done(self, task): task.state = TaskState.DONE def call_soon(self, task): + """Schedule a coroutine/callable on the loop; return its Task handle. + + Thread-safe. Unless the caller can be proven to be running *on* the IO + thread, the closed/errored guards are enforced and the loop is woken so + a blocked select() returns promptly. On the IO thread that wakeup is + pointless -- we're already inside the loop, nothing is blocked in + select() -- so it's skipped, and the hot path (transport read/write + re-scheduling) pays nothing extra. + + The gate is ``on_io_thread()``, not "is there an IO thread": a test may + drive ``poll()`` cross-thread on an unstarted selector, and that blocked + poll still needs waking. Skipping the wakeup only when we're certainly + on the loop keeps that case correct; the cost off the loop is one + socketpair byte (harmless, and only the started IO-thread hot path is + performance-sensitive). + """ + # Wake/guard unless we're certainly on the loop thread. + threadsafe = not self.on_io_thread() + if threadsafe: + if self._exception: + raise self._exception from None + elif self._closed: + raise RuntimeError('NetworkSelector closed!') if not isinstance(task, Task): task = Task(task) self._add_ready_task(task) self._pending_tasks.add(task) - return task - - def call_soon_threadsafe(self, callback): - if self._exception: - raise self._exception from None - elif self._closed: - raise RuntimeError('NetworkSelector closed!') - task = self.call_soon(callback) - self.wakeup() + if threadsafe: + self.wakeup() return task def call_soon_with_future(self, coro, *args): @@ -494,7 +510,7 @@ async def wrapper(): future.success(await self._invoke(coro, *args)) except BaseException as exc: future.failure(exc) - self.call_soon_threadsafe(wrapper) + self.call_soon(wrapper) return future async def _invoke(self, coro, *args): diff --git a/kafka/net/wakeup_notifier.py b/kafka/net/wakeup_notifier.py index fed4d655e..bf70534b7 100644 --- a/kafka/net/wakeup_notifier.py +++ b/kafka/net/wakeup_notifier.py @@ -4,7 +4,7 @@ class WakeupNotifier: """await wakeup(timeout_secs) when either ``timeout_secs`` elapses or notify() is called -- whichever first. The notifier is safe to call - from any thread (it routes through call_soon_threadsafe). + from any thread (it routes through the thread-safe call_soon). Level-triggered: notify() arriving while no one is awaiting is latched and consumed by the next ``__call__``. This closes a lost-wakeup race @@ -24,12 +24,12 @@ def __init__(self, net): self._fut = None # Set by ``_wakeup`` when no awaiter is registered; consumed by the # next ``__call__``. All accesses run on the IO thread (notify - # routes through call_soon_threadsafe), so no lock is needed. + # routes through the thread-safe call_soon), so no lock is needed. self._pending = False # Coalescing guard: True once a ``_wakeup`` has been scheduled via # ``notify()`` but has not yet run on the IO thread. Lets ``notify()`` - # skip the redundant ``call_soon_threadsafe`` (Task alloc + socketpair - # write + selector wakeup) when a wake is already in flight. Set on + # skip the redundant ``call_soon`` (Task alloc + socketpair write + + # selector wakeup) when a wake is already in flight. Set on # user threads, cleared by ``_wakeup`` on the IO thread; cross-thread # access is GIL-atomic and the check-then-set in ``notify()`` can at # worst schedule one redundant wake, never drop one (see ``notify``). @@ -78,6 +78,6 @@ def notify(self): return self._scheduled = True try: - self._net.call_soon_threadsafe(self._wakeup) + self._net.call_soon(self._wakeup) except ReferenceError: self._scheduled = False diff --git a/kafka/producer/sender.py b/kafka/producer/sender.py index 02f197cc1..849ee16c8 100644 --- a/kafka/producer/sender.py +++ b/kafka/producer/sender.py @@ -747,8 +747,8 @@ def _produce_request(self, node_id, acks, timeout, batches): def wakeup(self): """Wake the sender loop early (e.g. when a sendable batch is appended). - Thread-safe: ``WakeupNotifier.notify`` routes through - ``call_soon_threadsafe``, so user threads may call this directly. + Thread-safe: ``WakeupNotifier.notify`` routes through the thread-safe + ``call_soon``, so user threads may call this directly. """ self._wakeup.notify() diff --git a/test/admin/test_admin_concurrent.py b/test/admin/test_admin_concurrent.py index 1a5a5d1ef..d2487a524 100644 --- a/test/admin/test_admin_concurrent.py +++ b/test/admin/test_admin_concurrent.py @@ -3,7 +3,7 @@ Verifies that multiple caller threads can safely invoke admin methods concurrently while a dedicated IO thread owns the event loop. Exercises the thread-safety foundation in KafkaConnectionManager (start/stop, -cross-thread run via Event, call_soon_threadsafe). +cross-thread run via Event, the thread-safe call_soon). """ import threading diff --git a/test/mock_broker.py b/test/mock_broker.py index d0cc70753..95b7cafa6 100644 --- a/test/mock_broker.py +++ b/test/mock_broker.py @@ -353,10 +353,10 @@ def stop(self, error=None): if transport.is_closing(): continue # abort() must run on the event loop: connection_lost mutates - # state the loop owns. call_soon_threadsafe works both when the + # state the loop owns. The thread-safe call_soon works both when the # loop runs on an IO thread and when a test drives poll() inline. try: - transport._net.call_soon_threadsafe(lambda t=transport: t.abort(error)) + transport._net.call_soon(lambda t=transport: t.abort(error)) except RuntimeError: pass # selector already closed; nothing left to abort diff --git a/test/net/backend/test_abstract.py b/test/net/backend/test_abstract.py index 9c15cb0f6..1f76e442e 100644 --- a/test/net/backend/test_abstract.py +++ b/test/net/backend/test_abstract.py @@ -20,7 +20,7 @@ # The full contract surface, kept here so a missing/renamed method fails loudly. CONTRACT_METHODS = ( 'start', 'stop', 'close', 'on_io_thread', - 'call_soon', 'call_soon_threadsafe', 'call_soon_with_future', + 'call_soon', 'call_soon_with_future', 'call_at', 'call_later', 'cancel', 'sleep', 'create_connection', 'run', 'create_future', 'wakeup', @@ -58,6 +58,13 @@ def test_readiness_primitives_and_poll_excluded(self): assert name not in CONTRACT_METHODS assert hasattr(net, name), name # still present on the selector impl + def test_call_soon_threadsafe_folded_into_call_soon(self): + # call_soon_threadsafe was merged into the thread-safe call_soon, which + # wakes the loop only on a genuine cross-thread schedule. + assert 'call_soon_threadsafe' not in CONTRACT_METHODS + net = NetworkSelector() + assert not hasattr(net, 'call_soon_threadsafe') + class TestNetTransportContract: def test_kafkatcptransport_satisfies_transport(self): diff --git a/test/net/backend/test_selector.py b/test/net/backend/test_selector.py index 88b4989c3..730e72c65 100644 --- a/test/net/backend/test_selector.py +++ b/test/net/backend/test_selector.py @@ -607,7 +607,7 @@ def wake_after_delay(): time.sleep(0.05) net.wakeup() # Schedule a task that resolves the future - net.call_soon_threadsafe(lambda: f.success(True)) + net.call_soon(lambda: f.success(True)) t = threading.Thread(target=wake_after_delay) t.start() @@ -618,15 +618,15 @@ def wake_after_delay(): assert f.succeeded() assert elapsed < 1.0 - def test_call_soon_threadsafe(self): + def test_call_soon_cross_thread(self): net = NetworkSelector() results = [] f = Future() def background(): time.sleep(0.02) - net.call_soon_threadsafe(lambda: results.append('from_thread')) - net.call_soon_threadsafe(lambda: f.success(True)) + net.call_soon(lambda: results.append('from_thread')) + net.call_soon(lambda: f.success(True)) t = threading.Thread(target=background) t.start() @@ -957,7 +957,7 @@ async def wedge(): wedged.set() release.wait(timeout=5.0) # safety cap so the suite can't hang - net.call_soon_threadsafe(wedge) + net.call_soon(wedge) assert wedged.wait(timeout=1.0), 'IO thread never entered the wedge' def _run_in_thread(self, net, coro, **kw): diff --git a/test/net/test_wakeup_notifier.py b/test/net/test_wakeup_notifier.py index 134d0e00c..18072fd82 100644 --- a/test/net/test_wakeup_notifier.py +++ b/test/net/test_wakeup_notifier.py @@ -223,7 +223,7 @@ def producer(base): def test_notify_from_other_thread(self, net, notifier): """notify() is safe to call from another thread; the wakeup - routes through call_soon_threadsafe to the IO thread.""" + routes through the thread-safe call_soon to the IO thread.""" async def task(): def background(): # Slight delay so the notifier is definitely awaiting.