ライフゲームをElixir言語で並列化したかっただけなのに…

はじめに

（Elixir言語の並列化は）初投稿です。

古来より伝わる「ライフゲーム」をモダンな言語で実装するチャレンジやっております。

今回は以前投稿したElixir言語での実装を並列化でコーディングするチャレンジ結果です。

過去シリーズは以下です。

• ライフゲームをモダンな書き方で・Swift編・探究の章
• ライフゲームをモダンな書き方で・Swift編・発動の章
• ライフゲームをモダンな書き方で・Rust接触編
• ライフゲームをモダンな書き方で・Elixir接触編

ライフゲームとは

概要は先駆者様の解説に委ねます。

コーディングに必要なルールは以下となります。

• 2次元グリッド上で展開され、各セルは「生」または「死」の2つの状態を持つ
• 各世代で、「隣接（上下左右と斜め4方向の計8方向にある）」セルを以下のルールに従って状態が更新される
  • 誕生：死んでいるセルに隣接する生きたセルがちょうど3つあれば、次の世代で誕生する
  • 生存：生きているセルに隣接する生きたセルが2つまたは3つならば、次の世代でも生存する
  • 過疎：生きているセルに隣接する生きたセルが1つ以下ならば、過疎により死滅する
  • 過密:：生きているセルに隣接する生きたセルが4つ以上ならば、過密により死滅する
• 端は常に死んでいるとみなす（固定境界）

開発環境

• MacStudio2023(M2 Max)
• macOS Tahoe(26.1)
• Erlang/OTP 27
• Elixir 1.19.3

並列化のために変更した箇所

前回のコードから世代更新部分を並列化するだけ…のはずでしたが、事態は危険な領域へと突入することになりました。

二次元配列から一次元配列へ

前回のライフゲームのコードはセルの集合＝グリッドとみなして、二次元配列で演算から出力まで賄えていました。

しかし並列化する対象は一次元配列で格納しないと都合が悪いことが発覚しました。

そのため、二次元表現の座標と一次元表現のインデックスの変換処理が必須となりました。

  # 座標をインデックスに変換
  @spec get_index(number(), number(), number()) :: number()
  defp get_index(x, y, w), do: y * w + x

  # x座標を取得
  @spec get_pos_x(number(), number()) :: number()
  defp get_pos_x(index, w), do: rem(index, w)

  # y座標を取得
  @spec get_pos_y(number(), number()) :: number()
  defp get_pos_y(index, w), do: div(index, w)

いくつかの変換処理を経てようやく並列化にこぎ着けました。

  @doc """
  次の世代を生成する
  """
  @spec next_by_parallel([0 | 1], integer(), integer()) :: [0 | 1]
  def next_by_parallel(cells, w, h) do
    cell_count = length(cells)
    # セル個数ループ
    0..(cell_count - 1)
    # 並列化
    |> Task.async_stream(fn index ->
      # 現在のセルの生死
      cell = Enum.at(cells, index)
      # 隣接するセルの生死を数える
      alive_neighbors = count_alive_neighbors(cells, index, w, h)
      # 生存または誕生
      rule(cell, alive_neighbors)
    end)
    |> Enum.map(fn {:ok, cell} -> cell end)
  end

並列化した分コード数が多くなったのは、今後の並列化プログラミングの課題となりえるのを肝に命じます。

並列化とは別の機能追加

せっかくだから、俺は便利機能を追加するぜ！

初期配置のパターン定義

ライフゲームの文献で挙げられる初期配置パターンを設定できるようにしました。

