QPQ-AG/gmhive_client
11
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Keep a second session on new_server and skip Eureka-retry

Browse Source 
apply_new_server opens a keep connector to the advertised
host/port. A dead primary does not Eureka-retry while another
keep session is up. Reconnect preserves host/port.

Dialyzer: host type, watchdog tref, worker config, zx ppath.
This commit is contained in:
Ulf Wiger
2026-08-19 08:38:13 +02:00
parent c0442c117e
commit 4b4d0f8f02
8 changed files with 119 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rebar.config
+3 -1
View File
@@ -12,9 +12,11 @@
{deps, [
{enoise, {git, "https://git.qpq.swiss/QPQ-AG/enoise.git", {ref, "029292817e"}}},
{gm_enoise_session, {git, "https://git.qpq.swiss/QPQ-AG/gm_enoise_session.git",
{ref, "38b84133c7"}}},
{gmhive_protocol,
{git, "https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git",
{ref, "9eec871"}}},
{ref, "1852b20"}}},
{gmhive_worker, {git, "https://git.qpq.swiss/QPQ-AG/gmhive_worker", {ref, "cabd104114"}}},
{gmconfig, {git, "https://git.qpq.swiss/QPQ-AG/gmconfig.git",
{ref, "38620ff9e2"}}},
rebar.lock
+11 -4
View File
@@ -15,6 +15,10 @@
{git,"https://git.qpq.swiss/QPQ-AG/enoise.git",
{ref,"029292817ea1c685b6377bfae667976c0f4a36bc"}},
0},
{<<"gm_enoise_session">>,
{git,"https://git.qpq.swiss/QPQ-AG/gm_enoise_session.git",
{ref,"38b84133c7f09ee33daca3514d7770b9ac640445"}},
0},
{<<"gmconfig">>,
{git,"https://git.qpq.swiss/QPQ-AG/gmconfig.git",
{ref,"38620ff9e2289dff121377c382e4d5b08c719d87"}},
@@ -25,7 +29,7 @@
1},
{<<"gmhive_protocol">>,
{git,"https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git",
{ref,"9eec871e954b729cc5c9249da81abb91de03ad39"}},
{ref,"1852b20fe6666cd34ac08ec2bdf62540d817e8d0"}},
0},
{<<"gmhive_worker">>,
{git,"https://git.qpq.swiss/QPQ-AG/gmhive_worker",
@@ -33,9 +37,10 @@
0},
{<<"gmserialization">>,
{git,"https://git.qpq.swiss/QPQ-AG/gmserialization.git",
{ref,"0288719ae15814f3a53114c657502a24376bebfa"}},
{ref,"dda5cac7a91ba0ad105815ec3cb1de551a04d719"}},
1},
{<<"gproc">>,{pkg,<<"gproc">>,<<"1.0.0">>},0},
{<<"ranch">>,{pkg,<<"ranch">>,<<"1.8.0">>},1},
{<<"setup">>,
{git,"https://github.com/uwiger/setup",
{ref,"0430661811269d007285f817467cbab7c4c31e96"}},
@@ -47,7 +52,9 @@
0}]}.
[
{pkg_hash,[
{<<"gproc">>, <<"AA9EC57F6C9FF065B16D96924168D7C7157CD1FD457680EFE4B1274F456FA500">>}]},
{<<"gproc">>, <<"AA9EC57F6C9FF065B16D96924168D7C7157CD1FD457680EFE4B1274F456FA500">>},
{<<"ranch">>, <<"8C7A100A139FD57F17327B6413E4167AC559FBC04CA7448E9BE9057311597A1D">>}]},
{pkg_hash_ext,[
{<<"gproc">>, <<"109F253C2787DE8A371A51179D4973230CBEC6239EE673FA12216A5CE7E4F902">>}]}
{<<"gproc">>, <<"109F253C2787DE8A371A51179D4973230CBEC6239EE673FA12216A5CE7E4F902">>},
{<<"ranch">>, <<"49FBCFD3682FAB1F5D109351B61257676DA1A2FDBE295904176D5E521A2DDFE5">>}]}
].
src/gmhc_connector.erl
+18 -6
View File
@@ -45,8 +45,13 @@
, nowait => boolean() %% default: false
, tcp_opts => list()
, enoise_opts => list() %% if present, MUST contain {'noise', _}
, tcp_opts => list()
, connect_timeout => pos_integer()
, type => worker | monitor
, pubkey => binary() | string()
, extra => list()
, extra_pubkeys => list()
, pool_id => binary() | string()
, id => id()
}.
-export_type([ id/0
@@ -150,7 +155,7 @@ init(#{id := Id} = Opts) when is_map(Opts) ->
S0#st{econn = undefined, reconnect = false};
{error, _} = Error ->
?LOG_WARNING("Could not connect to core server: ~p", [Error]),
start_reconnect_timer(S0#st{econn = undefined})
start_reconnect_timer(S0#st{econn = undefined, opts = Opts}, Opts)
end;
false ->
S0
@@ -289,21 +294,27 @@ try_connect(Opts, S) ->
end.
try_connect_(Opts0, S) ->
case eureka_get_host_port() of
case connect_pool_opts(Opts0) of
{error, _} = Error ->
Error;
PoolOpts when is_map(PoolOpts) ->
case try_noise_connect(maps:merge(Opts0, PoolOpts)) of
%% Explicit host/port (new_server) win over Eureka / config.
case try_noise_connect(maps:merge(PoolOpts, Opts0)) of
{ok, EConn, Opts1} ->
S1 = protocol_connect(Opts1, S#st{ econn = EConn
, status = connecting
, reconnect_timer = undefined }),
, reconnect_timer = undefined }),
{ok, S1#st{status = connected}};
{error, _} = Error ->
Error
end
end.
connect_pool_opts(#{host := _, port := _} = Opts) ->
maps:with([host, port, pool_id], Opts);
connect_pool_opts(_) ->
eureka_get_host_port().
eureka_get_host_port() ->
case gmhc_eureka:get_pool_address() of
{ok, #{host := Host,
@@ -437,7 +448,8 @@ start_timer(T, Opts0) ->
erlang:start_timer(T, self(), {reconnect, Opts}).
retry_opts(Opts) ->
maps:with([deadline], Opts).
maps:with([deadline, host, port, pool_id, type, pubkey,
extra, extra_pubkeys], Opts).
protocol_connect(Opts, #st{econn = EConn} = S) ->
Pubkey = to_bin(opt(pubkey, Opts, [<<"pubkey">>])),
src/gmhc_connectors_sup.erl
+2 -2
View File
@@ -17,11 +17,11 @@ start_link() ->
case supervisor:start_link({local, ?MODULE}, ?MODULE, []) of
{ok, _} = Ok ->
RI0 = gmhc_watchdog:get_restart_info(),
?LOG_ERROR("Restart info: ~p", [RI0]),
?LOG_DEBUG("Restart info: ~p", [RI0]),
RI = sort_restart_info(RI0),
RemoveIds = maps:keys(maps:without(maps:keys(RI), RI0)),
gmhc_watchdog:remove_restart_info(RemoveIds),
?LOG_ERROR("Sorted restart info: ~p", [RI]),
?LOG_INFO("Sorted restart info: ~p", [RI]),
maps:foreach(fun restart_connector/2, RI),
Ok;
Other ->
src/gmhc_eureka.erl
+1 -1
View File
@@ -120,7 +120,7 @@ cache_dir() ->
cache_dir_zomp() ->
#{package_id := {Realm, App, _}} = zx_daemon:meta(),
filename:join(zx_lib:ppath(var, {Realm, App}), "gmhive.cache").
filename:join(zx_lib:ppath(var, {Realm, App, {z, z, z}}), "gmhive.cache").
get_pool_address_() ->
case gmconfig:find_config([<<"pool_admin">>, <<"url">>], [user_config]) of
src/gmhc_handler.erl
+78 -16
View File
@@ -19,10 +19,10 @@
-record(pool, { id :: gmhc_connector:id()
, pid :: pid() | 'undefined'
, mref :: reference()
, mref :: reference() | 'undefined'
, connected = false :: boolean()
, keep = true :: boolean()
, host :: string()
, host :: binary() | string()
, port :: pos_integer()
, opts = #{} :: map() }).
@@ -60,6 +60,7 @@ start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
init([]) ->
gmhc_events:ensure_subscribed(disconnected),
{ok, #st{}}.
handle_call({call, Req}, _From, #st{} = S) ->
@@ -71,13 +72,14 @@ handle_call(_Req, _From, S) ->
handle_cast({pool_connected, Id, Pid, Opts}, #st{pools = Pools} = S) ->
MRef = erlang:monitor(process, Pid),
S1 = remember_connect_opts(Opts, S),
case lists:keyfind(Id, #pool.id, Pools) of
#pool{} = P ->
P1 = P#pool{ connected = true
, pid = Pid
, mref = MRef },
Pools1 = lists:keyreplace(Id, #pool.id, Pools, P1),
{noreply, S#st{pools = Pools1}};
{noreply, S1#st{pools = Pools1}};
false ->
P = #pool{ id = Id
, pid = Pid
@@ -85,7 +87,7 @@ handle_cast({pool_connected, Id, Pid, Opts}, #st{pools = Pools} = S) ->
, host = maps:get(host, Opts)
, port = maps:get(port, Opts)
, opts = Opts },
{noreply, S#st{pools = [P | Pools]}}
{noreply, S1#st{pools = [P | Pools]}}
end;
handle_cast({notify, Msg}, #st{} = S) ->
notify_connector(Msg),
@@ -99,11 +101,20 @@ handle_info({from_pool, Msg}, S) ->
#{notification := #{new_server := #{ host := _
, port := _
, keep := _ } = Server}} ->
{noreply, start_pool_connector(Server, S)};
{noreply, apply_new_server(Server, S)};
_ ->
gmhc_server:from_pool(Msg),
{noreply, S}
end;
handle_info({gproc_ps_event, disconnected, #{info := #{id := Id}}}, S) ->
{noreply, on_pool_down(Id, S)};
handle_info({'DOWN', _MRef, process, Pid, _}, #st{pools = Pools} = S) ->
case lists:keyfind(Pid, #pool.pid, Pools) of
#pool{id = Id} ->
{noreply, on_pool_down(Id, S)};
false ->
{noreply, S}
end;
handle_info(Msg, S) ->
?LOG_DEBUG("Unknown msg: ~p", [Msg]),
{noreply, S}.
@@ -169,19 +180,35 @@ notify_connector(Msg0) ->
{Via, Msg} = maps:take(via, Msg0),
gmhc_connector:send(Via, #{notification => #{msg => Msg}}).
start_pool_connector(#{ host := Host
, port := Port
, keep := Keep }, #st{pools = Pools, opts = Opts} = S) ->
apply_new_server(#{host := Host0, port := Port, keep := false} = _Server,
#st{pools = Pools} = S) ->
Host = norm_host(Host0),
case [P || #pool{host = H1, port = P1} = P <- Pools,
H1 == Host,
P1 == Port] of
norm_host(H1) == Host, P1 == Port] of
[#pool{id = Id} | _] ->
catch gmhive_client:disconnect(Id),
S#st{pools = [P || #pool{id = I} = P <- Pools, I =/= Id]};
[] ->
case gmhive_client:connect(Opts#{ host => Host
, port => Port
, nowait => true }) of
S
end;
apply_new_server(#{host := Host0, port := Port, keep := Keep},
#st{pools = Pools, opts = Opts} = S)
when is_integer(Port), Port > 0 ->
Host = norm_host(Host0),
case [P || #pool{host = H1, port = P1} = P <- Pools,
norm_host(H1) == Host, P1 == Port] of
[] ->
ConnOpts = Opts#{ host => host_to_list(Host)
, port => Port
, type => worker
, nowait => true },
case gmhive_client:connect(ConnOpts) of
{ok, CId} ->
Pid = gmhc_connector:whereis_id(CId),
MRef = erlang:monitor(process, Pid),
MRef = case is_pid(Pid) of
true -> erlang:monitor(process, Pid);
false -> undefined
end,
P = #pool{ id = CId, host = Host, port = Port,
keep = Keep, pid = Pid, mref = MRef },
S#st{pools = [P | Pools]};
@@ -190,7 +217,42 @@ start_pool_connector(#{ host := Host
[Host, Port, Error]),
S
end;
[#pool{} = _P] ->
[#pool{} = P] ->
Pools1 = lists:keyreplace(P#pool.id, #pool.id, Pools,
P#pool{keep = Keep}),
?LOG_DEBUG("Already have a pool entry for ~p:~p", [Host, Port]),
S
S#st{pools = Pools1}
end.
remember_connect_opts(Opts, #st{opts = Old} = S) when is_map(Opts) ->
Keep = maps:with([pubkey, extra, extra_pubkeys, pool_id, type], Opts),
S#st{opts = maps:merge(Old, Keep)};
remember_connect_opts(_, S) ->
S.
on_pool_down(Id, #st{pools = Pools} = S) ->
Pools1 = case lists:keyfind(Id, #pool.id, Pools) of
#pool{} = P ->
lists:keyreplace(Id, #pool.id, Pools,
P#pool{connected = false, pid = undefined});
false ->
Pools
end,
case has_kept_connection(Id, Pools1) of
true ->
%% Another standby/keep session is up — do not Eureka-retry this one.
catch gmhc_connector:disconnect(Id);
false ->
ok
end,
S#st{pools = Pools1}.
has_kept_connection(ExceptId, Pools) ->
lists:any(fun(#pool{id = I, connected = C, keep = K}) ->
I =/= ExceptId andalso C andalso K
end, Pools).
norm_host(H) when is_list(H) -> list_to_binary(H);
norm_host(H) when is_binary(H) -> H.
host_to_list(H) when is_binary(H) -> binary_to_list(H).
src/gmhc_server.erl
+5 -1
View File
@@ -410,6 +410,9 @@ pick_nonce(N, [A, B]) when A < B ->
%% Dialyzer doesn't like that the fun passed to spawn_link/1
%% doesn't have a local return (it communicates its result via the exit reason).
-dialyzer({no_return, spawn_worker/2}).
%% Miner port success typing is only {error,{runtime,_}}; the ok / no_solution
%% clauses and worker_result/2 are the live path.
-dialyzer({nowarn_function, [init_worker/5, worker_result/2]}).
spawn_worker(#worker{nonce = undefined} = W, _) ->
W;
@@ -430,7 +433,8 @@ spawn_worker(W, _) ->
% Worker already has work. Don't disturb it.
W.
-spec init_worker(binary(), integer(), integer(), tuple(), pid()) -> no_return().
-spec init_worker(binary(), integer(), integer(),
gmhw_pow_cuckoo:config(), pid()) -> no_return().
init_worker(Data, Nonce, Target, Config, Parent) ->
gmhc_events:publish(puzzle, {Data, Target, Nonce, Config}),
Res = gmhc_workers:generate_from_hash(Data, Target, Nonce, Config, undefined),
src/gmhc_watchdog.erl
+1 -1
View File
@@ -21,7 +21,7 @@
-record(svc, { n = 5 :: pos_integer()
, n0 = 5 :: pos_integer()
, interval = 5000 :: pos_integer()
, tref :: reference()
, tref :: reference() | undefined
, mref :: reference()
}).