Consistent Models and Messaging Systems

Message Oriented Middleware (MOM)

• Message-oriented middleware (MOM) is a type of communication middleware that facilitates asynchronous communication between processes.
• A distributed system consists of a collection of processes that are spatially separated and communicate through message exchanges.
• An atomic bit string is a message. Its format and significance are explained by a communications protocol.
• The transport of a message from its source to destination is performed by a computer network
• Internet protocols:
  • Application Transport: Specific communication services between 2 or more processes
  • Network Interface: Communication between 2 computers not directly connected or directly connected.
• Properties of the Internet Transport Protocols:
  • UDP: Message, Connectionless, Non-reliable, No order, No flow control, Variable number of recipients
  • TCP: Stream, Connection-based, Reliable, Ordered, Flow control, 1 recipient

Distributed System

• A distributed system encompasses a collection of geographically dispersed processes that communicate by exchanging messages. A crucial aspect of distributed systems is the message transmission delay, which must be considered in comparison to the timeframe between events within a single process.

Internet Protocols

• Communication channels provided to an application depend on the transport protocol (e.g., UDP or TCP).
• The design of distributed applications depends on the characteristics offered by the chosen transport protocol.

TCP Reliability

• TCP reliability (message loss):
  • Network hardware misconfigurations or failures
  • Unplugged cables
  • Damaged cables (e.g., by road works, shark bites)
• TCP guarantees that the application will be notified if the local endpoint is unable to communicate with the remote end.
• Error codes (e.g., ENOTCONN) are returned.
• The application is responsible for handling data loss.
• Web browsers typically report the problem to the user and try to reconnect again.
• TCP itself doesn't retransmit lost data in other connections.

TCP Reliability (message duplication)

• Retransmission of potentially already delivered messages isn't always necessary.
• TCP is not equipped to filter data duplication originating from applications.
• Duplicated data segments could pose problems for non-idempotent operations (e.g. credit/debit transactions or purchase orders).

Remote Procedure Call (RPC)

• A mechanism for invoking procedures on a different machine.
• Local procedure call: A procedure call that takes place within a single machine.
• Remote procedure call: A procedure that takes place on a different machine.

Remote Procedure Call (RPC) System Architecture

• Client-side process and server-side process. The communication is handled by the RPC's core, which is usually implemented atop of a transport protocol like TCP/IP. It enables a client process to interact with a server.

Client Stub

• 1. Assembles message: parameter marshalling.
• 2. Sends message via write or sendto to server
• 3. Blocks waiting for a response via read or recvfrom.
• Not in synchronous RPC

Server Stub

• 1. Receives message with request
• 2. Parses message to determine arguments
• 3. Calls function
• 1. Assembles message with return value function
• 2. Sends message via write or sendto
• 3. Blocks for a new request

RPC Advantages

• Distributed applications are simplified. RPC has limitations

Asynchronous Communication

• Communication where sender and receiver don't need to be active simultaneously.
• Used to avoid delays caused by sender/receiver synchronization.
• Suitable for loosely coupled systems.

Message-Oriented Middleware (MOM)

• Asynchronous message-based communication
• Message storage is handled by middleware.
• The service resembles a postal service with varying guarantees, such as order and reliability.
• MOM provides similarities to message fora/news groups.

MOM: Basic Patterns

• Point-to-point
• Publish-subscribe:
  • Publishers put messages into a topic.
  • Subscribers read messages from a topic; one message can be received by many subscribers.
  • Unlike queues, messages are delivered to more than one process (subscriber).

Messaging Service Implementation

• Asynchronous communication is facilitated by a messaging service.
• There can be communication servers.
• Synchronization needed at the communication level to ensure it operates correctly.

Messaging Service Applications

• Enterprise application integration (EAI)
• Workflow applications
• Microservices
• Message-based communication between people (e.g., email, SMS)
• Real-time messaging

Java Message Service (JMS)

• API for MOM, primarily designed for J2EE (now Jakarta EE)
• Enables Java applications to access MOMs in a portable manner.
• The functionalities offered by well-known MOM providers (e.g., IBM MQSeries, TIBCO) are encapsulated and presented.
• Provides integration options with Java Transaction Service
• Re-placed by Jakarta Messaging, but JMS is still in use in some documentation.
• Includes the API for queues and topics.

