Modeling Distributed Systems

1. Modeling a Distributed System#

When we model a distributed system, we describe it from several complementary viewpoints, just as we model anything by focusing on different aspects. Four viewpoints, taken together, characterise a distributed system:

1. Software architecture: how the software is organised, in particular whether it is built directly on the network operating system or on top of middleware.
2. Runtime architecture: the dynamic structure of the running system, captured by an architectural style.
3. Interaction model: how the processes interact through message passing, and how process speed, communication delay, and clock drift affect the outcome.
4. Failure model: the ways in which components can fail, and how the system tolerates them.

This chapter follows that plan: the software architecture in this section, the runtime architectural styles in sections 2 to 10, the interaction model in section 11, and the failure model in sections 12 and 13.

1.1 Software vs Runtime Architecture#

The software architecture describes the system statically, that is, the structure of its code. This is the notion of software architecture studied in software engineering.

The runtime architecture describes the system dynamically, and it is captured by the notion of architectural style.

We will cover the most common and significant styles, not every possible one.

1.2 Software Architecture: Network-OS-Based vs Middleware-Based#

The software of a distributed system can be organised in two ways, which differ in who masks the platform differences:

• Network-OS-based: the system is built directly on the basic communication facilities offered by every operating system, typically TCP/IP. Different machines may run different network operating systems, and masking those platform differences is left to the application programmer.
• Middleware-based: a middleware layer provides advanced communication, coordination, and administration services, and masks most of the platform differences itself.

Throughout this course we reason at the level of the software architecture: we study the protocols, techniques, and approaches used to build middleware, so that in practice one can simply buy a middleware and use it directly. For now, though, we work directly on top of TCP/IP, because this is the minimal layer on which all the problems must be solved, whether by us as application programmers or by the programmers of the middleware services that are then sold to other application programmers.

1.3 Middleware: A Functional View#

Middleware provides “business-unaware” (general, non-application-specific) services through a standard API, raising the level at which applications communicate and masking most, if not all, of the differences at the operating-system layer and sometimes even at the language layer. The main categories of service are the following.

Communication services. These let distributed components communicate at a higher level of abstraction than raw networking. Sockets are arguably a very thin communication middleware: a small layer, part of the operating system, built on top of TCP or UDP, offering connection-based (TCP) or connectionless (UDP) communication. Higher-level middleware is far easier to use. RMI (Remote Method Invocation), for example, lets an object on one machine invoke a method on an object running on another machine thousands of kilometres away, exactly as if the call were local, with no opening of a connection, encoding of the data into a message, and decoding on the other side. Communication can be synchronous or asynchronous, point-to-point or multicast. Crucially, such middleware can also mask language differences: an object written in Java can invoke a method on an object that is, transparently, implemented in C++, Python, or any other language, and on a different operating system or processor architecture (for example x86 invoking ARM).

Coordination services. These help components coordinate. Examples include ordering of messages and acquiring a distributed lock: something similar to a mutex, but working across machines even though there is no single shared clock. Such a lock can be implemented, but it takes considerable effort and time.

Special application services. Higher-level, ready-made facilities such as distributed transaction management, groupware and workflow services, messaging services, and notification services.

Management services. Services such as naming, which gives a component a name that is independent of its location, so the component can be invoked by name even as it moves from one machine to another; security, to secure a message or open a secure connection between two machines (signing, encryption, compression); and failure handling.

We will see how to implement many of these services on top of the minimal layer offered by TCP/IP, so that in the future they can be bought on the market and used directly.

2. Client-Server Architectural Style#

The client-server style is the most widely used architectural style. It is characterised by:

• Two types of components: clients and servers.
• Servers are passive: they wait and respond to requests.
• Clients are active: they initiate requests and represent the interface to the user.
• Communication is either:
  • Message-based (e.g., TCP, UDP, HTTP connections), or
  • Remote Procedure Call (RPC).

A typical example is a web application: the client issues a request, the server responds, and the client displays the result to the user.

2.1 Multi-Tier Architectures#

Every application, even a centralised one, is composed of three types of services:

1. User Interface (UI) services: components the user interacts with.
2. Application-level services: components that handle processing logic.
3. Data storage and processing services: components that manage persistent data.

In a distributed client-server system, these services are split across tiers, and the split can be made at several points.

Two-Tier Architecture#

In a two-tier system there is one network boundary, between client and server, and the designer chooses where to cut the UI / application / data stack:

Examples:

• Traditional web (pre-2010s): all UI on the client, all logic and data on the server.
• Modern web applications: part of the application logic (typically JavaScript) runs on the client. Examples: Google Docs, Google Maps (which also caches map data locally on the client).

Three-Tier Architecture#

Real-world note: The university’s online services run the UI on the client, application servers on AWS, and the database on local infrastructure: a less obvious but economically motivated split.

Beyond Three Tiers#

More than three tiers are possible. For instance, stored procedures in a DBMS represent application logic running directly on the database machine, effectively adding another logical layer. In every case the structure stays the same: a single class of active components (the clients) and layers of passive components (application servers, databases), connected by message passing or RPC.

3. Service-Oriented Architecture (SOA)#

SOA is a specialisation of the client-server style built around two ideas: a service, a capability offered by one component and consumed by another through a formally described interface rather than a fixed implementation, and a service broker, an active registry that lets a consumer discover a suitable provider at runtime instead of being wired to one in advance. The service itself is a logical concept, not a machine: what is crucial is that it has an explicit description, written in a dedicated language, of its interface (name, operations, parameters), not of its implementation.

