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 PG
PG，對象訪問的上層控制，確定讀取的對象的位置等信息，對對象的實際的讀寫數據控制由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的組織管理
在寫操作過程中，創建新的對象的過程
刪除對象