Linux内核开发之异步通知与异步I/O《来自linux设备开发详解》

structxxx_dev
{
structcdevcdev;
...
structfasync_struct*async_queue;//异步结构体
}

staticintxxx_fasync(intfd,structfile*filp,intmode)
{
structxxx_dev*dev=filp->private_data;
returnfasync_helper(fd,filp,mode,&dev->async_queue);
}
一旦设备资源可以获得时,应该调用kill_fasync()释放SIGIO信号,可读时第三个参数设置为POLL_IN,可写时第三个参数设置为POLL_OUT,下边是释放信号的典型模版：

staticssize_txxx_write(structfile*filp,constchar__user*buf,size_tcount,loff_t*f_ops)
{
structxxx_dev*dev=filp->private_data;
....
//产生异步信号
if(dev->async_queue)
{
kill_fasync(&dev->async_queue,SIGIO,POLL_IN);
}
..
}

staticintxxx_release(structinode*inode,structfile*filp)
{
structxxx_dev*dev=filp->private_data;
//将文件从异步通知列表中删除
xxx_fasync(-1,filp,0);
...
return0;
}

structglobalfifo_dev
{
structcdevcdev;/*cdev结构体*/
unsignedintcurrent_len;/*fifo有效数据长度*/
unsignedcharmem[GLOBALFIFO_SIZE];/*全局内存*/
structsemaphoresem;/*并发控制用的信号量*/
wait_queue_head_tr_wait;/*阻塞读用的等待队列头*/
wait_queue_head_tw_wait;/*阻塞写用的等待队列头*/
structfasync_struct*async_queue;/*异步结构体指针，用于读*/
};
/*文件释放函数*/
intglobalfifo_release(structinode*inode,structfile*filp)
{
/*将文件从异步通知列表中删除*/
globalmem_fasync(-1,filp,0);
return0;
}

staticintglobalfifo_fasync(intfd,structfile*filp,intmode)
{
structglobalfifo_dev*dev=filp->private_data;
returnfasync_helper(fd,filp,mode,&dev->async_queue);
}

/*globalfifo写操作*/
staticssize_tglobalfifo_write(structfile*filp,constchar__user*buf,
size_tcount,loff_t*ppos)
{
structglobalfifo_dev*dev=filp->private_data;//获得设备结构体指针
intret;
DECLARE_WAITQUEUE(wait,current);//定义等待队列

down(&dev->sem);//获取信号量
add_wait_queue(&dev->w_wait,&wait);//进入写等待队列头

/*等待FIFO非满*/
if(dev->current_len==GLOBALFIFO_SIZE)
{
if(filp->f_flags&O_NONBLOCK)
//如果是非阻塞访问
{
ret=-EAGAIN;
gotoout;
}
__set_current_state(TASK_INTERRUPTIBLE);//改变进程状态为睡眠
up(&dev->sem);

schedule();//调度其他进程执行
if(signal_pending(current))
//如果是因为信号唤醒
{
ret=-ERESTARTSYS;
gotoout2;
}

down(&dev->sem);//获得信号量
}

/*从用户空间拷贝到内核空间*/
if(count>GLOBALFIFO_SIZE-dev->current_len)
count=GLOBALFIFO_SIZE-dev->current_len;

if(copy_from_user(dev->mem+dev->current_len,buf,count))
{
ret=-EFAULT;
gotoout;
}
else
{
dev->current_len+=count;
printk(KERN_INFO"written%dbytes(s),current_len:%d\n",count,dev
->current_len);

wake_up_interruptible(&dev->r_wait);//唤醒读等待队列
/*产生异步读信号*/
if(dev->async_queue)
kill_fasync(&dev->async_queue,SIGIO,POLL_IN);

ret=count;
}

out:up(&dev->sem);//释放信号量
out2:remove_wait_queue(&dev->w_wait,&wait);//从附属的等待队列头移除
set_current_state(TASK_RUNNING);
returnret;
}

