Skip to content
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

Jump to bottom

add support for run-granularity op concurrency #26458

Open
wants to merge 1 commit into
base: prha/concurrency_config
Choose a base branch
from
Open

add support for run-granularity op concurrency #26458

wants to merge 1 commit into from

Conversation

prha
Copy link
Member

@prha prha commented Dec 12, 2024 •
edited
Loading

Summary & Motivation

Enables setting run granularity enforcement of pools, to limit the concurrency of in-progress runs containing a particular op / asset.

How I Tested These Changes

BK

Changelog

  • Enables run granularity concurrency enforcement of pool limits

@prha prha marked this pull request as ready for review December 12, 2024 21:08
@prha Graphite App
Copy link
Member Author

prha commented Dec 12, 2024 •
edited
Loading

Warning

This pull request is not mergeable via GitHub because a downstack PR is open. Once all requirements are satisfied, merge this PR as a stack on Graphite.
Learn more

This stack of pull requests is managed by Graphite. Learn more about stacking.

@prha prha force-pushed the prha/asset_op_concurrency_key branch from 571ead4 to 238ba49 Compare December 13, 2024 17:46
@prha prha force-pushed the prha/concurrency_granularity branch from 34f9dcd to 1816e2f Compare December 13, 2024 17:46
@prha prha mentioned this pull request Dec 13, 2024
Open
@prha prha force-pushed the prha/asset_op_concurrency_key branch from 238ba49 to a36d157 Compare January 3, 2025 05:18
This was referenced Jan 3, 2025
Open
Open
Draft
@prha prha force-pushed the prha/concurrency_granularity branch from 1816e2f to e7e5973 Compare January 3, 2025 05:19
This was referenced Jan 3, 2025
Open
Open
@prha prha force-pushed the prha/asset_op_concurrency_key branch from 67085a7 to 21ae854 Compare January 6, 2025 20:51
@prha prha force-pushed the prha/concurrency_granularity branch 2 times, most recently from add1323 to 57f2db2 Compare January 6, 2025 22:06
@prha prha force-pushed the prha/asset_op_concurrency_key branch from a32b759 to be83c16 Compare January 7, 2025 17:16
@prha prha force-pushed the prha/concurrency_granularity branch from 57f2db2 to 896bb19 Compare January 7, 2025 17:16
@prha prha force-pushed the prha/asset_op_concurrency_key branch 3 times, most recently from f52347d to ce97f66 Compare January 8, 2025 19:51
@prha prha force-pushed the prha/concurrency_granularity branch from 896bb19 to 960f7c2 Compare January 9, 2025 00:26
@prha prha changed the base branch from prha/asset_op_concurrency_key to prha/concurrency_config January 9, 2025 00:26
This was referenced Jan 9, 2025
Open
Open
@github-actions GitHub Actions
Copy link

github-actions bot commented Jan 9, 2025

Deploy preview for dagit-core-storybook ready!

✅ Preview
https://dagit-core-storybook-qdm6nnk70-elementl.vercel.app
https://prha-concurrency-granularity.core-storybook.dagster-docs.io

Built with commit 960f7c2.
This pull request is being automatically deployed with vercel-action

