Kernel的环形Buffer(Ring Buffer)——以Logger Buffer为例_kernel的环形buffer-以logger buffer

在Android L之前的日志系统是Kernel层实现了若干个环形Buffer实现的。系统各个日志读写操作都是针对这几个RingBuffer来实现的。那就来一窥Kernel是怎么做的。相关源码是位于driver/staging/android/下面的logger.c和logger.h两个文件

1，在整个Android日志系统的位置

2，在logger.c中，入口函数

 
 
static int __init logger_init(void)
{
...
	ret = create_log(LOGGER_LOG_MAIN, 256*1024);
	ret = create_log(LOGGER_LOG_EVENTS, 256*1024);
	ret = create_log(LOGGER_LOG_RADIO, 256*1024);
	ret = create_log(LOGGER_LOG_SYSTEM, 256*1024);
...
}

分别来看create_log的实现和LOGGER_LOG_*的相关定义

 
 
static int __init create_log(char *log_name, int size)
{
...
log->buffer = buffer;
/* finally, initialize the misc device for this log */
	ret = misc_register(&log->misc);
...
}

 
 
#define LOGGER_LOG_RADIO	"log_radio"	/* radio-related messages */
#define LOGGER_LOG_EVENTS	"log_events"	/* system/hardware events */
#define LOGGER_LOG_SYSTEM	"log_system"	/* system/framework messages */
#define LOGGER_LOG_MAIN		"log_main"	/* everything else */
...

从上可以知道，create_log函数就是把创建4个misc设备，并注册上。在创建的过程中会初始化各个设备的logger_log结构体。也就是说有4个logger_log结构体的实例。这4个MISC设备（即对应4种Logger）的Buffer大小为256Kb，即RING BUFFER的大小为256Kb。

再来看这个十分重要的结构体logger_log的其它成员

 
 
/**
 * struct logger_log - represents a specific log, such as 'main' or 'radio'
 * @buffer:	The actual ring buffer
 * @misc:	The "misc" device representing the log
 * @wq:		The wait queue for @readers
 * @readers:	This log's readers
 * @mutex:	The mutex that protects the @buffer
 * @w_off:	The current write head offset
 * @head:	The head, or location that readers start reading at.
 * @size:	The size of the log
 * @logs:	The list of log channels
 *
 * This structure lives from module insertion until module removal, so it does
 * not need additional reference counting. The structure is protected by the
 * mutex 'mutex'.
 */
struct logger_log {
	unsigned char		*buffer;
	struct miscdevice	misc;
	wait_queue_head_t	wq;
	struct list_head	readers;
	struct mutex		mutex;
	size_t			w_off;
	size_t			head;
	size_t			size;
	struct list_head	logs;
};

不复杂，注释的解析也十分清楚。

用户层对于这4个MISC设备的操作，是通过/dev/log/*下面的几个设备实现的，它同样也有file_operations结构体中的读写函数指针，如下：

 
 
static const struct file_operations logger_fops = {
	.owner = THIS_MODULE,
	.read = logger_read,
	.aio_write = logger_aio_write,
	.poll = logger_poll,
	.unlocked_ioctl = logger_ioctl,
	.compat_ioctl = logger_ioctl,
	.open = logger_open,
	.release = logger_release,
};

3，MISC设备节点是怎么生成？

在运行的系统的/dev/log目录下有如下节点，

 
 
root@U:/dev/log # ls
events
main
radio
system

这是因为在上面提到的misc_register(&log->misc)函数，在前面的Kernel设备驱动相关文章可以知道，最终是调用device_create() -> device_add()->kobject_uevent(&dev->kobj, KOBJ_ADD);就会发送一个KOBJ_ADD的UEVENT给到上层，由《Android的根文件系统》可以知道，这是由init进程来处理的。在init项目中devices.cpp会来处理这个ADD事件

 
 
...  
  } else if(!strncmp(uevent->subsystem, "misc", 4) &&
                 !strncmp(name, "log_", 4)) {                                                                                                                                                
         INFO("kernel logger is deprecated\n");
         base = "/dev/log/";
         make_dir(base, 0755);
         name += 4;
...

如果设备的名字前缀是log_的话，就会创建/dev/log/目录。所以就出现dev/log/下面的4个文件。

4，写操作分析

写操作是写一条消息，封装在logger_entry为头的消息。一次写操作大概构成如下：

struct logger_entry | priority | tag | msg

比如logcat打印出来如下：

12-06 16:12:23.849  3700  3788 | D | WifiStateMachine | : handleMessage: X

logger_aio_write()函数如下：

 
 
static ssize_t logger_aio_write(struct kiocb *iocb, const struct iovec *iov,
			 unsigned long nr_segs, loff_t ppos)
{
//1，填充header头的结构体
	header.pid = current->tgid;
	header.tid = current->pid;
	header.sec = now.tv_sec;
	header.nsec = now.tv_nsec;
	header.euid = current_euid();
	header.len = min_t(size_t, iocb->ki_nbytes, LOGGER_ENTRY_MAX_PAYLOAD);
	header.hdr_size = sizeof(struct logger_entry);
	/* null writes succeed, return zero */
	if (unlikely(!header.len))
		return 0;
//如果写的一条信息，会覆盖正在进程读的信息，那么就丢弃这条读，把读的偏移指到下一条安全的信息。
	fix_up_readers(log, sizeof(struct logger_entry) + header.len);
//2，把header信息定写入
	do_write_log(log, &header, sizeof(struct logger_entry));
