Linux 通用块层之拥塞控制

为什么需要拥塞控制

前面已经介绍了 Linux 操作系统通用块层中作为数据流动的关键因素 BIO，从软件的角度来讲，只要代码执行得够快，理论上就可以不停的下发数据让后端去执行。当然，都知道这是不可能的，首先局限于底层的硬件限制，磁盘速率一般也不超过 300MB/s，换成 SSD 的话会更快一下但是也不可能无限，另外一方面，代码执行总是需要消耗 CPU 时间，假设底层硬件能力为无穷大，但是通用块层自身限制也只能让下发速率局限于大约 5GB/s，这也就是他的极限通道速率，所以目前社区正在积极改造更为快速的 blk-mq，他的通道速率可以达到更为快速的 130GB/s¹。

尽管通用块层给了多大 5GB/s 的通道速率，但是平常遇到更多的速率限制在于硬盘速度，所以需要一种反馈机制通知更上层目前数据处理不过来，需要缓一缓，这大概就是拥塞控制的由来。

如何使用拥塞控制

拥塞控制是个很容易理解的概念，就是文件系统或者 VFS 缓存系统检测到底层出现拥塞，那么他就等下来等待拥塞状态消失才继续下发。

static inline int wb_congested(struct bdi_writeback *wb, int cong_bits)
{
        struct backing_dev_info *bdi = wb->bdi;

        if (bdi->congested_fn)
                return bdi->congested_fn(bdi->congested_data, cong_bits);
        return wb->congested->state & cong_bits;
}

static inline int bdi_congested(struct backing_dev_info *bdi, int cong_bits)
{
        return wb_congested(&bdi->wb, cong_bits);
}

static inline int bdi_read_congested(struct backing_dev_info *bdi)
{
        return bdi_congested(bdi, 1 << WB_sync_congested);
}

static void ext2_preread_inode(struct inode *inode)
{
        ...
        bdi = inode_to_bdi(inode);
        if (bdi_read_congested(bdi))
                return;
        if (bdi_write_congested(bdi))
                return;
        ...
}

ext2_preread_inode 函数就会首先判断底层是否拥塞，如果是则返回。再比如内存回收部分，也会观察当前的系统 IO 栈是否拥塞，如果拥塞则等待唤醒。

static int current_may_throttle(void)
{
        return !(current->flags & PF_LESS_THROTTLE) ||
                current->backing_dev_info == NULL ||
                bdi_write_congested(current->backing_dev_info);
}

/*
 * shrink_inactive_list() is a helper for shrink_zone().  It returns the number
 * of reclaimed pages
 */
static noinline_for_stack unsigned long
shrink_inactive_list(unsigned long nr_to_scan, struct lruvec *lruvec,
             struct scan_control *sc, enum lru_list lru)
{
        ...
        /*
         * If kswapd scans pages marked marked for immediate
         * reclaim and under writeback (nr_immediate), it implies
         * that pages are cycling through the LRU faster than
         * they are written so also forcibly stall.
         */
        if (nr_immediate && current_may_throttle())
                    congestion_wait(BLK_RW_ASYNC, HZ/10);
        }
        ...
        return nr_reclaimed;
}

这只是其中的几个使用示例。但总的来说，内核会自动设置拥塞状态，是否使用需要开发者自行判断，这其实也是一种辅助手段，譬如有的文件系统尽管可以不论是否拥塞都往下下刷数据，但是他总是需要申请一些限制资源，比如 request，这些资源在底层数据未下刷完成之前总是不会被释放掉的，等全部获取完毕则一定会进入一个强制等待的状态，拥塞机制的主要作用其实是做一个警示作用，他可以在资源尚且没有完全使用完毕的情况下提供驱动一种自我判断的手段，这种手段对于需要灵活性强的，不愿意强制睡眠的进程上下文是很有意义的。

拥塞状态如何控制

介绍完如何使用拥塞控制，接下来介绍到底拥塞位是如何被设置上的。一开始我怀疑是否与硬件寄存器有关，是不是底层磁盘有通用寄存器接口或者 SCSI 命令可以得到一个这样的状态，但是我失败了，没有找到一个这样的硬件接口。那么他到底又是怎么实现拥塞检测的呢？内核有这样的一个接口：

void set_wb_congested(struct bdi_writeback_congested *congested, int sync)
{
        enum wb_state bit;

        bit = sync ? WB_sync_congested : WB_async_congested;
        if (!test_and_set_bit(bit, &congested->state))
                atomic_inc(&nr_wb_congested[sync]);
}
EXPORT_SYMBOL(set_wb_congested);

void clear_wb_congested(struct bdi_writeback_congested *congested, int sync)
{
        wait_queue_head_t *wqh = &congestion_wqh[sync];
        enum wb_state bit;

        bit = sync ? WB_sync_congested : WB_async_congested;
        if (test_and_clear_bit(bit, &congested->state))
                atomic_dec(&nr_wb_congested[sync]);
        smp_mb__after_atomic();
        if (waitqueue_active(wqh))
                wake_up(wqh);
}
EXPORT_SYMBOL(clear_wb_congested);

