greg/machi
1
0
Fork 0
Code Issues 11 Pull requests 1 Projects Releases Packages Wiki Activity Actions

Dialyzer tightening and subsequent cleanup

Browse source
This commit is contained in:
Scott Lystig Fritchie 2015-06-19 16:04:34 +09:00
parent 1372bd9594
commit 984b4f7a86
8 changed files with 91 additions and 217 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

5
Makefile
View file

@ -38,15 +38,16 @@ build_plt: deps compile
dialyzer --build_plt --output_plt $(PLT) --apps $(APPS) deps/*/ebin dialyzer --build_plt --output_plt $(PLT) --apps $(APPS) deps/*/ebin
DIALYZER_DEP_APPS = ebin/machi_pb.beam deps/protobuffs/ebin DIALYZER_DEP_APPS = ebin/machi_pb.beam deps/protobuffs/ebin
DIALYZER_FLAGS = -Wno_return -Wrace_conditions -Wunderspecs
dialyzer: deps compile dialyzer: deps compile
dialyzer -Wno_return --plt $(PLT) ebin $(DIALYZER_DEP_APPS) | \ dialyzer $(DIALYZER_FLAGS) --plt $(PLT) ebin $(DIALYZER_DEP_APPS) | \
egrep -v -f ./filter-dialyzer-dep-warnings egrep -v -f ./filter-dialyzer-dep-warnings
dialyzer-test: deps compile dialyzer-test: deps compile
echo Force rebar to recompile .eunit dir w/o running tests > /dev/null echo Force rebar to recompile .eunit dir w/o running tests > /dev/null
rebar skip_deps=true eunit suite=lamport_clock rebar skip_deps=true eunit suite=lamport_clock
dialyzer -Wno_return --plt $(PLT) .eunit $(DIALYZER_DEP_APPS) | \ dialyzer $(DIALYZER_FLAGS) --plt $(PLT) .eunit $(DIALYZER_DEP_APPS) | \
egrep -v -f ./filter-dialyzer-dep-warnings egrep -v -f ./filter-dialyzer-dep-warnings
clean_plt: clean_plt:

4
filter-dialyzer-dep-warnings
View file

@ -1,8 +1,12 @@
####################### patterns for general errors in dep modules: ####################### patterns for general errors in dep modules:
^protobuffs\.erl:
^protobuffs_[a-z_]*\.erl:
^leexinc\.hrl:[0-9][0-9]*: ^leexinc\.hrl:[0-9][0-9]*:
^machi_chain_manager1.erl:[0-9][0-9]*: Guard test RetrospectiveP::'false' =:= 'true' can never succeed
^machi_pb\.erl:[0-9][0-9]*: ^machi_pb\.erl:[0-9][0-9]*:
^pokemon_pb\.erl:[0-9][0-9]*: ^pokemon_pb\.erl:[0-9][0-9]*:
####################### patterns for unknown functions: ####################### patterns for unknown functions:
^ basho_bench_config:get/2 ^ basho_bench_config:get/2
^ erl_prettypr:format/1 ^ erl_prettypr:format/1
^ erl_syntax:form_list/1 ^ erl_syntax:form_list/1
^ machi_partition_simulator:get/1

17
src/machi_admin_util.erl
View file

