Decomposing Transactional Systems

A client transaction starts by contacting one of the Proxies to obtain a read version (i.e., a timestamp). The Proxy then asks the Sequencer for a read version that is guaranteed to be no less than any previously issued transaction commit version, and this read version is sent back to the client. Then the client may issue multiple reads to StorageServers and obtain values at that specific read version. Client writes are buffered locally without contacting the cluster. At commit time, the client sends the transaction data, including the read and write sets (i.e., key ranges), to one of the Proxies and waits for a commit or abort response from the Proxy. If the transaction cannot commit, the client may choose to restart the transaction from the beginning again.

A Proxy commits a client transaction in three steps. First, the Proxy contacts the Sequencer to obtain a commit version that is larger than any existing read versions or commit versions. The Sequencer chooses the commit version by advancing it at a rate of one million versions per second. Then, the Proxy sends the transaction information to range-partitioned Resolvers, which implement FDB’s optimistic concurrency control by checking for read-write conflicts. If all Resolvers return with no conflict, the transaction can proceed to the final commit stage. Otherwise, the Proxy marks the transaction as aborted. Finally, committed transactions are sent to a set of LogServers for persistence. A transaction is considered committed after all designated LogServers have replied to the Proxy, which reports the committed version to the Sequencer (to ensure that later transactions' read versions are after this commit) and then replies to the client. At the same time, StorageServers continuously pull mutation logs from LogServers and apply committed updates to disks.

Like Bigtable, writes that occur in a transaction are buffered at the client until commit. As a result, reads in a transaction do not see the effects of the transaction’s writes. This design works well in Spanner because a read returns the timestamps of any data read, and uncommitted writes have not yet been assigned timestamps.

Reads within read-write transactions use wound-wait to avoid deadlocks. The client issues reads to the leader replica of the appropriate group, which acquires read locks and then reads the most recent data. While a client transaction remains open, it sends keepalive messages to prevent participant leaders from timing out its transaction. When a client has completed all reads and buffered all writes, it begins two-phase commit. The client chooses a coordinator group and sends a commit message to each participant’s leader with the identity of the coordinator and any buffered writes. Having the client drive two-phase commit avoids sending data twice across wide-area links.

A non-coordinator-participant leader first acquires write locks. It then chooses a prepare timestamp that must be larger than any timestamps it has assigned to previous transactions (to preserve monotonicity), and logs a prepare record through Paxos. Each participant then notifies the coordinator of its prepare timestamp.

The coordinator leader also first acquires write locks, but skips the prepare phase. It chooses a timestamp for the entire transaction after hearing from all other participant leaders. The commit timestamp s must be greater or equal to all prepare timestamps (to satisfy the constraints discussed in Section 4.1.3), greater than TT.now().latest at the time the coordinator received its commit message, and greater than any timestamps the leader has assigned to previous transactions (again, to preserve monotonicity). The coordinator leader then logs a commit record through Paxos (or an abort if it timed out while waiting on the other participants).

Before allowing any coordinator replica to apply the commit record, the coordinator leader waits until TT.after(s), so as to obey the commit-wait rule described in Section 4.1.2. Because the coordinator leader chose s based on TT.now().latest, and now waits until that timestamp is guaranteed to be in the past, the expected wait is at least 2 * epsilon. This wait is typically overlapped with Paxos communication. After commit wait, the coordinator sends the commit timestamp to the client and all other participant leaders. Each participant leader logs the transaction’s outcome through Paxos. All participants apply at the same timestamp and then release locks.

We begin with TAPIR’s protocol for executing transactions.

1. For Write(key, object), the client buffers key and object in the write set until commit and returns immediately.

2. For Read(key), if key is in the transaction’s write set, the client returns object from the write set. If the transaction has already read key, it returns a cached copy. Otherwise, the client sends Read(key) to the replica.

3. On receiving Read, the replica returns object and version, where object is the latest version of key and version is the timestamp of the transaction that wrote that version.

4. On response, the client puts (key, version) into the transaction’s read set and returns object to the application.

Once the application calls Commit or Abort, the execution phase finishes. To commit, the TAPIR client coordinates across all participants — the shards that are responsible for the keys in the read or write set — to find a single timestamp, consistent with the strict serial order of transactions, to assign the transaction’s reads and writes, as follows:

1. The TAPIR client selects a proposed timestamp. Proposed timestamps must be unique, so clients use a tuple of their local time and their client id.

2. The TAPIR client invokes Prepare(txn, timestamp) as an IR consensus operation, where timestamp is the proposed timestamp and txn includes the transaction id (txn.id) and the transaction read (txn.read set) and write sets (txn.write set). The client invokes Prepare on all participants through IR as a consensus operations.

3. Each TAPIR replica that receives Prepare (invoked by IR through ExecConsensus) first checks its transaction log for txn.id. If found, it returns PREPARE-OK if the transaction committed or ABORT if the transaction aborted.

4. Otherwise, the replica checks if txn.id is already in its prepared list. If found, it returns PREPARE-OK.

5. Otherwise, the replica runs TAPIR’s OCC validation checks, which check for conflicts with the transaction’s read and write sets at timestamp, shown in Figure 8.

6. Once the TAPIR client receives results from all shards, the client sends Commit(txn, timestamp) if all shards replied PREPARE-OK or Abort(txn, timestamp) if any shards replied ABORT or ABSTAIN. If any shards replied RETRY, then the client retries with a new proposed timestamp (up to a set limit of retries).

7. On receiving a Commit, the TAPIR replica: (1) commits the transaction to its transaction log, (2) updates its versioned store with w, (3) removes the transaction from its prepared list (if it is there), and (4) responds to the client.

8. On receiving a Abort, the TAPIR replica: (1) logs the abort, (2) removes the transaction from its prepared list (if it is there), and (3) responds to the client.

The essence of Calvin lies in separating the system into three separate layers of processing:

• The sequencing layer (or “sequencer”) intercepts transactional inputs and places them into a global transactional input sequence—this sequence will be the order of transactions to which all replicas will ensure serial equivalence during their execution. The sequencer therefore also handles the replication and logging of this input sequence.

• The scheduling layer (or “scheduler”) orchestrates transaction execution using a deterministic locking scheme to guarantee equivalence to the serial order specified by the sequencing layer while allowing transactions to be executed concurrently by a pool of transaction execution threads. (Although they are shown below the scheduler components in Figure 1, these execution threads conceptually belong to the scheduling layer.)

• The storage layer handles all physical data layout. Calvin transactions access data using a simple CRUD interface; any storage engine supporting a similar interface can be plugged into Calvin fairly easily.

Client interaction with masters is generally the same as it would be without CURP. Clients send update RPC requests to masters. If a client cannot receive a response, it retries the update RPC. If the master crashes, the client may retry the RPC with a different server.

For 1 RTT updates, masters return to clients before replication to backups. To ensure durability, clients directly record their requests to witnesses concurrently while waiting for responses from masters. Once all f witnesses have accepted the requests, clients are assured that the requests will survive master crashes, so clients complete the operations with the results returned from masters.

A witness accepts a new record RPC from a client only if the new operation is commutative with all operations that are currently saved in the witness. If the new request doesn’t commute with one of the existing requests, the witness must reject the record RPC since the witness has no way to order the two noncommutative operations consistent with the execution order in masters. For example, if a witness already accepted “x←1”, it cannot accept “x←5”.

Each of f witnesses operates independently; witnesses need not agree on either ordering or durability of operations. In an asynchronous network, record RPCs may arrive at witnesses in different order, which can cause witnesses to accept and reject different sets of operations. However, this does not endanger consistency. First, as mentioned in §3.2.1, a client can proceed without waiting for sync to backups only if all f witnesses accepted its record RPCs. Second, requests in each witness are required to be commutative independently, and only one witness is selected and used during recovery (described in §3.3).

The role of masters in CURP is similar to their role in traditional primary-backup replications. Masters in CURP receive, serialize, and execute all update RPC requests from clients. If an executed operation updates the system state, the master synchronizes (syncs) its current state with backups by replicating the updated value or the log of ordered operations.

In the read phase, the DBMS maintains a separate read set and write set of tuples for each transaction. During this phase, accessed tuples are copied to the read set and modified tuples are written to the write set, which is only visible to the current transaction. Each entry in the read or write set is encoded as {tuple, data, wts, rts}, where tuple is a pointer to the tuple in the database, data is the data value of the tuple, and wts and rts are the timestamps copied from the tuple when it was accessed by the transaction. For a read set entry, TicToc maintains the invariant that the version is valid from wts to rts in timestamp order.

The first step of the validation phase is to lock all the tuples in the transaction’s write set in their primary key order to prevent other transactions from updating the rows concurrently. Using this fixed locking order guarantees that there are no deadlocks with other transactions committing at the same time.

The second step in the validation phase is to compute the transaction’s commit timestamp from the timestamps stored within each tuple entry in its read/write sets. For a tuple in the read set but not in the write set, the commit timestamp should be no less than its wts since the tuple would have a different version before this timestamp. For a tuple in the transaction’s write set, however, the commit timestamp needs to be no less than its current rts + 1 since the previous version was valid till rts.

In the last step, the algorithm validates the tuples in the transaction’s read set. If the transaction’s commit_ts is less than or equal to the rts of the read set entry, then the invariant wts ≤ commit_ts ≤ rts holds. This means that the tuple version read by the transaction is valid at commit_ts, and thus no further action is required. If the entry’s rts is less than commit_ts, however, it is not clear whether the local value is still valid or not at commit_ts. It is possible that another transaction has modified the tuple at a logical time between the local rts and commit_ts, which means the transaction has to abort. Otherwise, if no other transaction has modified the tuple, rts can be extended to be greater than or equal to commit_ts, making the version valid at commit_ts.

Finally, if all of the tuples that the transaction accessed pass validation, then the transaction enters the write phase. In this phase the transaction’s write set is written to the database.