文件系统学习8——文件系统MQ队列机制详解

素颜马尾好姑娘i 2022-03-10 00:43 995阅读 0赞

上一篇已经讲述了MQ多队列的机制，利用cpu的多核，配上多队列机制，并发的处理IO请求，提高效率。  
本篇详细讲述下从bio下发到IO调度器中，MQ队列机制是如何一步步完成的。

#### 1、MQ处理结构流图 ####

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTQxODk_size_16_color_FFFFFF_t_70]  
从整个流程图可以看到，主要是分为三个部分：初始化硬件设备的target参数、初始化请求队列request\_queue以及bio请求的处理过程。前面两个过程主要是完成底层存储设备向文件系统的注册，同时完成软硬队列映射关系等初始化，后一个部分是bio在MQ机制最后生成对应子请求并挂载在硬件队列上的过程。

#### 2、scsi设备初始化 ####

对于走scsi协议的底层存储设备，均完成此初始化过程，内核为每个scsi设备提供给一个target的设备参数，该参数主要包括：硬件队列深度，超时时间，硬件中断函数等。  
同时，target的设备参数后序作为参数传入requset\_queue结构体中，用来初始化硬件队列相关参数。  
具体过程如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTQxODk_size_16_color_FFFFFF_t_70 1]

#### 3、request\_queue队列的初始化 ####

之前已经提到过，linux内核支持单队列的机制，也支持多队列的机制，并且在内核中IO调用通过的调用函数都是make\_request\_fn，那么内核如何知道选用的是多队列机制还是单队列机制呢？？  
通过make\_request\_fn函数的注册。

初始化步骤如下：

*  当底层存储设备是单通道时，此时会向内核make\_request\_fn注册实体函数，函数名为blk\_sq\_make\_request；当设备是多通道时，此时会向内核make\_request\_fn注册实体函数，函数名为blk\_mq\_make\_request。
 *  当注册完成后，会根据各自队列的数量，按照内核提供的映射关系，形成cpu号(软件队列号)到硬件队列号的map数组。
 *  按照cpu个数和硬件队列数分别分配软硬件队列环境并进行初始化
 *  将互为映射的软硬件队列环境的参数相互关联。

具体函数流程图如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTQxODk_size_16_color_FFFFFF_t_70 2]  
关于队列初始化的源码阅读如下：  
1.首先在driver\\scsi文件夹下的scsi\_scan.c文件中，scsi\_alloc\_sdev函数会判断当前scsi支持的块设备是否支持multiqueue：

if (shost_use_blk_mq(shost))    // multiple queue is enabled   gaocm
         sdev->request_queue = scsi_mq_alloc_queue(sdev);    //MQ队列的定义   
     else
         sdev->request_queue = scsi_alloc_queue(sdev);

2.若上述设备支持mq则进入scsi\_mq\_alloc\_queue函数，该函数中主要进行设备队列的初始化，blk\_mq\_init\_queue，在blk\_mq\_init\_queue 函数中根据set信息进行与该设备队列相关的信息参数初始化，过程如下：

