From cf10ab22b91332fd88e1ac064eb5c9948f9df62b Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 21:47:23 +0000 Subject: [PATCH 1/2] Put the repaired Cython fast paths behind an opt-in setting Two of the Cython fast paths never ran, each guarded by a condition that could not become true, so the extensions quietly did more work than the Python they were meant to accelerate. Repairing them (in the two PRs below this one) activates code that has by definition never executed in production. Gate it. `cython_optimizations` defaults to False. With it off the extensions behave exactly as the released versions do, so upgrading changes nothing; users opt in per app: app = faust.App('myapp', cython_optimizations=True) or `CYTHON_OPTIMIZATIONS=1` in the environment (prefixed when `env_prefix` is set, like every other env-backed setting). The flag is read once per StreamIterator and once per ConductorHandler -- so once per stream and once per assigned TP, not per message -- into a `bint`, leaving a predictable branch on the hot path rather than an attribute lookup into `app.conf`. ## What it gates, and what it does not Gated: * `StreamIterator._try_get_quick_value` -- taking values already in the channel queue instead of always awaiting. * `ConductorHandler` event reuse -- decoding once and reusing the event across channels with matching key/value types, instead of deserializing once per subscribed channel. Not gated: the `on_topic_buffer_full` argument fix. That one was wrong in *both* implementations, is not Cython-specific, and produced a metric that was simply incorrect -- gating a wrong metric key behind a "Cython improvements" flag would be incoherent. It applies always. ## Consequence worth stating plainly While the setting is off, the Cython and pure-Python paths genuinely differ. That is not new -- it is what has shipped for years -- and the flag does not introduce the divergence, only makes it selectable. The sharpest case is 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 raises one when it is not, changing which channels receive a message and how many acks it takes. So the parity suites now run with the setting on, which is the configuration in which the two implementations are supposed to agree. Each suite also gains a test pinning the default-off behaviour, so the historical path -- the one most users will actually run -- stays covered: 5 awaits for 5 queued values in the iterator, one decode per channel in the conductor. ## Verification Suite green in every configuration: extensions built (2274 passed), absent (2208 passed), free-threaded 3.14t under PYTHON_GIL=0 (2278 passed). `mypy -p faust` clean, `extra/tools/verify_doc_defaults.py` clean, docs build clean with the setting rendered into the configuration reference. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr --- docs/developerguide/cython.rst | 44 +++++++++++++++++ docs/includes/settingref.txt | 47 +++++++++++++++++++ faust/_cython/streams.pyx | 11 +++++ faust/transport/_cython/conductor.pyx | 10 +++- faust/types/settings/settings.py | 45 ++++++++++++++++++ tests/unit/test_cython_parity.py | 31 ++++++++++++ tests/unit/transport/test_conductor_parity.py | 43 ++++++++++++++++- 7 files changed, 228 insertions(+), 3 deletions(-) diff --git a/docs/developerguide/cython.rst b/docs/developerguide/cython.rst index 4548fa4b0..844cd3649 100644 --- a/docs/developerguide/cython.rst +++ b/docs/developerguide/cython.rst @@ -24,6 +24,50 @@ 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-testing: Testing the compiled code diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt index 669894cff..84f5a91dd 100644 --- a/docs/includes/settingref.txt +++ b/docs/includes/settingref.txt @@ -282,6 +282,53 @@ 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) + +.. 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`` diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx index c2ccb3f23..b92b90337 100644 --- a/faust/_cython/streams.pyx +++ b/faust/_cython/streams.pyx @@ -35,6 +35,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 +54,9 @@ 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. + self.cython_optimizations = bool(self.app.conf.cython_optimizations) if isinstance(self.channel, ChannelT): self.chan_is_channel = True @@ -206,9 +210,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: diff --git a/faust/transport/_cython/conductor.pyx b/faust/transport/_cython/conductor.pyx index 302100c48..3aad2f8de 100644 --- a/faust/transport/_cython/conductor.pyx +++ b/faust/transport/_cython/conductor.pyx @@ -20,6 +20,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 +34,9 @@ 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. + self.cython_optimizations = bool(self.app.conf.cython_optimizations) # 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 +93,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 diff --git a/faust/types/settings/settings.py b/faust/types/settings/settings.py index 8f68b2cf4..b45892819 100644 --- a/faust/types/settings/settings.py +++ b/faust/types/settings/settings.py @@ -604,6 +604,51 @@ 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) + + .. 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", diff --git a/tests/unit/test_cython_parity.py b/tests/unit/test_cython_parity.py index c505885c2..86b5d6b39 100644 --- a/tests/unit/test_cython_parity.py +++ b/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" + ) diff --git a/tests/unit/transport/test_conductor_parity.py b/tests/unit/transport/test_conductor_parity.py index 4f3a8b928..54c3ef060 100644 --- a/tests/unit/transport/test_conductor_parity.py +++ b/tests/unit/transport/test_conductor_parity.py @@ -240,6 +240,7 @@ async def run_both(harness: Harness, scenario) -> Dict[str, Any]: # ------------------------------------------------------------------ delivery @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [1, 2, 3], indirect=True) async def test_parity__fan_out(harness) -> None: """Every subscribed channel gets the event, and refcount matches.""" @@ -259,6 +260,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) async def test_parity__no_channels(harness) -> None: """A TP with no subscribers must not touch the message.""" harness.channel_set = set() @@ -276,6 +278,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [3], indirect=True) async def test_parity__multiple_messages(harness) -> None: """A batch, to catch state carried between calls.""" @@ -292,6 +295,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [2, 4], indirect=True) async def test_parity__event_reuse_for_matching_keyid(harness) -> None: """Channels with the same (key_type, value_type) share one decode. @@ -319,6 +323,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [(2, True), (4, True)], indirect=True) async def test_parity__no_reuse_for_differing_keyid(harness) -> None: """Channels with different (key_type, value_type) each decode their own. @@ -348,6 +353,7 @@ async def scenario(handler, h): # -------------------------------------------------------------- decode errors @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [1, 3], indirect=True) @pytest.mark.parametrize( "exc_cls,bucket", @@ -378,6 +384,7 @@ async def scenario(handler, h): @requires_cython_conductor @pytest.mark.asyncio +@pytest.mark.conf(cython_optimizations=True) @pytest.mark.parametrize("harness", [3], indirect=True) async def test_parity__decode_error_on_one_channel(harness) -> None: """One channel's decode fails; the rest of the fan-out must match. @@ -400,7 +407,7 @@ async def scenario(handler, h): # ------------------------------------------------------------ buffer pressure @requires_cython_conductor @pytest.mark.asyncio -@pytest.mark.conf(stream_buffer_maxsize=2) +@pytest.mark.conf(cython_optimizations=True, stream_buffer_maxsize=2) async def test_parity__queue_full_path(harness) -> None: """When a channel queue is full the handler must await ``chan.put``. @@ -494,7 +501,7 @@ async def test_monitor_counts_buffer_full_by_tp(app, impl) -> None: @requires_cython_conductor @pytest.mark.asyncio -@pytest.mark.conf(stream_buffer_maxsize=8) +@pytest.mark.conf(cython_optimizations=True, stream_buffer_maxsize=8) async def test_parity__pressure_callbacks(harness) -> None: """High-pressure and pressure-drop callbacks must fire identically. @@ -518,3 +525,35 @@ async def scenario(handler, h): results = await run_both(harness, scenario) assert_parity(results) + + +@requires_cython_conductor +@pytest.mark.asyncio +@pytest.mark.parametrize("harness", [3], indirect=True) +async def test_event_reuse_is_off_by_default(harness) -> None: + """Without the opt-in, the conductor behaves as the released versions do. + + `cython_optimizations` defaults to False, so the repaired reuse stays + dormant and every channel deserializes its own event -- exactly as before + the fix. No `conf` marker here on purpose: this is what an unmodified app + gets. + + This is also where the Cython and pure-Python conductors legitimately + differ, so it is not a parity test. That divergence is not new; the flag + only makes it selectable. + """ + assert harness.app.conf.cython_optimizations is False + + handler = harness.build("cython") + message = harness.message() + await handler(message) + obs = harness.observations(message) + + n = len(harness.channels) + assert obs["n_decodes"] == n, ( + f"expected one decode per channel with the optimizations off, got " + f"{obs['n_decodes']} for {n} channels: reuse is no longer opt-in" + ) + # Delivery itself is unchanged -- only how many times the payload is read. + assert obs["n_delivered_total"] == n + assert obs["refcount"] == n From e0ca537b8ba1a5e0630beba9b1cd02a50a6071f7 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 7 Aug 2026 22:01:08 +0000 Subject: [PATCH 2/2] Make retiring cython_optimizations a two-line change The setting is transitional -- it exists so the repaired Cython fast paths are adopted deliberately rather than arriving in an upgrade, and it is meant to be removed, not kept. Retiring it naively has a trap in it, which this closes before anyone walks into it. `Param.__get__` emits a UserWarning on *every read* of a setting once `version_deprecated` is set, and faust reads this one itself: once per Stream, once per assigned partition. Setting `version_deprecated` would therefore make faust warn at itself, at a rate that scales with the deployment, about a setting the user most likely never set and cannot act on. Measured before the change: three StreamIterator constructions, three warnings. Both extensions now read the flag through `faust.utils.optin.cython_optimizations_enabled`, which takes the value the descriptor stores instead of going through the descriptor. Internal reads stay silent; `app.conf.cython_optimizations` still warns, which is the entire point of deprecating a setting -- a helper that disarmed that too would be worse than the noise, because nobody would ever be told to stop using it. 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 missing attribute. Deliberately not `warnings.catch_warnings()`: it manipulates global state and is not thread-safe, which matters on the free-threaded builds this branch series added support for. ## Tests tests/unit/utils/test_optin.py pins both halves of the contract -- the internal read silent under deprecation, the public read still warning -- plus an end-to-end check that three stream iterators and three conductor handlers produce zero warnings with the setting marked deprecated (three and three before). The deprecation is applied by a fixture that restores the param afterwards, so the tests need no released deprecation to run. ## Docs The developer guide gains the intended sequence: ships off, default flipped once there is real-world evidence (parity passing is necessary but not sufficient -- it only proves the two implementations agree under test), deprecated, then removed along with the branches, the helper and the default-off tests. The setting's own docstring says it is transitional, so it does not read as permanent API. Suite green in every configuration: extensions built (2280 passed), absent (2213 passed), free-threaded 3.14t under PYTHON_GIL=0 (2284 passed). mypy, verify_doc_defaults and the docs build all clean. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01K8qT5E3rnSXvw7ibNLXrVr --- docs/developerguide/cython.rst | 37 +++++++++ docs/includes/settingref.txt | 8 ++ faust/_cython/streams.pyx | 5 +- faust/transport/_cython/conductor.pyx | 7 +- faust/types/settings/settings.py | 8 ++ faust/utils/optin.py | 48 +++++++++++ tests/unit/utils/test_optin.py | 115 ++++++++++++++++++++++++++ 7 files changed, 225 insertions(+), 3 deletions(-) create mode 100644 faust/utils/optin.py create mode 100644 tests/unit/utils/test_optin.py diff --git a/docs/developerguide/cython.rst b/docs/developerguide/cython.rst index 844cd3649..7facfde9f 100644 --- a/docs/developerguide/cython.rst +++ b/docs/developerguide/cython.rst @@ -68,6 +68,43 @@ 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={'': '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 diff --git a/docs/includes/settingref.txt b/docs/includes/settingref.txt index 84f5a91dd..df39d53c8 100644 --- a/docs/includes/settingref.txt +++ b/docs/includes/settingref.txt @@ -324,6 +324,14 @@ 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 diff --git a/faust/_cython/streams.pyx b/faust/_cython/streams.pyx index b92b90337..ef0644ccc 100644 --- a/faust/_cython/streams.pyx +++ b/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: @@ -56,7 +57,9 @@ cdef class StreamIterator: 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. - self.cython_optimizations = bool(self.app.conf.cython_optimizations) + # Via the helper, not `app.conf.`, 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 diff --git a/faust/transport/_cython/conductor.pyx b/faust/transport/_cython/conductor.pyx index 3aad2f8de..cad2c3871 100644 --- a/faust/transport/_cython/conductor.pyx +++ b/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: @@ -35,8 +36,10 @@ cdef class ConductorHandler: 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. - self.cython_optimizations = bool(self.app.conf.cython_optimizations) + # handler (one per assigned TP), not per message. Via the helper, not + # `app.conf.`, 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. diff --git a/faust/types/settings/settings.py b/faust/types/settings/settings.py index b45892819..42f6746dd 100644 --- a/faust/types/settings/settings.py +++ b/faust/types/settings/settings.py @@ -643,6 +643,14 @@ def cython_optimizations(self) -> bool: 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 diff --git a/faust/utils/optin.py b/faust/utils/optin.py new file mode 100644 index 000000000..4c13cc5a2 --- /dev/null +++ b/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)) diff --git a/tests/unit/utils/test_optin.py b/tests/unit/utils/test_optin.py new file mode 100644 index 000000000..e021ae33f --- /dev/null +++ b/tests/unit/utils/test_optin.py @@ -0,0 +1,115 @@ +"""The internal read of ``cython_optimizations`` must survive deprecation. + +The setting is transitional and expected to be deprecated and then removed. +That plan has a trap in it: ``Param.__get__`` emits a ``UserWarning`` on every +read once ``version_deprecated`` is set, and faust reads this setting itself -- +once per stream, and once per assigned partition. Deprecating it naively would +make faust warn at itself, repeatedly, about a setting the user probably never +set. + +These tests pin the two halves of the contract: + +* internal reads (via ``cython_optimizations_enabled``) stay silent, and +* user-facing reads (``app.conf.cython_optimizations``) still warn, because + warning is the entire point of deprecating a setting. +""" + +import warnings + +import pytest + +from faust.utils.optin import cython_optimizations_enabled + + +@pytest.fixture() +def param(app): + """The ``cython_optimizations`` Param descriptor.""" + return type(app.conf).SETTINGS["cython_optimizations"] + + +@pytest.fixture() +def deprecated(param): + """Mark the setting deprecated for the duration of a test.""" + saved = (param.version_deprecated, param.deprecation_reason) + param.version_deprecated = "0.99.0" + param.deprecation_reason = "the fast paths are now unconditional" + try: + yield param + finally: + param.version_deprecated, param.deprecation_reason = saved + + +def test_matches_the_public_read(app) -> None: + assert cython_optimizations_enabled(app.conf) is app.conf.cython_optimizations + + +def test_default_is_false(app) -> None: + assert cython_optimizations_enabled(app.conf) is False + + +@pytest.mark.conf(cython_optimizations=True) +def test_reads_true_when_enabled(app) -> None: + assert cython_optimizations_enabled(app.conf) is True + + +def test_internal_read_is_silent_when_deprecated(app, deprecated) -> None: + """The whole reason this helper exists.""" + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + value = cython_optimizations_enabled(app.conf) + + assert value is False + ours = [w for w in caught if "cython_optimizations" in str(w.message)] + assert not ours, ( + f"reading the setting from inside faust warned: " + f"{[str(w.message) for w in ours]}. faust reads this per stream and " + f"per partition, so a deprecation would flood logs with a warning " + f"the user cannot act on." + ) + + +def test_public_read_still_warns_when_deprecated(app, deprecated) -> None: + """The helper must not disarm the deprecation for users. + + Without this, a future deprecation could be silently ineffective -- which + is worse than noisy, because nobody would ever be told to stop using it. + """ + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + app.conf.cython_optimizations + + ours = [w for w in caught if "cython_optimizations" in str(w.message)] + assert ours, "deprecating the setting no longer warns users who read it" + assert "deprecated" in str(ours[0].message) + + +def test_extensions_are_silent_when_deprecated(app, deprecated) -> None: + """The end-to-end version: neither extension warns per construction. + + The helper is only useful if the extensions actually go through it. This + builds the objects that read the setting -- one per stream, one per + assigned partition -- and asserts the deprecation stays quiet. + """ + from faust.streams import _CStreamIterator + from faust.transport.conductor import Conductor, ConductorHandler + from faust.types import TP + + if _CStreamIterator is None or ConductorHandler is None: + pytest.skip("extensions not built in place") + + conductor = Conductor(app) + topic = app.topic("foo") + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + for _ in range(3): + _CStreamIterator(app.stream(app.channel())) + for i in range(3): + ConductorHandler(conductor, TP("foo", i), {topic}) + + ours = [w for w in caught if "cython_optimizations" in str(w.message)] + assert not ours, ( + f"{len(ours)} deprecation warnings from 3 stream iterators and 3 " + f"conductor handlers. A real worker builds one of each per stream and " + f"per assigned partition, so this scales with the deployment." + )