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

Initialize FLU package with ranch listener

Browse source
This commit is contained in:
Shunichi Shinohara 2015-11-18 17:00:14 +09:00
parent 9579b1b8b2
commit 7614910f36
4 changed files with 193 additions and 181 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

29
src/machi_flu1.erl
View file

@ -58,10 +58,9 @@
-export([start_link/1, stop/1,
update_wedge_state/3, wedge_myself/2]).
-export([make_listener_regname/1, make_projection_server_regname/1]).
-export([make_projection_server_regname/1]).
%% TODO: remove or replace in OTP way after gen_*'ified
-export([main2/4, run_append_server/2,
%% run_listen_server/1,
current_state/1, format_state/1]).
-record(state, {
@ -69,14 +68,9 @@
proj_store :: pid(),
witness = false :: boolean(),
append_pid :: pid(),
tcp_port :: non_neg_integer(),
data_dir :: string(),
wedged = true :: boolean(),
etstab :: ets:tid(),
epoch_id :: 'undefined' | machi_dt:epoch_id(),
pb_mode = undefined :: 'undefined' | 'high' | 'low',
high_clnt :: 'undefined' | pid(),
trim_table :: ets:tid(),
props = [] :: list() % proplist
}).
@ -153,8 +147,6 @@ main2(FluName, TcpPort, DataDir, Props) ->
S0 = #state{flu_name=FluName,
proj_store=ProjectionPid,
tcp_port=TcpPort,
data_dir=DataDir,
wedged=Wedged_p,
witness=Witness_p,
etstab=ets_table_name(FluName),
@ -168,7 +160,8 @@ main2(FluName, TcpPort, DataDir, Props) ->
ok
end,
S1 = S0#state{append_pid=AppendPid},
{ok, ListenPid} = start_listen_server(S1),
{ok, ListenerPid} = start_listen_server(TcpPort, DataDir, S1),
io:format(user, "Listener started: ~w~n", [{FluName, ListenerPid}]),
Config_e = machi_util:make_config_filename(DataDir, "unused"),
ok = filelib:ensure_dir(Config_e),
@ -180,22 +173,23 @@ main2(FluName, TcpPort, DataDir, Props) ->
put(flu_flu_name, FluName),
put(flu_append_pid, S1#state.append_pid),
put(flu_projection_pid, ProjectionPid),
put(flu_listen_pid, ListenPid),
put(flu_listen_pid, ListenerPid),
proc_lib:init_ack({ok, self()}),
receive killme -> ok end,
(catch exit(S1#state.append_pid, kill)),
(catch exit(ProjectionPid, kill)),
(catch exit(ListenPid, kill)),
(catch exit(ListenerPid, kill)),
ok.
start_append_server(S, AckPid) ->
proc_lib:start_link(?MODULE, run_append_server, [AckPid, S], ?INIT_TIMEOUT).
start_listen_server(_S) ->
%% FIXMEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEEE
%% proc_lib:start_link(?MODULE, run_listen_server, [S], ?INIT_TIMEOUT).
{ok, dummy}.
start_listen_server(TcpPort, DataDir,
#state{flu_name=FluName, witness=Witness, etstab=EtsTab,
proj_store=ProjStore}=_S) ->
machi_listener_sup:start_listener(FluName, TcpPort, Witness, DataDir,
EtsTab, ProjStore).
run_append_server(FluPid, #state{flu_name=Name,
wedged=Wedged_p,epoch_id=EpochId}=S) ->
@ -307,9 +301,6 @@ handle_append(CoC_Namespace, CoC_Locator,
Error
end.
make_listener_regname(BaseName) ->
list_to_atom(atom_to_list(BaseName) ++ "_listener").
%% This is the name of the projection store that is spawned by the
%% *flu*, for use primarily in testing scenarios. In normal use, we
%% ought to be using the OTP style of managing processes, via

14
src/machi_flu_psup.erl
View file

@ -143,21 +143,19 @@ init([FluName, TcpPort, DataDir, Props0]) ->
FProxySupSpec = machi_file_proxy_sup:child_spec(FluName),
ListenerRegName = machi_flu1:make_listener_regname(FluName),
NbAcceptors = 100,
ListenerSpec = ranch:child_spec(ListenerRegName, NbAcceptors,
ranch_tcp, [{port, TcpPort}],
machi_pb_protocol, []),
ListenerSupSpec = {machi_listener_sup:make_listener_sup_name(FluName),
{machi_listener_sup, start_link, [FluName]},
permanent, ?SHUTDOWN, supervisor, []},
FluSpec = {FluName,
{machi_flu1, start_link,
[ [{FluName, TcpPort+1, DataDir}|Props] ]},
[ [{FluName, TcpPort, DataDir}|Props] ]},
permanent, ?SHUTDOWN, worker, []},
{ok, {SupFlags, [
ProjSpec, FitnessSpec, MgrSpec,
FProxySupSpec, FNameMgrSpec, MetaSupSpec,
FluSpec, ListenerSpec]}}.
ListenerSupSpec, FluSpec]}}.
make_flu_regname(FluName) when is_atom(FluName) ->
FluName.