ベーマガの掲載コードでよく見たデータの並びでキャラの造形がある程度理解できるあのやり方です。

  # パルサー配置
  @spec get_pulsar_matrix_layout() :: [[0 | 1]]
  defp get_pulsar_matrix_layout() do
    [
      [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 1, 0, 0, 0, 1, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
      [1, 1, 1, 0, 0, 1, 1, 0, 1, 1, 0, 0, 1, 1, 1],
      [0, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 0],
      [0, 0, 0, 0, 1, 1, 0, 0, 0, 1, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 1, 0, 0, 0, 1, 1, 0, 0, 0, 0],
      [0, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 0],
      [1, 1, 1, 0, 0, 1, 1, 0, 1, 1, 0, 0, 1, 1, 1],
      [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 1, 0, 0, 0, 1, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0]
    ]
  end

  @doc """
  初期配置を生成する
  """
  @spec init(integer(), integer(), [[0 | 1]]) :: [0 | 1]
  def init(w, h, layout_matrix) do
    # レイアウトデータ
    layout_length = length(layout_matrix)
    layout_width = length(hd(layout_matrix))
    # 配置は中央を基準にする
    start_x = div(w - layout_width, 2)
    start_y = div(h - layout_length, 2)
    # 配置ループ
    for y <- 0..(h - 1), x <- 0..(w - 1) do
      if y >= start_y and y < start_y + layout_length and
           x >= start_x and x < start_x + layout_width do
        Enum.at(Enum.at(layout_matrix, y - start_y), x - start_x)
      else
        0
      end
    end
  end

# 実行例
cells = init(w, h, get_pulsar_matrix_layout())

CUI上でリアルタイム的出力

「全体クリア」←→「出力」の繰り返しで実現。

  @doc """
  再帰的にセル群を表示する
  """
  @spec recursive([0 | 1], integer(), integer(), integer(), integer()) :: :ok
  def recursive(cells, w, h, generation, max_generation) do
    # 画面クリア (ANSI escape code)
    IO.write("\e[H\e[J")
    # 世代表示
    IO.puts("世代: #{generation} #{if generation == 0, do: "（初期状態）", else: ""}")
    # 現在のセル群表示
    IO.puts(show_cells(cells, w, h))
    # ミリ秒待機
    Process.sleep(100)
    # 有限実施
    if max_generation == 0 or generation < max_generation do
      # 次世代セル群生成
      new_cells = next_by_parallel(cells, w, h)
      # 再帰呼び出し
      recursive(new_cells, w, h, generation + 1, max_generation)
    else
      :ok
    end
  end

コード全体

defmodule Lifegame do
  @moduledoc """
  ライフゲームのコントローラーモジュール
  """

  # 座標をインデックスに変換
  @spec get_index(number(), number(), number()) :: number()
  defp get_index(x, y, w), do: y * w + x

  # x座標を取得
  @spec get_pos_x(number(), number()) :: number()
  defp get_pos_x(index, w), do: rem(index, w)

  # y座標を取得
  @spec get_pos_y(number(), number()) :: number()
  defp get_pos_y(index, w), do: div(index, w)

  @doc """
  初期配置を生成する
  """
  @spec init(integer(), integer(), [[0 | 1]]) :: [0 | 1]
  def init(w, h, layout_matrix) do
    # レイアウトデータ
    layout_length = length(layout_matrix)
    layout_width = length(hd(layout_matrix))
    # 配置は中央を基準にする
    start_x = div(w - layout_width, 2)
    start_y = div(h - layout_length, 2)
    # 配置ループ
    for y <- 0..(h - 1), x <- 0..(w - 1) do
      if y >= start_y and y < start_y + layout_length and
           x >= start_x and x < start_x + layout_width do
        Enum.at(Enum.at(layout_matrix, y - start_y), x - start_x)
      else
        0
      end
    end
  end

  # 生きているセルの隣接するセルの数を数える
  @spec count_alive_neighbors([0 | 1], integer(), integer(), integer()) :: integer()
  defp count_alive_neighbors(cells, index, w, h) do
    # 自身の座標
    x = get_pos_x(index, w)
    y = get_pos_y(index, w)
    # 隣接するセル座標でループ
    for dy <- -1..1, dx <- -1..1, {dx, dy} != {0, 0} do
      neighbor_x = x + dx
      neighbor_y = y + dy
      neighbor_index = get_index(neighbor_x, neighbor_y, w)
      # 隣接範囲内チェック
      if neighbor_x >= 0 and neighbor_x < w and neighbor_y >= 0 and neighbor_y < h do
        # 隣接範囲内のセルの生死
        Enum.at(cells, neighbor_index)
      else
        # 端は常に死んでいるとみなす（固定境界）
        0
      end
    end
    |> Enum.sum()
  end

  # 生存：生きているセルに隣接する生きたセルが2つかまたは3つあれば、生存維持
  # 過疎or過密：上記の条件を満たさない場合は、次の世代で死滅する
  # 誕生：死んでいるセルに隣接する生きたセルがちょうど3つあれば、次の世代で誕生する
  # 過疎or過密：上記の条件を満たさない場合は、次の世代で死滅する
  @spec rule(number(), number()) :: number()
  defp rule(1, alive_neighbors) when alive_neighbors in 2..3, do: 1
  defp rule(1, _), do: 0
  defp rule(0, 3), do: 1
  defp rule(0, _), do: 0

  @doc """
  次の世代を生成する
  """
  @spec next_by_parallel([0 | 1], integer(), integer()) :: [0 | 1]
  def next_by_parallel(cells, w, h) do
    cell_count = length(cells)
    # セル個数ループ
    0..(cell_count - 1)
    # 並列化
    |> Task.async_stream(fn index ->
      # 現在のセルの生死
      cell = Enum.at(cells, index)
      # 隣接するセルの生死を数える
      alive_neighbors = count_alive_neighbors(cells, index, w, h)
      # 生存または誕生
      rule(cell, alive_neighbors)
    end)
    |> Enum.map(fn {:ok, cell} -> cell end)
  end

  # セルの生死を文字列変換して出力
  defp get_cell_str(1), do: "■"
  defp get_cell_str(0), do: "□"

  @doc """
  セル群の状態を文字列に変換して返す
  """
  @spec show_cells([0 | 1], integer(), integer()) :: String.t()
  def show_cells(cells, w, h) do
    for pos_y <- 0..(h - 1) do
      for pos_x <- 0..(w - 1) do
        cell = Enum.at(cells, get_index(pos_x, pos_y, w))
        get_cell_str(cell)
      end
      |> Enum.join(" ")
    end
    |> Enum.join("\n")
  end

  @doc """
  再帰的にセル群を表示する
  """
  @spec recursive([0 | 1], integer(), integer(), integer(), integer()) :: :ok
  def recursive(cells, w, h, generation, max_generation) do
    # 画面クリア (ANSI escape code)
    IO.write("\e[H\e[J")
    # 世代表示
    IO.puts("世代: #{generation} #{if generation == 0, do: "（初期状態）", else: ""}")
    # 現在のセル群表示
    IO.puts(show_cells(cells, w, h))
    # ミリ秒待機
    Process.sleep(100)
    # 有限実施
    if max_generation == 0 or generation < max_generation do
      # 次世代セル群生成
      new_cells = next_by_parallel(cells, w, h)
      # 再帰呼び出し
      recursive(new_cells, w, h, generation + 1, max_generation)
    else
      :ok
    end
  end

  # パルサー配置
  @spec get_pulsar_matrix_layout() :: [[0 | 1]]
  defp get_pulsar_matrix_layout() do
    [
      [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 1, 0, 0, 0, 1, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
      [1, 1, 1, 0, 0, 1, 1, 0, 1, 1, 0, 0, 1, 1, 1],
      [0, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 0],
      [0, 0, 0, 0, 1, 1, 0, 0, 0, 1, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 1, 0, 0, 0, 1, 1, 0, 0, 0, 0],
      [0, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 1, 0, 0],
      [1, 1, 1, 0, 0, 1, 1, 0, 1, 1, 0, 0, 1, 1, 1],
      [0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 1, 0, 0, 0, 1, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0],
      [0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0]
    ]
  end

  # 銀河配置
  defp get_galaxy_matrix_layout() do
    [
      [1, 1, 0, 1, 1, 1, 1, 1, 1],
      [1, 1, 0, 1, 1, 1, 1, 1, 1],
      [1, 1, 0, 0, 0, 0, 0, 0, 0],
      [1, 1, 0, 0, 0, 0, 0, 1, 1],
      [1, 1, 0, 0, 0, 0, 0, 1, 1],
      [1, 1, 0, 0, 0, 0, 0, 1, 1],
      [0, 0, 0, 0, 0, 0, 0, 1, 1],
      [1, 1, 1, 1, 1, 1, 0, 1, 1],
      [1, 1, 1, 1, 1, 1, 0, 1, 1]
    ]
  end

  @doc """
  初期配置パターンを選択する
  """
  @spec get_matrix_layout(:galaxy | :pulsar) :: [[0 | 1]]
  def get_matrix_layout(pattern) do
    case pattern do
      :pulsar -> get_pulsar_matrix_layout()
      :galaxy -> get_galaxy_matrix_layout()
    end
  end

  @doc """
  ライフゲーム起動
  """
  @spec run(integer(), integer(), integer()) :: :ok
  def run(w \\ 25, h \\ 25, max_generation \\ 0) do
    cells = init(w, h, get_matrix_layout(:galaxy))
    recursive(cells, w, h, 0, max_generation)
  end
end

実行結果例

• 銀河パターン

今後の展開予定

• グラフィカルなプラットフォームでのリアルタイム実行
• 他の言語でのチャレンジ

参考文献

Wikipedia - ライフゲーム

あとがき

並列処理に優れたElixir言語ではありますが、今回の実装での恩恵というのはsleepを行っている関係上、観測はできませんでした。

とはいえ、並列コーディングの理解は深められたので学習教材としてはライフゲームは理想的でした。

おわりに

ここまでご閲覧いただきありがとうございます。

今後も思いつくままに記事投稿を続けて行きたい所存であります。

火を点けろ、燃え残った全てに。