Merge pull request 'Improve miner log and retry on error, add zx as dep' (#21) from uw-improve-miner-retry into master

Reviewed-on: #21

ebin/gmhive_client.app

@@ -1,6 +1,6 @@
 {application,gmhive_client,
              [{description,"Gajumaru Hive Client"},
-              {vsn,"0.9.2"},
+              {vsn,"0.10.1"},
               {registered,[]},
               {applications,[kernel,stdlib,sasl,gproc,inets,ssl,enoise,
                              gmconfig,gmhive_protocol,gmhive_worker]},

rebar.config

@@ -3,7 +3,10 @@
 {minimum_otp_vsn, "27.1"}.
 
 {erl_opts, [debug_info]}.
-{plugins, [rebar3_hex]}.
+{plugins, [ rebar3_hex
+          , {zx_rebar_plugin, {git, "https://git.qpq.swiss/QPQ-AG/zx_rebar_plugin",
+                               {branch, "master"}}}
+          ]}.
 
 {post_hooks, [{compile, "make schema"}]}.
 
@@ -16,12 +19,13 @@
         {gmconfig, {git, "https://git.qpq.swiss/QPQ-AG/gmconfig.git",
                     {ref, "38620ff9e2"}}},
         {gproc, "1.0.0"},
-        {setup, {git, "https://github.com/uwiger/setup", {ref, "d9b0b51"}}}
+        {setup, {git, "https://github.com/uwiger/setup", {ref, "0430661"}}},
+        {zx, {zx, "https://gitlab.com/zxq9/zx.git", {ref, "2a0437f4"}, "0.14.0"}}
        ]}.
 
 {relx, [
         {release, { gmhive_client, "0.1.0" },
-         [sasl, enacl, enoise, gmhive_worker,
+         [sasl, enacl, enoise, gmhive_worker, {zx, load},
           {gmhive_protocol, load}, gmserialization, gproc, gmconfig, gmhive_client ]},
         {dev_mode, true},
         {sys_config, "./config/sys.config"},
@@ -36,5 +40,5 @@
 
 {dialyzer, [ {warnings, [unknown]}
            , {plt_apps, all_deps}
-           , {base_plt_apps, [erts, kernel, stdlib, inets]}
+           , {base_plt_apps, [erts, kernel, stdlib, inets, zx]}
            ]}.

rebar.lock

@@ -38,7 +38,12 @@
  {<<"gproc">>,{pkg,<<"gproc">>,<<"1.0.0">>},0},
  {<<"setup">>,
   {git,"https://github.com/uwiger/setup",
-       {ref,"d9b0b51505ff75c459f1ab62bd907ca5e3a32291"}},
+       {ref,"0430661811269d007285f817467cbab7c4c31e96"}},
+  0},
+ {<<"zx">>,
+  {zx,"https://gitlab.com/zxq9/zx.git",
+      {ref,"2a0437f416f94596e4e3b403603dbd028727742d"},
+      "0.14.0"},
   0}]}.
 [
 {pkg_hash,[

src/gmhc_app.erl

@@ -17,11 +17,11 @@
 -spec start([{atom(), any()}]) -> {ok, [atom()]} | {error, any()}.
 start(Opts) ->
     application:load(gmhive_client),
-    {error,_} = application:stop(gmhive_client),
+    _ = application:stop(gmhive_client),
     _ = lists:foreach(fun({K, V}) ->
                               application:set_env(gmhive_client, K, V)
                       end, Opts),
-    application:ensure_all_started(gmhive_client).
+    application:ensure_all_started(gmhive_client, permanent).
 
 start(_StartType, _StartArgs) ->
     set_things_up(),
@@ -46,6 +46,7 @@ stop(_State) ->
 
 set_things_up() ->
     maybe_add_logger_handler(),
+    clean_up_log_directories(),
     gmhc_counters:initialize(),
     gmhc_config:load_config(),
     logger:set_module_level([gmhw_pow_cuckoo], notice),
@@ -61,6 +62,36 @@ set_things_up() ->
     end,
     ok.
 
+clean_up_log_directories() ->
+    case setup_zomp:is_zomp_context() of
+        true ->
+            %% This is a temporary measure to remove old auto-created log sub-directories
+            LogDir = setup:log_dir(),
+            SubDirs = filelib:wildcard(LogDir ++ "/*/"),
+            [delete_if_empty(D) || D <- SubDirs];
+        false ->
+            ok
+    end.
+
+delete_if_empty(Dir) ->
+    case filelib:wildcard(Dir ++ "/*") of
+        [] ->
+            Res = file:del_dir(Dir),
+            Res == ok;
+        Fs ->
+            case [filelib:is_regular(F) || F <- Fs] of
+                [] ->
+                    case [F || F <- Fs, delete_if_empty(F) == false] of
+                        [] ->
+                            delete_if_empty(Dir);
+                        [_|_] ->
+                            false
+                    end;
+                [_|_] ->
+                    false
+            end
+    end.
+
 maybe_add_logger_handler() ->
     case is_headless() orelse application:get_env(gmhive_client, tty_logger, false) of
         true ->

src/gmhc_connector.erl

@@ -57,6 +57,7 @@
             , auto_connect = true :: boolean()
             , econn
             , connected = false :: boolean()
+            , status = disconnected :: disconnected | connecting | connected
             , reconnect = true :: boolean()
             , reconnect_timer :: timer_ref() | 'undefined'
             , recache_timer   :: reference() | 'undefined'
@@ -295,8 +296,9 @@ try_connect_(Opts0, S) ->
             case try_noise_connect(maps:merge(Opts0, PoolOpts)) of
                 {ok, EConn, Opts1} ->
                     S1 = protocol_connect(Opts1, S#st{ econn = EConn
+                                                     , status = connecting
                                                      , reconnect_timer = undefined }), 
-                    {ok, S1};
+                    {ok, S1#st{status = connected}};
                 {error, _} = Error ->
                     Error
             end
@@ -442,14 +444,14 @@ protocol_connect(Opts, #st{econn = EConn} = S) ->
     Type   = to_atom(opt(type, Opts, [<<"type">>])),
     RId = erlang:unique_integer(),
     Vsns = gmhp_msgs:versions(),
-    %% Client = client_name(),
+    Client = client_name(),
     Protocols = gmhp_msgs:protocols(hd(Vsns)),
     ConnectReq = #{ protocols => Protocols
                   , versions  => Vsns
                   , pool_id => PoolId
                   , pubkey => Pubkey
                   , extra_pubkeys => Extra
-                  %% , client => Client
+                  , client => Client
                   , type => Type
                   , nonces => gmhc_server:total_nonces()
                   , signature => ""},
@@ -484,7 +486,10 @@ send_connect(EConn, RId, Msg, #{pubkey        := Pubkey,
                     notify_connected(S1#st{protocol = P, version = V, opts = Opts1});
                 #{error := #{code := _, message := ErrMsg}} = ErrReply ->
                     ?LOG_ERROR("Connect error: ~s", [ErrMsg]),
-                    disconnected(S#st.id, ErrReply, S),
+                    %% TODO: fix the flow so that we send one disconnected event,
+                    %% and set the reconnect in the right place. For now, stuff
+                    %% the `reconnect = false`.
+                    disconnected(S#st.id, ErrReply, S#st{reconnect = false}),
                     gmhc_eureka:invalidate_cache(),
                     error(rejected)
             end
@@ -493,37 +498,20 @@ send_connect(EConn, RId, Msg, #{pubkey        := Pubkey,
             error(protocol_connect_timeout)
     end.
 
-%% client_name() ->
-%%     MyStr = app_string(gmhive_client),
-%%     case app_string(gajumine) of
-%%         <<>>  -> MyStr;
-%%         GMStr -> <<MyStr/binary,",",GMStr/binary>>
-%%     end.
-%%     %% {ok, Vsn} = application:get_key(gmhive_client, vsn),
-%%     %% maybe_add_gajumine(unicode:characters_to_binary(["gmhive_client-", Vsn])).
+client_name() ->
+    MyStr = app_string(gmhive_client),
+    case app_string(gajumine) of
+        <<>>  -> MyStr;
+        GMStr -> <<MyStr/binary,",",GMStr/binary>>
+    end.
 
-%% maybe_add_gajumine(Str) ->
-%%     case setup_zomp:is_zomp_context() of
-%%         true ->
-%%             case zx_daemon:meta() of
-%%                 #{package_id := {"qpq","gajumine",_} = PId} ->
-%%                     {ok, PStr} = zx_lib:package_string(PId),
-%%                     GMStr = unicode:characters_to_binary([",", PStr]),
-%%                     <<Str/binary, GMStr/binary>>;
-%%                 _ ->
-%%                     Str
-%%             end;
-%%         false ->
-%%             Str
-%%     end.
-
-%% app_string(App) ->
-%%     case application:get_key(App, vsn) of
-%%         undefined ->
-%%             <<>>;
-%%         {ok, Vsn} ->
-%%             unicode:characters_to_binary([atom_to_binary(App),"-",Vsn])
-%%     end.
+app_string(App) ->
+    case application:get_key(App, vsn) of
+        undefined ->
+            <<>>;
+        {ok, Vsn} ->
+            unicode:characters_to_binary([atom_to_binary(App),"-",Vsn])
+    end.
 
 to_bin(A) when is_atom(A) ->
     atom_to_binary(A, utf8);
@@ -541,10 +529,11 @@ connected(Id, Type, S) when Type==worker; Type==monitor ->
 disconnected(Id, S) ->
     disconnected(Id, #{}, S).
 
-disconnected(Id, Msg, S) ->
-    gmhc_events:publish(disconnected, Msg#{id => Id}),
+disconnected(_, _, #st{status = disconnected} = S) -> S;
+disconnected(Id, Msg, #st{reconnect = Bool} = S) ->
+    gmhc_events:publish(disconnected, Msg#{id => Id, reconnecting => Bool}),
     gmhc_server:disconnected(Id),
-    S#st{connected = false}.
+    S#st{connected = false, status = disconnected}.
 
 opt_autoconnect(#{auto_connect := Bool}) when is_boolean(Bool) ->
     Bool;

src/gmhc_connectors_sup.erl

@@ -49,8 +49,8 @@ add_restart_info(Id, Opts) ->
 init([]) ->
     Mod = gmhc_connector,
     SupFlags = #{ strategy => simple_one_for_one
-                , intensity => 3
-                , period => 10 },
+                , intensity => 5
+                , period => 60 },
     ChildSpecs = [ #{ id => Mod
                     , start => {Mod, start_link, []}
                     , type => worker

src/gmhc_handler.erl

@@ -28,7 +28,7 @@
 
 -record(st, {pools = [], opts = #{}}).
 
--define(CALL_TIMEOUT, 5000).
+-define(CALL_TIMEOUT, 10_000).
 
 -include_lib("kernel/include/logger.hrl").
 

src/gmhc_server.erl

@@ -41,6 +41,7 @@
             , candidate         :: map() | 'undefined'
             , nonces = 1        :: pos_integer()
             , workers = []      :: [worker()]
+            , check_timer       :: 'undefined' | reference()
             }).
 
 -define(CONNECTED(S), map_size(S#st.connected) > 0).
@@ -80,6 +81,7 @@ init([]) ->
                                       N + Acc
                               end, 0, IdleWorkers),
     process_flag(trap_exit, true),
+    gproc_ps:subscribe(l, gmhw_miner_return),
     {ok, #st{workers = IdleWorkers, nonces = TotalNonces}}.
 
 handle_call(total_nonces, _From, #st{nonces = Nonces} = S) ->
@@ -158,18 +160,26 @@ handle_info({'EXIT', Pid, Reason}, #st{ workers = Workers
             gmhc_events:publish(error, ?ERR_EVT(#{error => worker_error,
                                                    data  => Reason})),
             Ws1 = incr_worker_error(W, Workers),
-            erlang:start_timer(100, self(), check_workers),
-            {noreply, S#st{workers = Ws1}};
+            S1 = start_check_timer(S#st{workers = Ws1}),
+            {noreply, S1};
         false ->
             %% ?LOG_DEBUG("EXIT apparently not from worker?? (~p)", [Pid]),
             {noreply, S}
     end;
 handle_info({timeout, _, check_workers}, #st{workers = Workers} = S) ->
-    S1 = maybe_request_nonces(S),
+    S1 = maybe_request_nonces(S#st{check_timer = undefined}),
     S2 = lists:foldl(fun(W, Sx) ->
                              maybe_restart_worker(W, Sx)
                      end, S1, Workers),
     {noreply, S2};
+handle_info({gproc_ps_event, gmhw_miner_return, Data}, S) ->
+    case Data of
+        #{info := #{result := {invalid_output, _}} = Info} ->
+            ?LOG_NOTICE("Bad miner return: ~p", [Info]);
+        _ ->
+            ok
+    end,
+    {noreply, S};
 handle_info(Msg, St) ->
     ?LOG_DEBUG("Unknown msg: ~p", [Msg]),
     {noreply, St}.
@@ -180,6 +190,12 @@ terminate(_Reason, _St) ->
 code_change(_FromVsn, S, _Extra) ->
     {ok, S}.
 
+start_check_timer(#st{check_timer = undefined} = S) ->
+    TRef = erlang:start_timer(100, self(), check_workers),
+    S#st{check_timer = TRef};
+start_check_timer(S) ->
+    S.
+
 report_solutions(Solutions, W, #st{} = S) when ?CONNECTED(S) ->
     #{via := Via, seq := Seq} = W#worker.cand,
     Nonces = all_nonces(W),
@@ -268,7 +284,7 @@ handle_worker_result({worker_result, Result}, W, S) ->
             gmhc_events:publish(error, ?ERR_EVT(#{error => worker_error,
                                                    data  => Result})),
             Ws = incr_worker_error(W, S#st.workers),
-            S#st{workers = Ws}
+            start_check_timer(S#st{workers = Ws})
     end;
 handle_worker_result(Error, W, S) ->
     ?LOG_DEBUG("Got worker error from ~p: ~p", [W#worker.index, Error]),

src/gmhc_sup.erl

@@ -20,8 +20,8 @@ init([]) ->
                  , worker(gmhc_handler)
                  , supervisor(gmhc_connectors_sup) ],
     SupFlags = #{ strategy      => rest_for_one
-                , intensity     => 1
-                , period        => 5
+                , intensity     => 5             %% We really want the hive client to sort itself out
+                , period        => 5*60          %% Timemout issues can happen infrequently
                 , auto_shutdown => never },
     {ok, {SupFlags, ChildSpecs}}.
 

zomp.meta

@@ -4,14 +4,14 @@
 {prefix,"gmhc"}.
 {desc,"Gajumaru Hive Client"}.
 {author,"Ulf Wiger, QPQ AG"}.
-{package_id,{"uwiger","gmhive_client",{0,9,2}}}.
-{deps,[{"uwiger","gmhive_protocol",{0,2,0}},
+{package_id,{"uwiger","gmhive_client",{0,10,1}}}.
+{deps,[{"uwiger","setup",{3,0,0}},
+       {"uwiger","gmhive_protocol",{0,3,1}},
        {"uwiger","gmhive_worker",{0,5,1}},
        {"uwiger","gmcuckoo",{1,2,4}},
        {"otpr","eblake2",{1,0,1}},
        {"otpr","base58",{0,1,1}},
        {"otpr","gmserialization",{0,1,3}},
-       {"uwiger","setup",{2,2,4}},
        {"uwiger","gproc",{1,0,1}},
        {"uwiger","gmconfig",{0,1,2}},
        {"uwiger","enoise",{1,3,0}}]}.