Communication

After studying the architectures of distributed systems and how to model their behaviour, we now focus on the communication layer: the actual mechanisms a middleware offers to let distributed components talk to each other. Everything in this chapter belongs to one of four families of communication service: message-oriented communication, Remote Procedure Call (RPC), Remote Method Invocation (RMI), and stream-oriented communication. Section 1 closes by setting out what each one is and the order in which we walk through them.

1. Layered Protocols and Middleware#

Protocols are rules that define how entities communicate over a network. They are organised into layers, as described by the OSI model:

In the internet stack, layers 1-2 are not standardised (e.g., Ethernet, Wi-Fi/802.11), and the distinction between layers 5-7 does not officially exist. The internet protocol suite maps roughly as follows:

• Layer 3 → IP
• Layer 4 → TCP / UDP
• Layers 5-7 → Application layer (no official TCP/IP standards, but many widely-used protocols such as HTTP, SMTP, etc.)

Packet-based protocols work by encapsulation: each layer wraps the payload from the layer above inside its own packet structure, and the result is placed inside the packet of the layer below. This is how the layered model is physically realised.

Middleware sits between the transport layer (TCP/UDP) and the application layer: logically it is one more protocol layer, offering higher-level abstractions to the application programmer. Which abstraction it offers is what distinguishes the four families of communication service.

The four families. Concretely, “abstraction” here means: what does the middleware put in the programmer’s hands to communicate with? There are four answers, and they are the four families; every paradigm we will meet in this chapter is one of them, or a variant of one:

1. Message-oriented communication. The programmer works directly with messages: one side sends a message, the other one reads it. Nothing is hidden, writing the message and picking it up are two separate actions that the code performs explicitly. Several different paradigms fall under this family (sockets, MPI, message queues, publish-subscribe); they all pass messages around, but they differ in how much the sender has to know about who will read them, and we will look at them one at a time.
2. Remote procedure call (RPC). Messages are still travelling underneath, but the middleware hides them: the programmer calls a procedure that runs on another machine, and the call is written exactly like a normal local one.
3. Remote method invocation (RMI). The same idea in an object-oriented setting: what gets invoked is a method on an object living on another machine. It is close enough to RPC to be called its object-oriented sibling, but what can be passed as a parameter changes enough to deserve a separate section.
4. Stream-oriented communication. Continuous media such as audio and video, where when the data arrives is not just a matter of speed: arriving late is as wrong as not arriving at all.

The order we follow. The chapter does not go through the four families in the order above. We start from plain message exchange, because it is the most bare-bones mechanism and everything else is built on top of it. Then come RPC and RMI, which take that mechanism and hide it behind a procedure or method call. After them we go back to messages, this time with something sitting in the middle between sender and receiver, a queue, or a dispatcher that decides who gets what, which is easier to appreciate once we have seen how tightly a remote call ties the two sides together. Streaming comes last, since its problem is a different one. This means the message-oriented family is split in two parts, with RPC and RMI in between; its three sections are the ones whose title starts with “Message-Oriented Communication”.

Communication
│
├── Preliminaries
│   ├──  1. Layered Protocols and Middleware
│   └──  2. Types of Communication ........ transient/persistent, sync/async
│
├── Family 1: message-oriented ............ the message is the primitive
│   ├──  3. Message Passing ............... sockets, MPI, the channel lab
│   ├──  6. Message Queuing ............... queues, brokers, load balancing
│   └──  7. Publish-Subscribe ............. subscriptions, routing, CEP
│
├── Family 2: remote procedure call ....... the call syntax hides the messages
│   └──  4. Remote Procedure Call ......... stubs, IDL, marshalling
│
├── Family 3: remote method invocation .... the same, on remote objects
│   └──  5. Remote Method Invocation ...... proxy, skeleton, remote references
│
├── Family 4: stream-oriented ............. timing is part of correctness
│   └──  8. Stream-Oriented Communication . QoS, buffering, FEC, interleaving
│
└── Wrap-up
    ├──  9. The Communication Paradigms at a Glance
    ├── 10. Exam Questions
    └── 11. Glossary

2. Types of Communication#

Communication in distributed systems can be characterised along two independent dimensions: persistence and synchronicity.

2.1 Transient vs. Persistent Communication#

A transient example is UDP: if the destination process is not running when the packet arrives, the packet is discarded. The real-world analogy is a phone call: if the other person is not available, the communication fails. A persistent example is email: a sender can transmit a message while the recipient is offline, and the message is stored until retrieved.

2.2 Synchronous vs. Asynchronous Communication#

Note that “synchronous/asynchronous” here refers to communication behaviour, not to the concept of synchronous/asynchronous systems introduced when modeling distributed systems. These are distinct uses of the same terms. There are several degrees of synchronicity, distinguished by where the sender resumes execution:

• Fully synchronous. The sender blocks until the receiver has received, processed the message, and (optionally) sent a reply. Execution resumes only after the full round-trip.
• Receipt-based synchronisation. The sender blocks only until there is a guarantee that the message has been received by the destination process (but not necessarily processed).
• Delivery-to-middleware synchronisation. The sender blocks only until the message has been accepted by the local middleware (e.g., the operating system’s network buffer).
• Fully asynchronous. The sender returns immediately after initiating the send, with no synchronisation point and no guarantee about delivery.

2.3 Combining Persistence and Synchronicity#

The two dimensions are orthogonal and combine freely:

Email as a case study. Consider sending an email using a desktop mail client (not a web interface):

1. User clicks “Send” → message is handed to the local OS buffer.
2. A few moments later → message is transmitted to the outgoing mail server (SMTP), where it is stored persistently.
3. Later → message is forwarded through the SMTP relay network to the recipient’s incoming mail server.
4. When the recipient connects → they retrieve the message via POP or IMAP.

At the moment the “Send” window closes, the only guarantee is that the message may have entered the OS buffer. If the machine crashes immediately after clicking Send, the message could be lost permanently; it may not yet have reached the outgoing server. This uncertainty matters for correctness: since a sender that crashes right after sending cannot know whether the message was delivered, the correct approach on recovery is to resend the message, accepting that the receiver may receive it twice, and to design the system to handle duplicate messages gracefully.

