同步、异步、阻塞、非阻塞 I/O 及 异步I/O实现

structiocb{

void*data;//用于存储回调函数或者字符串

unsignedkey;

shortaio_lio_opcode;//操作码，是读还是写，或者其他

shortaio_reqprio;

intaio_fildes;//

union{

structio_iocb_commonc;

structio_iocb_vectorv;

structio_iocb_pollpoll;

structio_iocb_sockaddrsaddr;

}u;

};

inlinevoidio_prep_pread(structiocb*iocb,intfd,void*buf,size_tcount,longlongoffset);

初始化一个异步IO写请求：

inlinevoidio_prep_pwrite(structiocb*iocb,intfd,void*buf,size_tcount,longlongoffset);

对buf进行对齐：

intposix_memalign(void**memptr,size_talignment,size_tsize);

intio_getevents(aio_context_tctx_id,longmin_nr,longnr,structio_event*events,structtimespec*timeout);

min_nr:等待的最小的事件数

nr：等待的最大事件数

timeout：超时

io_event结构：

#definePADDEDptr(x,y)x;unsignedy

#definePADDEDul(x,y)unsignedlongx;unsignedy

structio_event{

PADDEDptr(void*data,__pad1);//data用于存放回调函数，如果无回调函数，该data也可以存放入字符串这样的信息

PADDEDptr(structiocb*obj,__pad2);//存放iocb对象

PADDEDul(res,__pad3);//实际读到的字符串长度，当所传递的fd是以O_DIRECT打开，但是，buf长度不是AIO_BLKSIZE的时候，该参数被设置为-22，errno为22时，其意义为“Invalidargument”

PADDEDul(res2,__pad4);//出错码，如果出错，则不为0；若未出错，则为0

};

注意的地方：

对于不满足规定的异步IO块大小的文本，使用没有用O_DIRECT打开的文件描述符进行操作

实例：

该程序实现了一个复制文件的功能，运行程序时需要传入两个参数，第一个是原文件名，第二个是目标文件名。

//#define__USE_GNU

#undef_FILE_OFFSET_BITS

#define_FILE_OFFSET_BITS64

#include<stdlib.h>

#include<stdio.h>

#include<unistd.h>

#include<dirent.h>

#include<sys/types.h>

#include<sys/stat.h>

#include<sys/param.h>

#include<fcntl.h>

#include<errno.h>

#include<libaio.h>

#defineAIO_BLKSIZE1024

#defineAIO_MAXIO64

staticintbusy=0;

staticinttocopy=0;

staticintsrcfd=-1;

staticintsrcfd2=-1;

staticintdstfd=-1;

staticintdstfd2=-1;

staticconstchar*dstname=NULL;

staticconstchar*srcname=NULL;

/*fatalerrorhandler*/

staticvoidio_error(constchar*func,intrc)

{

if(rc==-ENOSYS)

fprintf(stdout,"AIOnotinthiskernel/n");

else

fprintf(stdout,"%s:errno%d",func,rc);

if(dstfd>0)

close(dstfd);

if(dstname)

unlink(dstname);

exit(1);

}

/*

*writecompletecallback

*adjustcountsandfreeresources

*/

staticvoidwr_done(io_context_tctx,structiocb*iocb,longres,longres2)

{

if(res2!=0)

{

io_error("aiowrite",res2);

}

if(res!=iocb->u.c.nbytes)

{

fprintf(stdout,"writemissedbytesexpect%dgot%d",iocb->u.c.nbytes,res);

exit(1);

}

//printf("res=%d,res2=%d/n",res,res2);

--tocopy;

--busy;

free(iocb->u.c.buf);

free(iocb);

}

/*

*readcompletecallback

*changereadiocbintoawriteiocbandstartit.

*/

staticvoidrd_done(io_context_tctx,structiocb*iocb,longres,longres2)

{

/*libraryneedsaccessorstolookatiocb*/

intiosize=iocb->u.c.nbytes;

char*buf=(char*)iocb->u.c.buf;

off_toffset=iocb->u.c.offset;

intfd,tmp;

char*wrbuff=NULL;

if(res2!=0)

{

io_error("aioread",res2);

}

if(res!=iosize)

{

fprintf(stdout,"readmissingbytesexpect%dgot%d",iocb->u.c.nbytes,res);

exit(1);

}

if(iocb->aio_fildes==srcfd)

{

fd=dstfd;

}

else

fd=dstfd2;

/*turnreadintowrite*/

tmp=posix_memalign((void**)&wrbuff,getpagesize(),AIO_BLKSIZE);

if(tmp<0)

{

perror("posix_memalign");

exit(1);

}

snprintf(wrbuff,iosize+1,"%s",buf);

//printf("wrbuff-len=%d:%s/n",strlen(wrbuff),wrbuff);

//printf("wrbuff_len=%d/n",strlen(wrbuff));

free(buf);

io_prep_pwrite(iocb,fd,wrbuff,iosize,offset);

io_set_callback(iocb,wr_done);

if(1!=(res=io_submit(ctx,1,&iocb)))

io_error("io_submitwrite",res);

printf("/nsubmit%dwriterequest/n",res);

}

