diff --git a/lix/libutil/hash.cc b/lix/libutil/hash.cc index af5728914..d4771a177 100644 --- a/lix/libutil/hash.cc +++ b/lix/libutil/hash.cc @@ -333,11 +333,6 @@ HashSink::HashSink(HashType ht) : ht(ht), ctx(start(ht)) bytes = 0; } -HashSink::~HashSink() -{ - bufPos = 0; -} - void HashSink::writeUnbuffered(std::string_view data) { bytes += data.size(); diff --git a/lix/libutil/hash.hh b/lix/libutil/hash.hh index 225a0da77..7e8910ed0 100644 --- a/lix/libutil/hash.hh +++ b/lix/libutil/hash.hh @@ -208,7 +208,6 @@ private: public: HashSink(HashType ht); HashSink(const HashSink & h); - ~HashSink(); void writeUnbuffered(std::string_view data) override; HashResult finish() override; HashResult currentHash(); diff --git a/lix/libutil/io-buffer.cc b/lix/libutil/io-buffer.cc new file mode 100644 index 000000000..bab3d45af --- /dev/null +++ b/lix/libutil/io-buffer.cc @@ -0,0 +1,38 @@ +#include "io-buffer.hh" +#include + +namespace nix { + +std::span IoBuffer::getReadBuffer() +{ + const auto used = std::min(bufSize - bufBegin, bufUsed); + return {buffer.get() + bufBegin, used}; +} + +void IoBuffer::consumed(size_t size) +{ + assert(size <= bufUsed); + bufBegin = (bufBegin + size) % bufSize; + bufUsed -= size; + if (bufUsed == 0) { + bufBegin = 0; + } +} + +std::span IoBuffer::getWriteBuffer() +{ + if (!buffer) { + buffer.reset(new char[bufSize]); + } + const auto bufEnd = (bufBegin + bufUsed) % bufSize; + const auto free = std::min(bufSize - bufEnd, bufSize - bufUsed); + return {buffer.get() + bufEnd, free}; +} + +void IoBuffer::added(size_t size) +{ + assert(size <= bufSize - bufUsed); + bufUsed += size; +} + +} diff --git a/lix/libutil/io-buffer.hh b/lix/libutil/io-buffer.hh new file mode 100644 index 000000000..0c4414658 --- /dev/null +++ b/lix/libutil/io-buffer.hh @@ -0,0 +1,62 @@ +#pragma once +///@file IO buffer abstraction for use by buffered IO types. + +#include +#include +#include + +namespace nix { + +/** + * A single-threaded read/write IO buffer of fixed size. Adding data to the write side + * of the buffer makes it available to the read buffer, consuming it from the read side + * makes it available for future writes. Once the buffer is full no further data may be + * added, once it is empty no further data can be removed. + */ +class IoBuffer +{ + size_t bufSize, bufBegin{0}, bufUsed{0}; + std::unique_ptr buffer; + +public: + explicit IoBuffer(size_t bufSize = 32 * 1024) : bufSize(bufSize) {} + + size_t size() const + { + return bufSize; + } + + size_t used() const + { + return bufUsed; + } + + /** + * Return a subspan of the buffer that contains valid data. The returned + * span might not cover the entire buffer if `used() > 0`. All reads must be + * followed by a call to `consumed()` to remove bytes from the read buffer + * and make them available for use by the write buffer. + */ + std::span getReadBuffer(); + + /** + * Mark the first `size` bytes of the read buffer as consumed. `size` may + * not exceed `used()`, but may exceed `getReadBuffer().size()`. + */ + void consumed(size_t size); + + /** + * Return a subspan of the buffer that may be written into. The returned + * span may not cover the entire buffer if `used() > 0`. All writes must be + * followed by calls to `added(n)` to mark the written bytes as readable. + */ + std::span getWriteBuffer(); + + /** + * Mark the first `size` bytes of the write buffer as readable. `size` may + * not exceed `size() - used()`, but may exceed `getWriteBuffer().size()`. + */ + void added(size_t size); +}; + +} diff --git a/lix/libutil/meson.build b/lix/libutil/meson.build index ea8c3b051..4881e56dd 100644 --- a/lix/libutil/meson.build +++ b/lix/libutil/meson.build @@ -22,6 +22,7 @@ libutil_sources = files( 'git.cc', 'hash.cc', 'hilite.cc', + 'io-buffer.cc', 'json-utils.cc', 'logging.cc', 'monitor-fd.cc', @@ -94,6 +95,7 @@ libutil_headers = files( 'hash.hh', 'hilite.hh', 'input-accessor.hh', + 'io-buffer.hh', 'json-fwd.hh', 'json.hh', 'linear-map.hh', diff --git a/lix/libutil/serialise.cc b/lix/libutil/serialise.cc index 735fe4fc5..a97087f4b 100644 --- a/lix/libutil/serialise.cc +++ b/lix/libutil/serialise.cc @@ -4,8 +4,6 @@ #include #include -#include - namespace nix { @@ -52,32 +50,34 @@ template long long readNum(Source & source); void BufferedSink::operator () (std::string_view data) { - if (!buffer) buffer = decltype(buffer)(new char[bufSize]); - while (!data.empty()) { /* Optimisation: bypass the buffer if the data exceeds the buffer size. */ - if (bufPos + data.size() >= bufSize) { + if (buffer.used() + data.size() >= buffer.size()) { flush(); writeUnbuffered(data); break; } /* Otherwise, copy the bytes to the buffer. Flush the buffer when it's full. */ - size_t n = bufPos + data.size() > bufSize ? bufSize - bufPos : data.size(); - memcpy(buffer.get() + bufPos, data.data(), n); - data.remove_prefix(n); bufPos += n; - if (bufPos == bufSize) flush(); + auto into = buffer.getWriteBuffer(); + size_t n = std::min(data.size(), into.size()); + memcpy(into.data(), data.data(), n); + data.remove_prefix(n); + buffer.added(n); + if (buffer.used() == buffer.size()) { + flush(); + } } } - void BufferedSink::flush() { - if (bufPos == 0) return; - size_t n = bufPos; - bufPos = 0; // don't trigger the assert() in ~BufferedSink() - writeUnbuffered({buffer.get(), n}); + if (buffer.used() > 0) { + auto from = buffer.getReadBuffer(); + writeUnbuffered({from.data(), from.size()}); + buffer.consumed(from.size()); + } } @@ -136,25 +136,24 @@ std::string Source::drain() return std::move(s.s); } - size_t BufferedSource::read(char * data, size_t len) { - if (!buffer) buffer = decltype(buffer)(new char[bufSize]); + if (buffer.used() == 0) { + auto into = buffer.getWriteBuffer(); + buffer.added(readUnbuffered(into.data(), into.size())); + } - if (!bufPosIn) bufPosIn = readUnbuffered(buffer.get(), bufSize); - - /* Copy out the data in the buffer. */ - size_t n = len > bufPosIn - bufPosOut ? bufPosIn - bufPosOut : len; - memcpy(data, buffer.get() + bufPosOut, n); - bufPosOut += n; - if (bufPosIn == bufPosOut) bufPosIn = bufPosOut = 0; - return n; + auto from = buffer.getReadBuffer(); + len = std::min(len, from.size()); + memcpy(data, from.data(), len); + buffer.consumed(len); + return len; } bool BufferedSource::hasData() { - return bufPosOut < bufPosIn; + return buffer.used() > 0; } diff --git a/lix/libutil/serialise.hh b/lix/libutil/serialise.hh index 9674aa61d..574510d75 100644 --- a/lix/libutil/serialise.hh +++ b/lix/libutil/serialise.hh @@ -5,6 +5,7 @@ #include "lix/libutil/charptr-cast.hh" #include "lix/libutil/generator.hh" +#include "lix/libutil/io-buffer.hh" #include "lix/libutil/types.hh" #include "lix/libutil/file-descriptor.hh" @@ -43,11 +44,9 @@ struct FinishSink : virtual Sink */ struct BufferedSink : virtual Sink { - size_t bufSize, bufPos; - std::unique_ptr buffer; + IoBuffer buffer; - BufferedSink(size_t bufSize = 32 * 1024) - : bufSize(bufSize), bufPos(0), buffer(nullptr) { } + BufferedSink(size_t bufSize = 32 * 1024) : buffer(bufSize) {} void operator () (std::string_view data) override; @@ -92,18 +91,15 @@ struct Source std::string drain(); }; - /** * A buffered abstract source. Warning: a BufferedSource should not be * used from multiple threads concurrently. */ struct BufferedSource : Source { - size_t bufSize, bufPosIn, bufPosOut; - std::unique_ptr buffer; + IoBuffer buffer; - BufferedSource(size_t bufSize = 32 * 1024) - : bufSize(bufSize), bufPosIn(0), bufPosOut(0), buffer(nullptr) { } + BufferedSource(size_t bufSize = 32 * 1024) : buffer(bufSize) {} size_t read(char * data, size_t len) override; diff --git a/tests/unit/libutil/io-buffer.cc b/tests/unit/libutil/io-buffer.cc new file mode 100644 index 000000000..f6b19a6e0 --- /dev/null +++ b/tests/unit/libutil/io-buffer.cc @@ -0,0 +1,62 @@ +#include "lix/libutil/io-buffer.hh" + +#include + +namespace nix { + +TEST(IoBuffer, works) +{ + IoBuffer buf{8}; + + // empty buffer doesn't return anything + ASSERT_EQ(buf.used(), 0); + ASSERT_EQ(buf.getReadBuffer().size(), 0); + + // write a bit, it's no longer empty + ASSERT_EQ(buf.getWriteBuffer().size(), 8); + memcpy(buf.getWriteBuffer().data(), "test", 5); + buf.added(5); + + // five bytes available now + ASSERT_EQ(buf.used(), 5); + ASSERT_EQ(buf.getReadBuffer().size(), 5); + ASSERT_STREQ(buf.getReadBuffer().data(), "test"); + buf.consumed(5); + ASSERT_EQ(buf.used(), 0); + + // write buffer resets to start of buffer when empty + ASSERT_EQ(buf.getWriteBuffer().size(), 8); + + // not adding anything does nothing to the buffer + ASSERT_EQ(buf.used(), 0); + + // buffer can wrap around the end, but in two segments + ASSERT_EQ(buf.getWriteBuffer().size(), 8); + memcpy(buf.getWriteBuffer().data(), "test", 5); + buf.added(5); + ASSERT_EQ(buf.getReadBuffer().size(), 5); + buf.consumed(4); + ASSERT_EQ(buf.getWriteBuffer().size(), 3); + memcpy(buf.getWriteBuffer().data(), "12", 3); + buf.added(3); + ASSERT_EQ(buf.getWriteBuffer().size(), 4); + memcpy(buf.getWriteBuffer().data(), "345", 4); + buf.added(4); + + // reading now also happens in two chunks + ASSERT_EQ(buf.used(), 8); + ASSERT_EQ(buf.getReadBuffer().size(), 4); + ASSERT_STREQ(buf.getReadBuffer().data(), ""); + buf.consumed(1); + ASSERT_STREQ(buf.getReadBuffer().data(), "12"); + buf.consumed(3); + ASSERT_EQ(buf.used(), 4); + ASSERT_STREQ(buf.getReadBuffer().data(), "345"); + buf.consumed(4); + + // buffer is now empty again + ASSERT_EQ(buf.used(), 0); + ASSERT_EQ(buf.getWriteBuffer().size(), 8); +} + +} diff --git a/tests/unit/meson.build b/tests/unit/meson.build index 957fe1f5b..ceba67704 100644 --- a/tests/unit/meson.build +++ b/tests/unit/meson.build @@ -58,6 +58,7 @@ libutil_tests_sources = files( 'libutil/git.cc', 'libutil/hash.cc', 'libutil/hilite.cc', + 'libutil/io-buffer.cc', 'libutil/json-utils.cc', 'libutil/linear-map.cc', 'libutil/logging.cc',