Page MenuHomePhorge
Files F85628926

No OneTemporary
Actions

Size
19 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/README.md b/README.md
index 80e41b7..4b813da 100644
--- a/README.md
+++ b/README.md
@@ -1,21 +1,87 @@
# Exile
-**TODO: Add description**
+Exile is an alternative to beam [ports](https://hexdocs.pm/elixir/Port.html) for running external programs. It provides back-pressure, non-blocking io, and tries to fix issues with ports.
-## Installation
+At high-level exile is built around the idea of having demand-driven, asynchronous interaction with external command. Think of streaming a video through `ffmpeg` to server a web request. It also provides stream abstraction for interacting with an external program. For example, getting audio out of a stream is as simple as
+``` elixir
+def audio_stream!(stream) do
+ # read from stdin and write to stdout
+ proc_stream = Exile.stream!("ffmpeg", ~w(-i - -f mp3 -))
-If [available in Hex](https://hex.pm/docs/publish), the package can be installed
-by adding `exile` to your list of dependencies in `mix.exs`:
+ Task.async(fn ->
+ Stream.into(stream, proc_stream)
+ |> Stream.run()
+ end)
-```elixir
-def deps do
- [
- {:exile, "~> 0.1.0"}
- ]
+ proc_stream
end
+
+File.stream!("music_video.mkv", [], 65535)
+|> audio_stream!()
+|> Stream.into(File.stream!("music.mp3"))
+|> Stream.run()
```
-Documentation can be generated with [ExDoc](https://github.com/elixir-lang/ex_doc)
-and published on [HexDocs](https://hexdocs.pm). Once published, the docs can
-be found at [https://hexdocs.pm/exile](https://hexdocs.pm/exile).
+`Exile.stream!` is a convenience wrapper around `Exile.Process`. If you want more control over stdin, stdout, and os process use `Exile.Process` directly.
+
+*Note: Exile is still work-in-progress, expect breaking changes. Exile is based on NIF, please know the implications before using it*
+
+## Overview
+
+Approaches, implementations, and issues
+
+#### Port
+
+It is the default way of running external commands. This is okay when you have control over the external program and the interaction is minimal. Port has several important issues.
+
+* it can end up creating [zombie process](https://hexdocs.pm/elixir/Port.html#module-zombie-operating-system-processes)
+* cannot selectively close stdin. This is required when the external programs act on EOF from stdin
+* it sends command output as a message to the beam process. This does not put back pressure on the external program and leads exhausting VM memory
+
+#### Port based solutions
+
+There are many port based libraries such as [Porcelain](https://github.com/alco/porcelain/), [Erlexec](https://github.com/saleyn/erlexec), [Rambo](https://github.com/jayjun/rambo), etc. These solve the first two issues associated with ports: zombie process and selectively closing STDIN. But it does not solve the third issue: having back-pressure. At a high level, these libraries solve port issues by spawning an external middleware program which in turn spawns the program we want to run. Internally uses the port for reading the output and writing input. This is a high-level overview, these libraries are solving a different subset of issues, please check the relevant project page for details.
+
+* additional os process (middleware) for every execution of your program
+* in few cases such as porcelain user has to install this external program explicitly
+* might not be suitable when the program requires constant communication between beam process and external program
+
+#### ExCmd
+
+This is my other stab at solving back pressure on the external program issue. This is also a middleware based solution and has all middleware associated concerns as mentioned above. But unlike the above libraries, it does not use the port for io. It uses named pipes (FIFO) for io and utilizes back-pressure created by the operating system.
+
+__Issue__
+
+ExCmd uses named FIFO for blocking input and output operation for building back-pressure. But the blocking here happens at the system call level. Reading the output from the program internally resolves to a blocking [`read()`](http://man7.org/linux/man-pages/man2/read.2.html) system call. This blocks the dirty io scheduler indefinitely. Since the beam scheduler does not preempt system call, that scheduler will be blocked until it `read()` returns. Worst case scenario: there are as many blocking read/write as there are dirty io schedulers. This can lead to starvation of other io operations, low throughput, and dreaded scheduler collapse.
+
+As of now, there are no non-blocking io file operations available in the beam. Beam didn't allow opening FIFO files before OTP 21 for the same reason. One can fix this temporarily by starting beam with more dirty io schedulers.
+
+## Exile
+
+Exile takes a different NIF based approach. As of now the only way to do non-blocking, asynchronous io operations in the beam is to make the system calls ourselves with NIF (or port-driver). Exile uses a non-blocking system calls for io, so schedulers are never blocked indefinitely. It also uses POSIX `select()`. ie so polling is not done in userland.
+
+Building back-pressure using non-blocking io is done by blocking the program at os level (using pipes) and blocking beam process *within beam*, unlike ExCmd which uses blocking system calls.
+
+**Advantages over other approaches:**
+
+* solves all three issues of the port
+* it does not use any middleware
+ * no additional os process. no performance or resource cost
+ * no need to install any external command
+* can run many external programs in parallel without adversely affecting schedulers
+* stream abstraction for interacting with the external program
+* should be portable across POSIX compliant operating systems (not tested)
+
+##### TODO
+* add benchmarks results
+
+
+### 🚨 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.
+
+
+### Usage
+If all you want is to run a command with no communication, then just sticking with `System.cmd` is a better option.
+For most of the use-cases using `Exile.stream!` abstraction should be enough. Use `Exile.Process` only if you need more control over the life-cycle of IO streams and OS process.
diff --git a/lib/exile/process.ex b/lib/exile/process.ex
index 1953f2c..4660531 100644
--- a/lib/exile/process.ex
+++ b/lib/exile/process.ex
@@ -1,353 +1,372 @@
defmodule Exile.Process do
+ @moduledoc """
+ GenServer which wraps spawned external command.
+
+ One should use `ExCmd.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 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, size) when is_integer(size) do
GenServer.call(process, {:read, size}, :infinity)
end
def read(process) do
GenServer.call(process, {:read, nil}, :infinity)
end
def kill(process, signal) when signal in [:sigkill, :sigterm] do
GenServer.call(process, {:kill, signal}, :infinity)
end
def await_exit(process, timeout \\ :infinity) do
GenServer.call(process, {:await_exit, timeout}, :infinity)
end
def os_pid(process, timeout \\ :infinity) do
GenServer.call(process, :os_pid, :infinity)
end
def stop(process), do: GenServer.call(process, :stop, :infinity)
## Server
defmodule Pending do
defstruct bin: [], remaining: 0, client_pid: nil
end
defstruct [
:cmd,
:cmd_args,
:opts,
:errno,
:context,
:status,
await: %{},
pending_read: nil,
pending_write: nil
]
alias __MODULE__
def init(%{cmd: cmd, args: args, opts: opts}) do
path = :os.find_executable(to_charlist(cmd))
unless path do
raise "Command not found: #{cmd}"
end
state = %__MODULE__{
cmd: path,
cmd_args: args,
opts: opts,
errno: nil,
status: :init,
await: %{},
pending_read: %Pending{},
pending_write: %Pending{}
}
{:ok, state, {:continue, nil}}
end
def handle_continue(nil, state) do
exec_args = Enum.map(state.cmd_args, &to_charlist/1)
stderr_to_console = if state.opts.stderr_to_console, do: 1, else: 0
case ProcessNif.exec_proc([state.cmd | exec_args], stderr_to_console) do
{:ok, context} ->
start_watcher(context)
{:noreply, %Process{state | context: context, status: :start}}
{:error, errno} ->
raise "Failed to start command: #{state.cmd}, errno: #{errno}"
end
end
def handle_call(:stop, _from, state) do
# watcher will 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, bytes}, from, state) do
pending = %Pending{remaining: bytes, client_pid: from}
do_read(%Process{state | pending_read: pending})
end
def handle_call(:close_stdin, _from, state), do: do_close(state, :stdin)
def handle_call(:os_pid, _from, state), do: {:reply, ProcessNif.os_pid(state.context), state}
def handle_call({:kill, signal}, _from, state) do
do_kill(state.context, signal)
{:reply, :ok, %{state | status: {:exit, :killed}}}
end
def handle_info({:check_exit, from}, state), do: check_exit(state, from)
def handle_info({:await_exit_timeout, from}, state) do
cancel_timer(state, from, :check)
receive do
{:check_exit, ^from} -> :ok
after
0 -> :ok
end
GenServer.reply(from, :timeout)
{:noreply, clear_await(state, from)}
end
def handle_info({:select, context, _ref, :ready_output}, state) do
do_write(%Process{state | context: context})
end
def handle_info({:select, context, _ref, :ready_input}, state) do
do_read(%Process{state | context: context})
end
def handle_info(msg, _state), do: raise(msg)
defp do_write(%Process{pending_write: pending} = state) do
case ProcessNif.write_proc(state.context, pending.bin) do
{:ok, size} ->
if size < byte_size(pending.bin) do
binary = binary_part(pending.bin, size, byte_size(pending.bin) - size)
{:noreply, %{state | pending_write: %Pending{bin: binary}}}
else
GenServer.reply(pending.client_pid, :ok)
{:noreply, %{state | pending_write: %Pending{}}}
end
{:error, eagain()} ->
{:noreply, state}
{:error, errno} ->
GenServer.reply(pending.client_pid, {:error, errno})
{:noreply, %{state | errno: errno}}
end
end
defp do_read(%Process{pending_read: %Pending{remaining: nil} = pending} = state) do
case ProcessNif.read_proc(state.context, -1) do
{:ok, <<>>} ->
GenServer.reply(pending.client_pid, {:eof, []})
{:noreply, state}
{:ok, binary} ->
GenServer.reply(pending.client_pid, {:ok, binary})
{:noreply, state}
{:error, eagain()} ->
{:noreply, state}
{:error, errno} ->
GenServer.reply(pending.client_pid, {:error, errno})
{:noreply, %{state | errno: errno}}
end
end
defp do_read(%Process{pending_read: pending} = state) do
case ProcessNif.read_proc(state.context, pending.remaining) do
{:ok, <<>>} ->
GenServer.reply(pending.client_pid, {:eof, pending.bin})
{:noreply, %Process{state | pending_read: %Pending{}}}
{:ok, binary} ->
if 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.wait_proc(state.context) do
{:ok, status} ->
GenServer.reply(from, {:ok, status})
cancel_timer(state, from, :timeout)
{:noreply, clear_await(state, from)}
{:error, {0, _}} ->
# Ideally we should not poll and we should handle this with SIGCHLD signal
tref = Elixir.Process.send_after(self(), {:check_exit, from}, state.opts.io_busy_wait)
{:noreply, put_timer(state, from, :check, tref)}
{:error, {-1, status}} ->
GenServer.reply(from, {:error, status})
cancel_timer(state, from, :timeout)
{:noreply, clear_await(state, from)}
end
end
defp do_kill(context, :sigkill), do: ProcessNif.kill_proc(context)
defp do_kill(context, :sigterm), do: ProcessNif.terminate_proc(context)
defp do_close(state, type) do
case ProcessNif.close_pipe(state.context, stream_type(type)) do
:ok ->
{:reply, :ok, state}
{:error, errno} ->
raise errno
{:reply, {:error, errno}, %Process{state | errno: errno}}
end
end
defp clear_await(state, from) do
%Process{state | await: Map.delete(state.await, from)}
end
defp cancel_timer(state, from, key) do
case get_timer(state, from, key) do
nil -> :ok
tref -> Elixir.Process.cancel_timer(tref)
end
end
defp put_timer(state, from, key, timer) do
if Map.has_key?(state.await, from) do
await = put_in(state.await, [from, key], timer)
%Process{state | await: await}
else
%Process{state | await: %{from => %{key => timer}}}
end
end
defp get_timer(state, from, key), do: get_in(state.await, [from, key])
# Try to gracefully terminate external proccess if the genserver associated with the process is killed
defp start_watcher(context) do
process_server = self()
watcher_pid = spawn(fn -> watcher(process_server, context) end)
receive do
{^watcher_pid, :done} -> :ok
end
end
defp stream_type(:stdin), do: 0
defp stream_type(:stdout), do: 1
defp process_exit?(context) do
match?({:ok, _}, ProcessNif.wait_proc(context))
end
defp process_exit?(context, timeout) do
if process_exit?(context) do
true
else
:timer.sleep(timeout)
process_exit?(context)
end
end
# for proper process exit parent of the child *must* wait() for
# child processes termination exit and "pickup" after the exit
# (receive child exit_status). Resources acquired by child such as
# file descriptors won't be released even if the child process
# itself is terminated.
defp watcher(process_server, context) do
ref = Elixir.Process.monitor(process_server)
send(process_server, {self(), :done})
receive do
{:DOWN, ^ref, :process, ^process_server, _reason} ->
try do
process_exit?(context) && throw(:done)
Logger.debug(fn -> "Stopping external program" end)
ProcessNif.close_pipe(context, stream_type(:stdin))
ProcessNif.close_pipe(context, stream_type(:stdout))
# at max we wait for 100ms for program to exit
process_exit?(context, 100) && throw(:done)
Logger.debug("Failed to stop external program gracefully. attempting SIGTERM")
ProcessNif.terminate_proc(context)
process_exit?(context, 100) && throw(:done)
Logger.debug("Failed to stop external program with SIGTERM. attempting SIGKILL")
ProcessNif.kill_proc(context)
process_exit?(context, 1000) && throw(:done)
Logger.error("[exile] failed to kill external process")
raise "Failed to kill external process"
catch
:done -> Logger.debug(fn -> "Exited external program successfully" end)
end
end
end
end
diff --git a/lib/exile/process_nif.ex b/lib/exile/process_nif.ex
index 98c7166..598bbb9 100644
--- a/lib/exile/process_nif.ex
+++ b/lib/exile/process_nif.ex
@@ -1,26 +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 exec_proc(_cmd, _stderr_to_console), do: :erlang.nif_error(:nif_library_not_loaded)
def write_proc(_context, _bin), do: :erlang.nif_error(:nif_library_not_loaded)
def read_proc(_context, _bytes), do: :erlang.nif_error(:nif_library_not_loaded)
def close_pipe(_context, _pipe), do: :erlang.nif_error(:nif_library_not_loaded)
def kill_proc(_context), do: :erlang.nif_error(:nif_library_not_loaded)
def terminate_proc(_context), do: :erlang.nif_error(:nif_library_not_loaded)
def wait_proc(_context), do: :erlang.nif_error(:nif_library_not_loaded)
def os_pid(_context), do: :erlang.nif_error(:nif_library_not_loaded)
def is_alive(_context), do: :erlang.nif_error(:nif_library_not_loaded)
end

File Metadata

Mime Type
text/x-diff
Expires
Sat, Aug 8, 3:59 PM (1 d, 20 h)
Storage Engine
blob
Storage Format
Raw Data
Storage Handle
1722645
Default Alt Text
(19 KB)

Event Timeline