cosmos-sajal

cosmos-sajal

Async Task not returning results

I have around 10-15 ecto queries which I want to run async in my API code.
I am using Task.async and Task.yeild_many

Following is the code for async task -

def get_tasks() do
  task_1 =
      Task.async(SomeModule, :some_function, [
          param_1,
          param_2
      ])

  task_2 =
      Task.async(SomeModule, :some_function, [
          param_1,
          param_2
      ])

  task_3 =
      Task.async(SomeModule, :some_function, [
          param_1,
          param_2
      ])

  [task_1, task_2, task_3]
end

I get the tasks in my main function as -

        [
          {_, task_1},
          {_, task_2},
          {_, task_3}
        ] =
          [
            task_1,
            task_2,
            task_3,
          ]
          |> MyCode.TaskHelper.yeild_multiple_tasks()

And my task helper code is given below -

defmodule MyCode.TaskHelper do

  def get_results_or_shutdown(tasks_with_results) do
    Enum.map(tasks_with_results, fn {task, res} ->
      res || Task.shutdown(task, :brutal_kill)
    end)
  end

  @doc """
  Returns the result of multiple tasks ran parallely

  ## Parameters
    - task_list: list, a list of all tasks
  """
  def yeild_multiple_tasks(task_list) do
    task_list
    |> Task.yield_many()
    |> get_results_or_shutdown()
  end
end

Each tasks are ecto queries.
The issue is the tasks are behaving randomly. Sometimes they return results, sometimes they don’t. But there was no time when all the tasks have return results (for the representational purpose I have written 3 tasks,
but I have around 10-15 async tasks).
I ran the code synchronously, and it returned the correct results (obviously). I tried changing the pool_size in config to 50 for Repo, but to no avail.

Can someone please help me out with this? I am quite stuck here.

First Post!

dimitarvp

dimitarvp

Are all tasks calling the same function?

Most Liked

peerreynders

peerreynders

How long does that take?

Task.yield_many/2 has a default timeout of 5 secs - maybe that isn’t enough time for Ecto to finish all the queries.

The issue is the tasks are behaving randomly.

I wouldn’t necessarily expect Ecto + RDBMS to behave in a deterministic fashion (order of work finished). So the apparent randomness wouldn’t be too surprising if the timeout is too short.

The other advantage of Task.async_stream/5 is that the level of concurrency can controlled with :max_concurrency (it defaults to the number of online cores).

There is also Task.Supervisor.async_stream_nolink/6.

code
explanation

Where Next?

Popular in Questions Top

minhajuddin
I have seen a lot of code which picks the first element from a list using Enum.at(0) instead of List.first. Is there a reason why people ...
New
baxterw3b
Hi guys, i’m new in the Elixir world, and i have to say, that i love it! i’m having some problem to understand anonymous functions with ...
New
jononomo
For some reason my phoenix channels are working for me in my local dev environment, but as soon as I deploy via Docker, I get a 403 error...
New
lessless
I believe there are people here who are dealing with CSV files import on the daily basis, and since Excel is a really popular tool there ...
New
fireproofsocks
Forgive me if this is obvious, but how does one delete a database record WITHOUT selecting it first? Ecto.Repo — Ecto v3.14.0 has exampl...
New
pmjoe
I have a relationship of love and hate with Elixir. Lots of things are just absolutely right, but there are some things that are kind of ...
New
jason.o
In the code below, if the create action is not set to accept “extra_key” as an input, it errors out with a message shown above. Is there ...
New

Other popular topics Top

jononomo
I am trying to figure out how Mix knows whether the environment is test, dev, or prod – where is this set? Thanks.
New
hariharasudhan94
Lets say I have map like this fetching from my database %{"_id" => #BSON.ObjectId<58eb1a7a9ad169198c3dXXXX>, "email" => ...
New
stefanchrobot
What’s the safe way to decode a JSON string into a struct? I want to avoid calling String.to_atom. Jason.decode can give me a map with st...
New
gshaw
What is the idiomatic way of matching for not nil in Elixir? E.g., First way: defp halt_if_not_signed_in(conn, signed_in_account) when...
New
Harrisonl
We have an ECS cluster with 4 services, where each task joins a single cluster, via discovery ECS discovery service. Currently when I de...
New
senggen
Erlang/OTP 25 [erts-13.2.2] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] 15:22:35.803 [error] gen_event {lager_file_backend...
New

We're in Beta

About us Mission Statement