ceph源码解析–osd篇

  • Post author:
  • Post category:其他


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 {

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 op_queue;

[2] recovery_tp

recovery_wq(this, g_conf->osd_recovery_thread_timeout, &recovery_tp)

struct RecoveryWQ : public ThreadPool::WorkQueue {

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 rep_scrub_queue;

snap_trim_wq(this, g_conf->osd_snap_trim_thread_timeout, &disk_tp)

     osd->snap_trim_queue

     // -- snap trimming --

xlist snap_trim_queue;

backlog_wq(this, g_conf->osd_backlog_thread_timeout, &disk_tp)

     osd->backlog_queue

     // backlogs

xlist backlog_queue;

[4] command_tp

command_wq(this, g_conf->osd_command_thread_timeout, &command_tp)

list 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 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 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 missing; // oid -> (need v, have v)

map<version_t, hobject_t> rmissing;  // v -> oid

}

list 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& 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& 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 > 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& 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(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的组织管理

在写操作过程中,创建新的对象的过程

     删除对象