class JournalingObjectStore : public ObjectStore {
protected:

  class SubmitManager {
    Mutex lock;
    uint64_t op_seq; // journal提交的序列号，全局唯一
    uint64_t op_submitted;
	......
  } submit_manager;

  class ApplyManager {
    Mutex apply_lock;
    bool blocked;
    Cond blocked_cond;
    int open_ops;
    uint64_t max_applied_seq; // apply到文件系统page cache的序列号

    Mutex com_lock;
    map<version_t, vector<Context*> > commit_waiters;
    uint64_t committing_seq, committed_seq; // 将文件系统page cache的数据fsync到磁盘的序列号，用来通知journal释放空间
	......
  } apply_manager;

  ......
};

int FileStore::queue_transactions(Sequencer *posr, list<Transaction*> &tls,
				  TrackedOpRef osd_op,
				  ThreadPool::TPHandle *handle)
{
  ......

  // posr，定义在PG类中，对于同一个PG，一定是一样的，这里的default_osr根本就不会用，Jewel新代码已经删除了
  // OpSequencer非常关键，同一个PG会使用同样的OpSequencer，保证PG操作串行化
  OpSequencer *osr;
  if (!posr)
    posr = &default_osr;
  if (posr->p) {
    osr = static_cast<OpSequencer *>(posr->p);
  } else {
    osr = new OpSequencer; // PG的第一次操作的时候，会创建一个OpSequencer，以后就会复用
    osr->parent = posr;
    posr->p = osr;
  }

  ......

  if (journal && journal->is_writeable() && !m_filestore_journal_trailing) {
    Op *o = build_op(tls, onreadable, onreadable_sync, osd_op);

    op_queue_reserve_throttle(o, handle); // filestore层对整个op的限流，释放的时候是在FileStore::_finish_op

    journal->throttle(); // 对journal的限流

    // 在journal层面为op生成唯一sequence, 因为journal是单线程写，所以写一定是串行的
    uint64_t op_num = submit_manager.op_submit_start();
    o->op = op_num;

    if (m_filestore_journal_parallel) {
		......
    } else if (m_filestore_journal_writeahead) { // ext4, xfs都需要wal
      
      osr->queue_journal(o->op); // journal层面的sequence记录在OpSequencer中的journal queue中, JournalingObjectStore::finisher线程中会deque_journal

      _op_journal_transactions(o->tls, o->op, // 提交OP，注意这里的callback以及sequence
			       new C_JournaledAhead(this, osr, o, ondisk),
			       osd_op);
    } else {
      assert(0);
    }
    submit_manager.op_submit_finish(op_num); // op提交到journal队列完成
    return 0;
  }

  ......
}

void JournalingObjectStore::_op_journal_transactions(
  list<ObjectStore::Transaction*>& tls, uint64_t op,
  Context *onjournal, TrackedOpRef osd_op)
{
  if (journal && journal->is_writeable()) {
	......
    journal->submit_entry(op, tbl, data_align, onjournal, osd_op); // 放入journal队列，等待write线程执行journal写请求
  } else if (onjournal) {
    apply_manager.add_waiter(op, onjournal);
  }
}

void FileJournal::submit_entry(uint64_t seq, bufferlist& e, int alignment,
			       Context *oncommit, TrackedOpRef osd_op)
{
  ......

  // 获取journal限流资源
  throttle_ops.take(1);
  throttle_bytes.take(e.length());

  {
	// 注意锁的顺序
    Mutex::Locker l1(writeq_lock);  // ** lock **
    Mutex::Locker l2(completions_lock);  // ** lock **

	// write线程执行完成后，会处理这里的completion
    completions.push_back(
      completion_item(
	seq, oncommit, ceph_clock_now(g_ceph_context), osd_op));

    if (writeq.empty())
      writeq_cond.Signal();

    writeq.push_back(write_item(seq, e, alignment, osd_op)); // 放入队列，等待write线程执行
  }
}

void FileJournal::queue_completions_thru(uint64_t seq)
{
  ......

  // journal的一次写可以同时写入多个op请求日志，所以这里是循环处理所有已经完成的op
  // 将回调全部放入finisher线程的队列
  while (!completions_empty()) {
    completion_item next = completion_peek_front();
    if (next.seq > seq) // sequence判断是否已经写入完成
      break;
    completion_pop_front();

	......
    if (next.finish) // 放入finisher队列，等待回调
      finisher->queue(next.finish); // finisher线程实际上是JournalingObjectStore中的finisher
	......
  }
  finisher_cond.Signal();
}

void FileStore::_journaled_ahead(OpSequencer *osr, Op *o, Context *ondisk)
{
  queue_op(osr, o); // 将op在filestore层面排队，准备写入文件系统

  list<Context*> to_queue;
  osr->dequeue_journal(&to_queue); // journal已经写成功，出队列

  // journal写好了，数据就真正落盘了，所以执行ondisk回调
  // 注意此时数据还未写入文件系统，所以不可读
  if (ondisk) {
    ondisk_finisher.queue(ondisk); // 放入ondisk_finisher的队列，等待回调
  }

  if (!to_queue.empty()) {
    ondisk_finisher.queue(to_queue);
  }
}

void FileStore::queue_op(OpSequencer *osr, Op *o)
{
  // queue_op按提交时候的顺序调用，必然导致属于同一个OpSequencer的OP按照提交顺序
  // 在OpSequencer内部排队, 保证了PG内部op的先后顺序
  osr->queue(o);
  op_wq.queue(osr);
}

