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

libstore: instantiate RemoteStore FdSources as needed

Browse Source 
in the future we will want to instantiate either a sink, a source, both,
or streams, depending on how the fd is used. to do this we need to share
read buffers among sync and async readers. removing the FdSource we kept
in the connection also helps prove that we always use this buffer for io

Change-Id: Ib678e128ed6c4a07d6ce5ec1d3cde9eb3f5fc4ca
This commit is contained in:
eldritch horrors
2025-06-17 14:34:05 +02:00
parent 687ea19e6f
commit 3f62905312
6 changed files with 77 additions and 58 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libstore/remote-store-connection.hh
+16 -21
View File
@@ -6,6 +6,7 @@
#include "lix/libstore/worker-protocol-impl.hh"
#include "lix/libutil/async.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"
@@ -20,20 +21,23 @@ namespace nix {
* Bidirectional connection (send and receive) used by the Remote Store
* implementation.
*
* Contains `Source` and `Sink` for actual communication, along with
* Contains a socket fd and IO buffer for actual communication, along with
* other information learned when negotiating the connection.
*/
struct RemoteStore::Connection
{
/**
* Send with this.
* Receive buffer, shared between sync Sources and async Streams.
* All buffered receiving sources or streams using the `getFD()`
* fd must use this buffer, or they will corrupt the connection.
*/
int toFD;
ref<IoBuffer> fromBuf{make_ref<IoBuffer>()};
/**
* Receive with this.
* Returns the file descriptors socket backing this connection. A
* connection must be backed by a socket, not by a pair of pipes.
*/
std::unique_ptr<FdSource> from;
virtual int getFD() const = 0;
/**
* The store this connection belongs to.
@@ -72,19 +76,6 @@ struct RemoteStore::Connection
*/
std::chrono::time_point<std::chrono::steady_clock> startTime;
/**
* Coercion to `WorkerProto::ReadConn`. This makes it easy to use the
* factored out worker protocol searlizers with a
* `RemoteStore::Connection`.
*
* The worker protocol connection types are unidirectional, unlike
* this type.
*/
operator WorkerProto::ReadConn ()
{
return WorkerProto::ReadConn{*from, *store, daemonVersion};
}
/**
* Coercion to `WorkerProto::WriteConn`. This makes it easy to use the
* factored out worker protocol searlizers with a
@@ -160,7 +151,7 @@ struct RemoteStore::ConnectionHandle
try {
StringSink msg;
((msg << std::forward<Args>(args)), ...);
writeFull(handle->toFD, msg.s);
writeFull(handle->getFD(), msg.s);
} catch (...) {
handle.markBad();
throw;
@@ -177,7 +168,7 @@ struct RemoteStore::ConnectionHandle
std::get<Ids>(allArgs)
)),
...);
writeFull(handle->toFD, msg.s);
writeFull(handle->getFD(), msg.s);
} catch (...) {
handle.markBad();
throw;
@@ -191,7 +182,11 @@ struct RemoteStore::ConnectionHandle
co_return result::success();
} else {
try {
co_return WorkerProto::Serialise<R>::read(*handle);
FdSource from{handle->getFD(), handle->fromBuf};
from.specialEndOfFileError = "Nix daemon disconnected while reading a response";
co_return WorkerProto::Serialise<R>::read(
{from, *handle->store, handle->daemonVersion}
);
} catch (...) {
handle.markBad();
throw;
lix/libstore/remote-store.cc
+49 -30
View File
@@ -80,17 +80,18 @@ try {
/* Send the magic greeting, check for the reply. */
try {
conn.store = this;
conn.from->specialEndOfFileError =
"Nix daemon disconnected unexpectedly (maybe it crashed?)";
FdSink to{conn.toFD};
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();
uint64_t magic = readLongLong(*conn.from);
uint64_t magic = readLongLong(from);
if (magic != WORKER_MAGIC_2)
throw Error("protocol mismatch");
*conn.from >> conn.daemonVersion;
from >> conn.daemonVersion;
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)
@@ -103,8 +104,10 @@ try {
to << false; // obsolete reserveSpace
to.flush();
conn.daemonNixVersion = readString(*conn.from);
conn.remoteTrustsUs = WorkerProto::Serialise<std::optional<TrustedFlag>>::read(conn);
conn.daemonNixVersion = readString(from);
conn.remoteTrustsUs = WorkerProto::Serialise<std::optional<TrustedFlag>>::read(
{from, *conn.store, conn.daemonVersion}
);
auto ex = TRY_AWAIT(conn.processStderr());
if (ex.e) {
@@ -158,7 +161,7 @@ try {
for (auto & i : overrides)
command << i.first << i.second.value;
writeFull(conn.toFD, command.s);
writeFull(conn.getFD(), command.s);
auto ex = TRY_AWAIT(conn.processStderr());
if (ex.e) {
std::rethrow_exception(ex.e);
@@ -705,16 +708,29 @@ try {
kj::Promise<Result<box_ptr<AsyncInputStream>>> RemoteStore::narFromPath(const StorePath & path)
try {
struct NarStream : AsyncInputStream
{
ConnectionHandle conn;
AsyncFdIoStream rawStream;
AsyncBufferedInputStream bufferedIn;
box_ptr<AsyncInputStream> narCopier;
NarStream(ConnectionHandle conn)
: conn(std::move(conn))
, rawStream(AsyncFdIoStream::shared_fd{}, this->conn->getFD())
, bufferedIn(this->rawStream, this->conn->fromBuf)
, narCopier(copyNAR(bufferedIn))
{
}
kj::Promise<Result<size_t>> read(void * buffer, size_t size) override
{
return narCopier->read(buffer, size);
}
};
auto conn(TRY_AWAIT(getConnection()));
TRY_AWAIT(conn.sendCommand(WorkerProto::Op::NarFromPath, printStorePath(path)));
co_return make_box_ptr<AsyncGeneratorInputStream>([](auto conn) -> WireFormatGenerator {
try {
co_yield copyNAR(*conn->from);
} catch (...) {
conn.handle.markBad();
throw;
}
}(std::move(conn)));
co_return make_box_ptr<NarStream>(std::move(conn));
} catch (...) {
co_return result::current_exception();
}
@@ -745,34 +761,37 @@ kj::Promise<Result<RemoteStore::Connection::RemoteError>> RemoteStore::Connectio
try {
while (true) {
auto msg = readNum<uint64_t>(*from);
FdSource from{getFD(), fromBuf};
from.specialEndOfFileError = "Nix daemon disconnected while waiting for a response";
auto msg = readNum<uint64_t>(from);
if (msg == STDERR_ERROR) {
co_return RemoteError{std::make_exception_ptr(readError(*from))};
co_return RemoteError{std::make_exception_ptr(readError(from))};
}
else if (msg == STDERR_NEXT)
printError(chomp(readString(*from)));
printError(chomp(readString(from)));
else if (msg == STDERR_START_ACTIVITY) {
auto act = readNum<ActivityId>(*from);
auto lvl = (Verbosity) readInt(*from);
auto type = (ActivityType) readInt(*from);
auto s = readString(*from);
auto fields = readFields(*from);
auto parent = readNum<ActivityId>(*from);
auto act = readNum<ActivityId>(from);
auto lvl = (Verbosity) readInt(from);
auto type = (ActivityType) readInt(from);
auto s = readString(from);
auto fields = readFields(from);
auto parent = readNum<ActivityId>(from);
logger->startActivity(act, lvl, type, s, fields, parent);
}
else if (msg == STDERR_STOP_ACTIVITY) {
auto act = readNum<ActivityId>(*from);
auto act = readNum<ActivityId>(from);
logger->stopActivity(act);
}
else if (msg == STDERR_RESULT) {
auto act = readNum<ActivityId>(*from);
auto type = (ResultType) readInt(*from);
auto fields = readFields(*from);
auto act = readNum<ActivityId>(from);
auto type = (ResultType) readInt(from);
auto fields = readFields(from);
logger->result(act, type, fields);
}
@@ -823,7 +842,7 @@ kj::Promise<Result<void>> RemoteStore::ConnectionHandle::withFramedSinkAsync(
)
try {
{
FdSink to{handle->toFD};
FdSink to{handle->getFD()};
FramedSinkHandler handler{*this, *handlerThreads.lock()};
FramedSink sink(to, handler.ex);
TRY_AWAIT(fun(sink));
lix/libstore/ssh-store.cc
+5 -2
View File
@@ -70,6 +70,11 @@ protected:
struct Connection : RemoteStore::Connection
{
std::unique_ptr<SSH::Connection> sshConn;
int getFD() const override
{
return sshConn->socket.get();
}
};
ref<RemoteStore::Connection> openConnection() override;
@@ -99,8 +104,6 @@ ref<RemoteStore::Connection> SSHStore::openConnection()
command += " --store " + shellEscape(config_.remoteStore.get());
conn->sshConn = ssh.startCommand(command);
conn->toFD = conn->sshConn->socket.get();
conn->from = std::make_unique<FdSource>(conn->sshConn->socket.get());
return conn;
}
lix/libstore/uds-remote-store.cc
-3
View File
@@ -60,9 +60,6 @@ ref<RemoteStore::Connection> UDSRemoteStore::openConnection()
nix::connect(conn->fd.get(), path ? *path : settings.nixDaemonSocketFile);
conn->from = std::make_unique<FdSource>(conn->fd.get());
conn->toFD = conn->fd.get();
conn->startTime = std::chrono::steady_clock::now();
return conn;
lix/libstore/uds-remote-store.hh
+5
View File
@@ -61,6 +61,11 @@ private:
struct Connection : RemoteStore::Connection
{
AutoCloseFD fd;
int getFD() const override
{
return fd.get();
}
};
ref<RemoteStore::Connection> openConnection() override;
lix/nix/daemon.cc
+2 -2
View File
@@ -409,8 +409,8 @@ static void daemonLoop(AsyncIoRoot & aio, std::optional<TrustedFlag> forceTrustC
*/
static void forwardStdioConnection(RemoteStore & store) {
auto conn = store.openConnectionWrapper();
int from = conn->from->fd;
int to = conn->toFD;
int from = conn->getFD();
int to = conn->getFD();
auto nfds = std::max(from, STDIN_FILENO) + 1;
while (true) {