h/fish-shell
1
Fork 0
mirror of https://github.com/fish-shell/fish-shell.git synced 2026-06-10 12:51:15 -03:00
Code Issues Packages Projects Releases Wiki Activity

io_buffer_t becomes io_bufferfill_t

Browse Source 
This makes some significant architectual improvements to io_pipe_t and
io_buffer_t.

Prior to this fix, io_buffer_t subclassed io_pipe_t. io_buffer_t is now
replaced with a class io_bufferfill_t, which does not subclass pipe.

io_pipe_t no longer remembers both fds. Instead it has an autoclose_fd_t,
so that the file descriptor ownership is clear.
This commit is contained in:
ridiculousfish
2019-01-31 16:05:42 -08:00
parent 7c256e7e51
commit 78bbcef356
8 changed files with 289 additions and 341 deletions
Download Patch File Download Diff File Expand all files Collapse all files

212
src/exec.cpp
View File

@@ -81,25 +81,6 @@ void exec_close(int fd) {
}
}
int exec_pipe(int fd[2]) {
ASSERT_IS_MAIN_THREAD();
int res;
while ((res = pipe(fd))) {
if (errno != EINTR) {
return res; // caller will call wperror
}
}
debug(4, L"Created pipe using fds %d and %d", fd[0], fd[1]);
// Pipes ought to be cloexec. Pipes are dup2'd the corresponding fds; the resulting fds are not
// cloexec.
set_cloexec(fd[0]);
set_cloexec(fd[1]);
return res;
}
/// Returns true if the redirection is a file redirection to a file other than /dev/null.
static bool redirection_is_to_real_file(const io_data_t *io) {
bool result = false;
@@ -246,8 +227,8 @@ static bool io_transmogrify(const io_chain_t &in_chain, io_chain_t *out_chain,
switch (in->io_mode) {
case io_mode_t::pipe:
case io_mode_t::bufferfill:
case io_mode_t::fd:
case io_mode_t::buffer:
case io_mode_t::close: {
// These redirections don't need transmogrification. They can be passed through.
out = in;
@@ -424,12 +405,12 @@ static bool exec_internal_builtin_proc(parser_t &parser, const std::shared_ptr<j
const io_chain_t &proc_io_chain, io_streams_t &streams) {
assert(p->type == INTERNAL_BUILTIN && "Process must be a builtin");
int local_builtin_stdin = STDIN_FILENO;
bool close_stdin = false;
autoclose_fd_t locally_opened_stdin{};
// If this is the first process, check the io redirections and see where we should
// be reading from.
if (pipe_read) {
local_builtin_stdin = pipe_read->pipe_fd[0];
local_builtin_stdin = pipe_read->pipe_fd();
} else if (const auto in = proc_io_chain.get_io_for_fd(STDIN_FILENO)) {
switch (in->io_mode) {
case io_mode_t::fd: {
@@ -448,20 +429,20 @@ static bool exec_internal_builtin_proc(parser_t &parser, const std::shared_ptr<j
}
case io_mode_t::pipe: {
const io_pipe_t *in_pipe = static_cast<const io_pipe_t *>(in.get());
local_builtin_stdin = in_pipe->pipe_fd[0];
if (in_pipe->fd == STDIN_FILENO) {
local_builtin_stdin = in_pipe->pipe_fd();
}
break;
}
case io_mode_t::file: {
// Do not set CLO_EXEC because child needs access.
const io_file_t *in_file = static_cast<const io_file_t *>(in.get());
local_builtin_stdin = open(in_file->filename_cstr, in_file->flags, OPEN_MASK);
if (local_builtin_stdin == -1) {
locally_opened_stdin =
autoclose_fd_t{open(in_file->filename_cstr, in_file->flags, OPEN_MASK)};
if (!locally_opened_stdin.valid()) {
debug(1, FILE_ERROR, in_file->filename_cstr);
wperror(L"open");
} else {
close_stdin = true;
}
local_builtin_stdin = locally_opened_stdin.fd();
break;
}
case io_mode_t::close: {
@@ -517,10 +498,6 @@ static bool exec_internal_builtin_proc(parser_t &parser, const std::shared_ptr<j
// execution so as not to confuse some job-handling builtins.
j->set_flag(job_flag_t::FOREGROUND, fg);
// If stdin has been redirected, close the redirection stream.
if (close_stdin) {
exec_close(local_builtin_stdin);
}
return true; // "success"
}
@@ -548,7 +525,11 @@ static bool handle_builtin_output(const std::shared_ptr<job_t> &j, process_t *p,
if (!must_fork && p->is_last_in_job) {
// We are handling reads directly in the main loop. Note that we may still end
// up forking.
const bool stdout_is_to_buffer = stdout_io && stdout_io->io_mode == io_mode_t::buffer;
const bool stdout_is_bufferfill =
(stdout_io && stdout_io->io_mode == io_mode_t::bufferfill);
const std::shared_ptr<io_buffer_t> stdout_buffer =
stdout_is_bufferfill ? static_cast<io_bufferfill_t *>(stdout_io.get())->buffer()
: nullptr;
const bool no_stdout_output = stdout_stream.empty();
const bool no_stderr_output = stderr_stream.empty();
const bool stdout_discarded = stdout_stream.buffer().discarded();
@@ -558,7 +539,7 @@ static bool handle_builtin_output(const std::shared_ptr<job_t> &j, process_t *p,
// need to fork or even output anything.
debug(4, L"Skipping fork: no output for internal builtin '%ls'", p->argv0());
fork_was_skipped = true;
} else if (no_stderr_output && stdout_is_to_buffer) {
} else if (no_stderr_output && stdout_buffer) {
// The builtin produced no stderr, and its stdout is going to an
// internal buffer. There is no need to fork. This helps out the
// performance quite a bit in complex completion code.
@@ -570,8 +551,7 @@ static bool handle_builtin_output(const std::shared_ptr<job_t> &j, process_t *p,
// also produce stderr.
debug(4, L"Skipping fork: buffered output for internal builtin '%ls'", p->argv0());
io_buffer_t *io_buffer = static_cast<io_buffer_t *>(stdout_io.get());
io_buffer->append_from_stream(stdout_stream);
stdout_buffer->append_from_stream(stdout_stream);
fork_was_skipped = true;
} else if (stdout_io.get() == NULL && stderr_io.get() == NULL) {
// We are writing to normal stdout and stderr. Just do it - no need to fork.
@@ -749,20 +729,16 @@ static bool exec_block_or_func_process(parser_t &parser, std::shared_ptr<job_t>
"Unexpected process type");
// Create an output buffer if we're piping to another process.
shared_ptr<io_buffer_t> block_output_io_buffer{};
shared_ptr<io_bufferfill_t> block_output_bufferfill{};
if (!p->is_last_in_job) {
// Be careful to handle failure, e.g. too many open fds.
block_output_io_buffer = io_buffer_t::create(STDOUT_FILENO, user_ios);
if (!block_output_io_buffer) {
block_output_bufferfill = io_bufferfill_t::create(user_ios);
if (!block_output_bufferfill) {
job_mark_process_as_failed(j, p);
return false;
} else {
// This looks sketchy, because we're adding this io buffer locally - they
// aren't in the process or job redirection list. Therefore select_try won't
// be able to read them. However we call block_output_io_buffer->read()
// below, which reads until EOF. So there's no need to select on this.
io_chain.push_back(block_output_io_buffer);
}
// Teach the job about its bufferfill, and add it to our io chain.
io_chain.push_back(block_output_bufferfill);
}
if (p->type == INTERNAL_FUNCTION) {
@@ -792,10 +768,8 @@ static bool exec_block_or_func_process(parser_t &parser, std::shared_ptr<job_t>
int status = proc_get_last_status();
// Handle output from a block or function. This usually means do nothing, but in the
// case of pipes, we have to buffer such io, since otherwise the internal pipe
// buffer might overflow.
if (!block_output_io_buffer.get()) {
// If we have a block output buffer, populate it now.
if (!block_output_bufferfill) {
// No buffer, so we exit directly. This means we have to manually set the exit
// status.
if (p->is_last_in_job) {
@@ -804,11 +778,16 @@ static bool exec_block_or_func_process(parser_t &parser, std::shared_ptr<job_t>
p->completed = 1;
return true;
}
assert(block_output_bufferfill && "Must have a block output bufferfiller");
// Here we must have a non-NULL block_output_io_buffer.
assert(block_output_io_buffer.get() != NULL);
io_chain.remove(block_output_io_buffer);
block_output_io_buffer->read();
// Remove our write pipe and forget it. This may close the pipe, unless another thread has
// claimed it (background write) or another process has inherited it.
auto block_output_buffer = block_output_bufferfill->buffer();
io_chain.remove(block_output_bufferfill);
block_output_bufferfill.reset();
// Make the buffer populate itself from whatever was written to the write pipe.
block_output_buffer->read_to_wouldblock();
// Resolve our IO chain to a sequence of dup2s.
auto dup2s = dup2_list_t::resolve_chain(io_chain);
@@ -816,7 +795,7 @@ static bool exec_block_or_func_process(parser_t &parser, std::shared_ptr<job_t>
return false;
}
const std::string buffer_contents = block_output_io_buffer->buffer().newline_serialized();
const std::string buffer_contents = block_output_buffer->buffer().newline_serialized();
const char *buffer = buffer_contents.data();
size_t count = buffer_contents.size();
if (count > 0) {
@@ -824,7 +803,7 @@ static bool exec_block_or_func_process(parser_t &parser, std::shared_ptr<job_t>
const char *fork_reason =
p->type == INTERNAL_BLOCK_NODE ? "internal block io" : "internal function io";
if (!fork_child_for_process(j, p, *dup2s, false, fork_reason, [&] {
exec_write_and_exit(block_output_io_buffer->fd, buffer, count, status);
exec_write_and_exit(STDOUT_FILENO, buffer, count, status);
})) {
return false;
}
@@ -844,19 +823,28 @@ static bool exec_process_in_job(parser_t &parser, process_t *p, std::shared_ptr<
autoclose_fd_t pipe_current_read,
autoclose_fd_t *out_pipe_next_read, const io_chain_t &all_ios,
size_t stdout_read_limit) {
// The IO chain for this process. It starts with the block IO, then pipes, and then gets any
// from the process.
io_chain_t process_net_io_chain = j->block_io_chain();
// The pipe this command will write to (if any).
shared_ptr<io_pipe_t> pipe_write;
// The pipe this command will read from (if any).
shared_ptr<io_pipe_t> pipe_read;
// See if we need a pipe.
// See if we need a pipe for the next command.
const bool pipes_to_next_command = !p->is_last_in_job;
if (pipes_to_next_command) {
// Construct our pipes.
auto local_pipes = make_autoclose_pipes(all_ios);
if (!local_pipes) {
debug(1, PIPE_ERROR);
wperror(L"pipe");
job_mark_process_as_failed(j, p);
return false;
}
// The write end of any pipe we create.
autoclose_fd_t pipe_current_write{};
pipe_write = std::make_shared<io_pipe_t>(p->pipe_write_fd, false /* not input */,
std::move(local_pipes->write));
*out_pipe_next_read = std::move(local_pipes->read);
}
// The pipes the current process write to and read from. Unfortunately these can't be just
// allocated on the stack, since j->io wants shared_ptr.
//
// The write pipe (destined for stdout) needs to occur before redirections. For example,
// with a redirection like this:
//
@@ -884,12 +872,10 @@ static bool exec_process_in_job(parser_t &parser, process_t *p, std::shared_ptr<
//
// which depends on the redirection being evaluated before the pipe. So the write end of the
// pipe comes first, the read pipe of the pipe comes last. See issue #966.
shared_ptr<io_pipe_t> pipe_write;
shared_ptr<io_pipe_t> pipe_read;
// Write pipe goes first.
if (pipes_to_next_command) {
pipe_write.reset(new io_pipe_t(p->pipe_write_fd, false));
// The IO chain for this process.
io_chain_t process_net_io_chain = j->block_io_chain();
if (pipe_write) {
process_net_io_chain.push_back(pipe_write);
}
@@ -897,10 +883,9 @@ static bool exec_process_in_job(parser_t &parser, process_t *p, std::shared_ptr<
process_net_io_chain.append(p->io_chain());
// Read pipe goes last.
if (!p->is_first_in_job) {
pipe_read.reset(new io_pipe_t(STDIN_FILENO, true));
// Record the current read in pipe_read.
pipe_read->pipe_fd[0] = pipe_current_read.fd();
if (pipe_current_read.valid()) {
pipe_read = std::make_shared<io_pipe_t>(STDIN_FILENO, true /* input */,
std::move(pipe_current_read));
process_net_io_chain.push_back(pipe_read);
}
@@ -918,36 +903,6 @@ static bool exec_process_in_job(parser_t &parser, process_t *p, std::shared_ptr<
parser.vars().export_arr();
}
// Set up fds that will be used in the pipe.
if (pipes_to_next_command) {
// debug( 1, L"%ls|%ls" , p->argv[0], p->next->argv[0]);
int local_pipe[2] = {-1, -1};
if (exec_pipe(local_pipe) == -1) {
debug(1, PIPE_ERROR);
wperror(L"pipe");
job_mark_process_as_failed(j, p);
return false;
}
// Ensure our pipe fds not conflict with any fd redirections. E.g. if the process is
// like 'cat <&5' then fd 5 must not be used by the pipe.
if (!pipe_avoid_conflicts_with_io_chain(local_pipe, all_ios)) {
// We failed. The pipes were closed for us.
wperror(L"dup");
job_mark_process_as_failed(j, p);
return false;
}
// This tells the redirection about the fds, but the redirection does not close them.
assert(local_pipe[0] >= 0);
assert(local_pipe[1] >= 0);
memcpy(pipe_write->pipe_fd, local_pipe, sizeof(int) * 2);
// Record our pipes.
pipe_current_write.reset(local_pipe[1]);
out_pipe_next_read->reset(local_pipe[0]);
}
// Execute the process.
switch (p->type) {
case INTERNAL_FUNCTION:
@@ -1008,18 +963,13 @@ bool exec_job(parser_t &parser, shared_ptr<job_t> j) {
}
}
// Verify that all io_mode_t::buffers are output. We used to support a (single, hacked-in)
// magical input io_mode_t::buffer used by fish_pager, but now the claim is that there are no
// more clients and it is removed. This assertion double-checks that.
size_t stdout_read_limit = 0;
const io_chain_t all_ios = j->all_io_redirections();
for (size_t idx = 0; idx < all_ios.size(); idx++) {
const shared_ptr<io_data_t> &io = all_ios.at(idx);
if ((io->io_mode == io_mode_t::buffer)) {
io_buffer_t *io_buffer = static_cast<io_buffer_t *>(io.get());
assert(!io_buffer->is_input);
stdout_read_limit = io_buffer->buffer().limit();
for (auto &io : all_ios) {
if ((io->io_mode == io_mode_t::bufferfill)) {
// The read limit is dictated by the last bufferfill.
const auto *bf = static_cast<io_bufferfill_t *>(io.get());
stdout_read_limit = bf->buffer()->buffer().limit();
}
}
@@ -1028,21 +978,6 @@ bool exec_job(parser_t &parser, shared_ptr<job_t> j) {
DIE("this should be unreachable");
}
// We may have block IOs that conflict with fd redirections. For example, we may have a command
// with a redireciton like <&3; we may also have chosen 3 as the fd for our pipe. Ensure we have
// no conflicts.
for (const auto io : all_ios) {
if (io->io_mode == io_mode_t::buffer) {
auto *io_buffer = static_cast<io_buffer_t *>(io.get());
if (!io_buffer->avoid_conflicts_with_io_chain(all_ios)) {
// We could not avoid conflicts, probably due to fd exhaustion. Mark an error.
exec_error = true;
job_mark_process_as_failed(j, j->processes.front().get());
break;
}
}
}
// This loop loops over every process_t in the job, starting it as appropriate. This turns out
// to be rather complex, since a process_t can be one of many rather different things.
//
@@ -1098,29 +1033,30 @@ static int exec_subshell_internal(const wcstring &cmd, parser_t &parser, wcstrin
// IO buffer creation may fail (e.g. if we have too many open files to make a pipe), so this may
// be null.
const shared_ptr<io_buffer_t> io_buffer(
io_buffer_t::create(STDOUT_FILENO, io_chain_t(), is_subcmd ? read_byte_limit : 0));
if (io_buffer.get() != NULL) {
size_t read_limit = is_subcmd ? read_byte_limit : 0;
std::shared_ptr<io_buffer_t> buffer;
if (auto bufferfill = io_bufferfill_t::create(io_chain_t{}, read_limit)) {
parser_t &parser = parser_t::principal_parser();
if (parser.eval(cmd, io_chain_t{io_buffer}, SUBST) == 0) {
if (parser.eval(cmd, io_chain_t{bufferfill}, SUBST) == 0) {
subcommand_status = proc_get_last_status();
}
io_buffer->read();
buffer = bufferfill->buffer();
bufferfill.reset();
buffer->read_to_wouldblock();
}
if (io_buffer->buffer().discarded()) subcommand_status = STATUS_READ_TOO_MUCH;
if (buffer && buffer->buffer().discarded()) subcommand_status = STATUS_READ_TOO_MUCH;
// If the caller asked us to preserve the exit status, restore the old status. Otherwise set the
// status of the subcommand.
proc_set_last_status(apply_exit_status ? subcommand_status : prev_status);
is_subshell = prev_subshell;
if (lst == NULL || io_buffer.get() == NULL) {
if (lst == NULL || !buffer) {
return subcommand_status;
}
// Walk over all the elements.
for (const auto &elem : io_buffer->buffer().elements()) {
for (const auto &elem : buffer->buffer().elements()) {
if (elem.is_explicitly_separated()) {
// Just append this one.
lst->push_back(str2wcstring(elem.contents));