• Provide atomic operations at servers that maintain shared data for clients
• Provide recoverability from server crashes

Properties (ACID)

• Atomicity
• Consistency
• Isolation
• Durability

• Without concurrency control, we have lost updates, inconsistent retrievals, dirty reads, etc. 
• Concurrency control schemes are designed to allow two or more transactions to be executed correctly while maintaining serial equivalence
• Serial Equivalence is correctness criterion
• Schedule produced by concurrency control scheme should be equivalent to a serial schedule in which transactions are executed one after the other.

Scheme

• Locking
• Optimistic concurrency control
• Time-stamp based concurrency control

• (1) If the object is not already locked, it is locked and the operation proceeds
• (2) If the object has a conflicting lock set by another transaction, the transaction must wait until it is unlocked
• (3) If the object has a non-conflicting lock set by another transaction, the lock is shared and the operation proceeds
• (4) If the object has already been locked in the same transaction, the lock will be promoted if necessary and the operation proceeds
• When promotion is prevented by a conflicting lock, rule 2 is used

Strict Two-Phase Locking

Deadlock

Example

Resolution of Deadlock

Optimistic Concurrency Control

Drawback of locking

Optimistic Concurrency Control

Timestamp Based Concurrency Control

Concurrency Control for Distributed Transactions

The Two-Phase Commit Protocol

Three Phase Commit

• Problem with two-phase commit
• If coordinator crashes, participants cannot reach a decision, stay blocked until coordinator recovers
• Three-phase commit
• There is no single state from which it is possible to make a transaction directly to another COMMIT or ABORT state
• There is not state in which it is not possible to make a final decision, and from which a transaction to COMMIT can be made.

Update Propagation

• A comparison between push-based and pull-based protocols in the case of multiple client, single server system

• Update propagation for systems that only need eventual consistency
• Randomized approaches based on the theory of epidemics
• Infective, susceptible and removed servers
• Anti-entropy propagation model
• A server P picks another server Q at random, and exchanges updates
• Three approaches
• P only pushes updates to Q
• P only pulls new updates from Q
• P and Q send updates to each other
• If many infective servers, pull-based approach is better
• If only one infective server, eitehr approach will eventually propagate all updates
• Rumor spreading (Gossiping) will speed up propagation
• If server P has been updated, it randomly contacts Q and tries to push the update to Q, if Q was already updated by another server, with some probability (1/k), P loses interest in spreading the update any further

The Gossip System

• Guarantees
• Each client obtains a consistent service over time
• i.e., replica managers only provide a client with data that reflect the updates the client has observed so far
• Relaxed consistency between replicas
• Primarily causal consistency, but support also provided for sequential consistency
• Choice up to the application designer
• Procedure
• Request
• Front Ends sends a query or update request to a replica manager that is reachable
• Update Response
• RM replies as soon as it receives updates
• Coordination
• RM does not process the request until it can meet the required ordering constraints
• This may involve receiving updates from other replica managers in gossip messages
• Execution
• Query Response
• If the request is a query, the RM replies at this point
• Agreement
• The replica managers update each other by exchanging gossip messages, which contains the most recent updates they have received
• This is one in lazy fashion
• Query and update operations in a gossip service

• Front ends propagate their timestamps whenever clients communicate directly

• A gossip replica manager, showing its main state components

Bayou: Eventual Consistency

• If no updates take place for a long time, all replicas will gradually become consistent
• Domain specific conflict detection and resolution
• Appropriate for applications like shared calendars

Motivation for eventual consistency

• Sequential consistency requires that at every point, every replica has a value that could be the result of the global-agreed sequential application of writes
• This does not require that all replicas agree at all times, just that they always take on the same sequence of value
• Writes
• When they arrive, are applied in the same order at all replicas
• Easily done with timestamps

Conflict Resolution

• Dealing with inconsistency
• Every Bayou update contains a dependency check and a merger procedure in addition to the operation’s specification
• Replication not transparent to application
• Only the application knows how to resolve conflicts
• Split of responsibility
• Replication system
• Propagate updates
• Application
• resolve conflicts
• Optimistic application of writes require that writes be “undoable“

Rolling Back Updates

• Keep log of updates
• Order by timestamp
• When  a new update comes in, place it in the correct order and repapply log of updates
• Need to establish when you can truncate the log
• Requires old updates to be “committed”, new ones tentative
• Committed order can be achieved by designing a replica manager as the primary replica manager

