dinit/src/baseproc-service.cc

#include <cstring>
#include <cstdlib>

#include <sys/un.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>

#include "dinit.h"
#include "dinit-log.h"
#include "dinit-socket.h"
#include "proc-service.h"

#include "baseproc-sys.h"

/*
 * Base process implementation (base_process_service).
 *
 * See proc-service.h for interface documentation.
 */

void base_process_service::do_smooth_recovery() noexcept
{
    if (!restart_ps_process()) {
        unrecoverable_stop();
        services->process_queues();
    }
}

bool base_process_service::bring_up() noexcept
{
    if (!open_socket()) {
        return false;
    }

    bool start_success;
    if (in_auto_restart) {
        start_success = restart_ps_process();
    }
    else {
        restart_interval_count = 0;
        start_success = start_ps_process(exec_arg_parts,
                onstart_flags.starts_on_console || onstart_flags.shares_console);
        // start_ps_process updates last_start_time, use it also for restart_interval_time:
        restart_interval_time = last_start_time;

        // Arm start timer. (For restarts, this is only done once the restart interval expires).
        if (start_success) {
            if (start_timeout != time_val(0,0)) {
                process_timer.arm_timer_rel(event_loop, start_timeout);
                waiting_stopstart_timer = true;
            }
            else if (waiting_stopstart_timer) {
                process_timer.stop_timer(event_loop);
                waiting_stopstart_timer = false;
            }
        }
    }

    return start_success;
}

void base_process_service::handle_unexpected_termination() noexcept
{
    // unexpected termination, with possible restart
    stop_reason = stopped_reason_t::TERMINATED;

    // We want to circumvent the normal process of waiting for dependents to stop before we
    // attempt to restart, for two reasons:
    // 1) we can restart more quickly
    // 2) we can use the restart rate-limiting logic from restart_ps_process rather than
    //    the usual start_ps_process (the usual bring-up).
    // But we need to issue a forced stop and process queues, to discover our eventual target
    // state (so we know whether we actually want to restart or not).
    // Note we can't call forced_stop() directly here, because we need to set in_auto_restart in
    // between do_stop() and processing queues (so that it is set correctly if restart occurs):

    force_stop = true;
    do_stop();
    services->process_queues();

    if (get_state() == service_state_t::STOPPING) {
        // We must be waiting for dependents;
        // If we're going to restart, we can kick that off now:
        if (get_target_state() == service_state_t::STARTED && !pinned_stopped) {
            initiate_start();
            services->process_queues();
        }
    }
}

