eyelight93
How to leverage concurrency and speed up the aggregate operation
Hi guys, I’m having this problem.
Let say I have a list of 10_000 items, each item is a map with a RRule column
[
%{rrules:
[
%{value: 1, rule: rrule},
%{value: 2, rule: rrule2}
]
},
........
]
I’m spawning 10_000 tasks, each will handle one item
Enum.map(list, fn item -> Task.async(fn -> handle_item(item.rrules) end) end)
|> Enum.reduce(%{}, fn task, acc ->
value = Task.await(task)
Map.put(acc, value, 1)
end)
handle_item/1 will look like this
Enum.map(rrules, fn element ->
element.rule
|> CockTail.occurrences() # This returns a stream of time-series data ~ 8000 records
|> Enum.reduce(%{}, fn time, acc -> Map.put(acc, time, element.value) end)
end)
obviously, this implementation is very inefficient. But I’m not sure how I can leverage the concurrent in this case, to speed up the aggregate operation.
I would be very appreciated if somebody could point me to the right direction, thanks in advance
Most Liked
kip
I’d consider using Task.async_stream/2 or Flow for this first part. I believe it will do a better job on managing throughput.
I assume this isn’t your real code because here you are using the map returned from handle_item/1 as a map key.
You could consider using an :ets table for the accumulator since that allows concurrent write access. However concurrency here isn’t a magic bullet since you are already running this code pithing a concurrent environment (the original 10_000 items).
This looks like you are exploding an rrule into individual event times (is that what CockTail.occurrences/1 does?). I think the key question here is whether you need to materialise all the times up front or whether there is an opportunity to lazily evaluate the individual times when you need them. The calendar_recurrence might help with this.
kip
I think switching the Flow would be your best bet since it is designed for exactly this kind of use case. Something like (not tested):
items()
|> Flow.from_enumerable(max_demand: :erlang.system_info(:schedulers_online))
|> Flow.map(&handle_item/1))
|> Flow.partition()
|> Flow.reduce(fn -> %{} end, fn time, acc -> Map.put(acc, time, element.value) end)
|> Enum.to_list()
dimitarvp
10k items? This should barely blip for 0.5 secs IMO. If you can put some example GitHub project that can produce the items so we can check a few algorithms, we should be able to help further.
Popular in Questions
Other popular topics
Categories:
Sub Categories:
Forums
Popular Tags
- #ecto
- #liveview
- #troubleshooting
- #learning-elixir
- #deployment
- #library
- #erlang
- #testing
- #genserver
- #mix
- #absinthe
- #remote-other
- #otp
- #plug
- #how-to-question
- #macros
- #postgres
- #channels
- #elixirconf
- #exunit
- #discussion
- #code-sync
- #javascript
- #podcasts
- #onsite
- #dialyzer
- #docker
- #authentication
- #umbrella
- #full-time-contract
- #podcasts-by-brainlid
- #ecto-query
- #elixir-ls
- #phoenix_html
- #iex
- #blog-post
- #graphql
- #genstage
- #ai
- #websockets
- #supervisor
- #advent-of-code
- #elixirconf-us
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #security
- #performance









