Merge pull request #2 from fernandoareias/feature/add-ranch

Feature/add ranch

Author: fernandoareias

load-tests/test-script.js

@@ -2,8 +2,8 @@ import http from 'k6/http';
 import { check } from 'k6';
 
 export let options = {
-  vus: 10, // Número de usuários virtuais
-  duration: '60s',
+  vus: 1000, // Número de usuários virtuais
+  duration: '300s',
 };
 
 export default function () {

rebar.config

@@ -1,5 +1,7 @@
 {erl_opts, [debug_info]}.
-{deps, []}.
+{deps, [
+    {ranch, "2.1.0"}
+]}.
 
 {shell, [
     %% {config, "config/sys.config"},

rebar.lock

@@ -1 +1,8 @@
-[].
+{"1.2.0",
+[{<<"ranch">>,{pkg,<<"ranch">>,<<"2.1.0">>},0}]}.
+[
+{pkg_hash,[
+ {<<"ranch">>, <<"2261F9ED9574DCFCC444106B9F6DA155E6E540B2F82BA3D42B339B93673B72A3">>}]},
+{pkg_hash_ext,[
+ {<<"ranch">>, <<"244EE3FA2A6175270D8E1FC59024FD9DBC76294A321057DE8F803B1479E76916">>}]}
+].

src/lib/stage_behaviour.erl

@@ -11,6 +11,9 @@
 
 -define(is_false(X), ((X) == false)).
 
+-define(BACKPRESSURE_THRESHOLD, 50).
+-define(BACKPRESSURE_ATTEMPTS, 5).
+
 -callback handle_command(Command :: term()) ->
     {ok, Result :: term()} | {error, Reason :: term()} |
     {forward, NextStagePid :: pid(), NewCommand :: term()}.
@@ -39,8 +42,12 @@ loop(StageModule, _)  ->
         {command, Command, From} ->
             StageKey = list_to_atom(atom_to_list(StageModule) ++ "_workers"),
             [{StageKey, WorkersList}] = ets:lookup(erlda_workers_table, StageKey),
-            WorkerPid = pick_random_worker(WorkersList),
-            WorkerPid ! {work, Command, From},
+            case pick_worker_under_threshold(WorkersList, ?BACKPRESSURE_THRESHOLD, ?BACKPRESSURE_ATTEMPTS) of
+                {ok, WorkerPid} ->
+                    WorkerPid ! {work, Command, From};
+                busy ->
+                    From ! {stage_error, busy}
+            end,
             loop(StageModule, true);
         {worker_ready, _WorkerPid} ->
             loop(StageModule, true);
@@ -72,13 +79,30 @@ start_workers(StageModule, Count) when Count > 0 ->
     [spawn_worker(StageModule) || _ <- lists:seq(1, Count)].
 
 spawn_worker(StageModule) ->
-    worker:spawn_worker(StageModule).
+    stage_worker:spawn_worker(StageModule).
 
 pick_random_worker(Workers) ->
     N = length(Workers),
     Index = rand:uniform(N),
     lists:nth(Index, Workers).
 
+pick_worker_under_threshold(_Workers, _Threshold, Attempts) when Attempts =< 0 ->
+    busy;
+pick_worker_under_threshold([], _Threshold, _Attempts) ->
+    busy;
+pick_worker_under_threshold(Workers, Threshold, Attempts) ->
+    WorkerPid = pick_random_worker(Workers),
+    case queue_len(WorkerPid) < Threshold of
+        true -> {ok, WorkerPid};
+        false -> pick_worker_under_threshold(Workers, Threshold, Attempts - 1)
+    end.
+
+queue_len(Pid) when is_pid(Pid) ->
+    case process_info(Pid, message_queue_len) of
+        {message_queue_len, N} -> N;
+        _ -> 0
+    end.
+
 
 add_worker(StageModule) ->
     StageKey = list_to_atom(atom_to_list(StageModule) ++ "_workers"),

src/lib/stage_controller.erl

@@ -5,17 +5,22 @@
 -record(controller_state, {
     stage_module,
     stage_pid,
-    min_workers = 1,
-    max_workers = 20,
-    scale_up_threshold = 10,  
-    scale_down_threshold = 2,
-    check_interval = 5000      % 5s
+    min_workers = 10,
+    max_workers = 200,
+    scale_up_threshold = 100,  
+    scale_down_threshold = 10,
+    check_interval = 1000,     % 1s
+    consecutive_hot = 0,
+    consecutive_cold = 0,
+    last_action_ms = 0,
+    action_cooldown_ms = 2000,
+    min_samples_to_act = 1
 }).
 
 spawn_controller(StageModule, StagePid) ->
     State = #controller_state{stage_module = StageModule, stage_pid = StagePid},
     Pid = spawn(?MODULE, loop, [State]),
-    ControllerName = list_to_atom(atom_to_list(StageModule) ++ "_controler"),
+    ControllerName = list_to_atom(atom_to_list(StageModule) ++ "_controller"),
     register(ControllerName, Pid),
     Pid.
 
@@ -42,25 +47,60 @@ check_stage_balance(State) ->
 
     case is_process_alive(StagePid) of
         true ->
-            {message_queue_len, QLen} = process_info(StagePid, message_queue_len),
+            MaxWorkerQLen = max_worker_queue(StageModule),
             NumWorkers = stage_behaviour:count_workers(StageModule),
 
-            io:format("[+][~p][~p] - Analisando métricas: QLen=~p, Workers=~p, Up=~p, Down=~p~n", 
-                      [calendar:local_time(), self(), QLen, NumWorkers, Up, Down]),
+            io:format("[+][~p][~p] - Métricas: MaxWorkerQLen=~p, Workers=~p, Up=~p, Down=~p~n", 
+                      [calendar:local_time(), self(), MaxWorkerQLen, NumWorkers, Up, Down]),
 
-            if
-                QLen > Up andalso NumWorkers < Max ->
-                    io:format("[+][~p][~p] - Aumento workers para ~p. QLen: ~p, Workers: ~p~n", [calendar:local_time(), self(), StageModule, QLen, NumWorkers]),
-                    stage_behaviour:add_worker(StageModule, self());
-                
-                QLen < Down andalso NumWorkers > Min ->
-                    io:format("[-][~p][~p] - Diminuindo workers para ~p. QLen: ~p, Workers: ~p~n", [calendar:local_time(), self(), StageModule, QLen, NumWorkers]),
-                    stage_behaviour:remove_worker(StageModule, any);
-                true ->
-                    io:format("[+][~p][~p] - Workers estáveis para ~p. QLen: ~p, Workers: ~p~n", [calendar:local_time(), self(), StageModule, QLen, NumWorkers])
-            end;
+            NewState1 =
+                if MaxWorkerQLen > Up -> incr_hot(reset_cold(State));
+                   MaxWorkerQLen < Down -> incr_cold(reset_hot(State));
+                   true -> reset_hot(reset_cold(State))
+                end,
+            NewState2 = maybe_scale(StageModule, NumWorkers, Min, Max, NewState1),
+            NewState2;
         false ->
             io:format("[-][~p][~p] - Stage ~p não está mais vivo~n", [calendar:local_time(), self(), StageModule])
     end,
     State.
+
+max_worker_queue(StageModule) ->
+    StageKey = list_to_atom(atom_to_list(StageModule) ++ "_workers"),
+    case ets:lookup(erlda_workers_table, StageKey) of
+        [{StageKey, Workers}] when is_list(Workers), Workers =/= [] ->
+            lists:max([queue_len(W) || W <- Workers]);
+        _ -> 0
+    end.
+
+queue_len(Pid) when is_pid(Pid) ->
+    case process_info(Pid, message_queue_len) of
+        {message_queue_len, N} -> N;
+        _ -> 0
+    end.
+
+now_ms() -> erlang:monotonic_time(millisecond).
+
+cooldown_over(#controller_state{ last_action_ms = T, action_cooldown_ms = C }) ->
+    now_ms() - T >= C.
+
+incr_hot(S = #controller_state{ consecutive_hot = H }) -> S#controller_state{ consecutive_hot = H + 1 }.
+incr_cold(S = #controller_state{ consecutive_cold = C }) -> S#controller_state{ consecutive_cold = C + 1 }.
+reset_hot(S) -> S#controller_state{ consecutive_hot = 0 }.
+reset_cold(S) -> S#controller_state{ consecutive_cold = 0 }.
+
+maybe_scale(StageModule, NumWorkers, Min, Max, S = #controller_state{ consecutive_hot = H, consecutive_cold = C, min_samples_to_act = K }) ->
+    CanAct = cooldown_over(S),
+    case true of
+        _ when H >= K andalso NumWorkers < Max andalso CanAct ->
+            io:format("[+][~p][~p] - Scale UP ~p (consecutive_hot=~p)~n", [calendar:local_time(), self(), StageModule, H]),
+            stage_behaviour:add_worker(StageModule),
+            S#controller_state{ consecutive_hot = 0, last_action_ms = now_ms() };
+        _ when C >= K andalso NumWorkers > Min andalso CanAct ->
+            io:format("[-][~p][~p] - Scale DOWN ~p (consecutive_cold=~p)~n", [calendar:local_time(), self(), StageModule, C]),
+            stage_behaviour:remove_worker(StageModule, any),
+            S#controller_state{ consecutive_cold = 0, last_action_ms = now_ms() };
+        _ ->
+            S
+    end.
 

src/lib/worker.erl src/lib/stage_worker.erlsrc/lib/worker.erl renamed to src/lib/stage_worker.erl

@@ -1,4 +1,4 @@
--module(worker).
+-module(stage_worker).
 
 -export([spawn_worker/1, loop/1]).
 
@@ -9,12 +9,16 @@ loop(StageModule) ->
     receive
         {work, Command, From} ->
             case StageModule:handle_command(Command) of
-                {ok, _} ->
+                {ok, Result} ->
+                    From ! {stage_result, Result},
                     ok;
                 {error, Reason} ->
-                    io:format("[-][~p][~p] - Erro motivo ~p ~n", [calendar:local_time(), self(), Reason]);
+                    io:format("[-][~p][~p] - Erro motivo ~p ~n", [calendar:local_time(), self(), Reason]),
+                    From ! {stage_error, Reason};
                 {forward, NextStagePid, NewCommand} ->
                     NextStagePid ! {command, NewCommand, From}
             end,
             loop(StageModule)
     end.
+
+

src/poc.app.src

@@ -2,8 +2,18 @@
  [
   {description, "POC Application"},
   {vsn, "0.1.0"},
-  {modules, [poc_app]},
+  {modules, [
+      poc_app,
+      stage_behaviour,
+      stage_controller,
+      stage_worker,
+      http_server,
+      http_server_parser_stage,
+      http_server_cache_stage,
+      http_server_get_stage,
+      http_server_response_writer_stage
+  ]},
   {registered, []},
   {mod, {poc_app, []}},
-  {applications, [kernel, stdlib]}
+  {applications, [kernel, stdlib, ranch]}
  ]}. 

src/poc_app.erl

@@ -5,8 +5,10 @@
 -export([start/2, stop/1]).
 
 start(_Type, _Args) -> 
-    get_stage:start_link(),
-    http_parser_stage:start_link(),
+    http_server_cache_stage:start_link(),
+    http_server_response_writer_stage:start_link(),
+    http_server_get_stage:start_link(),
+    http_server_parser_stage:start_link(),
     http_server:start(),
     {ok, self()}.