Distributed computing is a rich topic, and we cannot pretend to address it entirely in a single chapter. This short section gives a brief and incomplete description of distributed computing. We will try to give you an overview of the paradigm and point to some of the main benefits and challenges of distributed systems.

The idea behind distributed computing is to design a system involving several components, which runs on different machines and communicates with each other (for example, over a network) to achieve a task or provide a service. A distributed system can involve components of different natures, each component providing a specific service and participating in the realization of the task. For example, a web server can be deployed to receive HTTP requests. To service a request, the web server may communicate over the network to query an authentication service to validate credentials and a database server in order to store and retrieve data and complete the request. Together, the web server, the authentication service, and the database form a distributed system.

A distributed system can also involve several instances of the same component. These instances form a cluster of nodes, and they can be used to divide the work among them. This topology allows a system to scale out and support a higher load by adding more instances to the cluster. As an example, if a web server is able to handle 20,000 requests per second, it may be possible to run a cluster of three identical servers to handle 60,000 requests per second (assuming that your architecture allows your application to scale linearly). Distributed clusters also help achieve high availability. If one of the nodes crashes, the others are still up and able to fulfill requests while the crashed instance is restarted or recovers. As there is no single-point of failure, there is no interruption of service.

For all their benefits, distributed systems come with their drawbacks and challenges. The communication between components is subject to failure and network disruptions. The application needs to implement a retry mechanism and error handling, and then deal with lost messages. Another challenge is managing shared state. For example, if all the nodes use a single database server to save and retrieve information, the database has to implement some form of a locking mechanism to ensure that concurrent modifications do not collide. It is also possible that once the cluster node count grows sufficiently large, the database will not be able to serve them all efficiently and will become a bottleneck.

Now that you have been briefly introduced to distributed systems, we will go back to MVT. The team has decided to turn the automated trader into a distributed application to be able to scale the platform. You have been tasked with the design of the system. Time to go to the whiteboard.

Your first strategy is simple. You plan to deploy several instances of the automated trader to form a cluster of nodes. These nodes can share the work and handle each part of the incoming orders. A load balancer in front of the cluster can distribute the load evenly among the nodes. This new architecture helps scale out the automated trader. However, you are facing a common problem with distributed systems: the nodes have to share a common state to operate. To understand this requirement, we explore one of the features of the automated trader. To be able to use MVT's automated trading system, customers have to open an account with MVT and provision it with enough money to cover their trades. This is used as a safety net by MVT to execute orders on behalf of its clients without running the risk of a customer being unable to honor their transactions. To ensure that the automated strategies do not overspend, the automated trader keeps track of the current balance of each customer and checks the balance of a customer before placing an automated order on their behalf.

Your plan consists of deploying several instances of the automated trading system. Each instance receives a portion of the orders processed by the order book, runs a strategy and places matching order on behalf of a customer. Now that the system consists of several identical instances running in parallel, each instance can place orders on behalf of the same customer. To be able to perform the balance validation, they all need to be aware of the current balance of all customers. Customer balances become a shared state that has to be synchronized in the cluster. To solve this problem, you envision a balance monitor server deployed as an independent component and holding the state of each customer's balance. When a trade order is received by a node of the automated trading cluster, the node interrogates the balance monitor server to verify that a customer's account has enough funds to place an automated trade. Similarly, when a trade is executed, a node instructs the balance monitor server to update the balance of the customer.

The preceding diagram describes various interactions between the components of your architecture. Automated Trader 1 receives an incoming trade and queries the balance monitor server to check whether the client has enough funds to perform a trade. The balance monitor server either authorizes or rejects the order. At the same time, Automated Trader 3 sends an order that was previously approved by the balance monitor server and updates the client's balance.

