Span Buffer Multiprocess Enhancement with Health Monitoring - #5
Span Buffer Multiprocess Enhancement with Health Monitoring#5CodingKylo wants to merge 1 commit into
Conversation
![re-entry-ai[bot]](https://avatars.githubusercontent.com/in/3466347?s=60&v=4){kind=small}
There was a problem hiding this comment.
🛡️ re-entry.ai Code Review
@CodingKylo — Please review the following assessment:
🟡 Risk Score: 44/100 · MEDIUM
| Dimension | Level |
|---|---|
| Likelihood | 🟠 High |
| Impact | 🟡 Medium |
| Detectability | 🟢 Low |
Intent
Add a configurable, multiprocess span flusher with health/backpressure state tracking and tests to validate drift/processing behavior under the new model.
Summary
The PR wires a new CLI flag (--flusher-processes) through the process strategy factory into SpanFlusher, and updates SpanFlusher to manage multiple flusher processes (one per shard group) instead of a single shared thread/process. The main risks are (1) contract breakage: flusher health/backpressure state fields appear to have changed shape, which can silently break other consumers/callers that still reference the old attributes, and (2) correctness/flakiness: tests now rely on time.sleep(0.1) after drift changes, which is a classic source of CI flakiness under multiprocessing/spawn. You should verify all repo references to the old flusher state attributes and ensure the CLI option plumbing is actually honored end-to-end (not just in unit construction).
🎯 Review Focus
Verify the flusher state/health/backpressure contract and the end-to-end CLI plumbing: ensure no other code still expects the old backpressure_since/healthy_since/process_restarts attributes, and confirm --flusher-processes actually constrains the number of spawned flusher workers in production (not just in unit tests).
Key Findings
- 🚨 [src/sentry/spans/consumers/process/flusher.py:L44-L53] CRITICAL:
self.max_processes = max_processes or len(buffer.assigned_shards)— ifflusher_processesis passed as 0 (or any falsy value), this silently falls back tolen(buffer.assigned_shards), defeating the limit and potentially spawning far more processes than intended. Fix: validate and clampmax_processesexplicitly at the boundary (factory/CLI) and in SpanFlusher:if max_processes is None: self.max_processes = len(...); else: if max_processes < 1: raise ValueError(...); self.max_processes = max_processes. ⚠️ [src/sentry/spans/consumers/process/flusher.py:L44-L53] WARNING:self.num_processes = min(self.max_processes, len(buffer.assigned_shards))— ifbuffer.assigned_shardscan ever be empty,self.num_processesbecomes 0, which can lead to empty process maps and later code paths that assume at least one process exists. Fix: enforcelen(buffer.assigned_shards) >= 1or handle the empty case explicitly (e.g., don’t start flusher workers and ensure poll/join are no-ops).⚠️ [src/sentry/consumers/init.py:L427-L441] WARNING: CLI option plumbing usesclick.Option(["--flusher-processes", "flusher_processes"], ...)— the second entry is the parameter name, but the factory signature expectsflusher_processes: int | None = None. If click is configured elsewhere to pass different naming (or if the option is not included in the factory call), this will silently default to 1 and not actually limit processes. Fix: add an integration test that asserts the CLI-provided value reachesSpanFlusher.max_processes(not just factory construction), and ensure the factory invocation uses the same keyword (flusher_processes=...).⚠️ [tests/sentry/spans/consumers/process/test_consumer.py:L57-L66] WARNING:time.sleep(0.1)after mutatingcurrent_drift— this makes the test timing-sensitive and can be flaky under load/CI variance, especially with multiprocessing spawn. Fix: replace sleep with deterministic synchronization (e.g., poll a metric/state that indicates the flusher observed the drift change, or use an Event/Condition exposed by the flusher for tests).
✅ Action Checklist
- [ ] CRITICAL [src/sentry/spans/consumers/process/flusher.py:L44-L53] —
self.max_processes = max_processes or len(buffer.assigned_shards)— ifflusher_processesis passed as 0 (or any falsy value), this silently falls back tolen(buffer.assigned_shards), defeating the limit and potentially spawning far more processes than intended. Fix: validate and clampmax_processesexplicitly at the boundary (factory/CLI) and in SpanFlusher:if max_processes is None: self.max_processes = len(...); else: if max_processes < 1: raise ValueError(...); self.max_processes = max_processes. - [ ] WARNING [src/sentry/spans/consumers/process/flusher.py:L44-L53] —
self.num_processes = min(self.max_processes, len(buffer.assigned_shards))— ifbuffer.assigned_shardscan ever be empty,self.num_processesbecomes 0, which can lead to empty process maps and later code paths that assume at least one process exists. Fix: enforcelen(buffer.assigned_shards) >= 1or handle the empty case explicitly (e.g., don’t start flusher workers and ensure poll/join are no-ops). - [ ] WARNING [src/sentry/consumers/init.py:L427-L441] — CLI option plumbing uses
click.Option(["--flusher-processes", "flusher_processes"], ...)— the second entry is the parameter name, but the factory signature expectsflusher_processes: int | None = None. If click is configured elsewhere to pass different naming (or if the option is not included in the factory call), this will silently default to 1 and not actually limit processes. Fix: add an integration test that asserts the CLI-provided value reachesSpanFlusher.max_processes(not just factory construction), and ensure the factory invocation uses the same keyword (flusher_processes=...). - [ ] WARNING [tests/sentry/spans/consumers/process/test_consumer.py:L57-L66] —
time.sleep(0.1)after mutatingcurrent_drift— this makes the test timing-sensitive and can be flaky under load/CI variance, especially with multiprocessing spawn. Fix: replace sleep with deterministic synchronization (e.g., poll a metric/state that indicates the flusher observed the drift change, or use an Event/Condition exposed by the flusher for tests). - [ ] SUGGESTION — Contract audit: search for all references to the old flusher health/backpressure fields (e.g.,
backpressure_since,healthy_since,process_restarts) and update them or provide backward-compatible aliases. The cross-file analysis indicates the state shape changed to per-process/per-shard maps; ensure no other code still reads the old attributes. [src/sentry/spans/consumers/process/flusher.py:L44-L53] - [ ] SUGGESTION — Harden CLI validation: in
src/sentry/consumers/__init__.py, settype=click.IntRange(min=1)(or equivalent) for--flusher-processesinstead oftype=intwith a default. This prevents the falsy/invalid values that currently fall back due tomax_processes or ...logic. [src/sentry/consumers/init.py:L427-L441] - [ ] SUGGESTION — Make the flusher process distribution test assert externally observable behavior rather than internal fields. Current tests assert
len(flusher.processes),flusher.max_processes,flusher.num_processes(fragile if internals change). Prefer asserting which shards are flushed/produced to Kafka (or that at most N worker processes actually handle shard IDs). [tests/sentry/spans/consumers/process/test_consumer.py:L90-L120]
Suggestions
- Contract audit: search for all references to the old flusher health/backpressure fields (e.g.,
backpressure_since,healthy_since,process_restarts) and update them or provide backward-compatible aliases. The cross-file analysis indicates the state shape changed to per-process/per-shard maps; ensure no other code still reads the old attributes. [src/sentry/spans/consumers/process/flusher.py:L44-L53] - Harden CLI validation: in
src/sentry/consumers/__init__.py, settype=click.IntRange(min=1)(or equivalent) for--flusher-processesinstead oftype=intwith a default. This prevents the falsy/invalid values that currently fall back due tomax_processes or ...logic. [src/sentry/consumers/init.py:L427-L441] - Make the flusher process distribution test assert externally observable behavior rather than internal fields. Current tests assert
len(flusher.processes),flusher.max_processes,flusher.num_processes(fragile if internals change). Prefer asserting which shards are flushed/produced to Kafka (or that at most N worker processes actually handle shard IDs). [tests/sentry/spans/consumers/process/test_consumer.py:L90-L120]
Posted by re-entry.ai · Risk governance for autonomous engineering teams
There was a problem hiding this comment.
🛡️ re-entry.ai Code Review
🔴 Risk Score: 62/100 · HIGH
| Dimension | Level |
|---|---|
| Likelihood | 🟠 High |
| Impact | 🟠 High |
| Detectability | 🟡 Medium |
Intent
Enhance the span flusher to run in a multiprocessing model with configurable parallelism and per-process health/backpressure tracking.
Summary
The change replaces the prior single background-thread flusher with a multiprocessing manager that creates processes based on shard distribution, and adds a new CLI option (--flusher-processes) to cap the number of flusher processes. The highest risks are correctness around process/shard mapping and lifecycle state when shard sets are empty or when max_processes is 0/None due to falsy defaulting. The reviewer must verify that the flusher never computes num_processes/max_processes as 0 in ways that break modulo/division logic, and that all health/backpressure bookkeeping and tests consistently use the new per-process state fields. Also verify the new test’s wall-clock sleep is not masking race conditions and causing CI flakiness.
🎯 Review Focus
Verify the flusher’s process/shard mapping invariants: ensure max_processes/num_processes can never become 0 due to empty shard sets or falsy defaulting, and confirm all health/backpressure bookkeeping and tests consistently use the new per-process state model.
Key Findings
⚠️ [src/sentry/spans/consumers/process/flusher.py:L44] WARNING:self.max_processes = max_processes or len(buffer.assigned_shards)— This treats an explicitmax_processes=0as falsy and overrides it withlen(buffer.assigned_shards), creating cross-file ambiguity with the CLI/factory boundary (factory forwardsflusher_processesdirectly). Fix: change toif max_processes is None: self.max_processes = len(buffer.assigned_shards) else: self.max_processes = max_processesand validate at the boundary (CLI/factory) thatflusher_processesis eitherNoneor>= 1(or explicitly allow 0 and implement a safe no-op mode).
✅ Action Checklist
- [ ] WARNING [src/sentry/spans/consumers/process/flusher.py:L44] —
self.max_processes = max_processes or len(buffer.assigned_shards)— This treats an explicitmax_processes=0as falsy and overrides it withlen(buffer.assigned_shards), creating cross-file ambiguity with the CLI/factory boundary (factory forwardsflusher_processesdirectly). Fix: change toif max_processes is None: self.max_processes = len(buffer.assigned_shards) else: self.max_processes = max_processesand validate at the boundary (CLI/factory) thatflusher_processesis eitherNoneor>= 1(or explicitly allow 0 and implement a safe no-op mode). - [ ] SUGGESTION — [src/sentry/spans/consumers/process/flusher.py:L44] Add an invariant check right after computing
num_processesto prevent silent broken state:if self.num_processes < 1: raise ValueError("SpanFlusher requires at least one process")or implement a deterministic no-op path whenbuffer.assigned_shardsis empty. - [ ] SUGGESTION — [src/sentry/spans/consumers/process/factory.py:L69] Validate
flusher_processesbefore forwarding: inProcessSpansStrategyFactory.__init__, enforceif flusher_processes is not None and flusher_processes < 1: raise ValueError(...). This prevents the falsy-defaulting bug from ever being reachable and makes CLI behavior predictable. - [ ] SUGGESTION — [tests/sentry/spans/consumers/process/test_consumer.py:L57] Remove the wall-clock sleep: replace
time.sleep(0.1)with a deterministic synchronization mechanism (e.g., a threading/Event/Condition that the flusher sets when it has processed the drift change, or poll a condition with a bounded timeout). This will reduce CI flakiness and make the test assert the actual completion condition rather than “wait long enough.” - [ ] SUGGESTION — [src/sentry/spans/consumers/process/flusher.py:L44] After the refactor to per-process health/backpressure state, ensure all callers/tests use the new fields (e.g.,
process_backpressure_since) and not removed single-process fields (backpressure_since/healthy_since). Add a small internal method likeget_backpressure_since(process_id)or backward-compatible aliases if tests still reference old names.
Suggestions
- [src/sentry/spans/consumers/process/flusher.py:L44] Add an invariant check right after computing
num_processesto prevent silent broken state:if self.num_processes < 1: raise ValueError("SpanFlusher requires at least one process")or implement a deterministic no-op path whenbuffer.assigned_shardsis empty. - [src/sentry/spans/consumers/process/factory.py:L69] Validate
flusher_processesbefore forwarding: inProcessSpansStrategyFactory.__init__, enforceif flusher_processes is not None and flusher_processes < 1: raise ValueError(...). This prevents the falsy-defaulting bug from ever being reachable and makes CLI behavior predictable. - [tests/sentry/spans/consumers/process/test_consumer.py:L57] Remove the wall-clock sleep: replace
time.sleep(0.1)with a deterministic synchronization mechanism (e.g., a threading/Event/Condition that the flusher sets when it has processed the drift change, or poll a condition with a bounded timeout). This will reduce CI flakiness and make the test assert the actual completion condition rather than “wait long enough.” - [src/sentry/spans/consumers/process/flusher.py:L44] After the refactor to per-process health/backpressure state, ensure all callers/tests use the new fields (e.g.,
process_backpressure_since) and not removed single-process fields (backpressure_since/healthy_since). Add a small internal method likeget_backpressure_since(process_id)or backward-compatible aliases if tests still reference old names.
📝 This review includes 3 inline comments (1 critical, 2 warnings)
Posted by re-entry.ai · Risk governance for autonomous engineering teams
| @@ -42,27 +44,53 @@ def __init__( | |||
| self, | |||
There was a problem hiding this comment.
🚨 CRITICAL
max_processes now defaults to len(buffer.assigned_shards), but there is no guard for the empty-shard case. If assigned_shards is empty, self.max_processes becomes 0 and self.num_processes is also 0, which later feeds a modulo/division-based shard assignment path and can crash during initialization. The root cause is that the constructor assumes every buffer has at least one shard instead of validating the API boundary.
| } | ||
| self.process_restarts = {process_index: 0 for process_index in range(self.num_processes)} | ||
| self.buffers: dict[int, SpansBuffer] = {} | ||
|
|
There was a problem hiding this comment.
The flusher no longer stores the original buffer on self, but the rest of the class now manages per-process buffers and process state. That is only safe if every later code path was updated to use self.buffers/process_to_shards_map; otherwise restart/health logic will dereference a missing attribute or operate on the wrong buffer. The underlying issue is a state-model refactor that removed the single-buffer invariant without fully updating the rest of the class.
| @@ -56,6 +57,10 @@ def add_commit(offsets, force=False): | |||
| step.poll() | |||
There was a problem hiding this comment.
This test now introduces a real time.sleep(0.1) after advancing drift. That makes the test depend on wall-clock scheduling and can still be flaky under CI load, especially since the suite also monkeypatches time.sleep earlier in the test. Prefer a deterministic synchronization primitive or polling with a bounded timeout instead of a fixed sleep.
There was a problem hiding this comment.
🛡️ re-entry.ai Code Review
@CodingKylo — Please review the following assessment:
🔴 Risk Score: 53/100 · HIGH
| Dimension | Level |
|---|---|
| Likelihood | 🟠 High |
| Impact | 🟠 High |
| Detectability | 🟡 Medium |
Intent
Update the span flusher to run in a multiprocessing model with configurable process count and per-process health/backpressure tracking, and add tests to validate the new process limiting behavior.
Summary
Behaviorally, the flusher is refactored from a single background thread to a multiprocessing manager that creates multiple processes and maps shards to processes via modulo partitioning. The PR also introduces a new CLI flag (--flusher-processes) and wires it through the factory into SpanFlusher.max_processes. The highest risks to verify are (1) correctness/contract changes around health/backpressure state (single shared values vs per-process tracking) and (2) test reliability and brittleness due to timing sleeps and assertions against internal multiprocessing topology. Additionally, the new CLI wiring must be confirmed end-to-end so the configured flusher_processes value actually reaches SpanFlusher at runtime with correct default/None semantics.
🎯 Review Focus
Verify the end-to-end contract and runtime wiring of the new per-process health/backpressure state and the --flusher-processes configuration: ensure no remaining code paths still read the old single-value fields and that the configured process count deterministically controls SpanFlusher.max_processes in production.
Key Findings
- 💡 [src/sentry/spans/consumers/process/flusher.py:L44-L53] SEVERITY: Potential contract break—health/backpressure state appears to have been removed/changed from shared mp.Value fields (backpressure_since/healthy_since) to per-process tracking, but call sites/tests may still rely on the old single-value contract; fix by providing backward-compatible aliases on SpanFlusher (e.g., properties that aggregate per-process values) or updating all consumers of the old fields to use the new per-process structure.
✅ Action Checklist
- [ ] WARNING [src/sentry/spans/consumers/process/flusher.py:L44-L53] — SEVERITY: Potential contract break—health/backpressure state appears to have been removed/changed from shared mp.Value fields (backpressure_since/healthy_since) to per-process tracking, but call sites/tests may still rely on the old single-value contract; fix by providing backward-compatible aliases on SpanFlusher (e.g., properties that aggregate per-process values) or updating all consumers of the old fields to use the new per-process structure.
- [ ] SUGGESTION — [src/sentry/consumers/init.py:L427-L441] Add an integration/unit test that asserts the click option value for --flusher-processes is actually propagated into ProcessSpansStrategyFactory and then into SpanFlusher.max_processes (including default=1 and explicit overrides).
- [ ] SUGGESTION — [tests/sentry/spans/consumers/process/test_consumer.py:L92-L103] Stop asserting on private/internal factory fields like fac._flusher; instead expose a small read-only diagnostic API on the factory/flusher (e.g., get_process_count(), get_process_to_shards_map()) and assert on that stable surface.
- [ ] SUGGESTION — [tests/sentry/spans/consumers/process/test_consumer.py:L101-L105] Ensure shard-distribution assertions run after shard assignment is complete (move after step.join() or poll until total_shards matches expected) to avoid race conditions if shard mapping is populated asynchronously.
Suggestions
- [src/sentry/consumers/init.py:L427-L441] Add an integration/unit test that asserts the click option value for --flusher-processes is actually propagated into ProcessSpansStrategyFactory and then into SpanFlusher.max_processes (including default=1 and explicit overrides).
- [tests/sentry/spans/consumers/process/test_consumer.py:L92-L103] Stop asserting on private/internal factory fields like fac._flusher; instead expose a small read-only diagnostic API on the factory/flusher (e.g., get_process_count(), get_process_to_shards_map()) and assert on that stable surface.
- [tests/sentry/spans/consumers/process/test_consumer.py:L101-L105] Ensure shard-distribution assertions run after shard assignment is complete (move after step.join() or poll until total_shards matches expected) to avoid race conditions if shard mapping is populated asynchronously.
📝 This review includes 4 inline comments (4 warnings)
Posted by re-entry.ai · Risk governance for autonomous engineering teams
| @@ -1,3 +1,4 @@ | |||
| import time | |||
There was a problem hiding this comment.
Quote: import time
Issue: The new test relies on real wall-clock timing (time.sleep(0.1) below). In CI or under load, this can still be flaky (threads/processes may not have progressed yet), and in other environments it can slow the suite.
Fix: Prefer a deterministic synchronization mechanism (e.g., an Event/Condition exposed by the flusher, or poll until a condition is met with a timeout) instead of a fixed sleep. Example: replace time.sleep(0.1) with a loop that waits until len(flusher.processes)/process_to_shards_map is populated or until a timeout expires.
| step.poll() | ||
| fac._flusher.current_drift.value = 9000 # "advance" our "clock" | ||
|
|
||
| step.poll() |
There was a problem hiding this comment.
Quote: time.sleep(0.1)
Issue: Fixed sleep is a classic source of nondeterministic tests. The comment says it is to “Give flusher threads time to process after drift change”, but there is no guarantee 100ms is enough (or necessary), leading to intermittent failures.
Fix: Replace with polling on an observable condition with a bounded timeout. For example:
deadline = time.time() + 5
while time.time() < deadline:
if step.is_done():
break
time.sleep(0.01)
assert step.is_done()(Use whatever completion/observable hook exists in the test harness.)
|
|
||
| topic = Topic("test") | ||
| messages: list[KafkaPayload] = [] | ||
|
|
There was a problem hiding this comment.
Quote: flusher = fac._flusher
Issue: The test reaches into a private/internal attribute (_flusher). This makes the test brittle: any internal refactor (renaming, changing structure, lazy initialization) can break the test without changing externally observable behavior.
Fix: Assert via public API/behavior instead of private fields. If no public API exists, add a small test hook or expose read-only properties (e.g., get_process_count(), get_process_to_shards_map()) on the factory/flusher for testing. (see also L93, L94)
| input_block_size=None, | ||
| output_block_size=None, | ||
| flusher_processes=2, # Limit to 2 processes even if more shards | ||
| produce_to_pipe=messages.append, |
There was a problem hiding this comment.
Quote: total_shards = sum(len(shards) for shards in flusher.process_to_shards_map.values())
Issue: The test assumes process_to_shards_map is already fully populated when checked. If shard assignment happens after create_with_partitions returns (or after step.join()), this can be flaky.
Fix: Move this assertion after step.join() or poll until total_shards == 4 (with timeout).
There was a problem hiding this comment.
🛡️ re-entry.ai Code Review
@CodingKylo — Please review the following assessment:
🔴 Risk Score: 70/100 · HIGH
| Dimension | Level |
|---|---|
| Likelihood | 🔴 Critical |
| Impact | 🔴 Critical |
| Detectability | 🟡 Medium |
Intent
Make the span flusher run as a configurable multiprocessing manager with per-process health/backpressure tracking, and add tests for the new process limiting behavior.
Summary
The PR refactors the span flusher from a single background worker into a multiprocessing model, adding a new CLI flag (--flusher-processes) that caps how many flusher processes are created. It also changes health/backpressure tracking from single shared mp.Value fields to per-process state, which can silently break any code/tests still expecting the old attributes. The reviewer must verify that process lifecycle management correctly terminates unhealthy workers, that shard-to-process mapping and drift/flush scheduling remain correct under concurrency, and that the new CLI option is validated end-to-end (including edge cases like 0/None). Finally, the added tests appear to focus on internal bookkeeping and process counts rather than end-to-end flushing correctness.
🎯 Review Focus
Confirm the flusher’s multiprocessing lifecycle and shard-to-process mapping are correct under failure/health transitions—specifically that unhealthy processes are reliably terminated and that drift/flush scheduling still results in commits being produced for all shards when flusher_processes is less than the shard count.
✅ Action Checklist
- SUGGESTION — Validate and clamp
--flusher-processesbefore it reachesSpanFlusherto avoid pathological values (e.g., 0 or negative) causingnum_processes = min(self.max_processes, len(...))to become 0 and leaving no flusher processes. Concretely: insrc/sentry/consumers/__init__.pysettype=click.IntRange(1, ...)(or enforceflusher_processes = max(1, flusher_processes)insrc/sentry/spans/consumers/process/factory.pybefore passingmax_processes). - SUGGESTION — Add an integration-style test that exercises end-to-end flushing/drift behavior with
flusher_processes < num_shards(not justlen(flusher.processes)), and remove the wall-clocktime.sleep(0.1)dependency. Concretely: intests/sentry/spans/consumers/process/test_consumer.py, replace the sleep with polling on an observable state (e.g., wait untilmessageslength increases or until a health/backpressure metric/state changes) to make the test deterministic across CI load. - SUGGESTION — Search for and update any remaining references to the old single-value fields (
backpressure_since,healthy_since,process_restarts) to the new per-process structures, or provide backward-compatible properties. Concretely: add a small compatibility shim inSpanFlusher(e.g., properties that aggregate per-process values) or update all call sites/tests to useprocess_backpressure_since/process_healthy_since/process_restartsmaps.
Suggestions
- Validate and clamp
--flusher-processesbefore it reachesSpanFlusherto avoid pathological values (e.g., 0 or negative) causingnum_processes = min(self.max_processes, len(...))to become 0 and leaving no flusher processes. Concretely: insrc/sentry/consumers/__init__.pysettype=click.IntRange(1, ...)(or enforceflusher_processes = max(1, flusher_processes)insrc/sentry/spans/consumers/process/factory.pybefore passingmax_processes). - Add an integration-style test that exercises end-to-end flushing/drift behavior with
flusher_processes < num_shards(not justlen(flusher.processes)), and remove the wall-clocktime.sleep(0.1)dependency. Concretely: intests/sentry/spans/consumers/process/test_consumer.py, replace the sleep with polling on an observable state (e.g., wait untilmessageslength increases or until a health/backpressure metric/state changes) to make the test deterministic across CI load. - Search for and update any remaining references to the old single-value fields (
backpressure_since,healthy_since,process_restarts) to the new per-process structures, or provide backward-compatible properties. Concretely: add a small compatibility shim inSpanFlusher(e.g., properties that aggregate per-process values) or update all call sites/tests to useprocess_backpressure_since/process_healthy_since/process_restartsmaps.
📝 This review includes 2 inline comments (1 warning, 1 note)
Posted by re-entry.ai · Risk governance for autonomous engineering teams
|
|
||
| self.buffer.record_stored_segments() | ||
| for buffer in self.buffers.values(): | ||
| buffer.record_stored_segments() |
There was a problem hiding this comment.
Issue: memory_infos: list[ServiceMemory] = [] then memory_infos.extend(buffer.get_memory_info()) for each buffer. If get_memory_info() yields many entries, this builds an unbounded list each submit() call, increasing GC pressure and latency.
Fix: Avoid materializing all entries; compute sums incrementally:
used = available = 0
for buffer in self.buffers.values():
for x in buffer.get_memory_info():
used += x.used
available += x.available| self.healthy_since.value = int(time.time()) | ||
| self.process_healthy_since[process_index].value = int(time.time()) | ||
|
|
||
| # Create a buffer for these specific shards |
There was a problem hiding this comment.
Issue: self.processes: dict[int, multiprocessing.context.SpawnProcess | threading.Thread] = {} is populated with process objects, but the type union includes threading.Thread while later code uses multiprocessing.Process checks. In _ensure_processes_alive, the code calls process.kill() only when isinstance(process, multiprocessing.Process), but multiprocessing.context.SpawnProcess is not guaranteed to be an instance of multiprocessing.Process (it is a subclass in CPython, but this is not guaranteed by the code here). If the check fails, unhealthy processes may never be terminated.
Fix: Check against the actual class used to create processes (self.mp_context.Process) or use hasattr(process, "kill"):
if hasattr(process, "kill"):
process.kill()2 participants
Martian Code Review Benchmark PR (mirrored from source #6)