Initial implementation/release

.gitignore

@@ -0,0 +1,2 @@
+*.beam
+ebin/sidejob.app

rebar.config

@@ -0,0 +1,5 @@
+{erl_opts, [debug_info, warnings_as_errors]}.
+{cover_enabled, true}.
+{eunit_opts, [verbose]}.
+{edoc_opts, [{preprocess, true}]}.
+{deps, []}.

src/sidejob.app.src

@@ -0,0 +1,13 @@
+{application, sidejob,
+ [
+  {description, ""},
+  {vsn, "0.0.1"},
+  {registered, []},
+  {applications, [
+                  kernel,
+                  stdlib
+                 ]},
+  {registered, []},
+  {mod, {sidejob_app, []}},
+  {env, []}
+ ]}.

src/sidejob.erl

@@ -0,0 +1,132 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright (c) 2013 Basho Technologies, Inc.  All Rights Reserved.
+%%
+%% This file is provided to you 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(sidejob).
+-export([new_resource/3, new_resource/4, call/2, call/3, cast/2]).
+
+%%%===================================================================
+%%% API
+%%%===================================================================
+
+%% @doc
+%% Create a new sidejob resource that uses the provided worker module,
+%% enforces the requested usage limit, and is managed by the specified
+%% number of worker processes.
+%%
+%% This call will generate and load a new module, via {@link sidejob_config},
+%% that provides information about the new resource. It will also start up the
+%% supervision hierarchy that manages this resource: ensuring that the workers
+%% and stats aggregation server for this resource remain running.
+new_resource(Name, Mod, Limit, Workers) ->
+    ETS = sidejob_resource_sup:ets(Name),
+    StatsETS = sidejob_resource_sup:stats_ets(Name),
+    WorkerNames = sidejob_worker:workers(Name, Workers),
+    StatsName = sidejob_resource_stats:reg_name(Name),
+    WorkerLimit = Limit div Workers,
+    sidejob_config:load_config(Name, [{width, Workers},
+                                      {limit, Limit},
+                                      {worker_limit, WorkerLimit},
+                                      {ets, ETS},
+                                      {stats_ets, StatsETS},
+                                      {workers, list_to_tuple(WorkerNames)},
+                                      {stats, StatsName}]),
+    sidejob_sup:add_resource(Name, Mod).
+
+%% @doc
+%% Same as {@link new_resource/4} except that the number of workers defaults
+%% to the number of scheduler threads.
+new_resource(Name, Mod, Limit) ->
+    Workers = erlang:system_info(schedulers),
+    new_resource(Name, Mod, Limit, Workers).
+
+
+%% @doc
+%% Same as {@link call/3} with a default timeout of 5 seconds.
+call(Name, Msg) ->
+    call(Name, Msg, 5000).
+
+%% @doc
+%% Perform a synchronous call to the specified resource, failing if the
+%% resource has reached its usage limit.
+call(Name, Msg, Timeout) ->
+    case available(Name) of
+        none ->
+            overload;
+        Worker ->
+            gen_server:call(Worker, Msg, Timeout)
+    end.
+
+%% @doc
+%% Perform an asynchronous cast to the specified resource, failing if the
+%% resource has reached its usage limit.
+cast(Name, Msg) ->
+    case available(Name) of
+        none ->
+            overload;
+        Worker ->
+            gen_server:cast(Worker, Msg)
+    end.
+
+%%%===================================================================
+%%% Internal functions
+%%%===================================================================
+
+%% Find an available worker or return none if all workers at limit
+available(Name) ->
+    ETS = Name:ets(),
+    Width = Name:width(),
+    Limit = Name:worker_limit(),
+    Scheduler = erlang:system_info(scheduler_id),
+    Worker = Scheduler rem Width,
+    case is_available(ETS, Limit, Worker) of
+        true ->
+            worker_reg_name(Name, Worker);
+        false ->
+            available(Name, ETS, Width, Limit, Worker+1, Worker)
+    end.
+
+available(Name, _ETS, _Width, _Limit, End, End) ->
+    ets:update_counter(Name:stats_ets(), rejected, 1),
+    none;
+available(Name, ETS, Width, Limit, X, End) ->
+    Worker = X rem Width,
+    case is_available(ETS, Limit, Worker) of
+        false ->
+            available(Name, ETS, Width, Limit, Worker+1, End);
+        true ->
+            worker_reg_name(Name, Worker)
+    end.
+
+is_available(ETS, Limit, Worker) ->
+    case ets:lookup_element(ETS, {full, Worker}, 2) of
+        1 ->
+            false;
+        0 ->
+            Value = ets:update_counter(ETS, Worker, 1),
+            case Value >= Limit of
+                true ->
+                    ets:insert(ETS, {{full, Worker}, 1}),
+                    false;
+                false ->
+                    true
+            end
+    end.
+
+worker_reg_name(Name, Id) ->
+    element(Id+1, Name:workers()).

