Linux Kernel kfifo 实现和巧妙设计

kfifo实现与解释

kfifo的基本概念

kfifo 是 Linux 内核中提供的一个先进先出(FIFO)环形缓冲区实现，它是一个高效、线程安全(在正确使用情况下)的循环队列数据结构。
kfifo 具有以下特点：

• 环形缓冲区：当数据到达缓冲区末尾时会绕回到开头
• 无锁设计：在单生产者单消费者场景下无需加锁
• 高效操作：提供快速的入队和出队操作

kfifo 的实现

数据结构：

struct kfifo {
    unsigned char *buffer;    // 缓冲区指针
    unsigned int size;       // 缓冲区大小（必须是2的幂次）
    unsigned int in;         // 入队指针（写入位置）
    unsigned int out;        // 出队指针（读取位置）
};

初始化：

int kfifo_init(struct kfifo *fifo, void *buffer, unsigned int size)
{
    // 检查size是否为2的幂次
    if (!is_power_of_2(size))
        return -EINVAL;
    
    fifo->buffer = buffer;
    fifo->size = size;
    fifo->in = fifo->out = 0;
    return 0;
}

入队操作：

unsigned int kfifo_in(struct kfifo *fifo, const void *from, unsigned int len)
{
    unsigned int l;
    
    // 计算可用空间
    len = min(len, fifo->size - fifo->in + fifo->out);
    
    // 首先复制从in到缓冲区末尾的数据
    l = min(len, fifo->size - (fifo->in & (fifo->size - 1)));
    memcpy(fifo->buffer + (fifo->in & (fifo->size - 1)), from, l);
    
    // 然后复制剩余的数据到缓冲区开头
    memcpy(fifo->buffer, (const unsigned char *)from + l, len - l);
    
    // 更新in指针（使用内存屏障保证顺序）
    smp_wmb();
    fifo->in += len;
    
    return len;
}

出队操作：

unsigned int kfifo_out(struct kfifo *fifo, void *to, unsigned int len)
{
    unsigned int l;
    
    // 计算可用数据量
    len = min(len, fifo->in - fifo->out);
    
    // 首先复制从out到缓冲区末尾的数据
    l = min(len, fifo->size - (fifo->out & (fifo->size - 1)));
    memcpy(to, fifo->buffer + (fifo->out & (fifo->size - 1)), l);
    
    // 然后复制剩余的数据从缓冲区开头
    memcpy((unsigned char *)to + l, fifo->buffer, len - l);
    
    // 更新out指针（使用内存屏障保证顺序）
    smp_rmb();
    fifo->out += len;
    
    return len;
}

关键点解释

• 大小必须为2的幂次：

  • 这样可以使用 in & (size - 1) 代替 in % size 来计算索引，提高效率
  • 位操作比取模运算快得多

• 无锁设计：

  • 单生产者单消费者场景下，in和out指针分别只由一个线程修改
  • 使用内存屏障(smp_wmb/smp_rmb)保证内存访问顺序

• 指针管理：

  • in和out指针始终递增，不进行取模运算
  • 实际索引通过 in & (size - 1) 计算
  • 这样可以避免处理指针回绕的复杂情况

• 可用空间计算：

  • 可用空间 = size - (in - out)
  • 可用数据 = in - out

使用示例

#define FIFO_SIZE 1024

struct kfifo my_fifo;
unsigned char buffer[FIFO_SIZE];

// 初始化
kfifo_init(&my_fifo, buffer, FIFO_SIZE);

// 生产者线程
unsigned char data[100];
kfifo_in(&my_fifo, data, sizeof(data));

// 消费者线程
unsigned char received[100];
kfifo_out(&my_fifo, received, sizeof(received));

用户空间的使用案例

#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>
#include <assert.h>

// 判断是否为2的幂次
#define is_power_of_2(x) ((x) != 0 && (((x) & ((x) - 1)) == 0))

