working external handlers from config

Author: mojombo

bin/ernie

@@ -42,18 +42,14 @@ OptionParser.new do |opts|
   opts.banner = help
   opts.version = version
 
-  opts.on("-h HANDLER", "--handler HANDLER", "Handler ruby file") do |x|
-    options[:handler] = x
+  opts.on("-c CONFIG", "--config CONFIG", "Config file") do |x|
+    options[:config] = x
   end
 
   opts.on("-p PORT", "--port PORT", "Port") do |x|
     options[:port] = x
   end
 
-  opts.on("-n NUMBER", "--number NUMBER", "Number of handler instances") do |x|
-    options[:number] = x
-  end
-
   opts.on("-l LOGLEVEL", "--log-level LOGLEVEL", "Log level (0-4)") do |x|
     options[:log_level] = x
   end
@@ -81,14 +77,13 @@ if command = ARGV[0]
   svc = BERTRPC::Service.new('localhost', port)
   puts svc.call.__admin__.send(command.gsub(/-/, '_'))
 else
-  if !options[:handler]
-    puts "A handler must be specified: ernie -h /path/to/handler.rb"
+  if !options[:config]
+    puts "A config file must be specified: ernie -c /path/to/config.yml"
     exit(1)
   end
 
-  handler = options[:handler]
+  config = options[:config]
   port = options[:port] || DEFAULT_PORT
-  number = options[:number] || 1
   log_level = options[:log_level] || 2
   pidfile = options[:pidfile] ? "-ernie_server_app pidfile \"'#{options[:pidfile]}'\"" : ''
   detached = options[:detached] ? '-detached' : ''
@@ -101,8 +96,7 @@ else
                #{code_paths}
                #{pidfile} \
                -ernie_server_app port #{port} \
-               -ernie_server_app handler '"#{handler}"' \
-               -ernie_server_app number #{number} \
+               -ernie_server_app config '"#{config}"' \
                -ernie_server_app log_level #{log_level} \
                -run ernie_server_app boot}.squeeze(' ')
   puts cmd

elib/asset_pool.erl

@@ -2,7 +2,7 @@
 -behaviour(gen_server).
 
 %% api
