No OneTemporary
Actions

Size

18 KB

Referenced Files

None

Subscribers

None

View Options

	diff --git a/lib/exile.ex b/lib/exile.ex
	new file mode 100644
	index 0000000..ac83448
	--- /dev/null
	+++ b/lib/exile.ex
	@@ -0,0 +1,5 @@
	+defmodule Exile do
	+ def stream!(cmd, args \\ []) do
	+ Exile.Stream.__build__(cmd, args)
	+ end
	+end
	diff --git a/lib/exile/process.ex b/lib/exile/process.ex
	index f3fb851..44deaef 100644
	--- a/lib/exile/process.ex
	+++ b/lib/exile/process.ex
	@@ -1,178 +1,201 @@
	defmodule Exile.Process do
	alias Exile.ProcessHelper
	require Logger
	use GenServer

	- def start_link(cmd, args) do
	- GenServer.start(__MODULE__, %{cmd: cmd, args: args})
	+ # delay between retries when io is busy (in milliseconds)
	+ @default_opts %{io_busy_wait: 1}
	+
	+ 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}) do
	+ 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, read_acc: [], errno: nil}, {:continue, nil}}
	+ {: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)

	case ProcessHelper.exec_proc([state.cmd \| exec_args]) do
	{:ok, {pid, stdin, stdout}} ->
	- start_watcher(pid, stdin)
	- state = Map.merge(state, %{pid: pid, stdin: stdin, stdout: 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(:os_pid, _from, state), do: {:reply, state.pid, state}
	+ 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}
	-
	- {:error, errno} ->
	- {:reply, {:error, errno}, %{state \| errno: errno}}
	+ :ok -> {:reply, :ok, %{state \| stdin: :closed}}
	+ {:error, errno} -> {:reply, {:error, errno}, %{state \| errno: errno}}
	end
	end

	- def handle_call(:await_exit, from, state), do: do_await_exit(state, from)
	-
	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))
	- Process.send_after(self(), {:write, binary, from}, 100)
	+ 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}, 100)
	+ 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}, 100)
	+ 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}, 100)
	+ Process.send_after(self(), {:await_exit, from}, state.opts.io_busy_wait)
	{:noreply, state}

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

	- # def ps(pid) do
	- # {out, 0} = System.cmd("ps", [to_string(pid)])
	- # out
	- # 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) do
	+ 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 {:error, _} <- ProcessHelper.close_pipe(stdin),
	+ 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/stream.ex b/lib/exile/stream.ex
	new file mode 100644
	index 0000000..b2ab6c9
	--- /dev/null
	+++ b/lib/exile/stream.ex
	@@ -0,0 +1,88 @@
	+defmodule Exile.Stream do
	+ alias Exile.Process
	+
	+ defstruct [:proc_server]
	+
	+ @type t :: %__MODULE__{}
	+
	+ @doc false
	+ def __build__(cmd, args) do
	+ {:ok, proc} = Process.start_link(cmd, args)
	+ %Exile.Stream{proc_server: proc}
	+ 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
	+ @chunk_size 65535
	+
	+ def reduce(%{proc_server: proc}, acc, fun) do
	+ start_fun = fn -> :ok end
	+
	+ next_fun = fn :ok ->
	+ case Process.read(proc, @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)
	+
	+ if exit_type == :normal_exit do
	+ case result do
	+ {:ok, 0} -> :ok
	+ {:ok, status} -> raise "command exited with status: #{status}"
	+ end
	+ 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
	diff --git a/priv/exile_nif.c b/priv/exile_nif.c
	index 468157f..a029fd7 100644
	--- a/priv/exile_nif.c
	+++ b/priv/exile_nif.c
	@@ -1,262 +1,272 @@
	#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[]) {
	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 \|\|
	+ 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)
	+
	+ 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;

	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);
	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},
	{"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)
	diff --git a/test/exile_test.exs b/test/exile_test.exs
	index 1ed1c47..4577bc5 100644
	--- a/test/exile_test.exs
	+++ b/test/exile_test.exs
	@@ -1,8 +1,27 @@
	defmodule ExileTest do
	use ExUnit.Case
	- doctest Exile

	- test "greets the world" do
	- assert Exile.hello() == :world
	+ test "stream" do
	+ str = "hello"
	+
	+ proc_stream = Exile.stream!("cat")
	+
	+ Task.async(fn ->
	+ Stream.map(1..1000, fn _ -> str end)
	+ \|> Enum.into(proc_stream)
	+ end)
	+
	+ output =
	+ proc_stream
	+ \|> Enum.to_list()
	+ \|> IO.iodata_to_binary()
	+
	+ assert IO.iodata_length(output) == 1000 * String.length(str)
	+ end
	+
	+ test "stream without stdin" do
	+ proc_stream = Exile.stream!("echo", ["hello"])
	+ output = proc_stream \|> Enum.to_list()
	+ assert IO.iodata_to_binary(output) == "hello\n"
	end
	end

File Metadata

Mime Type: text/x-diff
Expires: Sat, Aug 8, 4:33 PM (1 d, 22 h)
Storage Engine: blob
Storage Format: Raw Data
Storage Handle: 1723705
Default Alt Text: (18 KB)

No OneTemporaryActions

View Options

File Metadata

Event Timeline

No OneTemporary
Actions