At the end of this design session, you take a short walk to clear your mind while drinking a bottle of carbonated water. As you return to the whiteboard, the crude reality hits you. Like a flat bottle of carbonated water, your idea has fizzled out. You realize that all these arrows linking rectangles are in reality messages that are traveling over the network. Currently, while a single automated trader relies on its internal state to execute strategies and place orders, this new design requires the automated trader to query an external system over the network and wait for the answer. This query happens on the critical path. This is another common issue with distributed systems: components with focused roles need to communicate with each other to accomplish their tasks. This communication comes at a cost. It involves serialization, I/O operations, and transfer over a network. You share your reflections with Alice, who confirms that this is a problem. The automated trader has to keep the internal latency as low as possible for its decisions to be relevant. After a short discussion, you agree that it would endanger performance for the automated trader to perform a remote call on the critical path. You are now left with the task of implementing a distributed system with components sharing a common state without communicating with each other on the critical path. This is where we can start talking about CRDTs.

CRDT stands for Conflict-free Replicated Data Types. CRDTs were formally defined by Marc Shapiro and Nuno Preguiça in their paper, Designing a commutative replicated data type (refer to https://hal.inria.fr/inria-00177693/document). A CRDT is a data structure that is specifically designed to ensure eventual consistency across multiple components without the need for synchronization. Eventual consistency is a well-known concept in distributed system, which is not exclusive to CRDTs. This model guarantees that eventually, if a piece of data is no longer modified, all nodes in a cluster will end up with the same value for this piece of data. Nodes send each other update notifications to keep their state synchronized. The difference with strong consistency is that at a given time, some nodes may see a slightly outdated state until they receive the update notice:

The preceding diagram shows an example of eventual consistency. All the nodes of the cluster hold the same piece of data (A = 0). Node 1 receives an update to set the value of A to 1. After updating its internal state, it broadcasts the update to the rest of the cluster. The messages reach their targets at different instants, which means that until we reach step 4, A has a different value depending on the node. If a client queries node 4 for the value of A at step 3, they receive an older value as the change has not yet been reflected in node 4.

A problem that may arise with eventual consistency is the resolution of conflicts. Imagine a simple example where nodes in a cluster share the state of an array of integers. The following table describes a sequence of events involving updating the state of this array:

Our cluster now needs to resolve the conflict. Should node 1 update its state when receiving the update from node 2? If node 2 does the same, we end up with two nodes holding a different state. What about the other nodes? Some may receive the broadcast from node 2 before the one from node 1 and vice versa.

Various strategies exist to deal with this problem. Some protocols use timestamps or vector clocks to determine which update was performed later in time and should take precedence. Others simply assume that the last writer wins. This is not a simple problem and CRDTs are designed to completely avoid conflicts altogether. Actually, CRDTs are defined to make conflicts mathematically impossible. To be defined as a CRDT, a data structure has to support only commutative updates. That is, regardless of the ordering in which the update operations are applied, the end state must always be the same. This is the secret of eventual consistency without merge conflict. When a system uses CRDTs, all the nodes can send each other update messages without a need for strict synchronization. The messages can be received in any order, and all the local states will converge to the same value eventually.

In the preceding diagram, we see that node 3 and node 1 receive two different changes. They send this update information to all the other nodes. Note that we are not concerned with the order in which the updates are received by the other nodes. As the updates are commutative, their order has no impact on the final state that will be computed by each node. They are guaranteed to hold the same piece of data once all of them have received all the update broadcasts.

There exist two types of CRDT:

They are equivalent in that it is always possible to define a state-based CRDT for each operation-based CRDT and vice-versa. However, their implementations differ and provide different guarantees in terms of error-recovery and performance. We define each type and consider its characteristics. As an example, we implement each version of the simplest CRDT: an increase-only counter.

case class CounterUpdate(i: Int) 
case class GCounterState(uid: Int, counter: Int) 
 
class StateBasedGCounter( 
 uid: Int, count: Int, otherCounters: Map[Int, Int]) { 
 
 def value: Int = count + otherCounters.values.sum 
 
 def update( 
   change: CounterUpdate): (StateBasedGCounter, GCounterState) = 
   (new StateBasedGCounter(uid, count + change.i, otherCounters), 
     GCounterState(uid, count)) 
 
 def merge(other: GCounterState): StateBasedGCounter = { 
   val newValue = other.counter max otherCounters.getOrElse(other.uid,0) 
   new StateBasedGCounter(uid, count, otherCounters.+(other.uid -> newValue) ) 
 } 
} 

class OperationBasedCounter(count: Int) { 
 
 def value: Int = count 
 
 def update(change: CounterUpdate): (OperationBasedCounter, CounterUpdate) 
 = 
   new OperationBasedCounter(count + change.i) -> change 
 
 def merge(operation: CounterUpdate): OperationBasedCounter = 
   update(operation)._1 
} 

Based on the requirements of our system, it seems that CRDTs are a good fit for our implementation. Each node can keep the current state of each customer's balance in memory as a counter, update it when placing orders, and broadcast update messages to the rest of the system. This broadcast can be done outside the critical path, and we do not have to worry about handling conflicts, as this is what CRDTs are designed for. Eventually, all nodes will have in memory the same value for each balance, and they will be able to locally check for trade authorization. The balance monitor server can be removed entirely.

To implement the state of the balance as a CRDT, we need a more sophisticated counter than the one we previously explored. The balance cannot be represented as an increase-only counter because, occasionally, orders are canceled and the system must credit the customer's account. The counter has to be able to handle both increment and decrement operations. Luckily, such a counter exists. Let's look at a simple implementation of a state-based counter:

case class PNCounterState(incState: GCounterState, decState: GCounterState) 
 
class StateBasedPNCounter private( 
 incCounter: StateBasedGCounter, 
 decCounter: StateBasedGCounter) { 
 
 def value = incCounter.value - decCounter.value 
 
 def update(change: CounterUpdate): (StateBasedPNCounter, PNCounterState) = { 
   val (newIncCounter, newDecCounter, stateUpdate) = 
     change match { 
       case CounterUpdate(c) if c >= 0 => 
         val (iC, iState) = incCounter.update(change) 
         val dState = GCounterState(decCounter.uid, decCounter.value) 
         (iC, decCounter, PNCounterState(iState, dState)) 
       case CounterUpdate(c) if c < 0 => 
         val (dC, dState) = decCounter.update(change) 
         val iState = GCounterState(incCounter.uid, incCounter.value) 
         (incCounter, dC, PNCounterState(iState, dState)) 
     } 
 
   (new StateBasedPNCounter(newIncCounter, newDecCounter), stateUpdate) 
 } 
 
 def merge(other: PNCounterState): StateBasedPNCounter = 
   new StateBasedPNCounter( 
     incCounter.merge(other.incState), 
     decCounter.merge(other.decState) 
   ) 
} 

The PN counter leverages our previous implementation of an increase-only counter to provide the decrement capability. To be able to represent a counter as a state-based CRDT, we need to keep track of the state of both increment and decrement operations. This is necessary to guarantee that we do not lose information if our update messages are received in the wrong order by other nodes.

This implementation shows you another interesting property of CRDTs: simple and basic structures can be composed to create more complex and feature-rich CRDTs. Should we proceed to demonstrate the implementation of an operation-based counter? As it turns out and we are sure you spotted this earlier, our previous increase-only counter already supports decrement operations. Applying a positive or a negative delta is handled by the operation-based counter.

You have finished the implementation of the proof-of-concept and call Alice to get some feedback. She spends a few minutes studying your new design and your code. "Looks good to me. Do not forget to synchronize the account blacklist as well." What is she talking about? "Checking the account balance is only one of the criteria to allow or block an automated trade. Other attributes of the client need to be taken into consideration. Today, the automated trader runs a trust algorithm in the background, and it calculates a score for each customer. If the score falls below a certain threshold, the account is blacklisted until the end of the trading day, and all automated orders are denied. I like your design, but you need to incorporate this blacklist into the new system." Faced with this new challenge, you think that the best solution would be to implement the blacklist as a CRDT as well, provided that it fits your current design.

case class AddElement[A](a: A) 
case class GSetState[A](set: Set[A]) 
 
class StateBasedGSet[A](val value: Set[A]) { 
 
 def contains(a: A): Boolean = value.contains(a) 
 
 def update(a: AddElement[A]): (StateBasedGSet[A], GSetState[A]) = { 
   val newSet = new StateBasedGSet(value + a.a) 
   (newSet, GSetState(newSet.value)) 
 } 
 
 def merge(other: GSetState[A]): StateBasedGSet[A] = { 
   new StateBasedGSet(value ++ other.set) 
 } 
 
}