• Front end issues the request
• Either sent to a single replica or multicast to all replicate managers
• Coordination
• Replica managers coordinate in preparation for the execution of the request
• i.e., agree is requests is to be performed and the ordering of the request relative to others
• Execution
• Agreement
• Reach consensus on effect of the request, e.g., agree to commit or abort in a transactional system
• Response

• Primary-based replication protocols
• Replicated-write protocols
• Active replication using multicast communication
• Quorum-based protocols

Primary-backup Model

• Front ends only communicate with primary

Procedures

• Request
• FE issues a request containing a unique identifier to the primary replica manager
• Coordination
• The primary takes each request in the order in which it receives it
• Execution
• The primary executes the request and store the response
• Agreement
• If the request is an update, the primary sends the updated state, the response and the unique id to all backups. The backups send an acknowledgement.
• Response
• The primary responds to the front end, which hands the response back to the client.

Implementation

• Implements linearizability if primary is correct, since primary sequences all the operations

Failures

• If primary fails, the system retains linearizabilty if a single back becomes the new primary and if the new system configuration takes over exactly where the last left off
• If primary fails, it should be replaced with a unique backup
• Replica managers that survive have to agree upon which operation has been performed when the replacement primayr is over
• Requirements met if replica managers organized as a group and if primary uses view-synchronous communication to propagate updates to backups.

Replicate-Write Protocol

Active replication

• Front end multicasts request to each replica using a totally ordered reliable multicast
• System achieves sequential consistency but not linearizability
• Total order in which replica managers process requests may not be the same as real-time order in which clients made request.

Implementing Ordered Multicast

• Incoming messages are held back in a queue until delivery guarantees can be met
• The hold-back queue for arriving multicast messages

• Coordinate all machines needed to determine delivery order
• FIFO-ordering
• Easy
• Use a separate sequence number for each process
• Total ordering
• Use a sequencer
• Distributed algorithm with three phases
• Causal order
• Use vector timestamps

The ISIS algorithm for total ordering

Causal Ordering using Vector Timestamps

Quorum-based Protocols

• Procedures
• Assign a number of votes to each replica
• Let N be the total number of votes
• Define R = read quorum, W = write quorum
• R+W > N
• W > N/2
• guarantee that no two writes at the same time 
• since if yes, than the vote for w_1 and w_2 are larger than N
• Only one writer at a time can achieve write quorum
• Every reader sees at least one copy of the most recent read (takes one with most recent version number)
• Examples

Scaling

• None of the protocols for sequential consistency scale
• To read or write, you have to either
• Contact a primary copy
• Use reliable totally ordered multicast
• Contact over half the replicas
• All this complexity is to ensure sequential consistency
• Even the protocols for causal consistency and FIFO consistency are difficult to scale if they use reliable multicast
• Can we weaken sequential consistency without losing some important features?

• Emphasis on giving clients access to the service with reasonable response time, even if some results do not conform to sequential consistency
• Examples
• Gossip
• Relaxed consistency
• Causal update ordering
• Bayou
• Eventual consistency
• Domain-specific conflict detection and resolution
• Coda (file system)
• Disconnected operation
• Use vector timestamp to detect conflicts

• Motivation

• Performance enhancement
• Enhanced availability/reliability
• Fault tolerance
• Scalability
• Tradeoff between benefits of replication and work required to keep replica consistent

• Requirements

• Consistency
• Depends upon applications
• In many applications, we want different clients making (read/write) requests to different replicas of the same logical items should not obtain different results
• Replica transparency
• Desirable for most application

2. Consistency Models

• Consistency model is a contract between processes and a data store
• If process follow certain rules, then the store will work correctly
• Needed for understanding how concurrent read and writes behave w.r.t shared data
• Relevant for shared memory multiprocessors
• Cache coherence algorithms
• Shared database, files
• Independent operations
• transactions

• Any read on a data item x returns a value corresponding to the result of the most recent write on x
• Challenge
• It requires absolute global time

Sequential Consistency

• The result of any execution is the same as if the read and write operation by all processes were executed in some sequential order and the operations of each individual process appear in this sequence in the order specified by its program.
• i.e., required to be seen by the same process in the same order
• Example

