Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Add file streams for teeing Job stdout/err to disc #1456

Draft
wants to merge 5 commits into
base: master
Choose a base branch
from
Draft

Add file streams for teeing Job stdout/err to disc #1456

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
dfeb6e5
Add file streams for teeing Job stdout/err to disc
ag-eitilt Oct 31, 2023
71d5c89
Postpone teefile-list splitting; better names
ag-eitilt Nov 3, 2023
70436c7
Prevent code from depending on the logs
ag-eitilt Nov 6, 2023
b7ddc4d
Track teefiles at the level of the JobTable
ag-eitilt Nov 6, 2023
99bbf86
FAILING: Replace null_split with modern code
ag-eitilt Nov 10, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
30 changes: 21 additions & 9 deletions share/wake/lib/system/job.wake
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -112,8 +112,12 @@ export tuple Plan =
export Stdin: String
# How should standard output be displayed during a build
export Stdout: LogLevel
# The workspace-relative paths to which the stdout will be appended
export TeeStdout: List String
# How should standard error be displayed during a build
export Stderr: LogLevel
# The workspace-relative paths to which the stderr will be appended
export TeeStderr: List String
# Echo the command to this stream
export Echo: LogLevel
# See Persistence table above
Expand Down Expand Up @@ -227,7 +231,7 @@ export def editPlanShare (f: Boolean => Boolean): Plan => Plan =

# Get a unique hash-code for the job
export def getPlanHash (plan: Plan): Integer =
def Plan _ cmd _ env dir stdin _ _ _ _ _ _ _ _ isAtty = plan
def Plan _ cmd _ env dir stdin _ _ _ _ _ _ _ _ _ _ isAtty = plan
def isAttyStr = if isAtty then "true" else "false"

def sig =
Expand Down Expand Up @@ -275,7 +279,7 @@ def bToInt b =

# Set reasonable defaults for all Plan arguments
export def makeExecPlan (cmd: List String) (visible: List Path): Plan =
Plan "" cmd visible environment "." "" logInfo logWarning logEcho Share Nil defaultUsage id id False
Plan "" cmd visible environment "." "" logInfo Nil logWarning Nil logEcho Share Nil defaultUsage id id False

export def makeShellPlan (script: String) (visible: List Path): Plan =
makeExecPlan (which "dash", "-c", script, Nil) visible
Expand Down Expand Up @@ -571,8 +575,8 @@ export def virtualRunner: Runner =
def implode l =
cat (foldr (_, "\0", _) Nil l)

def runAlways cmd env dir stdin res uusage finputs foutputs vis keep run echo stdout stderr label isatty: Job =
def create label dir stdin env cmd signature visible keep echo stdout stderr isatty =
def runAlways cmd env dir stdin res uusage finputs foutputs vis keep run echo stdout teeout stderr teeerr label isatty: Job =
def create label dir stdin env cmd signature visible keep echo stdout teeoutFlat stderr teeerrFlat isatty =
prim "job_create"

def finish job inputs outputs all_outputs status runtime cputime membytes ibytes obytes =
Expand All @@ -582,6 +586,8 @@ def runAlways cmd env dir stdin res uusage finputs foutputs vis keep run echo st
def cache dir stdin env cmd signature visible isatty = prim "job_cache"
def signature cmd res fni fno keep = prim "hash"
def hash = signature cmd res finputs foutputs keep
def teeoutFlat = catWith "\0" teeout
def teeerrFlat = catWith "\0" teeerr

def build Unit =
def visStrings = map getPathName vis
Expand All @@ -598,7 +604,9 @@ def runAlways cmd env dir stdin res uusage finputs foutputs vis keep run echo st
(bToInt keep)
echo
stdout
teeoutFlat
stderr
teeerrFlat
(bToInt isatty)

def prefix = str (getJobId job)
Expand Down Expand Up @@ -655,10 +663,10 @@ def runAlways cmd env dir stdin res uusage finputs foutputs vis keep run echo st
Pair Nil last -> confirm False last (build Unit)

