🥇Building blocks of distributed applications. Second approximation

Announcement

Colleagues, I plan to release another series of articles on the design of mass service systems in mid-summer: "Experiment VTrade" — an attempt to write a framework for trading systems. The series will cover the theory and practice of building an exchange, auction, and store. At the end of the article, I invite you to vote for the topics that interest you the most.

This is the concluding article in the series on distributed reactive applications in Erlang/Elixir. In first article) you can find the theoretical foundations of reactive architecture. The second article illustrates the main patterns and mechanisms for building such systems.

Today, we will raise questions about the development of the codebase and the projects as a whole.

Service Organization

In real life, when developing a service, it is often necessary to combine several interaction patterns in one controller. For example, the users service, which handles user profile management tasks for the project, should respond to req-resp requests and notify about profile updates via pub-sub. This case is quite simple: there is one controller behind messaging, implementing the service logic and publishing updates.

The situation becomes more complicated when we need to implement a fault-tolerant distributed service. Let’s imagine that the requirements for users have changed:

now the service must handle requests on 5 nodes in the cluster,
be capable of performing background processing tasks,
and also be able to dynamically manage subscription lists for profile updates.

Note: We are not addressing the issue of consistent data storage and replication. Let’s assume that these issues were resolved earlier, and there already exists a reliable and scalable storage layer in the system, and handlers have mechanisms for interacting with it.

The formal description of the users service has become more complicated. From a programmer’s perspective, due to the use of messaging, changes are minimal. To meet the first requirement, we need to set up load balancing at the req-resp exchange point.

The need for processing background tasks often arises. In users, this may include verifying user documents, processing uploaded multimedia, or syncing data with social networks. These tasks need to be distributed within the cluster and monitored for progress. Therefore, we have two options: either use the task distribution template from the previous article, or, if it doesn't fit, write a custom task scheduler that will manage the pool of handlers in the necessary way.

Point 3 requires extending the pub-sub template. To implement this, after creating the pub-sub exchange point, we need to additionally launch the controller of this point within our service. Thus, we essentially extract the logic of subscription and unsubscription handling from the messaging layer to the implementation in users.

As a result, the decomposition of the task showed that in order to meet the requirements, we need to start 5 instances of the service on different nodes and create an additional entity – a pub-sub controller responsible for subscriptions.
Launching 5 handlers does not require changing the service code. The only additional action is configuring the load balancing rules at the exchange point, which we will discuss shortly.
Additionally, there is a further complication: the pub-sub controller and the custom task scheduler must operate as a single instance. Again, the messaging service, being fundamental, must provide a leader selection mechanism.

Leader Selection

In distributed systems, leader selection is the procedure for appointing a single process responsible for scheduling distributed processing of some load.

In systems resistant to centralization, universal algorithms and consensus-based algorithms, such as Paxos or Raft, are applied.
Since messaging acts as a broker and central element, it is aware of all the service controllers – candidates for leadership. Messaging can appoint a leader without conducting a vote.

All services receive a system message after starting up and connecting to the exchange point #'$leader'{exchange = ?EXCHANGE, pid = LeaderPid, servers = Servers}. In the event that LeaderPid matches the pid current process, it is appointed as the leader, and the list Servers includes all nodes and their parameters.
When a new node is added and an active node is disabled in the cluster, all service controllers receive #'$slave_up'{exchange = ?EXCHANGE, pid = SlavePid, options = SlaveOpts} and #'$slave_down'{exchange = ?EXCHANGE, pid = SlavePid, options = SlaveOpts} respectively.

As a result, all components are aware of all changes, and there is guaranteed to be one leader in the cluster at all times.

Mediators

To implement complex distributed processing tasks and optimize existing architecture, it is convenient to use mediators.
To avoid changing the service code and to handle additional processing, routing, or logging of messages, a proxy handler can be introduced in front of the service to manage all additional tasks.

A classic example of pub-sub optimization is a distributed application with a business core generating update events, such as a market price change, and an access layer — N servers providing a websocket API for web clients.
If handled directly, client servicing would look like this:

the client establishes connections with the platform. On the server side, where traffic is terminated, a process is initiated to service this connection.
within the servicing process, authorization and subscription to updates occur. The process calls the subscribe method for the topics.
after the event is generated in the core, it is delivered to the processes servicing the connections.

Let’s say we have 50,000 subscribers for the topic 'news'. Subscribers are evenly distributed across 5 servers. As a result, each update arriving at the exchange point would be replicated 50,000 times: 10,000 times on each server, according to the number of subscribers on it. Not a very efficient scheme, is it?
To improve the situation, we introduce a proxy that has the same name as the exchange point. The global name registry should be able to return the nearest process by name; this is important.

We will run this proxy on the access layer servers, and all our websocket API servicing processes will subscribe to it instead of the original pub-sub exchange point in the core. The proxy subscribes to the core only in the case of unique subscriptions and replicates the incoming message to all its subscribers.
As a result, there will be 5 messages sent between the core and access servers instead of 50,000.

Routing and balancing

Req-Resp

In the current implementation of messaging, there are 7 request distribution strategies:

defaultThe request is sent to all controllers.
round-robinRequests are iterated and distributed cyclically among controllers.
consensusControllers servicing the service are divided into a leader and followers. Requests are sent only to the leader.
consensus & round-robinThere is a leader in the group, but requests are distributed among all members.
stickyA hash function is computed and assigned to a specific handler. Subsequent requests with this signature go to the same handler.
sticky-funDuring the initialization of the exchange point, a hash calculation function is additionally passed for sticky balancing.
funSimilar to sticky-fun, but it additionally allows redirecting, rejecting, or preprocessing.