bool base_process_service::start_ps_process(const std::vector<const char *> &cmd, bool on_console) noexcept
{
    // In general, you can't tell whether fork/exec is successful. We use a pipe to communicate
    // success/failure from the child to the parent. The pipe is set CLOEXEC so a successful
    // exec closes the pipe, and the parent sees EOF. If the exec is unsuccessful, the errno
    // is written to the pipe, and the parent can read it.

    event_loop.get_time(last_start_time, clock_type::MONOTONIC);

    int pipefd[2];
    if (bp_sys::pipe2(pipefd, O_CLOEXEC)) {
        log(loglevel_t::ERROR, get_name(), ": can't create status check pipe: ", strerror(errno));
        return false;
    }

    const char * logfile = this->logfile.c_str();
    if (this->log_type == log_type_id::LOGFILE || this->log_type == log_type_id::NONE) {
    	if (*logfile == 0) {
    		logfile = "/dev/null";
    	}
    }
    else /* log_type_id::BUFFER */ {
    	if (this->log_output_fd == -1) {
    		int logfd[2];
    		// Note: we set CLOEXEC on the file descriptors here; when the output file descriptor is dup'd
    		// to stdout, this will be effectively removed for the output end
    	    if (bp_sys::pipe2(logfd, O_CLOEXEC)) {
    	        log(loglevel_t::ERROR, get_name(), ": can't create output pipe: ", strerror(errno));
    	        goto out_p;
    	    }
    	    this->log_input_fd = logfd[0];
    	    this->log_output_fd = logfd[1];
    	    try {
    	    	this->log_output_listener.add_watch(event_loop, logfd[0], dasynq::IN_EVENTS,
    	    			false /* not enabled */);
    	    }
    	    catch (...) {
    	    	log(loglevel_t::ERROR, get_name(), ": can't add output watch (insufficient resources)");
    	    	bp_sys::close(this->log_input_fd);
    	    	bp_sys::close(this->log_output_fd);
    	    	this->log_input_fd = -1;
    	    	this->log_output_fd = -1;
    	    	goto out_p;
    	    }
    	}
    	// (More is done below, after we have performed additional setup)
    }

    {
		bool child_status_registered = false;
		control_conn_t *control_conn = nullptr;

		int control_socket[2] = {-1, -1};
		int notify_pipe[2] = {-1, -1};
		bool have_notify = !notification_var.empty() || force_notification_fd != -1;
		ready_notify_watcher * rwatcher = have_notify ? get_ready_watcher() : nullptr;
		bool ready_watcher_registered = false;

		if (onstart_flags.pass_cs_fd) {
			if (dinit_socketpair(AF_UNIX, SOCK_STREAM, /* protocol */ 0, control_socket, SOCK_NONBLOCK)) {
				log(loglevel_t::ERROR, get_name(), ": can't create control socket: ", strerror(errno));
				goto out_lfd;
			}

			// Make the server side socket close-on-exec:
			int fdflags = bp_sys::fcntl(control_socket[0], F_GETFD);
			bp_sys::fcntl(control_socket[0], F_SETFD, fdflags | FD_CLOEXEC);

			try {
				control_conn = new control_conn_t(event_loop, services, control_socket[0]);
			}
			catch (std::exception &exc) {
				log(loglevel_t::ERROR, get_name(), ": can't launch process; out of memory");
				goto out_cs;
			}
		}

		if (have_notify) {
			// Create a notification pipe:
			if (bp_sys::pipe2(notify_pipe, 0) != 0) {
				log(loglevel_t::ERROR, get_name(), ": can't create notification pipe: ", strerror(errno));
				goto out_cs_h;
			}

			// Set the read side as close-on-exec:
			int fdflags = bp_sys::fcntl(notify_pipe[0], F_GETFD);
			bp_sys::fcntl(notify_pipe[0], F_SETFD, fdflags | FD_CLOEXEC);

			// add, but don't yet enable, readiness watcher:
			try {
				rwatcher->add_watch(event_loop, notify_pipe[0], dasynq::IN_EVENTS, false);
				ready_watcher_registered = true;
			}
			catch (std::exception &exc) {
				log(loglevel_t::ERROR, get_name(), ": can't add notification watch: ", exc.what());
				goto out_cs_h;
			}
		}

		if (log_type == log_type_id::BUFFER) {
	    	// Set watcher enabled if space in buffer
			if (log_buf_size > 0) {
				// Append a "restarted" message to buffer contents
				const char *restarting_msg = "\n(dinit: note: service restarted)\n";
				unsigned restarting_msg_len = strlen(restarting_msg);
				bool trailing_nl = log_buffer[log_buf_size - 1] == '\n';
				if (trailing_nl) {
					++restarting_msg; // trim leading newline
					--restarting_msg_len;
				}
				if (log_buf_size + restarting_msg_len >= log_buf_max) {
					goto skip_enable_log_watch;
				}
				if (!ensure_log_buffer_backing(log_buf_size + restarting_msg_len)) {
					goto skip_enable_log_watch;
				}
				memcpy(log_buffer.data() + log_buf_size, restarting_msg, restarting_msg_len);
				log_buf_size += restarting_msg_len;
			}
			log_output_listener.set_enabled(event_loop, true);
		}
		skip_enable_log_watch: ;

		// Set up complete, now fork and exec:

		pid_t forkpid;

		try {
			child_status_listener.add_watch(event_loop, pipefd[0], dasynq::IN_EVENTS);
			child_status_registered = true;

			// We specify a high priority (i.e. low priority value) so that process termination is
			// handled early. This means we have always recorded that the process is terminated by the
			// time that we handle events that might otherwise cause us to signal the process, so we
			// avoid sending a signal to an invalid (and possibly recycled) process ID.
			forkpid = child_listener.fork(event_loop, reserved_child_watch, dasynq::DEFAULT_PRIORITY - 10);
			reserved_child_watch = true;
		}
		catch (std::exception &e) {
			log(loglevel_t::ERROR, get_name(), ": could not fork: ", e.what());
			goto out_cs_h;
		}

		if (forkpid == 0) {
			const char * working_dir_c = nullptr;
			if (! working_dir.empty()) working_dir_c = working_dir.c_str();
			after_fork(getpid());
			run_proc_params run_params{cmd.data(), working_dir_c, logfile, pipefd[1], run_as_uid, run_as_gid, rlimits};
			run_params.on_console = on_console;
			run_params.in_foreground = !onstart_flags.shares_console;
			run_params.csfd = control_socket[1];
			run_params.socket_fd = socket_fd;
			run_params.notify_fd = notify_pipe[1];
			run_params.force_notify_fd = force_notification_fd;
			run_params.notify_var = notification_var.c_str();
			run_params.env_file = env_file.c_str();
			run_params.output_fd = log_output_fd;
			#if SUPPORT_CGROUPS
			run_params.run_in_cgroup = run_in_cgroup.c_str();
			#endif
			run_child_proc(run_params);
		}
		else {
			// Parent process
			pid = forkpid;

			bp_sys::close(pipefd[1]); // close the 'other end' fd
			if (control_socket[1] != -1) bp_sys::close(control_socket[1]);
			if (notify_pipe[1] != -1) bp_sys::close(notify_pipe[1]);
			notification_fd = notify_pipe[0];
			waiting_for_execstat = true;
			return true;
		}

		// Failure exit:

		out_cs_h:
		if (child_status_registered) {
			child_status_listener.deregister(event_loop);
		}

		if (notify_pipe[0] != -1) bp_sys::close(notify_pipe[0]);
		if (notify_pipe[1] != -1) bp_sys::close(notify_pipe[1]);
		if (ready_watcher_registered) {
			rwatcher->deregister(event_loop);
		}

		if (onstart_flags.pass_cs_fd) {
			delete control_conn;

			out_cs:
			bp_sys::close(control_socket[0]);
			bp_sys::close(control_socket[1]);
		}
    }

    out_lfd:
	if (log_input_fd != -1) {
		log_output_listener.deregister(event_loop);
		bp_sys::close(log_input_fd);
		bp_sys::close(log_output_fd);
		log_input_fd = -1;
		log_output_fd = -1;
	}

    out_p:
    bp_sys::close(pipefd[0]);
    bp_sys::close(pipefd[1]);

    return false;
}