src/sidejob_app.erl

@@ -0,0 +1,48 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright (c) 2013 Basho Technologies, Inc.  All Rights Reserved.
+%%
+%% This file is provided to you 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(sidejob_app).
+-behaviour(application).
+
+%% Application callbacks
+-export([start/2, start_phase/3, prep_stop/1, stop/1, config_change/3]).
+
+%%%===================================================================
+%%% Application callbacks
+%%%===================================================================
+
+start(_Type, _StartArgs) ->
+    case sidejob_sup:start_link() of
+        {ok, Pid} ->
+            {ok, Pid};
+        Error ->
+            Error
+    end.
+
+stop(_State) ->
+    ok.
+
+start_phase(_Phase, _Type, _PhaseArgs) ->
+    ok.
+
+prep_stop(State) ->
+    State.
+
+config_change(_Changed, _New, _Removed) ->
+    ok.

src/sidejob_config.erl

@@ -0,0 +1,66 @@
+%% -------------------------------------------------------------------
+%%
+%% Copyright (c) 2013 Basho Technologies, Inc.  All Rights Reserved.
+%%
+%% This file is provided to you 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.
+%%
+%% -------------------------------------------------------------------
+
+%% @doc
+%% Utility that converts a given property list into a module that provides
+%% constant time access to the various key/value pairs.
+%%
+%% Example:
+%%   load_config(test, [{limit, 1000},
+%%                      {num_workers, 4},
+%%                      {workers, [{test_1, test_2, test_3, test_4}]}]).
+%%
+%% creates the module `test' such that:
+%%   test:limit().       => 1000
+%%   test:num_workers(). => 16
+%%   test:workers().     => [{test_1, test_2, test_3, test_4}]}]
+%%
+-module(sidejob_config).
+-export([load_config/2]).
+
+load_config(Resource, Config) ->
+    Module = make_module(Resource),
+    Exports = [make_export(Key) || {Key, _} <- Config],
+    Functions = [make_function(Key, Value) || {Key, Value} <- Config],
+    ExportAttr = make_export_attribute(Exports),
+    Abstract = [Module, ExportAttr | Functions],
+    Forms = erl_syntax:revert_forms(Abstract),
+    {ok, Resource, Bin} = compile:forms(Forms, [verbose, report_errors]),
+    code:purge(Resource),
+    {module, Resource} = code:load_binary(Resource,
+                                          atom_to_list(Resource) ++ ".erl",
+                                          Bin),
+    ok.
+
+make_module(Module) ->
+    erl_syntax:attribute(erl_syntax:atom(module),
+                         [erl_syntax:atom(Module)]).
+
+make_export(Key) ->
+    erl_syntax:arity_qualifier(erl_syntax:atom(Key),
+                               erl_syntax:integer(0)).
+
+make_export_attribute(Exports) ->
+    erl_syntax:attribute(erl_syntax:atom(export),
+                         [erl_syntax:list(Exports)]).
+
+make_function(Key, Value) ->
+    Constant = erl_syntax:clause([], none, [erl_syntax:abstract(Value)]),
+    erl_syntax:function(erl_syntax:atom(Key), [Constant]).