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

Compare commits

base: QPQ-AG:master
Branches Tags
QPQ-AG:master
QPQ-AG:uw-restart-logic
QPQ-AG:uw-restart-params
QPQ-AG:uw-limit-connectors
QPQ-AG:uw-watchdog
QPQ-AG:uw-cache-conn
QPQ-AG:uw-export-schema
QPQ-AG:uw-repeats-reporting
QPQ-AG:uw-progress-reporting
QPQ-AG:uw-miner-returns
QPQ-AG:uw-demonitor
QPQ-AG:uw-handle-empty-nonces
QPQ-AG:uw-reset-worker
..
compare: QPQ-AG:uw-reset-worker
Branches Tags
QPQ-AG:master
QPQ-AG:uw-restart-logic
QPQ-AG:uw-restart-params
QPQ-AG:uw-limit-connectors
QPQ-AG:uw-watchdog
QPQ-AG:uw-cache-conn
QPQ-AG:uw-export-schema
QPQ-AG:uw-repeats-reporting
QPQ-AG:uw-progress-reporting
QPQ-AG:uw-miner-returns
QPQ-AG:uw-demonitor
QPQ-AG:uw-handle-empty-nonces
QPQ-AG:uw-reset-worker

No commits in common. "master" and "uw-reset-worker" have entirely different histories.
master ... uw-reset-w

29 changed files with 110 additions and 993 deletions
Show Stats Expand all files Collapse all files

2
.gitignore vendored
View File

@ -19,7 +19,7 @@ VERSION
*.aes~ *.aes~
*.config~ *.config~
*.args~ *.args~
data/* data/mnesia
_checkouts*/ _checkouts*/
tmp/ tmp/
rebar3.crashdump rebar3.crashdump

10
Makefile
View File

@ -1,10 +0,0 @@
BUILD=_build/default/lib/gmhive_client
SCHEMA_OUT=$(BUILD)/priv/gmhc_schema.json
schema:
ERL_LIBS=_build/default/lib \
erl -run gmhc_config_schema export $(SCHEMA_OUT) -run init stop
scripts/update_readme_json.sh "README.md" "$(SCHEMA_OUT)"

30
README.md
View File