下面再给出测试程序：

#include...
//接收到异步读信号的动作
voidinput_handler(intsignum)
{
printf("Receiveasignalfromglobalfifo,signalnum:%d\n",signum);
}

intmain()
{
intfd,oflags;
fd=open("/dev/globalfifo",O_RDWR,S_IRUSR|S_IWUSR);
if(fd!=-1)
{
//启动信号驱动机制
signal(SIGIO,input_handler);//让input_handler()处理SIGIO信号
fcntl(fd,F_SETOWN,getpid());
oflags=fcntl(fd,F_GETFL);
fcntl(fd,F_SETFL,oflags|FASYNC);
while(1)
{
sleep(100);
}
}
else
{
printf("deviceopenfailure\n");
}
}

structaiocb{
int		aio_fildes;	/*Filedescriptor*/
off_t		aio_offset;	/*Fileoffset*/
volatilevoid*	aio_buf;	/*Locationofbuffer*/
size_t		aio_nbytes;	/*Lengthoftransfer*/
int		aio_reqprio;	/*Requestpriorityoffset*/
structsigevent	aio_sigevent;	/*Signalnumberandvalue*/
int		aio_lio_opcode;	/*Operationtobeperformed*/
};

#include<aio.h>
..
intfd,set;
structaiocbmy_aiocb;

fd=open("file.txt",O_RDONLY);
if(fd<0)
{
perror("open");
}
//清零aiocb结构体
bzero((char*)&my_aiocb,sizeof(structaiocb));
//为aiocb请求分配数据缓冲区
my_aiocb.aio_buf=malloc(BUFSIZE+1);
if(!my_aiocb.aio_buf)
perror("malloc");
//初始化aiocb的成员
my_aiocb.aio_fildes=fd;
my_aiocb.aio_nbytes=BUFSIZE;
my_aiocb.aio_offset=0;

ret=aio_read(&my_aiocb);
if(ret<0)
perror("aio_read");
while(aio_error(&my_aiocb)==EINPROGRESS)
;
if((ret=aio_return(&my_iocb)))
{
//获得异步读的返回值
}
else
{
读失败，分析errror
}

structaioct*cblist(MAX_LIST)
//清零aioct结构链表
bzero((char*)cblist,sizeof(cblist));
//将一个或更多的aiocb放入aioct结构体链表
cblist[0]=&my_aiocb;
ret=aio_read(&my_aiocb);
ret=aio_suspend(cblist,MAX_LIST,NULL);

structaiocbaiocb1,aiocb2;
structaiocb*list[MAX_LIST];
...
//准备第一个aiocb
aiocb1.aio_fildes=fd;
aiocb1.aio_buf=malloc(BUFSIZE+1);
aiocb1.aio_nbytes=BUFSIZE;
aiocb1.aio_offset=next_offset;
aiocb1.aio_lio_opcode=LIO_READ;//异步读操作
...//准备多个aiocb
bzero((char*)list,sizeof(list));
//将aiocb填入链表
list[0]=&aiocb1;
list[1]=&aiocb2;
...
ret=lio_listio(LIO_WAIT,list,MAX_LIST,NULL);//发起大量IO操作

