6861295443

Committed 14 Nov 2023 08:56AM UTC coverage: 82.708% (+0.04%) from 82.664%

Build # 6861295443

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #11945 from kjellwinblad/kjell/shared_con/better_names_and_specs

emqx_bridge_v2 module improvements

Run Details

114 of 123 new or added lines in 13 files covered. (92.68%)

82 existing lines in 18 files now uncovered.

35552 of 42985 relevant lines covered (82.71%)

6667.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.23

/apps/emqx_ft/src/emqx_ft_assembler.erl

%%--------------------------------------------------------------------
%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
%%
%% Licensed under the Apache License, Version 2.0 (the "License");
%% you may not use this file except in compliance with the License.
%% You may obtain a copy of the License at
%%
%%     http://www.apache.org/licenses/LICENSE-2.0
%%
%% Unless required by applicable law or agreed to in writing, software
%% distributed under the License is distributed on an "AS IS" BASIS,
%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
%% See the License for the specific language governing permissions and
%% limitations under the License.
%%--------------------------------------------------------------------

-module(emqx_ft_assembler).

-export([start_link/4]).

-behaviour(gen_statem).
-export([callback_mode/0]).
-export([init/1]).
-export([handle_event/4]).
-export([terminate/3]).

-export([where/1]).

-type stdata() :: #{
    storage := emqx_ft_storage_fs:storage(),
    transfer := emqx_ft:transfer(),
    finopts := emqx_ft:finopts(),
    assembly := emqx_ft_assembly:t(),
    export => emqx_ft_storage_exporter:export()
}.

-define(NAME(Transfer), {n, l, {?MODULE, Transfer}}).
-define(REF(Transfer), {via, gproc, ?NAME(Transfer)}).

%%

start_link(Storage, Transfer, Size, Opts) ->
    gen_statem:start_link(?REF(Transfer), ?MODULE, {Storage, Transfer, Size, Opts}, []).

where(Transfer) ->
    gproc:where(?NAME(Transfer)).

%%

-type state() ::
    idle
    | list_local_fragments
    | {list_remote_fragments, [node()]}
    | start_assembling
    | {assemble, [{node(), emqx_ft_storage_fs:filefrag()}]}
    | complete.

-define(internal(C), {next_event, internal, C}).

callback_mode() ->
    handle_event_function.

-spec init(_Args) -> {ok, state(), stdata()}.
init({Storage, Transfer, Size, Opts}) ->
    _ = erlang:process_flag(trap_exit, true),
    St = #{
        storage => Storage,
        transfer => Transfer,
        finopts => Opts,
        assembly => emqx_ft_assembly:new(Size)
    },
    {ok, idle, St}.

-spec handle_event(info | internal, _, state(), stdata()) ->
    {next_state, state(), stdata(), {next_event, internal, _}}
    | {stop, {shutdown, ok | {error, _}}, stdata()}.
handle_event(info, kickoff, idle, St) ->
    % NOTE
    % Someone's told us to start the work, which usually means that it has set up a monitor.
    % We could wait for this message and handle it at the end of the assembling rather than at
    % the beginning, however it would make error handling much more messier.
    {next_state, list_local_fragments, St, ?internal([])};
handle_event(info, kickoff, _, _St) ->
    keep_state_and_data;
handle_event(
    internal,
    _,
    list_local_fragments,
    St = #{storage := Storage, transfer := Transfer, assembly := Asm}
) ->
    % TODO: what we do with non-transients errors here (e.g. `eacces`)?
    {ok, Fragments} = emqx_ft_storage_fs:list(Storage, Transfer, fragment),
    NAsm = emqx_ft_assembly:update(emqx_ft_assembly:append(Asm, node(), Fragments)),
    NSt = St#{assembly := NAsm},
    case emqx_ft_assembly:status(NAsm) of
        complete ->
            {next_state, start_assembling, NSt, ?internal([])};
        {incomplete, _} ->
            Nodes = emqx:running_nodes() -- [node()],
            {next_state, {list_remote_fragments, Nodes}, NSt, ?internal([])};
        % TODO: recovery?
        {error, _} = Error ->
            {stop, {shutdown, Error}}
    end;
