Good News, Everyone!

Linux kernel 5.0 уже здесь и появляется в экспериментальных дистрибутивах, таких как Arch, openSUSE Tumbleweed, Fedora.



А если посмотреть на RC дистрибутивов Ubuntu Disko Dingo и Red Hat 8, то станет понятно: скоро kernel 5.0 с десктопов фанатов перекачует и на серьёзные сервера.
Кто-то скажет — ну и что. Очередной релиз, ничего особенного. Вот и сам Linus Torvalds сказал:

I’d like to point out (yet again) that we don’t do feature-based releases, and that “5.0” doesn’t mean anything more than that the 4.x numbers started getting big enough that I ran out of fingers and toes.

(Еще раз повторюсь — наши релизы не привязываются к каким-то определенным фичам, так что номер новой версии 5.0 означает только то, что для нумерования версий 4.х у меня уже не хватает пальцев на руках и ногах)

Однако модуль для floppy дисков (кто не знает — это такие диски размером c нагрудный карман рубашки, ёмкостью в 1,44 MB) — поправили…
И вот почему:

Всё дело в multi-queue block layer (blk-mq). Вводных статей про него в интернете предостаточно, так что давайте сразу к сути. Процесс перехода на blk-mq был начат давно и неспешно продвигался. Появился multi-queue scsi (параметр ядра scsi_mod.use_blk_mq), появились новые планировщики mq-deadline, bfq и прочее…

[root@fedora-29 sblkdev]# cat /sys/block/sda/queue/scheduler
[mq-deadline] none

Кстати, а какой у вас?

Сокращалось число драйверов блочных устройств, которые работают по старинке. А в 5.0 убрали функцию blk_init_queue() за ненадобностью. И теперь старый славный код lwn.net/Articles/58720 от 2003 года уже не только не собирается, но и потерял актуальность. Более того, новые дистрибутивы, которые готовятся к выпуску в этом году, в дефолтной конфигурации используют multi-queue block layer. Например, на 18-том Manjaro, ядро хоть и версии 4.19, но blk-mq по дефолту.

Поэтому можно считать, что в ядре 5.0 переход на blk-mq завершился. А для меня это важное событие, которое потребует переписывания кода и дополнительного тестирования. Что само по себе обещает появление багов больших и маленьких, а также несколько упавших серверов (Надо, Федя, надо! (с)).

Кстати, если кто-то думает, что для rhel8 этот переломный момент не настал, так как ядро там «зафризили» версией 4.18, то вы ошибаетесь. В свеженьком RC на rhel8 новинки из 5.0 уже мигрировали, и функцию blk_init_queue() тоже выпилили (наверное, при перетаскивании очередного чекина с github.com/torvalds/linux в свои исходники).
Вообще, «freeze» версии ядра для дистрибьютеров Linux, таких как SUSE и Red Hat, давно стало маркетинговым понятием. Система сообщает, что версия, к примеру, 4.4, а по факту функционал из свеженькой 4.8 vanilla. При этом на официальном сайте красуется надпись вроде: «В новом дистрибутиве мы сохранили для вас стабильное 4.4 ядро».

Но мы отвлеклись…

Так вот. Нам нужен новый simple block device driver, чтобы было понятнее, как это работает.
Итак, исходник на github.com/CodeImp/sblkdev. Предлагаю обсуждать, делать pull request-ы, заводить issue — буду чинить. QA пока не проверял.

Далее в статье я попробую описать что зачем. Поэтому дальше много кода.
Сразу прошу прощения, что в полной степени не соблюдается Linux kernel coding style, и да — я не люблю goto.

Итак, начнём с точек входа.

static int __init sblkdev_init(void)
{
    int ret = SUCCESS;

    _sblkdev_major = register_blkdev(_sblkdev_major, _sblkdev_name);
    if (_sblkdev_major <= 0){
        printk(KERN_WARNING "sblkdev: unable to get major number\n");
        return -EBUSY;
    }

    ret = sblkdev_add_device();
    if (ret)
        unregister_blkdev(_sblkdev_major, _sblkdev_name);
        
    return ret;
}

static void __exit sblkdev_exit(void)
{
    sblkdev_remove_device();

    if (_sblkdev_major > 0)
        unregister_blkdev(_sblkdev_major, _sblkdev_name);
}

module_init(sblkdev_init);
module_exit(sblkdev_exit);

