TD-927: progressor integration

apps/hellgate/src/hellgate.app.src

@@ -10,6 +10,9 @@
         stdlib,
         genlib,
         fault_detector_proto,
+        herd,
+        progressor,
+        hg_progressor,
         hg_proto,
         routing,
         cowboy,

apps/hellgate/src/hellgate.erl

@@ -51,6 +51,8 @@ init([]) ->
         {
             #{strategy => one_for_all, intensity => 6, period => 30},
             [
+                %% for debugging only
+                %% hg_profiler:get_child_spec(),
                 party_client:child_spec(party_client, PartyClient),
                 hg_machine:get_child_spec(MachineHandlers),
                 get_api_child_spec(MachineHandlers, Opts)

apps/hellgate/src/hg_invoice.erl

@@ -256,13 +256,16 @@ process_with_tag(Tag, F) ->
 
 -spec fail(hg_machine:id()) -> ok.
 fail(Id) ->
-    case hg_machine:call(?NS, Id, fail) of
+    try hg_machine:call(?NS, Id, fail) of
         {error, failed} ->
             ok;
         {error, Error} ->
             erlang:error({unexpected_error, Error});
         {ok, Result} ->
             erlang:error({unexpected_result, Result})
+    catch
+        _Class:_Term:_Trace ->
+            ok
     end.
 
 %%

apps/hellgate/src/hg_machine.erl

@@ -36,6 +36,11 @@
     auxst => auxst()
 }.
 
+-type backend() ::
+    machinegun
+    | progressor
+    | hybrid.
+
 -callback namespace() -> ns().
 
 -callback init(args(), machine()) -> result().
@@ -85,6 +90,8 @@
 -export([get_history/5]).
 -export([get_machine/5]).
 
+-export([call_automaton/3]).
+
 %% Dispatch
 
 -export([get_child_spec/1]).
@@ -220,6 +227,10 @@ do_call(Ns, Id, Args, After, Limit, Direction) ->
     end.
 
 call_automaton(Function, Args) ->
+    call_automaton(Function, Args, application:get_env(hellgate, backend, machinegun)).
+
+-spec call_automaton(woody:func(), woody:args(), backend()) -> term().
+call_automaton(Function, Args, machinegun) ->
     case hg_woody_wrapper:call(automaton, Function, Args) of
         {ok, _} = Result ->
             Result;
@@ -233,7 +244,11 @@ call_automaton(Function, Args) ->
             {error, working};
         {exception, #mg_stateproc_RepairFailed{reason = Reason}} ->
             {error, {repair, {failed, Reason}}}
-    end.
+    end;
+call_automaton(Function, Args, progressor) ->
+    hg_progressor:call_automaton(Function, Args);
+call_automaton(Function, Args, hybrid) ->
+    hg_hybrid:call_automaton(Function, Args).
 
 %%
 
@@ -400,7 +415,7 @@ start_link(MachineHandlers) ->
 
 -spec init([module()]) -> {ok, {supervisor:sup_flags(), [supervisor:child_spec()]}}.
 init(MachineHandlers) ->
-    _ = ets:new(?TABLE, [protected, named_table, {read_concurrency, true}]),
+    _ = ets:new(?TABLE, [named_table, {read_concurrency, true}]),
     true = ets:insert_new(?TABLE, [{MH:namespace(), MH} || MH <- MachineHandlers]),
     {ok, {#{}, []}}.
 

apps/hellgate/src/hg_profiler.erl

@@ -0,0 +1,144 @@
+-module(hg_profiler).
+
+-behaviour(gen_server).
+
+-export([start_link/0]).
+-export([
+    init/1,
+    handle_call/3,
+    handle_cast/2,
+    handle_info/2,
+    terminate/2,
+    code_change/3
+]).
+
+-export([get_child_spec/0]).
+-export([report/0]).
+-export([scan_proc/0]).
+-export([scan_proc/1]).
+
+-spec get_child_spec() -> _.
+get_child_spec() ->
+    #{
+        id => ?MODULE,
+        start => {?MODULE, start_link, []}
+    }.
+
+-spec report() -> _.
+report() ->
+    TargetItem = memory,
+    MFAs = [
+        {prg_worker, init, 1},
+        {prg_worker_sidecar, init, 1},
+        {epg_pool_wrk, init, 1},
+        {epgsql_sock, init, 1}
+    ],
+    lists:foreach(fun(MFA) -> do_report(MFA, TargetItem) end, MFAs).
+
+-spec scan_proc() -> _.
+scan_proc() ->
+    scan_proc(memory).
+
+-spec scan_proc(_) -> _.
+scan_proc(Item) ->
+    ScanResult = lists:foldl(
+        fun(P, Acc) ->
+            try maps:from_list(process_info(P, [dictionary, Item])) of
+                #{dictionary := Dict} = Info ->
+                    case maps:from_list(Dict) of
+                        #{'$initial_call' := InitialMFA} ->
+                            Value = maps:get(Item, Info),
+                            {Cnt, Sm} = maps:get(InitialMFA, Acc, {0, 0}),
+                            Acc#{InitialMFA => {Cnt + 1, Sm + Value}};
+                        _ ->
+                            io:format(user, "UNKNOWN '$initial_call': ~p~n", [P]),
+                            Acc
+                    end;
+                _ ->
+                    Acc
+            catch
+                _:_ ->
+                    Acc
+            end
+        end,
+        #{},
+        processes()
+    ),
+    Sorted = lists:sort(fun({_, {_, SumA}}, {_, {_, SumB}}) -> SumA >= SumB end, maps:to_list(ScanResult)),
+    lists:foreach(
+        fun({MFA, {Count, Sum}}) ->
+            io:format(user, "MFA: ~p Count: ~p Memory: ~p~n", [MFA, Count, Sum])
+        end,
+        Sorted
+    ).
+
+%%%===================================================================
+%%% Spawning and gen_server implementation
+%%%===================================================================
+-spec start_link() -> _.
+start_link() ->
+    gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
+
+-spec init(_) -> _.
+init(_) ->
+    erlang:start_timer(60000, self(), report),
+    {ok, #{}}.
+
+-spec handle_call(_, _, _) -> _.
+handle_call(_Request, _From, State) ->
+    {reply, ok, State}.
+
+-spec handle_cast(_, _) -> _.
+handle_cast(_Request, State) ->
+    {noreply, State}.
+
+-spec handle_info(_, _) -> _.
+handle_info({timeout, _TRef, report}, State) ->
+    TargetItem = memory,
+    MFAs = [
+        {prg_worker, init, 1},
+        {prg_worker_sidecar, init, 1},
+        {epg_pool_wrk, init, 1},
+        {epgsql_sock, init, 1}
+    ],
+    lists:foreach(fun(MFA) -> do_report(MFA, TargetItem) end, MFAs),
+    erlang:start_timer(60000, self(), report),
+    {noreply, State};
+handle_info(_Info, State) ->
+    {noreply, State}.
+
+-spec terminate(_, _) -> _.
+terminate(_Reason, _State) ->
+    ok.
+
+-spec code_change(_, _, _) -> _.
+code_change(_OldVsn, State, _Extra) ->
+    {ok, State}.
+
+%%%
+
+do_report(InitialMFA, Item) ->
+    {ProcCount, Summary} = lists:foldl(
+        fun(P, {Cnt, Sm} = Acc) ->
+            try maps:from_list(process_info(P, [dictionary, Item])) of
+                #{dictionary := Dict} = Info ->
+                    case maps:from_list(Dict) of
+                        #{'$initial_call' := InitialMFA} ->
+                            Value = maps:get(Item, Info),
+                            {Cnt + 1, Sm + Value};
+                        _ ->
+                            Acc
+                    end;
+                _ ->
+                    Acc
+            catch
+                _:_ ->
+                    Acc
+            end
+        end,
+        {0, 0},
+        processes()
+    ),
+    %% io:format(user, "MFA: ~p Item: ~p Count: ~p Summary: ~p~n", [InitialMFA, Item, ProcCount, Summary]),
+    logger:info("MFA: ~p Item: ~p Count: ~p Summary: ~p", [InitialMFA, Item, ProcCount, Summary]),
+    ok.

apps/hellgate/test/hg_ct_helper.erl

@@ -217,7 +217,8 @@ start_app(hellgate = AppName) ->
                     operation_time_limit => 1200000,
                     pre_aggregation_size => 2
                 }
-            }}
+            }},
+            {backend, progressor}
         ]),
         #{
             hellgate_root_url => get_hellgate_url()
