lix-project/lix-releng-staging
0
0
Fork 0
Code Packages Activity

Tunnel progress messages from the daemon to the client

Browse source 
This makes the progress bar work for non-root users.
This commit is contained in:
Eelco Dolstra 2017-08-28 18:49:42 +02:00
parent e681b1f064
commit fe34b91289
No known key found for this signature in database
GPG key ID: 8170B4726D7198DE
4 changed files with 124 additions and 30 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
src/libstore/remote-store.cc
View file

@ -629,17 +629,37 @@ RemoteStore::Connection::~Connection()
}
static Logger::Fields readFields(Source & from)
{
Logger::Fields fields;
size_t size = readInt(from);
for (size_t n = 0; n < size; n++) {
auto type = (decltype(Logger::Field::type)) readInt(from);
if (type == Logger::Field::tInt)
fields.push_back(readNum<uint64_t>(from));
else if (type == Logger::Field::tString)
fields.push_back(readString(from));
else
throw Error("got unsupported field type %x from Nix daemon", (int) type);
}
return fields;
}
void RemoteStore::Connection::processStderr(Sink * sink, Source * source)
{
to.flush();
unsigned int msg;
while ((msg = readInt(from)) == STDERR_NEXT
|| msg == STDERR_READ || msg == STDERR_WRITE) {
while (true) {
auto msg = readNum<uint64_t>(from);
if (msg == STDERR_WRITE) {
string s = readString(from);
if (!sink) throw Error("no sink");
(*sink)(s);
}
else if (msg == STDERR_READ) {
if (!source) throw Error("no source");
size_t len = readNum<size_t>(from);
@ -647,16 +667,43 @@ void RemoteStore::Connection::processStderr(Sink * sink, Source * source)
writeString(buf.get(), source->read(buf.get(), len), to);
to.flush();
}
else
else if (msg == STDERR_ERROR) {
string error = readString(from);
unsigned int status = readInt(from);
throw Error(status, error);
}
else if (msg == STDERR_NEXT)
printError(chomp(readString(from)));
else if (msg == STDERR_START_ACTIVITY) {
auto act = readNum<ActivityId>(from);
auto type = (ActivityType) readInt(from);
auto s = readString(from);
auto fields = readFields(from);
auto parent = readNum<ActivityId>(from);
logger->startActivity(act, type, s, fields, parent);
}
else if (msg == STDERR_STOP_ACTIVITY) {
auto act = readNum<ActivityId>(from);
logger->stopActivity(act);
}
else if (msg == STDERR_RESULT) {
auto act = readNum<ActivityId>(from);
auto type = (ResultType) readInt(from);
auto fields = readFields(from);
logger->result(act, type, fields);
}
else if (msg == STDERR_LAST)
break;
else
throw Error("got unknown message type %x from Nix daemon", msg);
}
if (msg == STDERR_ERROR) {
string error = readString(from);
unsigned int status = readInt(from);
throw Error(status, error);
}
else if (msg != STDERR_LAST)
throw Error("protocol error processing standard error");
}

3
src/libstore/worker-protocol.hh
View file

@ -57,6 +57,9 @@ typedef enum {
#define STDERR_WRITE 0x64617416 // data for sink
#define STDERR_LAST 0x616c7473
#define STDERR_ERROR 0x63787470
#define STDERR_START_ACTIVITY 0x53545254
#define STDERR_STOP_ACTIVITY 0x53544f50
#define STDERR_RESULT 0x52534c54
Path readStorePath(Store & store, Source & from);

2
src/libutil/logging.hh
View file

@ -47,7 +47,7 @@ public:
struct Field
{
// FIXME: use std::variant.
enum { tInt, tString } type;
enum { tInt = 0, tString = 1 } type;
uint64_t i = 0;
std::string s;
Field(const std::string & s) : type(tString), s(s) { }

80
src/nix-daemon/nix-daemon.cc
View file

@ -56,6 +56,21 @@ static FdSource from(STDIN_FILENO);
static FdSink to(STDOUT_FILENO);
Sink & operator << (Sink & sink, const Logger::Fields & fields)
{
sink << fields.size();
for (auto & f : fields) {
sink << f.type;
if (f.type == Logger::Field::tInt)
sink << f.i;
else if (f.type == Logger::Field::tString)
sink << f.s;
else abort();
}
return sink;
}
/* Logger that forwards log messages to the client, *if* we're in a
state where the protocol allows it (i.e., when canSendStderr is
true). */
@ -69,15 +84,14 @@ struct TunnelLogger : public Logger
Sync<State> state_;
void log(Verbosity lvl, const FormatOrString & fs) override
void enqueueMsg(const std::string & s)
{
if (lvl > verbosity) return;
auto state(state_.lock());
if (state->canSendStderr) {
assert(state->pendingMsgs.empty());
try {
to << STDERR_NEXT << (fs.s + "\n");
to(s);
to.flush();
} catch (...) {
/* Write failed; that means that the other side is
@ -86,7 +100,16 @@ struct TunnelLogger : public Logger
throw;
}
} else
state->pendingMsgs.push_back(fs.s);
state->pendingMsgs.push_back(s);
}
void log(Verbosity lvl, const FormatOrString & fs) override
{
if (lvl > verbosity) return;
StringSink buf;
buf << STDERR_NEXT << (fs.s + "\n");
enqueueMsg(*buf.s);
}
/* startWork() means that we're starting an operation for which we
@ -99,7 +122,7 @@ struct TunnelLogger : public Logger
state->canSendStderr = true;
for (auto & msg : state->pendingMsgs)
to << STDERR_NEXT << (msg + "\n");
to(msg);
state->pendingMsgs.clear();
@ -121,6 +144,28 @@ struct TunnelLogger : public Logger
if (status != 0) to << status;
}
}
void startActivity(ActivityId act, ActivityType type,
const std::string & s, const Fields & fields, ActivityId parent) override
{
StringSink buf;
buf << STDERR_START_ACTIVITY << act << type << s << fields << parent;
enqueueMsg(*buf.s);
}
void stopActivity(ActivityId act) override
{
StringSink buf;
buf << STDERR_STOP_ACTIVITY << act;
enqueueMsg(*buf.s);
}
void result(ActivityId act, ResultType type, const Fields & fields) override
{
StringSink buf;
buf << STDERR_RESULT << act << type << fields;
enqueueMsg(*buf.s);
}
};
@ -665,16 +710,15 @@ static void processConnection(bool trusted)
{
MonitorFdHup monitor(from.fd);
TunnelLogger tunnelLogger;
auto tunnelLogger = new TunnelLogger();
auto prevLogger = nix::logger;
logger = &tunnelLogger;
logger = tunnelLogger;
unsigned int opCount = 0;
Finally finally([&]() {
logger = prevLogger;
_isInterrupted = false;
debug("%d operations", opCount);
prevLogger->log(lvlDebug, fmt("%d operations", opCount));
});
/* Exchange the greeting. */
@ -693,7 +737,7 @@ static void processConnection(bool trusted)
readInt(from); // obsolete reserveSpace
/* Send startup error messages to the client. */
tunnelLogger.startWork();
tunnelLogger->startWork();
try {
@ -713,7 +757,7 @@ static void processConnection(bool trusted)
params["path-info-cache-size"] = "0";
auto store = make_ref<LocalStore>(params);
tunnelLogger.stopWork();
tunnelLogger->stopWork();
to.flush();
/* Process client requests. */
@ -730,28 +774,28 @@ static void processConnection(bool trusted)
opCount++;
try {
performOp(&tunnelLogger, store, trusted, clientVersion, from, to, op);
performOp(tunnelLogger, store, trusted, clientVersion, from, to, op);
} catch (Error & e) {
/* If we're not in a state where we can send replies, then
something went wrong processing the input of the
client. This can happen especially if I/O errors occur
during addTextToStore() / importPath(). If that
happens, just send the error message and exit. */
bool errorAllowed = tunnelLogger.state_.lock()->canSendStderr;
tunnelLogger.stopWork(false, e.msg(), e.status);
bool errorAllowed = tunnelLogger->state_.lock()->canSendStderr;
tunnelLogger->stopWork(false, e.msg(), e.status);
if (!errorAllowed) throw;
} catch (std::bad_alloc & e) {
tunnelLogger.stopWork(false, "Nix daemon out of memory", 1);
tunnelLogger->stopWork(false, "Nix daemon out of memory", 1);
throw;
}
to.flush();
assert(!tunnelLogger.state_.lock()->canSendStderr);
assert(!tunnelLogger->state_.lock()->canSendStderr);
};
} catch (std::exception & e) {
tunnelLogger.stopWork(false, e.what(), 1);
tunnelLogger->stopWork(false, e.what(), 1);
to.flush();
return;
}