The distribution strategy is set at the initialization of the exchange point.

In addition to balancing, messaging allows tagging entities. Let's consider the types of tags in the system:

Connection tag. This allows us to understand through which connection events came. It is used when the controller process connects to one exchange point but with different routing keys.
Service tag. This allows grouping handlers for one service and expanding routing and balancing capabilities. For the req-resp pattern, routing is linear. We send a request to the exchange point, and it then forwards it to the service. But if we need to split handlers into logical groups, the division is done using tags. When specifying a tag, the request will be directed to a specific group of controllers.
Request tag. This allows distinguishing responses. Since our system is asynchronous, to process responses from the service, it's necessary to specify a RequestTag when sending a request. This way, we can understand which request's response has arrived.

Pub-sub

For pub-sub, it's a bit simpler. We have an exchange point to which messages are published. The exchange point distributes messages among subscribers, who have subscribed to the routing keys they need (one could say this is similar to topics).

Scalability and fault tolerance

The overall scalability of the system depends on the degree of scalability of its layers and components:

Services scale by adding additional nodes with handlers of this service to the cluster. During practical operation, one can choose the optimal load balancing policy.
The messaging service within a separate cluster generally scales either by offloading particularly busy exchange points to separate nodes in the cluster or by adding proxy processes to high-load areas of the cluster.
The scalability of the entire system as a characteristic depends on the flexibility of the architecture and the ability to combine separate clusters into a single logical entity.

The simplicity and speed of scaling often determine the success of a project. Messaging in its current implementation grows alongside the application. Even if we lack a cluster of 50-60 machines, we can resort to federation. Unfortunately, the topic of federation is beyond the scope of this article.

Failover

In our discussion of load balancing, we already talked about service controller redundancy. However, messaging must also be redundant. In the event of a node or machine failure, messaging must automatically recover, and in the shortest possible time.

In my projects, I use additional nodes that pick up the load in case of a failure. Erlang has a standard implementation of distributed mode for OTP applications. The Distributed mode handles recovery in case of failure by starting the crashed application on another pre-launched node. The process is transparent; after a failure, the application automatically moves to the failover node. More information about this functionality can be read. here.

Performance

Let's at least roughly compare the performance of RabbitMQ and our custom messaging.
I found official results of RabbitMQ testing from the OpenStack team.

In section 6.14.1.2.1.2.2 of the original document, the result of RPC CAST is presented:

Preliminary, we will not make any additional settings to the OS kernel or Erlang VM. Testing conditions:

erl opts: +A1 +sbtu.
The test within a single Erlang node runs on a laptop with an old mobile i7.
Cluster tests are conducted on servers with a 10G network.
The code runs in Docker containers. The network is in NAT mode.

Test code:

req_resp_bench(_) ->
  W = perftest:comprehensive(10000,
    fun() ->
      messaging:request(?EXCHANGE, default, ping, self()),
      receive
        #'$msg'{message = pong} -> ok
      after 5000 ->
        throw(timeout)
      end
    end
  ),
  true = lists:any(fun(E) -> E >= 30000 end, W),
  ok.

Scenario 1: The test is run on a laptop with an old mobile i7 processor. The test, messaging, and service are running on the same node in a single docker container:

Sequential 10000 cycles in ~0 seconds (26987 cycles/s)
Sequential 20000 cycles in ~1 seconds (26915 cycles/s)
Sequential 100000 cycles in ~4 seconds (26957 cycles/s)
Parallel 2 100000 cycles in ~2 seconds (44240 cycles/s)
Parallel 4 100000 cycles in ~2 seconds (53459 cycles/s)
Parallel 10 100000 cycles in ~2 seconds (52283 cycles/s)
Parallel 100 100000 cycles in ~3 seconds (49317 cycles/s)

Scenario 2: 3 nodes running on different machines under docker (NAT).

Sequential 10000 cycles in ~1 seconds (8684 cycles/s)
Sequential 20000 cycles in ~2 seconds (8424 cycles/s)
Sequential 100000 cycles in ~12 seconds (8655 cycles/s)
Parallel 2 100000 cycles in ~7 seconds (15160 cycles/s)
Parallel 4 100000 cycles in ~5 seconds (19133 cycles/s)
Parallel 10 100000 cycles in ~4 seconds (24399 cycles/s)
Parallel 100 100000 cycles in ~3 seconds (34517 cycles/s)

In all cases, CPU utilization did not exceed 250%

Summary

I hope this cycle doesn't come off as a stream of consciousness and that my experience will genuinely benefit both researchers of distributed systems and practitioners just starting to build distributed architectures for their business systems, who are keenly looking at Erlang/Elixir but unsure if it's worth it...

Photo @chuttersnap

Only registered users can participate in the survey. Please log in, please.

What topics should I cover in greater detail within the 'VTrade Experiment' cycle?

Theory: Markets, orders, and their timeframes: DAY, GTD, GTC, IOC, FOK, MOO, MOC, LOO, LOC
Order book. Theory and practice of implementing the book with groupings
Trading visualization: Ticks, bars, resolutions. How to store and how to correlate
Back office. Planning and development. Staff monitoring and incident investigation
API. Let's discuss what interfaces are needed and how to implement them
Data storage: PostgreSQL, Timescale, Tarantool in trading systems
Reactivity in trading systems
Other. I will write in the comments