Глава 4. Cgroups

Под справедливостью (fairness) мы имеем ввиду, что каждая cgroup получает справедливую долю значения выдачи ввода/ вывода в это устройство. Для осуществления этого следует настроить планировщик CFQ (Complete Fair Queuing). В отсутствии cgroup планировщик CFQ назначает каждому процессу некую очередь, а затем предоставляем расслоение по времени для каждой из очередей, тем самым обрабатывая справедливость.

Некое дерево обслуживания это список активных очередей/ процессов над которым работает планировщик. Итак, в своей основе, планировщик CFQ обслуживает запросы из всех очередей в дереве обслуживания.

При работе с cgroup вводится понятие CFQ групп. Теперь вместо планирования по процессам собственно планирование выполняется на уровне групп. Это означает,что каждая cgroup обладает множеством деревьев обслуживания в которых планируются очереди этих групп. Затем имеется некое глобальное дерево обслуживания в котором планируются сами группы CFQ.

Группа CFQ определяется следующим образом:

struct cfq_group {
   /* должен быть самым первым участником */
   struct blkg_policy_data pd;

   /* участник группы service_tree */
   struct rb_node rb_node;

   /* ключ группы service_tree */
u64 vdisktime;

    /*
    * Значение числа cfqgs и сумма их весов под этой cfqg.
    * Она покрывает эти leaf_weight cfqg е веса всех потомков, 
    * однако не покрывает веса всех последующих потомков.
    *
    * Когда cfqg присутствует в дереве обслуживания, она активна. Некая активная cfqg
    * также активирует своих предков и вносит вклад в значение children_weight
    * своего предка.
    */
   int nr_active;

   unsigned int children_weight;

    /*
    * vfraction это доля vdisktime которые озаглавлены в этой
    * cfqg. Она определяется составлением значений отношений 
    * проходом вверх по этой cfqg к корню.
    *
    * Она пребывает в фиксированной точке с/ CFQ_SERVICE_SHIFT, а значение суммы всех
    * vfractions в неком дереве обслуживания примерно равно 1. Сумма может слегка отличаться 
    * по причине ошибок округления и отклонений, вызванных входом и выходом  
    * cfqgs из дерева обслуживания.
    */
   * unsigned int vfraction;

   /*
    * Имеются два веса - (внутренний) вес это вес всех этих
    * cfqg относительно cfqgs одного уровня. leaf_weight это значение веса 
    * такой cfqg относительно всех дочерних cfqgs. Для значения корневой cfqg, оба
    * веса выдерживаются синхронными для обратной совместимости.
    */

   unsigned int weight;
   unsigned int new_weight;
   unsigned int dev_weight;
   unsigned int leaf_weight;
   unsigned int new_leaf_weight;
   unsigned int dev_leaf_weight;

   /* число cfqq находящихся в данный момент вэтой группе */
   int nr_cfqq;
   /*
    * Среднее число занятых очередей в группе. Полезно при расчёте среза рабочей группы.
    * Мы создаём массив для каждого класса prio, но во время выполнения он используется
    * только для классов RT и BE, а слот для класса IDLE остаётся не использованным.
    * Это в первую очередь сделано во избежание путаницы и предупреждающих сообщений gcc.
    */

   unsigned int
   busy_queues_avg[CFQ_PRIO_NR]; /*
    *rr перечисляет все очереди с запросами. Мы поддерживаем деревья обслуживания для
    *классов RT и BE. Эти деревья подразделяются на подклассы
    * на базе SYNC, SYNC_NOIDLE и ASYNC в типе рабочей нагрузки. Для
    * класса IDLE нет никаких подклассов и все имеющиеся очереди CFQ идут в 
    * некое отдельное дерево обслуживания service_tree_idle.
    * Сч1тчики встраиваются в cfq_rb_root
    */

   struct cfq_rb_root service_trees[2][3];
   struct cfq_rb_root service_tree_idle;

   u64 saved_wl_slice;
   enum wl_type_t saved_wl_type;
   enum wl_class_t saved_wl_class;

     /* число запросов, которые пребывают в списке отправки или внутри драйвера */
   int dispatched;
   struct cfq_ttime ttime;
   struct cfqg_stats stats;       /* статистики для этой cfqg */

   /* асинхронная очередь для каждого варианта приоритета */ struct
   cfq_queue *async_cfqq[2][IOPRIO_BE_NR]; struct
   cfq_queue *async_idle_cfqq;

};
 	   

