arg_chunks = [
            args_list[i : i + chunk_size]  # noqa
            for i in range(0, len(args_list), chunk_size)
        ]
        args_lst = [
                    (args[0], args[1])
                    for chunk in arg_chunks
                    for args in chunk
                    ]
        # since 100 args / 10 workers, my agent will spawn 10 worker sub-flows
        <http://logger.info|logger.info>(f"Spawning {len(arg_chunks)} worker flows...")
        await asyncio.gather(
            *[
                run_deployment(
                    # returns a FlowRun object
                    timeout=0,  # allows non blocking
                    name=deploy_id,
                    parameters={
                                "project": project,
                                "arg_1": args[0],
                                "arg_2": args[1]
                                },
                )
                for args in args_lst
            ]
        )

@flow(name="dask-flow", task_runner=DaskTaskRunner(address="<tcp://dask-scheduler:8786>"), persist_result=True, result_storage=LocalFileSystem.load("local-storage"))
async def gpu_computation(project, arg_1,arg_2):
     .....
     .....
     project["col"] = gpu_compute(arg_1,arg_2)
     project.save(...)
     -----

so i have this dask flow which creates n workers based on how many chunks there are, and the flow is a long running task, that runs the args and does some gpu computation and then saves a file, the problem i am having is it is cancelling the tasks as some are long running, i dont want them to cancel, just to wait, or perhaps i should just have jobs start for first chunk and then have second chunk wait for completion of first? i dont know the best way to do this ? thoughts?

hi @Cody Webb

it is cancelling the tasks as some are long running

what is cancelling the tasks?

one sec running again to get the logs i ran at work and then power went out so i just got home lol

oh jeez! 👍 no rush

okay maybe it was because i was on battery, now it seems lke one will successfully complete but then most others error out ? is there way to stagger these out or only chunk them or am i missing something? im thinking that im overloading the api im hitting that does the gpu computation or something?

most others error out

do you have the trace for one that errored out?

is there way to stagger these out or only chunk them or am i missing something?

im not sure i understand, but you can use concurrency limits to enforce max parallelism

im thinking that im overloading the api

are you running against a local server or cloud? or ephemerally its worth mentioning that

run_deployment

is not a task, so the DaskTaskRunner wont change the execution of

run_deployment

calls at all, unless you're calling it from a task - hard to tell from above where you're awaiting your run_deployment calls

im running it inside a flow^

gotcha, so you could be running into trouble running lots of stuff against the ephemeral api (i.e. no dedicated API, just the one that we spin up if you're not talking to a server/cloud) and task_runners dont do anything to subflow execution, what is your expectation with the DaskTaskRunner here?

theres really no trace outside of the it saying ERROR : FAILED no reason as to why, and looking in the gpu service isnt saying much of anything either my intuition is what i was saying, as if i do a job of just 10 or so itll run completely fine, but anything more it dies, so was wondering if there a better approach here

yep i wrote that recipe 🙂

ha nice, yea its cool , basically each task doesnt depend on any of the others so was trying to do them concurrently

that makes sense! hmm

looking in the gpu service

as a sanity check, do things work as expected if you remove dask and your GPU service from the situation?

yea

oh wait, is the

gpu_computation

flow the one being kicked off with

run_deployment

?

ya exactly

so len_chunks flow runs are all pointing at the same dask scheduler ?

the run_deployment kicks off the run which i send the args to the gpu to do the inference on

im not sure how you're using tasks in the

gpu_computation

flow, but i suspect that the weirdness is related to the dask task runner somehow - where / how are you using dask in there?

thats all im using it for is to dispatch the tasks, its all basically just api code, hitting different endpoints of an api

just as a datapoint, did you try without the dask task runner?

no but i will, what one do you rec? just sequential?

id just rm

task_runner

kwarg and let it use the default

ConcurrentTaskRunner

okay

nice!

okay so, in that case, should i add another agent/pool if i want to go faster?

but seems like theres some bottlenecks on how many tasks/requests the gpu backend/endpoints can handle concurrently

i suppose that depends, is your gpu service still the bottleneck? if it were, im not sure extra worker/pools would help

kinda like horiz. scaling?

i might be misunderstanding

i think the concurrent requests are the bottleneck

(Still have like half of the gpu mem)