3.1 Roles and the Service Loop#

Three roles cooperate, as the diagram shows.

The interaction is a three-step loop: the provider publishes its interface description to the broker (1); the consumer finds a provider by querying the broker, which returns one or more matches (2); the consumer then binds and invokes a chosen provider directly, by message passing or RPC (3). Because the contract is the interface and not the implementation, the consumer can bind to any provider that satisfies it. A single component can play both roles at once, exposing a service to the outside while internally calling other services to implement it; composing services this way is called orchestration.

3.2 Web Services#

Web services are the most widely adopted realisation of SOA, standardised by the W3C as “a software system designed to support interoperable machine-to-machine interaction over a network.” Their three core standards line up neatly with the three steps of the loop:

A WSDL description resembles a class in an object-oriented language: it groups several related operations (for example addToBasket and removeFromBasket) under a single service.

3.3 Other Implementations#

Web services are not the only option. Other middleware frameworks implement the same SOA style, including OSGi, JAX-WS (Java API for XML Web Services), and Jini/Apache River.

4. REST: Representational State Transfer#

REST is a special, highly constrained case of the client-server architecture. It was conceived by Roy Thomas Fielding, one of the principal authors of HTTP, in his doctoral thesis, as a formal description of how the web should work to achieve its full potential.

REST is not a protocol or a technology, but a set of principles and constraints that, if followed, yield the best possible scalability, performance, and flexibility from a web-based system.

4.1 Goals of REST#

• Scalability of component interactions.
• Generality of interfaces, enabling independent deployment and substitution of components.
• Ease of introducing intermediary components (e.g., caches, security layers, legacy encapsulators) transparently to the rest of the system.

4.2 REST Constraints#

REST defines six architectural constraints. A system is only fully REST-compliant if it satisfies all of them.

Constraint 1: Client-Server#

All interactions must follow the client-server model: one component invokes a service, another provides it. This is shared with web services and client-server systems in general.

Constraint 2: Stateless Interactions#

This is one of the most important, and most demanding, constraints. Every interaction must be completely stateless: the server retains no session state between requests. Consequently, the same request must always return the same response.

Consider an e-commerce basket. A typical (non-REST) interaction: the client sends “add an iPhone to my basket”; the server replies with a basket containing one iPhone, keeping the basket as session state; if the client resends the same message, the server now replies with two iPhones. Same request, different response: the interaction is stateful.

Since most services are stateful at the application level, making the interactions stateless means that the state must travel with the messages, moving continuously between invoker and invoked:

1. Client sends: “Add an iPhone to this basket: [empty basket].”
2. Server replies with a basket containing one iPhone.
3. Client sends: “Add an iPhone to this basket: [basket with one iPhone].”
4. Server replies with a basket containing two iPhones.

Now resending the original request (iPhone + empty basket) always returns the same single-iPhone basket. Seen from above the overall service is still stateful; every single interaction, though, is stateless.

Why does statelessness matter so much? A stateful server is hard to replicate, because all replicas must keep their state synchronised. A stateless server is pure code with no state: requests can be served by any replica, in any order, without coordination, and if one replica crashes another takes over immediately with no state to recover. This is the primary motivation: statelessness makes horizontal scaling straightforward. It shifts complexity onto the programmers of client and server, but buys scalability for the system.

Constraint 3: Cacheable#

Responses must be explicitly or implicitly labelled as cacheable or non-cacheable. Stateless interactions are inherently cacheable by definition (same request, same response), so intermediate components can serve cached responses without forwarding requests to the origin server. This further enhances scalability.

Constraint 4: Layered System#

The system must be strongly layered: each component only interacts with the immediately adjacent layer and has no visibility beyond it. This enables the transparent introduction of intermediaries such as load balancers, caches, and security gateways.

Constraint 5: Uniform Interface#

This is the most distinctive REST constraint and the sharpest point of difference with web services.

In web services, each service provider defines its own interface via WSDL. Every service has a different, application-specific set of operations.

In REST, all components must expose the same, fixed, application-independent interface, defined by the HTTP verbs:

Any application-specific information (what to get, what to post, etc.) is encoded as parameters or in the URI, not in the verb itself.

Analogy: Imagine a Java version in which all methods must be named invoke, and specific behaviour is encoded as the first string parameter. It is more complex to write, but expressively complete: you move the expressiveness of the language into your code. And now any intermediary who intercepts a message knows exactly what kind of operation it is, without knowing anything about the application: a GET in a stateless system cannot change the destination’s state, so a caching layer can answer it from the cache; a security layer can encrypt a POST differently from a GET; and so on. Five verbs carry just enough application-independent information for intermediaries to act intelligently.

Additional requirements of the uniform interface (four sub-constraints):

• Resource identification via URI: every resource has a unique identifier (URI), and everything that has an identifier is a valid resource, including a service. Requests are made against resources.
• Manipulation through representations: there is a distinction between a logical resource and its representation, which consists of data plus metadata describing the data. For example, accessing https://www.repubblica.it retrieves the homepage resource, but the representation returned may differ depending on the client (a full HTML page for a desktop browser, a compact layout for a mobile device). Whether the representation coincides with the raw resource or is derived from it stays hidden behind the interface.
• Self-descriptive messages: each message carries enough control data to describe how it should be processed (the action requested, the meaning of a response, cache behaviour overrides).
• Hypermedia as the engine of application state (HATEOAS): the application state is driven by navigating links embedded in resource representations (see §4.3).