Очевидно, при загрузке модуля запускается функция sblkdev_init(), при выгрузке sblkdev_exit().
Функция register_blkdev() регистрирует блочное устройство. Ему выделяется major номер. unregister_blkdev() — освобождает этот номер.

Ключевой структурой нашего модуля является sblkdev_device_t.

// The internal representation of our device
typedef struct sblkdev_device_s
{
    sector_t capacity;			    // Device size in bytes
    u8* data;			    		// The data aray. u8 - 8 bytes
    atomic_t open_counter;			// How many openers

    struct blk_mq_tag_set tag_set;
    struct request_queue *queue;	// For mutual exclusion

    struct gendisk *disk;			// The gendisk structure
} sblkdev_device_t;

Она содержит всю необходимую модулю ядра информацию об устройстве, в частности: ёмкость блочного устройства, сами данные (это же simple), указатели на диск и очередь.

Вся инициализация блочного устройства выполняется в функции sblkdev_add_device().

static int sblkdev_add_device(void)
{
    int ret = SUCCESS;

    sblkdev_device_t* dev = kzalloc(sizeof(sblkdev_device_t), GFP_KERNEL);
    if (dev == NULL) {
        printk(KERN_WARNING "sblkdev: unable to allocate %ld bytes\n", sizeof(sblkdev_device_t));
        return -ENOMEM;
    }
    _sblkdev_device = dev;

    do{
        ret = sblkdev_allocate_buffer(dev);
        if(ret)
            break;

#if 0 //simply variant with helper function blk_mq_init_sq_queue. It`s available from kernel 4.20 (vanilla).
        {//configure tag_set
            struct request_queue *queue;

            dev->tag_set.cmd_size = sizeof(sblkdev_cmd_t);
            dev->tag_set.driver_data = dev;

            queue = blk_mq_init_sq_queue(&dev->tag_set, &_mq_ops, 128, BLK_MQ_F_SHOULD_MERGE | BLK_MQ_F_SG_MERGE);
            if (IS_ERR(queue)) {
                ret = PTR_ERR(queue);
                printk(KERN_WARNING "sblkdev: unable to allocate and initialize tag set\n");
                break;
            }
            dev->queue = queue;
        }
#else   // more flexible variant
        {//configure tag_set
            dev->tag_set.ops = &_mq_ops;
            dev->tag_set.nr_hw_queues = 1;
            dev->tag_set.queue_depth = 128;
            dev->tag_set.numa_node = NUMA_NO_NODE;
            dev->tag_set.cmd_size = sizeof(sblkdev_cmd_t);
            dev->tag_set.flags = BLK_MQ_F_SHOULD_MERGE | BLK_MQ_F_SG_MERGE;
            dev->tag_set.driver_data = dev;

            ret = blk_mq_alloc_tag_set(&dev->tag_set);
            if (ret) {
                printk(KERN_WARNING "sblkdev: unable to allocate tag set\n");
                break;
            }
        }

        {//configure queue
            struct request_queue *queue = blk_mq_init_queue(&dev->tag_set);
            if (IS_ERR(queue)) {
                ret = PTR_ERR(queue);
                printk(KERN_WARNING "sblkdev: Failed to allocate queue\n");
                break;
            }
            dev->queue = queue;
        }
#endif
        dev->queue->queuedata = dev;

        {// configure disk
            struct gendisk *disk = alloc_disk(1); //only one partition 
            if (disk == NULL) {
                printk(KERN_WARNING "sblkdev: Failed to allocate disk\n");
                ret = -ENOMEM;
                break;
            }

            disk->flags |= GENHD_FL_NO_PART_SCAN; //only one partition 
            //disk->flags |= GENHD_FL_EXT_DEVT;
            disk->flags |= GENHD_FL_REMOVABLE;

            disk->major = _sblkdev_major;
            disk->first_minor = 0;
            disk->fops = &_fops;
            disk->private_data = dev;
            disk->queue = dev->queue;
            sprintf(disk->disk_name, "sblkdev%d", 0);
            set_capacity(disk, dev->capacity);

            dev->disk = disk;
            add_disk(disk);
        }

        printk(KERN_WARNING "sblkdev: simple block device was created\n");
    }while(false);

    if (ret){
        sblkdev_remove_device();
        printk(KERN_WARNING "sblkdev: Failed add block device\n");
    }

    return ret;
}

