aio 运行在内核态的版本

简介：

      上边的文章以aio系列开头的函数都是在用户态运行的程序，是由glibc提供的，linux kernel也提供了一系列在内核态运行的程序，我们知道调用内核函数用的syscall，但是glibc并没有提供这几个调用的封装(wrapper)， 提供封装的是libaio。在运行之前，我们首先安装libaio，通过
       apt-get  install libaio1                        这个我感觉没什么用
       apt-get  install libaio-dev                 这个是安装libaio的开发文件的

描述： 

        首先我们从结构体上来看数据：
        首先是主结构体。
            struct iocb{                  
                   PADDEDptr(void *data, __pad1);                            /*保存的数据用来通过event返回，这里PADDEDptr会根据主机的字节序和
                                                                                                         为了字节对齐填充字段*/
                  PADDED(unsigned key, __pad2);                           //内核会设置aio_key
                   short  aio_lio_opcode                                            //操作的类型 IOCB_CMD_系列,下面介绍
                   short  aio_reqprio;                                                 //操作的优先级
                   int      aio_fieldes;                                                  //文件描述符
                   union {
                        struct   io_iocb_common    c;                        //暂时我们只会用到这个，对文件进行操作
                        struct   io_iocb_vector         v;
                        struct   io_iocb_poll             poll;
                        struct   io_iocb_sockaddr  saddr;
                   }u;                      
               };     
          iocb结构里面包含一个union，里面有三个结构体，暂时我们只关心io_iocb_common结构体，他的定义是：
             struct io_iocb_common{
                    PADDEDptr(void* buf, __pad1);                                        //保存数据的指针
                    PADDEDul(nbytes,   __pad2);                                           //数据指针的大小
                    long long                      offset;                                           //数据文件内的偏移量
                    long long                    __pad3;
                    unsigned                    flags;                                              //
                    unsigned                    resfd;                                             //如果flags设置了IOCB_FLAG_RESFD,这个字段保存eventfd。
              };        
          下面是当操作完成的通知事件的结构体：
              struct io_event{
                       PADDEDptr(void* data, __pad1);                                                  //iocb 的aio_data字段
                       PADDEDptr(struct iocb* obj, __pad2);                                           //iocb
                       PADDEDul(res,  __pad3);                                                               //操作完成的返回值
                       PADDEDul(res2, __pad4);                                                             //额外的返回值
                };
           下面是可以操作的类型
               typedef enum io_iocb_cmd{
                           IO_CMD_PREAD = 0;                                   //读操作
                           IO_CMD_PWRITE = 1;                                  //写操作
                         
                          IO_CMD_FSYNC = 2;                                     //串行话操作，把在iocb.fiedes文件上的操作，串行运行
                          IO_CMD_FDSYNC = 3;                                  
                         
                          IO_CMD_POLL = 5;
                          IO_CMD_NOOP = 6;
                          IO_CMD_PREADV = 7;                                    
                          IO_CMD_PWRITEV = 8；
                 };
      其余的结构体大家可以通过察看/usr/include/libaio.h文件得到。
      下面是相关的函数：libaio封装的系统调用：
      int io_setup(unsigned nr_events, io_context_t * ctxp);
           该函数初始化io_context_t结构，其中调用之前ctxp必须初始化为0，nr_event表示能够ctxp的容量。
      int io_submit(io_context_t ctx, long nr, struct iocb *iocb[]);
           该函数把nr个iocb放到ctx对应的执行队列中之后返回。
      int io_cancel(io_context_t ctx, struct iocb* iocb);
           该函数尝试取消之前通过io_submit传入的iocb。
      int io_getevents(io_context_t ctx,long min_nr, long nr, struct io_event* events[], struct timespec * timeout);
            该函数尝试从完成队列中去的至少min_nr个，至多nr个的消息。我们通过检查io_event的信息可以做我们想做的事情。

     下面的函数是libaio方便我们操作，提供的封装函数：具体代码参照后边的附录。
     static inline void io_set_callback(struct iocb *iocb, io_callback_t cb)；       //把cb复制到iocb的data数据中去。
     static inline void io_prep_pread(struct iocb *iocb, int fd, void *buf, size_t count, long long offset)；
     static inline void io_prep_pwrite(struct iocb *iocb, int fd, void *buf, size_t count, long long offset)；
     static inline void io_prep_preadv(struct iocb *iocb, int fd, const struct iovec *iov, int iovcnt, long long offset)；
     static inline void io_prep_pwritev(struct iocb *iocb, int fd, const struct iovec *iov, int iovcnt, long long offset)；
     static inline void io_prep_poll(struct iocb *iocb, int fd, int events)；
     static inline int io_poll(io_context_t ctx, struct iocb *iocb, io_callback_t cb, int fd, int events)
     static inline void io_prep_fsync(struct iocb *iocb, int fd)；
     static inline int io_fsync(io_context_t ctx, struct iocb *iocb, io_callback_t cb, int fd)；
     static inline void io_prep_fdsync(struct iocb *iocb, int fd)；
      static inline int io_fdsync(io_context_t ctx, struct iocb *iocb, io_callback_t cb, int fd)；
     static inline void io_set_eventfd(struct iocb *iocb, int eventfd)；
        其实带有prep只是帮助我们填充了结构体而已。其余的是帮助你先调用pre函数再调用io_submit说明一下io_set_eventfd可以填充一个eventfd，可以通过该结构和epoll联系起来。
   额外的libaio还提供了几个io系列的封装:详细看附录
  如下
    int io_queue_init(int maxevents, io_context_t *ctxp)；   相当于调用io_setup
    int io_queue_run(io_context_t ctx);                                     调用io_getevents 然后调用回调函数  

    int io_queue_wait(io_context_t ctx, struct timespec *timeout); 
    int io_queue_release(io_context_t ctx);                              调用io_destroy