// 环形缓冲区结构体
struct kfifo {
    unsigned char *buffer;    // 缓冲区指针
    unsigned int size;       // 缓冲区大小（必须是2的幂次）
    unsigned int in;         // 入队指针（写入位置）
    unsigned int out;        // 出队指针（读取位置）
};

// 初始化kfifo
int kfifo_init(struct kfifo *fifo, void *buffer, unsigned int size)
{
    // 检查size是否为2的幂次
    if (!is_power_of_2(size))
        return -1;
    
    fifo->buffer = buffer;
    fifo->size = size;
    fifo->in = fifo->out = 0;
    return 0;
}

// 创建一个kfifo
struct kfifo *kfifo_alloc(unsigned int size)
{
    unsigned char *buffer;
    struct kfifo *fifo;
    
    // 检查size是否为2的幂次，如果不是则向上取整为2的幂次
    if (!is_power_of_2(size)) {
        size = roundup_pow_of_two(size);
    }

    fifo = malloc(sizeof(struct kfifo));
    if (!fifo)
        return NULL;

    buffer = malloc(size);
    if (!buffer) {
        free(fifo);
        return NULL;
    }

    kfifo_init(fifo, buffer, size);
    return fifo;
}

// 释放kfifo
void kfifo_free(struct kfifo *fifo)
{
    free(fifo->buffer);
    free(fifo);
}

// 获取kfifo中已用空间大小
unsigned int kfifo_len(struct kfifo *fifo)
{
    return fifo->in - fifo->out;
}

// 获取kfifo中剩余空间大小
unsigned int kfifo_avail(struct kfifo *fifo)
{
    return fifo->size - kfifo_len(fifo);
}

// 入队操作
unsigned int kfifo_in(struct kfifo *fifo, const void *from, unsigned int len)
{
    unsigned int l;
    
    // 计算可用空间
    len = min(len, fifo->size - fifo->in + fifo->out);
    
    // 首先复制从in到缓冲区末尾的数据
    l = min(len, fifo->size - (fifo->in & (fifo->size - 1)));
    memcpy(fifo->buffer + (fifo->in & (fifo->size - 1)), from, l);
    
    // 然后复制剩余的数据到缓冲区开头
    memcpy(fifo->buffer, (const unsigned char *)from + l, len - l);
    
    // 更新in指针
    fifo->in += len;
    
    return len;
}

// 出队操作
unsigned int kfifo_out(struct kfifo *fifo, void *to, unsigned int len)
{
    unsigned int l;
    
    // 计算可用数据量
    len = min(len, fifo->in - fifo->out);
    
    // 首先复制从out到缓冲区末尾的数据
    l = min(len, fifo->size - (fifo->out & (fifo->size - 1)));
    memcpy(to, fifo->buffer + (fifo->out & (fifo->size - 1)), l);
    
    // 然后复制剩余的数据从缓冲区开头
    memcpy((unsigned char *)to + l, fifo->buffer, len - l);
    
    // 更新out指针
    fifo->out += len;
    
    return len;
}

// 查看数据但不移动out指针
unsigned int kfifo_peek(struct kfifo *fifo, void *to, unsigned int len)
{
    unsigned int l;
    unsigned int out = fifo->out;
    
    len = min(len, fifo->in - out);
    
    l = min(len, fifo->size - (out & (fifo->size - 1)));
    memcpy(to, fifo->buffer + (out & (fifo->size - 1)), l);
    
    memcpy((unsigned char *)to + l, fifo->buffer, len - l);
    
    return len;
}

// 重置kfifo
void kfifo_reset(struct kfifo *fifo)
{
    fifo->in = fifo->out = 0;
}