@ -82,20 +82,6 @@ only does this inform the client of the network for which the client should
mine, but it also changes what URL the client will use to find the hive mine, but it also changes what URL the client will use to find the hive
server. (For `"testnet"` this will be `"test.gajumining.com"`.) server. (For `"testnet"` this will be `"test.gajumining.com"`.)
### Caching of connection data
The client retrieves the Hive Server connection info from gajumining.com.
This information is account-specific, and may change over time.
To speed up the reconnecting process, should the Hive Server restart,
this connection information is cached locally for up to 24 hours.
The normal place for the cached data would be
`$ZOMP_DIR//var/uwiger/gmhive_client/Vsn/setup.data/mainnet/gmhc_eureka.XXXXXXXX.cache`,
where `Vsn` is the current version of `gmhive_client`, and `XXXXXXXX` is a fragment
of the current account ID. The data location can be changed with the command-line
option `-setup data_dir Dir`.
## JSON Schema ## JSON Schema
```json ```json
@ -103,10 +89,6 @@ option `-setup data_dir Dir`.
"$schema": "http://json-schema.org/draft-04/schema#", "$schema": "http://json-schema.org/draft-04/schema#",
"additionalProperties": false, "additionalProperties": false,
"properties": { "properties": {
"cache_dir": {
"description": "Location of cache, default is 'setup:data_dir()'",
"type": "string"
},
"extra_pubkeys": { "extra_pubkeys": {
"default": [], "default": [],
"description": "Additional worker pubkeys, sharing rewards", "description": "Additional worker pubkeys, sharing rewards",
@ -173,16 +155,6 @@ option `-setup data_dir Dir`.
"pattern": "^ak_[1-9A-HJ-NP-Za-km-z]*$", "pattern": "^ak_[1-9A-HJ-NP-Za-km-z]*$",
"type": "string" "type": "string"
}, },
"report": {
"default": "silent",
"description": "Progress reporting",
"enum": [
"debug",
"progress",
"silent"
],
"type": "string"
},
"type": { "type": {
"default": "worker", "default": "worker",
"description": "monitor mode can be used to see if a pool is alive", "description": "monitor mode can be used to see if a pool is alive",
@ -202,7 +174,7 @@ option `-setup data_dir Dir`.
"properties": { "properties": {
"executable": { "executable": {
"default": "mean29-generic", "default": "mean29-generic",
"description": "Executable binary of the worker. Can be a fully qualified path, \nbut the software may apply default logic to locate a plain basename.", "description": "Executable binary of the worker. Can be a fully qualified path, but the software may apply default logic to locate a plain basename.",
"type": "string" "type": "string"
}, },
"extra_args": { "extra_args": {

2
ebin/gmhive_client.app
View File

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

1
gmhc_config-testnet-local.eterm
View File

@ -1,5 +1,4 @@
#{ pubkey => <<"ak_zjyvDmMbXafMADYrxDs4uMiYM5zEVJBqWgv619NUQ37Gkwt7z">> #{ pubkey => <<"ak_zjyvDmMbXafMADYrxDs4uMiYM5zEVJBqWgv619NUQ37Gkwt7z">>
, report => <<"progress">>
, pool_admin => #{url => <<"local">>} , pool_admin => #{url => <<"local">>}
, pool => #{id => <<"ct_26xqeE3YKmZV8jrks57JSgZRCHSuG4RGzpnvdz6AAiSSTVbJRM">>, , pool => #{id => <<"ct_26xqeE3YKmZV8jrks57JSgZRCHSuG4RGzpnvdz6AAiSSTVbJRM">>,
host => <<"127.0.0.1">>} host => <<"127.0.0.1">>}

2
gmhc_config.eterm
View File

@ -1,5 +1,5 @@
#{ pubkey => <<"ak_2KAcA2Pp1nrR8Wkt3FtCkReGzAi8vJ9Snxa4PcmrthVx8AhPe8">> #{ pubkey => <<"ak_2KAcA2Pp1nrR8Wkt3FtCkReGzAi8vJ9Snxa4PcmrthVx8AhPe8">>
, pool => #{id => <<"ct_LRbi65kmLtE7YMkG6mvG5TxAXTsPJDZjAtsPuaXtRyPA7gnfJ">>} , pool => #{id => <<"ct_LRbi65kmLtE7YMkG6mvG5TxAXTsPJDZjAtsPuaXtRyPA7gnfJ">>}
, workers => , mining =>
[#{executable => <<"mean15-generic">>}] [#{executable => <<"mean15-generic">>}]
}. }.

139
priv/gmhc_schema.json
View File

@ -1,139 +0,0 @@
{
"$schema": "http://json-schema.org/draft-04/schema#",
"additionalProperties": false,
"properties": {
"cache_dir": {
"description": "Location of cache, default is 'setup:data_dir()'",
"type": "string"
},
"extra_pubkeys": {
"default": [],
"description": "Additional worker pubkeys, sharing rewards",
"items": {
"pattern": "^ak_[1-9A-HJ-NP-Za-km-z]*$",
"type": "string"
},
"type": "array"
},
"network": {
"default": "mainnet",
"type": "string"
},
"pool": {
"additionalProperties": false,
"properties": {
"host": {
"default": "127.0.0.1",
"description": "Hostname of hive server",
"example": "0.0.0.0",
"type": "string"
},
"id": {
"description": "Pool contract id",
"pattern": "^ct_[1-9A-HJ-NP-Za-km-z]*$",
"type": "string"
},
"port": {
"default": 17888,
"description": "Hive server listen port",
"minimum": 1,
"type": "integer"
}
},
"type": "object"
},
"pool_admin": {
"additionalProperties": false,
"properties": {
"default_per_network": {
"additionalProperties": false,
"properties": {
"mainnet": {
"default": "https://gajumining.com/api/workers/{CLIENT_ID}",
"type": "string"
},
"testnet": {
"default": "https://test.gajumining.com/api/workers/{CLIENT_ID}",
"type": "string"
}
},
"type": "object"
},
"url": {
"default": "https://test.gajumining.com/api/workers/{CLIENT_ID}",
"description": "URL of Eureka worker api",
"type": "string"
}
},
"type": "object"
},
"pubkey": {
"description": "Primary client pubkey",
"pattern": "^ak_[1-9A-HJ-NP-Za-km-z]*$",
"type": "string"
},
"report": {
"default": "silent",
"description": "Progress reporting",
"enum": [
"debug",
"progress",
"silent"
],
"type": "string"
},
"type": {
"default": "worker",
"description": "monitor mode can be used to see if a pool is alive",
"enum": [
"worker",
"monitor"
],
"type": "string"
},
"workers": {
"default": [
{ "executable": "mean29-generic" }
],
"description": "Definitions of workers' configurations. If no worker are configured one worker is used as default, i.e. 'mean29-generic' executable without any extra args.",
"items": {
"additionalProperties": false,
"properties": {
"executable": {
"default": "mean29-generic",
"description": "Executable binary of the worker. Can be a fully qualified path, \nbut the software may apply default logic to locate a plain basename.",
"type": "string"
},
"extra_args": {
"default": "",
"description": "Extra arguments to pass to the worker executable binary. The safest choice is specifying no arguments i.e. empty string.",
"type": "string"
},
"hex_encoded_header": {
"default": false,
"description": "Hexadecimal encode the header argument that is send to the worker executable. CUDA executables expect hex encoded header.",
"type": "boolean"
},
"instances": {
"description": "Instances used by the worker in case of Multi-GPU mining. Numbers on the configuration list represent GPU devices that are to be addressed by the worker.",
"example": [0,1,2,3],
"items": { "type": "integer" },
"minItems": 1,
"type": "array"
},
"repeats": {
"default": 1,
"description": "Number of tries to do in each worker context - WARNING: it should be set so the worker process runs for 3-5s or else the node risk missing out on new micro blocks.",
"type": "integer"
}
},
"required": [
"executable"
],
"type": "object"
},
"type": "array"
}
},
"type": "object"
}

6
rebar.config
View File

@ -5,14 +5,12 @@
{erl_opts, [debug_info]}. {erl_opts, [debug_info]}.
{plugins, [rebar3_hex]}. {plugins, [rebar3_hex]}.
{post_hooks, [{compile, "make schema"}]}.
{deps, [ {deps, [
{enoise, {git, "https://git.qpq.swiss/QPQ-AG/enoise.git", {ref, "029292817e"}}}, {enoise, {git, "https://git.qpq.swiss/QPQ-AG/enoise.git", {ref, "029292817e"}}},
{gmhive_protocol, {gmhive_protocol,
{git, "https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git", {git, "https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git",
{ref, "8d4652a"}}}, {ref, "818ce33"}}},
{gmhive_worker, {git, "https://git.qpq.swiss/QPQ-AG/gmhive_worker", {ref, "cabd104114"}}}, {gmhive_worker, {git, "https://git.qpq.swiss/QPQ-AG/gmhive_worker", {ref, "255ef59"}}},
{gmconfig, {git, "https://git.qpq.swiss/QPQ-AG/gmconfig.git", {gmconfig, {git, "https://git.qpq.swiss/QPQ-AG/gmconfig.git",
{ref, "38620ff9e2"}}}, {ref, "38620ff9e2"}}},
{gproc, "1.0.0"}, {gproc, "1.0.0"},

6
rebar.lock
View File

@ -21,15 +21,15 @@
0}, 0},
{<<"gmcuckoo">>, {<<"gmcuckoo">>,
{git,"https://git.qpq.swiss/QPQ-AG/gmcuckoo.git", {git,"https://git.qpq.swiss/QPQ-AG/gmcuckoo.git",
{ref,"256e14e7c88043132245be902ff9756070d285b4"}}, {ref,"106e1cd2e4ff81286f6bc7d7c85f83bc20e14b82"}},
1}, 1},
{<<"gmhive_protocol">>, {<<"gmhive_protocol">>,
{git,"https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git", {git,"https://git.qpq.swiss/QPQ-AG/gmhive_protocol.git",
{ref,"8d4652a79a2ad8f51e1fe560c15c698d4c92485c"}}, {ref,"818ce33cc1dec74c020515be48fb548a5207befd"}},
0}, 0},
{<<"gmhive_worker">>, {<<"gmhive_worker">>,
{git,"https://git.qpq.swiss/QPQ-AG/gmhive_worker", {git,"https://git.qpq.swiss/QPQ-AG/gmhive_worker",
{ref,"cabd104114691edb925aacd7e04d431d47bac420"}}, {ref,"255ef59ccd7f795d2d25f2d0ebcf24e3251b6f36"}},
0}, 0},
{<<"gmserialization">>, {<<"gmserialization">>,
{git,"https://git.qpq.swiss/QPQ-AG/gmserialization.git", {git,"https://git.qpq.swiss/QPQ-AG/gmserialization.git",

51
scripts/update_readme_json.sh
View File

@ -1,51 +0,0 @@
#!/usr/bin/env bash
set -euo pipefail
README="$1"
SCHEMA="$2"
if [[ ! -f "$README" || ! -f "$SCHEMA" ]]; then
echo "Usage: $0 README.md schema.json"
exit 1
fi
tmpfile=$(mktemp)
awk -v schema_file="$SCHEMA" '
BEGIN {
in_json_block = 0
in_schema_section = 0
}
/^##[[:space:]]+JSON[[:space:]]+Schema/ {
in_schema_section = 1
print
next
}
/^##[[:space:]]+/ && in_schema_section {
# Another section starts, end the schema section
in_schema_section = 0
}
in_schema_section && /^```json/ {
print "```json"
while ((getline line < schema_file) > 0) print line
close(schema_file)
in_json_block = 1
next
}
in_json_block && /^```$/ {
print "```"
in_json_block = 0
next
}
!(in_schema_section && in_json_block) {
print
}
' "$README" > "$tmpfile"
mv "$tmpfile" "$README"
echo "✅ Updated JSON schema block in $README"

28
src/gmhc_app.erl
View File

@ -1,6 +1,5 @@
%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*-
-module(gmhc_app). -module(gmhc_app).
-vsn("0.8.3").
-behaviour(application). -behaviour(application).
@ -17,11 +16,11 @@
-spec start([{atom(), any()}]) -> {ok, [atom()]} | {error, any()}. -spec start([{atom(), any()}]) -> {ok, [atom()]} | {error, any()}.
start(Opts) -> start(Opts) ->
application:load(gmhive_client), application:load(gmhive_client),
_ = application:stop(gmhive_client), {error,_} = application:stop(gmhive_client),
_ = lists:foreach(fun({K, V}) -> _ = lists:foreach(fun({K, V}) ->
application:set_env(gmhive_client, K, V) application:set_env(gmhive_client, K, V)
end, Opts), end, Opts),
application:ensure_all_started(gmhive_client, permanent). application:ensure_all_started(gmhive_client).
start(_StartType, _StartArgs) -> start(_StartType, _StartArgs) ->
set_things_up(), set_things_up(),
@ -45,31 +44,8 @@ stop(_State) ->
ok. ok.
set_things_up() -> set_things_up() ->
maybe_add_logger_handler(),
gmhc_counters:initialize(), gmhc_counters:initialize(),
gmhc_config:load_config(), gmhc_config:load_config(),
logger:set_module_level([gmhw_pow_cuckoo], notice), logger:set_module_level([gmhw_pow_cuckoo], notice),
?LOG_DEBUG("Config: ~p", [gmconfig:user_config()]), ?LOG_DEBUG("Config: ~p", [gmconfig:user_config()]),
case gmhc_config:get_config([<<"report">>]) of
<<"debug">> ->
?LOG_NOTICE("Starting debug reporter", []),
gmhc_events:debug();
<<"progress">> ->
?LOG_NOTICE("Starting progress reporter", []),
gmhc_events:progress();
_ -> ok
end,
ok. ok.
maybe_add_logger_handler() ->
case is_headless() orelse application:get_env(gmhive_client, tty_logger, false) of
true ->
Level = application:get_env(gmhive_client, tty_logger_level, error),
io:fwrite("Adding logger handler: ~p~n", [Level]),
logger:add_handler(gmhc_tty, logger_std_h, #{level => Level});
false ->
ok
end.
is_headless() ->
undefined == application:get_key(gajumine, vsn).

13
src/gmhc_config.erl
View File

@ -1,5 +1,4 @@
-module(gmhc_config). -module(gmhc_config).
-vsn("0.8.3").
-export([ load_config/0 -export([ load_config/0
, get_config/1 , get_config/1
@ -14,20 +13,8 @@ load_config() ->
gmconfig:apply_os_env(), gmconfig:apply_os_env(),
gmconfig:process_plain_args(), gmconfig:process_plain_args(),
check_application_env(), check_application_env(),
check_final_config(),
ok. ok.
check_final_config() ->
ExtraPubkeys = get_config([<<"extra_pubkeys">>]),
AllKeys = [get_config([<<"pubkey">>]) | ExtraPubkeys],
case AllKeys -- lists:usort(AllKeys) of
[] ->
ok;
Duplicates ->
?LOG_ERROR("Duplicate account ids found: ~p", [Duplicates]),
error({duplicate_account_ids, Duplicates})
end.
instrument_gmconfig() -> instrument_gmconfig() ->
gmconfig:set_gmconfig_env(gmconfig_env()). gmconfig:set_gmconfig_env(gmconfig_env()).

19
src/gmhc_config_schema.erl
View File

@ -1,9 +1,7 @@
-module(gmhc_config_schema). -module(gmhc_config_schema).
-vsn("0.8.3").
-export([ schema/0 -export([ schema/0
, to_json/0 , to_json/0 ]).
, export/1 ]).
-import(gmconfig_schema_helpers, -import(gmconfig_schema_helpers,
[ str/1 [ str/1
@ -23,17 +21,6 @@
to_json() -> to_json() ->
json:encode(schema()). json:encode(schema()).
export(ToFile) ->
case file:open(ToFile, [write]) of
{ok, Fd} ->
try ok = io:put_chars(Fd, json:format(schema(), #{indent => 4}))
after
file:close(Fd)
end;
{error, _} = Error ->
Error
end.
schema() -> schema() ->
obj(schema_init(), obj(schema_init(),
#{ #{
@ -51,10 +38,6 @@ schema() ->
, pool => pool() , pool => pool()
, pool_admin => pool_admin() , pool_admin => pool_admin()
, workers => workers() , workers => workers()
, cache_dir => str(#{description => <<"Location of cache, default is 'setup:data_dir()'">>})
, report => str(#{ enum => [<<"debug">>, <<"progress">>, <<"silent">>]
, default => <<"silent">>
, description => <<"Progress reporting">> })
}). }).
pool() -> pool() ->

138
src/gmhc_connector.erl
View File

@ -1,5 +1,4 @@
-module(gmhc_connector). -module(gmhc_connector).
-vsn("0.8.3").
-behaviour(gen_server). -behaviour(gen_server).
@ -56,11 +55,7 @@
id :: non_neg_integer() id :: non_neg_integer()
, auto_connect = true :: boolean() , auto_connect = true :: boolean()
, econn , econn
, connected = false :: boolean()
, status = disconnected :: disconnected | connecting | connected
, reconnect = true :: boolean()
, reconnect_timer :: timer_ref() | 'undefined' , reconnect_timer :: timer_ref() | 'undefined'
, recache_timer :: reference() | 'undefined'
, awaiting_connect = [] :: list() , awaiting_connect = [] :: list()
, protocol :: binary() | 'undefined' , protocol :: binary() | 'undefined'
, version :: binary() | 'undefined' , version :: binary() | 'undefined'
@ -145,9 +140,6 @@ init(#{id := Id} = Opts) when is_map(Opts) ->
{ok, S} -> {ok, S} ->
?LOG_DEBUG("Initial connect succeeded", []), ?LOG_DEBUG("Initial connect succeeded", []),
S; S;
{error, rejected} ->
?LOG_WARNING("Connection rejected; will not retry", []),
S0#st{econn = undefined, reconnect = false};
{error, _} = Error -> {error, _} = Error ->
?LOG_WARNING("Could not connect to core server: ~p", [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})
@ -186,7 +178,6 @@ handle_call({connect, Opts}, From, #st{awaiting_connect = Waiters} = S) ->
end, end,
S1 = start_reconnect_timer( S1 = start_reconnect_timer(
S#st{ auto_connect = true S#st{ auto_connect = true
, reconnect = true
, awaiting_connect = Waiters1 }, Opts), , awaiting_connect = Waiters1 }, Opts),
{noreply, S1}; {noreply, S1};
false -> false ->
@ -247,21 +238,12 @@ handle_info({timeout, TRef, {reconnect, Opts}}, #st{ reconnect_timer = {TRef, _,
end; end;
handle_info({tcp_closed, _Port}, #st{} = S) -> handle_info({tcp_closed, _Port}, #st{} = S) ->
?LOG_DEBUG("got tcp_closed", []), ?LOG_DEBUG("got tcp_closed", []),
S1 = disconnected(S#st.id, S), disconnected(S#st.id),
S2 = case S1#st.auto_connect of S1 = case S#st.auto_connect of
true -> start_reconnect_timer(S1#st{econn = undefined}); true -> start_reconnect_timer(S#st{econn = undefined});
false -> S1#st{econn = undefined} false -> S#st{econn = undefined}
end, end,
{noreply, S2}; {noreply, S1};
handle_info({timeout, _, recache}, #st{opts = Opts, econn = EConn} = S0) ->
S = S0#st{recache_timer = undefined},
case (EConn =/= undefined) andalso S#st.connected of
false ->
{noreply, S};
true ->
cache_eureka_info(Opts),
{noreply, ensure_recache_timer(S)}
end;
handle_info(Msg, S) -> handle_info(Msg, S) ->
?LOG_DEBUG("Discarding msg (auto_connect=~p): ~p", [S#st.auto_connect, Msg]), ?LOG_DEBUG("Discarding msg (auto_connect=~p): ~p", [S#st.auto_connect, Msg]),
{noreply, S}. {noreply, S}.
@ -281,8 +263,6 @@ code_change(_FromVsn, S, _Extra) ->
try_connect(Opts, S) -> try_connect(Opts, S) ->
try try_connect_(Opts, S) try try_connect_(Opts, S)
catch catch
error:rejected:_ ->
{error, rejected};
error:E:T -> error:E:T ->
?LOG_ERROR("Unexpected error connecting: ~p / ~p", [E, T]), ?LOG_ERROR("Unexpected error connecting: ~p / ~p", [E, T]),
{error, E} {error, E}
@ -296,9 +276,8 @@ try_connect_(Opts0, S) ->
case try_noise_connect(maps:merge(Opts0, PoolOpts)) of case try_noise_connect(maps:merge(Opts0, PoolOpts)) of
{ok, EConn, Opts1} -> {ok, EConn, Opts1} ->
S1 = protocol_connect(Opts1, S#st{ econn = EConn S1 = protocol_connect(Opts1, S#st{ econn = EConn
, status = connecting
, reconnect_timer = undefined }), , reconnect_timer = undefined }),
{ok, S1#st{status = connected}}; {ok, S1};
{error, _} = Error -> {error, _} = Error ->
Error Error
end end
@ -306,12 +285,12 @@ try_connect_(Opts0, S) ->
eureka_get_host_port() -> eureka_get_host_port() ->
case gmhc_eureka:get_pool_address() of case gmhc_eureka:get_pool_address() of
{ok, #{host := Host, #{<<"address">> := Host,
port := Port, <<"port">> := Port,
pool_id := PoolId}} -> <<"pool_id">> := PoolId} ->
#{host => unicode:characters_to_list(Host), #{host => binary_to_list(Host),
port => Port, port => Port,
pool_id => unicode:characters_to_list(PoolId)}; pool_id => binary_to_list(PoolId)};
{error, _} = Error -> {error, _} = Error ->
Error Error
end. end.
@ -353,8 +332,6 @@ default_tcp_opts() ->
enoise_opts() -> enoise_opts() ->
[{noise, <<"Noise_NN_25519_ChaChaPoly_BLAKE2b">>}]. [{noise, <<"Noise_NN_25519_ChaChaPoly_BLAKE2b">>}].
start_reconnect_timer(#st{reconnect = false} = S) ->
S;
start_reconnect_timer(#st{} = S) -> start_reconnect_timer(#st{} = S) ->
start_reconnect_timer(S, #{}). start_reconnect_timer(S, #{}).
@ -366,18 +343,11 @@ start_reconnect_timer(#st{} = S, Opts) ->
false -> false ->
?LOG_DEBUG("starting reconnect timer ...", []), ?LOG_DEBUG("starting reconnect timer ...", []),
TRef = start_timer(1000, Opts), TRef = start_timer(1000, Opts),
S#st{reconnect_timer = {TRef, 10, 8000 + gmhc_lib:rand(3000), Opts}} S#st{reconnect_timer = {TRef, 10, 1000, Opts}}
end. end.
restart_reconnect_timer(#st{reconnect = false} = S) ->
cancel_reconnect_timer(S);
restart_reconnect_timer(#st{reconnect_timer = {_, 0, T, Opts}} = S) -> restart_reconnect_timer(#st{reconnect_timer = {_, 0, T, Opts}} = S) ->
NewT = max(T + 5000 + gmhc_lib:rand(1000), ?MAX_RETRY_INTERVAL), NewT = max(T * 2, ?MAX_RETRY_INTERVAL),
if (NewT > T andalso NewT =:= ?MAX_RETRY_INTERVAL) ->
gmhc_eureka:invalidate_cache();
true ->
ok
end,
TRef = start_timer(NewT, Opts), TRef = start_timer(NewT, Opts),
S#st{reconnect_timer = {TRef, 10, NewT, Opts}}; S#st{reconnect_timer = {TRef, 10, NewT, Opts}};
restart_reconnect_timer(#st{reconnect_timer = {_, N, T, Opts}} = S) -> restart_reconnect_timer(#st{reconnect_timer = {_, N, T, Opts}} = S) ->
@ -402,25 +372,11 @@ notify_deadline(D, #st{awaiting_connect = Waiters} = S) ->
end, [], Waiters), end, [], Waiters),
S#st{awaiting_connect = Waiters1}. S#st{awaiting_connect = Waiters1}.
cache_eureka_info(Opts) -> notify_connected(#st{id = Id, awaiting_connect = Waiters} = S) ->
gmhc_eureka:cache_good_address(maps:with([host, port, pool_id], Opts)).
notify_connected(#st{id = Id, awaiting_connect = Waiters, opts = Opts} = S) ->
gmhc_events:publish(connected, #{id => Id}), gmhc_events:publish(connected, #{id => Id}),
[gen_server:reply(From, ok) || {From, _} <- Waiters], [gen_server:reply(From, ok) || {From, _} <- Waiters],
gmhc_handler:pool_connected(S#st.id, S#st.opts), gmhc_handler:pool_connected(S#st.id, S#st.opts),
cache_eureka_info(Opts), S#st{awaiting_connect = []}.
gmhc_connectors_sup:add_restart_info(Id, Opts),
ensure_recache_timer(S#st{awaiting_connect = []}).
ensure_recache_timer(#st{recache_timer = T} = S) ->
case T of
undefined ->
TRef = erlang:start_timer(1*60*1000, self(), recache),
S#st{recache_timer = TRef};
_ when is_reference(T) ->
S
end.
cancel_reconnect_timer(#st{reconnect_timer = T} = S) -> cancel_reconnect_timer(#st{reconnect_timer = T} = S) ->
case T of case T of
@ -444,32 +400,17 @@ protocol_connect(Opts, #st{econn = EConn} = S) ->
Type = to_atom(opt(type, Opts, [<<"type">>])), Type = to_atom(opt(type, Opts, [<<"type">>])),
RId = erlang:unique_integer(), RId = erlang:unique_integer(),
Vsns = gmhp_msgs:versions(), Vsns = gmhp_msgs:versions(),
Client = client_name(),
Protocols = gmhp_msgs:protocols(hd(Vsns)), Protocols = gmhp_msgs:protocols(hd(Vsns)),
ConnectReq = #{ protocols => Protocols ConnectReq = #{ protocols => Protocols
, versions => Vsns , versions => Vsns
, pool_id => PoolId , pool_id => PoolId
, pubkey => Pubkey , pubkey => Pubkey
, extra_pubkeys => Extra , extra_pubkeys => Extra
, client => Client
, type => Type , type => Type
, nonces => gmhc_server:total_nonces() , nonces => gmhc_server:total_nonces()
, signature => ""}, , signature => ""},
?LOG_DEBUG("ConnectReq = ~p", [ConnectReq]), ?LOG_DEBUG("ConnectReq = ~p", [ConnectReq]),
try gmhp_msgs:encode_connect(ConnectReq, RId) of Msg = gmhp_msgs:encode_connect(ConnectReq, RId),
Msg ->
send_connect(EConn, RId, Msg, ConnectReq, Opts, S)
catch error:Error ->
ErrMsg = unicode:characters_to_binary(io_lib:fwrite("~p", [Error])),
disconnected(S#st.id, #{error =>
#{code => gmhp_msgs:error_code(invalid_input),
message => ErrMsg}}, S)
end.
send_connect(EConn, RId, Msg, #{pubkey := Pubkey,
extra_pubkeys := Extra,
pool_id := PoolId,
type := Type}, Opts, S) ->
enoise:send(EConn, Msg), enoise:send(EConn, Msg),
receive receive
{noise, EConn, Data} -> {noise, EConn, Data} ->
@ -478,41 +419,20 @@ send_connect(EConn, RId, Msg, #{pubkey := Pubkey,
, result := #{connect_ack := #{ protocol := P , result := #{connect_ack := #{ protocol := P
, version := V }} , version := V }}
}} -> }} ->
S1 = connected(S#st.id, Type, S), connected(S#st.id, Type),
Opts1 = Opts#{ pubkey => Pubkey Opts1 = Opts#{ pubkey => Pubkey
, extra => Extra , extra => Extra
, pool_id => PoolId , pool_id => PoolId
, type => Type }, , type => Type },
notify_connected(S1#st{protocol = P, version = V, opts = Opts1}); notify_connected(S#st{protocol = P, version = V, opts = Opts1});
#{error := #{code := _, message := ErrMsg}} = ErrReply -> #{error := #{message := Msg}} ->
?LOG_ERROR("Connect error: ~s", [ErrMsg]), ?LOG_ERROR("Connect error: ~s", [Msg]),
%% TODO: fix the flow so that we send one disconnected event, error(protocol_connect)
%% 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 end
after 10000 -> after 10000 ->
gmhc_eureka:invalidate_cache(),
error(protocol_connect_timeout) error(protocol_connect_timeout)
end. end.
client_name() ->
MyStr = app_string(gmhive_client),
case app_string(gajumine) of
<<>> -> MyStr;
GMStr -> <<MyStr/binary,",",GMStr/binary>>
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) -> to_bin(A) when is_atom(A) ->
atom_to_binary(A, utf8); atom_to_binary(A, utf8);
to_bin(S) -> to_bin(S) ->
@ -522,18 +442,12 @@ to_atom(A) when is_atom(A) -> A;
to_atom(S) -> to_atom(S) ->
binary_to_existing_atom(iolist_to_binary(S), utf8). binary_to_existing_atom(iolist_to_binary(S), utf8).
connected(Id, Type, S) when Type==worker; Type==monitor -> connected(Id, Type) when Type==worker; Type==monitor ->
gmhc_server:connected(Id, Type), gmhc_server:connected(Id, Type).
S#st{connected = true}.
disconnected(Id, S) -> disconnected(Id) ->
disconnected(Id, #{}, S). gmhc_events:publish(disconnected, #{id => Id}),
gmhc_server:disconnected(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, status = disconnected}.
opt_autoconnect(#{auto_connect := Bool}) when is_boolean(Bool) -> opt_autoconnect(#{auto_connect := Bool}) when is_boolean(Bool) ->
Bool; Bool;

44
src/gmhc_connectors_sup.erl
View File

@ -1,37 +1,14 @@
-module(gmhc_connectors_sup). -module(gmhc_connectors_sup).
-vsn("0.8.3").
-behavior(supervisor). -behavior(supervisor).
-export([ start_link/0 -export([ start_link/0
, init/1 ]). , init/1 ]).
-export([ start_first_connector/0 -export([ start_first_connector/0
, start_connector/1 , start_connector/1]).
, add_restart_info/2 ]).
-export([ sort_restart_info/1 ]).
-include_lib("kernel/include/logger.hrl").
start_link() -> start_link() ->
case supervisor:start_link({local, ?MODULE}, ?MODULE, []) of supervisor:start_link({local, ?MODULE}, ?MODULE, []).
{ok, _} = Ok ->
RI0 = gmhc_watchdog:get_restart_info(),
?LOG_ERROR("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]),
maps:foreach(fun restart_connector/2, RI),
Ok;
Other ->
Other
end.
restart_connector({?MODULE, _}, Opts) ->
start_connector(Opts);
restart_connector(_, _) ->
ok.
start_first_connector() -> start_first_connector() ->
start_connector(#{}). start_connector(#{}).
@ -43,14 +20,11 @@ start_connector(Opts0) ->
end, end,
supervisor:start_child(?MODULE, [Opts]). supervisor:start_child(?MODULE, [Opts]).
add_restart_info(Id, Opts) ->
gmhc_watchdog:note_started({?MODULE, Id}, Opts).
init([]) -> init([]) ->
Mod = gmhc_connector, Mod = gmhc_connector,
SupFlags = #{ strategy => simple_one_for_one SupFlags = #{ strategy => simple_one_for_one
, intensity => 5 , intensity => 3
, period => 60 }, , period => 10 },
ChildSpecs = [ #{ id => Mod ChildSpecs = [ #{ id => Mod
, start => {Mod, start_link, []} , start => {Mod, start_link, []}
, type => worker , type => worker
@ -58,13 +32,3 @@ init([]) ->
, shutdown => 5000 , shutdown => 5000
, modules => [Mod] } ], , modules => [Mod] } ],
{ok, {SupFlags, ChildSpecs}}. {ok, {SupFlags, ChildSpecs}}.
sort_restart_info(RI) ->
L = lists:sort(
[{Id, {T,H,P}, I}
|| {{?MODULE,_} = Id, #{type := T, host := H, port := P} = I} <- maps:to_list(RI)]),
ConnTypes = [CT || {_, CT, _} <- lists:ukeysort(2, L)],
lists:foldl(fun(CT, Acc) ->
{Id, _, I} = lists:keyfind(CT, 2, L),
Acc#{Id => I}
end, #{}, ConnTypes).

1
src/gmhc_counters.erl
View File

@ -1,5 +1,4 @@
-module(gmhc_counters). -module(gmhc_counters).
-vsn("0.8.3").
-export([ initialize/0 ]). -export([ initialize/0 ]).

133
src/gmhc_eureka.erl
View File

@ -1,136 +1,19 @@
-module(gmhc_eureka). -module(gmhc_eureka).
-vsn("0.8.3").
-export([get_pool_address/0]). -export([get_pool_address/0]).
-export([cache_good_address/1,
invalidate_cache/0,
cached_address/0,
cache_filename/0,
cache_dir/0]).
-include_lib("kernel/include/logger.hrl"). -include_lib("kernel/include/logger.hrl").
-include("gmhc_events.hrl"). -include("gmhc_events.hrl").
get_pool_address() -> get_pool_address() ->
case cached_address() of
{ok, _} = Ok -> Ok;
{error, _} ->
get_pool_address_()
end.
cached_address() ->
I0 = cache_info(),
CacheF = cache_filename(I0),
?LOG_DEBUG("Eureka cache filename: ~p", [CacheF]),
case file:read_file(CacheF) of
{ok, Bin} ->
NowTS = erlang:system_time(seconds),
OldestTS = NowTS - 24*60*60,
try binary_to_term(Bin) of
#{ ts := TS
, network := N
, pubkey := PK
, host := _
, port := _
, pool_id := _} = Map when N == map_get(network, I0),
PK == map_get(pubkey, I0) ->
if TS >= OldestTS ->
Result = maps:remove(ts, Map),
?LOG_DEBUG("Cached eureka info: ~p", [Result]),
{ok, Result};
true ->
{error, outdated}
end;
Other ->
{error, {invalid_cache_term, Other}}
catch
error:E ->
{error, {invalid_cache_data, E}}
end;
{error, _} = Err ->
Err
end.
cache_good_address(#{host := _,
port := _,
pool_id := _} = I0) ->
CacheInfo = cache_info(I0),
CacheF = cache_filename(CacheInfo),
ToCache = CacheInfo#{ts => erlang:system_time(seconds)},
case filelib:ensure_dir(CacheF) of
ok ->
case file:write_file(CacheF, term_to_binary(ToCache)) of
ok ->
{ok, ToCache};
{error, _} = Err ->
?LOG_DEBUG("Couldn't cache eureka in ~p: ~p", [CacheF, Err]),
Err
end;
{error, _} = Err ->
?LOG_ERROR("Cannot save cached info to ~s", [CacheF]),
Err
end.
invalidate_cache() ->
CacheF = cache_filename(),
case file:delete(CacheF) of
ok ->
?LOG_DEBUG("Eureka cache file removed (~p)", [CacheF]),
ok;
{error, _} = Err ->
?LOG_DEBUG("Couldn't remove Eureka cache (~p): ~p", [CacheF, Err]),
Err
end.
cache_info(#{ host := Addr
, port := Port
, pool_id := PoolId }) ->
I0 = cache_info(),
I0#{ host => unicode:characters_to_binary(Addr)
, port => Port
, pool_id => unicode:characters_to_binary(PoolId)}.
cache_info() ->
Pubkey = gmhc_config:get_config([<<"pubkey">>]),
Network = gmhc_config:get_config([<<"network">>]),
#{ pubkey => Pubkey
, network => Network }.
cache_filename() ->
cache_filename(cache_info()).
cache_filename(#{network := Network, pubkey := Pubkey}) ->
Path = filename:join(cache_dir(), Network),
<<"ak_", PKShort:8/binary, _/binary>> = Pubkey,
filename:join(Path, "gmhc_eureka." ++ binary_to_list(PKShort) ++ ".cache").
cache_dir() ->
case gmconfig:find_config([<<"cache_dir">>]) of
{ok, D} ->
D;
undefined ->
case setup_zomp:is_zomp_context() of
true ->
cache_dir_zomp();
false ->
filename:join(setup:data_dir(), "gmhive.cache")
end
end.
cache_dir_zomp() ->
#{package_id := {Realm, App, _}} = zx_daemon:meta(),
filename:join(zx_lib:ppath(var, {Realm, App}), "gmhive.cache").
get_pool_address_() ->
case gmconfig:find_config([<<"pool_admin">>, <<"url">>], [user_config]) of case gmconfig:find_config([<<"pool_admin">>, <<"url">>], [user_config]) of
{ok, URL0} -> {ok, URL0} ->
case expand_url(URL0) of case expand_url(URL0) of
<<"local">> -> <<"local">> ->
{ok, #{host => <<"127.0.0.1">>, #{<<"address">> => <<"127.0.0.1">>,
port => gmconfig:get_config( <<"port">> => gmconfig:get_config(
[<<"pool">>, <<"port">>], [schema_default]), [<<"pool">>, <<"port">>], [schema_default]),
pool_id => gmhc_config:get_config([<<"pool">>, <<"id">>]) }}; <<"pool_id">> => gmhc_config:get_config([<<"pool">>, <<"id">>]) };
URL -> URL ->
?LOG_INFO("Trying to connect to ~p", [URL]), ?LOG_INFO("Trying to connect to ~p", [URL]),
connect1(URL) connect1(URL)
@ -165,13 +48,9 @@ connect1(URL0) ->
Error Error
end. end.
get_host_port(#{ <<"address">> := Addr get_host_port(Data) ->
, <<"port">> := Port
, <<"pool_id">> := PoolId } = Data) ->
?LOG_DEBUG("Data = ~p", [Data]), ?LOG_DEBUG("Data = ~p", [Data]),
{ok, #{ host => Addr maps:with([<<"address">>, <<"port">>, <<"pool_id">>], Data).
, port => Port
, pool_id => PoolId }}.
request(get, URL) -> request(get, URL) ->
case request(get, URL, []) of case request(get, URL, []) of

119
src/gmhc_events.erl
View File

@ -1,5 +1,4 @@
-module(gmhc_events). -module(gmhc_events).
-vsn("0.8.3").
-export([subscribe/1, -export([subscribe/1,
ensure_subscribed/1, ensure_subscribed/1,
@ -7,17 +6,7 @@
ensure_unsubscribed/1, ensure_unsubscribed/1,
publish/2]). publish/2]).
-export([debug/0, -export([debug/0]).
progress/0,
stop/0]).
-export([rpt_debug/2,
rpt_progress/2]).
%% internal
-export([init_reporter/2]).
-include_lib("kernel/include/logger.hrl").
-export_type([event/0]). -export_type([event/0]).
@ -70,100 +59,22 @@ ensure_unsubscribed(Event) ->
debug() -> debug() ->
ok = application:ensure_started(gproc), ok = application:ensure_started(gproc),
spawn_reporter(fun() -> spawn(fun() ->
sub(), subscribe(pool_notification),
gmhive_worker:subscribe_returns(), subscribe({pool_notification, new_generation}),
loop(fun rpt_debug/2, false) subscribe(connected),
end). subscribe(puzzle),
subscribe(result),
subscribe(error),
subscribe(disconnected),
loop()
end).
progress() -> loop() ->
ok = application:ensure_started(gproc),
spawn_reporter(fun() ->
sub(),
loop(fun rpt_progress/2, true)
end).
spawn_reporter(F) ->
Parent = self(),
proc_lib:start_link(?MODULE, init_reporter, [F, Parent]).
init_reporter(F, Parent) ->
try_register_reporter(),
proc_lib:init_ack(Parent, self()),
F().
stop() ->
case whereis(gmhc_reporter) of
undefined ->
not_running;
Pid ->
exit(Pid, kill)
end.
sub() ->
subscribe(pool_notification),
subscribe({pool_notification, new_generation}),
subscribe(connected),
subscribe(puzzle),
subscribe(result),
subscribe(error),
subscribe(disconnected).
loop(F, Ts) ->
receive receive
stop -> ok; stop -> ok;
{gproc_ps_event, E, Data} -> {gproc_ps_event, E, Data} ->
maybe_print(F(E, Data), Ts), io:fwrite("EVENT ~p: ~p~n", [E, Data]),
loop(F, Ts) loop()
end. end.
try_register_reporter() ->
try register(gmhc_reporter, self())
catch
error:_ ->
?LOG_ERROR("Reporter already running. Try gmhc_events:stop().", []),
error(already_running)
end.
maybe_print([], _) ->
ok;
maybe_print(String, Ts) when is_boolean(Ts) ->
TSstr = [[ts(), " "] || Ts],
io:put_chars([TSstr, String, "\n"]).
ts() ->
calendar:system_time_to_rfc3339(erlang:system_time(millisecond),
[{unit, millisecond}, {offset, "Z"}]).
rpt_debug(E, Data) ->
io_lib:fwrite("EVENT ~p: ~p", [E, Data]).
rpt_progress(puzzle, #{info := {_Data, _Target, Nonce, _Config}}) ->
w("Trying nonce: ~p", [Nonce]);
rpt_progress(result, #{info := Info}) ->
case Info of
{error, no_solution} ->
[];
{ok, Cycles} ->
w("Found! Reporting ~w cycles to leader.", [length(Cycles)]);
Other ->
w("Unexpected 'result': ~tp", [Other])
end;
rpt_progress(pool_notification, #{info := #{msg := Msg}}) ->
case Msg of
#{solution_accepted := #{seq := Seq}} ->
w("The hive has produced a solution! Sequence: ~w", [Seq]);
#{new_generation := _} -> [];
#{candidate := _} -> [];
Other ->
w("Unexpected 'pool_notification': ~tp", [Other])
end;
rpt_progress(connected, _) ->
w("Connected!", []);
rpt_progress(disconnected, _) ->
w("Disconnected!", []);
rpt_progress(_, _) ->
[].
w(Fmt, Args) ->
io_lib:fwrite(Fmt, Args).

50
src/gmhc_handler.erl
View File

@ -1,5 +1,4 @@
-module(gmhc_handler). -module(gmhc_handler).
-vsn("0.8.3").
-behavior(gen_server). -behavior(gen_server).
-export([ start_link/0 -export([ start_link/0
@ -12,7 +11,6 @@
]). ]).
-export([ call/1 -export([ call/1
, async_call/1
, notify/1 , notify/1
, pool_connected/2 , pool_connected/2
, from_pool/1 ]). , from_pool/1 ]).
@ -28,7 +26,7 @@
-record(st, {pools = [], opts = #{}}). -record(st, {pools = [], opts = #{}}).
-define(CALL_TIMEOUT, 10_000). -define(CALL_TIMEOUT, 5000).
-include_lib("kernel/include/logger.hrl"). -include_lib("kernel/include/logger.hrl").
@ -39,13 +37,6 @@ call(Req) ->
{error, Reason} {error, Reason}
end. end.
async_call(Req) ->
try gen_server:call(?MODULE, {async_call, Req}, ?CALL_TIMEOUT)
catch
exit:Reason ->
{error, Reason}
end.
notify(Msg) -> notify(Msg) ->
gen_server:cast(?MODULE, {notify, Msg}). gen_server:cast(?MODULE, {notify, Msg}).
@ -64,8 +55,6 @@ init([]) ->
handle_call({call, Req}, _From, #st{} = S) -> handle_call({call, Req}, _From, #st{} = S) ->
{reply, call_connector(Req), S}; {reply, call_connector(Req), S};
handle_call({async_call, Req}, _From, #st{} = S) ->
{reply, call_connector(Req, false), S};
handle_call(_Req, _From, S) -> handle_call(_Req, _From, S) ->
{reply, {error, unknown_method}, S}. {reply, {error, unknown_method}, S}.
@ -129,39 +118,24 @@ maybe_publish(_) ->
maybe_via(#{via := Via}, Info) -> maybe_via(#{via := Via}, Info) ->
Info#{via => Via}. Info#{via => Via}.
call_connector(Req) -> call_connector(Req0) ->
call_connector(Req, true).
call_connector(Req0, Wait) ->
{ViaId, Req} = maps:take(via, Req0), {ViaId, Req} = maps:take(via, Req0),
case gmhc_connector:whereis_id(ViaId) of case gmhc_connector:whereis_id(ViaId) of
undefined -> undefined ->
{error, no_connection}; {error, no_connection};
Pid when is_pid(Pid) -> Pid when is_pid(Pid) ->
Id = erlang:unique_integer(), Id = erlang:unique_integer(),
MRef = case Wait of MRef = erlang:monitor(process, Pid),
true -> erlang:monitor(process, Pid);
false -> none
end,
gmhc_connector:send(ViaId, #{call => Req#{ id => Id }}), gmhc_connector:send(ViaId, #{call => Req#{ id => Id }}),
case Wait of receive
true -> {from_pool, #{reply := #{ id := Id, result := Result }}} ->
receive Result;
{from_pool, #{reply := #{ id := Id {from_pool, #{error := #{ id := Id } = Error}} ->
, result := Result }}} -> {error, maps:remove(id, Error)};
erlang:demonitor(MRef), {'DOWN', MRef, _, _, _} ->
Result; {error, no_connection}
{from_pool, #{error := #{ id := Id } = Error}} -> after 5000 ->
erlang:demonitor(MRef), {error, {timeout, process_info(self(), messages)}}
{error, maps:remove(id, Error)};
{'DOWN', MRef, _, _, _} ->
{error, no_connection}
after 5000 ->
erlang:demonitor(MRef),
{error, {timeout, process_info(self(), messages)}}
end;
false ->
ok
end end
end. end.

9
src/gmhc_lib.erl
View File

@ -1,9 +0,0 @@
-module(gmhc_lib).
-vsn("0.8.3").
-export([ rand/1 ]).
rand(Range) ->
<<Rx:32>> = crypto:strong_rand_bytes(4),
Range - (Rx rem Range).

88
src/gmhc_server.erl
View File

@ -1,5 +1,4 @@
-module(gmhc_server). -module(gmhc_server).
-vsn("0.8.3").
-behaviour(gen_server). -behaviour(gen_server).
@ -45,7 +44,7 @@
-define(CONNECTED(S), map_size(S#st.connected) > 0). -define(CONNECTED(S), map_size(S#st.connected) > 0).
-define(MAX_ERRORS, 50). -define(MAX_ERRORS, 5).
connected(Id, Type) -> connected(Id, Type) ->
gen_server:call(?MODULE, {connected, Id, Type}). gen_server:call(?MODULE, {connected, Id, Type}).
@ -92,9 +91,7 @@ handle_call({connected, Id, Type}, {Pid,_}, #st{connected = Conn} = S) ->
monitor -> monitor ->
stop_workers(S1#st.workers), % shouldn't be any running stop_workers(S1#st.workers), % shouldn't be any running
S1#st{workers = [], working = false}; S1#st{workers = [], working = false};
worker -> worker -> S1#st{working = true}
gmhc_watchdog:watch(5*60_000, 1), %% 5 minutes, one-shot
S1#st{working = true}
end, end,
{reply, ok, S2}; {reply, ok, S2};
handle_call(_Req, _From, S) -> handle_call(_Req, _From, S) ->
@ -110,25 +107,21 @@ handle_cast({from_pool, #{via := Connector,
%% We could check whether we have already received the candidate ... %% We could check whether we have already received the candidate ...
%% For now, stop all workers, restart with new candidate %% For now, stop all workers, restart with new candidate
try try
%% Most of the time we don't want to stop the worker. If we do, though, then % Most of the time we don't want to stop the worker. If we do, though, then
%% we need to do it more carefully than this, or memory usage will triple. % we need to do it more carefully than this, or memory usage will triple.
%% Workers1 = stop_workers(Workers), % Workers1 = stop_workers(Workers),
%% {Workers2, Cand1} = assign_nonces(Workers, Cand),
%% Nonces may be [], in which case we need to request new nonces first. #st{candidate = Cand2} = S1 = maybe_request_nonces(S#st{candidate = Cand1}),
#st{candidate = Cand1} = S1 = maybe_request_nonces(S#st{candidate = Cand}), NewWorkers = [spawn_worker(W, Cand2) || W <- Workers2],
{Workers2, Cand2} = assign_nonces(Workers, Cand1), {noreply, S1#st{workers = NewWorkers}}
#st{candidate = Cand3} = S2 = maybe_request_nonces(S1#st{candidate = Cand2}),
NewWorkers = [spawn_worker(W, Cand3) || W <- Workers2],
{noreply, S2#st{workers = NewWorkers}}
catch catch
Cat:Err:St -> Cat:Err:St ->
?LOG_ERROR("CAUGHT ~p:~p / ~p", [Cat, Err, St]), ?LOG_ERROR("CAUGHT ~p:~p / ~p", [Cat, Err, St]),
{noreply, S} {noreply, S}
end; end;
handle_cast({disconnected, Id}, #st{connected = Conn} = S) -> handle_cast({disconnected, Id}, #st{connected = Conn} = S) ->
?LOG_DEBUG("disconnected: ~p", [Id]), ?LOG_DEBUG("disconnected: ~p", [Id]),
Conn1 = maps:remove(Id, Conn), Conn1 = maps:remove(Id, Conn),
gmhc_watchdog:unwatch(),
S1 = if map_size(Conn1) == 0 -> S1 = if map_size(Conn1) == 0 ->
Ws = stop_workers(S#st.workers), Ws = stop_workers(S#st.workers),
S#st{connected = Conn1, workers = Ws}; S#st{connected = Conn1, workers = Ws};
@ -158,18 +151,11 @@ handle_info({'EXIT', Pid, Reason}, #st{ workers = Workers
gmhc_events:publish(error, ?ERR_EVT(#{error => worker_error, gmhc_events:publish(error, ?ERR_EVT(#{error => worker_error,
data => Reason})), data => Reason})),
Ws1 = incr_worker_error(W, Workers), Ws1 = incr_worker_error(W, Workers),
erlang:start_timer(100, self(), check_workers),
{noreply, S#st{workers = Ws1}}; {noreply, S#st{workers = Ws1}};
false -> false ->
%% ?LOG_DEBUG("EXIT apparently not from worker?? (~p)", [Pid]), %% ?LOG_DEBUG("EXIT apparently not from worker?? (~p)", [Pid]),
{noreply, S} {noreply, S}
end; end;
handle_info({timeout, _, check_workers}, #st{workers = Workers} = S) ->
S1 = maybe_request_nonces(S),
S2 = lists:foldl(fun(W, Sx) ->
maybe_restart_worker(W, Sx)
end, S1, Workers),
{noreply, S2};
handle_info(Msg, St) -> handle_info(Msg, St) ->
?LOG_DEBUG("Unknown msg: ~p", [Msg]), ?LOG_DEBUG("Unknown msg: ~p", [Msg]),
{noreply, St}. {noreply, St}.
@ -182,9 +168,6 @@ code_change(_FromVsn, S, _Extra) ->
report_solutions(Solutions, W, #st{} = S) when ?CONNECTED(S) -> report_solutions(Solutions, W, #st{} = S) when ?CONNECTED(S) ->
#{via := Via, seq := Seq} = W#worker.cand, #{via := Via, seq := Seq} = W#worker.cand,
Nonces = all_nonces(W),
[report_no_solution_(Via, Seq, N)
|| N <- Nonces, not lists:keymember(N, 1, Solutions)],
gmhc_handler:call( gmhc_handler:call(
#{via => Via, #{via => Via,
solutions => #{ seq => Seq solutions => #{ seq => Seq
@ -192,28 +175,24 @@ report_solutions(Solutions, W, #st{} = S) when ?CONNECTED(S) ->
, evidence => Evd } , evidence => Evd }
|| {Nonce, Evd} <- Solutions] }}). || {Nonce, Evd} <- Solutions] }}).
report_no_solution(_Nonce, W, #st{} = S) when ?CONNECTED(S) -> %% report_solution(Nonce, Solution, W, #st{connected = true}) ->
%% #{seq := Seq} = W#worker.cand,
%% gmhc_handler:call(#{solution => #{ seq => Seq
%% , nonce => Nonce
%% , evidence => Solution }}).
report_no_solution(Nonce, W, #st{} = S) when ?CONNECTED(S) ->
#{via := Via, seq := Seq} = W#worker.cand, #{via := Via, seq := Seq} = W#worker.cand,
Nonces = all_nonces(W),
%% ?LOG_DEBUG("report no_solution Seq = ~p, Nonce = ~p", [Seq, Nonce]), %% ?LOG_DEBUG("report no_solution Seq = ~p, Nonce = ~p", [Seq, Nonce]),
[report_no_solution_(Via, Seq, Nonce1) || Nonce1 <- Nonces], gmhc_handler:call(#{via => Via,
ok. no_solution => #{ seq => Seq
, nonce => Nonce}}).
report_no_solution_(Via, Seq, Nonce) ->
gmhc_handler:async_call(#{via => Via,
no_solution => #{ seq => Seq
, nonce => Nonce}}).
all_nonces(#worker{nonce = Nonce, config = Config}) ->
case gmhw_pow_cuckoo:repeats(Config) of
1 -> [Nonce];
Rs -> lists:seq(Nonce, Nonce + Rs - 1)
end.
maybe_request_nonces(#st{ candidate = #{via := Via, seq := Seq, nonces := Nonces} maybe_request_nonces(#st{ candidate = #{via := Via, seq := Seq, nonces := Nonces}
, nonces = N} = S) when ?CONNECTED(S) -> , nonces = N} = S) when ?CONNECTED(S) ->
case Nonces == [] of case Nonces == [] of
true -> true ->
%% ?LOG_DEBUG("Request more nonces, Seq = ~p, N = ~p", [Seq, N]),
Res = gmhc_handler:call(#{via => Via, Res = gmhc_handler:call(#{via => Via,
get_nonces => #{ seq => Seq get_nonces => #{ seq => Seq
, n => N }}), , n => N }}),
@ -227,7 +206,7 @@ maybe_request_nonces(S) ->
nonces_result(#{nonces := #{seq := Seq, nonces := Nonces}}, Seq0, S) -> nonces_result(#{nonces := #{seq := Seq, nonces := Nonces}}, Seq0, S) ->
case Seq == Seq0 of case Seq == Seq0 of
true -> true ->
wd_ping(), %% ?LOG_DEBUG("Got nonces = ~p", [Nonces]),
#st{candidate = Cand} = S, #st{candidate = Cand} = S,
S#st{candidate = Cand#{nonces => Nonces}}; S#st{candidate = Cand#{nonces => Nonces}};
false -> false ->
@ -237,32 +216,21 @@ nonces_result(#{nonces := #{seq := Seq, nonces := Nonces}}, Seq0, S) ->
nonces_result({error, Reason}, Seq0, S) -> nonces_result({error, Reason}, Seq0, S) ->
?LOG_DEBUG("Got error on nonce request: ~p", [Reason]), ?LOG_DEBUG("Got error on nonce request: ~p", [Reason]),
Workers = stop_workers_for_seq(Seq0, S#st.workers), Workers = stop_workers_for_seq(Seq0, S#st.workers),
case Reason of
{timeout, _} ->
Timeout = retry_timeout(1000, 3000),
erlang:start_timer(Timeout, self(), check_workers);
_ ->
ok
end,
S#st{workers = Workers}. S#st{workers = Workers}.
retry_timeout(Floor, Range) ->
Floor + gmhc_lib:rand(Range).
handle_worker_result({worker_result, Result}, W, S) -> handle_worker_result({worker_result, Result}, W, S) ->
%% ?LOG_DEBUG("worker result: ~p", [Result]), %% ?LOG_DEBUG("worker result: ~p", [Result]),
wd_ping(),
case Result of case Result of
{solutions, Solutions} -> {solutions, Solutions} ->
{Cont, S1} = report_solutions_(Solutions, W, S), {Cont, S1} = report_solutions_(Solutions, W, S),
maybe_continue(Cont, reset_errors(W), S1); maybe_continue(Cont, W, S1);
{solution, Nonce, Solution} -> {solution, Nonce, Solution} ->
%% report_solution(Nonce, Solution, W, S), %% report_solution(Nonce, Solution, W, S),
{Cont, S1} = report_solutions_([{Nonce, Solution}], W, S), {Cont, S1} = report_solutions_([{Nonce, Solution}], W, S),
maybe_continue(Cont, reset_errors(W), S1); maybe_continue(Cont, W, S1);
{no_solution, Nonce} -> {no_solution, Nonce} ->
report_no_solution(Nonce, W, S), report_no_solution(Nonce, W, S),
maybe_restart_worker(reset_errors(W), S); maybe_restart_worker(W, S);
{error, S} -> {error, S} ->
?LOG_DEBUG("Worker ~p reported error as normal", [W#worker.index]), ?LOG_DEBUG("Worker ~p reported error as normal", [W#worker.index]),
gmhc_events:publish(error, ?ERR_EVT(#{error => worker_error, gmhc_events:publish(error, ?ERR_EVT(#{error => worker_error,
@ -290,9 +258,6 @@ report_solutions_(Solutions, W, S) ->
{error, S} {error, S}
end. end.
reset_errors(#worker{} = W) ->
W#worker{errors = 0}.
reset_worker(#worker{index = I} = W, Ws) -> reset_worker(#worker{index = I} = W, Ws) ->
W1 = reset_worker_(W), W1 = reset_worker_(W),
lists:keyreplace(I, #worker.index, Ws, W1). lists:keyreplace(I, #worker.index, Ws, W1).
@ -443,6 +408,3 @@ worker_result(Pid, Result) ->
decode_candidate_hash(#{candidate := C} = Cand) -> decode_candidate_hash(#{candidate := C} = Cand) ->
{ok, Hash} = gmser_api_encoder:safe_decode(bytearray, C), {ok, Hash} = gmser_api_encoder:safe_decode(bytearray, C),
Cand#{candidate := Hash}. Cand#{candidate := Hash}.
wd_ping() ->
gmhc_watchdog:ping().

10
src/gmhc_sup.erl
View File

@ -1,6 +1,5 @@
%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*-
-module(gmhc_sup). -module(gmhc_sup).
-vsn("0.8.3").
-behaviour(supervisor). -behaviour(supervisor).
@ -15,13 +14,12 @@ start_link() ->
supervisor:start_link({local, ?MODULE}, ?MODULE, []). supervisor:start_link({local, ?MODULE}, ?MODULE, []).
init([]) -> init([]) ->
ChildSpecs = [ worker(gmhc_watchdog) ChildSpecs = [ worker(gmhc_server)
, worker(gmhc_server)
, worker(gmhc_handler) , worker(gmhc_handler)
, supervisor(gmhc_connectors_sup) ], , supervisor(gmhc_connectors_sup) ],
SupFlags = #{ strategy => rest_for_one SupFlags = #{ strategy => one_for_one
, intensity => 5 %% We really want the hive client to sort itself out , intensity => 1
, period => 5*60 %% Timemout issues can happen infrequently , period => 5
, auto_shutdown => never }, , auto_shutdown => never },
{ok, {SupFlags, ChildSpecs}}. {ok, {SupFlags, ChildSpecs}}.

167
src/gmhc_watchdog.erl
View File

@ -1,167 +0,0 @@
-module(gmhc_watchdog).
-behavior(gen_server).
-export([ watch/2
, unwatch/0
, ping/0 ]).
-export([ note_started/2
, remove_restart_info/1
, get_restart_info/0 ]).
-export([ start_link/0
, init/1
, handle_call/3
, handle_cast/2
, handle_info/2
, terminate/2
, code_change/3 ]).
-record(st, {services = #{}}).
-record(svc, { n = 5 :: pos_integer()
, n0 = 5 :: pos_integer()
, interval = 5000 :: pos_integer()
, tref :: reference()
, mref :: reference()
}).
-include_lib("kernel/include/logger.hrl").
watch(Interval, N) ->
gen_server:call(?MODULE, {watch, self(), Interval, N}).
unwatch() ->
gen_server:cast(?MODULE, {unwatch, self()}).
ping() ->
gen_server:cast(?MODULE, {ping, self()}).
note_started(Id, Info) ->
gen_server:call(?MODULE, {note_started, Id, Info}).
get_restart_info() ->
gen_server:call(?MODULE, get_restart_info).
remove_restart_info([]) ->
ok;
remove_restart_info(IDs) ->
gen_server:call(?MODULE, {remove_restart_info, IDs}).
start_link() ->
gen_server:start_link({local, ?MODULE}, ?MODULE, [], []).
init([]) ->
{ok, #st{}}.
handle_call({note_started, Id, Info}, _From, S) ->
update_pt(Id, Info),
{reply, ok, S};
handle_call({remove_restart_info, IDs}, _From, S) ->
remove_restart_info_(IDs),
{reply, ok, S};
handle_call(get_restart_info, _From, S) ->
{reply, get_pt(), S};
handle_call({watch, Pid, Interval, N}, _From, S) ->
{reply, ok, add_watch(Pid, Interval, N, S)};
handle_call(_Req, _From, S) ->
{reply, {error, unknown_method}, S}.
handle_cast({ping, Pid}, S) ->
{noreply, reset_watch(Pid, S)};
handle_cast({unwatch, Pid}, S) ->
{noreply, delete_watch(Pid, S)};
handle_cast(Msg, S) ->
?LOG_DEBUG("Unknown cast: ~p", [Msg]),
{noreply, S}.
handle_info({timeout, TRef, Pid}, S) ->
?LOG_INFO("Timeout for pid ~p", [Pid]),
{noreply, ping_timeout(Pid, TRef, S)};
handle_info({'DOWN', _, process, Pid, _}, S) ->
{noreply, delete_watch(Pid, S)};
handle_info(Msg, S) ->
?LOG_DEBUG("Unknown msg: ~p", [Msg]),
{noreply, S}.
terminate(_, _) ->
ok.
code_change(_FromVsn, S, _Extra) ->
{ok, S}.
add_watch(Pid, Interval, N, #st{services = Svcs} = S) ->
MRef = erlang:monitor(process, Pid),
Svc0 = #svc{ interval = Interval
, mref = MRef
, n = N
, n0 = N},
Svc = start_timer(Pid, Svc0),
S#st{services = Svcs#{Pid => Svc}}.
reset_watch(Pid, #st{services = Svcs} = S) ->
case maps:find(Pid, Svcs) of
{ok, #svc{ n0 = N0 } = Svc} ->
Svc1 = restart_timer(Pid, Svc#svc{n = N0}),
S#st{services = Svcs#{Pid := Svc1}};
error ->
S
end.
delete_watch(Pid, #st{services = Svcs} = S) ->
case maps:find(Pid, Svcs) of
{ok, #svc{tref = TRef, mref = MRef}} ->
erlang:cancel_timer(TRef),
erlang:demonitor(MRef),
S#st{services = maps:remove(Pid, Svcs)};
error ->
S
end.
ping_timeout(Pid, TRef, #st{services = Svcs} = S) ->
case maps:find(Pid, Svcs) of
{ok, #svc{ n = N, tref = TRef} = Svc} ->
N1 = N-1,
if N1 =< 0 ->
?LOG_ERROR("Will exit Pid ~p", [Pid]),
exit(Pid, kill),
S#st{services = maps:remove(Pid, Svcs)};
true ->
Svc1 = restart_timer(Pid, Svc#svc{n = N1}),
S#st{services = Svcs#{Pid := Svc1}}
end;
{ok, _} ->
?LOG_DEBUG("Timeout didn't match TRef - ignoring", []),
S;
_ ->
S
end.
start_timer(Pid, #svc{interval = T} = Svc) ->
TRef = erlang:start_timer(T, self(), Pid),
Svc#svc{tref = TRef}.
restart_timer(Pid, #svc{tref = TRef} = Svc) ->
erlang:cancel_timer(TRef),
start_timer(Pid, Svc#svc{tref = undefined}).
update_pt(Id, Info) ->
Pt = get_pt(),
put_pt(Pt#{Id => Info}).
remove_restart_info_(IDs) ->
RI = get_pt(),
case maps:without(IDs, RI) of
RI ->
ok;
NewRI ->
put_pt(NewRI)
end.
get_pt() ->
persistent_term:get(pt_key(), #{}).
put_pt(Pt) ->
persistent_term:put(pt_key(), Pt).
pt_key() ->
{?MODULE, restart_info}.

1
src/gmhc_workers.erl
View File

@ -8,7 +8,6 @@
%%%------------------------------------------------------------------- %%%-------------------------------------------------------------------
-module(gmhc_workers). -module(gmhc_workers).
-vsn("0.8.3").
-export([ -export([
get_worker_configs/0 get_worker_configs/0

2
src/gmhive_client.app.src
View File

@ -1,7 +1,7 @@
%% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*- %% -*- mode: erlang; erlang-indent-level: 4; indent-tabs-mode: nil -*-
{application, gmhive_client, {application, gmhive_client,
[{description, "Gajumaru Hive Client"}, [{description, "Gajumaru Hive Client"},
{vsn, "zomp"}, {vsn, "0.1.0"},
{registered, []}, {registered, []},
{applications, {applications,
[ [

14
src/gmhive_client.app.src.script
View File

@ -1,14 +0,0 @@
%% -*- erlang-mode; erlang-indent-level: 4; indent-tabs-mode: nil -*-
[{application, Name, Opts}] = CONFIG.
case lists:keyfind(vsn, 1, Opts) of
{vsn, "zomp"} ->
ZompMetaF = filename:join(filename:dirname(filename:dirname(SCRIPT)), "zomp.meta"),
{ok, ZMeta} = file:consult(ZompMetaF),
{_, {_, _, {Vmaj,Vmin,Vpatch}}} = lists:keyfind(package_id, 1, ZMeta),
VsnStr = unicode:characters_to_list(io_lib:fwrite("~w.~w.~w", [Vmaj, Vmin, Vpatch])),
Opts1 = lists:keyreplace(vsn, 1, Opts, {vsn, VsnStr}),
[{application, Name, Opts1}];
_ ->
CONFIG
end.

1
src/gmhive_client.erl
View File

@ -1,5 +1,4 @@
-module(gmhive_client). -module(gmhive_client).
-vsn("0.8.3").
-export([ connect/1 -export([ connect/1
, disconnect/1 , disconnect/1

10
zomp.meta
View File

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

7
zompify.sh
View File

@ -38,12 +38,5 @@ rm "$IGNORE_TEMP"
cp "$PWD/zomp.meta" "$DST/" cp "$PWD/zomp.meta" "$DST/"
cp "$PWD/Emakefile" "$DST/" cp "$PWD/Emakefile" "$DST/"
# copy generated schema
SCHEMA="$SRC/priv/gmhc_schema.json"
if [ -e "$SCHEMA" ]; then
mkdir -p "$DST/priv"
cp -a "$SCHEMA" "$DST/priv/$(basename "$SCHEMA")"
fi
# Clean up beam files just in case # Clean up beam files just in case
[ -d "$DST/ebin" ] && find "$DST/ebin" -name '*.beam' -exec rm -f {} + || true [ -d "$DST/ebin" ] && find "$DST/ebin" -name '*.beam' -exec rm -f {} + || true