Page MenuHomePhorge

No OneTemporary

Size
46 KB
Referenced Files
None
Subscribers
None
diff --git a/c_src/exile.c b/c_src/exile.c
index efd399d..c985664 100644
--- a/c_src/exile.c
+++ b/c_src/exile.c
@@ -1,672 +1,683 @@
#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, __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 GET_CTX(env, arg, ctx) \
do { \
ExilePriv *data = enif_priv_data(env); \
if (enif_get_resource(env, arg, data->exec_ctx_rt, (void **)&ctx) == \
false) { \
return make_error(env, ATOM_INVALID_CTX); \
} \
} while (0);
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 MAX_ARGUMENT_LEN = 1024;
static const int UNBUFFERED_READ = -1;
static const int PIPE_BUF_SIZE = 65535;
/* We are choosing an exit code which is not reserved see:
* https://www.tldp.org/LDP/abs/html/exitcodes.html. */
static const int FORK_EXEC_FAILURE = 125;
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_CTX;
static ERL_NIF_TERM ATOM_PIPE_CLOSED;
static ERL_NIF_TERM ATOM_EAGAIN;
static ERL_NIF_TERM ATOM_ALLOC_FAILED;
/* 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 *exec_ctx_rt;
ErlNifResourceType *io_rt;
} ExilePriv;
typedef struct ExecContext {
int cmd_input_fd;
int cmd_output_fd;
int exit_status; // can be exit status or signal number depending on exit_type
enum exit_type exit_type;
pid_t pid;
// these are to hold enif_select resource objects
int *read_resource;
int *write_resource;
} ExecContext;
typedef struct StartProcessResult {
bool success;
int err;
ExecContext context;
} StartProcessResult;
/* TODO: assert if the external process is exit (?) */
static void exec_ctx_dtor(ErlNifEnv *env, void *obj) {
ExecContext *ctx = obj;
enif_release_resource(ctx->read_resource);
enif_release_resource(ctx->write_resource);
debug("Exile exec_ctx_dtor called");
}
static void exec_ctx_stop(ErlNifEnv *env, void *obj, int fd,
int is_direct_call) {
debug("Exile exec_ctx_stop called");
}
static void exec_ctx_down(ErlNifEnv *env, void *obj, ErlNifPid *pid,
ErlNifMonitor *monitor) {
debug("Exile exec_ctx_down called");
}
static ErlNifResourceTypeInit exec_ctx_rt_init = {exec_ctx_dtor, exec_ctx_stop,
exec_ctx_down};
static void io_resource_dtor(ErlNifEnv *env, void *obj) {
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) {
debug("Exile io_resource_down called");
}
static ErlNifResourceTypeInit io_rt_init = {io_resource_dtor, io_resource_stop,
io_resource_down};
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);
}
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]);
}
}
/* 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);
}
/* This is not ideal, but as of now there is no portable way to do this */
static void close_all_fds() {
int fd_limit = (int)sysconf(_SC_OPEN_MAX);
for (int i = STDERR_FILENO + 1; i < fd_limit; i++)
close(i);
}
-static StartProcessResult start_process(char *args[], bool stderr_to_console) {
+static StartProcessResult start_process(char *args[], bool stderr_to_console,
+ const char dir[1024]) {
StartProcessResult result = {.success = false};
pid_t pid;
int pipes[2][2] = {{0, 0}, {0, 0}};
if (pipe(pipes[STDIN_FILENO]) == -1 || pipe(pipes[STDOUT_FILENO]) == -1) {
result.err = errno;
perror("[exile] failed to create pipes");
close_all(pipes);
return result;
}
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) {
result.err = errno;
perror("[exile] failed to set flags for pipes");
close_all(pipes);
return result;
}
switch (pid = fork()) {
case -1:
result.err = errno;
perror("[exile] failed to fork");
close_all(pipes);
return result;
case 0: // child
+ if (dir[0] && chdir(dir) != 0) {
+ perror("[exile] failed to change directory");
+ _exit(FORK_EXEC_FAILURE);
+ }
+
close(STDIN_FILENO);
close(STDOUT_FILENO);
if (dup2(r_cmdin, STDIN_FILENO) < 0) {
perror("[exile] failed to dup to stdin");
/* We are assuming FORK_EXEC_FAILURE exit code wont be used by the command
* we are running. Technically we can not assume any exit code here. The
* parent can not differentiate between exit before `exec` and the normal
* command exit.
* One correct way to solve this might be to have a separate
* pipe shared between child and parent and signaling the parent by
* closing it or writing to it. */
_exit(FORK_EXEC_FAILURE);
}
if (dup2(w_cmdout, STDOUT_FILENO) < 0) {
perror("[exile] failed to dup to stdout");
_exit(FORK_EXEC_FAILURE);
}
if (stderr_to_console != true) {
close(STDERR_FILENO);
int dev_null = open("/dev/null", O_WRONLY);
if (dev_null == -1) {
perror("[exile] failed to open /dev/null");
_exit(FORK_EXEC_FAILURE);
}
if (dup2(dev_null, STDERR_FILENO) < 0) {
perror("[exile] failed to dup stderr");
_exit(FORK_EXEC_FAILURE);
}
close(dev_null);
}
close_all_fds();
execvp(args[0], args);
perror("[exile] execvp(): failed");
_exit(FORK_EXEC_FAILURE);
default: // parent
/* close file descriptors used by child */
close(r_cmdin);
close(w_cmdout);
result.success = true;
result.context.pid = pid;
result.context.cmd_input_fd = w_cmdin;
result.context.cmd_output_fd = r_cmdout;
return result;
}
}
/* TODO: return appropriate error instead returning generic "badarg" error */
static ERL_NIF_TERM execute(ErlNifEnv *env, int argc,
const ERL_NIF_TERM argv[]) {
char tmp[MAX_ARGUMENTS][MAX_ARGUMENT_LEN + 1];
char *exec_args[MAX_ARGUMENTS + 1];
+ char dir[1024] = {'\0'};
ErlNifTime start;
unsigned int args_len;
start = enif_monotonic_time(ERL_NIF_USEC);
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;
+ if (enif_get_string(env, argv[2], dir, 1024, ERL_NIF_LATIN1) < 0) {
+ return enif_make_badarg(env);
+ }
+
struct ExilePriv *data = enif_priv_data(env);
- StartProcessResult result = start_process(exec_args, stderr_to_console);
+ StartProcessResult result = start_process(exec_args, stderr_to_console, dir);
ExecContext *ctx = NULL;
ERL_NIF_TERM term;
if (result.success) {
ctx = enif_alloc_resource(data->exec_ctx_rt, sizeof(ExecContext));
ctx->cmd_input_fd = result.context.cmd_input_fd;
ctx->cmd_output_fd = result.context.cmd_output_fd;
ctx->read_resource = enif_alloc_resource(data->io_rt, sizeof(int));
ctx->write_resource = enif_alloc_resource(data->io_rt, sizeof(int));
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);
notify_consumed_timeslice(env, start, enif_monotonic_time(ERL_NIF_USEC));
return make_ok(env, term);
} else {
return make_error(env, 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->write_resource, NULL, ATOM_UNDEFINED);
if (retval != 0)
perror("select_write()");
return retval;
}
static ERL_NIF_TERM sys_write(ErlNifEnv *env, int argc,
const ERL_NIF_TERM argv[]) {
if (argc != 2)
enif_make_badarg(env);
ErlNifTime start;
start = enif_monotonic_time(ERL_NIF_USEC);
ExecContext *ctx = NULL;
GET_CTX(env, argv[0], ctx);
if (ctx->cmd_input_fd == PIPE_CLOSED)
return make_error(env, ATOM_PIPE_CLOSED);
ErlNifBinary bin;
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? */
ssize_t result = write(ctx->cmd_input_fd, bin.data, bin.size);
int write_errono = errno;
notify_consumed_timeslice(env, start, enif_monotonic_time(ERL_NIF_USEC));
/* TODO: branching is ugly, cleanup required */
if (result >= (ssize_t)bin.size) { // request completely satisfied
return make_ok(env, enif_make_int(env, result));
} else if (result >= 0) { // request partially satisfied
int retval = select_write(env, ctx);
if (retval != 0)
return make_error(env, enif_make_int(env, retval));
return make_ok(env, enif_make_int(env, result));
} else if (write_errono == EAGAIN) { // busy
int retval = select_write(env, ctx);
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_errono));
}
}
static ERL_NIF_TERM sys_close(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 {
enif_select(env, ctx->cmd_input_fd, ERL_NIF_SELECT_STOP,
ctx->write_resource, NULL, ATOM_UNDEFINED);
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(env, enif_make_int(env, errno));
}
}
case 1:
if (ctx->cmd_output_fd == PIPE_CLOSED) {
return ATOM_OK;
} else {
enif_select(env, ctx->cmd_output_fd, ERL_NIF_SELECT_STOP,
ctx->read_resource, NULL, ATOM_UNDEFINED);
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(env, 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->read_resource, NULL, ATOM_UNDEFINED);
if (retval != 0)
perror("select_read()");
return retval;
}
static ERL_NIF_TERM sys_read(ErlNifEnv *env, int argc,
const ERL_NIF_TERM argv[]) {
if (argc != 2)
enif_make_badarg(env);
ErlNifTime start;
start = enif_monotonic_time(ERL_NIF_USEC);
ExecContext *ctx = NULL;
GET_CTX(env, argv[0], ctx);
if (ctx->cmd_output_fd == PIPE_CLOSED)
return make_error(env, ATOM_PIPE_CLOSED);
int size, request;
enif_get_int(env, argv[1], &request);
size = request;
if (request == UNBUFFERED_READ) {
size = PIPE_BUF_SIZE;
} else if (request < 1) {
enif_make_badarg(env);
} else if (request > PIPE_BUF_SIZE) {
size = PIPE_BUF_SIZE;
}
unsigned char buf[size];
ssize_t result = read(ctx->cmd_output_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 request completely satisfied OR EOF OR its
* UNBUFFERED_READ */
if (result == request || result == 0 || request == UNBUFFERED_READ) {
return make_ok(env, bin_term);
} else { // request partially satisfied
int retval = select_read(env, ctx);
if (retval != 0)
return make_error(env, enif_make_int(env, retval));
return make_ok(env, bin_term);
}
} else {
if (read_errno == EAGAIN) { // busy
int retval = select_read(env, ctx);
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 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 make_ok(env, ATOM_TRUE);
int result = kill(ctx->pid, 0);
if (result == 0) {
return make_ok(env, ATOM_TRUE);
} else {
return make_ok(env, ATOM_FALSE);
}
}
static ERL_NIF_TERM sys_terminate(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(env, enif_make_int(env, 0));
return make_ok(env, enif_make_int(env, kill(ctx->pid, SIGTERM)));
}
static ERL_NIF_TERM sys_kill(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(env, enif_make_int(env, 0));
return make_ok(env, 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(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 ERL_NIF_TERM sys_wait(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_exit_term(env, ctx);
int status;
int wpid = waitpid(ctx->pid, &status, WNOHANG);
if (wpid == ctx->pid) {
ctx->pid = CMD_EXIT;
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(env, 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(env, enif_make_int(env, 0));
return make_ok(env, 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->exec_ctx_rt =
enif_open_resource_type_x(env, "exile_resource", &exec_ctx_rt_init,
ERL_NIF_RT_CREATE | ERL_NIF_RT_TAKEOVER, NULL);
data->io_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_CTX = enif_make_atom(env, "invalid_exile_exec_ctx");
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");
*priv = (void *)data;
return 0;
}
static void on_unload(ErlNifEnv *env, void *priv) {
debug("exile unload");
enif_free(priv);
}
static ErlNifFunc nif_funcs[] = {
- {"execute", 2, execute, USE_DIRTY_IO},
+ {"execute", 3, execute, USE_DIRTY_IO},
{"sys_write", 2, sys_write, USE_DIRTY_IO},
{"sys_read", 2, sys_read, USE_DIRTY_IO},
{"sys_close", 2, sys_close, USE_DIRTY_IO},
{"sys_terminate", 1, sys_terminate, USE_DIRTY_IO},
{"sys_wait", 1, sys_wait, USE_DIRTY_IO},
{"sys_kill", 1, sys_kill, USE_DIRTY_IO},
{"alive?", 1, is_alive, USE_DIRTY_IO},
{"os_pid", 1, os_pid, USE_DIRTY_IO},
};
ERL_NIF_INIT(Elixir.Exile.ProcessNif, nif_funcs, &on_load, NULL, NULL,
&on_unload)
diff --git a/lib/exile.ex b/lib/exile.ex
index 925e103..ddb2584 100644
--- a/lib/exile.ex
+++ b/lib/exile.ex
@@ -1,43 +1,44 @@
defmodule Exile do
@moduledoc """
Exile is an alternative for beam ports with back-pressure and non-blocking IO
"""
@doc """
Runs the given command with arguments and return an Enumerable to read command output.
First parameter must be a list containing command with arguments. example: `["cat", "file.txt"]`.
### Options
* `input` - Input can be either an `Enumerable` or a function which accepts `Collectable`.
1. input as Enumerable:
```elixir
# List
Exile.stream!(~w(bc -q), input: ["1+1\n", "2*2\n"]) |> Enum.to_list()
# Stream
Exile.stream!(~w(cat), input: File.stream!("log.txt", [], 65536)) |> Enum.to_list()
```
2. input as collectable:
If the input in a function with arity 1, Exile will call that function with a `Collectable` as the argument. The function must *push* input to this collectable. Return value of the function is ignored.
```elixir
Exile.stream!(~w(cat), input: fn sink -> Enum.into(1..100, sink, &to_string/1) end)
|> Enum.to_list()
```
By defaults no input will be given to the command
+ * `cd` - The directory to run the command in
* `exit_timeout` - Duration to wait for external program to exit after completion before raising an error. Defaults to `:infinity`
* `chunk_size` - Size of each iodata chunk emitted by Enumerable stream. When set to `nil` the output is unbuffered and chunk size will be variable. Defaults to 65535
All other options are passed to `Exile.Process.start_link/3`
### Examples
``` elixir
Exile.stream!(~w(ffmpeg -i pipe:0 -f mp3 pipe:1), input: File.stream!("music_video.mkv", [], 65535))
|> Stream.into(File.stream!("music.mp3"))
|> Stream.run()
```
"""
def stream!(cmd_with_args, opts \\ []) do
Exile.Stream.__build__(cmd_with_args, opts)
end
end
diff --git a/lib/exile/process.ex b/lib/exile/process.ex
index 7dff1be..54bf51e 100644
--- a/lib/exile/process.ex
+++ b/lib/exile/process.ex
@@ -1,379 +1,386 @@
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".
### Obligatory NIF warning
As with any NIF based solution, bugs or issues in Exile implementation can bring down the beam VM. But NIF implementation is comparatively small and mostly uses POSIX system calls, spawned external processes are still completely isolated at OS level and the port issues it tries to solve are critical.
"""
alias Exile.ProcessNif
require Logger
use GenServer
defmacro fork_exec_failure(), do: 125
# delay between exit_check when io is busy (in milliseconds)
@default_opts [io_exit_check_delay: 1, stderr_to_console: false]
def start_link(cmd_with_args, opts \\ []) do
opts = Keyword.merge(@default_opts, opts)
GenServer.start(__MODULE__, %{cmd_with_args: cmd_with_args, opts: opts})
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) 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 [
:cmd_with_args,
:opts,
:errno,
:context,
:status,
await: %{},
pending_read: nil,
pending_write: nil
]
alias __MODULE__
def init(args) do
%{cmd_with_args: [cmd | args], opts: opts} = args
path = System.find_executable(cmd)
unless path do
raise "Command not found: #{cmd}"
end
+ if opts[:cd] do
+ if !File.exists?(opts[:cd]) || !File.dir?(opts[:cd]) do
+ raise ":dir is not a valid path"
+ end
+ end
+
state = %__MODULE__{
cmd_with_args: [path | 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_with_args, &to_charlist/1)
stderr_to_console = if state.opts[:stderr_to_console], do: 1, else: 0
+ cd = to_charlist(state.opts[:cd] || "")
- case ProcessNif.execute(exec_args, stderr_to_console) do
+ case ProcessNif.execute(exec_args, stderr_to_console, cd) do
{:ok, context} ->
start_watcher(context)
{:noreply, %Process{state | context: context, status: :start}}
{:error, errno} ->
raise "Failed to start command: #{state.cmd_with_args}, errno: #{errno}"
end
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
{: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) 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(: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, _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(msg, _state), do: raise(msg)
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 ProcessNif.sys_write(state.context, 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 ProcessNif.sys_read(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.sys_read(state.context, 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 check_exit(state, from) do
case ProcessNif.sys_wait(state.context) do
{:ok, {:exit, fork_exec_failure()}} ->
GenServer.reply(from, {:error, :failed_to_execute})
cancel_timer(state, from, :timeout)
{:noreply, clear_await(state, from)}
{: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_exit_check_delay])
{: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.sys_kill(context)
defp do_kill(context, :sigterm), do: ProcessNif.sys_terminate(context)
defp do_close(state, type) do
case ProcessNif.sys_close(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 process 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.sys_wait(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
Logger.debug(fn -> "Stopping external program" end)
# sys_close is idempotent, calling it multiple times is okay
ProcessNif.sys_close(context, stream_type(:stdin))
ProcessNif.sys_close(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.sys_terminate(context)
process_exit?(context, 100) && throw(:done)
Logger.debug("Failed to stop external program with SIGTERM. attempting SIGKILL")
ProcessNif.sys_kill(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 -> "External program exited successfully" end)
end
end
end
end
diff --git a/lib/exile/process_nif.ex b/lib/exile/process_nif.ex
index ff80041..5db610e 100644
--- a/lib/exile/process_nif.ex
+++ b/lib/exile/process_nif.ex
@@ -1,28 +1,28 @@
defmodule Exile.ProcessNif do
@moduledoc false
@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 execute(_cmd, _stderr_to_console), do: :erlang.nif_error(:nif_library_not_loaded)
+ def execute(_cmd, _stderr_to_console, _dir), do: :erlang.nif_error(:nif_library_not_loaded)
def sys_write(_context, _bin), do: :erlang.nif_error(:nif_library_not_loaded)
def sys_read(_context, _bytes), do: :erlang.nif_error(:nif_library_not_loaded)
def sys_close(_context, _pipe), do: :erlang.nif_error(:nif_library_not_loaded)
def sys_kill(_context), do: :erlang.nif_error(:nif_library_not_loaded)
def sys_terminate(_context), do: :erlang.nif_error(:nif_library_not_loaded)
def sys_wait(_context), do: :erlang.nif_error(:nif_library_not_loaded)
def os_pid(_context), do: :erlang.nif_error(:nif_library_not_loaded)
def alive?(_context), do: :erlang.nif_error(:nif_library_not_loaded)
end
diff --git a/test/exile/process_nif_test.exs b/test/exile/process_nif_test.exs
index e68bb31..5be7845 100644
--- a/test/exile/process_nif_test.exs
+++ b/test/exile/process_nif_test.exs
@@ -1,10 +1,10 @@
defmodule Exile.ProcessNifTest do
use ExUnit.Case, async: false
alias Exile.ProcessNif
test "exit before exec" do
- {:ok, ctx} = ProcessNif.execute(['invalid'], 0)
+ {:ok, ctx} = ProcessNif.execute(['invalid'], 0, '')
:timer.sleep(500)
assert {:ok, {:exit, 125}} = ProcessNif.sys_wait(ctx)
end
end
diff --git a/test/exile/process_test.exs b/test/exile/process_test.exs
index e2dbb51..92f3699 100644
--- a/test/exile/process_test.exs
+++ b/test/exile/process_test.exs
@@ -1,281 +1,293 @@
defmodule Exile.ProcessTest do
use ExUnit.Case, async: true
alias Exile.Process
test "read" do
{:ok, s} = Process.start_link(~w(echo test))
assert {:eof, iodata} = Process.read(s, 100)
assert IO.iodata_to_binary(iodata) == "test\n"
assert :ok == Process.close_stdin(s)
assert {:ok, {:exit, 0}} == Process.await_exit(s, 500)
end
test "write" do
{:ok, s} = Process.start_link(~w(cat))
assert :ok == Process.write(s, "hello")
assert {:ok, iodata} = Process.read(s, 5)
assert IO.iodata_to_binary(iodata) == "hello"
assert :ok == Process.write(s, "world")
assert {:ok, iodata} = Process.read(s, 5)
assert IO.iodata_to_binary(iodata) == "world"
assert :ok == Process.close_stdin(s)
assert {:eof, []} == Process.read(s)
assert {:ok, {:exit, 0}} == Process.await_exit(s, 100)
end
test "stdin close" do
logger = start_events_collector()
# base64 produces output only after getting EOF from stdin. we
# collect events in order and assert that we can still read from
# stdout even after closing stdin
{:ok, s} = Process.start_link(~w(base64))
# parallel reader should be blocked till we close stdin
start_parallel_reader(s, logger)
:timer.sleep(100)
assert :ok == Process.write(s, "hello")
add_event(logger, {:write, "hello"})
assert :ok == Process.write(s, "world")
add_event(logger, {:write, "world"})
:timer.sleep(100)
assert :ok == Process.close_stdin(s)
add_event(logger, :input_close)
assert {:ok, {:exit, 0}} == Process.await_exit(s, 50)
assert [
{:write, "hello"},
{:write, "world"},
:input_close,
{:read, "aGVsbG93b3JsZA==\n"},
:eof
] == get_events(logger)
end
test "external command termination on stop" do
{:ok, s} = Process.start_link(~w(cat))
{:ok, os_pid} = Process.os_pid(s)
assert os_process_alive?(os_pid)
Process.stop(s)
:timer.sleep(100)
refute os_process_alive?(os_pid)
end
test "external command kill on stop" do
# cat command hangs waiting for EOF
{:ok, s} = Process.start_link([fixture("ignore_sigterm.sh")])
{:ok, os_pid} = Process.os_pid(s)
assert os_process_alive?(os_pid)
Process.stop(s)
if os_process_alive?(os_pid) do
:timer.sleep(3000)
refute os_process_alive?(os_pid)
else
:ok
end
end
test "exit status" do
{:ok, s} = Process.start_link(~w(sh -c "exit 2"))
assert {:ok, {:exit, 2}} == Process.await_exit(s, 500)
end
test "writing binary larger than pipe buffer size" do
large_bin = generate_binary(5 * 65535)
{:ok, s} = Process.start_link(~w(cat))
writer =
Task.async(fn ->
Process.write(s, large_bin)
Process.close_stdin(s)
end)
:timer.sleep(100)
{_, iodata} = Process.read(s, 5 * 65535)
Task.await(writer)
assert IO.iodata_length(iodata) == 5 * 65535
assert {:ok, {:exit, 0}} == Process.await_exit(s, 500)
end
test "back-pressure" do
logger = start_events_collector()
# we test backpressure by testing if `write` is delayed when we delay read
{:ok, s} = Process.start_link(~w(cat))
large_bin = generate_binary(65535 * 5)
writer =
Task.async(fn ->
Enum.each(1..10, fn i ->
Process.write(s, large_bin)
add_event(logger, {:write, i})
end)
Process.close_stdin(s)
end)
:timer.sleep(50)
reader =
Task.async(fn ->
Enum.each(1..10, fn i ->
Process.read(s, 5 * 65535)
add_event(logger, {:read, i})
# delay in reading should delay writes
:timer.sleep(10)
end)
end)
Task.await(writer)
Task.await(reader)
assert {:ok, {:exit, 0}} == Process.await_exit(s, 500)
assert [
write: 1,
read: 1,
write: 2,
read: 2,
write: 3,
read: 3,
write: 4,
read: 4,
write: 5,
read: 5,
write: 6,
read: 6,
write: 7,
read: 7,
write: 8,
read: 8,
write: 9,
read: 9,
write: 10,
read: 10
] == get_events(logger)
end
# this test does not work properly in linux
@tag :skip
test "if we are leaking file descriptor" do
{:ok, s} = Process.start_link(~w(sleep 60))
{:ok, os_pid} = Process.os_pid(s)
# we are only printing FD, TYPE, NAME with respective prefix
{bin, 0} = System.cmd("lsof", ["-F", "ftn", "-p", to_string(os_pid)])
Process.stop(s)
open_files = parse_lsof(bin)
assert [%{fd: "0", name: _, type: "PIPE"}, %{type: "PIPE", fd: "1", name: _}] = open_files
end
test "process kill with pending write" do
{:ok, s} = Process.start_link(~w(cat))
{:ok, os_pid} = Process.os_pid(s)
large_data =
Stream.cycle(["test"]) |> Stream.take(500_000) |> Enum.to_list() |> IO.iodata_to_binary()
task =
Task.async(fn ->
try do
Process.write(s, large_data)
catch
:exit, reason -> reason
end
end)
:timer.sleep(200)
Process.stop(s)
:timer.sleep(3000)
refute os_process_alive?(os_pid)
assert {:normal, _} = Task.await(task)
end
+ test "cd" do
+ parent = Path.expand("..", File.cwd!())
+ {:ok, s} = Process.start_link(~w(sh -c pwd), cd: parent)
+ {:ok, dir} = Process.read(s)
+ assert String.trim(dir) == parent
+ assert {:ok, {:exit, 0}} = Process.await_exit(s)
+ end
+
+ test "invalid path" do
+ assert {:error, _} = Process.start_link(~w(sh -c pwd), cd: "invalid")
+ end
+
def start_parallel_reader(proc_server, logger) do
spawn_link(fn -> reader_loop(proc_server, logger) end)
end
def reader_loop(proc_server, logger) do
case Process.read(proc_server) do
{:ok, data} ->
add_event(logger, {:read, data})
reader_loop(proc_server, logger)
{:eof, []} ->
add_event(logger, :eof)
end
end
def start_events_collector do
{:ok, ordered_events} = Agent.start(fn -> [] end)
ordered_events
end
def add_event(agent, event) do
:ok = Agent.update(agent, fn events -> events ++ [event] end)
end
def get_events(agent) do
Agent.get(agent, & &1)
end
defp os_process_alive?(pid) do
match?({_, 0}, System.cmd("ps", ["-p", to_string(pid)]))
end
defp fixture(script) do
Path.join([__DIR__, "../scripts", script])
end
defp parse_lsof(iodata) do
String.split(IO.iodata_to_binary(iodata), "\n", trim: true)
|> Enum.reduce([], fn
"f" <> fd, acc -> [%{fd: fd} | acc]
"t" <> type, [h | acc] -> [Map.put(h, :type, type) | acc]
"n" <> name, [h | acc] -> [Map.put(h, :name, name) | acc]
_, acc -> acc
end)
|> Enum.reverse()
|> Enum.reject(fn
%{fd: fd} when fd in ["255", "cwd", "txt"] ->
true
%{fd: "rtd", name: "/", type: "DIR"} ->
true
# filter libc and friends
%{fd: "mem", type: "REG", name: "/lib/x86_64-linux-gnu/" <> _} ->
true
%{fd: "mem", type: "REG", name: "/usr/lib/locale/C.UTF-8/" <> _} ->
true
%{fd: "mem", type: "REG", name: "/usr/lib/locale/locale-archive" <> _} ->
true
%{fd: "mem", type: "REG", name: "/usr/lib/x86_64-linux-gnu/gconv" <> _} ->
true
_ ->
false
end)
end
defp generate_binary(size) do
Stream.repeatedly(fn -> "A" end) |> Enum.take(size) |> IO.iodata_to_binary()
end
end

File Metadata

Mime Type
text/x-diff
Expires
Thu, Nov 28, 10:34 PM (1 d, 19 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
41110
Default Alt Text
(46 KB)

Event Timeline