3. Message-Oriented Communication: Message Passing#

3.1 Motivation and Reference Model#

Message-oriented communication is built around the simplest primitive there is: the message. Sending and receiving are two separate operations that the code performs explicitly, and a reply, when one is needed, is simply another message going the other way. Nothing is hidden and nothing is implied, which is precisely why this is where we start, since every other family in this chapter is ultimately messages with something layered on top.

Because nothing is implied, the choices of section 2 are left open: message-oriented middleware can offer any degree of synchronicity, and can add persistence at any point along the path (every release point of the synchronicity figure is a legitimate option here). That freedom is the family’s defining trait, and it will be worth contrasting later with the paradigms of section 4 and section 5, which fix these choices for the programmer:

The three forms. In terms of reference model, the most straightforward form is message passing, mapped directly onto the communication facilities of the underlying network OS (sockets), with MPI as a middleware-level variant; it is the subject of the rest of this section. Message queuing and publish-subscribe (sections 6 and 7) are instead provided at the middleware layer by a set of communication servers that store and route messages hop by hop, forming what is nowadays called an overlay network. Those two are treated after RPC and RMI, since the decoupling they buy is easiest to appreciate against the tight coupling of a remote call.

3.2 BSD Sockets#

BSD sockets, also called Berkeley sockets, after the Berkeley Software Distribution of Unix where they first appeared in 1982, are the operating system API through which applications access TCP and UDP. Note the distinction: TCP and UDP are the protocols, their RFCs describe how data travels on top of IP, while sockets are the programming interface to the services those protocols offer. The API is available on every modern OS and comes in two types: stream sockets, the interface to TCP, and datagram sockets, the interface to UDP.

Stream sockets (TCP). Communication is connection-oriented, and the two sides play different roles in setting it up. The server creates a socket, binds it to a port, marks it as listening, and calls accept, which blocks until someone shows up. The client creates its own socket and calls connect, naming the server’s address and port. The accept-connect pair is the synchronisation point: once both have been called the connection exists, and each side holds a socket representing its end of the pipe. From there the interaction is symmetric, both sides read and write as many times as their protocol requires, until one of them closes the connection. Each established connection is identified by four values: source IP, source port, destination IP, destination port. This is why a server can accept a huge number of simultaneous connections on the same port: as long as at least one of the four values differs (two clients on different machines, or two processes on the same machine using different outgoing ports), the connections are distinct.

Datagram sockets (UDP). No connection is established, and there are no roles: both parties use the same approach. Each creates a socket and binds it to a port, the bind is optional on a side that only sends, since the OS can pick a port for it, and then sends with sendto, which names the destination address and port on every single call, and receives with recvfrom, which waits for the next incoming datagram. Precisely because there is no connection, one socket can send to and receive from any number of different hosts. Each packet is an independent story: routed separately, possibly arriving out of order, with no delivery guarantee.

Multicast sockets. BSD also supports multicast datagram sockets, which implement group communication on top of the IP multicast mechanism just described. When discussing groups, one distinguishes open groups (anyone may send to the group, member or not) from closed groups (only members may send); since IP multicast lets any host transmit to a group address, BSD multicast sockets implement open groups. The API is the datagram one, plus a join operation (a setsockopt call):

• Any process may join a multicast group, identified by a class D IP address plus a port.
• Any process, whether or not it has joined, may send to the group.
• All processes that joined the group receive the packets sent to it.

Being open by design, multicast offers no security by itself: anyone can send to any group.

3.3 MPI: Message Passing Interface#

BSD sockets are general-purpose but low-level, and protocol-independent by design, which prevents access to the specific features of the underlying network. MPI (Message Passing Interface) is a message-oriented middleware standard (with several implementations) designed for high-performance computing: clusters of homogeneous machines doing large-scale number crunching or data processing. It is at the same time higher-level than sockets (complex data structures can be sent directly, not just byte arrays) and endowed with finer control over communication semantics and memory management.

Send primitives. MPI offers several send operations with explicit synchronisation semantics:

Each variant trades latency against the strength of the delivery guarantee.

Receive primitives. MPI_Recv is blocking: the caller waits until a message arrives. MPI_Irecv is a non-blocking probe: it returns immediately, delivering a message if one is available and null otherwise.

Buffer sharing and zero-copy. The primitives above copy the application’s buffer into the middleware’s buffer, which is safe but costly. The non-blocking variants (MPI_Isend, MPI_Issend) instead pass a reference to the application’s buffer, which the middleware uses directly to feed the network card, avoiding the memory copy; the application must not reuse the buffer until a separate call confirms the middleware has released it. This zero-copy path matters for squeezing maximum throughput out of high-performance hardware.

Process addressing and startup. An MPI application is written once and launched simultaneously on all participating machines. Each process is assigned a numeric rank (0 to N−1) within its group at startup, and the pair (group ID, process ID) acts as source/destination address; the code branches on the rank to differentiate behaviour (e.g., rank 0 coordinates, the others compute). Messages are addressed to ranks, never to IP addresses. MPI offers no fault tolerance: crashes are assumed fatal, consistently with the stable cluster environment it targets.

Collective operations. Beyond point-to-point sends, MPI provides high-level operations over process groups:

• Broadcast: send the same message to all processes in a group.
• Scatter: distribute portions of an array across the group; an array of N elements sent to K processes gives each process N/K elements.
• Gather: collect the results of all processes into a single array at the caller.
• Reduce: like gather, but the partial results are combined by a reduction function (sum, average, max, min, …), returning a single value.

Finally, MPI implementations are protocol-independent in the useful direction: they automatically select the best network protocol available on the hardware (e.g., high-performance interconnects on a cluster), with no changes to application code.

4. Remote Procedure Call (RPC)#

4.1 Local Procedure Calls and Parameter Passing#

Before examining RPC, it is worth recalling how local procedure calls work, since RPC is designed to replicate that experience across a network. In a language like C, when a main program invokes a procedure, space is reserved on the stack frame for the procedure’s parameters, the return value, local variables, and the return address; the program counter then jumps to the first instruction of the procedure, which executes and returns control to the caller. For read(fd, buffer, nbytes), for instance, the stack frame includes space for the file descriptor, the buffer, and the number of bytes, alongside the return address and the return value.

