Put the repaired Cython fast paths behind an opt-in setting

docs/developerguide/cython.rst

@@ -24,6 +24,87 @@ import falls back:
 That fallback is what makes the accelerators optional, and it is also the
 single biggest hazard in maintaining them.  This page is about the hazard.
 
+.. _cython-optin:
+
+The ``cython_optimizations`` opt-in
+===================================
+
+Two of the Cython fast paths never ran -- each was guarded by a condition that
+could not become true (see :ref:`cython-drift`).  Repairing them activates code
+that has, by definition, never executed in production, so the repairs are
+behind a setting that defaults to **off**:
+
+.. sourcecode:: python
+
+    app = faust.App('myapp', cython_optimizations=True)
+
+or ``CYTHON_OPTIMIZATIONS=1`` in the environment (``FAUST_CYTHON_OPTIMIZATIONS``
+when :setting:`env_prefix` is set).  With it off, the extensions behave exactly
+as the released versions do.
+
+What it gates:
+
+* ``StreamIterator._try_get_quick_value`` -- taking values already in the
+  channel queue instead of always awaiting.
+* ``ConductorHandler`` event reuse -- decoding a message once and reusing the
+  event across channels with matching key/value types, instead of
+  deserializing once per subscribed channel.
+
+What it does **not** gate: the ``on_topic_buffer_full`` argument fix.  That one
+was wrong in *both* implementations, is not a Cython-specific change, and
+produces a metric that was simply incorrect before -- so it applies
+unconditionally.
+
+One consequence to be aware of.  While the setting is off, the Cython path and
+the pure-Python path genuinely differ.  That is not new -- it is what has
+shipped for years -- and the flag does not introduce the divergence, it makes
+it selectable.  The sharpest case is in the conductor: a reused event is never
+decoded a second time, so a channel whose payload would fail to deserialize
+raises no error when the event is reused, and raises one when it is not.  That
+changes which channels receive a message, and how many acks it takes.
+
+Consequently the parity suites run with the setting **on** -- that is the
+configuration in which the two implementations are supposed to agree.  A
+separate test in each suite pins the default-off behaviour, so the historical
+path stays covered too.
+
+.. _cython-optin-lifecycle:
+
+Retiring the setting
+--------------------
+
+The setting is **transitional**.  It exists to make adopting the repaired
+paths a decision rather than something that arrives in an upgrade, and it is
+meant to be removed, not kept.  The intended sequence:
+
+1. **Now** -- ships defaulting to ``False``.  Upgrading changes nothing.
+2. **Default flipped to** ``True`` once there is real-world evidence the
+   repaired paths behave: the parity suites passing is necessary but not
+   sufficient, since they only prove the two implementations agree under test.
+   Record the flip as ``version_changed={'<ver>': 'Enabled by default.'}``.
+3. **Deprecated** -- set ``version_deprecated`` and ``deprecation_reason`` on
+   the setting.  Users who set it explicitly get a warning; nobody else
+   notices.
+4. **Removed** -- delete the setting, both ``bint`` attributes, the branches
+   guarding the fast paths, :mod:`faust.utils.optin`, and the two
+   default-off tests.  At that point the fast paths are simply the behaviour,
+   and the parity suites no longer need a ``conf`` marker.
+
+One thing to know before step 3.
+:meth:`~faust.types.settings.params.Param.__get__` emits a
+:exc:`UserWarning` on *every read* of a deprecated setting, and faust reads
+this one itself -- once per :class:`~faust.Stream`, once per assigned
+partition.  Deprecating it naively would make faust warn at itself, at a rate
+that scales with the deployment, about a setting the user most likely never
+set.
+
+That is why the two extensions read it through
+:func:`faust.utils.optin.cython_optimizations_enabled` rather than
+``app.conf.cython_optimizations``: the helper takes the stored value and so
+stays silent, while user-facing reads still warn, which is the point of
+deprecating it.  ``tests/unit/utils/test_optin.py`` pins both halves, so step 3
+is genuinely a two-line change.
+
 .. _cython-testing:
 
 Testing the compiled code

docs/includes/settingref.txt

@@ -282,6 +282,61 @@ the second version is 2, and so on.
    use: ``app.topic(..., internal=True)``.
 
 