例子：

     我只是写了一个简单的例子只是使用了系统的api。test.c
 

#include<libaio.h>
#include<stdio.h>
#include<fcntl.h>
#include<string.h>
#include<inttypes.h>
int main()
{
    io_context_t ctx;
    struct iocb cb;
    struct iocb * cbs[1];
    char data[4096];
    struct io_event events[1];
    int ret;
    int fd;
    fd = open("./testfile", O_RDWR|O_CREAT);
    if(fd < 0)
    {
        perror("open error");
        return -1;
    }

    ctx = 0;
    ret = io_setup(128, &ctx);    //设置能够容纳128的context
    if(ret < 0)
    {
        perror("io_setup error");
        return -1;
    }
    memset(&cb,0,sizeof(cb));
    cb.aio_fildes = fd;
    cb.aio_lio_opcode = IO_CMD_PWRITE;   //写操作
    cb.u.c.buf = (void*)data;
    cb.u.c.offset = 0;
    cb.u.c.nbytes = 4096;
    cbs[0] = &cb;
    ret =io_submit(ctx, 1, cbs);              // 提交该操作
    if(ret != 1)
    {
        if(ret < 0)
            perror("io_submit error");
          perror("io_submit error");
        else
            fprintf(stderr, "could not submit");
        return -1;
    }

    ret=io_getevents(ctx, 1,1,events,NULL);   //获取完成状态，没有之i那个time，阻塞直到完成。
    printf("%d\n",ret);
    ret = io_destroy(ctx);
    if(ret < 0)
    {perror("io_destroy error");
        return -1;}
    return 0;
}

     通过静态连接的话 通过gcc libaio.a test.c就可以。

附录：

//通过调用这个函数，当我们通过调用io_getevents的获取event的时候可以去的data结构，回调该函数。
static inline void io_set_callback(struct iocb *iocb, io_callback_t cb)
{
        iocb->data = (void *)cb;
}
//读操作
static inline void io_prep_pread(struct iocb *iocb, int fd, void *buf, size_t count, long long offset)
{
        memset(iocb, 0, sizeof(*iocb));
        iocb->aio_fildes = fd;
        iocb->aio_lio_opcode = IO_CMD_PREAD;
        iocb->aio_reqprio = 0;
        iocb->u.c.buf = buf;
        iocb->u.c.nbytes = count;
        iocb->u.c.offset = offset;
}
static inline void io_prep_pwrite(struct iocb *iocb, int fd, void *buf, size_t count, long long offset)
{
        memset(iocb, 0, sizeof(*iocb));
        iocb->aio_fildes = fd;
        iocb->aio_lio_opcode = IO_CMD_PWRITE;
        iocb->aio_reqprio = 0;
        iocb->u.c.buf = buf;
        iocb->u.c.nbytes = count;
        iocb->u.c.offset = offset;
}