# Only run if the first four arguments differ
target runOnce cmd env dir stdin vis isatty run \ res usage finputs foutputs keep echo stdout stderr label =
runAlways cmd env dir stdin res usage finputs foutputs vis keep run echo stdout stderr label isatty
target runOnce cmd env dir stdin vis isatty run \ res usage finputs foutputs keep echo stdout teeout stderr teeerr label =
runAlways cmd env dir stdin res usage finputs foutputs vis keep run echo stdout teeout stderr teeerr label isatty

export def runJobImp label cmd env dir stdin res usage finputs foutputs vis pers run (LogLevel echo) (LogLevel stdout) (LogLevel stderr) isatty =
export def runJobImp label cmd env dir stdin res usage finputs foutputs vis pers run (LogLevel echo) (LogLevel stdout) teeout (LogLevel stderr) teeerr isatty =
if isOnce pers then
runOnce
cmd
Expand All @@ -675,7 +683,9 @@ export def runJobImp label cmd env dir stdin res usage finputs foutputs vis pers
(isKeep pers)
echo
stdout
teeout
stderr
teeerr
label
else
runAlways
Expand All @@ -692,12 +702,14 @@ export def runJobImp label cmd env dir stdin res usage finputs foutputs vis pers
run
echo
stdout
teeout
stderr
teeerr
label
isatty

export def runJobWith (Runner _ _ run) (Plan label cmd vis env dir stdin stdout stderr echo pers res usage finputs foutputs isatty) =
runJobImp label cmd env dir stdin res usage finputs foutputs vis pers run echo stdout stderr isatty
export def runJobWith (Runner _ _ run) (Plan label cmd vis env dir stdin stdout teeout stderr teeerr echo pers res usage finputs foutputs isatty) =
runJobImp label cmd env dir stdin res usage finputs foutputs vis pers run echo stdout teeout stderr teeerr isatty

# Set the value of a tag on a Job
# This is useful for post-build reflection into the database
Expand Down
6 changes: 3 additions & 3 deletions share/wake/lib/system/path.wake
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -179,7 +179,7 @@ def computeHashes (prefix: String) (files: List String): List String =
def add f h = prim "add_hash"

def hashPlan cmd vis =
Plan "<hash>" cmd vis Nil "." "" logNever logError logDebug ReRun Nil hashUsage id id False
Plan "<hash>" cmd vis Nil "." "" logNever Nil logError Nil logDebug ReRun Nil hashUsage id id False

def stdin_file_path = "to_hash.{prefix}.stdin"

Expand Down Expand Up @@ -240,7 +240,7 @@ target hashcode (f: String): String =
reuse
else
def hashPlan cmd =
Plan "" cmd Nil Nil "." "" logNever logError logDebug ReRun Nil hashUsage id id False
Plan "" cmd Nil Nil "." "" logNever Nil logError Nil logDebug ReRun Nil hashUsage id id False

def job =
hashPlan ("<hash>", f, Nil)
Expand All @@ -259,7 +259,7 @@ target hashcode (f: String): String =
# Allow an untracked file to be removed via `wake --clean`
export target markFileCleanable (filepath: String): Result Unit Error =
def hashPlan cmd =
Plan "" cmd Nil Nil "." "" logNever logError logDebug ReRun Nil hashUsage id id False
Plan "" cmd Nil Nil "." "" logNever Nil logError Nil logDebug ReRun Nil hashUsage id id False

def job =
hashPlan ("<hash>", filepath, Nil)
Expand Down
10 changes: 6 additions & 4 deletions share/wake/lib/system/plan_scorer.wake
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,7 @@ publish runner =

# Run a job, via a Runner chosen based on 'score' functions.
export def runJob (p: Plan): Job = match p
Plan label cmd vis env dir stdin stdout stderr echo pers res usage finputs foutputs isatty ->
Plan label cmd vis env dir stdin stdout teeout stderr teeerr echo pers res usage finputs foutputs isatty ->
def implode l = cat (foldr (_, "\0", _) Nil l)
def bToInt b = if b then 1 else 0

Expand All @@ -50,16 +50,18 @@ export def runJob (p: Plan): Job = match p

