linux内核数据结构之kfifo

来源:一口Linux发布时间:2022-01-11274浏览

1、前言

最近项目中用到一个环形缓冲区（ring buffer），代码是由linux内核的kfifo改过来的。缓冲区在文件系统中经常用到，通过缓冲区缓解cpu读写内存和读写磁盘的速度。例如一个进程A产生数据发给另外一个进程B，进程B需要对进程A传的数据进行处理并写入文件，如果B没有处理完，则A要延迟发送。为了保证进程A减少等待时间，可以在A和B之间采用一个缓冲区，A每次将数据存放在缓冲区中，B每次冲缓冲区中取。这是典型的生产者和消费者模型，缓冲区中数据满足FIFO特性，因此可以采用队列进行实现。Linux内核的kfifo正好是一个环形队列，可以用来当作环形缓冲区。生产者与消费者使用缓冲区如下图所示：

环形缓冲区的详细介绍及实现方法可以参考http://en.wikipedia.org/wiki/Circular_buffer，介绍的非常详细，列举了实现环形队列的几种方法。环形队列的不便之处在于如何判断队列是空还是满。维基百科上给三种实现方法。

2、linux 内核kfifo

　　kfifo设计的非常巧妙，代码很精简，对于入队和出对处理的出人意料。首先看一下kfifo的数据结构：

struct kfifo {
unsignedchar*buffer;     * the buffer holding the data */
unsignedintsize;         * the size of the allocated buffer */
unsignedintin;           * data is added at offset (in % size) */
unsignedintout;          * data is extracted from off. (out % size) */
spinlock_t*lock;          * protects concurrent modifications */
};

kfifo提供的方法有：

1根据给定buffer创建一个kfifo
2struct kfifo *kfifo_init(unsignedchar*buffer, unsignedintsize,
3gfp_tgfp_mask, spinlock_t*lock);
4给定size分配buffer和kfifo
5struct kfifo *kfifo_alloc(unsignedintsize, gfp_tgfp_mask,
6spinlock_t*lock);
7释放kfifo空间
8void kfifo_free(struct kfifo *fifo)
9 //向kfifo中添加数据
10 unsigned int kfifo_put(struct kfifo *fifo,
11constunsignedchar*buffer, unsignedintlen)
12 //从kfifo中取数据
13 unsigned int kfifo_put(struct kfifo *fifo,
14constunsignedchar*buffer, unsignedintlen)
15 //获取kfifo中有数据的buffer大小
16 unsigned int kfifo_len(struct kfifo *fifo)

定义自旋锁的目的为了防止多进程/线程并发使用kfifo。因为in和out在每次get和out时，发生改变。初始化和创建kfifo的源代码如下：

1structkfifo*kfifo_init(unsignedchar*buffer,unsignedintsize,
2gfp_tgfp_mask,spinlock_t*lock)
3{
4structkfifo*fifo;
6*sizemustbeapowerof2*/
7BUG_ON(!is_power_of_2(size));
9fifokmalloc(sizeof(structkfifo),gfp_mask);
10if(!fifo)
11returnERR_PTR(-ENOMEM);
13fifo->bufferbuffer;
14fifo->sizesize;
15fifo->infifo->out0;
16fifo->locklock;
17
18returnfifo;
19}
20structkfifo*kfifo_alloc(unsignedintsize,gfp_tgfp_mask,spinlock_t*lock)
21{
22unsignedchar*buffer;
23structkfifo*ret;
29if(!is_power_of_2(size)){
30BUG_ON(size>0x80000000);
31sizeroundup_pow_of_two(size);
32}
34bufferkmalloc(size,gfp_mask);
35if(!buffer)
36returnERR_PTR(-ENOMEM);
38retkfifo_init(buffer,size,gfp_mask,lock);
39
40if(IS_ERR(ret))
41kfree(buffer);
43returnret;
44}

在kfifo_init和kfifo_calloc中，kfifo->size的值总是在调用者传进来的size参数的基础上向2的幂扩展，这是内核一贯的做法。这样的好处不言而喻--对kfifo->size取模运算可以转化为与运算，如：kfifo->in % kfifo->size 可以转化为 kfifo->in & (kfifo->size – 1)

kfifo的巧妙之处在于in和out定义为无符号类型，在put和get时，in和out都是增加，当达到最大值时，产生溢出，使得从0开始，进行循环使用。put和get代码如下所示：