Constraint 6: Code on Demand (optional)#

REST was among the first architectural styles to introduce the concept of mobile code. Servers may include executable code in their responses (e.g., JavaScript embedded in a web page), which the client downloads and runs locally. This is the basis of modern dynamic web pages. Code on demand is the only optional REST constraint; we return to it in the mobile code style.

4.3 Hypermedia as the Engine of Application State (HATEOAS)#

A REST interaction proceeds as follows:

1. The client starts from a well-known entry URI (e.g., the homepage of a service).
2. The server returns a representation of that resource, which includes links to other related resources.
3. The client follows those links to navigate to further resources, changing the application state with each step.

The client never needs to construct or guess URIs: it discovers them at runtime from the responses. This is what “hypermedia is the engine of application state” means: navigation through linked resources is the state machine of the application. The state of the interaction is not kept by the server; it is encoded in which resource you are currently looking at.

Example: An e-commerce system. The client requests the entry resource. The server returns a list of available products (e.g., iPhones in stock). The response includes a link to the resource representing the act of adding an iPhone to the basket. The client follows that link; the server returns a new resource representing the updated basket, and so on.

4.4 REST vs. Web Services: Summary#

4.5 REST in Practice#

Most modern web applications do not follow REST perfectly, but the closer a system adheres to the REST constraints, the greater the scalability, flexibility, and maintainability it achieves. REST should be seen as an ideal to aim for, offering concrete engineering benefits at each constraint that is adopted.

5. Peer-to-Peer Architectural Style#

5.1 From Client-Server to Peer-to-Peer#

Following the progression from client-server → multi-tier → web services (orchestration), the distinction between active clients and passive servers gradually blurs. The logical endpoint of this trend is the peer-to-peer (P2P) style, where no distinction exists between clients and servers. Every node can both offer and consume services.

Historical note: The term “peer-to-peer” actually predates client-server thinking: early networking systems were inherently peer-to-peer, since any node could send and receive. The term re-emerged at the end of the 20th century largely driven by file-sharing applications (e.g., Napster, BitTorrent), which needed to avoid centralised servers for both legal and technical reasons.

5.2 Motivation#

• Scalability: A central server becomes a bottleneck and a single point of failure. Distributing the workload among peers removes this constraint.
• Utilisation of edge resources: As computing power and connectivity spread to the “leaves” of the network (end-user machines), it becomes wasteful to funnel everything through central servers.
• Resilience: No single node is indispensable; the system can continue operating even if many peers leave.

The resources shared through direct exchange between peers can be of many kinds: processing cycles (SETI@home), collaborative work (ICQ, Skype), storage space (Freenet), network bandwidth (ad hoc networking), and, most commonly, data (file sharing).

5.3 The Role of Centralised Coordination#

In practice, most P2P systems retain a partially centralised element for coordination, even if the core service is fully distributed:

The defining characteristic is that the primary service (file transfer, computation, calls) is handled by the peers, while the central node only coordinates.

6. Object-Oriented Architectural Style#

In the object-oriented style, the system is structured as a collection of objects, each encapsulating a data structure behind an interface (its API). Objects can live on different machines and interact through Remote Method Invocation (RMI), the distributed equivalent of a local method call: any object may invoke any other object’s methods, with no fixed active/passive (client/server) distinction. In this sense the style is a typical example of the peer-to-peer model: each object is potentially both caller and callee.

All the usual benefits of object orientation carry over: information hiding, encapsulation, and reuse. Encapsulation also lowers management complexity: because an object hides its internals behind its interface, the objects that make up a server can be moved at run time to balance load, and legacy components can be wrapped inside an object and integrated into a new application.

In practice, distributed object systems are often used to implement client-server applications, organised in layers; the underlying interaction model, though, remains symmetric.

7. Data-Centred Architectural Style#

7.1 Core Concept#

In a data-centred system, components do not communicate with each other directly. Instead, they interact exclusively through a shared, logical data repository at the centre. Components can:

• Write data to the repository (out).
• Read a copy of data from the repository (read).
• Remove data from the repository (in).

The repository is a logical concept, physically it may be distributed, and it is usually a passive component: communication with it typically happens through RPC, and access to it is synchronised.

7.2 The Linda Model and Tuple Spaces#

The Linda model is the formal foundation of data-centred architectures. It was proposed in the 1980s by Carriero and Gelernter, originally for parallel computation, and was later revitalised for distributed computing, with implementations such as IBM TSpaces, Sun JavaSpaces, and GigaSpaces. Its communication is characterised as persistent, implicit, content-based, and generative, giving a high degree of decoupling.

Data is represented as tuples (ordered sequences of typed values). The shared repository is called a tuple space.

The standard primitive operations are:

Pattern matching is used for read and in: a pattern (or template) specifies values for some fields and wildcards for others. If multiple tuples match, one is chosen non-deterministically. If no tuple matches, the calling process blocks until a matching tuple is inserted by another component.

Variants exist beyond the standard primitives:

• Non-blocking probes (rdp(p), inp(p)) return null immediately if no matching tuple exists, rather than blocking.
• Bulk primitives (e.g., rdg(p)) operate on all matching tuples at once.

Some of these non-standard primitives have non-trivial distributed implementations: e.g., if atomicity must be preserved, a probe requires a distributed transaction to be sure no matching tuple exists anywhere in a distributed tuple space.

