From 8b6c3e71ab70ce739bc60073a892bbe9f7ba5fcd Mon Sep 17 00:00:00 2001 From: Lanzheng Liu Date: Tue, 15 Sep 2026 16:01:21 +0800 Subject: [PATCH 1/4] fix(lsmt): make a file's shared state safe across vCPUs The read-path statistics belong to the file, not to the vCPU that reads it, so with a multi-vCPU backend every worker serving the same image updates lsmt_io_cnt, lsmt_io_size and the allocated-block counter of the writable index at the same time, and their plain read-modify-writes lose updates. Make all three atomic and update them with relaxed ordering. The writable index itself is worse than a lost counter: it is a std::set that insert() mutates in place, and while pwrite() serialises the writers on m_rw_mtx, the readers never took it. On a single vCPU that was safe only because photon does not preempt and neither lookup() nor insert() yields; across vCPUs a reader walking the tree while another vCPU erases and reinserts a mapping resolves a hole or a foreign mapping, and one that catches a rebalance in progress follows nodes that are no longer on its search path. Take a photon::rwlock on every access to the index of a file: for reading in pread() and seek_data() and around the index snapshots of commit(), close_seal() and flatten(), for writing in index_insert(), and across the whole of restack(), which also reshuffles m_files. m_rw_mtx still comes first and the rwlock second, and readers never take m_rw_mtx, so the two cannot deadlock. The write lock covers the index mutation only, not the data I/O of a write, so appending does not stall the readers; a hybrid writer still rewrites the data of an existing mapping in place outside of it, so a read overlapping a write of the same LBA may return a mix of both, which is what a same-LBA read/write race is. A read-only file has an immutable index, so m_index_mutable stays false there and pread() skips the lock: that is the hot path of an image service, and a lock that is only ever shared still puts every vCPU on one cache line. For the same reason pread() delegates to do_pread(), which also lets the MAX_IO_SIZE splitting take the lock once instead of recursively -- photon's rwlock is not reentrant. Also close the writers that mutated the index with no lock at all: discard() inserted before taking m_rw_mtx, and the sparse and the warp files inserted and appended to the index log without it. Add two tests. multi_vcpu_io_counters reads a sealed layer from four vCPUs and checks the counters against the number of reads. multi_vcpu_concurrent_rw maps a whole 1MB layer, then has two vCPUs rewrite random blocks while two others read them, requiring every block to come back as one consistent version of its own pattern; with the index lock disabled it reads 5 corrupted blocks out of 40000 and fails. Signed-off-by: Lanzheng Liu --- src/overlaybd/lsmt/file.cpp | 104 ++++++++++++++++------ src/overlaybd/lsmt/index.cpp | 10 ++- src/overlaybd/lsmt/test/test.cpp | 142 +++++++++++++++++++++++++++++++ 3 files changed, 228 insertions(+), 28 deletions(-) diff --git a/src/overlaybd/lsmt/file.cpp b/src/overlaybd/lsmt/file.cpp index c92aee95..8f944be9 100644 --- a/src/overlaybd/lsmt/file.cpp +++ b/src/overlaybd/lsmt/file.cpp @@ -503,14 +503,21 @@ class LSMTReadOnlyFile : public IFileRW { vector m_files; vector m_uuid; IMemoryIndex *m_index = nullptr; + // guards the writable index, which is mutated in place while other vCPUs + // read it; mutable because commit() is const + mutable photon::rwlock m_index_rwlock; + // false for a read-only file, whose immutable index needs no lock + bool m_index_mutable = false; bool m_file_ownership = false; uint64_t m_data_offset = HeaderTrailer::SPACE / ALIGNMENT; - uint32_t lsmt_io_cnt = 0; - uint64_t lsmt_io_size = 0; + // read-path statistics, updated by every vCPU serving this file + atomic lsmt_io_cnt{0}; + atomic lsmt_io_size{0}; LSMTFileType m_filetype = LSMTFileType::RO; virtual ~LSMTReadOnlyFile() { - LOG_INFO("pread times: `, size: `M", lsmt_io_cnt, lsmt_io_size >> 20); + LOG_INFO("pread times: `, size: `M", lsmt_io_cnt.load(memory_order_relaxed), + lsmt_io_size.load(memory_order_relaxed) >> 20); close(); if (m_file_ownership) { LOG_DEBUG("m_file_ownership:`, m_files.size:`", m_file_ownership, m_files.size()); @@ -612,10 +619,19 @@ class LSMTReadOnlyFile : public IFileRW { LOG_ERROR_RETURN(EFAULT, -1, "arguments must be aligned!"); virtual ssize_t pread(void *buf, size_t count, off_t offset) override { + if (!m_index_mutable) + return do_pread(buf, count, offset); + photon::scoped_rwlock ilock(m_index_rwlock, photon::RLOCK); + return do_pread(buf, count, offset); + } + + // the caller holds m_index_rwlock for reading; the rwlock is not reentrant, + // so the MAX_IO_SIZE splitting recurses here rather than into pread() + ssize_t do_pread(void *buf, size_t count, off_t offset) { CHECK_ALIGNMENT(count, offset); auto nbytes = count; while (count > MAX_IO_SIZE) { - auto ret = pread(buf, MAX_IO_SIZE, offset); + auto ret = do_pread(buf, MAX_IO_SIZE, offset); if (ret < (ssize_t)MAX_IO_SIZE) return -1; if (buf != nullptr) { @@ -660,8 +676,8 @@ class LSMTReadOnlyFile : public IFileRW { memset((char *)buf + ret, 0, size - ret); } } - lsmt_io_size += ret; - lsmt_io_cnt++; + lsmt_io_size.fetch_add((uint64_t)ret, memory_order_relaxed); + lsmt_io_cnt.fetch_add(1, memory_order_relaxed); (char *&)buf += size; return 0; }); @@ -731,6 +747,7 @@ class LSMTReadOnlyFile : public IFileRW { begin /= ALIGNMENT; end /= ALIGNMENT; + photon::scoped_rwlock ilock(m_index_rwlock, photon::RLOCK); while (begin < end) { SegmentMapping mappings[128]; auto length = (end - begin < Segment::MAX_LENGTH ? end - begin : Segment::MAX_LENGTH); @@ -771,6 +788,7 @@ class LSMTFile : public LSMTReadOnlyFile { RWType m_rw_type = RWType::Append; Mutex m_rw_mtx; + // lock order: m_rw_mtx, then m_index_rwlock; readers take only the latter IFile *m_findex = nullptr; vector m_stacked_mappings; @@ -781,6 +799,7 @@ class LSMTFile : public LSMTReadOnlyFile { LSMTFile() { m_compacted_idx_size.store(0); m_filetype = LSMTFileType::RW; + m_index_mutable = true; } ~LSMTFile() { @@ -800,6 +819,17 @@ class LSMTFile : public LSMTReadOnlyFile { return static_cast(m_index); } + // every access to the writable index goes through these two, or takes + // m_index_rwlock explicitly + void index_insert(const SegmentMapping &m) { + photon::scoped_rwlock ilock(m_index_rwlock, photon::WLOCK); + rw_index()->insert(m); + } + size_t index_lookup_writable(Segment s, SegmentMapping *pm, size_t n) { + photon::scoped_rwlock ilock(m_index_rwlock, photon::RLOCK); + return rw_index()->lookup_writable_layer(s, pm, n); + } + virtual int vioctl(int request, va_list args) override { if (request == GetType) { return LSMTReadOnlyFile::vioctl(request, args); @@ -924,7 +954,7 @@ class LSMTFile : public LSMTReadOnlyFile { m.moffset = (uint64_t)moffset / ALIGNMENT; m_data_offset = max(m_data_offset, m.mend()); m.tag = m_rw_tag; - rw_index()->insert(m); + index_insert(m); append_index(m); return 0; }; @@ -936,8 +966,7 @@ class LSMTFile : public LSMTReadOnlyFile { while (m_rw_type == RWType::Hybrid && cursor < end_in_blocks) { SegmentMapping upper[128]; auto length = min(end_in_blocks - cursor, (uint64_t)Segment::MAX_LENGTH); - auto n = rw_index()->lookup_writable_layer({cursor, (uint32_t)length}, upper, - LEN(upper)); + auto n = index_lookup_writable({cursor, (uint32_t)length}, upper, LEN(upper)); if (n == 0) break; for (size_t i = 0; i < n; i++) { @@ -1003,8 +1032,9 @@ class LSMTFile : public LSMTReadOnlyFile { m.moffset = (uint64_t)(pos / ALIGNMENT); m.tag = m_rw_tag; LOG_DEBUG(m); - static_cast(m_index)->insert(m); + // the insert and its index log record are one critical section Lock lock(m_rw_mtx); + index_insert(m); append_index(m); return 0; } @@ -1040,8 +1070,14 @@ class LSMTFile : public LSMTReadOnlyFile { } auto m_index0 = (IMemoryIndex0 *)m_index; - unique_ptr mapping(m_index0->dump()); - CompactOptions opts(&m_files, mapping.get(), m_index->size(), m_vsize, &args); + unique_ptr mapping; + size_t index_size; + { + photon::scoped_rwlock ilock(m_index_rwlock, photon::RLOCK); + mapping.reset(m_index0->dump()); + index_size = m_index->size(); + } + CompactOptions opts(&m_files, mapping.get(), index_size, m_vsize, &args); atomic_uint64_t _no_use_var(0); return compact(opts, _no_use_var); @@ -1049,9 +1085,15 @@ class LSMTFile : public LSMTReadOnlyFile { virtual int close_seal(IFileRO **reopen_as = nullptr) override { auto m_index0 = (IMemoryIndex0 *)m_index; - unique_ptr mapping(m_index0->dump(ALIGNMENT)); + unique_ptr mapping; + size_t index_size; + { + photon::scoped_rwlock ilock(m_index_rwlock, photon::RLOCK); + mapping.reset(m_index0->dump(ALIGNMENT)); + index_size = m_index0->size(); + } uint64_t index_offset = m_files[m_rw_tag]->lseek(0, SEEK_END); - ssize_t index_bytes = m_index0->size() * sizeof(SegmentMapping); + ssize_t index_bytes = index_size * sizeof(SegmentMapping); index_bytes = (index_bytes + ALIGNMENT - 1) / ALIGNMENT * ALIGNMENT; auto ret = m_files[m_rw_tag]->write(mapping.get(), index_bytes); if (ret < index_bytes) @@ -1060,13 +1102,13 @@ class LSMTFile : public LSMTReadOnlyFile { LayerInfo layer; if (load_layer_info(&m_files[m_rw_tag], 1, layer, true) != 0) return -1; - ret = write_header_trailer(m_files[m_rw_tag], false, true, true, index_offset, - m_index0->size(), layer); + ret = write_header_trailer(m_files[m_rw_tag], false, true, true, index_offset, index_size, + layer); if (ret < 0) LOG_ERRNO_RETURN(0, -1, "failed to write trailer."); if (reopen_as) { auto new_index = - create_memory_index(mapping.release(), m_index0->size(), + create_memory_index(mapping.release(), index_size, HeaderTrailer::SPACE / ALIGNMENT, index_offset / ALIGNMENT); if (new_index == nullptr) { LOG_ERROR("create memory index of reopen file failed."); @@ -1125,7 +1167,11 @@ class LSMTFile : public LSMTReadOnlyFile { virtual int flatten(IFile *as) override { - unique_ptr pmi((IComboIndex*)(m_index->make_read_only_index())); + unique_ptr pmi; + { + photon::scoped_rwlock ilock(m_index_rwlock, photon::RLOCK); + pmi.reset((IComboIndex *)(m_index->make_read_only_index())); + } if (!pmi) LOG_ERROR_RETURN(0, -1, "failed to make read only index."); @@ -1137,6 +1183,8 @@ class LSMTFile : public LSMTReadOnlyFile { int reserve_top_layer(LSMTFile *top_layer) { + // restack also reshuffles m_files, so no reader may be in flight + photon::scoped_rwlock ilock(m_index_rwlock, photon::WLOCK); std::vector pmappings; // temp index for reserved layer /* ==== close_seal the top RW layer and reopen it. ==== */ IFileRO* gc_layer = nullptr; @@ -1226,7 +1274,7 @@ class LSMTSparseFile : public LSMTFile { m_files[m_rw_tag], ret, moffset, count); } LOG_DEBUG("insert segment: `", m); - static_cast(m_index)->insert(m); + index_insert(m); } return ret; } @@ -1235,7 +1283,7 @@ class LSMTSparseFile : public LSMTFile { virtual int discard(SegmentMapping &m) override { m.moffset = (uint64_t)(m.offset + (HeaderTrailer::SPACE / ALIGNMENT)); LOG_DEBUG(m); - static_cast(m_index)->insert(m); + index_insert(m); return m_files[m_rw_tag]->trim(m.offset * ALIGNMENT + HeaderTrailer::SPACE, m.length * ALIGNMENT); } @@ -1318,7 +1366,8 @@ class LSMTWarpFile : public LSMTFile { LOG_ERRNO_RETURN(0, -1, "write failed, file:`, ret:`, pos:`, count:`", file, ret, offset, count); } - static_cast(m_index)->insert(m); + Lock lock(m_rw_mtx); + index_insert(m); append_index(m); return count; } @@ -1337,6 +1386,7 @@ class LSMTWarpFile : public LSMTFile { LOG_DEBUG("RemoteMapping: {offset: `, count: `, roffset: `}", lba.offset, lba.count, lba.roffset); size_t nwrite = 0; + Lock lock(m_rw_mtx); while (lba.count > 0) { SegmentMapping m; m.offset = lba.offset / ALIGNMENT; @@ -1345,7 +1395,7 @@ class LSMTWarpFile : public LSMTFile { m.moffset = lba.roffset / ALIGNMENT; m.tag = m_rw_tag + (uint8_t)SegmentType::remoteData; LOG_DEBUG("insert segment: ` into findex: `", m, m_findex); - static_cast(m_index)->insert(m); + index_insert(m); append_index(m); nwrite += m.length * ALIGNMENT; lba.offset += m.length * ALIGNMENT; @@ -1398,8 +1448,14 @@ class LSMTWarpFile : public LSMTFile { int commit(const CommitArgs &args) const override { auto m_index0 = (IMemoryIndex0 *)m_index; - unique_ptr mapping(m_index0->dump()); - CompactOptions opts(&m_files, mapping.get(), m_index->size(), m_vsize, &args); + unique_ptr mapping; + size_t raw_index_size; + { + photon::scoped_rwlock ilock(m_index_rwlock, photon::RLOCK); + mapping.reset(m_index0->dump()); + raw_index_size = m_index->size(); + } + CompactOptions opts(&m_files, mapping.get(), raw_index_size, m_vsize, &args); LayerInfo info; info.virtual_size = m_vsize; info.uuid.clear(); diff --git a/src/overlaybd/lsmt/index.cpp b/src/overlaybd/lsmt/index.cpp index 139e3485..61f77e2d 100644 --- a/src/overlaybd/lsmt/index.cpp +++ b/src/overlaybd/lsmt/index.cpp @@ -19,6 +19,7 @@ #include #include #include +#include #include #include #include @@ -463,16 +464,17 @@ class Index0 : public IComboIndex { typedef set::iterator iterator; struct block_usage { - uint64_t m_alloc = 0; + // updated under the RW lock, read lock-free by block_count() + atomic m_alloc{0}; inline void operator-=(const SegmentMapping &m) { if (m.zeroed) return; - m_alloc = m_alloc - m.length; + m_alloc.fetch_sub(m.length, memory_order_relaxed); } inline void operator+=(const SegmentMapping &m) { if (m.zeroed) return; - m_alloc = m_alloc + m.length; + m_alloc.fetch_add(m.length, memory_order_relaxed); } } alloc_blk; @@ -594,7 +596,7 @@ class Index0 : public IComboIndex { } virtual uint64_t block_count() const override { - return alloc_blk.m_alloc; + return alloc_blk.m_alloc.load(memory_order_relaxed); } // returns the first and last mapping in the index diff --git a/src/overlaybd/lsmt/test/test.cpp b/src/overlaybd/lsmt/test/test.cpp index a6920a3f..314993e0 100644 --- a/src/overlaybd/lsmt/test/test.cpp +++ b/src/overlaybd/lsmt/test/test.cpp @@ -25,6 +25,7 @@ IMemoryIndex -> IMemoryIndex0 -> IComboIndex -> Index0 ( set ) -> Co #include "lsmt-filetest.h" #include "photon/fs/localfs.h" #include +#include #include #include #include @@ -1102,6 +1103,147 @@ TEST_F(FileTest3, photon_verify) { thread_join((photon::join_handle *)thd); } +// The read-path counters belong to the file, not to the vCPU that reads it, so +// all the vCPUs serving the same image update them concurrently. +TEST_F(FileTest, multi_vcpu_io_counters) { + const uint64_t VSIZE = 8 << 20; + const size_t WLEN = 1 << 20; // length of each write, hence of each mapping + const size_t BLOCK = 4096; // length of each read + const int NVCPU = 4; // one OS thread (vCPU) each + const int NREADS = 8192; // reads per vCPU + + name_next_layer(); + auto fdata = lfs->open(data_name.back().c_str(), O_RDWR | O_CREAT | O_TRUNC, S_IRWXU); + auto findex = lfs->open(idx_name.back().c_str(), O_RDWR | O_CREAT | O_TRUNC, S_IRWXU); + ASSERT_NE(nullptr, fdata); + ASSERT_NE(nullptr, findex); + LayerInfo args(fdata, findex); + args.virtual_size = VSIZE; + auto rw = LSMT::create_file_rw(args, true); + ASSERT_NE(nullptr, rw); + + // fill the image, so that every read below hits exactly one mapping + ALIGNED_MEM4K(wbuf, WLEN); + memset(wbuf, 0xab, WLEN); + for (uint64_t off = 0; off < VSIZE; off += WLEN) { + ASSERT_EQ((ssize_t)WLEN, rw->pwrite(wbuf, WLEN, off)); + } + ASSERT_EQ(0, rw->close_seal()); + delete rw; + + auto fro = lfs->open(data_name.back().c_str(), O_RDONLY); + ASSERT_NE(nullptr, fro); + auto file = dynamic_cast(LSMT::open_file_ro(fro, true)); + ASSERT_NE(nullptr, file); + DEFER(delete file); + ASSERT_EQ(0u, file->lsmt_io_cnt.load()); + ASSERT_EQ(0u, file->lsmt_io_size.load()); + + atomic failed{0}; + vector vcpus; + for (int i = 0; i < NVCPU; i++) { + vcpus.emplace_back([&] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_DEFAULT); + DEFER(photon::fini()); + ALIGNED_MEM4K(rbuf, BLOCK); + for (int j = 0; j < NREADS; j++) { + // keep every read within the first mapping: [0, WLEN) + auto off = (off_t)((j % (WLEN / BLOCK)) * BLOCK); + if (file->pread(rbuf, BLOCK, off) != (ssize_t)BLOCK) + failed.fetch_add(1); + } + }); + } + for (auto &t : vcpus) + t.join(); + + EXPECT_EQ(0, failed.load()); + EXPECT_EQ((uint64_t)NVCPU * NREADS, (uint64_t)file->lsmt_io_cnt.load()); + EXPECT_EQ((uint64_t)NVCPU * NREADS * BLOCK, file->lsmt_io_size.load()); +} + +// A writable layer mutates its index in place, so without the index lock the +// readers of the other vCPUs walk a tree that is being rebalanced. +TEST_F(FileTest, multi_vcpu_concurrent_rw) { + const uint64_t VSIZE = 1 << 20; + const size_t BLOCK = 4096; + const size_t NBLOCK = VSIZE / BLOCK; + const int NWRITER = 2, NREADER = 2; + const int NWRITES = 4000; // per writer + const int NREADS = 20000; // per reader + + // every block mixes its own number into its bytes, so neither a hole nor + // another block passes for it, and a block must agree on one version + auto byte_at = [](size_t b, size_t i, uint8_t v) { + return (uint8_t)(b * 31 + ((i ^ (b * 13)) * 7) + v + 1); + }; + + name_next_layer(); + auto fdata = lfs->open(data_name.back().c_str(), O_RDWR | O_CREAT | O_TRUNC, S_IRWXU); + auto findex = lfs->open(idx_name.back().c_str(), O_RDWR | O_CREAT | O_TRUNC, S_IRWXU); + ASSERT_NE(nullptr, fdata); + ASSERT_NE(nullptr, findex); + LayerInfo args(fdata, findex); + args.virtual_size = VSIZE; + auto file = LSMT::create_file_rw(args, true); + ASSERT_NE(nullptr, file); + DEFER(delete file); + + // map every block first, so that no read below may legitimately hit a hole + ALIGNED_MEM4K(wbuf, BLOCK); + for (size_t b = 0; b < NBLOCK; b++) { + for (size_t i = 0; i < BLOCK; i++) + wbuf[i] = byte_at(b, i, 0); + ASSERT_EQ((ssize_t)BLOCK, file->pwrite(wbuf, BLOCK, (off_t)(b * BLOCK))); + } + + atomic bad{0}; + vector vcpus; + for (int w = 0; w < NWRITER; w++) { + vcpus.emplace_back([&, w] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_DEFAULT); + DEFER(photon::fini()); + ALIGNED_MEM4K(buf, BLOCK); + uint32_t seed = 0x9e3779b9u * (w + 1); + for (int j = 0; j < NWRITES; j++) { + seed = seed * 1103515245 + 12345; + auto b = (size_t)((seed >> 16) % NBLOCK); + for (size_t i = 0; i < BLOCK; i++) + buf[i] = byte_at(b, i, (uint8_t)(j + 1)); + if (file->pwrite(buf, BLOCK, (off_t)(b * BLOCK)) != (ssize_t)BLOCK) + bad.fetch_add(1); + } + }); + } + for (int r = 0; r < NREADER; r++) { + vcpus.emplace_back([&, r] { + photon::init(photon::INIT_EVENT_DEFAULT, photon::INIT_IO_DEFAULT); + DEFER(photon::fini()); + ALIGNED_MEM4K(buf, BLOCK); + uint32_t seed = 0x85ebca6bu * (r + 1); + for (int j = 0; j < NREADS; j++) { + seed = seed * 1103515245 + 12345; + auto b = (size_t)((seed >> 16) % NBLOCK); + if (file->pread(buf, BLOCK, (off_t)(b * BLOCK)) != (ssize_t)BLOCK) { + bad.fetch_add(1); + continue; + } + auto v = (uint8_t)(((uint8_t *)buf)[0] - byte_at(b, 0, 0)); + for (size_t i = 0; i < BLOCK; i++) { + if (((uint8_t *)buf)[i] != byte_at(b, i, v)) { + bad.fetch_add(1); + break; + } + } + } + }); + } + for (auto &t : vcpus) + t.join(); + + EXPECT_EQ(0, bad.load()); +} + void WarpFileTest::randwrite_warpfile(IFile *file, size_t nwrites) { LOG_INFO("start randwrite ` times", nwrites); ALIGNED_MEM4K(buf, 1 << 20) From 837d826d45323cdd3cf4971a872b486d2e09ef78 Mon Sep 17 00:00:00 2001 From: Lanzheng Liu Date: Tue, 15 Sep 2026 16:01:21 +0800 Subject: [PATCH 2/4] fix(zfile): keep the compressor batch state request-local BaseCompressor held the per-block source and destination pointers of a batch in two member vectors, so two vCPUs compressing or decompressing through the same compressor overwrote each other's batch. Build them in std::arrays on the caller's stack and pass them down to do_compress() and do_decompress() instead, which also lets the QAT path drop the resize it needed once nbatch() started returning the real batch size. Bound the batch at MAX_BATCH (256) so those arrays cannot be overrun, and reject n == 0, which divided dst_buffer_capacity by zero. Signed-off-by: Lanzheng Liu --- src/overlaybd/zfile/compressor.cpp | 52 +++++++++++++++++++----------- 1 file changed, 33 insertions(+), 19 deletions(-) diff --git a/src/overlaybd/zfile/compressor.cpp b/src/overlaybd/zfile/compressor.cpp index d1dfee3e..3b83c41b 100644 --- a/src/overlaybd/zfile/compressor.cpp +++ b/src/overlaybd/zfile/compressor.cpp @@ -16,6 +16,7 @@ #include "compressor.h" #include "lz4/lz4.h" +#include #include #include #include @@ -45,13 +46,10 @@ static std::atomic g_qat_state{0}; #endif class BaseCompressor : public ICompressor { public: + enum : size_t { MAX_BATCH = 256 }; uint32_t max_dst_size = 0; uint32_t src_blk_size = 0; - // vector raw_data; - vector compressed_data; - vector uncompressed_data; - - const int DEFAULT_N_BATCH = 256; + enum { DEFAULT_N_BATCH = MAX_BATCH }; virtual int init(const CompressArgs *args) { auto opt = &args->opt; @@ -61,9 +59,6 @@ class BaseCompressor : public ICompressor { src_blk_size = opt->block_size; LOG_DEBUG("create batch buffer, size: `", nbatch()); - // raw_data.resize(nbatch()); - compressed_data.resize(nbatch()); - uncompressed_data.resize(nbatch()); return 0; } @@ -72,11 +67,15 @@ class BaseCompressor : public ICompressor { } virtual int do_compress(size_t *src_chunk_len /* uncompressed length per block */, - size_t *dst_chunk_len, size_t dst_buffer_capacity, size_t nblock) = 0; + size_t *dst_chunk_len, size_t dst_buffer_capacity, size_t nblock, + unsigned char **compressed_data, + unsigned char **uncompressed_data) = 0; virtual int do_decompress(size_t *src_chunk_len, /* compressed length per block */ size_t *dst_chunk_len, - size_t dst_buffer_capacity, size_t nblock) = 0; + size_t dst_buffer_capacity, size_t nblock, + unsigned char **compressed_data, + unsigned char **uncompressed_data) = 0; virtual int compress(const unsigned char *src, size_t src_len, unsigned char *dst, size_t dst_len) override { @@ -91,9 +90,14 @@ class BaseCompressor : public ICompressor { int compress_batch(const unsigned char *src, size_t *src_chunk_len, unsigned char *dst, size_t dst_buffer_capacity, size_t *dst_chunk_len, size_t n) override { + if (n == 0 || n > MAX_BATCH) { + LOG_ERROR_RETURN(EINVAL, -1, "invalid batch size (`), maximum is `", n, MAX_BATCH); + } if (dst_buffer_capacity / n < max_dst_size) { LOG_ERROR_RETURN(ENOBUFS, -1, "dst_len should be greater than `", max_dst_size - 1); } + std::array compressed_data; + std::array uncompressed_data; off_t src_offset = 0, dst_offset = 0; for (size_t i = 0; i < n; i++) { uncompressed_data[i] = ((unsigned char *)src + src_offset); @@ -101,7 +105,8 @@ class BaseCompressor : public ICompressor { src_offset += src_chunk_len[i]; dst_offset += dst_buffer_capacity / n; } - return do_compress(src_chunk_len, dst_chunk_len, dst_buffer_capacity, n); + return do_compress(src_chunk_len, dst_chunk_len, dst_buffer_capacity, n, + compressed_data.data(), uncompressed_data.data()); } virtual int decompress(const unsigned char *src, size_t src_len, unsigned char *dst, @@ -117,11 +122,16 @@ class BaseCompressor : public ICompressor { int decompress_batch(const unsigned char *src, size_t *src_chunk_len, unsigned char *dst, size_t dst_buffer_capacity, size_t *dst_chunk_len, size_t n) override { + if (n == 0 || n > MAX_BATCH) { + LOG_ERROR_RETURN(EINVAL, -1, "invalid batch size (`), maximum is `", n, MAX_BATCH); + } if (dst_buffer_capacity / n < src_blk_size) { LOG_ERROR_RETURN(ENOBUFS, -1, "dst_len (`) should be greater than compressed block size `", dst_buffer_capacity / n, src_blk_size); } + std::array compressed_data; + std::array uncompressed_data; off_t src_offset = 0, dst_offset = 0; for (size_t i = 0; i < n; i++) { compressed_data[i] = ((unsigned char *)src + src_offset); @@ -130,7 +140,8 @@ class BaseCompressor : public ICompressor { dst_offset += dst_buffer_capacity / n; } - return do_decompress(src_chunk_len, dst_chunk_len, dst_buffer_capacity, n); + return do_decompress(src_chunk_len, dst_chunk_len, dst_buffer_capacity, n, + compressed_data.data(), uncompressed_data.data()); } }; @@ -195,9 +206,6 @@ class LZ4Compressor : public BaseCompressor { if (qat_init(pQat) == 0) { qat_enable = true; g_qat_state.store(1, std::memory_order_release); - /* nbatch() now returns DEFAULT_N_BATCH (was 1 when BaseCompressor::init ran). */ - compressed_data.resize(DEFAULT_N_BATCH); - uncompressed_data.resize(DEFAULT_N_BATCH); } else { delete pQat; pQat = nullptr; @@ -213,7 +221,9 @@ class LZ4Compressor : public BaseCompressor { } virtual int do_compress(size_t *src_chunk_len, size_t *dst_chunk_len, - size_t dst_buffer_capacity, size_t nblock) override { + size_t dst_buffer_capacity, size_t nblock, + unsigned char **compressed_data, + unsigned char **uncompressed_data) override { int ret = 0; #ifdef ENABLE_QAT @@ -245,7 +255,8 @@ class LZ4Compressor : public BaseCompressor { } int do_decompress(size_t *src_chunk_len, size_t *dst_chunk_len, size_t dst_buffer_capacity, - size_t n) override { + size_t n, unsigned char **compressed_data, + unsigned char **uncompressed_data) override { int ret = 0; #ifdef ENABLE_QAT @@ -310,7 +321,8 @@ class Compressor_zstd : public BaseCompressor { virtual int do_compress(size_t *src_chunk_len /* uncompressed length per block */, size_t *dst_chunk_len, size_t dst_buffer_capacity, - size_t nblock) override { + size_t nblock, unsigned char **compressed_data, + unsigned char **uncompressed_data) override { int ret = 0; for (size_t i = 0; i < nblock; i++) { @@ -326,7 +338,9 @@ class Compressor_zstd : public BaseCompressor { virtual int do_decompress(size_t *src_chunk_len, /* compressed length per block */ size_t *dst_chunk_len, - size_t dst_buffer_capacity, size_t nblock) override { + size_t dst_buffer_capacity, size_t nblock, + unsigned char **compressed_data, + unsigned char **uncompressed_data) override { int ret = 0; for (size_t i = 0; i < nblock; i++) { From 7712b627efe3b05679f1138b8a7469f3d2b02390 Mon Sep 17 00:00:00 2001 From: Lanzheng Liu Date: Tue, 15 Sep 2026 16:01:21 +0800 Subject: [PATCH 3/4] feat(tcmu): support multi-vCPU backends with Photon v0.9 Dispatch TCMU commands across a shared Photon WorkPool and let each vCPU publish its own responses, allow the file cache in multi-vCPU mode, and add a benchmark that compares the backend against the single-vCPU implementation on the same runner. Commands beyond the ones the dispatcher starts locally go onto the pool's own task ring, so the device dispatcher never leaves its vCPU and keeps draining its mailbox: WorkPool spreads the overflow over whichever vCPUs are free, and its RingChannel pays for a wakeup only when a worker is parked. Each task just spawns the handler thread and returns, because the pool runs tasks inline on its own loop. Completions are published in place under a per-device spinlock with the uio notification coalesced by a counter, instead of being forwarded to the device's home vCPU one wakeup per command. libtcmu keeps the reader's cursor (dev->cmd_tail) separate from the response cursor (mb->cmd_tail) and the kernel matches responses by cmd_id, so completions only have to be mutually exclusive and must never run ahead of the reader; the in-flight count is released last because device teardown waits on it before freeing the device. Fan-out is load adaptive: each batch starts up to 8 of its commands on the current vCPU, limited further by that vCPU's idle handler slots, and sends only the rest through async_call for pool-wide distribution. This keeps available local capacity busy without holding overflow on a saturated vCPU. enableThread, which turned on one extra thread and refused the file cache, becomes workpoolSize: the number of vCPUs in the pool, 8 by default. main() now releases everything through DEFER, so the image service, the work pool, the tcmulib context and the main loop are also torn down on the error paths, and in the reverse order of their construction: the device loops stop, tcmulib_close() removes the devices and with them the image files and their photon threads, the workers are joined, the image service goes away, and photon::fini() runs last -- it waits for every photon thread of the main vCPU, so nothing may outlive it. The Release benchmark disables DSA and ISA-L in both builds to isolate the TCMU implementation, and is sized to the hosted runner's 4 vCPUs: the backend's overlaybd.json sets workpoolSize to 3, while fio runs a single job that takes all of its concurrency from iodepth, leaving the fourth CPU to fio and the kernel's LIO loopback path instead of oversubscribing it. It reads a range whose allocated file extents cover at least 95% of it, and gates the IOPS ratio against the single-vCPU baseline: at least 95% through a queue depth of 32, then 2x at 64, 4x at 128 and 8x at 256. The comparison table is published as a check-run annotation because job logs and artifacts require repository authentication. Use signed elapsed time in the TCMU and ublk read retry loops so a backward timestamp does not underflow into a spurious seven-day timeout. Signed-off-by: Lanzheng Liu --- .github/scripts/summarize-tcmu-performance.py | 166 ++++++++++ .github/scripts/tcmu-performance.sh | 306 ++++++++++++++++++ .github/workflows/cmake.yml | 78 ++++- .github/workflows/tcmu-performance.yml | 150 +++++++++ README.md | 3 +- src/config.h | 2 +- src/example_config/overlaybd-registryv2.json | 1 + src/example_config/overlaybd.json | 1 + src/image_service.cpp | 4 - src/main.cpp | 222 +++++++------ src/ublk/ublk_device.cpp | 4 +- 11 files changed, 827 insertions(+), 110 deletions(-) create mode 100644 .github/scripts/summarize-tcmu-performance.py create mode 100644 .github/scripts/tcmu-performance.sh create mode 100644 .github/workflows/tcmu-performance.yml diff --git a/.github/scripts/summarize-tcmu-performance.py b/.github/scripts/summarize-tcmu-performance.py new file mode 100644 index 00000000..4cab2237 --- /dev/null +++ b/.github/scripts/summarize-tcmu-performance.py @@ -0,0 +1,166 @@ +#!/usr/bin/env python3 + +import csv +import json +import os +import re +import statistics +import sys +from pathlib import Path + + +def load_runs(result_dir: Path): + rows = [] + for path in sorted(result_dir.glob("*.json")): + match = re.fullmatch(r"(baseline|current)-(.+)-(\d+)", path.stem) + if not match: + continue + label, profile, run = match.groups() + with path.open() as file: + job = json.load(file)["jobs"][0]["read"] + rows.append( + { + "label": label, + "profile": profile, + "run": int(run), + "iops": float(job["iops"]), + "bw_bytes": float(job["bw_bytes"]), + "clat_us": float(job["clat_ns"]["mean"]) / 1000, + } + ) + return rows + + +def median(rows, label, profile, metric): + return statistics.median( + row[metric] + for row in rows + if row["label"] == label and row["profile"] == profile + ) + + +def delta(current, baseline): + return (current / baseline - 1) * 100 + + +result_dir = Path(sys.argv[1]) +result_dir.mkdir(parents=True, exist_ok=True) +rows = load_runs(result_dir) +# One fio job per profile, so its queue depth is the total concurrency. +profiles = ("qd1", "qd8", "qd32", "qd64", "qd128", "qd256") +queue_depth = { + "qd1": 1, + "qd8": 8, + "qd32": 32, + "qd64": 64, + "qd128": 128, + "qd256": 256, +} +# Low depth measures per-command overhead, where the work pool must not cost +# more than the single-vCPU baseline. High depth is where fan-out has to pay +# off: the gate there is a multiple of the baseline, not parity with it. +minimum_baseline_ratio = { + "qd1": 0.95, + "qd8": 0.95, + "qd32": 0.95, + "qd64": 2.00, + "qd128": 4.00, + "qd256": 8.00, +} + + +def threshold_summary(): + """Describe minimum_baseline_ratio without repeating equal thresholds.""" + groups = [] + for profile in profiles: + if profile not in minimum_baseline_ratio: + continue + ratio = minimum_baseline_ratio[profile] + if groups and groups[-1][0] == ratio: + groups[-1][2] = profile + else: + groups.append([ratio, profile, profile]) + return "; ".join( + f"`{first}`-`{last}` >= {ratio:.0%}" if first != last else f"`{first}` >= {ratio:.0%}" + for ratio, first, last in groups + ) + + +failures = [] +# Check-run annotations are public; job logs and artifacts are not. +highlights = [] + +if rows: + with (result_dir / "raw-results.csv").open("w", newline="") as file: + writer = csv.DictWriter(file, fieldnames=rows[0].keys()) + writer.writeheader() + writer.writerows(rows) + +lines = [ + "# OverlayBD TCMU performance comparison", + "", + f"- Baseline: `{os.getenv('BASELINE_SHA', 'unknown')}` with its checked-in configuration", + f"- Current: `{os.getenv('GITHUB_SHA', 'working tree')}` with " + f"`workpoolSize={os.getenv('WORKPOOL_SIZE', 'unknown')}`, per-vCPU command queues and in-place completions", + f"- Build type: `{os.getenv('BUILD_TYPE', 'unknown')}` for both versions", + "- Runner: 4 vCPUs, 3 of them given to the backend's work pool through its `overlaybd.json`", + "- I/O: 4 KiB random reads from the image's largest allocated data extent, `libaio`, `O_DIRECT`; " + "one fio job per profile, all concurrency from `iodepth`, which the profile name gives", + f"- Required current/baseline IOPS: {threshold_summary()}", + "", + "Both versions ran sequentially on the same GitHub runner against the same prewarmed file cache. " + "Values are the median of three 15-second fio runs, each with a 3-second ramp that is not measured.", + "", + "| fio profile | queue depth | baseline IOPS | current IOPS | baseline ratio | IOPS change | baseline latency | current latency | latency change |", + "| --- | ---: | ---: | ---: | ---: | ---: | ---: | ---: | ---: |", +] +for profile in profiles: + baseline_runs = [row for row in rows if row["label"] == "baseline" and row["profile"] == profile] + current_runs = [row for row in rows if row["label"] == "current" and row["profile"] == profile] + if baseline_runs and current_runs: + baseline_iops = median(rows, "baseline", profile, "iops") + current_iops = median(rows, "current", profile, "iops") + baseline_lat = median(rows, "baseline", profile, "clat_us") + current_lat = median(rows, "current", profile, "clat_us") + iops_ratio = current_iops / baseline_iops + lines.append( + f"| `{profile}` | {queue_depth[profile]} | " + f"{baseline_iops:,.0f} | {current_iops:,.0f} | " + f"{iops_ratio:.1%} | {delta(current_iops, baseline_iops):+.1f}% | " + f"{baseline_lat:,.1f} us | " + f"{current_lat:,.1f} us | {delta(current_lat, baseline_lat):+.1f}% |" + ) + highlights.append( + f"{profile} iops {baseline_iops:,.0f} -> {current_iops:,.0f} " + f"baseline ratio {iops_ratio:.2f}, " + f"latency {baseline_lat:,.0f} us -> {current_lat:,.0f} us" + ) + if profile in minimum_baseline_ratio and iops_ratio < minimum_baseline_ratio[profile]: + failures.append( + f"{profile} current/baseline IOPS is {iops_ratio:.1%}; " + f"required >= {minimum_baseline_ratio[profile]:.0%}" + ) + else: + baseline_iops = f"{median(rows, 'baseline', profile, 'iops'):,.0f}" if baseline_runs else "timeout/no data" + current_iops = f"{median(rows, 'current', profile, 'iops'):,.0f}" if current_runs else "timeout/no data" + baseline_lat = f"{median(rows, 'baseline', profile, 'clat_us'):,.1f} us" if baseline_runs else "—" + current_lat = f"{median(rows, 'current', profile, 'clat_us'):,.1f} us" if current_runs else "—" + lines.append( + f"| `{profile}` | {queue_depth[profile]} | " + f"{baseline_iops} | {current_iops} | — | — | " + f"{baseline_lat} | {current_lat} | — |" + ) + failures.append(f"{profile} has incomplete benchmark data") + +summary = "\n".join(lines) + "\n" +(result_dir / "summary.md").write_text(summary) +print(summary) + +if highlights: + # A plain factor, since '%' must be percent-escaped in workflow commands. + print("::notice title=TCMU performance summary::" + "; ".join(highlights)) + +if failures: + for failure in failures: + print(f"::error title=TCMU performance regression::{failure}") + sys.exit(1) diff --git a/.github/scripts/tcmu-performance.sh b/.github/scripts/tcmu-performance.sh new file mode 100644 index 00000000..f6f9f45d --- /dev/null +++ b/.github/scripts/tcmu-performance.sh @@ -0,0 +1,306 @@ +#!/usr/bin/env bash + +set -Eeuo pipefail + +if [[ $# -ne 5 ]]; then + echo "usage: $0