• Linearizability
• Definition of sequential consistency says nothing about time
• There is no reference to the “most recent” write operation
• Liearizability
• Weaker than strict consistency, stronger than sequential consistency
• Operations are assumed to receive a time stamp with a global available lock that is loosely synchronized
• The result of any execution is the same as if the operations by all processes on the data store were executed in some sequential order and the operations of each individual process appear in this sequence in the order specified by its program. 

Causal Consistency

• Writes that are potentially causally related must be seen by all processes in the same order. Concurrent writes may be seen in a different order on different machines.
• Example
• (a) is not casual-consistent, since in process P2, a performs before b, so that in process P3 and P4, a should also performs before b
• While in (b) there is not casual relationship between a and b

FIFO consistency

• Writes done by a single process are seen by all other processes in the order in which there were issues.
• But writes from different processes may be seen in a different order by different process.

Weak Consistency

• Access to synchronization variable associated with a data store are sequential consistent.
• No operation on a synchronization variable is allowed to be performed until all previous writes have been completed everywhere
• No read or write operation on data items are allowed to be performed until all previous operations to synchronization variable have been performed.

Release Consistency

• Before a read or write operation on a shard data is performed, all previous requires done by the process must have completed successfully.
• Before a release is allowed to be performed, all previous reads and writes by the process must have completed.
• Access to synchronization variables are FIFO consistent (sequential consistent is not required).

Entry Consistency

• The weak consistency models that use synchronization variable (release, entry consistency) are mostly relevant to shared multiprocessor systems
• Also modern CPU with multiple pipelines, out-of-order instruction execution, asynchronous writes, etc.
• In distributed systems, weak consistency typically refers to weaker consistency models than sequential consistency
• Casual consistency
• e.g., used in the Gossip system
• Optimistic approaches such as those used in Bayou, Coda that use application specific operations to achieve eventual consistency

5. Eventual Consistency

• Session Guarantees

• When clients move around and connects to different replicas, strange things can happen 
• Updates you just made are missing
• Database goes back in time
• Design choice
• Insist strict consistency
• Enforce some session guarantees, client-centric consistency

• Monotonic Reads

• If a process reads the value of a data item x, any successive read operation on x by that process will always return that same value on a more recent value.
• Disallow reads to a database less current than previous read
• Example error
• Get a list of email message, when attempts to read one, pop put “message does not exist”

• Monotonic Writes

• A write operation by a process on a data item x is completed before any successive write operation x by the same process
• Writes must follow any previous writes that occurred within their session

• Read your Writes

• A read operation by a process on a data item x is completed before any successive write operation on x by the same process.
• Every read in a session should see all previous writes in that session.
• Example error
• Deleted email message re-appear

• Writes Follow Reads

• A write operation by a process on a data item x following a previous read operation on x by the same process is guaranteed to take place on the same or a more recent value of x that was read.
• If a write W followed by a read R at a server X, ten at all other servers
• If W is in T’s database, then any writes relevant to R are also avaialble
• After users outside session
• Traditional write/read dependencies preserved at all servers
• Two guarantees: ordering and propagation
• Order
• If a read precedes a write in a session, then that read depends on a previous non-session write, then previous write will never be seen after second write at any server. It may not be seen at all.
• Propagation
• Previous write will actually have propagated to any data base where the second write is applied.

• Responsibility of session manager, not servers
• Two sets
• Read set
• set of writes that are relevant to session reads
• Write set
• Set of writes that performed in session
• Update dependencies captured in read-sets and write-sets
• Causal order of writes
• Use Lamport clocks

• A name in a distributed system is a string of bits or characters that is used to refer to an entity 
• Type of names
• Address
• The name of an access point of an entity
• Identifiers
• A name that uniquely identifies an entity
• Location-independent name
• A name that is independent from its addresses

Name Resolution

• A naming system maintains a name-to-address binding for resources
• Main problem in naming
• How to do name resolution in a distributed systems in a scalable way
• Approaches for name resolution closely tied to naming scheme
• Flat vs Structure naming
• In a flat name space, identifiers are random bit strings
• No information on location embedded in name