JMS Architecture and Model

• Two types of destinations used in JMS:
• Queues: single-destination communication
• Topics: multi-destination communication
• Two important components:
• JMS Provider (MOM service implementation)
• JMS Client (application sends/receives messages)

JMS Messages

• Headers: essential fields for identification and routing.
• Properties: optional details (metadata).
• Body: the actual data exchanged.

JMS Queues

• The Queue model:
  • Several senders can store messages in a queue.
  • Several receivers collect messages from a queue.
  • A message is only delivered to one receiver.
  • This improves scalability.
  • Queues have a long lifespan and are administered, not created by clients.

JMS Queues: Communication Semantics

• Blocking vs non-blocking
• Sends() and receive() operations;
  • Blocking: blocks until the message is sent.
  • Non-blocking (via timeout): specify timeout
  • Asynchronous (via callback): response via callbacks
• Reliability depends on delivery mode, and other factors (e.g., non-volatile storage).
• Persistent (once-and-only-once semantics).
• Non-persistent (at-most-once semantics)

JMS Queues: Persistent Delivery Implementation

• Difficulty with maintaining consistency and delivery across failures.
• Challenges of message loss or duplication during sender-server-receiver transmissions.
• The Importance of consumer acknowledgment, for correct delivery semantics in a failure-prone environment.
• There are different modes:
• AUTO_ACKNOWLEDGE, DUPS_OK_ACKNOWLEDGE
• CLIENT_ACKNOWLEDGE.

JMS Queues: Persistent Delivery Implementation (Issue)

• Problems when the server recovers (message loss or acknowledgment loss).

JMS Queues: Persistent Delivery Implementation (Solution)

• Resending messages with uncertainty using the JMSRedelivered header and the JMSXDeliveryCount property.

JMS Queues: Once-and-only-once Guarantees

• JMS's once-and-only-once delivery isn't "exactly once"; instead, duplicates may be delivered.
• Apache Kafka, on the other hand, is claimed to support exact-once delivery due to its use of logs, durable storage, and offsets for messages positioning.

JMS Queues: Session-based Transactions

• Ensures atomicity of send() /receive() operations.
• Guarantees that either all operations within a transaction complete or none of them.
• Commit(): confirms successful transaction
• Rollback(): undoes unsuccessful transaction

JMS Queues: Distributed Transactions

• Guarantees the atomicity of all transactions across the system, even when dealing with concurrent operations and process failures
• Needs to use Java Transaction API
• Ensures consistency even if a server crash or a network fault occurs

JMS Queues: Message Order

• Message order is guaranteed mostly within the same session's requests.
• Message sorting based on priority, delivery time and message selectors may override this.

JMS Topics

• Delivers messages to zero or more subscribers.
• Publishers send messages to a topic.
• Subscribers receive messages from a topic.

JMS Topics: Sending and Receiving Messages

• The API structure for topics is similar to the queue structure.
• The send() and receive() functions allow for handling of various message types similar to the queue structure
• Message selectors conditions can be used to filter messages received by subscribers.

JMS Topic Subscription

• Unshared: Only one active consumer
• Shared: More than one active consumer, all consumers receive all messages from a topic
• Duplicate-free delivery to each consumer
• Durable vs. Non-durable Subscribers
• Durable : the subscription persists even if subscribers are unavailable
• Non-durable: subscription exists only while there's a subscriber connected to the topic

JMS Topic Subscription and Reliability

The reliability of messaging depends on the message's delivery mode and subscriber's durability

JMS Topic Message Consumption Order

• The ordering of message delivery in topics is similar to the one for queues.
• Delivery order might differ from topic to topic (e.g., with different delivery modes or other factors)
• Time-dependent message delivery ordering, and is outside the client's control

JMS...

• An API for message-oriented middleware (MOM).
• Allows Java applications to access MOM in a portative manner.
• Includes functionalities for enterprise applications.

JMS Architecture

• JMS supports two types of destinations: queues (single-destination) and topics (multi-destination)

JMS Messages

• JMS messages consist of three parts: header, properties, body

JMS Queues

• A queue is a long-lived messaging destination managed by an administrator.

JMS Queues: Communication Semantics

