Simplify backend api contract: merge call_soon w/ call_soon_threadsafe

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."""

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):

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):

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

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()
 

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
 

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
 

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):

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):

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.