パート3
パート1では、 10秒ごとにクライアントにメッセージを送信する（少し役に立たない）mochiwebアプリケーションを作成しました。 Linuxカーネルをセットアップし、メモリ使用量をチェックするための多くの接続を確立するツールを作成しました。 各接続には約45 Kbが必要であることがわかりました。
パート2では、アプリケーションを有用なものに変え、メモリ消費を削減します。
•ログイン/ログアウト/送信APIを使用したメッセージルーターの実装。
•mochiwebアプリケーションを更新して、ルーターを操作します。
•分散アーランシステムをインストールして、さまざまなノードでルーターを実行できるようにします。
•多数のメッセージを含むルーターテストツールの作成。
•メモリ使用を24時間スケジュールし、mochiwebアプリケーションを最適化してメモリを節約します。
これは、メッセージ配信ロジックとmochiwebアプリケーションを分離することを意味します。 パート1のFloodtestユーティリティと連携して、産業に近い条件でアプリケーションの動作をテストできます。
メッセージルーターの実装
ルーターAPIには3つの関数のみが含まれています。
•login（Id、Pid）は、メッセージを受信するプロセスを登録します。
•ログアウト（Pid）はメッセージの受信を停止します。
•send（Id、Msg）は、クライアントにメッセージを送信します。
1つのプロセスに対して、異なるIDでログインできることに注意してください。
この例のルーターモジュールは、2 etsテーブルを使用して、PidとId間の双方向マップを保存します。 （#stateエントリのpid2idおよびid2pid）：
-module(router). -behaviour(gen_server). -export([start_link/0]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3]). -export([send/2, login/2, logout/1]). -define(SERVER, global:whereis_name(?MODULE)). % will hold bidirectional mapping between id <–> pid -record(state, {pid2id, id2pid}). start_link() -> gen_server:start_link({global, ?MODULE}, ?MODULE, [], []). % sends Msg to anyone logged in as Id send(Id, Msg) -> gen_server:call(?SERVER, {send, Id, Msg}). login(Id, Pid) when is_pid(Pid) -> gen_server:call(?SERVER, {login, Id, Pid}). logout(Pid) when is_pid(Pid) -> gen_server:call(?SERVER, {logout, Pid}). init([]) -> % set this so we can catch death of logged in pids: process_flag(trap_exit, true), % use ets for routing tables {ok, #state{ pid2id = ets:new(?MODULE, [bag]), id2pid = ets:new(?MODULE, [bag]) } }. handle_call({login, Id, Pid}, _From, State) when is_pid(Pid) -> ets:insert(State#state.pid2id, {Pid, Id}), ets:insert(State#state.id2pid, {Id, Pid}), link(Pid), % tell us if they exit, so we can log them out io:format("~w logged in as ~w\n",[Pid, Id]), {reply, ok, State}; handle_call({logout, Pid}, _From, State) when is_pid(Pid) -> unlink(Pid), PidRows = ets:lookup(State#state.pid2id, Pid), case PidRows of [] -> ok; _ -> IdRows = [ {I,P} || {P,I} <- PidRows ], % invert tuples % delete all pid->id entries ets:delete(State#state.pid2id, Pid), % and all id->pid [ ets:delete_object(State#state.id2pid, Obj) || Obj <- IdRows ] end, io:format("pid ~w logged out\n",[Pid]), {reply, ok, State}; handle_call({send, Id, Msg}, _From, State) -> % get pids who are logged in as this Id Pids = [ P || { _Id, P } <- ets:lookup(State#state.id2pid, Id) ], % send Msg to them all M = {router_msg, Msg}, [ Pid ! M || Pid <- Pids ], {reply, ok, State}. % handle death and cleanup of logged in processes handle_info(Info, State) -> case Info of {'EXIT', Pid, _Why} -> % force logout: handle_call({logout, Pid}, blah, State); Wtf -> io:format("Caught unhandled message: ~w\n", [Wtf]) end, {noreply, State}. handle_cast(_Msg, State) -> {noreply, State}. terminate(_Reason, _State) -> ok. code_change(_OldVsn, State, _Extra) -> {ok, State}.
Mochiwebアプリケーションの更新
ユーザーがmochiwebに接続するURLに基づく整数IDで表され、この識別子を使用してメッセージルーターに登録すると仮定します。 mochiwebは、10秒ごとにブロックされるのではなく、ルーターからメッセージを受信するとブロックされ、ルーターが送信するすべてのリクエストに対してHTTPメッセージをクライアントに送信します。
•クライアントはhttp：// localhost：8000 / test / 123を介してmochiwebに接続します。
•Mochiwebアプリケーションは、この接続のPidをメッセージルーターの識別子「123」に登録します。
•アドレス「123」のルーターにメッセージを送信すると、正しいmochiwebプロセスに送信され、このユーザーのブラウザーに表示されます。
mochiconntest_web.erlの更新バージョンは次のとおりです。
-module(mochiconntest_web). -export([start/1, stop/0, loop/2]). %% External API start(Options) -> {DocRoot, Options1} = get_option(docroot, Options), Loop = fun (Req) -> ?MODULE:loop(Req, DocRoot) end, % we'll set our maximum to 1 million connections. (default: 2048) mochiweb_http:start([{max, 1000000}, {name, ?MODULE}, {loop, Loop} | Options1]). stop() -> mochiweb_http:stop(?MODULE). loop(Req, DocRoot) -> "/" ++ Path = Req:get(path), case Req:get(method) of Method when Method =:= 'GET'; Method =:= 'HEAD' -> case Path of "test/" ++ Id -> Response = Req:ok({"text/html; charset=utf-8", [{"Server","Mochiweb-Test"}], chunked}), % login using an integer rather than a string {IdInt, _} = string:to_integer(Id), router:login(IdInt, self()), feed(Response, IdInt, 1); _ -> Req:not_found() end; 'POST' -> case Path of _ -> Req:not_found() end; _ -> Req:respond({501, [], []}) end. feed(Response, Id, N) -> receive {router_msg, Msg} -> Html = io_lib:format("Recvd msg #~w: '~s'", [N, Msg]), Response:write_chunk(Html) end, feed(Response, Id, N+1). %% Internal API get_option(Option, Options) -> {proplists:get_value(Option, Options), proplists:delete(Option, Options)}.
うまくいく！
ここですべてを整理しましょう-mochiweb用とルーター用の2つのアーランシェルを使用します。 mochiwebの起動に使用されるstart-dev.shを変更し、次の追加パラメーターをerlに追加します。
•-sname n1からアーランノード「n1」への名前付け
•Kカーネルポールを有効にする場合はtrue。
•+ P 134217727-生成できるプロセスの最大数は32768です。接続ごとに1つのプロセスが必要ですが、具体的にどの程度必要かはわかりません。 134 217 727-「man erl」による最大値。
make && ./start-dev.shを実行すると、挨拶が表示されます：（n1 @ localhost）1>-mochiwebアプリケーションが動作し、erlangノードに名前が付けられました。
次に、別のerlangシェルを実行します。
erl -sname n2
現時点では、これらの2つのアーランノードは互いを認識していないため、これを修正します。
(n2@localhost)1> nodes(). [] (n2@localhost)2> net_adm:ping(n1@localhost). pong (n2@localhost)3> nodes(). [n1@localhost]
次に、ルーターをコンパイルして起動します。
(n2@localhost)4> c(router). {ok,router} (n2@localhost)5> router:start_link(). {ok,<0.38.0>}
楽しみのために、ブラウザでhttp：// localhost：8000 / test / 123を開きます（またはコンソールからlynx --source " http：// localhost：8000 / test / 123 "を使用します）。 ルーターを起動したシェルを確認すると、1人のユーザーがログインしていることがわかります。
これで、メッセージをルーターに送信して、ブラウザーでどのように表示されるかを確認できます。 出力には〜sパラメーターを使用し、アトムはエラーになりますので、とりあえず文字列のみを使用してください。
(n2@localhost)6> router:send(123, "Hello World"). (n2@localhost)7> router:send(123, "Why not open another browser window too?"). (n2@localhost)8> router:send(456, "This message will go into the void unless you are connected as /test/456 too").
ブラウザを確認してください、あなたはメッセージを受け取りました:)
分散アーランシステムの実行
異なるマシンでルーターとmochiwebを実行するのは理にかなっています。 テスト用に複数の予備のマシンがある場合、erlangシェルを分散ノードとして実行する必要があります。つまり、-sname n1の代わりに-name n1@host1.example.comを使用する必要があります（n2でも同じです）。 上記の例のように、net_adm：ping（...）でお互いが見えることを確認してください。
router.erlの行16には、ルータープロセスの名前（「ルーター」）がグローバルに登録されているため、次のマクロを使用して、分散システム上であっても、gen_server呼び出しでルーターの場所を識別します。
-define(SERVER, global:whereis_name(?MODULE)).
分散システムのプロセスのグローバルネームレジストリは、Erlangで無料で入手できる多くの機能の1つにすぎません。
大量のメッセージを生成する
実際の環境では、非常にアクティブなユーザーとパッシブなユーザーが多い「ロングテール」のようなパターンが見られる場合があります。 ただし、このテストでは、ランダムなユーザーに無差別に偽のメッセージを送信します。
-module(msggen). -export([start/3]). start(0, _, _) -> ok; start(Num, Interval, Max) -> Id = random:uniform(Max), router:send(Id, "Fake message Num = " ++ Num), receive after Interval -> start(Num -1, Interval, Max) end.
このコードは、Numメッセージを1〜Max間隔msのランダムなユーザーIDに送信します。
ルーターとmochiwebアプリケーションを実行し、 http：// localhost：8000 / test / 3に移動して実行すると、この動作を確認できます：
erl -sname test (test@localhost)1> net_adm:ping(n1@localhost). pong (test@localhost)2> c(msggen). {ok,msggen} (test@localhost)3> msggen:start(20, 10, 5). ok
20のメッセージが1〜5のランダムな識別子に10ミリ秒ごとに1メッセージずつ送信されます。 おそらくあなたは幸運で、いくつかのメッセージを受け取るでしょう。
それらのいくつかを並行して実行して、メッセージの複数のソースをモデル化することもできます。 10個のプロセスの例は次のとおりです。各プロセスは識別子1〜5に20個のメッセージを送信し、各メッセージ間の遅延は100ミリ秒です。
[ spawn(fun() -> msggen:start(20, 100, 5), io:format("~w finished.\n", [self()]) end) || _ <- lists:seq(1,10) ]. [<0.97.0>,<0.98.0>,<0.99.0>,<0.100.0>,<0.101.0>,<0.102.0>, <0.103.0>,<0.104.0>,<0.105.0>,<0.106.0>] <0.101.0> finished. <0.105.0> finished. <0.106.0> finished. <0.104.0> finished. <0.102.0> finished. <0.98.0> finished. <0.99.0> finished. <0.100.0> finished. <0.103.0> finished. <0.97.0> finished.
C10k
テスト用のすべてのパーツがより広範囲にあります。 クライアントはmochiwebアプリケーションに接続し、mochiwebアプリケーションはそれらをメッセージルーターに登録します。 大量の偽のメッセージを生成してルーターに送信し、ルーターから登録済みの顧客に送信できます。 パート1から10,000の同時接続を再度確認しましょう。ただし、今回はすべてのクライアントを接続したままにし、システムを介して多くのメッセージを実行します。
パート1の指示に従って、カーネルなどを構成するとします。 すでにmochiwebアプリケーションとルーターが実行されているので、それらにより多くのトラフィックを投入しましょう。
クライアントが接続されていない場合、mochiwebは約40 MBのメモリを使用します。
$ ps -o rss= -p `pgrep -f 'sname n1'` 40156
この嫌なコマンドを思いついて、時間、mochiwebアプリケーションの現在のメモリ使用量、および60秒ごとに確立された接続数を表示しました。
$ MOCHIPID=`pgrep -f 'name n1'`; while [ 1 ] ; do NUMCON=`netstat -n | awk '/ESTABLISHED/ && $4=="127.0.0.1:8000"' | wc -l`; MEM=`ps -o rss= -p $MOCHIPID`; echo -e "`date`\t`date +%s`\t$MEM\t$NUMCON"; sleep 60; done | tee -a mochimem.log
1つのプロセスのメモリ使用量をグラフィカルに表現する最適な方法を知っている人がいる場合は、コメントを残してください。
次に、新しいerlシェルでパート1のfloodtestを実行します。
erl> floodtest:start("/tmp/mochi-urls.txt", 10). 100 , 10 000 . Stats: {825,0,0} Stats: {1629,0,0} Stats: {2397,0,0} Stats: {3218,0,0} Stats: {4057,0,0} Stats: {4837,0,0} Stats: {5565,0,0} Stats: {6295,0,0} Stats: {7022,0,0} Stats: {7727,0,0} Stats: {8415,0,0} Stats: {9116,0,0} Stats: {9792,0,0} Stats: {10000,0,0} ...
メモリ使用量を確認します。
Mon Oct 20 16:57:24 BST 2008 1224518244 40388 1 Mon Oct 20 16:58:25 BST 2008 1224518305 41120 263 Mon Oct 20 16:59:27 BST 2008 1224518367 65252 5267 Mon Oct 20 17:00:32 BST 2008 1224518432 89008 9836 Mon Oct 20 17:01:37 BST 2008 1224518497 90748 10001 Mon Oct 20 17:02:41 BST 2008 1224518561 90964 10001 Mon Oct 20 17:03:46 BST 2008 1224518626 90964 10001 Mon Oct 20 17:04:51 BST 2008 1224518691 90964 10001
10,000の同時接続（およびFirefoxで開いた接続）に到達し、mochiwebのメモリ消費は約90 MB（90964 KB）です。
次に、いくつかのメッセージを送信します。
erl> [ spawn(fun() -> msggen:start(1000000, 100, 10000) end) || _ <- lists:seq(1,100) ]. [<0.65.0>,<0.66.0>,<0.67.0>,<0.68.0>,<0.69.0>,<0.70.0>, <0.71.0>,<0.72.0>,<0.73.0>,<0.74.0>,<0.75.0>,<0.76.0>, <0.77.0>,<0.78.0>,<0.79.0>,<0.80.0>,<0.81.0>,<0.82.0>, <0.83.0>,<0.84.0>,<0.85.0>,<0.86.0>,<0.87.0>,<0.88.0>, <0.89.0>,<0.90.0>,<0.91.0>,<0.92.0>,<0.93.0>|...]
100プロセスが1秒あたり10メッセージで100万メッセージを1〜10,000のランダムIDに送信します。これは、ルーターが1秒あたり1000メッセージを処理することを意味します。
floodtestの出力を確認すると、クライアントがhttpメッセージを受信していることがわかります（これらは{NumConnected、NumClosed、NumChunksRecvd}であることに注意してください）。
... Stats: {10000,0,5912} Stats: {10000,0,15496} Stats: {10000,0,25145} Stats: {10000,0,34755} Stats: {10000,0,44342} ...
各プロセスで1秒間に10個の100万件のメッセージを処理するには、27時間かかります。 以下は、最初の10分間のメモリ使用量です。
Mon Oct 20 16:57:24 BST 2008 1224518244 40388 1 Mon Oct 20 16:58:25 BST 2008 1224518305 41120 263 Mon Oct 20 16:59:27 BST 2008 1224518367 65252 5267 Mon Oct 20 17:00:32 BST 2008 1224518432 89008 9836 Mon Oct 20 17:01:37 BST 2008 1224518497 90748 10001 Mon Oct 20 17:02:41 BST 2008 1224518561 90964 10001 Mon Oct 20 17:03:46 BST 2008 1224518626 90964 10001 Mon Oct 20 17:04:51 BST 2008 1224518691 90964 10001 Mon Oct 20 17:05:55 BST 2008 1224518755 90980 10001 Mon Oct 20 17:07:00 BST 2008 1224518820 91120 10001 Mon Oct 20 17:08:05 BST 2008 1224518885 98664 10001 Mon Oct 20 17:09:10 BST 2008 1224518950 106752 10001 Mon Oct 20 17:10:15 BST 2008 1224519015 114044 10001 Mon Oct 20 17:11:20 BST 2008 1224519080 119468 10001 Mon Oct 20 17:12:25 BST 2008 1224519145 125360 10001
すべての10 kクライアントが接続されたときにサイズが40 MBから90 MBに既に増加し、しばらくしてから最大125 MBになっていることがわかります。
floodtestはCPUをほとんど使用せず、msggenはCPUの2％を使用し、ルーターとmochiwebは1％未満であることに注意してください。
24時間以内に完了した後の結果
mochiwebプロセスのメモリ使用量を監視しながら、アプリケーションは24時間動作しました。 10,000の接続クライアント、1秒あたり1000メッセージがランダムクライアントに送信されます。
次のトリックを使用して、gnuplotにグラフを描画させました。
(echo -e "set terminal png size 500,300\nset xlabel \"Minutes Elapsed\"\nset ylabel \"Mem (KB)\"\nset title \"Mem usage with 10k active connections, 1000 msg/sec\"\nplot \"-\" using 1:2 with lines notitle" ; awk 'BEGIN{FS="\t";} NR%10==0 {if(!t){t=$2} mins=($2-t)/60; printf("%d %d\n",mins,$3)}' mochimem.log ; echo -e "end" ) | gnuplot > mochimem.png
このグラフは、メモリ使用量（10kのアクティブな接続と1000 msg /秒）が24時間で250 MB以内に調整されていることを示しています。 私が興味のために行ったという事実のために、2つのより低い極端が現れました：
erl> [erlang:garbage_collect(P) || P <- erlang:processes()].
これにより、すべてのプロセスでガベージが収集され、約100 MBのメモリが解放されます。 現在、手動でガベージコレクションを強制することなくメモリを保持する方法を模索しています。
mochiwebのメモリ使用量を削減
mochiwebアプリケーションはメッセージを送信するだけで、すぐにそれらを忘れてしまうことに注意してください。送信されるメッセージの数に応じてメモリ使用量が増加することはありません。
Erlangのメモリ管理に関しては初心者ですが、ガベージをより頻繁に収集するように強制できれば、そのメモリのほとんどをリダイレクトでき、最終的には空のシステムでより多くのユーザーにサービスを提供できるようになることを提案しますメモリ。
文書の調査により、いくつかの結果が得られました。
erlang：system_flag（fullsweep_after、Number）
興味深いことに、これは新しいプロセスにのみ適用され、mochiwebプロセスだけでなく、VM内のすべてのプロセスに影響します。
次：
erlang：system_flag（min_heap_size、MinHeapSize）
役に立つかもしれませんが、mochiwebプロセスにはデフォルト値よりも多くのヒープが必要になると確信しています。 mochiwebのソースコードを変更する必要を回避したいと思います。
近くで、私は気づいた：
erlang：hibernate（モジュール、関数、引数）
理にかなっています-各メッセージの後にスリープ状態に移行して、何が起こるか見てみましょう。
mochiconntest_web.erlを編集し、次を変更します。
•関数の最後の行（Response、Id、N）を変更して、関数自体を呼び出す代わりにスリープモードに入るようにします。
•hibernate（）を呼び出して、受信時にブロックされるのではなく、すぐにルーターにメッセージを送信します。
•フィードをエクスポートすることを忘れないでください/ 3。
mochiconntest_web.erlをメッセージ間の休止状態に更新しました：
-module(mochiconntest_web). -export([start/1, stop/0, loop/2, feed/3]). %% External API start(Options) -> {DocRoot, Options1} = get_option(docroot, Options), Loop = fun (Req) -> ?MODULE:loop(Req, DocRoot) end, % we'll set our maximum to 1 million connections. (default: 2048) mochiweb_http:start([{max, 1000000}, {name, ?MODULE}, {loop, Loop} | Options1]). stop() -> mochiweb_http:stop(?MODULE). loop(Req, DocRoot) -> "/" ++ Path = Req:get(path), case Req:get(method) of Method when Method =:= 'GET'; Method =:= 'HEAD' -> case Path of "test/" ++ IdStr -> Response = Req:ok({"text/html; charset=utf-8", [{"Server","Mochiweb-Test"}], chunked}), {Id, _} = string:to_integer(IdStr), router:login(Id, self()), % Hibernate this process until it receives a message: proc_lib:hibernate(?MODULE, feed, [Response, Id, 1]); _ -> Req:not_found() end; 'POST' -> case Path of _ -> Req:not_found() end; _ -> Req:respond({501, [], []}) end. feed(Response, Id, N) -> receive {router_msg, Msg} -> Html = io_lib:format("Recvd msg #~w: '~w' ", [N, Msg]), Response:write_chunk(Html) end, % Hibernate this process until it receives a message: proc_lib:hibernate(?MODULE, feed, [Response, Id, N+1]). %% Internal API get_option(Option, Options) -> {proplists:get_value(Option, Options), proplists:delete(Option, Options)}.
これらの変更を行い、mochiwebを再構築してから、同じテストを再度行いました。
proc_libを使用した24時間以内の実行後の結果：hibernate（）
hibernate（）を使用するということは、mochiwebアプリケーションのメモリが10 MBの接続で78 MBに整列することを意味します。これは、パート1で見た450 MBよりもはるかに優れています。
だから...
Mochiweb用のCometアプリケーションを作成し、整数IDで識別されるユーザーに任意のメッセージを送信できるようにしました。 接続ユーザーが10,000人で、1秒間に1000メッセージを24時間駆動した後、80 MBのメモリ、またはユーザーあたり8 KBのメモリの使用が観察されました。 素敵なグラフィックも作成しました。
これは本当に進歩です。
次のステップ
パート3では、ユーザー数を100万人に増やします。 十分なメモリがあるマルチプロセッサマシンでテストを実施します。 また、100万の接続をシミュレートするためのいくつかの追加のトリックとカスタマイズも示します。
アプリケーションは一種の「pub-sub」システムに進化し、サブスクリプションはユーザーIDに関連付けられ、アプリケーションによって保存されます。 ソーシャルネットワークデータの典型的なセット、友人を使用します。 これにより、ユーザーは自分のユーザーIDでログインし、友人の1人が生成したイベントを自動的に受信できます。