1staticinlineunsignedintkfifo_put(structkfifo *fifo,
2constunsignedchar*buffer, unsignedintlen)
3{
4unsignedlongflags;
5unsignedintret;
6spin_lock_irqsave(fifo->lock, flags);
7ret = __kfifo_put(fifo, buffer, len);
8spin_unlock_irqrestore(fifo->lock, flags);
9returnret;
10}
11
12staticinlineunsignedintkfifo_get(structkfifo *fifo,
13unsignedchar*buffer, unsignedintlen)
14{
15unsignedlongflags;
16unsignedintret;
17spin_lock_irqsave(fifo->lock, flags);
18ret = __kfifo_get(fifo, buffer, len);
19当fifo->in == fifo->out时，buufer为空
20if(fifo->infifo->out)
21fifo->infifo->out0;
22spin_unlock_irqrestore(fifo->lock, flags);
23returnret;
24}
25
26
27unsignedint__kfifo_put(structkfifo *fifo,
28constunsignedchar*buffer, unsignedintlen)
29{
30unsignedintl;
31buffer中空的长度
32len = min(len, fifo->size - fifo->in+ fifo->out);
34*
35      * Ensure that we sample the fifo->out index -before- we
36      * start putting bytes into the kfifo.
37      */
39smp_mb();
41* first put the data starting from fifo->in to buffer end */
42l = min(len, fifo->size - (fifo->in& (fifo->size - 1)));
43memcpy(fifo->buffer + (fifo->in& (fifo->size - 1)), buffer, l);
45* then put the rest (if any) at the beginning of the buffer */
46memcpy(fifo->buffer, buffer + l, len - l);
47
48*
49      * Ensure that we add the bytes to the kfifo -before-
50      * we update the fifo->in index.
51      */
53smp_wmb();
55fifo->in+= len;  每次累加，到达最大值后溢出，自动转为0
57returnlen;
58}
59
60unsignedint__kfifo_get(structkfifo *fifo,
61unsignedchar*buffer, unsignedintlen)
62{
63unsignedintl;
64有数据的缓冲区的长度
65len = min(len, fifo->in- fifo->out);
67*
68      * Ensure that we sample the fifo->in index -before- we
69      * start removing bytes from the kfifo.
70      */
72smp_rmb();
74* first get the data from fifo->out until the end of the buffer */
75l = min(len, fifo->size - (fifo->out& (fifo->size - 1)));
76memcpy(buffer, fifo->buffer + (fifo->out& (fifo->size - 1)), l);
78* then get the rest (if any) from the beginning of the buffer */
79memcpy(buffer + l, fifo->buffer, len - l);
81*
82      * Ensure that we remove the bytes from the kfifo -before-
83      * we update the fifo->out index.
84      */
86smp_mb();
88fifo->out+= len; 每次累加，到达最大值后溢出，自动转为0
90returnlen;
91}

put和get在调用__put和__get过程都进行加锁，防止并发。从代码中可以看出put和get都调用两次memcpy，这针对的是边界条件。例如下图：蓝色表示空闲，红色表示占用。

（1）空的kfifo，

（2）put一个buffer后

（3）get一个buffer后

（4）当此时put的buffer长度超出in到末尾长度时，则将剩下的移到头部去

3、测试程序

仿照kfifo编写一个ring_buffer，现有线程互斥量进行并发控制。设计的ring_buffer如下所示：

