-
Notifications
You must be signed in to change notification settings - Fork 43
fix: thread safety issues #614
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
793123b
d1e88b7
67b5fc4
5d11dbc
22d9583
027a326
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -61,6 +61,13 @@ def add_client_handler( | |
| handlers = _client_handlers[client][event] | ||
| handlers.append(handler) | ||
|
|
||
| # outside the lock intentionally: the immediate-fire status check acquires the registry lock, so calling it | ||
| # under _client_lock risks lock-order inversion against run_handlers_for_provider (registry lock → _client_lock). | ||
| # As a consequence, a narrow double-fire is possible: if dispatch_event(client's event) runs concurrently, it | ||
| # sets the matching provider status (enabling the immediate fire below) and then re-runs every handler for this | ||
| # client. If _run_immediate_handler lands after that status set but before dispatch snapshots the handler list, | ||
| # the handler fires twice — once here, once from dispatch. Only happens when the registered event matches the event | ||
| # being dispatched; otherwise the immediate fire is a no-op. | ||
| _run_immediate_handler(client, event, handler) | ||
|
|
||
|
|
||
|
|
@@ -78,6 +85,7 @@ def add_global_handler(event: ProviderEvent, handler: EventHandler) -> None: | |
|
|
||
| from openfeature.api import get_client # noqa: PLC0415 | ||
|
|
||
| # See comment in add_client_handler for why this runs outside the lock. | ||
| _run_immediate_handler(get_client(), event, handler) | ||
|
|
||
|
|
||
|
|
@@ -133,7 +141,6 @@ def _run_handler(handler: EventHandler, details: EventDetails) -> None: | |
|
|
||
|
|
||
| def clear() -> None: | ||
| with _global_lock: | ||
| with _global_lock, _client_lock: | ||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. minor: this increases the length of critical sections, increasing the change of contention. Not that it's called too frequently but it's moving in the opposite direction |
||
| _global_handlers.clear() | ||
| with _client_lock: | ||
| _client_handlers.clear() | ||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,4 +1,5 @@ | ||
| import logging | ||
| import threading | ||
| import typing | ||
| from collections.abc import Awaitable, Mapping, Sequence | ||
| from dataclasses import dataclass | ||
|
|
@@ -86,6 +87,7 @@ def __init__( | |
| self.version = version | ||
| self.context = context or EvaluationContext() | ||
| self.hooks = hooks or [] | ||
| self._hooks_lock = threading.RLock() | ||
|
|
||
| @property | ||
| def provider(self) -> FeatureProvider: | ||
|
|
@@ -98,7 +100,10 @@ def get_metadata(self) -> ClientMetadata: | |
| return ClientMetadata(domain=self.domain) | ||
|
|
||
| def add_hooks(self, hooks: list[Hook]) -> None: | ||
| self.hooks = self.hooks + hooks | ||
| # Guards the read-concat-store against a lost update; this practically never races under the default 5ms GIL | ||
| # switch interval, but is essential under a no-GIL build. | ||
|
Comment on lines
+103
to
+104
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. minor: misleading comment again. It's not "never" — just rarely |
||
| with self._hooks_lock: | ||
| self.hooks = self.hooks + hooks | ||
|
|
||
| def get_boolean_value( | ||
| self, | ||
|
|
@@ -468,8 +473,9 @@ def _establish_hooks_and_provider( | |
|
|
||
| def _assert_provider_status( | ||
| self, | ||
| provider: FeatureProvider, | ||
| ) -> OpenFeatureError | None: | ||
| status = self.get_provider_status() | ||
| status = provider_registry.get_provider_status(provider) | ||
| if status == ProviderStatus.NOT_READY: | ||
| return ProviderNotReadyError() | ||
| if status == ProviderStatus.FATAL: | ||
|
|
@@ -589,7 +595,7 @@ async def evaluate_flag_details_async( | |
| ) | ||
|
|
||
| try: | ||
| if provider_err := self._assert_provider_status(): | ||
| if provider_err := self._assert_provider_status(provider): | ||
| error_hooks( | ||
| flag_type, | ||
| provider_err, | ||
|
|
@@ -765,7 +771,7 @@ def evaluate_flag_details( | |
| ) | ||
|
|
||
| try: | ||
| if provider_err := self._assert_provider_status(): | ||
| if provider_err := self._assert_provider_status(provider): | ||
| error_hooks( | ||
| flag_type, | ||
| provider_err, | ||
|
|
||
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -1,5 +1,6 @@ | ||||||
| from __future__ import annotations | ||||||
|
|
||||||
| import threading | ||||||
| import typing | ||||||
| from collections.abc import Mapping, MutableMapping, Sequence | ||||||
| from datetime import datetime | ||||||
|
|
@@ -24,6 +25,7 @@ | |||||
| ] | ||||||
|
|
||||||
| _hooks: list[Hook] = [] | ||||||
| _hooks_lock = threading.RLock() | ||||||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. minor: the critical sections are very short and don't invoke anything else, so we don't need a re-entrant lock here — a normal one would do
Suggested change
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. ok, I see @gruebel asked for the |
||||||
|
|
||||||
|
|
||||||
| # https://openfeature.dev/specification/sections/hooks/#requirement-461 | ||||||
|
|
@@ -151,14 +153,21 @@ def supports_flag_value_type(self, flag_type: FlagType) -> bool: | |||||
| return True | ||||||
|
|
||||||
|
|
||||||
| # while the lock guarantees safety, even without it there was never a loss within 50.000 runs (with the default GIL | ||||||
| # switch interval of 5ms). only when the switch interval was significantly shortened to 0.1 microseconds, losses were | ||||||
| # observed without locks every now and then. with a no-GIL python, the lock would be essential | ||||||
|
Comment on lines
+156
to
+158
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. minor: this comment seems trying to justify that it's "fine" to not have a lock in some cases? We know the bug is here, and the first rule of testings tells us that we can't prove the absence of bugs with tests — only their presence. Longer switch interval doesn't make it safe — just make less likely to be observed. And as one person put it: when you serve millions of users, once-in-a-million events happen every day Anyway, this comment doesn't look like it belongs to the code — once the lock is here, there's not much point in debating/defending it |
||||||
|
|
||||||
|
|
||||||
| def add_hooks(hooks: list[Hook]) -> None: | ||||||
| global _hooks | ||||||
| _hooks = _hooks + hooks | ||||||
| with _hooks_lock: | ||||||
| global _hooks | ||||||
| _hooks = _hooks + hooks | ||||||
|
|
||||||
|
|
||||||
| def clear_hooks() -> None: | ||||||
| global _hooks | ||||||
| _hooks = [] | ||||||
| with _hooks_lock: | ||||||
| global _hooks | ||||||
| _hooks = [] | ||||||
|
|
||||||
|
|
||||||
| def get_hooks() -> list[Hook]: | ||||||
|
|
||||||
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -261,5 +261,6 @@ def emit_provider_stale(self, details: ProviderEventDetails) -> None: | |
| self.emit(ProviderEvent.PROVIDER_STALE, details) | ||
|
|
||
| def emit(self, event: ProviderEvent, details: ProviderEventDetails) -> None: | ||
| if hasattr(self, "_on_emit"): | ||
| self._on_emit(self, event, details) | ||
| on_emit = getattr(self, "_on_emit", None) | ||
| if on_emit is not None: | ||
| on_emit(self, event, details) | ||
|
Comment on lines
-264
to
+266
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. why is this change needed?
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. because of what was pointed out in the original issue: |
||
| Original file line number | Diff line number | Diff line change | ||||||||||||||||||||||||||||||||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| @@ -1,3 +1,5 @@ | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| import threading | ||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||||||||||||||||||
| from openfeature.evaluation_context import EvaluationContext | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| from openfeature.transaction_context.context_var_transaction_context_propagator import ( | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| ContextVarsTransactionContextPropagator, | ||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
@@ -21,25 +23,28 @@ | |||||||||||||||||||||||||||||||||||||||||||||||||||
| _evaluation_transaction_context_propagator: TransactionContextPropagator = ( | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| NoOpTransactionContextPropagator() | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| ) | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| _propagator_lock = threading.RLock() | ||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||||||||||||||||||
| def set_transaction_context_propagator( | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| transaction_context_propagator: TransactionContextPropagator, | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| ) -> None: | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| global _evaluation_transaction_context_propagator | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| _evaluation_transaction_context_propagator = transaction_context_propagator | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| with _propagator_lock: | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| _evaluation_transaction_context_propagator = transaction_context_propagator | ||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||||||||||||||||||
| def clear_transaction_context_propagator() -> None: | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| set_transaction_context_propagator(NoOpTransactionContextPropagator()) | ||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||||||||||||||||||
| def get_transaction_context() -> EvaluationContext: | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| return _evaluation_transaction_context_propagator.get_transaction_context() | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| with _propagator_lock: | ||||||||||||||||||||||||||||||||||||||||||||||||||||
|
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. not sure about this one here, if this is really needed or not
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Same here, I tried to ask Claude but the answer was sometimes yes and sometimes no 😅 |
||||||||||||||||||||||||||||||||||||||||||||||||||||
| propagator = _evaluation_transaction_context_propagator | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| return propagator.get_transaction_context() | ||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||||||||||||||||||
|
|
||||||||||||||||||||||||||||||||||||||||||||||||||||
| def set_transaction_context(evaluation_context: EvaluationContext) -> None: | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| global _evaluation_transaction_context_propagator | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| _evaluation_transaction_context_propagator.set_transaction_context( | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| evaluation_context | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| ) | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| with _propagator_lock: | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| propagator = _evaluation_transaction_context_propagator | ||||||||||||||||||||||||||||||||||||||||||||||||||||
| propagator.set_transaction_context(evaluation_context) | ||||||||||||||||||||||||||||||||||||||||||||||||||||
|
Comment on lines
41
to
+50
Member
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. We're holding Holding it through the call serializes all transaction-context access across threads, and might be problematic especially if somebody was doing a lot in a custom propagator.
Suggested change
Member
Author
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. I can change it, I'm really not sure myself.. |
||||||||||||||||||||||||||||||||||||||||||||||||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
minor: this comment says that we're trading one concurrency issue for another — not a good place to be in.
_run_immediate_handleris not synchronized withrun_handlers_for_provider, so a provider event may race with the immediate handler and the user may receive events in reverse order. We'd need to do something about it