ceph rbd：qos

2020-02-19 本文已影响0人 chnmagnus

基本介绍

rbd qos控制采取了令牌桶算法来实现，最初版本及算法介绍见：
https://blog.csdn.net/Dongsheng_Yang/article/details/77689521

最初始的pull request：
https://github.com/ceph/ceph/pull/17032

代码流程

下面是当前master分支的实现（2018.09.19)。

初始化qos控制组件

1.设置相关参数

qos的相关参数是
rbd_qos_iops_limit
rbd_qos_bps_limit
rbd_qos_read_iops_limit
rbd_qos_write_iops_limit
rbd_qos_read_bps_limit
rbd_qos_write_bps_limit

上述参数设为0，表示关闭对应的操作的qos控制，如果>0，则表示开启控制。

2.创建qos控制组件

初始化是在ImageRequestWQ的构造函数中完成的，会为所有类型的qos创建一个TokenBucketThrottle对象，该对象实现了基于令牌桶算法的qos控制策略。
此时，所有qos控制组件的max和avg都是0，表示关闭qos控制。所以此时qos控制不会生效

static std::list<uint64_t> throttle_flags = {
  RBD_QOS_IOPS_THROTTLE,
  RBD_QOS_BPS_THROTTLE,
  RBD_QOS_READ_IOPS_THROTTLE,
  RBD_QOS_WRITE_IOPS_THROTTLE,
  RBD_QOS_READ_BPS_THROTTLE,
  RBD_QOS_WRITE_BPS_THROTTLE
};

ImageRequestWQ::ImageRequestWQ

  for (auto flag : throttle_flags) {
    m_throttles.push_back(make_pair(
      flag, new TokenBucketThrottle(cct, 0, 0, timer, timer_lock)));
  }

3.根据用户参数开启对应的qos控制

在ImageCtx中处理用户参数的函数apply_metadata中，通过ImageRequestWQ得apply_qos_limit函数，为上一步初始化的所有组件设置qos参数。
传入的limit如果为0，表示关闭对应操作的qos控制；大于0表示开启，limit的值会被设置到令牌桶算法的max和avg值上。max表示该桶最多有多少令牌，avg表示每秒向桶中加入多少令牌。

ImageCtx::apply_metadata

io_work_queue->apply_qos_limit(qos_iops_limit, RBD_QOS_IOPS_THROTTLE);
io_work_queue->apply_qos_limit(qos_bps_limit, RBD_QOS_BPS_THROTTLE);
io_work_queue->apply_qos_limit(qos_read_iops_limit, RBD_QOS_READ_IOPS_THROTTLE);
io_work_queue->apply_qos_limit(qos_write_iops_limit, RBD_QOS_WRITE_IOPS_THROTTLE);
io_work_queue->apply_qos_limit(qos_read_bps_limit, RBD_QOS_READ_BPS_THROTTLE);
io_work_queue->apply_qos_limit(qos_write_bps_limit, RBD_QOS_WRITE_BPS_THROTTLE);

qos控制的作用流程

以librbd的aio_write和aio_read为例。

1.发出读写请求到ImageRequestWQ

读、写过程的入口函数分别是：ImageRequestWQ::aio_read 和 ImageRequestWQ::aio_write。其基本流程是：

1)调用start_in_flight_io，将m_in_flight_ios加一
2)获取m_image_ctx.owner_lock的读写锁
3)判断请求是加入ImageRequestWQ异步执行，还是直接执行。
对于读请求，如果我们设置了non_blocking_aio参数，或者有写请求被qos控制组件阻塞（m_write_blockers > 0），或者ImageRequestWQ中存在写请求（m_queued_writes > 0）时，需要将该请求加入队列。
对于写请求，如果我们设置了non_blocking_aio参数，或者有写请求被qos控制组件阻塞（m_write_blockers > 0）时，需要将该请求加入队列。
4)调用ImageRequestWQ::queue函数将请求加入队列的同时，根据请求类型，分别将m_queued_writes或m_queued_reads加一
5)释放m_image_ctx.owner_lock的读写锁

2.ImageRequestWQ中请求的后续处理