1**@brief仿照linuxkfifo写的ringbuffer
2*@atuherAnkerdate:2013-12-18
3*ring_buffer.h
4**/
5
6#ifndef KFIFO_HEADER_H 
7#define KFIFO_HEADER_H
8
9#include <inttypes.h>
10#include <string.h>
11#include <stdlib.h>
12#include <stdio.h>
13#include <errno.h>
14#include <assert.h>
15
16判断x是否是2的次方
17#define is_power_of_2(x) ((x) != 0 && (((x) & ((x) - 1)) == 0))
18取a和b中最小值
19#define min(a, b) (((a) < (b)) ? (a) : (b))
20
21structring_buffer
22{
23void*buffer;缓冲区
24uint32_tsize;大小
25uint32_tin;入口位置
26uint32_tout;出口位置
27pthread_mutex_t*f_lock;互斥锁
28};
29初始化缓冲区
30structring_buffer*ring_buffer_init(void*buffer,uint32_tsize,pthread_mutex_t*f_lock)
31{
32assert(buffer);
33structring_buffer*ring_bufNULL;
34if(!is_power_of_2(size))
35{
36fprintf(stderr,"sizemustbepowerof2.\n");
37returnring_buf;
38}
39ring_buf(structring_buffer*)malloc(sizeof(structring_buffer));
40if(!ring_buf)
41{
42fprintf(stderr,"Failedtomallocmemory,errno:%u,reason:%s",
43errno,strerror(errno));
44returnring_buf;
45}
46memset(ring_buf,0,sizeof(structring_buffer));
47ring_buf->bufferbuffer;
48ring_buf->sizesize;
49ring_buf->in0;
50ring_buf->out0;
51ring_buf->f_lockf_lock;
52returnring_buf;
53}
54释放缓冲区
55voidring_buffer_free(structring_buffer*ring_buf)
56{
57if(ring_buf)
58{
59if(ring_buf->buffer)
60{
61free(ring_buf->buffer);
62ring_buf->bufferNULL;
63}
64free(ring_buf);
65ring_bufNULL;
66}
67}
68
69缓冲区的长度
70uint32_t__ring_buffer_len(conststructring_buffer*ring_buf)
71{
72return(ring_buf->in-ring_buf->out);
73}
74
75从缓冲区中取数据
76uint32_t__ring_buffer_get(structring_buffer*ring_buf,void*buffer,uint32_tsize)
77{
78assert(ring_buf||buffer);
79uint32_tlen0;
80sizemin(size,ring_buf->in-ring_buf->out);
81*firstgetthedatafromfifo->outuntiltheendofthebuffer*/
82lenmin(size,ring_buf->size-(ring_buf->out&(ring_buf->size-1)));
83memcpy(buffer,ring_buf->buffer+(ring_buf->out&(ring_buf->size-1)),len);
84*thengettherest(ifany)fromthebeginningofthebuffer*/
85memcpy(buffer+len,ring_buf->buffer,size-len);
86ring_buf->out+=size;
87returnsize;
88}
89向缓冲区中存放数据
90uint32_t__ring_buffer_put(structring_buffer*ring_buf,void*buffer,uint32_tsize)
91{
92assert(ring_buf||buffer);
93uint32_tlen0;
94sizemin(size,ring_buf->size-ring_buf->in+ring_buf->out);
95*firstputthedatastartingfromfifo->intobufferend*/
96lenmin(size,ring_buf->size-(ring_buf->in&(ring_buf->size-1)));
97memcpy(ring_buf->buffer+(ring_buf->in&(ring_buf->size-1)),buffer,len);
98*thenputtherest(ifany)atthebeginningofthebuffer*/
99memcpy(ring_buf->buffer,buffer+len,size-len);
100ring_buf->in+=size;
101returnsize;
102}
103
104uint32_tring_buffer_len(conststructring_buffer*ring_buf)
105{
106uint32_tlen0;
107pthread_mutex_lock(ring_buf->f_lock);
108len__ring_buffer_len(ring_buf);
109pthread_mutex_unlock(ring_buf->f_lock);
110returnlen;
111}
112
113uint32_tring_buffer_get(structring_buffer*ring_buf,void*buffer,uint32_tsize)
114{
115uint32_tret;
116pthread_mutex_lock(ring_buf->f_lock);
117ret__ring_buffer_get(ring_buf,buffer,size);
118buffer中没有数据
119if(ring_buf->inring_buf->out)
120ring_buf->inring_buf->out0;
121pthread_mutex_unlock(ring_buf->f_lock);
122returnret;
123}
124
125uint32_tring_buffer_put(structring_buffer*ring_buf,void*buffer,uint32_tsize)
126{
127uint32_tret;
128pthread_mutex_lock(ring_buf->f_lock);
129ret__ring_buffer_put(ring_buf,buffer,size);
130pthread_mutex_unlock(ring_buf->f_lock);
131returnret;
132}
133#endif

采用多线程模拟生产者和消费者编写测试程序，如下所示：

