Package go-craq
implements CRAQ (Chain Replication with Apportioned Queries)
as described in the CRAQ paper. MIT Licensed.
CRAQ is a replication protocol that allows reads from any replica while still maintaining strong consistency. CRAQ should provide better read throughput than Raft and Paxos. Read performance grows linearly with the number of nodes added to the system. Network chatter is significantly lower compared to Raft and Paxos.
Chain Replication: How to Build an Effective KV Storage
MIT 6.824 Distributed Systems Lecture on CRAQ (80mins)
+------------------+
| |
+-----+ Coordinator |
| | |
Write | +------------------+
|
v
+---+----+ +--------+ +--------+
| +---->+ +---->+ |
| Node | | Node | | Node |
| +<----+ +<----+ |
+---+-+--+ +---+-+--+ +---+-+--+
^ | ^ | ^ |
Read | | Read | | Read | |
| | | | | |
+ v + v + v
There are 3 packages that should be started to run the chain. The default node implementation in cmd/node uses the Go net/rpc package and bbolt for storage.
Facilitates new writes to the chain; allows nodes to announce themselves to the chain; manages the order of the nodes of the chain. One Coordinator should be run for each chain. For better resiliency, you could run a cluster of Coordinators and use something like Raft or Paxos for leader election, but that's outside the scope of this project.
-a # Local address to listen on. Default: :1234
Represents a single node in the chain. Responsible for storing writes, serving reads, and forwarding messages along the chain. In practice, you would probably have a single Node process running on a machine. Each Node should have it's own storage unit.
-a # Local address to listen on. Default: :1235
-p # Public address reachable by coordinator and the other nodes. Default: :1235
-c # Coordinator address. Default: :1234
-f # Bolt DB database file. Default: craq.db
Basic CLI tool for interacting with the chain. Allows writes and reads. The one included in this project uses the net/rpc package as the transport layer and bbolt as the storage layer.
-c # Address of coordinator. Default: :1234
-n # Address of node to send reads to. Default: :1235
./client write hello "world" # Write a new entry for key 'hello'
./client read hello # read the latest committed version of key 'hello'
go-craq processes communicate via RPC. The project is designed to be used with whatever RPC system shall be desired. The basic default client included in the go-craq package uses the net/rpc package from Go's stdlib; an easy-to-work-with package with a great API.
Pull requests for additional transport implementations are very welcome. Some common ones that would be great to have are gRPC and HTTP. Start by reading through transport/transport.go. Use transport/netrpc as an example.
go-craq is designed to make it easy to swap the persistance layer. CRAQ is
flexible and any storage unit that implements the Storer
interface in
store/store.go can be used. Some implementations for common
storage projects can be found in the store
package. store/kv
package is a very simple in-memory key/value store that is included as an
example to work off of when adding new storage implementations.
Pull requests for additional storage implementations are very welcome. Start by reading through the comments in store/store.go. Use the store/kv package as an example. CRAQ should work well with volatile and non-volatile storage but mixing should be avoided or else you may end up seeing long startup times due to data propagation. Mixing persistent storage mechanisms is an interesting idea I've been playing with myself. For example, one node storing items in the cloud and another storing items locally.
store/storetest should be used for testing new storage implementations. Run the test suite like this:
func TestStorer(t *testing.T) {
storetest.Run(t, func(name string, test storetest.Test) {
// New() is your store's constructor function.
test(t, New())
})
}
There are several places to start that'll give you a great understanding of how things work.
connectToCoordinator
method in node/node.go. This is the
method the Node must run during startup to connect to the Coordinator and
announce itself to the chain. The Coordinator responds with some metadata about
where in the chain the node is now located. The node uses this info to connect
to the predecessor in the chain.
Update
method in node/node.go. This is the method the
Coordinator uses to update the node's metadata. New data is sent to the node if
the node's predecessor or successor changes, and if the address of the tail node
changes.
ClientWrite
method in node/node.go. This is the method the
Coordinator uses to send writes to the head node. This is where the chain begins
the process of propagation.
A write request containing the key and value are sent to the Coordinator via the
Coordinator's Write
RPC method. If the chain is not empty, the Coordinator
will pass the write request to the head node via the node's ClientWrite
method.
The head node receives the key and value. The node looks up the key in it's
store to determine what the latest version should be. If the key already exists,
the version of the new write will be the existing version incremented by one.
The new value is written to the store. If the node is not connected to another
node in the chain, it commits the version. If the node does have a successor,
the key, value, and version are forwarded to the next node in the chain via the
successor's Write
RPC method.
The key, value, and version are passed along the chain one-by-one. Each node
adds the item to the store and sends a message to the successor in the chain.
When the write reaches the tail node, the tail marks the item as committed. The
tail sends a Commit
RPC method to it's predecessor. The tail's predecessor
commits that version of the item, then continues to forward the Commit
message
backwards through the chain, one node at a time, until every node has committed
the version.
When the node.Start
method is run, the Node will backfill it's list of latest
versions for all committed items in it's store, then it'll connect to the
coordinator. Each Node stores the latest version of each committed item in it's
store in-memory. This is done so that if the node is or becomes the tail node,
other nodes can query it for the latest committed version of an item. The
resources at the top of the readme provide some info on why this is important,
but basically it helps ensure strong consistency.
After backfilling the map of latest versions the Node connects to the Coordinator. The Coordinator adds the new Node to the list of Nodes in the chain, connects to the Node, responds with some metadata for the new Node, then sends updated metadata to the rest of the nodes in the chain to let it know there's a new tail.
The metadata that the Coordinator sends back to the new Node includes info to let the Node know if it's the head, the tail, and who it's predecessor in the chain is. The Node uses this info to connect to the predecessor in the chain.
After connecting to the predecessor, the Node asks the predecessor for all items it has that a) the Node has no record of, or b) the Node has older versions of. This ensures that the new Node is caught up with the chain. Because the new node is now the tail, any uncommitted items sent during propagation are immediately committed.
Once the Node is connected to it's neighbor and the coordinator, it starts listening for RPCs. The RPC server is setup and started in cmd/node.
It's worth mentioning that CRAQ works best with read-heavy workloads. One of it's best "features" is being able to read from any node in the chain. If a node receives a read request for key Y and the latest version of key Y in the store is not committed, the node will send a request to the tail to ask for the latest committed version of key Y; this helps ensure that all reads to any node returns the same value. In other words, it asserts strong consistency. As the chain grows, it takes longer for an item to be committed and the probability of needing to ask the tail for the latest version rises. Asking the tail for the latest version can quickly become a bottleneck. Therefore, storing these latest versions in-memory affords us higher throughput from the tail for requests for the latest version at the expense of slower startup times because the tail needs to backfill it's map of latest versions.
In the future, it may be beneficial to let the operator of the node signify whether they'd like to backfill at startup or serve 'latest version' requests directly from the store.
- Benchmarks based off the tests in the paper, as close as reasonably possible.
- gRPC transporter
- HTTP transporter
- Allow nodes to join at any location in the chain.