You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@couchdb.apache.org by da...@apache.org on 2014/02/05 15:51:02 UTC
[40/49] couchdb commit: updated refs/heads/1843-feature-bigcouch to
3069c01
Remove src/rexi
Project: http://git-wip-us.apache.org/repos/asf/couchdb/repo
Commit: http://git-wip-us.apache.org/repos/asf/couchdb/commit/834aeb0d
Tree: http://git-wip-us.apache.org/repos/asf/couchdb/tree/834aeb0d
Diff: http://git-wip-us.apache.org/repos/asf/couchdb/diff/834aeb0d
Branch: refs/heads/1843-feature-bigcouch
Commit: 834aeb0d785f7f700670222f71ba7d71c5c3ecf5
Parents: 21118e2
Author: Paul J. Davis <pa...@gmail.com>
Authored: Tue Feb 4 17:42:38 2014 -0600
Committer: Paul J. Davis <pa...@gmail.com>
Committed: Tue Feb 4 17:42:38 2014 -0600
----------------------------------------------------------------------
src/rexi/README.md | 23 ----
src/rexi/include/rexi.hrl | 20 ----
src/rexi/src/rexi.app.src | 36 -------
src/rexi/src/rexi.erl | 188 ---------------------------------
src/rexi/src/rexi_app.erl | 23 ----
src/rexi/src/rexi_gov_manager.erl | 156 ---------------------------
src/rexi/src/rexi_governor.erl | 70 ------------
src/rexi/src/rexi_monitor.erl | 64 -----------
src/rexi/src/rexi_server.erl | 188 ---------------------------------
src/rexi/src/rexi_sup.erl | 28 -----
src/rexi/src/rexi_utils.erl | 68 ------------
11 files changed, 864 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/README.md
----------------------------------------------------------------------
diff --git a/src/rexi/README.md b/src/rexi/README.md
deleted file mode 100644
index b2eeaea..0000000
--- a/src/rexi/README.md
+++ /dev/null
@@ -1,23 +0,0 @@
-Rexi is a tailor-made RPC server application for sending [CouchDB][1] operations to nodes in a cluster. It is used in [BigCouch][2] as the remote procedure vehicle to get [fabric][6] functions to execute on remote cluster nodes.
-
-Rexi better fits the needs of the BigCouch distributed data store by dropping some unneeded overhead in rex, the RPC server that ships with Erlang/OTP. Rexi is optimized for the case when you need to spawn a bunch of remote processes. Cast messages are sent from the origin to the remote rexi server, and local processes are spawned from there, which is vastly more efficient than spawning remote processes from the origin. You still get monitoring of the remote processes, but the request-handling process doesn't get stuck trying to connect to an overloaded/dead node. 'rexi_DOWN' messages will arrive at the client eventually. This has been an extremely advantageous mix of latency and failure detection, vastly improving the performance of BigCouch.
-
-Rexi is used in conjunction with 'Fabric' which is also an application within BigCouch, but can be used on a stand-alone basis.
-
-### Getting Started
-Rexi requires R13B03 or higher and can be built with [rebar][7], which comes bundled in the repository.
-
-### License
-[Apache 2.0][3]
-
-### Contact
- * [http://cloudant.com][4]
- * [info@cloudant.com][5]
-
-[1]: http://couchdb.apache.org
-[2]: http://github.com/cloudant/BigCouch
-[3]: http://www.apache.org/licenses/LICENSE-2.0.html
-[4]: http://cloudant.com
-[5]: mailto:info@cloudant.com
-[6]: http://github.com/cloudant/fabric
-[7]: http://github.com/basho/rebar
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/include/rexi.hrl
----------------------------------------------------------------------
diff --git a/src/rexi/include/rexi.hrl b/src/rexi/include/rexi.hrl
deleted file mode 100644
index a2d86b2..0000000
--- a/src/rexi/include/rexi.hrl
+++ /dev/null
@@ -1,20 +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.
-
--record(error, {
- timestamp,
- reason,
- mfa,
- nonce,
- stack
-}).
-
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/src/rexi.app.src
----------------------------------------------------------------------
diff --git a/src/rexi/src/rexi.app.src b/src/rexi/src/rexi.app.src
deleted file mode 100644
index 0691edd..0000000
--- a/src/rexi/src/rexi.app.src
+++ /dev/null
@@ -1,36 +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.
-
-{application, rexi, [
- {description, "Lightweight RPC server"},
- {vsn, git},
- {modules, [
- rexi,
- rexi_app,
- rexi_gov_manager,
- rexi_governor,
- rexi_monitor,
- rexi_server,
- rexi_sup,
- rexi_utils
- ]},
- {registered, [
- rexi_sup,
- rexi_server
- ]},
- {applications, [
- kernel,
- stdlib,
- config
- ]},
- {mod, {rexi_app,[]}}
-]}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/src/rexi.erl
----------------------------------------------------------------------
diff --git a/src/rexi/src/rexi.erl b/src/rexi/src/rexi.erl
deleted file mode 100644
index 8e53dba..0000000
--- a/src/rexi/src/rexi.erl
+++ /dev/null
@@ -1,188 +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).
--export([start/0, stop/0, restart/0]).
--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").
-
--define(SERVER, rexi_server).
-
-start() ->
- application:start(rexi).
-
-stop() ->
- application:stop(rexi).
-
-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().
-cast(Node, MFA) ->
- cast(Node, self(), MFA).
-
-%% @doc Executes apply(M, F, A) on Node.
-%% You might want to use this instead of rpc:cast/4 for two reasons. First,
-%% the Caller pid and the returned reference are inserted into the remote
-%% process' dictionary as `rexi_from', so it has a way to communicate with you.
-%% Second, the remote process is monitored. If it exits with a Reason other
-%% than normal, Caller will receive a message of the form
-%% `{Ref, {rexi_EXIT, Reason}}' where Ref is the returned reference.
--spec cast(node(), pid(), {atom(), atom(), list()}) -> reference().
-cast(Node, Caller, MFA) ->
- Ref = make_ref(),
- Msg = cast_msg({doit, {Caller, Ref}, get(nonce), MFA}),
- rexi_utils:send({?SERVER, Node}, Msg),
- Ref.
-
-%% @doc Executes apply(M, F, A) on Node.
-%% This version accepts a sync option which uses the erlang:send/2 call
-%% directly in process instead of deferring to a spawned process if
-%% erlang:send/2 were to block. If the sync option is omitted this call
-%% is identical to cast/3.
--spec cast(node(), pid(), {atom(), atom(), list()}, [atom()]) -> reference().
-cast(Node, Caller, MFA, Options) ->
- case lists:member(sync, Options) of
- true ->
- Ref = make_ref(),
- Msg = cast_msg({doit, {Caller, Ref}, get(nonce), MFA}),
- erlang:send({?SERVER, Node}, Msg),
- Ref;
- false ->
- cast(Node, Caller, MFA)
- end.
-
-%% @doc Sends an async kill signal to the remote process associated with Ref.
-%% No rexi_EXIT message will be sent.
--spec kill(node(), reference()) -> ok.
-kill(Node, Ref) ->
- rexi_utils:send({?SERVER, Node}, cast_msg({kill, Ref})),
- ok.
-
-%% @equiv async_server_call(Server, self(), Request)
--spec async_server_call(pid() | {atom(),node()}, any()) -> reference().
-async_server_call(Server, Request) ->
- async_server_call(Server, self(), Request).
-
-%% @doc Sends a properly formatted gen_server:call Request to the Server and
-%% returns the reference which the Server will include in its reply. The
-%% function acts more like cast() than call() in that the server process
-%% is not monitored. Clients who want to know if the server is alive should
-%% monitor it themselves before calling this function.
--spec async_server_call(pid() | {atom(),node()}, pid(), any()) -> reference().
-async_server_call(Server, Caller, Request) ->
- Ref = make_ref(),
- rexi_utils:send(Server, {'$gen_call', {Caller,Ref}, Request}),
- Ref.
-
-%% @doc convenience function to reply to the original rexi Caller.
--spec reply(any()) -> any().
-reply(Reply) ->
- {Caller, Ref} = get(rexi_from),
- erlang:send(Caller, {Ref,Reply}).
-
-%% @equiv sync_reply(Reply, 300000)
-sync_reply(Reply) ->
- sync_reply(Reply, 300000).
-
-%% @doc convenience function to reply to caller and wait for response. Message
-%% is of the form {OriginalRef, {self(),reference()}, Reply}, which enables the
-%% original caller to respond back.
--spec sync_reply(any(), pos_integer() | infinity) -> any().
-sync_reply(Reply, Timeout) ->
- {Caller, Ref} = get(rexi_from),
- Tag = make_ref(),
- erlang:send(Caller, {Ref, {self(),Tag}, Reply}),
- receive {Tag, Response} ->
- Response
- after Timeout ->
- timeout
- end.
-
-%% @equiv stream(Msg, 100, 300000)
-stream(Msg) ->
- stream(Msg, 10, 300000).
-
-%% @equiv stream(Msg, Limit, 300000)
-stream(Msg, Limit) ->
- stream(Msg, Limit, 300000).
-
-%% @doc convenience function to stream messages to caller while blocking when
-%% a specific number of messages are outstanding. Message is of the form
-%% {OriginalRef, self(), Reply}, which enables the original caller to ack.
--spec stream(any(), integer(), pos_integer() | infinity) -> any().
-stream(Msg, Limit, 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 ->
- timeout
- end.
-
-%% @equiv stream_ack(Client, 1)
-stream_ack(Client) ->
- erlang:send(Client, {rexi_ack, 1}).
-
-%% @doc Ack streamed messages
-stream_ack(Client, N) ->
- erlang:send(Client, {rexi_ack, N}).
-
-%% internal functions %%
-
-cast_msg(Msg) -> {'$gen_cast', Msg}.
-
-maybe_wait(Limit, Timeout) ->
- case get(rexi_unacked) of
- undefined ->
- {ok, 0};
- Count when Count >= Limit ->
- wait_for_ack(Count, Timeout);
- Count ->
- drain_acks(Count)
- end.
-
-wait_for_ack(Count, Timeout) ->
- receive
- {rexi_ack, N} -> drain_acks(Count-N)
- after Timeout ->
- throw(timeout)
- end.
-
-drain_acks(Count) when Count < 0 ->
- erlang:error(mismatched_rexi_ack);
-drain_acks(Count) ->
- receive
- {rexi_ack, N} -> drain_acks(Count-N)
- after 0 ->
- {ok, Count}
- end.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/src/rexi_app.erl
----------------------------------------------------------------------
diff --git a/src/rexi/src/rexi_app.erl b/src/rexi/src/rexi_app.erl
deleted file mode 100644
index c791d83..0000000
--- a/src/rexi/src/rexi_app.erl
+++ /dev/null
@@ -1,23 +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_app).
--behaviour(application).
--export([start/2, stop/1]).
-
--include_lib("eunit/include/eunit.hrl").
-
-start(_Type, StartArgs) ->
- rexi_sup:start_link(StartArgs).
-
-stop(_State) ->
- ok.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/src/rexi_gov_manager.erl
----------------------------------------------------------------------
diff --git a/src/rexi/src/rexi_gov_manager.erl b/src/rexi/src/rexi_gov_manager.erl
deleted file mode 100644
index 4fb69e3..0000000
--- a/src/rexi/src/rexi_gov_manager.erl
+++ /dev/null
@@ -1,156 +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).
--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/blob/834aeb0d/src/rexi/src/rexi_governor.erl
----------------------------------------------------------------------
diff --git a/src/rexi/src/rexi_governor.erl b/src/rexi/src/rexi_governor.erl
deleted file mode 100644
index ad62150..0000000
--- a/src/rexi/src/rexi_governor.erl
+++ /dev/null
@@ -1,70 +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).
-
-% gen_server callbacks
--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}};
-
-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}.
-
-handle_info({'DOWN', _, process, Pid, normal},
- #state{pids = Pids} = State) ->
- ets:delete(Pids, Pid),
- {noreply, State};
-
-handle_info({'DOWN', _, process, _Pid, killed}, State) ->
- {noreply, State}.
-
-terminate(_Reason, _State) ->
- ok.
-
-code_change(_OldVsn, State, _Extra) ->
- {ok, State}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/src/rexi_monitor.erl
----------------------------------------------------------------------
diff --git a/src/rexi/src/rexi_monitor.erl b/src/rexi/src/rexi_monitor.erl
deleted file mode 100644
index 7be3f0a..0000000
--- a/src/rexi/src/rexi_monitor.erl
+++ /dev/null
@@ -1,64 +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_monitor).
--export([start/1, stop/1]).
-
--include_lib("eunit/include/eunit.hrl").
-
-%% @doc spawn_links a process which monitors the supplied list of items and
-%% returns the process ID. If a monitored process exits, the caller will
-%% receive a {rexi_DOWN, MonitoringPid, DeadPid, Reason} message.
--spec start([pid() | atom() | {atom(),node()}]) -> pid().
-start(Procs) ->
- Parent = self(),
- Nodes = [node() | nodes()],
- {Mon, Skip} = lists:partition(fun(P) -> should_monitor(P, Nodes) end,
- Procs),
- spawn_link(fun() ->
- [notify_parent(Parent, P, noconnect) || P <- Skip],
- [erlang:monitor(process, P) || P <- Mon],
- wait_monitors(Parent)
- end).
-
-%% @doc Cleanly shut down the monitoring process and flush all rexi_DOWN
-%% messages from our mailbox.
--spec stop(pid()) -> ok.
-stop(MonitoringPid) ->
- MonitoringPid ! {self(), shutdown},
- flush_down_messages().
-
-%% internal functions %%
-
-notify_parent(Parent, Pid, Reason) ->
- erlang:send(Parent, {rexi_DOWN, self(), Pid, Reason}).
-
-should_monitor(Pid, Nodes) when is_pid(Pid) ->
- lists:member(node(Pid), Nodes);
-should_monitor({_, Node}, Nodes) ->
- lists:member(Node, Nodes).
-
-wait_monitors(Parent) ->
- receive
- {'DOWN', _, process, Pid, Reason} ->
- notify_parent(Parent, Pid, Reason),
- wait_monitors(Parent);
- {Parent, shutdown} ->
- ok
- end.
-
-flush_down_messages() ->
- receive {rexi_DOWN, _, _, _} ->
- flush_down_messages()
- after 0 ->
- ok
- end.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/src/rexi_server.erl
----------------------------------------------------------------------
diff --git a/src/rexi/src/rexi_server.erl b/src/rexi/src/rexi_server.erl
deleted file mode 100644
index c3fc508..0000000
--- a/src/rexi/src/rexi_server.erl
+++ /dev/null
@@ -1,188 +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_server).
--behaviour(gen_server).
--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]).
-
--include_lib("rexi/include/rexi.hrl").
--include_lib("eunit/include/eunit.hrl").
-
--record(job, {
- client::reference(),
- worker::reference(),
- client_pid::pid(),
- worker_pid::pid()
-}).
-
--record(st, {
- workers = ets:new(workers, [private, {keypos, #job.worker}]),
- clients = ets:new(clients, [private, {keypos, #job.client}]),
- errors = queue:new(),
- error_limit = 20,
- error_count = 0
-}).
-
-start_link() ->
- gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
-
-init([]) ->
- {ok, #st{}}.
-
-handle_call(get_errors, _From, #st{errors = Errors} = St) ->
- {reply, {ok, lists:reverse(queue:to_list(Errors))}, St};
-
-handle_call(get_last_error, _From, #st{errors = Errors} = St) ->
- try
- {reply, {ok, queue:get_r(Errors)}, St}
- catch error:empty ->
- {reply, {error, empty}, St}
- end;
-
-handle_call({set_error_limit, N}, _From, #st{error_count=Len, errors=Q} = St) ->
- if N < Len ->
- {NewQ, _} = queue:split(N, Q);
- true ->
- NewQ = Q
- end,
- NewLen = queue:len(NewQ),
- {reply, ok, St#st{error_limit=N, error_count=NewLen, errors=NewQ}};
-
-handle_call(_Request, _From, St) ->
- {reply, ignored, St}.
-
-
-handle_cast({doit, From, MFA}, St) ->
- handle_cast({doit, From, undefined, MFA}, St);
-
-handle_cast({doit, {ClientPid, ClientRef} = From, Nonce, MFA}, State) ->
- {LocalPid, Ref} = spawn_monitor(?MODULE, init_p, [From, MFA, Nonce]),
- Job = #job{
- client = ClientRef,
- worker = Ref,
- client_pid = ClientPid,
- worker_pid = LocalPid
- },
- {noreply, add_job(Job, State)};
-
-
-handle_cast({kill, FromRef}, #st{clients = Clients} = St) ->
- case find_worker(FromRef, Clients) of
- #job{worker = KeyRef, worker_pid = Pid} = Job ->
- erlang:demonitor(KeyRef),
- exit(Pid, kill),
- {noreply, remove_job(Job, St)};
- false ->
- {noreply, St}
- end;
-
-handle_cast(_, St) ->
- twig:log(notice, "rexi_server ignored_cast"),
- {noreply, St}.
-
-handle_info({'DOWN', Ref, process, _, normal}, #st{workers=Workers} = St) ->
- case find_worker(Ref, Workers) of
- #job{} = Job ->
- {noreply, remove_job(Job, St)};
- false ->
- {noreply, St}
- end;
-
-handle_info({'DOWN', Ref, process, Pid, Error}, #st{workers=Workers} = St) ->
- case find_worker(Ref, Workers) of
- #job{worker_pid=Pid, worker=Ref, client_pid=CPid, client=CRef} =Job ->
- case Error of #error{reason = {_Class, Reason}, stack = Stack} ->
- notify_caller({CPid, CRef}, {Reason, Stack}),
- St1 = save_error(Error, St),
- {noreply, remove_job(Job, St1)};
- _ ->
- notify_caller({CPid, CRef}, Error),
- {noreply, remove_job(Job, St)}
- end;
- false ->
- {noreply, St}
- end;
-
-handle_info(_Info, St) ->
- {noreply, St}.
-
-terminate(_Reason, St) ->
- ets:foldl(fun(#job{worker_pid=Pid},_) -> exit(Pid,kill) end, nil,
- 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}.
-
-init_p(From, MFA) ->
- init_p(From, MFA, undefined).
-
-%% @doc initializes a process started by rexi_server.
--spec init_p({pid(), reference()}, {atom(), atom(), list()},
- string() | undefined) -> any().
-init_p(From, {M,F,A}, Nonce) ->
- put(rexi_from, From),
- put(initial_call, {M,F,length(A)}),
- put(nonce, Nonce),
- try apply(M, F, A) catch exit:normal -> ok; Class:Reason ->
- Stack = clean_stack(),
- twig:log(error, "rexi_server ~p:~p ~100p", [Class, Reason, Stack]),
- exit(#error{
- timestamp = now(),
- reason = {Class, Reason},
- mfa = {M,F,A},
- nonce = Nonce,
- stack = Stack
- })
- end.
-
-%% internal
-
-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) ->
- St#st{errors = queue:in(E, Q), error_count = C+1}.
-
-clean_stack() ->
- lists:map(fun({M,F,A}) when is_list(A) -> {M,F,length(A)}; (X) -> X end,
- erlang:get_stacktrace()).
-
-add_job(Job, #st{workers = Workers, clients = Clients} = State) ->
- ets:insert(Workers, Job),
- ets:insert(Clients, Job),
- State.
-
-remove_job(Job, #st{workers = Workers, clients = Clients} = State) ->
- ets:delete_object(Workers, Job),
- ets:delete_object(Clients, Job),
- State.
-
-find_worker(Ref, Tab) ->
- case ets:lookup(Tab, Ref) of [] -> false; [Worker] -> Worker end.
-
-notify_caller({Caller, Ref}, Reason) ->
- rexi_utils:send(Caller, {Ref, {rexi_EXIT, Reason}}).
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/src/rexi_sup.erl
----------------------------------------------------------------------
diff --git a/src/rexi/src/rexi_sup.erl b/src/rexi/src/rexi_sup.erl
deleted file mode 100644
index a8aa800..0000000
--- a/src/rexi/src/rexi_sup.erl
+++ /dev/null
@@ -1,28 +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_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]}).
-
-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)]}}.
http://git-wip-us.apache.org/repos/asf/couchdb/blob/834aeb0d/src/rexi/src/rexi_utils.erl
----------------------------------------------------------------------
diff --git a/src/rexi/src/rexi_utils.erl b/src/rexi/src/rexi_utils.erl
deleted file mode 100644
index 1b11576..0000000
--- a/src/rexi/src/rexi_utils.erl
+++ /dev/null
@@ -1,68 +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_utils).
-
--export([send/2, recv/6]).
-
-%% @doc send a message as quickly as possible
-send(Dest, Msg) ->
- rexi_gov_manager:send(Dest, Msg).
-
-%% @doc set up the receive loop with an overall timeout
--spec recv([any()], integer(), function(), any(), timeout(), timeout()) ->
- {ok, any()} | {timeout, any()} | {error, atom()} | {error, atom(), any()}.
-recv(Refs, Keypos, Fun, Acc0, infinity, PerMsgTO) ->
- process_mailbox(Refs, Keypos, Fun, Acc0, nil, PerMsgTO);
-recv(Refs, Keypos, Fun, Acc0, GlobalTimeout, PerMsgTO) ->
- TimeoutRef = erlang:make_ref(),
- TRef = erlang:send_after(GlobalTimeout, self(), {timeout, TimeoutRef}),
- try
- process_mailbox(Refs, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO)
- after
- erlang:cancel_timer(TRef)
- end.
-
-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);
- {stop, Acc} ->
- {ok, Acc};
- Error ->
- Error
- end.
-
-process_message(RefList, Keypos, Fun, Acc0, TimeoutRef, PerMsgTO) ->
- receive
- {timeout, TimeoutRef} ->
- {timeout, Acc0};
- {Ref, Msg} ->
- case lists:keyfind(Ref, Keypos, RefList) of
- false ->
- % this was some non-matching message which we will ignore
- {ok, Acc0};
- Worker ->
- Fun(Msg, Worker, Acc0)
- end;
- {Ref, From, Msg} ->
- case lists:keyfind(Ref, Keypos, RefList) of
- false ->
- {ok, Acc0};
- Worker ->
- Fun(Msg, {Worker, From}, Acc0)
- end;
- {rexi_DOWN, _, _, _} = Msg ->
- Fun(Msg, nil, Acc0)
- after PerMsgTO ->
- {timeout, Acc0}
- end.