Почтовый ящик

Почтовый ящик — стандартный способ получения актором данных от других акторов в соответствии с моделью акторов. То же: mailbox.

Актор через почтовый ящик получает только копии данных. Можно не беспокоиться, например, о том, что эти данные через секунду-другую изменятся, потому что другой актор тоже имеет доступ к этому участку памяти.

Прочитать очередное сообщение в почтовом ящике актор может с помощью конструкции receive ... end. Пример:

receive 
    Mess ->
        io:format("Получено сообщение: ~ts~n", [Mess])
    after 20_000 ->
        io:format("Давно не приходило сообщений.~n")
end,

Этот оператор срабатывает лишь единожды — на одно сообщение. Когда он сработал, программа переходит к обработке следующих строк кода.

Как и всё другое в Эрланге, этот оператор является выражением, поэтому он возвращает терм. Можно, конечно, привязать этот терм к какой-либо переменной: A = receive ... end. Функция io:format возвращает атом ok, поэтому в нашем примере весь оператор в любом случае вернёт атом ok.

after — факультативная часть оператора. Если она есть, то после N миллисекунд будет выполнена соответствующая цепочка выражений. И наоборот, этот оператор можно вызвать так, чтобы он не вытаскивал очередное сообщение из почтового ящика. Оставив лишь after, мы можем создать таймер — программа засыпает на N миллисекунд (в нашем примере на 1 секунду):

1> receive after 1_000 -> [] end.
[]

В примере выше, где мы используем переменную Mess, “Mess” это паттерн. Можно использовать какой-нибудь более сложный паттерн, например {ok,Mess} или [A,B,C,D,E]. Это позволит сразу провести декомпозицию и оперировать в кляузе полученными переменными: Mess, A, B, C, D или E.

В нашем первом примере оператор receive ... end нуждается в сообщении. Поэтому, если в почтовом ящике пока ничего нет, актор в этом месте замирает. Если не использовать after, актор может замереть навечно, а для нас это может оказаться нежелательным. Поэтому мы используем after.

Если актор не успевает обрабатывать приходящие сообщения, почтовый ящик может очень сильно распухнуть. Лимитов на размер почтового ящика или количество сообщений в нём нет. Есть лишь общее ограничение — объём доступной оперативной памяти.

Следить, не разросся ли почтовый ящик у того или иного актора, удобно в Обсервере.

Скорость обработки сообщений

Пересылка сообщений осуществляется внутри виртуальной машины BEAM, поэтому это весьма скоростная операция. Интересно, за какое именно минимальное время происходит обработка одного сообщения.

Для этого напишем следующую программу. Общая идея такая. Спауним второй актор. Посылаем ему N сообщений. Пусть это будут случайные числа. Актор не сразу приступает к обработке сообщений, сначала он спит 5 секунд. Далее актор с помощью бифа process_info показывает нам, сколько у него сообщений в почтовом ящике, после чего приступает к обработке сообщений. Его задача — из этих N чисел найти максимальное. С помощью timer:tc мы засекаем общее время обработки сообщений, потом находим среднее время обработки сообщений.

-module(proba).
-export([main/1, aktor/1, loop/1]).

main(N) ->
    Pid = spawn(proba, aktor, [N]),
    [ Pid ! rand:uniform() || _X <- lists:seq(1,N) ],
    Pid ! konec,
    main_finished.

aktor(N) ->
    receive after 5_000 -> [] end,
    io:format("message_queue_len: ~p~n", [process_info(self(), message_queue_len)]),
    {Time,_Max} = timer:tc(proba, loop, [0], nanosecond), 
    io:format("Result for N=~p: ~p~n", [N, Time/(N+1)]).

loop(Max) ->
    receive
        konec -> Max;
        Num -> loop(max(Max,Num))
    end.

Время измеряется в наносекундах.

1> proba:main(10).
main_finished
message_queue_len: {message_queue_len,11}
Result for N=10: 474.09090909090907
2> proba:main(100).
main_finished
message_queue_len: {message_queue_len,101}
Result for N=100: 160.33663366336634
3> proba:main(1000).
main_finished
message_queue_len: {message_queue_len,1001}
Result for N=1000: 122.83416583416583
4> proba:main(1_000_000).
main_finished
message_queue_len: {message_queue_len,1000001}
Result for N=1000000: 95.50827849172151

Последним сообщением мы посылаем атом konec, чтобы можно было завершить обработку сообщений и посчитать время. Поэтому количество сообщений в почтовом ящике не 10, а 11, не 100, а 101 и т.д.

Как мы видим, при N=10 скорость обработки составляет 474 нс. При 100 — 160, при 1000 — 123, при 1_000_000 — 96 нс. В силу каких-то причин чем больше скопилось сообщения в п/я, тем быстрее они обрабатываются. Видимо, это как-то вызвано оптимизацией.

При N=1 время обработки превышает одну микросекунду:

5> proba:main(1).
main_finished
message_queue_len: {message_queue_len,2}
Result for N=1: 1792.0

Поскольку чаще всего так и происходит, что в одном ящике не более одного сообщения, то можно прийти к выводу, что время обработки одного сообщения около 1 мкс.