Example: A tuple space contains ("Italy", 60), ("France", 68), ("Germany", 84). A read("France", ?) returns ("France", 68). A read("Vietnam", ?) blocks until some process inserts a tuple starting with "Vietnam".

7.3 The Tuple Space as Communication and Synchronisation Medium#

The tuple space serves two roles simultaneously:

• Communication medium: Components exchange data by inserting and retrieving tuples, without ever addressing each other directly.
• Synchronisation medium: A blocking read/in on a missing tuple suspends the caller until the tuple appears, providing a natural synchronisation point between concurrent processes.

7.4 Decoupling and the Printing Example#

Consider implementing a printing service:

With web services (tightly coupled):

1. The client discovers which printer is available (via a broker).
2. The client selects a specific printer and invokes it directly.
3. The client is blocked while the printer processes the job.
4. If the printer fails, the client must handle the error explicitly.

With a tuple space (loosely coupled):

1. The client inserts ("print", document) into the tuple space: no discovery, no addressing.
2. Any available printer performs in("print", ?), retrieves the job, and prints it.
3. If multiple printers exist, the one that executes in first wins automatically.
4. If no printer exists yet, the tuple waits in the space until one becomes available.
5. If a printer fails, another one can pick up the job seamlessly.

This illustrates the key property of data-centred systems: loose coupling in both space and time. The service broker of SOA was invented to partially decouple consumer from provider, but after discovery the consumer still invokes one specific provider and expects it to stay there; with a tuple space nobody is invoked directly, and the architecture can change at runtime (printers added or removed) without informing the clients.

7.5 Persistent Communication#

An important property of the tuple space is persistence: a tuple inserted into the space remains there until explicitly removed, regardless of whether any consumer is currently active. A component can insert a tuple and terminate; another component can retrieve it hours later. This is persistent, asynchronous communication.

7.6 Limitations and Implementation Challenges#

• Single point of failure / bottleneck: The tuple space is the central element of the architecture. If it is slow, centralised, or unavailable, the entire system suffers. Distributing and replicating the tuple space (storing and replicating tuples efficiently, routing queries efficiently) is a non-trivial engineering challenge.
• Only proactive: Components must actively poll or block on reads; the base model has no reactive notification mechanism (unlike event-based systems, see §8). Reactive behaviour must be simulated with an extra process performing a blocking read.
• Less control: The loose coupling that makes the system flexible also means less visibility into who is consuming what.

As a consequence, commercial implementations typically (a) provide only client access to a server holding the tuple space rather than a fully distributed, decentralised implementation, and (b) introduce reactive primitives such as notify, which registers a listener invoked whenever a matching tuple is written.

7.7 Modern Relevance: Message Queues#

Message queue systems such as Apache Kafka are modern implementations of the data-centred principle: components publish data to a queue (the shared repository); other components consume from it. There is no direct interaction between producers and consumers. Kafka is widely used in production distributed systems for exactly the decoupling and scalability benefits described above.

8. Event-Based Architectural Style#

8.1 Core Concept#

In an event-based system, components communicate through two primitives:

• Publish: A component sends a message (an event) without specifying a destination.
• Subscribe: A component declares its interest in receiving certain types of messages by registering a subscription pattern.

A middleware component, the message broker, intercepts published events, matches them against active subscriptions, and delivers each event to all matching subscribers. This is inherently multicast: a single published event may be delivered to many subscribers simultaneously.

8.2 Subscription Patterns#

Subscriptions express interest in events based on their content. For example:

Given a published event “fire alarm at the first floor”, subscribers A and B receive it: A because it is a fire event, B because it is on the first floor. C does not, since it wants fire training, not an alarm. Given “fire training at the first floor”, subscribers B and C receive it; A does not (with fire* read strictly as the topics “fire alarm” and “fire”, A subscribes to alarms, not training).

A single component may hold multiple subscriptions simultaneously and may act as both publisher and subscriber over its lifetime.

8.3 Key Properties#

8.4 Comparison: Data-Centred vs. Event-Based#

8.5 The Message Broker#

Behind the scenes, event-based systems require a message broker, a centralised (or distributed) component that:

• Stores active subscriptions.
• Receives published events.
• Matches events against subscriptions.
• Delivers matched events to the appropriate subscribers.

Like the tuple space in Linda, the message broker is a potential single point of failure and bottleneck. Distributing and replicating the broker is an important engineering challenge that will be addressed in later lectures (e.g., Apache Kafka as a distributed event broker).

8.6 Origin of the Term “Event-Based”#

These systems originated in monitoring and sensing contexts: physical sensors detect events in the real world (temperature readings, smoke detection, etc.) and publish them; monitoring systems subscribe to the relevant event types. The term “event” captures the idea that a message represents something that happened in the external world. However, the model is general and applicable to any domain.

9. Mobile Code Architectural Style#

9.1 Core Concept#

Mobile code is an architectural style based on the ability to relocate parts of an application’s code at runtime: only the code, or the code together with the state of the execution. Rather than always sending data to where the code resides (as in client-server), mobile code allows code to move, either toward the data or toward the computational resources needed to run it. The paradigms differ in the location, before and after the interaction, of the know-how (the code), the resources (the data), and the computation (the processing capability, including the execution state).

9.2 The Four Paradigms#

The lecture introduces the paradigms with a cake-making analogy: the recipe is the code (know-how), the ingredients are the data (resources), and the oven is the processing capability.