voidsetup_io(..)
{
intfd;
structsigactionsig_act;
structaiocbmy_aiocb;
...
//设置信号处理函数
sigemptyset(&sig_act.sa_mask);
sig_act.sa_flags=SA_SIGINFO;
sig_act.sa_sigaction=aio_completion_handler;
//设置AIO请求
bzero((char*)&my_aiocb,sizeof(structaiocb));
my_aiocb.aio_flags=fd;
my_aiocb.aio_buf=malloc(BUF_SIZE+1);
my_aiocb.aio_nbytes=BUF_SIZE;
my_aiocb.offset=next_offset;
//连接AIO请求和信号处理函数
my_aiocb.aio_sigevent.sigev_notify=SIGEV_SIGNVAL;
my_aiocb.aio_sigevent.sigev_signo=SIGIO;
my_aiocb.aic_sigevent.sigev_value.sival_ptr=&my_aiocb;
//将信号和处理函数绑定
ret=sigaction(SIGION,&sig_act,NULL);
...
ret=aio_read(&my_aiocb);
｝
//信号处理函数
voidaio_completion_handler(intsigno,siginfo_t*info,void*context)
{
structaiocb*req;
//确定是我们需要的信号
if(info->si_signo==SIGIO)
{
req=(structaiocb*)info->si_value.sival_ptr;//获得aiocb;
//请求的操作是否完成
if(aio_error(req)==0)
{
ret=aio_return(req);
}
}
return;
}

voidsetup_io(..)
{
...//同上
//连接AIO请求和线程回调函数
my_aiocb.aio_sigevent.sigev_notify=SIGEV_THREAD;
my_aiocb.aio_sigevent.notify_function=aio_completion_handler;

//设置回调函数
my_aiocb.aio_sigevent.notify_attributes=NULL;
my_aiocb.aio_sigevent.sigev_value.sival_ptr=&my_aiocb;
...
ret=aio_read(&my_aiocb);
｝
//信号处理函数
voidaio_completion_handler(intsigno,siginfo_t*info,void*context)
{
structaiocb*req;
req=(structaiocb*)sigval.sival_ptr;//获得aiocb;
//请求的操作是否完成
if(aio_error(req)==0)
{
ret=aio_return(req);
}
return;
}

ssize_t(*aio_read)(structkiocb*iocb,char*buffer,size_tcount,loff_toffset);
ssize_t(*aio_write)(structkiocb*iocb,constchar*buffer,size_tcount,loff_toffset);
int(*aio_fsync)(structkiocb*iocb,intdatasync);

aio_read()和aio_write()与file_operation中的read()和write()中的offset参数不同，它直接传递值，而后者传递的是指针。这两个函数本身也不一定完成读写操作，它只是发起，初始化读写操作。

下面来看看实际的代码部分：

//异步读
staticssize_txxx_aio_read(structkiocb*iocb,char*buffer,size_tcount,loff_toffset)
{
returnxxx_defer_op(0,iocb,buf,count,pos);
}

//异步写
staticssize_txxx_aio_write(structkiocb*iocb,constchar*buffer,size_tcount,loff_toffset)
{
returnxxx_defer_op(1,iocb,(char*)buf,count,pos);
}

//初始化异步IO
staticintxxx_defer_op(intwrite,structkiocb*iocb,char*buf,size_tcount,loff_tpos)
{
structasync_work*async_wk;
intresult;
//当可以访问buffer时进行复制
if(write)
{
result=xxx_write(iocb->ki_filp,buf,count,&pos);
}
else
{
result=xxx_read(iocb->ki_filp,buf,count,&pos);
}
//如果是同步IOCB,立即返回状态
if(is_sync_kiocb(iocb))
returnresutl;
//否则，推后几us执行
async_wk=kmalloc(sizeof(*async_wk),GFP_KERNEL));
if(async_wk==NULL)
returnresult;
async_wk->aiocb=iocb;
async_wk->result=result;
INIT_WORK(&async_wk->work,xxx_do_deferred_op,async_wk);
schedule_delayed_work(&async_wk->work,HZ/100);
return-EIOCBOUEUED;//控制权限返回给用户空间
}
//延迟后执行
staticvoidxxx_do_deferred_op(void*p)
{
structasync_work*async_wk=(structasync_work*)p;
aio_complete(async_wk_iocb,async_wk->result,0);
kfree(async_wk);
}

structasync_work
{
structkiocb*iocb;//kiocb结构体指针
intresult;//执行结果
structwork_structwork;//工作结构体
};

在上边代码中最核心的是使用aync_work结构体将操作延迟，通过schedule_delayed_work可以调度其运行，而aio_complete的调用用于通知内核驱动程序已经完成了操作。