handle_event(
    internal,
    _,
    {list_remote_fragments, Nodes},
    St = #{transfer := Transfer, assembly := Asm}
) ->
    % TODO
    % Async would better because we would not need to wait for some lagging nodes if
    % the coverage is already complete.
    % TODO: portable "storage" ref
    Results = emqx_ft_storage_fs_proto_v1:multilist(Nodes, Transfer, fragment),
    NodeResults = lists:zip(Nodes, Results),
    NAsm = emqx_ft_assembly:update(
        lists:foldl(
            fun
                ({Node, {ok, {ok, Fragments}}}, Acc) ->
                    emqx_ft_assembly:append(Acc, Node, Fragments);
                ({_Node, _Result}, Acc) ->
                    % TODO: log?
                    Acc
            end,
            Asm,
            NodeResults
        )
    ),
    NSt = St#{assembly := NAsm},
    case emqx_ft_assembly:status(NAsm) of
        complete ->
            {next_state, start_assembling, NSt, ?internal([])};
        % TODO: retries / recovery?
        {incomplete, _} = Status ->
            {stop, {shutdown, {error, Status}}};
        {error, _} = Error ->
            {stop, {shutdown, Error}}
    end;
handle_event(
    internal,
    _,
    start_assembling,
    St = #{storage := Storage, transfer := Transfer, assembly := Asm}
) ->
    Filemeta = emqx_ft_assembly:filemeta(Asm),
    Coverage = emqx_ft_assembly:coverage(Asm),
    case emqx_ft_storage_exporter:start_export(Storage, Transfer, Filemeta) of
        {ok, Export} ->
            {next_state, {assemble, Coverage}, St#{export => Export}, ?internal([])};
        {error, _} = Error ->
            {stop, {shutdown, Error}}
    end;
handle_event(internal, _, {assemble, [{Node, Segment} | Rest]}, St = #{export := Export}) ->
    % TODO
    % Currently, race is possible between getting segment info from the remote node and
    % this node garbage collecting the segment itself.
    % TODO: pipelining
    % TODO: better error handling
    {ok, Content} = pread(Node, Segment, St),
    case emqx_ft_storage_exporter:write(Export, Content) of
        {ok, NExport} ->
            {next_state, {assemble, Rest}, St#{export := NExport}, ?internal([])};
        {error, _} = Error ->
            {stop, {shutdown, Error}, maps:remove(export, St)}
    end;
handle_event(internal, _, {assemble, []}, St = #{}) ->
    {next_state, complete, St, ?internal([])};
handle_event(internal, _, complete, St = #{export := Export, finopts := Opts}) ->
    Result = emqx_ft_storage_exporter:complete(Export, Opts),
    _ = maybe_garbage_collect(Result, St),
    {stop, {shutdown, Result}, maps:remove(export, St)}.

-spec terminate(_Reason, state(), stdata()) -> _.
terminate(_Reason, _StateName, #{export := Export}) ->
    emqx_ft_storage_exporter:discard(Export);
terminate(_Reason, _StateName, #{}) ->
    ok.

pread(Node, Segment, #{storage := Storage, transfer := Transfer}) when Node =:= node() ->
    emqx_ft_storage_fs:pread(Storage, Transfer, Segment, 0, segsize(Segment));
pread(Node, Segment, #{transfer := Transfer}) ->
    emqx_ft_storage_fs_proto_v1:pread(Node, Transfer, Segment, 0, segsize(Segment)).

%%

maybe_garbage_collect(ok, #{storage := Storage, transfer := Transfer, assembly := Asm}) ->
    Nodes = emqx_ft_assembly:nodes(Asm),
    emqx_ft_storage_fs_gc:collect(Storage, Transfer, Nodes);
maybe_garbage_collect({error, _}, _St) ->
    ok.

segsize(#{fragment := {segment, Info}}) ->
    maps:get(size, Info).

1	%%--------------------------------------------------------------------
2	%% Copyright (c) 2023 EMQ Technologies Co., Ltd. All Rights Reserved.
3	%%
4	%% Licensed under the Apache License, Version 2.0 (the "License");
5	%% you may not use this file except in compliance with the License.
6	%% You may obtain a copy of the License at
7	%%
8	%% http://www.apache.org/licenses/LICENSE-2.0
9	%%
10	%% Unless required by applicable law or agreed to in writing, software
11	%% distributed under the License is distributed on an "AS IS" BASIS,
12	%% WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	%% See the License for the specific language governing permissions and
14	%% limitations under the License.
15	%%--------------------------------------------------------------------
16
17	-module(emqx_ft_assembler).
18
19	-export([start_link/4]).
20
21	-behaviour(gen_statem).
22	-export([callback_mode/0]).
23	-export([init/1]).
24	-export([handle_event/4]).
25	-export([terminate/3]).
26
27	-export([where/1]).
28
29	-type stdata() :: #{
30	storage := emqx_ft_storage_fs:storage(),
31	transfer := emqx_ft:transfer(),
32	finopts := emqx_ft:finopts(),
33	assembly := emqx_ft_assembly:t(),
34	export => emqx_ft_storage_exporter:export()
35	}.
36
37	-define(NAME(Transfer), {n, l, {?MODULE, Transfer}}).
38	-define(REF(Transfer), {via, gproc, ?NAME(Transfer)}).
39
40	%%
41
42	start_link(Storage, Transfer, Size, Opts) ->
43	gen_statem:start_link(?REF(Transfer), ?MODULE, {Storage, Transfer, Size, Opts}, []).	1,120✔
44
45	where(Transfer) ->
46	gproc:where(?NAME(Transfer)).	1,147✔
47
48	%%
49
50	-type state() ::
51	idle
52	\| list_local_fragments
53	\| {list_remote_fragments, [node()]}
54	\| start_assembling
55	\| {assemble, [{node(), emqx_ft_storage_fs:filefrag()}]}
56	\| complete.
57
58	-define(internal(C), {next_event, internal, C}).
59
60	callback_mode() ->
61	handle_event_function.	1,120✔
62
63	-spec init(_Args) -> {ok, state(), stdata()}.
64	init({Storage, Transfer, Size, Opts}) ->
65	_ = erlang:process_flag(trap_exit, true),	1,120✔
66	St = #{	1,120✔
67	storage => Storage,
68	transfer => Transfer,
69	finopts => Opts,
70	assembly => emqx_ft_assembly:new(Size)
71	},
72	{ok, idle, St}.	1,120✔
73
74	-spec handle_event(info \| internal, _, state(), stdata()) ->
75	{next_state, state(), stdata(), {next_event, internal, _}}
76	\| {stop, {shutdown, ok \| {error, _}}, stdata()}.
77	handle_event(info, kickoff, idle, St) ->
78	% NOTE
79	% Someone's told us to start the work, which usually means that it has set up a monitor.
80	% We could wait for this message and handle it at the end of the assembling rather than at
81	% the beginning, however it would make error handling much more messier.
82	{next_state, list_local_fragments, St, ?internal([])};	1,119✔
83	handle_event(info, kickoff, _, _St) ->
84	keep_state_and_data;	×
85	handle_event(
86	internal,
87	_,
88	list_local_fragments,
89	St = #{storage := Storage, transfer := Transfer, assembly := Asm}
90	) ->
91	% TODO: what we do with non-transients errors here (e.g. `eacces`)?
92	{ok, Fragments} = emqx_ft_storage_fs:list(Storage, Transfer, fragment),	1,119✔
93	NAsm = emqx_ft_assembly:update(emqx_ft_assembly:append(Asm, node(), Fragments)),	1,119✔
94	NSt = St#{assembly := NAsm},	1,119✔
95	case emqx_ft_assembly:status(NAsm) of	1,119✔
96	complete ->
97	{next_state, start_assembling, NSt, ?internal([])};	1,104✔
98	{incomplete, _} ->
99	Nodes = emqx:running_nodes() -- [node()],	15✔
100	{next_state, {list_remote_fragments, Nodes}, NSt, ?internal([])};	15✔
101	% TODO: recovery?
102	{error, _} = Error ->
103	{stop, {shutdown, Error}}	×
104	end;
105	handle_event(
106	internal,
107	_,
108	{list_remote_fragments, Nodes},
109	St = #{transfer := Transfer, assembly := Asm}
110	) ->
111	% TODO
112	% Async would better because we would not need to wait for some lagging nodes if
113	% the coverage is already complete.
114	% TODO: portable "storage" ref
115	Results = emqx_ft_storage_fs_proto_v1:multilist(Nodes, Transfer, fragment),	15✔
116	NodeResults = lists:zip(Nodes, Results),	15✔
117	NAsm = emqx_ft_assembly:update(	15✔
118	lists:foldl(
119	fun
120	({Node, {ok, {ok, Fragments}}}, Acc) ->
121	emqx_ft_assembly:append(Acc, Node, Fragments);	22✔
122	({_Node, _Result}, Acc) ->
123	% TODO: log?
UNCOV 124	Acc	×
125	end,
126	Asm,
127	NodeResults
128	)
129	),
130	NSt = St#{assembly := NAsm},	15✔
131	case emqx_ft_assembly:status(NAsm) of	15✔
132	complete ->
133	{next_state, start_assembling, NSt, ?internal([])};	11✔
134	% TODO: retries / recovery?
135	{incomplete, _} = Status ->
136	{stop, {shutdown, {error, Status}}};	4✔
137	{error, _} = Error ->
138	{stop, {shutdown, Error}}	×
139	end;
140	handle_event(
141	internal,
142	_,
143	start_assembling,
144	St = #{storage := Storage, transfer := Transfer, assembly := Asm}
145	) ->
146	Filemeta = emqx_ft_assembly:filemeta(Asm),	1,115✔
147	Coverage = emqx_ft_assembly:coverage(Asm),	1,115✔
148	case emqx_ft_storage_exporter:start_export(Storage, Transfer, Filemeta) of	1,115✔
149	{ok, Export} ->
150	{next_state, {assemble, Coverage}, St#{export => Export}, ?internal([])};	1,115✔
151	{error, _} = Error ->
152	{stop, {shutdown, Error}}	×
153	end;
154	handle_event(internal, _, {assemble, [{Node, Segment} \| Rest]}, St = #{export := Export}) ->
155	% TODO
156	% Currently, race is possible between getting segment info from the remote node and
157	% this node garbage collecting the segment itself.
158	% TODO: pipelining
159	% TODO: better error handling
160	{ok, Content} = pread(Node, Segment, St),	1,189✔
161	case emqx_ft_storage_exporter:write(Export, Content) of	1,188✔
162	{ok, NExport} ->
163	{next_state, {assemble, Rest}, St#{export := NExport}, ?internal([])};	1,188✔
164	{error, _} = Error ->
165	{stop, {shutdown, Error}, maps:remove(export, St)}	×
166	end;
167	handle_event(internal, _, {assemble, []}, St = #{}) ->
168	{next_state, complete, St, ?internal([])};	1,114✔
169	handle_event(internal, _, complete, St = #{export := Export, finopts := Opts}) ->
170	Result = emqx_ft_storage_exporter:complete(Export, Opts),	1,114✔
171	_ = maybe_garbage_collect(Result, St),	1,114✔
172	{stop, {shutdown, Result}, maps:remove(export, St)}.	1,114✔
173
174	-spec terminate(_Reason, state(), stdata()) -> _.
175	terminate(_Reason, _StateName, #{export := Export}) ->
176	emqx_ft_storage_exporter:discard(Export);	1✔
177	terminate(_Reason, _StateName, #{}) ->
178	ok.	1,119✔
179
180	pread(Node, Segment, #{storage := Storage, transfer := Transfer}) when Node =:= node() ->
181	emqx_ft_storage_fs:pread(Storage, Transfer, Segment, 0, segsize(Segment));	1,177✔
182	pread(Node, Segment, #{transfer := Transfer}) ->
183	emqx_ft_storage_fs_proto_v1:pread(Node, Transfer, Segment, 0, segsize(Segment)).	12✔
184
185	%%
186
187	maybe_garbage_collect(ok, #{storage := Storage, transfer := Transfer, assembly := Asm}) ->
188	Nodes = emqx_ft_assembly:nodes(Asm),	1,112✔
189	emqx_ft_storage_fs_gc:collect(Storage, Transfer, Nodes);	1,112✔
190	maybe_garbage_collect({error, _}, _St) ->
191	ok.	2✔
192
193	segsize(#{fragment := {segment, Info}}) ->
194	maps:get(size, Info).	1,189✔

emqx / emqx / 6861295443

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous