Linux kernel aio funcionalidad

Question

Estoy probando kernel asynchronous io functions (no posix aio) y estoy tratando de descubrir cómo funciona. El siguiente código es un programa completo donde simplemente escribo una matriz repetidamente en un archivo abierto usando O_DIRECT. Aparece un error en la función de devolución de llamada: "escribir bytes perdidos esperar 1024 obtuvo 0" (ver la declaración fprintf en work_done()).

Para aquellos que no están familiarizados con AIO núcleo, el código de abajo hace lo siguiente:

1. Init algunas estructuras
2. Preparar AIO (io_prep_pwrite)
3. Presentar solicitudes io (io_submit)
4. Compruebe si hay finalización de evento (io_getevents)
5. Llamar 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 con O_DIRECT, las cosas funcionan bien, pero es mejor que tener las escrituras asincrónicas. ¿Puede alguien decirme qué estoy haciendo mal? ¿Es este el uso correcto de kernel aio, por ejemplo, mi uso de devoluciones de llamada es correcto? ¿Hay alguna restricción en el uso de O_DIRECT?

puedo compilar usando 'gcc -Wall test.c -laio'

Gracias de antemano.

/* 
* 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; 
} 

Answer 1

En primer lugar, buen trabajo utilizando libaio en lugar de POSIX aio.

¿Existen restricciones en el uso de O_DIRECT?

No estoy 100% seguro de que este es el verdadero problema, pero O_DIRECT tiene algunos requisitos (citando a su mayoría de TLPI):

• El búfer de datos que se transfieren deben estar alineados en una memoria límite que es un múltiplo del tamaño de bloque (utilizar posix_memalign)
• el archivo o dispositivo de desplazamiento en la que la transferencia de datos comienza debe ser un múltiplo del tamaño de bloque
• la longitud de los datos a transferir debe ser un múltiplo del tamaño de bloque

De un vistazo, veo que no se están tomando precauciones para alinear la memoria en allocate_2D_matrix.

Si no abro el archivo con O_DIRECT, las cosas funcionan bien, pero es mejor que la propósito de tener asíncrono escribe.

Esto no es el caso. La E/S asincrónica funciona bien sin O_DIRECT (por ejemplo, piense en la cantidad de llamadas al sistema recortadas).