🥇Building Blocks of Distributed Applications. First Approach

Previously article We have covered the theoretical foundations of reactive architecture. It is time to discuss data streams, the implementation paths of reactive Erlang/Elixir systems, and the messaging patterns within them:

Request-response
Request-Chunked Response
Response with Request
Publish-subscribe
Inverted Publish-subscribe
Task distribution

SOA, MSA and messaging

SOA and MSA are system architectures that define the rules for building systems, while messaging provides the primitives for their implementation.

I do not wish to promote any specific system architecture. I advocate for the application of the most effective and useful practices tailored for a particular project and business. Regardless of the paradigm we choose, it is better to create system components with a focus on the Unix way: components with minimal coupling that are responsible for distinct entities. API methods perform the simplest possible actions with these entities.

Messaging—as the name suggests—is a message broker. Its primary goal is to receive and deliver messages. It manages the interfaces for sending information, forming logical communication channels within the system, routing and load balancing, as well as handling failures at the system level.
The messaging being developed does not aim to compete with rabbitmq or replace it. Its main features are:

Distribution.
Exchange points can be created on all nodes of the cluster, as close to the code that uses them as possible.
Simplicity.
Focus on minimizing boilerplate code and enhancing usability.
Better performance.
We do not attempt to replicate rabbitmq's functionality; instead, we isolate only the architectural and transport layer, integrating it as simply as possible into OTP to minimize overhead.
Flexibility.
Each service can combine multiple messaging patterns.
Fault tolerance built into the design.
Scalability.
Messaging evolves alongside the application. As the load increases, messaging points can be moved to separate machines.

Note. From a code organization perspective, meta-projects are well suited for complex systems built on Erlang/Elixir. All project code resides in a single repository – an umbrella project. In this setup, microservices are maximally isolated and perform simple operations, each responsible for a distinct entity. This approach facilitates easy API maintenance across the entire system, makes modifications straightforward, and simplifies the writing of unit and integration tests.

The system components interact directly or through a broker. From a messaging standpoint, each service has several life phases:

Service initialization.
At this stage, the configuration and launch of the executing service process and its dependencies occur.
Creating an exchange.
The service can either use a static exchange point defined in the node's configuration or create exchanges dynamically.
Service registration.
To handle requests, the service must be registered at the exchange point.
Normal operation.
The service performs useful work.
Termination.
There are two types of termination: graceful and abrupt. In graceful termination, the service disconnects from the exchange and stops. In case of emergencies, the messaging system follows one of the failure-handling scenarios.

It looks quite complex, but the code is not that daunting. Code examples with comments will be provided in the templates analysis shortly.

Exchanges

An exchange is a messaging process that implements the interaction logic with components within the messaging pattern. In all the examples presented below, components interact through exchanges, the combination of which forms the messaging system.

Message exchange patterns (MEPs)

Globally, exchange patterns can be divided into bidirectional and unidirectional. The former involves a response to the incoming message, while the latter does not. A classic example of a bidirectional pattern in a client-server architecture is the Request-response pattern. Let's examine this pattern and its variants.

Request–response or RPC

RPC is used when we need to get a response from another process. This process can be running on the same node or located on another continent. Below is a diagram illustrating the interaction between the client and server through messaging.

Since messaging is fully asynchronous, the exchange for the client is divided into two phases:

Sending a request
```
messaging:request(Exchange, ResponseMatchingTag, RequestDefinition, HandlerProcess).
```
Exchange ‒ a unique name for the exchange point
ResponseMatchingTag ‒ a local tag for handling the response. For example, when sending multiple identical requests belonging to different users.
RequestDefinition ‒ the body of the request
HandlerProcess ‒ the handler's PID. This process will receive a response from the server.

Processing the response

