Add 'src/rexi/' from commit 'a327b7dbeb2b0050f7ca9072047bf8ef2d282833'

git-subtree-dir: src/rexi
git-subtree-mainline: 81ad7005d01fbc03bd7573aaa97f0b44c137e6c3
git-subtree-split: a327b7dbeb2b0050f7ca9072047bf8ef2d282833

13 files changed, 1085 insertions, 0 deletions

diff --git a/src/rexi/README.md b/src/rexi/README.md
new file mode 100644
index 000000000..b2eeaea2b
--- /dev/null
+++ b/src/rexi/README.md
@@ -0,0 +1,23 @@
+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
diff --git a/src/rexi/include/rexi.hrl b/src/rexi/include/rexi.hrl
new file mode 100644
index 000000000..a2d86b2ab
--- /dev/null
+++ b/src/rexi/include/rexi.hrl
@@ -0,0 +1,20 @@
+% 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
+}).
+
diff --git a/src/rexi/priv/stats_descriptions.cfg b/src/rexi/priv/stats_descriptions.cfg
new file mode 100644
index 000000000..93c29d95a
--- /dev/null
+++ b/src/rexi/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">>}
+]}.
diff --git a/src/rexi/src/rexi.app.src b/src/rexi/src/rexi.app.src
new file mode 100644
index 000000000..efe128ca0
--- /dev/null
+++ b/src/rexi/src/rexi.app.src
@@ -0,0 +1,38 @@
+% 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,
+        couch_log,
+        couch_stats,
+        config
+    ]},
+    {mod, {rexi_app,[]}}
+]}.
diff --git a/src/rexi/src/rexi.erl b/src/rexi/src/rexi.erl
new file mode 100644
index 000000000..fea4d6453
--- /dev/null
+++ b/src/rexi/src/rexi.erl
@@ -0,0 +1,286 @@
+% 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([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").
+
+start() ->
+    application:start(rexi).
+
+stop() ->
+    application:stop(rexi).
+
+restart() ->
+    stop(), start().
+
+
+%% @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(rexi_utils:server_pid(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(rexi_utils:server_pid(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(rexi_utils:server_pid(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_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 ->
+            couch_stats:increment_counter(
+                [rexi, streams, timeout, init_stream]
+            ),
+            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).
+
+%% @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 ->
+        couch_stats:increment_counter([rexi, streams, timeout, stream]),
+        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 ->
+        couch_stats:increment_counter([rexi, streams, timeout, stream]),
+        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),
+    ok.
+
+%% @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_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 ->
+            {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 ->
+        couch_stats:increment_counter([rexi, streams, timeout, wait_for_ack]),
+        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.
diff --git a/src/rexi/src/rexi_app.erl b/src/rexi/src/rexi_app.erl
new file mode 100644
index 000000000..0f1e892b5
--- /dev/null
+++ b/src/rexi/src/rexi_app.erl
@@ -0,0 +1,22 @@
+% 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]).
+
+
+start(_Type, StartArgs) ->
+    rexi_sup:start_link(StartArgs).
+
+stop(_State) ->
+    ok.
diff --git a/src/rexi/src/rexi_buffer.erl b/src/rexi/src/rexi_buffer.erl
new file mode 100644
index 000000000..d16dc8ba3
--- /dev/null
+++ b/src/rexi/src/rexi_buffer.erl
@@ -0,0 +1,104 @@
+% 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,
+    max_count
+}).
+
+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(_) ->
+    %% 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}.
+
+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}
+    end.
+
+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,
+    {{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}.
+
+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(#state{count = Count, max_count = Max}) ->
+    Count >= Max.
+
+get_node({_, Node}) when is_atom(Node) ->
+    Node;
+get_node(Pid) when is_pid(Pid) ->
+    node(Pid).
diff --git a/src/rexi/src/rexi_monitor.erl b/src/rexi/src/rexi_monitor.erl
new file mode 100644
index 000000000..da6dcf533
--- /dev/null
+++ b/src/rexi/src/rexi_monitor.erl
@@ -0,0 +1,64 @@
+% 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]).
+
+
+%% @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) ->
+    couch_stats:increment_counter([rexi, down]),
+    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.
diff --git a/src/rexi/src/rexi_server.erl b/src/rexi/src/rexi_server.erl
new file mode 100644
index 000000000..6cecdb8e7
--- /dev/null
+++ b/src/rexi/src/rexi_server.erl
@@ -0,0 +1,178 @@
+% 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).
+-vsn(1).
+-export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2,
+    code_change/3]).
+
+-export([start_link/1, init_p/2, init_p/3]).
+
+-include_lib("rexi/include/rexi.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 = 0,
+    error_count = 0
+}).
+
+start_link(ServerId) ->
+    gen_server:start_link({local, ServerId}, ?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) ->
+    couch_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{}=State, _Extra) ->
+    {ok, State}.
+
+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(),
+        couch_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{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) ->
+    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}}).
diff --git a/src/rexi/src/rexi_server_mon.erl b/src/rexi/src/rexi_server_mon.erl
new file mode 100644
index 000000000..e6b5eb98e
--- /dev/null
+++ b/src/rexi/src/rexi_server_mon.erl
@@ -0,0 +1,130 @@
+% 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
+% 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).
+-vsn(1).
+
+
+-export([
+    start_link/1,
+    status/0
+]).
+
+
+-export([
+    init/1,
+    terminate/2,
+    handle_call/3,
+    handle_cast/2,
+    handle_info/2,
+    code_change/3
+]).
+
+
+-define(INTERVAL, 60000).
+
+
+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(ChildMod) ->
+    net_kernel:monitor_nodes(true),
+    erlang:send(self(), check_nodes),
+    {ok, ChildMod}.
+
+
+terminate(_Reason, _St) ->
+    ok.
+
+
+handle_call(status, _From, ChildMod) ->
+    case missing_servers(ChildMod) of
+        [] ->
+            {reply, ok, ChildMod};
+        Missing ->
+            {reply, {waiting, length(Missing)}, ChildMod}
+    end;
+
+handle_call(Msg, _From, St) ->
+    couch_log:notice("~s ignored_call ~w", [?MODULE, Msg]),
+    {reply, ignored, St}.
+
+
+handle_cast(Msg, St) ->
+    couch_log:notice("~s ignored_cast ~w", [?MODULE, Msg]),
+    {noreply, St}.
+
+
+handle_info({nodeup, _}, ChildMod) ->
+    start_servers(ChildMod),
+    {noreply, ChildMod};
+
+handle_info({nodedown, _}, St) ->
+    {noreply, St};
+
+handle_info(check_nodes, ChildMod) ->
+    start_servers(ChildMod),
+    erlang:send_after(?INTERVAL, self(), check_nodes),
+    {noreply, ChildMod};
+
+handle_info(Msg, St) ->
+    couch_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_servers(ChildMod) ->
+    lists:foreach(fun(Id) ->
+        {ok, _} = start_server(ChildMod, Id)
+    end, missing_servers(ChildMod)).
+
+
+missing_servers(ChildMod) ->
+    ServerIds = [list_to_atom(lists:concat([ChildMod, "_", Node]))
+        || Node <- [node() | nodes()]],
+    SupModule = sup_module(ChildMod),
+    ChildIds = [Id || {Id, _, _, _} <- supervisor:which_children(SupModule)],
+    ServerIds -- ChildIds.
+
+
+start_server(ChildMod, ChildId) ->
+    ChildSpec = {
+        ChildId,
+        {ChildMod, start_link, [ChildId]},
+        permanent,
+        brutal_kill,
+        worker,
+        [ChildMod]
+    },
+    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"])).
diff --git a/src/rexi/src/rexi_server_sup.erl b/src/rexi/src/rexi_server_sup.erl
new file mode 100644
index 000000000..29c6ad60c
--- /dev/null
+++ b/src/rexi/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/1]).
+
+
+start_link(Name) ->
+    supervisor:start_link({local, Name}, ?MODULE, []).
+
+
+init([]) ->
+    {ok, {{one_for_one, 1, 1}, []}}.
diff --git a/src/rexi/src/rexi_sup.erl b/src/rexi/src/rexi_sup.erl
new file mode 100644
index 000000000..55c482998
--- /dev/null
+++ b/src/rexi/src/rexi_sup.erl
@@ -0,0 +1,64 @@
+% 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([start_link/1]).
+-export([init/1]).
+
+start_link(Args) ->
+    supervisor:start_link({local,?MODULE}, ?MODULE, Args).
+
+init([]) ->
+    {ok, {{one_for_one, 3, 10}, [
+        {
+            rexi_server,
+            {rexi_server, start_link, [rexi_server]},
+            permanent,
+            100,
+            worker,
+            [rexi_server]
+        },
+        {
+            rexi_server_sup,
+            {rexi_server_sup, start_link, [rexi_server_sup]},
+            permanent,
+            100,
+            supervisor,
+            [rexi_server_sup]
+        },
+        {
+            rexi_server_mon,
+            {rexi_server_mon, start_link, [rexi_server]},
+            permanent,
+            100,
+            worker,
+            [rexi_server_mon]
+        },
+        {
+            rexi_buffer_sup,
+            {rexi_server_sup, start_link, [rexi_buffer_sup]},
+            permanent,
+            100,
+            supervisor,
+            [rexi_server_sup]
+        },
+        {
+            rexi_buffer_mon,
+            {rexi_server_mon, start_link, [rexi_buffer]},
+            permanent,
+            100,
+            worker,
+            [rexi_server_mon]
+        }
+    ]}}.
diff --git a/src/rexi/src/rexi_utils.erl b/src/rexi/src/rexi_utils.erl
new file mode 100644
index 000000000..e3eaa6fcc
--- /dev/null
+++ b/src/rexi/src/rexi_utils.erl
@@ -0,0 +1,103 @@
+% 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([server_id/1, server_pid/1, send/2, recv/6]).
+
+%% @doc Return a rexi_server id for the given node.
+server_id(Node) ->
+    case config:get("rexi", "server_per_node", "false") of
+    "true" ->
+        list_to_atom("rexi_server_" ++ atom_to_list(Node));
+    _ ->
+        rexi_server
+    end.
+
+%% @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) ->
+    case erlang:send(Dest, Msg, [noconnect, nosuspend]) of
+    ok ->
+        ok;
+    _ ->
+        % treat nosuspend and noconnect the same
+        rexi_buffer:send(Dest, Msg)
+    end.
+
+%% @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);
+    {new_refs, NewRefList, Acc} ->
+        process_mailbox(NewRefList, 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};
+    {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 ->
+            % 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.