+.. setting:: cython_optimizations
+
+``cython_optimizations``
+------------------------
+
+:type: :class:`bool`
+:default: :const:`False`
+:environment: :envvar:`CYTHON_OPTIMIZATIONS`
+:version-introduced: 0.12.2
+
+Enable the repaired fast paths in the Cython extensions.
+
+Disabled by default, and has no effect at all unless the optional
+Cython extension modules were built.
+
+Faust ships a few hot paths twice: a pure-Python implementation, and a
+Cython one used instead when the extensions are available.  Two of the
+Cython fast paths never actually ran -- each was guarded by a condition
+that could not become true -- so for years the extensions quietly did
+more work than the Python they were meant to accelerate:
+
+* ``StreamIterator`` always awaited the channel rather than taking
+  values already sitting in the queue, and
+* ``ConductorHandler`` re-deserialized the payload once per subscribed
+  channel instead of decoding once and reusing the event.
+
+Both are repaired, but the repaired code has by definition never run in
+production, so it is opt-in.  Leaving this ``False`` keeps the
+extensions behaving exactly as the released versions do.
+
+Note this makes the Cython path differ from the pure-Python path while
+disabled -- which has always been true; the flag does not introduce the
+divergence, it just makes it selectable.  The most visible difference is
+in the conductor: a reused event is never decoded again, so a channel
+whose payload would fail to deserialize raises no error when the event
+is reused, and does when it is not.  That changes which channels receive
+a message and how many acks it takes.
+
+Enable it to get the fast paths::
+
+    app = faust.App('myapp', cython_optimizations=True)
+
+This setting is transitional: it is expected to default to ``True`` in
+a later release, then be deprecated and removed, at which point the
+fast paths are simply the behaviour.  See the developer guide for the
+sequence -- and for why faust reads this setting internally through
+:func:`faust.utils.optin.cython_optimizations_enabled` rather than
+directly, which is what keeps the eventual deprecation from warning
+once per stream and once per assigned partition.
+
+.. seealso::
+
+    The developer guide's :ref:`developers-cython` page, for what the
+    two faults were and how the implementations are held level.
+
 .. setting:: blocking_timeout
 
 ``blocking_timeout``

faust/_cython/streams.pyx

@@ -7,6 +7,7 @@ from mode.utils.futures import maybe_async, notify
 
 from faust.exceptions import Skip
 from faust.types import ChannelT, EventT
+from faust.utils.optin import cython_optimizations_enabled
 
 
 cdef class StreamIterator:
@@ -35,6 +36,7 @@ cdef class StreamIterator:
         object topics
         object acks_enabled_for
         object _skipped_value
+        bint cython_optimizations
 
     def __init__(self, object stream):
         self.stream = stream
@@ -53,6 +55,11 @@ cdef class StreamIterator:
         self.unacked = self.consumer._unacked_messages
         self.add_unacked = self.unacked.add
         self._skipped_value = self.stream._skipped_value
+        # Opt-in: see the `cython_optimizations` setting.  Read once here
+        # rather than per message, so the hot path costs a `bint` test.
+        # Via the helper, not `app.conf.<name>`, so that deprecating the
+        # setting does not emit a warning per stream -- see faust/utils/optin.py.
+        self.cython_optimizations = cython_optimizations_enabled(self.app.conf)
 
         if isinstance(self.channel, ChannelT):
             self.chan_is_channel = True
@@ -206,9 +213,16 @@ cdef class StreamIterator:
         # chan_queue_empty():`` ... ``channel_value = chan_quick_get()``), so
         # this restores the fast path the extension was meant to provide and
         # brings the two implementations back into agreement.
+        #
+        # Behind the `cython_optimizations` setting, off by default: the
+        # repaired path has never run in production, so taking it is opt-in.
+        # Disabled, this reproduces the released behaviour exactly -- always
+        # reporting "use the slow path", never reaching `get_nowait()`.
         if self.chan_is_channel:
             if self.chan_errors:
                 raise self.chan_errors.popleft()
+            if not self.cython_optimizations:
+                return (True, None)
             if self.chan_queue_empty():
                 return (True, None)
             else:

faust/transport/_cython/conductor.pyx