9.3 Remote Evaluation#

The client has the code but lacks the resources (data, processing power) to execute it locally. It ships the code to a remote site, which executes it on the client’s behalf.

Real-world examples:

• PostScript printing: PostScript is a full, Turing-complete programming language, not merely a page description. The client ships a PostScript program to the printer; the printer has all the necessary libraries and hardware. The program executes locally on the printer to produce the output.
• Serverless computing / FaaS (e.g., AWS Lambda, Google Cloud Functions): The client ships a function (code) to a cloud platform that provides the execution environment and infrastructure. The platform runs the function remotely. Google Colab can be seen as an extreme variant of this pattern (extreme because mobile code usually moves parts of an application, not the whole of it).

9.4 Code on Demand#

The client has the resources (data and processing power) but lacks the code to do something useful with them. It requests the code from a server and executes it locally.

Real-world examples:

• JavaScript on the web: This is the defining example of code on demand, and also one of REST’s optional constraints: part of the application is already running on the client, and the downloaded code enriches it with new functionality.
• Plugin/extension systems: Applications that download and install new components at runtime without restarting.

9.5 Mobile Agents#

A computation begins executing on one machine, then migrates, together with its accumulated state, to another machine and resumes from where it left off. This continues across multiple sites as needed.

Motivating example: Suppose you need to search several large, remote databases sequentially (because each query depends on the previous result). In a traditional client-server approach, each query requires a round trip over the network. With a mobile agent, you send the agent to the first database, it runs the query locally, collects the result, migrates to the second database with its state intact, runs the next query, and so on, returning at the end with all results. Network traffic is drastically reduced.

In practice: Mobile agents have remained largely theoretical. Several research prototypes (typically JVM-based) have demonstrated feasibility, but no widely adopted production system uses strong mobility. The implementation complexity and security risks outweigh the benefits in most real scenarios.

9.6 Weak vs. Strong Mobility#

The mobile code technologies are classified by what they can move:

In short: weak mobility moves code, strong mobility moves threads or processes. Strong mobility requires eradicating a running computation from one site, serialising it, transferring it, and restoring it at the destination: a formidable engineering challenge, provided only by a few research systems. Weak mobility is provided by several mainstream platforms, including Java, .NET, and the web.

9.7 Advantages of Mobile Code#

• Runtime flexibility: The architecture of an application can be changed while it is running; with mobile code even the code of the application can change at runtime. This continues the trend seen with data-centred and event-based styles, whose decoupling already made the runtime architecture easier to modify.
• Live updates: New versions of components can be deployed without downtime (e.g., firmware updates in Tesla vehicles, OS patches on Unix-like systems that do not require a restart).
• Enriching existing systems: New functionality can be injected into a running application, and existing services can be adapted to the client’s needs (e.g., a new JavaScript module enhancing an existing web app).
• Reduced network traffic (mobile agents): Moving computation to the data rather than moving data to the computation can be far more efficient when queries are small but data sets are large.

9.8 Challenges and Limitations#

Technical complexity:

• Linking new code into a running application at runtime is non-trivial. Languages compiled to native code (C, C++) typically resolve all links at compile or load time; you cannot unlink part of the application and relink a new version. Technologies that enable dynamic linking usually rely on a virtual machine or interpreter (e.g., the JVM, JavaScript engines), which adds overhead but provides the necessary flexibility. Java was among the first mainstream technologies designed to allow this.

Security:

• Accepting and executing code from a remote source is fundamentally risky: it is the same threat model as a virus or malware (“viruses, please enter my machine and do whatever you want”). Mitigations include:
  • Sandboxing: Restricting what the received code is allowed to do (e.g., JavaScript in a browser cannot access the local filesystem by default).
  • Code signing: Cryptographic verification that the code originates from a trusted source.
• Securing mobile code is inherently harder than securing a traditional application. Concerns are most severe for mobile agents (arbitrary code executing with full process privileges) and least severe for tightly sandboxed code-on-demand environments.

10. Combining Architectural Styles#

Architectural styles are not mutually exclusive, and the distinctions between them are not sharp. A single application may follow multiple styles simultaneously at different levels of abstraction: at a high level it may be described as client-server; more precisely it may implement REST (a specialisation of client-server); at the same time it may use mobile code patterns (JavaScript delivered as code on demand); and its internal components may communicate via an event-based broker or a data-centred queue (e.g., Kafka). These distinctions are descriptive tools, not rigid categories.

11. The Interaction Model#

11.1 Why Distributed Systems Are Fundamentally Different#

In a traditional, centralised system, the behaviour of a program is determined entirely by its algorithm. The same algorithm, correctly implemented in any language, produces exactly the same result on any machine; the speed of the CPU affects only performance, never correctness. Quicksort returns the same sorted array on a slow machine in five minutes and on a fast one in five seconds.

This ceases to be true in a distributed system.

A distributed system consists of multiple processes communicating via message passing. To describe it fully, we need not only the algorithm of each individual process, but also a model of how those processes interact. And critically, the final result of a distributed execution, not just its speed, can depend on:

1. The rate at which each process executes (CPU speed).
2. The performance of the communication channels (message transmission delay).
3. The clock drift rate of the independent clocks on each machine.

Clock drift deserves a comment. If all the clocks in the system were perfectly synchronised, having multiple clocks would be like having a single clock, and we would be back in the world of parallel systems, where the hardware supports primitives like a mutex or Java’s synchronized. It is precisely because no two clocks are ever perfectly synchronised that this escape route is closed, and the drift between clocks can affect the result of any protocol that relies on time.

