Help with `:file` blocking GenServer process

Right. I chose the port method and it seems to be working well!

Well the file:read operation blocks by design. Two solutions have already been mentioned: put the read in a separate process which sends you a message when done and which you can kill when you give up on it; use ports which are non-blocking.

Yes, but i understand what you are saying. I understand that file:read(Fifo, Size) will block until Size amount of bytes are available. This is fine, but in my case i didn’t want to block the gen_server process while waiting on those bytes to arrive. In my case they may never arrive for the life of this fifo, meaning i need to be able to cancel, or stop the existing read operation.

I have a stupid question. You are reading with:

{:ok, <<protocol_pattern_match>>} = :file.read(fifo, protocol_size)

Do you write enough bytes into the fifo so this read can complete? I expect yes but it is always best to ask. Start with the simple questions first.

For completeness here is my (mostly) final implementation:

defmodule PipeWorker do
  @moduledoc """
  Proxy for Pipe 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, amnt) do
    GenServer.call(pipe, {:read, amnt})
  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]),
         pipe <- :erlang.open_port(to_charlist(pipe_name), [:eof, :binary]) do
      {:ok, %{pipe_name: pipe_name, pipe: pipe, buffer: <<>>, caller: nil, size: nil}}
    else
      {:error, _} = error -> {:stop, error}
      {_, _num} -> {:stop, {:error, "mkfifo"}}
    end
  end

  def terminate(_, state) do
    Logger.warn("PipeWorker #{state.pipe_name} exit")
    :erlang.port_close(state.pipe)
    File.rm!(state.pipe_name)
  end

  def handle_call({:write, packet}, _from, state) do
    reply = :erlang.port_command(state.pipe, packet)
    {:reply, reply, state}
  end

  def handle_call({:read, amnt}, {_pid, ref} = from, %{caller: nil, size: nil} = state) do
    {:reply, ref, %{state | caller: from, size: amnt, buffer: nil}, {:continue, state.buffer}}
  end

  def handle_info({pipe, {:data, data}}, %{pipe: pipe, buffer: buffer} = state) do
    buffer = buffer <> data
    {:noreply, %{state | buffer: nil}, {:continue, buffer}}
  end

  def handle_continue(buffer, %{size: size} = state) when byte_size(buffer) >= size do
    {pid, ref} = state.caller
    {resp, buffer} = String.split_at(buffer, size)
    send(pid, {__MODULE__, ref, {:ok, resp}})
    {:noreply, %{state | caller: nil, size: nil, buffer: buffer}}
  end

  def handle_continue(buffer, state) do
    {:noreply, %{state | buffer: buffer}}
  end
end

and usage is something like:

# Start the Pipe
{:ok, pid} = PipeWorker.start_link("/tmp/test.pipe")
# Start an asynchronous read of 10 bytes
ref = PipeWorker.read(pid, 10)
# This can obviously be done in `handle_info` of a GenServer or other OTP behaviour.
receive do
   {PipeWorker, ^ref, {:ok, <<protocol_pattern_match>> = data} -> handle_data(data) 
end

For EVEN MORE completeness here’s the NIF implementation i won’t be using: https://github.com/ConnorRigby/pipe_fitter/blob/master/c_src/erl_pipe_nif.c
where usage is something like:

# Start the pipe
{:ok, pipe} = :erl_pipe_nif.start('/tmp/test.pipe')
# Tell the NIF we want a message as soon as data is ready. 
# This is powered by `enif_select`
:ok = :erl_pipe_nif.poll(pipe)
# This can obviously be done in `handle_info` of a GenServer or other OTP behaviour.
receive do
   {:select, ^pipe, _, _} -> 
      # Read the data. Still blocking here, but as long as it is only called after the `:select`
      # message, NBD. 
      :erl_pipe_nif.read(pipe, 10)
end

Anyway Thanks again @josevalim for making me check out the Port option again. Hope the information here is enough to answer anyone elses questions in the future.

I’m still investigating solutions. I whipped up a quick NIF that works similarly to :file.open('test.pipe', [:read, :write, :raw] but allows for calling read/2 within a Task. Unfortunately this is the best solution so far, however i don’t like putting NIFs into production unless there is no other options.

Good point. Can that be done at the pipe level? I.e. have some UNIX utility that buffers it for you or similar?

I can confirm that this does in fact allow reading and writing from a named pipe. However it does add the requirement of buffering now, since you are not telling the port how much data you want to read. Either way thanks for the help!

ah ha! i was trying to use spawn which was not needed. Will try this method out and report back.

So this worked:

~/OSS/elixir[jv-basic-releases %]$ mkfifo test.pipe
~/OSS/elixir[jv-basic-releases %]$ erl
Erlang/OTP 21 [erts-10.0] [source] [64-bit] [smp:8:8] [ds:8:8:10] [async-threads:1] [hipe]

Eshell V10.0  (abort with ^G)
1> Fifo = open_port("test.pipe", [eof]),
1> receive
1> Msg -> io:format("Got ~p", [Msg])
1> end.
Got {#Port<0.5>,{data,"foo\n"}}ok

After typing the receive, I ran echo "foo" > test.pipe in another terminal. Could it be something related to permissions as the error message says?