Под структуру выделяем память, аллоцируем буфер для хранения данных. Тут ничего особенного.
Далее инициализируем очередь обработки запросов или одной функцией blk_mq_init_sq_queue(), или сразу двумя: blk_mq_alloc_tag_set() + blk_mq_init_queue().

Кстати, если заглянуть в исходники функции blk_mq_init_sq_queue(), то увидим, что это всего лишь обёртка над функциями blk_mq_alloc_tag_set() и blk_mq_init_queue(), которая появилась в ядре 4.20. Кроме того, она скрывет он нас многие параметры очереди, однако выглядит значительно проще. Вам выбирать, какой вариант лучше, но я предпочитаю более явный.

Ключевым в данном коде является глобальная переменная _mq_ops.

static struct blk_mq_ops _mq_ops = {
    .queue_rq = queue_rq,
};

Именно здесь расположилась функция, которая обеспечивает обработку запросов, но подробнее о ней чуть позже. Главное, что точку входа в обработчик запросов мы обозначили.

Теперь, когда мы создали очередь — можно создавать экземпляр диска.

Здесь без особых изменений. Диск аллоцируется, задаются параметры, и диск добавляется в систему. Хочу пояснить насчет параметра disk->flags. Он позволяет указать системе, что диск removable, или, например, что он не содержит партиций и искать их там не надо.

Для управления диском есть структура _fops.

static const struct block_device_operations _fops = {
    .owner = THIS_MODULE,
    .open = _open,
    .release = _release,
    .ioctl = _ioctl,
#ifdef CONFIG_COMPAT
    .compat_ioctl = _compat_ioctl,
#endif
};

Точки входа _open и _release нам для simple block device модуля пока не сильно интересны. Кроме атомарного инкремента и декремента счётчика, там ничего нет. compat_ioctl я тоже оставил без реализации, так как вариант систем с 64-х битным ядром и 32-х битным user-space окружением мне не кажется перспективным.

А вот _ioctl позволяет обработать системные запросы к данному диску. При появлении диска система пытается побольше узнать о нём. По своему разумению вы можете отвечать на некоторые запросы (к примеру, чтобы прикинуться новым CD), но общее правило таково: если вы не хотите отвечать на неинтересующие вас запросы, просто верните код ошибки -ENOTTY. Кстати, если нужно, то здесь можно добавить и свои обработчики запросов, касающиеся именно этого диска.

Итак, устройство мы добавили — нужно позаботиться об освобождении ресурсов. Здесь вам не тутRust.

static void sblkdev_remove_device(void)
{
    sblkdev_device_t* dev = _sblkdev_device;
    if (dev){
        if (dev->disk)
            del_gendisk(dev->disk);

        if (dev->queue) {
            blk_cleanup_queue(dev->queue);
            dev->queue = NULL;
        }

        if (dev->tag_set.tags)
            blk_mq_free_tag_set(&dev->tag_set);

        if (dev->disk) {
            put_disk(dev->disk);
            dev->disk = NULL;
        }

        sblkdev_free_buffer(dev);

        kfree(dev);
        _sblkdev_device = NULL;

        printk(KERN_WARNING "sblkdev: simple block device was removed\n");
    }
}

В принципе, всё очевидно: удаляем объект диска из системы и освобождаем очередь, после чего освобождаем и свои буферы (области данных).

А теперь самое главное — обработка запросов в функции queue_rq().

static blk_status_t queue_rq(struct blk_mq_hw_ctx *hctx, const struct blk_mq_queue_data* bd)
{
    blk_status_t status = BLK_STS_OK;
    struct request *rq = bd->rq;

    blk_mq_start_request(rq);

    //we cannot use any locks that make the thread sleep
    {
        unsigned int nr_bytes = 0;

        if (do_simple_request(rq, &nr_bytes) != SUCCESS)
            status = BLK_STS_IOERR;

        printk(KERN_WARNING "sblkdev: request process %d bytes\n", nr_bytes);

#if 0 //simply and can be called from proprietary module 
        blk_mq_end_request(rq, status);
#else //can set real processed bytes count 
        if (blk_update_request(rq, status, nr_bytes)) //GPL-only symbol
            BUG();
        __blk_mq_end_request(rq, status);
#endif
    }

    return BLK_STS_OK;//always return ok
}

Для начала рассмотрим параметры. Первый — struct blk_mq_hw_ctx *hctx — состояние аппаратной очереди. В нашем случае мы обходимся без аппаратной очереди, так что unused.

