Source-linked AI summary
Using Paxos to Build a Scalable, Consistent, and Highly Available Datastore
Jun Rao, Eugene J. Shekita, Sandeep Tata
TL;DR
Spinnaker addresses consistent, highly available replication for a partitioned datastore running on commodity servers in one datacenter. It integrates Paxos with logging and recovery, preserving majority-based availability across failure sequences while remaining competitive with eventually consistent alternatives, with scope limited to single-operation transactions and single-datacenter deployment.
Problem
Master-slave replication can become unavailable or lose committed writes under certain failure sequences, while consistent replication across three or more replicas is difficult.
Method
Spinnaker uses a Paxos-based replication protocol integrated with commit logging and recovery for a range-partitioned, 3-way-replicated datastore.
Results
Compared to Cassandra, Spinnaker is as fast or faster on reads and 5% to 10% slower on writes.
Takeaways & Limitations
Paxos replication can provide majority-based read and write availability across failure sequences with performance competitive with weaker-consistency alternatives.
Takeaways & Limitations
Spinnaker is designed for a single datacenter and currently executes each API call as a single-operation transaction.
Abstract
from arXiv · showhide
Spinnaker is an experimental datastore that is designed to run on a large cluster of commodity servers in a single datacenter. It features key-based range partitioning, 3-way replication, and a transactional get-put API with the option to choose either strong or timeline consistency on reads. This paper describes Spinnaker's Paxos-based replication protocol. The use of Paxos ensures that a data partition in Spinnaker will be available for reads and writes as long a majority of its replicas are alive. Unlike traditional master-slave replication, this is true regardless of the failure sequence that occurs. We show that Paxos replication can be competitive with alternatives that provide weaker consistency guarantees. Compared to an eventually consistent datastore, we show that Spinnaker can be as fast or even faster on reads and only 5% to 10% slower on writes.
1. INTRODUCTION
Spinnaker addresses the need to scale and remain available on commodity-server clusters by combining partitioning, replication, and configurable consistency. Its Paxos-based replication is designed to preserve availability under arbitrary failure sequences while remaining competitive with weaker-consistency alternatives.
- Motivation: Sharding distributes data across commodity servers, but manual partitioning creates maintenance and load-balancing challenges.Partitioned database architectures automate these tasks using key-based hash or range partitioning.
- Why Paxos: Failure sequences can make master-slave replication unavailable even with only one node down, and committed writes may be lost after permanent master failure.The illustrated sequence has the slave fail, the master accept later writes, and then the master fail before the slave recovers.
- Motivation: 3-way replication is used because failures are inevitable at large cluster scale and 2-way replication can suffer catastrophic data loss from double-disk failures.Three replicas also simplify upgrades by allowing one replica to be taken offline while two remain online.
- Why Paxos: Paxos provides a proven way to reach consensus among 2F + 1 replicas while tolerating up to F failures.The paper presents Paxos as addressing the general consistency problem that becomes difficult with three or more replicas.
- Spinnaker: Spinnaker combines key-based range partitioning, 3-way replication, and a transactional get-put API with strong or timeline-consistent reads.Timeline consistency may return stale data in exchange for better performance.
- Spinnaker: Paxos keeps a data partition available for reads and writes while a majority of its replicas remain alive, regardless of failure sequence.The design targets a single datacenter and assumes a different strategy for cross-datacenter fault tolerance.
2. RELATED WORK
Related systems address replication, consistency, and scalability through different architectural choices. The paper contrasts Spinnaker’s tightly integrated Paxos replication with middleware, eventual-consistency, and distributed-storage approaches.
- Replication approaches: 2PC is overkill for replica consistency because a single node failure causes an abort, conflicting with the goal of availability during failures.The passage also notes that invoking 2PC for every transaction has additional disadvantages.
- Replication approaches: Spinnaker integrates transaction ordering with its commit log and recovery processing rather than relying on a separate replication coordinator.The paper contrasts this with approaches whose certifier replication or fault tolerance is unspecified.
- Replication approaches: Middleware replication solutions can recover from simple failures, but their ability to handle the complicated failures targeted by Paxos is unclear.Ganymed is described as using a single master and FIFO queues.
- Comparison with datastores: Unlike Dynamo, Spinnaker avoids conflict resolution by using Paxos to keep replicas synchronized.Dynamo uses vector clocks and background antientropy mechanisms to manage eventual-consistency conflicts.
- Comparison with datastores: Bigtable relies on GFS for data, log storage, and replication, whereas Spinnaker uses local components and integrated replication.The paper identifies centralized logging overhead and the absence of a hot standby as drawbacks for transactional workloads.
- Comparison with datastores: PNUTS supports timeline consistency and single-operation transactions while focusing more on cross-datacenter replication than Spinnaker.PNUTS relies on the centralized Yahoo Message Broker for replication.
3. DATA MODEL AND API
Spinnaker exposes a row-and-column API with single-operation transactions and selectable read consistency. Versioned conditional updates provide optimistic concurrency control for read-modify-write operations.
- Data model: Rows contain arbitrary numbers of columns, values, and version numbers, while column names and values are opaque bytes.The data model organizes rows in tables and identifies each row uniquely by key.
- Read API: The get operation returns a column value and version, using strong consistency for the latest value or timeline consistency for potentially stale data.The consistency choice is controlled by the consistent flag.
- Write API: The put and delete operations insert or remove a column value from a row.The API also provides a conditional delete corresponding to conditional put.
- Conditional updates: conditionalPut updates a column only when its current version equals the supplied version; otherwise it returns an error.This enables a simple form of optimistic concurrency control.
- Conditional updates: A transactional counter increment reads a value consistently and conditionally writes the incremented value using the returned version.Applications are expected to retry the increment if an error is returned.
- Transactions: Each API call is a single-operation transaction, with multi-column variants available for updating multiple columns in one call.The paper treats non-modifying calls as reads and modifying calls as writes.
4. ARCHITECTURE
Spinnaker partitions rows by key ranges and replicates each range across overlapping cohorts of nodes. Its node architecture combines logging, commit tracking, storage structures, and coordination services while keeping Zookeeper off the read/write critical path.
- Cluster layout: Spinnaker uses range partitioning, assigning each base key range to a node and replicating it on the next two nodes by default.The default replication factor is N = 3.
- Cluster layout: A cohort is the group of nodes replicating one key range, and cohorts overlap across adjacent ranges.For example, the [0, 199] cohort is A-B-C and the [200, 399] cohort is B-C-D.
- Node architecture: Each node supports multiple key ranges with thread-safe components and uses a shared write-ahead log with logical LSNs per cohort.A dedicated logging device can be used for performance.
- Node architecture: Writes remain in the commit queue until sufficient cohort acknowledgments arrive, after which they enter a memtable and are flushed into indexed immutable SSTables.Background SSTable merges reclaim deleted rows and improve read performance.
- Coordination: Zookeeper provides fault-tolerant coordination for metadata, failures, locks, barriers, and group membership.It is not on the critical path for reads and writes; normal node communication consists of heartbeats.
5. THE REPLICATION PROTOCOL
Spinnaker replicates each cohort with a leader and two followers using a Paxos protocol that combines leader election with quorum-based writes. Strong reads use the leader, while timeline reads may use any replica and can be temporarily stale.
- Protocol structure: Each cohort has an elected leader and two followers, with replication divided into leader election and a quorum phase.In failure-free operation, the leader remains unchanged and only the quorum phase runs.
- Write path: A client write is routed to the affected range’s leader, which logs it, proposes it to followers, and commits after receiving one follower acknowledgment.The leader applies the write to its memtable and then responds to the client.
- Read consistency: Strongly consistent reads go to the leader, whereas timeline-consistent reads may use any cohort node and temporarily return stale values.Follower staleness is bounded by the commit period, which can be reduced to decrease it.
- Performance: A write requires 3 log forces and 4 messages overall, but its critical path is 1 follower log force and 2 message delays.Group commit is also used to improve logging performance.
- Conditional writes: Conditional puts use the same replication and recovery protocol, with the leader first checking the column version before executing the write.A failed version check produces no data write and returns an error.
6. RECOVERY
Spinnaker recovers followers through local replay followed by leader-directed catch-up, and recovers leaders by synchronizing followers before re-proposing unresolved writes. SSTable metadata supplies committed data when log records have been rolled over.
- Follower recovery: Follower recovery has local recovery and catch-up phases, beginning with idempotent replay through the follower’s last committed LSN.Writes after that LSN remain ambiguous until reconciliation with the leader.
- Follower recovery: During catch-up, the follower advertises its last committed LSN and the leader sends all committed writes after it before briefly blocking new writes.This ensures the follower is fully caught up before recovery completes.
- Log rollover: When the leader’s log no longer contains required records, tagged SSTables are located by their minimum and maximum LSNs and sent to the follower.Log rollover is safe because the writes have already been captured in SSTables.
- Leader takeover: Leader takeover first catches followers up to the new leader’s last committed LSN, then re-proposes unresolved writes after obtaining a quorum.The cohort reopens for writes with an LSN larger than any previously used.
7. LEADER ELECTION
Spinnaker uses Zookeeper-backed, per-cohort leader election to choose the candidate with the most advanced log while preserving committed writes. The protocol relies on majority participation and quorum replication to establish consensus and avoid losing committed data.
- Election goals: Leader election runs after leader failure or local recovery and must reach consensus without losing committed writes.The protocol is run independently for each cohort.
- Candidate registration: Each cohort node advertises its last LSN in a sequential ephemeral Zookeeper candidate node and watches the candidate set.Election state is stored under the key range’s Zookeeper path.
- Leader selection: After two candidates appear, the candidate with the maximum last LSN becomes leader, with Zookeeper sequence numbers breaking ties.The new leader records its hostname and runs leader takeover.
- Safety: The protocol preserves committed writes because each committed write is forced to at least two logs and at least two nodes participate in election.With three nodes, those two sets must overlap in at least one node.
- Failure coordination: Zookeeper event handling invokes leader election and coordinates failures, recovery, and node rejoining.The event handler runs as a Zookeeper client on each node.
8. DISCUSSION
Spinnaker remains available for strongly consistent reads and writes with a replica majority, while timeline reads remain available with one replica. Its Paxos design trades stronger consistency and failure-sequence robustness against leader routing, write latency, and limited transaction scope.
- Availability: With N = 3, writes commit after reaching two logs, and strong reads and writes remain available while two cohort nodes are alive.Timeline-consistent reads remain available with only one surviving node.
- Durability boundary: Spinnaker can lose committed data after rapid permanent failures of the leader and one follower, despite normally surviving two permanent replica failures.The stated durability guarantee applies under normal circumstances.
- Transaction scope: Spinnaker currently executes each API call as a single-operation transaction; multi-operation transactions remain future work.The paper outlines protocol and recovery extensions that could support batching log records at commit.
- Design tradeoffs: Spinnaker routes all cohort writes, and strong-consistency reads, through the leader, which can reduce performance, scaling, and availability relative to eventually consistent systems.The experiments examine these design tradeoffs.
9. EXPERIMENTAL RESULTS
Spinnaker was evaluated against Cassandra under comparable 3-way-replicated datastore workloads. It matched or outperformed Cassandra on reads, while writes were modestly slower and both systems benefited substantially from SSD logging.
- Read results: Spinnaker’s consistent-read latency was 1.5x to 3.0x lower than Cassandra’s quorum-read latency.Cassandra accessed two replicas and checked conflicts, whereas Spinnaker contacted only the cohort leader.
- Read results: Spinnaker’s timeline-read latency was nearly identical to Cassandra’s weak-read latency.Cassandra’s weak read had the best latency, but its consistency guarantees could be unpredictable.
- Write results: Spinnaker’s write latency was 5% to 10% worse than Cassandra’s quorum-write latency across the tested load range.Spinnaker waited for acknowledgments from a cohort leader and one follower, while Cassandra waited for any two replicas.
- Write results: SSD logging reduced average write latency to 6 msec or less in most cases for both Spinnaker and Cassandra.The original poor latency was attributed mainly to Cassandra’s primitive log manager and unwanted disk seeks.
10. CONCLUSION
The paper presents Spinnaker’s Paxos-based replication protocol as a way to combine scalability, consistency, and availability on commodity servers. Its evaluation found competitive performance, while several capabilities and comparisons remained future work.
- Conclusion: Paxos with 3-way replication keeps a Spinnaker data partition available for reads and writes while a majority of replicas remain alive.This availability holds regardless of the failure sequence, unlike traditional master-slave replication.
- Conclusion: Spinnaker was as fast or faster than Cassandra on reads and only 5% to 10% slower on writes.Cassandra is described as an eventually consistent datastore.
- Future work: Future work includes multi-operation transactions, online load balancing, and comparison with a DFS-based datastore such as Bigtable.These areas were identified as open aspects of Spinnaker.
- Protocol design: Spinnaker’s replication protocol is based on a variation of Multi-Paxos integrated with database commits, recovery, and a shared write-ahead log.Its recovery catch-up phase ensures that returning nodes are not missing log entries.
- Protocol design: Spinnaker uses Zookeeper for leader election and reliable in-order TCP messages to simplify its replication protocol.The coordination service itself is based on Paxos.
C. EXPERIMENTAL SETUP
The experiments used separate ten-node clusters for datastore nodes and clients, with commodity hardware, local SATA disks, and a 1-Gbit Ethernet network. Measurements reported average end-to-end operation latency across increasing client load.
- Hardware and cluster: Experiments ran on a ten-node cluster with dual quad-core 2.1 GHz AMD processors, 16GB memory, and five locally attached SATA disks per node.One disk was dedicated to logging, and the others formed a striped logical volume.
- Hardware and cluster: A second ten-node cluster generated client requests, and the datastore nodes communicated through a rack-level 1-Gbit Ethernet switch.Write-back caching was disabled on the SATA disks to guarantee durability.
- Software configuration: The evaluation reused Cassandra’s SSTables, memtables, and log manager, while Spinnaker’s replication, recovery, and commit-queue components were implemented from scratch.Zookeeper version 3.2.0 was used.
- Measurement procedure: The experiments measured average read or write latency, including the client-to-datastore round trip, as system load increased.Load was varied by increasing threads per client node by powers of two.
- Measurement procedure: Most read experiments were CPU- and network-bottlenecked, while write experiments were usually bottlenecked by log forces required for commit processing.The paper characterizes both bottlenecks as typical for transactional workloads.
D.1 Availability Results
Spinnaker recovers availability after leader failure, with recovery tied to the commit period. Its performance remains competitive across scaling, mixed workloads, SSD logging, conditional puts, and durability settings.
- Availability: Recovery time was proportional to the commit period because the new leader re-proposes uncommitted log records.
- Scaling: Write latency remained roughly constant as the cluster grew from 20 to 80 EC2 instances for both Spinnaker and Cassandra.Each write affects only the three nodes storing the replicated value.
- Mixed Reads and Writes: Spinnaker’s timeline-read mix was 2% to 10% slower than Cassandra’s weak-read mix across workloads.The comparison used a mixed read-write workload with 4KB values and two client threads.
- Mixed Reads and Writes: With 10% writes, Spinnaker’s consistent-read mix was about 10% faster than Cassandra’s quorum-read mix, while Cassandra was roughly 7% faster with 50% writes.
- SSD Logging: Using SSD logs reduced average write latency to 6 msec or less in most cases and improved Spinnaker relatively more than Cassandra.Spinnaker was more sensitive to logging performance, and SSD logging could also simplify its architecture by removing the need for a shared log file.
- Conditional Puts: A conditional put performed marginally worse than a regular put because it reads a version number and compares it before writing.
- Durability and Write Latency: Cassandra quorum writes were about 40% to 50% slower than weak writes, while Spinnaker’s two-of-three main-memory-log mode reached about 2 msec.The main-memory approach can lose a small number of committed writes during a correlated power outage.