《大规模分布式存储系统》第9章 分布式存储引擎【9.1】
第9章 分布式存储引擎分布式存储引擎层负责处理分布式系统中的各种问题，例如数据分布、负载均
衡、容错、一致性协议等。与其他分布式存储系统类似，分布式存储引擎层支持根
据主键更新、插入、删除、随机读取以及范围查找等操作，数据库功能层构建在分
布式存储引擎层之上。
分布式存储引擎层包含三个模块：RootServer、UpdateServer以及ChunkServer。
其中，RootServer用于整体控制，实现子表分布、副本复制、负载均衡、机器管理以
及Schema管理；UpdateServer用于存储增量数据，数据结构为一个内存B树，并通过
主备实时同步实现高可用，另外，UpdateServer的网络框架也经过专门的优化；
ChunkServer用于存储基线数据，基线数据按照主键有序划分为一个个子表，每个子
表在ChunkServer上存储了一个或者多个SSTable，另外，定期合并和数据分发的主要
逻辑也由ChunkServer实现。
OceanBase包含一个公共模块，包含其他模块共用的网络框架、内存池、任务队
列、锁、基础数据结构等，本章将介绍分布式存储引擎以及公共模块的实现。
9.1 公共模块
OceanBase源代码中有一个公共模块，包含其他模块需要的公共类，例如公共数
据结构、内存管理、锁、任务队列、RPC框架、压缩/解压缩等。下面介绍其中部分
类的设计思路。
9.1.1 内存管理
内存管理是C++高性能服务器的核心问题。一些通用的内存管理库，比如Google
TCMalloc，在内存申请/释放速度、小内存管理、锁开销等方面都已经做得相当卓越
了，然而，我们并没有采用。这是因为，通用内存管理库在性能上毕竟不如专用的
内存池，更为严重的问题是，它鼓励了开发人员忽视内存管理的陋习，比如在服务
器程序中滥用C++标准模板库（STL）。
在分布式存储系统开发初期，内存相关的Bug相当常见，比如内存越界、服务器
出现Core Dump，这些Bug都非常难以调试。因此，这个时期内存管理的首要问题并
不是高效，而是可控性，并防止内存碎片。
OceanBase系统有一个全局的定长内存池，这个内存池维护了由64KB大小的定长
内存块组成的空闲链表，其工作原理如下：
●如果申请的内存不超过64KB，尝试从空闲链表中获取一个64KB的内存块返回
给申请者；如果空闲链表为空，需要首先从操作系统中申请一批大小为64KB的内存
块加入空闲链表。释放时将64KB的内存块加入到空闲链表中以便下次重用。
●如果申请的内存超过64KB，直接调用Glibc的内存分配（malloc）函数，向操
作系统申请用户所需大小的内存块。释放时直接调用Glibc的内存释放（free）函数，
将内存块归还操作系统。
OceanBase的全局内存池实现简单，但内存使用率比较低，即使申请几个字节的
内存，也需要占用大小为64KB的内存块。因此，全局内存池不适合管理小块内存，
每个需要申请内存的模块，比如UpdateServer中的MemTable,ChunkServer中的缓存
等，都只能从全局内存池中申请大块内存，每个模块内部再实现专用的内存池。每
个线程处理读写请求时需要使用临时内存，为了提高效率，每个线程会缓存若干个
大小分别为64KB和2MB的内存块，每个线程总是首先尝试从线程局部缓存中申请内
存，如果申请不到，再从全局内存池中申请。
class ObIAllocator
{
public:
//内存申请接口
virtual void*alloc(const int64_t sz)=0;
//内存释放接口
virtual void free(void*ptr)=0;
};
class ObMalloc:public ObIAllocator
{
public:
//设置模块号
void set_mod_id(int32_t mod_id);
//申请大小为sz的内存块
void*alloc(const int64_t sz);
//释放内存
void free(void*ptr);
}
class ObTCMalloc:public ObIAllocator
{
public:
//设置模块号
void set_mod_id(int32_t mod_id);
//申请大小为sz的内存块
void*alloc(const int64_t sz);
//释放内存
void free(void*ptr);
}
ObIAllocator是内存管理器的接口，包含alloc和free两个方法。ObMalloc和
ObTCMalloc是两个实现了ObIAllocator接口的全局内存池，不同点在于，ObMalloc不
支持线程缓存，ObTCMalloc支持线程缓存。ObTCMalloc首先尝试从线程局部的空闲
链表申请内存块，如果申请不到，再通过ObMalloc的alloc方法申请。释放内存时，
如果没有超出线程缓存的内存块个数限制，则将内存块还给线程局部的空闲链表；
否则，通过ObMalloc的free方法释放。另外，允许通过set_mod_id函数设置申请者所
在的模块编号，便于统计每个模块的内存使用情况。
全局内存池的意义如下：
●全局内存池可以统计每个模块的内存使用情况，如果出现内存泄露，可以很快
定位到发生问题的模块。
●全局内存池可用于辅助调试。例如，可以将全局内存池中申请到的内存块按字
节填充为某个非法的值（比如0xFE），当出现内存越界等问题时，服务器程序会很
快在出现问题的位置Core Dump，而不是带着错误运行一段时间后才Core Dump，从
而方便问题定位。
总而言之，OceanBase的内存管理没有采用高深的技术，也没有做到通用或者最
优，但是很好地满足了服务器程序开发的两个最主要的需求：可控性以及没有内存
碎片。
9.1.2 基础数据结构
1.哈希表
为了提高随机读取性能，UpdateServer支持创建哈希索引，这个哈希索引结构就
是LightyHashMap，代码如下：
template＜typename Key,typename Value＞
class LightyHashMap
{
public:
//插入一个＜key,value＞对到哈希表
inline int insert(const Key＆key,const Value＆value);
//根据key查找value
inline int get(const Key＆key,Value＆value);
//根据key删除一个＜key,value＞对，如果value不为空，那么，保存删除的值到value中
inline int erase(const Key＆key,Value*value=NULL);
private:
struct Node
{
Key key;
Value value;
union
{
Node*next;
int64_t flag;
};
};
Node*buckets_;//哈希桶指针
BitLock bit_lock_;//位锁，用于保护哈希桶
};
LightyHashMap采用链式冲突处理方法，即将所有哈希值相同的＜key,value＞对
链到同一个哈希桶中，它包含如下三个方法：
●insert：往哈希表中插入一个＜key,value＞对。这个函数首先根据key的哈希值
得到桶号，接着，往哈希桶中插入一个包含key和value值的Node节点。
●get：根据key查找value。这个函数首先根据key的哈希值得到桶号，接着，遍历
对应的链表，找到与传入key相同的Node节点，返回其中的value值。
●erase：根据key删除一个＜key,value＞对。这个函数首先根据key的哈希值得到
桶号，接着，遍历对应的链表，找到并删除与传入key相同的Node节点。
LightyHashMap设计用来存储几千万甚至几亿个元素，它与普通哈希表的不同点
在于以下两点：
1）位锁（BitLock）：LightyHashMap通过BitLock实现哈希桶的锁结构，每个哈
希桶的锁结构只需要占用一个位（Bit）。如果哈希桶对应的位锁值为0，表示没有锁
冲突；否则，表示出现锁冲突。需要注意的是，LightyHashMap没有区分读锁和写
锁，多个get请求也是冲突的。可以对LightyHashMap的BitLock做一些改进，例如用两
个位（Bit）表示哈希桶对应的锁，其中一个位表示是否有读冲突，另外一个位表示
是否有写冲突。
2）延迟初始化（Lazy Initialization）：LightyHashMap的哈希桶个数往往特别多
（默认为1000万个），即使仅仅对所有哈希桶执行一次memset操作，消耗的时间也
是相当可观的。因此，LightyHashMap采用延迟初始化策略，即将哈希桶划分为多个
单元，默认情况下每个单元包含65536个哈希桶。每次执行insert、get或者erase操作
时都会判断哈希桶所属的单元是否已经初始化，如果未初始化，则对该单元内的所
有哈希桶执行初始化操作。
2.B树
UpdateServer的MemTable结构底层采用B树结构索引其中的数据行，代码如下：
template＜class K,class V,class Alloc＞
class BTreeBase
{
public:
//把＜key,value＞对加到B树中，overwrite参数表示是否覆盖原有值
int put(const K＆key,const V＆value,const bool overwrite=false);
//获取key对应的value
int get(const K＆key,V＆value);
//获取扫描操作描述符
int get_scan_handle(TScanHandle＆handle);
//设置扫描的数据范围
int set_key_range(TScanHandle＆handle,const K＆start_key,int32_t
start_exclude,const K＆end_key,int32_t end_exclude);
//读取下一行数据
int get_next(TScanHandle＆handle,K＆key,V＆value);
};
支持的功能如下：
1）Put：插入一个＜key,value＞对。
2）Get：根据key获取对应的value。
3）Scan：扫描一段范围内的数据行。首先，调用get_scan_handle获取扫描操作
描述符，其次，调用set_key_range设置扫描的数据范围，最后，不断地调用get_next
读取下一行数据直到全部读完。
B树支持多线程并发修改。如图9-1所示，往MemTable插入数据行（Data）时，
将修改其B树索引结构（Index），分为两种情况：
图 9-1 并发修改B树
●两个线程分别插入Data1和Data2：由于Data1和Data2属于不同的索引节点，插
入Data1和Data2将影响B树的不同部分，两个线程可以并发执行，不会产生冲突。
●两个线程分别插入Data2和Data3：由于Data2和Data3属于相同的索引节点，因
此，插入操作将产生冲突。其中一个线程会执行成功，另外一个线程失败后将重
试。
每个索引节点满了以后将分裂为两个节点，并触发对该索引节点的父亲节点的
修改操作。分裂操作将增加插入线程冲突的概率，在图9-1中，如果Data1和Data2的
父亲节点都需要分裂，那么，两个插入线程都需要修改Data1和Data2的祖父节点，从
而产生冲突。
另外，为了提高读写并发能力，B树实现时采用了写时复制（Copy-on-write）技
术，修改每个索引节点时首先将该节点拷贝出来，接着在拷贝出来的节点上执行修
改操作，最后再原子地修改其父亲节点的指针使其指向拷贝出来的节点。这种实现
方式的好处在于修改操作不影响读取，读取操作永远不会被阻塞。
细心的读者可能会发现，这里的B树不支持更新（Update）以及删除操作，这是
由OceanBase MVCC存储引擎的实现机制决定的。对于更新操作，MVCC存储引擎会
在行的末尾追加一个单元记录更新的内容，而不会影响索引结构；对于删除操作，
MVCC存储引擎内部实现为标记删除，即在行的末尾追加一个单元记录行的删除时
间，而不会物理删除某行数据。
9.1.3 锁
为了实现并发控制，OceanBase需要对一行记录加共享锁或者互斥锁。为此，专
门实现了QLock，代码如下：
struct QLock
{
enum State
{
EXCLUSIVE_BIT=1UL＜＜31，
UID_MASK=～EXCLUSIVE_BIT
};
volatile uint32_t n_ref_;//表示持有共享锁的引用计数
volatile uint32_t uid_;//表示持有互斥锁的用户编号
//加共享锁，uid为用户编号，end_time为超时时间
int shared_lock(const uint32_t uid,const int64_t end_time=-1);
//解除共享锁
int shared_unlock();
//加互斥锁，uid为用户编号，end_time为超时时间
int exclusive_lock(const uint32_t uid,const int64_t end_time=-1);
//解除互斥锁
int exclusive_unlock(const uint32_t uid);
//共享锁升级为互斥锁，uid为用户编号，end_time为超时时间
int share2exclusive_lock(const uint32_t uid,const int64_t end_time=-1);
//互斥锁降级为共享锁
int exclusive2shared_lock(const uint32_t uid);
};
在QLock的实现中，每把锁占用8个字节，其中4个字节为n_ref_，表示持有共享
锁的引用计数，另外4个字节为uid_，表示持有互斥锁的用户编号（例如线程编
号）。uid_的最高位（EXCLUSIVE_BIT）表示是否为互斥锁，其余31位表示用户编
号。
share_lock用于加共享锁，实现时只需要将n_ref_原子加1；exclusive_lock用于加
互斥锁，实现时需要将EXCLUSIVE_BIT置1并等待持有共享锁的所有用户解锁完成。
另外，为了避免新用户不断产生并持有共享锁导致无法获取互斥锁的情况，
exclusive_lock实现步骤如下：
1）将EXCLUSIVE_BIT置为1；
2）等待持有共享锁的所有用户解锁完成；
3）如果第2）步无法在超时时间内完成，加锁失败，将EXCLUSIVE_BIT重新置
为0。
第1）步执行完成后，新产生的用户无法获取共享锁。这样，只需要等待已经持
有共享锁的用户解锁即可，不会出现获取互斥锁时“饿死”的现象。
share2exclusive_lock将共享锁升级为互斥锁，实现时首先升级为互斥锁，如果获
取成功，接着再解除共享锁，即引用计数减1。
9.1.4 任务队列
在生产者/消费者模型中，往往有一个任务队列，生产者将任务加入到任务队
列，消费者从任务队列中取出任务进行处理。例如，在网络框架中，网络线程接收
任务并加入到任务队列，工作线程不断地从任务队列取出任务进行处理。
最为常见的场景是系统有一个全局任务队列，所有网络线程和工作线程操作全
局任务队列都需要首先获取独占锁，这种方式的锁冲突严重，将导致大量操作系统
上下文切换（context switch）。为了解决这个问题，可以给每个工作线程分配一个任
务队列，网络线程按照一定的策略选择一个任务队列并加入任务，例如随机选择或
者选择已有任务个数最少的任务队列。
将任务加入到任务队列（随机选择）：
1）将total_task_num原子加1（total_task_num为全局任务计数值）；
2）通过total_task_num%工作线程数，计算出任务所属的工作线程；
3）将任务加入到该工作线程对应的任务队列中；
4）唤醒工作线程。
然而，如果某个任务的处理时间很长，就有可能出现任务不均衡的情况，即某
个线程的任务队列中还有很多任务未被处理，其他线程却处于空闲状态。OceanBase
采取了一种很简单的策略应对这种情况：每个工作线程首先尝试从对应的任务队列
中获取任务，如果获取失败（对应的任务队列为空），那么，遍历所有工作线程的
任务队列，直到获取任务成功或者遍历完成所有的任务队列为止。
除此之外，OceanBase还实现了LightyQueue用于解决全局任务队列锁冲突问题。
LightyQueue的设计思想如下：
假设系统中有3个工作线程t1，t2和t3，全局任务队列中共有10个槽位。首先，
t1，t2和t3分别等待1号，2号以及3号槽位。网络线程将任务加入1号槽位时唤醒t1，
加入2号槽位时唤醒t2，加入3号槽位时唤醒t3。接着，t2很快将任务处理完成后等待4
号槽位，t3等待5号槽位，t1等待6号槽位。网络线程将任务加入到4，5，6号槽位时
将分别唤醒t2，t3和t1。通过这样的方式，每个工作线程在不同的槽位上等待，避免
了全局锁冲突。
将任务加入到工作队列（push）的操作如下：
1）占据下一个push槽位；
2）将任务加入到该push槽位；
3）唤醒该push槽位上正在等待的工作线程。
工作线程从任务队列中获取任务（pop）的操作如下：
1）占据下一个pop槽位；
2）如果该pop槽位上有任务，则直接返回；
3）否则，工作线程在该pop槽位上等待直到被push操作唤醒或者超时。
9.1.5 网络框架
OceanBase的网络框架代码如下：
class ObSingleServer
{
public:
//设置工作线程个数
int set_thread_count(const int thread_count);
//设置网络IO线程个数
int set_io_thread_count(const int io_thread_count);
//设置监听端口
int set_listen_port(const int listen_port);
public:
//处理接收到的网络包，默认的处理逻辑是将网络包加入到全局任务队列中
virtual int handlePacket(ObPacket*packet);
//工作线程每次从全局任务队列中取出一个网络包并调用该函数进行处理
virtual do_request(ObPacket*packet);
};
OceanBase服务端接收客户端发送的网络包（ObPacket），并交给handlePacket处
理函数进行处理。默认情况下，handlePacket会将网络包加入到全局任务队列中。接
着，工作线程会从全局任务队列中不断获取网络包，并调用do_request进行处理，处
理完成后应答客户端。可以分别通过set_thread_count以及set_io_thread_count函数来设
置工作线程以及网络线程的个数。
客户端使用ObClientManager发送网络包：
class ObClientManager
{
public:
//异步发送请求包
//@paramserver服务器端地址
//@parampcode请求包的类型（packet code）
//@paramversion请求包的版本
//@paramin_buffer请求包实际内容缓冲区
int post_request(const ObServer＆server,const int32_t pcode,const int32_t
version,const ObDataBuffer＆in_buffer)const;
//同步发送请求包并等待应答
//@paramserver服务器端地址
//@parampcode请求包的类型（packet code）
//@paramversion请求包的版本
//@paramtimeout请求时间
//@paramin_buffer请求包实际内容缓冲区
//@paramout_buffer应答包的实际内容缓冲区
int send_request(const ObServer＆server,const int32_t pcode,const int32_t
version,const int64_t timeout,ObDataBuffer＆in_buffer,ObDataBuffer＆
out_buffer)const;
};
客户端发包分为两种情况：异步请求（post_request）以及同步请求
（send_request）。异步请求时，客户端将请求包加入到网络发送队列后立即返回，
不等待应答。同步请求时，客户端将请求包加入到网络发送队列后开始阻塞等待，
直到网络线程接收到服务端的应答包后才唤醒客户端，从而执行后续处理逻辑。
9.1.6 压缩与解压缩
class ObCompressor
{
public:
//数据压缩与解压缩接口
//@paramsrc_buff输入数据缓冲区
//@paramsrc_data_size输入数据大小
//@paramdst_buffer输出数据缓冲区
//@paramdst_buffer_size输出数据缓冲区大小
//@paramdst_data_size输出数据大小
virtual compress(const char*src_buffer,const int64_t
src_data_size,char*dst_buffer,const int64_t dst_buffer_size,int64_t＆
dst_data_size)=0;
virtual decompress(const char*src_buffer,const int64_t
src_data_size,char*dst_buffer,const int64_t dst_buffer_size,int64_t＆
dst_data_size)=0;
//获取压缩库名称
const char*get_compress_name()const;
//根据传入的大小计算压缩后最大可能的溢出大小
int64_t get_max_overflow_size(const int64_t src_data_size)const;
};
ObCompressor定义了压缩与解压缩的通用接口，具体的压缩库实现了这些接
口。压缩库以动态库（.so）的形式存在，每个工作线程第一次调用compress或者
decompress方法时将加载相应的动态库，这样便实现了压缩库的插件化。目前，支持
的压缩库包括LZO 以及Snappy 。
LZO：见http://www.oberhumer.com/opensource/lzo/
Snappy：Google开源的压缩库，见http://code.google.com/p/snappy/
页:
[1]