@@ -3,6 +3,7 @@
 from asyncio import ALL_COMPLETED, ensure_future, wait
 
 from faust.exceptions import KeyDecodeError, ValueDecodeError
+from faust.utils.optin import cython_optimizations_enabled
 
 
 cdef class ConductorHandler:
@@ -20,6 +21,7 @@ cdef class ConductorHandler:
         object wait_until_producer_ebb
         object consumer_on_buffer_full
         object consumer_on_buffer_drop
+        bint cython_optimizations
 
 
     def __init__(self, object conductor, object tp, object channels):
@@ -33,6 +35,11 @@ cdef class ConductorHandler:
         self.acquire_flow_control = self.app.flow_control.acquire
         self.wait_until_producer_ebb = self.app.producer.buffer.wait_until_ebb
         self.consumer = self.app.consumer
+        # Opt-in: see the `cython_optimizations` setting.  Read once per
+        # handler (one per assigned TP), not per message.  Via the helper, not
+        # `app.conf.<name>`, so that deprecating the setting does not emit a
+        # warning per partition -- see faust/utils/optin.py.
+        self.cython_optimizations = cython_optimizations_enabled(self.app.conf)
         # We divide `stream_buffer_maxsize` with Queue.pressure_ratio
         # find a limit to the number of messages we will buffer
         # before considering the buffer to be under high pressure.
@@ -89,9 +96,13 @@ cdef class ConductorHandler:
                         event = await chan.decode(message, propagate=True)
                         event_keyid = keyid
                         dest_event = event
-                    elif keyid == event_keyid:
+                    elif self.cython_optimizations and keyid == event_keyid:
                         dest_event = event
                     else:
+                        # Reuse is behind the `cython_optimizations` setting,
+                        # off by default: the repaired path has never run in
+                        # production.  Disabled, every channel deserializes its
+                        # own event, reproducing the released behaviour.
                         dest_event = await chan.decode(message, propagate=True)
                     if not self._put(dest_event, chan, full):
                         continue

faust/types/settings/settings.py