voidm_io_queue_run(io_context_tmyctx)

{

//printf("inm_io_queue_run:/n");

structio_eventevents[AIO_MAXIO];

io_callback_tcb;

intn,i;

n=io_getevents(myctx,1,AIO_MAXIO,events,NULL);

printf("/n%dio_requestcompleted/n/n",n);

for(i=0;i<n;i++)

{

cb=(io_callback_t)events[i].data;

structiocb*io=events[i].obj;

//useepoll_fdtoregisterEPOLLOUT

//heresegmentfault

//printf("events[%d].data=%x,res=%d,res2=%d/n",i,cb,events[i].res,events[i].res2);

cb(myctx,io,events[i].res,events[i].res2);

}

}

intmain(intargc,char*const*argv)

{

structstatst;

//off_tlength=0,offset=0;

longlonglength=0,offset=0;

io_context_tmyctx;

char*buff=NULL;

intleftover;

inttmp;

intlast;

if(argc!=3||argv[1][0]=='-')

{

fprintf(stdout,"Usage:aiocpSOURCEDEST/n");

exit(1);

}

if((srcfd=open(srcname=argv[1],O_RDONLY|O_DIRECT))<0)

{

perror(srcname);

exit(1);

}

if(fstat(srcfd,&st)<0)

{

perror("fstat");

exit(1);

}

length=(longlong)st.st_size;

if((dstfd=open(dstname=argv[2],O_WRONLY|O_CREAT|O_DIRECT,0666))<0)

{

close(srcfd);

perror(dstname);

exit(1);

}

leftover=length%AIO_BLKSIZE;

length-=leftover;

/*initializestatmachine*/

memset(&myctx,0,sizeof(myctx));

io_queue_init(AIO_MAXIO,&myctx);

/*insys/param.h#definehowmany(x,y)(((x)+((y)-1))/(y))*/

tocopy=howmany(length,AIO_BLKSIZE);

while(tocopy>0)

{

inti,rc;

/*submitasmanyreadsasonceaspossibleuptoAIO_MAXIO

*insys/param#defineMIN(a,b)(((a)<(b))?(a):(b))

*/

intn=MIN(MIN(AIO_MAXIO-busy,AIO_MAXIO/2),howmany(length-offset,AIO_BLKSIZE));

if(n>0)

{

structiocb*ioq
;

for(i=0;i<n;i++)

{

structiocb*io=(structiocb*)malloc(sizeof(structiocb));

intiosize=AIO_BLKSIZE;

tmp=posix_memalign((void**)&buff,getpagesize(),AIO_BLKSIZE);

if(tmp<0)

{

perror("posix_memalign");

exit(1);

}

if(NULL==io)

{

fprintf(stdout,"outofmemeory/n");

exit(1);

}

io_prep_pread(io,srcfd,buff,iosize,offset);

io_set_callback(io,rd_done);

ioq[i]=io;

offset+=iosize;

buff=NULL;

}

printf("START.../n/n");

rc=io_submit(myctx,n,ioq);

if(rc<0)

io_error("io_submit",rc);

printf("/nsubmit%dreadrequest/n",rc);

busy+=n;

}

//HandleIO'sthathavecompleted

m_io_queue_run(myctx);

//ifwehavemaximumnumberofi/o'sinflight

//thenwaitforonetocomplete

/*if(busy==AIO_MAXIO)

{

rc=io_queue_wait(myctx,NULL);

if(rc<0)

{

io_error("io_queue_wait",rc);

}

}

*/

}

if(leftover)

{

srcfd2=open(srcname,O_RDONLY);

if(srcfd2<0)

{

perror("srcfd2open");

exit(1);

}

dstfd2=open(dstname,O_WRONLY);

if(dstfd2<0)

{

perror("dstfd2open");

exit(1);

}

structiocb*io=(structiocb*)malloc(sizeof(structiocb));

if(io==NULL)

{

perror("mallocfailed");

exit(1);

}

buff=NULL;

tmp=posix_memalign((void**)&buff,getpagesize(),AIO_BLKSIZE);

if(tmp<0)

{

perror("posix_memalign");

exit(1);

}

io_prep_pread(io,srcfd2,buff,leftover,offset);

io_set_callback(io,rd_done);

if((last=io_submit(myctx,1,&io))<0)

{

perror("io_submit");

exit(1);

}

printf("/nsubmit%diorequest/n",last);

m_io_queue_run(myctx);

}

exit(0);

}

同步阻塞IO

在这个模型中，应用程序（application）为了执行这个read操作，会调用相应的一个systemcall，将系统控制权交给kernel，然后就进行等待（这其实就是被阻塞了）。kernel开始执行这个systemcall，执行完毕后会向应用程序返回响应，应用程序得到响应后，就不再阻塞，并进行后面的工作。