ImageRequestWQ对应的线程池中的线程，会从ImageRequestWQ中取出请求，开始做处理，取出请求的函数为ImageRequestWQ::_void_dequeue，在这里，实现了qos控制的分支处理。

1)查看wq队头第一个请求，对其调用needs_throttle函数，
如果需要被blocked，跳到3.;
如果不需要blocked，则直接跳到5.。

3.请求被qos控制组件加入阻塞队列

如果开启了流控，每个请求会有一个m_throttled_flag来标记这个请求被哪些流控组件放行，只有前文所述的六种qos控制类型的flag都被设置到m_throttled_flag，这个请求才会被调度到线程池执行。一个请求，可能被多种流控组件所限制，m_throttled_flag的意义就在此。

needs_throttle函数会遍历所有的qos控制组件，通过m_qos_enabled_flag（标识开启了哪些类型的qos控制）来确认rbd开启了哪些qos控制。对于未开启的qos控制类型，直接为请求设置flag到m_throttled_flag。

当遇到开启的qos控制类型，则需要先通过tokens_requested函数，获得该请求执行所需的令牌数，然后调用qos控制组件的get函数，判断是否有足够的令牌，如果有，设置m_throttled_flag，然后放行。如果没有足够令牌，则将该请求从ImageRequestWQ中取出，放入对应qos控制组件的blocker队列，并注册回调函数handle_throttle_ready。同时会将m_io_throttled加一。

template <typename I>
bool ImageRequestWQ<I>::needs_throttle(ImageDispatchSpec<I> *item) { 
  uint64_t tokens = 0;
  uint64_t flag = 0;
  bool blocked = false;
  TokenBucketThrottle* throttle = nullptr;

  for (auto t : m_throttles) {
    flag = t.first;
    // 判断该类型的qos控制是否已经放行
    if (item->was_throttled(flag))
      continue;
    // 设置flag，表示该类型的qos控制放行
    if (!(m_qos_enabled_flag & flag)) {
      item->set_throttled(flag);
      continue;
    }

    throttle = t.second;
    tokens = item->tokens_requested(flag);
    // 判断是否有足够令牌，不足则将请求放入阻塞队列
    if (throttle->get<ImageRequestWQ<I>, ImageDispatchSpec<I>,
          &ImageRequestWQ<I>::handle_throttle_ready>(
        tokens, this, item, flag)) {
      blocked = true;
    } else {
      item->set_throttled(flag);
    }
  }
  return blocked;
}

4.请求从qos控制组件阻塞队列requeue到ImageRequestWQ队列

qos控制组件存在一个每秒执行的定时器，执行函数为TokenBucketThrottle::schedule_timer。
这个函数会每秒向令牌桶中增加一定数目的令牌，增加令牌后，从前往后遍历阻塞队列中的请求，如果此时的令牌能够满足请求的执行，则将该请求从阻塞队列中取出，为每个取出的请求调用handle_throttle_ready函数。

void TokenBucketThrottle::schedule_timer() {
  add_tokens();

  m_token_ctx = new FunctionContext(
      [this](int r) {
        schedule_timer();
      });

  m_timer->add_event_after(1, m_token_ctx);
}

void TokenBucketThrottle::add_tokens() {
  list<Blocker> tmp_blockers;
  {
    // put m_avg tokens into bucket.
    Mutex::Locker lock(m_lock);
    m_throttle.put(m_avg);
    // check the m_blockers from head to tail, if blocker can get
    // enough tokens, let it go.
    while (!m_blockers.empty()) {
      Blocker blocker = m_blockers.front();
      uint64_t got = m_throttle.get(blocker.tokens_requested);
      if (got == blocker.tokens_requested) {
          // got enough tokens for front.
        tmp_blockers.splice(tmp_blockers.end(), m_blockers, m_blockers.begin());
      } else {
          // there is no more tokens.
        blocker.tokens_requested -= got;
        break;
      }
    }
  }

  for (auto b : tmp_blockers) {
    // 调用handle_throttle_ready函数
    b.ctx->complete(0);
  }
}

