rexi_governor.erl

% Copyright 2012 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_governor).

-behaviour(gen_server).

%% API
-export([start_link/0, send_on_behalf/2]).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2, handle_info/2,
         terminate/2, code_change/3]).

-define(SERVER, ?MODULE).

-define(SPAWN_MAX,1000).

-record(state, {spawn_cnt = 0}).

start_link() ->
    gen_server:start_link({local, ?SERVER}, ?MODULE, [], []).

send_on_behalf(Dest, Msg) ->
    gen_server:cast(?SERVER, {spawn_and_track, Dest, Msg}).

init([]) ->
    erlang:system_monitor(self(), [busy_dist_port]),
    {ok, #state{}}.

handle_call(_Request, _From, State) ->
    Reply = ok,
    {reply, Reply, State}.

handle_cast({spawn_and_track, {_, _Node} = Dest, Msg},
            #state{spawn_cnt=SC}=State) ->
    % go ahead and send if possible
    case erlang:send_nosuspend(Dest, Msg) of
    true -> ok;
    false when SC < ?SPAWN_MAX ->
        % would have suspended, spawn and monitor
        spawn_monitor(erlang, send, [Dest, Msg]);
    false ->
        ok
    end,
    {noreply, State}.

handle_info({monitor, _, busy_dist_port, _}, #state{spawn_cnt = Cnt} = State) ->
    {noreply, State#state{spawn_cnt = Cnt + 1}};

handle_info({'DOWN', _, process, _, _}, #state{spawn_cnt = Cnt} = State) ->
    {noreply, State#state{spawn_cnt = Cnt - 1}}.

terminate(_Reason, _State) ->
    ok.

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.