/* mark the queue as mq asap */
       q->mq_ops = set->ops;   //标记为MQ队列   
    
       q->queue_ctx = alloc_percpu(struct blk_mq_ctx); //获得per_cpu的地址 建立software queue环境 
       if (!q->queue_ctx)
           goto err_exit;
    
       q->queue_hw_ctx = kzalloc_node(nr_cpu_ids * sizeof(*(q->queue_hw_ctx)),
                           GFP_KERNEL, set->numa_node);    //获得hardware queue上下文环境 
       if (!q->queue_hw_ctx)
           goto err_percpu;
    
       q->mq_map = blk_mq_make_queue_map(set);   // 建立software queue和hardware queue之间的映射关系 
       if (!q->mq_map)
           goto err_map;
    
       blk_mq_realloc_hw_ctxs(set, q);
       if (!q->nr_hw_queues)
           goto err_hctxs;
    
       INIT_WORK(&q->timeout_work, blk_mq_timeout_work);
       blk_queue_rq_timeout(q, set->timeout ? set->timeout : 30 * HZ); //定义scsi设备队列超时设定    
    
       q->nr_queues = nr_cpu_ids;
    
       q->queue_flags |= QUEUE_FLAG_MQ_DEFAULT;
    
       if (!(set->flags & BLK_MQ_F_SG_MERGE))
           q->queue_flags |= 1 << QUEUE_FLAG_NO_SG_MERGE;
    
       q->sg_reserved_size = INT_MAX;
    
       INIT_WORK(&q->requeue_work, blk_mq_requeue_work);
       INIT_LIST_HEAD(&q->requeue_list);
       spin_lock_init(&q->requeue_lock);
    
       if (q->nr_hw_queues > 1)
           blk_queue_make_request(q, blk_mq_make_request); //注册q->make_request_fn函数        
       else
           blk_queue_make_request(q, blk_sq_make_request);
    
       /*
        * Do this after blk_queue_make_request() overrides it...
        */
       q->nr_requests = set->queue_depth;
    
       if (set->ops->complete)
           blk_queue_softirq_done(q, set->ops->complete);
    
       blk_mq_init_cpu_queues(q, set->nr_hw_queues);
    
       get_online_cpus();
       mutex_lock(&all_q_mutex);
    
       list_add_tail(&q->all_q_node, &all_q_list);
       blk_mq_add_queue_tag_set(set, q);
       blk_mq_map_swqueue(q, cpu_online_mask);
    
       mutex_unlock(&all_q_mutex);
       put_online_cpus();
    
       return q;
    
    err_hctxs:
       kfree(q->mq_map);
    err_map:
       kfree(q->queue_hw_ctx);
    err_percpu:
       free_percpu(q->queue_ctx);
    err_exit:
       q->mq_ops = NULL;
       return ERR_PTR(-ENOMEM);
    }

3.完成上述的初始化之后，在blk\_queue\_make\_request中将q->make\_request\_fn注册为blk\_mq\_make\_request。

#### 4、bio请求处理 ####

bio的请求处理是device mapper中提交一个个bio请求到IO schedule中之后完成的。整个处理的流图如下图所示：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTQxODk_size_16_color_FFFFFF_t_70 3]  
首先，bio请求被提交(submit\_bio)，此时进入generic\_make\_request，表明其将在块设备层中被进行相关处理工作。  
generic\_make\_request函数如下：

blk_qc_t generic_make_request(struct bio *bio)
    {
        struct bio_list bio_list_on_stack;
        blk_qc_t ret = BLK_QC_T_NONE;
    
        if (!generic_make_request_checks(bio))  //判断当前bio是否有效       gaocm
            goto out;
        if (current->bio_list) {
            bio_list_add(current->bio_list, bio);
            goto out;
        }
        //上述过程要求当前的make_request_fn每次只能被触发一次，因此，通过current->bio_list判断当前是否有bio在其中，若有则将当前这个加入到尾部等待被处理，若没有则可直接处理该bio   
    
        BUG_ON(bio->bi_next);
        bio_list_init(&bio_list_on_stack);      //初始化该双向链表  
        current->bio_list = &bio_list_on_stack; //当前为NULL   
        do {
            struct request_queue *q = bdev_get_queue(bio->bi_bdev); //获得bio对应的设备队列  
    
            if (likely(blk_queue_enter(q, false) == 0)) {   //判断当前的设备队列是否有效能够响应该请求  
                ret = q->make_request_fn(q, bio);   //将bio进行进一步处理，放入块设备层的处理队列中  
                blk_queue_exit(q);
    
                bio = bio_list_pop(current->bio_list);  
            } else {
                struct bio *bio_next = bio_list_pop(current->bio_list);
    
                bio_io_error(bio);
                bio = bio_next;
            }
        } while (bio);
        current->bio_list = NULL; /* deactivate */  //clear this bio list and make_request function is avalible again
    
    out:
        return ret;
    }

没错，从代码上分析和我们之前讲的一样，bio会通过队列机制进行一个管理，当处理一个bio时，首先或得当前bio的设备队列（即软硬件队列），然后判断当前的设备队列是否有效能够响应该请求 ，如果能够则调用make\_request\_fn，该函数即是硬件初始化时候注册的函数。

然后，对于MQ机制，make\_request\_fn的重载即为blk\_mq\_make\_request函数，该函数如下：

const int is_sync = rw_is_sync(bio->bi_rw); //判断是否为同步       
        const int is_flush_fua = bio->bi_rw & (REQ_FLUSH | REQ_FUA);//判断是否为屏障IO    
        struct blk_map_ctx data;
        struct request *rq;
        unsigned int request_count = 0;
        struct blk_plug *plug;
        struct request *same_queue_rq = NULL;
        blk_qc_t cookie;
    
        blk_queue_bounce(q, &bio);      //做DMA时的相关地址限制，可能该bio只能访问低端内存，因此需要将高端内存中的bio数据拷贝到低端内存中  
    
        if (bio_integrity_enabled(bio) && bio_integrity_prep(bio)) {    //bio完整性判断
            bio_io_error(bio);
            return BLK_QC_T_NONE;
        }
    
        blk_queue_split(q, &bio, q->bio_split); //判断当前的bio是否超过了预设最大处理大小，若是则进行拆分，拆分后会进行generic_make_request函数调用  
    
        if (!is_flush_fua && !blk_queue_nomerges(q) &&
            blk_attempt_plug_merge(q, bio, &request_count, &same_queue_rq)) //若非屏障IO并且设备队列支持合并且plug队列中可进行合并则进行合并工作  
            return BLK_QC_T_NONE;
    
        rq = blk_mq_map_request(q, bio, &data); //在mq中注册一个request       
        if (unlikely(!rq))
            return BLK_QC_T_NONE;
    
        cookie = blk_tag_to_qc_t(rq->tag, data.hctx->queue_num);
    
        if (unlikely(is_flush_fua)) {
            blk_mq_bio_to_request(rq, bio); //将bio转换为request    
            blk_insert_flush(rq);   //若是屏障IO则将其加入到flush队列中，该队列直接发送至driver   
            goto run_queue;
        }
    
        plug = current->plug;
        /*
         * If the driver supports defer issued based on 'last', then
         * queue it up like normal since we can potentially save some
         * CPU this way.
         */
        if (((plug && !blk_queue_nomerges(q)) || is_sync) &&    //有plug队列，且设备队列支持合并或者改请求是同步请求。。
            !(data.hctx->flags & BLK_MQ_F_DEFER_ISSUE)) {   //延迟发送  
            struct request *old_rq = NULL;
    
            blk_mq_bio_to_request(rq, bio); //转化为request    
    
            /*
             * We do limited pluging. If the bio can be merged, do that.
             * Otherwise the existing request in the plug list will be
             * issued. So the plug list will have one request at most
             */
            if (plug) {
                /*
                 * The plug list might get flushed before this. If that
                 * happens, same_queue_rq is invalid and plug list is
                 * empty
                 */
                if (same_queue_rq && !list_empty(&plug->mq_list)) {
                    old_rq = same_queue_rq;
                    list_del_init(&old_rq->queuelist);  //判断之前是否有能合并或者一样的请求，若有则删除之前的请求  
                }
                list_add_tail(&rq->queuelist, &plug->mq_list);  //将该请求加入到plug队列中
            } else /* is_sync */
                old_rq = rq;
            blk_mq_put_ctx(data.ctx);   
            if (!old_rq)    //无为处理请求    
                goto done;
            if (!blk_mq_direct_issue_request(old_rq, &cookie))  //直接加入到底层scsi层队列中，并发往driver?    
                goto done;
            blk_mq_insert_request(old_rq, false, true, true);   //加入到software queue中    
            goto done;
        }
    
        if (!blk_mq_merge_queue_io(data.hctx, data.ctx, rq, bio)) { //底层driver支持延迟发送或者为async请求  
            //能合并则进行合并，否则加入到software queue中 
            /*
             * For a SYNC request, send it to the hardware immediately. For
             * an ASYNC request, just ensure that we run it later on. The
             * latter allows for merging opportunities and more efficient
             * dispatching.
             */
    run_queue:
            blk_mq_run_hw_queue(data.hctx, !is_sync || is_flush_fua);   //执行hardware queue  
        }
        blk_mq_put_ctx(data.ctx);
    done:
        return cookie;
    }

