serialise.hh

Go to the documentation of this file.

#pragma once
 
#include <memory>
 
#include "lix/libutil/charptr-cast.hh"
#include "lix/libutil/generator.hh"
#include "lix/libutil/types.hh"
#include "lix/libutil/file-descriptor.hh"
 
namespace nix {
 

struct Sink
{
    virtual ~Sink() { }
    virtual void operator () (std::string_view data) = 0;
    virtual bool good() { return true; }
};

struct NullSink : Sink
{
    void operator () (std::string_view data) override
    { }
};
 
 
struct FinishSink : virtual Sink
{
    virtual void finish() = 0;
};
 

struct BufferedSink : virtual Sink
{
    size_t bufSize, bufPos;
    std::unique_ptr<char[]> buffer;
 
    BufferedSink(size_t bufSize = 32 * 1024)
        : bufSize(bufSize), bufPos(0), buffer(nullptr) { }
 
    void operator () (std::string_view data) override;
 
    void flush();
 
protected:
 
    virtual void writeUnbuffered(std::string_view data) = 0;
};
 

struct Source
{
    virtual ~Source() { }

    void operator () (char * data, size_t len);

    virtual size_t read(char * data, size_t len) = 0;
 
    virtual bool good() { return true; }
 
    void drainInto(Sink & sink);
 
    std::string drain();
};
 

struct BufferedSource : Source
{
    size_t bufSize, bufPosIn, bufPosOut;
    std::unique_ptr<char[]> buffer;
 
    BufferedSource(size_t bufSize = 32 * 1024)
        : bufSize(bufSize), bufPosIn(0), bufPosOut(0), buffer(nullptr) { }
 
    size_t read(char * data, size_t len) override;
 
    bool hasData();
 
protected:
    virtual size_t readUnbuffered(char * data, size_t len) = 0;
};
 

struct FdSink : BufferedSink
{
    int fd;
    size_t written = 0;
 
    FdSink() : fd(-1) { }
    FdSink(int fd) : fd(fd) { }
    FdSink(FdSink&&) = default;
 
    FdSink & operator=(FdSink && s)
    {
        flush();
        fd = s.fd;
        s.fd = -1;
        written = s.written;
        return *this;
    }
 
    ~FdSink();
 
    void writeUnbuffered(std::string_view data) override;
 
    bool good() override;
 
private:
    bool _good = true;
};
 

struct FdSource : BufferedSource
{
    int fd;
    size_t read = 0;
    std::optional<std::string> specialEndOfFileError;
 
    std::string endOfFileError() const;
 
    FdSource() : fd(-1) { }
    FdSource(int fd) : fd(fd) { }
    FdSource(FdSource &&) = default;
 
    FdSource & operator=(FdSource && s)
    {
        fd = s.fd;
        s.fd = -1;
        read = s.read;
        return *this;
    }
 
    bool good() override;
protected:
    size_t readUnbuffered(char * data, size_t len) override;
private:
    bool _good = true;
};
 

struct StringSink : Sink
{
    std::string s;
    StringSink() { }
    explicit StringSink(const size_t reservedSize)
    {
      s.reserve(reservedSize);
    };
    StringSink(std::string && s) : s(std::move(s)) { };
    void operator () (std::string_view data) override;
};
 

struct StringSource : Source
{
    std::string_view s;
    size_t pos;
    StringSource(std::string_view s) : s(s), pos(0) { }
    size_t read(char * data, size_t len) override;
};
 

struct TeeSink : Sink
{
    Sink & sink1, & sink2;
    TeeSink(Sink & sink1, Sink & sink2) : sink1(sink1), sink2(sink2) { }
    virtual void operator () (std::string_view data) override
    {
        sink1(data);
        sink2(data);
    }
};
 

struct TeeSource : Source
{
    Source & orig;
    Sink & sink;
    TeeSource(Source & orig, Sink & sink)
        : orig(orig), sink(sink) { }
    size_t read(char * data, size_t len) override
    {
        size_t n = orig.read(data, len);
        sink({data, n});
        return n;
    }
};

struct SizedSource : Source
{
    Source & orig;
    size_t remain;
    SizedSource(Source & orig, size_t size)
        : orig(orig), remain(size) { }
    size_t read(char * data, size_t len) override
    {
        if (this->remain <= 0) {
            throw EndOfFile("sized: unexpected end-of-file");
        }
        len = std::min(len, this->remain);
        size_t n = this->orig.read(data, len);
        this->remain -= n;
        return n;
    }

    size_t drainAll()
    {
        std::vector<char> buf(8192);
        size_t sum = 0;
        while (this->remain > 0) {
            size_t n = read(buf.data(), buf.size());
            sum += n;
        }
        return sum;
    }
};

struct LengthSink : Sink
{
    uint64_t length = 0;
 
    void operator () (std::string_view data) override
    {
        length += data.size();
    }
};

struct LambdaSink : Sink
{
    typedef std::function<void(std::string_view data)> lambda_t;
 
    lambda_t lambda;
 
    LambdaSink(const lambda_t & lambda) : lambda(lambda) { }
 
    void operator () (std::string_view data) override
    {
        lambda(data);
    }
};
 

struct LambdaSource : Source
{
    typedef std::function<size_t(char *, size_t)> lambda_t;
 
