はじめてなElixir(13) なんちゃって温度センサをやや改良する

はじめてなElixir(11) で spawn を使って温度センサもどきを作ってみました。調子に乗って はじめてなElixir(12) で spawn_link や spawn_monitor に移行しようとして失敗しました。このままだと悔しいので、温度センサもどきをちょっとだけ改良します。

なんちゃって温度センサに履歴をもたせる

はじめてなElixir(11) のプロセスは直前の値を関数の引数として持ち回っていました。これをリストにすれば、過去に得た値も全部センサ内に持っていることができます。

temp1.ex

defmodule Temp1 do
  def pseudo_temp_sensor(temp) do
    receive do
      {sender, {:init, init_temp}} ->
        IO.puts("start of :init")
        send(sender, init_temp)
        pseudo_temp_sensor([init_temp])
      {sender, :next} ->
        IO.puts("start of :next")
        IO.inspect(temp)
        send(sender, hd(temp))
        cur = hd(temp) + Enum.random(0..3) * Enum.random(-1..1)
        pseudo_temp_sensor([cur | temp])
      {sender, :past} ->
        IO.puts("start of :past")
        send(sender, temp)
        pseudo_temp_sensor(temp)
      {sender, :last} ->
        IO.puts("start of :last")
        send(sender, hd(temp))
        pseudo_temp_sensor(temp)
    end
  end

ちょっと IO.puts がウザいですが、動作の確認のために付けておきました。
以下のプログラムで動作を見ます。

  def test_next_temp(pid) do
    send(pid, {self(), :next})
    receive do mes -> mes end
  end

  def test_sensor(zero_th, num \\ 10) do
    pid = spawn(Temp1, :pseudo_temp_sensor, [nil])
    send(pid, {self(), {:init, zero_th}})
    receive do mes -> mes end
    1..num |> Enum.map(fn(_void) -> test_next_temp(pid) end)
    send(pid, {self(), :past})
    receive do mes ->
        IO.inspect(mes)
        IO.inspect(Enum.take(mes, div(num, 2)))
        :ok
    end
  end
end

:next で呼ばれるたびに新しい値がリストの頭に追加されていきます。

iex(5)> c "temp1.ex"        
warning: redefining module Temp1 (current version defined in memory)
  temp1.ex:1

[Temp1]
iex(6)> Temp1.test_sensor(0)
start of :init
start of :next
[0]
start of :next
[0, 0]
start of :next
[-2, 0, 0]
start of :next
[1, -2, 0, 0]
start of :next
[2, 1, -2, 0, 0]
start of :next
[5, 2, 1, -2, 0, 0]
start of :next
[5, 5, 2, 1, -2, 0, 0]
start of :next
[8, 5, 5, 2, 1, -2, 0, 0]
start of :next
[8, 8, 5, 5, 2, 1, -2, 0, 0]
start of :next
[9, 8, 8, 5, 5, 2, 1, -2, 0, 0]
start of :past
[12, 9, 8, 8, 5, 5, 2, 1, -2, 0, 0]
[12, 9, 8, 8, 5]
:ok

:past とやると履歴のリストが返ってきます。これ Enum.take すると頭から持ってくるので最近の値が出てきます。

事象発生プロセス |> 処理関数
というイメージで使うと値が生成された順番に処理関数に食わせることになるので、それとは順序が逆になります。

以下は余計なデバッグプリントを削除したものです。

temp1.ex