• Simple solutions that work in a LAN environment
• Broadcasting & Multicasting
• Message containing identifier of the entity is boardcast
• Machine with access point for the entity replies with the address of the access point
• ARP protocol for finding the data-link address of a machine given the IP address
• When network grows, multicast is more efficient
• Forwarding pointers
• When an entity moves from A to B, it leaves behind a reference to its new location at B
• Home-based Approaches
• Home agent keeps track of current location of mobile entity
• Hierarchical Approaches
• Similar to DNS

What is a DHT

• Distributed Hash Table
• key = hash(data)
• lookup(key) -> IP address
• send-RPC(IP address, PUT, key, value)
• Send-RPC(IP address, GET, key)  -> value
• Chord

Structure Naming Systems

• Names are organized into name spaces
• A name space can be represented as a labeled, directed graph wit two types of nodes
• Leaf nodes and directory nodes
• Absolute vs relative path names
• Local names vs global names
• Name resolution: the process of looking up a name
• Closure mechanism
• Knowing where and how to start name resolution
• Unix File Name Spaces

Linking and Mounting

• Symbolic link

• Mounting remote name spaces

Implementing Name Spaces

• Naming service
• A service that allows users and processes to add, remove, and lookup names
• Name spaces for large-scale widely distributed systems are typically organized hierarchically
• Three layers used to implement such distributed name spaces
• Global layer
• root node and its children
• Administrational layer
• Directory nodes within a single organization
• Managerial layer
• Example: DNS name space

Iterative vs Recursive Resolution

• Recursive

• puts a higher performance demand on each name server
• cons
• Too high for global layer name servers
• Pros
• Caching is more effective
• Communication costs may be reduces
• Example

• Iterative
• Example

Attribute-based Naming

• An entity is described in terms of (attribute, value) pairs
• Each entity can have several attributes
• Attribute-based naming service are called directory services
• LDAP (Lightweight directory access protocol) defacto industry standard
• Based on OSI X.500 directory service
• Another example: UDDI for web services

• Multithreaded clients
• Thread can block waiting for server response but application/process is not blocked
• Multithreaded servers
• Simplified server code as opposed to finite-state-machine approach that users non-blocking system calls
• Can handle multiple client requests in parallel (while making blocking ystem calls)
• Improved performance over iterative servers on multiprocessor systems

NFS Architecture

Google File System (GFS)

• Objects can be implemented in many different ways
• Compile-time objects
• e.g., instance of classes written in object-oriented language like Java, C++
• Data-base objects
• Procedural languages like C,with an appropriate “wrapper code” that gives it the appearance of an object
• System like Java RMI support compile-time objects
• Not possible or difficult in language-independent RMI middleware such as CORBA
• These systems use object adapters
• Implementations of object interfaces are registered at an object adapter, which acts as an intermediary between the client and the object implementation

Persistent vs. Transient Objects

• Persistent objects 
• continue to exist even if they are not contained in the address space of server process
• the “state” of a persistent object has to be stored on a persistent store, i.e., some second storage
• invocation requests result in an instance of the object being created in the address space of a running process
• many policies possible for object instantiation and (de)instantiation
• Transient objects
• Only exist as long as their container server process are running
• i.e., only exist in memory

Static vs Dynamic Remote Method Invocations

• Static invocation
• Typical ways for writing code that uses RMI is similar to the process for writing RPCC
• declare the interface in IDL, compile the IDL file to generate client and server stubs, link them to client and server side code to generate the client and the server executables
• requires the object interface to be known when the client is being developed
• Dynamic invocation
• The method invocation is composed at run-time
• invoke (object, method, input_parameters, output_parameters)
• Useful for applications where object interface are discovered at runtime
• e.g., object browser, batch processing systems for object invocations

Design Issues for RMI

• RMI invocation semantics
• Invocation semantics depend upon implementation of Request-Reply protocol used by RMI
• Could be Maybe, At-least-once, At-most-once

• Transparency
• Should remote invocations be transparent to the programmer?
• Partial failure, higher latency
• Different semantics for remote objects, e.g., difficult to implement “cloning” in the same way for local and remote objects or to support synchronization operations e.g., wait/notify
• Current consensus
• Access transparency
• Remote invocations should be made transparent in the sense that syntax of a remote invocation is the same as the syntax of local invocation
• Distinguish
• But programmers should be able to distinguish between remote and local objects by looking at their interfaces, 
• e.g., in Java RMI, remote objects implement the Remote interface

Implementing Issues for RMI

• Parameter Passing
• Representation of a remote object referece