    lambda_t lambda;
 
    LambdaSource(const lambda_t & lambda) : lambda(lambda) { }
 
    size_t read(char * data, size_t len) override
    {
        return lambda(data, len);
    }
};
 
struct GeneratorSource : Source
{
    GeneratorSource(Generator<Bytes> && g) : g(std::move(g)) {}
 
    virtual size_t read(char * data, size_t len) override
    {
        // we explicitly do not poll the generator multiple times to fill the
        // buffer, only to produce some output at all. this is allowed by the
        // semantics of read(), only operator() must fill the buffer entirely
        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;
    }
 
private:
    Generator<Bytes> g;
    Bytes buf{};
};
 
inline Sink & operator<<(Sink & sink, Generator<Bytes> && g)
{
    while (auto buffer = g.next()) {
        sink(std::string_view(buffer->data(), buffer->size()));
    }
    return sink;
}
 
struct SerializingTransform;
using WireFormatGenerator = Generator<Bytes, SerializingTransform>;
 
struct SerializingTransform
{
    std::array<unsigned char, 8> buf;
 
    Bytes operator()(uint64_t n)
    {
        buf[0] = n & 0xff;
        buf[1] = (n >> 8) & 0xff;
        buf[2] = (n >> 16) & 0xff;
        buf[3] = (n >> 24) & 0xff;
        buf[4] = (n >> 32) & 0xff;
        buf[5] = (n >> 40) & 0xff;
        buf[6] = (n >> 48) & 0xff;
        buf[7] = (unsigned char) (n >> 56) & 0xff;
        return {charptr_cast<const char *>(buf.begin()), 8};
    }
 
    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);
 
// NOLINTBEGIN(cppcoreguidelines-avoid-capturing-lambda-coroutines):
// These coroutines do their entire job before the semicolon and are not
// retained, so they live long enough.
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; }();
}
// NOLINTEND(cppcoreguidelines-avoid-capturing-lambda-coroutines)
 
MakeError(SerialisationError, Error);
 
template<typename T>
T readNum(Source & source);
 
inline unsigned int readInt(Source & source)
{
    return readNum<unsigned int>(source);
}
 
 
inline uint64_t readLongLong(Source & source)
{
    return readNum<uint64_t>(source);
}
 
 
void readPadding(size_t len, Source & source);
size_t readString(char * buf, size_t max, Source & source);
std::string readString(Source & source, size_t max = std::numeric_limits<size_t>::max());
template<class T> T readStrings(Source & source);
 
Source & operator >> (Source & in, std::string & s);
 
template<typename T>
Source & operator >> (Source & in, T & n)
{
    n = readNum<T>(in);
    return in;
}
 
template<typename T>
Source & operator >> (Source & in, bool & b)
{
    b = readNum<uint64_t>(in);
    return in;
}
 
Error readError(Source & source);
 

struct StreamToSourceAdapter : Source
{
    std::shared_ptr<std::basic_istream<char>> istream;
 
    StreamToSourceAdapter(std::shared_ptr<std::basic_istream<char>> istream)
        : istream(istream)
    { }
 
    size_t read(char * data, size_t len) override
    {
        if (!istream->read(data, len)) {
            if (istream->eof()) {
                if (istream->gcount() == 0)
                    throw EndOfFile("end of file");
            } else
                throw Error("I/O error in StreamToSourceAdapter");
        }
        return istream->gcount();
    }
};
 

struct FramedSource : Source
{
    Source & from;
    bool eof = false;
    std::vector<char> pending;
    size_t pos = 0;
 
    FramedSource(Source & from) : from(from)
    { }
 
    ~FramedSource()
    {
        try {
            if (!eof) {
                while (true) {
                    auto n = readInt(from);
                    if (!n) break;
                    std::vector<char> data(n);
                    from(data.data(), n);
                }
            }
        } catch (...) {
            ignoreExceptionInDestructor();
        }
    }
 
    size_t read(char * data, size_t len) override
    {
        if (eof) throw EndOfFile("reached end of FramedSource");
 
        if (pos >= pending.size()) {
            size_t len = readInt(from);
            if (!len) {
                eof = true;
                return 0;
            }
            pending = std::vector<char>(len);
            pos = 0;
            from(pending.data(), len);
        }
 
        auto n = std::min(len, pending.size() - pos);
        memcpy(data, pending.data() + pos, n);
        pos += n;
        return n;
    }
};

struct FramedSink : nix::BufferedSink
{
    BufferedSink & to;
    std::exception_ptr & ex;
 
    FramedSink(BufferedSink & to, std::exception_ptr & ex) : to(to), ex(ex)
    { }
 
    ~FramedSink()
    {
        try {
            to << 0;
            to.flush();
        } catch (...) {
            ignoreExceptionInDestructor();
        }
    }
 
    void writeUnbuffered(std::string_view data) override
    {
        /* Don't send more data if the remote has
            encountered an error. */
        if (ex) {
            auto ex2 = ex;
            ex = nullptr;
            std::rethrow_exception(ex2);
        }
        to << data.size();
        to(data);
    };
};
 
/* Disabling GC when entering a coroutine (without the boehm patch).
   mutable to avoid boehm gc dependency in libutil.
 */
extern std::shared_ptr<void> (*create_coro_gc_hook)();
 
 
}