set_wb_congested 就是设置 BDI 后端磁盘拥塞状态的函数，通过对他的调用可以让当前的 BDI 后端磁盘信息添加上一个拥塞的状态，状态位设置完毕之后，可以被前文提到的 wb_congested 函数调用，从而完成一个这样的闭环。clear_wb_congested 函数就是清除这个拥塞状态位的函数，他的作用就是清除这个拥塞标志。

什么条件下会被设置拥塞

前文已经知道如何控制拥塞状态位，那仅仅只是一个 API 函数，具体在什么条件下设置该标记位还没有提到，本节继续深入在什么条件下会设置拥塞位。

static struct request *__get_request(struct request_list *rl, int rw_flags,
                     struct bio *bio, gfp_t gfp_mask)
{
        ...
        if (rl->count[is_sync]+1 >= queue_congestion_on_threshold(q)) {
                ...
                blk_set_congested(rl, is_sync);
        }
        ...
}

/*
 * Return the threshold (number of used requests) at which the queue is
 * considered to be congested.  It include a little hysteresis to keep the
 * context switch rate down.
 */
static inline int queue_congestion_on_threshold(struct request_queue *q)
{
        return q->nr_congestion_on;
}

没错，就是在获取请求队列时，同步请求队列的数量超出预定义的 q->nr_congestion_on 值时，则设置拥塞状态，标志着当前后台很忙，前台适当做出反应。这个 q->nr_congestion_on 在初始化时定义为了 128。

还有一个动态调整的函数就是：

int blk_update_nr_requests(struct request_queue *q, unsigned int nr)
{
        struct request_list *rl;
        int on_thresh, off_thresh;

        spin_lock_irq(q->queue_lock);
        q->nr_requests = nr;
        blk_queue_congestion_threshold(q);
        on_thresh = queue_congestion_on_threshold(q);
        off_thresh = queue_congestion_off_threshold(q);

        blk_queue_for_each_rl(rl, q) {
                if (rl->count[BLK_RW_SYNC] >= on_thresh)
                        blk_set_congested(rl, BLK_RW_SYNC);
                else if (rl->count[BLK_RW_SYNC] < off_thresh)
                        blk_clear_congested(rl, BLK_RW_SYNC);

                if (rl->count[BLK_RW_ASYNC] >= on_thresh)
                        blk_set_congested(rl, BLK_RW_ASYNC);
                else if (rl->count[BLK_RW_ASYNC] < off_thresh)
                        blk_clear_congested(rl, BLK_RW_ASYNC);

                if (rl->count[BLK_RW_SYNC] >= q->nr_requests) {
                        blk_set_rl_full(rl, BLK_RW_SYNC);
                } else {
                        blk_clear_rl_full(rl, BLK_RW_SYNC);
                        wake_up(&rl->wait[BLK_RW_SYNC]);
                }

                if (rl->count[BLK_RW_ASYNC] >= q->nr_requests) {
                        blk_set_rl_full(rl, BLK_RW_ASYNC);
                } else {
                        blk_clear_rl_full(rl, BLK_RW_ASYNC);
                        wake_up(&rl->wait[BLK_RW_ASYNC]);
                }
        }

        spin_unlock_irq(q->queue_lock);
        return 0;
}

这个函数就更为简单明了，通过与一些预定义值比较从而动态设置拥塞状态是 yes 或者 no。

其他细节

本文提到的拥塞控制都是在未打开 CONFIG_CGROUP_WRITEBACK 的情况下做的分析，因为如果系统开启了 CGROUP_WRITEBACK，则又是走的另外一套拥塞控制逻辑。

在第一章提到的函数 wb_congested 进行拥塞判断时，其实他还有一个自定义拥塞函数的一个判断，也就是这边巴拉巴拉半天，如果后端的 BDI 设备自己要自定义一个拥塞判断的话，前文提到的这些逻辑都是不起效果的。目前我所知道的比如软件 RAID 的代码中他就自定义了 congested 函数，因为没有人会比 md 驱动层更了解他们自己是否拥塞，比如这是 RAID1 的拥塞判断函数。

static int raid1_congested(struct mddev *mddev, int bits)
{
        struct r1conf *conf = mddev->private;
        int i, ret = 0;

        if ((bits & (1 << WB_async_congested)) &&
        ┊   conf->pending_count >= max_queued_requests)
                return 1;

        rcu_read_lock();
        for (i = 0; i < conf->raid_disks * 2; i++) {
                struct md_rdev *rdev = rcu_dereference(conf->mirrors[i].rdev);
                if (rdev && !test_bit(Faulty, &rdev->flags)) {
                        struct request_queue *q = bdev_get_queue(rdev->bdev);

                        BUG_ON(!q);

                        /* Note the '|| 1' - when read_balance prefers
                        ┊* non-congested targets, it can be removed
                        ┊*/
                        if ((bits & (1 << WB_async_congested)) || 1)
                                ret |= bdi_congested(q->backing_dev_info, bits);
                        else
                                ret &= bdi_congested(q->backing_dev_info, bits);
                }
        }
        rcu_read_unlock();
        return ret;
}