Good point. Can that be done at the pipe level? I.e. have some UNIX utility that buffers it for you or similar?
josevalim
Creator of Elixir
ConnorRigby
Nerves Core Team
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.
0
ConnorRigby
Nerves Core Team
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.
1
rvirding
Creator of Erlang
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. 
2
ConnorRigby
Nerves Core Team
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.
0
rvirding
Creator of Erlang
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.
3
ConnorRigby
Nerves Core Team