match (opts | foldl best (Pair 0.0 None) | getPairSecond)
Some r ->
runJobImp label cmd env dir stdin res usage finputs foutputs vis pers r echo stdout stderr isatty
runJobImp label cmd env dir stdin res usage finputs foutputs vis pers r echo stdout teeout stderr teeerr isatty
None ->
def create label dir stdin env cmd signature visible keep echo stdout stderr isatty =
def create label dir stdin env cmd signature visible keep echo stdout teeoutFlat stderr teeerrFlat isatty =
prim "job_create"

def badfinish job e = prim "job_fail_finish"
def badlaunch job e = prim "job_fail_launch"
def teeoutFlat = catWith "\0" teeout
def teeerrFlat = catWith "\0" teeerr

def job =
create label dir stdin env.implode cmd.implode 0 "" 0 "echo" "info" "error" (bToInt isatty)
create label dir stdin env.implode cmd.implode 0 "" 0 "echo" "info" teeoutFlat "error" teeerrFlat (bToInt isatty)

def error =
def pretty = match _
Expand Down
84 changes: 68 additions & 16 deletions src/runtime/job.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,6 +39,7 @@

#include <algorithm>
#include <cstring>
#include <fstream>
#include <iostream>
#include <limits>
#include <list>
Expand Down Expand Up @@ -105,7 +106,9 @@ struct Job final : public GCObject<Job, Value> {
bool keep;
std::string echo;
std::string stream_out;
std::vector<std::string> tee_out;
std::string stream_err;
std::vector<std::string> tee_err;
JakeSiFive marked this conversation as resolved.
Show resolved Hide resolved
HeapPointer<Value> bad_launch;
HeapPointer<Value> bad_finish;
double pathtime;
Expand All @@ -124,8 +127,8 @@ struct Job final : public GCObject<Job, Value> {
HeapPointer<Continuation> q_report; // waken once job finished (inputs+outputs+report available)

Job(Database *db_, String *label_, String *dir_, String *stdin_file_, String *environ,
String *cmdline_, bool keep, const char *echo, const char *stream_out,
const char *stream_err);
String *cmdline_, bool keep, const char *echo, const char *stream_out, std::vector<std::string> tee_out,
const char *stream_err, std::vector<std::string> tee_err);

template <typename T, T (HeapPointerBase::*memberfn)(T x)>
T recurse(T arg);
Expand Down Expand Up @@ -275,17 +278,21 @@ struct JobEntry {
std::string echo_line;
std::list<Status>::iterator status;
std::unique_ptr<std::streambuf> stdout_linebuf;
std::vector<std::unique_ptr<std::ofstream>> stdout_teefiles;
std::unique_ptr<std::streambuf> stderr_linebuf;
std::vector<std::unique_ptr<std::ofstream>> stderr_teefiles;

JobEntry(JobTable::detail *imp_, RootPointer<Job> &&job_, std::unique_ptr<std::streambuf> stdout,
std::unique_ptr<std::streambuf> stderr)
std::vector<std::unique_ptr<std::ofstream>> stdout_teefiles_, std::unique_ptr<std::streambuf> stderr, std::vector<std::unique_ptr<std::ofstream>> stderr_teefiles_)
: imp(imp_),
job(std::move(job_)),
pid(0),
pipe_stdout(-1),
pipe_stderr(-1),
stdout_linebuf(std::move(stdout)),
stderr_linebuf(std::move(stderr)) {}
stdout_teefiles(std::move(stdout_teefiles_)),
stderr_linebuf(std::move(stderr)),
stderr_teefiles(std::move(stderr_teefiles_)) {}
~JobEntry();

double runtime(struct timespec now);
Expand Down Expand Up @@ -783,7 +790,6 @@ static void launch(JobTable *jobtable) {
// Make the raw output streams and the TermInfoBufs
// that jobs will use. We make one TermInfoBuf per
// file descriptor that we're outputting to.
// TODO: We could add file tee-ing here as well
if (!jobtable->imp->fd_bufs.count(fd_out)) {
std::unique_ptr<std::streambuf> fd_buf;
if (fd_out != -1) {
Expand Down Expand Up @@ -824,8 +830,18 @@ static void launch(JobTable *jobtable) {
} else {
err = std::make_unique<NullBuf>();
}

auto stdout_teefiles = std::vector<std::unique_ptr<std::ofstream>>();
for (auto tee_path : task.job->tee_out) {
stdout_teefiles.push_back(std::make_unique<std::ofstream>(tee_path, std::ios::out | std::ios::app));
JakeSiFive marked this conversation as resolved.
Show resolved Hide resolved
}
auto stderr_teefiles = std::vector<std::unique_ptr<std::ofstream>>();
for (auto tee_path : task.job->tee_err) {
stderr_teefiles.push_back(std::make_unique<std::ofstream>(tee_path, std::ios::out | std::ios::app));
}

std::shared_ptr<JobEntry> entry = std::make_shared<JobEntry>(
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Better to store an std::vector<std::ostream*> that acts as a list of weak pointers to the unique pointers stored by the job table that outlives these references. That way you don't have to do the job table string lookup every time, only at job launch time.

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more. 

    auto stdout_tee_names = std::vector<std::ostream*>();
    for (const auto& file : stdout_teefiles) {
      auto iter = jobtable...->teefiles.find(file);
      if (iter == ... .end()) {
          auto out = std::make_unique<std::ofstream>(stdout_teefiles[i], std::ios::out);
          iter = jobtable->imp->teefiles[stdout_teefiles[i]].emplace(file, std::move(out)).first
      }
      stdout_tee_names.push_back(iter->second.get());
    }

Copy link
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's a nice little tweak that makes me glad I don't write C++ on the regular...

jobtable->imp.get(), std::move(task.job), std::move(out), std::move(err));
jobtable->imp.get(), std::move(task.job), std::move(out), std::move(stdout_teefiles), std::move(err), std::move(stderr_teefiles));

int stdout_stream[2];
int stderr_stream[2];
Expand Down Expand Up @@ -974,6 +990,9 @@ bool JobTable::wait(Runtime &runtime) {
imp->poll.remove(fd);
close(fd);
entry->pipe_stdout = -1;
for (auto &tee_fd : entry->stdout_teefiles) {
tee_fd->close();
}
entry->status->wait_stdout = false;
entry->job->state |= STATE_STDOUT;
runtime.heap.guarantee(WJob::reserve());
Expand All @@ -983,6 +1002,9 @@ bool JobTable::wait(Runtime &runtime) {
entry->job->db->save_output(entry->job->job, 1, buffer, got, entry->runtime(now));
if (!imp->batch) {
entry->stdout_linebuf->sputn(buffer, got);
for (auto &tee_fd : entry->stdout_teefiles) {
tee_fd->write(buffer, got);
}
}
}
}
Expand All @@ -993,6 +1015,9 @@ bool JobTable::wait(Runtime &runtime) {
imp->poll.remove(fd);
close(fd);
entry->pipe_stderr = -1;
for (auto &tee_fd : entry->stderr_teefiles) {
tee_fd->close();
}
entry->status->wait_stderr = false;
entry->job->state |= STATE_STDERR;
runtime.heap.guarantee(WJob::reserve());
Expand All @@ -1002,6 +1027,9 @@ bool JobTable::wait(Runtime &runtime) {
entry->job->db->save_output(entry->job->job, 2, buffer, got, entry->runtime(now));
if (!imp->batch) {
entry->stderr_linebuf->sputn(buffer, got);
for (auto &tee_fd : entry->stderr_teefiles) {
tee_fd->write(buffer, got);
}
}
}
}
Expand Down Expand Up @@ -1082,8 +1110,8 @@ bool JobTable::wait(Runtime &runtime) {
}

Job::Job(Database *db_, String *label_, String *dir_, String *stdin_file_, String *environ,
String *cmdline_, bool keep_, const char *echo_, const char *stream_out_,
const char *stream_err_)
String *cmdline_, bool keep_, const char *echo_, const char *stream_out_, std::vector<std::string> tee_out_,
const char *stream_err_, std::vector<std::string> tee_err_)
: db(db_),
label(label_),
cmdline(cmdline_),
Expand All @@ -1096,7 +1124,9 @@ Job::Job(Database *db_, String *label_, String *dir_, String *stdin_file_, Strin
keep(keep_),
echo(echo_),
stream_out(stream_out_),
stream_err(stream_err_) {
tee_out(tee_out_),
stream_err(stream_err_),
tee_err(tee_err_) {
start.tv_sec = stop.tv_sec = 0;
start.tv_nsec = stop.tv_nsec = 0;

Expand Down Expand Up @@ -1286,18 +1316,35 @@ static PRIMFN(prim_job_virtual) {
}

static PRIMTYPE(type_job_create) {
return args.size() == 12 && args[0]->unify(Data::typeString) &&
return args.size() == 14 && args[0]->unify(Data::typeString) &&
args[1]->unify(Data::typeString) && args[2]->unify(Data::typeString) &&
args[3]->unify(Data::typeString) && args[4]->unify(Data::typeString) &&
args[5]->unify(Data::typeInteger) && args[6]->unify(Data::typeString) &&
args[7]->unify(Data::typeInteger) && args[8]->unify(Data::typeString) &&
args[9]->unify(Data::typeString) && args[10]->unify(Data::typeString) &&
args[11]->unify(Data::typeInteger) && out->unify(Data::typeJob);
args[11]->unify(Data::typeString) && args[12]->unify(Data::typeString) &&
args[13]->unify(Data::typeInteger) && out->unify(Data::typeJob);
}

std::vector<std::string> null_sep_list(const std::string &str) {
JakeSiFive marked this conversation as resolved.
Show resolved Hide resolved
size_t pos = 0;
std::vector<std::string> out = std::vector<std::string>();
do {
size_t next = str.find('\0', pos);
if (next == std::string::npos) {
out.push_back(str.substr(pos));
break;
} else {
out.push_back(str.substr(pos, next - pos));
}
pos = next + 1;
} while (pos < str.size());
return out;
}

static PRIMFN(prim_job_create) {
JobTable *jobtable = static_cast<JobTable *>(data);
EXPECT(12);
EXPECT(14);
STRING(label, 0);
STRING(dir, 1);
STRING(stdin_file, 2);
Expand All @@ -1308,16 +1355,21 @@ static PRIMFN(prim_job_create) {
INTEGER_MPZ(keep, 7);
STRING(echo, 8);
STRING(stream_out, 9);
STRING(stream_err, 10);
INTEGER_MPZ(is_atty, 11);
STRING(tee_out_flat, 10);
STRING(stream_err, 11);
STRING(tee_err_flat, 12);
INTEGER_MPZ(is_atty, 13);

Hash hash;
REQUIRE(mpz_sizeinbase(signature, 2) <= 8 * sizeof(hash.data));
mpz_export(&hash.data[0], 0, 1, sizeof(hash.data[0]), 0, 0, signature);

std::vector<std::string> tee_out = null_sep_list(tee_out_flat->as_str());
std::vector<std::string> tee_err = null_sep_list(tee_err_flat->as_str());

Job *out =
Job::alloc(runtime.heap, jobtable->imp->db, label, dir, stdin_file, env, cmd,
mpz_cmp_si(keep, 0), echo->c_str(), stream_out->c_str(), stream_err->c_str());
mpz_cmp_si(keep, 0), echo->c_str(), stream_out->c_str(), tee_out, stream_err->c_str(), tee_err);

out->record = jobtable->imp->db->predict_job(out->code.data[0], &out->pathtime);

Expand Down Expand Up @@ -1396,7 +1448,7 @@ static PRIMFN(prim_job_cache) {
Value *joblist;
if (reuse.found && !jobtable->imp->check) {
Job *jobp = Job::claim(runtime.heap, jobtable->imp->db, dir, dir, stdin_file, env, cmd, true,
STREAM_ECHO, STREAM_INFO, STREAM_WARNING);
STREAM_ECHO, STREAM_INFO, std::vector<std::string>(), STREAM_WARNING, std::vector<std::string>());
JakeSiFive marked this conversation as resolved.
Show resolved Hide resolved
jobp->state = STATE_FORKED | STATE_STDOUT | STATE_STDERR | STATE_MERGED | STATE_FINISHED;
jobp->job = job;
jobp->record = reuse;
Expand Down
Loading