fix(python): make async pipe transport fork-safe - #6679
Open
jeremyprime wants to merge 4 commits into
Open
Conversation
Signed-off-by: Jeremy Parr-Pearson <jeremy.parr-pearson@improving.com>
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughThe async pipe transport now detects process forks, resets inherited Python state, recreates the native pipe, and rejects parent-client reuse in child processes. Integration tests cover child operations, parent recovery, timeout handling, repeated fork cycles, and stale-client rejection. ChangesFork-safe async pipe transport
Sequence Diagram(s)sequenceDiagram
participant ParentProcess
participant ChildProcess
participant BaseClient
participant NativePipe
ParentProcess->>BaseClient: initialize async pipe
ParentProcess->>ChildProcess: fork process
ChildProcess->>BaseClient: create child client
BaseClient->>BaseClient: detect PID change and clear inherited state
BaseClient->>NativePipe: reinit_async_pipe(pipe_write_fd)
NativePipe-->>BaseClient: use recreated child pipe
ChildProcess->>BaseClient: execute commands
BaseClient-->>ChildProcess: return command results
ParentProcess->>BaseClient: execute commands after child exit
BaseClient-->>ParentProcess: return command results
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (5)
ffi/src/lib.rs (3)
1481-1487: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winDo not busy-spin on the pipe state without a yield.
This loop calls
get_async_pipe()continuously until the pipe exists. With the mutex-backed state, each iteration takes the global lock, so the spinning task competes with the response delivery path. Addstd::thread::yield_now()or a shorttokio::time::sleepinside the loop, or wait on a notification instead.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ffi/src/lib.rs` around lines 1481 - 1487, Update the ASYNC_PIPE initialization wait loop around get_async_pipe() to yield between failed checks, using std::thread::yield_now() or an equivalent short async delay/notification. Preserve the existing behavior of returning the pipe immediately once get_async_pipe() succeeds.
1142-1145: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueTake the pipe writer once.
Line 1142 and Line 1143 both call
get_async_pipe(). The second call repeats the state read and can observe a different writer than the first. Bind the writer once and branch on it.♻️ Proposed diff
- if cid != 0 && get_async_pipe().is_some() { - if let Some(w) = get_async_pipe() { - w.push_error(cid, request_id, error_type, error_string); - } - } else { + let pipe = if cid != 0 { get_async_pipe() } else { None }; + if let Some(w) = pipe { + w.push_error(cid, request_id, error_type, error_string); + } else {🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ffi/src/lib.rs` around lines 1142 - 1145, Update the async error handling around get_async_pipe so it retrieves the pipe writer once, then uses that single bound writer for the presence check and push_error call. Preserve the existing cid != 0 guard and error arguments.
831-838: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winLog the write failure before the flush thread exits.
When the flush thread returns, no further async response reaches Python. Every in-flight future then waits until its request timeout. A log record at this point names the cause (
EPIPE/EBADF) and shortens diagnosis of exactly the hang class this PR targets.🔎 Proposed diff
} else { let e = std::io::Error::last_os_error().raw_os_error().unwrap_or(0); if e == libc::EINTR || e == libc::EAGAIN { continue; } - // EPIPE/EBADF — fd is gone, exit the flush thread. - return; + // EPIPE/EBADF — fd is gone, exit the flush thread. + logger_core::log_warn( + "async-pipe", + format!("flush thread exiting: write to fd {fd} failed with errno {e}"), + ); + return; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ffi/src/lib.rs` around lines 831 - 838, In the write-failure branch of the flush thread, before the `return` reached after filtering `EINTR` and `EAGAIN`, log the non-retryable OS error using the existing logging mechanism. Include the captured error code and enough context to identify the failed write, while preserving the current retry behavior and thread exit.python/tests/async_tests/test_fork_safety.py (1)
61-61: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider
anyio.runin the child worker.The async client supports asyncio and trio.
asyncio.runlimits the child to asyncio, so the fork path is never exercised on trio.anyio.run(run)lets the worker take a backend parameter and cover both.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/tests/async_tests/test_fork_safety.py` at line 61, Replace asyncio.run in the child worker with anyio.run so run can execute using the selected async backend, including both asyncio and trio. Pass the worker’s backend parameter through the child execution path and preserve the existing run behavior.python/glide-async/python/glide/glide_client.py (1)
590-590: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winInitialize
_create_pidin__init__.
create()assigns_create_pid, so the attribute is missing on any instance that skipscreate()or fails before Line 590._execute_commandcovers this withgetattr(self, "_create_pid", 0). Assign the attribute in__init__next to_pipe_client_idinstead, and read it directly.♻️ Proposed diff
self._pipe_client_id: int = 0 + self._create_pid: int = 0 self._is_asyncio: bool = True🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@python/glide-async/python/glide/glide_client.py` at line 590, Initialize _create_pid in __init__ alongside _pipe_client_id, using the appropriate default value, and remove its initialization from create(). Update _execute_command to access self._create_pid directly instead of using getattr, while preserving the existing PID behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@ffi/src/lib.rs`:
- Around line 590-602: Replace the ASYNC_PIPE Mutex<Option<&'static
SharedPipeWriter>> with an AtomicPtr<SharedPipeWriter>. Update get_async_pipe()
to perform a lock-free atomic load and convert null to None, have
init_async_pipe use compare_exchange for first-call-wins behavior, and have
reinit_async_pipe use an atomic store to replace the inherited writer without
locking or poisoning risk.
In `@python/glide-async/python/glide/client_pool.py`:
- Around line 131-164: Make fork recovery in _create_pool(), _setup_pipe(), and
reinit_async_pipe() perform the PID-change check before acquiring
_async_pipe_lock, using a child-safe reset path that closes and clears inherited
pipe state without blocking on ASYNC_PIPE or its mutex. Reinitialize the pipe
and writer in the child only after this reset, preserving normal locking for
non-fork paths. Add a regression test that forks while transport initialization
is active and verifies child commands complete.
In `@python/glide-async/python/glide/glide_client.py`:
- Around line 368-400: The fork-reset logic is duplicated in
AsyncClientPool.__init__, allowing stale Trio state to survive. Replace the
inline glide_client global reset in AsyncClientPool.__init__ with a call to
_detect_fork_and_reset(), while preserving the required lock context and
ensuring the pool uses the shared helper as the single reset implementation.
- Around line 797-801: Extract the existing fork-process validation from
_execute_command into a reusable _check_same_process method. Invoke it
immediately after each _is_closed check in _execute_batch, _execute_script,
_update_connection_password, and _refresh_iam_token, while preserving the
existing ClosingError behavior for clients reused after fork.
In `@python/tests/async_tests/test_fork_safety.py`:
- Around line 109-123: Replace the non-blocking result reads in
python/tests/async_tests/test_fork_safety.py lines 109-123 and 185-186: in the
first result-collection block, perform exactly num_workers bounded
result_queue.get(timeout=5.0) calls and handle queue.Empty while preserving the
existing assertions; at the second site, replace result_queue.get_nowait() with
result_queue.get(timeout=5.0). Ensure queue.Empty is available for handling.
- Around line 64-68: Add a forked-child worker test in TestForkSafety that
reuses the inherited parent client and asserts client operations raise
ClosingError, covering the documented pre-fork client rejection. Integrate the
worker into the existing parametrized anyio test flow and preserve current
fork-safety coverage.
- Around line 58-59: Suppress Ruff BLE001 specifically on the broad exception
handler in the child failure-reporting path of test_fork_safety.py, while
preserving the existing catch-all behavior and result_queue reporting.
---
Nitpick comments:
In `@ffi/src/lib.rs`:
- Around line 1481-1487: Update the ASYNC_PIPE initialization wait loop around
get_async_pipe() to yield between failed checks, using std::thread::yield_now()
or an equivalent short async delay/notification. Preserve the existing behavior
of returning the pipe immediately once get_async_pipe() succeeds.
- Around line 1142-1145: Update the async error handling around get_async_pipe
so it retrieves the pipe writer once, then uses that single bound writer for the
presence check and push_error call. Preserve the existing cid != 0 guard and
error arguments.
- Around line 831-838: In the write-failure branch of the flush thread, before
the `return` reached after filtering `EINTR` and `EAGAIN`, log the non-retryable
OS error using the existing logging mechanism. Include the captured error code
and enough context to identify the failed write, while preserving the current
retry behavior and thread exit.
In `@python/glide-async/python/glide/glide_client.py`:
- Line 590: Initialize _create_pid in __init__ alongside _pipe_client_id, using
the appropriate default value, and remove its initialization from create().
Update _execute_command to access self._create_pid directly instead of using
getattr, while preserving the existing PID behavior.
In `@python/tests/async_tests/test_fork_safety.py`:
- Line 61: Replace asyncio.run in the child worker with anyio.run so run can
execute using the selected async backend, including both asyncio and trio. Pass
the worker’s backend parameter through the child execution path and preserve the
existing run behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: a873baef-9740-4447-b993-bbd18054d16e
📒 Files selected for processing (6)
CHANGELOG.mdffi/src/lib.rspython/glide-async/python/glide/client_pool.pypython/glide-async/python/glide/glide_client.pypython/glide-shared/glide_shared/_glide_ffi.pypython/tests/async_tests/test_fork_safety.py
![valkey-review-bot[bot]](https://avatars.githubusercontent.com/in/3633497?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
The PID-comparison + reinit_async_pipe mechanism is the right shape for this bug, but two things stop it short of the goal: the new Mutex that replaced OnceLock is itself the kind of state that does not survive fork(), so the recovery path can inherit it locked and hang exactly where the issue reports a hang; and the pre-fork-client guard is wired into _execute_command only, so exec()/script/password paths (and pool-created clients) still hang instead of raising, which is narrower than what the CHANGELOG entry promises.
… paths, reuse helper in pool Signed-off-by: Jeremy Parr-Pearson <jeremy.parr-pearson@improving.com>
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
ffi/src/lib.rs (1)
903-948: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueRedundant double load of
get_async_pipe()at each call site.At Line 906,
get_async_pipe().is_some()is checked, thenget_async_pipe()is called again inside the spawned future (lines 914, 924, 936). The same pattern appears at lines 1153-1154. BecauseASYNC_PIPEonly ever transitions from null to non-null or from one non-null writer to another (never back to null), the second call cannot observe a differentOptionvariant than a state already confirmed non-null by the first. Fetch the writer once and reuse theOptionto avoid the extra atomic load per call site.♻️ Example simplification for handle_custom_error
- if cid != 0 && get_async_pipe().is_some() { - if let Some(w) = get_async_pipe() { - w.push_error(cid, request_id, error_type, error_string); - } - } else { + if cid != 0 { + if let Some(w) = get_async_pipe() { + w.push_error(cid, request_id, error_type, error_string); + } else { + unsafe { + Self::send_async_custom_error( + failure_callback, + error_string, + error_type, + request_id, + ) + }; + } + } else { unsafe { Self::send_async_custom_error( failure_callback, error_string, error_type, request_id, ) }; }Also applies to: 1150-1156
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@ffi/src/lib.rs` around lines 903 - 948, Update the async pipe handling around the request completion task and the analogous handle_custom_error path to load get_async_pipe() once, retain the confirmed writer Option, and move or reuse it inside the spawned future instead of calling get_async_pipe() again at each push_success or push_error call. Preserve the existing null-pipe fallback and response behavior.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@ffi/src/lib.rs`:
- Around line 743-758: Update init_async_pipe and reinit_async_pipe so pipe
flush threads are started only after the writer becomes the active ASYNC_PIPE
owner, or explicitly shut down any thread before discarding a rejected or
replaced SharedPipeWriter. Ensure failed first-call-wins CAS attempts and
non-fork reinitialization do not leave background threads running for the
process lifetime.
---
Nitpick comments:
In `@ffi/src/lib.rs`:
- Around line 903-948: Update the async pipe handling around the request
completion task and the analogous handle_custom_error path to load
get_async_pipe() once, retain the confirmed writer Option, and move or reuse it
inside the spawned future instead of calling get_async_pipe() again at each
push_success or push_error call. Preserve the existing null-pipe fallback and
response behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: b4a297f1-528e-4511-8d71-6e743206dddd
📒 Files selected for processing (4)
ffi/src/lib.rspython/glide-async/python/glide/client_pool.pypython/glide-async/python/glide/glide_client.pypython/tests/async_tests/test_fork_safety.py
🚧 Files skipped from review as they are similar to previous changes (2)
- python/glide-async/python/glide/glide_client.py
- python/glide-async/python/glide/client_pool.py
Signed-off-by: Jeremy Parr-Pearson <jeremy.parr-pearson@improving.com>
Collaborator
Author
|
@coderabbitai help |
ChatThere are 3 ways to chat with CodeRabbit:
CodeRabbit commands
Other keywords and placeholders
CodeRabbit configuration file (
|
Collaborator
Author
|
@coderabbitai rate limit |
|
Your plan includes PR reviews subject to rate limits. Reviews are available now. |
Collaborator
Author
|
@coderabbitai configuration |
|
Configuration used: Organization UI CodeRabbit configuration# Source: Organization UI (base)
language: en-US
# Source: Organization UI (base)
tone_instructions: ''
# Source: Organization UI (base)
early_access: false
# Source: Organization UI (base)
enable_free_tier: true
# Source: Organization UI (base)
inheritance: false
reviews:
# Source: Organization UI (base)
profile: chill
# Source: Organization UI (base)
request_changes_workflow: false
# Source: Organization UI (base)
high_level_summary: false
# Source: Organization UI (base)
high_level_summary_instructions: ''
# Source: Organization UI (base)
high_level_summary_placeholder: '`@coderabbitai` summary'
# Source: Organization UI (base)
high_level_summary_in_walkthrough: false
# Source: Organization UI (base)
auto_title_placeholder: '`@coderabbitai`'
# Source: Organization UI (base)
auto_title_instructions: ''
# Source: Organization UI (base)
review_status: false
# Source: Organization UI (base)
review_details: false
# Source: defaults
review_progress: true
# Source: Organization UI (base)
commit_status: true
# Source: Organization UI (base)
fail_commit_status: false
# Source: Organization UI (base)
collapse_walkthrough: true
# Source: Organization UI (base)
changed_files_summary: true
# Source: Organization UI (base)
sequence_diagrams: true
# Source: Organization UI (base)
estimate_code_review_effort: false
# Source: Organization UI (base)
assess_linked_issues: true
# Source: Organization UI (base)
related_issues: true
# Source: Organization UI (base)
related_prs: true
# Source: Organization UI (base)
suggested_labels: true
# Source: defaults
labeling_instructions: []
# Source: defaults
mutually_exclusive_groups: {}
# Source: Organization UI (base)
auto_apply_labels: false
# Source: Organization UI (base)
suggested_reviewers: true
# Source: Organization UI (base)
auto_assign_reviewers: false
# Source: defaults
suggested_reviewers_instructions: []
# Source: Organization UI (base)
in_progress_fortune: true
# Source: Organization UI (base)
poem: false
# Source: Organization UI (base)
enable_prompt_for_ai_agents: true
# Source: defaults
path_filters: []
# Source: defaults
path_instructions: []
# Source: Organization UI (base)
abort_on_close: true
# Source: Organization UI (base)
disable_cache: false
slop_detection:
# Source: Organization UI (base)
enabled: true
auto_review:
# Source: Organization UI (base)
enabled: true
# Source: Organization UI (base)
description_keyword: ''
# Source: Organization UI (base)
auto_incremental_review: true
# Source: Organization UI (base)
auto_pause_after_reviewed_commits: 5
# Source: defaults
ignore_title_keywords: []
# Source: defaults
labels: []
# Source: Organization UI (base)
drafts: false
# Source: defaults
base_branches: []
# Source: defaults
ignore_usernames: []
finishing_touches:
docstrings:
# Source: Organization UI (base)
enabled: false
unit_tests:
# Source: Organization UI (base)
enabled: false
simplify:
# Source: Organization UI (base)
enabled: false
autofix:
# Source: defaults
enabled: true
fix_ci:
# Source: defaults
enabled: true
resolve_merge_conflict:
# Source: defaults
enabled: true
# Source: defaults
custom: []
pre_merge_checks:
# Source: Organization UI (base)
override_requested_reviewers_only: false
docstrings:
# Source: Organization UI (base)
mode: warning
# Source: Organization UI (base)
threshold: 80
title:
# Source: Organization UI (base)
mode: warning
# Source: Organization UI (base)
requirements: ''
description:
# Source: Organization UI (base)
mode: warning
issue_assessment:
# Source: Organization UI (base)
mode: warning
# Source: defaults
custom_checks: []
# Source: defaults
post_merge_actions: []
tools:
ast-grep:
# Source: defaults
rule_dirs: []
# Source: defaults
util_dirs: []
# Source: Organization UI (base)
essential_rules: true
# Source: defaults
packages: []
shellcheck:
# Source: Organization UI (base)
enabled: true
ruff:
# Source: Organization UI (base)
enabled: true
markdownlint:
# Source: Organization UI (base)
enabled: true
github-checks:
# Source: Organization UI (base)
enabled: true
# Source: Organization UI (base)
timeout_ms: 90000
languagetool:
# Source: Organization UI (base)
enabled: true
# Source: defaults
enabled_rules: []
# Source: defaults
disabled_rules: []
# Source: defaults
enabled_categories: []
# Source: defaults
disabled_categories: []
# Source: Organization UI (base)
enabled_only: false
# Source: Organization UI (base)
level: default
biome:
# Source: Organization UI (base)
enabled: true
hadolint:
# Source: Organization UI (base)
enabled: true
swiftlint:
# Source: Organization UI (base)
enabled: true
phpstan:
# Source: Organization UI (base)
enabled: true
# Source: Organization UI (base)
level: default
phpmd:
# Source: Organization UI (base)
enabled: true
phpcs:
# Source: Organization UI (base)
enabled: true
golangci-lint:
# Source: Organization UI (base)
enabled: true
yamllint:
# Source: Organization UI (base)
enabled: true
gitleaks:
# Source: Organization UI (base)
enabled: true
trufflehog:
# Source: Organization UI (base)
enabled: true
checkov:
# Source: Organization UI (base)
enabled: true
tflint:
# Source: Organization UI (base)
enabled: true
detekt:
# Source: Organization UI (base)
enabled: true
eslint:
# Source: Organization UI (base)
enabled: true
e18e:
# Source: defaults
enabled: true
flake8:
# Source: Organization UI (base)
enabled: true
fbinfer:
# Source: defaults
enabled: true
# Source: defaults
enable_java: false
fortitudeLint:
# Source: Organization UI (base)
enabled: true
rubocop:
# Source: Organization UI (base)
enabled: true
buf:
# Source: Organization UI (base)
enabled: true
regal:
# Source: Organization UI (base)
enabled: true
actionlint:
# Source: Organization UI (base)
enabled: true
zizmor:
# Source: defaults
enabled: true
pmd:
# Source: Organization UI (base)
enabled: true
clang:
# Source: Organization UI (base)
enabled: true
cppcheck:
# Source: Organization UI (base)
enabled: true
opengrep:
# Source: Organization UI (base)
enabled: true
semgrep:
# Source: Organization UI (base)
enabled: true
circleci:
# Source: Organization UI (base)
enabled: true
clippy:
# Source: Organization UI (base)
enabled: true
sqlfluff:
# Source: Organization UI (base)
enabled: true
squawk:
# Source: defaults
enabled: true
trivy:
# Source: Organization UI (base)
enabled: true
prismaLint:
# Source: Organization UI (base)
enabled: true
pylint:
# Source: Organization UI (base)
enabled: true
oxc:
# Source: Organization UI (base)
enabled: true
shopifyThemeCheck:
# Source: Organization UI (base)
enabled: true
luacheck:
# Source: Organization UI (base)
enabled: true
brakeman:
# Source: Organization UI (base)
enabled: true
dotenvLint:
# Source: Organization UI (base)
enabled: true
htmlhint:
# Source: Organization UI (base)
enabled: true
stylelint:
# Source: Organization UI (base)
enabled: true
checkmake:
# Source: Organization UI (base)
enabled: true
osvScanner:
# Source: Organization UI (base)
enabled: true
oasdiff:
# Source: defaults
enabled: true
reactDoctor:
# Source: defaults
enabled: true
presidio:
# Source: Organization UI (base)
enabled: false
blinter:
# Source: Organization UI (base)
enabled: true
smartyLint:
# Source: Organization UI (base)
enabled: true
emberTemplateLint:
# Source: Organization UI (base)
enabled: true
skillspector:
# Source: defaults
enabled: true
psscriptanalyzer:
# Source: Organization UI (base)
enabled: true
chat:
# Source: Organization UI (base)
art: true
# Source: Organization UI (base)
allow_non_org_members: true
# Source: Organization UI (base)
auto_reply: true
integrations:
jira:
# Source: Organization UI (base)
usage: auto
linear:
# Source: Organization UI (base)
usage: auto
knowledge_base:
# Source: Organization UI (base)
opt_out: false
web_search:
# Source: Organization UI (base)
enabled: true
code_guidelines:
# Source: Organization UI (base)
enabled: true
# Source: defaults
filePatterns: []
learnings:
# Source: Organization UI (base)
scope: auto
# Source: defaults
approval_delay: 0
issues:
# Source: Organization UI (base)
scope: auto
jira:
# Source: Organization UI (base)
usage: auto
# Source: defaults
project_keys: []
# Source: defaults
excluded_project_keys: []
linear:
# Source: Organization UI (base)
usage: auto
# Source: defaults
team_keys: []
pull_requests:
# Source: Organization UI (base)
scope: auto
mcp:
# Source: Organization UI (base)
usage: auto
# Source: defaults
disabled_servers: []
# Source: defaults
automatic_repository_linking: false
# Source: defaults
linked_repositories: []
code_generation:
docstrings:
# Source: Organization UI (base)
language: en-US
# Source: defaults
path_instructions: []
unit_tests:
# Source: defaults
path_instructions: []
issue_enrichment:
auto_enrich:
# Source: Organization UI (base)
enabled: false
planning:
# Source: Organization UI (base)
enabled: true
auto_planning:
# Source: Organization UI (base)
enabled: true
# Source: defaults
labels: []
labeling:
# Source: defaults
labeling_instructions: []
# Source: Organization UI (base)
auto_apply_labels: false
|
Signed-off-by: Jeremy Parr-Pearson <jeremy.parr-pearson@improving.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Make the Python async pipe transport fork-safe. After
fork(), the flush thread introduced in #5637 is gone butOnceLockprevented reinitialization, causing all commands in forked child processes (e.g. PySpark workers) to hang indefinitely.Issue link
Closes #6673
Features / Behaviour Changes
Implementation
Rust (
ffi/src/lib.rs):OnceLockwithMutex<Option<>>so the pipe can be reinitializedreinit_async_pipe()FFI function for post-fork recoverycreate_pipe_writer()helperPython (
glide_client.py,client_pool.py):_setup_pipe(), close inherited fds, callreinit_async_pipeClosingErrorif a pre-fork client is used in a child processCFFI (
_glide_ffi.py):reinit_async_pipedeclarationTesting
test_fork_safety.pywith 3 integration tests:test_forked_children_can_use_clients: parent inits pipe, forks 3 children, children create clients and issue commands (core regression test)test_parent_functional_after_fork: parent remains operational after children fork and completetest_multiple_fork_cycles: 3 sequential fork/use/exit cycles all workChecklist
make *-linttargets) and Prettier has been run (make prettier-fix).