@prha prha force-pushed the prha/concurrency_config branch from f572156 to 8cfee9c Compare January 9, 2025 03:43
@prha prha force-pushed the prha/concurrency_granularity branch from 960f7c2 to a85aa24 Compare January 9, 2025 03:43
@prha prha force-pushed the prha/concurrency_config branch from 8cfee9c to 152beca Compare January 9, 2025 03:58
@prha prha force-pushed the prha/concurrency_granularity branch from a85aa24 to d217d12 Compare January 9, 2025 03:58
@prha prha mentioned this pull request Jan 9, 2025
Open
@prha prha mentioned this pull request Jan 9, 2025
Open
@prha prha force-pushed the prha/concurrency_config branch from 152beca to 068a084 Compare January 10, 2025 01:17
@prha prha force-pushed the prha/concurrency_granularity branch from d217d12 to 72da007 Compare January 10, 2025 01:18
@prha prha force-pushed the prha/concurrency_config branch from 068a084 to 6927f22 Compare January 10, 2025 23:23
@prha prha force-pushed the prha/concurrency_granularity branch from 72da007 to 8782d96 Compare January 10, 2025 23:23
@prha prha force-pushed the prha/concurrency_config branch from 6927f22 to cb8dd1e Compare January 16, 2025 00:03
@prha prha force-pushed the prha/concurrency_granularity branch from 8782d96 to 5c83e2c Compare January 16, 2025 00:03
@prha prha force-pushed the prha/concurrency_config branch from cb8dd1e to bf19552 Compare January 16, 2025 00:19
@prha prha force-pushed the prha/concurrency_granularity branch from 5c83e2c to 9e5e36d Compare January 16, 2025 00:19
@prha prha force-pushed the prha/concurrency_config branch from bf19552 to 17ed4d9 Compare January 16, 2025 01:10
@prha prha force-pushed the prha/concurrency_granularity branch from 9e5e36d to eb04b1b Compare January 16, 2025 01:11
@prha prha mentioned this pull request Jan 16, 2025
Open
@prha prha force-pushed the prha/concurrency_config branch from 17ed4d9 to 6a116d5 Compare January 16, 2025 20:23
@prha prha force-pushed the prha/concurrency_granularity branch from eb04b1b to d80e191 Compare January 16, 2025 20:23
@prha prha force-pushed the prha/concurrency_config branch from 6a116d5 to 7309b15 Compare January 17, 2025 01:08
@prha prha force-pushed the prha/concurrency_granularity branch from d80e191 to e5ad4c3 Compare January 17, 2025 01:08
@prha prha force-pushed the prha/concurrency_config branch from 7309b15 to 256d726 Compare January 17, 2025 02:27
@prha prha force-pushed the prha/concurrency_granularity branch from e5ad4c3 to 111c97c Compare January 17, 2025 02:27
@prha prha force-pushed the prha/concurrency_config branch from 256d726 to 3efd17a Compare January 17, 2025 02:55
@prha prha force-pushed the prha/concurrency_granularity branch from 111c97c to 4e2eb36 Compare January 17, 2025 02:55
@prha prha requested a review from OwenKephart January 17, 2025 21:21
@prha prha force-pushed the prha/concurrency_config branch from 3efd17a to 3d2b288 Compare January 17, 2025 22:30
@prha prha force-pushed the prha/concurrency_granularity branch from 4e2eb36 to 647d5df Compare January 17, 2025 22:30
OwenKephart
OwenKephart approved these changes Jan 17, 2025
View reviewed changes
Copy link
Contributor

@OwenKephart OwenKephart left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

note comments, but looks good

python_modules/dagster/dagster/_core/run_coordinator/queued_run_coordinator.py
@@ -307,6 +344,7 @@ def submit_run(self, context: SubmitRunContext) -> DagsterRun:
run = self._instance.get_run_by_id(dagster_run.run_id)
if run is None:
check.failed(f"Failed to reload run {dagster_run.run_id}")
assert run
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is this necessary?

python_modules/dagster/dagster/_daemon/run_coordinator/queued_run_coordinator_daemon.py
Comment on lines +81 to +87
if not isinstance(workspace_process_context.instance.run_coordinator, QueuedRunCoordinator):
check.failed(
f"Expected QueuedRunCoordinator, got {workspace_process_context.instance.run_coordinator}"
)
run_coordinator = cast(
QueuedRunCoordinator, workspace_process_context.instance.run_coordinator
)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: is this fundamentally different from

run_coordinator = check.inst(workspace_process_context.instance.run_coordinator, QueuedRunCoordinator)?

python_modules/dagster/dagster/_core/op_concurrency_limits_counter.py
return {**run.run_op_concurrency.root_key_counts}

else:
assert self._pool_granularity == PoolGranularity.RUN
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nit: prefer check.invariant() to assert, or potentially just:

if granularity == OP: ...
elif granularity == RUN: ...
else: raise Exception()

python_modules/dagster/dagster/_core/op_concurrency_limits_counter.py

def get_blocked_run_debug_info(self, run: DagsterRun) -> Mapping:
if not run.run_op_concurrency:
return {}

log_info = {}
for pool in run.run_op_concurrency.root_key_counts.keys():
concurrency_info = self._concurrency_info_by_pool.get(pool)
concurrency_info = self._concurrency_info_by_key.get(pool)
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

any particular reason for this name change? nbd either way just curious

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@OwenKephart OwenKephart OwenKephart approved these changes

Assignees
No one assigned
Labels
None yet
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
2 participants
@prha @OwenKephart