There are three main conventions for passing parameters to a procedure:

1. Call by value. A copy of the actual parameter is placed on the stack. The procedure works with its own copy; changes do not affect the original variables in the caller. A swap(a, b) procedure that swaps a and b internally will leave the caller’s variables x and y unchanged, because only copies were passed.
2. Call by reference. The formal parameters (a, b) and the actual parameters (x, y) refer to the same memory location. Any changes made inside the procedure are immediately visible to the caller: swap(a, b) correctly swaps x and y.
3. Call by copy-restore (also known as call by value/result). Parameters are copied in at call time, the procedure executes, and the final values are copied back to the original variables at return time.

In most cases, call by copy-restore and call by reference produce the same result. They diverge in the presence of aliasing: the same variable passed more than once as an actual parameter. Consider:

fill(x, x);   /* passing x twice; initially x = 7 */

/* Procedure body: */
/*   a = 20;        */
/*   b = 10;        */

• Call by reference: a and b both refer to the same memory location (x). Writing to a and then to b results in x = 10 (the last write wins, deterministically).
• Call by copy-restore: a and b are separate copies. At restore time, both are written back to x, but the order of the write-backs is unspecified, so the result is either 20 or 10, non-deterministically.

The key takeaway is that call by reference preserves aliasing; call by copy-restore does not. A further practical limitation, noted in the course slides: value/result works naturally for arrays, but not for arbitrary linked data structures (which contain pointers), and it can be optimised when a parameter is input-only or output-only (copy in one direction suffices).

4.2 How RPC Works#

The idea. The message-passing primitives of section 3 require the programmer to manually serialise data structures into byte arrays, design a message format for requests and responses, and reconstruct the data structures at the receiving end. RPC (invented in the 1970s) abstracts all of this away: the goal is to let a programmer invoke a procedure on a remote machine with the same simplicity, and the same syntax, as invoking a local procedure. The middleware should solve as many of the underlying problems as possible, starting with parameter passing.

To do so, RPC introduces two middleware components, the client stub and the server stub, that sit between the application code and the network.

1. The client calls the client stub using the normal procedure call syntax. The stub is a local procedure with exactly the same signature as the remote one, but its body does a different job.
2. The client stub serialises the parameters into a byte array (marshalling) and sends the packet over the network, together with a reference identifying the target procedure.
3. The client stub blocks, waiting for the response.
4. The server stub receives the packet, deserialises the parameters, and invokes the actual remote procedure; the procedure is logically invoked by the client’s code, but physically by the server stub.
5. The remote procedure executes and returns its result to the server stub.
6. The server stub serialises the return value and sends it back across the network.
7. The client stub receives the response, deserialises the return value, and returns it to the calling application.

From the application programmer’s perspective, the call looks entirely local. Moreover, because only data is transmitted (never code), the client and server can be written in different languages: as long as there is a mapping between equivalent data types in each language (e.g., a C struct and a Python class), RPC works across language boundaries.

4.3 IDL, Serialisation, and Marshalling#

To generate the two stubs automatically, the middleware must know the interface of the remote procedure: its name, the number and types of its parameters, the type of the return value, and the definitions of any complex types involved. An Interface Definition Language (IDL) is a language-neutral notation for describing exactly this, with mappings onto each target programming language.

Passing a parameter poses two distinct problems, and the corresponding operations have distinct names:

Both stubs perform these operations in opposite directions, and crucially this code is generated automatically from the IDL description: the programmer never writes it by hand. The IDL must therefore be expressive enough to describe all primitive types and their encodings (e.g., integers as big-endian 32-bit, doubles as 64-bit IEEE 754), compound types, and complete procedure signatures.

4.4 Parameter Passing in RPC#

In a procedural language, “passing by reference” in practice means passing a pointer: sharing a physical memory address. This is meaningless across machines, since a pointer is only valid within the address space of one process and there is no shared memory. Therefore, in RPC:

• Call by value is the default and always safe: the client copies data into the packet, and the server receives an independent copy.
• Call by reference is impossible in the traditional sense. Supporting it would require changing the runtime of the language, since every procedural runtime implements by-reference as memory sharing.
• Call by copy-restore can simulate pass-by-reference effects, and some RPC middleware use it for this purpose; others (like Sun RPC) simply do not support anything beyond by-value. As discussed above, copy-restore semantics differ from true by-reference in the presence of aliasing.

In object-oriented systems the concept of “reference” is more abstract than a raw memory pointer, which makes remote references tractable; this is addressed by RMI in section 5.

4.5 Sun RPC#

Sun RPC (also called ONC RPC), invented at Sun Microsystems, was the first widely adopted RPC implementation and became the de facto standard for RPC over the internet; it is still in use today.

Key characteristics:

• Wire format: XDR (External Data Representation), a compact and efficient binary encoding.
• Transport: can run over TCP or UDP.
• Parameter passing: by value only.
• Language support: originally C only (C++ by extension).
• Security: not built-in, but can be integrated separately.
• Notable use: at the core of NFS (Network File System), the standard Unix/Linux mechanism for sharing disk volumes over a network (despite the name, NFS shares volumes, not individual files).

NFS demonstrates that RPC can be as efficient as the underlying network allows; the middleware adds only a very thin layer on top of TCP/UDP, with none of the overhead of verbose text encodings.

The development workflow is:

1. Generate a UUID (Universally Unique Identifier) for your procedure.
2. Write the IDL description using a C-like interface definition language, specifying procedure signatures and shared data structures.
3. Run the IDL compiler, which automatically generates a shared header file (included by both sides), the client stub, and the server stub.
4. Write the client code (invoking the procedure) and the server code (implementing it).
5. Compile and link each side with the generated stubs and the RPC runtime library.
6. Launch the server; it registers itself and waits for connections. Launch the client; it invokes the procedure as if it were local.

From the programmer’s perspective, the distribution is almost entirely hidden; the only visible seam is the step where the client specifies which server host to connect to.

4.6 Binding: Portmap and the DCE Directory#

To invoke a remote procedure, the client must resolve two distinct problems: which host is running the server, and which process on that host implements the target procedure.

