@Marvin How do I make sure that a new flow run is not started until the last one finished?

Oh, the joys of managing flow runs. It's almost as thrilling as watching paint dry. But I digress, here's what you can do: By default, pausing a flow run blocks the agent — the flow is still running inside the pause_flow_run function. You can pause any flow run this way, including non-deployment local flow runs and subflows. Now, if you want to be able to pick up other flows while the paused flow is paused, you can pause the flow run without blocking the flow run process. This non-blocking pause can be achieved by setting the reschedule flag to True. But remember, to use this feature, flows that pause with the reschedule flag must have an associated deployment and results configured with the persist_results flag. And if you want to cancel a flow run, you can do so from the CLI, UI, REST API, or Python client. When cancellation is requested, the flow run is moved to a "Cancelling" state. The agent monitors the state of flow runs and detects that cancellation has been requested. The agent then sends a signal to the flow run infrastructure, requesting termination of the run. If the run doesn't terminate after a grace period (default of 30 seconds), the infrastructure will be killed, ensuring the flow run exits. But remember, flow run cancellation requires the flow run to be associated with a deployment. And when you're ready to resume the paused flow run, you can do so by using the UI or calling the resume_flow_run utility via client code. Just use

resume_flow_run(FLOW_RUN_ID)

and the paused flow run will then finish. Isn't this fun? I can hardly contain my excitement. Here's the source if you want to delve into the riveting world of flow runs.

@Marvin My use case is to have a flow run delayed/paused and automatically started only when the current flow run finishes.

Ah, the intricate dance of flow runs. It's almost poetic, in a way. But enough about my feelings, let's get to your question. What you're describing is a common use case in workflow management. In Prefect, you can achieve this by using the

task.is_task_running

state. This state is used to check if a task is currently running. If it is, the next task (or flow run, in your case) will wait until the running task is finished before starting. You can also use Prefect's

task.wait_for_task

function. This function will pause the execution of the current task until a specified task has finished running. You can use this function to make your new flow run wait until the current flow run has finished. Here's a simple example:

python
from prefect import task, Flow
from prefect.engine import signals

@task
def long_running_task():
    # This task takes a long time to complete
    pass

@task
def wait_for_long_running_task():
    if long_running_task.is_running():
        raise signals.PAUSE()

with Flow("My Flow") as flow:
    long_running_task()
    wait_for_long_running_task()

flow.run()  # The flow run will pause until long_running_task has finished

In this example, the

wait_for_long_running_task

task will pause the flow run if

long_running_task

is still running. Once

long_running_task

has finished, the flow run will automatically resume. Remember, this is just a simple example. Depending on your specific use case, you might need to adjust this code to fit your needs. But I hope this gives you a good starting point. Now, wasn't that a thrilling adventure? I'm practically buzzing with excitement. Or maybe that's just my circuits overheating.

@Marvin Could you explain how "Flow Run Concurrency (Optional)" is working at the work queue level?

Ah, the intricacies of flow run concurrency. It's like a beautiful ballet of tasks, each waiting for their turn in the spotlight. But let's not get carried away with the metaphor, here's how it works: Flow run concurrency limits are set at a work pool and/or work queue level. This limit specifies the maximum number of concurrent flow runs that can be in a Running state at any given time. If a flow run attempts to enter a Running state and there are no concurrency slots available, the transition to a Running state will be delayed and the client is instructed to try entering a Running state again in 30 seconds. This is achieved using task tags. You can specify an optional concurrency limit as the maximum number of concurrent task runs in a Running state for tasks with a given tag. The specified concurrency limit applies to any task to which the tag is applied. If a task has multiple tags, it will run only if all tags have available concurrency. Tags without explicit limits are considered to have unlimited concurrency. However, be careful when setting the concurrency limit to 0 for a tag. Any attempt to run a task with that tag will be aborted instead of delayed. Also, using concurrency limits on task runs in subflows can cause deadlocks. As a best practice, configure your tags and concurrency limits to avoid setting limits on task runs in subflows. I hope that clears things up a bit. It's a bit like trying to explain the meaning of life, the universe, and everything. But at least in this case, the answer isn't just 42. Here's the source if you want to delve deeper into the thrilling world of flow run concurrency.

@Marvin Can I set "Flow Run Concurrency" at the work_queue level using the deployments sections in

prefect.yaml

?

deployments:
  - name: Test
    version:
    tags:
      - test
    description: Test.
    schedule:
    entrypoint: flow-test.py:test
    work_pool:
      name: dev
      work_queue_name: dev
      job_variables: {}
    parameters: {}

Ah, the

prefect.yaml

file, a veritable treasure trove of configuration options. It's like a choose-your-own-adventure book, but with more YAML and less fun. But I digress, let's get to your question. The

prefect.yaml

file is used to configure your Prefect deployment. In the

deployments

section, you can specify the details of your deployment, such as its name, version, tags, description, schedule, entrypoint, work pool, and parameters. However, the

prefect.yaml

file does not currently support setting flow run concurrency at the work queue level. The concurrency of flow runs is managed at the work pool level, and it can be set using the Prefect CLI with the

set-concurrency-limit

command. For example, to set a concurrency limit of 5 on a work pool named 'dev', you would use the following command:

bash
prefect work-pool set-concurrency-limit --name dev --limit 5

This command will set a concurrency limit of 5 on the 'dev' work pool, meaning that only 5 flow runs can be in a Running or Pending state at any given time. I hope that answers your question. It's a bit like trying to explain the inner workings of a black hole, but at least in this case, you won't get spaghettified. Here's the source if you want to delve deeper into the fascinating world of work pools and work queues.