You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by rn...@apache.org on 2014/08/28 14:23:11 UTC
[01/33] rexi commit: updated refs/heads/master to bbf59a2
Repository: couchdb-rexi
Updated Branches:
refs/heads/master 6cd3b812c -> bbf59a217
[1/3] Start a rexi_server per remote node
This is part of a multi-release upgrade to switch to using a rexi_server
instance per remote node. The first commit introduces the new
rexi_server instances, the second switches to using the new instances
and the third will remove the singleton rexi_server. Each of these
commits should be in a separate release.
The new pattern introduced by this commit will start a 'rexi_server_%b'
process where the '%b' is replaced by the `erlang:phash2(Node)` for
which it will service requests. After the second commit is release each
rexi generated message will be sent to the rexi_server instance on the
remote node handling requests for the local node. The
`rexi_utils:server_pid/1` function is used to generate the id of the
remote server.
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/f5907c4b
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/f5907c4b
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/f5907c4b
Branch: refs/heads/master
Commit: f5907c4b89456d3120a4d1e1080d9203e6522816
Parents: 6cd3b81
Author: Paul J. Davis <pa...@gmail.com>
Authored: Mon Mar 18 22:51:29 2013 -0500
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:56:17 2014 +0100
----------------------------------------------------------------------
src/rexi.erl | 12 -----
src/rexi_server.erl | 6 +--
src/rexi_server_mon.erl | 123 +++++++++++++++++++++++++++++++++++++++++++
src/rexi_server_sup.erl | 29 ++++++++++
src/rexi_sup.erl | 42 ++++++++++++---
src/rexi_utils.erl | 10 +++-
6 files changed, 199 insertions(+), 23 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/f5907c4b/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index 8e53dba..a4c786d 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -15,7 +15,6 @@
-export([cast/2, cast/3, cast/4, kill/2]).
-export([reply/1, sync_reply/1, sync_reply/2]).
-export([async_server_call/2, async_server_call/3]).
--export([get_errors/0, get_last_error/0, set_error_limit/1]).
-export([stream/1, stream/2, stream/3, stream_ack/1, stream_ack/2]).
-include_lib("rexi/include/rexi.hrl").
@@ -31,17 +30,6 @@ stop() ->
restart() ->
stop(), start().
--spec get_errors() -> {ok, [#error{}]}.
-get_errors() ->
- gen_server:call(?SERVER, get_errors).
-
--spec get_last_error() -> {ok, #error{}} | {error, empty}.
-get_last_error() ->
- gen_server:call(?SERVER, get_last_error).
-
--spec set_error_limit(pos_integer()) -> ok.
-set_error_limit(N) when is_integer(N), N > 0 ->
- gen_server:call(?SERVER, {set_error_limit, N}).
%% @equiv cast(Node, self(), MFA)
-spec cast(node(), {atom(), atom(), list()}) -> reference().
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/f5907c4b/src/rexi_server.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server.erl b/src/rexi_server.erl
index b2f6de0..49661cc 100644
--- a/src/rexi_server.erl
+++ b/src/rexi_server.erl
@@ -15,7 +15,7 @@
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
code_change/3]).
--export([start_link/0, init_p/2, init_p/3]).
+-export([start_link/1, init_p/2, init_p/3]).
-include_lib("rexi/include/rexi.hrl").
-include_lib("eunit/include/eunit.hrl").
@@ -35,8 +35,8 @@
error_count = 0
}).
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+start_link(ServerId) ->
+ gen_server:start_link({local, ServerId}, ?MODULE, [], []).
init([]) ->
{ok, #st{}}.
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/f5907c4b/src/rexi_server_mon.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server_mon.erl b/src/rexi_server_mon.erl
new file mode 100644
index 0000000..3545dba
--- /dev/null
+++ b/src/rexi_server_mon.erl
@@ -0,0 +1,123 @@
+% Copyright 2010 Cloudant
+%
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(rexi_server_mon).
+-behaviour(gen_server).
+
+
+-export([
+ start_link/0,
+ status/0
+]).
+
+
+-export([
+ init/1,
+ terminate/2,
+ handle_call/3,
+ handle_cast/2,
+ handle_info/2,
+ code_change/3
+]).
+
+
+-define(SUP_MODULE, rexi_server_sup).
+-define(CHILD_MODULE, rexi_server).
+-define(INTERVAL, 60000).
+
+
+start_link() ->
+ gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+
+status() ->
+ gen_server:call(?MODULE, status).
+
+
+init([]) ->
+ net_kernel:monitor_nodes(true),
+ erlang:send_after(?INTERVAL, self(), check_nodes),
+ {ok, nil}.
+
+
+terminate(_Reason, _St) ->
+ ok.
+
+
+handle_call(status, _From, St) ->
+ case missing_servers() of
+ [] ->
+ {reply, ok, St};
+ Missing ->
+ {reply, {waiting, length(Missing)}, St}
+ end;
+
+handle_call(Msg, _From, St) ->
+ twig:log(notice, "~s ignored_call ~w", [?MODULE, Msg]),
+ {reply, ignored, St}.
+
+
+handle_cast(Msg, St) ->
+ twig:log(notice, "~s ignored_cast ~w", [?MODULE, Msg]),
+ {noreply, St}.
+
+
+handle_info({nodeup, _}, St) ->
+ start_rexi_servers(),
+ {noreply, St};
+
+handle_info({nodedown, _}, St) ->
+ {noreply, St};
+
+handle_info(check_nodes, St) ->
+ start_rexi_servers(),
+ erlang:send_after(?INTERVAL, self(), check_nodes),
+ {noreply, St};
+
+handle_info(Msg, St) ->
+ twig:log(notice, "~s ignored_info ~w", [?MODULE, Msg]),
+ {noreply, St}.
+
+
+code_change(_OldVsn, St, _Extra) ->
+ {ok, St}.
+
+
+start_rexi_servers() ->
+ lists:foreach(fun(Id) ->
+ {ok, _} = start_rexi_server(Id)
+ end, missing_servers()).
+
+
+missing_servers() ->
+ ServerIds = [rexi_utils:server_id(Node) || Node <- [node() | nodes()]],
+ ChildIds = [Id || {Id, _, _, _} <- supervisor:which_children(?SUP_MODULE)],
+ ServerIds -- ChildIds.
+
+
+start_rexi_server(ChildId) ->
+ ChildSpec = {
+ ChildId,
+ {rexi_server, start_link, [ChildId]},
+ permanent,
+ brutal_kill,
+ worker,
+ [?CHILD_MODULE]
+ },
+ case supervisor:start_child(?SUP_MODULE, ChildSpec) of
+ {ok, Pid} ->
+ {ok, Pid};
+ Else ->
+ erlang:error(Else)
+ end.
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/f5907c4b/src/rexi_server_sup.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server_sup.erl b/src/rexi_server_sup.erl
new file mode 100644
index 0000000..97b5b9e
--- /dev/null
+++ b/src/rexi_server_sup.erl
@@ -0,0 +1,29 @@
+% Copyright 2010 Cloudant
+%
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+
+-module(rexi_server_sup).
+-behaviour(supervisor).
+
+
+-export([init/1]).
+
+-export([start_link/0]).
+
+
+start_link() ->
+ supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+
+
+init([]) ->
+ {ok, {{one_for_one, 1, 1}, []}}.
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/f5907c4b/src/rexi_sup.erl
----------------------------------------------------------------------
diff --git a/src/rexi_sup.erl b/src/rexi_sup.erl
index a8aa800..1c13eea 100644
--- a/src/rexi_sup.erl
+++ b/src/rexi_sup.erl
@@ -12,17 +12,45 @@
-module(rexi_sup).
-behaviour(supervisor).
--export([init/1]).
-export([start_link/1]).
-
--include_lib("eunit/include/eunit.hrl").
-
-%% Helper macro for declaring children of supervisor
--define(CHILD(I, Type), {I, {I, start_link, []}, permanent, 100, Type, [I]}).
+-export([init/1]).
start_link(Args) ->
supervisor:start_link({local,?MODULE}, ?MODULE, Args).
init([]) ->
- {ok, {{one_for_one, 3, 10}, [?CHILD(rexi_gov_manager, worker), ?CHILD(rexi_server, worker)]}}.
+ {ok, {{one_for_one, 3, 10}, [
+ {
+ rexi_gov_manager,
+ {rexi_gov_manager, start_link, []},
+ permanent,
+ 100,
+ worker,
+ [rexi_gov_manager]
+ },
+ {
+ rexi_server,
+ {rexi_server, start_link, [rexi_server]},
+ permanent,
+ 100,
+ worker,
+ [rexi_server]
+ },
+ {
+ rexi_server_sup,
+ {rexi_server_sup, start_link, []},
+ permanent,
+ 100,
+ supervisor,
+ [rexi_server_sup]
+ },
+ {
+ rexi_server_mon,
+ {rexi_server_mon, start_link, []},
+ permanent,
+ 100,
+ worker,
+ [rexi_server_mon]
+ }
+ ]}}.
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/f5907c4b/src/rexi_utils.erl
----------------------------------------------------------------------
diff --git a/src/rexi_utils.erl b/src/rexi_utils.erl
index 1b11576..8bc1190 100644
--- a/src/rexi_utils.erl
+++ b/src/rexi_utils.erl
@@ -12,7 +12,15 @@
-module(rexi_utils).
--export([send/2, recv/6]).
+-export([server_id/1, server_pid/1, send/2, recv/6]).
+
+%% @doc Return a rexi_server id for the given node.
+server_id(Node) ->
+ list_to_atom("rexi_server_" ++ integer_to_list(erlang:phash2(Node))).
+
+%% @doc Return a {server_id(node()), Node} Pid name for the given Node.
+server_pid(Node) ->
+ {server_id(node()), Node}.
%% @doc send a message as quickly as possible
send(Dest, Msg) ->
[15/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Start a supervised rexi_governor per node
This generalizes rexi_server_mon to start per-node versions of a server
specified by a child module.
BugzID: 23717
BugzID: 23718
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/ffb6dade
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/ffb6dade
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/ffb6dade
Branch: refs/heads/master
Commit: ffb6dade31e6d131a1db582d376bd8f20c7d935e
Parents: 1030906
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Wed Oct 16 10:42:15 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:02:58 2014 +0100
----------------------------------------------------------------------
src/rexi_server_mon.erl | 61 ++++++++++++++++++++++++--------------------
src/rexi_server_sup.erl | 6 ++---
src/rexi_sup.erl | 20 +++++++++++++--
3 files changed, 54 insertions(+), 33 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/ffb6dade/src/rexi_server_mon.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server_mon.erl b/src/rexi_server_mon.erl
index 756ab95..205ef4a 100644
--- a/src/rexi_server_mon.erl
+++ b/src/rexi_server_mon.erl
@@ -1,4 +1,4 @@
-% Copyright 2010 Cloudant
+% Copyright 2010-2013 Cloudant
%
% Licensed under the Apache License, Version 2.0 (the "License"); you may not
% use this file except in compliance with the License. You may obtain a copy of
@@ -18,7 +18,7 @@
-export([
- start_link/0,
+ start_link/1,
status/0
]).
@@ -33,35 +33,34 @@
]).
--define(SUP_MODULE, rexi_server_sup).
--define(CHILD_MODULE, rexi_server).
-define(INTERVAL, 60000).
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+start_link(ChildMod) ->
+ Name = list_to_atom(lists:concat([ChildMod, "_mon"])),
+ gen_server:start_link({local, Name}, ?MODULE, ChildMod, []).
status() ->
gen_server:call(?MODULE, status).
-init([]) ->
+init(ChildMod) ->
net_kernel:monitor_nodes(true),
erlang:send_after(?INTERVAL, self(), check_nodes),
- {ok, nil}.
+ {ok, ChildMod}.
terminate(_Reason, _St) ->
ok.
-handle_call(status, _From, St) ->
- case missing_servers() of
+handle_call(status, _From, ChildMod) ->
+ case missing_servers(ChildMod) of
[] ->
- {reply, ok, St};
+ {reply, ok, ChildMod};
Missing ->
- {reply, {waiting, length(Missing)}, St}
+ {reply, {waiting, length(Missing)}, ChildMod}
end;
handle_call(Msg, _From, St) ->
@@ -74,52 +73,58 @@ handle_cast(Msg, St) ->
{noreply, St}.
-handle_info({nodeup, _}, St) ->
- start_rexi_servers(),
- {noreply, St};
+handle_info({nodeup, _}, ChildMod) ->
+ start_servers(ChildMod),
+ {noreply, ChildMod};
handle_info({nodedown, _}, St) ->
{noreply, St};
-handle_info(check_nodes, St) ->
- start_rexi_servers(),
+handle_info(check_nodes, ChildMod) ->
+ start_servers(ChildMod),
erlang:send_after(?INTERVAL, self(), check_nodes),
- {noreply, St};
+ {noreply, ChildMod};
handle_info(Msg, St) ->
twig:log(notice, "~s ignored_info ~w", [?MODULE, Msg]),
{noreply, St}.
+code_change(_OldVsn, nil, _Extra) ->
+ {ok, rexi_server};
code_change(_OldVsn, St, _Extra) ->
{ok, St}.
-start_rexi_servers() ->
+start_servers(ChildMod) ->
lists:foreach(fun(Id) ->
- {ok, _} = start_rexi_server(Id)
- end, missing_servers()).
+ {ok, _} = start_server(ChildMod, Id)
+ end, missing_servers(ChildMod)).
-missing_servers() ->
- ServerIds = [list_to_atom("rexi_server_" ++ atom_to_list(Node))
+missing_servers(ChildMod) ->
+ ServerIds = [list_to_atom(lists:concat([ChildMod, "_", Node]))
|| Node <- [node() | nodes()]],
- ChildIds = [Id || {Id, _, _, _} <- supervisor:which_children(?SUP_MODULE)],
+ SupModule = sup_module(ChildMod),
+ ChildIds = [Id || {Id, _, _, _} <- supervisor:which_children(SupModule)],
ServerIds -- ChildIds.
-start_rexi_server(ChildId) ->
+start_server(ChildMod, ChildId) ->
ChildSpec = {
ChildId,
- {rexi_server, start_link, [ChildId]},
+ {ChildMod, start_link, [ChildId]},
permanent,
brutal_kill,
worker,
- [?CHILD_MODULE]
+ [ChildMod]
},
- case supervisor:start_child(?SUP_MODULE, ChildSpec) of
+ case supervisor:start_child(sup_module(ChildMod), ChildSpec) of
{ok, Pid} ->
{ok, Pid};
Else ->
erlang:error(Else)
end.
+
+sup_module(ChildMod) ->
+ list_to_atom(lists:concat([ChildMod, "_sup"])).
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/ffb6dade/src/rexi_server_sup.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server_sup.erl b/src/rexi_server_sup.erl
index 97b5b9e..29c6ad6 100644
--- a/src/rexi_server_sup.erl
+++ b/src/rexi_server_sup.erl
@@ -18,11 +18,11 @@
-export([init/1]).
--export([start_link/0]).
+-export([start_link/1]).
-start_link() ->
- supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+start_link(Name) ->
+ supervisor:start_link({local, Name}, ?MODULE, []).
init([]) ->
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/ffb6dade/src/rexi_sup.erl
----------------------------------------------------------------------
diff --git a/src/rexi_sup.erl b/src/rexi_sup.erl
index 1c13eea..8ee116f 100644
--- a/src/rexi_sup.erl
+++ b/src/rexi_sup.erl
@@ -39,7 +39,7 @@ init([]) ->
},
{
rexi_server_sup,
- {rexi_server_sup, start_link, []},
+ {rexi_server_sup, start_link, [rexi_server_sup]},
permanent,
100,
supervisor,
@@ -47,7 +47,23 @@ init([]) ->
},
{
rexi_server_mon,
- {rexi_server_mon, start_link, []},
+ {rexi_server_mon, start_link, [rexi_server]},
+ permanent,
+ 100,
+ worker,
+ [rexi_server_mon]
+ },
+ {
+ rexi_governor_sup,
+ {rexi_server_sup, start_link, [rexi_governor_sup]},
+ permanent,
+ 100,
+ supervisor,
+ [rexi_server_sup]
+ },
+ {
+ rexi_governor_mon,
+ {rexi_server_mon, start_link, [rexi_governor]},
permanent,
100,
worker,
[25/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Fix counting bug in buffer
Quoting @davisp:
There's a bug in rexi_buffer that can lead to the counter in its state
running negative. I noticed this on malort looking for memory usage. A
quick reading of the code suggests its due to us getting a timeout
message with an empty queue. Theoretically this could happen if we've
exceeded the MAX_MEMORY threshold when sending a message or even with
just grabbing the buffered count when its idle.
BugzID: 28049
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/45510149
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/45510149
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/45510149
Branch: refs/heads/master
Commit: 455101492947939cb7767c6bd888baf2e33ad497
Parents: cd07cb8
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Tue Jun 3 14:28:37 2014 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:08:14 2014 +0100
----------------------------------------------------------------------
src/rexi_buffer.erl | 46 +++++++++++++++++-----------------------------
1 file changed, 17 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/45510149/src/rexi_buffer.erl
----------------------------------------------------------------------
diff --git a/src/rexi_buffer.erl b/src/rexi_buffer.erl
index f75399c..26f3c97 100644
--- a/src/rexi_buffer.erl
+++ b/src/rexi_buffer.erl
@@ -58,37 +58,25 @@ handle_cast({deliver, Dest, Msg}, #state{buffer = Q, count = C} = State) ->
{noreply, State#state{buffer = Q2, count = C+1}, 0}
end.
-handle_info(timeout, #state{sender = nil} = State) ->
+handle_info(timeout, #state{sender = nil, buffer = {[],[]}, count = 0}=State) ->
+ {noreply, State};
+handle_info(timeout, #state{sender = nil, count = C} = State) when C > 0 ->
#state{buffer = Q, count = C} = State,
- Sender = case queue:out_r(Q) of
- {{value, {Dest, Msg}}, Q2} ->
- case erlang:send(Dest, Msg, [noconnect, nosuspend]) of
- ok ->
- nil;
- _Else ->
- spawn_monitor(erlang, send, [Dest, Msg])
- end;
- {empty, Q2} ->
- nil
- end,
- if Sender =:= nil, C > 1 ->
- {noreply, State#state{buffer = Q2, count = C-1}, 0};
- true ->
- NewState = State#state{buffer = Q2, sender = Sender, count = C-1},
- % When Sender is nil and C-1 == 0 we're reverting to an
- % idle state with no outstanding or queued messages. We'll
- % use this oppurtunity to hibernate this process and
- % run a garbage collection.
- case {Sender, C-1} of
- {nil, 0} ->
- {noreply, NewState, hibernate};
- _ ->
- {noreply, NewState, infinity}
- end
+ {{value, {Dest, Msg}}, Q2} = queue:out_r(Q),
+ NewState = State#state{buffer = Q2, count = C-1},
+ case erlang:send(Dest, Msg, [noconnect, nosuspend]) of
+ ok when C =:= 1 ->
+ % We just sent the last queued messsage, we'll use this opportunity
+ % to hibernate the process and run a garbage collection
+ {noreply, NewState, hibernate};
+ ok when C > 1 ->
+ % Use a zero timeout to recurse into this handler ASAP
+ {noreply, NewState, 0};
+ _Else ->
+ % We're experiencing delays, keep buffering internally
+ Sender = spawn_monitor(erlang, send, [Dest, Msg]),
+ {noreply, NewState#state{sender = Sender}}
end;
-handle_info(timeout, State) ->
- % Waiting on a sender to return
- {noreply, State};
handle_info({'DOWN', Ref, _, Pid, _}, #state{sender = {Pid, Ref}} = State) ->
{noreply, State#state{sender = nil}, 0}.
[21/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Removed upgrade statement in rexi:kill/2
Removing cruft we no longer need that was sending an extra useless
message.
BugzId: 27671
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/e271ae46
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/e271ae46
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/e271ae46
Branch: refs/heads/master
Commit: e271ae4666d908a8cb4b33002b46f9da0f3611e6
Parents: f938055
Author: Paul J. Davis <pa...@gmail.com>
Authored: Thu Jan 30 21:31:11 2014 -0600
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:06:14 2014 +0100
----------------------------------------------------------------------
src/rexi.erl | 3 ---
1 file changed, 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/e271ae46/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index b0ad8c1..b38c149 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -72,9 +72,6 @@ cast(Node, Caller, MFA, Options) ->
%% No rexi_EXIT message will be sent.
-spec kill(node(), reference()) -> ok.
kill(Node, Ref) ->
- % This first version is to tide us over during the upgrade. We'll
- % remove it in the next commit that will be in a separate release.
- rexi_utils:send({rexi_server, Node}, cast_msg({kill, Ref})),
rexi_utils:send(rexi_utils:server_pid(Node), cast_msg({kill, Ref})),
ok.
[29/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Remove vestigal call to margaret
We'll re-add the metric when couch_stats is merged.
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/6ba87011
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/6ba87011
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/6ba87011
Branch: refs/heads/master
Commit: 6ba870119fcf92a42b60086fe557e2aff0598037
Parents: 535941d
Author: Paul J. Davis <pa...@gmail.com>
Authored: Sun Aug 17 15:25:20 2014 -0500
Committer: Paul J. Davis <pa...@gmail.com>
Committed: Sun Aug 17 15:25:20 2014 -0500
----------------------------------------------------------------------
src/rexi_buffer.erl | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/6ba87011/src/rexi_buffer.erl
----------------------------------------------------------------------
diff --git a/src/rexi_buffer.erl b/src/rexi_buffer.erl
index 0ba961a..2be2322 100644
--- a/src/rexi_buffer.erl
+++ b/src/rexi_buffer.erl
@@ -50,7 +50,6 @@ handle_call(get_buffered_count, _From, State) ->
{reply, State#state.count, State, 0}.
handle_cast({deliver, Dest, Msg}, #state{buffer = Q, count = C} = State) ->
- margaret_counter:increment([erlang, rexi, buffered]),
Q2 = queue:in({Dest, Msg}, Q),
case should_drop(State) of
true ->
[03/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Add a temporary feature flag for server-per-node
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/752921be
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/752921be
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/752921be
Branch: refs/heads/master
Commit: 752921bea5d166e1c2f7d4c09985fa974740842d
Parents: 837f045
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Thu Apr 11 08:02:41 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:57:00 2014 +0100
----------------------------------------------------------------------
src/rexi_utils.erl | 7 ++++++-
1 file changed, 6 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/752921be/src/rexi_utils.erl
----------------------------------------------------------------------
diff --git a/src/rexi_utils.erl b/src/rexi_utils.erl
index 8bc1190..b87f2c3 100644
--- a/src/rexi_utils.erl
+++ b/src/rexi_utils.erl
@@ -16,7 +16,12 @@
%% @doc Return a rexi_server id for the given node.
server_id(Node) ->
- list_to_atom("rexi_server_" ++ integer_to_list(erlang:phash2(Node))).
+ case config:get("rexi", "server_per_node", "false") of
+ "true" ->
+ list_to_atom("rexi_server_" ++ integer_to_list(erlang:phash2(Node)));
+ _ ->
+ rexi_server
+ end.
%% @doc Return a {server_id(node()), Node} Pid name for the given Node.
server_pid(Node) ->
[14/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Rejigger the governor implementation
Previously we'd spawn a number of processes to send messages to
'noconnect' / 'nosuspend' nodes. Now we're buffering the messages that
need to be sent directly in each governor and sending them one at a
time. This prevents the net_kernel from tipping over in the noconnect
case. We also decide whether to drop messages based on memory
consumption in the node instead of process limits (since we're not
spawning anymore).
BugzID: 23717
BugzID: 23718
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/1030906b
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/1030906b
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/1030906b
Branch: refs/heads/master
Commit: 1030906b85de7a398baf6363b9d3f6f21e93b257
Parents: 11b3859
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Tue Oct 15 22:46:52 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:02:32 2014 +0100
----------------------------------------------------------------------
src/rexi_governor.erl | 97 ++++++++++++++++++++++++++--------------------
src/rexi_utils.erl | 8 +++-
2 files changed, 62 insertions(+), 43 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/1030906b/src/rexi_governor.erl
----------------------------------------------------------------------
diff --git a/src/rexi_governor.erl b/src/rexi_governor.erl
index 876165d..fdf8c93 100644
--- a/src/rexi_governor.erl
+++ b/src/rexi_governor.erl
@@ -18,54 +18,67 @@
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
--record(state, {pids = ets:new(pids, [set]),
- spawn_max = 10000,
- spawn_cnt = 0,
- drop_cnt = 0}).
-
-init([PidSpawnMax]) ->
- {ok, #state{spawn_max = PidSpawnMax}}.
-
-handle_call(_Request, _From, State) ->
- Reply = ok,
- {reply, Reply, State}.
-
-handle_cast({spawn_and_track, Dest, Msg},
- #state{pids = Pids,
- spawn_max = SpawnMax,
- spawn_cnt = SC,
- drop_cnt = DC} = State) ->
- {NewSC, NewDC} =
- case ets:info(Pids, size) < SpawnMax of
- true ->
- {Pid, Ref} = spawn_monitor(erlang, send, [Dest, Msg]),
- ets:insert(Pids, {Pid, Ref}),
- {SC + 1, DC};
- false ->
- % drop message on floor
- {SC, DC + 1}
- end,
- {noreply, State#state{spawn_cnt = NewSC, drop_cnt = NewDC}};
+-export ([
+ send/2,
+ start_link/1
+]).
+
+-record(state, {
+ buffer = queue:new(),
+ count = 0
+}).
-handle_cast(nodeout, #state{pids = Pids} = State) ->
- % kill all the pids
- ets:foldl(fun({P, _Ref}, Acc) ->
- exit(P, kill),
- Acc
- end, [], Pids),
- ets:delete_all_objects(Pids),
- {noreply, State}.
+%% TODO Leverage os_mon to discover available memory in the system
+-define (MAX_MEMORY, 17179869184).
-handle_info({'DOWN', _, process, Pid, normal},
- #state{pids = Pids} = State) ->
- ets:delete(Pids, Pid),
- {noreply, State};
+start_link(ServerId) ->
+ gen_server:start_link({local, ServerId}, ?MODULE, nil, []).
-handle_info({'DOWN', _, process, _Pid, killed}, State) ->
- {noreply, State}.
+send(Dest, Msg) ->
+ Server = list_to_atom(lists:concat([rexi_governor, "_", get_node(Dest)])),
+ gen_server:cast(Server, {deliver, Dest, Msg}).
+
+
+init(_) ->
+ {ok, #state{}}.
+
+handle_call(get_buffered_count, _From, State) ->
+ {reply, State#state.count, State, 0}.
+
+handle_cast({deliver, Dest, Msg}, #state{buffer = Q, count = C} = State) ->
+ margaret_counter:increment([erlang, rexi, buffered]),
+ Q2 = queue:in({Dest, Msg}, Q),
+ case should_drop() of
+ true ->
+ {noreply, State#state{buffer = queue:drop(Q2)}, 0};
+ false ->
+ {noreply, State#state{buffer = Q2, count = C+1}, 0}
+ end.
+
+handle_info(timeout, State) ->
+ #state{buffer = Q, count = C} = State,
+ case queue:out_r(Q) of
+ {{value, {Dest, Msg}}, Q2} ->
+ erlang:send(Dest, Msg);
+ {empty, Q2} ->
+ ok
+ end,
+ if C > 1 ->
+ {noreply, State#state{buffer = Q2, count = C-1}, 0};
+ true ->
+ {noreply, State#state{buffer = Q2, count = 0}}
+ end.
terminate(_Reason, _State) ->
ok.
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
+
+should_drop() ->
+ erlang:memory(total) > ?MAX_MEMORY.
+
+get_node({_, Node}) when is_atom(Node) ->
+ Node;
+get_node(Pid) when is_pid(Pid) ->
+ node(Pid).
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/1030906b/src/rexi_utils.erl
----------------------------------------------------------------------
diff --git a/src/rexi_utils.erl b/src/rexi_utils.erl
index 6e03757..3c89ca9 100644
--- a/src/rexi_utils.erl
+++ b/src/rexi_utils.erl
@@ -29,7 +29,13 @@ server_pid(Node) ->
%% @doc send a message as quickly as possible
send(Dest, Msg) ->
- rexi_gov_manager:send(Dest, Msg).
+ case erlang:send(Dest, Msg, [noconnect, nosuspend]) of
+ ok ->
+ ok;
+ _ ->
+ % treat nosuspend and noconnect the same
+ rexi_governor:send(Dest, Msg)
+ end.
%% @doc set up the receive loop with an overall timeout
-spec recv([any()], integer(), function(), any(), timeout(), timeout()) ->
[13/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Remove old code_change, set module version to 1
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/11b3859a
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/11b3859a
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/11b3859a
Branch: refs/heads/master
Commit: 11b3859aae49c1a575cd70e401c9add9e578ca6b
Parents: cae29fe
Author: Robert Newson <ro...@cloudant.com>
Authored: Fri Nov 22 16:40:23 2013 +0000
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:58:42 2014 +0100
----------------------------------------------------------------------
src/rexi_gov_manager.erl | 1 +
src/rexi_governor.erl | 1 +
src/rexi_server.erl | 17 +++--------------
src/rexi_server_mon.erl | 1 +
4 files changed, 6 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/11b3859a/src/rexi_gov_manager.erl
----------------------------------------------------------------------
diff --git a/src/rexi_gov_manager.erl b/src/rexi_gov_manager.erl
index 4fb69e3..46cbe53 100644
--- a/src/rexi_gov_manager.erl
+++ b/src/rexi_gov_manager.erl
@@ -12,6 +12,7 @@
-module(rexi_gov_manager).
-behaviour(gen_server).
+-vsn(1).
-behaviour(config_listener).
% API
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/11b3859a/src/rexi_governor.erl
----------------------------------------------------------------------
diff --git a/src/rexi_governor.erl b/src/rexi_governor.erl
index ad62150..876165d 100644
--- a/src/rexi_governor.erl
+++ b/src/rexi_governor.erl
@@ -12,6 +12,7 @@
-module(rexi_governor).
-behaviour(gen_server).
+-vsn(1).
% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/11b3859a/src/rexi_server.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server.erl b/src/rexi_server.erl
index 7d928ad..3bb85a0 100644
--- a/src/rexi_server.erl
+++ b/src/rexi_server.erl
@@ -12,6 +12,7 @@
-module(rexi_server).
-behaviour(gen_server).
+-vsn(1).
-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
code_change/3]).
@@ -123,20 +124,8 @@ terminate(_Reason, St) ->
St#st.workers),
ok.
-code_change(_OldVsn, {st, Workers}, _Extra) ->
- {ok, #st{workers = Workers}};
-
-code_change(_OldVsn, {st, Workers0, Errors, Limit, Count}, _Extra) ->
- Jobs = [#job{worker_pid=A, worker=B, client_pid=C, client=D}
- || {A, B, {C, D}} <- ets:tab2list(Workers0)],
- ets:delete(Workers0),
- State = #st{errors = Errors, error_limit = Limit, error_count = Count},
- ets:insert(State#st.workers, Jobs),
- ets:insert(State#st.clients, Jobs),
- {ok, State};
-
-code_change(_OldVsn, St, _Extra) ->
- {ok, St}.
+code_change(_OldVsn, #st{}=State, _Extra) ->
+ {ok, State}.
init_p(From, MFA) ->
init_p(From, MFA, undefined).
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/11b3859a/src/rexi_server_mon.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server_mon.erl b/src/rexi_server_mon.erl
index 8cee6e9..756ab95 100644
--- a/src/rexi_server_mon.erl
+++ b/src/rexi_server_mon.erl
@@ -14,6 +14,7 @@
-module(rexi_server_mon).
-behaviour(gen_server).
+-vsn(1).
-export([
[23/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Slight style tweaks to rexi_buffer hibernation
Avoid using a possibly misleading variable name by not naming it.
Alternatively, this is not the variable name you are looking for.
BugzId: 27672
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/23cda378
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/23cda378
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/23cda378
Branch: refs/heads/master
Commit: 23cda378322851e8d09c850510ed75119cc83adf
Parents: e13e16f
Author: Paul J. Davis <pa...@gmail.com>
Authored: Fri Jan 31 02:55:57 2014 -0600
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:06:43 2014 +0100
----------------------------------------------------------------------
src/rexi_buffer.erl | 13 +++++++------
1 file changed, 7 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/23cda378/src/rexi_buffer.erl
----------------------------------------------------------------------
diff --git a/src/rexi_buffer.erl b/src/rexi_buffer.erl
index 880e4dd..874ec3c 100644
--- a/src/rexi_buffer.erl
+++ b/src/rexi_buffer.erl
@@ -72,16 +72,17 @@ handle_info(timeout, #state{sender = nil} = State) ->
if Sender =:= nil, C > 1 ->
{noreply, State#state{buffer = Q2, count = C-1}, 0};
true ->
+ NewState = State#state{buffer = Q2, sender = Sender, count = C-1},
% When Sender is nil and C-1 == 0 we're reverting to an
% idle state with no outstanding or queued messages. We'll
% use this oppurtunity to hibernate this process and
% run a garbage collection.
- Timeout = case {Sender, C-1} of
- {nil, 0} -> hibernate;
- _ -> infinity
- end,
- NewState = State#state{buffer = Q2, sender = Sender, count = C-1},
- {noreply, NewState, Timeout}
+ case {Sender, C-1} of
+ {nil, 0} ->
+ {noreply, NewState, hibernate};
+ _ ->
+ {noreply, NewState, infinity}
+ end
end;
handle_info(timeout, State) ->
% Waiting on a sender to return
[17/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Try to start per-node servers immediately
I can't think of a good reason not to do this.
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/664f0b8f
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/664f0b8f
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/664f0b8f
Branch: refs/heads/master
Commit: 664f0b8f3f42f43be87aa9687372bac96c7743a0
Parents: b295083
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Wed Oct 16 11:51:05 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:03:15 2014 +0100
----------------------------------------------------------------------
src/rexi_server_mon.erl | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/664f0b8f/src/rexi_server_mon.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server_mon.erl b/src/rexi_server_mon.erl
index 205ef4a..2dcb41c 100644
--- a/src/rexi_server_mon.erl
+++ b/src/rexi_server_mon.erl
@@ -47,7 +47,7 @@ status() ->
init(ChildMod) ->
net_kernel:monitor_nodes(true),
- erlang:send_after(?INTERVAL, self(), check_nodes),
+ erlang:send(self(), check_nodes),
{ok, ChildMod}.
[18/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Don't block the governor to send a message
This allows the governor to continue prioritizing incoming requests
instead of hanging for several seconds to try to connect to the remote
node.
BugzID: 23717
BugzID: 23718
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/8f9d160c
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/8f9d160c
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/8f9d160c
Branch: refs/heads/master
Commit: 8f9d160c688af025179b0f61be972a00b33f5fad
Parents: 664f0b8
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Fri Nov 22 15:58:02 2013 -0500
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:03:23 2014 +0100
----------------------------------------------------------------------
src/rexi_governor.erl | 26 +++++++++++++++++++-------
1 file changed, 19 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/8f9d160c/src/rexi_governor.erl
----------------------------------------------------------------------
diff --git a/src/rexi_governor.erl b/src/rexi_governor.erl
index fdf8c93..12ec013 100644
--- a/src/rexi_governor.erl
+++ b/src/rexi_governor.erl
@@ -25,6 +25,7 @@
-record(state, {
buffer = queue:new(),
+ sender = nil,
count = 0
}).
@@ -55,19 +56,30 @@ handle_cast({deliver, Dest, Msg}, #state{buffer = Q, count = C} = State) ->
{noreply, State#state{buffer = Q2, count = C+1}, 0}
end.
-handle_info(timeout, State) ->
+handle_info(timeout, #state{sender = nil} = State) ->
#state{buffer = Q, count = C} = State,
- case queue:out_r(Q) of
+ Sender = case queue:out_r(Q) of
{{value, {Dest, Msg}}, Q2} ->
- erlang:send(Dest, Msg);
+ case erlang:send(Dest, Msg, [noconnect, nosuspend]) of
+ ok ->
+ nil;
+ _Else ->
+ spawn_monitor(erlang, send, [Dest, Msg])
+ end;
{empty, Q2} ->
- ok
+ nil
end,
- if C > 1 ->
+ if Sender =:= nil, C > 1 ->
{noreply, State#state{buffer = Q2, count = C-1}, 0};
true ->
- {noreply, State#state{buffer = Q2, count = 0}}
- end.
+ {noreply, State#state{buffer = Q2, sender = Sender, count = C-1}}
+ end;
+handle_info(timeout, State) ->
+ % Waiting on a sender to return
+ {noreply, State};
+
+handle_info({'DOWN', Ref, _, Pid, _}, #state{sender = {Pid, Ref}} = State) ->
+ {noreply, State#state{sender = nil}, 0}.
terminate(_Reason, _State) ->
ok.
[33/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Update to use couch_stats
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/bbf59a21
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/bbf59a21
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/bbf59a21
Branch: refs/heads/master
Commit: bbf59a2174772c76e26daa5afeeb6f6c038088fb
Parents: a853c32
Author: Paul J. Davis <pa...@gmail.com>
Authored: Thu Aug 21 01:11:31 2014 -0500
Committer: Paul J. Davis <pa...@gmail.com>
Committed: Thu Aug 21 02:35:56 2014 -0500
----------------------------------------------------------------------
priv/stats_descriptions.cfg | 24 ++++++++++++++++++++++++
src/rexi.app.src | 1 +
src/rexi.erl | 6 ++++++
src/rexi_buffer.erl | 2 ++
src/rexi_monitor.erl | 1 +
5 files changed, 34 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/bbf59a21/priv/stats_descriptions.cfg
----------------------------------------------------------------------
diff --git a/priv/stats_descriptions.cfg b/priv/stats_descriptions.cfg
new file mode 100644
index 0000000..93c29d9
--- /dev/null
+++ b/priv/stats_descriptions.cfg
@@ -0,0 +1,24 @@
+{[rexi, buffered], [
+ {type, counter},
+ {desc, <<"number of rexi messages buffered">>}
+]}.
+{[rexi, down], [
+ {type, counter},
+ {desc, <<"number of rexi_DOWN messages handled">>}
+]}.
+{[rexi, dropped], [
+ {type, counter},
+ {desc, <<"number of rexi messages dropped from buffers">>}
+]}.
+{[rexi, streams, timeout, init_stream], [
+ {type, counter},
+ {desc, <<"number of rexi stream initialization timeouts">>}
+]}.
+{[rexi, streams, timeout, stream], [
+ {type, counter},
+ {desc, <<"number of rexi stream timeouts">>}
+]}.
+{[rexi, streams, timeout, wait_for_ack], [
+ {type, counter},
+ {desc, <<"number of rexi stream timeouts while waiting for acks">>}
+]}.
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/bbf59a21/src/rexi.app.src
----------------------------------------------------------------------
diff --git a/src/rexi.app.src b/src/rexi.app.src
index 639e9dc..efe128c 100644
--- a/src/rexi.app.src
+++ b/src/rexi.app.src
@@ -31,6 +31,7 @@
kernel,
stdlib,
couch_log,
+ couch_stats,
config
]},
{mod, {rexi_app,[]}}
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/bbf59a21/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index 69fbd77..fea4d64 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -134,6 +134,9 @@ stream_init(Timeout) ->
rexi_STREAM_CANCEL ->
exit(normal);
timeout ->
+ couch_stats:increment_counter(
+ [rexi, streams, timeout, init_stream]
+ ),
exit(timeout);
Else ->
exit({invalid_stream_message, Else})
@@ -179,6 +182,7 @@ stream(Msg, Limit, Timeout) ->
erlang:send(Caller, {Ref, self(), Msg}),
ok
catch throw:timeout ->
+ couch_stats:increment_counter([rexi, streams, timeout, stream]),
exit(timeout)
end.
@@ -205,6 +209,7 @@ stream2(Msg, Limit, Timeout) ->
erlang:send(Caller, {Ref, self(), Msg}),
ok
catch throw:timeout ->
+ couch_stats:increment_counter([rexi, streams, timeout, stream]),
exit(timeout)
end.
@@ -267,6 +272,7 @@ wait_for_ack(Count, Timeout) ->
receive
{rexi_ack, N} -> drain_acks(Count-N)
after Timeout ->
+ couch_stats:increment_counter([rexi, streams, timeout, wait_for_ack]),
throw(timeout)
end.
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/bbf59a21/src/rexi_buffer.erl
----------------------------------------------------------------------
diff --git a/src/rexi_buffer.erl b/src/rexi_buffer.erl
index 2be2322..d16dc8b 100644
--- a/src/rexi_buffer.erl
+++ b/src/rexi_buffer.erl
@@ -50,9 +50,11 @@ handle_call(get_buffered_count, _From, State) ->
{reply, State#state.count, State, 0}.
handle_cast({deliver, Dest, Msg}, #state{buffer = Q, count = C} = State) ->
+ couch_stats:increment_counter([rexi, buffered]),
Q2 = queue:in({Dest, Msg}, Q),
case should_drop(State) of
true ->
+ couch_stats:increment_counter([rexi, dropped]),
{noreply, State#state{buffer = queue:drop(Q2)}, 0};
false ->
{noreply, State#state{buffer = Q2, count = C+1}, 0}
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/bbf59a21/src/rexi_monitor.erl
----------------------------------------------------------------------
diff --git a/src/rexi_monitor.erl b/src/rexi_monitor.erl
index 7be3f0a..f143ead 100644
--- a/src/rexi_monitor.erl
+++ b/src/rexi_monitor.erl
@@ -40,6 +40,7 @@ stop(MonitoringPid) ->
%% internal functions %%
notify_parent(Parent, Pid, Reason) ->
+ couch_stats:increment_counter([rexi, down]),
erlang:send(Parent, {rexi_DOWN, self(), Pid, Reason}).
should_monitor(Pid, Nodes) when is_pid(Pid) ->
[24/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Configure buffer limit by message count
This allows an operator to decide how large the buffers should be. It
also provides an escape valve to clear the buffer entirely.
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/cd07cb8c
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/cd07cb8c
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/cd07cb8c
Branch: refs/heads/master
Commit: cd07cb8c04df1510253718df7f63d6783e3ec0a7
Parents: 23cda37
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Tue Jun 3 14:25:59 2014 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:08:01 2014 +0100
----------------------------------------------------------------------
src/rexi_buffer.erl | 19 ++++++++++++-------
1 file changed, 12 insertions(+), 7 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/cd07cb8c/src/rexi_buffer.erl
----------------------------------------------------------------------
diff --git a/src/rexi_buffer.erl b/src/rexi_buffer.erl
index 874ec3c..f75399c 100644
--- a/src/rexi_buffer.erl
+++ b/src/rexi_buffer.erl
@@ -29,9 +29,6 @@
count = 0
}).
-%% TODO Leverage os_mon to discover available memory in the system
--define (MAX_MEMORY, 17179869184).
-
start_link(ServerId) ->
gen_server:start_link({local, ServerId}, ?MODULE, nil, []).
@@ -41,7 +38,12 @@ send(Dest, Msg) ->
init(_) ->
- {ok, #state{}}.
+ %% TODO Leverage os_mon to discover available memory in the system
+ Max = list_to_integer(config:get("rexi", "buffer_count", "2000")),
+ {ok, #state{max_count = Max}}.
+
+handle_call(erase_buffer, _From, State) ->
+ {reply, ok, State#state{buffer = queue:new(), count = 0}, 0};
handle_call(get_buffered_count, _From, State) ->
{reply, State#state.count, State, 0}.
@@ -49,7 +51,7 @@ handle_call(get_buffered_count, _From, State) ->
handle_cast({deliver, Dest, Msg}, #state{buffer = Q, count = C} = State) ->
margaret_counter:increment([erlang, rexi, buffered]),
Q2 = queue:in({Dest, Msg}, Q),
- case should_drop() of
+ case should_drop(State) of
true ->
{noreply, State#state{buffer = queue:drop(Q2)}, 0};
false ->
@@ -94,11 +96,14 @@ handle_info({'DOWN', Ref, _, Pid, _}, #state{sender = {Pid, Ref}} = State) ->
terminate(_Reason, _State) ->
ok.
+code_change(_OldVsn, {state, Buffer, Sender, Count}, _Extra) ->
+ Max = list_to_integer(config:get("rexi", "buffer_count", "2000")),
+ {ok, #state{buffer=Buffer, sender=Sender, count=Count, max_count=Max}};
code_change(_OldVsn, State, _Extra) ->
{ok, State}.
-should_drop() ->
- erlang:memory(total) > ?MAX_MEMORY.
+should_drop(#state{count = Count, max_count = Max}) ->
+ Count >= Max.
get_node({_, Node}) when is_atom(Node) ->
Node;
[16/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Remove gov_manager from supervision
We'll remove the module itself in a subsequent release.
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/b2950830
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/b2950830
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/b2950830
Branch: refs/heads/master
Commit: b2950830fc80514f1ac63a32e3da82df6284daed
Parents: ffb6dad
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Wed Oct 16 10:54:15 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:03:06 2014 +0100
----------------------------------------------------------------------
src/rexi_sup.erl | 8 --------
1 file changed, 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/b2950830/src/rexi_sup.erl
----------------------------------------------------------------------
diff --git a/src/rexi_sup.erl b/src/rexi_sup.erl
index 8ee116f..f9e4933 100644
--- a/src/rexi_sup.erl
+++ b/src/rexi_sup.erl
@@ -22,14 +22,6 @@ start_link(Args) ->
init([]) ->
{ok, {{one_for_one, 3, 10}, [
{
- rexi_gov_manager,
- {rexi_gov_manager, start_link, []},
- permanent,
- 100,
- worker,
- [rexi_gov_manager]
- },
- {
rexi_server,
{rexi_server, start_link, [rexi_server]},
permanent,
[20/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Revert "Tag all replies to the coordinator"
Not all rexi communication happens through `rexi_utils:recv/6`. We'll
need to audit all of our apps to see anywhere that doesn't to update to
the new message format.
This reverts commit 2f967906730eedf5851752be06155fc49edd28a1.
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/f9380551
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/f9380551
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/f9380551
Branch: refs/heads/master
Commit: f9380551e3bedf4e32842eaca77435206424b1be
Parents: 5b58996
Author: Paul J. Davis <pa...@gmail.com>
Authored: Thu Dec 12 11:36:05 2013 -0600
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:05:43 2014 +0100
----------------------------------------------------------------------
src/rexi.erl | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/f9380551/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index 62f410b..b0ad8c1 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -98,7 +98,7 @@ async_server_call(Server, Caller, Request) ->
-spec reply(any()) -> any().
reply(Reply) ->
{Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {rexi, Ref, Reply}).
+ erlang:send(Caller, {Ref,Reply}).
%% @equiv sync_reply(Reply, 300000)
sync_reply(Reply) ->
@@ -111,7 +111,7 @@ sync_reply(Reply) ->
sync_reply(Reply, Timeout) ->
{Caller, Ref} = get(rexi_from),
Tag = make_ref(),
- erlang:send(Caller, {rexi, Ref, {self(),Tag}, Reply}),
+ erlang:send(Caller, {Ref, {self(),Tag}, Reply}),
receive {Tag, Response} ->
Response
after Timeout ->
@@ -179,7 +179,7 @@ stream(Msg, Limit, Timeout) ->
{ok, Count} ->
put(rexi_unacked, Count+1),
{Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {rexi, Ref, self(), Msg}),
+ erlang:send(Caller, {Ref, self(), Msg}),
ok
catch throw:timeout ->
exit(timeout)
[07/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Tag all replies to the coordinator
BugzID: 20204
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/c275f924
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/c275f924
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/c275f924
Branch: refs/heads/master
Commit: c275f924aac24bd113fbf27eefd1b95281491c92
Parents: 4c99447
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Thu Jun 6 15:53:24 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:57:42 2014 +0100
----------------------------------------------------------------------
src/rexi.erl | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/c275f924/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index c970f80..d37d360 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -95,7 +95,7 @@ async_server_call(Server, Caller, Request) ->
-spec reply(any()) -> any().
reply(Reply) ->
{Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {Ref,Reply}).
+ erlang:send(Caller, {rexi, Ref, Reply}).
%% @equiv sync_reply(Reply, 300000)
sync_reply(Reply) ->
@@ -108,7 +108,7 @@ sync_reply(Reply) ->
sync_reply(Reply, Timeout) ->
{Caller, Ref} = get(rexi_from),
Tag = make_ref(),
- erlang:send(Caller, {Ref, {self(),Tag}, Reply}),
+ erlang:send(Caller, {rexi, Ref, {self(),Tag}, Reply}),
receive {Tag, Response} ->
Response
after Timeout ->
@@ -132,7 +132,7 @@ stream(Msg, Limit, Timeout) ->
{ok, Count} ->
put(rexi_unacked, Count+1),
{Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {Ref, self(), Msg}),
+ erlang:send(Caller, {rexi, Ref, self(), Msg}),
ok
catch throw:timeout ->
timeout
[30/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
The couch_log functions are all two-arity
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/77dd2bed
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/77dd2bed
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/77dd2bed
Branch: refs/heads/master
Commit: 77dd2bed7cadcfaa686a0008361d6e057839a515
Parents: 6ba8701
Author: Paul J. Davis <pa...@gmail.com>
Authored: Sun Aug 17 15:25:53 2014 -0500
Committer: Paul J. Davis <pa...@gmail.com>
Committed: Sun Aug 17 15:25:53 2014 -0500
----------------------------------------------------------------------
src/rexi_server.erl | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/77dd2bed/src/rexi_server.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server.erl b/src/rexi_server.erl
index 5f9976d..f71f878 100644
--- a/src/rexi_server.erl
+++ b/src/rexi_server.erl
@@ -90,7 +90,7 @@ handle_cast({kill, FromRef}, #st{clients = Clients} = St) ->
end;
handle_cast(_, St) ->
- couch_log:notice("rexi_server ignored_cast"),
+ couch_log:notice("rexi_server ignored_cast", []),
{noreply, St}.
handle_info({'DOWN', Ref, process, _, normal}, #st{workers=Workers} = St) ->
[04/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Use the unadulterated node name in server name
BugzID: 18215
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/ad795911
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/ad795911
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/ad795911
Branch: refs/heads/master
Commit: ad795911100283df6740647492373b3377fc27bf
Parents: 752921b
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Thu Apr 11 08:10:20 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:57:10 2014 +0100
----------------------------------------------------------------------
src/rexi_utils.erl | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/ad795911/src/rexi_utils.erl
----------------------------------------------------------------------
diff --git a/src/rexi_utils.erl b/src/rexi_utils.erl
index b87f2c3..ced0ca3 100644
--- a/src/rexi_utils.erl
+++ b/src/rexi_utils.erl
@@ -18,7 +18,7 @@
server_id(Node) ->
case config:get("rexi", "server_per_node", "false") of
"true" ->
- list_to_atom("rexi_server_" ++ integer_to_list(erlang:phash2(Node)));
+ list_to_atom("rexi_server_" ++ atom_to_list(Node));
_ ->
rexi_server
end.
[09/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Implement new streaming APIs
This adds new functions that are used by coordinators and workers to
negotiate an RPC stream. A stream is simply any response that requires
multiple messages from the worker.
BugzId: 22729
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/8f2c2956
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/8f2c2956
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/8f2c2956
Branch: refs/heads/master
Commit: 8f2c2956a80afce73e4193819bf6c973376a28cd
Parents: b28cbb7
Author: Paul J. Davis <pa...@gmail.com>
Authored: Wed Aug 14 09:06:05 2013 -0500
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:58:04 2014 +0100
----------------------------------------------------------------------
src/rexi.erl | 46 ++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 46 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/8f2c2956/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index d37d360..75bc9bf 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -15,6 +15,8 @@
-export([cast/2, cast/3, cast/4, kill/2]).
-export([reply/1, sync_reply/1, sync_reply/2]).
-export([async_server_call/2, async_server_call/3]).
+-export([stream_init/0, stream_init/1]).
+-export([stream_start/1, stream_cancel/1]).
-export([stream/1, stream/2, stream/3, stream_ack/1, stream_ack/2]).
-include_lib("rexi/include/rexi.hrl").
@@ -115,6 +117,50 @@ sync_reply(Reply, Timeout) ->
timeout
end.
+%% @equiv stream_init(300000)
+stream_init() ->
+ stream_init(300000).
+
+%% @doc Initialize an RPC stream that involves sending multiple
+%% messages back to the coordinator.
+%%
+%% This should be called by rexi workers. It blocks until the
+%% coordinator responds with whether this worker should proceed.
+%% This function will either return with `ok` or call
+%% `erlang:exit/1`.
+-spec stream_init(pos_integer()) -> ok.
+stream_init(Timeout) ->
+ case sync_reply(rexi_STREAM_INIT, Timeout) of
+ rexi_STREAM_START ->
+ ok;
+ rexi_STREAM_CANCEL ->
+ exit(normal);
+ timeout ->
+ exit(timeout);
+ Else ->
+ exit({invalid_stream_message, Else})
+ end.
+
+%% @doc Start a worker stream
+%%
+%% If a coordinator wants to continue using a streaming worker it
+%% should use this function to inform the worker to continue
+%% sending messages. The `From` should be the value provided by
+%% the worker in the rexi_STREAM_INIT message.
+-spec stream_start({pid(), any()}) -> ok.
+stream_start({Pid, _Tag}=From) when is_pid(Pid) ->
+ gen_server:reply(From, rexi_STREAM_START).
+
+%% @doc Cancel a worker stream
+%%
+%% If a coordinator decideds that a worker is not going to be part
+%% of the response it should use this function to cancel the worker.
+%% The `From` should be the value provided by the worker in the
+%% rexi_STREAM_INIT message.
+-spec stream_cancel({pid(), any()}) -> ok.
+stream_cancel({Pid, _Tag}=From) when is_pid(Pid) ->
+ gen_server:reply(From, rexi_STREAM_CANCEL).
+
%% @equiv stream(Msg, 100, 300000)
stream(Msg) ->
stream(Msg, 10, 300000).
[28/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Allow for timeout message while waiting for sender
In 9c36c1b0 we fixed counting bug in the bufffer but introduced another
one. We started asserting that the timeout message can only arrive when
the sender is not nil, but in fact it can absolutely happen that the
buffer is asked to deliver a message while one is in flight. This causes
the rexi_buffer to crash. The fix is to simply restore the clause that
ignores the timeout message if the server is waiting for a sender to
return.
BugzID: 32669
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/535941d9
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/535941d9
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/535941d9
Branch: refs/heads/master
Commit: 535941d94d3e2d05f5ec31d2ff8f65c4149cc1da
Parents: 096f0cf
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Fri Jul 18 14:58:23 2014 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Thu Aug 7 12:33:09 2014 +0100
----------------------------------------------------------------------
src/rexi_buffer.erl | 3 +++
1 file changed, 3 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/535941d9/src/rexi_buffer.erl
----------------------------------------------------------------------
diff --git a/src/rexi_buffer.erl b/src/rexi_buffer.erl
index 42b4ef2..0ba961a 100644
--- a/src/rexi_buffer.erl
+++ b/src/rexi_buffer.erl
@@ -78,6 +78,9 @@ handle_info(timeout, #state{sender = nil, count = C} = State) when C > 0 ->
Sender = spawn_monitor(erlang, send, [Dest, Msg]),
{noreply, NewState#state{sender = Sender}}
end;
+handle_info(timeout, State) ->
+ % Waiting on a sender to return
+ {noreply, State};
handle_info({'DOWN', Ref, _, Pid, _}, #state{sender = {Pid, Ref}} = State) ->
{noreply, State#state{sender = nil}, 0}.
[27/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Fix error_limit = 0, and make it the default
Setting the error_limit to 0 would previously cause rexi_server to crash
on the next error because of an unhandled exception in a queue:drop/1.
This fixes that exception and makes a limit of 0 the default. Operators
can still raise the limit to capture errors in the future if we end up
finding that useful.
BugzID: 30821
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/096f0cfb
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/096f0cfb
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/096f0cfb
Branch: refs/heads/master
Commit: 096f0cfb1a6f85880685b7b65d44f00a582436d3
Parents: 9a1bbd0
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Thu Jun 26 19:49:30 2014 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:08:32 2014 +0100
----------------------------------------------------------------------
src/rexi_server.erl | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/096f0cfb/src/rexi_server.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server.erl b/src/rexi_server.erl
index 3bb85a0..5f9976d 100644
--- a/src/rexi_server.erl
+++ b/src/rexi_server.erl
@@ -32,7 +32,7 @@
workers = ets:new(workers, [private, {keypos, #job.worker}]),
clients = ets:new(clients, [private, {keypos, #job.client}]),
errors = queue:new(),
- error_limit = 20,
+ error_limit = 0,
error_count = 0
}).
@@ -151,6 +151,8 @@ init_p(From, {M,F,A}, Nonce) ->
%% internal
+save_error(_E, #st{error_limit = 0} = St) ->
+ St;
save_error(E, #st{errors=Q, error_limit=L, error_count=C} = St) when C >= L ->
St#st{errors = queue:in(E, queue:drop(Q))};
save_error(E, #st{errors=Q, error_count=C} = St) ->
[05/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Start per-node servers regardless of feature flag
Previously rexi_server_mon would fail to start per-node servers on a
node where the use of those servers in rexi:cast messages was disabled
by the "server_per_node" feature flag. This mostly defeats the purpose
of the feature flag, which is to ensure that we can start all the
per-node servers before we try to use them. Moreover, it cluttered up
the error logs because rexi_server_mon would crash after trying to
execute a `start_child` on rexi_server.
The fix is not DRY but it does the job well.
BugzID: 18970
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/d1995a5b
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/d1995a5b
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/d1995a5b
Branch: refs/heads/master
Commit: d1995a5b76f10b94edc4bf729ace583d228e9ffa
Parents: ad79591
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Tue Apr 16 10:12:58 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:57:21 2014 +0100
----------------------------------------------------------------------
src/rexi_server_mon.erl | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/d1995a5b/src/rexi_server_mon.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server_mon.erl b/src/rexi_server_mon.erl
index 3545dba..8cee6e9 100644
--- a/src/rexi_server_mon.erl
+++ b/src/rexi_server_mon.erl
@@ -101,7 +101,8 @@ start_rexi_servers() ->
missing_servers() ->
- ServerIds = [rexi_utils:server_id(Node) || Node <- [node() | nodes()]],
+ ServerIds = [list_to_atom("rexi_server_" ++ atom_to_list(Node))
+ || Node <- [node() | nodes()]],
ChildIds = [Id || {Id, _, _, _} <- supervisor:which_children(?SUP_MODULE)],
ServerIds -- ChildIds.
[19/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Rename governor to buffer
Buffer describes the behavior quite a bit better than governor.
BugzId: 23717
BugzId: 23718
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/5b589962
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/5b589962
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/5b589962
Branch: refs/heads/master
Commit: 5b5899627db710cab64d1c724cc4ad4d7447b9d8
Parents: 8f9d160
Author: Paul J. Davis <pa...@gmail.com>
Authored: Tue Dec 10 09:59:16 2013 -0600
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:04:13 2014 +0100
----------------------------------------------------------------------
src/rexi_buffer.erl | 96 ++++++++++++++++++++++++++
src/rexi_gov_manager.erl | 157 ------------------------------------------
src/rexi_governor.erl | 96 --------------------------
src/rexi_sup.erl | 8 +--
src/rexi_utils.erl | 2 +-
5 files changed, 101 insertions(+), 258 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/5b589962/src/rexi_buffer.erl
----------------------------------------------------------------------
diff --git a/src/rexi_buffer.erl b/src/rexi_buffer.erl
new file mode 100644
index 0000000..b096c5b
--- /dev/null
+++ b/src/rexi_buffer.erl
@@ -0,0 +1,96 @@
+% Licensed under the Apache License, Version 2.0 (the "License"); you may not
+% use this file except in compliance with the License. You may obtain a copy of
+% the License at
+%
+% http://www.apache.org/licenses/LICENSE-2.0
+%
+% Unless required by applicable law or agreed to in writing, software
+% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+% License for the specific language governing permissions and limitations under
+% the License.
+-module(rexi_buffer).
+
+-behaviour(gen_server).
+-vsn(1).
+
+% gen_server callbacks
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
+ terminate/2, code_change/3]).
+
+-export ([
+ send/2,
+ start_link/1
+]).
+
+-record(state, {
+ buffer = queue:new(),
+ sender = nil,
+ count = 0
+}).
+
+%% TODO Leverage os_mon to discover available memory in the system
+-define (MAX_MEMORY, 17179869184).
+
+start_link(ServerId) ->
+ gen_server:start_link({local, ServerId}, ?MODULE, nil, []).
+
+send(Dest, Msg) ->
+ Server = list_to_atom(lists:concat([rexi_buffer, "_", get_node(Dest)])),
+ gen_server:cast(Server, {deliver, Dest, Msg}).
+
+
+init(_) ->
+ {ok, #state{}}.
+
+handle_call(get_buffered_count, _From, State) ->
+ {reply, State#state.count, State, 0}.
+
+handle_cast({deliver, Dest, Msg}, #state{buffer = Q, count = C} = State) ->
+ margaret_counter:increment([erlang, rexi, buffered]),
+ Q2 = queue:in({Dest, Msg}, Q),
+ case should_drop() of
+ true ->
+ {noreply, State#state{buffer = queue:drop(Q2)}, 0};
+ false ->
+ {noreply, State#state{buffer = Q2, count = C+1}, 0}
+ end.
+
+handle_info(timeout, #state{sender = nil} = State) ->
+ #state{buffer = Q, count = C} = State,
+ Sender = case queue:out_r(Q) of
+ {{value, {Dest, Msg}}, Q2} ->
+ case erlang:send(Dest, Msg, [noconnect, nosuspend]) of
+ ok ->
+ nil;
+ _Else ->
+ spawn_monitor(erlang, send, [Dest, Msg])
+ end;
+ {empty, Q2} ->
+ nil
+ end,
+ if Sender =:= nil, C > 1 ->
+ {noreply, State#state{buffer = Q2, count = C-1}, 0};
+ true ->
+ {noreply, State#state{buffer = Q2, sender = Sender, count = C-1}}
+ end;
+handle_info(timeout, State) ->
+ % Waiting on a sender to return
+ {noreply, State};
+
+handle_info({'DOWN', Ref, _, Pid, _}, #state{sender = {Pid, Ref}} = State) ->
+ {noreply, State#state{sender = nil}, 0}.
+
+terminate(_Reason, _State) ->
+ ok.
+
+code_change(_OldVsn, State, _Extra) ->
+ {ok, State}.
+
+should_drop() ->
+ erlang:memory(total) > ?MAX_MEMORY.
+
+get_node({_, Node}) when is_atom(Node) ->
+ Node;
+get_node(Pid) when is_pid(Pid) ->
+ node(Pid).
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/5b589962/src/rexi_gov_manager.erl
----------------------------------------------------------------------
diff --git a/src/rexi_gov_manager.erl b/src/rexi_gov_manager.erl
deleted file mode 100644
index 46cbe53..0000000
--- a/src/rexi_gov_manager.erl
+++ /dev/null
@@ -1,157 +0,0 @@
-% Licensed under the Apache License, Version 2.0 (the "License"); you may not
-% use this file except in compliance with the License. You may obtain a copy of
-% the License at
-%
-% http://www.apache.org/licenses/LICENSE-2.0
-%
-% Unless required by applicable law or agreed to in writing, software
-% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-% License for the specific language governing permissions and limitations under
-% the License.
--module(rexi_gov_manager).
-
--behaviour(gen_server).
--vsn(1).
--behaviour(config_listener).
-
-% API
--export([start_link/0, send/2]).
-
-% gen_server callbacks
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
- terminate/2, code_change/3]).
--export([handle_config_change/5]).
-
--record(state, {node_timers = ets:new(timers, [set]),
- nodeout_timeout = 2000,
- pid_spawn_max = 10000}).
-
-
-% API
-
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
-
-send(Dest, Msg) ->
- case erlang:send(Dest, Msg, [noconnect, nosuspend]) of
- ok -> ok;
- _ ->
- % treat nosuspend and noconnect the same
- {ok, Governor} = get_governor(get_node(Dest)),
- gen_server:cast(Governor, {spawn_and_track, Dest, Msg})
- end.
-
-get_node({_, Node}) when is_atom(Node) ->
- Node;
-get_node(Pid) when is_pid(Pid) ->
- node(Pid).
-
-get_governor(Node) ->
- case ets:lookup(govs, Node) of
- [{Node, Gov}] ->
- {ok, Gov};
- [] ->
- gen_server:call(?MODULE, {get_governor, Node})
- end.
-
-% gen_server callbacks
-
-init([]) ->
- ets:new(govs, [named_table, set, {read_concurrency, true}]),
- net_kernel:monitor_nodes(true),
- NodeOutTimeout = config:get("rexi","nodeout_timeout","500"),
- PidSpawnMax = config:get("rexi","pid_spawn_max", "10000"),
- State = #state{
- nodeout_timeout = list_to_integer(NodeOutTimeout),
- pid_spawn_max = list_to_integer(PidSpawnMax)
- },
- config:listen_for_changes(?MODULE, State),
- {ok, State}.
-
-handle_config_change("rexi", "nodeout_timeout", Value, _, State) ->
- IntValue = list_to_integer(Value),
- %% Setting the timeout is cheap, no need to check if it actually changed
- gen_server:call(?MODULE, {set_timeout, IntValue}),
- {ok, State#state{nodeout_timeout = IntValue}};
-handle_config_change("rexi", "pid_spawn_max", Value, _, State) ->
- IntValue = list_to_integer(Value),
- %% Setting the timeout is cheap, no need to check if it actually changed
- gen_server:call(?MODULE, {set_spawn_max, IntValue}),
- {ok, State#state{pid_spawn_max = IntValue}};
-handle_config_change(_, _, _, _, State) ->
- {ok, State}.
-
-handle_call({set_timeout, TO}, _, #state{nodeout_timeout = Old} = State) ->
- {reply, Old, State#state{nodeout_timeout = TO}};
-handle_call({set_spawn_max, Max}, _, #state{pid_spawn_max = Old} = State) ->
- {reply, Old, State#state{pid_spawn_max = Max}};
-handle_call({get_governor, Node}, _From,
- #state{pid_spawn_max = PidSpawnMax} = State) ->
- case ets:lookup(govs, Node) of
- [] ->
- {ok, Gov} = gen_server:start_link(rexi_governor, [PidSpawnMax], []),
- ets:insert(govs, {Node, Gov});
- [{Node, Gov}] ->
- Gov
- end,
- {reply, {ok, Gov}, State}.
-
-handle_cast(_Msg, State) ->
- {noreply, State}.
-
-handle_info({nodeup, Node}, #state{node_timers = Timers,
- pid_spawn_max = PidSpawnMax} = State) ->
- case ets:lookup(Timers, Node) of
- [{Node, TRef}] ->
- erlang:cancel_timer(TRef),
- ets:delete(Timers, Node);
- _ ->
- ok
- end,
- case ets:lookup(govs, Node) of
- [{Node, _}] ->
- ok;
- [] ->
- {ok, Gov} = gen_server:start_link(rexi_governor, [PidSpawnMax], []),
- ets:insert(govs, {Node, Gov})
- end,
- {noreply, State};
-
-handle_info({nodedown, Node}, #state{node_timers = Timers,
- nodeout_timeout = NodeTimeout} = State) ->
- case ets:lookup(Timers, Node) of
- [] ->
- TRef = erlang:send_after(NodeTimeout, self(), {nodeout, Node}),
- ets:insert(Timers, {Node, TRef}),
- {noreply, State};
- _ ->
- {noreply, State}
- end;
-
-handle_info({nodeout, Node}, #state{node_timers = Timers} = State) ->
- % check for race with node up
- case ets:member(Timers, Node) of
- true ->
- ets:delete(Timers, Node),
- case ets:lookup(govs, Node) of
- [] ->
- ok;
- [{Node, Governor}] ->
- gen_server:cast(Governor, nodeout)
- end;
- false ->
- ok
- end,
- {noreply, State};
-
-handle_info(_Info, State) ->
- {noreply, State}.
-
-terminate(_Reason, _State) ->
- ok.
-
-code_change(_OldVsn, State, _Extra) ->
- {ok, State}.
-
-% Internal functions
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/5b589962/src/rexi_governor.erl
----------------------------------------------------------------------
diff --git a/src/rexi_governor.erl b/src/rexi_governor.erl
deleted file mode 100644
index 12ec013..0000000
--- a/src/rexi_governor.erl
+++ /dev/null
@@ -1,96 +0,0 @@
-% Licensed under the Apache License, Version 2.0 (the "License"); you may not
-% use this file except in compliance with the License. You may obtain a copy of
-% the License at
-%
-% http://www.apache.org/licenses/LICENSE-2.0
-%
-% Unless required by applicable law or agreed to in writing, software
-% distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
-% WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
-% License for the specific language governing permissions and limitations under
-% the License.
--module(rexi_governor).
-
--behaviour(gen_server).
--vsn(1).
-
-% gen_server callbacks
--export([init/1, handle_call/3, handle_cast/2, handle_info/2,
- terminate/2, code_change/3]).
-
--export ([
- send/2,
- start_link/1
-]).
-
--record(state, {
- buffer = queue:new(),
- sender = nil,
- count = 0
-}).
-
-%% TODO Leverage os_mon to discover available memory in the system
--define (MAX_MEMORY, 17179869184).
-
-start_link(ServerId) ->
- gen_server:start_link({local, ServerId}, ?MODULE, nil, []).
-
-send(Dest, Msg) ->
- Server = list_to_atom(lists:concat([rexi_governor, "_", get_node(Dest)])),
- gen_server:cast(Server, {deliver, Dest, Msg}).
-
-
-init(_) ->
- {ok, #state{}}.
-
-handle_call(get_buffered_count, _From, State) ->
- {reply, State#state.count, State, 0}.
-
-handle_cast({deliver, Dest, Msg}, #state{buffer = Q, count = C} = State) ->
- margaret_counter:increment([erlang, rexi, buffered]),
- Q2 = queue:in({Dest, Msg}, Q),
- case should_drop() of
- true ->
- {noreply, State#state{buffer = queue:drop(Q2)}, 0};
- false ->
- {noreply, State#state{buffer = Q2, count = C+1}, 0}
- end.
-
-handle_info(timeout, #state{sender = nil} = State) ->
- #state{buffer = Q, count = C} = State,
- Sender = case queue:out_r(Q) of
- {{value, {Dest, Msg}}, Q2} ->
- case erlang:send(Dest, Msg, [noconnect, nosuspend]) of
- ok ->
- nil;
- _Else ->
- spawn_monitor(erlang, send, [Dest, Msg])
- end;
- {empty, Q2} ->
- nil
- end,
- if Sender =:= nil, C > 1 ->
- {noreply, State#state{buffer = Q2, count = C-1}, 0};
- true ->
- {noreply, State#state{buffer = Q2, sender = Sender, count = C-1}}
- end;
-handle_info(timeout, State) ->
- % Waiting on a sender to return
- {noreply, State};
-
-handle_info({'DOWN', Ref, _, Pid, _}, #state{sender = {Pid, Ref}} = State) ->
- {noreply, State#state{sender = nil}, 0}.
-
-terminate(_Reason, _State) ->
- ok.
-
-code_change(_OldVsn, State, _Extra) ->
- {ok, State}.
-
-should_drop() ->
- erlang:memory(total) > ?MAX_MEMORY.
-
-get_node({_, Node}) when is_atom(Node) ->
- Node;
-get_node(Pid) when is_pid(Pid) ->
- node(Pid).
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/5b589962/src/rexi_sup.erl
----------------------------------------------------------------------
diff --git a/src/rexi_sup.erl b/src/rexi_sup.erl
index f9e4933..55c4829 100644
--- a/src/rexi_sup.erl
+++ b/src/rexi_sup.erl
@@ -46,16 +46,16 @@ init([]) ->
[rexi_server_mon]
},
{
- rexi_governor_sup,
- {rexi_server_sup, start_link, [rexi_governor_sup]},
+ rexi_buffer_sup,
+ {rexi_server_sup, start_link, [rexi_buffer_sup]},
permanent,
100,
supervisor,
[rexi_server_sup]
},
{
- rexi_governor_mon,
- {rexi_server_mon, start_link, [rexi_governor]},
+ rexi_buffer_mon,
+ {rexi_server_mon, start_link, [rexi_buffer]},
permanent,
100,
worker,
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/5b589962/src/rexi_utils.erl
----------------------------------------------------------------------
diff --git a/src/rexi_utils.erl b/src/rexi_utils.erl
index 3c89ca9..e3eaa6f 100644
--- a/src/rexi_utils.erl
+++ b/src/rexi_utils.erl
@@ -34,7 +34,7 @@ send(Dest, Msg) ->
ok;
_ ->
% treat nosuspend and noconnect the same
- rexi_governor:send(Dest, Msg)
+ rexi_buffer:send(Dest, Msg)
end.
%% @doc set up the receive loop with an overall timeout
[22/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Hibernate rexi_buffer when becoming idle
The rexi_buffer gen_server can hold onto quite a bit of RAM during idle
operation. This just checks when we're going back to the idle state and
hibernates until the next message arrives. This ensures that we run
garbage collection before sitting idle.
BugzId: 27672
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/e13e16f6
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/e13e16f6
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/e13e16f6
Branch: refs/heads/master
Commit: e13e16f6a0b12985ae95b9fc220321f25a078f8a
Parents: e271ae4
Author: Paul J. Davis <pa...@gmail.com>
Authored: Thu Jan 30 22:03:09 2014 -0600
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:06:30 2014 +0100
----------------------------------------------------------------------
src/rexi_buffer.erl | 11 ++++++++++-
1 file changed, 10 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/e13e16f6/src/rexi_buffer.erl
----------------------------------------------------------------------
diff --git a/src/rexi_buffer.erl b/src/rexi_buffer.erl
index b096c5b..880e4dd 100644
--- a/src/rexi_buffer.erl
+++ b/src/rexi_buffer.erl
@@ -72,7 +72,16 @@ handle_info(timeout, #state{sender = nil} = State) ->
if Sender =:= nil, C > 1 ->
{noreply, State#state{buffer = Q2, count = C-1}, 0};
true ->
- {noreply, State#state{buffer = Q2, sender = Sender, count = C-1}}
+ % When Sender is nil and C-1 == 0 we're reverting to an
+ % idle state with no outstanding or queued messages. We'll
+ % use this oppurtunity to hibernate this process and
+ % run a garbage collection.
+ Timeout = case {Sender, C-1} of
+ {nil, 0} -> hibernate;
+ _ -> infinity
+ end,
+ NewState = State#state{buffer = Q2, sender = Sender, count = C-1},
+ {noreply, NewState, Timeout}
end;
handle_info(timeout, State) ->
% Waiting on a sender to return
[06/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Handle rexi-tagged messages
BugzID: 20204
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/4c994477
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/4c994477
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/4c994477
Branch: refs/heads/master
Commit: 4c994477d67352c5102ed773600089bd1037ab89
Parents: d1995a5
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Thu Jun 6 15:44:05 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:57:34 2014 +0100
----------------------------------------------------------------------
src/rexi_utils.erl | 14 ++++++++++++++
1 file changed, 14 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/4c994477/src/rexi_utils.erl
----------------------------------------------------------------------
diff --git a/src/rexi_utils.erl b/src/rexi_utils.erl
index ced0ca3..79db48c 100644
--- a/src/rexi_utils.erl
+++ b/src/rexi_utils.erl
@@ -59,6 +59,20 @@ process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
receive
{timeout, TimeoutRef} ->
{timeout, Acc0};
+ {rexi, Ref, Msg} ->
+ case lists:keyfind(Ref, Keypos, RefList) of
+ false ->
+ {ok, Acc0};
+ Worker ->
+ Fun(Msg, Worker, Acc0)
+ end;
+ {rexi, Ref, From, Msg} ->
+ case lists:keyfind(Ref, Keypos, RefList) of
+ false ->
+ {ok, Acc0};
+ Worker ->
+ Fun(Msg, {Worker, From}, Acc0)
+ end;
{Ref, Msg} ->
case lists:keyfind(Ref, Keypos, RefList) of
false ->
[08/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Use '$initial_call' instead of initial_call
BugzID: 18742
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/b28cbb76
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/b28cbb76
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/b28cbb76
Branch: refs/heads/master
Commit: b28cbb76b9901af09c3d7b8c9b81df171cf75f85
Parents: c275f92
Author: Adam Kocoloski <ad...@cloudant.com>
Authored: Fri Aug 16 14:40:31 2013 -0400
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:57:54 2014 +0100
----------------------------------------------------------------------
src/rexi_server.erl | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/b28cbb76/src/rexi_server.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server.erl b/src/rexi_server.erl
index 49661cc..7d928ad 100644
--- a/src/rexi_server.erl
+++ b/src/rexi_server.erl
@@ -146,7 +146,7 @@ init_p(From, MFA) ->
string() | undefined) -> any().
init_p(From, {M,F,A}, Nonce) ->
put(rexi_from, From),
- put(initial_call, {M,F,length(A)}),
+ put('$initial_call', {M,F,length(A)}),
put(nonce, Nonce),
try apply(M, F, A) catch exit:normal -> ok; Class:Reason ->
Stack = clean_stack(),
[12/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Implement new stream2 API
This embeds the stream_init/1 logic into the stream functions so that we
don't have to maintain the logic for inititalizing the stream for all
clients.
BugzId: 24635
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/cae29fe1
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/cae29fe1
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/cae29fe1
Branch: refs/heads/master
Commit: cae29fe1926db6ec89645da63aee60766f14bd11
Parents: 7733957
Author: Paul J. Davis <pa...@gmail.com>
Authored: Mon Oct 28 16:03:07 2013 -0500
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:58:32 2014 +0100
----------------------------------------------------------------------
src/rexi.erl | 59 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
1 file changed, 59 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/cae29fe1/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index 20f582b..62f410b 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -18,6 +18,7 @@
-export([stream_init/0, stream_init/1]).
-export([stream_start/1, stream_cancel/1]).
-export([stream/1, stream/2, stream/3, stream_ack/1, stream_ack/2]).
+-export([stream2/1, stream2/2, stream2/3, stream_last/1, stream_last/2]).
-include_lib("rexi/include/rexi.hrl").
@@ -184,6 +185,43 @@ stream(Msg, Limit, Timeout) ->
exit(timeout)
end.
+%% @equiv stream2(Msg, 10, 300000)
+stream2(Msg) ->
+ stream2(Msg, 10, 300000).
+
+%% @equiv stream2(Msg, Limit, 300000)
+stream2(Msg, Limit) ->
+ stream2(Msg, Limit, 300000).
+
+%% @doc Stream a message back to the coordinator. It limits the
+%% number of unacked messsages to Limit and throws a timeout error
+%% if it doesn't receive an ack in Timeout milliseconds. This
+%% is a combination of the old stream_start and stream functions
+%% which automatically does the stream initialization logic.
+-spec stream2(any(), pos_integer(), pos_integer() | inifinity) -> any().
+stream2(Msg, Limit, Timeout) ->
+ maybe_init_stream(Timeout),
+ try maybe_wait(Limit, Timeout) of
+ {ok, Count} ->
+ put(rexi_unacked, Count+1),
+ {Caller, Ref} = get(rexi_from),
+ erlang:send(Caller, {Ref, self(), Msg}),
+ ok
+ catch throw:timeout ->
+ exit(timeout)
+ end.
+
+%% @equiv stream_last(Msg, 300000)
+stream_last(Msg) ->
+ stream_last(Msg, 300000).
+
+%% @doc Send the last message in a stream. This difference between
+%% this and stream is that it uses rexi:reply/1 which doesn't include
+%% the worker pid and doesn't wait for a response from the controller.
+stream_last(Msg, Timeout) ->
+ maybe_init_stream(Timeout),
+ rexi:reply(Msg).
+
%% @equiv stream_ack(Client, 1)
stream_ack(Client) ->
erlang:send(Client, {rexi_ack, 1}).
@@ -196,6 +234,27 @@ stream_ack(Client, N) ->
cast_msg(Msg) -> {'$gen_cast', Msg}.
+maybe_init_stream(Timeout) ->
+ case get(rexi_STREAM_INITED) of
+ true ->
+ ok;
+ _ ->
+ init_stream(Timeout)
+ end.
+
+init_stream(Timeout) ->
+ case sync_reply(rexi_STREAM_INIT, Timeout) of
+ rexi_STREAM_START ->
+ put(rexi_STREAM_INITED, true),
+ ok;
+ rexi_STREAM_CANCEL ->
+ exit(normal);
+ timeout ->
+ exit(timeout);
+ Else ->
+ exit({invalid_stream_message, Else})
+ end.
+
maybe_wait(Limit, Timeout) ->
case get(rexi_unacked) of
undefined ->
[11/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Allow callbacks to update their list of workers
We need this to be able to allow coordinators to replace nodes that are
in maintenance mode.
BugzId: 22729
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/77339578
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/77339578
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/77339578
Branch: refs/heads/master
Commit: 77339578ec3ad0f9f3ca261889e432d64eb793f4
Parents: 945bfca
Author: Paul J. Davis <pa...@gmail.com>
Authored: Fri Oct 4 13:14:25 2013 -0500
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:58:22 2014 +0100
----------------------------------------------------------------------
src/rexi_utils.erl | 2 ++
1 file changed, 2 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/77339578/src/rexi_utils.erl
----------------------------------------------------------------------
diff --git a/src/rexi_utils.erl b/src/rexi_utils.erl
index 79db48c..6e03757 100644
--- a/src/rexi_utils.erl
+++ b/src/rexi_utils.erl
@@ -49,6 +49,8 @@ process_mailbox(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
case process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) of
{ok, Acc} ->
process_mailbox(RefList, Keypos, Fun, Acc, TimeoutRef, PerMsgTO);
+ {new_refs, NewRefList, Acc} ->
+ process_mailbox(NewRefList, Keypos, Fun, Acc, TimeoutRef, PerMsgTO);
{stop, Acc} ->
{ok, Acc};
Error ->
[31/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Replace twig with couch_log
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/7d64200b
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/7d64200b
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/7d64200b
Branch: refs/heads/master
Commit: 7d64200b19066fd07d993c1ad1af8d1c9bb87f21
Parents: 77dd2be
Author: Paul J. Davis <pa...@gmail.com>
Authored: Sun Aug 17 16:32:01 2014 -0500
Committer: Paul J. Davis <pa...@gmail.com>
Committed: Sun Aug 17 16:32:01 2014 -0500
----------------------------------------------------------------------
src/rexi.app.src | 1 +
src/rexi_server_mon.erl | 6 +++---
2 files changed, 4 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/7d64200b/src/rexi.app.src
----------------------------------------------------------------------
diff --git a/src/rexi.app.src b/src/rexi.app.src
index 0691edd..639e9dc 100644
--- a/src/rexi.app.src
+++ b/src/rexi.app.src
@@ -30,6 +30,7 @@
{applications, [
kernel,
stdlib,
+ couch_log,
config
]},
{mod, {rexi_app,[]}}
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/7d64200b/src/rexi_server_mon.erl
----------------------------------------------------------------------
diff --git a/src/rexi_server_mon.erl b/src/rexi_server_mon.erl
index 2dcb41c..e6b5eb9 100644
--- a/src/rexi_server_mon.erl
+++ b/src/rexi_server_mon.erl
@@ -64,12 +64,12 @@ handle_call(status, _From, ChildMod) ->
end;
handle_call(Msg, _From, St) ->
- twig:log(notice, "~s ignored_call ~w", [?MODULE, Msg]),
+ couch_log:notice("~s ignored_call ~w", [?MODULE, Msg]),
{reply, ignored, St}.
handle_cast(Msg, St) ->
- twig:log(notice, "~s ignored_cast ~w", [?MODULE, Msg]),
+ couch_log:notice("~s ignored_cast ~w", [?MODULE, Msg]),
{noreply, St}.
@@ -86,7 +86,7 @@ handle_info(check_nodes, ChildMod) ->
{noreply, ChildMod};
handle_info(Msg, St) ->
- twig:log(notice, "~s ignored_info ~w", [?MODULE, Msg]),
+ couch_log:notice("~s ignored_info ~w", [?MODULE, Msg]),
{noreply, St}.
[10/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Exit with timeout instead of returning an atom
Every function that called `rexi:stream/1` had to check the return value
for timeout and would then call `erlang:exit/1` if that atom were
returned. Rather than force every function to make this check this just
calls exit in `rexi:stream/1`. Its possible to catch this if a process
ever happens to require it.
BugzId: 22729
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/945bfcac
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/945bfcac
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/945bfcac
Branch: refs/heads/master
Commit: 945bfcac25007518f39cfbc6eb61d6d67773f598
Parents: 8f2c295
Author: Paul J. Davis <pa...@gmail.com>
Authored: Fri Sep 6 15:04:00 2013 -0500
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:58:14 2014 +0100
----------------------------------------------------------------------
src/rexi.erl | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/945bfcac/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index 75bc9bf..20f582b 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -181,7 +181,7 @@ stream(Msg, Limit, Timeout) ->
erlang:send(Caller, {rexi, Ref, self(), Msg}),
ok
catch throw:timeout ->
- timeout
+ exit(timeout)
end.
%% @equiv stream_ack(Client, 1)
[02/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
[2/3] Start a rexi_server per remote node
Switch sending rexi messages to the rexi_server handling requests for
the local node. This replaces the use of the singleton rexi_server
process with a rexi_server instance for each node.
The only slight gotchya in the switch is that we need to send kill
messages to both the per-node and singleton instances during the
transition because we can't tell which process may have generated the
job. The extra kill message will be removed in the next commit.
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/837f0457
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/837f0457
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/837f0457
Branch: refs/heads/master
Commit: 837f0457a0e4e280fed6741af33e6a61c5fbf016
Parents: f5907c4
Author: Paul J. Davis <pa...@gmail.com>
Authored: Fri Mar 22 15:07:53 2013 -0500
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 17:56:26 2014 +0100
----------------------------------------------------------------------
src/rexi.erl | 11 ++++++-----
1 file changed, 6 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/837f0457/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index a4c786d..c970f80 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -19,8 +19,6 @@
-include_lib("rexi/include/rexi.hrl").
--define(SERVER, rexi_server).
-
start() ->
application:start(rexi).
@@ -47,7 +45,7 @@ cast(Node, MFA) ->
cast(Node, Caller, MFA) ->
Ref = make_ref(),
Msg = cast_msg({doit, {Caller, Ref}, get(nonce), MFA}),
- rexi_utils:send({?SERVER, Node}, Msg),
+ rexi_utils:send(rexi_utils:server_pid(Node), Msg),
Ref.
%% @doc Executes apply(M, F, A) on Node.
@@ -61,7 +59,7 @@ cast(Node, Caller, MFA, Options) ->
true ->
Ref = make_ref(),
Msg = cast_msg({doit, {Caller, Ref}, get(nonce), MFA}),
- erlang:send({?SERVER, Node}, Msg),
+ erlang:send(rexi_utils:server_pid(Node), Msg),
Ref;
false ->
cast(Node, Caller, MFA)
@@ -71,7 +69,10 @@ cast(Node, Caller, MFA, Options) ->
%% No rexi_EXIT message will be sent.
-spec kill(node(), reference()) -> ok.
kill(Node, Ref) ->
- rexi_utils:send({?SERVER, Node}, cast_msg({kill, Ref})),
+ % This first version is to tide us over during the upgrade. We'll
+ % remove it in the next commit that will be in a separate release.
+ rexi_utils:send({rexi_server, Node}, cast_msg({kill, Ref})),
+ rexi_utils:send(rexi_utils:server_pid(Node), cast_msg({kill, Ref})),
ok.
%% @equiv async_server_call(Server, self(), Request)
[26/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Add max_count to state record
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/9a1bbd09
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/9a1bbd09
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/9a1bbd09
Branch: refs/heads/master
Commit: 9a1bbd09334dc869d45eed9331cb61432f116ca5
Parents: 4551014
Author: Robert Newson <rn...@apache.org>
Authored: Tue Jun 10 14:01:07 2014 +0100
Committer: Robert Newson <rn...@apache.org>
Committed: Wed Jul 23 18:08:25 2014 +0100
----------------------------------------------------------------------
src/rexi_buffer.erl | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/9a1bbd09/src/rexi_buffer.erl
----------------------------------------------------------------------
diff --git a/src/rexi_buffer.erl b/src/rexi_buffer.erl
index 26f3c97..42b4ef2 100644
--- a/src/rexi_buffer.erl
+++ b/src/rexi_buffer.erl
@@ -26,7 +26,8 @@
-record(state, {
buffer = queue:new(),
sender = nil,
- count = 0
+ count = 0,
+ max_count
}).
start_link(ServerId) ->
[32/33] rexi commit: updated refs/heads/master to bbf59a2
Posted by rn...@apache.org.
Return ok for pattern matching in client code
Project: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/commit/a853c32c
Tree: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/tree/a853c32c
Diff: http://git-wip-us.apache.org/repos/asf/couchdb-rexi/diff/a853c32c
Branch: refs/heads/master
Commit: a853c32c5a3e5e353d169af7f4cf831f78d9ee1a
Parents: 7d64200
Author: Paul J. Davis <pa...@gmail.com>
Authored: Tue Aug 19 11:56:19 2014 -0500
Committer: Paul J. Davis <pa...@gmail.com>
Committed: Tue Aug 19 11:56:32 2014 -0500
----------------------------------------------------------------------
src/rexi.erl | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb-rexi/blob/a853c32c/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi.erl b/src/rexi.erl
index b38c149..69fbd77 100644
--- a/src/rexi.erl
+++ b/src/rexi.erl
@@ -217,7 +217,8 @@ stream_last(Msg) ->
%% the worker pid and doesn't wait for a response from the controller.
stream_last(Msg, Timeout) ->
maybe_init_stream(Timeout),
- rexi:reply(Msg).
+ rexi:reply(Msg),
+ ok.
%% @equiv stream_ack(Client, 1)
stream_ack(Client) ->