《大规模分布式存储系统》第9章 分布式存储引擎【9.3】

javazx

9.3　UpdateServer实现机制
. g& |6 g3 y' C: Z- O, m9 TUpdateServer用于存储增量数据，它是一个单机存储系统，由如下几个部分组
- I* \" ^3 J8 b6 N0 u+ `成：
$ f( @! E# X+ D, h1 T! U3 Z$ B1 W●内存存储引擎，在内存中存储修改增量，支持冻结以及转储操作；
●任务处理模型，包括网络框架、任务队列、工作线程等，针对小数据包做了专
门的优化；
: I, ~. @) M; P! \$ x. q●主备同步模块，将更新事务以操作日志的形式同步到备UpdateServer。
8 o# [3 e* \4 IUpdateServer是OceanBase性能瓶颈点，核心是高效，实现时对锁（例如，无锁
数据结构）、索引结构、内存占用、任务处理模型以及主备同步都需要做专门的优
" L3 x( d1 e0 w7 J; E化。
9.3.1　存储引擎
UpdateServer存储引擎如图9-3所示。
8 v8 z1 ~' n2 S8 \1 b图　9-3　UpdateServer存储引擎
UpdateServer存储引擎与6.1节中提到的Bigtable存储引擎看起来很相似，不同点
% |' C7 k4 Y. ^" u在于：
- x; M  i# k1 J4 r●UpdateServer只存储了增量修改数据，基线数据以SSTable的形式存储在
ChunkServer上，而Bigtable存储引擎同时包含某个子表的基线数据和增量数据；
7 A( m- P3 \- K' a9 w●UpdateServer内部所有表格共用MemTable以及SSTable，而Bigtable中每个子表
的MemTable和SSTable分开存放；
8 I0 m8 D, X3 S; \0 ^7 _% m! H1 g●UpdateServer的SSTable存储在SSD磁盘中，而Bigtable的SSTable存储在 GFS
中。
UpdateServer存储引擎包含几个部分：操作日志、MemTable以及SSTable。更新
$ K+ E8 l0 S+ i0 d* ]" h' C% o操作首先记录到操作日志中，接着更新内存中活跃的MemTable（Active
MemTable），活跃的MemTable到达一定大小后将被冻结，称为Frozen MemTable，同
时创建新的Active MemTable。Frozen MemTable将以SSTable文件的形式转储到SSD磁
( Y) c+ E% t& I1 N% h3 T盘中。
" q( F' @. W3 l1 L8 j; O  x1 ~0 y1.操作日志
% ?1 h1 n+ n1 z% E7 Y9 aOceanBase中有一个专门的提交线程负责确定多个写事务的顺序（即事务id），
将这些写事务的操作追加到日志缓冲区，并将日志缓冲区的内容写入日志文件。为
1 c1 {* [$ A$ [, W; C9 w3 b1 x了防止写操作日志污染操作系统的缓存，写操作日志文件采用Direct IO的方式实现：
. a: X/ C7 b# x+ g' Q, }class ObLogWriter
( Y2 _3 u- K% a4 k* Z  h% v' T{
5 N8 L9 f$ D1 U# Z5 tpublic:
//write_log函数将操作日志存入日志缓冲区
int write_log(const LogCommand cmd,const char*log_data,const int64_t
data_len);
  ]& |: c& c7 R9 y//将日志缓冲区中的日志先同步到备机再写入主机磁盘
int flush_log(LogBuffer＆tlog_buffer,const bool sync_to_slave=true,const bool
) H: |' N1 Q2 D4 r7 w  k- wis_master=true);
6 H! ~6 i& x9 ?' d# _};
% F* _  \$ n3 r- O6 w3 j每条日志项由四部分组成：日志头+日志序号+日志类型（LogCommand）+日志
内容，其中，日志头中记录了每条日志的校验和（checksum）。ObLogWriter中的
write_log函数负责将操作日志拷贝到日志缓冲区中，如果日志缓冲区已满，则向调
- a7 }# E* s' J用者返回缓冲区不足（OB_BUF_NOT_ENOUGH）错误码。接着，调用者会通过
) m8 g8 b7 K/ x& B  |+ tflush_log将缓冲区中已有的日志内容同步到备机并写入主机磁盘。如果主机磁盘的最
6 B7 X8 t* Q1 A% T. O后一个日志文件超过指定大小（默认为64MB），还会调用switch_log函数切换日志
+ X/ o3 o' L' ?; V6 H* t' d文件。为了提高写性能，UpdateServer实现了成组提交（Group Commit）技术，即首
先多次调用write_log函数将多个写操作的日志拷贝到相同的日志缓冲区，接着再调
" D; w. M$ f" r! H用flush_log函数将日志缓冲区中的内容一次性写入到日志文件中。
" p9 {2 I$ [% L. E2.MemTable
MemTable底层是一个高性能内存B树。MemTable封装了B树，对外提供统一的读
* E+ E0 r: E: N7 j. H4 t* W  f写接口。
B树中的每个叶子节点对应MemTable中的一行数据，key为行主键，value为行操
作链表的指针。每行的操作按照时间顺序构成一个行操作链表。
" {5 x, d2 q) v& j% I% H: i如图9-4所示，MemTable的内存结构包含两部分：索引结构以及行操作链表，索
9 p( i. K+ M" G0 B; c引结构为9.1.2节中提到的B树，支持插入、删除、更新、随机读取以及范围查询操
0 _& V3 g* x7 O8 |作。行操作链表保存的是对某一行各个列（每个行和列确定一个单元，称为Cell）的
% x% ~( L5 ~  N) H0 w; N( }7 Q7 H修改操作。
图　9-4　MemTable的内存结构
例9-3　对主键为1的商品有3个修改操作，分别是：将商品购买人数修改为
100，删除该商品，将商品名称修改为“女鞋”，那么，该商品的行操作链中将保存三
个Cell，分别为：＜update，购买人数，100＞、＜delete，*＞以及＜update，商品
名，“女鞋”＞。
7 r$ U  B" u% CMemTable中存储的是对该商品的所有修改操作，而不是最终结果。另外，
! I( P+ @- l6 J$ H  Z+ K' s/ jMemTable删除一行也只是往行操作链表的末尾加入一个逻辑删除标记，即＜delete，
*＞，而不是实际删除索引结构或者行操作链表中的行内容。
MemTable实现时做了很多优化，包括：
●哈希索引：针对主要操作为随机读取的应用，MemTable不仅支持B树索引，还
! J; G1 _& {6 h2 l: t) n3 T支持哈希索引，UpdateServer内部会保证两个索引之间的一致性。
# R2 s2 c3 j' S; U●内存优化：行操作链表中每个cell操作都需要存储操作列的编号
（column_id）、操作类型（更新操作还是删除操作）、操作值以及指向下一个cell操
作的指针，如果不做优化，内存膨胀会很大。为了减少内存占用，MemTable实现时
  D% f& I0 A8 q/ E9 i- ]$ z" f会对整数值进行变长编码，并将多个cell操作编码后序列到同一块缓冲区中，共用一
# e1 b0 p5 \, L' ^2 g个指向下一批cell操作缓冲区的指针：
$ k+ R6 N$ B; G& a+ c  H. Fstruct ObCellMeta
{
2 e7 r# }# F* a% `const static int64_t TP_INT8=1;//int8整数类型
const static int64_t TP_INT16=2;//int16整数类型
- O( P  G8 t" Y7 Jconst static int64_t TP_INT32=3;//int32整数类型
const static int64_t TP_INT64=4;//int64整数类型
const static int64_t TP_VARCHAR=6;//变长字符串类型
6 z+ S# n; j9 `const static int64_t TP_DOUBLE=13;//双精度浮点类型
const static int64_t TP_ESCAPE=0x1f;//扩展类型
const static int64_t ES_DEL_ROW=1;//删除行操作
};
' m" ]- I4 C. N2 L; t' Cclass ObCompactCellWriter
2 R# r( B3 Y- t3 ^{
7 `4 }0 G) I" Ypublic:
3 p! {8 G' [! n' S; Q//写入更新操作，存储成压缩格式
int append(uint64_t column_id,const ObObj＆value);
& [7 p" e* F9 E, `//写入删除操作，存储成压缩格式
9 L9 j$ K) Y* ?; u! }$ h% Mint row_delete();
};
' S# J6 E  g5 B2 t) @0 MMemTable通过ObCompactCellWriter来将cell操作序列化到内存缓冲区中，如果为
更新操作，调用append函数；如果为删除操作，调用row_delete函数。更新操作的存
% c( E5 ]6 j+ u4 c; @! c储格式为：数据类型+值+列ID,TP_INT8/TP_INT16/TP_INT32/TP_INT64分别表示8
位/16位/32位/64位整数类型，TP_VARCHAR表示变长字符串类型，TP_DOUBLE表示
' n  d' M" f8 w" T  c4 b双精度浮点类型。删除操作为扩展操作，其存储格式为：
TP_ESCAPE+ES_DEL_ROW。例9-3中的三个Cell：＜update，购买人数，100＞、＜
* f! X$ Q' J0 M- Jdelete，*＞以及＜update，商品名，“女鞋”＞在内存缓冲区的存储格式为：
第1～3字节表示第一个Cell，即＜update，购买人数，100＞；第4～5字节表示第
+ \3 j* _" N  W2 k# d6 S( C二个cell，即＜delete，*＞；第6～8字节表示第三个Cell，即＜update，商品名，“女
, ^( F7 M: U2 r& m' D0 C  f" v. `4 {鞋”＞。
- Q, X2 k) Y& ~8 AMemTable的主要对外接口可以归结如下：
) Q) v3 y! M- p2 R//开启一个事务
//@param[in]trans_type事务类型，可能为读事务或者写事务
" M  t6 y+ I% v+ n6 Q$ h+ g4 {//@param[out]td返回的事务描述符
5 Z4 J1 v# K3 f" x+ R$ n& ]- k2 qint start_transaction(const TETransType trans_type,MemTableTransDescriptor＆
  @, F% W% c9 _! G2 Etd);
