内核sk_buff工作线程总结

对skb采用异步序列操作的一个总结, 写成一系列函数. 将skb出队列作为一个线程调用, skb处理则采用回调函数, 在线程里直接调用, 提高了通用性. 也可以将一个skb_task定义为PER_CPU变量, 达到并行处理目的. :)

#include <linux/kthread.h>

struct skb_task{
	struct task_struct   *task;
	struct completion    start_done;
	struct completion    end_done;
	volatile u32		 stop;
	u32                  state;
	struct sk_buff_head	 pkt_list;
	
	int			(*skb_callback)(struct sk_buff*);
};

enum {
	SKB_TASK_STOP,
	SKB_TASK_IDLE,
	SKB_TASK_RUNNING,
};
/*
** skb_task工作线程
*/
static int _skb_task_worker(void *arg)
{
	struct sk_buff *skb = NULL;
	struct skb_task *task = (struct skb_task*)arg;
	
	if ( NULL == task ) {
		return 0;
	}
	pr_emerg( "%s begin.\n", __func__ );
	
	set_current_state(TASK_INTERRUPTIBLE);	
	complete( &task->start_done );
	
	while( task->stop == 0 ) {
		
		task->state = SKB_TASK_IDLE;
		
		//schedule_timeout_interruptible(2);
		schedule();
		
		__set_current_state(TASK_RUNNING);
		
		task->state = SKB_TASK_RUNNING;
		do {
#if 1			
			spin_lock(&task->pkt_list.lock);
			skb = __skb_dequeue(&task->pkt_list);
			spin_unlock(&task->pkt_list.lock);
#else	
			skb = skb_dequeue( &task->pkt_list );
#endif			
			if ( NULL != skb ) {
				// 调用skb回调函数
				if ( NULL == task->skb_callback ) {
					kfree_skb( skb );
				}
				else {
					task->skb_callback( skb );
				}
			}
			
			if ( task->stop != 0 ) {			
				goto _out;
			}
		}while( NULL != skb );
	};
	
_out:
	pr_emerg( "%s end.\n", __func__ );
	task->state = SKB_TASK_STOP;
	complete( &task->end_done );
	return 0;
}
/*
** 初始化skb_task, callback为skb处理回调函数
*/
int _skb_task_init(struct skb_task *task, 
		int (*callback)(struct sk_buff*) )
{
	skb_queue_head_init( &task->pkt_list );
	init_completion( &task->start_done );
	init_completion( &task->end_done );
	
	task->skb_callback = callback;
	task->state = 0x0;
	task->stop = 0x0;

	return 0;
} 
/*
** 创建skb_task线程
*/
int _skb_task_create(struct skb_task *task)
{
	int    rc = 0;
	
	task->task = kthread_create( _skb_task_worker, task, "skb_task");
	if ( IS_ERR( task->task ) ) {
		task->task = NULL;
		dbg_err( "kthread_create occur error.\n" );
		rc = -1;
	}
	else {
		dbg_info( "kthread_create success.\n" );
		wake_up_process( task->task );
		wait_for_completion( &task->start_done );
		dbg_info( "skb task worker run success.\n" );
	}	
	return rc;
}
/*
** 终止skb_task线程
*/
void _skb_task_stop(struct skb_task *task)
{
	if ( NULL == task->task ){
		return ;
	}
	pr_emerg( "wait for skb task done\n" );
	task->stop = 0x1;
	
	if (task->state == SKB_TASK_IDLE ) {
		if ( !task_is_stopped(task->task)
			&& TASK_DEAD != task->task->state ) {
			wake_up_process(task->task);	
		}
	}
	wait_for_completion( &task->end_done );	
	
	printk( "taask->state = %d\n", task->task->state );
}
/*
** 清除释放skb_task所属skb队列
*/
void _skb_task_clean(struct skb_task *task)
{
	struct sk_buff *skb = NULL;
	
	pr_emerg( "clean skb task queue\n" );
	
	do {
		skb = __skb_dequeue( &task->pkt_list );
		if ( NULL != skb ) {
			kfree_skb( skb );
		}
	}while( NULL != skb );
}
/*
** 将skb加入skb_task所属skb队列
*/
int _skb_task_enqueue(struct skb_task *task, 
		struct sk_buff *skb )
{
#if 1	
	spin_lock(&task->pkt_list.lock);
	__skb_queue_tail(&task->pkt_list, skb );
	spin_unlock(&task->pkt_list.lock);
#else
	skb_queue_tail(&task->pkt_list, skb );
#endif	

	if ( NULL == task->task ){
		return -1;
	}
	if (task->state == SKB_TASK_IDLE ) {
		wake_up_process(task->task);	
	}
	return 0;
}
/*****************************************************************************/
// skb local_out 输出示例

static struct skb_task  skb_local_out;

int local_out_init(void)
{
	_skb_task_init( &skb_local_out, ip_local_out );
	_skb_task_create( &skb_local_out );
}

void local_out_fini(void)
{
	_skb_task_stop( &skb_local_out );
	_skb_task_clean( &skb_local_out );	
}

int local_out_enqueue( struct sk_buff *skb )
{
	return _skb_task_enqueue( &skb_local_out, skb );
}