foo
This commit is contained in:
parent
9aea2d189b
commit
e5308bbad0
@ -53,7 +53,9 @@
|
||||
-type username() :: label().
|
||||
-type lower0_9() :: [$a..$z | $0..$9 | $_].
|
||||
-type label() :: [$a..$z | $0..$9 | $_ | $- | $.].
|
||||
-type package_meta() :: map().
|
||||
-type package_meta() :: #{package_id := package_id(),
|
||||
deps := [package_id()],
|
||||
type := app | lib,}.
|
||||
|
||||
|
||||
|
||||
|
||||
@ -11,7 +11,7 @@
|
||||
-copyright("Craig Everett <zxq9@zxq9.com>").
|
||||
-license("GPL-3.0").
|
||||
|
||||
-export([start/1, stop/1]).
|
||||
-export([start_monitor/1, stop/1]).
|
||||
-export([start_link/1]).
|
||||
|
||||
-include("zx_logger.erl").
|
||||
@ -20,9 +20,9 @@
|
||||
|
||||
%%% Startup
|
||||
|
||||
-spec start(Target) -> Result
|
||||
-spec start_monitor(Target) -> Result
|
||||
when Target :: zx:host(),
|
||||
Result :: {ok, pid()}
|
||||
Result :: {ok, PID :: pid(), Mon :: reference()}
|
||||
| {error, Reason},
|
||||
Reason :: term().
|
||||
%% @doc
|
||||
@ -30,8 +30,14 @@
|
||||
%% but this process may fail to connect or crash immediately after spawning. Should
|
||||
%% only be called by zx_daemon.
|
||||
|
||||
start(Target) ->
|
||||
zx_conn_sup:start_conn(Target).
|
||||
start_monitor(Target) ->
|
||||
case zx_conn_sup:start_conn(Target) of
|
||||
{ok, Pid} ->
|
||||
Mon = monitor(process, Pid),
|
||||
{ok, Pid, Mon};
|
||||
Error ->
|
||||
Error
|
||||
end.
|
||||
|
||||
|
||||
-spec stop(Conn :: pid()) -> ok.
|
||||
|
||||
@ -107,93 +107,32 @@
|
||||
reqm = none :: none | reference(),
|
||||
action = none :: none | action(),
|
||||
actions = queue:new() :: queue:queue(action()),
|
||||
cx = #cx{} :: conn_index()}).
|
||||
cx = cx_load() :: conn_index()}).
|
||||
|
||||
|
||||
-record(cx,
|
||||
{realms = #{} :: #{zx:realm() := zx:serial()},
|
||||
primes = #{} :: #{zx:realm() := zx:host()},
|
||||
hosts = #{} :: #{zx:realm() := queue:queue(zx:host())},
|
||||
conns = #{} :: #{zx:realm() := pid()},
|
||||
attempts = #{} :: #{pid() := zx:host()},
|
||||
zx_conns = sets:new() :: sets:set({pid(), reference()})}).
|
||||
|
||||
-spec cx_connected(Available, Conn, CX) -> Result
|
||||
when Available :: [{zx:realm(), zx:serial()}],
|
||||
Conn :: pid(),
|
||||
CX :: conn_index(),
|
||||
Result :: {Assignment, NewCX},
|
||||
Assignment :: assigned | unassigned,
|
||||
NewCX :: conn_index().
|
||||
%% @private
|
||||
%% An abstract data handler which is called whenever a new connection is successfully
|
||||
%% established by a zx_conn. Any unconnected realms with a valid serial will be
|
||||
%% assigned to the new connection; if none are needed then the connection is closed.
|
||||
%% The successful host is placed back in the hosts queue for each available realm.
|
||||
%% The return value is a tuple that indicates whether the new connection was assigned
|
||||
%% or not and the updated CX data value.
|
||||
|
||||
cx_connected(Available, Conn, CX) ->
|
||||
cx_connected(unassigned, Available, Conn, CX).
|
||||
{realms = #{} :: #{zx:realm() := realm_meta()},
|
||||
assigned = [] :: [{zx:realm(), pid()}],
|
||||
attempts = [] :: [{pid(), reference(), zx:host()}],
|
||||
conns = [] :: [{pid(), reference(), zx:host()}]}).
|
||||
|
||||
|
||||
cx_connected(A, [{Realm, Serial} | Rest], Conn, CX = #cx{realms = Realms}) ->
|
||||
case maps:find(Realm, Realms) of
|
||||
{ok, S} when S < Serial ->
|
||||
NewRealms = maps:update(Realm, Serial, Realms),
|
||||
{NewA, NewCX} = cx_assign(A, Conn, Realm, CX#cx{realms = NewRealms}),
|
||||
cx_connected(NewA, Rest, Conn, NewCX);
|
||||
{ok, S} when S == Serial ->
|
||||
{NewA, NewCX} = cx_assign(A, Conn, Realm, CX),
|
||||
cx_connected(NewA, Rest, Conn, NewCX);
|
||||
{ok, S} when S > Serial ->
|
||||
cx_connected(A, Rest, Conn, CX);
|
||||
error ->
|
||||
cx_connected(A, Rest, Conn, CX)
|
||||
end;
|
||||
cx_connected(A, [], Conn, CX = #cx{attempts = Attempts}) ->
|
||||
NewCX = CX#cx{attempts = maps:remove(Conn, Attempts)},
|
||||
{A, NewCX}.
|
||||
|
||||
|
||||
cx_assign(A, Conn, Realm, CX = #cx{hosts = Hosts, conns = Conns, attempts = Attempts}) ->
|
||||
Host = maps:get(Conn, Attempts),
|
||||
Enqueue = fun(Q) -> queue:in(Host, Q) end,
|
||||
NewHosts = maps:update_with(Realm, Enqueue, Hosts),
|
||||
case maps:is_key(Realm, Conns) of
|
||||
true ->
|
||||
{A, CX#cx{hosts = NewHosts}};
|
||||
false ->
|
||||
NewConns = maps:put(Realm, Conn, Conns),
|
||||
{assigned, CX#cx{hosts = NewHosts, conns = NewConns}}
|
||||
end.
|
||||
|
||||
|
||||
-spec cx_disconnect(Conn, CX) -> NewCX
|
||||
|
||||
cx_disconnected(Conn, CX) ->
|
||||
|
||||
|
||||
-spec cx_failed(Conn, CX) -> NewCX
|
||||
|
||||
cx_failed(Conn, CX) ->
|
||||
|
||||
|
||||
-spec cx_next(Realm, CX) ->
|
||||
|
||||
cx_next(Realm, CX) ->
|
||||
|
||||
|
||||
-spec cx_resolve(Realm, CX) -> Conn
|
||||
|
||||
cx_resolve(Realm, CX) ->
|
||||
|
||||
-record(rmeta,
|
||||
{revision = 0 :: non_neg_integer(),
|
||||
serial = 0 :: non_neg_integer(),
|
||||
prime = {"zomp.psychobitch.party", 11311} :: zx:host(),
|
||||
private = [] :: [zx:host()],
|
||||
mirrors = queue:new() :: queue:queue(zx:host()),
|
||||
realm_keys = [] :: [zx:key_meta()],
|
||||
package_keys = [] :: [zx:key_meta()],
|
||||
sysops = [] :: [zx:sysop_meta()]}).
|
||||
|
||||
|
||||
-type state() :: #s{}.
|
||||
-type conn_index() :: #cx{}.
|
||||
-type realm_meta() :: #rmeta{}.
|
||||
|
||||
-type action() :: {subscribe, zx:package()},
|
||||
-type action() :: {subscribe, zx:package()}
|
||||
| unsubscribe
|
||||
| {list, zx:realm()}
|
||||
| {list, zx:realm(), zx:name()}
|
||||
@ -290,7 +229,7 @@ list_versions(Package) ->
|
||||
|
||||
-spec query_latest(Object) -> Result
|
||||
when Object :: zx:package() | zx:package_id(),
|
||||
Result :: {ok, version()}
|
||||
Result :: {ok, zx:version()}
|
||||
| {error, Reason},
|
||||
Reason :: bad_realm
|
||||
| bad_package
|
||||
@ -358,23 +297,30 @@ start_link() ->
|
||||
-spec init(none) -> {ok, state()}.
|
||||
|
||||
init(none) ->
|
||||
SerialFile = filename:join(zx_lib:zomp_home(), "realm.serials"),
|
||||
Serials =
|
||||
case file:consult(SerialFile) of
|
||||
{ok, Ss} ->
|
||||
maps:from_list(Ss);
|
||||
{error, enoent} ->
|
||||
ok = log(info, "Initializing zomp/realm.serials..."),
|
||||
maps:new();
|
||||
{error, Reason} ->
|
||||
Message = "Reading zomp/realm.serials failed with ~tp. Recreating..."
|
||||
ok = log(error, Message, [Reason]),
|
||||
maps:new()
|
||||
end,
|
||||
State = #s{serials = Serials},
|
||||
CX = cx_load(),
|
||||
{ok, NewCX} = init_connections(CX),
|
||||
State = #s{cx = NewCX},
|
||||
{ok, State}.
|
||||
|
||||
|
||||
init_connections(CX) ->
|
||||
Realms = cx_realms(CX),
|
||||
init_connections(Realms, CX).
|
||||
|
||||
|
||||
init_connections([Realm | Realms], CX}) ->
|
||||
{ok, Hosts, NextCX} = cx_next_hosts(3, Realm, CX),
|
||||
StartConn =
|
||||
fun(Host) ->
|
||||
{ok, Pid, Mon} = zx_conn:start_monitor(Host),
|
||||
{Pid, Mon, Host}
|
||||
end,
|
||||
NewAttempts = lists:map(StartConn, Hosts),
|
||||
NewCX = lists:map(fun cx_add_attempt/2, NextCX, NewAttempts),
|
||||
init_connections(Realms, NewCX);
|
||||
init_connections([], CX) ->
|
||||
{ok, CX}.
|
||||
|
||||
|
||||
%%% gen_server
|
||||
|
||||
@ -464,7 +410,7 @@ do_pass_meta(Meta, Dir, ArgV, State) ->
|
||||
|
||||
do_subscribe(Requestor,
|
||||
{Realm, Name},
|
||||
State = #s{name = none, connp = none, reqp = none,
|
||||
State = #s{meta = none, reqp = none,
|
||||
hosts = Hosts, serials = Serials}) ->
|
||||
Monitor = monitor(process, Requestor),
|
||||
{Host, NewHosts} = select_host(Realm, Hosts),
|
||||
@ -484,31 +430,6 @@ do_subscribe(_, _, State = #s{realm = Realm, name = Name}) ->
|
||||
{{error, {already_subscribed, {Realm, Name}}}, State}.
|
||||
|
||||
|
||||
-spec select_host(Realm, Hosts) -> {Host, NewHosts}
|
||||
when Realm :: zx:realm(),
|
||||
Hosts :: none | hosts(),
|
||||
Host :: zx:host(),
|
||||
NewHosts :: hosts().
|
||||
|
||||
select_host(Realm, none) ->
|
||||
List =
|
||||
case file:consult(zx_lib:hosts_cache_file(Realm)) of
|
||||
{ok, Cached} -> Cached;
|
||||
{error, enoent} -> [zx_lib:get_prime(Realm)]
|
||||
end,
|
||||
NewState = State#s{hosts = #{Realm => List}},
|
||||
select_host(Realm, NewState);
|
||||
select_host(Realm, Hosts) ->
|
||||
{Target, Rest} =
|
||||
case maps:find(Realm, Hosts) of
|
||||
{ok, [H | Hs]} -> {H, Hs};
|
||||
{ok, []} -> {zx_lib:get_prime(Realm), []};
|
||||
error -> {zx_lib:get_prime(Realm), []}
|
||||
end,
|
||||
NewHosts = maps:put(Realm, Hosts, Rest),
|
||||
{Target, NewHosts}.
|
||||
|
||||
|
||||
-spec do_query_latest(Object, State) -> {Result, NewState}
|
||||
when Object :: zx:package() | zx:package_id(),
|
||||
State :: state(),
|
||||
@ -556,7 +477,12 @@ do_unsubscribe(State = #s{connp = ConnP, connm = ConnM}) ->
|
||||
State :: state(),
|
||||
NewState :: state().
|
||||
|
||||
do_report(From, {connected, Realms}, State = #s{
|
||||
do_report(From, {connected, Realms}, State) ->
|
||||
% FIXME
|
||||
ok = log(info,
|
||||
"Would do_report(~tp, {connected, ~tp}, ~tp) here",
|
||||
[From, Realms, State]),
|
||||
State.
|
||||
|
||||
|
||||
|
||||
@ -635,3 +561,379 @@ scrub([]) ->
|
||||
[];
|
||||
scrub(Deps) ->
|
||||
lists:filter(fun(PackageID) -> not zx_lib:installed(PackageID) end, Deps).
|
||||
|
||||
|
||||
|
||||
%%% Connection Cache ADT Interface Functions
|
||||
%%%
|
||||
%%% Functions to manipulate the conn_index() data type are in this section. This
|
||||
%%% data should be treated as abstract by functions outside of this section, as it is
|
||||
%%% a deep structure and future requirements are likely to wind up causing it to
|
||||
%%% change a little. For the same reason, these functions are all pure functions,
|
||||
%%% testable independently of the rest of the module and having no side effects.
|
||||
%%% Return values usually carry some status information with them, and it is up to the
|
||||
%%% caller to react to those responses in an appropriate way.
|
||||
|
||||
-spec cx_load() -> conn_index().
|
||||
%% @private
|
||||
%% Used to load a connection index populated with necessary realm configuration data
|
||||
%% and cached mirror data, if such things can be found in the system, otherwise return
|
||||
%% a blank connection index structure.
|
||||
|
||||
cx_load() ->
|
||||
case cx_populate() of
|
||||
{ok, CX} ->
|
||||
CX;
|
||||
{error, Reason} ->
|
||||
ok = log(error, "Cache load failed with: ~tp", [Reason]),
|
||||
ok = log(warning, "No realms configured."),
|
||||
#cx{}
|
||||
end.
|
||||
|
||||
|
||||
-spec cx_populate() -> Result
|
||||
when Result :: {ok, conn_index()}
|
||||
| {error, Reason},
|
||||
Reason :: no_realms
|
||||
| file:posix().
|
||||
|
||||
cx_populate() ->
|
||||
Home = zx_lib:zomp_home(),
|
||||
Pattern = filename:join(Home, "*.realm"),
|
||||
case filelib:wildcard(Pattern) of
|
||||
[] -> {error, no_realms};
|
||||
RealmFiles -> cx_fetch_cache(RealmFiles, #cx{})
|
||||
end.
|
||||
|
||||
|
||||
cx_populate([File | Files], CX) ->
|
||||
case file:consult(File) of
|
||||
{ok, Meta} ->
|
||||
cx_load_realm_meta(Meta, CX);
|
||||
{error, Reason} ->
|
||||
Message = "Realm file ~tp could not be read. Failed with: ~tp. Skipping.",
|
||||
ok = log(warning, Message, [File, Reason]),
|
||||
populate(Files, CX)
|
||||
end.
|
||||
|
||||
|
||||
cx_load_realm_meta(Meta, CX = #cx{realms = Realms}) ->
|
||||
{realm, Realm} = lists:keyfind(realm, 1, Meta),
|
||||
{revision, Revision} = lists:keyfind(revision, 1, Meta),
|
||||
{prime, Prime} = lists:keyfind(prime, 1, Meta),
|
||||
{realm_keys, RealmKeys} = lists:keyfind(realm_keys, 1, Meta),
|
||||
{package_keys, PackageKeys} = lists:keyfind(packge_keys, 1, Meta),
|
||||
{sysops, Sysops} = lists:keyfind(sysops, 1, Meta),
|
||||
Basic =
|
||||
#rmeta{revision = Revision,
|
||||
serial = Serial,
|
||||
prime = Prime,
|
||||
realm_keys = RealmKeys,
|
||||
package_keys = PackageKeys,
|
||||
sysops = Sysops},
|
||||
Complete = cx_fetch_cache(Realm, Basic),
|
||||
NewRealms = maps:put(Realm, Complete, Realms),
|
||||
CX#cx{realms = NewRealms}.
|
||||
|
||||
|
||||
cx_fetch_cache(Realm, Basic) ->
|
||||
CacheFile = cx_cache_file(Realm),
|
||||
case file:consult(CacheFile) of
|
||||
{ok, Cache} ->
|
||||
{serial, Serial} = lists:keyfind(serial, 1, Cache),
|
||||
{private, Private} = lists:keyfind(private, 1, Cache),
|
||||
{mirrors, Mirrors} = lists:keyfind(mirrors, 1, Cache),
|
||||
PQueue = queue:from_list(Private),
|
||||
Enqueue = fun(H, Q) -> queue:in(H, Q) end,
|
||||
MQueue = lists:foldl(Enqueue, Mirrors, PQueue),
|
||||
Basic#rmeta{serial = Serial, private = Private, mirrors = MQueue};
|
||||
{error, enoent} ->
|
||||
Basic
|
||||
end.
|
||||
|
||||
|
||||
-spec cx_store_cache(CX) -> Result
|
||||
when CX :: conn_index(),
|
||||
Result :: ok
|
||||
| {error, file:posix()}.
|
||||
|
||||
cx_store_cache(#cx{realms = Realms}) ->
|
||||
lists:foreach(fun cx_write_cache/1, maps:to_list(Realms)).
|
||||
|
||||
|
||||
-spec cx_write_cache({zx:realm(), realm_meta()}) -> ok.
|
||||
|
||||
cx_write_cache({Realm,
|
||||
#rmeta{serial = Serial, private = Private, mirrors = Mirrors}}) ->
|
||||
CacheFile = cx_cache_file(Realm),
|
||||
MList = queue:to_list(Mirrors),
|
||||
ActualMirrors = lists:subtract(MList, Private),
|
||||
CacheMeta = [{serial, Serial}, {mirrors, ActualMirrors}],
|
||||
ok = zx_lib:write_terms(CacheFile, CacheMeta),
|
||||
log(info, "Wrote cache for realm ~ts", [Realm]).
|
||||
|
||||
|
||||
-spec cx_cache_file(zx:realm()) -> file:filename().
|
||||
|
||||
cx_cache_file(Realm) ->
|
||||
filename:join(zx_lib:zomp_home(), Realm ++ ".cache").
|
||||
|
||||
|
||||
-spec cx_realms(conn_index()) -> [zx:realms()].
|
||||
|
||||
cx_realms(#cx{realms = Realms}) ->
|
||||
maps:keys(Realms).
|
||||
|
||||
|
||||
-spec cx_next_host(Realm, CX) -> Result
|
||||
when Realm :: zx:realm(),
|
||||
CX :: conn_index(),
|
||||
Result :: {ok, Next, NewCX}
|
||||
| {prime, Prime, NewCX}
|
||||
| {error, Reason},
|
||||
Next :: zx:host(),
|
||||
Prime :: zx:host(),
|
||||
NewCX :: conn_index(),
|
||||
Reason :: bad_realm
|
||||
| connected.
|
||||
%% @private
|
||||
%% Given a realm, retun the next cached host location to which to connect. Returns an
|
||||
%% error if the realm is already assigned or if the realm is not configured.
|
||||
|
||||
cx_next_host(Realm, CX = #cx{assigned = Assigned}) ->
|
||||
case lists:keymember(Realm, 1, Assigned) of
|
||||
false -> cx_next_host2(Realm, CX);
|
||||
true -> {error, connected}
|
||||
end.
|
||||
|
||||
cx_next_host2(Realm, CX = #cx{realm = Realms}) ->
|
||||
case maps:find(Realm, Realms) of
|
||||
{ok, Meta} ->
|
||||
{Result, Host, NewMeta} = cx_next_host3(Meta),
|
||||
NewRealms = maps:put(Realm, NewMeta, Realms),
|
||||
{Result, Host, CX#cx{realms = NewRealms}};
|
||||
error ->
|
||||
{error, bad_realm}
|
||||
end.
|
||||
|
||||
|
||||
-spec cx_next_host3(RealmMeta) -> Result
|
||||
when RealmMeta :: realm_meta(),
|
||||
Result :: {ok, Next, NewRealmMeta}
|
||||
| {prime, Prime, NewRealmMeta},
|
||||
Next :: zx:host(),
|
||||
Prime :: zx:host(),
|
||||
NewRealmMeta :: realm_meta().
|
||||
%% @private
|
||||
%% Match on the important success cases first.
|
||||
%% If there is a locally configured set of private mirrors (usually on the same LAN,
|
||||
%% or public ones an organization hosts for its own users) then try those first.
|
||||
%% Trying "first" is a relative concept in long-lived systems that experience a high
|
||||
%% frequency of network churn. When the daemon is initialized a queue is created from
|
||||
%% the known public mirrors, and then the private mirrors are enqueued at the head of
|
||||
%% the mirrors so they will be encountered first.
|
||||
%% If the entire combined mirrors list is exhausted then the prime node will be
|
||||
%% returned, but also as a consequence the prime mirrors will be reloaded at the head
|
||||
%% once again so if the prime fails (or causes a redirect) the private mirrors will
|
||||
%% once again occur in the queue.
|
||||
%% If the prime node is returned it is indicated with the atom `prime', which should
|
||||
%% indicate to the caller that any iterative host scanning or connection procedures
|
||||
%% should treat this as the last value of interest (and stop spawning connections).
|
||||
|
||||
cx_next_host3(Meta = #rmeta{prime = Prime, private = Privae, mirrors = Mirrors}) ->
|
||||
case queue:is_empty(Mirrors) of
|
||||
false ->
|
||||
{{value, Next}, NewMirrors} = queue:out(Mirrors),
|
||||
{ok, Next, Meta#rmeta{mirrors = NewMirrors}};
|
||||
true ->
|
||||
Enqueue = fun(H, Q) -> queue:in(H, Q) end,
|
||||
NewMirrors = lists:foldl(Enqueue, Private, Mirrors),
|
||||
{prime, Prime, Meta#rmeta{mirrors = NewMirrors}}
|
||||
end.
|
||||
|
||||
|
||||
-spec cx_next_hosts(N, Realm, CX) -> Result
|
||||
when N :: non_neg_integer(),
|
||||
Realm :: zx:realm(),
|
||||
CX :: conn_index(),
|
||||
Result :: {ok, Hosts, NewCX}
|
||||
| {error, Reason},
|
||||
Hosts :: [zx:host()],
|
||||
NewCX :: conn_index(),
|
||||
Reason :: bad_realm
|
||||
| connected.
|
||||
%% @private
|
||||
%% This function allows recruiting an arbitrary number of hosts from the host cache
|
||||
%% of a given realm, taking private mirrors first, then public mirrors, and ending
|
||||
%% with the prime node for the realm if no others exist.
|
||||
|
||||
cx_next_hosts(N, Realm, CX = #cx{assigned = Assigned}) ->
|
||||
case lists:keymember(Realm, 1, Assigned) of
|
||||
false -> cx_next_hosts2(N, Realm, CX);
|
||||
true -> {error, connected}
|
||||
end.
|
||||
|
||||
|
||||
cx_next_hosts2(N, Realm, CX = #cx{realms = Realms}) ->
|
||||
case maps:find(Realm, Realms) of
|
||||
{ok, Meta} ->
|
||||
{ok, Hosts, NewMeta} = cx_next_hosts3(N, [], Meta),
|
||||
NewRealms = maps:put(Realm, NewMeta, Realms),
|
||||
{ok, Hosts, CX#cx{realms = NewRealms}};
|
||||
error ->
|
||||
{error, bad_realm}
|
||||
end.
|
||||
|
||||
|
||||
cx_next_hosts3(N, Hosts, Meta) when N < 1 ->
|
||||
{ok, Hosts, Meta};
|
||||
cx_next_hosts3(N, Hosts, Meta) ->
|
||||
case cx_next_host3(Meta) of
|
||||
{ok, Host, NewMeta} -> cx_next_hosts3(N - 1, [Host | Hosts], NewMeta);
|
||||
{prime, Prime, NewMeta} -> {ok, [Prime | Hosts], NewMeta}
|
||||
end.
|
||||
|
||||
|
||||
-spec cx_add_attempt(New, CX) -> NewCX
|
||||
when New :: {pid(), reference(), zx:host()},
|
||||
CX :: conn_index(),
|
||||
NewCX :: conn_index().
|
||||
|
||||
cx_add_attempt(New, CX = #cx{attempts = Attempts}) ->
|
||||
CX#cx{attempts = [New | Attempts]}.
|
||||
|
||||
|
||||
-spec cx_connected(Available, Conn, CX) -> Result
|
||||
when Available :: [{zx:realm(), zx:serial()}],
|
||||
Conn :: pid(),
|
||||
CX :: conn_index(),
|
||||
Result :: {Assignment, NewCX},
|
||||
Assignment :: assigned | unassigned,
|
||||
NewCX :: conn_index().
|
||||
%% @private
|
||||
%% An abstract data handler which is called whenever a new connection is successfully
|
||||
%% established by a zx_conn. Any unconnected realms with a valid serial will be
|
||||
%% assigned to the new connection; if none are needed then the connection is closed.
|
||||
%% The successful host is placed back in the hosts queue for each available realm.
|
||||
%% The return value is a tuple that indicates whether the new connection was assigned
|
||||
%% or not and the updated CX data value.
|
||||
|
||||
cx_connected(Available, Conn, CX) ->
|
||||
cx_connected(unassigned, Available, Conn, CX).
|
||||
|
||||
|
||||
cx_connected(A, [{Realm, Serial} | Rest], Conn, CX = #cx{realms = Realms}) ->
|
||||
case maps:find(Realm, Realms) of
|
||||
{ok, Meta = #rmeta{serial = S}} when S < Serial ->
|
||||
NewMeta = Meta#rmeta{serial = Serial},
|
||||
{NewA, NewCX} = cx_connected(A, Realm, Conn, NewMeta, CX),
|
||||
cx_connected(NewA, Rest, Conn, NewCX);
|
||||
{ok, Meta = #rmeta{serial = S}} when S == Serial ->
|
||||
{NewA, NewCX} = cx_connected(A, Realm, Conn, Meta, CX),
|
||||
cx_connected(NewA, Rest, Conn, NewCX);
|
||||
{ok, #rmeta{serial = S}} when S > Serial ->
|
||||
cx_connected(A, Rest, Conn, CX);
|
||||
error ->
|
||||
cx_connected(A, Rest, Conn, CX)
|
||||
end;
|
||||
cx_connected(A, [], Conn, CX = #cx{attempts = Attempts, conns = Conns}) ->
|
||||
{value, ConnRec, NewAttempts} = lists:keytake(Conn, 1, Attempts),
|
||||
NewConns = [ConnRec | Conns],
|
||||
NewCX = CX#cx{attempts = NewAttempts, conns = NewConns},
|
||||
{A, NewCX}.
|
||||
|
||||
|
||||
cx_connected(A,
|
||||
Realm,
|
||||
Conn,
|
||||
Meta = #rmeta{prime = Prime, mirrors = Mirrors},
|
||||
CX = #cx{realms = Realms, assigned = Assigned, attempts = Attempts}) ->
|
||||
{NewMirrors, Node} =
|
||||
case lists:keyfind(Conn, 1, Attempts) of
|
||||
{_, _, Prime} -> {Mirrors, Prime};
|
||||
{_, _, Host} -> {enqueue_unique(Host, Mirrors), Host}
|
||||
end,
|
||||
{NewA, NewAssigned} =
|
||||
case lists:keymember(Realm, 1, Assigned) of
|
||||
true -> {A, Assigned};
|
||||
false -> {assigned, [{Realm, Pid} | Assigned]}
|
||||
end,
|
||||
NewMeta = Meta#rmeta{mirrors = NewMirrors},
|
||||
NewRealms = maps:put(Realm, NewMeta, Realms),
|
||||
NewCX = CX#cx{realms = NewRealms, assigned = NewAssigned},
|
||||
{NewA, NewCX}.
|
||||
|
||||
|
||||
-spec enqueue_unique(term(), queue:queue()) -> queue:queue().
|
||||
%% @private
|
||||
%% Simple function to ensure that only unique elements are added to a queue. Obviously
|
||||
%% this operation is extremely general and O(n) in complexity due to the use of
|
||||
%% queue:member/2.
|
||||
|
||||
enqueue_unique(Element, Queue) ->
|
||||
case queue:member(Element, Queue) of
|
||||
true -> Queue;
|
||||
false -> queue:in(Element, Queue)
|
||||
end.
|
||||
|
||||
|
||||
-spec cx_disconnected(Conn, CX) -> Result
|
||||
when Conn :: pid(),
|
||||
CX :: conn_index(),
|
||||
Result :: {ok, Mon, UnassignedRealms, NewCX}
|
||||
| {error, unknown},
|
||||
Mon :: reference(),
|
||||
UnassignedRealms :: [zx:realm()],
|
||||
NewCX :: conn_index().
|
||||
%% @private
|
||||
%% An abstract data handler which is called whenever a connection terminates.
|
||||
%% This function removes all data related to the disconnected pid and its assigned
|
||||
%% realms, and returns the monitor reference of the pid, a list of realms that are now
|
||||
%% unassigned, and an updated connection index.
|
||||
|
||||
cx_disconnected(Conn, CX = #cx{assigned = Assigned, conns = Conns}) ->
|
||||
case lists:keytake(Con, Conns) of
|
||||
{value, {Pid, Mon, Conn}, NewConns} ->
|
||||
{UnassignedRealms, NewAssigned} = cx_scrub_assigned(Pid, Assigned),
|
||||
NewCX = CX#cx{assigned = NewAssigned, conns = NewConns},
|
||||
{ok, Mon, UnassignedRealms, NewCX};
|
||||
false ->
|
||||
{error, unknown}
|
||||
end.
|
||||
|
||||
|
||||
-spec cx_scrub_assigned(Pid, Assigned) -> {UnassignedRealms, NewAssigned}
|
||||
when Pid :: pid(),
|
||||
Assigned :: [{zx:realm(), pid()}],
|
||||
UnassignedRealms :: [zx:realm()],
|
||||
NewAssigned :: [{zx:realm(), pid()}].
|
||||
%% @private
|
||||
%% This could have been performed as a set of two list operations (a partition and a
|
||||
%% map), but to make the procedure perfectly clear it is written out explicitly.
|
||||
|
||||
cx_scrub_assigned(Pid, Assigned) ->
|
||||
cx_scrub_assigned(Pid, Assigned, [], []).
|
||||
|
||||
|
||||
cx_scrub_assigned(Pid, [{Realm, Pid} | Rest], Unassigned, Assigned) ->
|
||||
cx_scrub_assigned(Pid, Rest, [Realm | Unassigned], Assigned);
|
||||
cx_scrub_assigned(Pid, [A | Rest], Unassigned, Assigned) ->
|
||||
cx_scrub_assigned(Pid, Rest, Unassigned, [A | Assigned]);
|
||||
cx_scrub_assigned(_, [], Unassigned, Assigned) ->
|
||||
{Unassigned, Assigned}.
|
||||
|
||||
|
||||
-spec cx_resolve(Realm, CX) -> Result
|
||||
when Realm :: zx:realm(),
|
||||
CX :: conn_index(),
|
||||
Result :: {ok, Conn :: pid()}
|
||||
| unassigned.
|
||||
%% @private
|
||||
%% Check the registry of assigned realms and return the pid of the appropriate
|
||||
%% connection, or an `unassigned' indication if the realm is not yet connected.
|
||||
|
||||
cx_resolve(Realm, #cx{assigned = Assigned}) ->
|
||||
case lists:keyfind(Realm, 1, Assigned) of
|
||||
{Realm, Conn} -> {ok, Conn};
|
||||
false -> unassigned
|
||||
end.
|
||||
|
||||
@ -1,8 +1,6 @@
|
||||
{realm,"otpr"}.
|
||||
{revision,0}.
|
||||
{prime,{"otpr.psychobitch.party",11311}}.
|
||||
{private,[]}.
|
||||
{mirrors,[]}.
|
||||
{sysops,[{{"otpr","zxq9"},["zxq9.1"],"zxq9@zxq9.com","Craig Everett"}]}.
|
||||
{realm_keys,[{{"otpr","otpr.1.realm"},
|
||||
realm,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user