Hey guys, I am trying to create_flow_runs programatically using

from prefect.client import Client
import pandas as pd

if __name__ == "__main__":
    dates = pd.bdate_range(pd.Timestamp("20200601"), pd.Timestamp("20200630"))

    c = Client()
    for d in reversed(dates):
        c.create_flow_run(
            "e04181f3-86b3-4f6f-87cf-ed7c24cb95ec",
            parameters=dict(date=d.strftime("%Y-%m-%d")),
            context=dict(name="John Doe"),
            run_name=f"test_backfill_{d.strftime('%Y%m%d')}")

But some of the flow runs are stuck with

Queued due to concurrency limits. The local process will attempt to run the flow for the next 10 minutes, after which time it will be made available to other agents.

I am on the dev version of the cloud backend and understand that there are limits. I have the following questions: 1. Am I hitting this error because of the dev cloud version? 2. Although, the message says the flow will be rescheduled within the next 10 minutes, it takes around 30 mins for the rest of the tasks to get scheduled and run. 3. Is there a better way to create a backfill? I am avoiding

FlowRunTask

because context is currently not passed to the triggered flow runs

The

FlowRunTask

is internally using

Client()

. Unfortunately the

FlowRunTask

is broken in v0.12.4 when using server-mode (non-cloud). In general using the

FlowRunTask

is the preferred way to trigger (registered) flow runs. Please ignore this advice when you're using the cloud-mode flavor of Prefect.

As Sven said, can you confirm that you’re using the Cloud product? AFAIK (but maybe I missed it), there isn’t a dev version of that available. Specifically looking at the rescheduling message, what’s happening under the hood is that your flow runs are bound by a concurrency limit, let’s say X. When you get this message about concurrency limiting, what’s happening is that in the local process, the flow run will check with orchestration (cloud / server) every ~30 seconds to see if the number of running flows is < X. If so, the flow run will proceed, otherwise it’ll go back to waiting. When it says it’ll be made available to other agents after 10 minutes, it means that the flow run will attempt to run in a different process (helpful in case the original process dies for whatever reason), and this new process will also require doing the concurrency check with orchestration. The experience you’re getting with having ~30 minutes for the tasks to get scheduled and run is likely due to how long your processes are taking overall. The flow concurrency checks will see if there are any flows in the

Running

state, which means that tasks can run. If your tasks are taking a long time, or are retrying, the flow run will still be

Running

while that is happening, thus taking a concurrency slot.

@Alex Cano I can confirm that I am using the cloud product. I was making a distinction between the developer, team and enterprise versions as mentioned here https://www.prefect.io/get-prefect

@karteekaddanki Gotcha! Thought you had meant a development version, not the developer version! To answer your questions: 1. Yes, you are receiving that message because of the developer cloud account tiering works. 2. Since you’re on the developer tier, each flow run needs to complete running prior to the next starting, but there shouldn’t be anything causing a 30 minute delay between a flow run starting to run and the tasks getting generated and scheduled to run. This will likely require someone from the Prefect team to help debug 🙂 3. On best practices for doing backfills, I’d think doing it the way you’re doing it would work well (using the client & creating several flow runs). You could use

FlowRunTask

, but that’d mean making a new flow just to do the backfill, which is likely more overhead than most people would want.

@Alex Cano Thanks! I don't mind the overhead of

FlowRunTask

if there was a way to pass the context to the flows. All my backfills require date to be modified in context (and not as a parameter because of target caching) and I cannot change that bit right now.

Gotcha! Actually quick question @karteekaddanki, do you have heartbeating enabled for this flow?

@Alex Cano Not entirely sure. I haven't disabled it explicitly. I have a feeling I might be doing something wrong in how I am running my code. The code in my original message is in a python script and execute it from the command line passing flow_id and dates from the command line.

@karteekaddanki Are you able to share the main code itself? We may be able to help more if you can!

@Alex Cano I've edited my original post removing the

argparse

bits of the code for simplicity. Essentially, this is the script I am running. The Flow id corresponds to a flow with one trivial task and shouldn't take more than a second to run.

@karteekaddanki Gotcha! In this case, your script looks fine in terms of creating the flow runs. The next question is confirming your Agent does find the flow runs successfully (albeit slowly, correct?), and that whatever you’re using for execution is also executing the code correctly? My next thought on where things are getting held up is the execution layer, so maybe take a peek at those logs to see if you’re finding anything interesting

@Alex Cano Yup. Everything seems to be working correctly but after a delay of 30mins (almost to the dot). The tasks finish cleanly. I'll try to spend more time on a reproducible

foobar

example.

@karteekaddanki That sounds good! It also will likely be a good idea to loop in someone from the Prefect team as well… since this seems very odd, especially with the almost exactly 30 minute delay.

Hi! Just FYI I’m poking around a bit on this, this may be a bug on the cloud side and if so I will keep you updated 🙂

@Laura Lorenz (she/her) Thanks for the PR! It'll make my workflow a bit more simple. Regarding the concurrency bug, I think it maybe a race condition that is triggered by flows that take a very short time to finish. I haven't had any issues when I replaced my toy-flow examples with my real flows. I'll have another stab at it once I finish porting my flows to Prefect.