{"id":361,"date":"2017-08-15T00:00:00","date_gmt":"2017-08-15T00:00:00","guid":{"rendered":"https:\/\/en.pingcap.com\/blog\/design-and-implementation-of-multi-raft\/"},"modified":"2024-05-22T07:38:00","modified_gmt":"2024-05-22T14:38:00","slug":"design-and-implementation-of-multi-raft","status":"publish","type":"post","link":"https:\/\/www.pingcap.com\/ko\/blog\/design-and-implementation-of-multi-raft\/","title":{"rendered":"The Design and Implementation of Multi-raft"},"content":{"rendered":"

<\/span>Placement Driver<\/span><\/h2>\n

Placement Driver (PD), the global central controller of TiKV, stores the metadata information of the entire TiKV cluster, generates Global IDs, and is responsible for the scheduling of TiKV and the global TSO time service.<\/p>\n

PD is a critical central node. With the integration of etcd, it automatically supports the distributed scaling and failover as well as solves the problem of single point of failure. We will write another article to thoroughly introduce PD.<\/p>\n

In TiKV, the interaction with PD is placed in the pd<\/a> directory. You can interact with PD with your self-defined RPC and the protocol is quite simple. In pd\/mod.rs<\/a>, we provide the Client trait to interact with PD and have implemented the RPC Client.<\/p>\n

The Client trait of PD is easy to understand, most of which are the set\/get operations towards the metadata information of the cluster. But you need to pay extra attention to the operations below:<\/p>\n

bootstrap_cluster<\/code><\/strong>: When we start a TiKV service, we should firstly find out whether the TiKV cluster has been bootstrapped through is_cluster_bootstrapped<\/code>. If not, then create the first region on this TiKV service.<\/p>\n

region_heartbeat<\/code><\/strong>: Region reports its related information to PD regularly for the subsequent scheduling. For example, if the number of peers reported to PD is smaller than the predefined number of replica, then PD adds a new Peer replica to this Region.<\/p>\n

store_heartbeat<\/code><\/strong>: Store reports its related information to PD regularly for the subsequent scheduling. For example, Store informs PD of the current disk size and the free space. If PD considers it inadequate, it will not migrate other Peers to this Store.<\/p>\n

ask_split\/report_split<\/code><\/strong>: When a Region needs to split, it will inform PD through ask_split<\/code> and PD then generates the ID of the newly-split Region. After split successfully, Region informs PD through report_split<\/code>.<\/p>\n

By the way, we will make PD support gRPC protocol in the future, so the ClientAPI<\/code> will have some changes.<\/p>\n

<\/span>Raftstore<\/span><\/h2>\n

The goal of TiKV is to support 100 TB+ data and it is impossible for one Raft group to make it, we need to use multiple Raft groups, which is Multi-raft<\/strong>. In TiKV, the implementation of Multi-raft is completed in Raftstore and you can find the code in the raftstore\/store<\/a> directory.<\/p>\n

Region<\/h3>\n

To support Multi-raft, we perform data sharding and make each Raft store a portion of data.<\/p>\n

Hash and Range are commonly used for data sharding. TiKV uses Range and the main reason is that Range can better aggregate keys with the same prefix, which is convenient for operations like scan. Besides, Range outperforms in split\/merge than Hash. Usually, it only involves metadata modification and there is no need to move data around.<\/p>\n

The problem of Range is that a Region may probably become a performance hotspot due to frequent operations. But we can use PD to schedule these Regions onto better machines.<\/p>\n

To sum up, we use Range for data sharding in TiKV and split them into multiple Raft Groups, each of which is called a Region.<\/p>\n

Below is the protocol definition of Region’s protbuf:<\/p>\n

\nmessage RegionEpoch {\n\n     optional uint64 conf_ver  = 1 [(gogoproto.nullable) = false];\n\n     optional uint64 version     = 2 [(gogoproto.nullable) = false];\n\n}\n\nmessage Region {\n\n    optional uint64 id                  = 1 [(gogoproto.nullable) = false];\n\n    optional bytes  start_key           = 2;\n\n    optional bytes  end_key             = 3;\n\n    optional RegionEpoch region_epoch   = 4;\n\n    repeated Peer   peers               = 5;\n\n}\n\nmessage Peer {\n\n    optional uint64 id          = 1 [(gogoproto.nullable) = false];\n\n    optional uint64 store_id    = 2 [(gogoproto.nullable) = false];\n\n}\n\n<\/code><\/pre>\n
region_epoch<\/code><\/strong>: When a Region adds or deletes Peer or splits, we think that this Region’s epoch has changed. RegionEpoch’s conf_ver<\/code> increases during ConfChange while version<\/code> increases during split\/merge.<\/p>\n
id<\/code><\/strong>: Region’s only indication and PD allocates it in a globally unique way.<\/p>\n
start_key<\/code><\/strong>, end_key<\/code><\/strong>: Stand for the range of this Region [start_key, end_key). To the very first region, start and end key are both empty, and TiKV handles it in a special way internally.<\/p>\n
peers<\/code><\/strong>: The node information included in the current Region. To a Raft Group, we usually have three replicas, each of which is a Peer. Peer’s id<\/code> is also globally allocated by PD and store_id<\/code> indicates the Store of this Peer.<\/p>\n
RocksDB \/ Keys Prefix<\/h3>\n
In terms of actual data storage, whether it’s Raft Metadata, Log or the data in State Machine, we store them inside a RocksDB instance. More information about RocksDB, please refer to the RocksDB project on GitHub<\/a>.<\/p>\n
We use different prefixes to differentiate data of Raft and State Machine. For detailed information, please refer to keys\/src\/lib.rs<\/a>. As for the actual data of State Machine, we add “z” as the prefix and for other metadata stored locally, including Raft, we use the 0x01 prefix.<\/p>\n
I want to highlight the Key format of some important metadata and I’ll skip the first 0x01 prefix.<\/p>\n