/** * Copyright (c) 2007-2012, Timothy Stack * * All rights reserved. * * Redistribution and use in source and binary forms, with or without * modification, are permitted provided that the following conditions are met: * * * Redistributions of source code must retain the above copyright notice, this * list of conditions and the following disclaimer. * * Redistributions in binary form must reproduce the above copyright notice, * this list of conditions and the following disclaimer in the documentation * and/or other materials provided with the distribution. * * Neither the name of Timothy Stack nor the names of its contributors * may be used to endorse or promote products derived from this software * without specific prior written permission. * * THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ''AS IS'' AND ANY * EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE * DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE FOR ANY * DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. * * @file piper_proc.cc */ #include "piper_proc.hh" #include #include #include #include #include #include #include #include #include #include "base/fs_util.hh" #include "base/lnav_log.hh" #include "config.h" #include "line_buffer.hh" static const char* STDIN_EOF_MSG = "---- END-OF-STDIN ----"; static ssize_t write_timestamp(int fd, off_t woff) { char time_str[64]; struct timeval tv; char ms_str[10]; gettimeofday(&tv, nullptr); strftime(time_str, sizeof(time_str), "%FT%T", localtime(&tv.tv_sec)); snprintf(ms_str, sizeof(ms_str), ".%03d", (int) (tv.tv_usec / 1000)); strcat(time_str, ms_str); strcat(time_str, " "); return pwrite(fd, time_str, strlen(time_str), woff); } piper_proc::piper_proc(auto_fd pipefd, bool timestamp, auto_fd filefd) : pp_fd(std::move(filefd)), pp_child(-1) { require(pipefd.get() >= 0); require(this->pp_fd.get() >= 0); log_perror(fcntl(this->pp_fd.get(), F_SETFD, FD_CLOEXEC)); this->pp_child = fork(); switch (this->pp_child) { case -1: throw error(errno); case 0: { line_buffer lb; off_t woff = 0, last_woff = 0; file_range last_range; auto open_res = lnav::filesystem::open_file("/dev/null", O_RDWR); if (open_res.isErr()) { fprintf(stderr, "unable to open /dev/null: %s\n", open_res.unwrapErr().c_str()); exit(EXIT_FAILURE); } auto nullfd = open_res.unwrap(); if (pipefd != STDIN_FILENO) { dup2(nullfd, STDIN_FILENO); } dup2(nullfd, STDOUT_FILENO); for (int fd_to_close = 0; fd_to_close < 1024; fd_to_close++) { int flags; if (fd_to_close == this->pp_fd.get()) { continue; } if ((flags = fcntl(fd_to_close, F_GETFD)) == -1) { continue; } if (flags & FD_CLOEXEC) { close(fd_to_close); } } log_perror(fcntl(pipefd.get(), F_SETFL, O_NONBLOCK)); lb.set_fd(pipefd); do { struct pollfd pfd = {lb.get_fd(), POLLIN, 0}; poll(&pfd, 1, -1); while (true) { auto load_result = lb.load_next_line(last_range); if (load_result.isErr()) { break; } auto li = load_result.unwrap(); if (li.li_partial && !lb.is_pipe_closed()) { break; } if (li.li_file_range.empty()) { break; } auto read_result = lb.read_range(li.li_file_range); if (read_result.isErr()) { break; } auto sbr = read_result.unwrap(); ssize_t wrc; last_woff = woff; if (timestamp) { wrc = write_timestamp(this->pp_fd, woff); if (wrc == -1) { perror("Unable to write to output file for stdin"); break; } woff += wrc; } /* Need to do pwrite here since the fd is used by the main * lnav process as well. */ wrc = pwrite( this->pp_fd, sbr.get_data(), sbr.length(), woff); if (wrc == -1) { perror("Unable to write to output file for stdin"); break; } woff += wrc; last_range = li.li_file_range; if (sbr.get_data()[sbr.length() - 1] != '\n' && (last_range.next_offset() != lb.get_file_size())) { woff = last_woff; } } } while (lb.is_pipe() && !lb.is_pipe_closed()); if (timestamp) { ssize_t wrc; wrc = write_timestamp(this->pp_fd, woff); if (wrc == -1) { perror("Unable to write to output file for stdin"); break; } woff += wrc; wrc = pwrite( this->pp_fd, STDIN_EOF_MSG, strlen(STDIN_EOF_MSG), woff); if (wrc == -1) { perror("Unable to write to output file for stdin"); break; } } } _exit(0); break; default: break; } } bool piper_proc::has_exited() { if (this->pp_child > 0) { int rc, status; rc = waitpid(this->pp_child, &status, WNOHANG); if (rc == -1 || rc == 0) { return false; } this->pp_child = -1; } return true; } piper_proc::~piper_proc() { if (this->pp_child > 0) { int status; kill(this->pp_child, SIGTERM); while (waitpid(this->pp_child, &status, 0) < 0 && (errno == EINTR)) { ; } this->pp_child = -1; } }