guiwf/lix
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

libstore: use async io for remote store IO

Browse Source 
this means both the worker protocol and the serve protocol, i.e.
ssh-ng/local connections and legacy ssh connections. now we have
no blocking reads left anywhere in our client store connections.

Change-Id: I2f628d4d2e71ef0a7006918f175192f3f58eea95
This commit is contained in:
eldritch horrors
2025-08-06 14:43:47 +00:00
parent db0ed505e9
commit 22cf5eb989
3 changed files with 39 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libstore/legacy-ssh-store.cc
+6 -7
View File
@@ -170,8 +170,9 @@ struct LegacySSHStore final : public Store
}
});
AsyncFdIoStream stream(AsyncFdIoStream::shared_fd{}, sshConn->socket.get());
{
AsyncFdIoStream stream(AsyncFdIoStream::shared_fd{}, sshConn->socket.get());
StringSink buffer;
// can't use TRY_AWAIT here because macros break with variadic templates. sigh.
((co_await sendArg(stream, buffer, std::forward<Args>(args))).value(), ...);
@@ -182,12 +183,10 @@ struct LegacySSHStore final : public Store
invalidateOnCancel.cancel();
co_return result::success();
} else {
// 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{sshConn->socket.get(), fromBuf};
auto result = ServeProto::Serialise<R>::read({from, *store, remoteVersion});
AsyncBufferedInputStream from{stream, fromBuf};
auto result = TRY_AWAIT(ServeProto::readAsync(
from, *store, remoteVersion, ServeProto::Serialise<R>::read
));
invalidateOnCancel.cancel();
co_return result;
}
lix/libstore/remote-store-connection.hh
+7 -10
View File
@@ -10,6 +10,7 @@
#include "lix/libutil/io-buffer.hh"
#include "lix/libutil/pool.hh"
#include "lix/libutil/result.hh"
#include "lix/libutil/serialise-async.hh"
#include "lix/libutil/serialise.hh"
#include "lix/libutil/signals.hh"
#include <kj/async.h>
@@ -143,13 +144,14 @@ struct RemoteStore::ConnectionHandle
}
});
AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, handle->getFD()};
// 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<LastArgT>(); }) {
AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, handle->getFD()};
try {
StringSink msg;
((msg << std::forward<Args>(args)), ...);
@@ -163,7 +165,6 @@ struct RemoteStore::ConnectionHandle
using ImmediateArgsIdxs = std::make_index_sequence<sizeof...(Args) - 1>;
AllArgsT allArgs(std::forward<Args>(args)...);
AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, handle->getFD()};
try {
StringSink msg;
[&]<size_t... Ids>(std::integer_sequence<size_t, Ids...>) {
@@ -186,14 +187,10 @@ struct RemoteStore::ConnectionHandle
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";
auto result =
WorkerProto::Serialise<R>::read({from, *handle->store, handle->daemonVersion});
AsyncBufferedInputStream from{stream, handle->fromBuf};
auto result = LIX_TRY_AWAIT(WorkerProto::readAsync(
from, *handle->store, handle->daemonVersion, WorkerProto::Serialise<R>::read
));
invalidateOnCancel.cancel();
co_return result;
} catch (...) {
lix/libstore/remote-store.cc
+26 -23
View File
@@ -5,6 +5,7 @@
#include "lix/libutil/error.hh"
#include "lix/libutil/file-descriptor.hh"
#include "lix/libutil/result.hh"
#include "lix/libutil/serialise-async.hh"
#include "lix/libutil/serialise.hh"
#include "lix/libutil/signals.hh"
#include "lix/libstore/path-with-outputs.hh"
@@ -83,41 +84,43 @@ try {
kj::Promise<Result<void>> 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};
from.specialEndOfFileError =
"Nix daemon connection broke during setup phase (is it reachable?)";
FdSink to{conn.getFD()};
to << WORKER_MAGIC_1;
to.flush();
AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, conn.getFD()};
AsyncBufferedInputStream from{stream, conn.fromBuf};
uint64_t magic = readNum<uint64_t>(from);
conn.store = this;
{
StringSink packet;
packet << WORKER_MAGIC_1;
TRY_AWAIT(stream.writeFull(packet.s.data(), packet.s.size()));
}
uint64_t magic = TRY_AWAIT(readNum<uint64_t>(from));
if (magic != WORKER_MAGIC_2)
throw Error("protocol mismatch");
conn.daemonVersion = readNum<unsigned>(from);
conn.daemonVersion = TRY_AWAIT(readNum<unsigned>(from));
if (GET_PROTOCOL_MAJOR(conn.daemonVersion) != GET_PROTOCOL_MAJOR(PROTOCOL_VERSION))
throw Error("Nix daemon protocol version not supported");
if (GET_PROTOCOL_MINOR(conn.daemonVersion) < MIN_SUPPORTED_MINOR_WORKER_PROTO_VERSION)
throw Error("The remote Nix daemon version is too old");
to << PROTOCOL_VERSION;
// Obsolete CPU affinity.
to << 0;
{
StringSink packet;
packet << PROTOCOL_VERSION;
packet << 0; // Obsolete CPU affinity.
packet << false; // obsolete reserveSpace
TRY_AWAIT(stream.writeFull(packet.s.data(), packet.s.size()));
}
to << false; // obsolete reserveSpace
conn.daemonNixVersion = TRY_AWAIT(readString(from));
conn.remoteTrustsUs = TRY_AWAIT(WorkerProto::readAsync(
from,
*conn.store,
conn.daemonVersion,
WorkerProto::Serialise<std::optional<TrustedFlag>>::read
));
to.flush();
conn.daemonNixVersion = readString(from);
conn.remoteTrustsUs = WorkerProto::Serialise<std::optional<TrustedFlag>>::read(
{from, *conn.store, conn.daemonVersion}
);
AsyncFdIoStream stream{AsyncFdIoStream::shared_fd{}, conn.getFD()};
auto ex = TRY_AWAIT(conn.processStderr(stream));
if (ex.e) {
std::rethrow_exception(ex.e);