Add buffered line reads to BufferedSource

src/libstore/gc.cc

@@ -5,6 +5,7 @@
 #include "nix/util/finally.hh"
 #include "nix/util/unix-domain-socket.hh"
 #include "nix/util/signals.hh"
+#include "nix/util/serialise.hh"
 #include "nix/util/util.hh"
 #include "nix/store/posix-fs-canonicalise.hh"
 
@@ -576,9 +577,10 @@ void LocalStore::collectGarbage(const GCOptions & options, GCResults & results)
                     if (fcntl(fdClient.get(), F_SETFL, fcntl(fdClient.get(), F_GETFL) & ~O_NONBLOCK) == -1)
                         panic("Could not set non-blocking flag on client socket");
 
+                    FdSource source(fdClient.get());
                     while (true) {
                         try {
-                            auto path = readLine(fdClient.get());
+                            auto path = source.readLine();
                             auto storePath = maybeParseStorePath(path);
                             if (storePath) {
                                 debug("got new GC root '%s'", path);

src/libstore/unix/build/linux-derivation-builder.cc

@@ -4,6 +4,7 @@
 #  include "nix/util/cgroup.hh"
 #  include "nix/util/linux-namespaces.hh"
 #  include "nix/util/logging.hh"
+#  include "nix/util/serialise.hh"
 #  include "linux/fchmodat2-compat.hh"
 
 #  include <sys/ioctl.h>
@@ -385,7 +386,8 @@ struct ChrootLinuxDerivationBuilder : ChrootDerivationBuilder, LinuxDerivationBu
             userNamespaceSync.writeSide = -1;
         });
 
-        auto ss = tokenizeString<std::vector<std::string>>(readLine(sendPid.readSide.get()));
+        FdSource sendPidSource(sendPid.readSide.get());
+        auto ss = tokenizeString<std::vector<std::string>>(sendPidSource.readLine());
         assert(ss.size() == 1);
         pid = string2Int<pid_t>(ss[0]).value();
 

src/libutil-tests/file-descriptor.cc

@@ -0,0 +1,245 @@
+#include <gtest/gtest.h>
+
+#include "nix/util/file-descriptor.hh"
+#include "nix/util/serialise.hh"
+
+#include <cstring>
+
+namespace nix {
+
+// BufferedSource with configurable small buffer for precise boundary testing.
+struct TestBufferedStringSource : BufferedSource
+{
+    std::string_view data;
+    size_t pos = 0;
+
+    TestBufferedStringSource(std::string_view d, size_t bufSize)
+        : BufferedSource(bufSize)
+        , data(d)
+    {
+    }
+
+protected:
+    size_t readUnbuffered(char * buf, size_t len) override
+    {
+        if (pos >= data.size())
+            throw EndOfFile("end of test data");
+        size_t n = std::min(len, data.size() - pos);
+        std::memcpy(buf, data.data() + pos, n);
+        pos += n;
+        return n;
+    }
+};
+
+TEST(ReadLine, ReadsLinesFromPipe)
+{
+    Pipe pipe;
+    pipe.create();
+
+    writeFull(pipe.writeSide.get(), "hello\nworld\n", /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    EXPECT_EQ(readLine(pipe.readSide.get()), "hello");
+    EXPECT_EQ(readLine(pipe.readSide.get()), "world");
+    EXPECT_EQ(readLine(pipe.readSide.get(), /*eofOk=*/true), "");
+}
+
+TEST(ReadLine, ReturnsPartialLineOnEofWhenAllowed)
+{
+    Pipe pipe;
+    pipe.create();
+
+    writeFull(pipe.writeSide.get(), "partial", /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    EXPECT_EQ(readLine(pipe.readSide.get(), /*eofOk=*/true), "partial");
+    EXPECT_EQ(readLine(pipe.readSide.get(), /*eofOk=*/true), "");
+}
+
+TEST(ReadLine, ThrowsOnEofWhenNotAllowed)
+{
+    Pipe pipe;
+    pipe.create();
+    pipe.writeSide.close();
+
+    EXPECT_THROW(readLine(pipe.readSide.get()), EndOfFile);
+}
+
+TEST(ReadLine, EmptyLine)
+{
+    Pipe pipe;
+    pipe.create();
+
+    writeFull(pipe.writeSide.get(), "\n", /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    EXPECT_EQ(readLine(pipe.readSide.get()), "");
+}
+
+TEST(ReadLine, ConsecutiveEmptyLines)
+{
+    Pipe pipe;
+    pipe.create();
+
+    writeFull(pipe.writeSide.get(), "\n\n\n", /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    EXPECT_EQ(readLine(pipe.readSide.get()), "");
+    EXPECT_EQ(readLine(pipe.readSide.get()), "");
+    EXPECT_EQ(readLine(pipe.readSide.get()), "");
+    EXPECT_EQ(readLine(pipe.readSide.get(), /*eofOk=*/true), "");
+}
+
+TEST(ReadLine, LineWithNullBytes)
+{
+    Pipe pipe;
+    pipe.create();
+
+    std::string data(
+        "a\x00"
+        "b\n",
+        4);
+    writeFull(pipe.writeSide.get(), data, /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    auto line = readLine(pipe.readSide.get());
+    EXPECT_EQ(line.size(), 3);
+    EXPECT_EQ(
+        line,
+        std::string(
+            "a\x00"
+            "b",
+            3));
+}
+
+TEST(BufferedSourceReadLine, ReadsLinesFromPipe)
+{
+    Pipe pipe;
+    pipe.create();
+
+    writeFull(pipe.writeSide.get(), "hello\nworld\n", /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    FdSource source(pipe.readSide.get());
+
+    EXPECT_EQ(source.readLine(), "hello");
+    EXPECT_EQ(source.readLine(), "world");
+    EXPECT_EQ(source.readLine(/*eofOk=*/true), "");
+}
+
+TEST(BufferedSourceReadLine, ReturnsPartialLineOnEofWhenAllowed)
+{
+    Pipe pipe;
+    pipe.create();
+
+    writeFull(pipe.writeSide.get(), "partial", /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    FdSource source(pipe.readSide.get());
+
+    EXPECT_EQ(source.readLine(/*eofOk=*/true), "partial");
+    EXPECT_EQ(source.readLine(/*eofOk=*/true), "");
+}
+
+TEST(BufferedSourceReadLine, ThrowsOnEofWhenNotAllowed)
+{
+    Pipe pipe;
+    pipe.create();
+    pipe.writeSide.close();
+
+    FdSource source(pipe.readSide.get());
+
+    EXPECT_THROW(source.readLine(), EndOfFile);
+}
+
+TEST(BufferedSourceReadLine, EmptyLine)
+{
+    Pipe pipe;
+    pipe.create();
+
+    writeFull(pipe.writeSide.get(), "\n", /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    FdSource source(pipe.readSide.get());
+
+    EXPECT_EQ(source.readLine(), "");
+}
+
+TEST(BufferedSourceReadLine, ConsecutiveEmptyLines)
+{
+    Pipe pipe;
+    pipe.create();
+
+    writeFull(pipe.writeSide.get(), "\n\n\n", /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    FdSource source(pipe.readSide.get());
+
+    EXPECT_EQ(source.readLine(), "");
+    EXPECT_EQ(source.readLine(), "");
+    EXPECT_EQ(source.readLine(), "");
+    EXPECT_EQ(source.readLine(/*eofOk=*/true), "");
+}
+
+TEST(BufferedSourceReadLine, LineWithNullBytes)
+{
+    Pipe pipe;
+    pipe.create();
+
+    std::string data(
+        "a\x00"
+        "b\n",
+        4);
+    writeFull(pipe.writeSide.get(), data, /*allowInterrupts=*/false);
+    pipe.writeSide.close();
+
+    FdSource source(pipe.readSide.get());
+
+    auto line = source.readLine();
+    EXPECT_EQ(line.size(), 3);
+    EXPECT_EQ(
+        line,
+        std::string(
+            "a\x00"
+            "b",
+            3));
+}
+
+TEST(BufferedSourceReadLine, NewlineAtBufferBoundary)
+{
+    // "abc\n" with buf=4: newline is last byte, triggers buffer reset.
+    TestBufferedStringSource source("abc\nxyz\n", 4);
+
+    EXPECT_EQ(source.readLine(), "abc");
+    EXPECT_FALSE(source.hasData());
+    EXPECT_EQ(source.readLine(), "xyz");
+}
+
+TEST(BufferedSourceReadLine, DataRemainsAfterReadLine)
+{
+    // "ab\ncd\n" with buf=8: all fits in buffer, data remains after first read.
+    TestBufferedStringSource source("ab\ncd\n", 8);
+
+    EXPECT_EQ(source.readLine(), "ab");
+    EXPECT_TRUE(source.hasData());
+    EXPECT_EQ(source.readLine(), "cd");
+}
+
+TEST(BufferedSourceReadLine, LineSpansMultipleRefills)
+{
+    // 10-char line with buf=4 requires 3 buffer refills.
+    TestBufferedStringSource source("0123456789\n", 4);
+
+    EXPECT_EQ(source.readLine(), "0123456789");
+}
+
+TEST(BufferedSourceReadLine, BufferExhaustedThenEof)
+{
+    // 8 chars with buf=4: two refills, then EOF with partial line.
+    TestBufferedStringSource source("abcdefgh", 4);
+
+    EXPECT_EQ(source.readLine(/*eofOk=*/true), "abcdefgh");
+    EXPECT_EQ(source.readLine(/*eofOk=*/true), "");
+}
+
+} // namespace nix

src/libutil-tests/meson.build

@@ -59,6 +59,7 @@ sources = files(
   'config.cc',
   'executable-path.cc',
   'file-content-address.cc',
+  'file-descriptor.cc',
   'file-system.cc',
   'git.cc',
   'hash.cc',

src/libutil/include/nix/util/file-descriptor.hh

@@ -76,7 +76,8 @@ void readFull(Descriptor fd, char * buf, size_t count);
 void writeFull(Descriptor fd, std::string_view s, bool allowInterrupts = true);
 
 /**
- * Read a line from a file descriptor.
+ * Read a line from an unbuffered file descriptor.
+ * See BufferedSource::readLine for a buffered variant.
  *
  * @param fd The file descriptor to read from
  * @param eofOk If true, return an unterminated line if EOF is reached. (e.g. the empty string)

src/libutil/include/nix/util/serialise.hh

@@ -120,6 +120,8 @@ struct BufferedSource : virtual Source
 
     size_t read(char * data, size_t len) override;
 
+    std::string readLine(bool eofOk = false);
+
     /**
      * Return true if the buffer is not empty.
      */

src/libutil/serialise.cc

@@ -125,7 +125,7 @@ void Source::skip(size_t len)
 size_t BufferedSource::read(char * data, size_t len)
 {
     if (!buffer)
-        buffer = decltype(buffer)(new char[bufSize]);
+        buffer = std::make_unique_for_overwrite<char[]>(bufSize);
 
     if (!bufPosIn)
         bufPosIn = readUnbuffered(buffer.get(), bufSize);
@@ -139,6 +139,51 @@ size_t BufferedSource::read(char * data, size_t len)
     return n;
 }
 
+std::string BufferedSource::readLine(bool eofOk)
+{
+    if (!buffer)
+        buffer = std::make_unique_for_overwrite<char[]>(bufSize);
+
+    std::string line;
+    while (true) {
+        if (bufPosOut < bufPosIn) {
+            auto * start = buffer.get() + bufPosOut;
+            auto * end = buffer.get() + bufPosIn;
+            if (auto * newline = static_cast<char *>(memchr(start, '\n', end - start))) {
+                line.append(start, newline - start);
+                bufPosOut = (newline - buffer.get()) + 1;
+                if (bufPosOut == bufPosIn)
+                    bufPosOut = bufPosIn = 0;
+                return line;
+            }
+
+            line.append(start, end - start);
+            bufPosOut = bufPosIn = 0;
+        }
+
+        auto handleEof = [&]() -> std::string {
+            bufPosOut = bufPosIn = 0;
+            if (eofOk)
+                return line;
+            throw EndOfFile("unexpected EOF reading a line");
+        };
+
+        size_t n = 0;
+        try {
+            n = readUnbuffered(buffer.get(), bufSize);
+        } catch (EndOfFile & e) {
+            return handleEof();
+        }
+
+        if (n == 0) {
+            return handleEof();
+        }
+
+        bufPosIn = n;
+        bufPosOut = 0;
+    }
+}
+
 bool BufferedSource::hasData()
 {
     return bufPosOut < bufPosIn;