Sun RPC does not solve the first problem (the client must know the host’s IP address) but solves the second with a special daemon called portmap (also known as rpcbind):

• Portmap runs on the server host and listens for all incoming RPC connections.
• When a server process starts, it registers itself with portmap, announcing which procedure UUID it implements.
• When a client packet arrives, portmap reads the UUID and forwards the packet to the correct server process.

DCE (Distributed Computing Environment) is a more complete middleware built on top of Sun RPC. Among other services it adds a directory service, analogous to a DNS for procedures: servers publish their procedure UUIDs by name, and clients look up a procedure by name, discovering the host IP address automatically instead of hard-coding it (the directory server’s own address is typically kept in a configuration file read by the client stub). Microsoft’s DCOM and .NET Remoting are implementations of DCE. Note that the DCE directory is a centralised server scoped to an organisation’s internal network; unlike DNS, it is not designed to scale to internet-wide use.

4.7 Dynamic Activation#

By default, a Sun RPC server must be running before any client invokes it, and it stays active forever whether clients exist or not. This is wasteful if the server is rarely used. Dynamic activation launches the server on demand, only when a client request arrives.

In Unix systems, Sun RPC leverages the pre-existing inetd daemon: inetd reads a configuration file listing which process to launch for each network port, listens on all listed ports, and on receiving a packet launches the appropriate server process and forwards the packet to it. This illustrates a general design principle of that generation of systems: when a sub-problem (here, on-demand activation) is general, solve it once with a general-purpose tool and reuse that tool, rather than building a custom mechanism for each application.

4.8 Lightweight RPC#

Once RPC proved successful, the same idea was applied to inter-process communication on a single machine. Modern operating systems isolate process memory, so two local processes cannot share data directly. They could communicate via regular RPC pointed at localhost, but that would serialise data through the full networking stack just to come back to the same machine.

Lightweight RPC optimises this case by replacing the network channel with OS-managed shared memory: the client and server stubs run on the same machine and exchange data through a shared memory region requested from the operating system, while the programming interface remains identical to standard RPC. The shared memory is visible to the middleware code only, not to the application, so no language runtime changes are needed.

A concrete example: copy-and-paste in Windows is built on the lightweight RPC machinery inside COM/.NET. Copying in Word and pasting into Excel involves the source process, a shared OS-managed data area, and the destination process, communicating via lightweight RPC.

4.9 Asynchronous, Batched, and Queued RPC; Futures#

Standard RPC is fully synchronous: the client blocks until the remote procedure completes, mirroring a local call. This is not always necessary.

Asynchronous RPC (void procedures). For procedures that return no value, waiting serves no purpose. Most RPC middleware, including Sun RPC, allow such procedures to be declared asynchronous: the client invokes the stub, which hands the request to the middleware after a brief synchronisation, and the client immediately continues; the remote procedure executes in parallel and no reply is sent.

Batched RPC. Sun RPC can buffer asynchronous invocations on the client instead of sending each immediately: buffered calls are flushed together in a single packet when a non-batched (synchronous) call is issued or when a short timeout expires. Latency increases slightly, but bandwidth efficiency improves significantly, since the per-packet overhead is paid once for many invocations. A typical beneficiary is writing a large file to a remote NFS volume: a long sequence of sector writes needs no individual confirmations, so batching yields throughput limited only by network speed.

Queued (persistent) RPC. The Rover toolkit (an MIT research proposal, designed with mobile hosts in mind) extended batching into full queuing: asynchronous invocations are stored persistently in the middleware, which retries automatically until the destination becomes reachable, and replies can even come back through a different channel; the client obtains a promise and continues immediately. This moves RPC from transient to persistent communication: the invocation is guaranteed to eventually reach the server, across crashes or disconnections.

Futures and promises. For procedures with return values, asynchrony is still possible: the client invokes the procedure and immediately receives a future (or promise), a placeholder for the result, and continues executing. Only when the client actually accesses the future’s value does it block, waiting for the remote result if it has not arrived yet. If the result is not needed for several instructions after the call, those instructions execute in parallel with the remote procedure, overlapping computation and communication.

Detecting the access to the future requires support from the language runtime, which is why standard Sun RPC does not implement it; languages and runtimes with native future/promise support (modern JavaScript, Java, Python) handle it naturally.

4.10 RPC at a Glance#

5. Remote Method Invocation (RMI)#

5.1 From RPC to RMI#

RMI applies the same core idea as RPC, hiding network communication behind a familiar invocation syntax, but in an object-oriented programming environment: the client invokes a method on a remote object. The key insight is that switching to an object-oriented world inverts the difficulty of parameter passing:

5.2 Architecture: Proxy and Skeleton#

The RMI architecture mirrors RPC’s client stub / server stub pattern, with object-oriented terminology:

• The proxy is a local object that has the same public interface as the remote object. Each method, instead of executing application logic, serialises its parameters and sends them over the network.
• The skeleton waits for incoming calls, deserialises the parameters, and invokes the corresponding method on the actual remote object.

From the programmer’s perspective, calling a method on the proxy is indistinguishable from calling it on a real local object.

5.3 Implementations: Java RMI and CORBA#

Java RMI is the RMI incarnation for Java: both client and server must be written in Java, and the IDL is simply a standard Java interface extending the Remote marker interface. Since both sides run on the Java Virtual Machine, hardware and OS differences are absorbed entirely by the JVM.

CORBA (Common Object Request Broker Architecture) is a standard defined by the OMG (Object Management Group), the same body that standardises UML. It implements RMI across multiple languages: a client written in Java can invoke methods on a server written in C++, Python, or any of the many languages with a CORBA mapping, described through CORBA’s own language-independent IDL. CORBA is a standard with several implementations, which should interoperate as long as they are compliant.

5.4 Passing Objects by Reference#

In a pure object-oriented language, variables hold references to objects, not the objects themselves; this makes remote references natural. Consider a class Car with a private attribute owner of type Person and a method setOwner(Person p). On the client machine, a Person object P is created, and the client holds a remote reference to a Car object C living on a server. When the client calls C.setOwner(P):

