Linux内核跟踪之ring buffer的实现.doc

Linux内核跟踪之ring buffer的实现一: 前言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的数据结构框架如下所示:2010-8-13 17:29 上传下载附件 (35.65 KB) 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 CON*_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 CON*_HOTPLUG_CPUget_online_cpus();cpumask_copy(buffer-cpumask, cpu_online_mask);#elsecpumask_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-bufferscpu =rb_allocate_cpu_buffer(buffer, cpu);if (!buffer-bufferscpu)goto fail_free_buffers;#ifdef CON*_HOTPLUG_CPUbuffer-cpu_notify.notifier_call = rb_cpu_notify;buffer-cpu_notify.priority = 0;register_cpu_notifier(&buffer-cpu_notify);#endifput_online_cpus();mutex_init(&buffer-mutex);return buffer;fail_free_buffers:for_each_buffer_cpu(buffer, cpu) if (buffer-bufferscpu)rb_free_cpu_buffer(buffer-bufferscpu);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-bufferscpu =rb_allocate_cpu_buffer(buffer, cpu);if (!buffer-bufferscpu)goto fail_free_buffers;从此可以看到,它只为当时存在的CPU分配了缓存区.到这里,我们大概可以猜到怎么处理hotplug cpu的情况了: 在有CPU加入时,为这个CPU对应的 “框”对应分配内存,在CPU拨除或掉线的情况下,释放掉该CPU对应的内存. 到底是不是跟我们所想的一样呢? 我们继续看代码:#ifdef CON*_HOTPLUG_CPUbuffer-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-bufferscpu =rb_allocate_cpu_buffer(buffer, cpu);if (!buffer-bufferscpu) WARN(1, failed to allocate ring buffer on CPU %ldn,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 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 dont 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-bufferscpu;/* 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 dont 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 arry0 to store the data length */if (length RB_MAX_SMALL_DATA)length += sizeof(event.array0);/* 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只能表示从0xb00xb11100的长度,即 028的长度,那么,如果数据长度超过了28,那应该要怎么表示呢?在数据长度超过28的情况下,会使用ring_buffer_event中的arry0表示里面的数据长度,即从后面的数据部份取出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: 表示附加的时间差信息,这个信息会存放在arry0中.RINGBUF_TYPE_TIME_STAMP: 表示存放的是时间戳信息, array0用来存放tv_nsec, array1.2中存放 tv_sec.在现在的代码中还末用到.RINGBUF_TYPE_DATA:表示里面填充的数据,数据的长度表示方式在前面已经分析过了,这里就不再赘述了.好了,返回rb_calculate_event_length():RB_MAX_SMALL_DATA =28也就是我们上面分析的event中的最小长度,如果要存入的长度大于这个长度的,那么,就需要数据部份的一个32位数用来存放它的长度,因此这 种情况下,需要增加sizeof(event.array0)的长度.另外,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了.在这里要注意中断抢占的原则: 只有高优先的中断才能抢占低优先级的中断.也就是中断是不能