This is similar to what happens in a multi-threaded program, with one crucial difference: on a single machine there is a single clock, and the hardware offers atomic instructions on which synchronisation primitives are built; in a distributed system there is no such mechanism to rely on.

11.2 Distributed Algorithms#

A distributed algorithm must be designed to produce correct results for any possible combination of process speeds, channel speeds, and clock drift rates, or at the very least, for all combinations within explicitly stated bounds. These are three factors one never thinks about when designing a sorting algorithm; here they can change the outcome entirely.

11.3 Asynchronous vs. Synchronous Distributed Systems#

Asynchronous Systems (the real world)#

In a fully asynchronous distributed system, no bounds are placed on:

• Process execution speed.
• Message transmission delay (a message could, in principle, take years to arrive: it is delayed, not lost).
• Clock drift rate (two clocks could be years apart).

This is an accurate model of the internet and real networks. However, as we will see, it makes many fundamental problems impossible to solve correctly in general.

Synchronous Systems (the engineering approximation)#

In a synchronous distributed system, all three quantities are bounded, and the bounds are known:

• The time to execute each step of a process has known lower and upper bounds.
• Each message transmitted over a channel is received within a known bounded time.
• Each process has a local clock whose drift rate from real time has a known bound.

This is the engineering approach: just as a bridge is designed to withstand a maximum-specified earthquake rather than an infinitely powerful one, distributed algorithms are designed to be correct as long as the stated bounds are not violated. For example: any message not received within one minute is considered lost; any process that does not complete its next step within one second is considered crashed.

Key trade-off: The larger the bounds, the more reliable the algorithm, but also the slower to react (e.g., waiting longer before declaring a message lost). The tighter the bounds, the more efficient but the more fragile.

Any algorithm that is correct in an asynchronous system is also correct in a synchronous one, but not vice versa. The vast majority of practical distributed algorithms assume a synchronous model, with bounds set generously (if the typical transmission time is 100 ms, set the bound 100 times larger) so that violations are rare.

11.4 The Pepperland Example#

This classic thought experiment (due to Leslie Lamport, one of the founders of distributed systems theory) illustrates the limits of asynchronous distributed systems.

Setup: Two army divisions are positioned on two hills, with an enemy in the valley between them. They are safe while they stay in their encampments, and they win only if both charge at the same time. Generals communicate by messenger; messengers are reliable (never captured) but may take an arbitrary amount of time. The generals must solve two problems:

1. Elect a leader.
2. Agree on a time to charge.

Problem 1: Leader Election (solvable in asynchronous systems)#

Since neither general has a special distinguishing property, both must run the same algorithm (an algorithm of the form “the first general sends, the second waits” presumes they already know who is first, which is exactly what is being decided):

1. Each general picks a large random number and sends it to the other.
2. The general with the higher number becomes the leader.
3. If both numbers are equal (unlikely but possible), repeat.

Since messages are guaranteed to arrive eventually and ties are broken by repeating, this process terminates with probability 1, regardless of channel speed. It may take hours, days, or years, but sooner or later there is a winner. Leader election is solvable in an asynchronous system.

Problem 2: Agreeing on a time to charge (not solvable in asynchronous systems)#

The elected leader cannot simply propose “charge at 1 pm”: the messenger may take longer than the gap between now and the proposed time, and the two clocks may disagree arbitrarily. Even with perfectly synchronised clocks, the unbounded messenger delay alone makes the problem unsolvable: the follower cannot know how far in the future to place the attack.

In a synchronous Pepperland, bounds make the problem easy. Let min and max be the known bounds on the messenger’s travel time:

1. The leader sends the messenger with “charge!”, waits min minutes, then charges.
2. The other general charges immediately upon receiving the message.
3. The two charges then start within max − min minutes of each other, and if the charge is known to last longer than that window, victory is guaranteed.

Equivalently, with a known bound on clock drift and message delay, the leader can propose an attack time far enough in the future (at least the maximum delay plus the maximum drift). Either way, it is the existence of known bounds that turns an impossible problem into a trivial one.

Conclusion: Agreement problems that cannot be solved in an asynchronous system become tractable as soon as reasonable timing bounds are introduced. This is why the synchronous model is the foundation for almost all practical distributed algorithms.

11.5 Summary#

12. Failure Models#

12.1 Partial Failures: The Key Distinction from Centralised Systems#

In a centralised system, a failure typically brings the entire system down. Users accept this: if LibreOffice dies because the SSD or the RAM of the machine it runs on fails, nobody blames the programmer. In a distributed system, failures are generally partial: only some components fail while others continue operating, and users do not accept a total shutdown caused by a partial failure (if Microsoft 365 lost your document every time one of its thousands of machines crashed, you would rightly blame the programmers). The system must detect, tolerate, and recover from failures while continuing to serve its users. The failure model defines the ways in which failures may occur, to provide a better understanding of their effects.

12.2 A Taxonomy of Failures#

Failures can affect two kinds of components, processes and channels, and can be of three kinds: omission, Byzantine, and timing. The two distinctions are orthogonal.

Omission Failures#

An omission failure means a component omits performing its expected function.

For channels: The message is dropped and never delivered. This can occur at three points:

• Send omission: the message is lost at the sender before entering the network.
• Channel omission: the message is lost in transit.
• Receive omission: the message arrives but is lost before being passed to the application.

