No OneTemporary
Actions

Size

33 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/c_src/exile.c b/c_src/exile.c
	index 47d5337..6a87f37 100644
	--- a/c_src/exile.c
	+++ b/c_src/exile.c
	@@ -1,509 +1,571 @@
	#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 <unistd.h>

	//#define DEBUG

	#ifdef DEBUG
	#define debug(...) \
	do { \
	enif_fprintf(stderr, __VA_ARGS__); \
	enif_fprintf(stderr, "\n"); \
	} while (0)
	#define start_timing() ErlNifTime __start = enif_monotonic_time(ERL_NIF_USEC)
	#define elapsed_microseconds() (enif_monotonic_time(ERL_NIF_USEC) - __start)
	#else
	#define debug(...)
	#define start_timing()
	#define elapsed_microseconds() 0
	#endif

	#define error(...) \
	do { \
	enif_fprintf(stderr, __VA_ARGS__); \
	enif_fprintf(stderr, "\n"); \
	} while (0)

	#define ERL_TRUE enif_make_atom(env, "true")
	#define ERL_FALSE enif_make_atom(env, "false")
	#define ERL_UNDEFINED enif_make_atom(env, "undefined")

	#define MAKE_OK(term) enif_make_tuple2(env, ATOM_OK, term)
	#define MAKE_ERROR(term) enif_make_tuple2(env, ATOM_ERROR, term)

	#define GET_CTX(env, arg, ctx) \
	do { \
	ExilePriv *data = enif_priv_data(env); \
	if (enif_get_resource(env, arg, data->rt, (void **)&ctx) == false) { \
	return MAKE_ERROR(ATOM_INVALID_CTX); \
	} \
	} while (0);

	-static const int PIPE_READ = 0;
	-static const int PIPE_WRITE = 1;
	+static const int PIPE_READ = 0;
	+static const int PIPE_WRITE = 1;
	static const int PIPE_CLOSED = -1;
	-static const int CMD_EXIT = -1;
	-static const int MAX_ARGUMENTS = 20;
	+static const int CMD_EXIT = -1;
	+static const int MAX_ARGUMENTS = 20;
	static const int MAX_ARGUMENT_LEN = 1024;

	static ERL_NIF_TERM ATOM_OK;
	static ERL_NIF_TERM ATOM_ERROR;
	static ERL_NIF_TERM ATOM_UNDEFINED;
	static ERL_NIF_TERM ATOM_INVALID_CTX;
	static ERL_NIF_TERM ATOM_PIPE_CLOSED;

	+// command exit types
	+static ERL_NIF_TERM ATOM_EXIT;
	+static ERL_NIF_TERM ATOM_SIGNALED;
	+static ERL_NIF_TERM ATOM_STOPPED;
	+
	enum exec_status {
	SUCCESS,
	PIPE_CREATE_ERROR,
	PIPE_FLAG_ERROR,
	FORK_ERROR,
	PIPE_DUP_ERROR,
	NULL_DEV_OPEN_ERROR,
	};

	+enum exit_type { NORMAL_EXIT, SIGNALED, STOPPED };
	+
	typedef struct ExilePriv {
	ErlNifResourceType *rt;
	} ExilePriv;

	typedef struct ExecContext {
	int cmd_input_fd;
	int cmd_output_fd;
	- int cmd_exit_status;
	+ int exit_status; // can be exit status or signal number depending on exit_type
	+ enum exit_type exit_type;
	pid_t pid;
	} ExecContext;

	typedef struct ExecResult {
	enum exec_status status;
	int err;
	ExecContext context;
	} ExecResult;

	+// TODO: should we assert if external process exit here?
	static void rt_dtor(ErlNifEnv env, void obj) {
	debug("Exile rt_dtor called\n");
	}

	static void rt_stop(ErlNifEnv env, void obj, int fd, int is_direct_call) {
	debug("Exile rt_stop called\n");
	}

	static void rt_down(ErlNifEnv env, void obj, ErlNifPid *pid,
	ErlNifMonitor *monitor) {
	debug("Exile rt_down called\n");
	}

	static ErlNifResourceTypeInit rt_init = {rt_dtor, rt_stop, rt_down};

	static int set_flag(int fd, int flags) {
	return fcntl(fd, F_SETFL, fcntl(fd, F_GETFL) \| flags);
	}

	static void close_all(int pipes[2][2]) {
	for (int i = 0; i < 2; i++) {
	if (pipes[i][PIPE_READ] > 0)
	close(pipes[i][PIPE_READ]);
	if (pipes[i][PIPE_WRITE] > 0)
	close(pipes[i][PIPE_WRITE]);
	}
	}

	#define RETURN_ERROR(__err) \
	do { \
	result.err = errno; \
	result.status = __err; \
	close_all(pipes); \
	error("error in start_proccess(), %s:%d %s", __FILE__, __LINE__, \
	strerror(errno)); \
	return result; \
	} while (0);

	static ExecResult start_proccess(char *args[], bool stderr_to_console) {
	ExecResult result;
	pid_t pid;
	int pipes[2][2] = {{0, 0}, {0, 0}};

	if (pipe(pipes[STDIN_FILENO]) == -1 \|\| pipe(pipes[STDOUT_FILENO]) == -1) {
	debug("failed create pipes");
	RETURN_ERROR(PIPE_CREATE_ERROR)
	}

	const int r_cmdin = pipes[STDIN_FILENO][PIPE_READ];
	const int w_cmdin = pipes[STDIN_FILENO][PIPE_WRITE];

	const int r_cmdout = pipes[STDOUT_FILENO][PIPE_READ];
	const int w_cmdout = pipes[STDOUT_FILENO][PIPE_WRITE];

	if (set_flag(r_cmdin, O_CLOEXEC) < 0 \|\| set_flag(w_cmdout, O_CLOEXEC) < 0 \|\|
	set_flag(w_cmdin, O_CLOEXEC \| O_NONBLOCK) < 0 \|\|
	set_flag(r_cmdout, O_CLOEXEC \| O_NONBLOCK) < 0) {
	debug("failed to set flag for pipes");
	RETURN_ERROR(PIPE_FLAG_ERROR)
	}

	+ // TODO: fork() can be expensive. especially in mac os. we should report
	+ // correct reduction cost for this at avoid potential scheduler collapse
	switch (pid = fork()) {

	case -1:
	debug("failed to fork");
	RETURN_ERROR(FORK_ERROR)

	case 0: // child

	- // close default stdio fd
	+ // TODO: check if we are leaking any resources such as opened fd to child
	+ // program close default stdio fd
	+
	close(STDIN_FILENO);
	close(STDOUT_FILENO);

	if (dup2(r_cmdin, STDIN_FILENO) < 0) {
	debug("failed dup command input pipe to stdin");
	RETURN_ERROR(PIPE_DUP_ERROR)
	}
	if (dup2(w_cmdout, STDOUT_FILENO) < 0) {
	debug("failed dup command output pipe to stdout");
	RETURN_ERROR(PIPE_DUP_ERROR)
	}

	if (stderr_to_console != true) {
	close(STDERR_FILENO);
	int dev_null = open("/dev/null", O_WRONLY);

	if (dev_null == -1) {
	RETURN_ERROR(NULL_DEV_OPEN_ERROR);
	}

	if (dup2(dev_null, STDERR_FILENO) < 0) {
	debug("failed dup command error pipe to stderr");
	close(dev_null);
	RETURN_ERROR(PIPE_DUP_ERROR)
	}

	close(dev_null);
	}

	close_all(pipes);

	execvp(args[0], args);
	perror("execvp(): failed");

	default: // parent
	// close file descriptors used by child
	close(r_cmdin);
	close(w_cmdout);

	result.context.pid = pid;
	result.context.cmd_input_fd = w_cmdin;
	result.context.cmd_output_fd = r_cmdout;
	result.status = SUCCESS;

	return result;
	}
	}

	/* TODO: return appropriate error instead returning generic "badarg" error */
	static ERL_NIF_TERM exec_proc(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	char tmp[MAX_ARGUMENTS][MAX_ARGUMENT_LEN + 1];
	char *exec_args[MAX_ARGUMENTS + 1];

	unsigned int args_len;
	if (enif_get_list_length(env, argv[0], &args_len) != true)
	return enif_make_badarg(env);

	if (args_len > MAX_ARGUMENTS)
	return enif_make_badarg(env);

	ERL_NIF_TERM head, tail, list = argv[0];
	for (unsigned int i = 0; i < args_len; i++) {
	if (enif_get_list_cell(env, list, &head, &tail) != true)
	return enif_make_badarg(env);

	if (enif_get_string(env, head, tmp[i], MAX_ARGUMENT_LEN, ERL_NIF_LATIN1) <
	1)
	return enif_make_badarg(env);
	exec_args[i] = tmp[i];
	list = tail;
	}
	exec_args[args_len] = NULL;

	bool stderr_to_console = true;
	int tmp_int;
	if (enif_get_int(env, argv[1], &tmp_int) != true)
	return enif_make_badarg(env);
	stderr_to_console = tmp_int == 1 ? true : false;

	struct ExilePriv *data = enif_priv_data(env);
	ExecResult result = start_proccess(exec_args, stderr_to_console);
	ExecContext *ctx = NULL;
	ERL_NIF_TERM term;

	switch (result.status) {
	case SUCCESS:
	ctx = enif_alloc_resource(data->rt, sizeof(ExecContext));
	ctx->cmd_input_fd = result.context.cmd_input_fd;
	ctx->cmd_output_fd = result.context.cmd_output_fd;
	ctx->pid = result.context.pid;

	debug("pid: %d cmd_in_fd: %d cmd_out_fd: %d", ctx->pid, ctx->cmd_input_fd,
	ctx->cmd_output_fd);

	term = enif_make_resource(env, ctx);

	// resource should be collected beam GC when there are no more references
	enif_release_resource(ctx);

	return MAKE_OK(term);
	default:
	return MAKE_ERROR(enif_make_int(env, result.err));
	}
	}

	static int select_write(ErlNifEnv env, ExecContext ctx) {
	int retval = enif_select(env, ctx->cmd_input_fd, ERL_NIF_SELECT_WRITE, ctx,
	NULL, ERL_UNDEFINED);
	if (retval != 0)
	perror("select_write()");

	return retval;
	}

	static ERL_NIF_TERM write_proc(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	if (argc != 2)
	enif_make_badarg(env);

	ExecContext *ctx = NULL;
	GET_CTX(env, argv[0], ctx);

	if (ctx->cmd_input_fd == PIPE_CLOSED)
	return MAKE_ERROR(ATOM_PIPE_CLOSED);

	ErlNifBinary bin;
	+ // TODO: should not use enif_inspect_binary
	if (enif_inspect_binary(env, argv[1], &bin) != true)
	return enif_make_badarg(env);

	unsigned int result = write(ctx->cmd_input_fd, bin.data, bin.size);

	- // TODO: cleanup
	+ // TODO: branching is quite ugly, cleanup required
	if (result >= bin.size) { // request completely satisfied
	return MAKE_OK(enif_make_int(env, result));
	} else if (result >= 0) { // request partially satisfied
	int retval = select_write(env, ctx);
	if (retval != 0)
	return MAKE_ERROR(enif_make_int(env, retval));
	return MAKE_OK(enif_make_int(env, result));
	} else if (errno == EAGAIN) { // busy
	int retval = select_write(env, ctx);
	if (retval != 0)
	return MAKE_ERROR(enif_make_int(env, retval));
	return MAKE_ERROR(enif_make_int(env, EAGAIN));
	} else { // Error
	perror("write()");
	return MAKE_ERROR(enif_make_int(env, errno));
	}
	}

	static ERL_NIF_TERM close_pipe(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	ExecContext *ctx = NULL;
	GET_CTX(env, argv[0], ctx);

	int kind;
	enif_get_int(env, argv[1], &kind);

	int result;
	switch (kind) {
	case 0:
	if (ctx->cmd_input_fd == PIPE_CLOSED) {
	return ATOM_OK;
	} else {
	result = close(ctx->cmd_input_fd);
	if (result == 0) {
	ctx->cmd_input_fd = PIPE_CLOSED;
	return ATOM_OK;
	} else {
	perror("cmd_input_fd close()");
	return MAKE_ERROR(enif_make_int(env, errno));
	}
	}
	case 1:
	if (ctx->cmd_output_fd == PIPE_CLOSED) {
	return ATOM_OK;
	} else {
	result = close(ctx->cmd_output_fd);
	if (result == 0) {
	ctx->cmd_output_fd = PIPE_CLOSED;
	return ATOM_OK;
	} else {
	perror("cmd_output_fd close()");
	return MAKE_ERROR(enif_make_int(env, errno));
	}
	}
	default:
	debug("invalid file descriptor type");
	return enif_make_badarg(env);
	}
	}

	static int select_read(ErlNifEnv env, ExecContext ctx) {
	int retval = enif_select(env, ctx->cmd_output_fd, ERL_NIF_SELECT_READ, ctx,
	NULL, ERL_UNDEFINED);
	if (retval != 0)
	perror("select_read()");
	return retval;
	}

	static ERL_NIF_TERM read_proc(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	if (argc != 2)
	enif_make_badarg(env);

	ExecContext *ctx = NULL;
	GET_CTX(env, argv[0], ctx);

	if (ctx->cmd_output_fd == PIPE_CLOSED)
	return MAKE_ERROR(ATOM_PIPE_CLOSED);

	bool is_buffered = true;
	int size;
	enif_get_int(env, argv[1], &size);

	if (size == -1) {
	size = 65535;
	is_buffered = false;
	} else if (size > 65535 \|\| size < 1) {
	enif_make_badarg(env);
	}

	unsigned char buf[size];
	int result = read(ctx->cmd_output_fd, buf, sizeof(buf));

	ERL_NIF_TERM bin_term;
	if (result >= 0) {
	ErlNifBinary bin;
	enif_alloc_binary(result, &bin);
	- // TODO: we should use binary when reading itself instead of allocating
	- // again
	+ // TODO: we should use the erl binary for `read` itself instead of
	+ // allocating again
	memcpy(bin.data, buf, result);
	bin_term = enif_make_binary(env, &bin);
	}

	- // TODO: cleanup
	+ // TODO: branching is quite ugly, cleanup required
	if (result >= size \|\|
	(is_buffered == false && result >= 0)) { // request completely satisfied
	return MAKE_OK(bin_term);
	} else if (result > 0) { // request partially satisfied
	int retval = select_read(env, ctx);
	if (retval != 0)
	return MAKE_ERROR(enif_make_int(env, retval));
	return MAKE_OK(bin_term);
	} else if (result == 0) { // EOF
	return MAKE_OK(bin_term);
	} else if (errno == EAGAIN) { // busy
	int retval = select_read(env, ctx);
	if (retval != 0)
	return MAKE_ERROR(enif_make_int(env, retval));
	return MAKE_ERROR(enif_make_int(env, EAGAIN));
	} else { // Error
	perror("read()");
	return MAKE_ERROR(enif_make_int(env, errno));
	}
	}

	static ERL_NIF_TERM is_alive(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	ExecContext *ctx = NULL;
	GET_CTX(env, argv[0], ctx);

	if (ctx->pid == CMD_EXIT)
	return ERL_FALSE;

	int result = kill(ctx->pid, 0);

	if (result == 0) {
	return ERL_TRUE;
	} else {
	return ERL_FALSE;
	}
	}

	static ERL_NIF_TERM terminate_proc(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	ExecContext *ctx = NULL;
	GET_CTX(env, argv[0], ctx);
	if (ctx->pid == CMD_EXIT)
	return MAKE_OK(enif_make_int(env, 0));

	- return enif_make_int(env, kill(ctx->pid, SIGTERM));
	+ return MAKE_OK(enif_make_int(env, kill(ctx->pid, SIGTERM)));
	}

	static ERL_NIF_TERM kill_proc(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	ExecContext *ctx = NULL;
	GET_CTX(env, argv[0], ctx);
	if (ctx->pid == CMD_EXIT)
	return MAKE_OK(enif_make_int(env, 0));

	- return enif_make_int(env, kill(ctx->pid, SIGKILL));
	+ return MAKE_OK(enif_make_int(env, kill(ctx->pid, SIGKILL)));
	+}
	+
	+static ERL_NIF_TERM make_exit_term(ErlNifEnv env, ExecContext ctx) {
	+ switch (ctx->exit_type) {
	+ case NORMAL_EXIT:
	+ return MAKE_OK(
	+ 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(enif_make_tuple2(env, ATOM_SIGNALED,
	+ enif_make_int(env, ctx->exit_status)));
	+ case STOPPED:
	+ return MAKE_OK(enif_make_tuple2(env, ATOM_STOPPED,
	+ enif_make_int(env, ctx->exit_status)));
	+ default:
	+ error("Invalid wait status");
	+ return MAKE_ERROR(ATOM_UNDEFINED);
	+ }
	}

	static ERL_NIF_TERM wait_proc(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	ExecContext *ctx = NULL;
	GET_CTX(env, argv[0], ctx);

	if (ctx->pid == CMD_EXIT)
	- return MAKE_OK(enif_make_int(env, ctx->cmd_exit_status));
	+ return make_exit_term(env, ctx);

	int status;
	int wpid = waitpid(ctx->pid, &status, WNOHANG);

	if (wpid == ctx->pid) {
	ctx->pid = CMD_EXIT;
	- ctx->cmd_exit_status = status;
	- return MAKE_OK(enif_make_int(env, status));
	+
	+ if (WIFEXITED(status)) {
	+ ctx->exit_type = NORMAL_EXIT;
	+ ctx->exit_status = WEXITSTATUS(status);
	+ } else if (WIFSIGNALED(status)) {
	+ ctx->exit_type = SIGNALED;
	+ ctx->exit_status = WTERMSIG(status);
	+ } else if (WIFSTOPPED(status)) {
	+ ctx->exit_type = STOPPED;
	+ ctx->exit_status = 0;
	+ }
	+
	+ return make_exit_term(env, ctx);
	} else if (wpid != 0) {
	perror("waitpid()");
	}
	ERL_NIF_TERM term = enif_make_tuple2(env, enif_make_int(env, wpid),
	enif_make_int(env, status));
	return MAKE_ERROR(term);
	+}
	+
	+static ERL_NIF_TERM os_pid(ErlNifEnv *env, int argc,
	+ const ERL_NIF_TERM argv[]) {
	+ ExecContext *ctx = NULL;
	+ GET_CTX(env, argv[0], ctx);
	+ if (ctx->pid == CMD_EXIT)
	+ return MAKE_OK(enif_make_int(env, 0));

	+ return MAKE_OK(enif_make_int(env, ctx->pid));
	}

	static int on_load(ErlNifEnv env, void *priv, ERL_NIF_TERM load_info) {
	struct ExilePriv *data = enif_alloc(sizeof(struct ExilePriv));
	if (!data)
	return 1;

	data->rt =
	enif_open_resource_type_x(env, "exile_resource", &rt_init,
	ERL_NIF_RT_CREATE \| ERL_NIF_RT_TAKEOVER, NULL);

	ATOM_OK = enif_make_atom(env, "ok");
	ATOM_ERROR = enif_make_atom(env, "error");
	ATOM_UNDEFINED = enif_make_atom(env, "undefined");
	ATOM_INVALID_CTX = enif_make_atom(env, "invalid_exile_exec_ctx");
	- ATOM_INVALID_CTX = enif_make_atom(env, "closed_pipe");
	+ 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");

	priv = (void )data;

	return 0;
	}

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

	+// maybe we can use dirty schedulers conditionally by checking if they are
	+// available or not at compile time
	static ErlNifFunc nif_funcs[] = {
	- {"exec_proc", 2, exec_proc, 0}, {"write_proc", 2, write_proc, 0},
	- {"read_proc", 2, read_proc, 0}, {"close_pipe", 2, close_pipe, 0},
	- {"terminate_proc", 1, terminate_proc, 0}, {"wait_proc", 1, wait_proc, 0},
	- {"kill_proc", 1, kill_proc, 0}, {"is_alive", 1, is_alive, 0},
	+ {"exec_proc", 2, exec_proc, 0},
	+ {"write_proc", 2, write_proc, 0},
	+ {"read_proc", 2, read_proc, 0},
	+ {"close_pipe", 2, close_pipe, 0},
	+ {"terminate_proc", 1, terminate_proc, 0},
	+ {"wait_proc", 1, wait_proc, 0},
	+ {"kill_proc", 1, kill_proc, 0},
	+ {"is_alive", 1, is_alive, 0},
	+ {"os_pid", 1, os_pid, 0},
	};

	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 268a22c..1953f2c 100644
	--- a/lib/exile/process.ex
	+++ b/lib/exile/process.ex
	@@ -1,339 +1,353 @@
	defmodule Exile.Process do
	alias Exile.ProcessNif
	require Logger
	use GenServer

	defmacro eagain(), do: 35

	# delay between retries when io is busy (in milliseconds)
	@default_opts %{io_busy_wait: 1, stderr_to_console: false}

	def start_link(cmd, args, opts \\ %{}) do
	opts = Map.merge(@default_opts, opts)
	GenServer.start(__MODULE__, %{cmd: cmd, args: args, opts: opts})
	end

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

	def write(process, binary) do
	GenServer.call(process, {:write, binary}, :infinity)
	end

	- def read(process, bytes) do
	- GenServer.call(process, {:read, bytes}, :infinity)
	+ def read(process, size) when is_integer(size) do
	+ GenServer.call(process, {:read, size}, :infinity)
	+ end
	+
	+ def read(process) do
	+ GenServer.call(process, {:read, nil}, :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, timeout \\ :infinity) 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 [
	:cmd,
	:cmd_args,
	:opts,
	:errno,
	:context,
	:status,
	await: %{},
	pending_read: nil,
	pending_write: nil
	]

	alias __MODULE__

	def init(%{cmd: cmd, args: args, opts: opts}) do
	path = :os.find_executable(to_charlist(cmd))

	unless path do
	raise "Command not found: #{cmd}"
	end

	state = %__MODULE__{
	cmd: path,
	cmd_args: args,
	opts: opts,
	errno: nil,
	status: :init,
	await: %{},
	pending_read: %Pending{},
	pending_write: %Pending{}
	}

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

	def handle_continue(nil, state) do
	exec_args = Enum.map(state.cmd_args, &to_charlist/1)
	stderr_to_console = if state.opts.stderr_to_console, do: 1, else: 0

	case ProcessNif.exec_proc([state.cmd \| exec_args], stderr_to_console) do
	{:ok, context} ->
	start_watcher(context)
	{:noreply, %Process{state \| context: context, status: :start}}

	{:error, errno} ->
	raise "Failed to start command: #{state.cmd}, errno: #{errno}"
	end
	end

	def handle_call(:stop, _from, state) do
	- # watcher will terminate the external process
	+ # watcher will take care of termination of external process
	+
	+ # TODO: pending write and read should receive "stopped" return
	+ # value instead of exit signal
	{:stop, :normal, :ok, state}
	end

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

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

	state = put_timer(state, from, :timeout, tref)
	check_exit(state, from)
	end

	- def handle_call({:write, binary}, from, state) do
	+ 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, bytes}, from, state) do
	pending = %Pending{remaining: bytes, client_pid: from}
	do_read(%Process{state \| pending_read: pending})
	end

	def handle_call(:close_stdin, _from, state), do: do_close(state, :stdin)

	+ def handle_call(:os_pid, _from, state), do: {:reply, ProcessNif.os_pid(state.context), state}
	+
	def handle_call({:kill, signal}, _from, state) do
	do_kill(state.context, signal)
	{:reply, :ok, %{state \| status: {:exit, :killed}}}
	end

	def handle_info({:check_exit, from}, state), do: check_exit(state, from)

	def handle_info({:await_exit_timeout, from}, state) do
	cancel_timer(state, from, :check)

	receive do
	{:check_exit, ^from} -> :ok
	after
	0 -> :ok
	end

	GenServer.reply(from, :timeout)
	{:noreply, clear_await(state, from)}
	end

	def handle_info({:select, context, _ref, :ready_output}, state) do
	do_write(%Process{state \| context: context})
	end

	def handle_info({:select, context, _ref, :ready_input}, state) do
	do_read(%Process{state \| context: context})
	end

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

	defp do_write(%Process{pending_write: pending} = state) do
	case ProcessNif.write_proc(state.context, pending.bin) do
	{:ok, size} ->
	- if size < IO.iodata_length(pending.bin) do
	- binary = IO.iodata_to_binary(pending.bin)
	- binary = binary_part(binary, size, IO.iodata_length(pending.bin) - size)
	+ if size < byte_size(pending.bin) do
	+ binary = binary_part(pending.bin, size, byte_size(pending.bin) - size)
	{:noreply, %{state \| pending_write: %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: nil} = pending} = state) do
	case ProcessNif.read_proc(state.context, -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 ProcessNif.read_proc(state.context, pending.remaining) do
	{:ok, <<>>} ->
	GenServer.reply(pending.client_pid, {:eof, pending.bin})
	{:noreply, %Process{state \| pending_read: %Pending{}}}

	{:ok, binary} ->
	- if IO.iodata_length(binary) < pending.remaining do
	+ if byte_size(binary) < pending.remaining do
	pending = %Pending{
	pending
	\| bin: [pending.bin \| binary],
	- remaining: pending.remaining - IO.iodata_length(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 check_exit(state, from) do
	case ProcessNif.wait_proc(state.context) do
	{:ok, status} ->
	GenServer.reply(from, {:ok, status})
	cancel_timer(state, from, :timeout)
	{:noreply, clear_await(state, from)}

	{:error, {0, _}} ->
	# Ideally we should not poll and we should handle this with SIGCHLD signal
	tref = Elixir.Process.send_after(self(), {:check_exit, from}, state.opts.io_busy_wait)
	{:noreply, put_timer(state, from, :check, tref)}

	{:error, {-1, status}} ->
	GenServer.reply(from, {:error, status})
	cancel_timer(state, from, :timeout)
	{:noreply, clear_await(state, from)}
	end
	end

	defp do_kill(context, :sigkill), do: ProcessNif.kill_proc(context)

	defp do_kill(context, :sigterm), do: ProcessNif.terminate_proc(context)

	defp do_close(state, type) do
	case ProcessNif.close_pipe(state.context, stream_type(type)) do
	:ok ->
	{:reply, :ok, state}

	{:error, errno} ->
	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])

	# Try to gracefully terminate external proccess if the genserver associated with the process is killed
	defp start_watcher(context) do
	process_server = self()
	watcher_pid = spawn(fn -> watcher(process_server, context) end)

	receive do
	{^watcher_pid, :done} -> :ok
	end
	end

	defp stream_type(:stdin), do: 0
	defp stream_type(:stdout), do: 1

	defp process_exit?(context) do
	match?({:ok, _}, ProcessNif.wait_proc(context))
	end

	defp process_exit?(context, timeout) do
	if process_exit?(context) do
	true
	else
	:timer.sleep(timeout)
	process_exit?(context)
	end
	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 watcher(process_server, context) do
	ref = Elixir.Process.monitor(process_server)
	send(process_server, {self(), :done})

	receive do
	{:DOWN, ^ref, :process, ^process_server, _reason} ->
	try do
	process_exit?(context) && throw(:done)

	- Logger.debug(fn -> "Killing" end)
	+ Logger.debug(fn -> "Stopping external program" end)
	ProcessNif.close_pipe(context, stream_type(:stdin))
	ProcessNif.close_pipe(context, stream_type(:stdout))

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

	+ Logger.debug("Failed to stop external program gracefully. attempting SIGTERM")
	ProcessNif.terminate_proc(context)
	process_exit?(context, 100) && throw(:done)

	+ Logger.debug("Failed to stop external program with SIGTERM. attempting SIGKILL")
	ProcessNif.kill_proc(context)
	process_exit?(context, 1000) && throw(:done)

	Logger.error("[exile] failed to kill external process")
	raise "Failed to kill external process"
	catch
	:done -> Logger.debug(fn -> "Exited external program successfully" end)
	end
	end
	end
	end
	diff --git a/lib/exile/process_nif.ex b/lib/exile/process_nif.ex
	index af6a1de..98c7166 100644
	--- a/lib/exile/process_nif.ex
	+++ b/lib/exile/process_nif.ex
	@@ -1,24 +1,26 @@
	defmodule Exile.ProcessNif do
	@on_load :load_nifs

	def load_nifs do
	nif_path = :filename.join(:code.priv_dir(:exile), "exile")
	:erlang.load_nif(nif_path, 0)
	end

	def exec_proc(_cmd, _stderr_to_console), do: :erlang.nif_error(:nif_library_not_loaded)

	def write_proc(_context, _bin), do: :erlang.nif_error(:nif_library_not_loaded)

	def read_proc(_context, _bytes), do: :erlang.nif_error(:nif_library_not_loaded)

	def close_pipe(_context, _pipe), do: :erlang.nif_error(:nif_library_not_loaded)

	def kill_proc(_context), do: :erlang.nif_error(:nif_library_not_loaded)

	def terminate_proc(_context), do: :erlang.nif_error(:nif_library_not_loaded)

	def wait_proc(_context), do: :erlang.nif_error(:nif_library_not_loaded)

	+ def os_pid(_context), do: :erlang.nif_error(:nif_library_not_loaded)
	+
	def is_alive(_context), do: :erlang.nif_error(:nif_library_not_loaded)
	end
	diff --git a/lib/exile/stream.ex b/lib/exile/stream.ex
	index ad31d65..15b35bc 100644
	--- a/lib/exile/stream.ex
	+++ b/lib/exile/stream.ex
	@@ -1,102 +1,102 @@
	defmodule Exile.Stream do
	@moduledoc """
	Defines a `Exile.Stream` struct returned by `Exile.stream!/3`.
	"""

	alias Exile.Process

	defstruct [:proc_server, :stream_opts]

	@default_opts %{exit_timeout: :infinity, chunk_size: 65535}

	@type t :: %__MODULE__{}

	@doc false
	def __build__(cmd, args, opts) do
	opts = Map.merge(@default_opts, opts)
	{stream_opts, proc_opts} = Map.split(opts, [:exit_timeout, :chunk_size])
	{:ok, proc} = Process.start_link(cmd, args, proc_opts)
	%Exile.Stream{proc_server: proc, stream_opts: stream_opts}
	end

	defimpl Collectable do
	def into(%{proc_server: proc} = stream) do
	collector_fun = fn
	:ok, {:cont, x} ->
	:ok = Process.write(proc, x)

	:ok, :done ->
	:ok = Process.close_stdin(proc)
	stream

	:ok, :halt ->
	:ok = Process.close_stdin(proc)
	end

	{:ok, collector_fun}
	end
	end

	defimpl Enumerable do
	def reduce(%{proc_server: proc, stream_opts: stream_opts}, acc, fun) do
	start_fun = fn -> :ok end

	next_fun = fn :ok ->
	case Process.read(proc, stream_opts.chunk_size) do
	{:eof, []} ->
	{:halt, :normal}

	{:eof, x} ->
	# multiple reads on closed pipe always returns :eof
	{[x], :ok}

	{:ok, x} ->
	{[x], :ok}

	{:error, errno} ->
	raise "Failed to read from the process. errno: #{errno}"
	end
	end

	after_fun = fn exit_type ->
	try do
	# always close stdin before stoping to give the command chance to exit properly
	Process.close_stdin(proc)
	result = Process.await_exit(proc, stream_opts.exit_timeout)

	case {exit_type, result} do
	{_, :timeout} ->
	Process.kill(proc, :sigkill)
	raise "command fail to exit within timeout: #{stream_opts.exit_timeout}"

	- {:normal, {:ok, 0}} ->
	+ {:normal, {:ok, {:exit, 0}}} ->
	:ok

	{:normal, {:ok, exit_status}} ->
	raise "command exited with status: #{exit_status}"

	{_, error} ->
	Process.kill(proc, :sigkill)
	raise "command exited with error: #{inspect(error)}"
	end
	after
	Process.stop(proc)
	end
	end

	Stream.resource(start_fun, next_fun, after_fun).(acc, fun)
	end

	def count(_stream) do
	{:error, __MODULE__}
	end

	def member?(_stream, _term) do
	{:error, __MODULE__}
	end

	def slice(_stream) do
	{:error, __MODULE__}
	end
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sat, Aug 29, 2:18 PM (1 d, 15 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1737126
Default Alt Text: (33 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions