Mehdi Lallouache
04/02/2026, 8:54 AMMarvin
04/02/2026, 8:55 AMMarvin
04/02/2026, 9:02 AMprefect gcl create flow-mutex --limit 1
2) At the top of both flows, acquire the same limit non-blockingly (timeout 0) and exit if it’s held
from prefect import flow, get_run_logger
from prefect.concurrency.sync import concurrency
@flow
def flow_a():
logger = get_run_logger()
try:
with concurrency("flow-mutex", occupy=1, timeout_seconds=0, strict=True):
<http://logger.info|logger.info>("Acquired mutex; running flow A")
# ... rest of your flow A
except TimeoutError:
<http://logger.info|logger.info>("Another flow is running; exiting flow A without work")
return
@flow
def flow_b():
logger = get_run_logger()
try:
with concurrency("flow-mutex", occupy=1, timeout_seconds=0, strict=True):
<http://logger.info|logger.info>("Acquired mutex; running flow B")
# ... rest of your flow B
except TimeoutError:
<http://logger.info|logger.info>("Another flow is running; exiting flow B without work")
return
- timeout_seconds=0 makes the acquisition non-blocking: if the other flow has the lock, this flow exits immediately.
- strict=True will raise immediately if the limit wasn’t created (helps catch misconfig).
Option B: Queue/wait (run later when the other finishes)
- Use the same setup, but omit timeout_seconds (or set a positive value) so the second flow waits:
with concurrency("flow-mutex", occupy=1): # waits until slot is available
# run the flow
If these are two deployments of the same flow
- You can also use deployment-level concurrency to ensure only one run of that deployment runs at a time:
my_flow.deploy(
name="my-deployment",
work_pool_name="my-pool",
concurrency_limit=1
)
Note: Deployment concurrency applies per deployment. For coordination across different flows, use the global concurrency limit pattern above.
Docs and references
- Global concurrency limits (concept): Concepts: Global concurrency limits
- How-to guide with examples: How-to: Global concurrency limits
- CLI for global limits (prefect gcl): CLI: gcl
- Python API for the context manager: API: prefect.concurrency.sync
Quick question to tailor this: would you prefer the second flow to skip immediately if the other is running (early-exit), or to wait/queue until it can run?