cafkafk/lix
cafkafk/lix
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

sinkToSource(): Start the coroutine lazily

Browse source 
In particular this causes copyStorePath() from HttpBinaryCacheStore to
only start a download if needed. E.g. if the destination LocalStore
goes to sleep waiting for the path lock and another process creates
the path, then LocalStore::addToStore() will never read from the
source so we don't have to do the download.
This commit is contained in:
Eelco Dolstra 2018-09-26 21:19:34 +02:00
parent 44e86304b6
commit ed78582847
No known key found for this signature in database
GPG key ID: 8170B4726D7198DE
1 changed files with 15 additions and 11 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

26
src/libutil/serialise.cc
View file

@ -169,17 +169,13 @@ std::unique_ptr<Source> sinkToSource(
{ {
typedef boost::coroutines2::coroutine<std::string> coro_t; typedef boost::coroutines2::coroutine<std::string> coro_t;
std::function<void(Sink &)> fun;
std::function<void()> eof; std::function<void()> eof;
coro_t::pull_type coro; std::experimental::optional<coro_t::pull_type> coro;
bool started = false;
SinkToSource(std::function<void(Sink &)> fun, std::function<void()> eof) SinkToSource(std::function<void(Sink &)> fun, std::function<void()> eof)
: eof(eof) : fun(fun), eof(eof)
, coro([&](coro_t::push_type & yield) {
LambdaSink sink([&](const unsigned char * data, size_t len) {
if (len) yield(std::string((const char *) data, len));
});
fun(sink);
})
{ {
} }
@ -188,11 +184,19 @@ std::unique_ptr<Source> sinkToSource(
size_t read(unsigned char * data, size_t len) override size_t read(unsigned char * data, size_t len) override
{ {
if (!coro) { eof(); abort(); } 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));
});
fun(sink);
});
if (!*coro) { eof(); abort(); }
if (pos == cur.size()) { if (pos == cur.size()) {
if (!cur.empty()) coro(); if (!cur.empty()) (*coro)();
cur = coro.get(); cur = coro->get();
pos = 0; pos = 0;
} }