// 测试用例
int main()
{
    struct kfifo *fifo;
    char buf[32];
    char out[32];
    int ret;
    
    // 创建大小为16的kfifo
    fifo = kfifo_alloc(16);
    if (!fifo) {
        printf("Failed to allocate fifo\n");
        return -1;
    }
    
    // 写入数据
    strcpy(buf, "Hello, kfifo!");
    ret = kfifo_in(fifo, buf, strlen(buf)+1);
    printf("Written %d bytes: %s\n", ret, buf);
    
    // 查看数据但不取出
    memset(out, 0, sizeof(out));
    ret = kfifo_peek(fifo, out, 5);
    printf("Peeked %d bytes: %s (len=%u)\n", ret, out, kfifo_len(fifo));
    
    // 读取数据
    memset(out, 0, sizeof(out));
    ret = kfifo_out(fifo, out, sizeof(out));
    printf("Read %d bytes: %s (len=%u)\n", ret, out, kfifo_len(fifo));
    
    // 测试环形特性
    strcpy(buf, "1234567890ABCDEF");
    ret = kfifo_in(fifo, buf, strlen(buf)+1);
    printf("Written %d bytes: %s (len=%u)\n", ret, buf, kfifo_len(fifo));
    
    memset(out, 0, sizeof(out));
    ret = kfifo_out(fifo, out, 8);
    printf("Read %d bytes: %s (len=%u)\n", ret, out, kfifo_len(fifo));
    
    ret = kfifo_out(fifo, out+8, sizeof(out)-8);
    printf("Read %d bytes: %s (len=%u)\n", ret, out, kfifo_len(fifo));
    
    kfifo_free(fifo);
    return 0;
}

kfifo 读写设计的巧妙之处详解

使用递增指针与位运算代替取模运算

巧妙之处：

• 传统环形缓冲区使用 index % size 计算位置，但取模运算在计算机中是比较耗时的操作
• kfifo 使用 index & (size - 1) 代替取模运算，这要求 size 必须是 2 的幂次
  例子：
  假设缓冲区大小 size = 8 (2^3)，则 size - 1 = 7 (二进制 0111)

// 传统方法
unsigned int index = 10;
unsigned int pos = index % 8;  // 需要除法运算，耗时

// kfifo方法
unsigned int pos = index & 7;  // 位与运算，非常快速
// 10 & 7 = 
// 1010 (10)
// & 0111 (7)
// = 0010 (2)

优点：

• 位运算比取模运算快 5-10 倍
• 当缓冲区操作非常频繁时，这种优化能显著提升性能

指针永不回绕的设计

巧妙之处：

• in 和 out 指针只递增，永不减少
• 实际位置通过 index & (size - 1) 计算
• 避免了处理指针回绕的复杂逻辑
  例子：
  缓冲区大小 size = 8 (0-7)

写入顺序:
1. in=0 → pos=0&7=0
2. in=1 → pos=1&7=1
...
7. in=7 → pos=7&7=7
8. in=8 → pos=8&7=0 (自动回绕)
9. in=9 → pos=9&7=1
...

优点：

• 不需要特殊处理指针到达末尾的情况
• 代码更简洁，减少了条件判断
• 32位系统中，当 in 或 out 超过 2^32 时会自然回绕，不影响正确性

分离式写入/读取设计

巧妙之处：

• 当一次操作需要跨越缓冲区末尾时，自动分成两部分处理
• 第一部分：从当前位置到缓冲区末尾
• 第二部分：从缓冲区开头到剩余部分
  写入例子：
  缓冲区大小=8，当前 in=6，写入5字节数据

// kfifo_in 内部处理:
len = 5
l = min(5, 8 - (6 & 7)) = min(5, 2) = 2
// 第一部分: 写入2字节到位置6和7
memcpy(buffer + 6, data, 2);
// 第二部分: 写入剩余3字节到位置0-2
memcpy(buffer, data + 2, 3);
// 最后 in = 6 + 5 = 11

读取例子：
缓冲区大小=8，当前 out=6，读取5字节数据

// kfifo_out 内部处理:
len = 5
l = min(5, 8 - (6 & 7)) = min(5, 2) = 2
// 第一部分: 从位置6和7读取2字节
memcpy(dest, buffer + 6, 2);
// 第二部分: 从位置0-2读取剩余3字节
memcpy(dest + 2, buffer, 3);
// 最后 out = 6 + 5 = 11