handle_info(#'$msg'{exchange = EXCHANGE, tag = ResponseMatchingTag, message = ResponsePayload}, State)

ResponsePayload ‒ server response.

For the server, the process also consists of 2 phases:

Initialization of the exchange point
Processing incoming requests

Let's illustrate this template with code. Suppose we need to implement a simple service that provides a single method for getting the exact time.

Server code

Let's move the service API definition to api.hrl:

%% =====================================================
%%  entities
%% =====================================================
-record(time, {
  unixtime :: non_neg_integer(),
  datetime :: binary()
}).

-record(time_error, {
  code :: non_neg_integer(),
  error :: term()
}).

%% =====================================================
%%  methods
%% =====================================================
-record(time_req, {
  opts :: term()
}).
-record(time_resp, {
  result :: #time{} | #time_error{}
}).

Define the service controller in time_controller.erl

%% The example shows only the significant code. By inserting it into the gen_server template, a working service can be obtained.

%% gen_server initialization
init(Args) ->
  %% connection to the exchange point
  messaging:monitor_exchange(req_resp, ?EXCHANGE, default, self())
  {ok, #{}}.

%% handling the disconnection event from the exchange point. This event also comes if the exchange point hasn't started yet.
handle_info(#exchange_die{exchange = ?EXCHANGE}, State) ->
  erlang:send(self(), monitor_exchange),
  {noreply, State};

%% handling the API
handle_info(#time_req{opts = _Opts}, State) ->
  messaging:response_once(Client, #time_resp{
result = #time{unixtime = time_utils:unixtime(now()), datetime = time_utils:iso8601_fmt(now())}
  });
  {noreply, State};

%% terminating the gen_server
terminate(_Reason, _State) ->
  messaging:demonitor_exchange(req_resp, ?EXCHANGE, default, self()),
  ok.

Client code

To send a request to the service, anywhere in the client you can call the messaging request API:

case messaging:request(?EXCHANGE, tag, #time_req{opts = #{}}, self()) of
    ok -> ok;
    _ -> %% repeat or fail logic
end

In a distributed system, the configuration of components can vary greatly, and at the time of the request, the messaging may not yet be running, or the service controller may not be ready to handle the request. Therefore, we need to check the response from messaging and handle the failure case.
After a successful send, a response or error will come back to the client from the service.
We will handle both cases in handle_info:

handle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time{unixtime = Utime}}}, State) ->
  ?debugVal(Utime),
  {noreply, State};

handle_info(#'$msg'{exchange = ?EXCHANGE, tag = tag, message = #time_resp{result = #time_error{code = ErrorCode}}}, State) ->
  ?debugVal({error, ErrorCode}),
  {noreply, State};

Request-Chunked Response

It's better to avoid sending huge messages. This affects the responsiveness and stability of the entire system. If the response to a request takes up a lot of memory, splitting it into parts is mandatory.

Here are a couple of examples of such cases:

Components exchange binary data, such as files. Breaking down the response into smaller parts helps efficiently work with files of any size and prevents memory overflow.
Listings. For example, we need to select all records from a large table in the database and send them to another component.

I call such responses a train. In any case, 1024 messages of 1 MB are better than a single message of 1 GB.

In an Erlang cluster, we gain additional benefits—reducing the load on the exchange point and the network, as responses are immediately directed to the recipient, bypassing the exchange point.

Response with Request

This is a fairly rare modification of the RPC pattern for building dialog systems.

Event-driven systems deliver data to consumers as it becomes available. Thus, the systems are more inclined toward a push model rather than pull or poll. This feature prevents wasting resources by constantly querying and waiting for data.
The diagram shows the process of distributing messages to consumers subscribed to a particular topic.

Classic examples of using this pattern include the distribution of state: the game world in computer games, market data on exchanges, useful information in data feeds.

Let's look at the subscriber code:

init(_Args) ->
  %% subscribe to the exchange, key = key
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {ok, #{}}.

handle_info(#exchange_die{exchange = ?SUBSCRIPTION}, State) ->
  %% if the exchange point is unavailable, we try to reconnect
  messaging:subscribe(?SUBSCRIPTION, key, tag, self()),
  {noreply, State};

%% handle incoming messages
handle_info(#'$msg'{exchange = ?SUBSCRIPTION, message = Msg}, State) ->
  ?debugVal(Msg),
  {noreply, State};

%% when the consumer stops - unsubscribe from the exchange
terminate(_Reason, _State) ->
  messaging:unsubscribe(?SUBSCRIPTION, key, tag, self()),
  ok.

The source can trigger the message publishing function at any convenient location:

messaging:publish_message(Exchange, Key, Message).

Exchange ‒ exchange point name,
Key ‒ routing key
Message ‒ payload

By implementing pub-sub, you can achieve a pattern convenient for logging. The set of sources and consumers can vary greatly. The diagram shows a case with one consumer and multiple sources.

Task distribution pattern

In almost every project, there are tasks for deferred processing, such as report generation, notification delivery, and data retrieval from external systems. The throughput of the system carrying out these tasks can be easily scaled by adding handlers. All we need to do is create a cluster of handlers and distribute tasks evenly among them.

Let's consider the emerging situations with 3 handlers as an example. Even at the task distribution stage, the issue of equitable distribution and handler overflow arises. Round-robin distribution will ensure fairness, and to prevent handler overflow, we will introduce a limitation prefetch_limit. In transitional modes prefetch_limit it will prevent a single handler from receiving all tasks.

Messaging manages queues and processing priority. Handlers receive tasks as they arrive. The task execution may end successfully or with a failure:

messaging:ack(Tack) ‒ is called in the case of successful message processing
messaging:nack(Tack) ‒ is called in all exceptional situations. After the task is returned, messaging will pass it to another handler.

Let's assume that during the processing of three tasks there was a complex failure: handler 1 crashed after receiving the task without having time to notify the exchange point. In this case, the exchange point will reassign the task to another handler after the ack timeout expires. Handler 3, for some reason, rejected the task and sent a nack, resulting in the task also being passed to another handler that successfully completed it.

Preliminary Summary

We have discussed the basic building blocks of distributed systems and gained a fundamental understanding of their application in Erlang/Elixir.

By combining basic templates, complex paradigms can be built to address emerging tasks.

In the final part of the series, we will discuss general issues related to service organization, routing, and load balancing, as well as the practical aspects of scalability and system resilience.

End of the second part.

Photo Marius Christensen
Illustrations prepared using websequencediagrams.com

Source: habr.com