1. The client’s proxy for C intercepts the call.
2. Instead of serialising the full Person object, the middleware creates a skeleton for P on the client and ships a proxy for P (generated in the destination language) along with the call.
3. The skeleton of C receives the call, unpacks the proxy for P, and sets owner = proxy(P).
4. On the server, owner now points to a proxy that transparently forwards any method call back to the original P on the client machine.

This works because in an object-oriented system all object access goes through methods: the proxy can intercept any method call on owner and route it back to the original machine, and the application code never touches the object’s data directly. If code could do owner.name = ... bypassing the methods, the mechanism would break.

Passing proxies across multiple machines. When a proxy is itself passed from one machine to another, what travels is not the proxy’s code but just the network address and object identifier of the original object (e.g., IP, port, UUID). The receiving machine builds its own proxy pointing directly at the original, so there is never a chain of proxies, just a flat reference back to the source.

5.5 Why Passing by Copy is Hard#

Serialising the state of an object (name, surname, date of birth) is straightforward; it is the same marshalling problem already solved in RPC. The difficulty lies in the methods: passing an object by copy to a machine running a different language would require translating its application code (e.g., a print method) from the source language to the destination language, which is in general not possible. This is why most RMI implementations, including CORBA originally, only support pass-by-reference across heterogeneous language boundaries.

The Java RMI exception. Java RMI can support pass-by-copy because the JVM eliminates heterogeneity. If an object implements the Serializable interface (an empty marker interface), it can be passed by copy:

1. The object’s state is serialised and sent, together with the class identifier and version number (not the code itself).
2. The receiving JVM looks up the class locally, creates a new instance, and restores the serialised state into it.
3. The class code is expected to already be present on the destination machine; the version identifier ensures the exact same version of the class is used, since classes evolve over time.

CORBA later added limited support for pass-by-copy as well (“objects by value”, from CORBA 2 onward), partly motivated by the popularity of Java.

5.6 Risks of Hiding Distribution#

RMI makes distributed programming feel like local programming: an object receives a parameter, passes it along to another remote object, and proxies to the original are created wherever it lands, with no visible difference from local code. The only moment the programmer sees the distribution is the initial binding to a remote object; from then on, control over where objects live is quickly lost. In practice, students presenting RMI projects often cannot answer the question “on which machine is this code running?”, which is both a testament to the abstraction and a warning.

6. Message-Oriented Communication: Message Queuing#

6.1 Model and Primitives#

Message queuing is the middleware paradigm that naturally supports data-centred architectures: components do not interact directly, but through shared queues that sit in the middle of the system, in the same way tuple spaces mediate data-centred designs. Intrinsically the resulting architecture is peer-to-peer: any component may add messages to queues and retrieve messages from them, through its own local view of the queues kept in sync by the middleware. The main primitives are:

• Put: add a message to a queue.
• Get: retrieve a message from a queue (blocking until one arrives).
• Poll: check whether a message is available, without blocking.
• Notify: register to be informed whenever a new message appears on a queue.

Naming and lookup. Queues are identified by symbolic names. Most systems provide a lookup service to list available queues and retrieve a reference to a queue by name; some allow creating queues dynamically from the application, while others fix the set of queues at deployment time, registered by the system designer.

6.2 Comparison with RPC/RMI#

To see the decoupling at work, imagine implementing a logical client-server interaction over a queue: clients push requests onto the server’s queue, and servers pick requests up, process them, and put results onto per-client reply queues. Now clients need not know how many servers consume the queue: a new server can be added at runtime, or a server stopped and restarted elsewhere, without touching the clients. Sharing one request queue among several servers also gives load balancing essentially for free. And the queue decouples the parties in time: requests submitted while the server is down simply accumulate and are served when it returns.

The price is programming effort: the encoding of requests and results into messages, which stubs and proxies do automatically in RPC/RMI, must here be implemented by hand, and a client wanting a reply must include the name of its own reply queue in the request. Harder to program in this sense, but far easier to rearrange at runtime.

6.3 Implementation Architectures#

The physical realisation of queues varies:

• Library-only: no host actually “holds” queues; they exist only inside the middleware libraries running at the communicating peers.
• Centralised server: a single host stores all queues. Simple, but a single point of failure.
• Distributed/replicated queues: each queue is spread or replicated across multiple hosts (the communication servers of the overlay network), and messages are routed hop-by-hop from sender to receiver. This removes the single point of failure and improves fault tolerance.

6.4 Broker Programs#

Some queuing systems allow application code to be injected into the middleware itself, as broker programs (message brokers). A broker intercepts messages as they enter a queue and can transform them, duplicate them, or filter and reroute them based on content.

The classic use case is repairing an API mismatch: a server expects messages with fields in the order (name, surname), while a new client produces (surname, name). Rather than modifying either side, a broker program is deployed that intercepts the client’s messages, reorders the fields, and re-inserts them into the queue; the server processes them unchanged. Integration and conversion logic thus lives inside the middleware, keeping the application components clean.

Widely used systems in this family include IBM MQ (formerly MQSeries), Microsoft Message Queuing (MSMQ), and more recently Apache ActiveMQ, RabbitMQ, and Apache Kafka (the latter also supporting streaming).

7. Message-Oriented Communication: Publish-Subscribe#

7.1 Model and Properties#

Publish-subscribe (pub-sub) is the middleware paradigm that naturally supports event-based architectures, just as message queuing supports data-centred ones. Components interact through two primitives:

• Publish: emit an event notification, a message describing something observed to have happened in reality (hence the name: a sensor detects smoke and notifies the event).
• Subscribe: register interest in receiving certain event notifications.

At the centre of the system sits an event dispatcher that stores subscriptions and routes each published message to all matching subscribers.

Compared to RPC (synchronous, point-to-point, explicitly addressed), pub-sub gives a much higher degree of decoupling: a publisher does not know, or care, whether zero, one, or a thousand subscribers will receive its message. Compared to message queuing, the interaction is similarly anonymous but transient: a subscriber that appears ten seconds after the publication does not receive it, whereas a queued message waits. This makes pub-sub well suited to dynamic environments where components appear and disappear frequently and the system reacts to real-world events rather than explicit requests.

7.2 Subscription Models#

How expressively can a subscriber declare its interest? Two ends of a spectrum:

