Linux内核跟踪之ring buffer的实现

时间：2009-06-10 来源：xgr180

------------------------------------------ 本文系本站原创,欢迎转载! 转载请注明出处:http://ericxiao.cublog.cn/ ------------------------------------------ 一: 前言 Ring buffer是整个trace系统使用缓存管理的一种方式, 由于trace可能在内核运行的任何时候发生, 这种kernel的不确定状态决定了ring buffer的写操作中不能有任何引起睡眠的操作, 而且ring buffer的操作频率极高,所以在ring buffer实现里有很多高效的方式来处理多处理器, 读写同步的机制. 理解ring buffer是我们理解整个kernel trace的基础. 本文分析的源代码版本为linux kernel 2.6.30, 分析的代码基本位于kernel/trace/ring_buffer.c中.另外,为了描述的方便,下文ring buffer用RB来代替. 二: ring buffer的基本数据结构在深入到代码之前,我们先来看一下RB所用到的几个基本的数据结构,这样我们对RB就会有一个全局性的了解.整个RB的数据结构框架如下所示: 500)this.width=500;" border=0> ring buffer用struct ring_buffer来表示,数据结构定义如下: struct ring_buffer { /*RB中的页面数*/ unsigned pages; /*RB的标志,目前只有RB_FL_OVERWRITE可用*/ unsigned flags; /*ring buffer中包含的cpu个数*/ int cpus; /*整个ring buffer的禁用标志,用原子操作了防止竞争*/ atomic_t record_disabled; /* cpu位图*/ cpumask_var_t cpumask; /*RB访问锁*/ struct mutex mutex; /*CPU的缓存区页面,每个CPU对应一项*/ struct ring_buffer_per_cpu **buffers; #ifdef CONFIG_HOTPLUG_CPU /*多CPU情况下的cpu hotplug 通知链表*/ struct notifier_block cpu_notify; #endif /*RB所用的时间,用来计数时间戳*/ u64 (*clock)(void); } 在RB的操作中,我们可以禁止全局的RB操作,例如,完全禁用掉Trace功能后,整个RB都是不允许再操做的,这时,就可以将原子变量record_disabled 加1.相反的,如果启用的话,将其减1即可.只有当record_disabled的值等于0时,才允许操作RB. 同时,有些时候,要对RB的一些数据进行更新,比如,我要重新设置一下RB的缓存区大小,这都需要串行操作,因此,在ring_buffer结构中有mutex成员,用来避免这些更改RB的操作的竞争. 每个cpu的缓存区结构为: struct ring_buffer_per_cpu { /*该cpu buffer所在的CPU*/ int cpu; /*cpu buffer所属的RB*/ struct ring_buffer *buffer; /*读锁,用了避免读者的操行操作,有时在 *写者切换页面的时候,也需要持有此锁 */ spinlock_t reader_lock; /* serialize readers */ raw_spinlock_t lock; struct lock_class_key lock_key; /*cpu buffer的页面链表*/ struct list_head pages; /*起始读位置*/ struct buffer_page *head_page; /* read from head */ /*写位置*/ struct buffer_page *tail_page; /* write to tail */ /*提交位置,只有当被写的页面提交过后 *才允许被读 */ struct buffer_page *commit_page; /* committed pages */ /*reader页面, 用来交换读页面*/ struct buffer_page *reader_page; unsigned long nmi_dropped; unsigned long commit_overrun; unsigned long overrun; unsigned long read; local_t entries; /*最新的页面commit时间*/ u64 write_stamp; /*最新的页面read时间*/ u64 read_stamp; /*cpu buffer的禁用启用标志*/ atomic_t record_disabled; } 首先,对每一个cpu的操作限制是由ring_buffer_per_cpu->record_disabled来实现的.同ring_buffer一样,禁用加1,启用减1. 从上图的全局结构关联图中,我们也可以看到,每个cpu都有一系列的页面,这样页面都链入在pages中. 该页面的结构如下: struct buffer_page { /*用来形成链表*/ struct list_head list; /* list of buffer pages */ /*写的位置*/ local_t write; /* index for next write */ /*读的位置*/ unsigned read; /* index for next read */ /*页面中有多少项数据*/ local_t entries; /* entries on this page */ struct buffer_data_page *page; /* Actual data page */ }; 具体的缓存区是由struct buffer_data_page指向的,实际上,它是具体页面的管理头部,结构如下: struct buffer_data_page { /*页面第一次被写时的时间戳*/ u64 time_stamp; /* page time stamp */ /*提交的位置*/ local_t commit; /* write committed index */ /*用来存放数据的缓存区*/ unsigned char data[]; /* data of buffer page */ }; 这里就有一个疑问了,为什么提交页面要放到struct buffer_date_page中,而不放到struct buffer_page呢? 三: ring buffer的初始化 Ring buffer的初始化函数为ring_buffer_alloc(). 代码如下: struct ring_buffer *ring_buffer_alloc(unsigned long size, unsigned flags) { struct ring_buffer *buffer; int bsize; int cpu; /* Paranoid! Optimizes out when all is well */ /*如果struct buffer_page的大小超过了struct page的大小,编译时会报错 *因为ring_buffer_page_too_big()其实并不存在. */ if (sizeof(struct buffer_page) > sizeof(struct page)) ring_buffer_page_too_big(); /* keep it in its own cache line */ /*alloc and init struct ring_buffer*/ buffer = kzalloc(ALIGN(sizeof(*buffer), cache_line_size()), GFP_KERNEL); if (!buffer) return NULL; /*init cpumask*/ if (!alloc_cpumask_var(&buffer->cpumask, GFP_KERNEL)) goto fail_free_buffer; /*BUF_PAGE_SIZE means the data size of per page, *size/BUF_PAGE_SIZE can calculate page number of per cpu. */ buffer->pages = DIV_ROUND_UP(size, BUF_PAGE_SIZE); buffer->flags = flags; /* buffer->clock is the timestap of local cpu*/ buffer->clock = trace_clock_local; /* need at least two pages */ if (buffer->pages == 1) buffer->pages++; /* * In case of non-hotplug cpu, if the ring-buffer is allocated * in early initcall, it will not be notified of secondary cpus. * In that off case, we need to allocate for all possible cpus. */ #ifdef CONFIG_HOTPLUG_CPU get_online_cpus(); cpumask_copy(buffer->cpumask, cpu_online_mask); #else cpumask_copy(buffer->cpumask, cpu_possible_mask); #endif /*number of cpu*/ buffer->cpus = nr_cpu_ids; /* alloc and init buffer for per cpu,Notice:buffer->buffers is a double pointer*/ bsize = sizeof(void *) * nr_cpu_ids; buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()), GFP_KERNEL); if (!buffer->buffers) goto fail_free_cpumask; for_each_buffer_cpu(buffer, cpu) { buffer->buffers[cpu] = rb_allocate_cpu_buffer(buffer, cpu); if (!buffer->buffers[cpu]) goto fail_free_buffers; } #ifdef CONFIG_HOTPLUG_CPU buffer->cpu_notify.notifier_call = rb_cpu_notify; buffer->cpu_notify.priority = 0; register_cpu_notifier(&buffer->cpu_notify); #endif put_online_cpus(); mutex_init(&buffer->mutex); return buffer; fail_free_buffers: for_each_buffer_cpu(buffer, cpu) { if (buffer->buffers[cpu]) rb_free_cpu_buffer(buffer->buffers[cpu]); } kfree(buffer->buffers); fail_free_cpumask: free_cpumask_var(buffer->cpumask); put_online_cpus(); fail_free_buffer: kfree(buffer); return NULL; } 结合我们上面分析的数据结构,来看这个函数,应该很简单,首先,我们在这个函数中遇到的第一个疑问是: 为什么RB至少需要二个页面呢? 我们来假设一下只有一个页面的情况,RB开始写,因为head和tail是重合在一起的,当写完一个页面的时候,tail后移,因为只有一个页面,还是会指向这个页面,这样还是跟head重合在一起,如果带有RB_FL_OVERWRITE标志的话,head会后移试图清理这个页面,但后移之后还是指向这个页面,也就是说tail跟head还是会重合.假设此时有读操作,读完了head的数据,造成head后移,同样head和tail还是重合在一起.因此就造成了,第一次写完这个页面,就永远无法再写了,因为这时候永远都是一个满的状态. 也就是说,这里需要两个页面是为了满足缓存区是否满的判断,即tail->next == head 然后,我们面临的第二个问题是,RB怎么处理hotplug cpu的情况呢? 看下面的代码: /*number of cpu*/ buffer->cpus = nr_cpu_ids; /* alloc and init buffer for per cpu,Notice:buffer->buffers is a double pointer*/ bsize = sizeof(void *) * nr_cpu_ids; buffer->buffers = kzalloc(ALIGN(bsize, cache_line_size()), GFP_KERNEL); 从上面的代码看到,在初始化RB的时候,它为每个可能的CPU都准备了一个 “框”,下面来看下这个 “框”的初始化: for_each_buffer_cpu(buffer, cpu) { buffer->buffers[cpu] = rb_allocate_cpu_buffer(buffer, cpu); if (!buffer->buffers[cpu]) goto fail_free_buffers; } 从此可以看到,它只为当时存在的CPU分配了缓存区. 到这里,我们大概可以猜到怎么处理hotplug cpu的情况了: 在有CPU加入时,为这个CPU对应的 “框”对应分配内存,在CPU拨除或掉线的情况下,释放掉该CPU对应的内存. 到底是不是跟我们所想的一样呢? 我们继续看代码: #ifdef CONFIG_HOTPLUG_CPU buffer->cpu_notify.notifier_call = rb_cpu_notify; buffer->cpu_notify.priority = 0; register_cpu_notifier(&buffer->cpu_notify); #endif 如上代码片段,它为hotplug CPU注册了一个notifier, 它对优先级是0,对应的处理函数是rb_cpu_notify,代码如下: static int rb_cpu_notify(struct notifier_block *self, unsigned long action, void *hcpu) { struct ring_buffer *buffer = container_of(self, struct ring_buffer, cpu_notify); long cpu = (long)hcpu; switch (action) { /*CPU处理active 的notify*/ case CPU_UP_PREPARE: case CPU_UP_PREPARE_FROZEN: /*如果cpu已经位于RB的cpu位图,说明已经为其准备好了 *缓存区,直接退出 */ if (cpu_isset(cpu, *buffer->cpumask)) return NOTIFY_OK; /*否则,它是一个新的CPU, 则为其分配缓存，如果 *分配成功，则将其在cpu位图中置位*/ buffer->buffers[cpu] = rb_allocate_cpu_buffer(buffer, cpu); if (!buffer->buffers[cpu]) { WARN(1, "failed to allocate ring buffer on CPU %ld\n", cpu); return NOTIFY_OK; } smp_wmb(); cpu_set(cpu, *buffer->cpumask); break; case CPU_DOWN_PREPARE: case CPU_DOWN_PREPARE_FROZEN: /* * Do nothing. * If we were to free the buffer, then the user would * lose any trace that was in the buffer. */ /*如果是CPU处于deactive的notify，则不需要将其占的缓存 *释放，因为一旦释放，我们将失去该cpu上的trace 信息*/ break; default: break; } return NOTIFY_OK; } 首先,RB的结构体中内嵌了struct notifier_block,所以,我们利用其位移差就可以取得对应的RB结构,上面的代码比较简单,不过,与我们之前的估计有点差别,即,在CPU处理deactive状态的时候,并没有将其对应的缓存释放,这是为了避免丢失该CPU上的trace信息. 接下来我们看一下对每个CPU对应的缓存区的初始化,它是在rb_allocate_cpu_buffer()中完成的,代码如下: static struct ring_buffer_per_cpu * rb_allocate_cpu_buffer(struct ring_buffer *buffer, int cpu) { struct ring_buffer_per_cpu *cpu_buffer; struct buffer_page *bpage; unsigned long addr; int ret; /* alloc and init a struct ring_buffer_per_cpu */ cpu_buffer = kzalloc_node(ALIGN(sizeof(*cpu_buffer), cache_line_size()), GFP_KERNEL, cpu_to_node(cpu)); if (!cpu_buffer) return NULL; cpu_buffer->cpu = cpu; cpu_buffer->buffer = buffer; spin_lock_init(&cpu_buffer->reader_lock); cpu_buffer->lock = (raw_spinlock_t)__RAW_SPIN_LOCK_UNLOCKED; INIT_LIST_HEAD(&cpu_buffer->pages); /* alloc and init cpubuffer->reader_page */ bpage = kzalloc_node(ALIGN(sizeof(*bpage), cache_line_size()), GFP_KERNEL, cpu_to_node(cpu)); if (!bpage) goto fail_free_buffer; cpu_buffer->reader_page = bpage; addr = __get_free_page(GFP_KERNEL); if (!addr) goto fail_free_reader; bpage->page = (void *)addr; rb_init_page(bpage->page); INIT_LIST_HEAD(&cpu_buffer->reader_page->list); /* alloc and init the page list, head_page, tail_page and commit_page are all point to the fist page*/ ret = rb_allocate_pages(cpu_buffer, buffer->pages); if (ret < 0) goto fail_free_reader; cpu_buffer->head_page = list_entry(cpu_buffer->pages.next, struct buffer_page, list); cpu_buffer->tail_page = cpu_buffer->commit_page = cpu_buffer->head_page; return cpu_buffer; fail_free_reader: free_buffer_page(cpu_buffer->reader_page); fail_free_buffer: kfree(cpu_buffer); return NULL; } 这段代码的逻辑比较清晰,首先,它分配并初始化了ring_buffer_per_cpu结构,然后对其缓存区进行初始化.在这里我们需要注意,reader_page单独占一个页面,并末与其它页面混在一起.初始化状态下,head_pages,commit_page,tail_page都指向同一个页面,即ring_buffer_per_cpu->pages链表中的第一个页面. 四:ring buffer的写操作一般来说,trace子系统往ring buffer中写数据通常分为两步,一是从ring buffer是取出一块缓冲区,然后再将数据写入到缓存区,然后再将缓存区提交.当然ring buffer也提供了一个接口直接将数据写入ring buffer,两种方式的实现都是一样的,在这里我们分析第一种做法,后一种方式对应的接口为ring_buffer_write().可自行对照分析. 4.1:ring_buffer_lock_reserver()分析 ring_buffer_lock_reserve()用于从ring buffer中取出一块缓存,函数如下: struct ring_buffer_event * ring_buffer_lock_reserve(struct ring_buffer *buffer, unsigned long length) { struct ring_buffer_per_cpu *cpu_buffer; struct ring_buffer_event *event; int cpu, resched; /* jude wheter ring buffer is off ,can use trace_on/trace_off to enable/disable it */ if (ring_buffer_flags != RB_BUFFERS_ON) return NULL; /* if the ring buffer is disabled, maybe some have other operate in this ring buffer currently */ if (atomic_read(&buffer->record_disabled)) return NULL; /* If we are tracing schedule, we don't want to recurse */ resched = ftrace_preempt_disable(); cpu = raw_smp_processor_id(); /* not trace this cpu? */ if (!cpumask_test_cpu(cpu, buffer->cpumask)) goto out; /*get the cpu buffer which associated with this CPU*/ cpu_buffer = buffer->buffers[cpu]; /* if the cpu buffer is disabled */ if (atomic_read(&cpu_buffer->record_disabled)) goto out; /* change the data length to ring buffer length, include a head in this buffer */ length = rb_calculate_event_length(length); if (length > BUF_PAGE_SIZE) goto out; /* get the length buffer from cpu_buffer */ event = rb_reserve_next_event(cpu_buffer, RINGBUF_TYPE_DATA, length); if (!event) goto out; /* * Need to store resched state on this cpu. * Only the first needs to. */ /* if the preempt is enable and need sched in this cpu, set the resched bit */ if (preempt_count() == 1) per_cpu(rb_need_resched, cpu) = resched; return event; out: ftrace_preempt_enable(resched); return NULL; } 在进行写操作之前,要首先确认RB是否能被所在的CPU操作. 在这里要经过四个步骤的确认: 1: 确认全局ring_buffer_flags标志是否为RB_BUFFERS_ON. 该标志是一个全局的RB控制,它控制着任何一个RB的操作, RB_BUFFERS_ON为允许, RB_BUFFERS_OFF为禁用.对应的接口为trace_on()和trace_off(). 2: 确认该RB的record_disabled是否为0. 我们在前面分析RB的结构体时分析过,该成员是控制对应RB的操作 3:确认所在的CPU是否在RB的CPU位图中. 所在不在RB的CPU位图,表示还尚末为这个CPU分配缓存,暂时不能进行任何操作 4:确认该CPU对应的ring_buffer_per_cpu->record_disabled是否为0. 它是对单个CPU的控制此外,在RB中的禁用/启用抢占也很有意思,如下代码片段如示: ...... /* If we are tracing schedule, we don't want to recurse */ resched = ftrace_preempt_disable(); ...... ...... /* if the preempt is enable and need sched in this cpu, set the resched bit */ if (preempt_count() == 1) per_cpu(rb_need_resched, cpu) = resched; ...... 这段代码的逻辑是: 在禁用抢占之前先检查当前进程是否有抢占,如果有,resched为1,否则为0.然后禁止抢占在操作完了之后,如果当前是第一次禁止抢占,则将resched保存在RB的per-cpu变量中. 为什么要弄得如此复杂呢? 我们来看一下ftrace_preempt_disable()的代码就明白了: /** * ftrace_preempt_disable - disable preemption scheduler safe * * When tracing can happen inside the scheduler, there exists * cases that the tracing might happen before the need_resched * flag is checked. If this happens and the tracer calls * preempt_enable (after a disable), a schedule might take place * causing an infinite recursion. * * To prevent this, we read the need_resched flag before * disabling preemption. When we want to enable preemption we * check the flag, if it is set, then we call preempt_enable_no_resched. * Otherwise, we call preempt_enable. * * The rational for doing the above is that if need_resched is set * and we have yet to reschedule, we are either in an atomic location * (where we do not need to check for scheduling) or we are inside * the scheduler and do not want to resched. */ static inline int ftrace_preempt_disable(void) { int resched; resched = need_resched(); preempt_disable_notrace(); return resched; } 这段代码的注释说得很明显了,它是为了防止了无限递归的trace scheduler和防止在原子环境中有进程切换的动作. 其实,说白了,它做这么多动作,就是为了防止在启用抢占的时候,避免调用schedule()进行进程切换. 那,就有一个疑问了,既然无论在当前是否有抢占都要防止有进程切换,为什么不干脆调用preempt_enable_no_resched()来启用抢占呢? 我们要分配长度为length的数据长度,那是否它在RB中占的长度就是length呢?肯定不是,因为RB中的数据还是自己的管理头部.至少,在RB中读数据的时候,它需要知道这个数据有多长. 那它究竟在RB中占用多少的长度呢?我们来跟踪rb_calculate_event_length(): static unsigned rb_calculate_event_length(unsigned length) { struct ring_buffer_event event; /* Used only for sizeof array */ /* zero length can cause confusions */ if (!length) length = 1; /* if length is more than RB_MAX_SMALL_DATA,it need arry[0] to store the data length */ if (length > RB_MAX_SMALL_DATA) length += sizeof(event.array[0]); /* add the length of struct ring_buffer_event */ length += RB_EVNT_HDR_SIZE; /* must align by 4 */ length = ALIGN(length, RB_ALIGNMENT); return length; } 别看这个函数很短小,却暗含乾坤.从代码中看到,其实我们存入到RB中的数据都是用struct ring_buffer_event来表示的,理解了这个数据结构,上面的代码逻辑自然就清晰了. 该结构体定义如下: struct ring_buffer_event { u32 type:2, len:3, time_delta:27; u32 array[]; }; Type表示这块数据的类型,len有时是表示这块数据的长度,time_delta表示这块数据与上一块数据的时间差.从上面的定义可以看出: struct ring_buffer_event的len定义只占三位.它最多只能表示0xb11100的数据大小.另外,在RB中有一个约束,event中的数据必须按4对齐的,那么数据长度的低二位肯定为0,那么ring_buffer_event中的len只能表示从0xb0~0xb11100的长度,即0~28的长度,那么,如果数据长度超过了28,那应该要怎么表示呢? 在数据长度超过28的情况下,会使用ring_buffer_event中的arry[0]表示里面的数据长度,即从后面的数据部份取出4字来额外表示它的长度. Ring buffer event有以下面这几种类型,也就是type的可能值: enum ring_buffer_type { RINGBUF_TYPE_PADDING, RINGBUF_TYPE_TIME_EXTEND, /* FIXME: RINGBUF_TYPE_TIME_STAMP not implemented */ RINGBUF_TYPE_TIME_STAMP, RINGBUF_TYPE_DATA, }; RINGBUF_TYPE_PADDING: 是指往ring buffer中填充的数据, 这用在页面有剩余或者当前event无效的情况. RINGBUF_TYPE_TIME_EXTEND: 表示附加的时间差信息,这个信息会存放在arry[0]中. RINGBUF_TYPE_TIME_STAMP: 表示存放的是时间戳信息, array[0]用来存放tv_nsec, array[1..2]中存放 tv_sec.在现在的代码中还末用到. RINGBUF_TYPE_DATA:表示里面填充的数据,数据的长度表示方式在前面已经分析过了,这里就不再赘述了. 好了,返回rb_calculate_event_length(): RB_MAX_SMALL_DATA = 28也就是我们上面分析的event中的最小长度,如果要存入的长度大于这个长度的,那么,就需要数据部份的一个32位数用来存放它的长度,因此这种情况下,需要增加sizeof(event.array[0])的长度.另外,event本身也要占用RB的长度,所以需要加上event占的空间,也就是代码中的RB_EVNT_HDR_SIZE. 最后,数据要按4即RB_ALIGNMENT对齐. 那,我们来思考一下,为什么在length为0的情况,需要将其设为1呢? 我们来做个假设,如果length为0,且末做调整,因为event占的大小是两个32位,也就是8.它跟4已经是对齐的了.此时加上length,也就是0.经过4对齐后,它计算出来的长度仍然是event的大小. 在rb_update_event()中对event的各项数据进行赋值时,它的len对象为0. 而对于数据长度超过RB_MAX_SMALL_DATA来说,它的len对象也为0. 此时就无法区别这个对象是长度超过RB_MAX_SMALL_DATA的对象,还是长度为0的对象,也就是无法确定数据后面的一个32位的空间是否是属于这个对象(这里提到了rb_update_event(),我们在后面遇到它再进行详细分析,在这里只需要知道就是调用它来对event的各成员进行初始化就可以了). 现在要到ring buffer中去分配存放的空间了,它是在rb_reserve_next_event()中完成的. 可以说,这个函数就是ring buffer的精华部份了.首先,我们要明确一下,ring buffer它要实现的功能是什么? Ring buffer是用来做存放trace信息用的,既然是做trace.那它就不能对执行效率产生过多的影响,但是它可以占据稍微多一点的空间.然后,每个CPU的每个执行路径trace数据都是放在同一个buffer中的,所以在写数据的时候,要考虑多CPU的竞争情况. 另外,只要我们稍加注意就会发现, ring_buffer_lock_reserve()中调用的rb_reserve_next_event()函数是在所在CPU对应的缓存区上进行操作的. ring_buffer_lock_reserve()和ring_buffer_unlock_commit()是一对函数.从这两个函数的字面意思看来,一个是lock,另一个是unlock.这里的lock机制不是我们之前所讲的类似于mutex, spin_lock之类的lock.因为每个cpu都对应一个缓存区(struct ring_buffer_per_cpu),每个CPU只能读写属于它的缓存区,这样就不需要考虑SMP上的竞争了.因此就不需要使用spinlock, 在这里也不能使用mutex.因为trace在很多不确定情况下会用到,例如function tracer 在每个函数里都会用到,这样就会造成CPU上的所有执行线程去抢用一个mutex的情况.这样会大大降低系统效率,甚至会造成CPU空运转.另外,如果使用mutex,可能会在原子环境中引起睡眠操作. Ring buffer中的lock是指内核抢占,在调用ring_buffer_lock_reserver()时禁止内核抢占,在调用ring_buffer_unlock_commit()是恢复内核抢占.这样在竞争的时候,就只需要考虑中断和NMI了.在这里要注意中断抢占的原则:只有高优先的中断才能抢占低优先级的中断.也就是中断是不能相互嵌套的.例如,A线程正在执行,中断线程B发生了,因此从AàB.在B没有执行完的时候是不可能会切换到A的.如下所示: 500)this.width=500;" border=0> 另外,在代码注释中经常看到first commit,这个first commit 到底是什么意思呢? 其实它就表示对应CPU缓存区commit之后,第一个从缓存区中的取动作. 对应到上图的 “正常的中断抢占序列”, A是first commit,它被B中断了,B就不是fist commit. 判断是否是first commit是通过下列语句来判断的,代码如下: cpu_buffer->tail_page == cpu_buffer->commit_page && rb_page_write(cpu_buffer->tail_page) == rb_commit_index(cpu_buffer) 在上面我们分析过,每个CPU的缓存区是从tail页面开始写,从head页面开始读,commit_page则是表示已经提交到的页面. 上面的语句中,如果提交页面是写页面,写序号等于提交序号.就表示当前的位置就是commit的位置也就是first commit. 经过上面的分析,相信对该函数的流程有大概的了解了,下面来分析下具体的代码,可以说该函数的每一句代码都值得推敲,采用分段分析的方法,如下: static struct ring_buffer_event * rb_reserve_next_event(struct ring_buffer_per_cpu *cpu_buffer, unsigned type, unsigned long length) { struct ring_buffer_event *event; u64 ts, delta; int commit = 0; int nr_loops = 0; again: /* * We allow for interrupts to reenter here and do a trace. * If one does, it will cause this original code to loop * back here. Even with heavy interrupts happening, this * should only happen a few times in a row. If this happens * 1000 times in a row, there must be either an interrupt * storm or we have something buggy. * Bail! */ if (RB_WARN_ON(cpu_buffer, ++nr_loops > 1000)) return NULL; 在这里,在调用这个函数之前禁止了抢止,中断和NMI在这里存在着竞争,因此在下面的运行中,随时都会被中断/NMI所抢占. 由于在从struct ring_buffer_per_cpu中取页面的时候,会有当前页面空间不足,需要前进一个页面的情况.每次前进一个页面都会跳转到again,此时nr_loops都会增加1, 如果在一次请求中,这样的情况出现了1000次,说明中断抢占的次数太多了,很可能是由于中断风暴(interrupte storm)或者是bug造成的. /*取当前的时间戳*/ ts = ring_buffer_time_stamp(cpu_buffer->cpu); /* * Only the first commit can update the timestamp. * Yes there is a race here. If an interrupt comes in * just after the conditional and it traces too, then it * will also check the deltas. More than one timestamp may * also be made. But only the entry that did the actual * commit will be something other than zero. */ /*只有第一次处于提交状态的请求才能够更新cpu_buffer->write_stamp*/ if (cpu_buffer->tail_page == cpu_buffer->commit_page && rb_page_write(cpu_buffer->tail_page) == rb_commit_index(cpu_buffer)) { delta = ts - cpu_buffer->write_stamp; /* make sure this delta is calculated here */ barrier(); /* Did the write stamp get updated already? */ /*如果之前取的当前时间戳小于cpu_buffer->write_stamp说明 *ring_buffer的write_stamp已经更新过了,也就是在发生了抢占 */ /* ----------NOTIC HERE----------*/ if (unlikely(ts < cpu_buffer->write_stamp)) delta = 0; /* 如果更新时间差值大于1 << 27,那就必须要插入一个表示时间 * 的ring_buffer_event */ if (test_time_stamp(delta)) { commit = rb_add_time_stamp(cpu_buffer, &ts, &delta); if (commit == -EBUSY) return NULL; if (commit == -EAGAIN) goto again; RB_WARN_ON(cpu_buffer, commit < 0); } } else /* Non commits have zero deltas */ /*在commit时发生的抢占,它的time stamp delta为0*/ delta = 0; 从上面的if判断可以看到,只有在fist commit的时候才会计算delta,其它的情况下,delta都是0. 我们来思考一下,为什么在确认了是fist commit,进入到了if,还需要进行: if (unlikely(ts < cpu_buffer->write_stamp)) delta = 0; 的判断呢? 什么情况下会有当前时间戳小于cpu_buffer最新提交时的时间戳呢? 对应到上面的”正常中断抢占序列”的图,只有在A处才会计算delta时间,在被B,C抢占后,它的delta是为0的. 这个delta到底是用来做什么的呢?它为什么要用这样的判断方式呢? 我们在之前说过,在ring_buffer_per_cpu中的每一块数据都带有一个event的头部,即: struct ring_buffer_event { u32 type:2, len:3, time_delta:27; u32 array[]; }; 它里面有一个time_delta的成员,占27位. 在每一个页面的头部,即Struct buffer_data_page里面也有一个时间戳,即: struct buffer_data_page { u64 time_stamp; /* page time stamp */ local_t commit; /* write commited index */ unsigned char data[]; /* data of buffer page */ } 那这几个时间戳有什么样的关联呢? 在ring_buffer_per_cpu中有一个timestamp,它表示最近commit时的时间戳. 每次切换进一个新页面时,该页面对应的: buffer_data_page->time_stamp会记录当前的时间戳. 即buffer_date_page->time_stamp记录页面被切换成写页面时的时间戳. 而ring_buffer_event->time_delta表示当前时间和上一次commit时间即ring_buffer_per_cpu->time_stamp的差值. 综上所述,存在以下关系: 页面中的第一个event, event1在进行写操作时的时间戳为: buffer_data_page->time_stamp + ring_buffer_event1->time_delta. 第二个event,event2在进行写操作时的时间戳为: buffer_data_page->time_stamp+ring_buffer_event1->time_delta+ ring_buffer_event2->time_delta. 依次类推,不过有种情况是特别的,即RINGBUF_TYPE_TIME_EXTEND类型的EVENT,它是为了有时delta时间超过27位时,扩展了一个32位用来存放的时间戳.这也就是上面代码中的if (test_time_stamp(delta)).另外需要注意,这里的返回值commit,只有在fist commit的时候才会为1. 这段代码有个值得思考的地方,也就是上面代码的”----------NOTIC HERE----------“注释处. 在这里判断了它是first commit,为什么会有可能出现当前的时间戳比最近提交的时间戳还要小呢? 试想一下,如果在”----------NOTIC HERE----------”处发生了中断,这个中断执行路径反而会先从RB中取得空间,然后commit,就会出现这样的情况了.因此,我们要注意,fist commit是指最近commit后的一个状态,而不是第一个进入rb_reserve_next_event()的状态. 上面代码中的rb_add_time_stamp()子函数的执行流程跟我们在后面要分析的部份差不多,因此在这里就不对它进行详细分析了. /*从ring_buffer中取出event*/ event = __rb_reserve_next(cpu_buffer, type, length, &ts); if (PTR_ERR(event) == -EAGAIN) goto again; if (!event) { if (unlikely(commit)) /* * Ouch! We needed a timestamp and it was commited. But * we didn't get our event reserved. */ rb_set_commit_to_write(cpu_buffer); return NULL; } /* * If the timestamp was commited, make the commit our entry * now so that we will update it when needed. */ if (commit) rb_set_commit_event(cpu_buffer, event); else if (!rb_is_commit(cpu_buffer, event)) delta = 0; /*event->time_delta表示的是距离上次commit时的时间差*/ event->time_delta = delta; return event } 剩下的代码就很好理解了,如果取得的event为空,说明发生了错误,另外在commit为1的情况下是必须要commit的,因为它已经更新了ring_buffer_per_cpu. 另外,在上面的:
if (commit) rb_set_commit_event(cpu_buffer, event); else if (!rb_is_commit(cpu_buffer, event)) delta = 0; 为什么需要再次判断rb_is_commit()呢? 这是因为,可能在__rb_reserve_next()之前有中断抢占了当前执行路径,而先从RB取得空间,这种情况下,先取得RB空间的,成了first commit. 注意在这里的条件是” __rb_reserve_next()”之前,因为在后面我们会看到,在__rb_reserve_next()中是用原子操作来避免竞争的,实际上, __rb_reserve_next()使写操作串行化,即一个一个按顺序通过它.因为只有第一个通过__rb_reserve_next()才会是first commit状态,所以,在它的后面的话,就无所谓抢占了. 这个函数就分析到这里了,在里面有一个重要的子函数,即__rb_reserve_next(),代码较长,用分段分析的方式如下: static struct ring_buffer_event * __rb_reserve_next(struct ring_buffer_per_cpu *cpu_buffer, unsigned type, unsigned long length, u64 *ts) { struct buffer_page *tail_page, *head_page, *reader_page, *commit_page; unsigned long tail, write; struct ring_buffer *buffer = cpu_buffer->buffer; struct ring_buffer_event *event; unsigned long flags; commit_page = cpu_buffer->commit_page; /* we just need to protect against interrupts */ barrier(); tail_page = cpu_buffer->tail_page; /*local_add_return()是一个原子操作,起保护作用. *write是加上了length之后的位置,tail是之前的位置 */ write = local_add_return(length, &tail_page->write); tail = write - length; 注意这里的临界条件,对struct buffer_page->write的更新是采用的原子操作,即它的操作是不能被打断的,这也是一种临界区的保护方式. 在这里的临界操作,需要注意: write = local_add_return(length, &tail_page->write); tail = write - length; 它是先进行原子加,然后再write-length取得加之前的位置,这样就保证tail对应的刚好是取出来的那缓存区. /* See if we shot pass the end of this buffer page */ /*如果超过了一个页面,那就需要前进一个页面了*/ if (write > BUF_PAGE_SIZE) { struct buffer_page *next_page = tail_page; /*因此这里要更改tail_page的指向了,不能再有竞争的情况了 *因为调用local_irq_save()来禁止中断,调用__raw_spin_lock()来保护 *ring_buffer 的读操作 */ /*---------NOTICE HERE----------*/ local_irq_save(flags); /* * Since the write to the buffer is still not * fully lockless, we must be careful with NMIs. * The locks in the writers are taken when a write * crosses to a new page. The locks protect against * races with the readers (this will soon be fixed * with a lockless solution). * * Because we can not protect against NMIs, and we * want to keep traces reentrant, we need to manage * what happens when we are in an NMI. * * NMIs can happen after we take the lock. * If we are in an NMI, only take the lock * if it is not already taken. Otherwise * simply fail. */ if (unlikely(in_nmi())) { if (!__raw_spin_trylock(&cpu_buffer->lock)) goto out_reset; } else __raw_spin_lock(&cpu_buffer->lock); 如果write > BUF_PAGE_SIZE,说明当前的页面已经不够空间来存放一个event了,因此,我们需要切换到下一个页面.既然要切换页面,那就需要有同步措施了,在这里采用的是禁中断和自旋锁cpu_buffer->lock, reader在读RB的时候也会持有该锁,这样就同步了写者与读者. 在这里要注意,禁中断只是禁止外部设备的中断响应,并不能禁止NMI, 所以在这里还需要NMI的情况特殊考虑,如果是在NMI的情况,如果自旋锁被占用,就立即返回,我们不能在这个里面等待太久. /*使next_page指向tail_page的下一个页面*/ rb_inc_page(cpu_buffer, &next_page); head_page = cpu_buffer->head_page; reader_page = cpu_buffer->reader_page; /* we grabbed the lock before incrementing */ if (RB_WARN_ON(cpu_buffer, next_page == reader_page)) goto out_unlock; /* * If for some reason, we had an interrupt storm that made * it all the way around the buffer, bail, and warn * about it. */ /*可能是前进的次数太多了也可能是因为ring_buffer的页面太少了 *导致了next_page和commit_page重合的情况 */ if (unlikely(next_page == commit_page)) { WARN_ON_ONCE(1); goto out_unlock; } 进入到自旋锁的保护区之后,我们就可以前进一个页面了.这里有几种情况: 1: 前进之后的页面不可能和reader_page重合, 我们在后面可以看到,reader_page是一个孤立的页面,不位于cpu_buffer->pages链表中. 2: commit页面与前进之后的页面重合,这有可能是前进次数太多,即中断次数太多(还来不及commit),也有可能是RB的页面数目太少. /*前进一个页面之后碰到了head_page,即ring_buffer已经满了*/ /*如果带有RB_FL_OVEWRITE标志,就将旧的数据清除掉*/ if (next_page == head_page) { if (!(buffer->flags & RB_FL_OVERWRITE)) goto out_unlock; /* tail_page has not moved yet? */ if (tail_page == cpu_buffer->tail_page) { /* count overflows */ rb_update_overflow(cpu_buffer); rb_inc_page(cpu_buffer, &head_page); cpu_buffer->head_page = head_page; cpu_buffer->head_page->read = 0; } } 如果前进一个页面之后,跟head_page重合了,说明cpu buffer已经满了,如果带有RB_FL_OVERWRITE标志的话,我们就可以将head_page中的数据冲刷掉. 在这里需要注意,为什么这里要再判断? if (tail_page == cpu_buffer->tail_page) 这是因为在持有自旋锁之前是有竞争的,在上面代码的 ”/*---------NOTICE HERE----------*/”注释处,我们来考虑一下,如果此时有路径 A运行到了NOTICE HERE,发生了中断,中断路径B也进入了NOTICE HERE,然后更新了cpu_buffer->tail_page,而后退出.此时再切换回路径A,A会继续往下执行,实际上,这时候cpu_buffer->tail_page已经更新过了. 如果出现重合的现象,我们只需要将head_page前移就可以了,这里不需要将head_page的内容清空,因为在取到下一个页面的时候,会调用: local_set(&next_page->write, 0); local_set(&next_page->page->commit, 0); 将其重置. 我们跟踪看一下rb_update_overflow()函数的代码片段,这个函数很简单,但里面有一个值得我们注意的地方: static void rb_update_overflow(struct ring_buffer_per_cpu *cpu_buffer) { ...... for (head = 0; head < rb_head_size(cpu_buffer); head += rb_event_length(event)) { event = __rb_page_index(cpu_buffer->head_page, head); if (RB_WARN_ON(cpu_buffer, rb_null_event(event))) return; ．．．．．． } } 疑问,这通外里为什么不可能遍历出是null的event呢? Null的event是指: event->type == RINGBUF_TYPE_PADDING && event->time_delta == 0; 它是一个页面不足以存放一个event后的填充数据. 我们来看一下它的循环判断条件: head < rb_head_size(cpu_buffer); rb_head_size()定义如下: static inline unsigned rb_head_size(struct ring_buffer_per_cpu *cpu_buffer) { return rb_page_commit(cpu_buffer->head_page); } 即commit的位置. 其实,这里断定不可能出现null event的原因是,在commit的时候,不会commit 填充数据.它只会commit有效数据,这在我们后面的分析中可以得到确认. /* * If the tail page is still the same as what we think * it is, then it is up to us to update the tail * pointer. */ /*如果tail_page没有发生更改,就可以更改tail_page的指向了 *即前进一个页面 */ if (tail_page == cpu_buffer->tail_page) { local_set(&next_page->write, 0); local_set(&next_page->page->commit, 0); cpu_buffer->tail_page = next_page; /* reread the time stamp */ /*更新页面转入写页面时的时间戳*/ *ts = ring_buffer_time_stamp(cpu_buffer->cpu); cpu_buffer->tail_page->page->time_stamp = *ts; } 分析这小段代码要结合在上段代码中的抢占分析,如果没有抢占,或者是第一个递增此页面,就可以更新cpu_buffer->tail_page的指向了.在这里要注意,对每个取到的页面都是进行初始化,但没必要将整个页面都清零.只需要将它的写位置和提交位置置为0就可以了. 另外,我们在这里也可以看到,我们将这个页面的时间戳置为了该页面切换成写页面的时间戳.这个时间戳后面还会调整,接着看. /* * The actual tail page has moved forward. */ /*ring_buffer后还有一段空闲的区域,将它赋为RINGBUF_TYPE_PADDING*/ if (tail < BUF_PAGE_SIZE) { /* Mark the rest of the page with padding */ event = __rb_page_index(tail_page, tail); event->type = RINGBUF_TYPE_PADDING; } /* 恢复tail_page->write的值,因为在local_add_return()之后存在竞争, * 即在很多个执行路径中多次相加后,才会有禁中断 * 和自旋锁保护 */ if (tail <= BUF_PAGE_SIZE) /* Set the write back to the previous setting */ local_set(&tail_page->write, tail); /* * If this was a commit entry that failed, * increment that too */ /*如果是一个fist commit状态的页面,commit it*/ if (tail_page == cpu_buffer->commit_page && tail == rb_commit_index(cpu_buffer)) { rb_set_commit_to_write(cpu_buffer); } /*更新状态完成,释放临界区*/ __raw_spin_unlock(&cpu_buffer->lock); local_irq_restore(flags); /* fail and let the caller try again */ /*返回EAGAIN,表示重新到ring_buffer中分配event*/ return ERR_PTR(-EAGAIN); } 可能看到这里,大家都有点疑问,为什么需要对tail做这么多次判断呢? 这是因为,这里有个特殊的情况,如下图示: 当运行到1的时候,是会有竞争情况的,如下示: 1:执行路径A运行到1的时候,有write > BUF_PAGE_SIZE的情况,假设A是在这次执行中,最先发生write > BUF_PAGE_SIZE的.这时竞争发生,有其它的中断过来了. 2:执行路径B抢占了执行路径A, 此时经过local_add_return()计算后,仍然会有: write >BUF_PAGE_SIZE的情况.,运行到1处,又有其它中断过来了.其实它这里计算出来的write值是在上一次的基础上计算出来的 3:执行路径C抢点了执行路径B, ...... 经过上面的分析,我们可得知,只有A,也就是第一个发生write > BUF_PAGE_SIZE的路径的tail才会小于BUF_PAGE_SIZE,因为其它的路径都是在超过BUF_PAGE_SIZE的基础上计算出来的. 正是因为这样,所以在2处才有tail < BUG_PAGE_SIZE的判断,并且更新tail_page->write.这样,在恢复到A的时候,就会将tail_page->write回复到原始值了. /* We reserved something on the buffer */ /*不可能会出现write > BUF_PAGE_SIZE的情况*/ if (RB_WARN_ON(cpu_buffer, write > BUF_PAGE_SIZE)) return NULL; /*取出并更新event*/ event = __rb_page_index(tail_page, tail); rb_update_event(event, type, length); /* * If this is a commit and the tail is zero, then update * this page's time stamp. */ /* 如果当前是一个新页面,而且是一个fist commit. * 则更新Struct buffer_data_page -> time_stamp,即该页面开始写时的时间戳 */ if (!tail && rb_is_commit(cpu_buffer, event)) cpu_buffer->commit_page->page->time_stamp = *ts; return event; out_unlock: /* reset write */ if (tail <= BUF_PAGE_SIZE) local_set(&tail_page->write, tail); __raw_spin_unlock(&cpu_buffer->lock); local_irq_restore(flags); return NULL; } 剩下的代码就简单了,运行到这里,表明当前页面有足够的空间容纳要分配的event, 直接取得tail对应的空间即可. 在这里需要注意的是,如果是第一次从这个页面分配空间且处于first commit的状态,需要将页面的时间戳更改成当前的时间戳. 4.2: ring_buffer_unlock_commit()分析在前面的分析中, ring_buffer_lock_reserve()从RB中取出了空间,可以调用rb_event_data()返回event中实际存放数据的位置,将数据写入event之后,我们就需要进行commit了.这个动作就是在ring_buffer_unlock_commit()中完成的,代码如下: int ring_buffer_unlock_commit(struct ring_buffer *buffer, struct ring_buffer_event *event, unsigned long flags) { struct ring_buffer_per_cpu *cpu_buffer; int cpu = raw_smp_processor_id(); cpu_buffer = buffer->buffers[cpu]; rb_commit(cpu_buffer, event); /* * Only the last preempt count needs to restore preemption. */ if (preempt_count() == 1) ftrace_preempt_enable(per_cpu(rb_need_resched, cpu)); else preempt_enable_no_resched_notrace(); return 0; } Ftrace中的抢占恢复我们在前面分析ring_buffer_lock_reserve()的时候就已经分析过了,这里就不再重复,然后调用rb_commit()进行具体的commit动作,包括更新cpu_buffer的write_tamp,将commit迁移到write位置,这个函数比较简单,这里就不做详细分析了. 五: ring_buffer的读操作 RB的读操作没有写操作那么复杂,具体的读操作有两种方式,一种是迭代器的读,另一种采用reader_page进行切换读,下面分别对这两种方式进行详细的分析. 5.1: 迭代器方式的读操作这种读操作从它的名称上就可以看出来,它就是遍历每一个commit页面,然后将页数中的event读出来,这种读方式不会更改RB中的数据,下面看一下具体的实现: 5.1.1: ring_buffer_read_start() ring_buffer_read_start()用来初始化一个读ring buffer的迭代器,代码如下: struct ring_buffer_iter * ring_buffer_read_start(struct ring_buffer *buffer, int cpu) { struct ring_buffer_per_cpu *cpu_buffer; struct ring_buffer_iter *iter; unsigned long flags; /*如果该CPU不是ring buffer的有效CPU,非法*/ if (!cpumask_test_cpu(cpu, buffer->cpumask)) return NULL; /*分配一个迭代器*/ iter = kmalloc(sizeof(*iter), GFP_KERNEL); if (!iter) return NULL; /* 取得对应cpu的ring_buffer_per_cpu */ cpu_buffer = buffer->buffers[cpu]; /*迭代器的cpu_buffer指向对应CPU的ring_buffer_per_cpu*/ iter->cpu_buffer = cpu_buffer; /*禁止该buffer的写操作*/ atomic_inc(&cpu_buffer->record_disabled); /*等待所有的写操作退出*/ synchronize_sched(); /* 读操作加锁, 因为可以从不同的CPU上读因此 * 需要持有cpu_buffer->reader_lock *另外,为了避免对ring_buffer_per_cpu的竞争操作,需要 *持有cpu_buffer->lock */ spin_lock_irqsave(&cpu_buffer->reader_lock, flags); __raw_spin_lock(&cpu_buffer->lock); /*初始化迭代器,即将iter->head_page指向读页面 *iter->head指向读页面的读取位置*/ rb_iter_reset(iter); __raw_spin_unlock(&cpu_buffer->lock); spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); return iter; } 该操作比较简单,就是分配并初始化了一个ring_buffer_iter, 我们来看一下具体的初始化过程: static void rb_iter_reset(struct ring_buffer_iter *iter) { struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; /* Iterator usage is expected to have record disabled */ /*如果reader_page->list里空的,就将读的起始页面 *指向head_page *否则指向reader_page */ if (list_empty(&cpu_buffer->reader_page->list)) { iter->head_page = cpu_buffer->head_page; iter->head = cpu_buffer->head_page->read; } else { iter->head_page = cpu_buffer->reader_page; iter->head = cpu_buffer->reader_page->read; } /*如果该页面已经操作了, 取cpu_buffer->read_stamp, *否则取page->timer_stamp */ if (iter->head) iter->read_stamp = cpu_buffer->read_stamp; else iter->read_stamp = iter->head_page->page->time_stamp; } 从我们上面分析的ring_buffer初始化过程中看到,reader_page是一个单独的面面,且它的链表初始化是为空的,如果没有对reader_page进行特殊操作的话,那就是从head_page开始读. 那这个read_page怎么用呢? 它是用来做reader方式的读操作的,我们在后面再来进行分析. 再来看一下时间戳的信息,如果页面没有被读过,那就将read_stamp置为页面的时间戳.在上面已经分析过,页面的时间戳,是将页面切换成写页面时的时间戳(或者是页面第一次写时的时间戳),那就是这个页面的时间起始点. 那cpu_buffer->read_stamp是什么意思呢? 先将它放一边,接下来继续看它的读操作. 5.1.2: ring_buffer_iter_peek() 在上面初始化了一个迭代器后,现在就要开始真正的读操作了,代码如下: struct ring_buffer_event * ring_buffer_iter_peek(struct ring_buffer_iter *iter, u64 *ts) { struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; struct ring_buffer_event *event; unsigned long flags; again: spin_lock_irqsave(&cpu_buffer->reader_lock, flags); event = rb_iter_peek(iter, ts); spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); /*这个event是填充数据,比如一个页面已经容不下 *一个event了,这时,它的剩余空间就是RINGBUF_TYPE_PADDING *类型 *遇到这样的情况,继续读下一个即可 */ if (event && event->type == RINGBUF_TYPE_PADDING) { cpu_relax(); goto again; } return event; } 一眼就可以看出,它的实际操作是在rb_iter_peek()中完成的,代码如下: static struct ring_buffer_event * rb_iter_peek(struct ring_buffer_iter *iter, u64 *ts) { struct ring_buffer *buffer; struct ring_buffer_per_cpu *cpu_buffer; struct ring_buffer_event *event; int nr_loops = 0; /*如果CPU buffer的数据已经读完了*/ if (ring_buffer_iter_empty(iter)) return NULL; cpu_buffer = iter->cpu_buffer; buffer = cpu_buffer->buffer; again: /* * We repeat when a timestamp is encountered. It is possible * to get multiple timestamps from an interrupt entering just * as one timestamp is about to be written. The max times * that this can happen is the number of nested interrupts we * can have. Nesting 10 deep of interrupts is clearly * an anomaly. */ /*如果重试次数超过了10次,那表示ring buffer中有太多的 * 的附加数据(比如忽略的数据,时间戳数据,等) */ if (RB_WARN_ON(cpu_buffer, ++nr_loops > 10)) return NULL; /* 如果该CPU buffer已经空了*/ if (rb_per_cpu_empty(cpu_buffer)) return NULL; /*取iter->head_page对应的第一个event*/ event = rb_iter_head_event(iter); switch (event->type) { /*如果是一段填充数据*/ case RINGBUF_TYPE_PADDING: /*是一段完全的空闲整充区*/ if (rb_null_event(event)) { rb_inc_iter(iter); goto again; } /*是被忽略的内容*/ rb_advance_iter(iter); return event; /*如果附加的是一段时间戳*/ case RINGBUF_TYPE_TIME_EXTEND: /* Internal data, OK to advance */ rb_advance_iter(iter); goto again; case RINGBUF_TYPE_TIME_STAMP: /* FIXME: not implemented */ rb_advance_iter(iter); goto again; /*取出来的是一段数据*/ case RINGBUF_TYPE_DATA: /*取到数据了,更新时间戳之后退出*/ if (ts) { *ts = iter->read_stamp + event->time_delta; ring_buffer_normalize_time_stamp(buffer, cpu_buffer->cpu, ts); } return event; default: BUG(); } return NULL; } 这段代码比较简单,就是从ring buffer中取数据,然后更新时间戳.在这里我们需要注意一种RINGBUF_TYPE_PADDING类型的特例,如下所示: case RINGBUF_TYPE_PADDING: /*是一段完全的空闲整充区*/ if (rb_null_event(event)) { rb_inc_iter(iter); goto again; } /*是被忽略的内容*/ rb_advance_iter(iter); return event; 什么叫rb_null_event呢,代码中的判断是这样的: static inline int rb_null_event(struct ring_buffer_event *event) { return event->type == RINGBUF_TYPE_PADDING && event->time_delta == 0; } 可以得到,类型是RINGBUF_TYPE_PADDING,时间戳间隔是0, 这样的情况通常是填充页面的空闲部份. 比如一个页面不够放一个event了,就将该页面的剩余部份置为null event,然后将event存入下一个页面中. RINGBUF_TYPE_PADDING还有一种类型是rb_discarded_event, 代码中的判断如下: static inline int rb_discarded_event(struct ring_buffer_event *event) { return event->type == RINGBUF_TYPE_PADDING && event->time_delta; } 它跟null event的差别是时间戳不为空. 这样的情况经常是,用户不想显示ring buffer中的对应event,就将其设为这种类型, (比如event trace中的filter), 它的设置接口是ring_buffer_event_discard(),如下所示: void ring_buffer_event_discard(struct ring_buffer_event *event) { event->type = RINGBUF_TYPE_PADDING; /* time delta must be non zero */ if (!event->time_delta) event->time_delta = 1; } 回到上面的rb_iter_peek()中,这个函数里面有个重要的子函数rb_advance_iter(),代码如下: static void rb_advance_iter(struct ring_buffer_iter *iter) { struct ring_buffer *buffer; struct ring_buffer_per_cpu *cpu_buffer; struct ring_buffer_event *event; unsigned length; cpu_buffer = iter->cpu_buffer; buffer = cpu_buffer->buffer; /* * Check if we are at the end of the buffer. */ /*检查该页面是否已经读完了*/ if (iter->head >= rb_page_size(iter->head_page)) { /*前面的if加上这里的RB_WARN_ON()表示读时不可能 *会超过commit_page的范围, commit_page的提交序号最多 *只会到page_size的位置 *在正常情况下,RB中没数据了,就不会进入到ring_buffer_iter_peek(): *在rb_iter_peek()刚开始的判断中就会被返回 */ if (RB_WARN_ON(buffer, iter->head_page == cpu_buffer->commit_page)) return; /*必须要前进一个页面*/ rb_inc_iter(iter); return; } /*取得当前的event*/ event = rb_iter_head_event(iter); /*计算event的长度*/ length = rb_event_length(event); /* * This should not be called to advance the header if we are * at the tail of the buffer. */ /*如果超过了提交的范围, 这是不可能存在的*/ if (RB_WARN_ON(cpu_buffer, (iter->head_page == cpu_buffer->commit_page) && (iter->head + length > rb_commit_index(cpu_buffer)))) return; /*更新iter的时间戳*/ rb_update_iter_read_stamp(iter, event); iter->head += length; /* check for end of page padding */ /*如果该页面已经读完了而且没有超过commit page *再将iter前进 */ if ((iter->head >= rb_page_size(iter->head_page)) && (iter->head_page != cpu_buffer->commit_page)) rb_advance_iter(iter); } 该接口用来在struct ring_buffer_per_cpu中前进一个event,它有三种可能的情况: 1: 如果该页面已经处理完了,那就转入下一个页面 2: 如果页面已经读完了(读位置等于提交位置),不需要进行任何操作了,返回. 3: 将页面的读位置更新到下一个event的位置,然后更新迭代器的时间戳,返回另外,如果下一个event落到了下一个页面中,那就再调用一下本函数,那迁移到一个页面,如下代码所示: if ((iter->head >= rb_page_size(iter->head_page)) && (iter->head_page != cpu_buffer->commit_page)) rb_advance_iter(iter) 5.1.3: ring_buffer_read_finish 取完缓存区中的数据之后,还需要做清理的工作,这是在ring_buffer_read_finish()中完成的,代码如下: void ring_buffer_read_finish(struct ring_buffer_iter *iter) { struct ring_buffer_per_cpu *cpu_buffer = iter->cpu_buffer; /*恢复ring_buffer_per_cpu的使用*/ atomic_dec(&cpu_buffer->record_disabled); /*释放迭代器占用的空间*/ kfree(iter); } 现在迭代器方式的读操作分析完了,我们来总结一下读操作的同步机制: 1):在读操作开始时,就会禁用该CPU上的RB写 2):在整个读操作时,持有cpu_buffer->reader_lock,且禁中断为什么需要这样做呢? 这个问题值得好好挖掘一下: 1): 为什么在读的时候要禁止该CPU上的写? 在write的时候,如果缓存区满了,会清空head,然后将head转向下一个位置. 假若进行这个操作的时候,Read正在读head这个页面,那读操作就紊乱了. 有人可能会说,那在write的时候,如果要清空head就持有reader_lock锁不就行了么? 这样当然是可以的,只是相于来说比较繁锁 2): 为什么要持有reader_lock锁呢? 对于不同的迭代器读操作来说,它们是没有竞争的,因为它们操作的是同一个迭代器,这里持有reader_lock锁主要是为了跟reader方式的读操作保持同步,因为在reader方式下,会更改head页面,这些操作我们在稍后的分析中会看到. 5.2: reader方式的读操作我们在上面的分析中多次提到了reader方式的读操作,这种方式要使用struct ring_buffer_per_cpu中的reader_page成员它的接口为rb_buffer_peek(), 代码如下: struct ring_buffer_event * ring_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts) { struct ring_buffer_per_cpu *cpu_buffer = buffer->buffers[cpu]; struct ring_buffer_event *event; unsigned long flags; /*如果ring buffer中不含此CPU,退出*/ if (!cpumask_test_cpu(cpu, buffer->cpumask)) return NULL; again: /*加锁,从ring buffer中取数据,然后解锁*/ spin_lock_irqsave(&cpu_buffer->reader_lock, flags); event = rb_buffer_peek(buffer, cpu, ts); spin_unlock_irqrestore(&cpu_buffer->reader_lock, flags); /*如果取到的数据是填充数据,再取*/ if (event && event->type == RINGBUF_TYPE_PADDING) { cpu_relax(); goto again; } return event; } 这段代码很简单,首先为了保持Read操作的同步,持有reader_lock锁,然后调用rb_buffer_peek()到RB中取数据,如果取出的数据是填充数据,则再取一次. 核心操作是在rb_buffer_peek()中完成的,代码如下: static struct ring_buffer_event * rb_buffer_peek(struct ring_buffer *buffer, int cpu, u64 *ts) { struct ring_buffer_per_cpu *cpu_buffer; struct ring_buffer_event *event; struct buffer_page *reader; int nr_loops = 0; /* 取CPU对应的ring_buffer_per_cpu */ cpu_buffer = buffer->buffers[cpu]; again: /* * We repeat when a timestamp is encountered. It is possible * to get multiple timestamps from an interrupt entering just * as one timestamp is about to be written. The max times * that this can happen is the number of nested interrupts we * can have. Nesting 10 deep of interrupts is clearly * an anomaly. */ /*如果重复次数超过了10, 说明ring buffer中存放了太多的 *无用数据*/ if (RB_WARN_ON(cpu_buffer, ++nr_loops > 10)) return NULL; /*取出当前的reader页面*/ reader = rb_get_reader_page(cpu_buffer); if (!reader) return NULL; /*从reader页面中取数据*/ event = rb_reader_event(cpu_buffer); /*下面的逻辑跟iter方式的一样,只是这里记录时候戳采用的 *是cpu_buffer->read_stamp */ switch (event->type) { case RINGBUF_TYPE_PADDING: if (rb_null_event(event)) RB_WARN_ON(cpu_buffer, 1); /* * Because the writer could be discarding every * event it creates (which would probably be bad) * if we were to go back to "again" then we may never * catch up, and will trigger the warn on, or lock * the box. Return the padding, and we will release * the current locks, and try again. */ rb_advance_reader(cpu_buffer); return event; case RINGBUF_TYPE_TIME_EXTEND: /* Internal data, OK to advance */ rb_advance_reader(cpu_buffer); goto again; case RINGBUF_TYPE_TIME_STAMP: /* FIXME: not implemented */ rb_advance_reader(cpu_buffer); goto again; case RINGBUF_TYPE_DATA: if (ts) { *ts = cpu_buffer->read_stamp + event->time_delta; ring_buffer_normalize_time_stamp(buffer, cpu_buffer->cpu, ts); } return event; default: BUG(); } return NULL; } 这里的代码逻辑跟iter方式的读操作很相似,结合代码中的注释应该很容易理解这段代码,我们来看一下里面的几个重要的操作: static struct buffer_page * rb_get_reader_page(struct ring_buffer_per_cpu *cpu_buffer) { struct buffer_page *reader = NULL; unsigned long flags; int nr_loops = 0; /*因为写操作会更改RB的页面,所以这里必须要跟 *写操作保持同步,即持有cpu_buffer->lock锁 */ local_irq_save(flags); __raw_spin_lock(&cpu_buffer->lock); again: /* * This should normally only loop twice. But because the * start of the reader inserts an empty page, it causes * a case where we will loop three times. There should be no * reason to loop four times (that I know of). */ if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) { reader = NULL; goto out; } reader = cpu_buffer->reader_page; /* If there's more to read, return this page */ /*如果该页面中还有数据,返回该页面*/ if (cpu_buffer->reader_page->read < rb_page_size(reader)) goto out; /* Never should we have an index greater than the size */ /*读位置超过了commit 位置,这是不可能出现的*/ if (RB_WARN_ON(cpu_buffer, cpu_buffer->reader_page->read > rb_page_size(reader))) goto out; /* check if we caught up to the tail */ reader = NULL; /*读页面就是提交页面, 而且读页面中没有数据可读了, *说明缓存区中已经没有数据可读了 */ if (cpu_buffer->commit_page == cpu_buffer->reader_page) goto out; /* * Splice the empty reader page into the list around the head. * Reset the reader page to size zero. */ /*cpu_buffer->reader_page中的数据已经全部都读完了,将它和 *cpu_buffer->head_page调换一下位置 */ reader = cpu_buffer->head_page; cpu_buffer->reader_page->list.next = reader->list.next; cpu_buffer->reader_page->list.prev = reader->list.prev; local_set(&cpu_buffer->reader_page->write, 0); local_set(&cpu_buffer->reader_page->page->commit, 0); /* Make the reader page now replace the head */ reader->list.prev->next = &cpu_buffer->reader_page->list; reader->list.next->prev = &cpu_buffer->reader_page->list; /* * If the tail is on the reader, then we must set the head * to the inserted page, otherwise we set it one before. */ cpu_buffer->head_page = cpu_buffer->reader_page; /*因为切换进来的reader_page是可写的,因此,在不越过commit_page的 *情况下,head_page前进一个页面 */ if (cpu_buffer->commit_page != reader) rb_inc_page(cpu_buffer, &cpu_buffer->head_page); /* Finally update the reader page to the new head */ cpu_buffer->reader_page = reader; rb_reset_reader_page(cpu_buffer); goto again; out: __raw_spin_unlock(&cpu_buffer->lock); local_irq_restore(flags); return reader; } 关于reader操作的原理可以看一下ring_buffer.c中自带的注释: /* * The ring buffer is made up of a list of pages. A separate list of pages is * allocated for each CPU. A writer may only write to a buffer that is * associated with the CPU it is currently executing on. A reader may read * from any per cpu buffer. * * The reader is special. For each per cpu buffer, the reader has its own * reader page. When a reader has read the entire reader page, this reader * page is swapped with another page in the ring buffer. * * Now, as long as the writer is off the reader page, the reader can do what * ever it wants with that page. The writer will never write to that page * again (as long as it is out of the ring buffer). * * Here's some silly ASCII art. * * +------+ * |reader| RING BUFFER * |page | * +------+ +---+ +---+ +---+ * | |-->| |-->| | * +---+ +---+ +---+ * ^ | * | | * +---------------+ * * * +------+ * |reader| RING BUFFER * |page |------------------v * +------+ +---+ +---+ +---+ * | |-->| |-->| | * +---+ +---+ +---+ * ^ | * | | * +---------------+ * * * +------+ * |reader| RING BUFFER * |page |------------------v * +------+ +---+ +---+ +---+ * ^ | |-->| |-->| | * | +---+ +---+ +---+ * | | * | | * +------------------------------+ * * * +------+ * |buffer| RING BUFFER * |page |------------------v * +------+ +---+ +---+ +---+ * ^ | | | |-->| | * | New +---+ +---+ +---+ * | Reader------^ | * | page | * +------------------------------+ * * * After we make this swap, the reader can hand this page off to the splice * code and be done with it. It can even allocate a new page if it needs to * and swap that into the ring buffer. * * We will be using cmpxchg soon to make all this lockless. * */ 其实,它就是利用reader_page来替换了当前head_page页,这样就不会影响到写操作. 在这里,我们需要注意的是,经过上面的操作之后, 有可能tail_page和reader_page在同一个页面上,这种情况会在RB开始写的时候,这时候的tail_page, head_page都是重合在一起的,我们可以根据上面的代码逻辑推理一下,假设在刚开始的状态,写了一个不足一个页面的数据,reader开始读,reader_page替换掉head_page之后,页面会成为这个样子: 500)this.width=500;" border=0> 那么,在这时候,tail_page commit_page和reader_page是重合在一起的. 在上面的这种情况下,因为切换出来的head_page,也就是new_reader_page没有更改它的页面前向指针与后向指针,因此,tail_page在前进的时候,会前进到head_page的下一个页面.但是在这种情况下,要注意head_page是没有数据的,所以,在iter读方式下,如果reader_page是被切换出来的head_page(也就是代码中说的,reader_page的链表不为空),读页面是从reader_page开始的. 另外,还需要注意一点的是,代码中一个比较诡异的注释: /* * This should normally only loop twice. But because the * start of the reader inserts an empty page, it causes * a case where we will loop three times. There should be no * reason to loop four times (that I know of). */ if (RB_WARN_ON(cpu_buffer, ++nr_loops > 3)) { reader = NULL; goto out; } 代码注释说,通常情况下, nr_loops会等于2,在开始读的时候,因为reader_page会插入一个空页面,造成该值等于3的情况. 两次循环的情况我们都知道: 刚开始进这个函数,nr_loops加1,假设reader_page中没有数据可读,切换进head_page,然后goto到起点,nr_loops变为2. 那如果nr_loops变为3的话,必须是切换进来的head_page为空,并且RB中有数据. 什么时候会满足这样的情况呢? 其实,这种情况,只需要在上一次reader操作时,head_page和commit_page重合,造成head_page不能前进,然后写操作继续进行,写下一个页面了.此时,再有reader去读的时候,会有head_page是空的,但它后面还有数据的情况. 另外,我们这里还需要注意一下这个问题,如下代码所示: switch (event->type) { /*如果是一段填充数据*/ case RINGBUF_TYPE_PADDING: /*是一段完全的空闲整充区*/ if (rb_null_event(event)) { rb_inc_iter(iter); goto again; } /*是被忽略的内容*/ rb_advance_iter(iter); return event; 为什么这里如果取出来的是填充数据,会被BUG_ON()呢,而在iter中,却没有呢? 其实,我们注意看一下代码,在rb_reader_event()中取页面的时候,如果发现read==commit,也就是取到填充位置,就会切换下一个页面进来,因此,在rb_reader_event()不可能会返回一个没有数据的页面. 另外,从代码中看来,reader读操作和写操作是可以同时进行的,因为reader不会修改页面的数据,但是write在写的时候就必须要跟reader保持同步了,因为reader会切换页面,write也会清空header页面,所以这时需要持有自旋锁保护. 从上面的代码中也可以看出,reader方式的读会依次清空RB中的数据页. 六: 小结 RB是一种高效的缓冲区操作,在理解代码的时候,需要考虑到各种临界条件,另外,在阅读代码的过程中,不是很简洁,很多函数都是的功能极其相似.