ceph源码分析之读写操作流程（1）

ceph是一个存储集群，它拥有ceph-mon，ceph-mds和ceph-osd三种进程，本文主要从源码层面分析ceph-osd中的数据读写操作流程。

ceph-osd数据层级是这样的：

最上层：OSD，OSDServer

第二层：PG（具体实现类是ReplicatedPG）

第三层：ObjectStore(具体实现类是FileStore类，该类中有成员变量FileJournal类)

最底层：具体的Journal device或者file  （即系统读写函数）

介绍一下FileJournal，FileStore之间关系

ObjectStore定义了一个对象存储的抽象类

JournalingObjectStore继承ObjectStore定义了一个带有Journal的对象存储的类

Filestore最终继承成为一个实际工作的数据存储的类。

在JournalingObjectStore中定义了两个重要的成员变量：

[cpp] view plain copy

1. Journal *journal   //指向了日志操作类  
2. Finisher finisher   //日志操作完成后的处理线程，在后面会提到  

Journal定义了一个日志操作的抽象类

FileJournal继承Journal定义了一个实际工作的日志操作的类

两者之间的联系：OSD在init时，会new出一个FileStore类对象，并且在FileStore进行mount时候new出FileJournal类对象。

在第一层和第三层中拥有消息队列和执行线程，执行线程从消息队列中拿出数据来进行处理。

Finisher类是一个定义操作完成后的处理类，它内部自带了线程和消息队列。

OSD中处理读写操作是线程池和消息队列（有很多，其他暂时不讨论）：

[cpp] view plain copy

1. ThreadPool  op_tp;  
2. ThreadPool::WorkQueueVal<pair<PGRef,OpRequestRef>, PGRef>  &op_wq;  

FileJournal中拥有的线程和消息队列：

[cpp] view plain copy

1. Write  write_thread;  
2. deque<write_item> writeq;  

其父类Journal中拥有线程和消息队列（引用自之前说的JournalingObjectStore类中）：

[cpp] view plain copy

1. Finisher        finisher_thread;  

FileStore中拥有的线程和消息队列：

[cpp] view plain copy

1. ThreadPool  op_tp;  
2. OpWQ        op_wq;//Filestore中实现，继承自ThreadPool::WorkQueue<OpSequencer>  
3. Finisher    ondisk_finisher;  
4. Finisher    op_finisher;  

本章由于介绍的是上两层流程，故只用到了OSD::op_tp 和OSD::op_wq，在ceph源码分析之读写操作流程（2）文章中用到了上述的其他线程。

关于线程是如何具体工作的，可以参见ceph源码分析之线程介绍。

简单说一下线程池,所谓线程池就是根据配置文件中的配置，一次性创建了多个线程来处理消息队列中数据，它会轮询的处理在它之中的所有消息队列。即一个线程可能处理多个消息队列。

同时它定义了三种不同的WorkQueue,不同的WQ实现会在申明她们的类中自行实现需要的功能函数（如_process，_process_finish），线程从ThreadPool::worker为入口函数，开始等待op_wq中是否有数据，如果来了操作消息，则调用OSD::_process函数来执行。

为了方便，将数据操作分成两层来讲解，第一层是图1中的上两层，即数据操作的消息流程走向。第二层是图1中的下两层，即数据操作的实际读写流程。

在源码中，它们也被分别放在了osd和os两个文件夹中。

消息流程走向：

OSD获得消息后分发将消息送入了osd->op_wq队列。

[cpp] view plain copy

1. void OSD::handle_op(OpRequestRef op)  
2. {  
3.     ……  
4.     enqueue_op(pg,op);  
5. }  

osd->op_tp线程从队列中拿出消息开始工作（op_tp线程是在OSD::init时创建，线程具体的工作在OSD::OpWQ::_process中）。

之后会调用到ReplicatePG::do_request，此函数根据不同的消息会选择不同的操作方案，关于操作主要有这三个：

[cpp] view plain copy

1. void ReplicatedPG::do_request(OpRequestRef op,ThreadPool::TPHandle &handle)  
2. {  
3.     ……  
4.     switch(op->get_req()->get_type()){  
5.     caseCEPH_MSG_OSD_OP:  
6.         ……  
7.         do_op(op);//1.主osd上的pg操作  
8.         break;  
9.     caseMSG_OSD_SUBOP:  
10.         do_sub_op(op);//2.从osd上的pg操作  
11.         break;  
12.     caseMSG_OSD_SUBOPREPLY:  
13.         do_sub_op_reply(op);//3.主osd接收到从osd上的pg操作返回  
14.         break;  
15.     ……  
16.     }  
17. }  

然后分成了三步：

1.主osd：ReplicatedPG::do_opà ReplicatePG::execute_ctxà ReplicatedPG::prepare_transactionà ReplicatedPG::do_osd_ops

2.从osd：ReplicatedPG::do_sub_opàReplicatedPG::sub_op_modify写好后给主osd回应

3.主osd：ReplicatedPG::do_sub_op_replyà ReplicatedPG::sub_op_modify_reply-à ReplicatedPG::repop_ack-à ReplicatedPG::eval_repop

当第一步主osd调用ReplicatedPG::do_osd_ops处理完自己上面的pg操作后，会调用ReplicatedPG::issue_repop给副本pg的osd发消息。

于是第二步开始执行，从osd处理完后又会给主osd发送回应，

第三步开始执行，ReplicatedPG::eval_repop中会先通过ReplicatePG::waitfor_ack判断是否所有拥有pg的osd都回复了自己，

如果都回复了，则会调用OSD::sent_message_osd_client通知client端操作完成。

 

到此为止，关于读写操作的上层读写流程就已经完整了，再往后的调用就是ReplicatedPG层调用Filestore层的具体数据操作了。