From c3bc0d35ddf9f4dc774576959d03d3ae41b8d4fa Mon Sep 17 00:00:00 2001 From: eldritch horrors Date: Mon, 16 Jun 2025 18:51:59 +0200 Subject: [PATCH] libstore: use async streams in RemoteStore this is a large step towards making RemoteStore a proper capnp rpc interface, and it lets us get rid of the RemoteStore error handler thread pool. this does mean we make six or more extra syscalls per operation to set and clear socket non-blocking flags, but they are pretty cheap compared to cross-thread wakeups and scheduling. once we have real capnp rpc for store wires we can drop them again too. Change-Id: I67dfebc8644a407cd4a8221ffcad02a938ac5abe --- lix/libstore/remote-store-connection.hh | 86 ++++++++-------- lix/libstore/remote-store.cc | 129 +++++++++++++----------- lix/libstore/remote-store.hh | 8 -- lix/libutil/async-io.cc | 21 ++++ lix/libutil/async-io.hh | 15 +++ lix/libutil/serialise.hh | 49 --------- 6 files changed, 151 insertions(+), 157 deletions(-) diff --git a/lix/libstore/remote-store-connection.hh b/lix/libstore/remote-store-connection.hh index 3ff7bd620..64bccde6d 100644 --- a/lix/libstore/remote-store-connection.hh +++ b/lix/libstore/remote-store-connection.hh @@ -5,11 +5,13 @@ #include "lix/libstore/worker-protocol.hh" #include "lix/libstore/worker-protocol-impl.hh" #include "lix/libutil/async.hh" +#include "lix/libutil/async-io.hh" #include "lix/libutil/file-descriptor.hh" #include "lix/libutil/io-buffer.hh" #include "lix/libutil/pool.hh" #include "lix/libutil/result.hh" #include "lix/libutil/serialise.hh" +#include "lix/libutil/signals.hh" #include #include #include @@ -98,7 +100,7 @@ struct RemoteStore::Connection std::exception_ptr e; }; - kj::Promise> processStderr(); + kj::Promise> processStderr(AsyncFdIoStream & stream); }; /** @@ -111,82 +113,89 @@ struct RemoteStore::Connection struct RemoteStore::ConnectionHandle { Pool::Handle handle; - Sync & handlerThreads; - ConnectionHandle( - Pool::Handle && handle, Sync & handlerThreads - ) - : handle(std::move(handle)) - , handlerThreads(handlerThreads) - { - } - - ConnectionHandle(ConnectionHandle && h) - : handle(std::move(h.handle)) - , handlerThreads(h.handlerThreads) - { - } + ConnectionHandle(Pool::Handle && handle) : handle(std::move(handle)) {} RemoteStore::Connection & operator * () { return *handle; } RemoteStore::Connection * operator -> () { return &*handle; } - kj::Promise> processStderr(); + kj::Promise> processStderr(AsyncFdIoStream & stream); - kj::Promise> - withFramedSinkAsync(std::function>(Sink & sink)> fun); + kj::Promise> withFramedStream( + AsyncFdIoStream & stream, + std::function>(AsyncOutputStream & stream)> fun + ); template - kj::Promise> sendCommand(Args &&... args) + kj::Promise> sendCommandUninterruptible(Args &&... args) try { constexpr auto LastArgIdx = sizeof...(Args) - 1; using AllArgsT = std::tuple; using LastArgT = std::tuple_element_t; + // invalidate this connection if we're cancelled early, e.g. by a user ^C. + // regular exceptions must be handled elsewhere due the subframe requests. + // this also invalidates connections if a request was sent while unwinding + // the stack, but that's sufficiently suspect to warrant being as careful. + auto invalidateOnCancel = kj::defer([&] { + if (std::uncaught_exceptions() == 0) { + handle.markBad(); + } + }); + // if the last argument can be serialized normally we will serialize *all* // arguments at once and hand off to the remote. if the last argument does // not have a serializer we assume it's a callback for a subframe protocol // and serialize all *preceding* arguments normally before handing over to // the subframing layer (which is then responsible for any error handling) if constexpr (requires(StringSink s) { s << std::declval(); }) { + AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, handle->getFD()}; try { StringSink msg; ((msg << std::forward(args)), ...); - writeFull(handle->getFD(), msg.s); + TRY_AWAIT(stream.writeFull(msg.s.data(), msg.s.size())); } catch (...) { handle.markBad(); throw; } - LIX_TRY_AWAIT(processStderr()); + LIX_TRY_AWAIT(processStderr(stream)); } else { using ImmediateArgsIdxs = std::make_index_sequence; AllArgsT allArgs(std::forward(args)...); - [&](std::integer_sequence) { + AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, handle->getFD()}; try { StringSink msg; - ((msg << std::forward>( - std::get(allArgs) - )), - ...); - writeFull(handle->getFD(), msg.s); + [&](std::integer_sequence) { + ((msg << std::forward>( + std::get(allArgs) + )), + ...); + }(ImmediateArgsIdxs{}); + TRY_AWAIT(stream.writeFull(msg.s.data(), msg.s.size())); } catch (...) { handle.markBad(); throw; } - }(ImmediateArgsIdxs{}); - LIX_TRY_AWAIT(withFramedSinkAsync(std::get(allArgs))); + LIX_TRY_AWAIT(withFramedStream(stream, std::get(allArgs))); } if constexpr (std::is_void_v) { + invalidateOnCancel.cancel(); co_return result::success(); } else { try { + // NOTE while no async streams are using the fd it is fully synchronous. + // we need either sync sources or async sources, and async sources would + // require a *lot* of code duplication. response messages are mostly not + // large enough to block us for long, so we just accept the hit for now. FdSource from{handle->getFD(), handle->fromBuf}; from.specialEndOfFileError = "Nix daemon disconnected while reading a response"; - co_return WorkerProto::Serialise::read( - {from, *handle->store, handle->daemonVersion} - ); + auto result = + WorkerProto::Serialise::read({from, *handle->store, handle->daemonVersion}); + invalidateOnCancel.cancel(); + co_return result; } catch (...) { handle.markBad(); throw; @@ -196,16 +205,11 @@ struct RemoteStore::ConnectionHandle co_return result::current_exception(); } -private: - struct FramedSinkHandler + template + kj::Promise> sendCommand(Args &&... args) { - std::exception_ptr ex; - std::packaged_task stderrHandler; - - explicit FramedSinkHandler(ConnectionHandle & conn, ThreadPool & handlerThreads); - - ~FramedSinkHandler() noexcept(false); - }; + return makeInterruptible(sendCommandUninterruptible(std::forward(args)...)); + } }; } diff --git a/lix/libstore/remote-store.cc b/lix/libstore/remote-store.cc index a3f87f3cb..7016423d4 100644 --- a/lix/libstore/remote-store.cc +++ b/lix/libstore/remote-store.cc @@ -1,5 +1,7 @@ +#include "lix/libutil/async-collect.hh" #include "lix/libutil/async-io.hh" #include "lix/libutil/async.hh" +#include "lix/libutil/box_ptr.hh" #include "lix/libutil/error.hh" #include "lix/libutil/result.hh" #include "lix/libutil/serialise.hh" @@ -20,13 +22,14 @@ #include "lix/libutil/logging.hh" #include "lix/libstore/filetransfer.hh" #include "lix/libutil/strings.hh" -#include "lix/libutil/thread-name.hh" -#include "lix/libutil/thread-pool.hh" #include "lix/libutil/types.hh" #include "path-info.hh" +#include #include +#include #include +#include #include namespace nix { @@ -78,6 +81,9 @@ try { kj::Promise> RemoteStore::initConnection(Connection & conn) try { /* Send the magic greeting, check for the reply. */ + // NOTE: this is synchronous until we call processStderr. this is intentional; + // reading the response would be synchronous anyway, and making sending of the + // greeting synchronous also makes this code path significantly more readable. try { conn.store = this; FdSource from{conn.getFD(), conn.fromBuf}; @@ -109,7 +115,8 @@ try { {from, *conn.store, conn.daemonVersion} ); - auto ex = TRY_AWAIT(conn.processStderr()); + AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, conn.getFD()}; + auto ex = TRY_AWAIT(conn.processStderr(stream)); if (ex.e) { std::rethrow_exception(ex.e); } @@ -161,8 +168,9 @@ try { for (auto & i : overrides) command << i.first << i.second.value; - writeFull(conn.getFD(), command.s); - auto ex = TRY_AWAIT(conn.processStderr()); + AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, conn.getFD()}; + TRY_AWAIT(stream.writeFull(command.s.data(), command.s.size())); + auto ex = TRY_AWAIT(conn.processStderr(stream)); if (ex.e) { std::rethrow_exception(ex.e); } @@ -171,9 +179,9 @@ try { co_return result::current_exception(); } -kj::Promise> RemoteStore::ConnectionHandle::processStderr() +kj::Promise> RemoteStore::ConnectionHandle::processStderr(AsyncFdIoStream & stream) try { - auto ex = TRY_AWAIT(handle->processStderr()); + auto ex = TRY_AWAIT(handle->processStderr(stream)); if (ex.e) { co_return result::failure(ex.e); } @@ -185,7 +193,7 @@ try { kj::Promise> RemoteStore::getConnection() try { - co_return ConnectionHandle(TRY_AWAIT(connections->get()), handlerThreads); + co_return ConnectionHandle(TRY_AWAIT(connections->get())); } catch (...) { co_return result::current_exception(); } @@ -372,7 +380,7 @@ try { caMethod.render(hashType), WorkerProto::write(*conn, references), repair, - [&](Sink & sink) { return dump.drainInto(sink); } + [&](AsyncOutputStream & stream) { return dump.drainInto(stream); } ))); } catch (...) { co_return result::current_exception(); @@ -417,7 +425,7 @@ try { renderContentAddress(info.ca), repair, !checkSigs, - [&](Sink & sink) { return copier->drainInto(sink); } + [&](AsyncOutputStream & stream) { return copier->drainInto(stream); } )); co_return result::success(); } catch (...) { @@ -439,14 +447,20 @@ try { repair, !checkSigs, // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) - [&](Sink & sink) -> kj::Promise> { + [&](AsyncOutputStream & stream) -> kj::Promise> { try { - sink << pathsToCopy.size(); + auto send = [&](T && value) { + auto tmp = make_box_ptr(); + *tmp << std::forward(value); + return stream.writeFull(tmp->s.data(), tmp->s.size()).attach(std::move(tmp)); + }; + + TRY_AWAIT(send(pathsToCopy.size())); for (auto & [pathInfo, pathSource] : pathsToCopy) { - sink << WorkerProto::Serialise::write( + TRY_AWAIT(send(WorkerProto::Serialise::write( WorkerProto::WriteConn{*this, remoteVersion}, pathInfo - ); - TRY_AWAIT(TRY_AWAIT(pathSource())->drainInto(sink)); + ))); + TRY_AWAIT(TRY_AWAIT(pathSource())->drainInto(stream)); } co_return result::success(); } catch (...) { @@ -664,7 +678,7 @@ try { TRY_AWAIT(conn.sendCommand( WorkerProto::Op::AddBuildLog, drvPath.to_string(), - [&](Sink & sink) { return source.drainInto(sink); } + [&](AsyncOutputStream & stream) { return source.drainInto(stream); } )); co_return result::success(); } catch (...) { @@ -757,9 +771,28 @@ static Logger::Fields readFields(Source & from) return fields; } -kj::Promise> RemoteStore::Connection::processStderr() +kj::Promise> +RemoteStore::Connection::processStderr(AsyncFdIoStream & stream) try { while (true) { + // fill the read buffer asynchronously until we have at least a message type. + // once we have this we'll continue synchronously as in the sendCommand case. + while (fromBuf->used() < sizeof(uint64_t)) { + const auto available = fromBuf->getWriteBuffer(); + fromBuf->added(TRY_AWAIT(stream.read(available.data(), available.size()))); + } + + // SAFETY NOTE: while we're running we own the executor, and thus the stream. + // setting these flags is unsafe if the stream is shared with another thread. + const int oldFlags = fcntl(getFD(), F_GETFL); + if (oldFlags == -1 || fcntl(getFD(), F_SETFL, oldFlags & ~O_NONBLOCK) < 0) { + throw SysError("making connection blocking"); + } + KJ_DEFER({ + if (fcntl(getFD(), F_SETFL, oldFlags) < 0) { + throw SysError("restoring connection flags"); + } + }); FdSource from{getFD(), fromBuf}; from.specialEndOfFileError = "Nix daemon disconnected while waiting for a response"; @@ -807,52 +840,30 @@ try { co_return result::current_exception(); } -RemoteStore::ConnectionHandle::FramedSinkHandler::FramedSinkHandler( - ConnectionHandle & conn, ThreadPool & handlerThreads -) - : stderrHandler([&](AsyncIoRoot & aio) { - try { - aio.blockOn(conn.processStderr()); - } catch (...) { - ex = std::current_exception(); - } - }) -{ - handlerThreads.enqueueWithAio([&](AsyncIoRoot & aio) { stderrHandler(aio); }); -} - -RemoteStore::ConnectionHandle::FramedSinkHandler::~FramedSinkHandler() noexcept(false) -{ - stderrHandler.get_future().get(); - // if we're handling an Interrupted exception we must be careful: it's - // possible that the exception was thrown by the withFramedSink framed - // function, but not by the FramedSink itself. in this case our stderr - // handler thread may race with FramedSink::writeUnbuffered, catch the - // Interrupted exception independently, store it into ex, and have our - // own destructor rethrow a second copy of Interrupted. since we can't - // handle multiple exceptions anyway the safest path is to simply drop - // the remote (possibly Interrupted) exception when called for unwind. - if (ex && std::uncaught_exceptions() == 0) { - throw FramedSink::RemoteError(ex); - } -} - -kj::Promise> RemoteStore::ConnectionHandle::withFramedSinkAsync( - std::function>(Sink & sink)> fun +kj::Promise> RemoteStore::ConnectionHandle::withFramedStream( + AsyncFdIoStream & stream, + std::function>(AsyncOutputStream & stream)> fun ) try { - { - FdSink to{handle->getFD()}; - FramedSinkHandler handler{*this, *handlerThreads.lock()}; - FramedSink sink(to, handler.ex); - TRY_AWAIT(fun(sink)); - sink.flush(); - } + AsyncBufferedOutputStream to(stream); + AsyncFramedStream sink(to); + + // NOLINTNEXTLINE(cppcoreguidelines-avoid-capturing-lambda-coroutines) + auto send = [&]() -> kj::Promise> { + try { + TRY_AWAIT(fun(sink)); + TRY_AWAIT(sink.finish()); + TRY_AWAIT(to.flush()); + co_return result::success(); + } catch (...) { + handle.markBad(); + co_return result::current_exception(); + } + }; + + TRY_AWAIT(asyncJoin(send(), processStderr(stream))); co_return result::success(); -} catch (FramedSink::RemoteError & e) { - co_return result::failure(e.e); } catch (...) { - handle.markBad(); co_return result::current_exception(); } } diff --git a/lix/libstore/remote-store.hh b/lix/libstore/remote-store.hh index e76c6929c..415a047a9 100644 --- a/lix/libstore/remote-store.hh +++ b/lix/libstore/remote-store.hh @@ -9,7 +9,6 @@ #include "lix/libstore/gc-store.hh" #include "lix/libstore/log-store.hh" #include "lix/libutil/async-io.hh" -#include "lix/libutil/thread-pool.hh" #include "lix/libutil/types.hh" @@ -204,13 +203,6 @@ private: std::atomic_bool failed{false}; - // NOTE we rely on the thread pool not starting threads eagerly. if it ever starts - // doing that we're certainly going to fail due to the immense thread count, which - // we need to satisfy temporary `incCapacity` calls by some RemoteStore functions. - Sync handlerThreads{ - std::in_place, "remote stderr", std::numeric_limits::max() - }; - kj::Promise> copyDrvsFromEvalStore( const std::vector & paths, std::shared_ptr evalStore); diff --git a/lix/libutil/async-io.cc b/lix/libutil/async-io.cc index 7488c3a62..9028c2070 100644 --- a/lix/libutil/async-io.cc +++ b/lix/libutil/async-io.cc @@ -190,4 +190,25 @@ kj::Promise> AsyncFdIoStream::write(const void * src, size_t size return {result::failure(std::make_exception_ptr(SysError(errno, "write failed")))}; } } + +kj::Promise> AsyncFramedStream::finish() +try { + StringSink tmp; + tmp << 0; + TRY_AWAIT(to.writeFull(tmp.s.data(), tmp.s.size())); + co_return result::success(); +} catch (...) { + co_return result::current_exception(); +} + +kj::Promise> AsyncFramedStream::write(const void * buffer, size_t size) +try { + StringSink tmp; + tmp << size; + TRY_AWAIT(to.writeFull(tmp.s.data(), tmp.s.size())); + TRY_AWAIT(to.writeFull(buffer, size)); + co_return size; +} catch (...) { + co_return result::current_exception(); +} } diff --git a/lix/libutil/async-io.hh b/lix/libutil/async-io.hh index 680307ca5..31ccec42a 100644 --- a/lix/libutil/async-io.hh +++ b/lix/libutil/async-io.hh @@ -182,4 +182,19 @@ public: kj::Promise> read(void * tgt, size_t size) override; kj::Promise> write(const void * src, size_t size) override; }; + +/** + * Write as chunks in the format expected by FramedSource. + */ +class AsyncFramedStream : public AsyncOutputStream +{ + AsyncOutputStream & to; + +public: + explicit AsyncFramedStream(AsyncOutputStream & to) : to(to) {} + + kj::Promise> finish(); + + kj::Promise> write(const void * src, size_t size) override; +}; } diff --git a/lix/libutil/serialise.hh b/lix/libutil/serialise.hh index 6cbe94670..58d9f436f 100644 --- a/lix/libutil/serialise.hh +++ b/lix/libutil/serialise.hh @@ -544,53 +544,4 @@ struct FramedSource : Source return n; } }; - -/** - * Write as chunks in the format expected by FramedSource. - * - * The exception_ptr reference can be used to terminate the stream when you - * detect that an error has occurred on the remote end. - */ -struct FramedSink : nix::BufferedSink -{ - BufferedSink & to; - std::exception_ptr & ex; - - struct RemoteError : BaseException - { - std::exception_ptr e; - - RemoteError(std::exception_ptr e) - : e(e) // NOLINT(bugprone-throw-keyword-missing): intentional copy - { - } - }; - - FramedSink(BufferedSink & to, std::exception_ptr & ex) : to(to), ex(ex) - { } - - ~FramedSink() - { - try { - to << 0; - to.flush(); - } catch (...) { - ignoreExceptionInDestructor(); - } - } - - void writeUnbuffered(std::string_view data) override - { - /* Don't send more data if the remote has - encountered an error. */ - if (ex) { - auto ex2 = ex; - ex = nullptr; - throw RemoteError{ex2}; - } - to << data.size(); - to(data); - }; -}; - }