Task で一定数ずつ並列で外部サイトにアクセス

Elixir(Erlang)の並列処理の威力を確かめるため、「テストサーバに外部からアクセスできないことを確認する」スクリプトを Elixir で書いてみました。

PythonやRubyなどで普通に直列で実行している場合、途中でつまるサイトがあると、タイムアウトまでそこで止まってしまいます。そういうサイトが複数あった場合にはそりゃもう大変です。並列実行することにより、変なサイトがあっても上限は最悪タイムアウトまでで済むため、精神衛生上とても良いです。しかも Elixir ならとっても簡単に書けます。

ソースコード一式は GitHub に置いてあります。

実行ファイルの作成手順

• mix でプロジェクトのガワを作成
• mix.exs を編集して依存ファイルをダウンロード
• Elixirのソースコードを書く
• コンパイル
• 実行

mix の設定

mix.exs の生成と編集

$ mix new confirm_accessRestriction

• mix.exs を編集します。
  • project に escript を追加
  • application に :httpoison を追加(ロジック側で HTTPoison.start() を呼んでもOK）
  • deps に :httpoison を追加
  • escript_config を定義

mix.exs

defmodule ConfirmAccessRestriction.Mixfile do
  use Mix.Project

  def project do
    [app: :confirm_accessRestriction,
     version: "0.0.1",
     elixir: "~> 1.1",
     escript: escript_config,                  # ←
     build_embedded: Mix.env == :prod,
     start_permanent: Mix.env == :prod,
     deps: deps]
  end

  def application do
    [applications: [:logger, :httpoison]]      # ←
  end

  defp deps do
    [
      {:httpoison, "~> 0.8.0"}                 # ←
      ]
  end

  defp escript_config do                       # ←
    [ main_module: ConfirmAccessRestriction ]
  end

end

依存ファイルのダウンロード

$ mix deps.get
$ mix deps.compile

$ mix deps

実装とコンパイル

ソースコードは後述

$ vi lib/confirm_accessRestriction.ex
$ mix escript.build

ハマりどころ

「Task.async 超便利、簡単に並列処理できる」と思って、何も考えずにプロセスを作りまくって全部並列でアクセスした所、タイムアウトしまくりました w

Elixir(= Erlang)のプロセスは実態は１つのErlangプロセスとして動いています。カーネルモジュールがあるわけでもないので、OSから見たら単なるユーザランドで動く１つのプロセスにすぎないので、プロセスに対するOSの制約をモロに受けてしまいます。

例えば CentOS6とかだとデフォルトのファイルディスクリプタが 1024 だったりします。そのため、

• ulimit -a
• sysctl -a
• cat /proc/【ErlangのプロセスID】/limits

などにより、どういう設定になっているのかを確認する必要があります。

The Road to 2 Million Websocket Connections in Phoenix の WebSocket で200万アクセスの例だと、こんな感じに設定しているようです。128GBのメモリの40core というかなり強力なマシンなので、そのままコピーするわけにはいきませんが。

sysctl -w fs.file-max=12000500
sysctl -w fs.nr_open=20000500
ulimit -n 20000000
sysctl -w net.ipv4.tcp_mem='10000000 10000000 10000000'
sysctl -w net.ipv4.tcp_rmem='1024 4096 16384'
sysctl -w net.ipv4.tcp_wmem='1024 4096 16384'
sysctl -w net.core.rmem_max=16384
sysctl -w net.core.wmem_max=16384

ちなみに CentOS 6.7 のデフォルト値はこうなっていました。

fs.file-max: 386560
fs.nr_open: 1048576
ulimit -n: 1024
net.ipv4.tcp_mem: 365280    487040  730560
net.ipv4.tcp_rmem: 4096 87380   4194304
net.ipv4.tcp_wmem: 4096 16384   4194304
net.core.rmem_max: 124928
net.core.wmem_max: 124928

ソースコード

並列数 x ソケットの問題があるので、アクセス数が多い場合は単純に Enum.map で Task.async() して並列処理するとサチります。

Enum.chunk 関数を使うとリストを一定数ずつ分割できるので、処理する関数をラップしてあげるのが良さそうです。さくらのVPSの4G 環境で動く CentOS 6.7 のデフォルトの設定の場合、ソケットの同時利用数はおよそ CPU core * 12 個ぐらいが安定して動作する限界っぽかったです（もちろん他の環境だと違うハズ）。

      @cpu_core 4
           ・
           ・
     proc_nums = @cpu_core * 12

     url_lists
      |> Enum.chunk(proc_nums, proc_nums, [])
      |> Enum.map(&(do_check/1))
    end

    defp do_check(urls) do
      urls
      |> Enum.map(&(Task.async(fn -> access_url(&1) end)))

ソース全体はこれです。同じものは GitHub にも置いてあります。

url_lists()にアクセス対象のURLを記載して mix escript.build してやれば、confirm_accessRestriction という escript ファイルが生成されます。これはモジュールやらを全部包含したファイル（内部的に zip になっているという話を聞いたことがあるような）のため、Erlang がインストールされている環境ならこのファイルだけあれば ./confirm_accessRestriction のようにして、実行することができます。

lib/confirm_accessRestriction.ex

defmodule ConfirmAccessRestriction do

  require Logger
  @cpu_core 4

  def main(), do: main(:dummy)

  def main(_) do
    # テストした環境では並列度は CPU core * 12 ぐらいが限界っぽい
    proc_nums = @cpu_core * 12

    url_lists
    |> Enum.chunk(proc_nums, proc_nums, [])
    |> Enum.map(&(do_check/1))
  end

  defp do_check(urls) do
    urls
    |> Enum.map(&(Task.async(fn -> access_url(&1) end)))
    |> Enum.map(&(Task.await/1))
    |> Enum.map(&(build_results/1))
    |> Enum.sort(&(&1 > &2))
    |> Enum.map(&(IO.puts/1))
  end

  defp access_url(url) do
    # 遅めのサイトが多いなど、必要に応じて 1000～2000位まで
    # 上げるのがいいかと思います。
    timeout = 500
    http_opt = [{:timeout, timeout}, {:recv_timeout, timeout}, {:max_redirect, 1}]

    user_agent = [{"User-agent", "elixir access check"}]

    # HEAD/GET で挙動が違うサイトがあるので GET でアクセスする
    case HTTPoison.get(url, user_agent, http_opt) do
      {:ok, res} -> {:ok, res.status_code, url}
      {:error, reason} -> {:error, reason.reason, url}
    end
  end

  defp build_results(result) do
    # アクセスできなければ OK(つまり 200 はエラー)
    case result do
      {:ok, 200, url} -> "x NG: #{url}(status_code = #{200})"

      # 200以外(401, 404とか)、タイムアウト、その他のエラー（ドメインが存在しないとか）
      # なら、OKとする
      {:ok, status_code, url} -> "o OK: #{url}(status_code = #{status_code})"
      {:error, _connect_timeout, url} -> "o OK: #{url}(timeout)"
      {_, reason, url} -> "o OK: #{url}(reason = #{reason}"
    end
  end


  defp url_lists() do

    # アクセスチェックするURL

    ~w"""
    http://example.co.jp/login/
    http://example.co.jp/admin/

    """
  end

end