No OneTemporary
Actions

Size

16 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/exile/process.ex b/lib/exile/process.ex
	index 44deaef..25d771d 100644
	--- a/lib/exile/process.ex
	+++ b/lib/exile/process.ex
	@@ -1,201 +1,202 @@
	defmodule Exile.Process do
	alias Exile.ProcessHelper
	require Logger
	use GenServer

	# delay between retries when io is busy (in milliseconds)
	- @default_opts %{io_busy_wait: 1}
	+ @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) do
	GenServer.call(process, :await_exit, :infinity)
	end

	def stop(process) do
	GenServer.stop(process, :normal, :infinity)
	end

	## Server

	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

	{:ok, %{cmd: path, args: args, opts: opts, read_acc: [], errno: nil, status: :init},
	{:continue, nil}}
	end

	def handle_continue(nil, state) do
	exec_args = Enum.map(state.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]) do
	+ 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}

	{:error, errno} ->
	raise "Failed to start command: #{state.cmd}, errno: #{errno}"
	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, from, state), do: do_await_exit(state, from)

	def handle_call({:write, _binary}, _from, %{stdin: :closed} = state),
	do: {:reply, {:error, :closed}, state}

	def handle_call({:write, binary}, from, state), do: do_write(state, from, binary)

	def handle_call({:read, bytes}, from, state), do: do_read(state, from, bytes)

	def handle_call(:close_stdin, _from, %{stdin: :closed} = state), do: {:reply, :closed, state}

	def handle_call(:close_stdin, _from, state) do
	case ProcessHelper.close_pipe(state.stdin) do
	:ok -> {:reply, :ok, %{state \| stdin: :closed}}
	{:error, errno} -> {:reply, {:error, errno}, %{state \| errno: errno}}
	end
	end

	def handle_info({:read, bytes, from}, state), do: do_read(state, from, bytes)
	def handle_info({:write, binary, from}, state), do: do_write(state, from, binary)
	def handle_info({:await_exit, from}, state), do: do_await_exit(state, from)

	defp do_write(state, from, binary) do
	case ProcessHelper.write_proc(state.stdin, binary) do
	{:ok, bytes} ->
	# Logger.info("Wrote: #{bytes} length: #{IO.iodata_length(binary)}")

	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)
	else
	GenServer.reply(from, :ok)
	end

	{:noreply, state}

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

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

	defp do_read(state, from, bytes) do
	case ProcessHelper.read_proc(state.stdout, bytes) do
	{:ok, <<>>} ->
	GenServer.reply(from, {:eof, state.read_acc})
	{:noreply, %{state \| read_acc: []}}

	{: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]}}
	else
	GenServer.reply(from, {:ok, [state.read_acc \| binary]})
	{:noreply, %{state \| read_acc: []}}
	end

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

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

	defp do_await_exit(%{pid: pid} = state, from) do
	case ProcessHelper.wait_proc(pid) do
	{^pid, status} ->
	{:reply, {:ok, status}, state}

	{0, _} ->
	Process.send_after(self(), {:await_exit, from}, state.opts.io_busy_wait)
	{:noreply, state}

	{-1, status} ->
	{:reply, {:error, status}, state}
	end
	end

	@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
	parent = self()

	watcher_pid =
	spawn(fn ->
	ref = Process.monitor(parent)
	send(parent, {self(), :done})

	# TODO: should check if process is alreayd exit
	receive do
	{:DOWN, ^ref, :process, ^parent, _reason} ->
	with true <- ProcessHelper.is_alive(pid),
	_ <- ProcessHelper.close_pipe(stdin),
	_ <- ProcessHelper.close_pipe(stdout),
	_ <- :timer.sleep(@stdin_close_wait),
	{p, _} <- ProcessHelper.wait_proc(pid),
	false <- p != pid,
	_ <- ProcessHelper.terminate_proc(pid),
	_ <- :timer.sleep(@sigterm_wait),
	{p, _} <- ProcessHelper.wait_proc(pid),
	false <- p != pid,
	_ <- ProcessHelper.kill_proc(pid) do
	Logger.debug(fn -> "Killed process: #{pid}" end)
	end
	end
	end)

	receive do
	{^watcher_pid, :done} -> :ok
	end
	end
	end
	diff --git a/lib/exile/process_helper.ex b/lib/exile/process_helper.ex
	index 0c063a9..e105f17 100644
	--- a/lib/exile/process_helper.ex
	+++ b/lib/exile/process_helper.ex
	@@ -1,39 +1,39 @@
	defmodule Exile.ProcessHelper do
	@on_load :load_nifs

	def load_nifs do
	:erlang.load_nif('./priv/exile_nif', 0)
	end

	- def exec_proc(_cmd) do
	+ def exec_proc(_cmd, _stderr_to_console) do
	raise "NIF exec_proc/0 not implemented"
	end

	def write_proc(_pipe, _bin) do
	raise "NIF write_proc/2 not implemented"
	end

	def read_proc(_pipe, _bytes) do
	raise "NIF read_proc/0 not implemented"
	end

	def close_pipe(_pipe) do
	raise "NIF close_pipe/1 not implemented"
	end

	def kill_proc(_pid) do
	raise "NIF kill_proc/1 not implemented"
	end

	def terminate_proc(_pid) do
	raise "NIF terminate_proc/1 not implemented"
	end

	def wait_proc(_pid) do
	raise "NIF wait_proc/1 not implemented"
	end

	def is_alive(_pid) do
	raise "NIF is_alive/1 not implemented"
	end
	end
	diff --git a/priv/exile_nif.c b/priv/exile_nif.c
	index a029fd7..effb6aa 100644
	--- a/priv/exile_nif.c
	+++ b/priv/exile_nif.c
	@@ -1,272 +1,278 @@
	#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_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
	};

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

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

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

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

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

	if (pipe(pipes[STDIN_FILENO]) == -1 \|\| pipe(pipes[STDOUT_FILENO]) == -1 \|\|
	pipe(pipes[STDERR_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 \|\|
	set_flag(pipes[STDERR_FILENO][PIPE_READ], O_CLOEXEC \| O_NONBLOCK) < 0 \|\|
	set_flag(pipes[STDERR_FILENO][PIPE_WRITE], O_CLOEXEC \| O_NONBLOCK) < 0) {
	RETURN_ERROR(PIPE_FLAG_ERROR)
	}

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

	case 0:
	close(STDIN_FILENO);
	close(STDOUT_FILENO);
	- close(STDERR_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)

	- int dev_null = open("/dev/null", O_WRONLY);
	- if (dup2(dev_null, STDERR_FILENO) < 0)
	- RETURN_ERROR(PIPE_DUP_ERROR)
	+ if (stderr_to_console != true) {
	+ close(STDERR_FILENO);
	+ int dev_null = open("/dev/null", O_WRONLY);
	+ if (dup2(dev_null, STDERR_FILENO) < 0)
	+ RETURN_ERROR(PIPE_DUP_ERROR)
	+ close(dev_null);
	+ }

	- 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;
	}
	}

	static ERL_NIF_TERM exec_proc(ErlNifEnv *env, int argc,
	const ERL_NIF_TERM argv[]) {
	char _temp[MAX_ARGUMENTS][MAX_ARGUMENT_LEN];
	char *exec_args[MAX_ARGUMENTS + 1];
	char *arg = NULL;
	+ bool stderr_to_console = true;

	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 (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, _temp[i], sizeof(_temp[i]), ERL_NIF_LATIN1) <
	1)
	return enif_make_badarg(env);

	exec_args[i] = _temp[i];
	list = tail;
	}
	exec_args[args_len] = NULL;

	- ExecResult result = start_proccess(exec_args);
	+ int temp;
	+ if (enif_get_int(env, argv[1], &temp) != true)
	+ return enif_make_badarg(env);
	+ stderr_to_console = temp == 1 ? true : false;
	+
	+ ExecResult result = start_proccess(exec_args, stderr_to_console);
	ERL_NIF_TERM ret;

	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));

	return ERL_OK(ret);
	default:
	ret = enif_make_int(env, result.err);
	return ERL_ERROR(ret);
	}
	}

	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);

	ErlNifBinary bin;
	bool is_success = enif_inspect_binary(env, argv[1], &bin);
	int result = write(pipe_in, bin.data, bin.size);

	if (result >= 0) {
	return ERL_OK(enif_make_int(env, result));
	} else if (errno == EAGAIN) {
	return ERL_ERROR(enif_make_int(env, errno));
	} else {
	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);

	int result = close(pipe);

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

	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 (bytes > 65535 \|\| bytes < 1)
	enif_make_badarg(env);

	char buf[bytes];
	int result = read(pipe_out, buf, sizeof(buf));

	if (result >= 0) {
	ErlNifBinary bin;
	enif_alloc_binary(result, &bin);
	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 {
	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);

	int result = kill(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));
	}

	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));
	}

	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);

	int wpid = waitpid(pid, &status, WNOHANG);
	if (wpid != pid) {
	perror("waitpid()");
	}

	return enif_make_tuple2(env, enif_make_int(env, wpid),
	enif_make_int(env, status));
	}

	static ErlNifFunc nif_funcs[] = {
	- {"exec_proc", 1, exec_proc}, {"write_proc", 2, write_proc},
	+ {"exec_proc", 2, exec_proc}, {"write_proc", 2, write_proc},
	{"read_proc", 2, read_proc}, {"close_pipe", 1, close_pipe},
	{"terminate_proc", 1, terminate_proc}, {"wait_proc", 1, wait_proc},
	{"kill_proc", 1, kill_proc}, {"is_alive", 1, is_alive},
	};

	ERL_NIF_INIT(Elixir.Exile.ProcessHelper, nif_funcs, NULL, NULL, NULL, NULL)

File Metadata

Mime Type: text/x-diff
Expires: Sat, Aug 8, 10:55 AM (1 d, 5 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1723108
Default Alt Text: (16 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions