Commit 1ac44862 authored by jan.koester's avatar jan.koester
Browse files

test

parent b82c3fa2

include/paritypp/server.h

+3 −0
Original line number Diff line number Diff line
@@ -101,8 +101,11 @@ private: 
    std::unordered_map<uint64_t,

        std::unordered_map<uint32_t, block_entry>> index_;

    uint32_t total_blocks_ = 0;

    off_t dead_bytes_ = 0;     // accumulated tombstoned bytes (header + data)

    off_t file_size_  = 0;    // current file size



    bool scan_and_build_index();

    void maybe_auto_vacuum();

};



// QUIC-based parity storage server

src/client.cpp

+10 −0
Original line number Diff line number Diff line
@@ -697,6 +697,16 @@ void client::store(uint64_t group_id, const uint8_t* data, size_t data_len) { 
    DBG_LOG("[PARITY] store gid=" << group_id << " data_len=" << data_len

              << " stripes=" << total_stripes << "\n");



    // Remove old blocks for this group before writing new ones.

    // This prevents orphan stripes when data size changes and avoids

    // unbounded tombstone accumulation in the block store file.

    try {

        remove(group_id);

    } catch (...) {

        // Best-effort: if removal fails, store_stripe will overwrite

        // same-index blocks via tombstone+append anyway.

    }



    // Retry once on failure: stale QUIC connections from a previous operation

    // cause the first stripe to fail; resetting connections fixes it.

    for (int attempt = 0; attempt < 2; ++attempt) {

src/server.cpp

+19 −0
Original line number Diff line number Diff line
@@ -100,10 +100,12 @@ file_block_store::~file_block_store() { 
bool file_block_store::scan_and_build_index() {

    index_.clear();

    total_blocks_ = 0;

    dead_bytes_ = 0;

    off_t pos = 0;

    off_t end = ::lseek(fd_, 0, SEEK_END);

    if (end < 0) return false;

    ::lseek(fd_, 0, SEEK_SET);

    file_size_ = end;



    uint8_t hdr[RECORD_HEADER_SIZE];

    while (pos + static_cast<off_t>(RECORD_HEADER_SIZE) <= end) {
@@ -126,6 +128,8 @@ bool file_block_store::scan_and_build_index() { 
            entry.offset   = pos;

            entry.data_len = data_len;

            ++total_blocks_;

        } else {

            dead_bytes_ += static_cast<off_t>(RECORD_HEADER_SIZE + data_len);

        }



        pos += static_cast<off_t>(RECORD_HEADER_SIZE + data_len);
@@ -144,6 +148,7 @@ bool file_block_store::store(uint64_t group_id, uint32_t block_index, 
        if (bit != git->second.end()) {

            uint8_t del = RECORD_DELETED;

            (void)::pwrite(fd_, &del, 1, bit->second.offset);

            dead_bytes_ += static_cast<off_t>(RECORD_HEADER_SIZE + bit->second.data_len);

            git->second.erase(bit);

            --total_blocks_;

        }
@@ -171,6 +176,8 @@ bool file_block_store::store(uint64_t group_id, uint32_t block_index, 
    e.data_len = dlen;

    index_[group_id][block_index] = e;

    ++total_blocks_;

    file_size_ = pos + RECORD_HEADER_SIZE + dlen;

    maybe_auto_vacuum();

    return true;

}
@@ -201,6 +208,7 @@ bool file_block_store::remove_group(uint64_t group_id) { 
    for (auto& [idx, entry] : git->second) {

        uint8_t del = RECORD_DELETED;

        (void)::pwrite(fd_, &del, 1, entry.offset);

        dead_bytes_ += static_cast<off_t>(RECORD_HEADER_SIZE + entry.data_len);

        --total_blocks_;

    }

    index_.erase(git);
@@ -292,9 +300,20 @@ bool file_block_store::vacuum() { 


    index_ = std::move(new_index);

    total_blocks_ = new_total;

    dead_bytes_ = 0;

    file_size_ = new_pos;

    return true;

}



void file_block_store::maybe_auto_vacuum() {

    // Auto-vacuum when dead space exceeds 10MB and is more than 50% of the file

    static constexpr off_t MIN_DEAD_BYTES = 10 * 1024 * 1024; // 10 MB

    if (dead_bytes_ >= MIN_DEAD_BYTES && file_size_ > 0 &&

        dead_bytes_ > file_size_ / 2) {

        vacuum();

    }

}



// --- server ---



static std::map<std::string, netplus::ssl::CertificateBundle> load_certs(