// Step 4 -- scan 8 bytes at a time: SWAR (SIMD Within A Register).
//
// Step 3's hot loop examines one byte per iteration: load, compare against
// ',', '\r', '\n' (or '"'), branch. On a 50 MB file that is 50M dependent
// branchy iterations. The classic fix is to test WHOLE WORDS:
//
//   * Load 8 bytes as one uint64_t (memcpy -- the portable unaligned load).
//   * XOR with a "broadcast" of the target byte (0x0101...01 * b): any byte
//     equal to the target becomes 0x00.
//   * The zero-byte trick, (v - 0x01..01) & ~v & 0x80..80, lights up bit 7 of
//     every zero byte. OR the masks for all four delimiters; if the result is
//     zero, all 8 bytes are boring and we skip them in one iteration.
//   * When the mask is nonzero, __builtin_ctzll(mask) >> 3 is the index of
//     the first interesting byte (little-endian).
//
// Why SWAR and not memchr or NEON/AVX intrinsics? memchr finds ONE byte
// value; an unquoted field ends at any of ',' '\r' '\n' and a quoted one at
// '"', so we would need multiple passes. Platform SIMD does 16-64 bytes per
// step but forks the code per architecture; SWAR is plain C++, runs anywhere,
// and captures most of the win -- the file is consumed 8 bytes per loop
// instead of 1. The last <8 bytes fall back to the byte loop, which also
// keeps every load inside the mapping.
//
// Everything else -- views, deferred unescape, the record loop -- is step 3
// unchanged, so the delta measures the scanner alone.
//
// THE MEASURED LESSON: on this data the win is ~5% (1293 -> 1354 MB/s), not
// the 2-4x the technique is famous for. Why: bench.csv's mean field is 7.1
// bytes, so most scans hit a delimiter inside the FIRST 8-byte word -- the
// fixed costs per field (call, mask, ctz) are the same as the byte loop's,
// and there are almost no "boring" words to skip. SWAR earns its keep on
// long fields (documents, JSON blobs, base64 columns); on short-field CSV
// the byte loop was already near the structural limit. Measure, then decide.
//
// Build:  c++ -std=c++20 -O2 step4_swar.cpp -o step4
#include "csv_test_common.hpp"
#include <fcntl.h>
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
#include <cstdint>
#include <cstring>
#include <string>
#include <string_view>
#include <vector>

struct Mapped {
    const char* data = nullptr;
    size_t size = 0;
    Mapped(const char* path) {
        int fd = ::open(path, O_RDONLY);
        if (fd < 0) { std::perror(path); std::exit(1); }
        struct stat st{};
        ::fstat(fd, &st);
        size = size_t(st.st_size);
        data = (const char*)::mmap(nullptr, size, PROT_READ, MAP_PRIVATE, fd, 0);
        ::close(fd);
        if (data == MAP_FAILED) { std::perror("mmap"); std::exit(1); }
    }
    ~Mapped() { if (data) ::munmap((void*)data, size); }
};

struct Field {
    std::string_view raw;
    bool needs_unescape;
};

static void unescape_into(std::string_view raw, std::string& out) {
    out.clear();
    for (size_t i = 0; i < raw.size(); ++i) {
        out += raw[i];
        if (raw[i] == '"') ++i;
    }
}

// ---- the SWAR kernel ---------------------------------------------------------
static constexpr uint64_t kOnes = 0x0101010101010101ull;
static constexpr uint64_t kHighs = 0x8080808080808080ull;

// bit 7 set in every byte of x that equals b
static inline uint64_t match_byte(uint64_t x, uint8_t b) {
    uint64_t v = x ^ (kOnes * b);
    return (v - kOnes) & ~v & kHighs;
}

// first p in [p, end) whose byte is ',' '\r' or '\n'; end if none.
// (A stray '"' inside an unquoted field is not a terminator -- same lenient
// behavior as step 3, and one fewer mask to compute.)
static inline const char* scan_unquoted(const char* p, const char* end) {
    while (end - p >= 8) {
        uint64_t x;
        std::memcpy(&x, p, 8);
        uint64_t m = match_byte(x, ',') | match_byte(x, '\r') | match_byte(x, '\n');
        if (m) return p + (__builtin_ctzll(m) >> 3);
        p += 8;
    }
    while (p < end && *p != ',' && *p != '\r' && *p != '\n') ++p;
    return p;
}

// first '"' in [p, end); end if none
static inline const char* scan_quoted(const char* p, const char* end) {
    while (end - p >= 8) {
        uint64_t x;
        std::memcpy(&x, p, 8);
        if (uint64_t m = match_byte(x, '"')) return p + (__builtin_ctzll(m) >> 3);
        p += 8;
    }
    while (p < end && *p != '"') ++p;
    return p;
}

template <typename OnRecord>
static void parse_swar(const char* p, const char* end, OnRecord on_record) {
    std::vector<Field> rec;
    rec.reserve(16);
    while (p < end) {
        rec.clear();
        for (;;) {
            const char* begin;
            if (p < end && *p == '"') {               // quoted field
                begin = ++p;
                bool esc = false;
                for (;;) {
                    p = scan_quoted(p, end);
                    if (p + 1 < end && p[1] == '"') { esc = true; p += 2; }
                    else break;
                }
                rec.push_back({{begin, size_t(p - begin)}, esc});
                if (p < end) ++p;                     // closing quote
            } else {                                   // unquoted field
                begin = p;
                p = scan_unquoted(p, end);
                rec.push_back({{begin, size_t(p - begin)}, false});
            }
            if (p >= end) break;
            char c = *p++;
            if (c == ',') continue;
            if (c == '\r' && p < end && *p == '\n') ++p;
            break;
        }
        on_record(rec);
    }
}

static std::vector<std::vector<std::string>> parse_all_materialized(const char* path) {
    Mapped m(path);
    std::vector<std::vector<std::string>> rows;
    std::string scratch;
    parse_swar(m.data, m.data + m.size, [&](const std::vector<Field>& rec) {
        std::vector<std::string> out;
        for (const Field& f : rec) {
            if (f.needs_unescape) { unescape_into(f.raw, scratch); out.push_back(scratch); }
            else out.emplace_back(f.raw);
        }
        rows.push_back(std::move(out));
    });
    return rows;
}

int main() {
    if (!run_edge_checks("edge.csv", parse_all_materialized)) return 1;

    Mapped m("bench.csv");
    const double mb = m.size / 1e6;

    BenchResult res;
    for (int rep = 0; rep < 3; ++rep) {
        unsigned long long a0 = allocs_now();
        double t0 = now_secs();
        unsigned long long rows = 0, fields = 0, byte_sum = 0;
        parse_swar(m.data, m.data + m.size, [&](const std::vector<Field>& rec) {
            ++rows;
            fields += rec.size();
            for (const Field& f : rec) byte_sum += f.raw.size();
        });
        double dt = now_secs() - t0;
        if (dt < res.best_secs) res.best_secs = dt;
        res.rows = rows; res.fields = fields; res.byte_sum = byte_sum;
        res.last_pass_allocs = allocs_now() - a0;
    }
    print_bench("step4_swar", res, mb);
    return 0;
}