JMS queues have blocking, non-blocking, and asynchronous send/receive semantics.

JMS Topics

• Topics support a publish-subscribe model.

MOM: Basic Patterns

• Point-to-point
• Publish/subscribe

JMS Messages

• Header
• Properties
• Body

Internet Protocols

• Application Transport
• Network Interface

Message Queuing Protocols

• AMQP: Advanced Message Queuing Protocol, an open standard protocol approved by OASIS.
• MQTT: A transport protocol for industrial applications, also an OASIS transport protocol.
• OpenWire: A public protocol for Apache ActiveMQ, which provides a JMS API.
• ActiveMQ supports AMQP, MQTT and other protocols.

Architecture

• Larger scale systems use message relays. E.g. , if applications/services run on different data centers

Message Brokers

• MOM is often used in the integration of applications.
• Different applications might have distinct syntax for messages, thus requiring a broker to translate between them.

Further Reading

• van Steen and Tanenbaum, Distributed Systems, 3rd Edition

Replication and Consistency Models

• Replicating data across many nodes offers:
• Enhanced performance (local reads)
• Increased reliability (avoiding data loss if a single replica is lost)
• Improved scalability (distributed load balancing across the nodes)
• Data availability (if not all replicas fail)
• Challenges include data consistency and concurrency if there are multiple updates made on the replicated data

Replication

• Updates to multiple replicas require careful coordination to maintain consistency.
• Protocols for synchronization needed to maintain consistency

Strong Consistency Models

• Sequential consistency is consistent if every execution of operations by the system can be equivalent to some serial execution of operations.
• Serializability: A collection of operations is serializable if the outcomes of the concurrent execution of these operations are similar to the outcome of some sequential execution sequence of these operations
• Linearizability: A system's operations are linearizable if the outcome of concurrent operations is equivalent to some serial execution of these operations.

Sequential Consistency Model

• An execution is sequentially consistent if it is equivalent to some serial execution of all the operations in that execution.
• Operations are executed in a certain order by each thread
• Used by multi-threaded systems.
• Not composable, as shown by examples with two arrays

Linearizability

• An execution of operations is linearizable if it's equivalent to some ordering where each operation has a start and end time on the timeline.
• Necessary that communication delays be predictable
• Critical for applications relying on a total order.

One-copy Serializability

• Essential for transaction-based systems. It focuses on transactions ensuring the outcome of concurrent transactions is equivalent to the order of some sequential execution of these operations.

Weak Consistency Models

• Improves scalability and availability, often at the expense of strict consistency.

Further Reading

• Fekete and Ramamritham (2010): Consistency Models for Replicated Data
• van Steen and Tanenbaum (Distributed Systems, 3rd Ed.): data-centric and client-centric consistency models.

Replication for Fault Tolerance

• Quorum Consensus protocols guarantee that a majority of replicas must agree

Quorum Consensus Protocols

• A quorum is a subset of replicas.
• Operations (e.g., reading, writing) require a quorum of replicas to complete successfully.
• Quorums help to enforce consistency, while allowing a certain degree of fault tolerance.

Read-Write Quorums

• Read Quorums and Write quorums are needed in order to maintain system consistency in a fault-tolerant environment.
• Must overlap such that common replicas exist in the sets involved
• Different quorum sizes give different trade-offs between consistency and availability

Quorum Consensus Implementation

• Read operation requires querying a quorum of replicas to find the current version and read the object value from an up-to-date replica.
• Write operation requires sending the new value to a write quorum; all replicas update the value.

Naïve Implementation with Faults

• Failures in partitions might cause the operations to be performed only on a subset of replicas, resulting in inconsistent data

Ensuring Consistency with Transactions

• Transactions, like the two-phase commit, are used to resolve data inconsistencies in multi-process environments

Playing with Quorums (1/2)

• The possibility of trade-offs between performance and availability depending upon the number of replicas in a quorum
• Assigning weights to replicas allows for different quorum sizes for read and write operations

Playing with Quorums (2/2)

• Weighted voting systems allow for variable weights assigned to the replicas in quorums; this can be used to tune availability and performance tradeoffs

Dynamo Quorums

