Есть ли способ сделать асинхронный вызов gen_server в Erlang?

Например. предположим, у меня есть модуль, реализующий поведение gen_server, и он имеет

handle_call({foo, Foo}, _From, State) ->
  {reply, result(Foo), State}
;

Я могу связаться с этим обработчиком, выполнив gen_server:call(Server, {foo, Foo}) из какого-то другого процесса (думаю, если gen_server попытается выполнить gen_server:call сам, он заблокируется). Но gen_server:call блокируется при ответе (или тайм-ауте). Что делать, если я не хочу блокировать ответ?

Воображаемый вариант использования: Предположим, у меня есть 5 таких gen_servers, и мне достаточно ответа от любых 2 из них. Я хочу сделать что-то вроде этого:

OnResponse -> fun(Response) ->
  % blah
end,
lists:foreach(
  fun(S) ->
    gen_server:async_call(S, {foo, Foo}, OnResponse)
  end, 
  Servers),
Result = wait_for_two_responses(Timeout),
lol_i_dunno()

Я знаю, что gen_server выполнил приведение, но приведение не может предоставить какой-либо ответ, поэтому я не думаю, что это то, что мне нужно в данном случае. Кроме того, похоже, что gen_server не должно беспокоить, хочет ли вызывающий абонент обрабатывать ответ синхронно (используя gen_server:call) или асинхронно (похоже, не существует?).

Кроме того, серверу разрешено предоставлять ответ асинхронно, если handle_call возвращает no_reply, а затем вызывает gen_server:reply. Так почему бы не поддерживать асинхронную обработку ответа на другой стороне? Или это есть, но я просто не могу найти??


erlang otp
person allyourcode    schedule 17.02.2015    source источник
comment
не для этого ли предназначен handle_cast?   -  person CoderDennis    schedule 17.02.2015
comment
О, я вижу, вы упомянули приведение в вопросе, но это асинхронный способ. Если другому процессу нужен ответ, то сервер должен отправить еще одно асинхронное сообщение.   -  person CoderDennis    schedule 17.02.2015
comment
Если вы собираетесь отправлять ответ асинхронно через handle_cast, то кажется, что весь gen_server бесполезен, потому что в конечном итоге вам нужно отправить отправителя в бросках. handle_call не должен беспокоиться о том, блокирует ли другой процесс ответ или нет. Я имею в виду, что я мог бы запустить новый процесс, чтобы превратить синхронные вызовы в асинхронные, но это кажется расточительным.   -  person allyourcode    schedule 18.02.2015
comment
Я не думаю, что вам нужен новый процесс, чтобы сделать его асинхронным. Похоже, вы уже передаете функцию обратного вызова. У меня недостаточно опыта работы с erlang, чтобы быть полезным, но я знаю, что handle_call для синхронных методов, а handle_cast для асинхронных. Отправка сообщения обратно отправителю — идеальный способ асинхронного выполнения. Не борись с этим.   -  person CoderDennis    schedule 18.02.2015

Ответы (4)


arrow_upward
2
arrow_downward

gen_server:call в основном представляет собой последовательность

send a message to the server (with identifier)
wait for the response of that particular message

завернутый в одну функцию.

для вашего примера вы можете разложить поведение на 2 этапа: цикл, который использует gen_server:cast(Server,{Message,UniqueID,self()} со всеми серверами, а затем цикл приема, который ожидает как минимум 2 ответа в форме {UniqueID,Answer}. Но вы должны позаботиться о том, чтобы в какой-то момент очистить свой почтовый ящик. Лучшим решением должно быть делегирование этого отдельному процессу, который просто умрет, когда получит необходимое количество ответов:

[edit] внесите некоторые исправления в код, теперь он должен работать :o)

get_n_answers(Msg,ServerList,N) when N =< length(ServerList) ->
    spawn(?MODULE,get_n_answers,[Msg,ServerList,N,[],self()]).

get_n_answers(_Msg,[],0,Rep,Pid) -> 
    Pid ! {Pid,Rep};
get_n_answers(_Msg,[],N,Rep,Pid) -> 
    NewRep = receive
        Answ -> [Answ|Rep]
    end,
    get_n_answers(_Msg,[],N-1,NewRep,Pid);
get_n_answers(Msg,[H|T],N,Rep,Pid) -> 
    %gen_server:cast(H,{Msg,Pid}),
    H ! {Msg,self()},
    get_n_answers(Msg,T,N,Rep,Pid).

и вы можете использовать его так:

ID = get_n_answers(Msg,ServerList,2),
% insert some code here
Answer = receive
    {ID,A} -> A % tagged with ID to do not catch another message in the mailbox
end
person Pascal    schedule 18.02.2015

arrow_upward
2
arrow_downward

Вы можете легко реализовать это, отправив каждый вызов в отдельный процесс и ожидая ответов от необходимого количества (по сути, это то, что касается асинхронности, не так ли? :-)

Взгляните на эту простую реализацию параллельного вызова, которая основан на async_call из rpc библиотека в OTP.

Вот как это работает на простом английском языке.

  • Вам нужно сделать 5 вызовов, чтобы (в родительском процессе) вы породили 5 дочерних процессов Erlang.
  • Каждый процесс отправляет обратно родительскому процессу кортеж, содержащий его PID и результат вызова.
  • Кортеж можно построить и отправить обратно только после того, как желаемый вызов будет завершен.
  • В родительском процессе вы перебираете ответы в цикле приема.
  • Можно дождаться всех ответов или только 2 или 3 из начатых 5.

Родительский процесс (который порождает рабочие процессы) в конечном итоге получит все ответы (я имею в виду те, которые вы хотите игнорировать). Вам нужен способ отбросить их, если вы не хотите, чтобы очередь сообщений росла бесконечно. Есть два варианта:

  1. Сам родительский процесс может быть временным процессом, созданным только для вызова других 5 дочерних процессов. Как только желаемое количество ответов будет собрано, он может отправить ответ вызывающему абоненту и умереть. Сообщения, отправленные умершему процессу, будут отброшены.

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

person Greg    schedule 10.02.2016

arrow_upward
0
arrow_downward

gen_server не имеют концепции асинхронных вызовов на стороне клиента. Нетривиально, как реализовать последовательно, потому что gen_server:call представляет собой комбинацию монитора для серверного процесса, отправки сообщения запроса и ожидания либо ответа, либо монитора, либо тайм-аута. Если вы сделаете что-то вроде того, что вы упомянули, вам нужно будет каким-то образом обрабатывать сообщения DOWN с сервера ... поэтому гипотетический async_call должен возвращать некоторый ключ для yeld, а также внутреннюю ссылку на монитор для случая, когда вы обрабатываете сообщения DONW из других процессов. .. и не хочу смешивать это с желтыми ошибками.

Не очень хорошая, но возможная альтернатива - использовать rpc:async_call(gen_server, call, [....])

Но у этого подхода есть ограничение на процесс вызова, который будет недолговечным rex дочерним элементом, поэтому, если ваш gen-сервер использует pid вызывающего абонента каким-либо образом, кроме его отправки, логика ответа будет нарушена.

person Andrey Ivanov    schedule 17.02.2015

arrow_upward
0
arrow_downward

gen_sever:call к самому процессу наверняка заблокируется до истечения времени ожидания. Чтобы понять причину, нужно знать тот факт, что gen_server фреймворк фактически объединяет ваш конкретный код в один модуль, а gen_server:call будет "переводится" как форма "pid ! Msg".

Итак, представьте, как этот блок кода действует, процесс фактически остается в цикле, продолжая получать сообщения, и когда поток управления сталкивается с функцией обработки, процесс получения временно прерывается, поэтому, если вы вызываете gen_server:call к самому процессу, поскольку это синхронная функция, она ожидает ответа, который, однако, никогда не придет, пока функция обработки не вернется, чтобы процесс мог продолжать получать сообщения, поэтому код находится в тупике.

person nordicwing    schedule 28.01.2016