Привет!
Имеется gen_event который слушает udp порт, получает пакеты и преобразует их в erlang'овый record.
Пакетов _очень много_ и бывают сильные всплески — доходят они далеко не все, очень много потерь.
Так как позднее нужно будет писать данные в БД и проводить над ними различные вычисления (причем и писать и считать сразу), то встает вопрос об архитуктуре приложения, как это все грамотно распараллелить, господа?!
Приветствую объяснения "на пальцах" ибо пока еще императивная голова набекрень и отсутствует опыт реализации распределенных систем.

-module(module_in_awesome_erlang)
-behaviour(gen_event).

...

init([]) ->
  {ok, Socket} = gen_udp:open(?DEFAULT_PORT, ?UDP_OPTIONS),
  {ok, #state {socket = Socket}}.

...

handle_info({udp, Socket, _, _, Packet}, State) ->
  {ok, Record} = netflow5:decode_packet(Packet),
  gen_event:notify(?MODULE, {packet, Record}),
  {ok, State}.

handle_event({packet, Record}, State) ->
  % will save to database in future ;)
  {ok, State}.

Здравствуйте, ZveN, Вы писали:

ZN>Привет!
ZN>Имеется gen_event который слушает udp порт, получает пакеты и преобразует их в erlang'овый record.
ZN>Пакетов _очень много_ и бывают сильные всплески — доходят они далеко не все, очень много потерь.
ZN>Так как позднее нужно будет писать данные в БД и проводить над ними различные вычисления (причем и писать и считать сразу), то встает вопрос об архитуктуре приложения, как это все грамотно распараллелить, господа?!
ZN>Приветствую объяснения "на пальцах" ибо пока еще императивная голова набекрень и отсутствует опыт реализации распределенных систем.

Если у тебя нет чёткой задачи, то и архитектуру писать под неё смысла нет имхо
Императивность/функциональность здесь не особо причём по-моему.
Основной принцип, который я бы посоветовал — пляши от потоков данных, возможно обработку стоит перенести над другие хосты (если она тяжеловатая).
Кстати, что такое "очень много"?
Сколько в секунду?
Исследования лимитов сокетов Эрланга в разрезе операционок читал?
А по коду: нафига у тебя пакет потом передаётся в тот же самый процесс? Какой в этом смысл? Не проще напрямую вызвать функцию?

А почему у вас gen_event вообще этим занимается?


Задача как я понимаю — netflow collector?

Здравствуйте, odobenus-rosmarus, Вы писали:

OR>А почему у вас gen_event вообще этим занимается?

Да это не суть важно, переделать не долго:

loop(Socket, State) ->
    receive
        {udp, Socket, _, _, Packet} ->
            {ok, Record} = decode_packet(Packet),
            loop(Socket, ...);
        Unhandled ->
            io:format("Wtf?~n", []);
        stop ->
            ok = gen_udp:close(Socket)
    end.

OR>Задача как я понимаю — netflow collector?

Телепат Верно.

Здравствуйте, Курилка, Вы писали:

К>Здравствуйте, ZveN, Вы писали:

К>Если у тебя нет чёткой задачи, то и архитектуру писать под неё смысла нет имхо

Задача навесить функцию на каждый полученный пакет по UDP.

К>Основной принцип, который я бы посоветовал — пляши от потоков данных, возможно обработку стоит перенести над другие хосты (если она тяжеловатая).

После получения пакета, разворачивать его в структуру уже в отдельном процессе верно?

К>Кстати, что такое "очень много"?

Пара тысяч человек активно качает данные, собираю netflow.

К>Исследования лимитов сокетов Эрланга в разрезе операционок читал?

Нет, а где можно это почитать?. (OS Linux)

К>А по коду: нафига у тебя пакет потом передаётся в тот же самый процесс? Какой в этом смысл? Не проще напрямую вызвать функцию?

Согласен.

Здравствуйте, ZveN, Вы писали:

ZN>Задача навесить функцию на каждый полученный пакет по UDP.

Дак навешивай, куда ещё тебе тут лишняя "мегаархитектура", если у тебя по идее скорость реакции должна быть вменяемой. Чем больше ты мусора навесишь, тем медленней оно работать будет

К>>Основной принцип, который я бы посоветовал — пляши от потоков данных, возможно обработку стоит перенести над другие хосты (если она тяжеловатая).