• Request/Reply protocol
• Handling failures at client and/or server
• Issues in marshaling of parameters and results
• Input, output, inout parameters
• Data representation
• handling reference parameters
• Distributed object references
• handling failures in request-reply protocol
• Partial failure
• Client, server, network
• Supporting persistent objects, object adapters, dynamic invocations, etc

Marshalling

• Client unable to locate server
• Reasons
• Server has crashes
• Server has moved
• (RPC systems) client compiled using old version of service interfance
• System must report error (remote exception) to client 
• Loss of transparency
• Request message lost
• Retransmit a fixed number of times before throwing an exception
• Reply message lost
• Client resubmits request
• Server choices
• Re-execute procedure
• Server should be idempotent so that it can be repeated safely
• Filter duplicates
• Server should hold on to results until ackowledged
• Server crashes after receiving a request
• At least once
• Keep trying till server comes up again
• At most once
• Return immediately
• Exactly once impossible to achieve
• Client crashes after sending a request
• If a client crashes before RPC returns, we have an “orphan” computation at server
• Waste resources, could also start other comutations
• Orphan detection
• Reincarnation
• Client broadcasts new epoch when it comes up again
• Expiration
• RPC has fixed amount of time to do work

Note

• Implementing the request-reply protocol on top of TCP
• Does not help in providing applications with different invocation semantics
• TCP does not help with server crashes
• If a connection is broken, the end points do not have any guarantees about the delivery of messages that may have been in transit

RMI Software Components

• Communication module
• Implements the request-reply protocol
• Remote reference module
• Responsible for translating between local and remote object references and for creating remote object references
• Maintains remote object table that maintains a mapping between local&remote object references
• E.g., Object Adapter in CORBA

• Activation of remote objects
• Some applications require that information survive for long periods of time
• However, objects not in user all the time, so keeping them in running processes is a potential waste of resources
• Object can be activated on demand
• E.g., standard TCP services such as FTP on UNIX machines are activated by inetd
• Active and passive objects
• Active objects
• Instantiated in a running processes
• Passive objects
• Not currently active but can be made active
• Implementation of its methods, and marshalled state stored on disk
• Activator responsible for
• Registering passive objects that are available for activation
• Starting named server processes and activating remote objects in them
• Keeping track of locations of servers for mote objects that it has already activated
• Examples
• CORBA implementation repository
• JAVA RMI has once activator on each server computer

RMI – Other Topics

• Persistent object stores
• An object that is guaranteed to live between activations of process is called a persistent object
• Stored the state of an object in a marshalled (serialized) form on disk
• Location service
• Objects can be migrated from one system to another during their lifetime
• Maintains mapping between object references and the location of an object
• Distributed Garbage Collection
• Needed for reclaiming space on servers
• Passing “behavior”
• Java allows objects (data+code) to be passed by value
• If the class for an object passed by value is not present in a JVM, its code is downloaded automatically
• Use of reflection in Java RMI
• Allows construction of generic dispatcher and skeleton

• Java approach based on reference counting
• Each server process maintains a list of remote processes that hold remote object references for its remote objects
• When a client first acquires a remote reference to an object, it make addRef() invocation to server before creating a proxy
• When a clients local garbage collector notices that a proxy is no longer reachable, it makes a removeRef() invocation to the server before deleting the proxy
• When the local garbage collector on the server notices that the list of client processes that have a more reference to an object is empty, it will delete the object (unless there are any local objects that have a reference to the object)
• Other approaches
• Evictor pattern
• Leases

• Integrate with Java language and libraries
• Security, write once run anywhere, multithreaded
• Object oriented
• Can pass “behavior”
• Mobile code
• Not possible in CORBA, traditional RPC systems
• Distributed garbage collection
• Remoteness of objects intentionally not transparent
• Good for handling failures

Remote Interfaces, Objects, and Methods

• Object becomes remote by implementing a remote interface
• A remote interface extends the interface java.rmi.Remote
• Each method of the interface declares java.rmi.RemoteException in its throws clause in addition to any application-specific clauses

Creating distributed applications using RMI

1. Define the remote interfaces
2. Implement the remote objects and server
3. Implement the client
4. Compile the remote interface, server and client 
5. Generate the stub and skeleton using rmic
6. Start the RMI registry
7. Start the server
8. Run the client