High scalable message buffer for Erlang/Elixir

A generic message buffer behaviour with support for pooling, back-pressure, sharding, and distribution.

The gen_buffer can be illustrated as:

                              |--[gen_buffer_worker]
                              |
      +--[gen_buffer_sup]--+  |--[gen_buffer_worker]
      |                    |  |
      |    <ets_buffer>    |--|--[gen_buffer_worker] -> <your message handler>
      |                    |  |
      +--------------------+  |--[gen_buffer_worker]
                    ^         |
                    |         |--[gen_buffer_worker]
  (enqueue message) |                   ^
                    |                   | (sends message to worker)
  [gen_buffer]--------------------------+
       ^       (worker available?)
       |
    messages

Some implementation notes:

• A buffer is represented by its own supervision tree. The main supervisor creates the buffer itself (using an ETS table) and a pool of workers for it; each worker is a gen_server.

• At the moment a message is sent to the buffer, it tries to find an available worker (gen_server) and then dispatch the message directly to it. If there is not any available worker, the message is stored into a buffer data struct created and handled using ets_buffer. Once a worker becomes available, those buffered messages are processed.

• In order to get better and/or higher scalability, the buffer can be partitioned, it supports sharding under-the-hood. Therefore, in a partitioned buffer, the incoming messages are spread across the configured partitions (for load balancing); you can configure the desired number of partitions with the option n_partitions when you start the buffer for the first time.

• It is also possible to run the buffer in a distributed fashion on multiple nodes using the gen_buffer_dist module instead.

In your rebar.config:

{deps, [
  {gen_buffer, {git, "https://github.com/cabol/gen_buffer.git", {branch, "master"}}}
]}.

In your mix.exs:

def deps do
  [{:gen_buffer, github: "cabol/gen_buffer", branch: "master"}]
end

First of all, we have to create a message handler module:

-module(my_message_handler).

-behaviour(gen_buffer).

%% gen_buffer callbacks
-export([
  init/1,
  handle_message/3
]).

%%%===================================================================
%%% gen_buffer callbacks
%%%===================================================================

%% @hidden
init(_Args) ->
  % initialize your handler state
  {ok, #{}}.

%% @hidden
handle_message(Buffer, Message, State) ->
  % your logic to process incoming messages goes here
  Response = {Buffer, Message},
  {ok, Response, State}.

%% Optionally you can implement `handle_info/3` and `terminate/3`

Now we can create our buffer calling gen_buffer:start_link(BufferName, Opts)

gen_buffer:start_link(my_buffer, #{message_handler => my_message_handler}).

It is also possible to start the buffer as part of your supervision tree in your app. In your supervisor, within the init/1 callback, you can add the buffer spec to the supervisor's children list:

%% @hidden
init(_) ->
  Children = [
    gen_buffer:child_spec(#{buffer => my_buffer, message_handler => my_message_handler})
  ],

  {ok, {{one_for_one, 0, 1}, Children}}.

You can run observer:start() to see how the buffer looks like.

The following are the options for gen_buffer:start_link/2:

Messaging is asynchronous by nature, as well as gen_buffer, then when you send a message to a buffer, it is dispatched to another process to be processed asynchronously via your message handler. If you want to receive the result of your message handler, you have to start the buffer with send_replies option to true. For sending messages we use the function gen_buffer:send/2,3 as follows:

% if option send_replies has been set to true, the buffer sends the reply to
% the caller process
Ref1 = gen_buffer:send(my_buffer, "hello").

% or you can specify explicitly to what process the buffer should reply to;
% but send_replies has to be set to true
Ref2 = gen_buffer:send(my_buffer, "hello", ReplyToThisPID).

When we send a message calling gen_buffer:send/2,3, a reference is returned and it can be used to receive the reply, like so:

% this is a blocking call that waits for your reply for 5000 milliseconds
% by default. If none reply is received during that time, {error, timeout}
% is returned
gen_buffer:recv(my_buffer, Ref).

% or you can pass the timeout explicitly
gen_buffer:recv(my_buffer, Ref, 1000).

There is a function gen_buffer:send_recv/2,3 which combines the previous two functions in one, meaning that, when you send a message using this function, it gets blocked until the reply arrives or until the timeout expires.

% by default, the timeout is 5000 milliseconds
gen_buffer:send_recv(my_buffer, "hello").

% or you can pass it explicitly
gen_buffer:send_recv(my_buffer, "hello", 1000).

It is also possible to increase or decrease the number of workers in runtime for traffic throttling. Controlling the number of workers we can control the throughput and/or processing rate too.

% adding one worker inheriting the initial options at startup time
gen_buffer:incr_worker(my_buffer).

% passing/changing the options
gen_buffer:incr_worker(my_buffer, OtherOpts).

% removing one worker (a random worker)
gen_buffer:decr_worker(my_buffer).

% adding 5 workers at once inheriting the initial options at startup time
gen_buffer:set_workers(my_buffer, 5).

% passing/changing the options
gen_buffer:set_workers(my_buffer, 5, OtherOpts).

There are several functions to get info about the buffer, such as:

% getting the buffer size (number of buffered messages)
gen_buffer:size(my_buffer).

% getting all available info about the buffer
% check gen_buffer:buffer_info() typespec
gen_buffer:info(my_buffer)

% getting all available info about all created buffers
% check gen_buffer:buffers_info() typespec
gen_buffer:info()

% getting one random worker PID
gen_buffer:get_worker(my_buffer)

% getting all buffer workers (list of PIDs)
gen_buffer:get_worker(my_buffer)

This is not recommended since it breaks the essence of messaging; remember we mentioned before the messaging is asynchronous by nature. Nevertheless, there is a way to execute the handle_message logic for a message bypassing the normal path, it just finds a worker and executing the logic directly with it.

% if the evaluation fails, it retries one more time by default
gen_buffer:eval(my_buffer, "hello")

% or you can pass the desired number of retries
gen_buffer:eval(my_buffer, "hello", 5)

The main use cases are for testing, debugging, etc.

$ make test

You can find tests results in _build/test/logs, and coverage in _build/test/cover.

NOTE: gen_buffer comes with a helper Makefile, but it is just a simple wrapper on top of rebar3, therefore, you can run the tests using rebar3 directly, like so: rebar3 do ct, cover.

$ make doc

NOTE: Once you run the previous command, a new folder doc is created, and you'll have a pretty nice HTML documentation.

Copyright (c) 2019 Carlos Bolanos.

gen_buffer source code is licensed under the MIT License.