Каждая группа CFQ содержит некое значение "веса ввода/ вывода", которое можно настраивать в cgroup. Значения vdisktime CFQG (групп CFQG) определяются их положением в "дереве обслуживания cfqg", а затем они изменяются в соответствии с "весом ввода/ вывода".

Дросселирование (throttling) предоставляет некое средство для применения ограничения ресурсов к имеющемуся блочному вводу/ выводу. Это делает возможным для ядра осуществлять контроль над значением максимального блочного ввода/ вывода, который может получать процесс пространства пользователя. Имеющееся ядро реализует его через cgroup блочного ввода/ вывода.

Дросселирование имеющегося блочного ввода/ вывода для cgroup выполняется с применением некого набора различных функций. Самой первой функцией является blk_throttl_bio и она определена в blk-throttle.c:

bool blk_throtl_bio(struct request_queue *q, struct blkcg_gq
            *blkg, struct bio *bio)
{
   struct throtl_qnode *qn = NULL;
   struct throtl_grp *tg = blkg_to_tg(blkg ?: q->root_blkg); struct throtl_service_queue *sq; bool rw = bio_data_dir(bio);
   bool throttled = false;
   struct throtl_data *td = tg->td;
   WARN_ON_ONCE(!rcu_read_lock_held());

   /* см. throtl_charge_bio() */
   if (bio_flagged(bio, BIO_THROTTLED) || !tg->has_rules[rw])
   goto out;

   spin_lock_irq(q->queue_lock);

   throtl_update_latency_buckets(td);

   if (unlikely(blk_queue_bypass(q)))
   goto out_unlock;

   blk_throtl_assoc_bio(tg, bio);
   blk_throtl_update_idletime(tg);

   sq = &tg->service_queue;

   again :

     while (true) {
        if (tg->last_low_overflow_time[rw] == 0) tg->last_low_overflow_time[rw] = jiffies;
         throtl_downgrade_check(tg);
         throtl_upgrade_check(tg);
        /* throtl выполнен как FIFO - если bios уже в очереди, он должен там присутствовать */
        if (sq->nr_queued[rw])
            break;

        /* прерываем очередь в случае превышения предела */
        if (!tg_may_dispatch(tg, bio, NULL)) {
            tg->last_low_overflow_time[rw] = jiffies;
            if (throtl_can_upgrade(td, tg)) {
                   throtl_upgrade_state(td);
                goto again;
            }
            break;
   }

   /* находясь в рамках предела, давайте выполним начисление и отправим напрямую */
   throtl_charge_bio(tg, bio);
   /*
    * Необходимо подрезать сечение, даже когда bio не были поставлен в очередь,
    * в противном случае может получиться, что bio находится вне очереди 
    * длительное время и сечение продолжит расширяться, а обрезка не
    * вызывается очень долго. Теперь, если пределы внезапно уменьшаются, 
    * мы принимаем во внимание все операции ввода/ вывода, отправленные до сих пор с новой
    * низкой скоростью, а * вновь поставленные в очередь операции ввода/ вывода получают действительно
    * продолжительное время отправки.
    *

    * Так что продолжайте подрезать сечение, даже если bio не стоит в очереди. */
   throtl_trim_slice(tg, rw);

    /*
    * @bio передаётся через этот уровень без дросселирования.
    * Идём вверх по лестнице.  Если мы уже на самом верху, 
    * можно выполнять непосредственно.
    **/
   qn = &tg->qnode_on_parent[rw];

   sq = sq->parent_sq;
   tg = sq_to_tg(sq);
   if (!tg)
       goto out_unlock;
   }

   /* out-of-limit, ставим в очередь к @tg */
   throtl_log(sq, "[%c] bio. bdisp=%llu sz=%u bps=%lluiodisp=%u iops=%u queued=%d/%d",

      rw == READ ? 'R' : 'W',
     tg->bytes_disp[rw], bio->bi_iter.bi_size,
      tg_bps_limit(tg, rw),
     tg->io_disp[rw], tg_iops_limit(tg, rw),
      sq->nr_queued[READ], sq->nr_queued[WRITE]);

   tg->last_low_overflow_time[rw] = jiffies;

   td->nr_queued[rw]++;
   throtl_add_bio_tg(bio, qn, tg);
   throttled = true;

     /*
     * Обновите время отправки @tg и принудительно отправьте расписание если @tg
     * был пуст до @bio. Принудительное планирование вряд ли 
     * вызовет неоправданную задержку, поскольку @bio, скорее всего, будет отправлен напрямую, 
     * если запланированное время @tg не пребывает в будущем.
     */
       if (tg->flags & THROTL_TG_WAS_EMPTY) {
           tg_update_disptime(tg);
           throtl_schedule_next_dispatch(tg->service_queue.parent_sq, true);
      }

    out_unlock:
      spin_unlock_irq(q->queue_lock);

    out:
     bio_set_flag(bio, BIO_THROTTLED);

    #ifdef CONFIG_BLK_DEV_THROTTLING_LOW
      if (throttled || !td->track_bio_latency) bio->
          bi_issue.value |= BIO_ISSUE_THROTL_SKIP_LATENCY;
     #endif

    return throttled;
   }
 	   