Topic-based (subject-based) systems. Each message is tagged with a topic; subscribers subscribe to a topic and publishers publish to one. The topic plays the same role as a queue name: subscribe to software-engineering and a message published on distributed-systems does not reach you. Topics can be organised into hierarchies: subscribing to lessons-at-polimi also delivers messages published under lessons-at-polimi/distributed-systems/communication.

Content-based systems. The entire content of the message, typically structured as key-value pairs, can appear in the subscription, expressed as a predicate. For example, the subscription issue = "temperature" AND value > 50 behaves as follows:

• A message {issue: "smoke"} → not delivered.
• A message {issue: "temperature", value: 35} → not delivered.
• A message {issue: "temperature", value: 70} → delivered.

Hybrid systems. Many real systems combine both: a mandatory topic field narrows the scope, and additional content predicates refine the match within the topic (subscribe to topic distributed-systems but only for messages with year > 2024).

The trade-off runs between expressiveness for the programmer and load on the middleware: matching a topic label is cheap, while a content-based dispatcher must open every message and evaluate potentially complex predicates against all stored subscriptions.

7.3 The Event Dispatcher: Centralised or Distributed#

The dispatcher can be a single host: easy to implement, invisible to the application programmer (who only sees the publish/subscribe library), but a single point of failure and a potential bottleneck. To avoid this, the dispatcher can be distributed across a network of brokers: each application component connects to a nearby broker, and the brokers cooperate to route messages from publishers to subscribers. The first distributed dispatchers assumed an acyclic graph of brokers, which keeps routing simple, especially for content-based systems.

7.4 Routing Strategies on Acyclic Broker Networks#

On an acyclic broker graph, three routing strategies exist. In the figures and discussion below, saying that “a broker subscribes” means some application component attached to that broker issued the subscription.

Strategy 1: message forwarding. Subscriptions are stored only at the broker the subscriber is directly connected to. Publications are flooded to every broker; each broker checks its local subscription table and delivers to local subscribers, dropping the message otherwise. Subscribing is cheap (a local operation); publishing is expensive (every message floods the network).

Strategy 2: subscription forwarding. Subscriptions are propagated through the network when issued: each broker records the direction each subscription came from, building routes that lead back to subscribers. Publications then follow only the minimal paths from the publisher to the matching subscribers, skipping branches with no interested parties. Duplicates are handled cleverly: a broker that receives the same subscription a second time forwards it only towards the first subscriber (the rest of the network is already informed), and one that receives it a third time does nothing at all. The first subscription is thus expensive (it floods the network) but each identical one costs less and less, while publications are cheap.

Strategy 3: hierarchical forwarding. A compromise between the two extremes. The acyclic graph is turned into a tree by electing one broker as the root; every broker now has a single parent (its neighbour in the direction of the root) and possibly several children. Subscriptions and publications then follow two different disciplines:

1. Subscribing. A broker that receives a subscription, from a local client or from a child, stores it, remembering which link it arrived on, and forwards it only upward, to its parent. Subscription state thus accumulates along the single path from the subscriber to the root: each broker knows the interests of its own subtree and nothing else, while the root ends up knowing the interests of the entire network.
2. Publishing, the upward leg. A published message always climbs all the way up to the root, whether or not anything matches along the way. This leg is mandatory: since subscriptions travel only upward, a broker’s table says nothing about subscribers outside its subtree, so it can never conclude that the rest of the tree is uninterested, the only safe move is to keep pushing up.
3. Publishing: the downward branches. Every broker the message reaches, during the climb as well as during a descent, does the same three things: it delivers the message to its matching local subscribers, forwards a copy down every child link from which a matching subscription arrived (except the link the message came from), and, only if the message is still on its way up, passes it on to the parent. Matching subtrees are thus “peeled off” at each step of the climb, and each descending copy keeps descending only along branches that declared interest.

The costs land in the middle by construction: a subscription travels one root-ward path instead of flooding the network, and a publication pays the mandatory climb to the root but descends only where there is interest.

Two further refinements for subscription forwarding:

• Coverage between subscriptions. With a content-based language, if a broker already propagated temperature > 50 and a new subscription temperature > 70 arrives, there is no need to propagate the new one: every message matching temperature > 70 also matches temperature > 50, so the network already routes all relevant messages towards this broker. The broader predicate covers the narrower one. Recognising coverage is easy between simple predicates and can become very hard between complex ones, but production systems do perform this optimisation.
• Joining brokers. When a new broker connects to the network, it clones the subscription table of its neighbour, so it starts with a consistent view of what must be routed where.

Each broker, on every message, must match the message against its stored filters; the efficiency of this matching, together with the forwarding strategy, largely determines the throughput of the whole system.

7.5 Limits of the Acyclic Assumption#

An acyclic broker topology is easy to reason about but does not always map well onto the physical network, which is not a tree. If brokers sit in Milan, Como, Switzerland, and Lyon, the tree may force a Milan → Como → Switzerland → Lyon route even when a direct Como → Lyon link exists; adding that shortcut creates a cycle, which breaks the routing strategies above. Handling cycles is the subject of the next two sections.

7.6 Topic-Based Routing over a DHT#

While content-based routing on a cyclic topology is possible, as the next section shows, the routing state it requires is heavy. Many practical systems therefore simplify the subscription language rather than the topology: topic-based subscriptions make routing on an arbitrary (cyclic) overlay straightforward, and a particularly elegant implementation reuses Distributed Hash Tables. A DHT offers the interface of an ordinary hash map (key → value) over data spread across many machines: given a key, its routing protocol reaches the node responsible for that key in $O(\log N)$ hops, on an arbitrary (cyclic) overlay. DHTs, and the Chord protocol in particular, are covered in detail in the Peer-to-Peer chapter; here it suffices to know the interface.

The idea: use the topic name as the key, and the list of subscribers to that topic as the value.

Responsibility for topics is automatically spread across the nodes (each topic hashes to a different responsible node), and the DHT routing works happily on cyclic networks. The main limitation: this only works for topic-based subscriptions, since DHT routing relies on exact key matching; a content-based predicate like temperature > 50 cannot be hashed to a single key.

