Implementing fixed-size tasks pool

[AndreiPashkin]

Hello!

I'm trying to implement the pool pattern which would process a fixed amount of tasks at a time. Basically a more advanced alternative to task_all() that would not launch all the tasks at once but process them gradually, ensuring that only N amount of tasks are launched at once. The exact problem that I'm trying to solve are OOM errors that we are getting when we schedule all the tasks with task_all() at once.

I thought that I could employ task_any() to check for completion of individual tasks and dequeue them but it turned out that repeated calls to task_any() on the same tasks lead to an exception like that:

e1c231a2ce8345fcb2f01c5fea36be7d: Function 'MyOrchestratorFunction (Orchestrator)' failed with an error. Reason: Message: 'AtomicTask' object has no attribute 'append', StackTrace:    at System.Threading.Tasks.Task.ThrowIfExceptional(Boolean includeTaskCanceledExceptions)
   at System.Threading.Tasks.Task`1.GetResultCore(Boolean waitCompletionNotification)
   at System.Threading.Tasks.Task`1.get_Result()
   at Microsoft.Azure.WebJobs.Extensions.DurableTask.TaskOrchestrationShim.InvokeUserCodeAndHandleResults(RegisteredFunctionInfo orchestratorInfo, OrchestrationContext innerContext) in D:\a\_work\1\s\src\WebJobs.Extensions.DurableTask\Listener\TaskOrchestrationShim.cs:line 150. IsReplay: False. State: Failed. HubName: TestHubName. AppName: . SlotName: . ExtensionVersion: 2.7.1. SequenceNumber: 24. TaskEventId: -1
Executed 'Functions.OrchestrateClipExtraction' (Failed, Id=b82e25de-fe1d-4141-8561-c6f85a5f5e54, Duration=35ms)

And it seems that there is no other way to get the status of an existing task - so there is no way to monitor it and remove from the pool. So I'm stuck because of that.

I wonder if anyone could kindly provide any pointers? Maybe I'm just doing it wrong and this is not how concurrency limiting should be done in AZ Functions? Or maybe there is a way to implement pooling?

To be more specific, here are my questions:

1. Is it possible to get task's status from an orchestrator function, without waiting for it's completion?
2. What is the recommended way to limit concurrent execution of activity tasks in AZ Functions?

p.s.
I'm also aware of batching technique when task_all() is used on small batches of tasks - but it's not very efficient.

[cgillum]

Take a look and see if any of the suggestions on this thread work: #596. The code samples are primarily in C# or JavaScript, but in theory the same approach should work for Python.

Adding @davidmrdavid in case he has any thoughts about the 'AtomicTask' object has no attribute 'append' error message, which sounds suspicious.

[AndreiPashkin]

@davidmrdavid, it's pretty easy to reproduce - try to call task_all() more than once for the same task.

[davidmrdavid]

Hi @AndreiPashkin,

I just tried to repro this with this code:

import logging
import azure.durable_functions as df

def orchestrator_function(context: df.DurableOrchestrationContext):

    task = context.call_activity('Hello', "Tokyo")
    result = yield context.task_all([task])
    logging.info(f"result is {result}")
    result = yield context.task_all([task])
    logging.info(f"result is {result}")
    return ""

main = df.Orchestrator.create(orchestrator_function)

and I experienced no exception regarding 'AtomicTask' object has no attribute 'append' .
I did experience a warning ("UserWarning: Potential duplicate Task completion for TaskId: 1") that seems to be causing some issues though, and I'm looking at this now.

Please let me know if I'm repro'ing your issue correctly. To be sure, please ensure that you're using the latest version of the SDK on PyPI

[davidmrdavid]

I just opened a PR with a fix to the warning listed above in here: Azure/azure-functions-durable-python#391

[AndreiPashkin]

@davidmrdavid, I can say that I definitely was able to reproduce it locally and in Azure. I'll try to do it again and post the results.

[davidmrdavid]

Thanks! Having a .zip'ed version of your reproducer would help me ensure I'm running the right code. Thanks again!