root
/
rabbitmq-server
mirror of https://github.com/rabbitmq/rabbitmq-server.git
1
0
Fork 0
Code Issues Actions 3 Packages Projects Releases Wiki Activity

Introduce batching (again - same diff as 5f7d8d07f94f)

This commit is contained in:
Matthew Sackman 2011-03-21 17:57:54 +00:00
parent 0cd4e2f3dd
commit 4ba94e18c6
1 changed files with 91 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

132
src/gm.erl
View File

@ -376,15 +376,16 @@
confirmed_broadcast/2, group_members/1]). confirmed_broadcast/2, group_members/1]).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
code_change/3, prioritise_info/2]). code_change/3, prioritise_cast/2, prioritise_info/2]).
-export([behaviour_info/1]). -export([behaviour_info/1]).
-export([table_definitions/0]). -export([table_definitions/0, flush/1]).
-define(GROUP_TABLE, gm_group). -define(GROUP_TABLE, gm_group).
-define(HIBERNATE_AFTER_MIN, 1000). -define(HIBERNATE_AFTER_MIN, 1000).
-define(DESIRED_HIBERNATE, 10000). -define(DESIRED_HIBERNATE, 10000).
-define(BROADCAST_TIMER, 25).
-define(SETS, ordsets). -define(SETS, ordsets).
-define(DICT, orddict). -define(DICT, orddict).
@ -398,7 +399,9 @@
pub_count, pub_count,
members_state, members_state,
callback_args, callback_args,
confirms confirms,
broadcast_buffer,
broadcast_timer
}). }).
-record(gm_group, { name, version, members }). -record(gm_group, { name, version, members }).
@ -508,21 +511,26 @@ confirmed_broadcast(Server, Msg) ->
group_members(Server) -> group_members(Server) ->
gen_server2:call(Server, group_members, infinity). gen_server2:call(Server, group_members, infinity).
flush(Server) ->
gen_server2:cast(Server, flush).
init([GroupName, Module, Args]) -> init([GroupName, Module, Args]) ->
random:seed(now()), random:seed(now()),
gen_server2:cast(self(), join), gen_server2:cast(self(), join),
Self = self(), Self = self(),
{ok, #state { self = Self, {ok, #state { self = Self,
left = {Self, undefined}, left = {Self, undefined},
right = {Self, undefined}, right = {Self, undefined},
group_name = GroupName, group_name = GroupName,
module = Module, module = Module,
view = undefined, view = undefined,
pub_count = 0, pub_count = 0,
members_state = undefined, members_state = undefined,
callback_args = Args, callback_args = Args,
confirms = queue:new() }, hibernate, confirms = queue:new(),
broadcast_buffer = [],
broadcast_timer = undefined }, hibernate,
{backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}. {backoff, ?HIBERNATE_AFTER_MIN, ?HIBERNATE_AFTER_MIN, ?DESIRED_HIBERNATE}}.
@ -620,7 +628,11 @@ handle_cast(join, State = #state { self = Self,
{Module:joined(Args, all_known_members(View)), State1}); {Module:joined(Args, all_known_members(View)), State1});
handle_cast(leave, State) -> handle_cast(leave, State) ->
{stop, normal, State}. {stop, normal, State};
handle_cast(flush, State) ->
noreply(
flush_broadcast_buffer(State #state { broadcast_timer = undefined })).
handle_info({'DOWN', MRef, process, _Pid, _Reason}, handle_info({'DOWN', MRef, process, _Pid, _Reason},
@ -662,14 +674,17 @@ handle_info({'DOWN', MRef, process, _Pid, _Reason},
end. end.
terminate(Reason, #state { module = Module, terminate(Reason, State = #state { module = Module,
callback_args = Args }) -> callback_args = Args }) ->
flush_broadcast_buffer(State),
Module:terminate(Args, Reason). Module:terminate(Args, Reason).
code_change(_OldVsn, State, _Extra) -> code_change(_OldVsn, State, _Extra) ->
{ok, State}. {ok, State}.
prioritise_cast(flush, _State) -> 1;
prioritise_cast(_ , _State) -> 0.
prioritise_info({'DOWN', _MRef, process, _Pid, _Reason}, _State) -> 1; prioritise_info({'DOWN', _MRef, process, _Pid, _Reason}, _State) -> 1;
prioritise_info(_ , _State) -> 0. prioritise_info(_ , _State) -> 0.
@ -782,33 +797,62 @@ handle_msg({activity, _NotLeft, _Activity}, State) ->
noreply(State) -> noreply(State) ->
{noreply, State, hibernate}. {noreply, ensure_broadcast_timer(State), hibernate}.
reply(Reply, State) -> reply(Reply, State) ->
{reply, Reply, State, hibernate}. {reply, Reply, ensure_broadcast_timer(State), hibernate}.
internal_broadcast(Msg, From, State = #state { self = Self, ensure_broadcast_timer(State = #state { broadcast_buffer = [],
pub_count = PubCount, broadcast_timer = undefined }) ->
members_state = MembersState, State;
module = Module, ensure_broadcast_timer(State = #state { broadcast_buffer = [],
confirms = Confirms, broadcast_timer = TRef }) ->
callback_args = Args }) -> timer:cancel(TRef),
PubMsg = {PubCount, Msg}, State #state { broadcast_timer = undefined };
Activity = activity_cons(Self, [PubMsg], [], activity_nil()), ensure_broadcast_timer(State = #state { broadcast_timer = undefined }) ->
ok = maybe_send_activity(activity_finalise(Activity), State), {ok, TRef} = timer:apply_after(?BROADCAST_TIMER, ?MODULE, flush, [self()]),
MembersState1 = State #state { broadcast_timer = TRef };
with_member( ensure_broadcast_timer(State) ->
fun (Member = #member { pending_ack = PA }) -> State.
Member #member { pending_ack = queue:in(PubMsg, PA) }
end, Self, MembersState), internal_broadcast(Msg, From, State = #state { self = Self,
pub_count = PubCount,
module = Module,
confirms = Confirms,
callback_args = Args,
broadcast_buffer = Buffer }) ->
Result = Module:handle_msg(Args, Self, Msg),
Buffer1 = [{PubCount, Msg} | Buffer],
Confirms1 = case From of Confirms1 = case From of
none -> Confirms; none -> Confirms;
_ -> queue:in({PubCount, From}, Confirms) _ -> queue:in({PubCount, From}, Confirms)
end, end,
handle_callback_result({Module:handle_msg(Args, Self, Msg), State1 = State #state { pub_count = PubCount + 1,
State #state { pub_count = PubCount + 1, confirms = Confirms1,
members_state = MembersState1, broadcast_buffer = Buffer1 },
confirms = Confirms1 }}). case From =/= none of
true ->
handle_callback_result({Result, flush_broadcast_buffer(State1)});
false ->
handle_callback_result(
{Result, State1 #state { broadcast_buffer = Buffer1 }})
end.
flush_broadcast_buffer(State = #state { broadcast_buffer = [] }) ->
State;
flush_broadcast_buffer(State = #state { self = Self,
members_state = MembersState,
broadcast_buffer = Buffer }) ->
Pubs = lists:reverse(Buffer),
Activity = activity_cons(Self, Pubs, [], activity_nil()),
ok = maybe_send_activity(activity_finalise(Activity), State),
MembersState1 = with_member(
fun (Member = #member { pending_ack = PA }) ->
PA1 = queue:join(PA, queue:from_list(Pubs)),
Member #member { pending_ack = PA1 }
end, Self, MembersState),
State #state { members_state = MembersState1,
broadcast_buffer = [] }.
%% --------------------------------------------------------------------------- %% ---------------------------------------------------------------------------
@ -1093,16 +1137,22 @@ maybe_monitor(Self, Self) ->
maybe_monitor(Other, _Self) -> maybe_monitor(Other, _Self) ->
erlang:monitor(process, Other). erlang:monitor(process, Other).
check_neighbours(State = #state { self = Self, check_neighbours(State = #state { self = Self,
left = Left, left = Left,
right = Right, right = Right,
view = View }) -> view = View,
broadcast_buffer = Buffer }) ->
#view_member { left = VLeft, right = VRight } #view_member { left = VLeft, right = VRight }
= fetch_view_member(Self, View), = fetch_view_member(Self, View),
Ver = view_version(View), Ver = view_version(View),
Left1 = ensure_neighbour(Ver, Self, Left, VLeft), Left1 = ensure_neighbour(Ver, Self, Left, VLeft),
Right1 = ensure_neighbour(Ver, Self, Right, VRight), Right1 = ensure_neighbour(Ver, Self, Right, VRight),
State1 = State #state { left = Left1, right = Right1 }, Buffer1 = case Right1 of
{Self, undefined} -> [];
_ -> Buffer
end,
State1 = State #state { left = Left1, right = Right1,
broadcast_buffer = Buffer1 },
ok = maybe_send_catchup(Right, State1), ok = maybe_send_catchup(Right, State1),
State1. State1.