tfwright

tfwright

Managing multiple related async tasks

I have a large amount of user data that I want to run some expensive analysis on every time the user makes a change. Obviously I don’t want the user to to have to wait for the result of this analysis so I want to run it in it’s own process using Task.async. However, I also want to “debounce” these tasks because once a new update occurs the previous task becomes obsolete so I want to kill it and start a new process. To add further complexity the analysis has several independent parts each of which I would also like to run concurrently, with the parent waiting on all of them to finish. Any suggestions about the best architecture for something like this?

My current idea is to use an Agent to create a registry of user “analyses” that handles an update by killing any existing task for that user (and thus killing its children) and starting a new one and adding it to the registry. When the task, uninterrupted by further updates, is allowed to finish, it will use PubSub to report the results back to the user’s process (a LiveView if that’s important). But I wasn’t sure if Task.Supervisor would be a better fit?

Marked As Solved

tfwright

tfwright

Here’s what I came up with. It all seems to work as expected but there’s a lot I’m unsure about:

I added an analysis registry and supervisor:

      {Registry, keys: :unique, name: MyApp.Analysis.Registry},
      {Task.Supervisor, name: MyApp.Analysis.Supervisor}

I wondered briefly whether to namespace them since it seems fairly easy to use them in different contexts, or at least Task.Supervisor, but it seemed like most examples I could find used namespaces by default.

Next I added a new function to my analysis module to handle the “launch or kill” spec

  def request_analysis(%Sources.Source{id: source_id} = source) do
    case Registry.register(Analysis.Registry, "source-#{source_id}-analyzer", :value) do
      {:ok, _} ->
        analyze(source_id)

      {:error, {:already_registered, pid}} ->
        Process.exit(pid, :kill)
        request_analysis(source)
    end

    :ok
  end

I’m not sure I understand the point of the “value” argument in Registry.register/3 and I’m just using a placeholder there. I even considered just using nil. My naive expectation would be that the “value” of this function would naturally be the pid itself. The Registry docs were a bit confusing because they assumed you wanted to register a new process by naming it instead of using register.

I then updated my LiveView to call this function instead of the analsys:

  defp request_reanalysis(socket) do
    Task.Supervisor.async_nolink(
      Analysis.Supervisor,
      Analysis.Analyzer,
      :request_analysis,
      [socket.assigns.source],
      shutdown: :brutal_kill
    )

    socket
    |> assign(analyzer_data: nil)
  end

The analyze function I left mostly intact, aside from adding the PubSub call. It uses Task.async and Task.await to concurrently build the data and compose it when each sub-part is complete. I suppose now that I have introduced Task.Supervisor I should use that given the advice above? But then I’m not sure what the purpose of Task.await would be. Poking around a bit it just seems like people mostly just always use one or the other. The documentation doesn’t seem to focus on the trade offs involved, of which surely there must be some?

Also Liked

ityonemo

ityonemo

This should be doable using Registry (don’t write an Agent that duplicates Registry functionality). Also don’t track the analyses separately. Wrap all of the subtasks into a single parent task, then kill that parent task, which should in turn kill the child tasks (assuming you’ve started them linked). I think you got that part.

Your parent task should

  1. check the registry if a running analysis exists, if so, unregister, then kill the running analysis.
  2. always register itself.

There’s a race condition here where two parent tasks could have been launched really close together, and one of them beats the other to re-registration; I recommend check if the registration fails, if so quit out, under the assumption that another system has obtained the registry slot.

Could be like 5-6 lines of elixir if you do it right.

ityonemo

ityonemo

Negative. Going to steal a page from James Grey and Bruce Tate here: Supervision is about lifecycle management, not just about restarting. Most importantly, when you supervise, your process gets tracked as part of the supervision hierarchy. Even if nothing else “depends on it”, and you get clean process domain coupling using links and monitors, many tools commonly use the supervision tree to monitor “everything” using a sane tree digraph, instead of a graph of links and monitors which could look like a ball of hair. Even if you never use it, the cost of organizing things into a tree is so low, you might as well do it. Just make sure your Tasks are supervised as transient (which I think is the default for Task.Supervisor)

tfwright

tfwright

That looks very promising! Thanks for the tip.

Where Next?

Popular in Questions Top

_russellb
I want to try my hand at web scraping. What tools/libraries do I need to use. I’m hoping to turn this into something professional so don’...
New
vertexbuffer
Hello, can anybody help here..? I have a list of players and I what to delete an element, but every for loop the list is reverting to ori...
New
Patoshizzle
After calling mix ecto.create I get this error: 17:00:32.162 [error] GenServer #PID<0.412.0> terminating ** (Postgrex.Error) FATAL...
New
Fl4m3Ph03n1x
About me? ( if you have nothing better to do than reading about some random guy in the internet :stuck_out_tongue: ) Hello all, this is ...
New
LegitStack
I’m trying to make a websocket server in Phoenix or raw Elixir. I heard about gun, I think I could use cowboy, but since I’m not that sma...
New
Lily
In templates/appointment/index.html.eex: <%= for appointment <- @appointments do %> <tr> <td><%= appoi...
New
dokuzbir
I want to highlight html closing tags when i click a html tag. That works in .html files but doesnt work for html.eex templates. How can...
New
nobody
Hi! In PHP: $_SERVER[‘SERVER_ADDR’] - in Elixir? Searched the docs for ip address and the web, no good results. Thanks!
New
lanycrost
Hi everyone! I need implement if…else if…else condition from my elixir code, and anymore of this control flow structures not work proper...
New
vonH
In asking this question I am more interested about the expressiveness of the language itself and less concerned about the availability of...
New

Other popular topics Top

marius95
Hello everyone, I try to use an Javascript Event Handler in my root.html.leex file. Therefore I created a function in the app.js file: ...
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
SoCreat
i’m a new one to elixir which editor can i use vs code? or atom? Thanks! :smiley:
New
fayddelight
I tried installing elixir 1.11.2 erlang 23.3.4 via asdf in my zsh shell. Enabled the versions locally and globally. When I list them ...
New
nobody
Hi! In PHP: $_SERVER[‘SERVER_ADDR’] - in Elixir? Searched the docs for ip address and the web, no good results. Thanks!
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
romenigld
I am trying to run a deploy with docker and I successfully runned with this command: docker build -t romenigld/blog-prod . but when I t...
New
joaquinalcerro
Hi there, I am working with Ecto-Postgresql and I need to call all of the records from a specific table but the table has 40,000 records...
New
svb
Hi! Currently I want to submit a form by pressing the Enter key. However, since my input field is of type “textarea” this is just adds a...
New
lanycrost
Hi everyone! I need implement if…else if…else condition from my elixir code, and anymore of this control flow structures not work proper...
New

We're in Beta

About us Mission Statement