The critical observable effect is the same in all three cases: the message does not reach its destination.

For processes: The process omits executing its program: it crashes. Two sub-cases:

• Fail-stop: The process halts and there is an out-of-band mechanism that allows other processes to detect the crash reliably.
• Crash: The process halts but detection requires in-band messaging. Without fail-stop guarantees, distinguishing a crashed process from a slow one is non-trivial.

In real systems, crash is the norm; fail-stop is an idealisation that simplifies algorithm design considerably.

Byzantine (Arbitrary) Failures#

A Byzantine failure means a component continues operating but produces incorrect or inconsistent results.

For channels: The message content may be corrupted in transit (effectively a different message arrives), non-existent messages may be delivered, or real messages may be delivered more than once.

For processes: Instead of halting, the process executes an incorrect program: it may skip intended processing steps, add extra ones, or produce wrong outputs. This can happen due to memory corruption (e.g., a hardware fault flipping bits in RAM, which in practice usually leads to a crash within milliseconds), or, more practically, due to a security compromise (a virus or attacker modifying the running code).

Byzantine failures are significantly harder to tolerate than omission failures, because:

• You cannot distinguish a Byzantine process from a correct one by simply checking whether it responds.
• You must verify the correctness of responses, not just their presence.

Timing Failures#

Timing failures are specific to distributed systems operating under the synchronous model. They occur when one of the assumed timing bounds is violated: a message takes longer than the specified maximum transmission time, or a process takes longer than the specified maximum step time. When this happens, the correctness guarantees of algorithms designed for the synchronous model no longer hold.

12.3 The Transformation Trick: Byzantine → Omission (for Channels)#

At the physical layer, all channel failures are Byzantine: electrical interference corrupts bit patterns; the signal always propagates but may arrive garbled.

However, every network protocol transforms these Byzantine channel failures into omission failures using CRC (Cyclic Redundancy Check) codes:

1. A CRC checksum is computed over the packet and appended before transmission.
2. At the receiver, the CRC is recomputed and compared to the received value.
3. If they differ, a Byzantine failure is detected and the packet is silently discarded, transforming it into an omission failure.
4. If they match, the packet is accepted. (The probability that a corrupted packet produces an identical CRC is negligibly small in practice.)

This is why, at all layers above the physical, Byzantine channel failures are essentially never observed.

The analogous trick for processes, detecting that a process is executing incorrect code, is far harder. Research (largely in the security field, since the realistic cause of a process behaving Byzantine is a compromise) has explored pairing code with formal correctness proofs that can be re-verified at the receiving side; if the code is changed, the proof fails. This remains too complex and costly for general production use. The only practical approach is cryptographic code signing, which verifies the origin of code but not the correctness of its execution.

12.4 Failure Detection#

To tolerate a failure, it must first be detected. Failure detection strategies differ between synchronous and asynchronous systems. In Pepperland terms: how does one division detect that the other has been attacked and defeated?

In Synchronous Systems: Keep-Alive Messages#

Each process periodically broadcasts an “I am alive” message. Since all timing bounds are known, a receiver can calculate the maximum interval between messages. If no keep-alive arrives within that window, the sender is presumed to have failed.

Implementation note: Prefer keep-alive (heartbeat) messages over ping-pong mechanisms. With ping-pong, a dedicated thread must send a ping, then block waiting for the pong, coordinating with the rest of the application. With keep-alive, each process simply listens passively for periodic heartbeats: it is the pong without the ping, and it is far simpler to implement and reason about.

In Asynchronous Systems: Fundamental Impossibility#

Without timing bounds, it is impossible to reliably distinguish between:

• A process that has crashed.
• A process that is slow.
• A message that has been delayed.

You cannot know how long to wait for the next “I am alive”: the messenger may still be travelling, or the sender’s slow clock may not yet have triggered the next send. This is a fundamental theoretical impossibility, not an engineering limitation. And if, in addition, messages can be lost, things get worse: a lost message and a crashed process become indistinguishable.

Fail-Stop vs. Crash#

The distinction matters precisely here. In a fail-stop system, an out-of-band mechanism (e.g., hardware monitoring, a dedicated watchdog) guarantees that a crash is detectable regardless of message loss. In a crash system, detection depends entirely on in-band messages, and if messages can be lost, a crashed process becomes indistinguishable from a slow one.

12.5 Agreement Under Message Loss: The Failing Pepperland Problem#

Can two generals agree on a binary decision (charge or retreat) if messengers can be captured (messages can be lost permanently)?

The answer is no. The intuition given in class is an induction: suppose an algorithm reaches agreement using at minimum $N$ messages; since any message can be lost, it must also work when the last message is lost, hence with $N-1$ messages; repeating the argument, it must work with $N-2$, …, down to 0 messages, and with zero messages exchanged no agreement can be reached (no communication, no agreement).

Therefore, no algorithm can guarantee agreement in a system where messages can be lost indefinitely.

In practice, this is handled with a pragmatic assumption: if a network partition persists indefinitely, the partition is treated as permanent. Each side continues operating independently, discarding any expectation of coordinating with the unreachable partition. If the link is later restored, a reconciliation protocol is needed to merge the diverged states, but this is treated as a separate problem.

12.6 Summary of Failure Types#

13. The FLP Impossibility Theorem#

Fischer, Lynch, and Paterson (1985) formally proved, in the result known as the FLP theorem, that distributed consensus is impossible in an asynchronous system even with a single crash failure.

