No OneTemporary
Actions

Size

28 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/c_src/exile.c b/c_src/exile.c
	index 647a731..fbb7836 100644
	--- a/c_src/exile.c
	+++ b/c_src/exile.c
	@@ -1,392 +1,357 @@
	#ifndef _POSIX_C_SOURCE
	#define _POSIX_C_SOURCE 200809L
	#endif

	#include "erl_nif.h"
	#include <errno.h>
	#include <fcntl.h>
	#include <signal.h>
	#include <stdbool.h>
	#include <stdio.h>
	#include <string.h>
	#include <sys/types.h>
	#include <sys/wait.h>
	#include <unistd.h>

	#ifdef ERTS_DIRTY_SCHEDULERS
	#define USE_DIRTY_IO ERL_NIF_DIRTY_JOB_IO_BOUND
	#else
	#define USE_DIRTY_IO 0
	#endif

	// #define DEBUG

	#ifdef DEBUG
	#define debug(...) \
	do { \
	enif_fprintf(stderr, "%s:%d\t(fn \"%s\") - ", __FILE__, __LINE__, \
	__func__); \
	enif_fprintf(stderr, __VA_ARGS__); \
	enif_fprintf(stderr, "\n"); \
	} while (0)
	#else
	#define debug(...)
	#endif

	#define error(...) \
	do { \
	enif_fprintf(stderr, "%s:%d\t(fn: \"%s\") - ", __FILE__, __LINE__, \
	__func__); \
	enif_fprintf(stderr, __VA_ARGS__); \
	enif_fprintf(stderr, "\n"); \
	} while (0)

	#define assert_argc(argc, count) \
	if (argc != count) { \
	error("number of arguments must be %d", count); \
	return enif_make_badarg(env); \
	}

	static const int UNBUFFERED_READ = -1;
	static const int PIPE_BUF_SIZE = 65535;

	static const int FD_CLOSED = -1;

	static ERL_NIF_TERM ATOM_TRUE;
	static ERL_NIF_TERM ATOM_FALSE;
	static ERL_NIF_TERM ATOM_OK;
	static ERL_NIF_TERM ATOM_ERROR;
	static ERL_NIF_TERM ATOM_UNDEFINED;
	static ERL_NIF_TERM ATOM_INVALID_FD;
	-static ERL_NIF_TERM ATOM_PIPE_CLOSED;
	static ERL_NIF_TERM ATOM_EAGAIN;
	-static ERL_NIF_TERM ATOM_ALLOC_FAILED;

	static ERL_NIF_TERM ATOM_SIGKILL;
	static ERL_NIF_TERM ATOM_SIGTERM;

	-/* command exit types */
	-static ERL_NIF_TERM ATOM_EXIT;
	-static ERL_NIF_TERM ATOM_SIGNALED;
	-static ERL_NIF_TERM ATOM_STOPPED;
	-
	-enum exit_type { NORMAL_EXIT, SIGNALED, STOPPED };
	-
	static void close_fd(int *fd) {
	if (*fd != FD_CLOSED) {
	close(*fd);
	*fd = FD_CLOSED;
	}
	}

	static void io_resource_dtor(ErlNifEnv env, void obj) {
	int fd = (int )obj;
	close_fd(fd);
	debug("Exile io_resource_dtor called");
	}

	static void io_resource_stop(ErlNifEnv env, void obj, int fd,
	int is_direct_call) {
	debug("Exile io_resource_stop called %d", fd);
	}

	static void io_resource_down(ErlNifEnv env, void obj, ErlNifPid *pid,
	ErlNifMonitor *monitor) {
	int fd = (int )obj;
	close_fd(fd);
	debug("Exile io_resource_down called");
	}

	static ErlNifResourceTypeInit io_rt_init = {io_resource_dtor, io_resource_stop,
	io_resource_down};

	static ErlNifResourceType *FD_RT;

	static inline ERL_NIF_TERM make_ok(ErlNifEnv *env, ERL_NIF_TERM term) {
	return enif_make_tuple2(env, ATOM_OK, term);
	}

	static inline ERL_NIF_TERM make_error(ErlNifEnv *env, ERL_NIF_TERM term) {
	return enif_make_tuple2(env, ATOM_ERROR, term);
	}

	/* time is assumed to be in microseconds */
	static void notify_consumed_timeslice(ErlNifEnv *env, ErlNifTime start,
	ErlNifTime stop) {
	ErlNifTime pct;

	pct = (ErlNifTime)((stop - start) / 10);
	if (pct > 100)
	pct = 100;
	else if (pct == 0)
	pct = 1;
	enif_consume_timeslice(env, pct);
	}

	static int select_write(ErlNifEnv env, int fd) {
	int ret =
	enif_select(env, *fd, ERL_NIF_SELECT_WRITE, fd, NULL, ATOM_UNDEFINED);

	if (ret != 0)
	perror("select_write()");
	return ret;
	}

	static ERL_NIF_TERM nif_write(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	assert_argc(argc, 2);

	ErlNifTime start;
	ssize_t size;
	ErlNifBinary bin;
	int write_errno;
	int *fd;

	start = enif_monotonic_time(ERL_NIF_USEC);

	if (!enif_get_resource(env, argv[0], FD_RT, (void **)&fd))
	return make_error(env, ATOM_INVALID_FD);

	if (enif_inspect_binary(env, argv[1], &bin) != true)
	return enif_make_badarg(env);

	if (bin.size == 0)
	return enif_make_badarg(env);

	/* should we limit the bin.size here? */
	size = write(*fd, bin.data, bin.size);
	write_errno = errno;

	notify_consumed_timeslice(env, start, enif_monotonic_time(ERL_NIF_USEC));

	if (size >= (ssize_t)bin.size) { // request completely satisfied
	return make_ok(env, enif_make_int(env, size));
	} else if (size >= 0) { // request partially satisfied
	int retval = select_write(env, fd);
	if (retval != 0)
	return make_error(env, enif_make_int(env, retval));
	return make_ok(env, enif_make_int(env, size));
	} else if (write_errno == EAGAIN \|\| write_errno == EWOULDBLOCK) { // busy
	int retval = select_write(env, fd);
	if (retval != 0)
	return make_error(env, enif_make_int(env, retval));
	return make_error(env, ATOM_EAGAIN);
	} else { // Error
	perror("write()");
	return make_error(env, enif_make_int(env, write_errno));
	}
	}

	static int select_read(ErlNifEnv env, int fd) {
	int ret =
	enif_select(env, *fd, ERL_NIF_SELECT_READ, fd, NULL, ATOM_UNDEFINED);

	if (ret != 0)
	perror("select_read()");
	return ret;
	}

	static ERL_NIF_TERM nif_create_fd(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	assert_argc(argc, 1);

	ERL_NIF_TERM term;
	int *fd;

	fd = enif_alloc_resource(FD_RT, sizeof(int));

	if (!enif_get_int(env, argv[0], fd))
	goto error_exit;

	term = enif_make_resource(env, fd);
	enif_release_resource(fd);

	return make_ok(env, term);

	error_exit:
	enif_release_resource(fd);
	return ATOM_ERROR;
	}

	static ERL_NIF_TERM nif_read(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	assert_argc(argc, 2);

	ErlNifTime start;
	int size, demand;
	int *fd;

	start = enif_monotonic_time(ERL_NIF_USEC);

	if (!enif_get_resource(env, argv[0], FD_RT, (void **)&fd))
	return make_error(env, ATOM_INVALID_FD);

	if (!enif_get_int(env, argv[1], &demand))
	return enif_make_badarg(env);

	size = demand;

	if (demand == UNBUFFERED_READ) {
	size = PIPE_BUF_SIZE;
	} else if (demand < 1) {
	return enif_make_badarg(env);
	} else if (demand > PIPE_BUF_SIZE) {
	size = PIPE_BUF_SIZE;
	}

	unsigned char buf[size];
	ssize_t result = read(*fd, buf, size);
	int read_errno = errno;

	ERL_NIF_TERM bin_term = 0;
	if (result >= 0) {
	/* no need to release this binary */
	unsigned char *temp = enif_make_new_binary(env, result, &bin_term);
	memcpy(temp, buf, result);
	}

	notify_consumed_timeslice(env, start, enif_monotonic_time(ERL_NIF_USEC));

	if (result >= 0) {
	/* we do not 'select' if demand completely satisfied OR EOF OR its
	* UNBUFFERED_READ */
	if (result == demand \|\| result == 0 \|\| demand == UNBUFFERED_READ) {
	return make_ok(env, bin_term);
	} else { // demand partially satisfied
	int retval = select_read(env, fd);
	if (retval != 0)
	return make_error(env, enif_make_int(env, retval));
	return make_ok(env, bin_term);
	}
	} else {
	if (read_errno == EAGAIN \|\| read_errno == EWOULDBLOCK) { // busy
	int retval = select_read(env, fd);
	if (retval != 0)
	return make_error(env, enif_make_int(env, retval));
	return make_error(env, ATOM_EAGAIN);
	} else { // Error
	perror("read()");
	return make_error(env, enif_make_int(env, read_errno));
	}
	}
	}

	static ERL_NIF_TERM nif_close(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	assert_argc(argc, 1);

	int *fd;

	if (!enif_get_resource(env, argv[0], FD_RT, (void **)&fd))
	return make_error(env, ATOM_INVALID_FD);

	close_fd(fd);

	return ATOM_OK;
	}

	static ERL_NIF_TERM nif_is_os_pid_alive(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	assert_argc(argc, 1);

	pid_t pid;

	// we should not assume pid type to be `int`?
	if (!enif_get_int(env, argv[0], (int *)&pid))
	return enif_make_badarg(env);

	int result = kill(pid, 0);

	if (result == 0)
	return ATOM_TRUE;
	else
	return ATOM_FALSE;
	}

	static ERL_NIF_TERM nif_kill(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	assert_argc(argc, 2);

	pid_t pid;
	int ret;

	// we should not assume pid type to be `int`?
	if (!enif_get_int(env, argv[0], (int *)&pid))
	return enif_make_badarg(env);

	if (enif_compare(argv[1], ATOM_SIGKILL) == 0)
	ret = kill(pid, SIGKILL);
	else if (enif_compare(argv[1], ATOM_SIGTERM) == 0)
	ret = kill(pid, SIGTERM);
	else
	return enif_make_badarg(env);

	if (ret != 0) {
	perror("[exile] failed to send signal");
	return make_error(
	env, enif_make_string(env, "failed to send signal", ERL_NIF_LATIN1));
	}

	return ATOM_OK;
	}

	-/* static ERL_NIF_TERM make_exit_term(ErlNifEnv env, ExecContext ctx) { */
	-/* switch (ctx->exit_type) { */
	-/* case NORMAL_EXIT: */
	-/* return make_ok(env, enif_make_tuple2(env, ATOM_EXIT, */
	-/* enif_make_int(env,
	- * ctx->exit_status))); */
	-/* case SIGNALED: */
	-/* /\* exit_status here points to signal number \/ /
	-/* return make_ok(env, enif_make_tuple2(env, ATOM_SIGNALED, */
	-/* enif_make_int(env,
	- * ctx->exit_status))); */
	-/* case STOPPED: */
	-/* return make_ok(env, enif_make_tuple2(env, ATOM_STOPPED, */
	-/* enif_make_int(env,
	- * ctx->exit_status))); */
	-/* default: */
	-/* error("Invalid wait status"); */
	-/* return make_error(env, ATOM_UNDEFINED); */
	-/* } */
	-/* } */
	-
	static int on_load(ErlNifEnv env, void *priv, ERL_NIF_TERM load_info) {
	FD_RT =
	enif_open_resource_type_x(env, "exile_resource", &io_rt_init,
	ERL_NIF_RT_CREATE \| ERL_NIF_RT_TAKEOVER, NULL);

	ATOM_TRUE = enif_make_atom(env, "true");
	ATOM_FALSE = enif_make_atom(env, "false");
	ATOM_OK = enif_make_atom(env, "ok");
	ATOM_ERROR = enif_make_atom(env, "error");
	ATOM_UNDEFINED = enif_make_atom(env, "undefined");
	ATOM_INVALID_FD = enif_make_atom(env, "invalid_fd_resource");
	- ATOM_PIPE_CLOSED = enif_make_atom(env, "closed_pipe");
	- ATOM_EXIT = enif_make_atom(env, "exit");
	- ATOM_SIGNALED = enif_make_atom(env, "signaled");
	- ATOM_STOPPED = enif_make_atom(env, "stopped");
	ATOM_EAGAIN = enif_make_atom(env, "eagain");
	- ATOM_ALLOC_FAILED = enif_make_atom(env, "alloc_failed");

	ATOM_SIGTERM = enif_make_atom(env, "sigterm");
	ATOM_SIGKILL = enif_make_atom(env, "sigkill");

	return 0;
	}

	static void on_unload(ErlNifEnv env, void priv) {
	debug("exile unload");
	enif_free(priv);
	}

	static ErlNifFunc nif_funcs[] = {
	{"nif_read", 2, nif_read, USE_DIRTY_IO},
	{"nif_create_fd", 1, nif_create_fd, USE_DIRTY_IO},
	{"nif_write", 2, nif_write, USE_DIRTY_IO},
	{"nif_close", 1, nif_close, USE_DIRTY_IO},
	{"nif_is_os_pid_alive", 1, nif_is_os_pid_alive, USE_DIRTY_IO},
	{"nif_kill", 2, nif_kill, USE_DIRTY_IO}};

	ERL_NIF_INIT(Elixir.Exile.ProcessNif, nif_funcs, &on_load, NULL, NULL,
	&on_unload)
	diff --git a/lib/exile/process.ex b/lib/exile/process.ex
	index 7770787..188f00f 100644
	--- a/lib/exile/process.ex
	+++ b/lib/exile/process.ex
	@@ -1,459 +1,455 @@
	defmodule Exile.Process do
	@moduledoc """
	GenServer which wraps spawned external command.

	One should use `Exile.stream!` over `Exile.Process`. stream internally manages this server for you. Use this only if you need more control over the life-cycle OS process.

	## Overview
	`Exile.Process` is an alternative primitive for Port. It has different interface and approach to running external programs to solve the issues associated with the ports.

	### When compared to Port
	* it is demand driven. User explicitly has to `read` output of the command and the progress of the external command is controlled using OS pipes. so unlike Port, this never cause memory issues in beam by loading more than we can consume
	* it can close stdin of the program explicitly
	* does not create zombie process. It always tries to cleanup resources

	At high level it makes non-blocking asynchronous system calls to execute and interact with the external program. It completely bypasses beam implementation for the same using NIF. It uses `select()` system call for asynchronous IO. Most of the system calls are non-blocking, so it does not has adverse effect on scheduler. Issues such as "scheduler collapse".
	"""

	alias Exile.ProcessNif, as: Nif
	require Logger
	use GenServer

	@default_opts [env: []]

	@doc """
	Starts `Exile.ProcessServer`

	Starts external program using `cmd_with_args` with options `opts`

	`cmd_with_args` must be a list containing command with arguments. example: `["cat", "file.txt"]`.

	### Options
	* `cd` - the directory to run the command in
	* `env` - an enumerable of tuples containing environment key-value. These can be accessed in the external program
	"""
	def start_link(cmd_with_args, opts \\ []) do
	opts = Keyword.merge(@default_opts, opts)

	with {:ok, args} <- normalize_args(cmd_with_args, opts) do
	GenServer.start(__MODULE__, args)
	end
	end

	def close_stdin(process) do
	GenServer.call(process, :close_stdin, :infinity)
	end

	def write(process, iodata) do
	GenServer.call(process, {:write, IO.iodata_to_binary(iodata)}, :infinity)
	end

	def read(process, size) when (is_integer(size) and size > 0) or size == :unbuffered do
	GenServer.call(process, {:read, size}, :infinity)
	end

	def read(process) do
	GenServer.call(process, {:read, :unbuffered}, :infinity)
	end

	def kill(process, signal) when signal in [:sigkill, :sigterm] do
	GenServer.call(process, {:kill, signal}, :infinity)
	end

	def await_exit(process, timeout \\ :infinity) do
	GenServer.call(process, {:await_exit, timeout}, :infinity)
	end

	def os_pid(process) do
	GenServer.call(process, :os_pid, :infinity)
	end

	def stop(process), do: GenServer.call(process, :stop, :infinity)

	## Server

	defmodule Pending do
	defstruct bin: [], remaining: 0, client_pid: nil
	end

	defstruct [
	:args,
	:errno,
	:port,
	:socket_path,
	:stdin,
	:stdout,
	:context,
	:status,
	await: %{},
	pending_read: nil,
	pending_write: nil
	]

	alias __MODULE__

	def init(args) do
	state = %__MODULE__{
	args: args,
	errno: nil,
	status: :init,
	await: %{},
	pending_read: %Pending{},
	pending_write: %Pending{}
	}

	{:ok, state, {:continue, nil}}
	end

	def handle_continue(nil, state) do
	%{cmd_with_args: cmd_with_args, cd: cd, env: env} = state.args
	socket_path = socket_path()

	{:ok, uds} = :socket.open(:local, :stream, :default)
	_ = :file.delete(socket_path)
	{:ok, _} = :socket.bind(uds, %{family: :local, path: socket_path})
	:ok = :socket.listen(uds)

	port = exec(cmd_with_args, socket_path, env, cd)
	{:os_pid, os_pid} = Port.info(port, :os_pid)
	Exile.Watcher.watch(self(), os_pid, socket_path)

	{write_fd_int, read_fd_int} = receive_fds(uds)

	{:ok, write_fd} = Nif.nif_create_fd(write_fd_int)
	{:ok, read_fd} = Nif.nif_create_fd(read_fd_int)

	{:noreply,
	%Process{
	state
	\| port: port,
	status: :start,
	socket_path: socket_path,
	stdin: read_fd,
	stdout: write_fd
	}}
	end

	def handle_call(:stop, _from, state) do
	- # watcher will take care of termination of external process
	-
	# TODO: pending write and read should receive "stopped" return
	# value instead of exit signal
	case state.status do
	{:exit, _} -> :ok
	_ -> Port.close(state.port)
	end

	{:stop, :normal, :ok, state}
	end

	def handle_call(:close_stdin, _from, state) do
	case state.status do
	{:exit, _} -> {:reply, :ok, state}
	_ -> do_close(state, :stdin)
	end
	end

	def handle_call({:await_exit, _}, _from, %{status: {:exit, status}} = state) do
	{:reply, {:ok, {:exit, status}}, state}
	end

	def handle_call({:await_exit, timeout}, from, %{status: :start} = state) do
	tref =
	if timeout != :infinity do
	Elixir.Process.send_after(self(), {:await_exit_timeout, from}, timeout)
	else
	nil
	end

	{:noreply, put_timer(state, from, :timeout, tref)}
	end

	def handle_call(_, _from, %{status: {:exit, status}} = state),
	do: {:reply, {:error, {:exit, status}}, state}

	def handle_call({:write, binary}, from, state) when is_binary(binary) do
	pending = %Pending{bin: binary, client_pid: from}
	do_write(%Process{state \| pending_write: pending})
	end

	def handle_call({:read, size}, from, state) do
	pending = %Pending{remaining: size, client_pid: from}
	do_read(%Process{state \| pending_read: pending})
	end

	def handle_call(:os_pid, _from, state) do
	case Port.info(state.port, :os_pid) do
	{:os_pid, os_pid} ->
	{:reply, {:ok, os_pid}, state}

	:undefined ->
	Logger.debug("Process not alive")
	{:reply, :undefined, state}
	end
	end

	def handle_call({:kill, signal}, _from, state) do
	{:reply, signal(state.port, signal), state}
	end

	def handle_info({:await_exit_timeout, from}, state) do
	GenServer.reply(from, :timeout)
	{:noreply, clear_await(state, from)}
	end

	def handle_info({:select, _write_resource, _ref, :ready_output}, state), do: do_write(state)

	def handle_info({:select, _read_resource, _ref, :ready_input}, state), do: do_read(state)

	def handle_info({port, {:exit_status, exit_status}}, %{port: port} = state),
	do: handle_port_exit(exit_status, state)

	def handle_info(msg, _state), do: raise(msg)

	defp handle_port_exit(exit_status, %{await: await} = state) do
	state =
	Enum.reduce(await, state, fn {from, _}, state ->
	GenServer.reply(from, {:ok, {:exit, exit_status}})
	cancel_timer(state, from, :timeout)
	clear_await(state, from)
	end)

	{:noreply, %{state \| status: {:exit, exit_status}}}
	end

	defp do_write(%Process{pending_write: %Pending{bin: <<>>}} = state) do
	GenServer.reply(state.pending_write.client_pid, :ok)
	{:noreply, %{state \| pending_write: %Pending{}}}
	end

	defp do_write(%Process{pending_write: pending} = state) do
	case Nif.nif_write(state.stdin, pending.bin) do
	{:ok, size} ->
	if size < byte_size(pending.bin) do
	binary = binary_part(pending.bin, size, byte_size(pending.bin) - size)
	{:noreply, %{state \| pending_write: %Pending{pending \| bin: binary}}}
	else
	GenServer.reply(pending.client_pid, :ok)
	{:noreply, %{state \| pending_write: %Pending{}}}
	end

	{:error, :eagain} ->
	{:noreply, state}

	{:error, errno} ->
	GenServer.reply(pending.client_pid, {:error, errno})
	{:noreply, %{state \| errno: errno}}
	end
	end

	defp do_read(%Process{pending_read: %Pending{remaining: :unbuffered} = pending} = state) do
	case Nif.nif_read(state.stdout, -1) do
	{:ok, <<>>} ->
	GenServer.reply(pending.client_pid, {:eof, []})
	{:noreply, state}

	{:ok, binary} ->
	GenServer.reply(pending.client_pid, {:ok, binary})
	{:noreply, state}

	{:error, :eagain} ->
	{:noreply, state}

	{:error, errno} ->
	GenServer.reply(pending.client_pid, {:error, errno})
	{:noreply, %{state \| errno: errno}}
	end
	end

	defp do_read(%Process{pending_read: pending} = state) do
	case Nif.nif_read(state.stdout, pending.remaining) do
	{:ok, <<>>} ->
	GenServer.reply(pending.client_pid, {:eof, pending.bin})
	{:noreply, %Process{state \| pending_read: %Pending{}}}

	{:ok, binary} ->
	if byte_size(binary) < pending.remaining do
	pending = %Pending{
	pending
	\| bin: [pending.bin \| binary],
	remaining: pending.remaining - byte_size(binary)
	}

	{:noreply, %Process{state \| pending_read: pending}}
	else
	GenServer.reply(pending.client_pid, {:ok, [state.pending_read.bin \| binary]})
	{:noreply, %Process{state \| pending_read: %Pending{}}}
	end

	{:error, :eagain} ->
	{:noreply, state}

	{:error, errno} ->
	GenServer.reply(pending.client_pid, {:error, errno})
	{:noreply, %{state \| pending_read: %Pending{}, errno: errno}}
	end
	end

	defp do_close(state, type) do
	fd =
	if type == :stdin do
	state.stdin
	else
	state.stdout
	end

	case Nif.nif_close(fd) do
	:ok ->
	{:reply, :ok, state}

	{:error, errno} ->
	+ # FIXME: correct
	raise errno
	{:reply, {:error, errno}, %Process{state \| errno: errno}}
	end
	end

	defp clear_await(state, from) do
	%Process{state \| await: Map.delete(state.await, from)}
	end

	defp cancel_timer(state, from, key) do
	case get_timer(state, from, key) do
	nil -> :ok
	tref -> Elixir.Process.cancel_timer(tref)
	end
	end

	defp put_timer(state, from, key, timer) do
	if Map.has_key?(state.await, from) do
	await = put_in(state.await, [from, key], timer)
	%Process{state \| await: await}
	else
	%Process{state \| await: %{from => %{key => timer}}}
	end
	end

	defp get_timer(state, from, key), do: get_in(state.await, [from, key])

	defp normalize_cmd(cmd) do
	path = System.find_executable(cmd)

	if path do
	{:ok, to_charlist(path)}
	else
	{:error, "command not found: #{inspect(cmd)}"}
	end
	end

	defp normalize_cmd_args(args) do
	if is_list(args) do
	{:ok, Enum.map(args, &to_charlist/1)}
	else
	{:error, "command arguments must be list of strings. #{inspect(args)}"}
	end
	end

	defp normalize_cd(nil), do: {:ok, ''}

	defp normalize_cd(cd) do
	if File.exists?(cd) && File.dir?(cd) do
	{:ok, to_charlist(cd)}
	else
	{:error, "`:cd` must be valid directory path"}
	end
	end

	defp normalize_env(nil), do: {:ok, []}

	defp normalize_env(env) do
	env =
	Enum.map(env, fn {key, value} ->
	{to_charlist(key), to_charlist(value)}
	end)

	{:ok, env}
	end

	defp validate_opts_fields(opts) do
	{_, additional_opts} = Keyword.split(opts, [:cd, :env])

	if Enum.empty?(additional_opts) do
	:ok
	else
	{:error, "invalid opts: #{inspect(additional_opts)}"}
	end
	end

	defp normalize_args([cmd \| args], opts) when is_list(opts) do
	with {:ok, cmd} <- normalize_cmd(cmd),
	{:ok, args} <- normalize_cmd_args(args),
	:ok <- validate_opts_fields(opts),
	{:ok, cd} <- normalize_cd(opts[:cd]),
	{:ok, env} <- normalize_env(opts[:env]) do
	{:ok, %{cmd_with_args: [cmd \| args], cd: cd, env: env}}
	end
	end

	defp normalize_args(_, _), do: {:error, "invalid arguments"}

	@spawner_path :filename.join(:code.priv_dir(:exile), "spawner")

	defp exec(cmd_with_args, socket_path, env, cd) do
	opts = []
	opts = if cd, do: [{:cd, cd} \| opts], else: []
	opts = if env, do: [{:env, env} \| opts], else: opts

	opts =
	[
	:nouse_stdio,
	:exit_status,
	:binary,
	args: [socket_path \| cmd_with_args]
	] ++ opts

	Port.open({:spawn_executable, @spawner_path}, opts)
	end

	defp socket_path do
	dir = System.tmp_dir!()
	Path.join(dir, random_string())
	end

	defp random_string do
	:crypto.strong_rand_bytes(16) \|> Base.url_encode64() \|> binary_part(0, 16)
	end

	defp signal(port, sig) when sig in [:sigkill, :sigterm] do
	case Port.info(port, :os_pid) do
	- {:os_pid, os_pid} ->
	- Nif.nif_kill(os_pid, sig)
	-
	- :undefined ->
	- {:error, :process_not_alive}
	+ {:os_pid, os_pid} -> Nif.nif_kill(os_pid, sig)
	+ :undefined -> {:error, :process_not_alive}
	end
	end

	@socket_timeout 2000
	defp receive_fds(uds) do
	case :socket.accept(uds, @socket_timeout) do
	{:ok, usock} ->
	{:ok, msg} = :socket.recvmsg(usock)

	%{
	ctrl: [
	%{
	data: <<read_fd_int::native-32, write_fd_int::native-32, _rest::binary>>,
	level: :socket,
	type: :rights
	}
	]
	} = msg

	:socket.close(usock)
	{write_fd_int, read_fd_int}

	error ->
	raise error
	end
	end
	end
	diff --git a/lib/exile/watcher.ex b/lib/exile/watcher.ex
	index 7a1d548..070a7ea 100644
	--- a/lib/exile/watcher.ex
	+++ b/lib/exile/watcher.ex
	@@ -1,82 +1,77 @@
	defmodule Exile.Watcher do
	use GenServer, restart: :temporary
	require Logger
	alias Exile.ProcessNif, as: Nif

	def start_link(args) do
	{:ok, _pid} = GenServer.start_link(__MODULE__, args)
	end

	def watch(pid, os_pid, socket_path) do
	spec = {Exile.Watcher, %{pid: pid, os_pid: os_pid, socket_path: socket_path}}
	DynamicSupervisor.start_child(Exile.WatcherSupervisor, spec)
	end

	def init(args) do
	%{pid: pid, os_pid: os_pid, socket_path: socket_path} = args
	Process.flag(:trap_exit, true)
	ref = Elixir.Process.monitor(pid)
	{:ok, %{pid: pid, os_pid: os_pid, socket_path: socket_path, ref: ref}}
	end

	def handle_info({:DOWN, ref, :process, pid, _reason}, %{
	pid: pid,
	socket_path: socket_path,
	os_pid: os_pid,
	ref: ref
	}) do
	File.rm!(socket_path)
	attempt_graceful_exit(os_pid)
	{:stop, :normal, nil}
	end

	def handle_info({:EXIT, _, reason}, nil), do: {:stop, reason, nil}

	# when watcher is attempted to be killed, we forcefully kill external os process.
	# This can happen when beam receive SIGTERM
	def handle_info({:EXIT, _, reason}, %{pid: pid, socket_path: socket_path, os_pid: os_pid}) do
	Logger.debug(fn -> "Watcher exiting. reason: #{inspect(reason)}" end)
	File.rm!(socket_path)
	Exile.Process.stop(pid)
	attempt_graceful_exit(os_pid)
	{:stop, reason, nil}
	end

	- # for proper process exit parent of the child must wait() for
	- # child processes termination exit and "pickup" after the exit
	- # (receive child exit_status). Resources acquired by child such as
	- # file descriptors won't be released even if the child process
	- # itself is terminated.
	defp attempt_graceful_exit(os_pid) do
	try do
	Logger.debug(fn -> "Stopping external program" end)

	# at max we wait for 100ms for program to exit
	process_exit?(os_pid, 100) && throw(:done)

	Logger.debug("Failed to stop external program gracefully. attempting SIGTERM")
	Nif.nif_kill(os_pid, :sigterm)
	process_exit?(os_pid, 100) && throw(:done)

	Logger.debug("Failed to stop external program with SIGTERM. attempting SIGKILL")
	Nif.nif_kill(os_pid, :sigkill)
	process_exit?(os_pid, 1000) && throw(:done)

	- Logger.error("[exile] failed to kill external process")
	+ Logger.error("failed to kill external process")
	raise "Failed to kill external process"
	catch
	:done -> Logger.debug(fn -> "External program exited successfully" end)
	end
	end

	defp process_exit?(os_pid), do: !Nif.nif_is_os_pid_alive(os_pid)

	defp process_exit?(os_pid, timeout) do
	if process_exit?(os_pid) do
	true
	else
	:timer.sleep(timeout)
	process_exit?(os_pid)
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Thu, Nov 28, 8:37 AM (1 d, 20 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 41006
Default Alt Text: (28 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions