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

M>>Вариант номер раз: перед чтением первого сообщения, вычитываешь все в локальный кеш и затем работаешь с кешем. Кстати, так ты сможешь и приоритеты реализовать.
N>Если за это время навалят ещё сообщений — не сработает. На практике именно так и происходило.

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

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

M>>>Вариант номер раз: перед чтением первого сообщения, вычитываешь все в локальный кеш и затем работаешь с кешем. Кстати, так ты сможешь и приоритеты реализовать.
N>>Если за это время навалят ещё сообщений — не сработает. На практике именно так и происходило.
F>вообще в энларге это используют. вычитывают весь мейлбокс и обрабатывают сразу.

Да-да. Например, gen_server2 уже хрестоматийный.

F>но это подходит не для всех задач.

Да. Оно эффективно только для сервера, который сам только отвечает, но ничего не спрашивает у других. Когда начинается работа по переформированию и перекладыванию дальше, полезность этого метода снижается в разы.

Ну и само его применение в качестве хака по исправлению отдельных симптомов вместо лечения причины — повод грустить.

Мы ещё применяли передачу экстренных управляющих сообщений полями в ETS. Например, на сервис запускалось два gen_server — один для основной нагрузки и один контрольно-управляющий. При этом контрольно-управляющий складывал предписания нагрузочному исполнителю в ETS, а нагрузочный заглядывал туда раз в N обработанных сообщений (N было от 100 до 1000) и ещё получал предписания заглянуть от таймера (который тупо слал что-то раз в секунду).
Но перевести всё управляющее общение на ETS было в принципе невозможно, так же как и полечить управляемость релейного сервера, который на один запрос к gen_server может создать некоторое количество (пусть даже с коэффициентом размножения сильно меньше 1) аналогичных запросов к другим.

>>> Чтобы перебрать всю очередь и перейти в ожидание, ему нужно прошерстить N-1 сообщение

M>>Вариант номер раз: перед чтением первого сообщения, вычитываешь все в локальный кеш и затем работаешь с кешем. Кстати, так ты сможешь и приоритеты реализовать.

N>Если за это время навалят ещё сообщений — не сработает. На практике именно так и происходило.

С тем, что за время ожидания ответа по tcp тебе могут прийти новые сообщения мне все понятно, но вот с асимптотикой N^2 совсем не понятно.

Если при обработке каждого сообщения тебе придется из очереди выгребать N сообщений, то очередь будет расти (обработали одно сообщение, а пришло N). Больше похоже на линейную сложность.

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

>>>> Чтобы перебрать всю очередь и перейти в ожидание, ему нужно прошерстить N-1 сообщение

M>>>Вариант номер раз: перед чтением первого сообщения, вычитываешь все в локальный кеш и затем работаешь с кешем. Кстати, так ты сможешь и приоритеты реализовать.

N>>Если за это время навалят ещё сообщений — не сработает. На практике именно так и происходило.

M>С тем, что за время ожидания ответа по tcp тебе могут прийти новые сообщения мне все понятно, но вот с асимптотикой N^2 совсем не понятно.

M>Если при обработке каждого сообщения тебе придется из очереди выгребать N сообщений, то очередь будет расти (обработали одно сообщение, а пришло N). Больше похоже на линейную сложность.

Снова теряешь контекст.
Линейная сложность возникает в ситуации, когда данный процесс не требует синхронных ответов. Как именно он это делает — тут уже неважно. Тогда цикл работы — выхватил сообщение из головы(!) очереди (O(1) на сообщение на это действие), отослал ответ, забыл.
Теперь пусть у тебя есть необходимость на каждое сообщение что-то передавать кому-то другому (это может быть процесс, или порт, как в случае gen_tcp) и ждать ответ. Предположим, у тебя есть постоянный темп сообщений. В какой-то момент выгреб всё, это N сообщений. Начинаешь отрабатывать, пока ты отработал K сообщений, в очереди уже ждёт K*p, где p — некоторое число, соответствующее темпам поступлений и расчистки. И на каждое отработанное следующее тебе нужно потратить времени пропорционально K*p для того, чтобы синхронный ответ (другого сервера, порта) извлечь из _конца_ очереди, перебрав все, что были перед ним. Закончил разборку N сообщений — на входе уже стоит N*p. И снова возвращаемся к той формуле, что была раньше, которая и даёт квадрат.

И поэтому надёжно работает только один метод — не пытаться вычитывать никого в обход головы очереди.

Мой алгоритм такой: вычитываешь все сообщения из очереди и кладешь их в кеш. Обрабатываешь первое сообщение из кеша. Вычитываешь, что накопилось снова в кеш. Обрабатываешь из головы кеша. Тогда в очереди (невычитанной) будет накапливаться порядка p сообщений, причем p около 1, иначе реальная очередь, которая в кеше будет расти (в результате сложность становится линейной). При этом, насколько я понимаю, при обработке сообщения, требующего синхронного ответа, нужно будет перебрать только невычитанные сообщения (те самые, которых ~p). Суть с том, чтобы перед обработкой каждого сообщения считывать все накопившиеся сообщения в кеш (если мы сейчас говорим про Ерланг, то не в курсе, как там это делать, т.к. не люблю функциональщину с ее искусственными ограничениями).

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

M>Мой алгоритм такой: вычитываешь все сообщения из очереди и кладешь их в кеш. Обрабатываешь первое сообщение из кеша. Вычитываешь, что накопилось снова в кеш. Обрабатываешь из головы кеша. Тогда в очереди (невычитанной) будет накапливаться порядка p сообщений, причем p около 1, иначе реальная очередь, которая в кеше будет расти (в результате сложность становится линейной). При этом, насколько я понимаю, при обработке сообщения, требующего синхронного ответа, нужно будет перебрать только невычитанные сообщения (те самые, которых ~p). Суть с том, чтобы перед обработкой каждого сообщения считывать все накопившиеся сообщения в кеш (если мы сейчас говорим про Ерланг, то не в курсе, как там это делать, т.к. не люблю функциональщину с ее искусственными ограничениями).

Вот ядро gen_call — синхронного вызова другого процесса, отрабатывающего интерфейс gen_server/gen_fsm:

    try erlang:monitor(process, Process) of
        Mref ->
            catch erlang:send(Process, {Label, {self(), Mref}, Request},
                  [noconnect]),
            receive
                {Mref, Reply} ->
                    erlang:demonitor(Mref, [flush]),
                    {ok, Reply};
                {'DOWN', Mref, _, _, noconnection} ->
                    Node = get_node(Process),
                    exit({nodedown, Node});
                {'DOWN', Mref, _, _, Reason} ->
                    exit(Reason)
            after Timeout ->
                    erlang:demonitor(Mref, [flush]),
                    exit(timeout)
            end
<... скипнул ситуации "нода лежит" и т.п.>

Здесь явный receive. Чтобы сделать по твоей схеме, надо заменить это на попытку вычитать вначале из кэша. Мы говорим о библиотечной функции (которую вызвать, кстати, может другое действие, которое вроде бы не относится к внешнему взаимодействию — но оно так сделано в текущем OTP). Значит, уже нужна хаченая OTP, чтобы это гарантированно работало на всех уровнях.

Далее, аналогично в gen_server и прочих придётся вместо его receive на следующее сообщение делать вычитку из этого кэша. И ещё в 100500 местах.

Да, это всё не имеет отношения к тому, функциональный язык или нет. Зато прямое — к рантайму. Если ты вместо стандартной очереди рантайма делаешь свои — будь готов переделать на них всё. (Чуть более, чем всё, потому что есть ещё куча стороннего софта, который не в курсе твоих заморочек.)

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

N>Линейная сложность возникает в ситуации, когда данный процесс не требует синхронных ответов. Как именно он это делает — тут уже неважно. Тогда цикл работы — выхватил сообщение из головы(!) очереди (O(1) на сообщение на это действие), отослал ответ, забыл.
N>Теперь пусть у тебя есть необходимость на каждое сообщение что-то передавать кому-то другому (это может быть процесс, или порт, как в случае gen_tcp) и ждать ответ. Предположим, у тебя есть постоянный темп сообщений. В какой-то момент выгреб всё, это N сообщений. Начинаешь отрабатывать, пока ты отработал K сообщений, в очереди уже ждёт K*p, где p — некоторое число, соответствующее темпам поступлений и расчистки. И на каждое отработанное следующее тебе нужно потратить времени пропорционально K*p для того, чтобы синхронный ответ (другого сервера, порта) извлечь из _конца_ очереди, перебрав все, что были перед ним. Закончил разборку N сообщений — на входе уже стоит N*p. И снова возвращаемся к той формуле, что была раньше, которая и даёт квадрат.

N>И поэтому надёжно работает только один метод — не пытаться вычитывать никого в обход головы очереди.

Вообще по этой задаче у нас есть неконтролируемый входной поток данных и их обработка, и при этом в общем случае отсутствуют какие-либо гарантии, что обработка происходит быстрее, чем поступление данных на вход. То есть проблема уже заложена в постановку задачи. Selective receive это возможный катализатор (а если используется gen:call, то и не факт), а никак не причина. Рано или поздно очередь будет расти, а это всегда проблема (даже если не дойдет до OOM, то те же control messages будут приходить слишком поздно и т.п.).

Буферный процесс для входящего потока мог бы решить проблему. Он бы просто складывал сообщения в буфер, а мы бы загребали их у него пачками (посылая асинхронное уведомление о готовности обработать не более N сообщений). В итоге наш отправляющий процесс имеет контролируемый входной поток сообщений. Что до буферного, то придется решить, что делать при переполнении — но здесь мы хотя бы это сами контролируем. Переполнение может произойти в любом случае, нельзя просто надеяться что мы отправляем быстрее чем получаем, и selective receive здесь не виноват.

Вообще, здесь на rsdn уже несколько раз читал (в сообщениях разных людей), что медленный selective receive — это родовая травма эрланга и чуть ли не повод его не использовать. Мне это кажется сомнительным, там, где selective receive приводит к проблемам, проблемы возникли бы (чуть позже, но со всей вероятностью) и имей selective receive константную сложность. Просто неконтролируемый входной поток надо делать контролируемым.

Сейчас пробежался глазами, кажется, то же самое предложил Masterspline в этом сообщении:

M>Вариант номер два: делаешь отправку в отдельной корутине, туда же придет и ответ по tcp, и там не будет такой очереди.

N>Проходили. Не работает. Сам догадаешься, почему?

Вот значит я не догадался. Почему? (На случай если я неверно понял, что именно предложил Masterspline, и это совсем другое, то меня конечно больше интересуют недостатки моего собственного предложения)

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

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

Если бы реакция на входную очередь была линейной, твои слова были бы совершенно правильны. Пусть есть некоторая пропускная способность системы. Неважно, чему это равно — тысяча или миллион сообщений в секунду — но она есть, её можно измерить. Если где-то возникает затор, накапливается очередь. Торможение не происходит мгновенно, но если нет сильного роста времени обработки одного сообщения в зависимости от уже накопленного — мы с этим справимся в тех же пределах. В реальности надо, конечно, давать какой-то запас — ну, например, 10% обычно достаточно.

Но описанные факторы дают квадратичную зависимость. И из них любое даже небольшое внутреннее переполнение приводит к лавинному эффекту: временная точка затора начинает разбухать.

В последнем из проектов, где я столкнулся с этим, цифры выглядели примерно так: нагрузка растёт ровно до достижения примерно 25K сообщений в секунду. (В разные пробы были заметно разные цифры, от 15K до 30K, и это тоже показательно.) Пока среднее значение длины очереди релейного процесса — до пары десятков сообщений, всё работает. Но чем выше, тем больше вероятность того, что очередь скакнёт до сотен просто из-за того, что шедулинг не успевает переключать процессы; это чисто стохастический процесс. По достижению этого уровня (сотни сообщений в очереди) релеинг фактически останавливается, снижая скорость, причём до чудовищно низких величин (когда 3K, а когда и просто сотни сообщений в секунду). Далее срабатывал регулятор нагрузки в тесте и снижал её, но привести систему к исходному состоянию можно было только добившись снижения темпа (или полной остановки) до того, что очередь на релейном процессе падала почти до нуля. Тогда его скорость восстанавливалась.

Мы пробовали ставить реакцию на переполнение на приёмной точке (куда входит TCP соединение). Да, это работало. Надо было нарисовать более-менее умную полиси типа "если обнаружили в какой-то момент затор (затор определили как более 1000 сообщений в очереди у приёмного процесса), прекратить приём и ждать, пока не упадёт до 0 (на самом деле поставили — до менее 10 сообщений). Да, в таком варианте работало. Но, за счёт частых впаданий в переполнение, суммарная скорость оказывалась ниже возможной устойчивой. Чтобы получить более-менее устойчивую устойчивость потока, извините за каламбур, пришлось принудительно снижать скорость потока до менее чем половины от предельно известной до начала турбулентности. (Валкин делал что-то похожее.)

_> Selective receive это возможный катализатор (а если используется gen:call, то и не факт), а никак не причина. Рано или поздно очередь будет расти, а это всегда проблема (даже если не дойдет до OOM, то те же control messages будут приходить слишком поздно и т.п.).

Это именно что причина. Другие решения, где я устраняю последовательный перебор очереди за счёт разных очередей, не имеют такой проблемы, потому что обработка одного сообщения не выходит за O(1).
(Разумеется, я не имею в виду ситуацию типа "очередь выжрала всю RAM". Но когда у тебя начинается торможение на 1/100-1/1000 от доступной RAM — это, мягко говоря, неадекватно.)

_>Буферный процесс для входящего потока мог бы решить проблему. Он бы просто складывал сообщения в буфер, а мы бы загребали их у него пачками (посылая асинхронное уведомление о готовности обработать не более N сообщений). В итоге наш отправляющий процесс имеет контролируемый входной поток сообщений.

Плохо. Потому что ещё есть проблема задержки. Задержку тоже надо уменьшать.
Если мы будем действовать по принципу "всю предыдущую порцию отработали и тогда просим следующую" — она будет большой и неравномерной. Чтобы она была равномерной — надо делать по принципу "пока одно сообщение отработали и подтвердили, ещё NN ползёт по буферам и очередям". А в этом случае, как показала практика, регулировка просто не успевает срабатывать до того, как очередь вырастет раза в 3-4, а время обработки соответственно на порядок.

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

Виноват. См. выше.

_>Вообще, здесь на rsdn уже несколько раз читал (в сообщениях разных людей), что медленный selective receive — это родовая травма эрланга и чуть ли не повод его не использовать. Мне это кажется сомнительным, там, где selective receive приводит к проблемам, проблемы возникли бы (чуть позже, но со всей вероятностью) и имей selective receive константную сложность. Просто неконтролируемый входной поток надо делать контролируемым.

В теории, может, и получается. На практике регуляторы не успевают до начала существенных заторов. Как уже сказал, после начала затора единственное, что сработает — "задушить" вход до полной расчистки затора. Результат — низкая средняя скорость и чудовищный jitter.

_>Сейчас пробежался глазами, кажется, то же самое предложил Masterspline в этом сообщении:

M>>Вариант номер два: делаешь отправку в отдельной корутине, туда же придет и ответ по tcp, и там не будет такой очереди.
N>>Проходили. Не работает. Сам догадаешься, почему?
_>Вот значит я не догадался. Почему? (На случай если я неверно понял, что именно предложил Masterspline, и это совсем другое, то меня конечно больше интересуют недостатки моего собственного предложения)

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

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


N>Но описанные факторы дают квадратичную зависимость. И из них любое даже небольшое внутреннее переполнение приводит к лавинному эффекту: временная точка затора начинает разбухать.

N>В последнем из проектов, где я столкнулся с этим, цифры выглядели примерно так: нагрузка растёт ровно до достижения примерно 25K сообщений в секунду. (В разные пробы были заметно разные цифры, от 15K до 30K, и это тоже показательно.) Пока среднее значение длины очереди релейного процесса — до пары десятков сообщений, всё работает. Но чем выше, тем больше вероятность того, что очередь скакнёт до сотен просто из-за того, что шедулинг не успевает переключать процессы; это чисто стохастический процесс. По достижению этого уровня (сотни сообщений в очереди) релеинг фактически останавливается, снижая скорость, причём до чудовищно низких величин (когда 3K, а когда и просто сотни сообщений в секунду). Далее срабатывал регулятор нагрузки в тесте и снижал её, но привести систему к исходному состоянию можно было только добившись снижения темпа (или полной остановки) до того, что очередь на релейном процессе падала почти до нуля. Тогда его скорость восстанавливалась.

Это похоже на network congestion очень сильно. Без backpressure это никак не решить. Даже с разными очердями, рано или поздно, наступит момент когда без backpressure это перестанет работать.

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

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

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

N>>В последнем из проектов, где я столкнулся с этим, цифры выглядели примерно так: нагрузка растёт ровно до достижения примерно 25K сообщений в секунду. (В разные пробы были заметно разные цифры, от 15K до 30K, и это тоже показательно.) Пока среднее значение длины очереди релейного процесса — до пары десятков сообщений, всё работает. Но чем выше, тем больше вероятность того, что очередь скакнёт до сотен просто из-за того, что шедулинг не успевает переключать процессы; это чисто стохастический процесс. По достижению этого уровня (сотни сообщений в очереди) релеинг фактически останавливается, снижая скорость, причём до чудовищно низких величин (когда 3K, а когда и просто сотни сообщений в секунду). Далее срабатывал регулятор нагрузки в тесте и снижал её, но привести систему к исходному состоянию можно было только добившись снижения темпа (или полной остановки) до того, что очередь на релейном процессе падала почти до нуля. Тогда его скорость восстанавливалась.

CK>Это похоже на network congestion очень сильно. Без backpressure это никак не решить. Даже с разными очердями, рано или поздно, наступит момент когда без backpressure это перестанет работать.

Спасибо, кэп. Да, наступит. Вопрос в том, что именно будет, когда оно наступит.
Если зависимость времени обработки одного сообщения от длины очереди отсутствует или не выходит за фиксированные пределы, и можно говорить про O(1), регулировка работает беспроблемно. Например, мы устанавливаем, что сумма очередей в цепочке после входного процесса не превышает 20K, а каждый из этих процессов в цепочке шлёт обновление своего статуса на 1000 обработанных сообщений или раз в секунду. Тогда при превышении этого порога мы просто прекратим принимать, оно пойдёт рассасываться, и как только упадёт ниже порога — снова начнём приём. (В случае TCP, для этого можно, например, с момента видения превышения перестать обновлять active у gen_tcp порта, а для возобновления послать ему что-то вроде {active,10}.)
Если же описанное квадратичное — то будет картина, как я описал в предыдущем сообщении. N сообщений отрабатывается за время T, а 2*N — за 4*T. Если ожидаешь скорости N/T, нужно немедленно останавливать приём, как только очередь достигла N, или даже числа меньшего, чем N, потому что пока до приёмника извне дойдёт команда остановиться, он успеет ещё прислать неизвестно сколько. А вот насколько меньше — надо вычислять по свойствам реальной системы. И всё равно колебания потока приведут к тому, что зашкалы будут, и систему будет клинить...