ZN>После получения пакета, разворачивать его в структуру уже в отдельном процессе верно?

Абстрактно говорить не буду, чисто просто выделение доп. процесса тебе ничего кроме накладных расходов (пусть и небольших для Эрланга) не даст (хотя, наверное, "отложенная" обработка данных станет возможна). По-моему первичная задача — "словить" все пакеты, потом исследовать много ли процессорного времени останется, если будет жестковато, то вынести обработку (если она сложная) на другой хост. В общем — мерять всё надо и смотреть уже в свете полученных цифр.

К>>Кстати, что такое "очень много"?

ZN>Пара тысяч человек активно качает данные, собираю netflow.

Ммм, речь про это? Как-то это мне не говорит сколько пакетов в секунду.

К>>Исследования лимитов сокетов Эрланга в разрезе операционок читал?

ZN>Нет, а где можно это почитать?. (OS Linux)

Блин, у тебя же UDP, запарил, вспомнилось просто это

Здравствуйте, odobenus-rosmarus, Вы писали:

OR>А почему у вас gen_event вообще этим занимается?
да и тому же получим поддержку supervisor'a

Здравствуйте, ZveN, Вы писали:

ZN>Здравствуйте, odobenus-rosmarus, Вы писали:

OR>>А почему у вас gen_event вообще этим занимается?
ZN>да и тому же получим поддержку supervisor'a

Супервайзору не обязателен стандартный тип процесса, а лишний функционал — это лишний функционал.

Здравствуйте, ZveN, Вы писали:

ZN>

ZN>loop(Socket, State) ->
ZN>    receive
ZN>        {udp, Socket, _, _, Packet} ->
ZN>            {ok, Record} = decode_packet(Packet),
ZN>            loop(Socket, ...);
ZN>        Unhandled ->
ZN>            io:format("Wtf?~n", []);
ZN>        stop ->
ZN>            ok = gen_udp:close(Socket)
ZN>    end.
ZN>

Код апгрейдиться не будет?

Здравствуйте, Курилка, Вы писали:

К>Дак навешивай, куда ещё тебе тут лишняя "мегаархитектура", если у тебя по идее скорость реакции должна быть вменяемой. Чем больше ты мусора навесишь, тем медленней оно работать будет

Не совсем понял, получение пакетов синхронно, и если все операции проводить в том же процессе то это же плохо.

К>Ммм, речь про это? Как-то это мне не говорит сколько пакетов в секунду.

Ага. Мне тоже не говорит, все зависит от кол-ва клиентов.

Здравствуйте, Курилка, Вы писали:

К>Здравствуйте, ZveN, Вы писали:

К>Код апгрейдиться не будет?

Нет.

Здравствуйте, ZveN, Вы писали:

ZN>Здравствуйте, Курилка, Вы писали:

К>>Здравствуйте, ZveN, Вы писали:

К>>Код апгрейдиться не будет?

ZN>Нет.

Точнее сказать я пока этим не заморачиваюсь.

Здравствуйте, ZveN, Вы писали:

ZN>Здравствуйте, Курилка, Вы писали:

К>>Дак навешивай, куда ещё тебе тут лишняя "мегаархитектура", если у тебя по идее скорость реакции должна быть вменяемой. Чем больше ты мусора навесишь, тем медленней оно работать будет

ZN>Не совсем понял, получение пакетов синхронно, и если все операции проводить в том же процессе то это же плохо.

Можешь аргументированно показать чем это плохо?
Пакеты эрланг будет в mailbox складывать и если ты их обработать не сможешь в одном потоке, то несколько потоков тебя не спасут (если только ты обработку по процессорам/ядрам не разнесёшь)

Здравствуйте, ZveN, Вы писали:

ZN>Точнее сказать я пока этим не заморачиваюсь.

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

Здравствуйте, Курилка, Вы писали:

ZN>>Не совсем понял, получение пакетов синхронно, и если все операции проводить в том же процессе то это же плохо.

К>Можешь аргументированно показать чем это плохо?
К>(если только ты обработку по процессорам/ядрам не разнесёшь)

Вот этого и хочется. Так как я только начинаю курить чудоязык, меня смущает "не разнесёшь". Разве при создании N процессов рантайм сам по камням/ядрам не разносит?

Здравствуйте, ZveN, Вы писали:

ZN>Здравствуйте, Курилка, Вы писали:

ZN>>>Не совсем понял, получение пакетов синхронно, и если все операции проводить в том же процессе то это же плохо.

К>>Можешь аргументированно показать чем это плохо?
К>>(если только ты обработку по процессорам/ядрам не разнесёшь)

ZN>Вот этого и хочется. Так как я только начинаю курить чудоязык, меня смущает "не разнесёшь". Разве при создании N процессов рантайм сам по камням/ядрам не разносит?

Ммм, надеяться на "чудо-рантайм", который за тебя всё сделает — очень хороший способ загубить решение. Рантайм не может за тебя знать специфику задачи.
Вот тот же ГЦ в Яве в серьёзных системах приходится порой тонко тюнить, чтоб выжать нужное поведение.
В эрланге есть SMP, только вот на многоядерных процах люди давно уже тупо запускали по рантайму на ядро. Со стороны это, конечно, выглядит как усложение, но зато программист (а лучше архитектор), понимает каким образом разнесена обработка, в результате чего много проще найти ботлнеки и т.п. Ну и в случае чего поставить ещё один "ящик" рядом, если текущий начинает перегружаться.
В общем, если тебе нужно выжать хорошую производительность системы, то надо понимать всю кухню, а если просто надо чтоб "что-то работало", то делай как придётся.

Здравствуйте, ZveN, Вы писали:

ZN>Привет!
ZN>Имеется gen_event который слушает udp порт, получает пакеты и преобразует их в erlang'овый record.
ZN>Пакетов _очень много_ и бывают сильные всплески — доходят они далеко не все, очень много потерь.
ZN>Так как позднее нужно будет писать данные в БД и проводить над ними различные вычисления (причем и писать и считать сразу), то встает вопрос об архитуктуре приложения, как это все грамотно распараллелить, господа?!
ZN>Приветствую объяснения "на пальцах" ибо пока еще императивная голова набекрень и отсутствует опыт реализации распределенных систем.

Параллелить можно по любому признаку в теле пакета. Главное в этом деле — чтобы он проще извлекался, и чтобы тебе не важен был относительный порядок пакетов для разных признаков. Затем делаешь следующее. Вычленяешь признак из полученного пакета (по возможности — не проводя полного рабора пакета), пускаешь запрос по своему словарю процесса, вытягивая по признаку ID процесса, который его обрабатывает, и посылаешь ему сообщение. Если хочется применять OTP — то можешь сделать их gen_server и посылать сообщения через cast. Обрабатывающий процесс уже выполнит полный парсинг и обработку.

Если тебе важен их относительный порядок — тогда раздавай им timestamps в последовательном процессе, и отложи их "склеивание" на возможно более поздний этап.

Вот и все. Здесь Эрланг, собственно, не особо причем. Тут надо думать скорее в терминах потока данных, это сильно помогает, потому что такая модель параллельна по своей семантике. Data Flow Diagrams — построй для своей задачи такую модель — она тривиально реализуема в Эрланге. Просто вродную — процесс есть процесс, данные — это данные, а сторадж это сторадж .

Потом — глядя на граф потока данных, находишь в нем критическую чепочку, и начинаешь сокращать ее, откладывая работу на поздние, параллельные этапы.

Здравствуйте, ZveN, Вы писали:

ZN>Привет!

Вообще — есть один аспект, в котором проектирование для Эрланга сильно отличается от ОО проектирования — хотя в целом там все близко. Этот аспект — в Эрланге инкапсуляция функциональности и состояния не связана, как в ОО, а разделена. В ОО — у вас функциональность нарезана на классы вместе с состояние. В Эрланге — состояние инкапсулируется отдельно — при помощи процессов, а функциональность тоже отдельно — при помощи модулей.

В результате, для мало-мальского описания дизайна для Эрланга вам потребуется _две_ модели (в случае ОО в простых ситуациях хватает только одной — диаграммы классов, а здесь ни в каких не хватит одной). Первая — диаграмма функциональной связи модулей — кто про кого знает. Она — статическая, как диаграмма классов в ОО — только сильно проще, и нечего не говорит о стейте, только связность кода показывает. Вторая — диаграмма потока данных, которая показывает процессы и их взаимодействие. Это уже "динамическая" диаграмма, показывающая вам с одной стороны как завернут и инкапсулирован стейт, с другой — поток данных. При этом, один процесс Эрланга может в общем случае реализовывать группу сущностей из DFD — скажем, внутри него есть и storage, и несколько "процессов" из диаграммы.