--export([start_link/1, start/1, lease/0, return/1, reload_assets/0, idle_worker_count/0]).
+-export([start_link/2, lease/1, return/2, reload_assets/1, idle_worker_count/1]).
 
 %% gen_server callbacks
 -export([init/1, handle_call/3, handle_cast/2, handle_info/2,
@@ -16,23 +16,20 @@
 %% API
 %%====================================================================
 
-start_link(Args) ->
-  gen_server:start_link({global, ?MODULE}, ?MODULE, Args, []).
+start_link(Handler, Count) ->
+  gen_server:start_link(?MODULE, [Handler, Count], []).
 
-start(Args) ->
-  gen_server:start({global, ?MODULE}, ?MODULE, Args, []).
+lease(Pid) ->
+  gen_server:call(Pid, lease).
 
-lease() ->
-  gen_server:call({global, ?MODULE}, {lease}).
+return(Pid, Asset) ->
+  gen_server:call(Pid, {return, Asset}).
 
-return(Asset) ->
-  gen_server:call({global, ?MODULE}, {return, Asset}).
+reload_assets(Pid) ->
+  gen_server:call(Pid, {reload_assets}).
 
-reload_assets() ->
-  gen_server:call({global, ?MODULE}, {reload_assets}).
-
-idle_worker_count() ->
-  gen_server:call({global, ?MODULE}, {idle_worker_count}).
+idle_worker_count(Pid) ->
+  gen_server:call(Pid, {idle_worker_count}).
 
 %%====================================================================
 %% gen_server callbacks
@@ -45,11 +42,12 @@ idle_worker_count() ->
 %%                         {stop, Reason}
 %% Description: Initiates the server
 %%--------------------------------------------------------------------
-init([Count, Handler]) ->
+init([Handler, Count]) ->
   process_flag(trap_exit, true),
   error_logger:info_msg("~p starting~n", [?MODULE]),
   Token = make_ref(),
   Assets = start_handlers(Count, Handler, Token),
+  logger:debug("Assets = ~p~n", [Assets]),
   {ok, #state{assets = Assets, handler = Handler, token = Token}}.
 
 %%--------------------------------------------------------------------
@@ -61,7 +59,8 @@ init([Count, Handler]) ->
 %%                                      {stop, Reason, State}
 %% Description: Handling call messages
 %%--------------------------------------------------------------------
-handle_call({lease}, _From, State) ->
+handle_call(lease, _From, State) ->
+  logger:debug("Leasing...~n", []),
   Token = State#state.token,
   case queue:out(State#state.assets) of
     {{value, Asset}, Assets2} ->
@@ -140,4 +139,4 @@ start_handlers(Assets, Count, Handler, Token) ->
   start_handlers(Assets2, Count - 1, Handler, Token).
 
 create_asset(Handler, Token) ->
-  {asset, port_wrapper:wrap_link("ruby " ++ Handler), Token}.
+  {asset, port_wrapper:wrap_link(Handler), Token}.

elib/asset_pool_sup.erl

@@ -1,15 +1,13 @@
 -module(asset_pool_sup).
 -behaviour(supervisor).
--export([start_link/0, init/1]).
+-export([start_link/2, init/1]).
 
-start_link() ->
-  supervisor:start_link({local, ?MODULE}, ?MODULE, []).
+start_link(Handler, Number) ->
+  supervisor:start_link(?MODULE, [Handler, Number]).
 
-init([]) ->
-  {ok, Handler} = application:get_env(ernie_server_app, handler),
+init([Handler, Number]) ->
   io:format("Using handler ~p~n", [Handler]),
-  {ok, Number} = application:get_env(ernie_server_app, number),
   io:format("Using ~p handler instances~n", [Number]),
   {ok, {{one_for_one, 1, 60},
-    [{asset_pool, {asset_pool, start_link, [[Number, Handler]]},
+    [{asset_pool, {asset_pool, start_link, [Handler, Number]},
     permanent, brutal_kill, worker, [asset_pool]}]}}.

elib/config.erl

@@ -0,0 +1,28 @@
+-module(config).
+-export([load/1]).
+
+load(ConfigFile) ->
+  {ok, Configs} = file:consult(ConfigFile),
+  Configs2 = lists:map((fun load_single/1), Configs),
+  {ok, Configs2}.
+
+load_single(Config) ->
+  case proplists:get_value(type, Config) of
+    native ->
+      verify(native, Config),
+      CodePaths = proplists:get_value(codepaths, Config),
+      lists:each(fun(X) -> code:add_patha(X) end, CodePaths),
+      Config;
+    extern ->
+      verify(extern, Config),
+      Handler = proplists:get_value(command, Config),
+      Number = proplists:get_value(count, Config),
+      {ok, SupPid} = asset_pool_sup:start_link(Handler, Number),
+      [{_Id, ChildPid, _Type, _Modules}] = supervisor:which_children(SupPid),
+      [{pid, ChildPid} | Config]
+  end.
+
+verify(native, _Config) ->
+  ok;
+verify(extern, _Config) ->
+  ok.

elib/ernie_server.erl

@@ -10,7 +10,8 @@
 
 -record(state, {lsock = undefined,
                 pending = queue:new(),
-                count = 0}).
+                count = 0,
+                map = undefined}).
 
 -record(request, {sock = undefined,
                   info = undefined,
@@ -43,12 +44,14 @@ asset_freed() ->
 %%                         {stop, Reason}
 %% Description: Initiates the server
 %%--------------------------------------------------------------------
-init([Port]) ->
+init([Port, Configs]) ->
   process_flag(trap_exit, true),
   error_logger:info_msg("~p starting~n", [?MODULE]),
   {ok, LSock} = try_listen(Port, 500),
   spawn(fun() -> loop(LSock) end),
-  {ok, #state{lsock = LSock}}.
+  Map = init_map(Configs),
+  io:format("pidmap = ~p~n", [Map]),
+  {ok, #state{lsock = LSock, map = Map}}.
 
 %%--------------------------------------------------------------------
 %% Function: %% handle_call(Request, From, State) -> {reply, Reply, State} |
@@ -73,19 +76,11 @@ handle_cast({process, Sock}, State) ->
   State2 = receive_term(Request, State),
   {noreply, State2};
 handle_cast({asset_freed}, State) ->
-  case queue:is_empty(State#state.pending) of
-    false ->
-      case asset_pool:lease() of
-        {ok, Asset} ->
-          {{value, Request}, Pending2} = queue:out(State#state.pending),
-          % io:format("d", []),
-          spawn(fun() -> process_now(Request, Asset) end),
-          {noreply, State#state{pending = Pending2}};
-        empty ->
-          % io:format(".", []),
-          {noreply, State}
-      end;
-    true ->
+  case queue:out(State#state.pending) of
+    {{value, Request}, Pending2} ->
+      State2 = process_request(Request, State#state{pending = Pending2}),
+      {noreply, State2};
+    {empty, _Pending} ->
       {noreply, State}
   end;
 handle_cast(_Msg, State) -> {noreply, State}.
@@ -101,6 +96,14 @@ code_change(_OldVersion, State, _Extra) -> {ok, State}.
 %% Internal
 %%====================================================================
 
+extract_mapping(Config) ->
+  Pid = proplists:get_value(pid, Config),
+  Mods = proplists:get_value(modules, Config),
+  lists:map(fun(X) -> {X, Pid} end, Mods).
+
+init_map(Configs) ->
+  lists:foldl(fun(X, Acc) -> Acc ++ extract_mapping(X) end, [], Configs).
+
 try_listen(Port, 0) ->
   error_logger:error_msg("Could not listen on port ~p~n", [Port]),
   {error, "Could not listen on port"};
@@ -166,45 +169,64 @@ receive_term(Request, State) ->
 
 process_request(Request, State) ->
   ActionTerm = binary_to_term(Request#request.action),
+  close_if_cast(ActionTerm, Request),
+  {_Type, Mod, _Fun, _Args} = ActionTerm,
+  Pid = proplists:get_value(Mod, State#state.map),
+  case Pid of
+    undefined ->
+      logger:debug("Dispatching to native module~n", []),
+      process_native_request(Request, State);
+    ValidPid ->
+      logger:debug("Found extern pid ~p~n", [ValidPid]),
+      process_extern_request(ValidPid, Request, State)
+  end.
+
+close_if_cast(ActionTerm, Request) ->
   case ActionTerm of
     {cast, _Mod, _Fun, _Args} ->
       Sock = Request#request.sock,
       gen_tcp:send(Sock, term_to_binary({noreply})),
       ok = gen_tcp:close(Sock),
-      logger:debug("Closing cast.~n", []);
+      logger:debug("Closed cast.~n", []);
     _Any ->
       ok
-  end,
+  end.
+
+process_native_request(_Request, State) ->
+  State.
+
+process_extern_request(Pid, Request, State) ->
   case queue:is_empty(State#state.pending) of
     false ->
+      logger:debug("Pre queueing request for pool ~p~n", [Pid]),
       Pending2 = queue:in(Request, State#state.pending),
-      % io:format("Q", []),
       State#state{pending = Pending2};
     true ->
-      try_process_now(Request, State)
+      try_process_now(Pid, Request, State)
   end.
 
-try_process_now(Request, State) ->
+try_process_now(Pid, Request, State) ->
   Count = State#state.count,
   State2 = State#state{count = Count + 1},
-  case asset_pool:lease() of
+  logger:debug("Count = ~p~n", [Count + 1]),
+  case asset_pool:lease(Pid) of
     {ok, Asset} ->
-      % io:format("i", []),
-      spawn(fun() -> process_now(Request, Asset) end),
+      logger:debug("Leased asset for pool ~p~n", [Pid]),
+      spawn(fun() -> process_now(Pid, Request, Asset) end),
       State2;
     empty ->
-      % io:format("q", []),
+      logger:debug("Post queueing request for pool ~p~n", [Pid]),
       Pending2 = queue:in(Request, State#state.pending),
       State2#state{pending = Pending2}
   end.
 
-process_now(Request, Asset) ->
+process_now(Pid, Request, Asset) ->
   try unsafe_process_now(Request, Asset) of
     _AnyResponse -> ok
   catch
     _AnyError -> ok
   after
-    asset_pool:return(Asset),
+    asset_pool:return(Pid, Asset),
     ernie_server:asset_freed(),
     gen_tcp:close(Request#request.sock)
   end.

elib/ernie_server_app.erl

@@ -8,8 +8,7 @@ boot() ->
 
 start(_Type, _Args) ->
   logger_sup:start_link(),
-  ernie_server_sup:start_link(),
-  asset_pool_sup:start_link().
+  ernie_server_sup:start_link().
 
 stop(_State) ->
   ok.

elib/ernie_server_sup.erl

@@ -14,6 +14,9 @@ init([]) ->
       ok = file:write_file(Location, list_to_binary(Pid));
     undefined -> ok
   end,
+  {ok, Config} = application:get_env(ernie_server_app, config),
+  {ok, Configs} = config:load(Config),
+  io:format("~p~n", [Configs]),
   {ok, {{one_for_one, 1, 60},
-    [{ernie_server, {ernie_server, start_link, [[Port]]},
+    [{ernie_server, {ernie_server, start_link, [[Port, Configs]]},
     permanent, brutal_kill, worker, [ernie_server]}]}}.

examples/dsl.cfg

@@ -0,0 +1 @@
+[{modules, [test]}, {type, extern}, {command, "ruby examples/dsl.rb"}, {count, 2}].