3FanYu
Golang singleflght concept in Elixir
As what I’ve learned so far, if I build an API server using Elixir and Phoenix. Each requests against the API server is an independent process.
Let’s say if I have an GET API to retrieve items and it takes a long time to query from the database, therefore I decided to cache it. When a cache is missed, the process should proceed to query from database, and save the result to cache on the way responding back the result to client. To prevent large amount of concurrent requests hitting the DB when the cache is penetrated, In golang, they would implement singleflight in the cache process. I’m curious how Elixir and Phoenix would deal with this situation.
Most Liked
jhogberg
Assuming that you want several clients each making the same request to get the same response, it should be very simple to roll it yourself. The general idea would be along these lines:
%% "Cache process"
receive_loop(State) ->
receive
{request, Pid, Ref, Request} ->
receive_loop(make_request(Pid, Ref, Request, State));
{database_response, DbRef, Result} ->
receive_loop(handle_response(DbRef, Result, State))
end.
make_request(Pid, Ref, Request, {Cache, In, Out}) ->
case {Cache, In} of
{#{ Request := Cached }, #{}} ->
%% Cache hit, reply with what we have.
Pid ! {reply, Ref, Cached},
{Cache, In, Out};
{_, #{ Request := Others }} ->
%% This request has already been made but we haven't received a
%% reply yet. Tell the previously made request that we want the reply,
%% too.
{Cache, In#{ Request => [{Pid, Ref} | Others] }, Out};
{#{}, #{}} ->
%% Not in cache, no outstanding request: make a new request.
%%
%% It is very important that this request is made asynchronously,
%% otherwise everyone will have to wait for each request to finish
%% which makes the whole thing pointless.
DbRef = call_your_database_asynchronously(self(), Request),
{Cache,
In#{ Request => [{Pid, Ref}] },
Out#{ DbRef => Request } }
end.
handle_response(DbRef, Result, {Cache, In, Out}) ->
#{ DbRef := Request } = Out,
#{ Request := Clients } = In,
%% Respond to all clients that made the same request, then
%% cache the result and remove this outstanding request
%% from the state.
%%
%% If you only want the "single flight" functionality, just skip
%% the cache parts in this example.
[Pid ! {reply, ClientRef, Result} || {Pid, ClientRef} <- Clients],
{Cache#{ Request => Result },
maps:remove(Request, In),
maps:remove(DbRef, Out)}.
(Apologies for putting it in Erlang, hopefully you’ll get the gist of it though
.)
dimitarvp
Cachex has got you covered as @kokolegorille mentioned.
Here’s a demo that shows you that the expensive work is done only once. You can paste this in e.g. dont_repeat_work.exs and just do elixir dont_repeat_work.exs after.
There:
Mix.install([{:cachex, "~> 3.6"}])
Cachex.start_link(name: :our_cache)
defmodule Worker do
def expensive_work(id) do
Cachex.fetch(:our_cache, "expensive_data", fn _key ->
IO.puts(
"#{inspect(NaiveDateTime.utc_now())}: doing expensive work on behalf of worker #{inspect(id)}"
)
:timer.sleep(1000)
{:commit, "EXPENSIVE_VALUE"}
end)
end
def use_expensive_work(id) do
IO.puts("#{inspect(NaiveDateTime.utc_now())}: requesting access from worker #{inspect(id)}")
value = expensive_work(id)
IO.puts("#{inspect(NaiveDateTime.utc_now())}: worker #{id} received value: #{inspect(value)}")
end
end
1..3
|> Task.async_stream(fn id ->
Worker.use_expensive_work(id)
end)
|> Stream.run()
On my machine this returned:
~N[2024-04-09 11:31:13.743650]: requesting access from worker 1
~N[2024-04-09 11:31:13.743645]: requesting access from worker 2
~N[2024-04-09 11:31:13.743642]: requesting access from worker 3
~N[2024-04-09 11:31:13.746957]: doing expensive work on behalf of worker 3
~N[2024-04-09 11:31:14.748989]: worker 2 received value: {:ok, "EXPENSIVE_VALUE"}
~N[2024-04-09 11:31:14.748980]: worker 3 received value: {:commit, "EXPENSIVE_VALUE"}
~N[2024-04-09 11:31:14.749017]: worker 1 received value: {:ok, "EXPENSIVE_VALUE"}
Notice the {:commit, ...} tuple. That means the work has been done only once and the other 2 {:ok, ...} tuples mean a cache hit. The important part however is that this does NOT lead to all 3 workers concurrently doing the expensive work and all 3 writing to the cache at (nearly) the same time. Cachex took special care to prevent this.
kokolegorille
There is the Cachex library…
As mentionned by the previous answers, it is simple to implement
in cache? → cache
not in cache? → query → cache the result
Last Post!
dimitarvp
Yeah, it’s educational. Though note that the BEAM VM has no mutexes per se, they can be emulated via various means. Cachex chose to do it with an :ets table and a Process dictionary. If you are curious here’s the meat of the implementation:
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
- #elixirconf-us
- #advent-of-code
- #distillery
- #processes
- #forms
- #api
- #metaprogramming
- #security
- #hex