static inline void io_prep_preadv(struct iocb *iocb, int fd, const struct iovec *iov, int iovcnt, long long offset)
{
        memset(iocb, 0, sizeof(*iocb));
        iocb->aio_fildes = fd;
        iocb->aio_lio_opcode = IO_CMD_PREADV;
        iocb->aio_reqprio = 0;
        iocb->u.c.buf = (void *)iov;
        iocb->u.c.nbytes = iovcnt;
        iocb->u.c.offset = offset;
}

static inline void io_prep_pwritev(struct iocb *iocb, int fd, const struct iovec *iov, int iovcnt, long long offset)
{
        memset(iocb, 0, sizeof(*iocb));
        iocb->aio_fildes = fd;
        iocb->aio_lio_opcode = IO_CMD_PWRITEV;
        iocb->aio_reqprio = 0;
        iocb->u.c.buf = (void *)iov;
        iocb->u.c.nbytes = iovcnt;
        iocb->u.c.offset = offset;
}
/* Jeff Moyer says this was implemented in Red Hat AS2.1 and RHEL3.
 * AFAICT, it was never in mainline, and should not be used. --RR */
static inline void io_prep_poll(struct iocb *iocb, int fd, int events)
{
        memset(iocb, 0, sizeof(*iocb));
        iocb->aio_fildes = fd;
        iocb->aio_lio_opcode = IO_CMD_POLL;
        iocb->aio_reqprio = 0;
        iocb->u.poll.events = events;
}

static inline int io_poll(io_context_t ctx, struct iocb *iocb, io_callback_t cb, int fd, int events)
{
        io_prep_poll(iocb, fd, events);
        io_set_callback(iocb, cb);
        return io_submit(ctx, 1, &iocb);
}

static inline void io_prep_fsync(struct iocb *iocb, int fd)
{
        memset(iocb, 0, sizeof(*iocb));
        iocb->aio_fildes = fd;
        iocb->aio_lio_opcode = IO_CMD_FSYNC;
        iocb->aio_reqprio = 0;
}
static inline int io_fsync(io_context_t ctx, struct iocb *iocb, io_callback_t cb, int fd)
{
        io_prep_fsync(iocb, fd);
        io_set_callback(iocb, cb);
        return io_submit(ctx, 1, &iocb);
}

static inline void io_prep_fdsync(struct iocb *iocb, int fd)
{
        memset(iocb, 0, sizeof(*iocb));
        iocb->aio_fildes = fd;
        iocb->aio_lio_opcode = IO_CMD_FDSYNC;
        iocb->aio_reqprio = 0;
}

static inline int io_fdsync(io_context_t ctx, struct iocb *iocb, io_callback_t cb, int fd)
{
        io_prep_fdsync(iocb, fd);
        io_set_callback(iocb, cb);
        return io_submit(ctx, 1, &iocb);
}

static inline void io_set_eventfd(struct iocb *iocb, int eventfd)
{
        iocb->u.c.flags |= (1 << 0) /* IOCB_FLAG_RESFD */;
        iocb->u.c.resfd = eventfd;
}

  queue系列函数：

int io_queue_init(int maxevents, io_context_t *ctxp)
{
    if (maxevents > 0) {
        *ctxp = NULL;
        return io_setup(maxevents, ctxp);
    }
    return -EINVAL;
}
int io_queue_run(io_context_t ctx)
{
    static struct timespec timeout = { 0, 0 };
    struct io_event event;
    int ret;

    /* FIXME: batch requests? */
    while (1 == (ret = io_getevents(ctx, 0, 1, &event, &timeout))) {
        io_callback_t cb = (io_callback_t)event.data;
        struct iocb *iocb = event.obj;

        cb(ctx, iocb, event.res, event.res2);
    }

    return ret;
}
//稍微做了下修改
int io_queue_wait(io_context_t ctx, struct timespec *timeout)
{
    return io_getevents(ctx, 0, 0, NULL, timeout);
}
int io_queue_release(io_context_t ctx)
{
    return io_destroy(ctx);
}