base_process_service::base_process_service(service_set *sset, string name,
        service_type_t service_type_p, ha_string &&command,
        const std::list<std::pair<unsigned,unsigned>> &command_offsets,
        const std::list<prelim_dep> &deplist_p)
     : service_record(sset, name, service_type_p, deplist_p), child_listener(this),
       child_status_listener(this), process_timer(this), log_output_listener(this)
{
    program_name = std::move(command);
    exec_arg_parts = separate_args(program_name, command_offsets);

    restart_interval_count = 0;
    restart_interval_time = {0, 0};
    process_timer.service = this;
    process_timer.add_timer(event_loop);

    // By default, allow a maximum of 3 restarts within 10.0 seconds:
    restart_interval.seconds() = 10;
    restart_interval.nseconds() = 0;
    max_restart_interval_count = 3;

    waiting_restart_timer = false;
    waiting_stopstart_timer = false;
    reserved_child_watch = false;
    tracking_child = false;
}

void base_process_service::do_restart() noexcept
{
    // Actually perform process restart. We may be in smooth recovery (state = STARTED) or this may
    // be a regular restart.

    waiting_restart_timer = false;
    auto service_state = get_state();

    if (!start_ps_process(exec_arg_parts, have_console || onstart_flags.shares_console)) {
        if (service_state == service_state_t::STARTING) {
            failed_to_start();
        }
        else {
            // smooth recovery failure
            unrecoverable_stop();
        }
        services->process_queues();
    }
    else {
        // started process successfully (at least as far as fork)
        if (start_timeout != time_val(0,0)) {
            process_timer.arm_timer_rel(event_loop, start_timeout);
            waiting_stopstart_timer = true;
        }
    }
}