78
src/machi_listener_sup.erl Normal file
View file

@ -0,0 +1,78 @@
%% -------------------------------------------------------------------
%%
%% Copyright (c) 2007-2015 Basho Technologies, Inc. All Rights Reserved.
%%
%% This file is provided to you 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.
%%
%% -------------------------------------------------------------------
%% @doc This is the supervisor to hold ranch listener for sigle FLU,
%% holds at most one child worker.
%% TODO: This supervisor is maybe useless. First introduced by workaround
%% to start listener dynamically in flu1 initialization time.
%% Because psup is blocked in flu1 initialization time, adding a child
%% to psup leads to deadlock.
%% By the result of refactoring process, if initialization can be done
%% only by static arguments, then this supervisor should be removed
%% and add listener as a direct child of psup.
-module(machi_listener_sup).
-behaviour(supervisor).
%% public API
-export([start_link/1,
start_listener/6,
stop_listener/1,
make_listener_sup_name/1,
make_listener_name/1]).
%% supervisor callback
-export([init/1]).
-define(BACKLOG, 8192).
start_link(FluName) ->
supervisor:start_link({local, make_listener_sup_name(FluName)}, ?MODULE, []).
start_listener(FluName, TcpPort, Witness, DataDir, EpochTab, ProjStore) ->
supervisor:start_child(make_listener_sup_name(FluName),
child_spec(FluName, TcpPort, Witness, DataDir,
EpochTab, ProjStore)).
stop_listener(FluName) ->
SupName = make_listener_sup_name(FluName),
ListenerName = make_listener_name(FluName),
ok = supervisor:terminate_child(SupName, ListenerName),
ok = supervisor:delete_child(SupName, ListenerName).
make_listener_sup_name(FluName) when is_atom(FluName) ->
list_to_atom(atom_to_list(FluName) ++ "_listener_sup").
make_listener_name(FluName) ->
list_to_atom(atom_to_list(FluName) ++ "_listener").
init([]) ->
SupFlags = {one_for_one, 1000, 10},
{ok, {SupFlags, []}}.
child_spec(FluName, TcpPort, Witness, DataDir, EpochTab, ProjStore) ->
ListenerName = make_listener_name(FluName),
NbAcceptors = 100,
TcpOpts = [{port, TcpPort}, {backlog, ?BACKLOG}],
ProtoOpts = [FluName, Witness, DataDir, EpochTab, ProjStore],
ranch:child_spec(ListenerName, NbAcceptors,
ranch_tcp, TcpOpts,
machi_pb_protocol, ProtoOpts).

253
src/machi_pb_protocol.erl
View file

