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

libstore: use AsyncInputStream for reading builder output

Browse Source 
DerivationGoal::InputStream existed only because we did not have an
error-reporting AsyncInputStream of our own yet. we do have one now
though and can thus delete old code in favor of the generic variant

Change-Id: I01c7c564554f8794bdf54603b239b7a808faeda0
This commit is contained in:
eldritch horrors
2025-07-14 17:02:30 +00:00
parent 334b8e2b20
commit 9c1e7d4d07
2 changed files with 24 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lix/libstore/build/derivation-goal.cc
+21 -42
View File
@@ -1193,45 +1193,23 @@ Goal::WorkResult DerivationGoal::tooMuchLogs()
getName(), settings.maxLogSize));
}
struct DerivationGoal::InputStream final : private kj::AsyncObject
{
int fd;
kj::UnixEventPort::FdObserver observer;
InputStream(kj::UnixEventPort & ep, int fd)
: fd(fd)
, observer(ep, fd, kj::UnixEventPort::FdObserver::OBSERVE_READ)
{
makeNonBlocking(fd);
}
kj::Promise<std::string_view> read(kj::ArrayPtr<char> buffer)
{
const auto res = ::read(fd, buffer.begin(), buffer.size());
// closing a pty endpoint causes EIO on the other endpoint. stock kj streams
// do not handle this and throw exceptions we can't ask for errno instead :(
// (we can't use `errno` either because kj may well have mangled it by now.)
if (res == 0 || (res == -1 && errno == EIO)) {
return std::string_view{};
}
KJ_NONBLOCKING_SYSCALL(res) {}
if (res > 0) {
return std::string_view{buffer.begin(), static_cast<size_t>(res)};
}
return observer.whenBecomesReadable().then([this, buffer] {
return read(buffer);
});
}
};
kj::Promise<Outcome<void, Goal::WorkResult>> DerivationGoal::handleBuilderOutput(InputStream & in) noexcept
kj::Promise<Outcome<void, Goal::WorkResult>>
DerivationGoal::handleBuilderOutput(AsyncInputStream & in) noexcept
try {
auto buf = kj::heapArray<char>(4096);
while (true) {
auto data = co_await in.read(buf);
std::string_view data;
try {
data = {buf.begin(), TRY_AWAIT(in.read(buf.begin(), buf.size()))};
} 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.
if (e.errNo == EIO) {
data = {};
} else {
throw;
}
}
lastChildActivity = AIO().provider.getTimer().now();
if (data.empty()) {
@@ -1260,11 +1238,12 @@ try {
co_return result::current_exception();
}
kj::Promise<Outcome<void, Goal::WorkResult>> DerivationGoal::handleHookOutput(InputStream & in) noexcept
kj::Promise<Outcome<void, Goal::WorkResult>> DerivationGoal::handleHookOutput(AsyncInputStream & in
) noexcept
try {
auto buf = kj::heapArray<char>(4096);
while (true) {
auto data = co_await in.read(buf);
std::string_view data = {buf.begin(), TRY_AWAIT(in.read(buf.begin(), buf.size()))};
lastChildActivity = AIO().provider.getTimer().now();
if (data.empty()) {
@@ -1316,12 +1295,12 @@ try {
kj::Promise<Outcome<void, Goal::WorkResult>> DerivationGoal::handleChildOutput() noexcept
try {
kj::Own<InputStream> builderIn, hookIn;
kj::Own<AsyncInputStream> builderIn, hookIn;
if (builderOutFD) {
builderIn = kj::heap<InputStream>(AIO().unixEventPort, builderOutFD->get());
builderIn = kj::heap<AsyncFdIoStream>(AsyncFdIoStream::shared_fd{}, builderOutFD->get());
}
if (hook) {
hookIn = kj::heap<InputStream>(AIO().unixEventPort, hook->fromHook.get());
hookIn = kj::heap<AsyncFdIoStream>(AsyncFdIoStream::shared_fd{}, hook->fromHook.get());
}
auto handlers = handleChildStreams(builderIn.get(), hookIn.get())
@@ -1362,7 +1341,7 @@ kj::Promise<Outcome<void, Goal::WorkResult>> DerivationGoal::monitorForSilence()
}
kj::Promise<Outcome<void, Goal::WorkResult>>
DerivationGoal::handleChildStreams(InputStream * builderIn, InputStream * hookIn) noexcept
DerivationGoal::handleChildStreams(AsyncInputStream * builderIn, AsyncInputStream * hookIn) noexcept
{
assert(builderIn || hookIn);
lix/libstore/build/derivation-goal.hh
+3 -5
View File
@@ -71,8 +71,6 @@ struct InitialOutput {
*/
struct DerivationGoal : public Goal
{
struct InputStream;
/**
* Whether this goal has completed. Completed goals can not be
* asked for more outputs, a new goal must be created instead.
@@ -317,9 +315,9 @@ protected:
kj::Promise<Outcome<void, WorkResult>> handleChildOutput() noexcept;
kj::Promise<Outcome<void, WorkResult>>
handleChildStreams(InputStream * builderIn, InputStream * hookIn) noexcept;
kj::Promise<Outcome<void, WorkResult>> handleBuilderOutput(InputStream & in) noexcept;
kj::Promise<Outcome<void, WorkResult>> handleHookOutput(InputStream & in) noexcept;
handleChildStreams(AsyncInputStream * builderIn, AsyncInputStream * hookIn) noexcept;
kj::Promise<Outcome<void, WorkResult>> handleBuilderOutput(AsyncInputStream & in) noexcept;
kj::Promise<Outcome<void, WorkResult>> handleHookOutput(AsyncInputStream & in) noexcept;
kj::Promise<Outcome<void, WorkResult>> monitorForSilence() noexcept;
WorkResult tooMuchLogs();
void flushLine();