<commit_before>%% @author Northscale <<EMAIL>>
%% @copyright 2010 NorthScale, Inc.
%%
%% 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(ebucketmigrator_srv).

-behaviour(gen_server).

-include("ns_common.hrl").

-define(SERVER, ?MODULE).
-define(CONNECT_TIMEOUT, ns_config_ets_dup:get_timeout(ebucketmigrator_connect, 60000)).
% Microseconds because we use timer:now_diff
-define(UPSTREAM_TIMEOUT, ns_config_ets_dup:get_timeout(ebucketmigrator_upstream_us, 600000000)).
-define(TIMEOUT_CHECK_INTERVAL, 15000).
-define(TERMINATE_TIMEOUT, ns_config_ets_dup:get_timeout(ebucketmigrator_terminate, 30000)).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2,
         handle_info/2, terminate/2, code_change/3]).

-type vb_filter_change_state() :: not_started | started | completed.

-record(had_backfill, {value :: boolean() | undefined,
                       waiters = [] :: list()}).

-record(state, {upstream :: port(),
                upstream_aux :: port(),
                downstream :: port(),
                downstream_aux :: port(),
                upstream_sender :: pid(),
                upbuf = <<>> :: binary(),
                downbuf = <<>> :: binary(),
                vbuckets,
                last_sent_seqno = -1 :: integer(),
                takeover :: boolean(),
                takeover_done :: boolean(),
                takeover_msgs_seen = 0 :: non_neg_integer(),
                last_seen :: erlang:timestamp(),

                vb_filter_change_state = not_started :: vb_filter_change_state(),
                vb_filter_change_owner = undefined :: {pid(), any()} | undefined,

                tap_name :: binary(),
                pid :: pid(),          % our own pid for informational purposes

                %% from perspective of ns_server we define backfill as
                %% tap stream that is resetting (i.e. deleting and
                %% recreating) it's only vbucket. In practice when
                %% there's multiple vbuckets we'll always set it to
                %% false.
                had_backfill = #had_backfill{} :: #had_backfill{}
               }).

%% external API
-export([start_link/3, start_link/4,
         build_args/5, add_args_option/3, get_args_option/2,
         start_vbucket_filter_change/2,
         start_old_vbucket_filter_change/1,
         set_controlling_process/2,
         had_backfill/1]).

-include("mc_constants.hrl").
-include("mc_entry.hrl").

%%
%% gen_server callback implementation
%%

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

queue_data(State, Element, Data) ->
    OldBuffer = erlang:element(Element, State),
    erlang:setelement(Element, State, <<OldBuffer/binary, Data/binary>>).

handle_leftover_message({tcp, Socket, Data},
                        #state{downstream=Downstream,
                               upstream=Upstream} = State) ->
    queue_data(State, case Socket of
                          Downstream ->
                              #state.downbuf;
                          Upstream ->
                              #state.upbuf
                      end, Data);
handle_leftover_message(Msg, State) ->
    handle_leftover_message_common(Msg, State).

handle_leftover_message_common({tcp_closed, Socket}, State) ->
    erlang:exit({connect_close_during_vbucket_filter_change,
                 case State#state.downstream =:= Socket of
                     true -> downstream;
                     false -> upstream
                 end});
handle_leftover_message_common({check_for_timeout, _}, State) ->
    ?log_debug("Ignoring check_for_timeout "
               "in the middle of vbucket filter change"),
    State;
handle_leftover_message_common(retry_not_ready_vbuckets, State) ->
    ?log_debug("Ignoring retry_not_ready_vbuckets "
               "in the middle of vbucket filter change"),
    State;
handle_leftover_message_common({system, _, _}, State) ->
    ?log_debug("Got erlang system message that I'll drop on the floor. "
               "I'll be dead soon anyways."),
    State;
handle_leftover_message_common({'EXIT', _Pid, Reason} = ExitMsg, _State) ->
    case Reason =:= normal orelse Reason =:= shutdown of
        true ->
            ok;
        false ->
            ?log_error("Killing myself due to exit message: ~p", [ExitMsg])
    end,
    erlang:exit(Reason);
handle_leftover_message_common(Msg, _State) ->
    ?log_error("Got unexpected message ~p", [Msg]),
    erlang:exit({unexpected_message, Msg}).

