Golang singleflght concept in Elixir

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:

https://github.com/whitfin/cachex/blob/31c3fd0d1b243803ae7503a720c4d498902a6750/lib/cachex/services/locksmith.ex

The 3 workers were scheduled to start at the same time but the runtime chose to start worker #3 first. Pretty normal for async runtimes for things to be out of order when jobs start. Important part is that eventually everything converges… if you write your algorithm correctly, of course.

Take a look at the timestamps. It seems the logs are out of order in regards to time. Worker 3 requested work first.

Porting patterns from one language to another is really helpful.

Wow, this is fantastic! Thanks for the demo code.
one additional question tho, why was the worker 3 doing the expensive work instead of worker 1? worker 1 was the first to be invoked so I thought it would make sense if it reaches the expensive work first

One thing to watch out for with Cachex.fetch and Ecto’s sandbox - the function passed to Cachex.fetch runs in a separate process from the caller.

Happy to help!

omg, TIL about Cachex.fetch. I don’t know how many versions of this I handrolled over the years.

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.

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