diff options
Diffstat (limited to 'src/libutil/serialise.cc')
| -rw-r--r-- | src/libutil/serialise.cc | 240 |
1 files changed, 194 insertions, 46 deletions
diff --git a/src/libutil/serialise.cc b/src/libutil/serialise.cc index c8b71188f..16f3476c2 100644 --- a/src/libutil/serialise.cc +++ b/src/libutil/serialise.cc @@ -11,23 +11,23 @@ namespace nix { -void BufferedSink::operator () (const unsigned char * data, size_t len) +void BufferedSink::operator () (std::string_view data) { - if (!buffer) buffer = decltype(buffer)(new unsigned char[bufSize]); + if (!buffer) buffer = decltype(buffer)(new char[bufSize]); - while (len) { + while (!data.empty()) { /* Optimisation: bypass the buffer if the data exceeds the buffer size. */ - if (bufPos + len >= bufSize) { + if (bufPos + data.size() >= bufSize) { flush(); - write(data, len); + write(data); break; } /* Otherwise, copy the bytes to the buffer. Flush the buffer when it's full. */ - size_t n = bufPos + len > bufSize ? bufSize - bufPos : len; - memcpy(buffer.get() + bufPos, data, n); - data += n; bufPos += n; len -= n; + size_t n = bufPos + data.size() > bufSize ? bufSize - bufPos : data.size(); + memcpy(buffer.get() + bufPos, data.data(), n); + data.remove_prefix(n); bufPos += n; if (bufPos == bufSize) flush(); } } @@ -38,7 +38,7 @@ void BufferedSink::flush() if (bufPos == 0) return; size_t n = bufPos; bufPos = 0; // don't trigger the assert() in ~BufferedSink() - write(buffer.get(), n); + write({buffer.get(), n}); } @@ -52,16 +52,13 @@ size_t threshold = 256 * 1024 * 1024; static void warnLargeDump() { - logWarning({ - .name = "Large path", - .description = "dumping very large path (> 256 MiB); this may run out of memory" - }); + warn("dumping very large path (> 256 MiB); this may run out of memory"); } -void FdSink::write(const unsigned char * data, size_t len) +void FdSink::write(std::string_view data) { - written += len; + written += data.size(); static bool warned = false; if (warn && !warned) { if (written > threshold) { @@ -70,7 +67,7 @@ void FdSink::write(const unsigned char * data, size_t len) } } try { - writeFull(fd, data, len); + writeFull(fd, data); } catch (SysError & e) { _good = false; throw; @@ -84,7 +81,7 @@ bool FdSink::good() } -void Source::operator () (unsigned char * data, size_t len) +void Source::operator () (char * data, size_t len) { while (len) { size_t n = read(data, len); @@ -93,26 +90,33 @@ void Source::operator () (unsigned char * data, size_t len) } -std::string Source::drain() +void Source::drainInto(Sink & sink) { std::string s; - std::vector<unsigned char> buf(8192); + std::vector<char> buf(8192); while (true) { size_t n; try { n = read(buf.data(), buf.size()); - s.append((char *) buf.data(), n); + sink({buf.data(), n}); } catch (EndOfFile &) { break; } } - return s; } -size_t BufferedSource::read(unsigned char * data, size_t len) +std::string Source::drain() +{ + StringSink s; + drainInto(s); + return *s.s; +} + + +size_t BufferedSource::read(char * data, size_t len) { - if (!buffer) buffer = decltype(buffer)(new unsigned char[bufSize]); + if (!buffer) buffer = decltype(buffer)(new char[bufSize]); if (!bufPosIn) bufPosIn = readUnbuffered(buffer.get(), bufSize); @@ -131,12 +135,12 @@ bool BufferedSource::hasData() } -size_t FdSource::readUnbuffered(unsigned char * data, size_t len) +size_t FdSource::readUnbuffered(char * data, size_t len) { ssize_t n; do { checkInterrupt(); - n = ::read(fd, (char *) data, len); + n = ::read(fd, data, len); } while (n == -1 && errno == EINTR); if (n == -1) { _good = false; throw SysError("reading from file"); } if (n == 0) { _good = false; throw EndOfFile("unexpected end-of-file"); } @@ -151,10 +155,10 @@ bool FdSource::good() } -size_t StringSource::read(unsigned char * data, size_t len) +size_t StringSource::read(char * data, size_t len) { if (pos == s.size()) throw EndOfFile("end of string reached"); - size_t n = s.copy((char *) data, len, pos); + size_t n = s.copy(data, len, pos); pos += n; return n; } @@ -164,6 +168,95 @@ size_t StringSource::read(unsigned char * data, size_t len) #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::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) + 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()) { + return (size_t)0; + } + } + + 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()) (*coro)(false); + } + + void finish() override + { + if (!coro) return; + if (!*coro) abort(); + (*coro)(true); + if (*coro) abort(); + } + }; + + return std::make_unique<SourceToSink>(fun); +} + + std::unique_ptr<Source> sinkToSource( std::function<void(Sink &)> fun, std::function<void()> eof) @@ -175,7 +268,6 @@ std::unique_ptr<Source> sinkToSource( std::function<void(Sink &)> fun; std::function<void()> eof; std::optional<coro_t::pull_type> coro; - bool started = false; SinkToSource(std::function<void(Sink &)> fun, std::function<void()> eof) : fun(fun), eof(eof) @@ -185,13 +277,13 @@ std::unique_ptr<Source> sinkToSource( std::string cur; size_t pos = 0; - size_t read(unsigned char * data, size_t len) override + size_t read(char * data, size_t len) override { if (!coro) - coro = coro_t::pull_type([&](coro_t::push_type & yield) { - LambdaSink sink([&](const unsigned char * data, size_t len) { - if (len) yield(std::string((const char *) data, len)); - }); + 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); }); @@ -204,7 +296,7 @@ std::unique_ptr<Source> sinkToSource( } auto n = std::min(cur.size() - pos, len); - memcpy(data, (unsigned char *) cur.data() + pos, n); + memcpy(data, cur.data() + pos, n); pos += n; return n; @@ -218,24 +310,24 @@ std::unique_ptr<Source> sinkToSource( void writePadding(size_t len, Sink & sink) { if (len % 8) { - unsigned char zero[8]; + char zero[8]; memset(zero, 0, sizeof(zero)); - sink(zero, 8 - (len % 8)); + sink({zero, 8 - (len % 8)}); } } -void writeString(const unsigned char * buf, size_t len, Sink & sink) +void writeString(std::string_view data, Sink & sink) { - sink << len; - sink(buf, len); - writePadding(len, sink); + sink << data.size(); + sink(data); + writePadding(data.size(), sink); } Sink & operator << (Sink & sink, const string & s) { - writeString((const unsigned char *) s.data(), s.size(), sink); + writeString(s, sink); return sink; } @@ -259,11 +351,28 @@ Sink & operator << (Sink & sink, const StringSet & s) return sink; } +Sink & operator << (Sink & sink, const Error & ex) +{ + auto info = ex.info(); + sink + << "Error" + << info.level + << info.name + << info.msg.str() + << 0 // FIXME: info.errPos + << info.traces.size(); + for (auto & trace : info.traces) { + sink << 0; // FIXME: trace.pos + sink << trace.hint.str(); + } + return sink; +} + void readPadding(size_t len, Source & source) { if (len % 8) { - unsigned char zero[8]; + char zero[8]; size_t n = 8 - (len % 8); source(zero, n); for (unsigned int i = 0; i < n; i++) @@ -272,7 +381,7 @@ void readPadding(size_t len, Source & source) } -size_t readString(unsigned char * buf, size_t max, Source & source) +size_t readString(char * buf, size_t max, Source & source) { auto len = readNum<size_t>(source); if (len > max) throw SerialisationError("string is too long"); @@ -287,7 +396,7 @@ string readString(Source & source, size_t max) auto len = readNum<size_t>(source); if (len > max) throw SerialisationError("string is too long"); std::string res(len, 0); - source((unsigned char*) res.data(), len); + source(res.data(), len); readPadding(len, source); return res; } @@ -312,15 +421,54 @@ template Paths readStrings(Source & source); template PathSet readStrings(Source & source); -void StringSink::operator () (const unsigned char * data, size_t len) +Error readError(Source & source) +{ + auto type = readString(source); + assert(type == "Error"); + auto level = (Verbosity) readInt(source); + auto name = readString(source); + auto msg = readString(source); + ErrorInfo info { + .level = level, + .name = name, + .msg = hintformat(std::move(format("%s") % msg)), + }; + auto havePos = readNum<size_t>(source); + assert(havePos == 0); + auto nrTraces = readNum<size_t>(source); + for (size_t i = 0; i < nrTraces; ++i) { + havePos = readNum<size_t>(source); + assert(havePos == 0); + info.traces.push_back(Trace { + .hint = hintformat(std::move(format("%s") % readString(source))) + }); + } + return Error(std::move(info)); +} + + +void StringSink::operator () (std::string_view data) { static bool warned = false; if (!warned && s->size() > threshold) { warnLargeDump(); warned = true; } - s->append((const char *) data, len); + s->append(data); } +size_t ChainSource::read(char * data, size_t len) +{ + if (useSecond) { + return source2.read(data, len); + } else { + try { + return source1.read(data, len); + } catch (EndOfFile &) { + useSecond = true; + return this->read(data, len); + } + } +} } |