du ioring comparison

du-cpp17.cpp

#include <iostream>
#include <fstream>
#include <filesystem>

#include <chrono>

using namespace std;

using Timer = std::chrono::high_resolution_clock;
using TimePoint = std::chrono::time_point<Timer>;
using ms = std::chrono::milliseconds;
using ns = std::chrono::nanoseconds;
namespace fs = std::filesystem;

int main(int argc, char* argv[]) {
    fs::path cwd = fs::current_path();

    TimePoint start, end;

    unsigned long total_size_bytes = 0;

    start = Timer::now();

    for (auto& de: fs::recursive_directory_iterator(cwd)) {
        const fs::path& p = de.path();
        if (de.is_regular_file()) {
            total_size_bytes += fs::file_size(p);
        }
    }

    end = Timer::now();

    unsigned duration_ms = std::chrono::duration_cast<ms>(end - start).count();

    cout << argv[0] <<": path=" << cwd << " total_size=" << total_size_bytes << " duration_ms=" << duration_ms << endl;
}

du-io_uring.cpp

#include <iostream>
#include <fstream>
#include <filesystem>
#include <future>
#include <vector>
#include <list>
#include <map>

#include <chrono>

#include <liburing.h>
#include <sys/stat.h>

using namespace std;

using Timer = std::chrono::high_resolution_clock;
using TimePoint = std::chrono::time_point<Timer>;
using ms = std::chrono::milliseconds;
using ns = std::chrono::nanoseconds;
namespace fs = std::filesystem;

class IoManager {
public:
    IoManager() {
        int ret;
        ret = io_uring_queue_init(IO_URING_QUEUE_MAX_SIZE, &_ring, 0);
        if (ret) {
            cerr << "ring setup failed" << endl;
            exit(1);
        }
    }
    ~IoManager() {
        io_uring_queue_exit(&_ring);
    }

    /**
     * @return How many items were drained
     */
    size_t DrainIoRing(int expected) {
        //cout << "Calling DrainIoRing" << endl;
        int drained_items = 0;
        // While there are events for us
        for (int i = 0; i < expected; ++i) {
            struct io_uring_cqe *cqe; // completion queue entry
            int ret = io_uring_wait_cqe(&_ring, &cqe);
            if (ret < 0) {
                cerr << "waiting for completion failed: " << ret << endl;
                exit(1);
            }

            // Retrieve pointer into _in_flights
            fs::path* path_key_completed = (fs::path*)(cqe->user_data);
            // Find in in_flights
            auto it = _in_flights.find(*path_key_completed);
            unsigned long file_size = it->second.stx_size;

            _total_size += file_size;

            //cout << "CQE received for path=" << *path_key_completed << " size=" << file_size << endl;

            // Mark as seen so kernel can re-use slot in ring
            io_uring_cqe_seen(&_ring, cqe);
            // Clear in flights (commented out for perf testing)
            //_in_flights.erase(it);
            drained_items++;
        }
        return drained_items;
    }

    size_t AsyncGetFileSizes(const vector<fs::path>& paths) {
        struct statx x1;
        const auto total_items = paths.size();
        int inserted_items = 0;
        int completed_items = 0;

        // Compute and send requests
        for (const auto& path: paths) {
            //cout << "io_uring_sq_space_left=" << io_uring_sq_space_left(&_ring) << endl;
            if (inserted_items == IO_URING_QUEUE_MAX_SIZE) {
                inserted_items = 0;
                /*int ret = io_uring_submit(&_ring);
                if (ret <= 0) {
                    cerr << "seq submit failed: " << ret << endl;
                    exit(1);
                }*/
                while (io_uring_submit(&_ring) > 0) {
                    //cerr << "unable to submit, need to sleep!" << endl;
                    std::this_thread::sleep_for(std::chrono::milliseconds(1));
                };
                completed_items += DrainIoRing(IO_URING_QUEUE_MAX_SIZE);
            }

            // ASSUMPTION: this list doesn't get reallocated and pointers stay
            // valid!
            auto [it, emplace_success] = \
                _in_flights.emplace(path, x1);
            // Used to identify requests in SQE/CQE
            const fs::path* path_key = &(it->first);
            struct statx* path_statx = &(it->second);

            //cout << "Enqueing path=" << *path_key << endl;

            

            struct io_uring_sqe *sqe;  // submission queue entry;
            sqe = io_uring_get_sqe(&_ring);
            if (!sqe) {
                cerr << "get sqe failed" << endl;
                exit(1);
            }
            
            //cout << "Processing path=" << path << endl;
            io_uring_prep_statx(sqe, -1, path.c_str(), 0, STATX_ALL, path_statx);
            io_uring_sqe_set_data(sqe, const_cast<fs::path*>(path_key));
            inserted_items++;
        }

        /*cout << "End of processing total_items=" << total_items
             << " paths, remaining_items=" << (total_items - completed_items)
             << " so far, total_size=" << _total_size << endl;*/

        /*cout << "Remaining items for which we await CQEs:";
        for (const auto& [path, statx]: _in_flights)
            cout << "  " << path << endl;*/

        while (io_uring_submit(&_ring) > 0) {
            //cerr << "unable to submit, need to sleep!" << endl;
            std::this_thread::sleep_for(std::chrono::milliseconds(1));
        };

        if (completed_items != total_items)
            completed_items += DrainIoRing(total_items - completed_items);

        return _total_size;
    }
private:
    const unsigned IO_URING_QUEUE_MAX_SIZE = 64;
    struct io_uring _ring;
    std::map<fs::path, struct statx> _in_flights;
    unsigned long _total_size = 0;
};



int main(int argc, char* argv[]) {
    fs::path cwd = fs::current_path();

    TimePoint start, end;
    IoManager iom;

    unsigned long total_size_bytes = 0;

    start = Timer::now();

    int count = 0;
    std::vector<fs::path> paths;
    for (auto& de: fs::recursive_directory_iterator(cwd)) {
        const fs::path& p = de.path();
        if (de.is_regular_file()) {
            paths.emplace_back(p);
        }
    }

    total_size_bytes += iom.AsyncGetFileSizes(paths);

    end = Timer::now();

    unsigned duration_ms = std::chrono::duration_cast<ms>(end - start).count();

    cout << argv[0] << ": path=" << cwd << " total_size=" << total_size_bytes << " duration_ms=" << duration_ms << endl;
}