@@ -604,6 +604,59 @@ def agent_supervisor(self) -> Type[SupervisorStrategyT]:
             restarted).
         """
 
+    @sections.Common.setting(
+        params.Bool,
+        version_introduced="0.12.2",
+        env_name="CYTHON_OPTIMIZATIONS",
+        default=False,
+    )
+    def cython_optimizations(self) -> bool:
+        """Enable the repaired fast paths in the Cython extensions.
+
+        Disabled by default, and has no effect at all unless the optional
+        Cython extension modules were built.
+
+        Faust ships a few hot paths twice: a pure-Python implementation, and a
+        Cython one used instead when the extensions are available.  Two of the
+        Cython fast paths never actually ran -- each was guarded by a condition
+        that could not become true -- so for years the extensions quietly did
+        more work than the Python they were meant to accelerate:
+
+        * ``StreamIterator`` always awaited the channel rather than taking
+          values already sitting in the queue, and
+        * ``ConductorHandler`` re-deserialized the payload once per subscribed
+          channel instead of decoding once and reusing the event.
+
+        Both are repaired, but the repaired code has by definition never run in
+        production, so it is opt-in.  Leaving this ``False`` keeps the
+        extensions behaving exactly as the released versions do.
+
+        Note this makes the Cython path differ from the pure-Python path while
+        disabled -- which has always been true; the flag does not introduce the
+        divergence, it just makes it selectable.  The most visible difference is
+        in the conductor: a reused event is never decoded again, so a channel
+        whose payload would fail to deserialize raises no error when the event
+        is reused, and does when it is not.  That changes which channels receive
+        a message and how many acks it takes.
+
+        Enable it to get the fast paths::
+
+            app = faust.App('myapp', cython_optimizations=True)
+
+        This setting is transitional: it is expected to default to ``True`` in
+        a later release, then be deprecated and removed, at which point the
+        fast paths are simply the behaviour.  See the developer guide for the
+        sequence -- and for why faust reads this setting internally through
+        :func:`faust.utils.optin.cython_optimizations_enabled` rather than
+        directly, which is what keeps the eventual deprecation from warning
+        once per stream and once per assigned partition.
+
+        .. seealso::
+
+            The developer guide's :ref:`developers-cython` page, for what the
+            two faults were and how the implementations are held level.
+        """
+
     @sections.Common.setting(
         params.Seconds,
         env_name="BLOCKING_TIMEOUT",

faust/utils/optin.py

@@ -0,0 +1,48 @@
+"""Reading opt-in settings from faust's own internals.
+
+``cython_optimizations`` is transitional: it exists so the repaired Cython
+fast paths can be adopted deliberately rather than arriving in an upgrade, and
+it is expected to be deprecated and removed once they are the default (see
+``docs/developerguide/cython.rst``).
+
+That plan has a trap in it, which is what this module exists to avoid.
+:meth:`faust.types.settings.params.Param.__get__` emits a :exc:`UserWarning`
+on **every read** of a setting once ``version_deprecated`` is set on it -- and
+faust reads this one itself, once per :class:`~faust.Stream` and once per
+assigned partition.  Deprecating the setting would therefore make faust warn
+at itself, repeatedly, about a setting the user very likely never set and
+cannot act on.
+
+So internal reads go through :func:`cython_optimizations_enabled`, which takes
+the stored value rather than the descriptor.  User-facing reads of
+``app.conf.cython_optimizations`` are untouched and *should* warn once the
+setting is deprecated -- that is the whole point of deprecating it.
+
+Note this deliberately does not use :func:`warnings.catch_warnings` to
+suppress the warning instead: that manipulates global state and is not
+thread-safe, which matters on the free-threaded builds faust now supports.
+
+When the setting is finally removed, delete this module and the two calls to
+it.
+"""
+
+from typing import Any
+
+__all__ = ["cython_optimizations_enabled"]
+
+
+def cython_optimizations_enabled(conf: Any) -> bool:
+    """Return whether the repaired Cython fast paths are enabled.
+
+    Arguments:
+        conf: The app's :class:`~faust.types.settings.Settings`.
+
+    Reads the value the descriptor stores rather than going through the
+    descriptor, so that deprecating the setting does not make every stream and
+    every partition assignment emit a warning from inside faust.  The storage
+    attribute is looked up through the settings registry rather than
+    hard-coded, so renaming the setting cannot silently turn this into a
+    read of a non-existent attribute.
+    """
+    param = type(conf).SETTINGS["cython_optimizations"]
+    return bool(getattr(conf, param.storage_name))

tests/unit/test_cython_parity.py

@@ -155,6 +155,7 @@ def counting_anext():
 
 @requires_cython
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 async def test_cython_stream_uses_queue_fast_path(*, app) -> None:
     """The compiled iterator must take the non-blocking queue path.
 
@@ -187,6 +188,7 @@ async def test_cython_stream_uses_queue_fast_path(*, app) -> None:
 
 @requires_cython
 @pytest.mark.asyncio
+@pytest.mark.conf(cython_optimizations=True)
 async def test_cython_stream_falls_back_to_slow_path_when_empty(*, app) -> None:
     """An empty queue must still take the awaiting path.
 
@@ -215,3 +217,32 @@ async def test_cython_stream_falls_back_to_slow_path_when_empty(*, app) -> None:
         pending.cancel()
         with pytest.raises(asyncio.CancelledError):
             await pending
+
+
+@requires_cython
+@pytest.mark.asyncio
+async def test_cython_stream_fast_path_is_off_by_default(*, app) -> None:
+    """Without the opt-in, the iterator behaves as the released versions do.
+
+    `cython_optimizations` defaults to False, so the repaired fast path stays
+    dormant: every value goes through `await Channel.__anext__` exactly as it
+    did before the fix.  No `conf` marker here on purpose -- this is the
+    default an unmodified app gets.
+    """
+    assert app.conf.cython_optimizations is False
+
+    it, queue, anext_calls = _new_iterator(app)
+    for i in range(5):
+        queue.put_nowait(i)
+
+    seen = []
+    for _ in range(5):
+        value, _sensor_state = await asyncio.wait_for(it.next(), timeout=5)
+        seen.append(value)
+
+    # Same values either way; only the route differs.
+    assert seen == [0, 1, 2, 3, 4]
+    assert len(anext_calls) == 5, (
+        f"expected the slow path for all 5 values with the optimizations off, "
+        f"got {len(anext_calls)} awaits: the fast path is no longer opt-in"
+    )