Page MenuHomePhorge
Files F85629432

No OneTemporary
Actions

Size
13 KB
Referenced Files
None
Subscribers
None

View Options

diff --git a/lib/exile/process.ex b/lib/exile/process.ex
index afad489..2e828d6 100644
--- a/lib/exile/process.ex
+++ b/lib/exile/process.ex
@@ -1,223 +1,312 @@
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, 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)
+ GenServer.call(process, {:await_exit, timeout}, :infinity)
end
def stop(process) do
- GenServer.stop(process, :normal, :infinity)
+ if exited?(process) do
+ GenServer.stop(process, :normal, :infinity)
+ else
+ GenServer.stop(process, :kill, :infinity)
+ end
+ end
+
+ def exited?(process) do
+ GenServer.call(process, :exited?, :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}}
+ state = %{
+ cmd: path,
+ args: args,
+ opts: opts,
+ read_acc: [],
+ errno: nil,
+ status: :init,
+ await: %{}
+ }
+
+ {:ok, state, {: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], 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(:exited?, _from, state), do: {:reply, !ProcessHelper.is_alive(state.pid), state}
+
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({:await_exit, timeout}, from, state) do
+ tref =
+ if timeout != :infinity do
+ 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),
do: {:reply, {:error, :closed}, state}
def handle_call({:write, binary}, from, state), do: do_write(state, binary, from)
def handle_call({:read, bytes}, from, state), do: do_read(state, bytes, from)
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_call({:kill, signal}, _from, state) do
+ do_kill(state.pid, 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({:await_exit, from}, state), do: do_await_exit(state, 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(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)
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, nil, from) do
case ProcessHelper.read_proc(state.stdout, 65535) do
{:ok, <<>>} ->
GenServer.reply(from, {:eof, []})
{:noreply, state}
{:ok, binary} ->
GenServer.reply(from, {:ok, binary})
{:noreply, state}
# EAGAIN
{:error, 35} ->
Process.send_after(self(), {:read, nil, 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, bytes, from) 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
+ defp check_exit(%{pid: pid} = state, from) do
case ProcessHelper.wait_proc(pid) do
{^pid, status} ->
- {:reply, {:ok, status}, state}
+ GenServer.reply(from, {:ok, status})
+ cancel_timer(state, from, :timeout)
+ {:noreply, clear_await(state, from)}
{0, _} ->
- Process.send_after(self(), {:await_exit, from}, state.opts.io_busy_wait)
- {:noreply, state}
+ tref = Process.send_after(self(), {:check_exit, from}, state.opts.io_busy_wait)
+ {:noreply, put_timer(state, from, :check, tref)}
{-1, status} ->
- {:reply, {:error, status}, state}
+ 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(pid, :sigterm), do: ProcessHelper.terminate_proc(pid)
+
+ defp clear_await(state, from) do
+ %{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)
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)
+ else
+ put_in(state, [:await], %{from => %{key => timer}})
+ end
+ end
+
+ 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
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, :normal} ->
+ :ok
+
{: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)
+ case ProcessHelper.wait_proc(pid) do
+ {^pid, _status} ->
+ # TODO: check stauts
+ nil
+
+ _ ->
+ Logger.debug(fn -> "Killing #{pid}" end)
+
+ with _ <- 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
end)
receive do
{^watcher_pid, :done} -> :ok
end
end
end
diff --git a/lib/exile/stream.ex b/lib/exile/stream.ex
index a792b6b..ad31d65 100644
--- a/lib/exile/stream.ex
+++ b/lib/exile/stream.ex
@@ -1,94 +1,102 @@
defmodule Exile.Stream do
@moduledoc """
Defines a `Exile.Stream` struct returned by `Exile.stream!/3`.
"""
alias Exile.Process
defstruct [:proc_server, :stream_opts]
@default_opts %{exit_timeout: :infinity, chunk_size: 65535}
@type t :: %__MODULE__{}
@doc false
def __build__(cmd, args, opts) do
opts = Map.merge(@default_opts, opts)
{stream_opts, proc_opts} = Map.split(opts, [:exit_timeout, :chunk_size])
{:ok, proc} = Process.start_link(cmd, args, proc_opts)
%Exile.Stream{proc_server: proc, stream_opts: stream_opts}
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
def reduce(%{proc_server: proc, stream_opts: stream_opts}, acc, fun) do
start_fun = fn -> :ok end
next_fun = fn :ok ->
case Process.read(proc, stream_opts.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, stream_opts.exit_timeout)
- if exit_type == :normal_exit do
- case result do
- {:ok, 0} -> :ok
- {:ok, status} -> raise "command exited with status: #{status}"
- end
+ case {exit_type, result} do
+ {_, :timeout} ->
+ Process.kill(proc, :sigkill)
+ raise "command fail to exit within timeout: #{stream_opts.exit_timeout}"
+
+ {:normal, {:ok, 0}} ->
+ :ok
+
+ {:normal, {:ok, exit_status}} ->
+ raise "command exited with status: #{exit_status}"
+
+ {_, error} ->
+ Process.kill(proc, :sigkill)
+ raise "command exited with error: #{inspect(error)}"
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

File Metadata

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

Event Timeline