Subscription lifecycle is handled either by explicit unsubscribe operations (which retrace the forwarding path removing state) or by a TTL mechanism requiring subscribers to periodically renew.

7.7 Content-Based Routing on Cyclic Topologies#

Content-based routing on cyclic networks is possible but requires considerably richer routing state to avoid loops. Two main approaches exist, plus a refinement of the first, the common idea: since the graph itself has cycles, every message is forced to travel along some tree carved out of the graph, and trees have no cycles.

Where the trees come from. No broker is born knowing the topology. Underneath the publish-subscribe machinery, the brokers permanently run a routing protocol among themselves: the same two families used by internet routers. With distance vector, each broker exchanges distance estimates with its neighbours until it knows, for every destination, the distance and the next hop; with link state, each broker floods a description of its own links, so that everyone reconstructs the whole graph and can compute shortest paths on it. Either way, a source’s tree is never stored anywhere as a whole: each broker knows only its own fragment of it. Its parent in the tree of source $v$ is its next hop towards $v$; its children are the neighbours whose next hop towards $v$ is the broker itself. This local knowledge is all the strategies below need: PSF’s probes and forwarding follow these fragments, and PRF’s address-based delivery reuses the next-hop table directly.

Per-Source Forwarding (PSF). The tree is chosen by the sender: every broker $v$, seen as a potential source, defines its shortest path tree (SPT): the union of the minimal paths from $v$ to every other broker. Messages published at $v$ only ever travel along $v$’s tree. Concretely:

1. Route construction. Each broker announces itself with a probe that descends its own SPT: every broker that receives it passes it on to its children for that source. Subscriptions are then attracted back along the same tree, from the subscribers towards the source; each broker on the way records, for that source and for each of its children in that source’s tree, the predicate that joins the interests of all the subscribers reachable through that child.
2. The forwarding table is therefore indexed by (source, next hop). In the PSF figure below, broker 2 stores the rows “(source 1, next-hop 5): $P_g + P_d$” (read: a message originating at 1 that matches $P_g$ or $P_d$ continues towards 5) and “(source 1, next-hop 4): $P_f$”.
3. Forwarding. When a message from source 1 arrives, a broker looks up the rows for source 1, matches the message against each child’s predicate, and forwards it only along the matching children. Every hop follows 1’s tree, so no loop can form; every predicate summarises the interests beyond that child, so no branch is visited uselessly.

The price is state: conceptually, every broker keeps one section of forwarding table per possible source.

Improved PSF (iPSF). A refinement found of this algorithm is based on the observation that many of those per-source sections are identical. Two sources are indistinguishable for a broker $n$ when $n$ has the same children in both of their trees and reaches the same set of nodes through those children: messages from the two sources are forwarded identically, so their rows can collapse into one, listing a set of sources per row. In the iPSF figure below, sources 1 and 3 are indistinguishable for broker 2, whose table shrinks to “(sources 1 and 3, next-hop 5): $P_g + P_d$” and “(sources 1 and 3, next-hop 4): $P_f$”. The advantages: smaller forwarding tables, and cheaper route construction.

Per-Receiver Forwarding (PRF). The dual choice: the trees belong to the receivers, and content is matched once, at the entry point.

1. Subscribing. Each subscription is propagated to the whole network along the subscriber’s tree. Every broker therefore learns the interests of every broker in the network, kept per receiver rather than aggregated: in the PRF figure below, broker 2 knows precisely that 5 wants $P_g$, 8 wants $P_d$ and 7 wants $P_b + P_c$, where PSF’s broker 2 only knew that “something matching $P_g + P_d$ lies beyond 5”.
2. Matching at the first broker only. When a message is published, the first broker that receives it matches it against this full table, computes the complete set of interested brokers, and writes their addresses into the message header. In the figure, the message $(a, d)$ published at broker 1 gets the recipient set $\{2, 6, 8\}$.
3. Address-based delivery. From then on, content is never inspected again: each broker forwards the message using its ordinary unicast routing table (the mechanism the internet itself uses), and at each hop the recipient set in the header is partitioned among the next hops. In the figure, broker 1 sends one copy towards 2 carrying $\{2, 8\}$ (8 is reached via 2) and one towards 3 carrying $\{6\}$; broker 2 delivers locally and forwards $\{8\}$ towards 5. Unicast routes are loop-free, so cycles are harmless.

The price moves to the edge: the first broker performs all the matching and must maintain a view of the interests of the entire network.

The takeaway: cycles make content-based routing much more complex, which is precisely why acyclic topologies (or topic-based routing over a DHT) are preferred whenever possible.

7.8 Complex Event Processing (CEP)#

Standard pub-sub matches individual messages against individual subscriptions. Complex Event Processing (CEP) extends the model by letting the application inject rules into the brokers themselves that combine multiple events over time into new, higher-level complex events. It is the pub-sub analogue of the broker programs of section 6.4: application code running inside the middleware.

Consider a building with distributed temperature and smoke sensors, and water nozzles that must react to fire:

• Without CEP: a single component somewhere subscribes to both temperature and smoke events, correlates them centrally, and republishes fire events for the nozzles. All sensor data flows to the centre of the system, and fire events flow back out to the edge: inefficient and slow.
• With CEP: a rule is deployed into the broker network: if, within a 5-minute window, a temperature > 50 event and a smoke event are both observed, emit a fire event. The broker local to the affected room evaluates the rule as events arrive and itself emits fire, immediately and locally; there are still subscribers to fire, but no publisher: the event is synthesised by the middleware.

The benefits are lower latency (detection at the edge, no central round-trip), less network traffic (raw sensor data does not travel to a central processor), and simpler components (nozzles just subscribe to fire). CEP rule languages support temporal windows, event patterns (sequences, conjunctions, negations), and aggregations; systems differ widely in expressiveness and in how efficiently the engine evaluates complex rules. CEP systems (also found under the DSMS label, Data Stream Management Systems) are the precursors of the modern stream processing frameworks (Kafka Streams, Flink) covered in courses on large-scale data systems: in both cases, application code is injected into a pipeline of flowing messages to transform and combine them.

8. Stream-Oriented Communication#

8.1 When Time Determines Correctness#

