diff --git a/lix/libstore/binary-cache-store.cc b/lix/libstore/binary-cache-store.cc index 85232525c..a05052ca5 100644 --- a/lix/libstore/binary-cache-store.cc +++ b/lix/libstore/binary-cache-store.cc @@ -399,12 +399,12 @@ try { { } - kj::Promise> read(void * buffer, size_t size) override + kj::Promise>> read(void * buffer, size_t size) override { return decompressed->read(buffer, size).then([&](auto r) { if (r.has_value()) { - if (r.value() > 0) { - total += r.value(); + if (r.value()) { + total += *r.value(); } else { stats.narRead++; // stats.narReadCompressedBytes += nar->size(); // FIXME diff --git a/lix/libstore/build/derivation-goal.cc b/lix/libstore/build/derivation-goal.cc index 0349b503c..b93f851d3 100644 --- a/lix/libstore/build/derivation-goal.cc +++ b/lix/libstore/build/derivation-goal.cc @@ -1229,7 +1229,11 @@ try { while (true) { std::string_view data; try { - data = {buf.begin(), TRY_AWAIT(in.read(buf.begin(), buf.size()))}; + if (const auto got = TRY_AWAIT(in.read(buf.begin(), buf.size()))) { + data = {buf.begin(), *got}; + } else { + co_return std::nullopt; + } } catch (SysError & e) { // the builder output stream may be a pty fd, and closing one pty // endpoint sends EIO to the other endpoint. this is a good exit. @@ -1272,13 +1276,14 @@ DerivationGoal::handleHookOutput(AsyncInputStream & in) noexcept try { auto buf = kj::heapArray(4096); while (true) { - std::string_view data = {buf.begin(), TRY_AWAIT(in.read(buf.begin(), buf.size()))}; - lastChildActivity = AIO().provider.getTimer().now(); - - if (data.empty()) { + const auto got = TRY_AWAIT(in.read(buf.begin(), buf.size())); + if (!got) { co_return std::nullopt; } + std::string_view data = {buf.begin(), *got}; + lastChildActivity = AIO().provider.getTimer().now(); + for (auto c : data) if (c == '\n') { auto json = parseJSONMessage(currentHookLine, "the derivation builder"); diff --git a/lix/libstore/filetransfer.cc b/lix/libstore/filetransfer.cc index fdd7b12a0..06ae7c54a 100644 --- a/lix/libstore/filetransfer.cc +++ b/lix/libstore/filetransfer.cc @@ -802,12 +802,17 @@ struct curlFileTransfer : public FileTransfer { AutoCloseFD fd; OwningFdStream(AutoCloseFD fd) : fd(std::move(fd)) {} - kj::Promise> read(void * buffer, size_t size) override + kj::Promise>> + read(void * buffer, size_t size) override { // NOTE the synchronous implementation used to have a buffer for // file data, but we cannot be bothered to treat this edge case. if (const auto got = ::read(fd.get(), buffer, size); got >= 0) { - return {result::success(got)}; + if (got == 0) { + return {result::success(std::nullopt)}; + } else { + return {result::success(got)}; + } } else { return {result::failure(std::make_exception_ptr(SysError("reading file") ))}; @@ -1071,7 +1076,7 @@ struct curlFileTransfer : public FileTransfer co_return result::current_exception(); } - kj::Promise> read(void * buffer, size_t len) override + kj::Promise>> read(void * buffer, size_t len) override try { TRACE(LIX_STORE_FILETRANSFER_READ(uri.c_str(), len)); @@ -1084,7 +1089,11 @@ struct curlFileTransfer : public FileTransfer total += available; } - co_return total; + if (total == 0) { + co_return std::nullopt; + } else { + co_return total; + } } catch (...) { co_return result::current_exception(); } diff --git a/lix/libstore/legacy-ssh-store.cc b/lix/libstore/legacy-ssh-store.cc index 369446747..942f8e637 100644 --- a/lix/libstore/legacy-ssh-store.cc +++ b/lix/libstore/legacy-ssh-store.cc @@ -356,7 +356,7 @@ struct LegacySSHStore final : public Store NarStream(Pool::Handle conn) : conn(std::move(conn)) {} - kj::Promise> read(void * buffer, size_t size) override + kj::Promise>> read(void * buffer, size_t size) override { return copier->read(buffer, size); } diff --git a/lix/libstore/local-store.cc b/lix/libstore/local-store.cc index 307430afb..15106fe59 100644 --- a/lix/libstore/local-store.cc +++ b/lix/libstore/local-store.cc @@ -1211,12 +1211,10 @@ try { free(toRealloc); throw std::bad_alloc(); } - auto got = 0; - Finally cleanup([&]() { - dump = {dumpBuffer.get(), dump.size() + got}; - }); + std::optional got = std::nullopt; + Finally cleanup([&]() { dump = {dumpBuffer.get(), dump.size() + got.value_or(0)}; }); got = TRY_AWAIT(source.read(dumpBuffer.get() + oldSize, want)); - if (got == 0) { + if (!got) { inMemory = true; break; } @@ -1234,7 +1232,7 @@ try { bool useSecond = false; ChainSource(AsyncInputStream & s1, AsyncInputStream & s2) : source1(s1), source2(s2) {} - kj::Promise> read(void * data, size_t len) override + kj::Promise>> read(void * data, size_t len) override try { if (useSecond) { co_return TRY_AWAIT(source2.read(data, len)); diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index aba00c790..354cc9187 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -737,7 +737,7 @@ try { { } - kj::Promise> read(void * buffer, size_t size) override + kj::Promise>> read(void * buffer, size_t size) override { return narCopier->read(buffer, size); } @@ -788,7 +788,12 @@ try { KJ_DEFER(resetBlockingState(getFD(), oldState)); while (fromBuf->used() < sizeof(uint64_t)) { const auto available = fromBuf->getWriteBuffer(); - fromBuf->added(TRY_AWAIT(stream.read(available.data(), available.size()))); + const auto got = TRY_AWAIT(stream.read(available.data(), available.size())); + if (got) { + fromBuf->added(*got); + } else { + throw Error("Nix daemon disconnected while waiting for a response"); + } } } diff --git a/lix/libstore/store-api.cc b/lix/libstore/store-api.cc index 56f1e37f3..8b709f47e 100644 --- a/lix/libstore/store-api.cc +++ b/lix/libstore/store-api.cc @@ -1020,11 +1020,13 @@ struct CopyPathStream : AsyncInputStream { } - kj::Promise> read(void * data, size_t len) override + kj::Promise>> read(void * data, size_t len) override try { auto result = TRY_AWAIT(inner->read(data, len)); - copied += result; - act.progress(copied, expected); + if (result) { + copied += *result; + act.progress(copied, expected); + } co_return result; } catch (...) { co_return result::current_exception(); diff --git a/lix/libutil/archive.cc b/lix/libutil/archive.cc index 6d63abd5a..637225b44 100644 --- a/lix/libutil/archive.cc +++ b/lix/libutil/archive.cc @@ -596,31 +596,32 @@ struct AsyncCopier : AsyncInputStream explicit AsyncCopier(AsyncInputStream & source) : source(source) {} - kj::Promise> read(void * buffer, size_t size) override + kj::Promise>> read(void * buffer, size_t size) override try { while (current.pending == 0) { if (auto want = stream.next()) { current = *want; } else { - co_return 0; + co_return std::nullopt; } } size = std::min(current.pending, size); if (size == 0) { - co_return 0; + co_return std::nullopt; } auto got = TRY_AWAIT(source.read(buffer, size)); - current.pending -= got; - if (got == 0) { + if (!got) { throw badArchive("truncated NAR encountered"); - } else if (!current.pendingFileContents) { - auto end = this->buffer.size(); - this->buffer.resize(end + got); - memcpy(this->buffer.data() + end, buffer, got); } - co_return got; + current.pending -= *got; + if (!current.pendingFileContents) { + auto end = this->buffer.size(); + this->buffer.resize(end + *got); + memcpy(this->buffer.data() + end, buffer, *got); + } + co_return *got; } catch (...) { co_return result::current_exception(); } @@ -673,11 +674,12 @@ struct AsyncParser try { while (n > 0) { auto got = TRY_AWAIT(source.read(buffer, n)); - if (got == 0) { + if (!got) { throw badArchive("unexpected end of nar encountered"); } - buffer += got; - n -= got; + + buffer += *got; + n -= *got; } co_return result::success(); } catch (...) { @@ -814,10 +816,10 @@ try { NarSource(AsyncInputStream & source) : source(source) {} - kj::Promise> read(void * data, size_t len) override + kj::Promise>> read(void * data, size_t len) override try { auto n = TRY_AWAIT(source.read(data, len)); - pos += n; + pos += n.value_or(0); co_return n; } catch (...) { co_return result::current_exception(); diff --git a/lix/libutil/async-io.cc b/lix/libutil/async-io.cc index d4f680e0a..8dbc19ac7 100644 --- a/lix/libutil/async-io.cc +++ b/lix/libutil/async-io.cc @@ -13,7 +13,7 @@ try { constexpr size_t BUF_SIZE = 65536; auto buf = std::make_unique(BUF_SIZE); while (auto r = TRY_AWAIT(read(buf.get(), BUF_SIZE))) { - sink(std::string_view(buf.get(), r)); + sink(std::string_view(buf.get(), *r)); } co_return result::success(); } catch (...) { @@ -25,7 +25,7 @@ try { constexpr size_t BUF_SIZE = 65536; auto buf = std::make_unique(BUF_SIZE); while (auto r = TRY_AWAIT(read(buf.get(), BUF_SIZE))) { - TRY_AWAIT(stream.writeFull(buf.get(), r)); + TRY_AWAIT(stream.writeFull(buf.get(), *r)); } co_return result::success(); } catch (...) { @@ -41,7 +41,7 @@ try { co_return result::current_exception(); } -kj::Promise> AsyncSourceInputStream::read(void * buffer, size_t size) +kj::Promise>> AsyncSourceInputStream::read(void * buffer, size_t size) try { while (true) { if (auto got = inner.read(static_cast(buffer), size); got > 0) { @@ -49,37 +49,45 @@ try { } } } catch (EndOfFile &) { - return {result::success(0)}; + return {result::success(std::nullopt)}; } catch (...) { return {result::current_exception()}; } -kj::Promise> AsyncStringInputStream::read(void * buffer, size_t size) +kj::Promise>> AsyncStringInputStream::read(void * buffer, size_t size) { size = std::min(size, s.size()); - if (size > 0) { - memcpy(buffer, s.data(), size); - s.remove_prefix(size); + if (size == 0) { + return {result::success(std::nullopt)}; } + + memcpy(buffer, s.data(), size); + s.remove_prefix(size); return {result::success(size)}; } -kj::Promise> AsyncTeeInputStream::read(void * buffer, size_t size) +kj::Promise>> AsyncTeeInputStream::read(void * buffer, size_t size) try { auto got = TRY_AWAIT(inner.read(buffer, size)); - sink({static_cast(buffer), got}); + if (got) { + sink({static_cast(buffer), *got}); + } co_return got; } catch (...) { co_return result::current_exception(); } -kj::Promise> AsyncGeneratorInputStream::read(void * data, size_t len) +kj::Promise>> AsyncGeneratorInputStream::read(void * data, size_t len) try { + if (len == 0) { + return {result::success(std::nullopt)}; + } + while (!buf.size()) { if (auto next = g.next()) { buf = *next; } else { - return {result::success(0)}; + return {result::success(std::nullopt)}; } } @@ -91,15 +99,15 @@ try { return {result::current_exception()}; } -kj::Promise> AsyncBufferedInputStream::read(void * data, size_t size) +kj::Promise>> AsyncBufferedInputStream::read(void * data, size_t size) try { while (buffer->used() == 0) { const auto space = buffer->getWriteBuffer(); const auto got = TRY_AWAIT(inner.read(space.data(), space.size())); - if (got == 0) { - co_return 0; + if (!got) { + co_return std::nullopt; } - buffer->added(got); + buffer->added(*got); } const auto available = buffer->getReadBuffer(); @@ -162,11 +170,13 @@ AsyncFdIoStream::~AsyncFdIoStream() noexcept(false) } } -kj::Promise> AsyncFdIoStream::read(void * tgt, size_t size) +kj::Promise>> AsyncFdIoStream::read(void * tgt, size_t size) { auto got = ::read(fd, tgt, size); - if (got >= 0) { + if (got > 0) { return {result::success(got)}; + } else if (got == 0) { + return {result::success(std::nullopt)}; } else if (errno == EAGAIN || errno == EWOULDBLOCK) { return observer.whenBecomesReadable().then([=, this] { return read(tgt, size); }); } else { diff --git a/lix/libutil/async-io.hh b/lix/libutil/async-io.hh index e2074a402..7ff20d767 100644 --- a/lix/libutil/async-io.hh +++ b/lix/libutil/async-io.hh @@ -27,8 +27,8 @@ class AsyncInputStream : private kj::AsyncObject public: virtual ~AsyncInputStream() noexcept(false) {} - // expected to return 0 only on EOF or when `size = 0` was explicitly set. - virtual kj::Promise> read(void * buffer, size_t size) = 0; + // expected to return none only on EOF or when `size = 0` was explicitly set. + virtual kj::Promise>> read(void * buffer, size_t size) = 0; kj::Promise> drainInto(Sink & sink); kj::Promise> drainInto(AsyncOutputStream & stream); @@ -48,7 +48,7 @@ public: AsyncSourceInputStream(Source & inner) : inner(inner) {} AsyncSourceInputStream(box_ptr inner) : inner(*inner), owned(std::move(inner).take()) {} - kj::Promise> read(void * buffer, size_t size) override; + kj::Promise>> read(void * buffer, size_t size) override; }; class AsyncStringInputStream : public AsyncInputStream @@ -58,7 +58,7 @@ class AsyncStringInputStream : public AsyncInputStream public: explicit AsyncStringInputStream(std::string_view s) : s(s) {} - kj::Promise> read(void * buffer, size_t size) override; + kj::Promise>> read(void * buffer, size_t size) override; }; // this writes to sources instead of async streams because none of the sinks @@ -71,7 +71,7 @@ class AsyncTeeInputStream : public AsyncInputStream public: AsyncTeeInputStream(AsyncInputStream & inner, Sink & sink) : inner(inner), sink(sink) {} - kj::Promise> read(void * buffer, size_t size) override; + kj::Promise>> read(void * buffer, size_t size) override; }; class AsyncGeneratorInputStream : public AsyncInputStream @@ -83,7 +83,7 @@ private: public: AsyncGeneratorInputStream(Generator && g) : g(std::move(g)) {} - kj::Promise> read(void * data, size_t len) override; + kj::Promise>> read(void * data, size_t len) override; }; class AsyncBufferedInputStream : public AsyncInputStream @@ -105,7 +105,7 @@ public: KJ_DISALLOW_COPY_AND_MOVE(AsyncBufferedInputStream); - kj::Promise> read(void * data, size_t size) override; + kj::Promise>> read(void * data, size_t size) override; }; class AsyncOutputStream : private kj::AsyncObject @@ -180,7 +180,7 @@ public: return fd; } - kj::Promise> read(void * tgt, size_t size) override; + kj::Promise>> read(void * tgt, size_t size) override; kj::Promise> write(const void * src, size_t size) override; }; diff --git a/lix/libutil/compression.cc b/lix/libutil/compression.cc index 3e2dfbf87..7e63b6fca 100644 --- a/lix/libutil/compression.cc +++ b/lix/libutil/compression.cc @@ -345,8 +345,8 @@ struct DecompressionStream : DecompressorPipes, AsyncInputStream if (buf.used() == 0) { const auto space = buf.getWriteBuffer(); const auto got = TRY_AWAIT(inner->read(space.data(), space.size())); - if (got > 0) { - buf.added(got); + if (got) { + buf.added(*got); } else { co_return; } @@ -371,7 +371,7 @@ struct DecompressionStream : DecompressorPipes, AsyncInputStream feedExc = std::current_exception(); } - kj::Promise> read(void * buffer, size_t size) override + kj::Promise>> read(void * buffer, size_t size) override try { while (true) { if (const auto got = ::read(uncompressed.readSide.get(), buffer, size); got > 0) { @@ -379,7 +379,7 @@ struct DecompressionStream : DecompressorPipes, AsyncInputStream } else if (got == 0) { // decompresser must have finished, poll for any errors and return EOF thread.get(); - co_return 0; + co_return std::nullopt; } else if (errno == EAGAIN || errno == EWOULDBLOCK) { co_await readObserver->whenBecomesReadable(); } else { diff --git a/lix/libutil/file-system.cc b/lix/libutil/file-system.cc index 1b71caeb1..60f2e91ff 100644 --- a/lix/libutil/file-system.cc +++ b/lix/libutil/file-system.cc @@ -450,7 +450,7 @@ try { try { while (true) { if (auto n = TRY_AWAIT(source.read(buf.data(), buf.size()))) { - writeFull(fd.get(), {buf.data(), n}); + writeFull(fd.get(), {buf.data(), *n}); } else { break; } diff --git a/tests/unit/libstore/filetransfer.cc b/tests/unit/libstore/filetransfer.cc index b12baa2be..b1e0b85c6 100644 --- a/tests/unit/libstore/filetransfer.cc +++ b/tests/unit/libstore/filetransfer.cc @@ -243,7 +243,7 @@ TEST(FileTransfer, exceptionAbortsRead) aio.blockOn( aio.blockOn(ft->download(fmt("http://[::1]:%d/index", port))).second->read(buf, 10) ), - 0 + std::nullopt ); } @@ -329,10 +329,11 @@ TEST(FileTransfer, stalledReaderDoesntBlockOthers) size_t dropped = 0; while (size > 0) { auto round = std::min(size, sizeof(buf)); - round = aio.blockOn(source.read(buf, round)); - if (round == 0) { + auto got = aio.blockOn(source.read(buf, round)); + if (!got) { break; } + round = *got; size -= round; dropped += round; } diff --git a/tests/unit/libutil/archive.cc b/tests/unit/libutil/archive.cc index 509c1d23d..274ac3515 100644 --- a/tests/unit/libutil/archive.cc +++ b/tests/unit/libutil/archive.cc @@ -421,7 +421,7 @@ TEST_P(NarTest, parseCopiedAsync) } char buf; - ASSERT_EQ(input.read(&buf, 1).wait(ws).value(), 0); + ASSERT_EQ(input.read(&buf, 1).wait(ws).value(), std::nullopt); } TEST_P(NarTest, index)