void FileStore::_do_op(OpSequencer *osr, ThreadPool::TPHandle &handle)
{
  wbthrottle.throttle(); // filestore层面writeback的限流

  ......

  // op_tp线程池的多个线程可以并发对同一个OpSequencer执行请求
  // 锁保证同一个OpSequencer中(也即PG中）只能有一个OP在执行
  osr->apply_lock.Lock();

  Op *o = osr->peek_queue(); // 获取一个op

  apply_manager.op_apply_start(o->op);
  int r = _do_transactions(o->tls, o->op, &handle); // 执行写请求到文件系统
  apply_manager.op_apply_finish(o->op);
}

void FileStore::_finish_op(OpSequencer *osr)
{
  list<Context*> to_queue;
  Op *o = osr->dequeue(&to_queue); // 将op从OpSequencer出队列
  osr->apply_lock.Unlock();  // 释放锁，这时候其他线程就可以继续对此QpSequencer执行apply操作

  op_queue_release_throttle(o); // 释放filestore的throttle，见queue_transactions

  if (o->onreadable_sync) {
    o->onreadable_sync->complete(0);
  }
  if (o->onreadable) {
    op_finisher.queue(o->onreadable);
  }
  if (!to_queue.empty()) {
    op_finisher.queue(to_queue); // 放入op_finisher队列，等待执行apply回调，标志数据可读
  }
  delete o;
}

void FileStore::sync_entry()
{
  lock.Lock();
  while (!stop) {
	......

    op_tp.pause(); // 暂停apply线程池的处理
    if (apply_manager.commit_start()) { // 如果有新的请求需要commit, 返回true

      uint64_t cp = apply_manager.get_committing_seq(); // 获取已经apply过的序列号

	  ......

      if (backend->can_checkpoint()) {
		  ......

      } else {
		apply_manager.commit_started(); // 设置block为false，主要是为journal replay服务
		op_tp.unpause(); // 恢复线程池

		int err = backend->syncfs(); // 这里会sync osd的整个current目录

		err = write_op_seq(op_fd, cp); // 记录下commit的序列号
	
		err = ::fsync(op_fd); // 保证更新序列号的操作落盘
      }
      
      apply_manager.commit_finish(); // 完成commit，通知journal
      wbthrottle.clear();

	  ......

    } else {
      op_tp.unpause();
    }
	......
  }
  stop = false;
  lock.Unlock();
}

bool JournalingObjectStore::ApplyManager::commit_start()
{
  bool ret = false;

  uint64_t _committing_seq = 0;
  {
    Mutex::Locker l(apply_lock);

    blocked = true; // 这个仅仅为journal replay起作用

    while (open_ops > 0) { // 等待其他inflight apply 完成
      blocked_cond.Wait(apply_lock);
    }

    {
      Mutex::Locker l(com_lock);
      if (max_applied_seq == committed_seq) {
		blocked = false;
		goto out;
      }

      _committing_seq = committing_seq = max_applied_seq; // 更新序列号
    }
  }
  ret = true;

 out:
  if (journal)
    journal->commit_start(_committing_seq);  // tell the journal too
  return ret;
}

uint64_t JournalingObjectStore::ApplyManager::op_apply_start(uint64_t op)
{
  Mutex::Locker l(apply_lock);
  while (blocked) { // 新的apply操作将会阻塞
    blocked_cond.Wait(apply_lock);
  }

  assert(!blocked);
  assert(op > committed_seq);
  open_ops++;
  return op;
}

void JournalingObjectStore::ApplyManager::commit_started()
{
  blocked = false; // 设置回false
  blocked_cond.Signal(); // 唤醒
}

void JournalingObjectStore::ApplyManager::op_apply_finish(uint64_t op)
{
  ......
  if (blocked) {
    blocked_cond.Signal(); // 唤醒sync线程
  }
  ......
}

int FileStore::queue_transactions(Sequencer *posr, list<Transaction*> &tls,
				  TrackedOpRef osd_op,
				  ThreadPool::TPHandle *handle)
{
	......
    journal->throttle();
	......
}

int FileJournal::prepare_multi_write(bufferlist& bl, uint64_t& orig_ops, uint64_t& orig_bytes)
{
	......
	put_throttle(1, peek_write().bl.length());
	......

}

int FileStore::queue_transactions(Sequencer *posr, list<Transaction*> &tls,
				  TrackedOpRef osd_op,
				  ThreadPool::TPHandle *handle)
{
	......
	op_queue_reserve_throttle(o, handle);
	......
}

void FileStore::_finish_op(OpSequencer *osr)
{
	......
    op_queue_release_throttle(o);
	......
}

struct SequencerPosition {
  uint64_t seq;  ///< seq
  uint32_t trans; ///< transaction in that seq (0-based)
  uint32_t op;    ///< op in that transaction (0-based)
  ......
};

int FileStore::_clone(const coll_t& cid, const ghobject_t& oldoid, const ghobject_t& newoid,
		      const SequencerPosition& spos)
{
  ......
  if (_check_replay_guard(cid, newoid, spos) < 0)
    return 0;

  ......

  // clone is non-idempotent; record our work.
  _set_replay_guard(**n, spos, &newoid);

  ......
}