libutil: add io buffer abstraction

the rpc transition will require sync and async objects to share a single
io buffer (since defining serializers on async is an immense pain in the
tail, slow, and ultimately not necessary). a generic buffer class allows
us to reuse existing serializers more readily (reuse them at all, even).

Change-Id: I5ebba8449f26f2bb76016818928183c7e0123be0

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();

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();

lix/libutil/io-buffer.cc

@@ -0,0 +1,38 @@
+#include "io-buffer.hh"
+#include <cassert>
+
+namespace nix {
+
+std::span<const char> 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<char> 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;
+}
+
+}

lix/libutil/io-buffer.hh

@@ -0,0 +1,62 @@
+#pragma once
+///@file IO buffer abstraction for use by buffered IO types.
+
+#include <cstddef>
+#include <memory>
+#include <span>
+
+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<char[]> 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<const char> 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<char> 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);
+};
+
+}

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',

lix/libutil/serialise.cc

@@ -4,8 +4,6 @@
 
 #include <cstring>
 #include <cerrno>
-#include <memory>
-
 
 namespace nix {
 
@@ -52,32 +50,34 @@ template long long readNum<long long>(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;
 }
 
 

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<char[]> 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<char[]> 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;
 

tests/unit/libutil/io-buffer.cc

@@ -0,0 +1,62 @@
+#include "lix/libutil/io-buffer.hh"
+
+#include <gtest/gtest.h>
+
+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);
+}
+
+}

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',