tets (e509140a) · Commits · tuxist / libparitypp

include/paritypp/server.h

+5 −0

Original line number	Diff line number	Diff line
		@@ -16,6 +16,11 @@
		#include <atomic>
		#include <thread>

		#ifdef _WIN32
		#include <sys/types.h>
		typedef intptr_t ssize_t;
		#endif

		namespace paritypp {

		// Block storage backend interface

src/server.cpp

+69 −21

Original line number	Diff line number	Diff line
		@@ -15,11 +15,59 @@
		#include <chrono>
		#include <set>

		#ifdef _WIN32
		#include <io.h>
		#include <fcntl.h>
		#include <sys/types.h>
		#include <sys/stat.h>
		#include <cstdio>

		static inline int pp_open(const char* path, int flags, int mode) {
		return ::_open(path, flags \| _O_BINARY, mode);
		}
		static inline int pp_close(int fd) { return ::_close(fd); }
		static inline off_t pp_lseek(int fd, off_t off, int whence) {
		return ::_lseek(fd, off, whence);
		}
		static inline ssize_t pp_pread(int fd, void* buf, size_t count, off_t offset) {
		off_t prev = ::_lseek(fd, 0, SEEK_CUR);
		if (prev < 0) return -1;
		if (::_lseek(fd, offset, SEEK_SET) < 0) return -1;
		int n = ::_read(fd, buf, static_cast<unsigned int>(count));
		::_lseek(fd, prev, SEEK_SET);
		return n;
		}
		static inline ssize_t pp_pwrite(int fd, const void* buf, size_t count, off_t offset) {
		off_t prev = ::_lseek(fd, 0, SEEK_CUR);
		if (prev < 0) return -1;
		if (::_lseek(fd, offset, SEEK_SET) < 0) return -1;
		int n = ::_write(fd, buf, static_cast<unsigned int>(count));
		::_lseek(fd, prev, SEEK_SET);
		return n;
		}
		static inline int pp_unlink(const char* path) { return ::_unlink(path); }
		#else
		#include <sys/types.h>
		#include <sys/stat.h>
		#include <fcntl.h>
		#include <unistd.h>

		static inline int pp_open(const char* path, int flags, int mode) {
		return ::open(path, flags, mode);
		}
		static inline int pp_close(int fd) { return ::close(fd); }
		static inline off_t pp_lseek(int fd, off_t off, int whence) {
		return ::lseek(fd, off, whence);
		}
		static inline ssize_t pp_pread(int fd, void* buf, size_t count, off_t offset) {
		return ::pread(fd, buf, count, offset);
		}
		static inline ssize_t pp_pwrite(int fd, const void* buf, size_t count, off_t offset) {
		return ::pwrite(fd, buf, count, offset);
		}
		static inline int pp_unlink(const char* path) { return ::unlink(path); }
		#endif

		namespace paritypp {

		// --- memory_block_store ---
		@@ -88,13 +136,13 @@ std::vector<uint64_t> memory_block_store::list_groups() {
		file_block_store::file_block_store(const std::string& base_dir)
		: data_path_(base_dir + "/blocks.bin") {
		std::filesystem::create_directories(base_dir);
		fd_ = ::open(data_path_.c_str(), O_RDWR \| O_CREAT, 0644);
		fd_ = pp_open(data_path_.c_str(), O_RDWR \| O_CREAT, 0644);
		if (fd_ >= 0)
		scan_and_build_index();
		}

		file_block_store::~file_block_store() {
		if (fd_ >= 0) ::close(fd_);
		if (fd_ >= 0) pp_close(fd_);
		}

		bool file_block_store::scan_and_build_index() {
		@@ -102,14 +150,14 @@ bool file_block_store::scan_and_build_index() {
		total_blocks_ = 0;
		dead_bytes_ = 0;
		off_t pos = 0;
		off_t end = ::lseek(fd_, 0, SEEK_END);
		off_t end = pp_lseek(fd_, 0, SEEK_END);
		if (end < 0) return false;
		::lseek(fd_, 0, SEEK_SET);
		pp_lseek(fd_, 0, SEEK_SET);
		file_size_ = end;

		uint8_t hdr[RECORD_HEADER_SIZE];
		while (pos + static_cast<off_t>(RECORD_HEADER_SIZE) <= end) {
		ssize_t n = ::pread(fd_, hdr, RECORD_HEADER_SIZE, pos);
		ssize_t n = pp_pread(fd_, hdr, RECORD_HEADER_SIZE, pos);
		if (n != static_cast<ssize_t>(RECORD_HEADER_SIZE)) break;

		uint8_t flags = hdr[0];
		@@ -147,14 +195,14 @@ bool file_block_store::store(uint64_t group_id, uint32_t block_index,
		auto bit = git->second.find(block_index);
		if (bit != git->second.end()) {
		uint8_t del = RECORD_DELETED;
		(void)::pwrite(fd_, &del, 1, bit->second.offset);
		(void)pp_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_;
		}
		}

		off_t pos = ::lseek(fd_, 0, SEEK_END);
		off_t pos = pp_lseek(fd_, 0, SEEK_END);
		if (pos < 0) return false;

		uint8_t hdr[RECORD_HEADER_SIZE];
		@@ -164,10 +212,10 @@ bool file_block_store::store(uint64_t group_id, uint32_t block_index,
		std::memcpy(&hdr[9], &block_index, 4);
		std::memcpy(&hdr[13], &dlen, 4);

		ssize_t w = ::pwrite(fd_, hdr, RECORD_HEADER_SIZE, pos);
		ssize_t w = pp_pwrite(fd_, hdr, RECORD_HEADER_SIZE, pos);
		if (w != static_cast<ssize_t>(RECORD_HEADER_SIZE)) return false;
		if (len > 0) {
		w = ::pwrite(fd_, data, len, pos + RECORD_HEADER_SIZE);
		w = pp_pwrite(fd_, data, len, pos + RECORD_HEADER_SIZE);
		if (w != static_cast<ssize_t>(len)) return false;
		}

		@@ -192,7 +240,7 @@ bool file_block_store::fetch(uint64_t group_id, uint32_t block_index,
		const auto& e = bit->second;
		out.resize(e.data_len);
		if (e.data_len > 0) {
		ssize_t n = ::pread(fd_, out.data(), e.data_len,
		ssize_t n = pp_pread(fd_, out.data(), e.data_len,
		e.offset + RECORD_HEADER_SIZE);
		if (n != static_cast<ssize_t>(e.data_len)) return false;
		}
		@@ -207,7 +255,7 @@ bool file_block_store::remove_group(uint64_t group_id) {
		// Tombstone all records in this group
		for (auto& [idx, entry] : git->second) {
		uint8_t del = RECORD_DELETED;
		(void)::pwrite(fd_, &del, 1, entry.offset);
		(void)pp_pwrite(fd_, &del, 1, entry.offset);
		dead_bytes_ += static_cast<off_t>(RECORD_HEADER_SIZE + entry.data_len);
		--total_blocks_;
		}
		@@ -251,7 +299,7 @@ bool file_block_store::vacuum() {
		std::lock_guard<std::mutex> lock(mutex_);

		std::string tmp_path = data_path_ + ".tmp";
		int tmp_fd = ::open(tmp_path.c_str(), O_RDWR \| O_CREAT \| O_TRUNC, 0644);
		int tmp_fd = pp_open(tmp_path.c_str(), O_RDWR \| O_CREAT \| O_TRUNC, 0644);
		if (tmp_fd < 0) return false;

		off_t new_pos = 0;
		@@ -264,18 +312,18 @@ bool file_block_store::vacuum() {
		// Read old header + data
		size_t rec_size = RECORD_HEADER_SIZE + entry.data_len;
		std::vector<uint8_t> buf(rec_size);
		ssize_t n = ::pread(fd_, buf.data(), rec_size, entry.offset);
		ssize_t n = pp_pread(fd_, buf.data(), rec_size, entry.offset);
		if (n != static_cast<ssize_t>(rec_size)) {
		::close(tmp_fd);
		::unlink(tmp_path.c_str());
		pp_close(tmp_fd);
		pp_unlink(tmp_path.c_str());
		return false;
		}
		buf[0] = RECORD_LIVE; // ensure live flag

		ssize_t w = ::pwrite(tmp_fd, buf.data(), rec_size, new_pos);
		ssize_t w = pp_pwrite(tmp_fd, buf.data(), rec_size, new_pos);
		if (w != static_cast<ssize_t>(rec_size)) {
		::close(tmp_fd);
		::unlink(tmp_path.c_str());
		pp_close(tmp_fd);
		pp_unlink(tmp_path.c_str());
		return false;
		}

		@@ -290,12 +338,12 @@ bool file_block_store::vacuum() {
		}

		// Swap files
		::close(fd_);
		pp_close(fd_);
		if (::rename(tmp_path.c_str(), data_path_.c_str()) != 0) {
		fd_ = ::open(data_path_.c_str(), O_RDWR \| O_CREAT, 0644);
		fd_ = pp_open(data_path_.c_str(), O_RDWR \| O_CREAT, 0644);
		return false;
		}
		fd_ = ::open(data_path_.c_str(), O_RDWR, 0644);
		fd_ = pp_open(data_path_.c_str(), O_RDWR, 0644);
		if (fd_ < 0) return false;

		index_ = std::move(new_index);