/ j  c% C; j& I0 p( h2 j//提交或者回滚一个事务
  ?: V; }9 ?4 Q% j//@param[in]td事务描述符
//@param[in]rollback是否回滚，默认为false
int end_transaction(const MemTableTransDescriptor td,bool rollback=false);
//执行随机读取操作，返回一个迭代器
4 j  S7 {# i: l# D- H//@param[in]td事务描述符
//@param[in]table_id表格编号
+ J" b4 ]5 i. G% b* ]* L: p3 ?//@param[in]row_key待查询的主键
0 t+ ^9 k2 c+ T" j8 f//@param[out]iter返回的迭代器
7 r! c5 t$ L: n6 _int get(const MemTableTransDescriptor td,const uint64_t table_id,const
ObRowkey＆row_key,MemTableIterator＆iter);
//执行范围查询操作，返回一个迭代器
, W- g  r: }7 H- r5 f0 s& e/ t' P" q//@param[in]td事务描述符
4 D" ]8 ?7 q) X  T+ Y//@param[in]range查询范围，包括起始行、结束行，开区间或者闭区间
0 Y  B/ D# R2 S- g! ?* V//@param[out]iter返回的迭代器
int scan(const MemTableTransDescriptor td,const ObRange＆
range,MemTableIterator＆iter);
/ q3 _9 t$ u# z5 L) W( h, w//开始执行一次修改操作
! U, I6 g/ K$ j* h//@param[in]td事务描述符
int start_mutation(const MemTableTransDescriptor td);
( B/ Z  e+ o, h& L0 u* F6 j//提交或者回滚一次修改操作
. K5 j6 A( V1 j! g) J- G( x//@param[in]td事务描述符
/ y4 o- ~9 D0 P7 e5 l  q//@param[in]rollback是否回滚
1 j1 w) x# k7 @int end_mutation(const MemTableTransDescriptor td,bool rollback);
: G$ [$ N) D. O9 P. k" B7 F% G2 M//执行修改操作
//@param[in]td事务描述符
//@param[in]mutator修改操作，包含一个或者多个对多个表格的cell操作
int set(const MemTableTransDescriptor td,ObUpsMutator＆mutator);
! c9 h# P# ?3 K! J- n对于读事务，操作步骤如下：
. ^: }1 H% S) X& C+ d2 n0 ^1）调用start_transaction开始一个读事务，获得事务描述符；
2）执行随机读取或者扫描操作，返回一个迭代器；接着可以从迭代器不断迭代
% A) N# S/ t. B0 A! }6 }5 [数据；
3）调用end_transaction提交或者回滚一个事务。
class MemTableIterator
  n+ t& w  M6 d6 G. c{
! d& \% Q; Z2 Z* C. `6 k6 s' `public:
//迭代器移动到下一个cell
int next_cell();
//获取当前cell的内容
, `) g$ L; x8 a2 X//@param[out]cell_info当前cell的内容，包括表名（table_id），行主键（row_
key)，列编号(column_id)以及列值(column_value)
int get_cell(ObCellInfo**cell_info);
# S7 d1 P! W, D9 t  V8 ]- B' ^//获取当前cell的内容
: R, n' K) H6 C" I$ W0 C//@param[out]cell_info当前cell的内容
//@param is_row_changed是否迭代到下一行
, M/ m9 P2 C4 ?) bint get_cell(ObCellInfo**cell_info,bool*is_row_changed);
2 K2 b5 [' f" @; _};
读事务返回一个迭代器MemTableIterator，通过它可以不断地获取下一个读到的
cell。在例9-3中，读取编号为1的商品可以得到一个迭代器，从这个迭代器中可以读
" E: j$ T  b% R5 z; j0 ~" R6 \出行操作链中保存的3个Cell，依次为：＜update，购买人数，100＞，＜delete，*
' \/ i3 a% F7 J* V＞，＜update，商品名，“女鞋”＞。
写事务总是批量执行，步骤如下：
1）调用start_transaction开始一批写事务，获得事务描述符；
- P' Q; [0 O$ l5 ^2）调用start_mutation开始一次写操作；
: ~& k0 E; c7 w0 v) \  |3）执行写操作，将数据写入到MemTable中；
4）调用end_mutation提交或者回滚一次写操作；如果还有写事务，转到步骤
+ ]4 g( l7 {' r! c/ Z2）；
5）调用end_transaction提交写事务。
3.SSTable
, N# J1 r9 A/ ]! k- b8 m# p当活跃的MemTable超过一定大小或者管理员主动发起冻结命令时，活跃的
MemTable将被冻结，生成冻结的MemTable，并同时以SSTable的形式转储到SSD磁盘
中。
SSTable的详细格式请参考9.4节ChunkServer实现机制，与ChunkServer中的
9 g# |. x; T  e& `+ @SSTable不同的是，UpdateServer中所有的表格共用一个SSTable，且SSTable为稀疏格
6 s7 Y# F7 F  O9 X; _: \. U0 s式，也就是说，每一行数据的每一列可能存在，也可能不存在修改操作。
. i0 C4 Y; [6 i  f9 n6 n5 P  P8 ~另外，OceanBase设计时也尽量避免读取UpdateServer中的SSTable，只要内存足
够，冻结的MemTable会保留在内存中，系统会尽快将冻结的数据通过定期合并或者
9 r& x+ [2 H! A. e, [( E' d$ k数据分发的方式转移到ChunkServer中去，以后不再需要访问UpdateServer中的
SSTable数据。
当然，如果内存不够需要丢弃冻结MemTable，大量请求只能读取SSD磁盘，
0 |5 ?" S+ O: f6 J! X4 LUpdateServer性能将大幅下降。因此，希望能够在丢弃冻结MemTable之前将SSTable
' g/ S4 Y# U: Y. Q的缓存预热。
4 V& q/ d3 q5 z  h* M* D! \UpdateServer的缓存预热机制实现如下：在丢弃冻结MemTable之前的一段时间
（比如10分钟），每隔一段时间（比如30秒），将一定比率（比如5%）的请求发给
SSTable，而不是冻结MemTable。这样，SSTable上的读请求将从5%到10%，再到
4 @6 m! Z( {: g" T15%，依次类推，直到100%，很自然地实现了缓存预热。
9.3.2　任务模型
任务模型包括网络框架、任务队列、工作线程，UpdateServer最初的任务模型基
* H, P6 o' T- I3 I; b! @& @于淘宝网实现的Tbnet框架（已开源，见http://code.taobao.org/p/tb-common-
utils/src/trunk/tbnet/）。Tbnet封装得很好，使用比较方便，每秒收包个数最多可以达
到接近10万，不过仍然无法完全发挥UpdateServer收发小数据包以及内存服务的特
点。OceanBase后来采用优化过的任务模型Libeasy，小数据包处理能力得到进一步提
升。
+ P+ [% f3 H: B1.Tbnet
如图9-5所示，Tbnet队列模型本质上是一个生产者—消费者队列模型，有两个线
* V/ E* X! }% e+ `程：网络读写线程以及超时检查线程，其中，网络读写线程执行事件循环，当服务
/ p% v3 f$ z3 }! F! o器端有可读事件时，调用回调函数读取请求数据包，生成请求任务，并加入到任务
0 X$ g1 n' a( I: _' Y队列中。工作线程从任务队列中获取任务，处理完成后触发可写事件，网络读写线
: x( i0 S( B9 M% S4 ?程会将处理结果发送给客户端。超时检查线程用于将超时的请求移除。
; i, U. J" ?) X, R  ]图　9-5　Tbnet队列模型
) g8 ^: S& G- P7 E; \Tbnet模型的问题在于多个工作线程从任务队列获取任务需要加锁互斥，这个过
程将产生大量的上下文切换（context switch），测试发现，当UpdateServer每秒处理
包的数量超过8万个时，UpdateServer每秒的上下文切换次数接近30万次，在测试环
境中已经达到极限（测试环境配置：Linux内核2.6.18，CPU为2*Intel Nehalem
E5520，共8核16线程）。
4 d0 t" J9 E5 Z9 i$ b1 M2.Libeasy
为了解决收发小数据包带来的上下文切换问题，OceanBase目前采用Libeasy任务
- S( C1 K, E1 q0 _  a% C: z模型。Libeasy采用多个线程收发包，增强了网络收发能力，每个线程收到网络包后
立即处理，减少了上下文切换，如图9-6所示。
图　9-6　Libeasy任务模型
1 Z2 B6 v6 d$ _/ k7 M- g" [UpdateServer有多个网络读写线程，每个线程通过Linux epool监听一个套接字集
合上的网络读写事件，每个套接字只能同时分配给一个线程。当网络读写线程收到
% T. p& x1 k; C6 Z) S- q& H  k网络包后，立即调用任务处理函数，如果任务处理时间很短，可以很快完成并回复
' H1 Q' E/ }* L4 O客户端，不需要加锁，避免了上下文切换。UpdateServer中大部分任务为短任务，比
- x. i7 R0 X! ~9 v# O/ y8 O如随机读取内存表，另外还有少量任务需要等待共享资源上的锁，可以将这些任务
加入到长任务队列中，交给专门的长任务处理线程处理。
由于每个网络读写线程处理一部分预先分配的套接字，这就可能出现某些套接
字上请求特别多而导致负载不均衡的情况。例如，有两个网络读写线程thread1和
. ~; [& S6 A$ O/ ]thread2，其中thread1处理套接字fd1、fd2，thread2处理套接字fd3、fd4，fd1和fd2上每
8 L  Y5 e: f8 e6 m/ e/ b) r1 x秒1000次请求，fd3和fd4上每秒10次请求，两个线程之间的负载很不均衡。为了处理
( R. j; @8 m2 p1 K( `8 a" P2 {* x0 J这种情况，Libeasy内部会自动在网络读写线程之间执行负载均衡操作，将套接字从
  n: s9 T7 z9 r3 I' n负载较高的线程迁移到负载较低的线程。
9.3.3　主备同步
8.4.1节已经介绍了UpdateServer的一致性选择。OceanBase选择了强一致性，主
& G; G" X4 G" r) VUpdateServer往备UpdateServer同步操作日志，如果同步成功，主UpdateServer操作本
+ p% K  G" H* `- g地后返回客户端更新成功，否则，主UpdateServer会把备UpdateServer从同步列表中
/ f0 _! {# C0 d7 t/ p8 R% C' D1 J剔除。另外，剔除备UpdateServer之前需要通知RootServer，从而防止RootServer将不
一致的备UpdateServer选为主UpdateServer。
如图9-7所示，主UpdateServer往备机推送操作日志，备UpdateServer的接收线程
接收日志，并写入到一块全局日志缓冲区中。备UpdateServer只要接收到日志就可以
& @3 V/ T0 j6 ^; U; ^/ y回复主UpdateServer同步成功，主UpdateServer接着更新本地内存并将日志刷到磁盘
文件中，最后回复客户端写入操作成功。这种方式实现了强一致性，如果主
2 e# M# t5 `: P  C% o% yUpdateServer出现故障，备UpdateServer包含所有的修改操作，因而能够完全无缝地
' X% s9 [' P" Z切换为主UpdateServer继续提供服务。另外，主备同步过程中要求主机刷磁盘文件，
备机只需要写内存缓冲区，强同步带来的额外延时也几乎可以忽略。
图　9-7　UpdateServer主备同步原理
# W: H: e& p) B- q) i, h* n正常情况下，备UpdateServer的日志回放线程会从全局日志缓冲区中读取操作日
2 [; Y% w0 P) k, T6 u) X% I志，在内存中回放并同时将操作日志刷到备机的日志文件中。如果发生异常，比如
& t. K0 G9 O2 e备UpdateServer刚启动或者主备之间网络刚恢复，全局日志缓冲区中没有日志或者日
. g& e+ T8 }$ R9 ^$ e  M  s: ?志不连续，此时，备UpdateServer需要主动请求主UpdateServer拉取操作日志。主
UpdateServer首先查找日志缓冲区，如果缓冲区中没有数据，还需要读取磁盘日志文
& N$ \" _4 W9 z+ j- x件，并将操作日志回复备UpdateServer。代码如下：
class ObReplayLogSrc
{
public:
8 p. M8 B' Y6 L: V* y( d7 U* x# F- \, e//读取一批待回放的操作日志
/ O6 Z: t8 ]. C3 N. D: H//@param[in]start_cursor日志起始点
//@param[out]end_id读取到的最大日志号加1，即下一次读取的起始日志号
//@param[in]buf日志缓冲区
//@param[in]len日志缓冲区长度
//@param[out]read_count读取到的有效字节数
int get_log(const ObLogCursor＆start_cursor,int64_t＆end_id,char*buf,const
& U$ `1 V+ q* M7 n& I4 ?int64_t len,int64_t＆read_count);
};
class ObUpsLogMgr
( J. ?; p3 [- s& r7 w{
public:
enum WAIT_SYNC_TYPE
, {( d7 n) S+ z: {& a) h9 w) q" `' D/ p{
- N0 P4 V) z) T8 I  {) lWAIT_NONE=0，
WAIT_COMMIT=1，
WAIT_FLUSH=2，
. t. w) E4 v( {% v: N7 t};
public:
$ B4 E: X0 {5 D5 [//备UpdateServer接收主UpdateServer发送的操作日志
int slave_receive_log(const char*buf,int64_t len,const int64_t
wait_sync_time_us,const WAIT_SYNC_TYPE wait_event_type);
" v0 R- E3 _" R//备UpdateServer获取并回放操作日志
5 p; u3 S+ [6 w! V$ Jint replay_log();
4 l4 n+ z; J8 B, A: J# r};
备UpdateServer接收到主UpdateServer发送的操作日志后，调用ObUpsLogMgr类的
slave_receive_log将操作日志保存到日志缓冲区中。备UpdateServer可以配置成不等待
（WAIT_NONE）、等待提交到MemTable（WAIT_COMMIT）或者等待提交到
. P* U/ v7 @& b; i3 @, I3 RMemTable且写入磁盘（WAIT_FLUSH）。另外，备UpdateServer有专门的日志回放线
程不断地调用ObUpsLogMgr中的replay_log函数获取并回放操作日志。
+ i. }5 U7 b; n8 i备UpdateServer执行replay_log函数时，首先调用ObReplayLogSrc的get_log函数读
取一批待回放的操作日志，接着，将操作日志应用到MemTable中并写入日志文件持
8 |9 v$ ~0 `4 O+ L9 }: X/ T久化。Get_log函数执行时首先查看本机的日志缓冲区，如果缓冲区中不存在日志起
始点（start_cursor）开始的操作日志，那么，生成一个异步任务，读取主
UpdateServer。一般情况下，slave_receive_log接收的日志刚加入日志缓冲区就被
3 M; f) E4 ?' V8 U3 }) A5 Kget_log读走了，不需要读取主UpdateServer。
- m, W7 Q& t$ g1 z) J% l9 w& M0 U