feat(compression-coordinator): Introduce max_concurrent_tasks config to limit job-handler concurrency. - #2435
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
WalkthroughThe change adds ChangesCompression coordinator concurrency
Estimated code review effort: 3 (Moderate) | ~20 minutes Sequence Diagram(s)sequenceDiagram
participant CompressionCoordinator
participant JobDatabase
participant JobHandler
CompressionCoordinator->>CompressionCoordinator: Check semaphore capacity
CompressionCoordinator->>JobDatabase: Fetch pending jobs within capacity
CompressionCoordinator->>JobHandler: Spawn handlers with retained permits
JobHandler->>CompressionCoordinator: Release permit on completion
Possibly related issues
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
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 |
Bill-hbrhbr
changed the title
max_concurrent_tasks config to limit job-handler task concurrency.
Bill-hbrhbr
changed the title
max_concurrent_tasks config to limit job-handler task concurrency.max_concurrent_tasks config to limit job-handler concurrency.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
🤖 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 `@components/clp-py-utils/clp_py_utils/clp_config.py`:
- Line 798: Add a shared Tokio semaphore upper-bound validation for
max_concurrent_tasks in components/clp-py-utils/clp_py_utils/clp_config.py:798
and components/clp-rust-utils/src/clp_config/package/config.rs:488, preserving
the existing positive-value validation. In
components/compression-coordinator/src/coordination.rs:137-138, validate the
configured value before constructing job_handler_sem and return the established
configuration error when it exceeds the supported limit.
In `@components/compression-coordinator/src/coordination.rs`:
- Around line 49-52: Update the recovery admission logic around the
coordinator’s recovered-handler tracking so every recovered handler contributes
to the active count, including handlers that do not hold a permit. Prevent new
jobs from being admitted until the total active recovered and newly started
handlers is below max_concurrent_tasks, and revise the recovery documentation
near the constructor to describe this limit-enforced behavior.
- Around line 270-275: Update the pending-job refill flow around
fetch_new_job_rows so it retrieves eligible jobs in bounded pages rather than
loading and retaining the entire backlog in pending_job_queue. Add durable
paging state, such as a cursor that advances only after successful processing or
equivalent recovery-safe state, and preserve first-fetch recovery semantics so
retries resume without skipping jobs or causing unbounded memory growth.
In `@components/package-template/src/etc/clp-config.template.json.yaml`:
- Line 72: Update the commented max_concurrent_tasks configuration entry in the
template to explicitly state that its value must be greater than zero,
distinguishing it from compression_scheduler.max_concurrent_tasks_per_job where
zero disables the limit.
- Line 67: Update the compression coordinator configuration note near the
“Compression coordinator config” comment to document every runtime prerequisite:
`spider`, `logs_input.type: s3`, and `archive_output.storage.type: s3`; clarify
that the related template examples must use these required S3 settings rather
than `fs` when enabling the configuration.
🪄 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: ASSERTIVE
Plan: Pro Plus
Run ID: 8fd3f43d-d433-437d-ab55-235ab39cfb5f
📒 Files selected for processing (4)
components/clp-py-utils/clp_py_utils/clp_config.pycomponents/clp-rust-utils/src/clp_config/package/config.rscomponents/compression-coordinator/src/coordination.rscomponents/package-template/src/etc/clp-config.template.json.yaml
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
components/compression-coordinator/src/coordination.rs (1)
294-300: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winRecord a job as dispatched only after handler creation succeeds.
dispatched_job_ids.push(job_id)runs before configuration deserialisation andcreate_job_handle. If either operation fails,runstill passes the ID tomark_jobs_dispatched. AnError::UnsupportedInputConfigjob therefore receivesdispatch_timewithout a coordinator handler, which removes it from the subsequent pending-job query and can interfere with the documented legacy-scheduler hand-off.Move the push after
create_job_handlereturnsOk:Proposed fix
let job_id = job_row.id; - dispatched_job_ids.push(job_id); let clp_io_config: ClpIoConfig = match BrotliMsgpack::deserialize(&job_row.serialized_clp_io_config) { Ok(clp_io_config) => clp_io_config, Err(e) => { ... } }; let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { continue; }; +dispatched_job_ids.push(job_id);🤖 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 `@components/compression-coordinator/src/coordination.rs` around lines 294 - 300, Move dispatched_job_ids.push(job_id) in the run flow to after configuration deserialization and successful create_job_handle completion, so only jobs with a coordinator handler are marked dispatched; preserve failure handling without passing failed job IDs to mark_jobs_dispatched.
🤖 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 `@components/package-template/src/etc/clp-config.template.json.yaml`:
- Line 67: Update the compression coordinator configuration note near the
existing logs_input.type and spider requirements to also state that
archive_output.storage.type must be "s3". Keep the documentation aligned with
the coordinator prerequisites and the archive output setting shown in the
template.
---
Outside diff comments:
In `@components/compression-coordinator/src/coordination.rs`:
- Around line 294-300: Move dispatched_job_ids.push(job_id) in the run flow to
after configuration deserialization and successful create_job_handle completion,
so only jobs with a coordinator handler are marked dispatched; preserve failure
handling without passing failed job IDs to mark_jobs_dispatched.
🪄 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: ASSERTIVE
Plan: Pro Plus
Run ID: b74354c7-b2d8-4659-a5ac-82902ed1fabb
📒 Files selected for processing (3)
components/compression-coordinator/src/coordination.rscomponents/compression-coordinator/src/error.rscomponents/package-template/src/etc/clp-config.template.json.yaml
| job_polling_interval: Duration, | ||
| cancellation_token: CancellationToken, | ||
| job_handler_sem: Arc<Semaphore>, | ||
| pending_job_queue: VecDeque<PendingJobRowProjection>, |
There was a problem hiding this comment.
I'm not sure if we need this. Can you explain why we can't do the following instead:
- On the main loop, fetch compression jobs with a limit and ID ordering (sth like
SELECT * FROM t ORDER BY id ASC LIMIT 100;) - Get a permit before spawning the coroutine. The permit automatially drops itself if the coroutine exits, aborts, or crashed.
- Let the semaphore to block coroutine creation implicitly. The main loop may be blocked, which is fine because it shouldn't push more jobs into Spider.
This should lead to the behavior we expect iiuc.
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
There's a risk that we won't be able to merge this PR before the release since it's not in an expected shape yet. Let's fix the recovery path bug first.
| async fn try_deserialize_clp_io_config( | ||
| &self, | ||
| job_id: CompressionJobId, | ||
| serialized_config: &[u8], | ||
| ) -> Option<ClpIoConfig> { | ||
| match BrotliMsgpack::deserialize(serialized_config) { | ||
| Ok(clp_io_config) => Some(clp_io_config), | ||
| Err(e) => { | ||
| tracing::error!( | ||
| error = % e, | ||
| job_id = % job_id, | ||
| "Failed to deserialize CLP I/O config. Skipping." | ||
| ); | ||
| self.mark_job_failed( | ||
| job_id, | ||
| &format!("Failed to deserialize CLP I/O config: {e}"), | ||
| ) | ||
| .await; | ||
| None | ||
| } | ||
| } | ||
| } |
There was a problem hiding this comment.
It's a really bad idea to return an option instead of a result, and log inside the helper. The best practice for designing this type of helper is:
- Return a result. Let the caller decide whether it wants to inspect the error or just propagate.
- Log on the caller side.
- You may see
create_job_handlelogs inside the helper. That is because it needs to print different logs based on the type of the error and execute different reaction calls accordingly. This makes it an exception since logging it inside would actually make things cleaner. In this helper, errors are printed unconditionally so it should be logged outside.
- You may see
| let Ok(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { | ||
| let Some(job_handle) = self.create_job_handle(job_id, clp_io_config).await else { |
There was a problem hiding this comment.
Similar to https://github.com/y-scope/clp/pull/2435/changes#r3736535562: you didn't change the implementation of create_job_handle, then why would you need to update the return type? This is polluting the diff in some sense and making the PR harder to review.
| /// A projection of the columns read from a compression job row. | ||
| #[derive(Debug, sqlx::FromRow)] | ||
| struct RunningJobRowProjection { | ||
| struct JobRowProjection { |
There was a problem hiding this comment.
It's a bad idea to merge PendingJobRowProjection and RunningJobRowProjection into one type: the user needs to check whether spider_job_id is NULL now. Since we will have https://github.com/y-scope/clp/pull/2435/changes#r3736798212, let roll back to two separate types instead of one.
In Rust, if you really want a type to cover both, a better idea is probably to make an enum containing both types.
| continue; | ||
| }; | ||
|
|
||
| let permit = self.job_handler_sem.clone().try_acquire_owned().ok(); |
There was a problem hiding this comment.
I'm not sure if we really need to enforce the permit here: the recovery may not necessarily happen from a failure recovery, but a restart with clp config changed. If the new max_concurrent_jobs is changed to a smaller number, this recovery path will be blocked for an improper reason: eventually, a number of compression jobs are already submitted to Spider that exceeds the configured concurrency.
My proposal:
- For now, it's probably easier if we don't do anything to the permit: let the already-submitted jobs run in background. As we would assume the recovery is a rare case, the number of jobs should be roughly bounded by
old_max_concurrent_jobs+new_max_concurrent_jobs. There are extreme cases where this bound can be broken by a series of frequent restarts, but it should be safe to ignore this case for now. - To improve the naive implementation, we could compare the number of jobs to recover vs. the currently configured
max_concurrent_jobs:- If
num_jobs_to_recover>max_concurrent_jobs: print a warning, make a join of all the recovered job handlers, so the main loop only starts when all previously submitted jobs are finished. - Otherwise, do the current implementation to recover jobs while each job holds an acquired permit.
- If
There was a problem hiding this comment.
@coderabbitai we decide to stick to the first option for this PR. Can you create an issue to keep track of it, documenting that on restart, recovered jobs may disobey the configured rate limit? You should also document my section option as the proper way to implement it.
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
repo="y-scope/clp"
body=$(cat <<'EOF'
## Summary
The compression coordinator does not apply `max_concurrent_tasks` to jobs recovered during startup.
After a restart, the coordinator resumes already submitted Spider jobs without acquiring `job_handler_sem` permits. If the configuration changed to a smaller `max_concurrent_tasks` value, the recovered jobs can exceed the configured concurrency limit.
This behaviour is intentional for now. It avoids blocking recovery when Spider already has more submitted compression jobs than the new limit permits.
## Required follow-up
Implement bounded startup recovery in `components/compression-coordinator/src/coordination.rs`.
1. Count the jobs that require recovery before spawning recovery handlers.
2. Compare that count with the configured `max_concurrent_tasks`.
3. If the recovery count is less than or equal to the configured limit, recover each job while it holds a semaphore permit.
4. If the recovery count exceeds the configured limit:
- Emit a warning that recovered jobs exceed the configured concurrency limit.
- Start all recovery handlers.
- Join all recovered job handlers before the normal coordinator poll loop starts.
5. Ensure normal scheduling starts only after the required recovery behaviour completes.
## Rationale
The current behaviour is acceptable because coordinator restart recovery should be rare. It can temporarily run approximately the prior limit plus the new limit. Repeated rapid restarts can exceed that approximate bound. The follow-up implementation must make restart behaviour explicit and bounded.
## Affected area
- `components/compression-coordinator/src/coordination.rs`
- Startup recovery and `job_handler_sem` permit management.
## Acceptance criteria
- Recovery behaviour is selected by comparing the number of recovery jobs with `max_concurrent_tasks`.
- Recovery jobs hold permits when the count does not exceed the limit.
- When the count exceeds the limit, the coordinator logs a warning and waits for all recovered handlers before normal scheduling.
- Tests cover both recovery-count cases.
## Context
Requested by `@LinZhihao-723`.
- Pull request: https://github.com/y-scope/clp/pull/2435
- Review discussion: https://github.com/y-scope/clp/pull/2435#discussion_r3736993293
EOF
)
gh issue create \
--repo "$repo" \
--title "Bound compression coordinator recovery after configuration changes" \
--body "$body" \
--assignee "LinZhihao-723"Length of output: 4193
@LinZhihao-723, I created and assigned #2472: Bound compression coordinator recovery after configuration changes.
The issue documents the current restart behaviour and the proposed recovery-count design.
🧬 Code Graph Analysis Results
components/compression-coordinator/src/job_handle.rs
-
Lines 30–35:
SpiderOptiondefines compression retry settings, commit execution policy, and result polling backoff parameters passed to compression job handles. -
Lines 42–61:
S3CompressionJobHandlestores the database, job ID, Spider submitter/resource group, compression configuration, and sharedSpiderOption. -
Lines 72–119:
S3CompressionJobHandle::newvalidates and transforms the CLP I/O configuration, returningError::UnsupportedInputConfigorError::InvalidDatasetfor unsupported or invalid inputs; otherwise constructs a handle with the shared Spider options. -
Lines 133–142:
S3CompressionJobHandle::runsubmits the job and waits for completion; failures are reported throughreport_failure.
components/clp-rust-utils/src/clp_config/package/config.rs
- Lines 483–494:
CompressionCoordinatorconfiguration defines the job polling interval, maximum concurrent jobs, retry settings, polling backoff, and commit-task timeouts.
components/compression-coordinator/src/coordination.rs
-
Lines 72–147:
Coordinator::newretrieves running jobs from the database and spawns detached recovery handles for each previously submitted Spider job. -
Lines 326–335:
fetch_new_job_rowsreturns all pending jobs on the first fetch, including jobs previously dispatched but not started; subsequent fetches return only pending jobs with no dispatch time.
You are interacting with an AI system.
… coordinator/limit-job-submission-concurrency
Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com>
… coordinator/limit-job-submission-concurrency
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
Overall it's in a good shape now. Some more comments.
| let permit = self | ||
| .job_handler_sem | ||
| .clone() | ||
| .acquire_owned() | ||
| .await | ||
| .map_err(|e| { | ||
| Error::Semaphore(format!("Failed to acquire a job handler permit: {e}")) | ||
| })?; |
There was a problem hiding this comment.
Since this is mainly for spider job submission rate control, we could do it after job IO dispatching. Let's move this after job handle creation.
| async fn fetch_new_job_rows( | ||
| &mut self, | ||
| limit: usize, | ||
| ) -> Result<Vec<PendingJobRowProjection>, Error> { |
There was a problem hiding this comment.
This API design is a bit error-prone: limit is actually not used in the first fetch, which is not documented in the docstring.
Took a closer look, it makes more sense to move self.job_handler_sem.available_permits inside this method and drop limit parameter. In this way, this method is self-contained where it's behavior is easier to document (we might need to rename it to fetch_new_job_rows_with_concurrency_limit).
| let query = if self.is_first_fetch { | ||
| self.is_first_fetch = false; | ||
| let is_first_fetch = self.is_first_fetch; | ||
| self.is_first_fetch = false; | ||
|
|
||
| let query = if is_first_fetch { | ||
| FIRST_FETCH_QUERY | ||
| } else { | ||
| SUBSEQUENT_FETCH_QUERY | ||
| }; | ||
| let rows = sqlx::query_as::<_, PendingJobRowProjection>(query) | ||
| .bind(CompressionJobStatus::Pending) | ||
| .fetch_all(&self.db_pool) | ||
| .await?; | ||
|
|
||
| let mut query_builder = | ||
| sqlx::query_as::<_, PendingJobRowProjection>(query).bind(CompressionJobStatus::Pending); | ||
| if !is_first_fetch { | ||
| query_builder = query_builder.bind( | ||
| i64::try_from(limit) | ||
| .expect("limit is bounded by Semaphore::MAX_PERMITS, which fits in i64"), | ||
| ); | ||
| } |
There was a problem hiding this comment.
let query = if self.is_first_fetch {
self.is_first_fetch = false;
sqlx::query_as::<_, PendingJobRowProjection>(FIRST_FETCH_QUERY)
.bind(CompressionJobStatus::Pending)
} else {
sqlx::query_as::<_, PendingJobRowProjection>(SUBSEQUENT_FETCH_QUERY)
.bind(CompressionJobStatus::Pending)
.bind(
i64::try_from(limit)
.expect("limit is bounded by Semaphore::MAX_PERMITS, which fits in i64"),
)
};Rewrite in a cleaner way.
- Avoid making a variable mutable in Rust whenever possible.
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
Please add a comment in the recover to mention that we don't bound recovered jobs, referring to this issue: #2472
Co-authored-by: Lin Zhihao <59785146+LinZhihao-723@users.noreply.github.com>
|
@coderabbitai review |
✅ Action performedReview finished.
|
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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 `@components/compression-coordinator/src/coordination.rs`:
- Around line 440-453: Remove the is_first_fetch branch in the scheduling flow
and use the dispatch_time IS NULL query with an available-permits limit on every
iteration, preventing fetch_all from loading the entire pending backlog. Run
stale-dispatch recovery before normal scheduling, and preserve the existing
permit conversion and query execution through the scheduling method around
FIRST_FETCH_QUERY and SUBSEQUENT_FETCH_QUERY.
🪄 Autofix
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: ASSERTIVE
Plan: Pro Plus
Run ID: e7f9a89f-68e9-40f1-a34f-883e786d3b04
📒 Files selected for processing (4)
components/clp-py-utils/clp_py_utils/clp_config.pycomponents/clp-rust-utils/src/clp_config/package/config.rscomponents/compression-coordinator/src/coordination.rscomponents/compression-coordinator/src/error.rs
LinZhihao-723
left a comment
LinZhihao-723
left a comment
There was a problem hiding this comment.
Approved with some final updates.
For the PR title, how about:
feat(compression-coordinator): Add configurable limit for concurrent compression jobs handled by the coordinator.
| // NOTE: The current implementation does not enforce concurrency limits for recovered jobs | ||
| // since they were already submitted to Spider. See #2472. |
There was a problem hiding this comment.
Downgrade this to a code-level comment: this should be more like a TODO instead of a formal behavior to documenet in the method-level docstring.
| /// * This query runs only once, so limiting it could leave previously dispatched jobs | ||
| /// unfetched. | ||
| /// * The recovery set is bounded by the previous coordinator's concurrency limit. |
There was a problem hiding this comment.
We should use bullet point since they are unordered.
| //! | ||
| //! The coordinator is responsible for the compression jobs in the `compression_jobs` table that | ||
| //! are in one of the following states: | ||
| //! | ||
| //! | `status` | `spider_id` | `dispatch_time` | Description | | ||
| //! |----------|-------------|-----------------|--------------------------------------------------| | ||
| //! | PENDING | NULL | NULL | New jobs awaiting dispatch. | | ||
| //! | PENDING | NULL | NOT NULL | Jobs dispatched but not yet submitted to Spider. | | ||
| //! | RUNNING | NOT NULL | NOT NULL | Jobs submitted to Spider. | | ||
| //! | ||
| //! NOTE: | ||
| //! | ||
| //! * These are the only legal states for a job that hasn't terminated. | ||
| //! * A non-NULL `dispatch_time` indicates that the coordinator has picked up the job and granted it | ||
| //! permission to run under the concurrency limit. |
There was a problem hiding this comment.
Adding a section to show all legal states.
3 participants
Description
This PR limits the number of compression jobs processed concurrently to prevent the coordinator from creating an unbounded number of job-handler tasks and consuming excessive memory when a large backlog accumulates.
The coordinator now only fetches as many new jobs as it has capacity to process. Once the configured concurrency limit is reached, no additional jobs are fetched until existing jobs finish and capacity becomes available.
On startup, the coordinator recovers work from the previous instance before scheduling new jobs. This includes both jobs that were already submitted to Spider and jobs that were dispatched but did not progress far enough to be marked as running. The former resume tracking their existing Spider jobs, while the latter are re-dispatched.
NOTE
The
is_first_fetchflag and its two-query fetch logic have been removed because they do not work well with bounded concurrency. Previously, the first fetch after startup returned all pending jobs, including those with an existingdispatch_time, to recover jobs that may not have completed dispatch before the previous coordinator exited.With bounded concurrency, recovery can no longer rely on the first fetch returning everything, since the coordinator should only fetch as many new jobs as it has capacity to process. Recovery is therefore handled explicitly by
recover_previous_jobs, which identifies all pending jobs with an existingdispatch_timeand re-dispatches them before normal scheduling begins.This leaves the steady-state dispatcher with a single, clear eligibility rule: only pending jobs without a
dispatch_timeare considered new work.The concurrency limit is configured through a new
max_concurrent_tasksfield inClpConfig, with a default of 1000.This PR opens #2472 to be resolved in a future PR.
Checklist
breaking change.
Validation performed
Summary by CodeRabbit
New Features
Bug Fixes