Skip to content

feat(job-orchestration): Allow concurrent compression job processing by processing batches of compression tasks per job. #1637

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.

Sign up for GitHub

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

Merged
merged 36 commits into from
Nov 28, 2025
+402 −158
Merged

feat(job-orchestration): Allow concurrent compression job processing by processing batches of compression tasks per job. #1637

Show file tree
Hide file tree
Changes from 23 commits
Commits
Show all changes
36 commits
Select commit Hold shift + click to select a range
c7cf648
feat(job-orchestration): Improve scheduling fairness by batching comp…
junhaoliao Nov 20, 2025
9b9e360
fix lint
junhaoliao Nov 20, 2025
31ab2a4
fix all G004 in VCS changed lines
junhaoliao Nov 20, 2025
b8c2936
fix all TRY401 in VCS changed lines
junhaoliao Nov 20, 2025
0cb7fae
fix all F841 in VCS changed lines
junhaoliao Nov 20, 2025
76423df
refactor into helpers and add docs
junhaoliao Nov 20, 2025
cd6930e
refactor search_and_schedule_new_tasks and poll_running_jobs for clar…
junhaoliao Nov 20, 2025
6fb54f2
refactor compression job handling to use DbContext for database opera…
junhaoliao Nov 20, 2025
50361f9
refactor compression job handling to improve task batching and metada…
junhaoliao Nov 20, 2025
2413d53
fix all DTZ005
junhaoliao Nov 20, 2025
a80c829
lint
junhaoliao Nov 20, 2025
a46b0ea
line after desc
junhaoliao Nov 20, 2025
ac310a6
add docstring to DbContext
junhaoliao Nov 20, 2025
672e81b
fix docstring for job scheduling function
junhaoliao Nov 20, 2025
4fe7076
refactor parameter order in batch and dispatch functions
junhaoliao Nov 20, 2025
e8847ee
enhance compression scheduler with additional comments and remove unu…
junhaoliao Nov 20, 2025
e1b14f5
remove redundant comment in task dispatch logic
junhaoliao Nov 20, 2025
4f764df
reorder
junhaoliao Nov 20, 2025
bcc5e75
add comment to clarify task submission process in compression scheduler
junhaoliao Nov 20, 2025
261b206
restore main entry point in compression scheduler
junhaoliao Nov 20, 2025
f8aa1d1
rename `num_tasks_per_sub_job` to `max_concurrent_tasks_per_job`
junhaoliao Nov 21, 2025
7e6ac2c
Allow unlimited concurrent tasks per job
junhaoliao Nov 21, 2025
2acde6d
Merge branch 'main' into batch-compression-tasks
junhaoliao Nov 21, 2025
8b0b5d2
Merge branch 'main' into batch-compression-tasks
junhaoliao Nov 25, 2025
0a2735e
update task running status for current batch only; Extract task statu…
junhaoliao Nov 25, 2025
9da4ea1
revert changes in query_scheduler.py
junhaoliao Nov 26, 2025
5196cf0
lint
junhaoliao Nov 26, 2025
d639446
Specify type hints for `job_id` and `kv` in compression scheduler fun…
junhaoliao Nov 27, 2025
1df8f05
Merge branch 'main' into batch-compression-tasks
junhaoliao Nov 27, 2025
dc28f53
Refactor SQL queries in `compression_scheduler.py` to use parameteriz…
junhaoliao Nov 27, 2025
f8acfe2
add comment about setting 0 to disable the task scheduling limit per …
junhaoliao Nov 27, 2025
78df192
Fix spacing in `clp_config.py` comment for `max_concurrent_tasks_per_…
junhaoliao Nov 27, 2025
8b456ae
Merge branch 'main' into batch-compression-tasks
junhaoliao Nov 27, 2025
1065b1e
docs - Apply suggestions from code review
junhaoliao Nov 28, 2025
24edafc
refactor(clp_config): Use a constant for unlimited concurrent tasks p…
junhaoliao Nov 28, 2025
9282174
Merge branch 'main' into batch-compression-tasks
junhaoliao Nov 28, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions components/clp-py-utils/clp_py_utils/clp_config.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@
StrEnumSerializer = PlainSerializer(serialize_str_enum)
# Generic types
NonEmptyStr = Annotated[str, Field(min_length=1)]
NonNegativeInt = Annotated[int, Field(ge=0)]
PositiveFloat = Annotated[float, Field(gt=0)]
PositiveInt = Annotated[int, Field(gt=0)]
# Specific types
Expand Down Expand Up @@ -253,6 +254,7 @@ def transform_for_container(self):

class CompressionScheduler(BaseModel):
jobs_poll_delay: PositiveFloat = 0.1 # seconds
max_concurrent_tasks_per_job: NonNegativeInt = 0
logging_level: LoggingLevel = "INFO"


Expand Down
Loading
Loading