defmodule Temp1 do
  def pseudo_temp_sensor(temp) do
    receive do
      {sender, {:init, init_temp}} ->
        send(sender, init_temp)
        pseudo_temp_sensor([init_temp])
      {sender, :next} ->
        send(sender, hd(temp))
        cur = hd(temp) + Enum.random(0..3) * Enum.random(-1..1)
        pseudo_temp_sensor([cur | temp])
      {sender, :past} ->
        send(sender, temp)
        pseudo_temp_sensor(temp)
      {sender, :last} ->
        send(sender, hd(temp))
        pseudo_temp_sensor(temp)
    end
  end

  def test_next_temp(pid) do
    send(pid, {self(), :next})
    receive do mes -> mes end
  end

  def test_sensor(zero_th, num \\ 10) do
    pid = spawn(Temp1, :pseudo_temp_sensor, [nil])
    send(pid, {self(), {:init, zero_th}})
    receive do mes -> mes end
    1..num |> Enum.map(fn(_void) -> test_next_temp(pid) end)
    send(pid, {self(), :past})
    receive do mes ->
      IO.inspect(mes)
    end
    send(pid, {self(), :last})
    receive do mes ->
      IO.inspect(mes)
    end
    :ok
  end
end

:last で最後の値のみ持ってきます。こんな感じで動作します。

iex(16)> c "temp1.ex"        
warning: redefining module Temp1 (current version defined in memory)
  temp1.ex:1

[Temp1]
iex(17)> Temp1.test_sensor(0)
[-1, -4, -4, -4, -7, -7, -5, -6, -4, -1, 0]
-1
:ok
iex(18)> Temp1.test_sensor(0)
[-3, -3, -3, -3, -3, -4, -4, -4, -3, 0, 0]
-3
:ok
iex(19)> Temp1.test_sensor(0)
[-3, -2, 0, 0, -2, -2, -1, -2, -2, -2, 0]
-3
:ok
iex(20)> Temp1.test_sensor(0)
[-3, -6, -6, -6, -4, -4, -4, -1, -2, -2, 0]
-3
:ok
iex(21)> Temp1.test_sensor(0)
[2, 0, 0, 0, -1, 0, 0, 0, 0, 0, 0]
2
:ok
iex(22)> Temp1.test_sensor(0)
[2, 2, 2, 2, 2, -1, -1, 0, 0, 0, 0]
2
:ok

これだとプロセスがだんだん太っちゃいますが、それはまた後で考えましょう。

遅延評価してみる

この温度センサもどきは自分で履歴を貯め込む一方で、値を毎度生成しています。生成するのに1000msかかるようにして、それを受け取った方で遅延評価できてるか見てみましょう。

temp2.ex

defmodule Temp2 do
  def pseudo_temp_sensor(temp) do
    receive do
      {sender, {:init, init_temp}} ->
        send(sender, init_temp)
        pseudo_temp_sensor([init_temp])
      {sender, :next} ->
        :timer.sleep(1000)
        send(sender, hd(temp))
        cur = hd(temp) + Enum.random(0..3) * Enum.random(-1..1)
        pseudo_temp_sensor([cur | temp])
      {sender, :past} ->
        send(sender, temp)
        pseudo_temp_sensor(temp)
      {sender, :last} ->
        send(sender, hd(temp))
        pseudo_temp_sensor(temp)
    end
  end

  def test_next_temp(pid) do
    send(pid, {self(), :next})
    receive do mes -> mes end
  end

  def test_sensor(zero_th, num \\ 10) do
    pid = spawn(Temp2, :pseudo_temp_sensor, [nil])
    send(pid, {self(), {:init, zero_th}})
    receive do mes -> mes end
    1..num
    |> Stream.map(fn(_void) -> test_next_temp(pid) end)
    |> Enum.take(2)
    |> IO.inspect
    :ok
  end
end

ポイントは最後のテストのプログラムのところです。Enum.map だったのを Stream.map にしてあります。データ生成には1つにつき1000msかかるように Erlang の :timer.sleep 関数を入れてます。
以下はデータを 10 個生成する例です。Eager な評価なら10秒かかるはずですが、全部の値が返ってくる前にパイプ越しに評価が進んでしまうので、この例では2秒で終了します。

iex(15)> Temp2.test_sensor(0, 10)
[0, 1]
:ok

ということでプロセスからダラダラと出てくる値を遅延評価で処理できそうなことがわかりました。
P.S.
この md で記述すると tab が 4文字スペースになるらしく、私の 8 文字インデントと違う体裁になるので、tab を 8文字スペースに置き換えました。ちょっとみやすくなったかと。