directio.cpp

#include <iostream>
#include <cstdio>
#include <stdexcept>
#include <unistd.h>
#include <thread>
#include <vector>
#include <cstring>
#include <time.h>

#include <sys/fcntl.h>
#include <unistd.h>
#include <sys/uio.h>
#include <uv.h>

void on_open(uv_fs_t *req);
void on_read(uv_fs_t *req);
void on_write(uv_fs_t *req);
const int N_STREAMS=8;
uv_fs_t open_req;
uv_fs_t read_req;
uv_fs_t write_req;
uv_fs_t writes[N_STREAMS];
uv_fs_t close_req;

uv_buf_t buffer;

class PerfTimer
{
public:
    PerfTimer() {
        clock_gettime(CLOCK_MONOTONIC_RAW, &_start_time);
    }

    void restart() {
        clock_gettime(CLOCK_MONOTONIC_RAW, &_start_time);
    }

    double elapsed() const {
        timespec curr;
        clock_gettime(CLOCK_MONOTONIC_RAW, &curr);
        return double(curr.tv_sec - _start_time.tv_sec) +
               double(curr.tv_nsec - _start_time.tv_nsec)/1000000000.0;
    }
private:
    timespec _start_time;
};


namespace {
    const char* PATH = "/tmp/perftest.tmp";
    const size_t FSIZE = 4*1024*1024*1024l;
    const size_t BUFFER_SIZE = 4096;

    int create_file() {
        auto fd = open(PATH, O_DIRECT|O_RDWR|O_SYNC);
        if (fd == -1) {
            throw std::runtime_error(std::string("open O_DIRECT error: ") + strerror(errno));
        }
        return fd;
    }

    void truncate(int fd) {
        if (ftruncate(fd, FSIZE) == -1) {
            throw std::runtime_error(std::string("truncate error: ") + strerror(errno));
        }
    }

    void close_file(int fd) {
        if (close(fd) != 0) {
            throw std::runtime_error(std::string("close error: ") + strerror(errno));
        }
    }

    void init_file() {
        system("rm /tmp/perftest.tmp");
        int fd = open(PATH, O_CREAT|O_WRONLY|O_EXCL, S_IRUSR|S_IWUSR);
        if (fd == -1) {
            if (errno != EEXIST) {
            throw std::runtime_error(std::string("create error: ") + strerror(errno));
            }
        } else {
            truncate(fd);
            close_file(fd);
        }
    }

    void create_buf(uv_buf_t* iov, int n) {
        for (int i = 0; i < n; i++) {
            char* p = (char*)aligned_alloc(BUFFER_SIZE, BUFFER_SIZE);
            iov[i] = uv_buf_init(p, BUFFER_SIZE);
            for (size_t i = 0; i < BUFFER_SIZE; i++) {
                p[i] = 32 + rand()%10;
            }
        }
    }
}

struct Req {
    uv_buf_t*       buffers;
    size_t          numbuffers;
    const char*     path;
    size_t          current_offset;
    size_t          stride;
    size_t          cnt;
};


void on_write(uv_fs_t* req) {
    static int streams = N_STREAMS;
    Req* ureq = (Req*)req->data;
    if (req->result < 0) {
        std::cout << "Write error: " << uv_strerror(req->result) << std::endl;
    } else {
        auto off = ureq->current_offset;
        ureq->current_offset += ureq->stride;
        ureq->cnt++;
        if (off >= FSIZE) {
            // Done writing
            std::cout << "Done!" << std::endl;
            std::cout << "Final offset = " << off << ", cnt: " << ureq->cnt << std::endl;
            streams--;
            if (streams == 0) {
                uv_fs_close(uv_default_loop(), req, open_req.result, nullptr);
            }
            return;
        }
        // Perform next write
        uv_fs_write(uv_default_loop(), req, open_req.result, ureq->buffers, ureq->numbuffers, off, on_write);
    }
    //uv_fs_req_cleanup(req);
}

void on_open(uv_fs_t* req) {
    if (req->result != -1) {
        for (int i = 0; i < N_STREAMS; i++) {
            auto ureq = (Req*)writes[i].data;
            auto off = ureq->current_offset;
            ureq->current_offset += ureq->stride;
            uv_fs_write(uv_default_loop(),
                        &writes[i],
                        open_req.result,
                        ureq->buffers,
                        ureq->numbuffers,
                        off,
                        on_write);
        }
    } else {
        std::cout << "Open error: " << uv_strerror(req->result) << std::endl;
    }
    uv_fs_req_cleanup(req);
}

int main() {
    std::cout << "Creating " << PATH << " file" << std::endl;
    try {
        init_file();
        for (int i = 0; i < N_STREAMS; i++) {
            auto ureq = Req {
                    (uv_buf_t*)malloc(sizeof(uv_buf_t)),
                    1,
                    PATH,
                    i*BUFFER_SIZE,  // initial offset
                    N_STREAMS*BUFFER_SIZE,  // offset increment (for one stream it's just one buffer size)
                    0
            };
            create_buf(ureq.buffers, ureq.numbuffers);
            writes[i].data = &ureq;
        }
        PerfTimer total;
        uv_fs_open(uv_default_loop(), &open_req, PATH, O_DIRECT|O_RDWR, 0, &on_open);
        uv_run(uv_default_loop(), UV_RUN_DEFAULT);
        uv_loop_close(uv_default_loop());
        std::cout << "Done writing " << ((double)FSIZE/1024.0/1024.0) << "MB in "
                  << total.elapsed() << "s" << std::endl;
    } catch (const std::exception& e) {
        std::cout << "error main " << e.what() << std::endl;
    }
    return 0;
}