@ -22,10 +22,6 @@
-module(machi_admin_util). -module(machi_admin_util).
%% TODO Move these types to a common header file? (also machi_flu1_client.erl?)
-type inet_host() :: inet:ip_address() | inet:hostname().
-type inet_port() :: inet:port_number().
-export([ -export([
verify_file_checksums_local/3, verify_file_checksums_local/4, verify_file_checksums_local/3, verify_file_checksums_local/4,
verify_file_checksums_remote/3, verify_file_checksums_remote/4 verify_file_checksums_remote/3, verify_file_checksums_remote/4
@ -35,15 +31,16 @@
-include("machi.hrl"). -include("machi.hrl").
-include("machi_projection.hrl"). -include("machi_projection.hrl").
-define(T, machi_dt).
-define(FLU_C, machi_flu1_client). -define(FLU_C, machi_flu1_client).
-spec verify_file_checksums_local(port(), machi_flu1_client:epoch_id(), binary()|list()) -> -spec verify_file_checksums_local(port(), ?T:epoch_id(), binary()|list()) ->
{ok, [tuple()]} | {error, term()}. {ok, [tuple()]} | {error, term()}.
verify_file_checksums_local(Sock1, EpochID, Path) when is_port(Sock1) -> verify_file_checksums_local(Sock1, EpochID, Path) when is_port(Sock1) ->
verify_file_checksums_local2(Sock1, EpochID, Path). verify_file_checksums_local2(Sock1, EpochID, Path).
-spec verify_file_checksums_local(inet_host(), inet_port(), -spec verify_file_checksums_local(?T:inet_host(), ?T:inet_port(),
machi_flu1_client:epoch_id(), binary()|list()) -> ?T:epoch_id(), binary()|list()) ->
{ok, [tuple()]} | {error, term()}. {ok, [tuple()]} | {error, term()}.
verify_file_checksums_local(Host, TcpPort, EpochID, Path) -> verify_file_checksums_local(Host, TcpPort, EpochID, Path) ->
Sock1 = ?FLU_C:connect(#p_srvr{address=Host, port=TcpPort}), Sock1 = ?FLU_C:connect(#p_srvr{address=Host, port=TcpPort}),
@ -53,13 +50,13 @@ verify_file_checksums_local(Host, TcpPort, EpochID, Path) ->
catch ?FLU_C:disconnect(Sock1) catch ?FLU_C:disconnect(Sock1)
end. end.
-spec verify_file_checksums_remote(port(), machi_flu1_client:epoch_id(), binary()|list()) -> -spec verify_file_checksums_remote(port(), ?T:epoch_id(), binary()|list()) ->
{ok, [tuple()]} | {error, term()}. {ok, [tuple()]} | {error, term()}.
verify_file_checksums_remote(Sock1, EpochID, File) when is_port(Sock1) -> verify_file_checksums_remote(Sock1, EpochID, File) when is_port(Sock1) ->
verify_file_checksums_remote2(Sock1, EpochID, File). verify_file_checksums_remote2(Sock1, EpochID, File).
-spec verify_file_checksums_remote(inet_host(), inet_port(), -spec verify_file_checksums_remote(?T:inet_host(), ?T:inet_port(),
machi_flu1_client:epoch_id(), binary()|list()) -> ?T:epoch_id(), binary()|list()) ->
{ok, [tuple()]} | {error, term()}. {ok, [tuple()]} | {error, term()}.
verify_file_checksums_remote(Host, TcpPort, EpochID, File) -> verify_file_checksums_remote(Host, TcpPort, EpochID, File) ->
Sock1 = ?FLU_C:connect(#p_srvr{address=Host, port=TcpPort}), Sock1 = ?FLU_C:connect(#p_srvr{address=Host, port=TcpPort}),

6
src/machi_basho_bench_driver.erl
View file

@ -67,9 +67,9 @@
-define(ETS_TAB, machi_keys). -define(ETS_TAB, machi_keys).
-define(THE_TIMEOUT, 60*1000). -define(THE_TIMEOUT, 60*1000).
-define(INFO(Str, Args), lager:info(Str, Args)). -define(INFO(Str, Args), (_ = lager:info(Str, Args))).
-define(WARN(Str, Args), lager:warning(Str, Args)). -define(WARN(Str, Args), (_ = lager:warning(Str, Args))).
-define(ERROR(Str, Args), lager:error(Str, Args)). -define(ERROR(Str, Args), (_ = lager:error(Str, Args))).
new(Id) -> new(Id) ->
Ps = find_server_info(Id), Ps = find_server_info(Id),

4
src/machi_flu1.erl
View file

@ -83,6 +83,8 @@
update_wedge_state/3]). update_wedge_state/3]).
-export([make_listener_regname/1, make_projection_server_regname/1]). -export([make_listener_regname/1, make_projection_server_regname/1]).
-define(T, machi_dt).
-record(state, { -record(state, {
flu_name :: atom(), flu_name :: atom(),
proj_store :: pid(), proj_store :: pid(),
@ -91,7 +93,7 @@
data_dir :: string(), data_dir :: string(),
wedged = true :: boolean(), wedged = true :: boolean(),
etstab :: ets:tid(), etstab :: ets:tid(),
epoch_id :: 'undefined' | pv1_epoch(), epoch_id :: 'undefined' | ?T:epoch_id(),
dbg_props = [] :: list(), % proplist dbg_props = [] :: list(), % proplist
props = [] :: list() % proplist props = [] :: list() % proplist
}). }).

161
src/machi_flu1_client.erl
View file

@ -85,47 +85,24 @@
trunc_hack/3, trunc_hack/4 trunc_hack/3, trunc_hack/4
]). ]).
%% TODO: Hrm, this kind of API use ... is it a bad idea? We really want to -define(T, machi_dt).
%% encourage client-side checksums; thus it ought to be dead easy.
-type chunk() :: chunk_bin() | {chunk_csum(), chunk_bin()}.
-type chunk_bin() :: binary() | iolist(). % client can use either
-type chunk_csum() :: binary(). % 1 byte tag, N-1 bytes checksum
-type chunk_summary() :: {file_offset(), chunk_size(), binary()}.
-type chunk_s() :: binary(). % server always uses binary()
-type chunk_pos() :: {file_offset(), chunk_size(), file_name_s()}.
-type chunk_size() :: non_neg_integer().
-type error_general() :: 'bad_arg' | 'wedged' | 'bad_checksum'.
-type epoch_csum() :: binary().
-type epoch_num() :: -1 | non_neg_integer().
-type epoch_id() :: {epoch_num(), epoch_csum()}.
-type file_info() :: {file_size(), file_name_s()}.
-type file_name() :: binary() | list().
-type file_name_s() :: binary(). % server reply
-type file_offset() :: non_neg_integer().
-type file_size() :: non_neg_integer().
-type file_prefix() :: binary() | list().
-type inet_host() :: inet:ip_address() | inet:hostname().
-type inet_port() :: inet:port_number().
-type port_wrap() :: {w,atom(),term()}.
-type projection() :: #projection_v1{}.
-type projection_type() :: 'public' | 'private'.
-export_type([epoch_id/0]). -type port_wrap() :: {w,atom(),term()}.
%% @doc Append a chunk (binary- or iolist-style) of data to a file %% @doc Append a chunk (binary- or iolist-style) of data to a file
%% with `Prefix'. %% with `Prefix'.
-spec append_chunk(port_wrap(), epoch_id(), file_prefix(), chunk()) -> -spec append_chunk(port_wrap(), ?T:epoch_id(), ?T:file_prefix(), ?T:chunk()) ->
{ok, chunk_pos()} | {error, error_general()} | {error, term()}. {ok, ?T:chunk_pos()} | {error, ?T:error_general()} | {error, term()}.
append_chunk(Sock, EpochID, Prefix, Chunk) -> append_chunk(Sock, EpochID, Prefix, Chunk) ->
append_chunk2(Sock, EpochID, Prefix, Chunk, 0). append_chunk2(Sock, EpochID, Prefix, Chunk, 0).
%% @doc Append a chunk (binary- or iolist-style) of data to a file %% @doc Append a chunk (binary- or iolist-style) of data to a file
%% with `Prefix'. %% with `Prefix'.
-spec append_chunk(inet_host(), inet_port(), -spec append_chunk(?T:inet_host(), ?T:inet_port(),
epoch_id(), file_prefix(), chunk()) -> ?T:epoch_id(), ?T:file_prefix(), ?T:chunk()) ->
{ok, chunk_pos()} | {error, error_general()} | {error, term()}. {ok, ?T:chunk_pos()} | {error, ?T:error_general()} | {error, term()}.
append_chunk(Host, TcpPort, EpochID, Prefix, Chunk) -> append_chunk(Host, TcpPort, EpochID, Prefix, Chunk) ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
try try
@ -142,8 +119,8 @@ append_chunk(Host, TcpPort, EpochID, Prefix, Chunk) ->
%% be reserved by the file sequencer for later write(s) by the %% be reserved by the file sequencer for later write(s) by the
%% `write_chunk()' API. %% `write_chunk()' API.
-spec append_chunk_extra(port_wrap(), epoch_id(), file_prefix(), chunk(), chunk_size()) -> -spec append_chunk_extra(port_wrap(), ?T:epoch_id(), ?T:file_prefix(), ?T:chunk(), ?T:chunk_size()) ->
{ok, chunk_pos()} | {error, error_general()} | {error, term()}. {ok, ?T:chunk_pos()} | {error, ?T:error_general()} | {error, term()}.
append_chunk_extra(Sock, EpochID, Prefix, Chunk, ChunkExtra) append_chunk_extra(Sock, EpochID, Prefix, Chunk, ChunkExtra)
when is_integer(ChunkExtra), ChunkExtra >= 0 -> when is_integer(ChunkExtra), ChunkExtra >= 0 ->
append_chunk2(Sock, EpochID, Prefix, Chunk, ChunkExtra). append_chunk2(Sock, EpochID, Prefix, Chunk, ChunkExtra).
@ -156,9 +133,9 @@ append_chunk_extra(Sock, EpochID, Prefix, Chunk, ChunkExtra)
%% be reserved by the file sequencer for later write(s) by the %% be reserved by the file sequencer for later write(s) by the
%% `write_chunk()' API. %% `write_chunk()' API.
-spec append_chunk_extra(inet_host(), inet_port(), -spec append_chunk_extra(?T:inet_host(), ?T:inet_port(),
epoch_id(), file_prefix(), chunk(), chunk_size()) -> ?T:epoch_id(), ?T:file_prefix(), ?T:chunk(), ?T:chunk_size()) ->
{ok, chunk_pos()} | {error, error_general()} | {error, term()}. {ok, ?T:chunk_pos()} | {error, ?T:error_general()} | {error, term()}.
append_chunk_extra(Host, TcpPort, EpochID, Prefix, Chunk, ChunkExtra) append_chunk_extra(Host, TcpPort, EpochID, Prefix, Chunk, ChunkExtra)
when is_integer(ChunkExtra), ChunkExtra >= 0 -> when is_integer(ChunkExtra), ChunkExtra >= 0 ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
@ -170,9 +147,9 @@ append_chunk_extra(Host, TcpPort, EpochID, Prefix, Chunk, ChunkExtra)
%% @doc Read a chunk of data of size `Size' from `File' at `Offset'. %% @doc Read a chunk of data of size `Size' from `File' at `Offset'.
-spec read_chunk(port_wrap(), epoch_id(), file_name(), file_offset(), chunk_size()) -> -spec read_chunk(port_wrap(), ?T:epoch_id(), ?T:file_name(), ?T:file_offset(), ?T:chunk_size()) ->
{ok, chunk_s()} | {ok, ?T:chunk_s()} |
{error, error_general() | 'not_written' | 'partial_read'} | {error, ?T:error_general() | 'not_written' | 'partial_read'} |
{error, term()}. {error, term()}.
read_chunk(Sock, EpochID, File, Offset, Size) read_chunk(Sock, EpochID, File, Offset, Size)
when Offset >= ?MINIMUM_OFFSET, Size >= 0 -> when Offset >= ?MINIMUM_OFFSET, Size >= 0 ->
@ -180,10 +157,10 @@ read_chunk(Sock, EpochID, File, Offset, Size)
%% @doc Read a chunk of data of size `Size' from `File' at `Offset'. %% @doc Read a chunk of data of size `Size' from `File' at `Offset'.
-spec read_chunk(inet_host(), inet_port(), epoch_id(), -spec read_chunk(?T:inet_host(), ?T:inet_port(), ?T:epoch_id(),
file_name(), file_offset(), chunk_size()) -> ?T:file_name(), ?T:file_offset(), ?T:chunk_size()) ->
{ok, chunk_s()} | {ok, ?T:chunk_s()} |
{error, error_general() | 'not_written' | 'partial_read'} | {error, ?T:error_general() | 'not_written' | 'partial_read'} |
{error, term()}. {error, term()}.
read_chunk(Host, TcpPort, EpochID, File, Offset, Size) read_chunk(Host, TcpPort, EpochID, File, Offset, Size)
when Offset >= ?MINIMUM_OFFSET, Size >= 0 -> when Offset >= ?MINIMUM_OFFSET, Size >= 0 ->
@ -196,18 +173,18 @@ read_chunk(Host, TcpPort, EpochID, File, Offset, Size)
%% @doc Fetch the list of chunk checksums for `File'. %% @doc Fetch the list of chunk checksums for `File'.
-spec checksum_list(port_wrap(), epoch_id(), file_name()) -> -spec checksum_list(port_wrap(), ?T:epoch_id(), ?T:file_name()) ->
{ok, [chunk_summary()]} | {ok, [?T:chunk_summary()]} |
{error, error_general() | 'no_such_file' | 'partial_read'} | {error, ?T:error_general() | 'no_such_file' | 'partial_read'} |
{error, term()}. {error, term()}.
checksum_list(Sock, EpochID, File) -> checksum_list(Sock, EpochID, File) ->
checksum_list2(Sock, EpochID, File). checksum_list2(Sock, EpochID, File).
%% @doc Fetch the list of chunk checksums for `File'. %% @doc Fetch the list of chunk checksums for `File'.
-spec checksum_list(inet_host(), inet_port(), epoch_id(), file_name()) -> -spec checksum_list(?T:inet_host(), ?T:inet_port(), ?T:epoch_id(), ?T:file_name()) ->
{ok, [chunk_summary()]} | {ok, [?T:chunk_summary()]} |
{error, error_general() | 'no_such_file'} | {error, term()}. {error, ?T:error_general() | 'no_such_file'} | {error, term()}.
checksum_list(Host, TcpPort, EpochID, File) when is_integer(TcpPort) -> checksum_list(Host, TcpPort, EpochID, File) when is_integer(TcpPort) ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
try try
@ -218,15 +195,15 @@ checksum_list(Host, TcpPort, EpochID, File) when is_integer(TcpPort) ->
%% @doc Fetch the list of all files on the remote FLU. %% @doc Fetch the list of all files on the remote FLU.
-spec list_files(port_wrap(), epoch_id()) -> -spec list_files(port_wrap(), ?T:epoch_id()) ->
{ok, [file_info()]} | {error, term()}. {ok, [?T:file_info()]} | {error, term()}.
list_files(Sock, EpochID) -> list_files(Sock, EpochID) ->
list2(Sock, EpochID). list2(Sock, EpochID).
%% @doc Fetch the list of all files on the remote FLU. %% @doc Fetch the list of all files on the remote FLU.
-spec list_files(inet_host(), inet_port(), epoch_id()) -> -spec list_files(?T:inet_host(), ?T:inet_port(), ?T:epoch_id()) ->
{ok, [file_info()]} | {error, term()}. {ok, [?T:file_info()]} | {error, term()}.
list_files(Host, TcpPort, EpochID) when is_integer(TcpPort) -> list_files(Host, TcpPort, EpochID) when is_integer(TcpPort) ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
try try
@ -238,15 +215,15 @@ list_files(Host, TcpPort, EpochID) when is_integer(TcpPort) ->
%% @doc Fetch the wedge status from the remote FLU. %% @doc Fetch the wedge status from the remote FLU.
-spec wedge_status(port_wrap()) -> -spec wedge_status(port_wrap()) ->
{ok, {boolean(), pv1_epoch()}} | {error, term()}. {ok, {boolean(), ?T:epoch_id()}} | {error, term()}.
wedge_status(Sock) -> wedge_status(Sock) ->
wedge_status2(Sock). wedge_status2(Sock).
%% @doc Fetch the wedge status from the remote FLU. %% @doc Fetch the wedge status from the remote FLU.
-spec wedge_status(inet_host(), inet_port()) -> -spec wedge_status(?T:inet_host(), ?T:inet_port()) ->
{ok, {boolean(), pv1_epoch()}} | {error, term()}. {ok, {boolean(), ?T:epoch_id()}} | {error, term()}.
wedge_status(Host, TcpPort) when is_integer(TcpPort) -> wedge_status(Host, TcpPort) when is_integer(TcpPort) ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
try try
@ -257,16 +234,16 @@ wedge_status(Host, TcpPort) when is_integer(TcpPort) ->
%% @doc Get the latest epoch number + checksum from the FLU's projection store. %% @doc Get the latest epoch number + checksum from the FLU's projection store.
-spec get_latest_epochid(port_wrap(), projection_type()) -> -spec get_latest_epochid(port_wrap(), ?T:projection_type()) ->
{ok, epoch_id()} | {error, term()}. {ok, ?T:epoch_id()} | {error, term()}.
get_latest_epochid(Sock, ProjType) get_latest_epochid(Sock, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
get_latest_epochid2(Sock, ProjType). get_latest_epochid2(Sock, ProjType).
%% @doc Get the latest epoch number + checksum from the FLU's projection store. %% @doc Get the latest epoch number + checksum from the FLU's projection store.
-spec get_latest_epochid(inet_host(), inet_port(), projection_type()) -> -spec get_latest_epochid(?T:inet_host(), ?T:inet_port(), ?T:projection_type()) ->
{ok, epoch_id()} | {error, term()}. {ok, ?T:epoch_id()} | {error, term()}.
get_latest_epochid(Host, TcpPort, ProjType) get_latest_epochid(Host, TcpPort, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
@ -278,17 +255,17 @@ get_latest_epochid(Host, TcpPort, ProjType)
%% @doc Get the latest projection from the FLU's projection store for `ProjType' %% @doc Get the latest projection from the FLU's projection store for `ProjType'
-spec read_latest_projection(port_wrap(), projection_type()) -> -spec read_latest_projection(port_wrap(), ?T:projection_type()) ->
{ok, projection()} | {error, not_written} | {error, term()}. {ok, ?T:projection()} | {error, not_written} | {error, term()}.
read_latest_projection(Sock, ProjType) read_latest_projection(Sock, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
read_latest_projection2(Sock, ProjType). read_latest_projection2(Sock, ProjType).
%% @doc Get the latest projection from the FLU's projection store for `ProjType' %% @doc Get the latest projection from the FLU's projection store for `ProjType'
-spec read_latest_projection(inet_host(), inet_port(), -spec read_latest_projection(?T:inet_host(), ?T:inet_port(),
projection_type()) -> ?T:projection_type()) ->
{ok, projection()} | {error, not_written} | {error, term()}. {ok, ?T:projection()} | {error, not_written} | {error, term()}.
read_latest_projection(Host, TcpPort, ProjType) read_latest_projection(Host, TcpPort, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
@ -300,17 +277,17 @@ read_latest_projection(Host, TcpPort, ProjType)
%% @doc Read a projection `Proj' of type `ProjType'. %% @doc Read a projection `Proj' of type `ProjType'.
-spec read_projection(port_wrap(), projection_type(), epoch_num()) -> -spec read_projection(port_wrap(), ?T:projection_type(), ?T:epoch_num()) ->
{ok, projection()} | {error, not_written} | {error, term()}. {ok, ?T:projection()} | {error, not_written} | {error, term()}.
read_projection(Sock, ProjType, Epoch) read_projection(Sock, ProjType, Epoch)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
read_projection2(Sock, ProjType, Epoch). read_projection2(Sock, ProjType, Epoch).
%% @doc Read a projection `Proj' of type `ProjType'. %% @doc Read a projection `Proj' of type `ProjType'.
-spec read_projection(inet_host(), inet_port(), -spec read_projection(?T:inet_host(), ?T:inet_port(),
projection_type(), epoch_num()) -> ?T:projection_type(), ?T:epoch_num()) ->
{ok, projection()} | {error, not_written} | {error, term()}. {ok, ?T:projection()} | {error, not_written} | {error, term()}.
read_projection(Host, TcpPort, ProjType, Epoch) read_projection(Host, TcpPort, ProjType, Epoch)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
@ -322,7 +299,7 @@ read_projection(Host, TcpPort, ProjType, Epoch)
%% @doc Write a projection `Proj' of type `ProjType'. %% @doc Write a projection `Proj' of type `ProjType'.
-spec write_projection(port_wrap(), projection_type(), projection()) -> -spec write_projection(port_wrap(), ?T:projection_type(), ?T:projection()) ->
'ok' | {error, 'written'} | {error, term()}. 'ok' | {error, 'written'} | {error, term()}.
write_projection(Sock, ProjType, Proj) write_projection(Sock, ProjType, Proj)
when ProjType == 'public' orelse ProjType == 'private', when ProjType == 'public' orelse ProjType == 'private',
@ -331,8 +308,8 @@ write_projection(Sock, ProjType, Proj)
%% @doc Write a projection `Proj' of type `ProjType'. %% @doc Write a projection `Proj' of type `ProjType'.
-spec write_projection(inet_host(), inet_port(), -spec write_projection(?T:inet_host(), ?T:inet_port(),
projection_type(), projection()) -> ?T:projection_type(), ?T:projection()) ->
'ok' | {error, 'written'} | {error, term()}. 'ok' | {error, 'written'} | {error, term()}.
write_projection(Host, TcpPort, ProjType, Proj) write_projection(Host, TcpPort, ProjType, Proj)
when ProjType == 'public' orelse ProjType == 'private', when ProjType == 'public' orelse ProjType == 'private',
@ -346,17 +323,16 @@ write_projection(Host, TcpPort, ProjType, Proj)
%% @doc Get all projections from the FLU's projection store. %% @doc Get all projections from the FLU's projection store.
-spec get_all_projections(port_wrap(), projection_type()) -> -spec get_all_projections(port_wrap(), ?T:projection_type()) ->
{ok, [projection()]} | {error, term()}. {ok, [?T:projection()]} | {error, term()}.
get_all_projections(Sock, ProjType) get_all_projections(Sock, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
get_all_projections2(Sock, ProjType). get_all_projections2(Sock, ProjType).
%% @doc Get all projections from the FLU's projection store. %% @doc Get all projections from the FLU's projection store.
-spec get_all_projections(inet_host(), inet_port(), -spec get_all_projections(?T:inet_host(), ?T:inet_port(), ?T:projection_type()) ->
projection_type()) -> {ok, [?T:projection()]} | {error, term()}.
{ok, [projection()]} | {error, term()}.
get_all_projections(Host, TcpPort, ProjType) get_all_projections(Host, TcpPort, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
@ -368,7 +344,7 @@ get_all_projections(Host, TcpPort, ProjType)
%% @doc Get all epoch numbers from the FLU's projection store. %% @doc Get all epoch numbers from the FLU's projection store.
-spec list_all_projections(port_wrap(), projection_type()) -> -spec list_all_projections(port_wrap(), ?T:projection_type()) ->
{ok, [non_neg_integer()]} | {error, term()}. {ok, [non_neg_integer()]} | {error, term()}.
list_all_projections(Sock, ProjType) list_all_projections(Sock, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
@ -376,8 +352,7 @@ list_all_projections(Sock, ProjType)
%% @doc Get all epoch numbers from the FLU's projection store. %% @doc Get all epoch numbers from the FLU's projection store.
-spec list_all_projections(inet_host(), inet_port(), -spec list_all_projections(?T:inet_host(), ?T:inet_port(), ?T:projection_type()) ->
projection_type()) ->
{ok, [non_neg_integer()]} | {error, term()}. {ok, [non_neg_integer()]} | {error, term()}.
list_all_projections(Host, TcpPort, ProjType) list_all_projections(Host, TcpPort, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
@ -419,8 +394,8 @@ disconnect(_) ->
%% @doc Restricted API: Write a chunk of already-sequenced data to %% @doc Restricted API: Write a chunk of already-sequenced data to
%% `File' at `Offset'. %% `File' at `Offset'.
-spec write_chunk(port_wrap(), epoch_id(), file_name(), file_offset(), chunk()) -> -spec write_chunk(port_wrap(), ?T:epoch_id(), ?T:file_name(), ?T:file_offset(), ?T:chunk()) ->
ok | {error, error_general()} | {error, term()}. ok | {error, ?T:error_general()} | {error, term()}.
write_chunk(Sock, EpochID, File, Offset, Chunk) write_chunk(Sock, EpochID, File, Offset, Chunk)
when Offset >= ?MINIMUM_OFFSET -> when Offset >= ?MINIMUM_OFFSET ->
write_chunk2(Sock, EpochID, File, Offset, Chunk). write_chunk2(Sock, EpochID, File, Offset, Chunk).
@ -428,9 +403,9 @@ write_chunk(Sock, EpochID, File, Offset, Chunk)
%% @doc Restricted API: Write a chunk of already-sequenced data to %% @doc Restricted API: Write a chunk of already-sequenced data to
%% `File' at `Offset'. %% `File' at `Offset'.
-spec write_chunk(inet_host(), inet_port(), -spec write_chunk(?T:inet_host(), ?T:inet_port(),
epoch_id(), file_name(), file_offset(), chunk()) -> ?T:epoch_id(), ?T:file_name(), ?T:file_offset(), ?T:chunk()) ->
ok | {error, error_general()} | {error, term()}. ok | {error, ?T:error_general()} | {error, term()}.
write_chunk(Host, TcpPort, EpochID, File, Offset, Chunk) write_chunk(Host, TcpPort, EpochID, File, Offset, Chunk)
when Offset >= ?MINIMUM_OFFSET -> when Offset >= ?MINIMUM_OFFSET ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
@ -443,16 +418,16 @@ write_chunk(Host, TcpPort, EpochID, File, Offset, Chunk)
%% @doc Restricted API: Delete a file after it has been successfully %% @doc Restricted API: Delete a file after it has been successfully
%% migrated. %% migrated.
-spec delete_migration(port_wrap(), epoch_id(), file_name()) -> -spec delete_migration(port_wrap(), ?T:epoch_id(), ?T:file_name()) ->
ok | {error, error_general() | 'no_such_file'} | {error, term()}. ok | {error, ?T:error_general() | 'no_such_file'} | {error, term()}.
delete_migration(Sock, EpochID, File) -> delete_migration(Sock, EpochID, File) ->
delete_migration2(Sock, EpochID, File). delete_migration2(Sock, EpochID, File).
%% @doc Restricted API: Delete a file after it has been successfully %% @doc Restricted API: Delete a file after it has been successfully
%% migrated. %% migrated.
-spec delete_migration(inet_host(), inet_port(), epoch_id(), file_name()) -> -spec delete_migration(?T:inet_host(), ?T:inet_port(), ?T:epoch_id(), ?T:file_name()) ->
ok | {error, error_general() | 'no_such_file'} | {error, term()}. ok | {error, ?T:error_general() | 'no_such_file'} | {error, term()}.
delete_migration(Host, TcpPort, EpochID, File) when is_integer(TcpPort) -> delete_migration(Host, TcpPort, EpochID, File) when is_integer(TcpPort) ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
try try
@ -464,16 +439,16 @@ delete_migration(Host, TcpPort, EpochID, File) when is_integer(TcpPort) ->
%% @doc Restricted API: Truncate a file after it has been successfully %% @doc Restricted API: Truncate a file after it has been successfully
%% erasure coded. %% erasure coded.
-spec trunc_hack(port_wrap(), epoch_id(), file_name()) -> -spec trunc_hack(port_wrap(), ?T:epoch_id(), ?T:file_name()) ->
ok | {error, error_general() | 'no_such_file'} | {error, term()}. ok | {error, ?T:error_general() | 'no_such_file'} | {error, term()}.
trunc_hack(Sock, EpochID, File) -> trunc_hack(Sock, EpochID, File) ->
trunc_hack2(Sock, EpochID, File). trunc_hack2(Sock, EpochID, File).
%% @doc Restricted API: Truncate a file after it has been successfully %% @doc Restricted API: Truncate a file after it has been successfully
%% erasure coded. %% erasure coded.
-spec trunc_hack(inet_host(), inet_port(), epoch_id(), file_name()) -> -spec trunc_hack(?T:inet_host(), ?T:inet_port(), ?T:epoch_id(), ?T:file_name()) ->
ok | {error, error_general() | 'no_such_file'} | {error, term()}. ok | {error, ?T:error_general() | 'no_such_file'} | {error, term()}.
trunc_hack(Host, TcpPort, EpochID, File) when is_integer(TcpPort) -> trunc_hack(Host, TcpPort, EpochID, File) when is_integer(TcpPort) ->
Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}), Sock = connect(#p_srvr{proto_mod=?MODULE, address=Host, port=TcpPort}),
try try

6
src/machi_util.erl
View file

@ -48,7 +48,7 @@
%% @doc Create a registered name atom for FLU sequencer internal %% @doc Create a registered name atom for FLU sequencer internal
%% rendezvous/message passing use. %% rendezvous/message passing use.
-spec make_regname(binary()|list()) -> -spec make_regname(binary()|string()) ->
atom(). atom().
make_regname(Prefix) when is_binary(Prefix) -> make_regname(Prefix) when is_binary(Prefix) ->
erlang:binary_to_atom(Prefix, latin1); erlang:binary_to_atom(Prefix, latin1);
@ -231,13 +231,13 @@ make_tagged_csum(server_regen, SHA) ->
%% @doc Log a verbose message. %% @doc Log a verbose message.
-spec verb(string()) -> term(). -spec verb(string()) -> ok.
verb(Fmt) -> verb(Fmt) ->
verb(Fmt, []). verb(Fmt, []).
%% @doc Log a verbose message. %% @doc Log a verbose message.
-spec verb(string(), list()) -> term(). -spec verb(string(), list()) -> ok.
verb(Fmt, Args) -> verb(Fmt, Args) ->
case application:get_env(kernel, verbose) of case application:get_env(kernel, verbose) of
{ok, true} -> io:format(Fmt, Args); {ok, true} -> io:format(Fmt, Args);

105
src/machi_yessir_client.erl
View file

@ -64,45 +64,15 @@
chunk_size chunk_size
}). }).
-type chunk() :: chunk_bin() | {chunk_csum(), chunk_bin()}.
-type chunk_bin() :: binary() | iolist(). % client can use either
-type chunk_csum() :: binary(). % 1 byte tag, N-1 bytes checksum
%% -type chunk_summary() :: {file_offset(), chunk_size(), binary()}.
-type chunk_s() :: binary(). % server always uses binary()
-type chunk_pos() :: {file_offset(), chunk_size(), file_name_s()}.
-type chunk_size() :: non_neg_integer().
-type error_general() :: 'bad_arg' | 'wedged'.
-type epoch_csum() :: binary().
-type epoch_num() :: -1 | non_neg_integer().
-type epoch_id() :: {epoch_num(), epoch_csum()}.
-type file_info() :: {file_size(), file_name_s()}.
-type file_name() :: binary() | list().
-type file_name_s() :: binary(). % server reply
-type file_offset() :: non_neg_integer().
-type file_size() :: non_neg_integer().
-type file_prefix() :: binary() | list().
-type inet_host() :: inet:ip_address() | inet:hostname().
-type inet_port() :: inet:port_number().
-type port_wrap() :: #yessir{}. % yessir non-standard!
-type projection() :: #projection_v1{}.
-type projection_type() :: 'public' | 'private'.
-export_type([epoch_id/0]).
%% @doc Append a chunk (binary- or iolist-style) of data to a file %% @doc Append a chunk (binary- or iolist-style) of data to a file
%% with `Prefix'. %% with `Prefix'.
-spec append_chunk(port_wrap(), epoch_id(), file_prefix(), chunk()) ->
{ok, chunk_pos()} | {error, error_general()} | {error, term()}.
append_chunk(Sock, EpochID, Prefix, Chunk) -> append_chunk(Sock, EpochID, Prefix, Chunk) ->
append_chunk_extra(Sock, EpochID, Prefix, Chunk, 0). append_chunk_extra(Sock, EpochID, Prefix, Chunk, 0).
%% @doc Append a chunk (binary- or iolist-style) of data to a file %% @doc Append a chunk (binary- or iolist-style) of data to a file
%% with `Prefix'. %% with `Prefix'.
-spec append_chunk(inet_host(), inet_port(),
epoch_id(), file_prefix(), chunk()) ->
{ok, chunk_pos()} | {error, error_general()} | {error, term()}.
append_chunk(_Host, _TcpPort, EpochID, Prefix, Chunk) -> append_chunk(_Host, _TcpPort, EpochID, Prefix, Chunk) ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
try try
@ -119,8 +89,6 @@ append_chunk(_Host, _TcpPort, EpochID, Prefix, Chunk) ->
%% be reserved by the file sequencer for later write(s) by the %% be reserved by the file sequencer for later write(s) by the
%% `write_chunk()' API. %% `write_chunk()' API.
-spec append_chunk_extra(port_wrap(), epoch_id(), file_prefix(), chunk(), chunk_size()) ->
{ok, chunk_pos()}. %%%% | {error, error_general()} | {error, term()}.
append_chunk_extra(#yessir{name=Name,start_bin=StartBin}, append_chunk_extra(#yessir{name=Name,start_bin=StartBin},
_EpochID, Prefix, Chunk, ChunkExtra) _EpochID, Prefix, Chunk, ChunkExtra)
when is_integer(ChunkExtra), ChunkExtra >= 0 -> when is_integer(ChunkExtra), ChunkExtra >= 0 ->
@ -140,9 +108,6 @@ append_chunk_extra(#yessir{name=Name,start_bin=StartBin},
%% be reserved by the file sequencer for later write(s) by the %% be reserved by the file sequencer for later write(s) by the
%% `write_chunk()' API. %% `write_chunk()' API.
-spec append_chunk_extra(inet_host(), inet_port(),
epoch_id(), file_prefix(), chunk(), chunk_size()) ->
{ok, chunk_pos()}. %%%% | {error, error_general()} | {error, term()}.
append_chunk_extra(_Host, _TcpPort, EpochID, Prefix, Chunk, ChunkExtra) append_chunk_extra(_Host, _TcpPort, EpochID, Prefix, Chunk, ChunkExtra)
when is_integer(ChunkExtra), ChunkExtra >= 0 -> when is_integer(ChunkExtra), ChunkExtra >= 0 ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
@ -154,10 +119,6 @@ append_chunk_extra(_Host, _TcpPort, EpochID, Prefix, Chunk, ChunkExtra)
%% @doc Read a chunk of data of size `Size' from `File' at `Offset'. %% @doc Read a chunk of data of size `Size' from `File' at `Offset'.
-spec read_chunk(port_wrap(), epoch_id(), file_name(), file_offset(), chunk_size()) ->
{ok, chunk_s()} |
{error, error_general() | 'no_such_file' | 'partial_read'} |
{error, term()}.
read_chunk(#yessir{name=Name}, _EpochID, File, Offset, Size) read_chunk(#yessir{name=Name}, _EpochID, File, Offset, Size)
when Offset >= ?MINIMUM_OFFSET, Size >= 0 -> when Offset >= ?MINIMUM_OFFSET, Size >= 0 ->
case get({Name,offset,File}) of case get({Name,offset,File}) of
@ -201,11 +162,6 @@ make_csum(Name,Size) ->
%% @doc Read a chunk of data of size `Size' from `File' at `Offset'. %% @doc Read a chunk of data of size `Size' from `File' at `Offset'.
-spec read_chunk(inet_host(), inet_port(), epoch_id(),
file_name(), file_offset(), chunk_size()) ->
{ok, chunk_s()} |
{error, error_general() | 'no_such_file' | 'partial_read'} |
{error, term()}.
read_chunk(_Host, _TcpPort, EpochID, File, Offset, Size) read_chunk(_Host, _TcpPort, EpochID, File, Offset, Size)
when Offset >= ?MINIMUM_OFFSET, Size >= 0 -> when Offset >= ?MINIMUM_OFFSET, Size >= 0 ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
@ -217,10 +173,6 @@ read_chunk(_Host, _TcpPort, EpochID, File, Offset, Size)
%% @doc Fetch the list of chunk checksums for `File'. %% @doc Fetch the list of chunk checksums for `File'.
-spec checksum_list(port_wrap(), epoch_id(), file_name()) ->
{ok, [chunk_csum()]} |
{error, error_general() | 'no_such_file' | 'partial_read'} |
{error, term()}.
checksum_list(#yessir{name=Name,chunk_size=ChunkSize}, _EpochID, File) -> checksum_list(#yessir{name=Name,chunk_size=ChunkSize}, _EpochID, File) ->
case get({Name,offset,File}) of case get({Name,offset,File}) of
undefined -> undefined ->
@ -234,9 +186,6 @@ checksum_list(#yessir{name=Name,chunk_size=ChunkSize}, _EpochID, File) ->
%% @doc Fetch the list of chunk checksums for `File'. %% @doc Fetch the list of chunk checksums for `File'.
-spec checksum_list(inet_host(), inet_port(), epoch_id(), file_name()) ->
{ok, [chunk_csum()]} |
{error, error_general() | 'no_such_file'} | {error, term()}.
checksum_list(_Host, _TcpPort, EpochID, File) -> checksum_list(_Host, _TcpPort, EpochID, File) ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
try try
@ -247,16 +196,12 @@ checksum_list(_Host, _TcpPort, EpochID, File) ->
%% @doc Fetch the list of all files on the remote FLU. %% @doc Fetch the list of all files on the remote FLU.
-spec list_files(port_wrap(), epoch_id()) ->
{ok, [file_info()]} | {error, term()}.
list_files(#yessir{name=Name}, _EpochID) -> list_files(#yessir{name=Name}, _EpochID) ->
Files = [{Offset, File} || {{N,offset,File}, Offset} <- get(), N == Name], Files = [{Offset, File} || {{N,offset,File}, Offset} <- get(), N == Name],
{ok, Files}. {ok, Files}.
%% @doc Fetch the list of all files on the remote FLU. %% @doc Fetch the list of all files on the remote FLU.
-spec list_files(inet_host(), inet_port(), epoch_id()) ->
{ok, [file_info()]} | {error, term()}.
list_files(_Host, _TcpPort, EpochID) -> list_files(_Host, _TcpPort, EpochID) ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
try try
@ -267,16 +212,11 @@ list_files(_Host, _TcpPort, EpochID) ->
%% @doc Fetch the wedge status from the remote FLU. %% @doc Fetch the wedge status from the remote FLU.
-spec wedge_status(port_wrap()) ->
{ok, {boolean(), pv1_epoch()}} | {error, term()}.
wedge_status(_Sock) -> wedge_status(_Sock) ->
{ok, {false, ?DUMMY_PV1_EPOCH}}. {ok, {false, ?DUMMY_PV1_EPOCH}}.
%% @doc Fetch the wedge status from the remote FLU. %% @doc Fetch the wedge status from the remote FLU.
-spec wedge_status(inet_host(), inet_port()) ->
{ok, {boolean(), pv1_epoch()}} | {error, term()}.
wedge_status(_Host, _TcpPort) -> wedge_status(_Host, _TcpPort) ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
try try
@ -287,8 +227,6 @@ wedge_status(_Host, _TcpPort) ->
%% @doc Get the latest epoch number + checksum from the FLU's projection store. %% @doc Get the latest epoch number + checksum from the FLU's projection store.
-spec get_latest_epoch(port_wrap(), projection_type()) ->
{ok, epoch_id()} | {error, term()}.
get_latest_epoch(Sock, ProjType) get_latest_epoch(Sock, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
case read_latest_projection(Sock, ProjType) of case read_latest_projection(Sock, ProjType) of
@ -298,9 +236,6 @@ get_latest_epoch(Sock, ProjType)
%% @doc Get the latest epoch number + checksum from the FLU's projection store. %% @doc Get the latest epoch number + checksum from the FLU's projection store.
-spec get_latest_epoch(inet_host(), inet_port(),
projection_type()) ->
{ok, epoch_id()} | {error, term()}.
get_latest_epoch(_Host, _TcpPort, ProjType) get_latest_epoch(_Host, _TcpPort, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
@ -312,8 +247,6 @@ get_latest_epoch(_Host, _TcpPort, ProjType)
%% @doc Get the latest projection from the FLU's projection store for `ProjType' %% @doc Get the latest projection from the FLU's projection store for `ProjType'
-spec read_latest_projection(port_wrap(), projection_type()) ->
{ok, projection()} | {error, not_written} | {error, term()}.
read_latest_projection(#yessir{name=Name}, ProjType) read_latest_projection(#yessir{name=Name}, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Ps = [P || {{N,proj,PT,_Epoch}, P} <- get(), N == Name, PT == ProjType], Ps = [P || {{N,proj,PT,_Epoch}, P} <- get(), N == Name, PT == ProjType],
@ -330,9 +263,6 @@ read_latest_projection(#yessir{name=Name}, ProjType)
%% @doc Get the latest projection from the FLU's projection store for `ProjType' %% @doc Get the latest projection from the FLU's projection store for `ProjType'
-spec read_latest_projection(inet_host(), inet_port(),
projection_type()) ->
{ok, projection()} | {error, not_written} | {error, term()}.
read_latest_projection(_Host, _TcpPort, ProjType) read_latest_projection(_Host, _TcpPort, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
@ -344,8 +274,6 @@ read_latest_projection(_Host, _TcpPort, ProjType)
%% @doc Read a projection `Proj' of type `ProjType'. %% @doc Read a projection `Proj' of type `ProjType'.
-spec read_projection(port_wrap(), projection_type(), epoch_num()) ->
{ok, projection()} | {error, not_written} | {error, term()}.
read_projection(#yessir{name=Name}, ProjType, Epoch) read_projection(#yessir{name=Name}, ProjType, Epoch)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
case get({Name,proj,ProjType,Epoch}) of case get({Name,proj,ProjType,Epoch}) of
@ -357,9 +285,6 @@ read_projection(#yessir{name=Name}, ProjType, Epoch)
%% @doc Read a projection `Proj' of type `ProjType'. %% @doc Read a projection `Proj' of type `ProjType'.
-spec read_projection(inet_host(), inet_port(),
projection_type(), epoch_num()) ->
{ok, projection()} | {error, written} | {error, term()}.
read_projection(_Host, _TcpPort, ProjType, Epoch) read_projection(_Host, _TcpPort, ProjType, Epoch)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
@ -371,8 +296,6 @@ read_projection(_Host, _TcpPort, ProjType, Epoch)
%% @doc Write a projection `Proj' of type `ProjType'. %% @doc Write a projection `Proj' of type `ProjType'.
-spec write_projection(port_wrap(), projection_type(), projection()) ->
'ok' | {error, written} | {error, term()}.
write_projection(#yessir{name=Name}=Sock, ProjType, Proj) write_projection(#yessir{name=Name}=Sock, ProjType, Proj)
when ProjType == 'public' orelse ProjType == 'private', when ProjType == 'public' orelse ProjType == 'private',
is_record(Proj, projection_v1) -> is_record(Proj, projection_v1) ->
@ -387,9 +310,6 @@ write_projection(#yessir{name=Name}=Sock, ProjType, Proj)
%% @doc Write a projection `Proj' of type `ProjType'. %% @doc Write a projection `Proj' of type `ProjType'.
-spec write_projection(inet_host(), inet_port(),
projection_type(), projection()) ->
'ok' | {error, written} | {error, term()}.
write_projection(_Host, _TcpPort, ProjType, Proj) write_projection(_Host, _TcpPort, ProjType, Proj)
when ProjType == 'public' orelse ProjType == 'private', when ProjType == 'public' orelse ProjType == 'private',
is_record(Proj, projection_v1) -> is_record(Proj, projection_v1) ->
@ -402,8 +322,6 @@ write_projection(_Host, _TcpPort, ProjType, Proj)
%% @doc Get all projections from the FLU's projection store. %% @doc Get all projections from the FLU's projection store.
-spec get_all_projections(port_wrap(), projection_type()) ->
{ok, [projection()]} | {error, term()}.
get_all_projections(#yessir{name=Name}, ProjType) get_all_projections(#yessir{name=Name}, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Ps = [Proj || {{N,proj,PT,_}, Proj} <- get(), N == Name, PT == ProjType], Ps = [Proj || {{N,proj,PT,_}, Proj} <- get(), N == Name, PT == ProjType],
@ -411,9 +329,6 @@ get_all_projections(#yessir{name=Name}, ProjType)
%% @doc Get all projections from the FLU's projection store. %% @doc Get all projections from the FLU's projection store.
-spec get_all_projections(inet_host(), inet_port(),
projection_type()) ->
{ok, [projection()]} | {error, term()}.
get_all_projections(_Host, _TcpPort, ProjType) get_all_projections(_Host, _TcpPort, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
@ -425,8 +340,6 @@ get_all_projections(_Host, _TcpPort, ProjType)
%% @doc Get all epoch numbers from the FLU's projection store. %% @doc Get all epoch numbers from the FLU's projection store.
-spec list_all_projections(port_wrap(), projection_type()) ->
{ok, [non_neg_integer()]} | {error, term()}.
list_all_projections(Sock, ProjType) list_all_projections(Sock, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
case get_all_projections(Sock, ProjType) of case get_all_projections(Sock, ProjType) of
@ -436,9 +349,6 @@ list_all_projections(Sock, ProjType)
%% @doc Get all epoch numbers from the FLU's projection store. %% @doc Get all epoch numbers from the FLU's projection store.
-spec list_all_projections(inet_host(), inet_port(),
projection_type()) ->
{ok, [non_neg_integer()]} | {error, term()}.
list_all_projections(_Host, _TcpPort, ProjType) list_all_projections(_Host, _TcpPort, ProjType)
when ProjType == 'public' orelse ProjType == 'private' -> when ProjType == 'public' orelse ProjType == 'private' ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
@ -450,8 +360,6 @@ list_all_projections(_Host, _TcpPort, ProjType)
%% @doc Quit &amp; close the connection to remote FLU. %% @doc Quit &amp; close the connection to remote FLU.
-spec quit(port_wrap()) ->
ok.
quit(_) -> quit(_) ->
ok. ok.
@ -460,8 +368,6 @@ quit(_) ->
%% @doc Restricted API: Write a chunk of already-sequenced data to %% @doc Restricted API: Write a chunk of already-sequenced data to
%% `File' at `Offset'. %% `File' at `Offset'.
-spec write_chunk(port_wrap(), epoch_id(), file_name(), file_offset(), chunk()) ->
ok | {error, error_general()} | {error, term()}.
write_chunk(#yessir{name=Name}, _EpochID, File, Offset, Chunk) write_chunk(#yessir{name=Name}, _EpochID, File, Offset, Chunk)
when Offset >= ?MINIMUM_OFFSET -> when Offset >= ?MINIMUM_OFFSET ->
Pos = case get({Name,offset,File}) of Pos = case get({Name,offset,File}) of
@ -474,9 +380,6 @@ write_chunk(#yessir{name=Name}, _EpochID, File, Offset, Chunk)
%% @doc Restricted API: Write a chunk of already-sequenced data to %% @doc Restricted API: Write a chunk of already-sequenced data to
%% `File' at `Offset'. %% `File' at `Offset'.
-spec write_chunk(inet_host(), inet_port(),
epoch_id(), file_name(), file_offset(), chunk()) ->
ok | {error, error_general()} | {error, term()}.
write_chunk(_Host, _TcpPort, EpochID, File, Offset, Chunk) write_chunk(_Host, _TcpPort, EpochID, File, Offset, Chunk)
when Offset >= ?MINIMUM_OFFSET -> when Offset >= ?MINIMUM_OFFSET ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
@ -489,8 +392,6 @@ write_chunk(_Host, _TcpPort, EpochID, File, Offset, Chunk)
%% @doc Restricted API: Delete a file after it has been successfully %% @doc Restricted API: Delete a file after it has been successfully
%% migrated. %% migrated.
-spec delete_migration(port_wrap(), epoch_id(), file_name()) ->
ok | {error, error_general() | 'no_such_file'} | {error, term()}.
delete_migration(#yessir{name=Name}, _EpochID, File) -> delete_migration(#yessir{name=Name}, _EpochID, File) ->
case get({Name,offset,File}) of case get({Name,offset,File}) of
undefined -> undefined ->
@ -503,8 +404,6 @@ delete_migration(#yessir{name=Name}, _EpochID, File) ->
%% @doc Restricted API: Delete a file after it has been successfully %% @doc Restricted API: Delete a file after it has been successfully
%% migrated. %% migrated.
-spec delete_migration(inet_host(), inet_port(), epoch_id(), file_name()) ->
ok | {error, error_general() | 'no_such_file'} | {error, term()}.
delete_migration(_Host, _TcpPort, EpochID, File) -> delete_migration(_Host, _TcpPort, EpochID, File) ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
try try
@ -516,16 +415,12 @@ delete_migration(_Host, _TcpPort, EpochID, File) ->
%% @doc Restricted API: Truncate a file after it has been successfully %% @doc Restricted API: Truncate a file after it has been successfully
%% erasure coded. %% erasure coded.
-spec trunc_hack(port_wrap(), epoch_id(), file_name()) ->
ok | {error, error_general() | 'no_such_file'} | {error, term()}.
trunc_hack(#yessir{name=Name}, _EpochID, File) -> trunc_hack(#yessir{name=Name}, _EpochID, File) ->
put({Name,offset,File}, ?MINIMUM_OFFSET). put({Name,offset,File}, ?MINIMUM_OFFSET).
%% @doc Restricted API: Truncate a file after it has been successfully %% @doc Restricted API: Truncate a file after it has been successfully
%% erasure coded. %% erasure coded.
-spec trunc_hack(inet_host(), inet_port(), epoch_id(), file_name()) ->
ok | {error, error_general() | 'no_such_file'} | {error, term()}.
trunc_hack(_Host, _TcpPort, EpochID, File) -> trunc_hack(_Host, _TcpPort, EpochID, File) ->
Sock = connect(#p_srvr{proto_mod=?MODULE}), Sock = connect(#p_srvr{proto_mod=?MODULE}),
try try