该函数完成bio转换为请求的下发，主要分为以下四种情况：  
1、对于flush等屏障IO，不通过软件队列，直接下发对应的flush队列，因为屏障IO具有非延时的特性，需要写入到driver中，不能被队列所阻塞。  
2、对于非屏障IO，首先判断队列里是否能合并，如果不能合并，则产生一个request，下发到plug队列中，plug队列命中，则直接返回。  
3、plug队列未命中，则将该请求加入到plug队列中，注意，实际在MQ机制中plug队列没有泄洪的作用，实际其队列深度不超过1.对于这类加入到plug队列的请求，最终还是下发到软件队列中。  
4、下发到软件队列的请求同样判断是否能合并，合并则直接返回，不能合并则受软件队列的调度器调度，最后加入到硬件队列中。

这四种情况分别对应的源码如下图所示：

*  blk\_mq\_direct\_issue\_request 将请求直接加入到底层driver中，判断当前scsi设备能否处理该请求

static int blk_mq_direct_issue_request(struct request *rq, blk_qc_t *cookie)
    {
        int ret;
        struct request_queue *q = rq->q;
        struct blk_mq_hw_ctx *hctx = q->mq_ops->map_queue(q,
                rq->mq_ctx->cpu);
        struct blk_mq_queue_data bd = {
            .rq = rq,
            .list = NULL,
            .last = 1
        };
        blk_qc_t new_cookie = blk_tag_to_qc_t(rq->tag, hctx->queue_num);
    
        /*
         * For OK queue, we are done. For error, kill it. Any other
         * error (busy), just add it to our list as we previously
         * would have done
         */
        ret = q->mq_ops->queue_rq(hctx, &bd);   //直接放入scsi队列中，返回是否能够被处理 
        if (ret == BLK_MQ_RQ_QUEUE_OK) {
            *cookie = new_cookie;
            return 0;
        }
    
        __blk_mq_requeue_request(rq);   //标记该request的nr_phys_segments减1 
    
        if (ret == BLK_MQ_RQ_QUEUE_ERROR) {
            *cookie = BLK_QC_T_NONE;
            rq->errors = -EIO;
            blk_mq_end_request(rq, rq->errors);
            return 0;
        }
    
        return -1;
    }

*  blk\_mq\_insert\_request 将请求加入到software queue队列中：

void blk_mq_insert_request(struct request *rq, bool at_head, bool run_queue,
            bool async)
    {
        struct request_queue *q = rq->q;
        struct blk_mq_hw_ctx *hctx;
        struct blk_mq_ctx *ctx = rq->mq_ctx, *current_ctx;
    
        current_ctx = blk_mq_get_ctx(q);    //获得software queue环境    
        if (!cpu_online(ctx->cpu))
            rq->mq_ctx = ctx = current_ctx;
    
        hctx = q->mq_ops->map_queue(q, ctx->cpu);   //找到对应的hardware queue上下文环境  
    
        spin_lock(&ctx->lock);
        __blk_mq_insert_request(hctx, rq, at_head); //通过rq找到ctx，加入到software queue中  
        spin_unlock(&ctx->lock);
    
        if (run_queue)
            blk_mq_run_hw_queue(hctx, async);   //运行hardware queue，用异步方式执行  
    
        blk_mq_put_ctx(current_ctx);
    }

*  blk\_mq\_merge\_queue\_io 判断能否与当前software queue中的请求进行合并

static inline bool blk_mq_merge_queue_io(struct blk_mq_hw_ctx *hctx,
                         struct blk_mq_ctx *ctx,
                         struct request *rq, struct bio *bio)
    {
        if (!hctx_allow_merges(hctx) || !bio_mergeable(bio)) {  //不允许merge  
            blk_mq_bio_to_request(rq, bio);
            spin_lock(&ctx->lock);
            __blk_mq_insert_request(hctx, rq, false);   //加入到software queue中    
            spin_unlock(&ctx->lock);
            return false;
        } else {
            struct request_queue *q = hctx->queue;
    
            spin_lock(&ctx->lock);
            if (!blk_mq_attempt_merge(q, ctx, bio)) {   //进行合并尝试    
                blk_mq_bio_to_request(rq, bio); //无法合并则转向加入software queue中  
                goto insert_rq;
            }
    
            spin_unlock(&ctx->lock);
            __blk_mq_free_request(hctx, ctx, rq);   //将刚刚在software queue和hardware queue中注册的request去除，因为请求已经加入到software queue中       
            return true;
        }
    }