@ -19,7 +19,7 @@
%% -------------------------------------------------------------------
%% @doc Ranch protocol callback module to handle PB protocol over
%% transport
%% transport, including both high and low modes.
%% TODO
%% - Two modes, high and low should be separated at listener level?
@ -28,172 +28,114 @@
-behaviour(gen_server).
-behaviour(ranch_protocol).
-export([start_link/4]).
-export([init/1]).
-export([handle_call/3, handle_cast/2, handle_info/2,
terminate/2, code_change/3]).
-include_lib("kernel/include/file.hrl").
-include("machi.hrl").
-include("machi_pb.hrl").
-include("machi_projection.hrl").
-define(V(X,Y), ok).
%% -include("machi_verbose.hrl").
-ifdef(TEST).
-include_lib("eunit/include/eunit.hrl").
-endif. % TEST
-record(state, {ref,
socket,
transport,
opts,
pb_mode,
data_dir,
witness,
%% - Used in projection related requests in low mode
%% - Used in spawning CR client in high mode
proj_store,
%%%% Low mode only
%% Current best knowledge, used for wedge_self / bad_epoch check
epoch_id,
%% Used in dispatching append_chunk* reqs to the
%% append serializing process
flu_name,
%% Stored in ETS before factorization, can be stored in the recored?
wedged,
%% Used in server_wedge_status to lookup the table
etstab,
%%%% High mode only
high_clnt,
%%%% to be removed
eof
}).
-record(state, {
%% Transport related items passed from Ranch
ref :: ranch:ref(),
socket :: socket(),
transport :: module(),
%% -record(state, {
%% used in append serializer to trigger chain mgr react_to_env
%% flu_name :: atom(),
%% proj_store :: pid(),
%% witness = false :: boolean(),
%% append_pid :: pid(),
%% tcp_port :: non_neg_integer(),
%% data_dir :: string(),
%% wedged = true :: boolean(),
%% etstab :: ets:tid(),
%% epoch_id :: 'undefined' | machi_dt:epoch_id(),
%% pb_mode = undefined :: 'undefined' | 'high' | 'low',
%% high_clnt :: 'undefined' | pid(),
%% trim_table :: ets:tid(),
%% props = [] :: list() % proplist
%% }).
%% Machi application related items below
data_dir :: string(),
witness :: boolean(),
pb_mode :: undefined | high | low,
%% - Used in projection related requests in low mode
%% - Used in spawning CR client in high mode
proj_store :: pid(),
-spec start_link(ranch:ref(), any(), module(), any()) -> {ok, pid()}.
start_link(Ref, Socket, Transport, Opts) ->
proc_lib:start_link(?MODULE, init, [#state{ref=Ref, socket=Socket,
transport=Transport,
opts=Opts}]).
%% Low mode only
%% Current best knowledge, used for wedge_self / bad_epoch check
epoch_id :: undefined | machi_dt:epoch_id(),
%% Used in dispatching append_chunk* reqs to the
%% append serializing process
flu_name :: atom(),
%% Used in server_wedge_status to lookup the table
epoch_tab :: ets:tid(),
init(#state{ref=Ref, socket=Socket, transport=Transport, opts=_Opts}=State) ->
ok = proc_lib:init_ack({ok, self()}),
%% TODO: Perform any required state initialization here.
ok = ranch:accept_ack(Ref),
ok = Transport:setopts(Socket, [{active, once}]),
gen_server:enter_loop(?MODULE, [], State).
%% High mode only
high_clnt :: pid(),
%% anything you want
props = [] :: list() % proplist
}).
-type socket() :: any().
-type state() :: #state{}.
-spec start_link(ranch:ref(), socket(), module(), [term()]) -> {ok, pid()}.
start_link(Ref, Socket, Transport, [FluName, Witness, DataDir, EpochTab, ProjStore]) ->
proc_lib:start_link(?MODULE, init, [#state{ref=Ref,
socket=Socket,
transport=Transport,
flu_name=FluName,
witness=Witness,
data_dir=DataDir,
epoch_tab=EpochTab,
proj_store=ProjStore}]).
-spec init(state()) -> no_return().
init(#state{ref=Ref, socket=Socket, transport=Transport}=State) ->
ok = proc_lib:init_ack({ok, self()}),
ok = ranch:accept_ack(Ref),
{_Wedged_p, CurrentEpochID} = lookup_epoch(State),
ok = Transport:setopts(Socket, [{active, once}|?PB_PACKET_OPTS]),
gen_server:enter_loop(?MODULE, [], State#state{epoch_id=CurrentEpochID}).
handle_call(Request, _From, S) ->
lager:warning("~s:handle_call UNKNOWN message: ~w", [?MODULE, Request]),
Reply = {error, {unknown_message, Request}},
{reply, Reply, S}.
handle_cast(_Msg, S) ->
io:format(user, "~s:handle_cast: ~p\n", [?MODULE, _Msg]),
lager:warning("~s:handle_cast UNKNOWN message: ~w", [?MODULE, _Msg]),
{noreply, S}.
handle_info({tcp, Sock, Data}=_Info, S) ->
io:format(user, "~s:handle_info: ~p\n", [?MODULE, _Info]),
%% TODO: Other transport support needed?? TLS/SSL, SCTP
handle_info({tcp, Sock, Data}=_Info, #state{socket=Sock}=S) ->
lager:debug("~s:handle_info: ~w", [?MODULE, _Info]),
transport_received(Sock, Data, S);
handle_info({tcp_closed, Sock}=_Info, S) ->
io:format(user, "~s:handle_info: ~p\n", [?MODULE, _Info]),
handle_info({tcp_closed, Sock}=_Info, #state{socket=Sock}=S) ->
lager:debug("~s:handle_info: ~w", [?MODULE, _Info]),
transport_closed(Sock, S);
handle_info({tcp_error, Sock, Reason}=_Info, S) ->
io:format(user, "~s:handle_info: ~p\n", [?MODULE, _Info]),
handle_info({tcp_error, Sock, Reason}=_Info, #state{socket=Sock}=S) ->
lager:debug("~s:handle_info: ~w", [?MODULE, _Info]),
transport_error(Sock, Reason, S);
handle_info(_Info, S) ->
io:format(user, "~s:handle_info: ~p\n", [?MODULE, _Info]),
lager:warning("~s:handle_info UNKNOWN message: ~w", [?MODULE, _Info]),
{noreply, S}.
terminate(_Reason, _S) ->
io:format(user, "~s:terminate: ~p\n", [?MODULE, _Reason]),
terminate(_Reason, #state{socket=undefined}=_S) ->
lager:debug("~s:terminate: ~w", [?MODULE, _Reason]),
ok;
terminate(_Reason, #state{socket=Socket}=_S) ->
lager:debug("~s:terminate: ~w", [?MODULE, _Reason]),
(catch gen_tcp:close(Socket)),
ok.
code_change(_OldVsn, S, _Extra) ->
{ok, S}.
%% Internal functions, or copy-n-paste'd thingie
%%%% Just copied and will be removed %%%%
%% TODO: sock opts should be migrated to ranch equivalent
%% run_listen_server(#state{flu_name=FluName, tcp_port=TcpPort}=S) ->
%% register(make_listener_regname(FluName), self()),
%% SockOpts = ?PB_PACKET_OPTS ++
%% [{reuseaddr, true}, {mode, binary}, {active, false},
%% {backlog,8192}],
%% case gen_tcp:listen(TcpPort, SockOpts) of
%% {ok, LSock} ->
%% proc_lib:init_ack({ok, self()}),
%% listen_server_loop(LSock, S);
%% Else ->
%% error_logger:warning_msg("~s:run_listen_server: "
%% "listen to TCP port ~w: ~w\n",
%% [?MODULE, TcpPort, Else]),
%% exit({?MODULE, run_listen_server, tcp_port, TcpPort, Else})
%% end.
%% listen_server_loop(LSock, S) ->
%% {ok, Sock} = gen_tcp:accept(LSock),
%% spawn_link(fun() -> net_server_loop(Sock, S) end),
%% listen_server_loop(LSock, S).
%% net_server_loop(Sock, S) ->
%% case gen_tcp:recv(Sock, 0, ?SERVER_CMD_READ_TIMEOUT) of
%% {ok, Bin} ->
%% {RespBin, S2} =
%% case machi_pb:decode_mpb_ll_request(Bin) of
%% LL_req when LL_req#mpb_ll_request.do_not_alter == 2 ->
%% {R, NewS} = do_pb_ll_request(LL_req, S),
%% {maybe_encode_response(R), mode(low, NewS)};
%% _ ->
%% HL_req = machi_pb:decode_mpb_request(Bin),
%% 1 = HL_req#mpb_request.do_not_alter,
%% {R, NewS} = do_pb_hl_request(HL_req, make_high_clnt(S)),
%% {machi_pb:encode_mpb_response(R), mode(high, NewS)}
%% end,
%% if RespBin == async_no_response ->
%% net_server_loop(Sock, S2);
%% true ->
%% case gen_tcp:send(Sock, RespBin) of
%% ok ->
%% net_server_loop(Sock, S2);
%% {error, _} ->
%% (catch gen_tcp:close(Sock)),
%% exit(normal)
%% end
%% end;
%% {error, SockError} ->
%% Msg = io_lib:format("Socket error ~w", [SockError]),
%% R = #mpb_ll_response{req_id= <<>>,
%% generic=#mpb_errorresp{code=1, msg=Msg}},
%% _Resp = machi_pb:encode_mpb_ll_response(R),
%% %% TODO: Weird that sometimes neither catch nor try/catch
%% %% can prevent OTP's SASL from logging an error here.
%% %% Error in process <0.545.0> with exit value: {badarg,[{erlang,port_command,.......
%% %% TODO: is this what causes the intermittent PULSE deadlock errors?
%% %% _ = (catch gen_tcp:send(Sock, _Resp)), timer:sleep(1000),
%% (catch gen_tcp:close(Sock)),
%% exit(normal)
%% end.
%% -- private
%%%% Common transport handling
-spec transport_received(socket(), machi_dt:chunk(), state()) ->
{noreply, state()}.
transport_received(Sock, Bin, #state{transport=Transport}=S) ->
{RespBin, S2} =
case machi_pb:decode_mpb_ll_request(Bin) of
@ -207,33 +149,36 @@ transport_received(Sock, Bin, #state{transport=Transport}=S) ->
{machi_pb:encode_mpb_response(R), mode(high, NewS)}
end,
if RespBin == async_no_response ->
Transport:setopts(Sock, [{active, once}]),
{noreply, S2};
true ->
case Transport:send(Sock, RespBin) of
ok ->
Transport:setopts(Sock, [{active, once}]),
{noreply, S2};
{error, Reason} ->
transport_error(Sock, Reason, S2)
end
end.
transport_closed(Sock, S) ->
(catch gen_tcp:close(Sock)),
{stop, normal, S#state{sock=undefined}}.
-spec transport_closed(socket(), state()) -> {stop, term(), state()}.
transport_closed(_Socket, S) ->
{stop, normal, S}.
transport_error(Sock, Reason, S) ->
Msg = io_lib:format("Socket error ~w", [SockError]),
-spec transport_error(socket(), term(), state()) -> no_return().
transport_error(Sock, Reason, #state{transport=Transport}=_S) ->
Msg = io_lib:format("Socket error ~w", [Reason]),
R = #mpb_ll_response{req_id= <<>>,
generic=#mpb_errorresp{code=1, msg=Msg}},
_Resp = machi_pb:encode_mpb_ll_response(R),
%% TODO of TODO comments: comments below with four %s are copy-n-paste'd,
%% TODO for TODO comments: comments below with four %s are copy-n-paste'd,
%% then it should be considered they are still open and should be addressed.
%%%% TODO: Weird that sometimes neither catch nor try/catch
%%%% can prevent OTP's SASL from logging an error here.
%%%% Error in process <0.545.0> with exit value: {badarg,[{erlang,port_command,.......
%%%% TODO: is this what causes the intermittent PULSE deadlock errors?
%%%% _ = (catch gen_tcp:send(Sock, _Resp)), timer:sleep(1000),
(catch gen_tcp:close(Sock)),
(catch Transport:close(Sock)),
%% TODO: better to exit with `Reason'?
exit(normal).
@ -242,14 +187,11 @@ maybe_encode_response(async_no_response=X) ->
maybe_encode_response(R) ->
machi_pb:encode_mpb_ll_response(R).
%%%% Not categorized / not-yet-well-understood items
%% TODO: may be external API
mode(Mode, #state{pb_mode=undefined}=S) ->
S#state{pb_mode=Mode};
mode(_, S) ->
S.
%%%% Low PB mode %%%%
do_pb_ll_request(#mpb_ll_request{req_id=ReqID}, #state{pb_mode=high}=S) ->
@ -257,24 +199,27 @@ do_pb_ll_request(#mpb_ll_request{req_id=ReqID}, #state{pb_mode=high}=S) ->
{machi_pb_translate:to_pb_response(ReqID, unused, Result), S};
do_pb_ll_request(PB_request, S) ->
Req = machi_pb_translate:from_pb_request(PB_request),
%% io:format(user, "[~w] do_pb_ll_request Req: ~w~n", [S#state.flu_name, Req]),
{ReqID, Cmd, Result, S2} =
case Req of
{RqID, {LowCmd, _}=CMD}
when LowCmd == low_proj;
LowCmd == low_wedge_status; LowCmd == low_list_files ->
{RqID, {LowCmd, _}=Cmd0}
when LowCmd =:= low_proj;
LowCmd =:= low_wedge_status;
LowCmd =:= low_list_files ->
%% Skip wedge check for projection commands!
%% Skip wedge check for these unprivileged commands
{Rs, NewS} = do_pb_ll_request3(CMD, S),
{RqID, CMD, Rs, NewS};
{RqID, CMD} ->
EpochID = element(2, CMD), % by common convention
{Rs, NewS} = do_pb_ll_request2(EpochID, CMD, S),
{RqID, CMD, Rs, NewS}
{Rs, NewS} = do_pb_ll_request3(Cmd0, S),
{RqID, Cmd0, Rs, NewS};
{RqID, Cmd0} ->
EpochID = element(2, Cmd0), % by common convention
{Rs, NewS} = do_pb_ll_request2(EpochID, Cmd0, S),
{RqID, Cmd0, Rs, NewS}
end,
{machi_pb_translate:to_pb_response(ReqID, Cmd, Result), S2}.
do_pb_ll_request2(EpochID, CMD, S) ->
{Wedged_p, CurrentEpochID} = ets:lookup_element(S#state.etstab, epoch, 2),
{Wedged_p, CurrentEpochID} = lookup_epoch(S),
%% io:format(user, "{Wedged_p, CurrentEpochID}: ~w~n", [{Wedged_p, CurrentEpochID}]),
if Wedged_p == true ->
{{error, wedged}, S#state{epoch_id=CurrentEpochID}};
is_tuple(EpochID)
@ -287,7 +232,7 @@ do_pb_ll_request2(EpochID, CMD, S) ->
true ->
%% We're at same epoch # but different checksum, or
%% we're at a newer/bigger epoch #.
_ = wedge_myself(S#state.flu_name, CurrentEpochID),
_ = machi_flu1:wedge_myself(S#state.flu_name, CurrentEpochID),
ok
end,
{{error, bad_epoch}, S#state{epoch_id=CurrentEpochID}};
@ -295,6 +240,9 @@ do_pb_ll_request2(EpochID, CMD, S) ->
do_pb_ll_request3(CMD, S#state{epoch_id=CurrentEpochID})
end.
lookup_epoch(#state{epoch_tab=T}) ->
ets:lookup_element(T, epoch, 2).
%% Witness status does not matter below.
do_pb_ll_request3({low_echo, _BogusEpochID, Msg}, S) ->
{Msg, S};
@ -498,7 +446,7 @@ do_server_list_files(#state{data_dir=DataDir}=_S) ->
end || File <- Files]}.
do_server_wedge_status(S) ->
{Wedged_p, CurrentEpochID0} = ets:lookup_element(S#state.etstab, epoch, 2),
{Wedged_p, CurrentEpochID0} = lookup_epoch(S),
CurrentEpochID = if CurrentEpochID0 == undefined ->
?DUMMY_PV1_EPOCH;
true ->
@ -589,9 +537,6 @@ check_or_make_tagged_checksum(?CSUM_TAG_CLIENT_SHA, Client_CSum, Chunk) ->
throw({bad_csum, CS})
end.
%%%% High PB mode %%%%
do_pb_hl_request(#mpb_request{req_id=ReqID}, #state{pb_mode=low}=S) ->