eat_leftover_messages(State) ->
    fold_messages(
      fun handle_leftover_message/2, State,
      fun () ->
              receive
                  Msg ->
                      {ok, Msg}
              after 0 ->
                      stop
              end
      end).

fold_messages(Fn, Acc, Producer) ->
    case Producer() of
        {ok, Msg} ->
            fold_messages(Fn, Fn(Msg, Acc), Producer);
        stop ->
            Acc
    end.

complete_native_vb_filter_change(#state{downstream=Downstream,
                                        upstream=Upstream,
                                        vb_filter_change_state=ChangeState,
                                        vb_filter_change_owner=Owner} = State) ->
    completed = ChangeState,
    true = Owner =/= undefined,

    inc_counter(native_vbucket_filter_changes),

    %% ok so first lets disable socket's active flags
    ok = inet:setopts(Downstream, [{active, false}]),
    ok = inet:setopts(Upstream, [{active, false}]),

    ?log_debug("Proceeding with reading unread binaries"),
    %% now we need to process pending messages
    State2 = eat_leftover_messages(State),
    State3 = confirm_downstream(State2),
    reply_and_die(State3).

complete_old_vb_filter_change(#state{downstream=Downstream,
                                     upstream=Upstream} = State) ->
    inc_counter(non_native_vbucket_filter_changes),
    (catch master_activity_events:note_vbucket_filter_change_old()),

    ok = gen_tcp:close(Upstream),
    ?log_debug("Closed upstream connection"),
    ok = inet:setopts(Downstream, [{active, false}]),

    State2 = confirm_downstream(State),

    State3 = State2#state{upstream=undefined,
                          upbuf= <<>>,
                          downbuf= <<>>},
    reply_and_die(State3).<commit_msg>MB-5307: silence upstream sender during vbucket filter change<commit_after>%% @author Northscale <<EMAIL>>
%% @copyright 2010 NorthScale, Inc.
%%
%% 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(ebucketmigrator_srv).

-behaviour(gen_server).

-include("ns_common.hrl").

-define(SERVER, ?MODULE).
-define(CONNECT_TIMEOUT, ns_config_ets_dup:get_timeout(ebucketmigrator_connect, 60000)).
% Microseconds because we use timer:now_diff
-define(UPSTREAM_TIMEOUT, ns_config_ets_dup:get_timeout(ebucketmigrator_upstream_us, 600000000)).
-define(TIMEOUT_CHECK_INTERVAL, 15000).
-define(TERMINATE_TIMEOUT, ns_config_ets_dup:get_timeout(ebucketmigrator_terminate, 30000)).

%% gen_server callbacks
-export([init/1, handle_call/3, handle_cast/2,
         handle_info/2, terminate/2, code_change/3]).

-type vb_filter_change_state() :: not_started | started | completed.

-record(had_backfill, {value :: boolean() | undefined,
                       waiters = [] :: list()}).

-record(state, {upstream :: port(),
                upstream_aux :: port(),
                downstream :: port(),
                downstream_aux :: port(),
                upstream_sender :: pid(),
                upbuf = <<>> :: binary(),
                downbuf = <<>> :: binary(),
                vbuckets,
                last_sent_seqno = -1 :: integer(),
                takeover :: boolean(),
                takeover_done :: boolean(),
                takeover_msgs_seen = 0 :: non_neg_integer(),
                last_seen :: erlang:timestamp(),

                vb_filter_change_state = not_started :: vb_filter_change_state(),
                vb_filter_change_owner = undefined :: {pid(), any()} | undefined,

                tap_name :: binary(),
                pid :: pid(),          % our own pid for informational purposes

                %% from perspective of ns_server we define backfill as
                %% tap stream that is resetting (i.e. deleting and
                %% recreating) it's only vbucket. In practice when
                %% there's multiple vbuckets we'll always set it to
                %% false.
                had_backfill = #had_backfill{} :: #had_backfill{}
               }).

%% external API
-export([start_link/3, start_link/4,
         build_args/5, add_args_option/3, get_args_option/2,
         start_vbucket_filter_change/2,
         start_old_vbucket_filter_change/1,
         set_controlling_process/2,
         had_backfill/1]).

-include("mc_constants.hrl").
-include("mc_entry.hrl").

%%
%% gen_server callback implementation
%%

