I have a GenServer that reads/writes to a Linux Pipe or Fifo. I need to do:
{:ok, fifo} = :file.open('/path/to/fifo', [:read, :write, :binary])
{:ok, <<protocol_pattern_match>>} = :file.read(fifo, protocol_size)
Except :file.read/2 blocks the entire calling process until it is complete similar to read() in C. This is fine because i should be able to just do task = Task.async(:file, :read, [protocol_size]) and get the result in
handle_info/2. Maybe i misunderstood the docs, but that doesn’t seem to be working for me. I’m expecting to get handle_info({ref, {:ok, <<protocol_pattern_match>>}, %{task: %{ref: ref}}) but that doesn’t seem to happen.
My other issue is that Task.shutdown/2 or :file.close/1 do not seem to be working as expected.
the docs for Task.shutdown/2 say when a calling process exits, the task should exit, but even using
Task.shutdown(state.task, :brutal_kill) doesn’t allow me to call :file.close(state.fifo) in terminate/2. (It just blocks forever)
anyway here’s the entire GenServer implementation:
defmodule PipeWorker do
@moduledoc """
Proxy for IO operations.
"""
use GenServer
require Logger
def start_link(pipe_name) do
GenServer.start_link(__MODULE__, pipe_name)
end
def close(pipe) do
GenServer.stop(pipe, :normal)
end
def read(pipe, size) do
GenServer.call(pipe, {:read, [size]}, :infinity)
end
def write(pipe, packet) do
GenServer.call(pipe, {:write, [packet]}, :infinity)
end
def init(pipe_name) do
with {_, 0} <- System.cmd("mkfifo", [pipe_name]),
{:ok, pipe} <- :file.open(to_charlist(pipe_name), [:read, :write, :binary]) do
{:ok, %{pipe_name: pipe_name, pipe: pipe, task: nil, caller: nil}}
else
{:error, _} = error -> {:stop, error}
{_, _num} -> {:stop, {:error, "mkfifo"}}
end
end
def terminate(_, state) do
Logger.warn("PipeWorker #{state.pipe_name} exit")
state.task && Task.shutdown(state.task, :brutal_kill)
Logger.warn("Pipe Task shut down")
IO.inspect(state.pipe, label: "pipe")
# :file.close(state.pipe) # blocks indefinitely no matter what. Shell becomes unresponsive.
# Logger.warn("Pipe closed")
File.rm!(state.pipe_name)
Logger.warn("Pipe removed")
end
def handle_call({cmd, args}, {pid, _} = _from, state) do
IO.inspect([state.pipe | args], label: "Pipe task args")
task = Task.async(:file, cmd, [state.pipe | args])
IO.inspect(task, label: "Pipe task")
{:reply, task.ref, %{state | task: task, caller: pid}}
end
# This is never called?
def handle_info({ref, result}, %{task: %{ref: ref}, caller: pid} = state) do
IO.inspect({ref, result}, label: "Task result")
send(pid, {__MODULE__, ref, result})
{:noreply, %{state | task: nil, caller: nil}}
end
end
