We have a workflow that acts on a batch of data and needs to perform several processing actions on that data. We use Batch APIs for most of our processing but there are some which don’t support batch. In those cases we execute a single activity for each item, but we prefer to run them in parallel and wait for all of the results so we write something like this:
tasks = [self._do_something(item_id) for item_id in item_ids]
results = await asyncio.gather(*tasks)
where _do_something is:
async def _do_something(item_id: str):
return await workflow.execute_activity(
"do_something_activity",
item_id,
schedule_to_close_timeout=timedelta(seconds=30),
)
When we do that for say 1000 items we can see all the activities being scheduled at once and then executed when workers are available to pick them up, but after all activity tasks are done we sometimes get an error [TMPRL1101] Potential deadlock detected: workflow didn't yield within 2 second(s).
We suspect maybe it’s because those activities return data and the accumulative size of that data becomes quite big, so when all activities are done, the Workflow Task is actually now to get all that data from the activities.
Does this assumption make sense? Or is it something more basic in how the python/temporal even loop operates and we are doing something unconventional?