aio_queue.c

#define _POSIX_C_SOURCE 200809L
#include <stdio.h>
#include <stdlib.h>
#include <aio.h>
#include <unistd.h>
#include <fcntl.h>
#include <errno.h>
#include <string.h>

#define NREQ 4
#define BUFSZ 4096

int main(int argc, char **argv)
{
    if (argc < 2) {
        fprintf(stderr, "usage: %s file\n", argv[0]);
        return 1;
    }
    const char *path = argv[1];
    int fd = open(path, O_RDONLY);
    if (fd < 0) {
        perror("open");
        return 1;
    }
    struct aiocb cbs[NREQ];
    char bufs[NREQ][BUFSZ];
    off_t offset = 0;
    int inflight = 0;
    for (int i = 0; i < NREQ; i++) {
        memset(&cbs[i], 0, sizeof(struct aiocb));
        cbs[i].aio_fildes = fd;
        cbs[i].aio_buf = bufs[i];
        cbs[i].aio_nbytes = BUFSZ;
        cbs[i].aio_offset = offset;
        if (aio_read(&cbs[i]) < 0) {
            perror("aio_read");
            close(fd);
            return 1;
        }
        inflight++;
        offset += BUFSZ;
    }
    while (inflight > 0) {
        const struct aiocb *list[NREQ];
        for (int i = 0; i < NREQ; i++) {
            list[i] = &cbs[i];
        }
        int r = aio_suspend(list, NREQ, NULL);
        if (r < 0 && errno != EINTR) {
            perror("aio_suspend");
            break;
        }
        for (int i = 0; i < NREQ; i++) {
            int err = aio_error(&cbs[i]);
            if (err == EINPROGRESS) continue;
            if (err != 0 && err != ECANCELED) {
                fprintf(stderr, "aio_error: %s\n", strerror(err));
                inflight--;
                cbs[i].aio_fildes = -1;
                continue;
            }
            ssize_t n = aio_return(&cbs[i]);
            if (n <= 0) {
                inflight--;
                cbs[i].aio_fildes = -1;
                continue;
            }
            write(STDOUT_FILENO, bufs[i], n);
            cbs[i].aio_offset = offset;
            cbs[i].aio_nbytes = BUFSZ;
            if (aio_read(&cbs[i]) < 0) {
                inflight--;
                cbs[i].aio_fildes = -1;
            } else {
                offset += BUFSZ;
            }
        }
    }
    close(fd);
    return 0;
}