Does this matter in real life? Yes. Virtually every distributed system relies on consensus: committing or aborting a transaction in a distributed database (e.g., when you withdraw money at an ATM), agreeing on the values of replicated distributed sensors, agreeing on whether a system component is faulty. Components that coordinate and collaborate are, constantly, reaching consensus on something: no consensus means no reliable distributed system.

13.1 How We Cope in Practice#

Since pure asynchronous systems make consensus impossible, real systems escape the impossibility in one of two ways:

• Change the assumptions: Assume reliable-enough channels (e.g., TCP assumes that sooner or later a message gets through as long as the physical connection exists) or add timing bounds to move into the synchronous model.
• Reduce the guarantees: Accept probabilistic correctness: the system is correct with very high probability rather than deterministically.

Both approaches are used in practice. Consensus algorithms (e.g., Paxos, Raft) are covered in the Agreement chapter.

14. Exam Questions#

Mobile code is the topic of this chapter that written exams target most directly; the architectural styles appear in older exams as descriptive questions. The course does not publish official solutions: the worked answers below are unofficial, reconstructed from the course material.

import java.util.*;

/* Complete the three methods so the harness prints ALL OK.
   - out(t): deposit t and wake the waiters.
   - rd(p):  return when p is present WITHOUT removing it; block otherwise.
   - in(p):  like rd(p) but REMOVE one occurrence; while readers of p are
             still waiting, in(p) must let them go first.                  */
class IntSpace {
    private final List<Integer> tuples = new ArrayList<>();
    private final Map<Integer, Integer> waitingReaders = new HashMap<>();

    public synchronized void out(int t) {
        // TODO
    }

    public synchronized int rd(int p) throws InterruptedException {
        // TODO
        return p;
    }

    public synchronized int in(int p) throws InterruptedException {
        // TODO
        return p;
    }
}

public class Playground {
    static IntSpace space = new IntSpace();

    static Thread spawn(String name, Runnable r) {
        Thread t = new Thread(r, name); t.start(); return t;
    }

    public static void main(String[] args) throws Exception {
        // phase 1: non-blocking path on a present tuple
        space.out(1);
        space.rd(1);                       // must return at once, NOT consume
        space.in(1);                       // must return at once and consume
        System.out.println("phase 1 ok: rd/in on a present tuple return immediately");

        // phase 2: in(5) on a missing tuple must block until out(5)
        Thread probe = spawn("probe", () -> {
            try { space.in(5); } catch (InterruptedException e) { }
        });
        Thread.sleep(400);
        if (!probe.isAlive()) { System.out.println("FAILED: in(5) returned on an empty space"); return; }
        space.out(5);
        probe.join(2000);
        if (probe.isAlive()) { System.out.println("FAILED: in(5) still blocked after out(5)"); return; }
        System.out.println("phase 2 ok: in(5) blocked until out(5)");

        // phase 3: many readers + one inputter on 7; a single out(7) must
        // satisfy ALL of them (readers first, then the inputter consumes)
        List<Thread> ts = new ArrayList<>();
        for (int i = 0; i < 40; i++)
            ts.add(spawn("reader", () -> {
                try { space.rd(7); } catch (InterruptedException e) { }
            }));
        ts.add(spawn("inputter", () -> {
            try { space.in(7); } catch (InterruptedException e) { }
        }));
        Thread.sleep(400);                 // let everybody block
        space.out(7);
        boolean stuck = false;
        for (Thread t : ts) { t.join(3000); stuck |= t.isAlive(); }
        if (stuck) { System.out.println("FAILED: some rd(7)/in(7) never woke up (lost the tuple to the inputter?)"); return; }
        System.out.println("phase 3 ok: 40 readers and 1 inputter all served by one out(7)");

        // phase 4: the inputter must have consumed 7 -> a new in(7) blocks
        Thread again = spawn("again", () -> {
            try { space.in(7); } catch (InterruptedException e) { }
        });
        Thread.sleep(400);
        if (!again.isAlive()) { System.out.println("FAILED: 7 still in the space, in() did not consume it"); return; }
        again.interrupt();
        System.out.println("phase 4 ok: 7 was consumed exactly once");
        System.out.println("ALL OK");
    }
}

import java.util.*;

public class IntSpace {
    private final List<Integer> tuples = new ArrayList<>();        // multiset of tuples
    private final Map<Integer, Integer> waitingReaders = new HashMap<>();

    public synchronized void out(int t) {
        tuples.add(t);
        notifyAll();                       // wakes everybody; the guard in in() lets readers go first
    }

    public synchronized int rd(int p) throws InterruptedException {
        waitingReaders.merge(p, 1, Integer::sum);                  // register as a reader of p
        try {
            while (!tuples.contains(p))
                wait();
        } finally {
            waitingReaders.merge(p, -1, Integer::sum);             // done (or interrupted)
        }
        notifyAll();                       // a waiting in(p) may proceed now
        return p;
    }

    public synchronized int in(int p) throws InterruptedException {
        while (!tuples.contains(p) || waitingReaders.getOrDefault(p, 0) > 0)
            wait();                        // defer to pending readers of p
        tuples.remove(Integer.valueOf(p)); // remove the tuple, not the index!
        return p;
    }
}

15. Glossary#

Compiled from G. Cugola's lectures (A.Y. 2025/26), slides + class transcriptions. Not official course material.