https://prefect.io
Join Slack
what is the simplest way to prevent concurrent runs at the flow-name level? E.g. if bad timing crea...
# ask-community
d
what is the simplest way to prevent concurrent runs at the flow-name level? E.g. if bad timing creates overlapped deployment runs of the same flow that should only run independently, or if a flow is running locally and a deployment of that flow is trying to kick off while the local run of the flow is running
👍 1
✅ 1
r
set concurrency to 1 on the work queue
d
would that not also prevent unrelated flows from running at the same time?
r
you can have a dedicated queue for a flow
d
so if i don’t want ANY unique flow to be running more than once at a time, every flow would have to live in its own work queue? That’s… silly
r
e
@David Michael Carter we recently introduced some new concurrency constructs that would allow you to have greater control.
🙌 1
Copy code
from prefect import flow
from prefect.context import get_run_context
from prefect.concurrency.sync import concurrency

@flow
def my_flow():
    with concurrency(get_run_context().flow.name, occupy=1):
        ...
r
a neat new feature 🙌
💯 1
e
Yessir! New in 2.13.0
f
@Emil Christensen Sorry to resurect this thread, but I believe I might have found a bug with this feature. When using the context-manager a concurrency limit is created with name of the function, as opposed to the deployed flow-name. In my use case; I reuse the same function and name the flows pr database-table. The context flow-name does not match the function-name and the concurrency-limitation does nothing.
e
@Fredrik Hoem Grelland Do you mean the deployment name i.e. (
my-flow/my-deployment
)? I’m not 100% sure I follow. If you could give an example, that would help clarify. There’s a lot of other metadata you can find in the return of 
get_run_context()
.
f
Yes, sorry, I will try again. I programmatically create hundreds of flows using the same flow function, but setting the name and parameter attribute differently for each one. This way I get one flow pr database table in my ETL-process. The documentation states that, when using a context manager, a "concurrency limit is implicitly created" https://docs.prefect.io/2.13.5/guides/global-concurrency-limits/#using-the-concurrency-context-manager. However, when running the flows, only one limit is created - with the function name, rather than the flow name that is passed as a parameter to the flow.
@Emil Christensen Ok, So when trying to create a reproduceable case to post here I found some additional limitations, that renders the functionality unusable for my case. Findings: • The global concurrency limit is implicitly created, with the name of the function. NOT the parameter -name- to the flow function. • The limit is created in an inactive state, as stated in the documentation, but there is no way to activate it in the python api as far as I can see. • There is no obvious way to create global limits programatically. So the way I see this, the only way to use global concurrency limits is to create and activate them in the UI, or allow the code to implicitly create limits and then manually toggle them active after first run. Am I mistaken?
@Emil Christensen Would you suggest that I create github issues for these limitations/deficiencies? We are currently in the process of figuring out how to utilize prefect effectively as an end-to-end orchestrator, and being able to control process concurrency is a key architectural tenet.
26 Views
Open in Slack
PreviousNext
Powered by

Prefect Community

Bring your towel and join one of the fastest growing data communities. Welcome to our second-generation open source orchestration platform, a completely rethought approach to dataflow automation.

Powered by