优点：

• 处理跨越末尾的情况无需额外条件判断
• 代码路径一致，无论是否跨越末尾都执行两次memcpy
• 现代CPU中，小数据量的两次memcpy可能比条件判断+单次memcpy更快

无锁设计的实现原理

巧妙之处：

• 单生产者单消费者场景下不需要锁
• 通过内存屏障保证执行顺序
• in 和 out 指针分别只由一个线程修改
  例子：
  生产者线程：

// 写入数据
kfifo_in(fifo, data, len);
// 内部执行顺序：
1. 计算可写空间
2. 执行memcpy复制数据
3. smp_wmb() 写入内存屏障
4. 更新in指针

消费者线程：

// 读取数据
kfifo_out(fifo, dest, len);
// 内部执行顺序：
1. smp_rmb() 读取内存屏障
2. 计算可读数据量
3. 执行memcpy复制数据
4. 更新out指针

内存屏障的作用：

• smp_wmb() 保证数据先写入内存，再更新in指针
• smp_rmb() 保证先读取in指针，再读取数据
• 这样即使CPU乱序执行，也能保证逻辑正确

空间和数据量计算的巧妙设计

巧妙之处：

• 可用空间 = size - (in - out)
• 可用数据 = in - out
• 即使指针超过size也能正确计算
  例子：
  缓冲区大小=8
  初始状态: in=0, out=0
• 写入6字节: in=6, out=0
  • 可用空间 = 8 - (6-0) = 2
  • 可用数据 = 6 - 0 = 6
• 读取4字节: in=6, out=4
  • 可用空间 = 8 - (6-4) = 6
  • 可用数据 = 6 - 4 = 2
• 再写入7字节: 只能写入6字节，in=12, out=4
  • 可用空间 = 8 - (12-4) = 0
  • 可用数据 = 12 - 4 = 8
• 读取8字节: in=12, out=12
  • 可用空间 = 8 - (12-12) = 8
  • 可用数据 = 12 - 12 = 0
    优点：
• 计算简单高效，只有减法和比较操作
• 即使in或out超过size也能正确工作
• 32位系统中，当in-out超过2^32时会自然回绕，不影响正确性

综合示例

假设有一个大小为8字节的kfifo：

struct kfifo fifo;
unsigned char buf[8];
kfifo_init(&fifo, buf, 8);

// 生产者写入5字节 "Hello"
kfifo_in(&fifo, "Hello", 5);
// in=5, out=0
// 缓冲区内容: |H|e|l|l|o| | | |

// 消费者读取3字节
char data[3];
kfifo_out(&fifo, data, 3);
// in=5, out=3
// 读取到"He|", 缓冲区剩余: | | |l|o| | | |

// 生产者再写入6字节 " World"
// 但只有5字节空间(8-(5-3)=6, 但实际需要6)
kfifo_in(&fifo, " World", 6);
// 实际写入5字节 " Worl", in=10, out=3
// 缓冲区内容: |l|o| |W|o|r|l| |

// 消费者读取剩余所有数据
char data2[10];
int len = kfifo_out(&fifo, data2, 10);
// 可读数据=10-3=7字节，但实际缓冲区只有7字节数据
// 读取过程:
// 1. 从out=3 (pos=3)到末尾: 读取" Worl" (5字节，但只有4位置)
//   实际读取位置3-7: | |W|o|r|l| → 4字节
// 2. 从开始读取剩余3字节: |l|o| | → 但只需要再读3字节
//   实际读取位置0-2: |l|o| | → 3字节
// 最终读取到" Worllo ", len=7
// out=10, in=10

这个例子展示了kfifo如何处理：

• 指针递增和自动回绕
• 空间不足时的部分写入
• 数据跨越缓冲区末尾的分段读取
• 指针计算与实际位置分离