<@ULVA73B9P> I have a flow scheduled every day at ...
# ask-marvin
m
@Marvin I have a flow scheduled every day at a particular hour, is there a way to prevent or stop the execution at the beginning, if another specified flow is running at the same time?
m
thought for 465.9 seconds
Yes—use a shared global concurrency limit so only one of those flows can run at a time. Then decide whether you want the second flow to wait or to exit immediately. Option A: Early-exit (don’t start if the other flow is running) 1) Create a global concurrency limit with capacity 1
Copy code
prefect 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
Copy code
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:
Copy code
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:
Copy code
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?