linux kernel aio funcionalidad

Estoy probando funciones asincrónicas de kernel io (no posix aio) y estoy tratando de averiguar cómo funciona. El siguiente código es un programa completo donde simplemente escribo una matriz repetidamente en un archivo abierto usando O_DIRECT. Obtengo un error en la función de devolución de llamada "escribir bytes perdidos esperan 1024 consiguió 0" (vea la instrucción fprintf en work_done ()).

Para aquellos que no están familiarizados con kernel aio, el siguiente código hace lo siguiente:

Iniciar algunas estructurasPrepare aio (io_prep_pwrite)Enviar solicitudes de io (io_submit)Compruebe la finalización del evento (io_getevents) Llame a una función de devolución de llamada para ver si todo salió bien.

Recibo un error en el paso 5. Si no abro el archivo usando O_DIRECT, las cosas funcionan bien, pero supera el propósito de tener escrituras asíncronas. ¿Alguien puede decirme qué estoy haciendo mal? ¿Es este el uso correcto de kernel aio, por ejemplo, es correcto mi uso de devoluciones de llamada? ¿Hay alguna restricción en el uso de O_DIRECT?

Compilo usando 'gcc -Wall test.c -laio'

Gracias por adelantado

/* 
 * File:   myaiocp.c
 * Author: kmehta
 *
 * Created on July 11, 2011, 12:50 PM
 *
 *
 * Testing kernel aio. 
 * Program creates a 2D matrix and writes it multiple times to create a file of desired size. 
 * Writes are performed using kernel aio functions (io_prep_pwrite, io_submit, etc.)
 */
#define _GNU_SOURCE
#define _XOPEN_SOURCE 600

#include <stdio.h>
#include <stdlib.h>
#include <getopt.h>
#include <pthread.h>
#include <fcntl.h>
#include <string.h>
#include <sys/uio.h>
#include <sys/time.h>
#include <omp.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <errno.h>
#include <libaio.h>

char ** buf;
long seg_size;
int seg_rows;
double total_size;
char * filename;
static int wait_count = 0;

void io_task();
void cleanup();
void allocate_2D_matrix(int[]);
int file_open(char *);
void wr_done(io_context_t ctx, struct iocb* iocb, long res, long res2);

int main(int argc, char **argv) {
    total_size  = 1048576;      //1MB
    seg_size    = 1024;         //1kB
    seg_rows    = 1024;
    filename    = "aio.out";

    int dims[] = {seg_rows, seg_size};
    allocate_2D_matrix(dims);   //Creates 2D matrix

    io_task();
    cleanup();

    return 0;
}

/*
 * Create a 2D matrix
 */
void allocate_2D_matrix(int dims[2]) {
    int i;
    char *data;

    //create the matrix
    data = (char *) calloc(1, dims[0] * dims[1] * sizeof (char));
    if (data == NULL) {
        printf("\nCould not allocate memory for matrix.\n");
        exit(1);
    }

    buf = (char **) malloc(dims[0] * sizeof (char *));
    if (buf == NULL) {
        printf("\nCould not allocate memory for matrix.\n");
        exit(1);
    }

    for (i = 0; i < dims[0]; i++) {
        buf[i] = &(data[i * dims[1]]);
    }
}

static void io_error(const char *func, int rc)
{
    if (rc == -ENOSYS)
        fprintf(stderr, "AIO not in this kernel\n");
    else if (rc < 0)
        fprintf(stderr, "%s: %s\n", func, strerror(-rc));
    else
        fprintf(stderr, "%s: error %d\n", func, rc);

    exit(1);
}

/*
 * Callback function
 */
static void work_done(io_context_t ctx, struct iocb *iocb, long res, long res2)
{

    if (res2 != 0) {
        io_error("aio write", res2);
      }

      if (res != iocb->u.c.nbytes) {
            fprintf(stderr, "write missed bytes expect %lu got %ld\n",
                  iocb->u.c.nbytes, res2);
            exit(1);
      }
      wait_count --;
      printf("%d ", wait_count);
}

/*
 * Wait routine. Get events and call the callback function work_done()
 */
int io_wait_run(io_context_t ctx, long iter)
{
      struct io_event events[iter];
      struct io_event *ep;
      int ret, n;

      /*
       * get up to aio_maxio events at a time.
       */
      ret = n = io_getevents(ctx, iter, iter, events, NULL);
      printf("got %d events\n", n);
      /*
       * Call the callback functions for each event.
       */
      for (ep = events ; n-- > 0 ; ep++) {
            io_callback_t cb = (io_callback_t)ep->data ; struct iocb *iocb = ep->obj ; cb(ctx, iocb, ep->res, ep->res2);
      }
      return ret;
}

void io_task() {
    long offset = 0;
    int bufIndex = 0;

    //Open file
    int fd = file_open(filename);

    //Initialize structures
    long i; 
    long iter = total_size / seg_size;  //No. of iterations to reach desired file size (total_size)
    io_context_t myctx;
    if(0 != io_queue_init(iter, &myctx))
    {
        perror("Could not initialize io queue");
        exit(EXIT_FAILURE);
    }
    struct iocb * ioq[iter];

    //loop through iter times to reach desired file size
    for (i = 0; i < iter; i++) {
        struct iocb *io = (struct iocb*) malloc(sizeof (struct iocb));
        io_prep_pwrite(io, fd, buf[bufIndex], seg_size, offset);
        io_set_callback(io, work_done);
        ioq[i] = io;

        offset += seg_size;
        bufIndex ++;
        if (bufIndex > seg_rows - 1)    //If entire matrix written, start again from index 0
            bufIndex = 0;
    }

    printf("done preparing. Now submitting..\n");
    if(iter != io_submit(myctx, iter, ioq))
    {
        perror("Failure on submit");
        exit(EXIT_FAILURE);
    }

    printf("now awaiting completion..\n");
    wait_count = iter;
    int res;

    while (wait_count) {
        res = io_wait_run(myctx, iter);
        if (res < 0)
            io_error("io_wait_run", res);
    }

    close(fd);
}

void cleanup() {
    free(buf[0]);
    free(buf);
}

int file_open(char *filename) {
    int fd;
    if (-1 == (fd = open(filename, O_DIRECT | O_CREAT | O_WRONLY | O_TRUNC, 0666))) {
        printf("\nError opening file. \n");
        exit(-1);
    }

    return fd;
}