从handle_client_mkdir函数操作来分析MDlog与LogSegment过程

// MKDIR
/* This function takes responsibility for the passed mdr*/
void Server::handle_client_mkdir(MDRequestRef& mdr)
{
  MClientRequest *req = mdr->client_request;
  set<SimpleLock*> rdlocks, wrlocks, xlocks;
  CDentry *dn = rdlock_path_xlock_dentry(mdr, 0, rdlocks, wrlocks, xlocks, false, false, false);  //获取该directory的目录项
  if (!dn) return;
  if (mdr->snapid != CEPH_NOSNAP) {
    respond_to_request(mdr, -EROFS);
    return;
  }
  CInode *diri = dn->get_dir()->get_inode();
  rdlocks.insert(&diri->authlock);
  if (!mds->locker->acquire_locks(mdr, rdlocks, wrlocks, xlocks))
    return;

  // new inode
  SnapRealm *realm = dn->get_dir()->inode->find_snaprealm();
  snapid_t follows = realm->get_newest_seq();

  unsigned mode = req->head.args.mkdir.mode;
  mode &= ~S_IFMT;
  mode |= S_IFDIR;
  CInode *newi = prepare_new_inode(mdr, dn->get_dir(), inodeno_t(req->head.ino), mode);    //创建该dir的新节点newi
  assert(newi);

  // it‘s a directory.
  dn->push_projected_linkage(newi);

  newi->inode.version = dn->pre_dirty();
  newi->inode.rstat.rsubdirs = 1;
  newi->inode.update_backtrace();

  dout(12) << " follows " << follows << dendl;
  if (follows >= dn->first)
    dn->first = follows + 1;
  newi->first = dn->first;

  // ...and that new dir is empty.
  CDir *newdir = newi->get_or_open_dirfrag(mds->mdcache, frag_t());      //获取或创建newdir 
  newdir->mark_complete();
  newdir->fnode.version = newdir->pre_dirty();

  // prepare finisher
　// 以下工作为将上面的改变记入日志MDlog中
  mdr->ls = mdlog->get_current_segment();
  EUpdate *le = new EUpdate(mdlog, "mkdir");  //为类EUpdate对象le分配空间并记录到mdlog此操作为mkdir
  mdlog->start_entry(le);        //找到将要写入日志的起始地址
  le->metablob.add_client_req(req->get_reqid(), req->get_oldest_client_tid());    //将req加入到用于记录客户端请求的pair容器中；metablob类注释为a bunch of metadata in the journal (日志中的一堆元数据)
  journal_allocated_inos(mdr, &le->metablob);     //为inode的成员变量赋予对应的值
  mdcache->predirty_journal_parents(mdr, &le->metablob, newi, dn->get_dir(), PREDIRTY_PRIMARY|PREDIRTY_DIR, 1);  //为将当前目录的目录项标记为脏
  le->metablob.add_primary_dentry(dn, newi, true, true);  //将此inode的目录项加入到链表lump中，该链表记录相关的dentry+inode，并标记此inode was last journaled
  le->metablob.add_new_dir(newdir); // dirty AND complete AND new  //标记newdir为dirty AND complete AND new
  
  // issue a cap on the directory
  int cmode = CEPH_FILE_MODE_RDWR;  //赋予其读写权限
  Capability *cap = mds->locker->issue_new_caps(newi, cmode, mdr->session, realm, req->is_replay());
  if (cap) {
    cap->set_wanted(0);

    // put locks in excl mode
    newi->filelock.set_state(LOCK_EXCL);
    newi->authlock.set_state(LOCK_EXCL);
    newi->xattrlock.set_state(LOCK_EXCL);
    cap->issue_norevoke(CEPH_CAP_AUTH_EXCL|CEPH_CAP_AUTH_SHARED|
            CEPH_CAP_XATTR_EXCL|CEPH_CAP_XATTR_SHARED);
  }

  // make sure this inode gets into the journal
  le->metablob.add_opened_ino(newi->ino());    //将newi的ino赋予类emetablob中的opened_ino
  LogSegment *ls = mds->mdlog->get_current_segment();  
  ls->open_files.push_back(&newi->item_open_file);　　//将newi添加到ls->open_files链表

  journal_and_reply(mdr, newi, dn, le, new C_MDS_mknod_finish(mds, mdr, dn, newi, follows));  //见下边的函数
}

/*******
 * some generic stuff for finishing off requests
 */
void Server::journal_and_reply(MDRequestRef& mdr, CInode *in, CDentry *dn, LogEvent *le, MDSInternalContextBase *fin)
{
  dout(10) << "journal_and_reply tracei " << in << " tracedn " << dn << dendl;

  // note trace items for eventual reply.
  mdr->tracei = in;
  if (in)
    mdr->pin(in);

  mdr->tracedn = dn;
  if (dn)
    mdr->pin(dn);

  early_reply(mdr, in, dn);   //此函数判断一些条件若符合则将mdlog刷出并返回，否则进行reply
  
  mdr->committing = true;
  submit_mdlog_entry(le, fin, mdr, __func__);
  
  if (mdr->client_request && mdr->client_request->is_replay()) {
    if (mds->queue_one_replay()) {         //若队列不为空则排队
      dout(10) << " queued next replay op" << dendl;
    } else {          //队列为空则将mdlog刷出
      dout(10) << " journaled last replay op, flushing" << dendl;
      mdlog->flush();
    }
  } else if (mdr->did_early_reply)      //若执行了early_reply则drop_rdlocks
    mds->locker->drop_rdlocks(mdr.get());
  else        //以上不符合则将mdlog刷出
    mdlog->flush();
}