1**@briefringbuffer测试程序，创建两个线程，一个生产者，一个消费者。
2*生产者每隔1秒向buffer中投入数据，消费者每隔2秒去取数据。
3*@atuherAnkerdate:2013-12-18
4**/
5#include "ring_buffer.h"
6#include <pthread.h>
7#include <time.h>
8
9#define BUFFER_SIZE  1024 * 1024
10
11typedefstructstudent_info
12{
13uint64_tstu_id;
14uint32_tage;
15uint32_tscore;
16}student_info;
17
18
19voidprint_student_info(conststudent_info*stu_info)
20{
21assert(stu_info);
22printf("id:%lu\t",stu_info->stu_id);
23printf("age:%u\t",stu_info->age);
24printf("score:%u\n",stu_info->score);
25}
26
27student_info*get_student_info(time_ttimer)
28{
29student_info*stu_info(student_info*)malloc(sizeof(student_info));
30if(!stu_info)
31{
32fprintf(stderr,Failed to malloc memory.\n");
33returnNULL;
34}
35srand(timer);
36stu_info->stu_id10000+rand()%9999;
37stu_info->agerand()%30;
38stu_info->scorerand()%101;
39print_student_info(stu_info);
40returnstu_info;
41}
42
43void*consumer_proc(void*arg)
44{
45structring_buffer*ring_buf(structring_buffer*)arg;
46student_infostu_info;
47while(1)
48{
49sleep(2);
50printf("------------------------------------------\n");
51printf("getastudentinfofromringbuffer.\n");
52ring_buffer_get(ring_buf,(void*)&stu_info,sizeof(student_info));
53printf("ringbuffer length:%u\n",ring_buffer_len(ring_buf));
54print_student_info(&stu_info);
55printf("------------------------------------------\n");
56}
57return(void*)ring_buf;
58}
59
60void*producer_proc(void*arg)
61{
62time_tcur_time;
63structring_buffer*ring_buf(structring_buffer*)arg;
64while(1)
65{
66time(&cur_time);
67srand(cur_time);
68intseedrand()%11111;
69printf("******************************************\n");
70student_info*stu_infoget_student_info(cur_time+seed);
71printf("putastudentinfotoringbuffer.\n");
72ring_buffer_put(ring_buf,(void*)stu_info,sizeof(student_info));
73printf("ringbuffer length:%u\n",ring_buffer_len(ring_buf));
74printf("******************************************\n");
75sleep(1);
76}
77return(void*)ring_buf;
78}
79
80intconsumer_thread(void*arg)
81{
82interr;
83pthread_ttid;
84errpthread_create(&tid,NULL,consumer_proc,arg);
85if(err!=0)
86{
87fprintf(stderr,Failed to create consumer thread.errno:%u, reason:%s\n",
88errno,strerror(errno));
89return-1;
90}
91returntid;
92}
93intproducer_thread(void*arg)
94{
95interr;
96pthread_ttid;
97errpthread_create(&tid,NULL,producer_proc,arg);
98if(err!=0)
99{
100fprintf(stderr,Failed to create consumer thread.errno:%u, reason:%s\n",
101errno,strerror(errno));
102return-1;
103}
104returntid;
105}
106
107
108intmain()
109{
110void*bufferNULL;
111uint32_tsize0;
112structring_buffer*ring_bufNULL;
113pthread_tconsume_pid,produce_pid;
114
115pthread_mutex_t*f_lock(pthread_mutex_t*)malloc(sizeof(pthread_mutex_t));
116if(pthread_mutex_init(f_lock,NULL)!=0)
117{
118fprintf(stderr,Failed init mutex,errno:%u,reason:%s\n",
119errno,strerror(errno));
120return-1;
121}
122buffer(void*)malloc(BUFFER_SIZE);
123if(!buffer)
124{
125fprintf(stderr,Failed to malloc memory.\n");
126return-1;
127}
128sizeBUFFER_SIZE;
129ring_bufring_buffer_init(buffer,size,f_lock);
130if(!ring_buf)
131{
132fprintf(stderr,Failed to init ring buffer.\n");
133return-1;
134}
135#if 0
136student_info*stu_infoget_student_info(638946124);
137ring_buffer_put(ring_buf,(void*)stu_info,sizeof(student_info));
138stu_infoget_student_info(976686464);
139ring_buffer_put(ring_buf,(void*)stu_info,sizeof(student_info));
140ring_buffer_get(ring_buf,(void*)stu_info,sizeof(student_info));
141print_student_info(stu_info);
142#endif
143printf("multithreadtest.......\n");
144produce_pidproducer_thread((void*)ring_buf);
145consume_pidconsumer_thread((void*)ring_buf);
146pthread_join(produce_pid,NULL);
147pthread_join(consume_pid,NULL);
148ring_buffer_free(ring_buf);
149free(f_lock);
150return0;
151}