@@ -271,6 +272,106 @@ start_app(snowflake = AppName) ->
         ]),
         #{}
     };
+start_app(epg_connector = AppName) ->
+    {
+        start_app(AppName, [
+            {databases, #{
+                default_db => #{
+                    host => "postgres",
+                    port => 5432,
+                    database => "progressor_db",
+                    username => "progressor",
+                    password => "progressor"
+                }
+            }},
+            {pools, #{
+                default_pool => #{
+                    database => default_db,
+                    size => 200
+                },
+                default_front_pool => #{
+                    database => default_db,
+                    size => 50
+                },
+                default_scan_pool => #{
+                    database => default_db,
+                    size => 8
+                }
+            }}
+        ]),
+        #{}
+    };
+start_app(progressor = AppName) ->
+    {
+        start_app(AppName, [
+            {call_wait_timeout, 20},
+            {defaults, #{
+                storage => #{
+                    client => prg_pg_backend,
+                    options => #{
+                        pool => default_pool,
+                        front_pool => default_front_pool,
+                        scan_pool => default_scan_pool
+                    }
+                },
+                retry_policy => #{
+                    initial_timeout => 5,
+                    backoff_coefficient => 1.0,
+                    %% seconds
+                    max_timeout => 180,
+                    max_attempts => 3,
+                    non_retryable_errors => []
+                },
+                task_scan_timeout => 1,
+                worker_pool_size => 30,
+                process_step_timeout => 30
+            }},
+            {namespaces, #{
+                invoice => #{
+                    processor => #{
+                        client => hg_progressor,
+                        options => #{
+                            party_client => #{},
+                            ns => <<"invoice">>,
+                            handler => hg_machine
+                        }
+                    },
+                    worker_pool_size => 150
+                },
+                invoice_template => #{
+                    processor => #{
+                        client => hg_progressor,
+                        options => #{
+                            party_client => #{},
+                            ns => <<"invoice_template">>,
+                            handler => hg_machine
+                        }
+                    }
+                },
+                customer => #{
+                    processor => #{
+                        client => hg_progressor,
+                        options => #{
+                            party_client => #{},
+                            ns => <<"customer">>,
+                            handler => hg_machine
+                        }
+                    }
+                },
+                recurrent_paytools => #{
+                    processor => #{
+                        client => hg_progressor,
+                        options => #{
+                            party_client => #{},
+                            ns => <<"recurrent_paytools">>,
+                            handler => hg_machine
+                        }
+                    }
+                }
+            }}
+        ]),
+        #{}
+    };
 start_app(AppName) ->
     {start_application(AppName), #{}}.
 

apps/hellgate/test/hg_customer_tests_SUITE.erl

@@ -68,6 +68,8 @@ init_per_suite(C) ->
         bender_client,
         party_client,
         hg_proto,
+        epg_connector,
+        progressor,
         hellgate,
         snowflake,
         {cowboy, CowboySpec}
@@ -94,6 +96,8 @@ init_per_suite(C) ->
 -spec end_per_suite(config()) -> _.
 end_per_suite(C) ->
     _ = hg_domain:cleanup(),
+    _ = application:stop(progressor),
+    _ = hg_progressor:cleanup(),
     [application:stop(App) || App <- cfg(apps, C)].
 
 -spec all() -> [{group, test_case_name()}].

apps/hellgate/test/hg_direct_recurrent_tests_SUITE.erl

@@ -100,6 +100,8 @@ init_per_suite(C) ->
         bender_client,
         party_client,
         hg_proto,
+        epg_connector,
+        progressor,
         hellgate,
         {cowboy, CowboySpec}
     ]),
@@ -129,6 +131,8 @@ init_per_suite(C) ->
 -spec end_per_suite(config()) -> config().
 end_per_suite(C) ->
     _ = hg_domain:cleanup(),
+    _ = application:stop(progressor),
+    _ = hg_progressor:cleanup(),
     [application:stop(App) || App <- cfg(apps, C)].
 
 -spec init_per_group(group_name(), config()) -> config().