CxLeaderboard - a comprehensive leaderboard library

For the term storage, :gb_trees might be a better fit than lists in maps.

I think DemonWare used them together with ets to keep rankings in call of duty. But that was before maps were introduced.

Wow that presentation is full of interesting insights, thank you! And gb_trees looks like a useful structure, will read more about it (esp. once erlang documentation site starts loading for me).

I wanted to link it but the website was down for me as well … You can do erl -man gb_trees in your shell.

I’ll probably give it a try soon

Let me know how it goes or if you need any help/have questions. Happy to help.

Can I get a example for Postgres auto-populate on startup.

I got the following but can’t get it to work. I think my issue is the connection is closed when the transaction ended.
related issue. Using Postgrex.stream with transactions · Issue #373 · elixir-ecto/postgrex · GitHub

worker(CxLeaderboard.Leaderboard, [:global, [data: loadDataToLeaderboard(pid)]])

def loadDataToLeaderboard(pid) do
{:ok, stream} = Postgrex.transaction(pid, fn(conn) →
Postgrex.stream(conn, “SELECT id, elo FROM users”, , max_rows: :infinity, timeout: :infinity)
end)
stream
end

@jonathanleang Interesting. I’ve never tried using these Ecto streams, I don’t think they play too well with a case where the stream should live for a long time. In our case I wrote my own module called StreamExt and implemented a couple of streaming functions in it. One based on offset/limit, and another (more efficient one) based on id comparisons.

defmodule StreamExt do
  @doc """
  Stream data in batches from an Ecto repo using a query.
  """
  def ecto_in_batches(repo, query, batch_size \\ 1000) do
    import Ecto.Query, only: [from: 1, from: 2]

    Stream.unfold(0, fn
      :done ->
        nil

      offset ->
        results =
          repo.all(from(_ in query, offset: ^offset, limit: ^batch_size))

        if length(results) < batch_size,
          do: {results, :done},
          else: {results, offset + batch_size}
    end)
  end

  def ecto_in_batches_by_id(repo, query, batch_size \\ 1000) do
    import Ecto.Query, only: [from: 2]

    Stream.unfold(-1, fn
      :done ->
        nil

      last_id ->
        results =
          repo.all(
            from(
              r in query,
              where: r.id > ^last_id,
              limit: ^batch_size
            )
          )

        {new_last_id, count} =
          results
          |> Enum.reduce({-1, 0}, fn r, {_, i} -> {elem(r, 0), i + 1} end)

        if count < batch_size,
          do: {results, :done},
          else: {results, new_last_id}
    end)
  end
end

Hey, just following up, did you manage to make it work?

@maxim
Thanks for following up, I have been trying for some time now, but kind of stuck.

here is what I got now.

   worker(CxLeaderboard.Leaderboard,[
    :global, 
    [data: loadDataToLeaderboard()]
  ])

 def loadDataToLeaderboard() do
     StreamExt.ecto_in_batches_by_id(Darkmoor.Repo, Darkmoor.User, 5)
     |> Stream.map(fn(batch) ->
       Enum.map(batch, fn({id, elo, name}) -> 
           {{elo, id}, name}
           end)
       end)
       # [{{100, 1}, "name1"}, {{200, 2}, "name2"}] this works
   end



  def ecto_in_batches_by_id(repo, query, batch_size \\ 1000) do
    import Ecto.Query, only: [from: 2]

    Stream.unfold(-1, fn
      :done ->
        nil

      last_id ->
        results =
          repo.all(
            from(
              r in query,
              select: {r.id, r.elo, r.name},
              where: r.id > ^last_id,
              limit: ^batch_size
            )
          )

        {new_last_id, count} =
          results
          |> Enum.reduce({-1, 0}, fn r, {_, i} -> {elem(r, 0), i + 1} end)
        if count < batch_size,
          do: {results, :done},
          else: {results, new_last_id}
    end)
  end

Ah I see, let me give you a specific solution here. I kind of forgot that my batch function wasn’t for direct usage. Try the following:

1. Add this file to your elixir project: stream_ext.ex · GitHub

2. Use this function:

   def leaderboard_stream() do
     import Ecto.Query, only: [from: 2]
     query = from(r in Darkmoor.User, select: {{r.elo, r.id}, r.name})
     StreamExt.ecto(Darkmoor.Repo, query, batch_size: 5, strategy: :id)
   end
   

3. And finally declare your worker like this:

   worker(CxLeaderboard.Leaderboard, [:global, [data: leaderboard_stream()]])
   

Let me know if this works.