Второй параметр — const struct blk_mq_queue_data* bd — параметр с очень лаконичной структурой, которую я не побоюсь представить вашему вниманию целиком:

struct blk_mq_queue_data {
	struct request *rq;
	bool last;
};

Получается, что по сути это всё тот-же request, пришедший к нам из врёмен, о которых уже не помнит летописец elixir.bootlin.com. Так что берём запрос и начинаем его обрабатывать, о чём уведомляем ядро вызовом blk_mq_start_request(). По завершению обработки запроса сообщим об этом ядру вызовом функции blk_mq_end_request().

Тут маленькое замечание: функция blk_mq_end_request() — это, по сути, обёртка над вызовами blk_update_request() + __blk_mq_end_request(). При использовании функции blk_mq_end_request() нельзя задать, сколько конкретно байт было действительно обработано. Считает, что обработано всё.

У альтернативного варианта есть другая особенность: функция blk_update_request экспортируется только для GPL-only модулей. То есть, если вы захотите создать проприетарный модуль ядра (да избавит вас PM от этого тернистого пути), вы не сможете использовать blk_update_request(). Так что здесь выбор за вами.

Непосредственно саму перекладку байтиков из запроса в буфер и обратно я вынес в функцию do_simple_request().

static int do_simple_request(struct request *rq, unsigned int *nr_bytes)
{
    int ret = SUCCESS;
    struct bio_vec bvec;
    struct req_iterator iter;
    sblkdev_device_t *dev = rq->q->queuedata;
    loff_t pos = blk_rq_pos(rq) << SECTOR_SHIFT;
    loff_t dev_size = (loff_t)(dev->capacity << SECTOR_SHIFT);

    printk(KERN_WARNING "sblkdev: request start from sector %ld \n", blk_rq_pos(rq));
    
    rq_for_each_segment(bvec, rq, iter)
    {
        unsigned long b_len = bvec.bv_len;

        void* b_buf = page_address(bvec.bv_page) + bvec.bv_offset;

        if ((pos + b_len) > dev_size)
            b_len = (unsigned long)(dev_size - pos);

        if (rq_data_dir(rq))//WRITE
            memcpy(dev->data + pos, b_buf, b_len);
        else//READ
            memcpy(b_buf, dev->data + pos, b_len);

        pos += b_len;
        *nr_bytes += b_len;
    }

    return ret;
}

Тут ничего нового: rq_for_each_segment перебирает все bio, а в них все bio_vec структуры, позволяя нам добраться до страниц с данными запроса.

Как впечатления? Кажется, всё просто? Обработка запроса вообще представляет из себя просто копирование данных между страницами запроса и внутренним буфером. Вполне достойно для simple block device driver, да?

Но есть проблема: Это не для реального использования!

Суть проблемы в том, что функция обработки запроса queue_rq() вызывается в цикле, обрабатывающем запросы из списка. Уж не знаю, какая именно блокировка для этого списка там используется, Spin или RCU (врать не хочу — кто знает, поправьте меня), но при попытке воспользоваться, к примеру, mutex-ом в функции обработки запроса отладочное ядро ругается и предупреждает: дремать тут нельзя. То есть пользоваться обычными средствами синхронизации или виртуальной памятью (virtually contiguous memory) — той, что аллоцируется с помощью vmalloc и может выпасть в swap со всем вытекающими — нельзя, так как процесс не может перейти в состояние ожидания.

Поэтому либо только Spin или RCU блокировки и буфер в виде массива страниц, или списка, или дерева, как это реализовано в ..\linux\drivers\block\brd.c, либо отложенная обработка в другом потоке, как это реализовано в ..\linux\drivers\block\loop.c.

Я думаю, не надо описывать, как собрать модуль, как его загрузить в систему и как выгрузить. На этом фронте без новинок, и на том спасибо :) Так что если кто-то хочет опробовать, уверен разберётся. Только не делайте это сразу на любимом ноутбуке! Поднимите виртуалочку или хотя бы сделайте бэкап на шару.

Кстати, Veeam Backup for Linux 3.0.1.1046 уже доступен. Только не пытайтесь запускать VAL 3.0.1.1046 на ядре 5.0 или старше. veeamsnap не соберётся. А некоторые multi-queue новшества ещё пока находятся на этапе тестирования.