No OneTemporary
Actions

Size

30 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/c_src/exile.c b/c_src/exile.c
	index 309c3cf..68284b1 100644
	--- a/c_src/exile.c
	+++ b/c_src/exile.c
	@@ -1,283 +1,431 @@
	#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 ERL_TRUE enif_make_atom(env, "true")
	#define ERL_FALSE enif_make_atom(env, "false")
	+#define ERL_UNDEFINED enif_make_atom(env, "undefined")
	#define ERL_OK(term) enif_make_tuple2(env, enif_make_atom(env, "ok"), term)
	#define ERL_ERROR(term) \
	enif_make_tuple2(env, enif_make_atom(env, "error"), term)

	static const int PIPE_READ = 0;
	static const int PIPE_WRITE = 1;
	static const int MAX_ARGUMENTS = 20;
	static const int MAX_ARGUMENT_LEN = 1024;

	enum exec_status {
	SUCCESS,
	PIPE_CREATE_ERROR,
	PIPE_FLAG_ERROR,
	FORK_ERROR,
	PIPE_DUP_ERROR,
	NULL_DEV_OPEN_ERROR,
	};

	typedef struct ExecResults {
	enum exec_status status;
	int err;
	pid_t pid;
	int pipe_in;
	int pipe_out;
	} ExecResult;

	+struct ExilePriv {
	+ /* ERL_NIF_TERM atom_ok; */
	+ /* ERL_NIF_TERM atom_undefined; */
	+
	+ ErlNifResourceType *rt;
	+ /* void read_resource; /
	+ /* void write_resource; /
	+};
	+
	+static void rt_dtor(ErlNifEnv env, void obj) {
	+ printf("----- rt_dtor called\n");
	+}
	+
	+static void rt_stop(ErlNifEnv env, void obj, int fd, int is_direct_call) {
	+ printf("----- rt_stop called\n");
	+}
	+
	+static void rt_down(ErlNifEnv env, void obj, ErlNifPid *pid,
	+ ErlNifMonitor *monitor) {
	+ printf("----- rt_down called\n");
	+}
	+
	+static ErlNifResourceTypeInit rt_init = {rt_dtor, rt_stop, rt_down};
	+
	+typedef struct ExecContext {
	+ int cmd_input_fd;
	+ int cmd_output_fd;
	+ pid_t pid;
	+} ExecContext;
	+
	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(error) \
	do { \
	fprintf(stderr, "error in start_proccess(), %s:%d %s\n", __FILE__, \
	__LINE__, strerror(errno)); \
	result.status = error; \
	result.err = errno; \
	close_all(pipes); \
	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) {
	RETURN_ERROR(PIPE_CREATE_ERROR)
	}

	if (set_flag(pipes[STDIN_FILENO][PIPE_READ], O_CLOEXEC) < 0 \|\|
	set_flag(pipes[STDOUT_FILENO][PIPE_WRITE], O_CLOEXEC) < 0 \|\|
	set_flag(pipes[STDIN_FILENO][PIPE_WRITE], O_CLOEXEC \| O_NONBLOCK) < 0 \|\|
	set_flag(pipes[STDOUT_FILENO][PIPE_READ], O_CLOEXEC \| O_NONBLOCK) < 0) {
	RETURN_ERROR(PIPE_FLAG_ERROR)
	}

	switch (pid = fork()) {
	case -1:
	RETURN_ERROR(FORK_ERROR)

	case 0:
	close(STDIN_FILENO);
	close(STDOUT_FILENO);

	if (dup2(pipes[STDIN_FILENO][PIPE_READ], STDIN_FILENO) < 0)
	RETURN_ERROR(PIPE_DUP_ERROR)
	if (dup2(pipes[STDOUT_FILENO][PIPE_WRITE], STDOUT_FILENO) < 0)
	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) {
	close(dev_null);
	RETURN_ERROR(PIPE_DUP_ERROR)
	}

	close(dev_null);
	}

	close_all(pipes);

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

	default:
	close(pipes[STDIN_FILENO][PIPE_READ]);
	close(pipes[STDOUT_FILENO][PIPE_WRITE]);
	result.pid = pid;
	result.pipe_in = pipes[STDIN_FILENO][PIPE_WRITE];
	result.pipe_out = pipes[STDOUT_FILENO][PIPE_READ];
	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)
	+ 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);
	ERL_NIF_TERM ret;
	+ ExecContext *ctx = NULL;

	switch (result.status) {
	case SUCCESS:
	- ret = enif_make_tuple3(env, enif_make_int(env, result.pid),
	- enif_make_int(env, result.pipe_in),
	- enif_make_int(env, result.pipe_out));
	+ ctx = enif_alloc_resource(data->rt, sizeof(ExecContext));
	+ ctx->cmd_input_fd = result.pipe_in;
	+ ctx->cmd_output_fd = result.pipe_out;
	+ ctx->pid = result.pid;

	- return ERL_OK(ret);
	+ printf("cmd_in: %d cmd_out: %d pid: %d\n", result.pipe_in, result.pipe_out,
	+ result.pid);
	+
	+ // TODO: exit the command gracefully when resource is released by GC
	+ /* enif_release_resource(ctx); */
	+
	+ return ERL_OK(enif_make_resource(env, ctx));
	default:
	ret = enif_make_int(env, result.err);
	return ERL_ERROR(ret);
	}
	}

	+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[]) {
	- int pipe_in;
	- enif_get_int(env, argv[0], &pipe_in);
	-
	if (argc != 2)
	enif_make_badarg(env);

	+ struct ExilePriv *data = enif_priv_data(env);
	+ ExecContext *ctx = NULL;
	+ if (enif_get_resource(env, argv[0], data->rt, (void **)&ctx) == false) {
	+ return enif_make_badarg(env);
	+ }
	+
	ErlNifBinary bin;
	if (enif_inspect_binary(env, argv[1], &bin) != true)
	return enif_make_badarg(env);

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

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

	static ERL_NIF_TERM close_pipe(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	- int pipe;
	- enif_get_int(env, argv[0], &pipe);
	+ struct ExilePriv *data = enif_priv_data(env);
	+ ExecContext *ctx = NULL;
	+ if (enif_get_resource(env, argv[0], data->rt, (void **)&ctx) == false) {
	+ return enif_make_badarg(env);
	+ }

	- int result = close(pipe);
	+ int kind;
	+ enif_get_int(env, argv[1], &kind);
	+
	+ int result;
	+ switch (kind) {
	+ case 0:
	+ result = close(ctx->cmd_input_fd);
	+ break;
	+ case 1:
	+ result = close(ctx->cmd_output_fd);
	+ break;
	+ default:
	+ return enif_make_badarg(env);
	+ }

	if (result == 0) {
	return enif_make_atom(env, "ok");
	} else {
	perror("close()");
	return ERL_ERROR(enif_make_int(env, errno));
	}
	}

	+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[]) {
	- int pipe_out, bytes;
	- enif_get_int(env, argv[0], &pipe_out);
	- enif_get_int(env, argv[1], &bytes);
	+ if (argc != 2)
	+ enif_make_badarg(env);

	- if (bytes > 65535 \|\| bytes < 1)
	+ struct ExilePriv *data = enif_priv_data(env);
	+ ExecContext *ctx = NULL;
	+ if (enif_get_resource(env, argv[0], data->rt, (void **)&ctx) == false) {
	+ return enif_make_badarg(env);
	+ }
	+
	+ 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);
	+ }

	- char buf[bytes];
	- int result = read(pipe_out, buf, sizeof(buf));
	+ 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
	memcpy(bin.data, buf, result);
	- return ERL_OK(enif_make_binary(env, &bin));
	- } else if (errno == EAGAIN) {
	- return ERL_ERROR(enif_make_int(env, errno));
	- } else {
	+ bin_term = enif_make_binary(env, &bin);
	+ }
	+
	+ // TODO: cleanup
	+ if (result >= size \|\|
	+ (is_buffered == false && result >= 0)) { // request completely satisfied
	+ return ERL_OK(bin_term);
	+ } else if (result > 0) { // request partially satisfied
	+ int retval = select_read(env, ctx);
	+ if (retval != 0)
	+ return ERL_ERROR(enif_make_int(env, retval));
	+ return ERL_OK(bin_term);
	+ } else if (result == 0) { // EOF
	+ return ERL_OK(bin_term);
	+ } else if (errno == EAGAIN) { // busy
	+ int retval = select_read(env, ctx);
	+ if (retval != 0)
	+ return ERL_ERROR(enif_make_int(env, retval));
	+ return ERL_ERROR(enif_make_int(env, EAGAIN));
	+ } else { // Error
	perror("read()");
	return ERL_ERROR(enif_make_int(env, errno));
	}
	}

	static ERL_NIF_TERM is_alive(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	- int pid;
	- enif_get_int(env, argv[0], &pid);
	+ struct ExilePriv *data = enif_priv_data(env);
	+ ExecContext *ctx = NULL;
	+ if (enif_get_resource(env, argv[0], data->rt, (void **)&ctx) == false) {
	+ return enif_make_badarg(env);
	+ }

	- int result = kill(pid, 0);
	+ 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[]) {
	- int pid;
	- enif_get_int(env, argv[0], &pid);
	- return enif_make_int(env, kill(pid, SIGTERM));
	+ struct ExilePriv *data = enif_priv_data(env);
	+ ExecContext *ctx = NULL;
	+ if (enif_get_resource(env, argv[0], data->rt, (void **)&ctx) == false) {
	+ return enif_make_badarg(env);
	+ }
	+ return enif_make_int(env, kill(ctx->pid, SIGTERM));
	}

	static ERL_NIF_TERM kill_proc(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	- int pid;
	- enif_get_int(env, argv[0], &pid);
	- return enif_make_int(env, kill(pid, SIGKILL));
	+ struct ExilePriv *data = enif_priv_data(env);
	+ ExecContext *ctx = NULL;
	+ if (enif_get_resource(env, argv[0], data->rt, (void **)&ctx) == false) {
	+ return enif_make_badarg(env);
	+ }
	+ return enif_make_int(env, kill(ctx->pid, SIGKILL));
	}

	static ERL_NIF_TERM wait_proc(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	- int pid, status;
	- enif_get_int(env, argv[0], &pid);
	+ struct ExilePriv *data = enif_priv_data(env);
	+ ExecContext *ctx = NULL;
	+ if (enif_get_resource(env, argv[0], data->rt, (void **)&ctx) == false) {
	+ return enif_make_badarg(env);
	+ }

	- int wpid = waitpid(pid, &status, WNOHANG);
	- if (wpid != pid) {
	+ int status;
	+ int wpid = waitpid(ctx->pid, &status, WNOHANG);
	+
	+ if (wpid == ctx->pid) {
	+ return ERL_OK(enif_make_int(env, status));
	+ } else {
	perror("waitpid()");
	+ ERL_NIF_TERM term = enif_make_tuple2(env, enif_make_int(env, wpid),
	+ enif_make_int(env, status));
	+ return ERL_ERROR(term);
	}
	+}
	+
	+static int load(ErlNifEnv env, void *priv, ERL_NIF_TERM load_info) {
	+ struct ExilePriv *data = enif_alloc(sizeof(struct ExilePriv));
	+ if (!data)
	+ return 1;
	+
	+ /* data->atom_ok = enif_make_atom(env, "ok"); */
	+ /* data->atom_undefined = enif_make_atom(env, "undefined"); */
	+
	+ data->rt = enif_open_resource_type_x(env, "exile_resource", &rt_init,
	+ ERL_NIF_RT_CREATE, NULL);
	+
	+ priv = (void )data;

	- return enif_make_tuple2(env, enif_make_int(env, wpid),
	- enif_make_int(env, status));
	+ return 0;
	}

	static ErlNifFunc nif_funcs[] = {
	{"exec_proc", 2, exec_proc, 0}, {"write_proc", 2, write_proc, 0},
	- {"read_proc", 2, read_proc, 0}, {"close_pipe", 1, close_pipe, 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},
	};

	-ERL_NIF_INIT(Elixir.Exile.ProcessHelper, nif_funcs, NULL, NULL, NULL, NULL)
	+ERL_NIF_INIT(Elixir.Exile.ProcessHelper, nif_funcs, &load, NULL, NULL, NULL)
	diff --git a/lib/exile/process.ex b/lib/exile/process.ex
	index d8355ea..705f230 100644
	--- a/lib/exile/process.ex
	+++ b/lib/exile/process.ex
	@@ -1,317 +1,340 @@
	defmodule Exile.Process do
	alias Exile.ProcessHelper
	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)
	end

	- def os_pid(process) do
	- GenServer.call(process, :os_pid, :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 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,
	+ :stdin_closed,
	+ 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 = %{
	+ state = %__MODULE__{
	cmd: path,
	- args: args,
	+ cmd_args: args,
	opts: opts,
	- read_acc: [],
	errno: nil,
	status: :init,
	- await: %{}
	+ await: %{},
	+ pending_read: %Pending{},
	+ pending_write: %Pending{}
	}

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

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

	case ProcessHelper.exec_proc([state.cmd \| exec_args], stderr_to_console) do
	- {:ok, {pid, stdin, stdout}} ->
	- start_watcher(pid, stdin, stdout)
	- state = Map.merge(state, %{pid: pid, stdin: stdin, stdout: stdout, status: :start})
	- {:noreply, state}
	+ {: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
	- do_close(state, :stdin)
	- do_close(state, :stdout)
	+ # do_close(state, :stdin)
	+ # do_close(state, :stdout)

	- if ProcessHelper.is_alive(state.pid) do
	- do_kill(state.pid, :sigkill)
	+ if ProcessHelper.is_alive(state.context) do
	+ do_kill(state.context, :sigkill)
	{:stop, :process_killed, :ok, %{state \| status: {:exit, :killed}}}
	else
	{:stop, :normal, :ok, state}
	end
	end

	- def handle_call(:os_pid, _from, state), do: {:reply, state.pid, state}
	-
	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
	- Process.send_after(self(), {:await_exit_timeout, from}, timeout)
	+ 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, %{stdin: :closed} = state),
	+ def handle_call({:write, _binary}, _from, %Process{stdin_closed: true} = state),
	do: {:reply, {:error, :closed}, state}

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

	- def handle_call({:read, bytes}, from, state), do: do_read(state, bytes, from)
	+ 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({:kill, signal}, _from, state) do
	- do_kill(state.pid, signal)
	+ do_kill(state.context, signal)
	{:reply, :ok, %{state \| status: {:exit, :killed}}}
	end

	- def handle_info({:read, bytes, from}, state), do: do_read(state, bytes, from)
	-
	- def handle_info({:write, binary, from}, state), do: do_write(state, binary, from)
	-
	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(state, binary, from) do
	- case ProcessHelper.write_proc(state.stdin, binary) do
	- {:ok, bytes} ->
	- if bytes < IO.iodata_length(binary) do
	- binary = IO.iodata_to_binary(binary)
	- binary = binary_part(binary, bytes, IO.iodata_length(binary) - bytes)
	- Process.send_after(self(), {:write, binary, from}, state.opts.io_busy_wait)
	+ defp do_write(%Process{pending_write: pending} = state) do
	+ case ProcessHelper.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)
	+ {:noreply, %{state \| pending_write: %Pending{bin: binary}}}
	else
	- GenServer.reply(from, :ok)
	+ GenServer.reply(pending.client_pid, :ok)
	+ {:noreply, %{state \| pending_write: %Pending{}}}
	end

	- {:noreply, state}
	-
	- # EAGAIN
	- {:error, 35} ->
	- Process.send_after(self(), {:write, binary, from}, state.opts.io_busy_wait)
	+ {:error, eagain()} ->
	{:noreply, state}

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

	- defp do_read(state, nil, from) do
	- case ProcessHelper.read_proc(state.stdout, 65535) do
	+ defp do_read(%Process{pending_read: %Pending{remaining: nil} = pending} = state) do
	+ case ProcessHelper.read_proc(state.context, -1) do
	{:ok, <<>>} ->
	- GenServer.reply(from, {:eof, []})
	+ GenServer.reply(pending.client_pid, {:eof, []})
	{:noreply, state}

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

	- # EAGAIN
	- {:error, 35} ->
	- Process.send_after(self(), {:read, nil, from}, state.opts.io_busy_wait)
	+ {:error, eagain()} ->
	{:noreply, state}

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

	- defp do_read(state, bytes, from) do
	- case ProcessHelper.read_proc(state.stdout, bytes) do
	+ defp do_read(%Process{pending_read: pending} = state) do
	+ case ProcessHelper.read_proc(state.context, pending.remaining) do
	{:ok, <<>>} ->
	- GenServer.reply(from, {:eof, state.read_acc})
	- {:noreply, %{state \| read_acc: []}}
	+ GenServer.reply(pending.client_pid, {:eof, pending.bin})
	+ {:noreply, %Process{state \| pending_read: %Pending{}}}

	{:ok, binary} ->
	- if IO.iodata_length(binary) < bytes do
	- Process.send_after(
	- self(),
	- {:read, bytes - IO.iodata_length(binary), from},
	- state.opts.io_busy_wait
	- )
	-
	- {:noreply, %{state \| read_acc: [state.read_acc \| binary]}}
	+ if IO.iodata_length(binary) < pending.remaining do
	+ pending = %Pending{
	+ pending
	+ \| bin: [pending.bin \| binary],
	+ remaining: pending.remaining - IO.iodata_length(binary)
	+ }
	+
	+ {:noreply, %Process{state \| pending_read: pending}}
	else
	- GenServer.reply(from, {:ok, [state.read_acc \| binary]})
	- {:noreply, %{state \| read_acc: []}}
	+ GenServer.reply(pending.client_pid, {:ok, [state.pending_read.bin \| binary]})
	+ {:noreply, %Process{state \| pending_read: %Pending{}}}
	end

	- # EAGAIN
	- {:error, 35} ->
	- Process.send_after(self(), {:read, bytes, from}, state.opts.io_busy_wait)
	+ {:error, eagain()} ->
	{:noreply, state}

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

	- defp check_exit(%{pid: pid} = state, from) do
	- case ProcessHelper.wait_proc(pid) do
	- {^pid, status} ->
	+ defp check_exit(state, from) do
	+ case ProcessHelper.wait_proc(state.context) do
	+ {:ok, status} ->
	GenServer.reply(from, {:ok, status})
	cancel_timer(state, from, :timeout)
	{:noreply, clear_await(state, from)}

	- {0, _} ->
	- tref = Process.send_after(self(), {:check_exit, from}, state.opts.io_busy_wait)
	+ {: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)}

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

	- defp do_kill(pid, :sigkill), do: ProcessHelper.kill_proc(pid)
	+ defp do_kill(context, :sigkill), do: ProcessHelper.kill_proc(context)

	- defp do_kill(pid, :sigterm), do: ProcessHelper.terminate_proc(pid)
	+ defp do_kill(context, :sigterm), do: ProcessHelper.terminate_proc(context)
	+
	+ defp do_close(%Process{stdin_closed: true} = state, type) do
	+ {:reply, :ok, state}
	+ end

	defp do_close(state, type) do
	- case state[type] do
	- :closed ->
	- {:reply, :ok, %{state \| type => :closed}}
	-
	- pipe ->
	- case ProcessHelper.close_pipe(pipe) do
	- :ok -> {:reply, :ok, %{state \| type => :closed}}
	- {:error, errno} -> {:reply, {:error, errno}, %{state \| errno: errno}}
	- end
	+ case ProcessHelper.close_pipe(state.context, stream_type(type)) do
	+ :ok ->
	+ {:reply, :ok, %Process{state \| stdin_closed: true}}
	+
	+ {:error, errno} ->
	+ raise errno
	+ {:reply, {:error, errno}, %Process{state \| errno: errno}}
	end
	end

	defp clear_await(state, from) do
	- %{state \| await: Map.delete(state.await, from)}
	+ %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 -> Process.cancel_timer(tref)
	+ tref -> Elixir.Process.cancel_timer(tref)
	end
	end

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

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

	@stdin_close_wait 3000
	@sigterm_wait 1000

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

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

	- defp watcher(process_server, pid, stdin, stdout) do
	- ref = Process.monitor(process_server)
	+ defp stream_type(:stdin), do: 0
	+ defp stream_type(:stdout), do: 1
	+
	+ defp watcher(process_server, context) do
	+ ref = Elixir.Process.monitor(process_server)
	send(process_server, {self(), :done})

	receive do
	{:DOWN, ^ref, :process, ^process_server, :normal} ->
	:ok

	{:DOWN, ^ref, :process, ^process_server, _reason} ->
	- case ProcessHelper.wait_proc(pid) do
	- {^pid, _status} ->
	+ case ProcessHelper.wait_proc(context) do
	+ {:ok, _status} ->
	# TODO: check stauts
	nil

	- _ ->
	- Logger.debug(fn -> "Killing #{pid}" end)
	+ {:error, {_, _}} ->
	+ Logger.debug(fn -> "Killing" end)

	- with _ <- ProcessHelper.close_pipe(stdin),
	- _ <- ProcessHelper.close_pipe(stdout),
	+ with _ <- ProcessHelper.close_pipe(context, stream_type(:stdin)),
	+ _ <- ProcessHelper.close_pipe(context, stream_type(:stdout)),
	_ <- :timer.sleep(@stdin_close_wait),
	- {p, _} <- ProcessHelper.wait_proc(pid),
	- false <- p != pid,
	- _ <- ProcessHelper.terminate_proc(pid),
	+ {:error, _} <- ProcessHelper.wait_proc(context),
	+ _ <- ProcessHelper.terminate_proc(context),
	_ <- :timer.sleep(@sigterm_wait),
	- {p, _} <- ProcessHelper.wait_proc(pid),
	- false <- p != pid,
	- _ <- ProcessHelper.kill_proc(pid) do
	- Logger.debug(fn -> "Killed process: #{pid}" end)
	+ {:error, _} <- ProcessHelper.wait_proc(context),
	+ _ <- ProcessHelper.kill_proc(context) do
	+ Logger.debug(fn -> "Killed process" end)
	end
	end
	end
	end
	end
	diff --git a/lib/exile/process_helper.ex b/lib/exile/process_helper.ex
	index cc5fec2..f8353de 100644
	--- a/lib/exile/process_helper.ex
	+++ b/lib/exile/process_helper.ex
	@@ -1,24 +1,24 @@
	defmodule Exile.ProcessHelper 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(_pipe, _bin), do: :erlang.nif_error(:nif_library_not_loaded)
	+ def write_proc(_context, _bin), do: :erlang.nif_error(:nif_library_not_loaded)

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

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

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

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

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

	- def is_alive(_pid), do: :erlang.nif_error(:nif_library_not_loaded)
	+ def is_alive(_context), do: :erlang.nif_error(:nif_library_not_loaded)
	end

File Metadata

Mime Type: text/x-diff
Expires: Sat, Aug 8, 9:52 PM (1 d, 18 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1722712
Default Alt Text: (30 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions