Compare commits

...
Author SHA1 Message Date
eldritch horrors e0beb9b118 [WIP] remove boost coroutine uses
we no longer need these since sinkToSource and sourceToSink are gone.

Change-Id: Ibbf440e2cf71bf3e9f3b833af2d78a21fb1b3193
2024-06-30 21:21:58 +02:00
eldritch horrors f274bfddad [WIP] remove sinkToSource and related boehm patch
Change-Id: I1379841299713175d0225b82a67f50660f9eb5e2
2024-06-30 21:21:58 +02:00
eldritch horrors d08787e42e [WIP] remove remaining sinkToSource uses
Change-Id: Id1ee0d2ad4a3774f4bbb960d76f0f76ac4f3eff9
2024-06-30 21:21:58 +02:00
eldritch horrors 0a1520c602 [WIP] generatorize protocol serializers
this is cursed. deeply and profoundly cursed. under NO CIRCUMSTANCES
must protocol serializer helpers be applied to temporaries! doing so
will inevitably cause dangling references and cause the entire thing
to crash. we need to do this even so to get rid of boost coroutines,
and likewise to encapsulate the serializers we suffer today at least
a little bit to allow a gradual migration to an actual IPC protocol.

Change-Id: I2921ba451e04d86798752d140885d3c5cc08e146
2024-06-30 21:21:58 +02:00
eldritch horrors 5264d72846 [WIP] narFromPath as generator
Change-Id: Ifa783c2c65c06ddd1d0212016d5bfd07666ea91c
2024-06-30 21:21:58 +02:00
eldritch horrors 8fb3b89d0f [WIP] libstore: remove a sinkToSouce from old daemon protocol
this doesn't have a test because this code path is only reached by
clients that predate 2.4, and we really should not be caring about
those any more right now. even the test suite doesn't, and the few
tests that might care are disabled because they will not even work

Change-Id: Id9eb190065138fedb2c7d90c328ff9eb9d97385b
2024-06-30 21:21:58 +02:00
eldritch horrors 425f3e6964 [WIP] libstore: rewrite the nar parser as a contents generator
this is not completely necessary at this point because the parser right
now already returns a generator to pass through all input data it read,
but the nar parser *was* very lax and would accept nars that weren't in
canonical form (defined as the form dumpPath would return). nar hashing
depends on these things, and as such rewriting the parser now allows us
to reject non-canonical nars that extract to the same store contents as
their canonical counterpart, but have different nar hashes.

gcc 12 is broken and causes double frees when yielding certain types of
move-only values, so this also pins gcc to 13 for the default stdenv in
23.11-based builds.

Change-Id: Iccd319e3bd5912d8297014c84c495edc59019bb7
2024-06-30 21:21:58 +02:00
eldritch horrors e11945065b libstore: turn copyNAR into a generator
Change-Id: Id452f6a03faa1037ff13af0f63e32883966ff40d
2024-06-30 21:21:58 +02:00
eldritch horrors 2112a39d2e libstore: turn the NAR parser into a passthrough generator
this will let us turn copyNAR into a generator as well, which in turn is
necessary to turn the users of copyNAR into generators without resorting
to sinkToSource coroutines. currently this uses the SerializingTransform
in all cases, even for copyNAR where it is not necessary. should this be
a performance problem we can easily swap out the transform for one which
does not produce any bytes of its own, but that should not be necessary.

Change-Id: I7e685879318fcbb78d8b88abfddd7752360eb0ce
2024-06-30 21:21:58 +02:00
eldritch horrors 48e3637aa3 libutil: remove sourceToSink
Change-Id: I2867ba2b3a7b59af44c1ec51b08329fccba41ca1
2024-06-30 21:21:58 +02:00
eldritch horrors d658cf5a0e libutil: remove makeDecompressionSink
Change-Id: I4258227b5dbbb735a75b477d8a57007bfca305e9
2024-06-30 21:21:58 +02:00
eldritch horrors 93344a1379 libstore: make BinaryCacheStore::getFile return a source
this lets us remove the last true remaining uses of
makeDecompressionSink.

Change-Id: I146ca2bbe1a9ae9a367117a7b8a304b23a63e5e2
2024-06-30 21:21:58 +02:00
eldritch horrors 9f769794d9 libutil: rewrite RewritingSink as source
the rewriting sink was just broken. when given a rewrite set that
contained a key that is also a proper infix of another key it was
possible to produce an incorrectly rewritten result if the writer
used the wrong block size. fixing this duplicates rewriteStrings,
to avoid this we'll rewrite rewriteStrings to use RewritingSource
in a new mode that'll allow rewrites we had previously forbidden.

Change-Id: I57fa0a9a994e654e11d07172b8e31d15f0b7e8c0
2024-06-30 21:21:58 +02:00
eldritch horrors d1230c7218 libutil: remove RewritingSink match/size tracking
size tracking can be done with a LengthSink and a tee. match tracking
was defeated by never having done any match tracking, all users would
see the same (empty) set of matches at all times. match tracking with
bytes offsets alone would not be sufficient in the general case, only
because computeHashModulo uses a single rewrite would it have worked.

Change-Id: Idb214b5222e0ea24f450f5505712a342b63d7570
2024-06-30 21:21:45 +02:00
eldritch horrors 50e1364a80 libutil: turn HashModuloSink into a free function
Change-Id: I5878007502fa68c2816a0f4c61f7d0e60bdde702
2024-06-30 21:08:17 +02:00
eldritch horrors 3e29a98903 libutil: return sources from runProgram2
this much more closely mimics what is actually happening: we're reading
data from somewhere else, actively, rather than passively waiting. with
the data flow matching the underlying system interactions better we can
remove a few sinkToSource calls that merely exists to undo the mismatch
caused by not treating subprocess output as a data source to begin with

Change-Id: If4abfc2f8398fb5e88c9b91a8bdefd5504bb2d11
2024-06-30 21:05:06 +02:00
eldritch horrors 14d539a274 libutil: return a program handle from runProgram2
this will let us also return a source for the program output later,
which will in turn make sinkToSource unnecessary for program output
processing. this may also reopen a path for provigin program input,
but that still needs a proper async io framework to avoid problems.

Change-Id: Iaf93f47db99c38cfaf134bd60ed6a804d7ddf688
2024-06-30 20:58:02 +02:00
eldritch horrors a5ae26726b enable -Werror=unused-result
we do not have any of these warnings appearing at the moment, but it
seems like a good idea to enable [[nodiscard]] checking anyway. once
we start introducing more functions with must-use conditions we will
need such checking, and the rust stdlib has proven them very useful.

Change-Id: Ibb6b042ae1ec5f527f8dc2809a7816a4c1548ae2
2024-06-30 20:51:10 +02:00
eldritch horrors 5f826bc600 libutil: convert readFileSource to a generator
Change-Id: I5f92b15fd367d46eb047d74ab6e317b4f51a46d3
2024-06-30 19:29:07 +02:00
eldritch horrors f659e93c5b libstore: convert dumpPath to a generator
Change-Id: Ic4cf5562504aa29130304469936f958c0426e5ef
2024-06-30 19:29:07 +02:00
eldritch horrors 1540bdbcfc libutil: allow construction of sources from generators
Change-Id: I78ff8d0720f06bce731e26d5e1c53b1382bbd589
2024-06-30 19:29:07 +02:00
eldritch horrors 379a892afb libutil: begin porting serialization to generators
generators are a better basis for serializers than streaming into sinks
as we do currently for many reasons, such as being usable as sources if
one wishes to (without requiring an intermediate sink to serialize full
data sets into memory, or boost coroutines to turn sinks into sources),
composing more naturally (as one can just yield a sub-generator instead
of being forced to wrap entire substreams into clunky functions or even
more clunky custom types to implement operator<< on), allowing wrappers
to transform data with clear ownership semantics (removing the need for
explicit memory allocations and Source wrappers), and many other things

Change-Id: I361d89ff556354f6930d9204f55117565f2f7f20
2024-06-30 19:29:06 +02:00
eldritch horrors 5f18b52274 libutil: convert drainFD to a Bytes generator
the `*Source` name is a slight misnomer since we do also have a
Source type, but we can probably live with this for time being.

Change-Id: I54eb2e59a4009014e324797f16b80b962759c7d3
2024-06-30 17:38:11 +02:00
eldritch horrors d96601fe55 libutil: allow draining Generator<Bytes> into sinks
Change-Id: I442d03a5399096d4baca9a2618b4c4b64db36c4b
2024-06-30 17:33:28 +02:00
eldritch horrors 2521f52465 libutil: add Bytes type
not used anywhere yet, but we'll use this a lot soon for generators that
return file contents, wire protocol fragments, or indeed any byte stream

Change-Id: I01a46f9bf9d75aaf4a5d7662773b99f498862a28
2024-06-30 17:30:34 +02:00
eldritch horrors 11307e04d1 libutil: generator type with on-yield value mapping
this will be the basis of non-boost coroutines in lix. anything that is
a boost coroutine *should* be representable with a Generator coroutine,
and many things that are not currently boost coroutines but behave much
like one (such as, notably, serializers) should be as well. this allows
us to greatly simplify many things that look like iteration but aren't.

Change-Id: I2cebcefa0148b631fb30df4c8cfa92167a407e34
2024-06-30 17:23:41 +02:00
77 changed files with 1585 additions and 1083 deletions
-54
View File
@@ -1,54 +0,0 @@
diff --git a/pthread_stop_world.c b/pthread_stop_world.c
index 2b45489..0e6d8ef 100644
--- a/pthread_stop_world.c
+++ b/pthread_stop_world.c
@@ -776,6 +776,8 @@ STATIC void GC_restart_handler(int sig)
/* world is stopped. Should not fail if it isn't. */
GC_INNER void GC_push_all_stacks(void)
{
+ size_t stack_limit;
+ pthread_attr_t pattr;
GC_bool found_me = FALSE;
size_t nthreads = 0;
int i;
@@ -868,6 +870,40 @@ GC_INNER void GC_push_all_stacks(void)
hi = p->altstack + p->altstack_size;
# endif
/* FIXME: Need to scan the normal stack too, but how ? */
+ } else {
+ #ifdef HAVE_PTHREAD_ATTR_GET_NP
+ if (pthread_attr_init(&pattr) != 0) {
+ ABORT("GC_push_all_stacks: pthread_attr_init failed!");
+ }
+ if (pthread_attr_get_np(p->id, &pattr) != 0) {
+ ABORT("GC_push_all_stacks: pthread_attr_get_np failed!");
+ }
+ #else
+ if (pthread_getattr_np(p->id, &pattr)) {
+ ABORT("GC_push_all_stacks: pthread_getattr_np failed!");
+ }
+ #endif
+ if (pthread_attr_getstacksize(&pattr, &stack_limit)) {
+ ABORT("GC_push_all_stacks: pthread_attr_getstacksize failed!");
+ }
+ if (pthread_attr_destroy(&pattr)) {
+ ABORT("GC_push_all_stacks: pthread_attr_destroy failed!");
+ }
+ // When a thread goes into a coroutine, we lose its original sp until
+ // control flow returns to the thread.
+ // While in the coroutine, the sp points outside the thread stack,
+ // so we can detect this and push the entire thread stack instead,
+ // as an approximation.
+ // We assume that the coroutine has similarly added its entire stack.
+ // This could be made accurate by cooperating with the application
+ // via new functions and/or callbacks.
+ #ifndef STACK_GROWS_UP
+ if (lo >= hi || lo < hi - stack_limit) { // sp outside stack
+ lo = hi - stack_limit;
+ }
+ #else
+ #error "STACK_GROWS_UP not supported in boost_coroutine2 (as of june 2021), so we don't support it in Nix."
+ #endif
}
# ifdef STACKPTR_CORRECTOR_AVAILABLE
if (GC_sp_corrector != 0)
+4 -3
View File
@@ -99,7 +99,8 @@
stdenvs = [
"gccStdenv"
"clangStdenv"
"stdenv"
# FIXME: gcc 12 (default in 23.11) has bugs that break the nar parser.
"gcc13Stdenv" # "stdenv"
"libcxxStdenv"
"ccacheStdenv"
];
@@ -145,7 +146,7 @@
];
};
stdenvs = forAllStdenvs (make-pkgs null);
native = stdenvs.stdenvPackages;
native = stdenvs.gcc13StdenvPackages;
in
{
inherit stdenvs native;
@@ -420,7 +421,7 @@
makeShell pkgs pkgs.stdenv
))
// {
default = self.devShells.${system}.native-stdenvPackages;
default = self.devShells.${system}.native-gcc13StdenvPackages;
}
);
};
+12 -1
View File
@@ -142,6 +142,16 @@ else
cpp_pch = []
endif
# gcc 12 is known to miscompile some coroutine-based code quite horribly,
# causing (among other things) copies of move-only objects and the double
# frees one would expect when the objects are unique_ptrs. these problems
# often show up as memory corruption when nesting generators (since we do
# treat generators like owned memory) and will cause inexplicable crashs.
assert(
cxx.get_id() != 'gcc' or cxx.version().version_compare('>=13'),
'GCC 12 and earlier are known to miscompile lix coroutines, use GCC 13 or clang.'
)
# Translate some historical and Mesony CPU names to Lixy CPU names.
# FIXME(Qyriad): the 32-bit x86 code is not tested right now, because cross compilation for Lix
@@ -194,7 +204,7 @@ configdata += {
'HAVE_BOEHMGC': boehm.found().to_int(),
}
boost = dependency('boost', required : true, modules : ['context', 'coroutine', 'container'])
boost = dependency('boost', required : true, modules : ['container'])
# cpuid only makes sense on x86_64
cpuid_required = is_x64 ? get_option('cpuid') : false
@@ -432,6 +442,7 @@ add_project_arguments(
'-Wimplicit-fallthrough',
'-Werror=switch',
'-Werror=switch-enum',
'-Werror=unused-result',
'-Wdeprecated-copy',
'-Wignored-qualifiers',
# Enable assertions in libstdc++ by default. Harmless on libc++. Benchmarked
+1 -10
View File
@@ -62,15 +62,7 @@
__forDefaults ? {
canRunInstalled = stdenv.buildPlatform.canExecute stdenv.hostPlatform;
boehmgc-nix = (boehmgc.override { enableLargeConfig = true; }).overrideAttrs {
patches = [
# We do *not* include prev.patches (which doesn't exist in normal pkgs.boehmgc anyway)
# because if the caller of this package passed a patched boehm as `boehmgc` instead of
# `boehmgc-nix` then this will almost certainly have duplicate patches, which means
# the patches won't apply and we'll get a build failure.
./boehmgc-coroutine-sp-fallback.diff
];
};
boehmgc-nix = boehmgc.override { enableLargeConfig = true; };
editline-lix = editline.overrideAttrs (prev: {
configureFlags = prev.configureFlags or [ ] ++ [ (lib.enableFeature true "sigstop") ];
@@ -167,7 +159,6 @@ stdenv.mkDerivation (finalAttrs: {
functionalTestFiles
]
++ lib.optionals (!finalAttrs.dontBuild || internalApiDocs) [
./boehmgc-coroutine-sp-fallback.diff
./doc
./misc
./src
+1
View File
@@ -65,5 +65,6 @@ if cxx.get_linker_id() in ['ld.bfd', 'ld.gold']
endif
libstore = dependency('lixstore', 'lix-store', required : true)
libutil = dependency('lixutil', 'lix-util', required : true)
subdir('lib/Nix')
+1 -1
View File
@@ -244,7 +244,7 @@ StorePath ProfileManifest::build(ref<Store> store)
/* Add the symlink tree to the store. */
StringSink sink;
dumpPath(tempDir, sink);
sink << dumpPath(tempDir);
auto narHash = hashString(htSHA256, sink.s);
+2 -2
View File
@@ -252,7 +252,7 @@ void runNix(Path program, const Strings & args)
.program = settings.nixBinDir+ "/" + program,
.args = args,
.environment = subprocessEnv,
});
}).wait();
return;
}
@@ -651,7 +651,7 @@ ProcessLineResult NixRepl::processLine(std::string line)
// runProgram redirects stdout to a StringSink,
// using runProgram2 to allow editors to display their UI
runProgram2(RunOptions { .program = editor, .searchPath = true, .args = args });
runProgram2(RunOptions { .program = editor, .searchPath = true, .args = args }).wait();
// Reload right after exiting the editor
state->resetFileCache();
-68
View File
@@ -42,10 +42,6 @@
#include <gc/gc.h>
#include <gc/gc_cpp.h>
#include <boost/coroutine2/coroutine.hpp>
#include <boost/coroutine2/protected_fixedsize_stack.hpp>
#include <boost/context/stack_context.hpp>
#endif
using json = nlohmann::json;
@@ -205,42 +201,6 @@ static void * oomHandler(size_t requested)
throw std::bad_alloc();
}
class BoehmGCStackAllocator : public StackAllocator {
boost::coroutines2::protected_fixedsize_stack stack {
// We allocate 8 MB, the default max stack size on NixOS.
// A smaller stack might be quicker to allocate but reduces the stack
// depth available for source filter expressions etc.
std::max(boost::context::stack_traits::default_size(), static_cast<std::size_t>(8 * 1024 * 1024))
};
// This is specific to boost::coroutines2::protected_fixedsize_stack.
// The stack protection page is included in sctx.size, so we have to
// subtract one page size from the stack size.
std::size_t pfss_usable_stack_size(boost::context::stack_context &sctx) {
return sctx.size - boost::context::stack_traits::page_size();
}
public:
boost::context::stack_context allocate() override {
auto sctx = stack.allocate();
// Stacks generally start at a high address and grow to lower addresses.
// Architectures that do the opposite are rare; in fact so rare that
// boost_routine does not implement it.
// So we subtract the stack size.
GC_add_roots(static_cast<char *>(sctx.sp) - pfss_usable_stack_size(sctx), sctx.sp);
return sctx;
}
void deallocate(boost::context::stack_context sctx) override {
GC_remove_roots(static_cast<char *>(sctx.sp) - pfss_usable_stack_size(sctx), sctx.sp);
stack.deallocate(sctx);
}
};
static BoehmGCStackAllocator boehmGCStackAllocator;
#endif
@@ -256,23 +216,6 @@ static Symbol getName(const AttrName & name, EvalState & state, Env & env)
}
}
#if HAVE_BOEHMGC
/* Disable GC while this object lives. Used by CoroutineContext.
*
* Boehm keeps a count of GC_disable() and GC_enable() calls,
* and only enables GC when the count matches.
*/
class BoehmDisableGC {
public:
BoehmDisableGC() {
GC_disable();
};
~BoehmDisableGC() {
GC_enable();
};
};
#endif
static bool gcInitialised = false;
void initGC()
@@ -294,17 +237,6 @@ void initGC()
GC_set_oom_fn(oomHandler);
StackAllocator::defaultAllocator = &boehmGCStackAllocator;
#if NIX_BOEHM_PATCH_VERSION != 1
printTalkative("Unpatched BoehmGC, disabling GC inside coroutines");
/* Used to disable GC when entering coroutines on macOS */
create_coro_gc_hook = []() -> std::shared_ptr<void> {
return std::make_shared<BoehmDisableGC>();
};
#endif
/* Set the initial heap size to something fairly big (25% of
physical RAM, up to a maximum of 384 MiB) so that in most cases
we don't need to garbage collect at all. (Collection has a
+4 -7
View File
@@ -691,17 +691,14 @@ struct GitInputScheme : InputScheme
filter = isNotDotGitDirectory;
} else {
// FIXME: should pipe this, or find some better way to extract a
// revision.
auto source = sinkToSource([&](Sink & sink) {
runProgram2({
auto proc = runProgram2({
.program = "git",
.args = { "-C", repoDir, "--git-dir", gitDir, "archive", input.getRev()->gitRev() },
.standardOut = &sink
});
.captureStdout = true,
});
Finally const _wait([&] { proc.wait(); });
unpackTarfile(*source, tmpDir);
unpackTarfile(*proc.stdout(), tmpDir);
}
auto storePath = store->addToStore(name, tmpDir, FileIngestionMethod::Recursive, htSHA256, filter);
+2 -4
View File
@@ -130,10 +130,8 @@ struct PathInputScheme : InputScheme
time_t mtime = 0;
if (!storePath || storePath->name() != "source" || !store->isValidPath(*storePath)) {
// FIXME: try to substitute storePath.
auto src = sinkToSource([&](Sink & sink) {
mtime = dumpPathAndGetMtime(absPath, sink, defaultPathFilter);
});
storePath = store->addToStoreFromDump(*src, "source");
auto src = GeneratorSource{dumpPathAndGetMtime(absPath, mtime, defaultPathFilter)};
storePath = store->addToStoreFromDump(src, "source");
}
input.attrs.insert_or_assign("lastModified", uint64_t(mtime));
+1 -1
View File
@@ -71,7 +71,7 @@ DownloadFileResult downloadFile(
storePath = std::move(cached->storePath);
} else {
StringSink sink;
dumpString(res.data, sink);
sink << dumpString(res.data);
auto hash = hashString(htSHA256, res.data);
ValidPathInfo info {
*store,
+28 -22
View File
@@ -67,16 +67,18 @@ void BinaryCacheStore::upsertFile(const std::string & path,
upsertFile(path, std::make_shared<std::stringstream>(std::move(data)), mimeType);
}
void BinaryCacheStore::getFile(const std::string & path, Sink & sink)
box_ptr<Source> BinaryCacheStore::getFile(const std::string & path)
{
sink(*getFileContents(path));
return make_box_ptr<GeneratorSource>([](std::string data) -> Generator<Bytes> {
co_yield std::span{data.data(), data.size()};
}(std::move(*getFileContents(path))));
}
std::optional<std::string> BinaryCacheStore::getFileContents(const std::string & path)
{
StringSink sink;
try {
getFile(path, sink);
return getFile(path)->drain();
} catch (NoSuchBinaryCacheFile &) {
return std::nullopt;
}
@@ -327,26 +329,32 @@ std::optional<StorePath> BinaryCacheStore::queryPathFromHashPart(const std::stri
}
}
void BinaryCacheStore::narFromPath(const StorePath & storePath, Sink & sink)
WireFormatGenerator BinaryCacheStore::narFromPath(const StorePath & storePath)
{
auto info = queryPathInfo(storePath).cast<const NarInfo>();
LengthSink narSize;
TeeSink tee { sink, narSize };
auto decompressor = makeDecompressionSink(info->compression, tee);
try {
getFile(info->url, *decompressor);
} catch (NoSuchBinaryCacheFile & e) {
throw SubstituteGone(std::move(e.info()));
auto file = getFile(info->url);
return [](auto info, auto file, auto & stats) -> WireFormatGenerator {
char buf[65536];
size_t total = 0;
auto decompressor = makeDecompressionSource(info->compression, *file);
try {
while (true) {
const auto len = decompressor->read(buf, sizeof(buf));
co_yield std::span{buf, len};
total += len;
}
} catch (EndOfFile &) {
}
decompressor->finish();
stats.narRead++;
//stats.narReadCompressedBytes += nar->size(); // FIXME
stats.narReadBytes += narSize.length;
stats.narReadBytes += total;
}(std::move(info), std::move(file), stats);
} catch (NoSuchBinaryCacheFile & e) {
throw SubstituteGone(std::move(e.info()));
}
}
std::shared_ptr<const ValidPathInfo> BinaryCacheStore::queryPathInfoUncached(const StorePath & storePath)
@@ -383,16 +391,14 @@ StorePath BinaryCacheStore::addToStore(
HashSink sink { hashAlgo };
if (method == FileIngestionMethod::Recursive) {
dumpPath(srcPath, sink, filter);
sink << dumpPath(srcPath, filter);
} else {
readFileSource(srcPath)->drainInto(sink);
sink << readFileSource(srcPath);
}
auto h = sink.finish().first;
auto source = sinkToSource([&](Sink & sink) {
dumpPath(srcPath, sink, filter);
});
return addToStoreCommon(*source, repair, CheckSigs, [&](HashResult nar) {
auto source = GeneratorSource{dumpPath(srcPath, filter)};
return addToStoreCommon(source, repair, CheckSigs, [&](HashResult nar) {
ValidPathInfo info {
*this,
name,
@@ -425,7 +431,7 @@ StorePath BinaryCacheStore::addTextToStore(
return path;
StringSink sink;
dumpString(s, sink);
sink << dumpString(s);
StringSource source(sink.s);
return addToStoreCommon(source, repair, CheckSigs, [&](HashResult nar) {
ValidPathInfo info {
+2 -2
View File
@@ -83,7 +83,7 @@ public:
/**
* Dump the contents of the specified file to a sink.
*/
virtual void getFile(const std::string & path, Sink & sink);
virtual box_ptr<Source> getFile(const std::string & path);
virtual std::optional<std::string> getFileContents(const std::string & path);
@@ -136,7 +136,7 @@ public:
std::shared_ptr<const Realisation> queryRealisationUncached(const DrvOutput &) override;
void narFromPath(const StorePath & path, Sink & sink) override;
WireFormatGenerator narFromPath(const StorePath & path) override;
ref<FSAccessor> getFSAccessor() override;
+8 -4
View File
@@ -923,12 +923,16 @@ void runPostBuildHook(
};
LogSink sink(act);
runProgram2({
auto proc = runProgram2({
.program = settings.postBuildHook,
.environment = hookEnvironment,
.standardOut = &sink,
.captureStdout = true,
.mergeStderrToStdout = true,
});
Finally const _wait([&] { proc.wait(); });
// FIXME just process the data, without a wrapper sink class
proc.stdout()->drainInto(sink);
}
void DerivationGoal::buildDone()
@@ -1204,7 +1208,7 @@ HookReply DerivationGoal::tryBuildHook()
/* Tell the hook all the inputs that have to be copied to the
remote system. */
CommonProto::write(worker.store, conn, inputPaths);
conn.to << CommonProto::write(worker.store, conn, inputPaths);
/* Tell the hooks the missing outputs that have to be copied back
from the remote system. */
@@ -1215,7 +1219,7 @@ HookReply DerivationGoal::tryBuildHook()
if (buildMode != bmCheck && status.known && status.known->isValid()) continue;
missingOutputs.insert(outputName);
}
CommonProto::write(worker.store, conn, missingOutputs);
conn.to << CommonProto::write(worker.store, conn, missingOutputs);
}
hook->sink = FdSink();
+13 -19
View File
@@ -1324,11 +1324,11 @@ struct RestrictedStore : public virtual RestrictedStoreConfig, public virtual In
return path;
}
void narFromPath(const StorePath & path, Sink & sink) override
WireFormatGenerator narFromPath(const StorePath & path) override
{
if (!goal.isAllowed(path))
throw InvalidPath("cannot dump unknown path '%s' in recursive Nix", printStorePath(path));
LocalFSStore::narFromPath(path, sink);
return LocalFSStore::narFromPath(path);
}
void ensurePath(const StorePath & path) override
@@ -2389,14 +2389,10 @@ SingleDrvOutputs LocalDerivationGoal::registerOutputs()
if (!rewrites.empty()) {
debug("rewriting hashes in '%1%'; cross fingers", actualPath);
/* FIXME: Is this actually streaming? */
auto source = sinkToSource([&](Sink & nextSink) {
RewritingSink rsink(rewrites, nextSink);
dumpPath(actualPath, rsink);
rsink.flush();
});
GeneratorSource dump{dumpPath(actualPath)};
RewritingSource rewritten(rewrites, dump);
Path tmpPath = actualPath + ".tmp";
restorePath(tmpPath, *source);
restorePath(tmpPath, rewritten);
deletePath(actualPath);
movePath(tmpPath, actualPath);
@@ -2450,23 +2446,21 @@ SingleDrvOutputs LocalDerivationGoal::registerOutputs()
rewriteOutput(outputRewrites);
/* FIXME optimize and deduplicate with addToStore */
std::string oldHashPart { scratchPath->hashPart() };
HashModuloSink caSink { outputHash.hashType, oldHashPart };
std::visit(overloaded {
[&](const TextIngestionMethod &) {
readFileSource(actualPath)->drainInto(caSink);
auto input = std::visit(overloaded {
[&](const TextIngestionMethod &) -> GeneratorSource {
return GeneratorSource(readFileSource(actualPath));
},
[&](const FileIngestionMethod & m2) {
[&](const FileIngestionMethod & m2) -> GeneratorSource {
switch (m2) {
case FileIngestionMethod::Recursive:
dumpPath(actualPath, caSink);
break;
return GeneratorSource(dumpPath(actualPath));
case FileIngestionMethod::Flat:
readFileSource(actualPath)->drainInto(caSink);
break;
return GeneratorSource(readFileSource(actualPath));
}
assert(false);
},
}, outputHash.method.raw);
auto got = caSink.finish().first;
auto got = computeHashModulo(outputHash.hashType, oldHashPart, input).first;
auto optCA = ContentAddressWithReferences::fromPartsOpt(
outputHash.method,
+5 -9
View File
@@ -32,23 +32,19 @@ void builtinFetchurl(const BasicDerivation & drv, const std::string & netrcData)
auto fetch = [&](const std::string & url) {
auto source = sinkToSource([&](Sink & sink) {
/* No need to do TLS verification, because we check the hash of
the result anyway. */
FileTransferRequest request(url);
request.verifyTLS = false;
auto decompressor = makeDecompressionSink(
unpack && mainUrl.ends_with(".xz") ? "xz" : "none", sink);
fileTransfer->download(std::move(request))->drainInto(*decompressor);
decompressor->finish();
});
auto raw = fileTransfer->download(std::move(request));
auto decompressor = makeDecompressionSource(
unpack && mainUrl.ends_with(".xz") ? "xz" : "none", *raw);
if (unpack)
restorePath(storePath, *source);
restorePath(storePath, *decompressor);
else
writeFile(storePath, *source);
writeFile(storePath, *decompressor);
auto executable = drv.env.find("executable");
if (executable != drv.env.end() && executable->second == "1") {
+2 -2
View File
@@ -20,9 +20,9 @@ namespace nix {
{ \
return LengthPrefixedProtoHelper<CommonProto, T >::read(store, conn); \
} \
TEMPLATE void CommonProto::Serialise< T >::write(const Store & store, CommonProto::WriteConn conn, const T & t) \
TEMPLATE [[nodiscard]] WireFormatGenerator CommonProto::Serialise< T >::write(const Store & store, CommonProto::WriteConn conn, const T & t) \
{ \
LengthPrefixedProtoHelper<CommonProto, T >::write(store, conn, t); \
return LengthPrefixedProtoHelper<CommonProto, T >::write(store, conn, t); \
}
COMMON_USE_LENGTH_PREFIX_SERIALISER(template<typename T>, std::vector<T>)
+18 -14
View File
@@ -16,9 +16,9 @@ std::string CommonProto::Serialise<std::string>::read(const Store & store, Commo
return readString(conn.from);
}
void CommonProto::Serialise<std::string>::write(const Store & store, CommonProto::WriteConn conn, const std::string & str)
WireFormatGenerator CommonProto::Serialise<std::string>::write(const Store & store, CommonProto::WriteConn conn, const std::string & str)
{
conn.to << str;
co_yield str;
}
@@ -27,9 +27,9 @@ StorePath CommonProto::Serialise<StorePath>::read(const Store & store, CommonPro
return store.parseStorePath(readString(conn.from));
}
void CommonProto::Serialise<StorePath>::write(const Store & store, CommonProto::WriteConn conn, const StorePath & storePath)
WireFormatGenerator CommonProto::Serialise<StorePath>::write(const Store & store, CommonProto::WriteConn conn, const StorePath & storePath)
{
conn.to << store.printStorePath(storePath);
co_yield store.printStorePath(storePath);
}
@@ -38,9 +38,9 @@ ContentAddress CommonProto::Serialise<ContentAddress>::read(const Store & store,
return ContentAddress::parse(readString(conn.from));
}
void CommonProto::Serialise<ContentAddress>::write(const Store & store, CommonProto::WriteConn conn, const ContentAddress & ca)
WireFormatGenerator CommonProto::Serialise<ContentAddress>::write(const Store & store, CommonProto::WriteConn conn, const ContentAddress & ca)
{
conn.to << renderContentAddress(ca);
co_yield renderContentAddress(ca);
}
@@ -53,9 +53,9 @@ Realisation CommonProto::Serialise<Realisation>::read(const Store & store, Commo
);
}
void CommonProto::Serialise<Realisation>::write(const Store & store, CommonProto::WriteConn conn, const Realisation & realisation)
WireFormatGenerator CommonProto::Serialise<Realisation>::write(const Store & store, CommonProto::WriteConn conn, const Realisation & realisation)
{
conn.to << realisation.toJSON().dump();
co_yield realisation.toJSON().dump();
}
@@ -64,9 +64,9 @@ DrvOutput CommonProto::Serialise<DrvOutput>::read(const Store & store, CommonPro
return DrvOutput::parse(readString(conn.from));
}
void CommonProto::Serialise<DrvOutput>::write(const Store & store, CommonProto::WriteConn conn, const DrvOutput & drvOutput)
WireFormatGenerator CommonProto::Serialise<DrvOutput>::write(const Store & store, CommonProto::WriteConn conn, const DrvOutput & drvOutput)
{
conn.to << drvOutput.to_string();
co_yield drvOutput.to_string();
}
@@ -76,9 +76,11 @@ std::optional<StorePath> CommonProto::Serialise<std::optional<StorePath>>::read(
return s == "" ? std::optional<StorePath> {} : store.parseStorePath(s);
}
void CommonProto::Serialise<std::optional<StorePath>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<StorePath> & storePathOpt)
WireFormatGenerator CommonProto::Serialise<std::optional<StorePath>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<StorePath> & storePathOpt)
{
conn.to << (storePathOpt ? store.printStorePath(*storePathOpt) : "");
return [](std::string s) -> WireFormatGenerator {
co_yield s;
}(storePathOpt ? store.printStorePath(*storePathOpt) : "");
}
@@ -87,9 +89,11 @@ std::optional<ContentAddress> CommonProto::Serialise<std::optional<ContentAddres
return ContentAddress::parseOpt(readString(conn.from));
}
void CommonProto::Serialise<std::optional<ContentAddress>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<ContentAddress> & caOpt)
WireFormatGenerator CommonProto::Serialise<std::optional<ContentAddress>>::write(const Store & store, CommonProto::WriteConn conn, const std::optional<ContentAddress> & caOpt)
{
conn.to << (caOpt ? renderContentAddress(*caOpt) : "");
return [](std::string s) -> WireFormatGenerator {
co_yield s;
}(caOpt ? renderContentAddress(*caOpt) : "");
}
}
+4 -3
View File
@@ -48,9 +48,10 @@ struct CommonProto
* infer the type instead of having to write it down explicitly.
*/
template<typename T>
static void write(const Store & store, WriteConn conn, const T & t)
[[nodiscard]]
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t)
{
CommonProto::Serialise<T>::write(store, conn, t);
return CommonProto::Serialise<T>::write(store, conn, t);
}
};
@@ -58,7 +59,7 @@ struct CommonProto
struct CommonProto::Serialise< T > \
{ \
static T read(const Store & store, CommonProto::ReadConn conn); \
static void write(const Store & store, CommonProto::WriteConn conn, const T & str); \
[[nodiscard]] static WireFormatGenerator write(const Store & store, CommonProto::WriteConn conn, const T & str); \
}
template<>
+41 -28
View File
@@ -160,8 +160,7 @@ struct TunnelSink : Sink
TunnelSink(Sink & to) : to(to) { }
void operator () (std::string_view data)
{
to << STDERR_WRITE;
writeString(data, to);
to << STDERR_WRITE << data;
}
};
@@ -292,7 +291,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
}
auto res = store->queryValidPaths(paths, substitute);
logger->stopWork();
WorkerProto::write(*store, wconn, res);
wconn.to << WorkerProto::write(*store, wconn, res);
break;
}
@@ -301,7 +300,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
logger->startWork();
auto res = store->querySubstitutablePaths(paths);
logger->stopWork();
WorkerProto::write(*store, wconn, res);
wconn.to << WorkerProto::write(*store, wconn, res);
break;
}
@@ -366,7 +365,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
#pragma GCC diagnostic pop
logger->stopWork();
WorkerProto::write(*store, wconn, paths);
wconn.to << WorkerProto::write(*store, wconn, paths);
break;
}
@@ -386,7 +385,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
logger->startWork();
auto outputs = store->queryPartialDerivationOutputMap(path);
logger->stopWork();
WorkerProto::write(*store, wconn, outputs);
wconn.to << WorkerProto::write(*store, wconn, outputs);
break;
}
@@ -433,7 +432,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
}();
logger->stopWork();
WorkerProto::Serialise<ValidPathInfo>::write(*store, wconn, *pathInfo);
wconn.to << WorkerProto::Serialise<ValidPathInfo>::write(*store, wconn, *pathInfo);
} else {
HashType hashAlgo;
std::string baseName;
@@ -454,7 +453,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
hashAlgo = parseHashType(hashAlgoRaw);
}
auto dumpSource = sinkToSource([&](Sink & saved) {
GeneratorSource dumpSource{[&]() -> WireFormatGenerator {
if (method == FileIngestionMethod::Recursive) {
/* We parse the NAR dump through into `saved` unmodified,
so why all this extra work? We still parse the NAR so
@@ -464,18 +463,33 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
command. (We don't trust `addToStoreFromDump` to not
eagerly consume the entire stream it's given, past the
length of the Nar. */
copyNAR(from, saved);
co_yield copyNAR(from);
} else {
/* Incrementally parse the NAR file, stripping the
metadata, and streaming the sole file we expect into
`saved`. */
RetrieveRegularNARSink savedRegular { saved };
parseDump(savedRegular, from);
if (!savedRegular.regular) throw Error("regular file expected");
auto parser = nar::parse(from);
nar::File * file = nullptr;
while (auto entry = parser.next()) {
file = std::visit(
overloaded{
[](nar::MetadataString) -> nar::File * { return nullptr; },
[](nar::MetadataRaw) -> nar::File * { return nullptr; },
[](nar::File & f) -> nar::File * { return &f; },
[](auto &) -> nar::File * { throw Error("regular file expected"); },
},
*entry
);
if (file) {
break;
}
});
}
assert(file); // should never fail unless the nar is empty, which would be invalid
co_yield std::move(file->contents);
}
}()};
logger->startWork();
auto path = store->addToStoreFromDump(*dumpSource, baseName, method, hashAlgo);
auto path = store->addToStoreFromDump(dumpSource, baseName, method, hashAlgo);
logger->stopWork();
to << store->printStorePath(path);
@@ -549,7 +563,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
auto results = store->buildPathsWithResults(drvs, mode);
logger->stopWork();
WorkerProto::write(*store, wconn, results);
wconn.to << WorkerProto::write(*store, wconn, results);
break;
}
@@ -627,7 +641,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
auto res = store->buildDerivation(drvPath, drv, buildMode);
logger->stopWork();
WorkerProto::write(*store, wconn, res);
wconn.to << WorkerProto::write(*store, wconn, res);
break;
}
@@ -761,7 +775,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
else {
to << 1
<< (i->second.deriver ? store->printStorePath(*i->second.deriver) : "");
WorkerProto::write(*store, wconn, i->second.references);
wconn.to << WorkerProto::write(*store, wconn, i->second.references);
to << i->second.downloadSize
<< i->second.narSize;
}
@@ -784,7 +798,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
for (auto & i : infos) {
to << store->printStorePath(i.first)
<< (i.second.deriver ? store->printStorePath(*i.second.deriver) : "");
WorkerProto::write(*store, wconn, i.second.references);
wconn.to << WorkerProto::write(*store, wconn, i.second.references);
to << i.second.downloadSize << i.second.narSize;
}
break;
@@ -794,7 +808,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
logger->startWork();
auto paths = store->queryAllValidPaths();
logger->stopWork();
WorkerProto::write(*store, wconn, paths);
wconn.to << WorkerProto::write(*store, wconn, paths);
break;
}
@@ -811,7 +825,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
logger->stopWork();
if (info) {
to << 1;
WorkerProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
wconn.to << WorkerProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
} else {
to << 0;
}
@@ -851,7 +865,7 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
auto path = store->parseStorePath(readString(from));
logger->startWork();
logger->stopWork();
dumpPath(store->toRealPath(path), to);
to << dumpPath(store->toRealPath(path));
break;
}
@@ -885,7 +899,6 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
else {
std::unique_ptr<Source> source;
StringSink saved;
source = std::make_unique<TunnelSource>(from, to);
logger->startWork();
@@ -907,9 +920,9 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
uint64_t downloadSize, narSize;
store->queryMissing(targets, willBuild, willSubstitute, unknown, downloadSize, narSize);
logger->stopWork();
WorkerProto::write(*store, wconn, willBuild);
WorkerProto::write(*store, wconn, willSubstitute);
WorkerProto::write(*store, wconn, unknown);
wconn.to << WorkerProto::write(*store, wconn, willBuild);
wconn.to << WorkerProto::write(*store, wconn, willSubstitute);
wconn.to << WorkerProto::write(*store, wconn, unknown);
to << downloadSize << narSize;
break;
}
@@ -937,11 +950,11 @@ static void performOp(TunnelLogger * logger, ref<Store> store,
if (GET_PROTOCOL_MINOR(clientVersion) < 31) {
std::set<StorePath> outPaths;
if (info) outPaths.insert(info->outPath);
WorkerProto::write(*store, wconn, outPaths);
wconn.to << WorkerProto::write(*store, wconn, outPaths);
} else {
std::set<Realisation> realisations;
if (info) realisations.insert(*info);
WorkerProto::write(*store, wconn, realisations);
wconn.to << WorkerProto::write(*store, wconn, realisations);
}
break;
}
@@ -1022,7 +1035,7 @@ void processConnection(
? store->isTrustedClient()
: std::optional { NotTrusted };
WorkerProto::WriteConn wconn {to, clientVersion};
WorkerProto::write(*store, wconn, temp);
wconn.to << WorkerProto::write(*store, wconn, temp);
}
/* Send startup error messages to the client. */
+1 -1
View File
@@ -994,7 +994,7 @@ void writeDerivation(Sink & out, const Store & store, const BasicDerivation & dr
},
}, i.second.raw);
}
CommonProto::write(store,
out << CommonProto::write(store,
CommonProto::WriteConn { .to = out },
drv.inputSrcs);
out << drv.platform << drv.builder << drv.args;
+1 -1
View File
@@ -63,7 +63,7 @@ struct DummyStore : public virtual DummyStoreConfig, public virtual Store
RepairFlag repair) override
{ unsupported("addTextToStore"); }
void narFromPath(const StorePath & path, Sink & sink) override
WireFormatGenerator narFromPath(const StorePath & path) override
{ unsupported("narFromPath"); }
std::shared_ptr<const Realisation> queryRealisationUncached(const DrvOutput &) override
+5 -6
View File
@@ -33,7 +33,7 @@ void Store::exportPath(const StorePath & path, Sink & sink)
HashSink hashSink(htSHA256);
TeeSink teeSink(sink, hashSink);
narFromPath(path, teeSink);
teeSink << narFromPath(path);
/* Refuse to export paths that have changed. This prevents
filesystem corruption from spreading to other machines.
@@ -45,11 +45,10 @@ void Store::exportPath(const StorePath & path, Sink & sink)
teeSink
<< exportMagic
<< printStorePath(path);
CommonProto::write(*this,
<< printStorePath(path)
<< CommonProto::write(*this,
CommonProto::WriteConn { .to = teeSink },
info->references);
teeSink
info->references)
<< (info->deriver ? printStorePath(*info->deriver) : "")
<< 0;
}
@@ -64,7 +63,7 @@ StorePaths Store::importPaths(Source & source, CheckSigsFlag checkSigs)
/* Extract the NAR from the source. */
StringSink saved;
copyNAR(source, saved);
saved << copyNAR(source);
uint32_t magic = readInt(source);
if (magic != exportMagic)
+2 -2
View File
@@ -150,12 +150,12 @@ protected:
}
void getFile(const std::string & path, Sink & sink) override
box_ptr<Source> getFile(const std::string & path) override
{
checkEnabled();
auto request(makeRequest(path));
try {
getFileTransfer()->download(std::move(request))->drainInto(sink);
return getFileTransfer()->download(std::move(request));
} catch (FileTransferError & e) {
if (e.error == FileTransfer::NotFound || e.error == FileTransfer::Forbidden)
throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache '%s'", path, getUri());
+10 -8
View File
@@ -185,7 +185,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
<< printStorePath(info.path)
<< (info.deriver ? printStorePath(*info.deriver) : "")
<< info.narHash.to_string(Base16, false);
ServeProto::write(*this, *conn, info.references);
conn->to << ServeProto::write(*this, *conn, info.references);
conn->to
<< info.registrationTime
<< info.narSize
@@ -193,7 +193,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
<< info.sigs
<< renderContentAddress(info.ca);
try {
copyNAR(source, conn->to);
conn->to << copyNAR(source);
} catch (...) {
conn->good = false;
throw;
@@ -206,7 +206,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
<< ServeProto::Command::ImportPaths
<< 1;
try {
copyNAR(source, conn->to);
conn->to << copyNAR(source);
} catch (...) {
conn->good = false;
throw;
@@ -214,7 +214,7 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
conn->to
<< exportMagic
<< printStorePath(info.path);
ServeProto::write(*this, *conn, info.references);
conn->to << ServeProto::write(*this, *conn, info.references);
conn->to
<< (info.deriver ? printStorePath(*info.deriver) : "")
<< 0
@@ -227,13 +227,15 @@ struct LegacySSHStore : public virtual LegacySSHStoreConfig, public virtual Stor
throw Error("failed to add path '%s' to remote host '%s'", printStorePath(info.path), host);
}
void narFromPath(const StorePath & path, Sink & sink) override
WireFormatGenerator narFromPath(const StorePath & path) override
{
auto conn(connections->get());
conn->to << ServeProto::Command::DumpStorePath << printStorePath(path);
conn->to.flush();
copyNAR(conn->from, sink);
return [] (auto conn) -> WireFormatGenerator {
co_yield copyNAR(conn->from);
}(std::move(conn));
}
std::optional<StorePath> queryPathFromHashPart(const std::string & hashPart) override
@@ -364,7 +366,7 @@ public:
conn->to
<< ServeProto::Command::QueryClosure
<< includeOutputs;
ServeProto::write(*this, *conn, paths);
conn->to << ServeProto::write(*this, *conn, paths);
conn->to.flush();
for (auto & i : ServeProto::Serialise<StorePathSet>::read(*this, *conn))
@@ -380,7 +382,7 @@ public:
<< ServeProto::Command::QueryValidPaths
<< false // lock
<< maybeSubstitute;
ServeProto::write(*this, *conn, paths);
conn->to << ServeProto::write(*this, *conn, paths);
conn->to.flush();
return ServeProto::Serialise<StorePathSet>::read(*this, *conn);
+27 -15
View File
@@ -7,6 +7,7 @@
*/
#include "types.hh"
#include "serialise.hh"
namespace nix {
@@ -45,7 +46,7 @@ struct LengthPrefixedProtoHelper;
struct LengthPrefixedProtoHelper< Inner, T > \
{ \
static T read(const Store & store, typename Inner::ReadConn conn); \
static void write(const Store & store, typename Inner::WriteConn conn, const T & str); \
[[nodiscard]] static WireFormatGenerator write(const Store & store, typename Inner::WriteConn conn, const T & str); \
private: \
template<typename U> using S = typename Inner::template Serialise<U>; \
}
@@ -78,13 +79,13 @@ LengthPrefixedProtoHelper<Inner, std::vector<T>>::read(
}
template<class Inner, typename T>
void
WireFormatGenerator
LengthPrefixedProtoHelper<Inner, std::vector<T>>::write(
const Store & store, typename Inner::WriteConn conn, const std::vector<T> & resSet)
{
conn.to << resSet.size();
co_yield resSet.size();
for (auto & key : resSet) {
S<T>::write(store, conn, key);
co_yield S<T>::write(store, conn, key);
}
}
@@ -102,13 +103,13 @@ LengthPrefixedProtoHelper<Inner, std::set<T>>::read(
}
template<class Inner, typename T>
void
WireFormatGenerator
LengthPrefixedProtoHelper<Inner, std::set<T>>::write(
const Store & store, typename Inner::WriteConn conn, const std::set<T> & resSet)
{
conn.to << resSet.size();
co_yield resSet.size();
for (auto & key : resSet) {
S<T>::write(store, conn, key);
co_yield S<T>::write(store, conn, key);
}
}
@@ -128,14 +129,14 @@ LengthPrefixedProtoHelper<Inner, std::map<K, V>>::read(
}
template<class Inner, typename K, typename V>
void
WireFormatGenerator
LengthPrefixedProtoHelper<Inner, std::map<K, V>>::write(
const Store & store, typename Inner::WriteConn conn, const std::map<K, V> & resMap)
{
conn.to << resMap.size();
co_yield resMap.size();
for (auto & i : resMap) {
S<K>::write(store, conn, i.first);
S<V>::write(store, conn, i.second);
co_yield S<K>::write(store, conn, i.first);
co_yield S<V>::write(store, conn, i.second);
}
}
@@ -150,13 +151,24 @@ LengthPrefixedProtoHelper<Inner, std::tuple<Ts...>>::read(
}
template<class Inner, typename... Ts>
void
WireFormatGenerator
LengthPrefixedProtoHelper<Inner, std::tuple<Ts...>>::write(
const Store & store, typename Inner::WriteConn conn, const std::tuple<Ts...> & res)
{
std::apply([&]<typename... Us>(const Us &... args) {
(S<Us>::write(store, conn, args), ...);
}, res);
auto fullArgs = std::apply(
[&](auto &... rest) {
return std::tuple<const Store &, typename Inner::WriteConn &, const Ts &...>(
std::cref(store), conn, rest...
);
},
res
);
return std::apply(
[]<typename... Us>(auto & store, auto conn, const Us &... args) -> WireFormatGenerator {
(co_yield S<Us>::write(store, conn, args), ...);
},
fullArgs
);
}
}
+2 -2
View File
@@ -68,10 +68,10 @@ protected:
del.cancel();
}
void getFile(const std::string & path, Sink & sink) override
box_ptr<Source> getFile(const std::string & path) override
{
try {
readFileSource(binaryCacheDir + "/" + path)->drainInto(sink);
return make_box_ptr<GeneratorSource>(readFileSource(binaryCacheDir + "/" + path));
} catch (SysError & e) {
if (e.errNo == ENOENT)
throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache", path);
+2 -2
View File
@@ -78,11 +78,11 @@ ref<FSAccessor> LocalFSStore::getFSAccessor()
std::dynamic_pointer_cast<LocalFSStore>(shared_from_this())));
}
void LocalFSStore::narFromPath(const StorePath & path, Sink & sink)
WireFormatGenerator LocalFSStore::narFromPath(const StorePath & path)
{
if (!isValidPath(path))
throw Error("path '%s' does not exist in store", printStorePath(path));
dumpPath(getRealStoreDir() + std::string(printStorePath(path), storeDir.size()), sink);
return dumpPath(getRealStoreDir() + std::string(printStorePath(path), storeDir.size()));
}
const std::string LocalFSStore::drvsLogDir = "drvs";
+1 -1
View File
@@ -42,7 +42,7 @@ public:
LocalFSStore(const Params & params);
void narFromPath(const StorePath & path, Sink & sink) override;
WireFormatGenerator narFromPath(const StorePath & path) override;
ref<FSAccessor> getFSAccessor() override;
/**
+11 -13
View File
@@ -1406,7 +1406,7 @@ StorePath LocalStore::addToStoreFromDump(Source & source0, std::string_view name
auto narHash = std::pair { hash, size };
if (method != FileIngestionMethod::Recursive || hashAlgo != htSHA256) {
HashSink narSink { htSHA256 };
dumpPath(realPath, narSink);
narSink << dumpPath(realPath);
narHash = narSink.finish();
}
@@ -1461,7 +1461,7 @@ StorePath LocalStore::addTextToStore(
canonicalisePathMetaData(realPath, {});
StringSink sink;
dumpString(s, sink);
sink << dumpString(s);
auto narHash = hashString(htSHA256, sink.s);
optimisePath(realPath, repair);
@@ -1601,7 +1601,7 @@ bool LocalStore::verifyStore(bool checkContents, RepairFlag repair)
auto hashSink = HashSink(info->narHash.type);
dumpPath(Store::toRealPath(i), hashSink);
hashSink << dumpPath(Store::toRealPath(i));
auto current = hashSink.finish();
if (info->narHash != nullHash && info->narHash != current.first) {
@@ -1887,25 +1887,23 @@ ContentAddress LocalStore::hashCAPath(
const std::string_view pathHash
)
{
HashModuloSink caSink ( hashType, std::string(pathHash) );
std::visit(overloaded {
[&](const TextIngestionMethod &) {
readFileSource(path)->drainInto(caSink);
auto data = std::visit(overloaded {
[&](const TextIngestionMethod &) -> GeneratorSource {
return GeneratorSource(readFileSource(path));
},
[&](const FileIngestionMethod & m2) {
[&](const FileIngestionMethod & m2) -> GeneratorSource {
switch (m2) {
case FileIngestionMethod::Recursive:
dumpPath(path, caSink);
break;
return GeneratorSource(dumpPath(path));
case FileIngestionMethod::Flat:
readFileSource(path)->drainInto(caSink);
break;
return GeneratorSource(readFileSource(path));
}
assert(false);
},
}, method.raw);
return ContentAddress {
.method = method,
.hash = caSink.finish().first,
.hash = computeHashModulo(hashType, std::string(pathHash), data).first,
};
}
+8 -11
View File
@@ -23,7 +23,7 @@ std::map<StorePath, StorePath> makeContentAddressed(
std::string oldHashPart(path.hashPart());
StringSink sink;
srcStore.narFromPath(path, sink);
sink << srcStore.narFromPath(path);
StringMap rewrites;
@@ -43,10 +43,10 @@ std::map<StorePath, StorePath> makeContentAddressed(
sink.s = rewriteStrings(sink.s, rewrites);
HashModuloSink hashModuloSink(htSHA256, oldHashPart);
hashModuloSink(sink.s);
auto narModuloHash = hashModuloSink.finish().first;
auto narModuloHash = [&] {
StringSource source{sink.s};
return computeHashModulo(htSHA256, oldHashPart, source).first;
}();
ValidPathInfo info {
dstStore,
@@ -61,15 +61,12 @@ std::map<StorePath, StorePath> makeContentAddressed(
printInfo("rewriting '%s' to '%s'", pathS, dstStore.printStorePath(info.path));
StringSink sink2;
RewritingSink rsink2(oldHashPart, std::string(info.path.hashPart()), sink2);
rsink2(sink.s);
rsink2.flush();
const auto rewritten = rewriteStrings(sink.s, {{oldHashPart, std::string(info.path.hashPart())}});
info.narHash = hashString(htSHA256, sink2.s);
info.narHash = hashString(htSHA256, rewritten);
info.narSize = sink.s.size();
StringSource source(sink2.s);
StringSource source(rewritten);
dstStore.addToStore(info, source);
remappings.insert_or_assign(std::move(path), std::move(info.path));
+1 -1
View File
@@ -61,7 +61,7 @@ StorePathSet scanForReferences(
TeeSink sink { refsSink, toTee };
/* Look for the hashes in the NAR dump of the path. */
dumpPath(path, sink);
sink << dumpPath(path);
return refsSink.getResultPaths();
}
+1 -1
View File
@@ -97,7 +97,7 @@ std::pair<ref<FSAccessor>, Path> RemoteFSAccessor::fetch(const Path & path_, boo
}
StringSink sink;
store->narFromPath(storePath, sink);
sink << store->narFromPath(storePath);
return {addToCache(storePath.hashPart(), std::move(sink.s)), restPath};
}
+34 -24
View File
@@ -203,7 +203,7 @@ StorePathSet RemoteStore::queryValidPaths(const StorePathSet & paths, Substitute
{
auto conn(getConnection());
conn->to << WorkerProto::Op::QueryValidPaths;
WorkerProto::write(*this, *conn, paths);
conn->to << WorkerProto::write(*this, *conn, paths);
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 27) {
conn->to << maybeSubstitute;
}
@@ -225,7 +225,7 @@ StorePathSet RemoteStore::querySubstitutablePaths(const StorePathSet & paths)
{
auto conn(getConnection());
conn->to << WorkerProto::Op::QuerySubstitutablePaths;
WorkerProto::write(*this, *conn, paths);
conn->to << WorkerProto::write(*this, *conn, paths);
conn.processStderr();
return WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
}
@@ -243,9 +243,9 @@ void RemoteStore::querySubstitutablePathInfos(const StorePathCAMap & pathsMap, S
StorePathSet paths;
for (auto & path : pathsMap)
paths.insert(path.first);
WorkerProto::write(*this, *conn, paths);
conn->to << WorkerProto::write(*this, *conn, paths);
} else
WorkerProto::write(*this, *conn, pathsMap);
conn->to << WorkerProto::write(*this, *conn, pathsMap);
conn.processStderr();
size_t count = readNum<size_t>(conn->from);
for (size_t n = 0; n < count; n++) {
@@ -377,7 +377,7 @@ ref<const ValidPathInfo> RemoteStore::addCAToStore(
<< WorkerProto::Op::AddToStore
<< name
<< caMethod.render(hashType);
WorkerProto::write(*this, *conn, references);
conn->to << WorkerProto::write(*this, *conn, references);
conn->to << repair;
// The dump source may invoke the store, so we need to make some room.
@@ -402,7 +402,7 @@ ref<const ValidPathInfo> RemoteStore::addCAToStore(
name, printHashType(hashType));
std::string s = dump.drain();
conn->to << WorkerProto::Op::AddTextToStore << name << s;
WorkerProto::write(*this, *conn, references);
conn->to << WorkerProto::write(*this, *conn, references);
conn.processStderr();
},
[&](const FileIngestionMethod & fim) -> void {
@@ -422,7 +422,7 @@ ref<const ValidPathInfo> RemoteStore::addCAToStore(
dump.drainInto(conn->to);
} else {
std::string contents = dump.drain();
dumpString(contents, conn->to);
conn->to << dumpString(contents);
}
}
conn.processStderr();
@@ -462,14 +462,14 @@ void RemoteStore::addToStore(const ValidPathInfo & info, Source & source,
<< printStorePath(info.path)
<< (info.deriver ? printStorePath(*info.deriver) : "")
<< info.narHash.to_string(Base16, false);
WorkerProto::write(*this, *conn, info.references);
conn->to << WorkerProto::write(*this, *conn, info.references);
conn->to << info.registrationTime << info.narSize
<< info.ultimate << info.sigs << renderContentAddress(info.ca)
<< repair << !checkSigs;
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 23) {
conn.withFramedSink([&](Sink & sink) {
copyNAR(source, sink);
sink << copyNAR(source);
});
} else {
conn.processStderr(0, &source);
@@ -485,17 +485,25 @@ void RemoteStore::addMultipleToStore(
{
auto remoteVersion = getProtocol();
auto source = sinkToSource([&](Sink & sink) {
sink << pathsToCopy.size();
GeneratorSource source{[](auto self, auto & pathsToCopy, auto remoteVersion) -> WireFormatGenerator {
NullSink null; // TODO remove .to from WriteConn instead
co_yield pathsToCopy.size();
for (auto & [pathInfo, pathSource] : pathsToCopy) {
WorkerProto::Serialise<ValidPathInfo>::write(*this,
WorkerProto::WriteConn {sink, remoteVersion},
co_yield WorkerProto::Serialise<ValidPathInfo>::write(*self,
WorkerProto::WriteConn {null, remoteVersion},
pathInfo);
pathSource->drainInto(sink);
try {
char buf[65536];
while (true) {
const auto read = pathSource->read(buf, sizeof(buf));
co_yield std::span{buf, read};
}
});
} catch (EndOfFile &) {
}
}
}(this, pathsToCopy, remoteVersion)};
addMultipleToStore(*source, repair, checkSigs);
addMultipleToStore(source, repair, checkSigs);
}
void RemoteStore::addMultipleToStore(
@@ -536,7 +544,7 @@ void RemoteStore::registerDrvOutput(const Realisation & info)
conn->to << info.id.to_string();
conn->to << std::string(info.outPath.to_string());
} else {
WorkerProto::write(*this, *conn, info);
conn->to << WorkerProto::write(*this, *conn, info);
}
conn.processStderr();
}
@@ -597,7 +605,7 @@ void RemoteStore::buildPaths(const std::vector<DerivedPath> & drvPaths, BuildMod
auto conn(getConnection());
conn->to << WorkerProto::Op::BuildPaths;
WorkerProto::write(*this, *conn, drvPaths);
conn->to << WorkerProto::write(*this, *conn, drvPaths);
conn->to << buildMode;
conn.processStderr();
readInt(conn->from);
@@ -615,7 +623,7 @@ std::vector<KeyedBuildResult> RemoteStore::buildPathsWithResults(
if (GET_PROTOCOL_MINOR(conn->daemonVersion) >= 34) {
conn->to << WorkerProto::Op::BuildPathsWithResults;
WorkerProto::write(*this, *conn, paths);
conn->to << WorkerProto::write(*this, *conn, paths);
conn->to << buildMode;
conn.processStderr();
return WorkerProto::Serialise<std::vector<KeyedBuildResult>>::read(*this, *conn);
@@ -740,7 +748,7 @@ void RemoteStore::collectGarbage(const GCOptions & options, GCResults & results)
conn->to
<< WorkerProto::Op::CollectGarbage << options.action;
WorkerProto::write(*this, *conn, options.pathsToDelete);
conn->to << WorkerProto::write(*this, *conn, options.pathsToDelete);
conn->to << options.ignoreLiveness
<< options.maxFreed
/* removed options */
@@ -792,7 +800,7 @@ void RemoteStore::queryMissing(const std::vector<DerivedPath> & targets,
{
auto conn(getConnection());
conn->to << WorkerProto::Op::QueryMissing;
WorkerProto::write(*this, *conn, targets);
conn->to << WorkerProto::write(*this, *conn, targets);
conn.processStderr();
willBuild = WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
willSubstitute = WorkerProto::Serialise<StorePathSet>::read(*this, *conn);
@@ -848,12 +856,14 @@ RemoteStore::Connection::~Connection()
}
}
void RemoteStore::narFromPath(const StorePath & path, Sink & sink)
WireFormatGenerator RemoteStore::narFromPath(const StorePath & path)
{
auto conn(connections->get());
conn->to << WorkerProto::Op::NarFromPath << printStorePath(path);
conn->processStderr();
copyNAR(conn->from, sink);
return [](auto conn) -> WireFormatGenerator {
co_yield copyNAR(conn->from);
}(std::move(conn));
}
ref<FSAccessor> RemoteStore::getFSAccessor()
@@ -897,7 +907,7 @@ std::exception_ptr RemoteStore::Connection::processStderr(Sink * sink, Source *
if (!source) throw Error("no source");
size_t len = readNum<size_t>(from);
auto buf = std::make_unique<char[]>(len);
writeString({(const char *) buf.get(), source->read(buf.get(), len)}, to);
to << std::string_view((const char *) buf.get(), source->read(buf.get(), len));
to.flush();
}
+1 -1
View File
@@ -183,7 +183,7 @@ protected:
virtual ref<FSAccessor> getFSAccessor() override;
virtual void narFromPath(const StorePath & path, Sink & sink) override;
virtual WireFormatGenerator narFromPath(const StorePath & path) override;
private:
+6 -2
View File
@@ -455,7 +455,7 @@ struct S3BinaryCacheStoreImpl : virtual S3BinaryCacheStoreConfig, public virtual
uploadFile(path, istream, mimeType, "");
}
void getFile(const std::string & path, Sink & sink) override
box_ptr<Source> getFile(const std::string & path) override
{
stats.get++;
@@ -469,7 +469,11 @@ struct S3BinaryCacheStoreImpl : virtual S3BinaryCacheStoreConfig, public virtual
printTalkative("downloaded 's3://%s/%s' (%d bytes) in %d ms",
bucketName, path, res.data->size(), res.durationMs);
sink(*res.data);
return make_box_ptr<GeneratorSource>(
[](std::string data) -> Generator<Bytes> {
co_yield std::span{data.data(), data.size()};
}(std::move(*res.data))
);
} else
throw NoSuchBinaryCacheFile("file '%s' does not exist in binary cache '%s'", path, getUri());
}
+5 -4
View File
@@ -20,9 +20,9 @@ namespace nix {
{ \
return LengthPrefixedProtoHelper<ServeProto, T >::read(store, conn); \
} \
TEMPLATE void ServeProto::Serialise< T >::write(const Store & store, ServeProto::WriteConn conn, const T & t) \
TEMPLATE [[nodiscard]] WireFormatGenerator ServeProto::Serialise< T >::write(const Store & store, ServeProto::WriteConn conn, const T & t) \
{ \
LengthPrefixedProtoHelper<ServeProto, T >::write(store, conn, t); \
return LengthPrefixedProtoHelper<ServeProto, T >::write(store, conn, t); \
}
SERVE_USE_LENGTH_PREFIX_SERIALISER(template<typename T>, std::vector<T>)
@@ -46,9 +46,10 @@ struct ServeProto::Serialise
return CommonProto::Serialise<T>::read(store,
CommonProto::ReadConn { .from = conn.from });
}
static void write(const Store & store, ServeProto::WriteConn conn, const T & t)
[[nodiscard]]
static WireFormatGenerator write(const Store & store, ServeProto::WriteConn conn, const T & t)
{
CommonProto::Serialise<T>::write(store,
return CommonProto::Serialise<T>::write(store,
CommonProto::WriteConn { .to = conn.to },
t);
}
+20 -23
View File
@@ -34,23 +34,22 @@ BuildResult ServeProto::Serialise<BuildResult>::read(const Store & store, ServeP
return status;
}
void ServeProto::Serialise<BuildResult>::write(const Store & store, ServeProto::WriteConn conn, const BuildResult & status)
WireFormatGenerator ServeProto::Serialise<BuildResult>::write(const Store & store, ServeProto::WriteConn conn, const BuildResult & status)
{
conn.to
<< status.status
<< status.errorMsg;
co_yield status.status;
co_yield status.errorMsg;
if (GET_PROTOCOL_MINOR(conn.version) >= 3)
conn.to
<< status.timesBuilt
<< status.isNonDeterministic
<< status.startTime
<< status.stopTime;
if (GET_PROTOCOL_MINOR(conn.version) >= 3) {
co_yield status.timesBuilt;
co_yield status.isNonDeterministic;
co_yield status.startTime;
co_yield status.stopTime;
}
if (GET_PROTOCOL_MINOR(conn.version) >= 6) {
DrvOutputs builtOutputs;
for (auto & [output, realisation] : status.builtOutputs)
builtOutputs.insert_or_assign(realisation.id, realisation);
ServeProto::write(store, conn, builtOutputs);
co_yield ServeProto::write(store, conn, builtOutputs);
}
}
@@ -80,21 +79,19 @@ UnkeyedValidPathInfo ServeProto::Serialise<UnkeyedValidPathInfo>::read(const Sto
return info;
}
void ServeProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & info)
WireFormatGenerator ServeProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & info)
{
conn.to
<< (info.deriver ? store.printStorePath(*info.deriver) : "");
co_yield (info.deriver ? store.printStorePath(*info.deriver) : "");
ServeProto::write(store, conn, info.references);
co_yield ServeProto::write(store, conn, info.references);
// !!! Maybe we want compression?
conn.to
<< info.narSize // downloadSize, lie a little
<< info.narSize;
if (GET_PROTOCOL_MINOR(conn.version) >= 4)
conn.to
<< info.narHash.to_string(Base32, true)
<< renderContentAddress(info.ca)
<< info.sigs;
co_yield info.narSize; // downloadSize, lie a little
co_yield info.narSize;
if (GET_PROTOCOL_MINOR(conn.version) >= 4) {
co_yield info.narHash.to_string(Base32, true);
co_yield renderContentAddress(info.ca);
co_yield info.sigs;
}
}
}
+5 -4
View File
@@ -79,7 +79,7 @@ struct ServeProto
#if 0
{
static T read(const Store & store, ReadConn conn);
static void write(const Store & store, WriteConn conn, const T & t);
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t);
};
#endif
@@ -88,9 +88,10 @@ struct ServeProto
* infer the type instead of having to write it down explicitly.
*/
template<typename T>
static void write(const Store & store, WriteConn conn, const T & t)
[[nodiscard]]
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t)
{
ServeProto::Serialise<T>::write(store, conn, t);
return ServeProto::Serialise<T>::write(store, conn, t);
}
};
@@ -142,7 +143,7 @@ inline std::ostream & operator << (std::ostream & s, ServeProto::Command op)
struct ServeProto::Serialise< T > \
{ \
static T read(const Store & store, ServeProto::ReadConn conn); \
static void write(const Store & store, ServeProto::WriteConn conn, const T & t); \
[[nodiscard]] static WireFormatGenerator write(const Store & store, ServeProto::WriteConn conn, const T & t); \
};
template<>
+32 -35
View File
@@ -275,13 +275,11 @@ StorePath Store::addToStore(
const StorePathSet & references)
{
Path srcPath(absPath(_srcPath));
auto source = sinkToSource([&](Sink & sink) {
if (method == FileIngestionMethod::Recursive)
dumpPath(srcPath, sink, filter);
else
readFileSource(srcPath)->drainInto(sink);
});
return addToStoreFromDump(*source, name, method, hashAlgo, repair, references);
auto source = GeneratorSource{
method == FileIngestionMethod::Recursive ? dumpPath(srcPath, filter).decay()
: readFileSource(srcPath)
};
return addToStoreFromDump(source, name, method, hashAlgo, repair, references);
}
void Store::addMultipleToStore(
@@ -337,9 +335,9 @@ void Store::addMultipleToStore(
info.ultimate = false;
/* Make sure that the Source object is destroyed when
we're done. In particular, a SinkToSource object must
be destroyed to ensure that the destructors on its
stack frame are run; this includes
we're done. In particular, a coroutine object must
be destroyed to ensure that the destructors in its
state are run; this includes
LegacySSHStore::narFromPath()'s connection lock. */
auto source = std::move(source_);
@@ -425,13 +423,11 @@ ValidPathInfo Store::addToStoreSlow(std::string_view name, const Path & srcPath,
/* Functionally, this means that fileSource will yield the content of
srcPath. The fact that we use scratchpadSink as a temporary buffer here
is an implementation detail. */
auto fileSource = sinkToSource([&](Sink & scratchpadSink) {
dumpPath(srcPath, scratchpadSink);
});
auto fileSource = GeneratorSource{dumpPath(srcPath)};
/* tapped provides the same data as fileSource, but we also write all the
information to narSink. */
TeeSource tapped { *fileSource, narSink };
TeeSource tapped { fileSource, narSink };
ParseSink blank;
auto & parseSink = method == FileIngestionMethod::Flat
@@ -466,10 +462,8 @@ ValidPathInfo Store::addToStoreSlow(std::string_view name, const Path & srcPath,
info.narSize = narSize;
if (!isValidPath(info.path)) {
auto source = sinkToSource([&](Sink & scratchpadSink) {
dumpPath(srcPath, scratchpadSink);
});
addToStore(info, *source);
auto source = GeneratorSource{dumpPath(srcPath)};
addToStore(info, source);
}
return info;
@@ -1065,16 +1059,17 @@ void copyStorePath(
info = info2;
}
auto source = sinkToSource([&](Sink & sink) {
LambdaSink progressSink([&, total = 0ULL](std::string_view data) mutable {
total += data.size();
GeneratorSource source{[](auto & act, auto & info, auto & srcStore, auto & storePath) -> WireFormatGenerator {
auto nar = srcStore.narFromPath(storePath);
uint64_t total = 0;
while (auto data = nar.next()) {
total += data->size();
act.progress(total, info->narSize);
});
TeeSink tee { sink, progressSink };
srcStore.narFromPath(storePath, tee);
});
co_yield *data;
}
}(act, info, srcStore, storePath)};
dstStore.addToStore(*info, *source, repair, checkSigs);
dstStore.addToStore(*info, source, repair, checkSigs);
}
@@ -1186,8 +1181,8 @@ std::map<StorePath, StorePath> copyPaths(
ValidPathInfo infoForDst = *info;
infoForDst.path = storePathForDst;
auto source =
sinkToSource([&srcStore, &dstStore, missingPath = missingPath, info = std::move(info)](Sink & sink) {
auto source = [](auto & srcStore, auto & dstStore, auto missingPath, auto info
) -> WireFormatGenerator {
// We can reasonably assume that the copy will happen whenever we
// read the path, so log something about that at that point
auto srcUri = srcStore.getUri();
@@ -1202,15 +1197,17 @@ std::map<StorePath, StorePath> copyPaths(
);
PushActivity pact(act.id);
LambdaSink progressSink([&, total = 0ULL](std::string_view data) mutable {
total += data.size();
auto nar = srcStore.narFromPath(missingPath);
uint64_t total = 0;
while (auto data = nar.next()) {
total += data->size();
act.progress(total, info->narSize);
co_yield *data;
}
};
pathsToCopy.push_back(std::pair{
infoForDst, std::make_unique<GeneratorSource>(source(srcStore, dstStore, missingPath, info))
});
TeeSink tee{sink, progressSink};
srcStore.narFromPath(missingPath, tee);
});
pathsToCopy.push_back(std::pair{infoForDst, std::move(source)});
}
dstStore.addMultipleToStore(pathsToCopy, act, repair, checkSigs);
+2 -2
View File
@@ -577,9 +577,9 @@ public:
{ return registerDrvOutput(output); }
/**
* Write a NAR dump of a store path.
* Generate a NAR dump of a store path.
*/
virtual void narFromPath(const StorePath & path, Sink & sink) = 0;
virtual WireFormatGenerator narFromPath(const StorePath & path) = 0;
/**
* For each path, if it's a derivation, build it. Building a
+2 -2
View File
@@ -38,8 +38,8 @@ public:
ref<FSAccessor> getFSAccessor() override
{ return LocalFSStore::getFSAccessor(); }
void narFromPath(const StorePath & path, Sink & sink) override
{ LocalFSStore::narFromPath(path, sink); }
WireFormatGenerator narFromPath(const StorePath & path) override
{ return LocalFSStore::narFromPath(path); }
/**
* Implementation of `IndirectRootStore::addIndirectRoot()` which
+5 -4
View File
@@ -20,9 +20,9 @@ namespace nix {
{ \
return LengthPrefixedProtoHelper<WorkerProto, T >::read(store, conn); \
} \
TEMPLATE void WorkerProto::Serialise< T >::write(const Store & store, WorkerProto::WriteConn conn, const T & t) \
TEMPLATE [[nodiscard]] WireFormatGenerator WorkerProto::Serialise< T >::write(const Store & store, WorkerProto::WriteConn conn, const T & t) \
{ \
LengthPrefixedProtoHelper<WorkerProto, T >::write(store, conn, t); \
return LengthPrefixedProtoHelper<WorkerProto, T >::write(store, conn, t); \
}
WORKER_USE_LENGTH_PREFIX_SERIALISER(template<typename T>, std::vector<T>)
@@ -46,9 +46,10 @@ struct WorkerProto::Serialise
return CommonProto::Serialise<T>::read(store,
CommonProto::ReadConn { .from = conn.from });
}
static void write(const Store & store, WorkerProto::WriteConn conn, const T & t)
[[nodiscard]]
static WireFormatGenerator write(const Store & store, WorkerProto::WriteConn conn, const T & t)
{
CommonProto::Serialise<T>::write(store,
return CommonProto::Serialise<T>::write(store,
CommonProto::WriteConn { .to = conn.to },
t);
}
+34 -37
View File
@@ -28,17 +28,17 @@ std::optional<TrustedFlag> WorkerProto::Serialise<std::optional<TrustedFlag>>::r
}
}
void WorkerProto::Serialise<std::optional<TrustedFlag>>::write(const Store & store, WorkerProto::WriteConn conn, const std::optional<TrustedFlag> & optTrusted)
WireFormatGenerator WorkerProto::Serialise<std::optional<TrustedFlag>>::write(const Store & store, WorkerProto::WriteConn conn, const std::optional<TrustedFlag> & optTrusted)
{
if (!optTrusted)
conn.to << (uint8_t)0;
co_yield (uint8_t)0;
else {
switch (*optTrusted) {
case Trusted:
conn.to << (uint8_t)1;
co_yield (uint8_t)1;
break;
case NotTrusted:
conn.to << (uint8_t)2;
co_yield (uint8_t)2;
break;
default:
assert(false);
@@ -57,23 +57,23 @@ DerivedPath WorkerProto::Serialise<DerivedPath>::read(const Store & store, Worke
}
}
void WorkerProto::Serialise<DerivedPath>::write(const Store & store, WorkerProto::WriteConn conn, const DerivedPath & req)
WireFormatGenerator WorkerProto::Serialise<DerivedPath>::write(const Store & store, WorkerProto::WriteConn conn, const DerivedPath & req)
{
if (GET_PROTOCOL_MINOR(conn.version) >= 30) {
conn.to << req.to_string_legacy(store);
co_yield req.to_string_legacy(store);
} else {
auto sOrDrvPath = StorePathWithOutputs::tryFromDerivedPath(req);
std::visit(overloaded {
[&](const StorePathWithOutputs & s) {
conn.to << s.to_string(store);
co_yield std::visit(overloaded {
[&](const StorePathWithOutputs & s) -> std::string {
return s.to_string(store);
},
[&](const StorePath & drvPath) {
[&](const StorePath & drvPath) -> std::string {
throw Error("trying to request '%s', but daemon protocol %d.%d is too old (< 1.29) to request a derivation file",
store.printStorePath(drvPath),
GET_PROTOCOL_MAJOR(conn.version),
GET_PROTOCOL_MINOR(conn.version));
},
[&](std::monostate) {
[&](std::monostate) -> std::string {
throw Error("wanted to build a derivation that is itself a build product, but protocols do not support that. Try upgrading the Nix implementation on the other end of this connection");
},
}, sOrDrvPath);
@@ -91,10 +91,10 @@ KeyedBuildResult WorkerProto::Serialise<KeyedBuildResult>::read(const Store & st
};
}
void WorkerProto::Serialise<KeyedBuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const KeyedBuildResult & res)
WireFormatGenerator WorkerProto::Serialise<KeyedBuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const KeyedBuildResult & res)
{
WorkerProto::write(store, conn, res.path);
WorkerProto::write(store, conn, static_cast<const BuildResult &>(res));
co_yield WorkerProto::write(store, conn, res.path);
co_yield WorkerProto::write(store, conn, static_cast<const BuildResult &>(res));
}
@@ -120,23 +120,21 @@ BuildResult WorkerProto::Serialise<BuildResult>::read(const Store & store, Worke
return res;
}
void WorkerProto::Serialise<BuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const BuildResult & res)
WireFormatGenerator WorkerProto::Serialise<BuildResult>::write(const Store & store, WorkerProto::WriteConn conn, const BuildResult & res)
{
conn.to
<< res.status
<< res.errorMsg;
co_yield res.status;
co_yield res.errorMsg;
if (GET_PROTOCOL_MINOR(conn.version) >= 29) {
conn.to
<< res.timesBuilt
<< res.isNonDeterministic
<< res.startTime
<< res.stopTime;
co_yield res.timesBuilt;
co_yield res.isNonDeterministic;
co_yield res.startTime;
co_yield res.stopTime;
}
if (GET_PROTOCOL_MINOR(conn.version) >= 28) {
DrvOutputs builtOutputs;
for (auto & [output, realisation] : res.builtOutputs)
builtOutputs.insert_or_assign(realisation.id, realisation);
WorkerProto::write(store, conn, builtOutputs);
co_yield WorkerProto::write(store, conn, builtOutputs);
}
}
@@ -150,10 +148,10 @@ ValidPathInfo WorkerProto::Serialise<ValidPathInfo>::read(const Store & store, R
};
}
void WorkerProto::Serialise<ValidPathInfo>::write(const Store & store, WriteConn conn, const ValidPathInfo & pathInfo)
WireFormatGenerator WorkerProto::Serialise<ValidPathInfo>::write(const Store & store, WriteConn conn, const ValidPathInfo & pathInfo)
{
WorkerProto::write(store, conn, pathInfo.path);
WorkerProto::write(store, conn, static_cast<const UnkeyedValidPathInfo &>(pathInfo));
co_yield WorkerProto::write(store, conn, pathInfo.path);
co_yield WorkerProto::write(store, conn, static_cast<const UnkeyedValidPathInfo &>(pathInfo));
}
@@ -173,18 +171,17 @@ UnkeyedValidPathInfo WorkerProto::Serialise<UnkeyedValidPathInfo>::read(const St
return info;
}
void WorkerProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & pathInfo)
WireFormatGenerator WorkerProto::Serialise<UnkeyedValidPathInfo>::write(const Store & store, WriteConn conn, const UnkeyedValidPathInfo & pathInfo)
{
conn.to
<< (pathInfo.deriver ? store.printStorePath(*pathInfo.deriver) : "")
<< pathInfo.narHash.to_string(Base16, false);
WorkerProto::write(store, conn, pathInfo.references);
conn.to << pathInfo.registrationTime << pathInfo.narSize;
co_yield (pathInfo.deriver ? store.printStorePath(*pathInfo.deriver) : "");
co_yield pathInfo.narHash.to_string(Base16, false);
co_yield WorkerProto::write(store, conn, pathInfo.references);
co_yield pathInfo.registrationTime;
co_yield pathInfo.narSize;
conn.to
<< pathInfo.ultimate
<< pathInfo.sigs
<< renderContentAddress(pathInfo.ca);
co_yield pathInfo.ultimate;
co_yield pathInfo.sigs;
co_yield renderContentAddress(pathInfo.ca);
}
}
+5 -4
View File
@@ -122,7 +122,7 @@ struct WorkerProto
#if 0
{
static T read(const Store & store, ReadConn conn);
static void write(const Store & store, WriteConn conn, const T & t);
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t);
};
#endif
@@ -131,9 +131,10 @@ struct WorkerProto
* infer the type instead of having to write it down explicitly.
*/
template<typename T>
static void write(const Store & store, WriteConn conn, const T & t)
[[nodiscard]]
static WireFormatGenerator write(const Store & store, WriteConn conn, const T & t)
{
WorkerProto::Serialise<T>::write(store, conn, t);
return WorkerProto::Serialise<T>::write(store, conn, t);
}
};
@@ -219,7 +220,7 @@ inline std::ostream & operator << (std::ostream & s, WorkerProto::Op op)
struct WorkerProto::Serialise< T > \
{ \
static T read(const Store & store, WorkerProto::ReadConn conn); \
static void write(const Store & store, WorkerProto::WriteConn conn, const T & t); \
[[nodiscard]] static WireFormatGenerator write(const Store & store, WorkerProto::WriteConn conn, const T & t); \
};
template<>
+205 -127
View File
@@ -1,5 +1,6 @@
#include <cerrno>
#include <algorithm>
#include <string_view>
#include <vector>
#include <map>
@@ -13,6 +14,8 @@
#include "archive.hh"
#include "file-system.hh"
#include "finally.hh"
#include "serialise.hh"
#include "config.hh"
#include "logging.hh"
#include "signals.hh"
@@ -40,10 +43,10 @@ static GlobalConfig::Register rArchiveSettings(&archiveSettings);
PathFilter defaultPathFilter = [](const Path &) { return true; };
static void dumpContents(const Path & path, off_t size,
Sink & sink)
static WireFormatGenerator dumpContents(const Path & path, off_t size)
{
sink << "contents" << size;
co_yield "contents";
co_yield size;
AutoCloseFD fd{open(path.c_str(), O_RDONLY | O_CLOEXEC)};
if (!fd) throw SysError("opening file '%1%'", path);
@@ -55,31 +58,35 @@ static void dumpContents(const Path & path, off_t size,
auto n = std::min(left, buf.size());
readFull(fd.get(), buf.data(), n);
left -= n;
sink({buf.data(), n});
co_yield std::span{buf.data(), n};
}
writePadding(size, sink);
co_yield SerializingTransform::padding(size);
}
static time_t dump(const Path & path, Sink & sink, PathFilter & filter)
static WireFormatGenerator dump(const Path & path, time_t & mtime, PathFilter & filter)
{
checkInterrupt();
auto st = lstat(path);
time_t result = st.st_mtime;
mtime = st.st_mtime;
sink << "(";
co_yield "(";
if (S_ISREG(st.st_mode)) {
sink << "type" << "regular";
if (st.st_mode & S_IXUSR)
sink << "executable" << "";
dumpContents(path, st.st_size, sink);
co_yield "type";
co_yield "regular";
if (st.st_mode & S_IXUSR) {
co_yield "executable";
co_yield "";
}
co_yield dumpContents(path, st.st_size);
}
else if (S_ISDIR(st.st_mode)) {
sink << "type" << "directory";
co_yield "type";
co_yield "directory";
/* If we're on a case-insensitive system like macOS, undo
the case hack applied by restorePath(). */
@@ -101,41 +108,55 @@ static time_t dump(const Path & path, Sink & sink, PathFilter & filter)
for (auto & i : unhacked)
if (filter(path + "/" + i.first)) {
sink << "entry" << "(" << "name" << i.first << "node";
auto tmp_mtime = dump(path + "/" + i.second, sink, filter);
if (tmp_mtime > result) {
result = tmp_mtime;
co_yield "entry";
co_yield "(";
co_yield "name";
co_yield i.first;
co_yield "node";
time_t tmp_mtime;
co_yield dump(path + "/" + i.second, tmp_mtime, filter);
if (tmp_mtime > mtime) {
mtime = tmp_mtime;
}
sink << ")";
co_yield ")";
}
}
else if (S_ISLNK(st.st_mode))
sink << "type" << "symlink" << "target" << readLink(path);
else if (S_ISLNK(st.st_mode)) {
co_yield "type";
co_yield "symlink";
co_yield "target";
co_yield readLink(path);
}
else throw Error("file '%1%' has an unsupported type", path);
sink << ")";
return result;
co_yield ")";
}
time_t dumpPathAndGetMtime(const Path & path, Sink & sink, PathFilter & filter)
WireFormatGenerator dumpPathAndGetMtime(Path path, time_t & mtime, PathFilter & filter)
{
sink << narVersionMagic1;
return dump(path, sink, filter);
co_yield narVersionMagic1;
co_yield dump(std::move(path), mtime, filter);
}
void dumpPath(const Path & path, Sink & sink, PathFilter & filter)
WireFormatGenerator dumpPath(Path path, PathFilter & filter)
{
dumpPathAndGetMtime(path, sink, filter);
time_t ignored;
co_yield dumpPathAndGetMtime(std::move(path), ignored, filter);
}
void dumpString(std::string_view s, Sink & sink)
WireFormatGenerator dumpString(std::string_view s)
{
sink << narVersionMagic1 << "(" << "type" << "regular" << "contents" << s << ")";
co_yield narVersionMagic1;
co_yield "(";
co_yield "type";
co_yield "regular";
co_yield "contents";
co_yield s;
co_yield ")";
}
@@ -156,28 +177,6 @@ static void skipGeneric(Source & source)
#endif
static void parseContents(ParseSink & sink, Source & source, const Path & path)
{
uint64_t size = readLongLong(source);
sink.preallocateContents(size);
uint64_t left = size;
std::array<char, 65536> buf;
while (left) {
checkInterrupt();
auto n = buf.size();
if ((uint64_t)n > left) n = left;
source(buf.data(), n);
sink.receiveContents({buf.data(), n});
left -= n;
}
readPadding(size, source);
}
struct CaseInsensitiveCompare
{
bool operator() (const std::string & a, const std::string & b) const
@@ -186,80 +185,94 @@ struct CaseInsensitiveCompare
}
};
namespace nar {
static void parse(ParseSink & sink, Source & source, const Path & path)
static Generator<Entry> parseObject(Source & source, const Path & path)
{
std::string s;
#define EXPECT(raw, kind) \
do { \
const auto s = readString(source); \
if (s != raw) { \
throw badArchive("expected " kind " tag"); \
} \
co_yield MetadataString{s}; \
} while (0)
s = readString(source);
if (s != "(") throw badArchive("expected open tag");
EXPECT("(", "open");
EXPECT("type", "type");
enum { tpUnknown, tpRegular, tpDirectory, tpSymlink } type = tpUnknown;
std::map<Path, int, CaseInsensitiveCompare> names;
while (1) {
checkInterrupt();
s = readString(source);
if (s == ")") {
break;
}
else if (s == "type") {
if (type != tpUnknown)
throw badArchive("multiple type fields");
std::string t = readString(source);
const auto t = readString(source);
co_yield MetadataString{t};
if (t == "regular") {
type = tpRegular;
sink.createRegularFile(path);
}
else if (t == "directory") {
sink.createDirectory(path);
type = tpDirectory;
}
else if (t == "symlink") {
type = tpSymlink;
}
else throw badArchive("unknown file type " + t);
}
else if (s == "contents" && type == tpRegular) {
parseContents(sink, source, path);
sink.closeRegularFile();
}
else if (s == "executable" && type == tpRegular) {
auto contentsOrFlag = readString(source);
co_yield MetadataString{contentsOrFlag};
const bool executable = contentsOrFlag == "executable";
if (executable) {
auto s = readString(source);
if (s != "") throw badArchive("executable marker has non-empty value");
sink.isExecutable();
co_yield MetadataString{s};
if (s != "") {
throw badArchive("executable marker has non-empty value");
}
contentsOrFlag = readString(source);
co_yield MetadataString{contentsOrFlag};
}
if (contentsOrFlag == "contents") {
uint64_t size = readLongLong(source);
co_yield MetadataRaw{SerializingTransform()(size)};
auto reader = [](Source & source, uint64_t left) -> Generator<Bytes> {
std::array<char, 65536> buf;
else if (s == "entry" && type == tpDirectory) {
std::string name, prevName;
s = readString(source);
if (s != "(") throw badArchive("expected open tag");
while (left) {
checkInterrupt();
auto n = std::min<uint64_t>(buf.size(), left);
source(buf.data(), n);
co_yield std::span{buf.data(), n};
left -= n;
}
}(source, size);
co_yield File{path, executable, size, reader};
while (reader.next()) {
// ignore remainder of contents
}
readPadding(size, source);
co_yield MetadataRaw{SerializingTransform::padding(size)};
} else {
throw badArchive("file without contents found: " + path);
}
} else if (t == "directory") {
auto reader = [](Source & source, const Path & path) -> Generator<Entry> {
std::map<Path, int, CaseInsensitiveCompare> names;
std::string prevName;
while (1) {
checkInterrupt();
s = readString(source);
{
const auto s = readString(source);
co_yield MetadataString{s};
if (s == ")") {
break;
} else if (s == "name") {
name = readString(source);
if (name.empty() || name == "." || name == ".." || name.find('/') != std::string::npos || name.find((char) 0) != std::string::npos)
} else if (s != "entry") {
throw badArchive("expected entry tag");
}
EXPECT("(", "open");
}
EXPECT("name", "name");
auto name = readString(source);
co_yield MetadataString{name};
if (name.empty() || name == "." || name == ".."
|| name.find('/') != std::string::npos
|| name.find((char) 0) != std::string::npos)
{
throw Error("NAR contains invalid file name '%1%'", name);
if (name <= prevName)
}
if (name <= prevName) {
throw Error("NAR directory is not sorted");
}
prevName = name;
if (archiveSettings.useCaseHack) {
auto i = names.find(name);
@@ -267,42 +280,109 @@ static void parse(ParseSink & sink, Source & source, const Path & path)
debug("case collision between '%1%' and '%2%'", i->first, name);
name += caseHackSuffix;
name += std::to_string(++i->second);
} else
} else {
names[name] = 0;
}
} else if (s == "node") {
if (name.empty()) throw badArchive("entry name missing");
parse(sink, source, path + "/" + name);
} else
throw badArchive("unknown field " + s);
}
}
else if (s == "target" && type == tpSymlink) {
EXPECT("node", "node");
co_yield parseObject(source, path + "/" + name);
EXPECT(")", "close");
}
}(source, path);
co_yield Directory{path, reader};
while (reader.next()) {
// ignore remaining entries
}
// directories are terminated already, don't try to read another ")"
co_return;
} else if (t == "symlink") {
EXPECT("target", "target");
std::string target = readString(source);
sink.createSymlink(path, target);
co_yield MetadataString{target};
co_yield Symlink{path, target};
} else {
throw badArchive("unknown file type " + t);
}
else
throw badArchive("unknown field " + s);
}
EXPECT(")", "close");
#undef EXPECT
}
void parseDump(ParseSink & sink, Source & source)
Generator<Entry> parse(Source & source)
{
std::string version;
try {
version = readString(source, narVersionMagic1.size());
co_yield MetadataString{version};
} catch (SerialisationError & e) {
/* This generally means the integer at the start couldn't be
decoded. Ignore and throw the exception below. */
}
if (version != narVersionMagic1)
throw badArchive("input doesn't look like a Nix archive");
parse(sink, source, "");
co_yield parseObject(source, "");
}
}
static WireFormatGenerator restore(ParseSink & sink, Generator<nar::Entry> nar)
{
while (auto entry = nar.next()) {
co_yield std::visit(
overloaded{
[](nar::MetadataString m) -> WireFormatGenerator {
co_yield m.data;
},
[](nar::MetadataRaw r) -> WireFormatGenerator {
co_yield r.raw;
},
[&](nar::File f) {
return [](auto f, auto & sink) -> WireFormatGenerator {
sink.createRegularFile(f.path);
sink.preallocateContents(f.size);
if (f.executable) {
sink.isExecutable();
}
while (auto block = f.contents.next()) {
sink.receiveContents(std::string_view{block->data(), block->size()});
co_yield *block;
}
sink.closeRegularFile();
}(std::move(f), sink);
},
[&](nar::Symlink sl) {
return [](auto sl, auto & sink) -> WireFormatGenerator {
sink.createSymlink(sl.path, sl.target);
co_return;
}(std::move(sl), sink);
},
[&](nar::Directory d) {
return [](auto d, auto & sink) -> WireFormatGenerator {
sink.createDirectory(d.path);
return restore(sink, std::move(d.contents));
}(std::move(d), sink);
},
},
std::move(*entry)
);
}
}
WireFormatGenerator parseAndCopyDump(ParseSink & sink, Source & source)
{
return restore(sink, nar::parse(source));
}
void parseDump(ParseSink & sink, Source & source)
{
auto parser = parseAndCopyDump(sink, source);
while (parser.next()) {
// ignore the actual item
}
}
struct RestoreSink : ParseSink
{
@@ -377,16 +457,14 @@ void restorePath(const Path & path, Source & source)
}
void copyNAR(Source & source, Sink & sink)
WireFormatGenerator copyNAR(Source & source)
{
// FIXME: if 'source' is the output of dumpPath() followed by EOF,
// we should just forward all data directly without parsing.
ParseSink parseSink; /* null sink; just parse the NAR */
static ParseSink parseSink; /* null sink; just parse the NAR */
TeeSource wrapper { source, sink };
parseDump(parseSink, wrapper);
return parseAndCopyDump(parseSink, source);
}
+56 -5
View File
@@ -1,6 +1,7 @@
#pragma once
///@file
#include "generator.hh"
#include "types.hh"
#include "serialise.hh"
#include "file-system.hh"
@@ -57,13 +58,13 @@ namespace nix {
* `+` denotes string concatenation.
* ```
*/
void dumpPath(const Path & path, Sink & sink,
WireFormatGenerator dumpPath(Path path,
PathFilter & filter = defaultPathFilter);
/**
* Same as dumpPath(), but returns the last modified date of the path.
*/
time_t dumpPathAndGetMtime(const Path & path, Sink & sink,
WireFormatGenerator dumpPathAndGetMtime(Path path, time_t & mtime,
PathFilter & filter = defaultPathFilter);
/**
@@ -71,7 +72,7 @@ time_t dumpPathAndGetMtime(const Path & path, Sink & sink,
*
* @param s Contents of the file.
*/
void dumpString(std::string_view s, Sink & sink);
WireFormatGenerator dumpString(std::string_view s);
/**
* \todo Fix this API, it sucks.
@@ -116,14 +117,64 @@ struct RetrieveRegularNARSink : ParseSink
}
};
namespace nar {
// FIXME: the Generator & are needed here because gcc 12 has a bug that copies
// move-only values that are yielded through types without user-defined ctors,
// such as File and Directory. gcc 13 fixes this, but even then it's easier to
// use refs anyway because it lets us drain incompletely read entries of a nar
// without resorting to Finally trickery. eventually this should be fixed with
// a more reasonable format though that doesn't need quite this weird a parser
struct MetadataString;
struct MetadataRaw;
struct File;
struct Symlink;
struct Directory;
using Entry = std::variant<MetadataString, MetadataRaw, File, Symlink, Directory>;
struct MetadataString
{
std::string_view data;
};
struct MetadataRaw
{
Bytes raw;
};
struct File
{
const Path & path;
bool executable;
uint64_t size;
Generator<Bytes> & contents;
};
struct Symlink
{
const Path & path, target;
};
struct Directory
{
const Path & path;
Generator<Entry> & contents;
};
Generator<Entry> parse(Source & source);
}
WireFormatGenerator parseAndCopyDump(ParseSink & sink, Source & source);
void parseDump(ParseSink & sink, Source & source);
void restorePath(const Path & path, Source & source);
/**
* Read a NAR from 'source' and write it to 'sink'.
* Read a NAR from 'source' and return it as a generator.
*/
void copyNAR(Source & source, Sink & sink);
WireFormatGenerator copyNAR(Source & source);
inline constexpr std::string_view narVersionMagic1 = "nix-archive-1";
-25
View File
@@ -198,31 +198,6 @@ std::string decompress(const std::string & method, std::string_view in)
return filter->drain();
}
std::unique_ptr<FinishSink> makeDecompressionSink(const std::string & method, Sink & nextSink)
{
if (method == "none" || method == "")
return std::make_unique<NoneSink>(nextSink);
else if (method == "br")
return sourceToSink([&](Source & source) {
BrotliDecompressionSource wrapped{source};
wrapped.drainInto(nextSink);
// special handling because sourceToSink is screwy: try
// to read the source one final time and fail when that
// succeeds (to reject trailing garbage in input data).
try {
char buf;
source(&buf, 1);
throw Error("garbage at end of brotli stream detected");
} catch (EndOfFile &) {
}
});
else
return sourceToSink([&](Source & source) {
auto decompressionSource = std::make_unique<ArchiveDecompressionSource>(source);
decompressionSource->drainInto(nextSink);
});
}
std::unique_ptr<Source> makeDecompressionSource(const std::string & method, Source & inner)
{
if (method == "none" || method == "") {
-1
View File
@@ -18,7 +18,6 @@ struct CompressionSink : BufferedSink, FinishSink
std::string decompress(const std::string & method, std::string_view in);
std::unique_ptr<FinishSink> makeDecompressionSink(const std::string & method, Sink & nextSink);
std::unique_ptr<Source> makeDecompressionSource(const std::string & method, Source & inner);
std::string compress(const std::string & method, std::string_view in, const bool parallel = false, int level = -1);
+3 -3
View File
@@ -81,12 +81,12 @@ std::string drainFD(int fd, bool block, const size_t reserveSize)
// the parser needs two extra bytes to append terminating characters, other users will
// not care very much about the extra memory.
StringSink sink(reserveSize + 2);
drainFD(fd, sink, block);
sink << drainFDSource(fd, block);
return std::move(sink.s);
}
void drainFD(int fd, Sink & sink, bool block)
Generator<Bytes> drainFDSource(int fd, bool block)
{
// silence GCC maybe-uninitialized warning in finally
int saved = 0;
@@ -115,7 +115,7 @@ void drainFD(int fd, Sink & sink, bool block)
throw SysError("reading from file");
}
else if (rd == 0) break;
else sink({(char *) buf.data(), (size_t) rd});
else co_yield std::span{(char *) buf.data(), (size_t) rd};
}
}
+2 -1
View File
@@ -2,6 +2,7 @@
///@file
#include "error.hh"
#include "generator.hh"
namespace nix {
@@ -35,7 +36,7 @@ void writeFull(int fd, std::string_view s, bool allowInterrupts = true);
*/
std::string drainFD(int fd, bool block = true, const size_t reserveSize=0);
void drainFD(int fd, Sink & sink, bool block = true);
Generator<Bytes> drainFDSource(int fd, bool block = true);
class AutoCloseFD
{
+4 -7
View File
@@ -289,17 +289,14 @@ std::string readFile(const Path & path)
}
box_ptr<Source> readFileSource(const Path & path)
Generator<Bytes> readFileSource(const Path & path)
{
AutoCloseFD fd{open(path.c_str(), O_RDONLY | O_CLOEXEC)};
if (!fd)
throw SysError("opening file '%s'", path);
struct FileSource : FdSource {
AutoCloseFD fd;
explicit FileSource(AutoCloseFD fd) : FdSource(fd.get()), fd(std::move(fd)) {}
};
return make_box_ptr<FileSource>(std::move(fd));
return [](AutoCloseFD fd) -> Generator<Bytes> {
co_yield drainFDSource(fd.get());
}(std::move(fd));
}
+1 -1
View File
@@ -143,7 +143,7 @@ unsigned char getFileType(const Path & path);
* Read the contents of a file into a string.
*/
std::string readFile(const Path & path);
box_ptr<Source> readFileSource(const Path & path);
Generator<Bytes> readFileSource(const Path & path);
/**
* Write a string to a file.
+250
View File
@@ -0,0 +1,250 @@
#pragma once
///@file
#include "types.hh"
#include <coroutine>
#include <exception>
#include <optional>
#include <utility>
#include <variant>
namespace nix {
template<typename T, typename Transform>
struct Generator;
namespace _generator {
template<typename T>
struct promise_state;
template<typename T>
struct GeneratorBase;
struct finished {};
template<typename T>
struct link
{
std::coroutine_handle<> handle{};
promise_state<T> * state{};
};
struct failure
{
std::exception_ptr e;
};
template<typename T>
struct promise_state
{
std::variant<T, link<T>, failure, finished> value{};
link<T> parent{};
};
template<typename T, typename Transform>
struct promise : promise_state<T>
{
using transform_t = std::conditional_t<std::is_void_v<Transform>, std::identity, Transform>;
transform_t convert;
std::optional<GeneratorBase<T>> inner;
Generator<T, Transform> get_return_object()
{
auto h = std::coroutine_handle<promise>::from_promise(*this);
return Generator<T, Transform>(GeneratorBase<T>(h, h.promise()));
}
std::suspend_always initial_suspend()
{
return {};
}
std::suspend_always final_suspend() noexcept
{
return {};
}
void unhandled_exception()
{
this->value = failure{std::current_exception()};
}
template<typename From>
requires requires(transform_t t, From && f) {
{
t(std::forward<From>(f))
} -> std::convertible_to<T>;
}
std::suspend_always yield_value(From && from)
{
this->value.template emplace<0>(convert(std::forward<From>(from)));
return {};
}
template<typename From>
requires requires(transform_t t, From && f) {
static_cast<Generator<T, void>>(t(std::forward<From>(f)));
}
std::suspend_always yield_value(From && from)
{
inner = static_cast<Generator<T, void>>(convert(std::forward<From>(from))).impl;
this->value = inner->active;
return {};
}
void return_void()
{
this->value = finished{};
}
};
template<typename T>
struct GeneratorBase
{
template<typename, typename>
friend struct Generator;
template<typename, typename>
friend struct promise;
// NOTE coroutine handles are LiteralType, own a memory resource (that may
// itself own unique resources), and are "typically TriviallyCopyable". we
// need to take special care to wrap this into a less footgunny interface.
GeneratorBase(GeneratorBase && other)
{
swap(other);
}
GeneratorBase & operator=(GeneratorBase && other)
{
GeneratorBase(std::move(other)).swap(*this);
return *this;
}
~GeneratorBase()
{
if (h) {
h.destroy();
}
}
std::optional<T> next()
{
while (active.handle) {
active.handle.resume();
auto & p = *active.state;
auto result = std::visit(
overloaded{
[&](finished) -> std::optional<T> {
active = p.parent;
return {};
},
[&](link<T> & inner) -> std::optional<T> {
auto base = inner.state;
while (base->parent.handle) {
base = base->parent.state;
}
base->parent = active;
active = inner;
return {};
},
[&](T & value) -> std::optional<T> { return std::move(value); },
[&](failure & f) -> std::optional<T> {
active = {};
std::rethrow_exception(f.e);
},
},
p.value
);
if (result) {
return std::move(result);
}
}
return std::nullopt;
}
protected:
std::coroutine_handle<> h{};
link<T> active{};
GeneratorBase(std::coroutine_handle<> h, promise_state<T> & state)
: h(h)
, active(h, &state)
{
}
void swap(GeneratorBase & other)
{
std::swap(h, other.h);
std::swap(active, other.active);
}
};
}
/// Coroutine-based iterator modeled loosely on Rust [`std::iter::Iterator`][iter]
/// interface. Like Rust's `Iterator` and unlike common C++ iterators, a Generator
/// returns `std::optional<T>` values from its next() function, but unlike both it
/// can also transform items produced within using a Transform function object the
/// Generator holds before returning them via next(). To allow generator nesting a
/// Transform may also return another Generator instance for any yielded value, in
/// this case the new Generator will temporarily take priority over the previously
/// running one and have its values returned until it is exhausted, then return to
/// the previous Generator. This mechanism may nest Generator to arbitrary depths.
///
/// \tparam T item type
/// \tparam Transform transform function object type, or `void` for no transform
///
/// [iter]: https://doc.rust-lang.org/stable/std/iter/trait.Iterator.html
template<typename T, typename Transform = void>
struct Generator
{
template<typename, typename>
friend struct _generator::promise;
template<typename, typename>
friend struct Generator;
using promise_type = _generator::promise<T, Transform>;
Generator(const Generator &) = delete;
Generator & operator=(const Generator &) = delete;
Generator(Generator &&) = default;
Generator & operator=(Generator &&) = default;
/// If the coroutine held by the Generator has not finished, runs it until it
/// yields a value, throws and exception, or returns. If the coroutine yields
/// a value this value is passed to a persistent instance of `Transform` that
/// is held by the Generator, and the result of this call is returned. If the
/// coroutine throws an exception, or the Transform throws an exception while
/// processing an item, that exception is rethrown and the Generator will not
/// return any more non-`std::nullopt` values from next(). Once the contained
/// coroutine has completed or an exception has been thrown the Generator can
/// no longer return any valid values, only `std::nullopt`. Exceptions thrown
/// are thrown only once, further invocations of next() return `std::nullopt`.
///
/// \returns `std::nullopt` if the coroutine has completed, or a value
std::optional<T> next()
{
return impl.next();
}
/// Type-erases the `Transform`.
///
/// \return a new Generator with the `Transform` type-erased
Generator<T, void> decay() &&
{
return Generator<T, void>(std::move(impl));
}
/// \copydoc decay()
operator Generator<T, void>() &&
{
return std::move(*this).decay();
}
private:
_generator::GeneratorBase<T> impl;
explicit Generator(_generator::GeneratorBase<T> b) : impl(std::move(b)) {}
};
}
+2 -2
View File
@@ -324,7 +324,7 @@ Hash hashString(HashType ht, std::string_view s)
Hash hashFile(HashType ht, const Path & path)
{
HashSink sink(ht);
readFileSource(path)->drainInto(sink);
sink << readFileSource(path);
return sink.finish().first;
}
@@ -370,7 +370,7 @@ HashResult hashPath(
HashType ht, const Path & path, PathFilter & filter)
{
HashSink sink(ht);
dumpPath(path, sink, filter);
sink << dumpPath(path, filter);
return sink.finish();
}
+7
View File
@@ -203,5 +203,12 @@ public:
HashResult currentHash();
};
inline HashResult hashSource(HashType ht, Source & source)
{
HashSink h(ht);
source.drainInto(h);
return h.finish();
}
}
+1
View File
@@ -72,6 +72,7 @@ libutil_headers = files(
'file-system.hh',
'finally.hh',
'fmt.hh',
'generator.hh',
'git.hh',
'hash.hh',
'hilite.hh',
+42 -15
View File
@@ -243,27 +243,57 @@ std::string runProgram(Path program, bool searchPath, const Strings & args, bool
// Output = error code + "standard out" output stream
std::pair<int, std::string> runProgram(RunOptions && options)
{
StringSink sink;
options.standardOut = &sink;
options.captureStdout = true;
int status = 0;
std::string stdout;
try {
runProgram2(options);
auto proc = runProgram2(options);
Finally const _wait([&] { proc.wait(); });
stdout = proc.stdout()->drain();
} catch (ExecError & e) {
status = e.status;
}
return {status, std::move(sink.s)};
return {status, std::move(stdout)};
}
void runProgram2(const RunOptions & options)
RunningProgram::RunningProgram(Path program, Pid pid, AutoCloseFD stdout)
: program(std::move(program))
, pid(std::move(pid))
, stdoutSource(stdout ? std::make_unique<FdSource>(stdout.get()) : nullptr)
, stdout_(std::move(stdout))
{
}
RunningProgram::~RunningProgram()
{
if (pid) {
// we will not kill a subprocess because we *can't* kill a subprocess
// reliably without placing it in its own process group, and cleaning
// up a subprocess only when `separatePG` is set is a loaded footgun.
assert(false && "destroying un-wait()ed running process");
std::terminate();
}
}
void RunningProgram::wait()
{
/* Wait for the child to finish. */
int status = pid.wait();
if (status)
throw ExecError(status, "program '%1%' %2%", program, statusToString(status));
}
RunningProgram runProgram2(const RunOptions & options)
{
checkInterrupt();
/* Create a pipe. */
Pipe out;
if (options.standardOut) out.create();
if (options.captureStdout) out.create();
ProcessOptions processOptions;
@@ -281,7 +311,7 @@ void runProgram2(const RunOptions & options)
Pid pid{startProcess([&]() {
if (options.environment)
replaceEnv(*options.environment);
if (options.standardOut && dup2(out.writeSide.get(), STDOUT_FILENO) == -1)
if (options.captureStdout && dup2(out.writeSide.get(), STDOUT_FILENO) == -1)
throw SysError("dupping stdout");
if (options.mergeStderrToStdout)
if (dup2(STDOUT_FILENO, STDERR_FILENO) == -1)
@@ -314,14 +344,11 @@ void runProgram2(const RunOptions & options)
out.writeSide.close();
if (options.standardOut)
drainFD(out.readSide.get(), *options.standardOut);
/* Wait for the child to finish. */
int status = pid.wait();
if (status)
throw ExecError(status, "program '%1%' %2%", options.program, statusToString(status));
return RunningProgram{
options.program,
std::move(pid),
options.captureStdout ? std::move(out.readSide) : AutoCloseFD{}
};
}
std::string statusToString(int status)
+25 -2
View File
@@ -3,6 +3,7 @@
#include "types.hh"
#include "error.hh"
#include "file-descriptor.hh"
#include <sys/types.h>
#include <sys/stat.h>
@@ -82,14 +83,36 @@ struct RunOptions
std::optional<uid_t> gid;
std::optional<Path> chdir;
std::optional<std::map<std::string, std::string>> environment;
Sink * standardOut = nullptr;
bool captureStdout = false;
bool mergeStderrToStdout = false;
bool isInteractive = false;
};
struct RunningProgram
{
friend RunningProgram runProgram2(const RunOptions & options);
private:
Path program;
Pid pid;
std::unique_ptr<Source> stdoutSource;
AutoCloseFD stdout_;
RunningProgram(Path program, Pid pid, AutoCloseFD stdout);
public:
RunningProgram() = default;
~RunningProgram();
void wait();
Source * stdout() const { return stdoutSource.get(); }
};
std::pair<int, std::string> runProgram(RunOptions && options);
void runProgram2(const RunOptions & options);
[[nodiscard("you must call wait() on the returned handle")]]
RunningProgram runProgram2(const RunOptions & options);
class ExecError : public Error
{
+97 -44
View File
@@ -65,74 +65,127 @@ void RefScanSink::operator () (std::string_view data)
}
RewritingSink::RewritingSink(const std::string & from, const std::string & to, Sink & nextSink)
: RewritingSink({{from, to}}, nextSink)
RewritingSource::RewritingSource(const std::string & from, const std::string & to, Source & inner)
: RewritingSource({{from, to}}, inner)
{
}
RewritingSink::RewritingSink(const StringMap & rewrites, Sink & nextSink)
: rewrites(rewrites), nextSink(nextSink)
RewritingSource::RewritingSource(StringMap rewrites, Source & inner)
: RewritingSource(may_change_size, std::move(rewrites), inner)
{
std::string::size_type maxRewriteSize = 0;
for (auto & [from, to] : rewrites) {
for (auto & [from, to] : this->rewrites) {
assert(from.size() == to.size());
maxRewriteSize = std::max(maxRewriteSize, from.size());
}
this->maxRewriteSize = maxRewriteSize;
}
void RewritingSink::operator () (std::string_view data)
{
std::string s(prev);
s.append(data);
s = rewriteStrings(s, rewrites);
prev = s.size() < maxRewriteSize
? s
: maxRewriteSize == 0
? ""
: std::string(s, s.size() - maxRewriteSize + 1, maxRewriteSize - 1);
auto consumed = s.size() - prev.size();
pos += consumed;
if (consumed) nextSink(s.substr(0, consumed));
RewritingSource::RewritingSource(may_change_size_t, StringMap rewrites, Source & inner)
: maxRewriteSize([&, result = size_t(0)]() mutable {
for (auto & [k, v] : rewrites) {
result = std::max(result, k.size());
}
void RewritingSink::flush()
{
if (prev.empty()) return;
pos += prev.size();
nextSink(prev);
prev.clear();
return result;
}())
, initials([&]() -> std::string {
std::string initials;
for (const auto & [k, v] : rewrites) {
assert(!k.empty());
initials.push_back(k[0]);
}
HashModuloSink::HashModuloSink(HashType ht, const std::string & modulus)
: hashSink(ht)
, rewritingSink(modulus, std::string(modulus.size(), 0), hashSink)
std::ranges::sort(initials);
auto [firstDupe, _end] = std::ranges::unique(initials);
return {initials.begin(), firstDupe};
}())
, rewrites(std::move(rewrites))
, inner(&inner)
{
}
void HashModuloSink::operator () (std::string_view data)
size_t RewritingSource::read(char * data, size_t len)
{
rewritingSink(data);
size_t used = 0;
if (rewrites.empty()) {
used += inner->read(data, len);
return used;
}
HashResult HashModuloSink::finish()
if (!unreturned.empty()) {
used = std::min(len, unreturned.size());
memcpy(data, unreturned.data(), used);
unreturned.remove_prefix(used);
data += used;
len -= used;
if (len == 0) {
return used;
}
}
if (!inner) {
if (used > 0) {
return used;
} else {
throw EndOfFile("rewritten source exhausted");
}
}
// always make sure to have at least *two* full rewrites in the buffer,
// otherwise we may end up incorrectly rewriting if the replacement map
// contains keys that are proper infixes of other keys in the map. take
// for example the set { ab -> cc, babb -> bbbb } on the input babb. if
// we feed the input bytewise without additional windowing we will miss
// the full babb match once the second b has been seen and bab has been
// rewritten to ccb.
while (buffered.size() < std::max(2 * maxRewriteSize, len)) {
try {
auto read = inner->read(data, std::min(2 * maxRewriteSize, len));
buffered.append(data, read);
} catch (EndOfFile &) {
inner = nullptr;
break;
}
}
const size_t reserved = inner ? maxRewriteSize : 0;
size_t j = 0;
while ((j = buffered.find_first_of(initials, j)) < buffered.size() - reserved) {
size_t skip = 1;
for (auto & [from, to] : rewrites) {
if (buffered.compare(j, from.size(), from) == 0) {
buffered.replace(j, from.size(), to);
skip = to.size();
break;
}
}
j += skip;
}
rewritten = std::move(buffered);
buffered = rewritten.substr(rewritten.size() - reserved);
unreturned = rewritten;
unreturned.remove_suffix(reserved);
return used + read(data, len);
}
HashResult computeHashModulo(HashType ht, const std::string & modulus, Source & source)
{
rewritingSink.flush();
HashSink hashSink(ht);
LengthSink lengthSink;
RewritingSource rewritingSource(modulus, std::string(modulus.size(), 0), source);
TeeSink tee{hashSink, lengthSink};
rewritingSource.drainInto(tee);
/* Hash the positions of the self-references. This ensures that a
NAR with self-references and a NAR with some of the
self-references already zeroed out do not produce a hash
collision. FIXME: proof. */
for (auto & pos : rewritingSink.matches)
hashSink(fmt("|%d", pos));
// NOTE(horrors) actually, RewritinSink never tracked any matches.
//for (auto & pos : rewritingSource.matches)
// hashSink(fmt("|%d", pos));
auto h = hashSink.finish();
return {h.first, rewritingSink.pos};
return {h.first, lengthSink.length};
}
}
+14 -22
View File
@@ -23,34 +23,26 @@ public:
void operator () (std::string_view data) override;
};
struct RewritingSink : Sink
struct RewritingSource : Source
{
const std::string::size_type maxRewriteSize;
const std::string initials;
const StringMap rewrites;
std::string::size_type maxRewriteSize;
std::string prev;
Sink & nextSink;
uint64_t pos = 0;
std::string rewritten, buffered;
std::string_view unreturned;
Source * inner;
std::vector<uint64_t> matches;
static constexpr struct may_change_size_t {
explicit may_change_size_t() = default;
} may_change_size{};
RewritingSink(const std::string & from, const std::string & to, Sink & nextSink);
RewritingSink(const StringMap & rewrites, Sink & nextSink);
RewritingSource(const std::string & from, const std::string & to, Source & inner);
RewritingSource(StringMap rewrites, Source & inner);
RewritingSource(may_change_size_t, StringMap rewrites, Source & inner);
void operator () (std::string_view data) override;
void flush();
size_t read(char * data, size_t len) override;
};
struct HashModuloSink : AbstractHashSink
{
HashSink hashSink;
RewritingSink rewritingSink;
HashModuloSink(HashType ht, const std::string & modulus);
void operator () (std::string_view data) override;
HashResult finish() override;
};
HashResult computeHashModulo(HashType ht, const std::string & modulus, Source & source);
}
+21 -209
View File
@@ -5,8 +5,6 @@
#include <cerrno>
#include <memory>
#include <boost/coroutine2/coroutine.hpp>
namespace nix {
@@ -149,177 +147,6 @@ size_t StringSource::read(char * data, size_t len)
}
#if BOOST_VERSION >= 106300 && BOOST_VERSION < 106600
#error Coroutines are broken in this version of Boost!
#endif
/* A concrete datatype allow virtual dispatch of stack allocation methods. */
struct VirtualStackAllocator {
StackAllocator *allocator = StackAllocator::defaultAllocator;
boost::context::stack_context allocate() {
return allocator->allocate();
}
void deallocate(boost::context::stack_context sctx) {
allocator->deallocate(sctx);
}
};
/* This class reifies the default boost coroutine stack allocation strategy with
a virtual interface. */
class DefaultStackAllocator : public StackAllocator {
boost::coroutines2::default_stack stack;
boost::context::stack_context allocate() {
return stack.allocate();
}
void deallocate(boost::context::stack_context sctx) {
stack.deallocate(sctx);
}
};
static DefaultStackAllocator defaultAllocatorSingleton;
StackAllocator *StackAllocator::defaultAllocator = &defaultAllocatorSingleton;
std::shared_ptr<void> (*create_coro_gc_hook)() = []() -> std::shared_ptr<void> {
return {};
};
/* This class is used for entry and exit hooks on coroutines */
class CoroutineContext {
/* Disable GC when entering the coroutine without the boehm patch,
* since it doesn't find the main thread stack in this case.
* std::shared_ptr<void> performs type-erasure, so it will call the right
* deleter. */
const std::shared_ptr<void> coro_gc_hook = create_coro_gc_hook();
public:
CoroutineContext() {};
~CoroutineContext() {};
};
std::unique_ptr<FinishSink> sourceToSink(std::function<void(Source &)> fun)
{
struct SourceToSink : FinishSink
{
typedef boost::coroutines2::coroutine<bool> coro_t;
std::function<void(Source &)> fun;
std::optional<coro_t::push_type> coro;
SourceToSink(std::function<void(Source &)> fun) : fun(fun)
{
}
std::string_view cur;
void operator () (std::string_view in) override
{
if (in.empty()) return;
cur = in;
if (!coro) {
CoroutineContext ctx;
coro = coro_t::push_type(VirtualStackAllocator{}, [&](coro_t::pull_type & yield) {
LambdaSource source([&](char *out, size_t out_len) {
if (cur.empty()) {
yield();
if (yield.get()) {
throw EndOfFile("coroutine exhausted");
}
}
size_t n = std::min(cur.size(), out_len);
memcpy(out, cur.data(), n);
cur.remove_prefix(n);
return n;
});
fun(source);
});
}
if (!*coro) { abort(); }
if (!cur.empty()) {
CoroutineContext ctx;
(*coro)(false);
}
}
void finish() override
{
if (!coro) return;
if (!*coro) abort();
{
CoroutineContext ctx;
(*coro)(true);
}
if (*coro) abort();
}
};
return std::make_unique<SourceToSink>(fun);
}
std::unique_ptr<Source> sinkToSource(std::function<void(Sink &)> fun)
{
struct SinkToSource : Source
{
typedef boost::coroutines2::coroutine<std::string> coro_t;
std::function<void(Sink &)> fun;
std::optional<coro_t::pull_type> coro;
SinkToSource(std::function<void(Sink &)> fun)
: fun(fun)
{
}
std::string cur;
size_t pos = 0;
size_t read(char * data, size_t len) override
{
if (!coro) {
CoroutineContext ctx;
coro = coro_t::pull_type(VirtualStackAllocator{}, [&](coro_t::push_type & yield) {
LambdaSink sink([&](std::string_view data) {
if (!data.empty()) yield(std::string(data));
});
fun(sink);
});
}
if (!*coro) {
throw EndOfFile("coroutine has finished");
}
if (pos == cur.size()) {
if (!cur.empty()) {
CoroutineContext ctx;
(*coro)();
}
cur = coro->get();
pos = 0;
}
auto n = std::min(cur.size() - pos, len);
memcpy(data, cur.data() + pos, n);
pos += n;
return n;
}
};
return std::make_unique<SinkToSource>(fun);
}
void writePadding(size_t len, Sink & sink)
{
if (len % 8) {
@@ -330,55 +157,40 @@ void writePadding(size_t len, Sink & sink)
}
void writeString(std::string_view data, Sink & sink)
WireFormatGenerator SerializingTransform::operator()(std::string_view s)
{
sink << data.size();
sink(data);
writePadding(data.size(), sink);
co_yield s.size();
co_yield Bytes(s.begin(), s.size());
co_yield SerializingTransform::padding(s.size());
}
Sink & operator << (Sink & sink, std::string_view s)
WireFormatGenerator SerializingTransform::operator()(const Strings & ss)
{
writeString(s, sink);
return sink;
co_yield ss.size();
for (const auto & s : ss)
co_yield std::string_view(s);
}
template<class T> void writeStrings(const T & ss, Sink & sink)
WireFormatGenerator SerializingTransform::operator()(const StringSet & ss)
{
sink << ss.size();
for (auto & i : ss)
sink << i;
co_yield ss.size();
for (const auto & s : ss)
co_yield std::string_view(s);
}
Sink & operator << (Sink & sink, const Strings & s)
{
writeStrings(s, sink);
return sink;
}
Sink & operator << (Sink & sink, const StringSet & s)
{
writeStrings(s, sink);
return sink;
}
Sink & operator << (Sink & sink, const Error & ex)
WireFormatGenerator SerializingTransform::operator()(const Error & ex)
{
auto & info = ex.info();
sink
<< "Error"
<< info.level
<< "Error" // removed
<< info.msg.str()
<< 0 // FIXME: info.errPos
<< info.traces.size();
co_yield "Error";
co_yield info.level;
co_yield "Error"; // removed
co_yield info.msg.str();
co_yield 0; // FIXME: info.errPos
co_yield info.traces.size();
for (auto & trace : info.traces) {
sink << 0; // FIXME: trace.pos
sink << trace.hint.str();
co_yield 0; // FIXME: trace.pos
co_yield trace.hint.str();
}
return sink;
}
+92 -17
View File
@@ -1,8 +1,10 @@
#pragma once
///@file
#include <concepts>
#include <memory>
#include "generator.hh"
#include "strings.hh"
#include "types.hh"
#include "file-descriptor.hh"
@@ -332,21 +334,48 @@ struct ChainSource : Source
size_t read(char * data, size_t len) override;
};
std::unique_ptr<FinishSink> sourceToSink(std::function<void(Source &)> fun);
struct GeneratorSource : Source
{
GeneratorSource(Generator<Bytes> && g) : g(std::move(g)) {}
/**
* Convert a function that feeds data into a Sink into a Source. The
* Source executes the function as a coroutine.
*/
std::unique_ptr<Source> sinkToSource(std::function<void(Sink &)> fun);
virtual size_t read(char * data, size_t len)
{
while (!buf.size()) {
if (auto next = g.next()) {
buf = *next;
} else {
throw EndOfFile("coroutine has finished");
}
}
len = std::min(len, buf.size());
memcpy(data, buf.data(), len);
buf = buf.subspan(len);
return len;
}
void writePadding(size_t len, Sink & sink);
void writeString(std::string_view s, Sink & sink);
private:
Generator<Bytes> g;
Bytes buf{};
};
inline Sink & operator << (Sink & sink, uint64_t n)
inline Sink & operator<<(Sink & sink, Generator<Bytes> && g)
{
while (auto bit = g.next()) {
sink(std::string_view(bit->data(), bit->size()));
}
return sink;
}
struct SerializingTransform;
using WireFormatGenerator = Generator<Bytes, SerializingTransform>;
struct SerializingTransform
{
std::array<char, 8> buf;
Bytes operator()(uint64_t n)
{
unsigned char buf[8];
buf[0] = n & 0xff;
buf[1] = (n >> 8) & 0xff;
buf[2] = (n >> 16) & 0xff;
@@ -355,19 +384,65 @@ inline Sink & operator << (Sink & sink, uint64_t n)
buf[5] = (n >> 40) & 0xff;
buf[6] = (n >> 48) & 0xff;
buf[7] = (unsigned char) (n >> 56) & 0xff;
sink({(char *) buf, sizeof(buf)});
return sink;
return {buf.begin(), 8};
}
Sink & operator << (Sink & in, const Error & ex);
Sink & operator << (Sink & sink, std::string_view s);
Sink & operator << (Sink & sink, const Strings & s);
Sink & operator << (Sink & sink, const StringSet & s);
static Bytes padding(size_t unpadded)
{
return Bytes("\0\0\0\0\0\0\0", unpadded % 8 ? 8 - unpadded % 8 : 0);
}
// opt in to generator chaining. without this co_yielding
// another generator of any type will cause a type error.
auto operator()(Generator<Bytes> && g)
{
return std::move(g);
}
// only choose this for *exactly* char spans, do not allow implicit
// conversions. this would cause ambiguities with strings literals,
// and resolving those with more string-like overloads needs a lot.
template<typename Span>
requires std::same_as<Span, std::span<char>> || std::same_as<Span, std::span<const char>>
Bytes operator()(Span s)
{
return s;
}
WireFormatGenerator operator()(std::string_view s);
WireFormatGenerator operator()(const Strings & s);
WireFormatGenerator operator()(const StringSet & s);
WireFormatGenerator operator()(const Error & s);
};
void writePadding(size_t len, Sink & sink);
inline Sink & operator<<(Sink & sink, uint64_t u)
{
return sink << [&]() -> WireFormatGenerator { co_yield u; }();
}
inline Sink & operator<<(Sink & sink, std::string_view s)
{
return sink << [&]() -> WireFormatGenerator { co_yield s; }();
}
inline Sink & operator<<(Sink & sink, const Strings & s)
{
return sink << [&]() -> WireFormatGenerator { co_yield s; }();
}
inline Sink & operator<<(Sink & sink, const StringSet & s)
{
return sink << [&]() -> WireFormatGenerator { co_yield s; }();
}
inline Sink & operator<<(Sink & sink, const Error & ex)
{
return sink << [&]() -> WireFormatGenerator { co_yield ex; }();
}
MakeError(SerialisationError, Error);
template<typename T>
T readNum(Source & source)
{
+1 -1
View File
@@ -101,7 +101,7 @@ struct SourcePath
void dumpPath(
Sink & sink,
PathFilter & filter = defaultPathFilter) const
{ return nix::dumpPath(path.abs(), sink, filter); }
{ sink << nix::dumpPath(path.abs(), filter); }
/**
* Return the location of this path in the "real" filesystem, if
+4 -20
View File
@@ -1,4 +1,5 @@
#include "strings.hh"
#include "references.hh"
#include <boost/lexical_cast.hpp>
#include <stdint.h>
@@ -65,30 +66,13 @@ std::string replaceStrings(
Rewriter::Rewriter(std::map<std::string, std::string> rewrites)
: rewrites(std::move(rewrites))
{
for (const auto & [k, v] : this->rewrites) {
assert(!k.empty());
initials.push_back(k[0]);
}
std::ranges::sort(initials);
auto [firstDupe, end] = std::ranges::unique(initials);
initials.erase(firstDupe, end);
}
std::string Rewriter::operator()(std::string s)
{
size_t j = 0;
while ((j = s.find_first_of(initials, j)) != std::string::npos) {
size_t skip = 1;
for (auto & [from, to] : rewrites) {
if (s.compare(j, from.size(), from) == 0) {
s.replace(j, from.size(), to);
skip = to.size();
break;
}
}
j += skip;
}
return s;
StringSource src{s};
RewritingSource inner{RewritingSource::may_change_size, rewrites, src};
return inner.drain();
}
template<class N>
+5
View File
@@ -11,6 +11,7 @@
#include <map>
#include <variant>
#include <vector>
#include <span>
namespace nix {
@@ -19,6 +20,10 @@ typedef std::set<std::string> StringSet;
typedef std::map<std::string, std::string> StringMap;
typedef std::map<std::string, std::string> StringPairs;
// TODO this should be a std::byte span, but too much of the
// current codebase predates std::byte and uses char instead
using Bytes = std::span<const char>;
/**
* Paths are just strings.
*/
+7 -7
View File
@@ -671,7 +671,7 @@ static void opDump(Strings opFlags, Strings opArgs)
FdSink sink(STDOUT_FILENO);
std::string path = *opArgs.begin();
dumpPath(path, sink);
sink << dumpPath(path);
sink.flush();
}
@@ -763,7 +763,7 @@ static void opVerifyPath(Strings opFlags, Strings opArgs)
printMsg(lvlTalkative, "checking path '%s'...", store->printStorePath(path));
auto info = store->queryPathInfo(path);
HashSink sink(info->narHash.type);
store->narFromPath(path, sink);
sink << store->narFromPath(path);
auto current = sink.finish();
if (current.first != info->narHash) {
printError("path '%s' was modified! expected hash '%s', got '%s'",
@@ -880,7 +880,7 @@ static void opServe(Strings opFlags, Strings opArgs)
store->substitutePaths(paths);
}
ServeProto::write(*store, wconn, store->queryValidPaths(paths));
wconn.to << ServeProto::write(*store, wconn, store->queryValidPaths(paths));
break;
}
@@ -891,7 +891,7 @@ static void opServe(Strings opFlags, Strings opArgs)
try {
auto info = store->queryPathInfo(i);
out << store->printStorePath(info->path);
ServeProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
wconn.to << ServeProto::write(*store, wconn, static_cast<const UnkeyedValidPathInfo &>(*info));
} catch (InvalidPath &) {
}
}
@@ -900,7 +900,7 @@ static void opServe(Strings opFlags, Strings opArgs)
}
case ServeProto::Command::DumpStorePath:
store->narFromPath(store->parseStorePath(readString(in)), out);
out << store->narFromPath(store->parseStorePath(readString(in)));
break;
case ServeProto::Command::ImportPaths: {
@@ -950,7 +950,7 @@ static void opServe(Strings opFlags, Strings opArgs)
MonitorFdHup monitor(in.fd);
auto status = store->buildDerivation(drvPath, drv);
ServeProto::write(*store, wconn, status);
wconn.to << ServeProto::write(*store, wconn, status);
break;
}
@@ -959,7 +959,7 @@ static void opServe(Strings opFlags, Strings opArgs)
StorePathSet closure;
store->computeFSClosure(ServeProto::Serialise<StorePathSet>::read(*store, rconn),
closure, false, includeOutputs);
ServeProto::write(*store, wconn, closure);
wconn.to << ServeProto::write(*store, wconn, closure);
break;
}
+2 -2
View File
@@ -30,14 +30,14 @@ struct CmdAddToStore : MixDryRun, StoreCommand
if (!namePart) namePart = baseNameOf(path);
StringSink sink;
dumpPath(path, sink);
sink << dumpPath(path);
auto narHash = hashString(htSHA256, sink.s);
Hash hash = narHash;
if (ingestionMethod == FileIngestionMethod::Flat) {
HashSink hsink(htSHA256);
readFileSource(path)->drainInto(hsink);
hsink << readFileSource(path);
hash = hsink.finish().first;
}
+2 -2
View File
@@ -23,7 +23,7 @@ struct CmdDumpPath : StorePathCommand
{
stopProgressBar();
FdSink sink(STDOUT_FILENO);
store->narFromPath(storePath, sink);
sink << store->narFromPath(storePath);
sink.flush();
}
};
@@ -59,7 +59,7 @@ struct CmdDumpPath2 : Command
{
stopProgressBar();
FdSink sink(STDOUT_FILENO);
dumpPath(path, sink);
sink << dumpPath(path);
sink.flush();
}
};
+8 -12
View File
@@ -76,23 +76,19 @@ struct CmdHashBase : Command
void run() override
{
for (auto path : paths) {
std::unique_ptr<AbstractHashSink> hashSink;
if (modulus)
hashSink = std::make_unique<HashModuloSink>(ht, *modulus);
else
hashSink = std::make_unique<HashSink>(ht);
auto source = [&] () -> GeneratorSource {
switch (mode) {
case FileIngestionMethod::Flat:
readFileSource(path)->drainInto(*hashSink);
break;
return GeneratorSource(readFileSource(path));
case FileIngestionMethod::Recursive:
dumpPath(path, *hashSink);
break;
return GeneratorSource(dumpPath(path));
}
assert(false);
}();
Hash h = hashSink->finish().first;
Hash h = modulus
? computeHashModulo(ht, *modulus, source).first
: hashSource(ht, source).first;
if (truncate && h.hashSize > 20) h = compressHash(h, 20);
logger->cout(h.to_string(base, base == SRI));
}
+1 -1
View File
@@ -100,7 +100,7 @@ struct CmdVerify : StorePathsCommand
auto hashSink = HashSink(info->narHash.type);
store->narFromPath(info->path, hashSink);
hashSink << store->narFromPath(info->path);
auto hash = hashSink.finish();
+1 -1
View File
@@ -50,7 +50,7 @@ public:
auto file = goldenMaster(testStem);
StringSink to;
CommonProto::write(
to << CommonProto::write(
*store,
CommonProto::WriteConn { .to = to },
value);
+1 -1
View File
@@ -56,7 +56,7 @@ public:
auto file = ProtoTest<Proto, protocolDir>::goldenMaster(testStem);
StringSink to;
Proto::write(
to << Proto::write(
*LibStoreTest::store,
typename Proto::WriteConn {to, version},
value);
+214
View File
@@ -0,0 +1,214 @@
#include "generator.hh"
#include <concepts>
#include <cstdint>
#include <gtest/gtest.h>
namespace nix {
TEST(Generator, yields)
{
auto g = []() -> Generator<int> {
co_yield 1;
co_yield 2;
}();
ASSERT_EQ(g.next(), 1);
ASSERT_EQ(g.next(), 2);
ASSERT_FALSE(g.next().has_value());
}
TEST(Generator, returns)
{
{
auto g = []() -> Generator<int> { co_return; }();
ASSERT_FALSE(g.next().has_value());
}
{
auto g = []() -> Generator<int> {
co_yield 1;
co_yield []() -> Generator<int> { co_return; }();
co_yield 2;
co_yield []() -> Generator<int> { co_yield 10; }();
co_yield 3;
(void) "dummy statement to force some more execution";
}();
ASSERT_EQ(g.next(), 1);
ASSERT_EQ(g.next(), 2);
ASSERT_EQ(g.next(), 10);
ASSERT_EQ(g.next(), 3);
ASSERT_FALSE(g.next().has_value());
}
}
TEST(Generator, nests)
{
auto g = []() -> Generator<int> {
co_yield 1;
co_yield []() -> Generator<int> {
co_yield 9;
co_yield []() -> Generator<int> {
co_yield 99;
co_yield 100;
}();
}();
auto g2 = []() -> Generator<int> {
co_yield []() -> Generator<int> {
co_yield 2000;
co_yield 2001;
}();
co_yield 1001;
}();
co_yield g2.next().value();
co_yield std::move(g2);
co_yield 2;
}();
ASSERT_EQ(g.next(), 1);
ASSERT_EQ(g.next(), 9);
ASSERT_EQ(g.next(), 99);
ASSERT_EQ(g.next(), 100);
ASSERT_EQ(g.next(), 2000);
ASSERT_EQ(g.next(), 2001);
ASSERT_EQ(g.next(), 1001);
ASSERT_EQ(g.next(), 2);
ASSERT_FALSE(g.next().has_value());
}
TEST(Generator, nestsExceptions)
{
auto g = []() -> Generator<int> {
co_yield 1;
co_yield []() -> Generator<int> {
co_yield 9;
throw 1;
co_yield 10;
}();
co_yield 2;
}();
ASSERT_EQ(g.next(), 1);
ASSERT_EQ(g.next(), 9);
ASSERT_THROW(g.next(), int);
}
TEST(Generator, exception)
{
{
auto g = []() -> Generator<int> {
co_yield 1;
throw 1;
}();
ASSERT_EQ(g.next(), 1);
ASSERT_THROW(g.next(), int);
ASSERT_FALSE(g.next().has_value());
}
{
auto g = []() -> Generator<int> {
throw 1;
co_return;
}();
ASSERT_THROW(g.next(), int);
ASSERT_FALSE(g.next().has_value());
}
}
namespace {
struct Transform
{
int state = 0;
std::pair<uint32_t, int> operator()(std::integral auto x)
{
return {x, state++};
}
Generator<std::pair<uint32_t, int>, Transform> operator()(const char *)
{
co_yield 9;
co_yield 19;
}
Generator<std::pair<uint32_t, int>, Transform> operator()(Generator<int> && inner)
{
return [](auto g) mutable -> Generator<std::pair<uint32_t, int>, Transform> {
while (auto i = g.next()) {
co_yield *i;
}
}(std::move(inner));
}
};
}
TEST(Generator, transform)
{
auto g = []() -> Generator<std::pair<uint32_t, int>, Transform> {
co_yield int32_t(-1);
co_yield "";
co_yield []() -> Generator<int> { co_yield 7; }();
co_yield 20;
}();
ASSERT_EQ(g.next(), (std::pair<unsigned, int>{4294967295, 0}));
ASSERT_EQ(g.next(), (std::pair<unsigned, int>{9, 0}));
ASSERT_EQ(g.next(), (std::pair<unsigned, int>{19, 1}));
ASSERT_EQ(g.next(), (std::pair<unsigned, int>{7, 0}));
ASSERT_EQ(g.next(), (std::pair<unsigned, int>{20, 1}));
ASSERT_FALSE(g.next().has_value());
}
namespace {
struct ThrowTransform
{
int operator()(int x)
{
return x;
}
int operator()(bool)
{
throw 2;
}
Generator<int, void> operator()(Generator<int> && inner)
{
throw false;
}
};
}
TEST(Generator, transformThrows)
{
{
auto g = []() -> Generator<int, ThrowTransform> {
co_yield 1;
co_yield false;
co_yield 2;
}();
ASSERT_EQ(g.next(), 1);
ASSERT_THROW(g.next(), int);
ASSERT_FALSE(g.next().has_value());
}
{
auto g = []() -> Generator<int, ThrowTransform> {
co_yield 1;
co_yield []() -> Generator<int> {
co_yield 2;
}();
co_yield 3;
}();
ASSERT_EQ(g.next(), 1);
ASSERT_THROW(g.next(), bool);
ASSERT_FALSE(g.next().has_value());
}
}
}
+4 -6
View File
@@ -25,11 +25,8 @@ class RewriteTest : public ::testing::TestWithParam<RewriteParams> {
TEST_P(RewriteTest, IdentityRewriteIsIdentity) {
RewriteParams param = GetParam();
StringSink rewritten;
auto rewriter = RewritingSink(param.rewrites, rewritten);
rewriter(param.originalString);
rewriter.flush();
ASSERT_EQ(rewritten.s, param.finalString);
StringSource src{param.originalString};
ASSERT_EQ(RewritingSource(param.rewrites, src).drain(), param.finalString);
}
INSTANTIATE_TEST_CASE_P(
@@ -38,7 +35,8 @@ INSTANTIATE_TEST_CASE_P(
::testing::Values(
RewriteParams{ "foooo", "baroo", {{"foo", "bar"}, {"bar", "baz"}}},
RewriteParams{ "foooo", "bazoo", {{"fou", "bar"}, {"foo", "baz"}}},
RewriteParams{ "foooo", "foooo", {}}
RewriteParams{ "foooo", "foooo", {}},
RewriteParams{ "babb", "bbbb", {{"ab", "aa"}, {"babb", "bbbb"}}}
)
);
+84 -33
View File
@@ -2,30 +2,47 @@
#include "error.hh"
#include "fmt.hh"
#include "pos-table.hh"
#include "generator.hh"
#include "ref.hh"
#include "types.hh"
#include <concepts>
#include <cstdint>
#include <initializer_list>
#include <limits.h>
#include <gtest/gtest.h>
#include <numeric>
#include <stdexcept>
#include <string_view>
#include <type_traits>
namespace nix {
TEST(Sink, uint64_t)
// don't deduce the type of `val` for added insurance.
template<typename T>
static std::string toWire(const std::type_identity_t<T> & val)
{
StringSink s;
s << 42;
ASSERT_EQ(s.s, std::string({42, 0, 0, 0, 0, 0, 0, 0}));
std::string result;
auto g = [] (const auto & val) -> WireFormatGenerator { co_yield val; }(val);
while (auto bit = g.next()) {
result.append(bit->data(), bit->size());
}
return result;
}
TEST(Sink, string_view)
TEST(WireFormatGenerator, uint64_t)
{
StringSink s;
s << "";
auto s = toWire<uint64_t>(42);
ASSERT_EQ(s, std::string({42, 0, 0, 0, 0, 0, 0, 0}));
}
TEST(WireFormatGenerator, string_view)
{
auto s = toWire<std::string_view>("");
// clang-format off
ASSERT_EQ(
s.s,
s,
std::string({
// length
0, 0, 0, 0, 0, 0, 0, 0,
@@ -34,11 +51,10 @@ TEST(Sink, string_view)
);
// clang-format on
s = {};
s << "test";
s = toWire<std::string_view>("test");
// clang-format off
ASSERT_EQ(
s.s,
s,
std::string({
// length
4, 0, 0, 0, 0, 0, 0, 0,
@@ -50,11 +66,10 @@ TEST(Sink, string_view)
);
// clang-format on
s = {};
s << "longer string";
s = toWire<std::string_view>("longer string");
// clang-format off
ASSERT_EQ(
s.s,
s,
std::string({
// length
13, 0, 0, 0, 0, 0, 0, 0,
@@ -67,13 +82,12 @@ TEST(Sink, string_view)
// clang-format on
}
TEST(Sink, StringSet)
TEST(WireFormatGenerator, StringSet)
{
StringSink s;
s << StringSet{};
auto s = toWire<StringSet>({});
// clang-format off
ASSERT_EQ(
s.s,
s,
std::string({
// length
0, 0, 0, 0, 0, 0, 0, 0,
@@ -82,11 +96,10 @@ TEST(Sink, StringSet)
);
// clang-format on
s = {};
s << StringSet{"a", ""};
s = toWire<StringSet>({"a", ""});
// clang-format off
ASSERT_EQ(
s.s,
s,
std::string({
// length
2, 0, 0, 0, 0, 0, 0, 0,
@@ -99,13 +112,12 @@ TEST(Sink, StringSet)
// clang-format on
}
TEST(Sink, Strings)
TEST(WireFormatGenerator, Strings)
{
StringSink s;
s << Strings{};
auto s = toWire<Strings>({});
// clang-format off
ASSERT_EQ(
s.s,
s,
std::string({
// length
0, 0, 0, 0, 0, 0, 0, 0,
@@ -114,11 +126,10 @@ TEST(Sink, Strings)
);
// clang-format on
s = {};
s << Strings{"a", ""};
s = toWire<Strings>({"a", ""});
// clang-format off
ASSERT_EQ(
s.s,
s,
std::string({
// length
2, 0, 0, 0, 0, 0, 0, 0,
@@ -131,23 +142,22 @@ TEST(Sink, Strings)
// clang-format on
}
TEST(Sink, Error)
TEST(WireFormatGenerator, Error)
{
PosTable pt;
auto o = pt.addOrigin(Pos::String{make_ref<std::string>("test")}, 4);
StringSink s;
s << Error{ErrorInfo{
auto s = toWire<Error>(Error{ErrorInfo{
.level = lvlInfo,
.msg = HintFmt("foo"),
.pos = pt[pt.add(o, 1)],
.traces = {{.pos = pt[pt.add(o, 2)], .hint = HintFmt("b %1%", "foo")}},
}};
}});
// NOTE position of the error and all traces are ignored
// by the wire format
// clang-format off
ASSERT_EQ(
s.s,
s,
std::string({
5, 0, 0, 0, 0, 0, 0, 0, 'E', 'r', 'r', 'o', 'r', 0, 0, 0,
3, 0, 0, 0, 0, 0, 0, 0,
@@ -163,4 +173,45 @@ TEST(Sink, Error)
// clang-format on
}
TEST(WireFormatGenerator, exampleMessage)
{
auto gen = []() -> WireFormatGenerator {
std::set<std::string> foo{"a", "longer string", ""};
co_yield 42;
co_yield foo;
co_yield std::string_view("test");
co_yield true;
}();
std::vector<char> full;
while (auto s = gen.next()) {
full.insert(full.end(), s->begin(), s->end());
}
ASSERT_EQ(
full,
(std::vector<char>{
// clang-format off
// 42
42, 0, 0, 0, 0, 0, 0, 0,
// foo
3, 0, 0, 0, 0, 0, 0, 0,
/// ""
0, 0, 0, 0, 0, 0, 0, 0,
/// a
1, 0, 0, 0, 0, 0, 0, 0,
'a', 0, 0, 0, 0, 0, 0, 0,
/// longer string
13, 0, 0, 0, 0, 0, 0, 0,
'l', 'o', 'n', 'g', 'e', 'r', ' ', 's', 't', 'r', 'i', 'n', 'g', 0, 0, 0,
// foo done
// test
4, 0, 0, 0, 0, 0, 0, 0,
't', 'e', 's', 't', 0, 0, 0, 0,
// true
1, 0, 0, 0, 0, 0, 0, 0,
//clang-format on
}));
}
}
+1
View File
@@ -40,6 +40,7 @@ libutil_tests_sources = files(
'libutil/compression.cc',
'libutil/config.cc',
'libutil/escape-string.cc',
'libutil/generator.cc',
'libutil/git.cc',
'libutil/hash.cc',
'libutil/hilite.cc',