Следующий фрагмент кода проверяет может ли данный bio быть отправлен для активной доставки в драйвер своего устройства:

   if (!tg_may_dispatch(tg, bio, NULL)) { tg->last_low_overflow_time[rw] = jiffies;
               if (throtl_can_upgrade(td, tg)) {
                       throtl_upgrade_state(td);
                   goto again;
               }
               break;
   }
 	   

Здесь приводится определение tg_may_dispatch:

static bool tg_may_dispatch(struct throtl_grp *tg, struct bio  *bio, unsigned long *wait)
{
    bool rw = bio_data_dir(bio);
    unsigned long bps_wait = 0, iops_wait = 0, max_wait = 0;

    /*
     * В настоящий момент вся машина состояний группы целиком зависит от первого bio
     * находящегося в очереди в списке bio этой группы. Поэтому не следует вызывать 
     * эту функцию с другим bio когда в очереди имеются прочие bio.
     * /
     BUG_ON(tg-&service_queue.nr_queued[rw] &&
       bio != throtl_peek_queued(&tg-&service_queue.queued[rw]));

     /* Когда tg-&bps = -1, BW неограничен */
        if (tg_bps_limit(tg, rw) == U64_MAX &&
             tg_iops_limit(tg, rw) == UINT_MAX) {
            if (wait)
                 *wait = 0;
            return true;
       }

     /*
     * Когда предыдущий срез закончился, начните новый, в противном случае
     * обновите/расширьте уже имеющийся срез чтобы быть уверенным, что он по крайней мере пребывает в интервале
     * throtl_slice достаточно давно. Новый срез запускается только для пустой группы 
     * дросселирования. Когда в очереди имеется bio, это означает, что должен иметься некий
     * активный срез и вместо этого его надлежит расширить.
     * /

    if (throtl_slice_used(tg, rw) &&
       !(tg-&service_queue.nr_queued[rw]))
           throtl_start_new_slice(tg, rw);
       else {
            if (time_before(tg-&slice_end[rw],
               jiffies + tg-&td-&throtl_slice))
                 throtl_extend_slice(tg, rw,
                   jiffies + tg-&td-&throtl_slice);
    }

    if (tg_with_in_bps_limit(tg, bio, &bps_wait) &&
        tg_with_in_iops_limit(tg, bio, &iops_wait)) {
         if (wait)
           *wait = 0 ;
        return true;
   }

   max_wait = max(bps_wait, iops_wait);

   if (wait)
         *wait = max_wait;

    if (time_before(tg-&slice_end[rw], jiffies + max_wait))
       throtl_extend_slice(tg, rw, jiffies + max_wait);

    return false;
 	   

Фрагмент кода

   if (tg_with_in_bps_limit(tg, bio, &bps_wait) &&
      tg_with_in_iops_limit(tg, bio, &iops_wait)) {
        if (wait)
         *wait = 0;
        return true;
   }
 	   

определяет пребывает ли этот bio внутри установленных пределов данной cgruop или нет. Как видно, он проверяет для данной cgroup как ограничение в байтах в секунду, так и предел числа вводав/ выводов в секунду.

Когда предел не превышен, этот bio загружается в данную cgroup первым:

/* находясь в рамках, давайте заполним и отправим непосредственно and dispatch directly */ throtl_charge_bio(tg, bio);

static void throtl_charge_bio(struct throtl_grp *tg, struct bio *bio) {
   bool rw = bio_data_dir(bio);
   unsigned int bio_size =
                           throtl_bio_data_size (bio);

    /* Заносим этот bio в его группу */
    tg->bytes_disp[rw] += bio_size;
    tg->io_disp[rw]++;
    tg->last_bytes_disp[rw] += bio_size;
    tg->last_io_disp[rw]++;

     /*
    * BIO_THROTTLED применяется для предотвращения дросселирования того же самого bio 
    * более одного раза, поскольку дросселированный bio проследует через blk-throtl 
    * второй раз когда он в конечном счёте будет выдан. Установите его когда bio
    * загружается до tg.
    */

     if (!bio_flagged(bio, BIO_THROTTLED))
          bio_set_flag(bio, BIO_THROTTLED);
     }
 	   

Данная функция загружает этот bio (в bytes и iops) в группу throttle. Затем она передаёт этот bio вверх своему предку, что видно из приводимого ниже кода:

	/*
     * @bio передаётся через этот уровень без дросселирования.
     * Поднимаемся вверх по лестнице.  Если мы уже на самом верху, 
     * Можно выполнять непосредственно.
     */
     qn = &tg->qnode_on_parent[rw];
     sq = sq->parent_sq;
     tg = sq_to_tg(sq);
 	   

Если превышены ограничения, этот код принимает другой ход. Вызывается следующий фрагмент кода:

throtl_add_bio_tg(bio, qn, tg);
throttled = true;
 	   

Давайте взглянем на функцию throtl_add_bio_tg более внимательно:

     /**
     * throtl_add_bio_tg - добавить bio в предписанный throtl_grp
     * @bio: подлежащий добавлению bio 
     * @qn: qnode для применения
     * @tg: целевая throtl_grp
     *

    Добавить @bio в service_queue @tg при помощи @qn. Если @qn не задан,
    применяется tg->qnode_on_self[].
    */

static void throtl_add_bio_tg(struct bio *bio, struct
throtl_qnode *qn,
             struct throtl_grp *tg)
{

    struct throtl_service_queue *sq = &tg->service_queue; bool rw = bio_data_dir(bio);
    if (!qn)
          qn = &tg->qnode_on_self[rw];

     /*
     * Если @tg в настоящий момент не обладает никакими очередями bio в том же самом 
     * направлении, устанавливаемая очередь @bio может измениться когда должен быть отправлен 
     * @tg. Отметьте что @tg был пустым.  Это автоматически
     * очищается при следующем tg_update_disptime().
     */
   if (!sq->nr_queued[rw])
       tg->flags |= THROTL_TG_WAS_EMPTY;

   throtl_qnode_add_bio(bio, qn, &sq->queued[rw]);

   sq->nr_queued[rw]++;
   throtl_enqueue_tg(tg);
   }
 	   

Данная функция добавляет полученный bio в очередь обслуживания throttle. Такая очередь действует как механизм для дросселирования запросов bio. Этот запрос на обслуживание затем позднее осушается.

/**

     *blk_throtl_drain - осушить дросселированные bio
     *@q: request_queue чтобы осушить дросселированные bio для
     *отправки всех дросселированных в данный момент bio в @q через
     - >make_request_fn().
    */

   void blk_throtl_drain(struct request_queue *q)
   __releases(q->queue_lock) __acquires(q->queue_lock)
{
   struct throtl_data *td = q->td; struct blkcg_gq *blkg;

   struct cgroup_subsys_state *pos_css;
   struct bio *bio;
   int rw;

   queue_lockdep_assert_held(q);

   rcu_read_lock();
     /*
     * Осушить все tg при выполнении проход в обратном порядке по дереву blkg с тем,
     * чтобы все bio распространились в td->service_queue.  Было бы лучше
     * непосредственно пройти по дереву service_queue, но проход по blkg 
     * выполняется проще.
     */

   blkg_for_each_descendant_post(blkg, pos_css, td->queue->root_blkg)
      tg_drain_bios(&blkg_to_tg(blkg)->service_queue);

    /* наконец, передаём bio из tg верхнего уровня в установленный td */
    tg_drain_bios(&td->service_queue);

    rcu_read_unlock();
    spin_unlock_irq(q->queue_lock);

   /* все bio теперь должны пребывать в td->service_queue, выдаём их
   */ for (rw = READ; rw <= WRITE; rw++)
          while ((bio = throtl_pop_queued(&td->service_queue.queued[rw],
                                 NULL)))
                    generic_make_request(bio);
    spin_lock_irq(q->queue_lock) ;