Здравствуйте, Gaperton, Вы писали:

G>Здравствуйте, ZveN, Вы писали:

G>Параллелить можно по любому признаку в теле пакета.
Главное в этом деле — чтобы он проще извлекался, и чтобы тебе не важен был относительный порядок пакетов для разных признаков. Затем делаешь следующее. Вычленяешь признак из полученного пакета (по возможности — не проводя полного рабора пакета), пускаешь запрос по своему словарю процесса, вытягивая по признаку ID процесса, который его обрабатывает, и посылаешь ему сообщение. Если хочется применять OTP — то можешь сделать их gen_server и посылать сообщения через cast. Обрабатывающий процесс уже выполнит полный парсинг и обработку.

У меня все проще, я принимаю данные с сенсора netflow, порядок их мне не важен и пакеты идентичные.

Пакет состоит из заголовка и до 30 однотипных записей. Мне нужно принять его и разобрать в record. Далее залогировать все его записи в хранилище, а так же провести рассчеты над каждой записью и сохранить результаты рассчетов в хранилище.

Решение вырисовывается следующее:
1. принимаю пакет с сокета
2. разбираю пакет в record, для этого модуль уже есть (клево все-таки это делается средствами pattern matching)
3. логирую в хранилище gen_event:notify
4. провожу рассчеты / пишу результаты в хранилище.

4-му пуннкту привлечено все внимание. Пологаю нужно будет запустить N нод, выбирать random:uniform(N) и выполнить процесс на этом узле. Реализовать его как gen_server и посылать сообщения через cast например

G>Вот и все. Здесь Эрланг, собственно, не особо причем. Тут надо думать скорее в терминах потока данных, это сильно помогает, потому что такая модель параллельна по своей семантике. Data Flow Diagrams — построй для своей задачи такую модель — она тривиально реализуема в Эрланге. Просто вродную — процесс есть процесс, данные — это данные, а сторадж это сторадж .

Как хранилище пока выступать будет mnesia (кстати чем кошернее работать с postgresql, кроме odbc?).

G>Потом — глядя на граф потока данных, находишь в нем критическую чепочку, и начинаешь сокращать ее, откладывая работу на поздние, параллельные этапы.

Здравствуйте, ZveN, Вы писали:

ZN>У меня все проще, я принимаю данные с сенсора netflow, порядок их мне не важен и пакеты идентичные.

ZN>Пакет состоит из заголовка и до 30 однотипных записей. Мне нужно принять его и разобрать в record. Далее залогировать все его записи в хранилище, а так же провести рассчеты над каждой записью и сохранить результаты рассчетов в хранилище.

ZN>Решение вырисовывается следующее:
ZN>1. принимаю пакет с сокета

Вот с этого момента уже можешь параллелить, если порядок тебе не важен и пакеты идентичны. Посмотри внимательно на модуль pool в документации.
ZN>2. разбираю пакет в record, для этого модуль уже есть (клево все-таки это делается средствами pattern matching)
ZN>3. логирую в хранилище gen_event:notify
ZN>4. провожу рассчеты / пишу результаты в хранилище.

ZN>4-му пуннкту привлечено все внимание. Пологаю нужно будет запустить N нод, выбирать random:uniform(N) и выполнить процесс на этом узле. Реализовать его как gen_server и посылать сообщения через cast например

Посмотри внимательно на модуль pool в документации, перед тем как это делать .

G>>Вот и все. Здесь Эрланг, собственно, не особо причем. Тут надо думать скорее в терминах потока данных, это сильно помогает, потому что такая модель параллельна по своей семантике. Data Flow Diagrams — построй для своей задачи такую модель — она тривиально реализуема в Эрланге. Просто вродную — процесс есть процесс, данные — это данные, а сторадж это сторадж .

ZN>Как хранилище пока выступать будет mnesia (кстати чем кошернее работать с postgresql, кроме odbc?).

Если тебе надо будет потом читать данные последовательно, используй лучше disk_log. Это гораздо эффективнее. Mnesia не очень хорошо работает для накопления логов — не заточена под такой сценарий.