*  硬件队列请求执行blk\_mq\_run\_hw\_queue函数，下发请求到scsi驱动层中

void blk_mq_run_hw_queue(struct blk_mq_hw_ctx *hctx, bool async)
    {
        if (unlikely(test_bit(BLK_MQ_S_STOPPED, &hctx->state) ||
            !blk_mq_hw_queue_mapped(hctx)))
            return;
    
        if (!async) {   // false and run  gaocm
        //若async为flash则说明该处理是同步的，需要马上处理，若是异步则将该操作交由kblocked进行处理 
            int cpu = get_cpu();
            if (cpumask_test_cpu(cpu, hctx->cpumask)) {  // cpu is set in cpumask    
                __blk_mq_run_hw_queue(hctx);    //运行hardware queue  
                put_cpu();
                return;
            }
    
            put_cpu();
        }
    
        kblockd_schedule_delayed_work_on(blk_mq_hctx_next_cpu(hctx),
                &hctx->run_work, 0);
    }

以上就是MQ队列的执行过程。

详细的执行流程见博客[Linux Block Layer块设备层基于MultiQueue的部分源码分析][Linux Block Layer_MultiQueue]

#### 5、MQ队列调度器 ####

完成上述的MQ队列的执行过程，但是没有涉及到调度器这一块，经过请教，发现MQ也存在调度器模块，调度器主要完成同一个task中的调度，也就是一个或者多个软件队列上的请求的调度，调度的目的和传统CFQ类似，即考虑底层磁盘特性，更高效的利用存储介质的IO读写。

注意：软队列的数目和cpu核数对应，在应用执行过程中，上层cpu的调度中就为该应用绑定了对应cpu执行的个数，一个应用可以运行在多个核上，那么这个应用下发的IO将会执行在多个软件队列中，即在多个软件队列中完成调度。个人猜测，未经验证。

#### 6、疑问与解答 ####

在内核支持了MQ机制之后，倘若底层硬件某个通道任务繁忙，此时IO被阻塞的非常严重，上层应用如何感知到呢？此时软硬件队列的映射能够改变吗？  
——软硬件队列的映射在硬件初始化阶段时就已经确定了，而且是不能改的。当某个硬件通道IO密集时，此时会获取该硬件队列对应的软件队列，从而获取IO较为繁忙的CPU，CPU在得知各个核的状态时，在应用任务下发时，cpu会重新调度，避免将过重的IO绑定到任务繁忙的cpu核上。

#### 参考博客： ####

[https://blog.csdn.net/g382112762/article/details/79606485][Linux Block Layer_MultiQueue]  
[https://blog.csdn.net/notbaron/article/details/81147591][https_blog.csdn.net_notbaron_article_details_81147591]  
[https://blog.csdn.net/yedushu/article/details/82050933][https_blog.csdn.net_yedushu_article_details_82050933]  
[https://hyunyoung2.github.io/2016/09/14/Multi\_Queue/][https_hyunyoung2.github.io_2016_09_14_Multi_Queue]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTQxODk_size_16_color_FFFFFF_t_70]: /images/20220310/74cd7a7d6e564a0a95c1873ff8ac5d42.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTQxODk_size_16_color_FFFFFF_t_70 1]: /images/20220310/53fec7e6a2134f07b9a00176c095c279.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTQxODk_size_16_color_FFFFFF_t_70 2]: /images/20220310/377d310e118648e39f0e212668b1dcd1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTQxODk_size_16_color_FFFFFF_t_70 3]: /images/20220310/0546576281da4b46ad7dc090e8d5fcb5.png
[Linux Block Layer_MultiQueue]: https://blog.csdn.net/g382112762/article/details/79606485
[https_blog.csdn.net_notbaron_article_details_81147591]: https://blog.csdn.net/notbaron/article/details/81147591
[https_blog.csdn.net_yedushu_article_details_82050933]: https://blog.csdn.net/yedushu/article/details/82050933
[https_hyunyoung2.github.io_2016_09_14_Multi_Queue]: https://hyunyoung2.github.io/2016/09/14/Multi_Queue/