//3，把用户传过来的Message信息写入
	while (nr_segs-- > 0) {
		size_t len;
		ssize_t nr;
		/* figure out how much of this vector we can keep */
    //这里计算剩余的信息是否超过了MAX_PAYLOAD的长度，即4K，超过部分将会丢弃。
    //一条信息基本上也不会超过4K，基本是取实际信息长度
		len = min_t(size_t, iov->iov_len, header.len - ret);
		/* write out this segment's payload */
		nr = do_write_log_from_user(log, iov->iov_base, len);
		if (unlikely(nr < 0)) {
			log->w_off = orig;
			mutex_unlock(&log->mutex);
			return nr;
		}
		iov++;
		ret += nr;
	}
//4，唤醒读的队列
	/* wake up any blocked readers */
	wake_up_interruptible(&log->wq);
	return ret;
}

一条log日志的头部信息(12-06 16:12:23.849  3700  3788)是在kernel中写入的。do_write_log()

 
 
static void do_write_log(struct logger_log *log, const void *buf, size_t count)
{
	size_t len;
//Buffer剩余的长度和消息长度取较小的值
//如果buffer剩余长度不够写完一条信息，就进入下面的if，把剩余的部分，从头部开始写进去。
	len = min(count, log->size - log->w_off); 
	memcpy(log->buffer + log->w_off, buf, len);
//如果一条Log，不够剩余的buffer的话，上面memcpy就是写了前半部分，
//下面的memcpy就把剩下的后半部分（cout-len），写到log->buffer的头。
//这里体现了循环Buffer。
	if (count != len)
		memcpy(log->buffer, buf + len, count - len);
//移动w_off
	log->w_off = logger_offset(log, log->w_off + count);
}

写完头部，再写用户层传过来的信息

 
 
static ssize_t do_write_log_from_user(struct logger_log *log,
				      const void __user *buf, size_t count)
{
//同理会作判断，是否剩余buffer能够写完，否则就把剩余的部分从头开始写
	len = min(count, log->size - log->w_off);
	if (len && copy_from_user(log->buffer + log->w_off, buf, len))
		return -EFAULT;
	if (count != len)
		if (copy_from_user(log->buffer, buf + len, count - len))
	log->w_off = logger_offset(log, log->w_off + count);
	return count;
}

5，写操作基本上完成了，接下来看读操作。

 
 
static ssize_t logger_read(struct file *file, char __user *buf,
			   size_t count, loff_t *pos)
{
	//1，如果读的偏移r_off与写的偏移w_off相等，则循环阻塞，直到write最后唤醒
	while (1) {
		prepare_to_wait(&log->wq, &wait, TASK_INTERRUPTIBLE);//write函数的wake_up_interruptible(&log->wq);会唤醒
		//Write的off存在Logger_log即内存buffer中，而r_off存在reder，即读的进程中。所以执行两次不同的logcat，都是从头开始读的。
        ret = (log->w_off == reader->r_off);
		mutex_unlock(&log->mutex);
		if (!ret)
			break;//如果不相等就跳出去处理。
		schedule(); //把cpu调试出去，等待队列的唤醒
	}
	if (!reader->r_all)
		reader->r_off = get_next_entry_by_uid(log,
			reader->r_off, current_euid());
	/* get the size of the next entry */
	//2，获取要读的整个一条日志的长度，同样由两部分组成，一个head长度+信息的长度。
    ret = get_user_hdr_len(reader->r_ver) +
		get_entry_msg_len(log, reader->r_off);
	
	/* get exactly one entry from the log */
    //3，把log读ret长度的信息到buf中。
	ret = do_read_log_to_user(log, reader, buf, ret);
out:
	mutex_unlock(&log->mutex);
	return ret;
}

来看看第3步do_read_log_to_user的实现

 
 
static ssize_t do_read_log_to_user(struct logger_log *log,
				   struct logger_reader *reader,
				   char __user *buf,
				   size_t count)
{

    //先把header copy到用户空间
	entry = get_entry_header(log, reader->r_off, &scratch);
	if (copy_header_to_user(reader->r_ver, entry, buf))
		return -EFAULT;
	count -= get_user_hdr_len(reader->r_ver);
	buf += get_user_hdr_len(reader->r_ver);
	msg_start = logger_offset(log,
		reader->r_off + sizeof(struct logger_entry));
	/*
	 * We read from the msg in two disjoint operations. First, we read from
	 * the current msg head offset up to 'count' bytes or to the end of
	 * the log, whichever comes first.
	 */
//同样做是否到log buffer尾部的判断，否则就要读两次
	len = min(count, log->size - msg_start);
	if (copy_to_user(buf, log->buffer + msg_start, len))
		return -EFAULT;
...
	if (count != len)
		if (copy_to_user(buf + len, log->buffer, count - len))
			return -EFAULT;
//移动reader的off偏移
	reader->r_off = logger_offset(log, reader->r_off +
		sizeof(struct logger_entry) + count);
	return count + get_user_hdr_len(reader->r_ver);
}

这样Logger的实现基本就完成了，循环Buffer的实现，也就是在读与写的操作过程，是否已经到了buffer尾的判断。

 
 
len = min(count, log->size - msg_start);
	if (copy_to_user(buf, log->buffer + msg_start, len))
if (count != len)
		if (copy_to_user(buf + len, log->buffer, count - len))
			return -EFAULT;

RingBuffer是Kernel一个常用的数据结构，从Logger系统中看它的实现，看到其它模块使用，心里也有就数了。