handle_throttle_ready函数会设置该流控组件的flag到请求的m_throttled_flag，表示该请求被该组件放行，然后通过item->were_all_throttled()函数判断，该请求是否被所有流控组件放行，如果是，则将请求requeue到ImageRequestWQ的front端，并将m_io_throttled减一；如果否，则不处理（此时该请求已经从该流控组件的阻塞队列中移除，其requeue动作交由阻塞该请求的最后一个流控组件完成）。

template <typename I>
void ImageRequestWQ<I>::handle_throttle_ready(int r, ImageDispatchSpec<I> *item, uint64_t flag) {
  CephContext *cct = m_image_ctx.cct;
  ldout(cct, 15) << "r=" << r << ", " << "req=" << item << dendl;

  ceph_assert(m_io_throttled.load() > 0);
  item->set_throttled(flag);
  if (item->were_all_throttled()) {
    this->requeue(item);
    --m_io_throttled;
    this->signal();
  }
}

5.请求出ImageRequestWQ队列被执行

1）如果是写请求，且!lock_required && !refresh_required，将m_in_flight_writes加一
2）将请求从ImageRequestWQ队列中取出，依次调用ImageDispatchSpec::start_op函数和ImageRequestWQ::process函数完成请求。
3）在process函数中，完成请求后，还会：

调用finish_queued_io将m_queued_reads或m_queued_writes减一（根据请求类型）；

如果是写请求，调用finish_in_flight_write将m_in_flight_writes减一，如果此时m_in_flight_writes减为0，且m_write_blocker_contexts非空，调用flush_image函数，执行flush操作；flush操作结束后会调用handle_blocked_writes函数，将m_write_blocker_contexts中所有context执行完成。

调用finish_in_flight_io，将m_in_flight_ios减一。

一些问题

注：下面是个人理解，可能不正确。
1.设置qos后，客户端超出qos发送请求，有没有相关机制，阻塞请求的发送，如果没有，请求会堆积在哪里？

没有，通过lidrbd，用户可以无限发送aio请求，这些请求会堆积在ImageRequestQueue或qos控制组件的阻塞队列中。取决于qos的限制和线程池处理请求的速度。

比如，用户每秒发送2000个请求，线程池每秒处理10000个请求，qos控制为1000，则每秒阻塞队列都会增加1000个请求。
比如，用户每秒发送20000个请求，线程池每秒处理10000个请求，qos控制为1000，则每秒有9000个请求加入阻塞队列，有10000个请求滞留在ImageRequestQueue。

2.对同一image的混合读写请求，是否有完成顺序的保证？

没有

两个点：
1）线程池从ImageRequestWQ取出请求的过程是顺序的，但取出后的执行过程没有顺序保证。
2）因为流控阻塞队列的存在，被要求流控的类型的请求，在令牌不足时会被加入流控阻塞队列，此时，其后面的不需要流控的请求，会被优先执行。

3.qos是可以针对不同类型的请求设置的，当对写请求设置qos后，读写请求是否有完成顺序的保证？

没有

两个点：
1）读请求是否会被写请求的流控组件所限制；不会
2）设置写请求流控组件会不会同步设置读请求的流控组件；不会

第一点可从下面类看出。该仿函数根据传入的不同请求，返回该请求所需的令牌数，如代码所示，读请求在遇到写类型的流控组件时，返回的所需令牌数为0；反之亦然。

template <typename I>
struct ImageDispatchSpec<I>::TokenRequestedVisitor
  : public boost::static_visitor<uint64_t> {
  ImageDispatchSpec* spec;
  uint64_t flag;

  TokenRequestedVisitor(ImageDispatchSpec* spec, uint64_t _flag)
    : spec(spec), flag(_flag) {
  }

  uint64_t operator()(const Read&) const {
    if (flag & RBD_QOS_WRITE_MASK) {
      return 0;
    }

    if (flag & RBD_QOS_BPS_MASK) {
      return spec->extents_length();
    }
    return 1;
  }

  uint64_t operator()(const Flush&) const {
    return 0;
  }

  template <typename T>
  uint64_t operator()(const T&) const {
    if (flag & RBD_QOS_READ_MASK) {
      return 0;
    }

    if (flag & RBD_QOS_BPS_MASK) {
      return spec->extents_length();
    }
    return 1;
  }
};