storage R&D guy.
全部博文（1000）
分类： 服务器与存储
2015-09-14 14:09:36
Ceph分布式文件系统的代码分析的文章网上是比较少的，本团队成员对ceph做过详细的代码阅读，包括mds、osd、client等模块，但是缺少条理清晰的文档总结。暂且先放上OSD的代码分析，等后续整理陆续放上其它模块的。
主要的类，涉及的线程，工作的方式
该类主要用以处理网络消息，与mds客户端等之间的网络连接的维护。当收到客户端或者mds对对象的数据请求后，交给相关的类进行处理。
ObjectStore *store; /*对object访问接口的封装**/
OSDSuperblock superblock; 主要是版本号等信息
OSDMapRef osdmap;
[1] op_tp:
op_wq(this, g_conf->osd_op_thread_timeout, &op_tp)
scrub_finalize_wq(this, g_conf->osd_scrub_finalize_thread_timeout, &op_tp)
这里的op_wq是当OSD中当有请求操作时，会将该操作分配给所属的PG处理：
涉及的操作类型包括：CEPH_MSG_OSD_OP(client op) , MSG_OSD_SUBOP(for replication etc.) ,MSG_OSD_SUBOPREPLY。这些操作都要交给PG处理。
通过方法enqueue_op(pg, op);加入队列
// add to pg's op_queue
pg->op_queue.push_back(op); //该pg中加入该操作
op_wq.queue(pg); //由于该pg有了操作，将pg入队，op_tp中的线程会处理
其中op_wq的定义如下：
struct OpWQ : public ThreadPool::WorkQueue<PG> {
OSD *osd;
OpWQ(OSD *o, time_t ti, ThreadPool *tp)
: ThreadPool::WorkQueue<PG>("OSD::OpWQ", ti, ti*10, tp), osd(o) {}
bool _enqueue(PG *pg);
void _dequeue(PG *pg) {
assert(0);
}
bool _empty() {
return osd->op_queue.empty();
}
PG *_dequeue();
void _process(PG *pg) {
osd->dequeue_op(pg);
}
void _clear() {
assert(osd->op_queue.empty());
}
} op_wq;
OpWQ主要操作osd->op_queue，即deque<OpSequencer*> op_queue;
[2] recovery_tp
recovery_wq(this, g_conf->osd_recovery_thread_timeout, &recovery_tp)
struct RecoveryWQ : public ThreadPool::WorkQueue<PG> {
OSD *osd;
RecoveryWQ(OSD *o, time_t ti, ThreadPool *tp)
: ThreadPool::WorkQueue<PG>("OSD::RecoveryWQ", ti, ti*10, tp), osd(o) {}
RecoveryWQ 主要操作osd->recovery_queue，实际上封装与recovery相关的操作，这里recovery操作具体由每个PG执行。
void _process(PG *pg) {
osd->do_recovery(pg);
}
[3] disk_tp
remove_wq(this, g_conf->osd_remove_thread_timeout, &disk_tp)
osd->backlog_queue
// backlogs
xlist<PG*> backlog_queue;
rep_scrub_wq(this, g_conf->osd_scrub_thread_timeout, &disk_tp)
struct RepScrubWQ : public ThreadPool::WorkQueue<MOSDRepScrub> {
private:
OSD *osd;
list<MOSDRepScrub*> rep_scrub_queue;
snap_trim_wq(this, g_conf->osd_snap_trim_thread_timeout, &disk_tp)
osd->snap_trim_queue
// -- snap trimming --
xlist<PG*> snap_trim_queue;
backlog_wq(this, g_conf->osd_backlog_thread_timeout, &disk_tp)
osd->backlog_queue
// backlogs
xlist<PG*> backlog_queue;
[4] command_tp
command_wq(this, g_conf->osd_command_thread_timeout, &command_tp)
list<Command*> command_queue;
osd->command_queue
void _process(Command *c) {
osd->osd_lock.Lock();
osd->do_command(c->con, c->tid, c->cmd, c->indata);
osd->osd_lock.Unlock();
delete c;
}
PG，对象访问的上层控制，确定读取的对象的位置等信息，对对象的实际的读写数据控制由FileStore完成。
Ceph系统中为了管理对象，将对象进行了分组。PG即place_group就是ceph中的分组。
class PG {
struct Info { 描述一个PG的基本信息
pg_t pgid;
pg_stat_t stats;
struct History {} 创建的版本号，修改时间等
}
struct Query { Query - used to ask a peer for information about a pg.向其他OSD查询一个pg的信息
__s32 type;
eversion_t since;
Info::History history;
}
struct Log { incremental log of recent pg changes. pg修改的日志
struct Entry {
__s32 op;
hobject_t soid;
osd_reqid_t reqid;
uint64_t offset; // [soft state] my offset on disk
}
list<Entry> log; // the actual log.
}
IndexLog - adds in-memory index of the log, by oid. 日志在内存中的索引
struct IndexedLog : public Log {
hash_map<hobject_t,Entry*> objects; // ptrs into log. be careful! 每个对象对应的日志
hash_map<osd_reqid_t,Entry*> caller_ops;
list<Entry>::iterator complete_to; // recovery pointers
}
class OndiskLog {
uint64_t tail; // first byte of log.
uint64_t head;
}
struct Missing { //summary of missing objects.
//kept in memory, as a supplement to Log.
map<hobject_t, item> missing; // oid -> (need v, have v)
map<version_t, hobject_t> rmissing; // v -> oid
}
list<Message*> op_queue; // op queue PG操作的队列
// pg state
Info info;
const coll_t coll;
IndexedLog log;
hobject_t log_oid;
hobject_t biginfo_oid;
OndiskLog ondisklog;
Missing missing;
int role; // 0 = primary, 1 = replica, -1=none. 该pg的角色，主，备
/* Encapsulates PG recovery process */ PG recover处理的过程
class RecoveryState {
RecoveryMachine machine;
RecoveryCtx *rctx;
}
}
父类PG主要是用以对PG本身的维护，对PG的修改，日志的管理等。
Srcub的过程：
PG收集其管理的所有的objects，并向PG的副本请求对象的信息，进行对象状态的异常检查。
ReplicatedPG主要用以操作对象，对象操作接口的封装。
负责向osd设备中数据的读写，作为类OSD的成员对象store出现。
负责日志的管理，通过日志恢复数据等，作为类OSD的成员对象journal出现。
int Client::ll_read(Fh *fh, loff_t off, loff_t len, bufferlist *bl)
int Client::_read(Fh *f, int64_t offset, uint64_t size, bufferlist *bl)
int Client::_read_sync(Fh *f, uint64_t off, uint64_t len, bufferlist *bl)
//前几个参数均在结构体Inode中
Inode *in = f->inode;
filer->read_trunc(in->ino, &in->layout, in->snapid,
pos, left, &tbl, 0,
in->truncate_size, in->truncate_seq,
onfinish);
int read_trunc(inodeno_t ino,
ceph_file_layout *layout,
snapid_t snap,
uint64_t offset,
uint64_t len,
bufferlist *bl, // ptr to data
int flags,
uint64_t truncate_size,
__u32 truncate_seq,
Context *onfinish)
向osd读取数据的过程：
1 将要读取数据的长度和偏移转化为要访问的对象
file_to_extents(ino, layout, offset, len, extents);
2 向osd发起请求
objecter->sg_read(extents, snap, bl, flags, onfinish);
Filer.h
//计算需要读取的数据所在的extent,extent沿用了brtfs文件系统的概念
// ino ==> extents, extent实际上是object,offset
根据文件偏移访问对象的过程：
void Filer::file_to_extents(inodeno_t ino, ceph_file_layout *layout,
uint64_t offset, uint64_t len,
vector<ObjectExtent>& extents)
__u32 object_size = layout->fl_object_size;
__u32 su = layout->fl_stripe_unit;
__u32 stripe_count = layout->fl_stripe_count;
uint64_t stripes_per_object = object_size / su;
每个对象有两部分ino和objectno
// layout into objects
uint64_t blockno = cur / su; // which block
uint64_t stripeno = blockno / stripe_count; // which horizontal stripe (Y)
uint64_t stripepos = blockno % stripe_count; // which object in the object set (X)
uint64_t objectsetno = stripeno / stripes_per_object; // which object set
uint64_t objectno = objectsetno * stripe_count + stripepos; // object id
object_t oid = file_object_t(ino, objectno);
ObjectExtent *ex = 0;//主要由下面的两个参数组成
ex->oloc = objecter->osdmap->file_to_object_locator(*layout);
ex->oid = oid;
object_locator_t file_to_object_locator(const ceph_file_layout& layout) const {
return object_locator_t(layout.fl_pg_pool, layout.fl_pg_preferred);
}
Objecter.h
void sg_read_trunc(vector<ObjectExtent>& extents, snapid_t snap, bufferlist *bl, int flags,
uint64_t trunc_size, __u32 trunc_seq, Context *onfinish)
//对集合中的每个ObjectExtent进行处理
Objecter.h tid_t read_trunc(const object_t& oid, const object_locator_t& oloc,
uint64_t off, uint64_t len, snapid_t snap, bufferlist *pbl, int flags,
uint64_t trunc_size, __u32 trunc_seq,
Context *onfinish,
eversion_t *objver = NULL, ObjectOperation *extra_ops = NULL)
//该函数发出请求
Objecter.h tid_t read_trunc(const object_t& oid, const object_locator_t& oloc,
uint64_t off, uint64_t len, snapid_t snap, bufferlist *pbl, int flags,
uint64_t trunc_size, __u32 trunc_seq,
Context *onfinish,
eversion_t *objver = NULL, ObjectOperation *extra_ops = NULL)
处理的过程如下：
OpWQ的 void _process(PG *pg) 到 osd->dequeue_op(pg);中的代码如下：
if (op->get_type() == CEPH_MSG_OSD_OP) {
if (op_is_discardable((MOSDOp*)op))
op->put();
else
pg->do_op((MOSDOp*)op); // do it now
àvoid ReplicatedPG::do_op(MOSDOp *op)
à ReplicatedPG::do_op(MOSDOp *op)
à prepare_transaction(ctx); int ReplicatedPG::prepare_transaction(OpContext *ctx)
àint result = do_osd_ops(ctx, ctx->ops, ctx->outdata);
int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops, bufferlist& odata)
该函数的case CEPH_OSD_OP_READ: 分支
int r = osd->store->read(coll, soid, op.extent.offset, op.extent.length, bl);
可以看到最终到了FileStore对象中。
int FileStore::read(coll_t cid, const hobject_t& oid,
uint64_t offset, size_t len, bufferlist& bl)
read函数中主要调用了int fd = lfn_open(cid, oid, O_RDONLY);
我们可以看到定位一个对象需要的参数：
int FileStore::lfn_open(coll_t cid, const hobject_t& oid, int flags, mode_t mode)
r = get_index(cid, &index);
get_index的过程：在当前正在使用的index集合中判断是否正在被使用，如果被使用需要等待释放，否则建立索引。
int IndexManager::get_index(coll_t c, const char *path, Index *index) {
Mutex::Locker l(lock);
while (1) {
/// Currently in use CollectionIndices
// map<coll_t,std::tr1::weak_ptr<CollectionIndex> > col_indices;
if (!col_indices.count(c)) {
int r = build_index(c, path, index);
if (r < 0)
return r;
(*index)->set_ref(*index);
col_indices[c] = (*index);
break;
}else {
cond.Wait(lock);
}
}
return 0;
}
建立索引的过程：
int IndexManager::build_index(coll_t c, const char *path, Index *index) {
*index = Index(new FlatIndex(path),
RemoveOnDelete(c, this));
或者：
*index = Index(new HashIndex(path, g_conf->filestore_merge_threshold,
g_conf->filestore_split_multiple, version),
RemoveOnDelete(c, this));
这里coll_t的定义为：
class coll_t {
public:
const static coll_t META_COLL;
const static coll_t TEMP_COLL;
coll_t()
: str("meta")
{ }
std::string str;
coll_t实际上代表了一个目录，目录中是对象的集合。HashIndex在一定的条件下会拆分或者合并其拥有的子集合。
r = index->lookup(oid, &path, &exist);
r = ::open(path->path(), flags, mode);
这里对对象的写或者修改操作最终会交给FileStore对象处理，提交到该对象的嵌套类OpSequencer中的链表q中，日志的序列号加入到链表jq中。在flush时，根据日志的序列号保证了日志未flush前，操作不会写入磁盘。
在一个操作的处理过程中，最终由PG发出处理该动作。上述的序列关系记录在PG对象中的ObjectStore::Sequencer osr;中。
对object的操作最终由PG类进行处理，过程如下：
ReplicatedPG::do_op
1 如果是CEPH_OSD_FLAG_PGOP，由do_pg_op处理返回。
2 如果该pg状态为: finalizing_scrub并且有写操作(CEPH_OSD_FLAG_WRITE)，加入到waiting_for_active。
3 如果该对象在missing列表中：is_missing_object，加入等待列表wait_for_missing_object。
4 如果该对象在degraded列表并且有写操作，加入对一个的等待列表wait_for_degraded_object。
5 从磁盘或者缓存中读取对象的属性信息：find_object_context
6 如果失败，不能找到，将操作加入到miss等待列表：wait_for_missing_object
7 根据得到的对象的信息判断，如果是读请求并且是lost状态，返回出错
8 根据pg的mode判断该osd_op的合法性，如果不成功加入到mode的等待列表中
9 遍历该op中的ops，获得每个操作涉及的对象的信息，加入集合src_obc中。
10 如果是write操作，相应的检查snap version
11 通过加读锁，进行操作prepare_transaction，操作完后解除读锁。ObjectContext:: ondisk_read_lock
该函数中如果是读操作读取该对象的信息
写操作只进行基本的检查
ReplicatedPG::prepare_transaction 执行操作，此时数据、日志都在内存中。
1> do_osd_ops
int ReplicatedPG::do_osd_ops(OpContext *ctx, vector<OSDOp>& ops,bufferlist& odata)
CEPH_OSD_OP_WRITE分支：
/**将数据写入到事务缓存中*/
t.write(coll, soid, op.extent.offset, op.extent.length, nbl);
2> do_osd_op_effects
3> 如果是读请求返回
4> 修改操作添加日志
ctx->log.push_back(Log::Entry(logopcode, soid, ctx->at_version, old_version, ctx->reqid, ctx->mtime));
12 准备回应MOSDOpReply，如果是read操作或者是上一步出错，回应。
13 执行到这里只能是写操作。
append_log(ctx->log, pg_trim_to, ctx->local_t);
PG::append_log
1> 将ctx中的log加入到事务ctx->local_t中的缓存中。
创建新的RepGather，rep_op,并执行：
14 向该pg的副本发送此次请求：
ReplicatedPG::issue_repop
向PG的acting列表中的osd发送消息MOSDSubOp。
当其他的osd收到该请求后：
1> OSD::handle_sub_op此时只是将该op压入队列中
2> 在函数OSD::dequeue_op处理该请求：
ReplicatedPG::do_sub_op
ReplicatedPG::sub_op_modify ------------------------此时执行对osd的数据修改动作
将修改操作作为事务提交到队列中：
int r = osd->store->queue_transactions(&osr, rm->tls, onapply, oncommit);
这里将该操作提交给了两个线程池的，第一个线程池负责将日志写入磁盘。第二个负责执行该操作。如果没有使用btrfs文件系统作为osd存储，会先进行日志的过程，即将操作加入到日志队列中，当日志写入磁盘后，通过回调将操作加入到操作队列中。
这里注册的两个回调：
Context *oncommit = new C_OSD_RepModifyCommit(rm); 当日志写入磁盘后被调用
Context *onapply = new C_OSD_RepModifyApply(rm); 当该操作被处理后被调用
ReplicatedPG::sub_op_modify_applied
MOSDSubOpReply CEPH_OSD_FLAG_ACK
ReplicatedPG::sub_op_modify_commit
MOSDSubOpReply CEPH_OSD_FLAG_ONDISK
当收到其他的osd的回应时：
OSD::handle_sub_op_reply
ReplicatedPG::do_sub_op_reply
sub_op_modify_reply(r);
ReplicatedPG::repop_ack
如果是CEPH_OSD_FLAG_ONDISK，则从下面集合中删除：
repop->waitfor_disk.erase(fromosd);
repop->waitfor_ack.erase(fromosd);
否则：
repop->waitfor_ack.erase(fromosd);
每收到一次ack，都会调用函数eval_repop
15 eval_repop
当已经收到其他的osd回应时(代码中的注释的意思)：
apply_repop 执行此次动作。执行的过程与其他的osd执行过程类似。该函数将 repop->applying = true;
多注册了一个回调：ReplicatedPG::C_OSD_OndiskWriteUnlock::finish
当repop->waitfor_disk.empty()为空时：
此时向请求的发出者回应：MOSDOpReply CEPH_OSD_FLAG_ACK | CEPH_OSD_FLAG_ONDISK
当repop->waitfor_ack.empty()为空时：
向此次请求的发出者回应：MOSDOpReply CEPH_OSD_FLAG_ACK
此时写入的数据已经可读，但未commit
注意，两个回应中，第一个如果回应了就包含了第二个。两种回应只存在一个。
当repop->waitfor_ack.empty() && repop->waitfor_disk.empty()两者都为空时，将此次的repop操作从队列中删除。
可以看到对于修改操作，需要通过日志、事务进行处理，将操作加入到日志，事务的过程为：
FileStore::queue_transactions的过程：
这里将该操作提交给了两个线程池的，第一个线程池负责将日志写入磁盘。第二个负责执行该操作。如果没有使用btrfs文件系统作为osd存储，会先进行日志的过程，即将操作加入到日志队列中，当日志写入磁盘后，通过回调将操作加入到操作队列中。
当日志可写时：
1 创建FileStore:: Op op = build_op(tls, onreadable, onreadable_sync);
2 op_queue_reserve_throttle(o);
==> FileStore::_op_queue_reserve_throttle 当队列的操作数过多，或者队列中操作数据长度过大，阻塞等待。在某个操作处理结束后，_void_process_finish会唤醒。
3 o->op = op_submit_start(); ==>ops_submitting.push_back 获得操作的序列号
4如果m_filestore_journal_parallel，即这里将该操作同时加入到日志队列和FileStore的操作队列中。
1>_op_journal_transactions(o->tls, o->op, ondisk); 日志提交到日志队列的过程
如果日志可写
journal->submit_entry(op, tbl, data_align, onjournal);
->completions.push_back(onjournal)
-> writeq.push_back (write_item(seq, e, alignment))
否则加入等待队列：commit_waiters[op].push_back(onjournal);
2>queue_op(osr, o);
_op_apply_start(o->op);àJournalingObjectStore::_op_apply_start
当不是blocked状态时，没有处理，如果是blocked状态，等待被唤醒
osr->queue(o); 加入到OpSequencer的队列q中
op_wq.queue(osr); 此时将该操作加入到FileStore对象的op_wq队列中。
5如果m_filestore_journal_writeahead(当btrfs没有enable时为true)
osr->queue_journal(o->op);
_op_journal_transactions(o->tls, o->op, new C_JournaledAhead(this, osr, o, ondisk));
即当日志写入成功后，执行回调函数：
C_JournaledAhead::finish
fs->_journaled_ahead(osr, o, ondisk);
queue_op(osr, o); 此时将操作加入到操作队列中
osr->dequeue_journal(); 从日志中去除
ondisk_finisher.queue(ondisk); 调用回调
6 op_submit_finish(o->op); ==> ops_submitting.pop_front();
此时返回。
这里不考虑btrfs的情况，对于一个操作首先提交到日志中，日志flush之后操作提交到队列中。
而提交日志，可以看到在函数_op_journal_transactions中，日志最终被提交到了FileJournal类中的writeq队列里。
该队列由下面的线程处理：
FileJournal::write_thread_entry
对writeq进行循环：
1 int r = prepare_multi_write(bl, orig_ops, orig_bytes);
prepare_single_write
check_for_full
journalq.push_back(pair<uint64_t,off64_t>(seq, queue_pos)); 这里只记录了该事务的序列号以及在日志中的位置。
2 do_write(bl); bl缓存中记录了wrteq取出的事务的信息，以及在日志中的相关信息。
FileJournal::do_write
1> FileJournal::write_bl将缓存中的数据写入磁盘文件中。
bl.write_fd(fd);
==>buffer::list::write_fd
2> 如果不是directio，flush数据：fdatasync
3> queue_completions_thru:
将completions中的对象加入到finisher中。这里是之前注册的ondisk回调
即：ReplicatedPG::sub_op_modify_commit
3 put_throttle(orig_ops, orig_bytes); 唤醒因为日志中操作数过多或者数据过大而阻塞的对象。
FileStore中的op_tp线程池在该类的mount方法被调用时启动。
Op_tp负责管理FileStore的op_wq。也就是说在FileStore::queue_transactions中，将操作加入到op_wq中，会有线程去处理。处理的过程为：
根据调用栈，可以看到对于一个osd的操作最终由op_tp线程池处理，处理的主循环为：
ThreadPool::worker
WorkQueue_* wq;
wq = work_queues[last_work_queue];
wq->_void_process(item);
==> OSD::OpWQ::_process
==> FileStore::_do_op
wq->_void_process_finish(item);
==> OSD::OpWQ::_process
==> FileStore::_finish_op
1 _op_queue_release_throttle
调整op_queue,并唤醒 op_throttle_cond.Signal();
2如果有onreadable_sync回调，调用。
3 op_finisher.queue(o->onreadable); 交给finisher线程处理。
一个操作处理的过程：
FileStore::_do_op(OpSequencer *osr)
do_transactions(o->tls, o->op);
_transaction_start (bytes, ops) 当为brtfs时该函数才有实质性动作
对于tls中的每个transaction调用：
_do_transaction(**p, op_seq) 对于write操作调用_write方法，将数据写入到对应的对象中。
FileStore::_write
此时会将数据写入到文件，但不是sync，会尝试加入到flush队列中进行sync写。
_transaction_finish 同样，当为btrfs时该函数才有实质性的动作
op_apply_finish(o->op); 唤醒操作
在FileStore::mount方法中，会创建sync线程 sync_thread.create();
该线程的入口函数为：
void FileStore::sync_entry()
主要通过sync函数，将FileStore打开的文件进行数据的flush磁盘操作。
::fsync(op_fd);
或者 sync_filesystem(basedir_fd);
函数FileStore::_do_transaction的末尾：
即执行了实际操作之后trigger_commit 可以看到该函数中通过cond唤醒了sync线程。
Sync后，日志如何进行trim？
在FileStore::mount()函数中,打开日志后，会进行数据的恢复：
ret = journal_replay(initial_op_seq);
journal->read_entry(bl, seq) 每次从日志中读取一个entry出来
list<Transaction*> tls; 将entry所有的Transaction加入其中
do_transactions(tls, seq); 执行事务
journal->make_writeable(); 恢复完毕，重新启动写线程
在写操作过程中，创建新的对象的过程
删除对象