• Dynamo, an Amazon-developed key-value storage system; provides high availability by enabling the use of quorums across a large number of servers.
• Data is replicated across many nodes
• Dynamo uses a preference list to determine which nodes are responsible for different data items
• Each operation has a "coordinator" node from the preference list that manages the actions; the rest of the replicas may be involved
• Data consistency is less strict than in Gifford's quorum approach– it uses "sloppy" or "consistent" quorums based on need

Dynamo's "Sloppy" Quorums and Hinted Handoff

• Allows the system to continue functioning despite replica failures
• Creates a backup copy of data in backup replicas, which is used when the primary replicas are unavailable
• When primary replicas recover the backup replicas automatically transfer their data

XA-based Quorum Consensus Implementation

• Each object's access is performed in the context of a transaction.
• The Read and Write operations need to know the current state and which replicas are up-to-date.
• Data modification is done only for a quorum.
• This guarantees atomic operations.

Transaction-based Quorum Consensus Replication

• Transactions offer improvements in consistency and support for more complex operations. However, possible deadlocks need to be considered

Quorum-Consensus and Replicated Abstract Data Types

• Herlihy generalizes quorum consensus concepts to ADTs (abstract data types), including data structures like queues. This technique employs initial and final quorum definitions to govern read and write operations; one suitable example is the queue.

Herlihy's Replicated Queue (example, execution, tracings)

• This technique makes sure that the operations (including read and write operations) within a transaction are executed concurrently and atomically.
• The outcome of the transaction must be equivalent to the order of some sequential execution of these operations

Herlihy's Replication Method

• Use of timestamps instead of version numbers.
• Reduction of quorum constraints.
• Reduction in message size when handling updates.
• Using message logs instead of state versions

Replicated Read/Write Objects with Time Stamps

• Using timestamps for read/write consistency rather than maintaining version numbers.
• Ensures a total order consistent with an omniscient observer.
• Client only needs to write to a final quorum for the entire state changes, and not just specific bits or locations.

Replicated Event Logs vs Replicated State

• Using event logs for replication instead of maintaining full states.
• Event logs are sequences of timestamped events.

Herlihy's Replicated Queue: Constraints

• Initial and final quorum must overlap.
• These constraints ensure that the queue maintains consistent state across replicas.

Herlihy's Replicated Queue: Optimizations

• Using a horizon time stamp for queue execution.
• Reducing the amount of data transmitted while processing dequeue requests

Issues with Replicated ADTs

• Timestamp generation for consistency across clients.
• Handling state changes when initial quorums are empty.

(Herlihy's) Replicated ADTs vs. CvRDTs

• Differences:
• CRDTs don't ensure strong consistency but strong eventual consistency.
• Replicated ADTs require a quorum to perform an operation.

Quorum Consensus: Final Thoughts

• Quorum-based systems are frequently restricted to "simple" data storage systems.
• Herlihy's generalization extends these approaches to other abstract data types as well.
• Techniques like majority voting are critical.

Further Reading

• Research papers by Leslie Lamport, Maurice Herlihy, et al., on distributed consensus, reliable broadcast, and state machine replication.

Blockchain

• A distributed, immutable ledger system recording transactions across a network of computers.

Bitcoin Blockchain

• Stores a complete history of all transactions.

Bitcoin Proof-of-Work

• A consensus protocol in Bitcoin. Nodes solve a computational puzzle (cryptographic hash) to add new blocks to the chain.

Bitcoin Forks

• Fork occurs when a disagreement arises about the next block in the chain.

Bitcoin Scalability and Energy Consumption

• Challenges in handling transactions per second, block sizes and storage.

Proof-of-Stake

• An alternative to Proof-of-Work
• Uses a lottery to select the next block proposer based on "coinage".

Permissioned Blockchains

• Restrict access to the blockchain network.

Further Reading

• Papers and resources on Bitcoin, including Nakamoto's original Bitcoin whitepaper.

Other Resources

• General Information regarding blockchain technology
• Resources on the CAP theorem
• Resources on various distributed consensus algorithms, like Paxos, and other distributed agreement protocols.

This quiz explores the differences between weak and strong consistency models, focusing on their advantages and implications in transaction-based systems. It also covers message delivery mechanisms and the roles of components in JMS message delivery. Test your knowledge on these essential concepts in distributed systems!