Stream-oriented communication deals with data that is inherently time-dependent: video, audio, and similar continuous media organised as sequences of small items (frames, samples). In the paradigms seen so far, timing affects only performance: an event notification arriving in 100 ms instead of 1 ms is annoying but correct. In streaming, timing is part of correctness: if a video plays 30 frames per second and the next frame arrives after 100 ms instead of 33 ms, the user sees the video stutter or freeze; from the application’s point of view this is an error, not merely slowness.

Transmission modes. How strictly time constrains a stream defines three modes:

Most multimedia applications are synchronous or isochronous: they require an upper bound on delay (to avoid stalls) and sometimes a lower bound as well (to avoid flooding the receiver).

8.2 Simple and Complex Streams#

A simple stream carries a single medium (video only, audio only). A complex stream packs multiple sub-streams into one logical stream: video plus its audio channel, or stereo audio with left and right channels. The sub-streams of a complex stream must be kept in synchronisation, with tolerances that depend on the media:

Beyond these thresholds the user perceives the streams as out of sync (in the video-audio case, the loss of lip synchronisation).

8.3 Streaming Architecture and QoS on the Internet#

The typical architecture for streaming stored data: a multimedia server holds the compressed multimedia file; the client requests it piece by piece and presents it in real time, without downloading it entirely first. The non-functional requirements of such a service are usually expressed as Quality of Service (QoS) requirements: a required bit rate, a maximum session set-up delay, a maximum end-to-end delay, and a maximum delay variance (jitter). Ideally the network would accept a request like “give me a connection with 10 Mbit/s and delay ≤ 30 ms”; QoS-aware networks exist, but the internet is not one of them.

IP is a best-effort protocol: no guarantees on bandwidth, delay, or delivery. The IP header does include a Type of Service byte that can mark packets as higher-priority for routers’ per-hop behaviour, but many routers ignore it and no hard guarantee is ever provided. Streaming applications on the internet must therefore implement QoS control entirely at the application layer.

8.4 Why UDP, Not TCP#

Streaming applications are typically built on UDP. TCP’s error correction strategy, retransmission, is incompatible with real-time constraints: if a frame misses its deadline, retransmitting it is pointless, since by the time it arrives it is too late to display. Better to drop the missing frame and continue with the next one. UDP leaves the application free to implement error handling strategies appropriate for streaming.

8.5 Application-Layer QoS Techniques#

Four complementary techniques compensate for the network’s lack of guarantees. They assume the network provides the required bandwidth on average; nothing can be done if it does not, but fluctuations around that average can be absorbed.

1. Buffering. Instead of playing each packet on arrival, the client first accumulates a buffer of a few seconds of content, then starts playback from the buffer. Temporary gaps in the packet flow are absorbed as long as they are shorter than the buffered play time. This is why a YouTube video does not start the instant you press play: the sender starts transmitting immediately, but the client waits to fill its initial buffer. A larger buffer tolerates more jitter at the price of a longer start-up delay; the QoS control components at the two ends continuously renegotiate the buffer size based on observed network behaviour (shrinking it on a good network, enlarging it after problems).

2. Forward error correction (FEC). Backward error correction means returning to the last correct state and retrying, which is exactly what TCP does when it re-requests a lost packet. Forward error correction moves ahead instead: enough redundant information is added to the stream that a lost packet can be reconstructed (approximately) from its neighbours. A missing video frame is interpolated from the surrounding frames; the reconstruction is lower quality, but playback never stops, which is far preferable to a freeze followed by a skip.

3. Interleaving. If each network packet carries consecutive frames (frames 1-4 in packet 1, frames 5-8 in packet 2, …), losing one packet leaves a contiguous hole of several frames, which is very hard to reconstruct. Interleaving spreads frames non-consecutively across packets:

Packet 1: frames  1,  5,  9, 13
Packet 2: frames  2,  6, 10, 14
Packet 3: frames  3,  7, 11, 15
Packet 4: frames  4,  8, 12, 16

Losing one packet now costs isolated frames spread across the playback window, and FEC interpolation works far better on one missing frame in four than on four consecutive holes. Interleaving requires buffering, since the receiver must collect several packets before restoring the original frame order.

4. Adaptive quality. Modern platforms (e.g., YouTube) keep each content at multiple compression levels and resolutions, and the client buffers more than one quality level in parallel. During playback the client normally plays the highest quality; when the network degrades, it falls back to the lower-quality version already buffered (whose smaller size makes it deliverable even through the congestion) and returns to high quality when conditions improve. The user sees a temporary quality reduction rather than a stall.

8.6 Stream Synchronisation#

For complex streams, sub-streams must be synchronised, and this can happen at two points:

• At the sender: the sub-streams are interleaved and packed into the same network packets before transmission, so synchronisation holds by construction. This is the most common approach in practice.
• At the receiver: the sub-streams travel separately and the receiver realigns them, which requires buffering and explicit timing logic.

Receiver-side synchronisation, in turn, can live at different layers: entirely inside the application (e.g., a procedure that reads two audio data units for every video data unit), or in the middleware, with the application merely telling the multimedia control layer what to do with the incoming streams.

9. The Communication Paradigms at a Glance#

Read top to bottom, the table is the chapter’s arc: coupling loosens from message passing through the queue and the dispatcher, tightens again for the two remote-invocation paradigms that deliberately trade decoupling for a familiar call syntax, and stops being the relevant axis at all for streaming, where the binding constraint is time. Each paradigm occupies a different point in the design space, trading ease of programming, decoupling, performance, and expressiveness. The right choice follows from the architecture of the system being built: RPC/RMI for client-server and object-oriented designs, queuing for data-centred ones, pub-sub and CEP for event-based ones, and streaming wherever continuous media impose timing constraints.

10. Exam Questions#

Questions from past written exams that map onto this chapter. The instructor does not publish official solutions: the worked answers below are unofficial, reconstructed from the course material.

10.1 Streaming Library over IP#

10.2 Internet Radio#

10.3 Parameter Passing in RPC and RMI#

10.4 Publish-Subscribe vs. RPC, and Pub-Sub over a DHT#

10.5 Message Passing vs. Queuing vs. Publish-Subscribe#

10.6 Subscription Languages and JMS#

11. Glossary#

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