|
Ceph分布式文件系统的代码分析的文章网上是比较少的，本团队成员对ceph做过详细的代码阅读，包括mds、osd、client等模块，但是缺少条理清晰的文档总结。暂且先放上OSD的代码分析，等后续整理陆续放上其它模块的。
1 OSD的基本结构主要的类，涉及的线程，工作的方式 1.1 类OSD该类主要用以处理网络消息，与mds客户端等之间的网络连接的维护。当收到客户端或者mds对对象的数据请求后，交给相关的类进行处理。 1.1.1 主要对象ObjectStore *store; /*对object访问接口的封装**/ OSDSuperblock superblock; 主要是版本号等信息 OSDMapRef osdmap;
1.1.2 OSD中的线程池[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; }
1.2 PGPG，对象访问的上层控制，确定读取的对象的位置等信息，对对象的实际的读写数据控制由FileStore完成。 Ceph系统中为了管理对象，将对象进行了分组。PG即place_group就是ceph中的分组。
1.2.1 主要对象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主要用以操作对象，对象操作接口的封装。
1.3 FileStore负责向osd设备中数据的读写，作为类OSD的成员对象store出现。 1.4 FileJournal负责日志的管理，通过日志恢复数据等，作为类OSD的成员对象journal出现。
2 OSD读写数据的过程2.1 客户端发起请求的过程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) 2.2 OSD的op_tp线程处理数据读取处理的过程如下： 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);
3 OSD中的日志、事务这里对对象的写或者修改操作最终会交给FileStore对象处理，提交到该对象的嵌套类OpSequencer中的链表q中，日志的序列号加入到链表jq中。在flush时，根据日志的序列号保证了日志未flush前，操作不会写入磁盘。
在一个操作的处理过程中，最终由PG发出处理该动作。上述的序列关系记录在PG对象中的ObjectStore::Sequencer osr;中。 3.1 对于对象的操作的处理过程对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操作从队列中删除。
3.2 修改操作的处理可以看到对于修改操作，需要通过日志、事务进行处理，将操作加入到日志，事务的过程为：
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之后操作提交到队列中。 3.3 日志的写入而提交日志，可以看到在函数_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); 唤醒因为日志中操作数过多或者数据过大而阻塞的对象。
3.4 写操作的处理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); 唤醒操作
3.5 事务的sync过程：在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？
3.6 日志的恢复过程
在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(); 恢复完毕，重新启动写线程
4 PG对object的组织管理在写操作过程中，创建新的对象的过程 删除对象
|
|
|
来自： hello_worldw6 > 《待分类》