code_change(_OldVsn, State, _Extra) ->
    {ok, State}.

queue_data(State, Element, Data) ->
    OldBuffer = erlang:element(Element, State),
    erlang:setelement(Element, State, <<OldBuffer/binary, Data/binary>>).

handle_leftover_message({tcp, Socket, Data},
                        #state{downstream=Downstream,
                               upstream=Upstream} = State) ->
    queue_data(State, case Socket of
                          Downstream ->
                              #state.downbuf;
                          Upstream ->
                              #state.upbuf
                      end, Data);
handle_leftover_message(Msg, State) ->
    handle_leftover_message_common(Msg, State).

handle_leftover_message_common({tcp_closed, Socket}, State) ->
    erlang:exit({connect_close_during_vbucket_filter_change,
                 case State#state.downstream =:= Socket of
                     true -> downstream;
                     false -> upstream
                 end});
handle_leftover_message_common({check_for_timeout, _}, State) ->
    ?log_debug("Ignoring check_for_timeout "
               "in the middle of vbucket filter change"),
    State;
handle_leftover_message_common(retry_not_ready_vbuckets, State) ->
    ?log_debug("Ignoring retry_not_ready_vbuckets "
               "in the middle of vbucket filter change"),
    State;
handle_leftover_message_common({system, _, _}, State) ->
    ?log_debug("Got erlang system message that I'll drop on the floor. "
               "I'll be dead soon anyways."),
    State;
handle_leftover_message_common({'EXIT', _Pid, Reason} = ExitMsg, _State) ->
    case Reason =:= normal orelse Reason =:= shutdown of
        true ->
            ok;
        false ->
            ?log_error("Killing myself due to exit message: ~p", [ExitMsg])
    end,
    erlang:exit(Reason);
handle_leftover_message_common(Msg, _State) ->
    ?log_error("Got unexpected message ~p", [Msg]),
    erlang:exit({unexpected_message, Msg}).

eat_leftover_messages(State) ->
    fold_messages(
      fun handle_leftover_message/2, State,
      fun () ->
              receive
                  Msg ->
                      {ok, Msg}
              after 0 ->
                      stop
              end
      end).

fold_messages(Fn, Acc, Producer) ->
    case Producer() of
        {ok, Msg} ->
            fold_messages(Fn, Fn(Msg, Acc), Producer);
        stop ->
            Acc
    end.

complete_native_vb_filter_change(#state{downstream=Downstream,
                                        upstream=Upstream,
                                        upstream_sender=UpstreamSender,
                                        vb_filter_change_state=ChangeState,
                                        vb_filter_change_owner=Owner} = State) ->
    completed = ChangeState,
    true = Owner =/= undefined,

    inc_counter(native_vbucket_filter_changes),

    %% ok so first lets disable socket's active flags
    ok = inet:setopts(Downstream, [{active, false}]),
    ok = inet:setopts(Upstream, [{active, false}]),

    ok = gen_server:call(UpstreamSender, silence_upstream),
    ?log_debug("Silenced upstream sender"),

    ?log_debug("Proceeding with reading unread binaries"),
    %% now we need to process pending messages
    State2 = eat_leftover_messages(State),
    State3 = confirm_downstream(State2),
    reply_and_die(State3).

complete_old_vb_filter_change(#state{downstream=Downstream,
                                     upstream=Upstream,
                                     upstream_sender=UpstreamSender} = State) ->
    inc_counter(non_native_vbucket_filter_changes),
    (catch master_activity_events:note_vbucket_filter_change_old()),

    ok = gen_server:call(UpstreamSender, silence_upstream),
    ok = gen_tcp:close(Upstream),
    ?log_debug("Closed upstream connection"),

    ok = inet:setopts(Downstream, [{active, false}]),

    State2 = confirm_downstream(State),

    State3 = State2#state{upstream=undefined,
                          upbuf= <<>>,
                          downbuf= <<>>},
    reply_and_die(State3).
{'fraction_non_alphanumeric': 0.10833273626137423, 'fraction_numerical': 0.00874113348140718, 'mean_word_length': 2.3464397027091826, 'pattern_counts': {'":': 0, '<': 27, '<?xml version=': 0, '>': 77, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 10, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4909916', 'n_tokens_mistral': 4150, 'n_tokens_neox': 3893, 'n_words': 903}