bool base_process_service::restart_ps_process() noexcept
{
    using time_val = dasynq::time_val;

    time_val current_time;
    event_loop.get_time(current_time, clock_type::MONOTONIC);

    // Check if enough time has lapsed since the previous restart. If not, start a timer:
    time_val tdiff = current_time - last_start_time;
    if (restart_delay <= tdiff) {
        // > restart delay (normally 200ms)
        do_restart();
    }
    else {
        time_val timeout = restart_delay - tdiff;
        process_timer.arm_timer_rel(event_loop, timeout);
        waiting_restart_timer = true;
    }
    return true;
}

bool base_process_service::interrupt_start() noexcept
{
    if (waiting_restart_timer) {
        process_timer.stop_timer(event_loop);
        waiting_restart_timer = false;
        return service_record::interrupt_start();
    }
    else {
        log(loglevel_t::WARN, "Interrupting start of service ", get_name(), " with pid ", pid,
                " (with SIGINT).");
        kill_pg(SIGINT);

        if (stop_timeout != time_val(0,0)) {
            process_timer.arm_timer_rel(event_loop, stop_timeout);
            waiting_stopstart_timer = true;
        }
        else if (waiting_stopstart_timer) {
            process_timer.stop_timer(event_loop);
            waiting_stopstart_timer = false;
        }

        set_state(service_state_t::STOPPING);
        return false;
    }
}

void base_process_service::kill_with_fire() noexcept
{
    if (pid != -1) {
        log(loglevel_t::WARN, "Service ", get_name(), " with pid ", pid,
                " exceeded allowed stop time; killing.");
        kill_pg(SIGKILL);
    }
}

void base_process_service::kill_pg(int signo) noexcept
{
    if (onstart_flags.signal_process_only) {
        bp_sys::kill(pid, signo);
    }
    else {
        pid_t pgid = bp_sys::getpgid(pid);
        if (pgid == -1) {
            // On some OSes (eg OpenBSD) we aren't allowed to get the pgid of a process in a different
            // session. If the process is in a different session, however, it must be a process group
            // leader and the pgid must equal the process id.
            pgid = pid;
        }
        bp_sys::kill(-pgid, signo);
    }
}

void base_process_service::timer_expired() noexcept
{
    waiting_stopstart_timer = false;

    // Timer expires if:
    // We are stopping, including after having startup cancelled (stop timeout, state is STOPPING); We are
    // starting (start timeout, state is STARTING); We are waiting for restart timer before restarting,
    // including smooth recovery (restart timeout, state is STARTING or STARTED).
    if (get_state() == service_state_t::STOPPING) {
        kill_with_fire();
    }
    else if (pid != -1) {
        // Starting, start timed out.
        log(loglevel_t::WARN, "Service ", get_name(), " with pid ", pid,
                " exceeded allowed start time; cancelling.");
        interrupt_start();
        stop_reason = stopped_reason_t::TIMEDOUT;
        failed_to_start(false, false);
    }
    else {
        // STARTING / STARTED, and we have no pid: must be restarting (smooth recovery if STARTED)
        do_restart();
    }
}

void base_process_service::becoming_inactive() noexcept
{
    if (socket_fd != -1) {
        close(socket_fd);
        socket_fd = -1;
    }
}

bool base_process_service::open_socket() noexcept
{
    if (socket_path.empty() || socket_fd != -1) {
        // No socket, or already open
        return true;
    }

    const char * saddrname = socket_path.c_str();

    // Check the specified socket path
    struct stat stat_buf;
    if (stat(saddrname, &stat_buf) == 0) {
        if ((stat_buf.st_mode & S_IFSOCK) == 0) {
            // Not a socket
            log(loglevel_t::ERROR, get_name(), ": activation socket file exists (and is not a socket)");
            return false;
        }
    }
    else if (errno != ENOENT) {
        // Other error
        log(loglevel_t::ERROR, get_name(), ": error checking activation socket: ", strerror(errno));
        return false;
    }

    // Remove stale socket file (if it exists).
    // We won't test the return from unlink - if it fails other than due to ENOENT, we should get an
    // error when we try to create the socket anyway.
    unlink(saddrname);

    uint sockaddr_size = offsetof(struct sockaddr_un, sun_path) + socket_path.length() + 1;
    struct sockaddr_un * name = static_cast<sockaddr_un *>(malloc(sockaddr_size));
    if (name == nullptr) {
        log(loglevel_t::ERROR, get_name(), ": opening activation socket: out of memory");
        return false;
    }

    name->sun_family = AF_UNIX;
    strcpy(name->sun_path, saddrname);

    int sockfd = dinit_socket(AF_UNIX, SOCK_STREAM, 0, SOCK_NONBLOCK | SOCK_CLOEXEC);
    if (sockfd == -1) {
        log(loglevel_t::ERROR, get_name(), ": error creating activation socket: ", strerror(errno));
        free(name);
        return false;
    }

    if (bind(sockfd, (struct sockaddr *) name, sockaddr_size) == -1) {
        log(loglevel_t::ERROR, get_name(), ": error binding activation socket: ", strerror(errno));
        close(sockfd);
        free(name);
        return false;
    }

    free(name);

    // POSIX (1003.1, 2013) says that fchown and fchmod don't necessarily work on sockets. We have to
    // use chown and chmod instead.
    if (chown(saddrname, socket_uid, socket_gid)) {
        log(loglevel_t::ERROR, get_name(), ": error setting activation socket owner/group: ",
                strerror(errno));
        close(sockfd);
        return false;
    }

    if (chmod(saddrname, socket_perms) == -1) {
        log(loglevel_t::ERROR, get_name(), ": Error setting activation socket permissions: ",
                strerror(errno));
        close(sockfd);
        return false;
    }

    if (listen(sockfd, 128) == -1) { // 128 "seems reasonable".
        log(loglevel_t::ERROR, ": error listening on activation socket: ", strerror(errno));
        close(sockfd);
        return false;
    }

    socket_fd = sockfd;
    return true;
}

bool base_process_service::ensure_log_buffer_backing(unsigned new_size) noexcept
{
	//  Note: we manage capacity manually to avoid it exceeding maximum
    if (log_buffer.size() < new_size) {
        if (log_buffer.capacity() < new_size) {
        	try {
				unsigned new_capacity = std::max((unsigned)log_buffer.capacity() * 2, new_size);
				new_capacity = std::min(new_capacity, log_buf_max);
				log_buffer.reserve(new_capacity);
				log_buffer.resize(new_capacity);
        	}
        	catch (std::bad_alloc &badalloc) {
        		log(loglevel_t::WARN, get_name(), ": cannot increase log buffer; out-of-memory");
        		return false;
        	}
        }
        else {
        	log_buffer.resize(new_size);
        }
    }
    return true;
}