<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(ns_vbucket_mover).

-behavior(gen_server).

-include("ns_common.hrl").

-include_lib("eunit/include/eunit.hrl").

-define(MAX_MOVES_PER_NODE, 1).

%% API
-export([start_link/4]).

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

-type progress_callback() :: fun((dict()) -> any()).

-record(state, {bucket::nonempty_string(),
                previous_changes,
                initial_counts::dict(),
                max_per_node::pos_integer(),
                map::array(),
                moves::dict(), movers::dict(),
                progress_callback::progress_callback()}).

%%
%% API
%%

%% @doc Start the mover.
-spec start_link(string(), map(), map(), progress_callback()) ->
                        {ok, pid()} | {error, any()}.
start_link(Bucket, OldMap, NewMap, ProgressCallback) ->
    gen_server:start_link(?MODULE, {Bucket, OldMap, NewMap, ProgressCallback},
                          []).

%%
%% gen_server callbacks
%%

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

init({Bucket, OldMap, NewMap, ProgressCallback}) ->
    erlang:put(i_am_master_mover, true),
    erlang:put(replicas_changes, []),
    erlang:put(bucket_name, Bucket),
    erlang:put(total_changes, 0),
    erlang:put(actual_changes, 0),
    erlang:put(child_processes, []),

    %% Dictionary mapping old node to vbucket and new node
    {MoveDict, TrivialMoves} =
        lists:foldl(fun ({V, [M1|_] = C1, C2}, {D, TrivialMoves}) ->
                            if C1 =:= C2 ->
                                    {D, TrivialMoves + 1};
                               true ->
                                    {dict:append(M1, {V, C1, C2}, D), TrivialMoves}
                            end
                    end, {dict:new(), 0},
                    lists:zip3(lists:seq(0, length(OldMap) - 1), OldMap,
                               NewMap)),
    ?rebalance_info("The following count of vbuckets do not need to be moved at all: ~p", [TrivialMoves]),
    ?rebalance_info("The following moves are planned:~n~p", [dict:to_list(MoveDict)]),
    Movers = dict:map(fun (_, _) -> 0 end, MoveDict),
    self() ! spawn_initial,
    process_flag(trap_exit, true),
    erlang:start_timer(3000, self(), maybe_sync_changes),
    erlang:start_timer(30000, self(), log_tap_stats),
    {ok, #state{bucket=Bucket,
                previous_changes = [],
                initial_counts=count_moves(MoveDict),
                max_per_node=?MAX_MOVES_PER_NODE,
                map = map_to_array(OldMap),
                moves=MoveDict, movers=Movers,
                progress_callback=ProgressCallback}}.


handle_call(_, _From, _State) ->
    exit(not_supported).


handle_cast(unhandled, unhandled) ->
    unhandled.


%% We intentionally don't handle other exits so we'll die if one of
%% the movers fails.
handle_info({_, _, maybe_sync_changes}, #state{previous_changes = PrevChanges} = State) ->
    Changes = erlang:get('replicas_changes'),
    case Changes =:= [] orelse Changes =/= PrevChanges of
        true -> {noreply, State#state{previous_changes = Changes}};
        _ ->
            sync_replicas(),
            {noreply, State#state{previous_changes = []}}
    end;
handle_info({timeout, _, log_tap_stats}, State) ->
    rpc:eval_everywhere(diag_handler, log_all_tap_and_checkpoint_stats, []),
    misc:flush(log_tap_stats),
    {noreply, State};
handle_info(spawn_initial, State) ->
    spawn_workers(State);
handle_info({move_done, {Node, VBucket, OldChain, NewChain}},
            #state{movers=Movers, map=Map, bucket=Bucket} = State) ->
    master_activity_events:note_move_done(Bucket, VBucket),
    %% Update replication
    update_replication_post_move(VBucket, OldChain, NewChain),
    %% Free up a mover for this node
    Movers1 = dict:update(Node, fun (N) -> N - 1 end, Movers),
    %% Pull the new chain from the target map
    %% Update the current map
    Map1 = array:set(VBucket, NewChain, Map),
    ns_bucket:set_map(Bucket, array_to_map(Map1)),
    RepSyncRV = (catch begin
                           ns_config:sync_announcements(),
                           ns_config_rep:synchronize_remote()
                       end),
    case RepSyncRV of
        ok -> ok;
        _ ->
            ?log_error("Config replication sync failed: ~p", [RepSyncRV])
    end,
    sync_replicas(),
    OldCopies = OldChain -- [undefined | NewChain],
    DeleteRVs = misc:parallel_map(
                  fun (CopyNode) ->
                          {CopyNode, (catch ns_memcached:delete_vbucket(CopyNode, Bucket, VBucket))}
                  end, OldCopies, infinity),
    BadDeletes = [P || {_, RV} = P <- DeleteRVs, RV =/= ok],
    case BadDeletes of
        [] -> ok;
        _ ->
            ?log_error("Deleting some old copies of vbucket failed: ~p", [BadDeletes])
    end,<commit_msg>remove all undefineds from list for old vbucket deletion.MB-5058<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(ns_vbucket_mover).

-behavior(gen_server).

-include("ns_common.hrl").

-include_lib("eunit/include/eunit.hrl").

-define(MAX_MOVES_PER_NODE, 1).

%% API
-export([start_link/4]).

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

-type progress_callback() :: fun((dict()) -> any()).

-record(state, {bucket::nonempty_string(),
                previous_changes,
                initial_counts::dict(),
                max_per_node::pos_integer(),
                map::array(),
                moves::dict(), movers::dict(),
                progress_callback::progress_callback()}).

%%
%% API
%%

%% @doc Start the mover.
-spec start_link(string(), map(), map(), progress_callback()) ->
                        {ok, pid()} | {error, any()}.
start_link(Bucket, OldMap, NewMap, ProgressCallback) ->
    gen_server:start_link(?MODULE, {Bucket, OldMap, NewMap, ProgressCallback},
                          []).

%%
%% gen_server callbacks
%%

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

init({Bucket, OldMap, NewMap, ProgressCallback}) ->
    erlang:put(i_am_master_mover, true),
    erlang:put(replicas_changes, []),
    erlang:put(bucket_name, Bucket),
    erlang:put(total_changes, 0),
    erlang:put(actual_changes, 0),
    erlang:put(child_processes, []),

    %% Dictionary mapping old node to vbucket and new node
    {MoveDict, TrivialMoves} =
        lists:foldl(fun ({V, [M1|_] = C1, C2}, {D, TrivialMoves}) ->
                            if C1 =:= C2 ->
                                    {D, TrivialMoves + 1};
                               true ->
                                    {dict:append(M1, {V, C1, C2}, D), TrivialMoves}
                            end
                    end, {dict:new(), 0},
                    lists:zip3(lists:seq(0, length(OldMap) - 1), OldMap,
                               NewMap)),
    ?rebalance_info("The following count of vbuckets do not need to be moved at all: ~p", [TrivialMoves]),
    ?rebalance_info("The following moves are planned:~n~p", [dict:to_list(MoveDict)]),
    Movers = dict:map(fun (_, _) -> 0 end, MoveDict),
    self() ! spawn_initial,
    process_flag(trap_exit, true),
    erlang:start_timer(3000, self(), maybe_sync_changes),
    erlang:start_timer(30000, self(), log_tap_stats),
    {ok, #state{bucket=Bucket,
                previous_changes = [],
                initial_counts=count_moves(MoveDict),
                max_per_node=?MAX_MOVES_PER_NODE,
                map = map_to_array(OldMap),
                moves=MoveDict, movers=Movers,
                progress_callback=ProgressCallback}}.


handle_call(_, _From, _State) ->
    exit(not_supported).


handle_cast(unhandled, unhandled) ->
    unhandled.


%% We intentionally don't handle other exits so we'll die if one of
%% the movers fails.
handle_info({_, _, maybe_sync_changes}, #state{previous_changes = PrevChanges} = State) ->
    Changes = erlang:get('replicas_changes'),
    case Changes =:= [] orelse Changes =/= PrevChanges of
        true -> {noreply, State#state{previous_changes = Changes}};
        _ ->
            sync_replicas(),
            {noreply, State#state{previous_changes = []}}
    end;
handle_info({timeout, _, log_tap_stats}, State) ->
    rpc:eval_everywhere(diag_handler, log_all_tap_and_checkpoint_stats, []),
    misc:flush(log_tap_stats),
    {noreply, State};
handle_info(spawn_initial, State) ->
    spawn_workers(State);
handle_info({move_done, {Node, VBucket, OldChain, NewChain}},
            #state{movers=Movers, map=Map, bucket=Bucket} = State) ->
    master_activity_events:note_move_done(Bucket, VBucket),
    %% Update replication
    update_replication_post_move(VBucket, OldChain, NewChain),
    %% Free up a mover for this node
    Movers1 = dict:update(Node, fun (N) -> N - 1 end, Movers),
    %% Pull the new chain from the target map
    %% Update the current map
    Map1 = array:set(VBucket, NewChain, Map),
    ns_bucket:set_map(Bucket, array_to_map(Map1)),
    RepSyncRV = (catch begin
                           ns_config:sync_announcements(),
                           ns_config_rep:synchronize_remote()
                       end),
    case RepSyncRV of
        ok -> ok;
        _ ->
            ?log_error("Config replication sync failed: ~p", [RepSyncRV])
    end,
    sync_replicas(),
    OldCopies0 = OldChain -- NewChain,
    OldCopies = [OldCopyNode || OldCopyNode <- OldCopies0,
                                OldCopyNode =/= undefined],
    DeleteRVs = misc:parallel_map(
                  fun (CopyNode) ->
                          {CopyNode, (catch ns_memcached:delete_vbucket(CopyNode, Bucket, VBucket))}
                  end, OldCopies, infinity),
    BadDeletes = [P || {_, RV} = P <- DeleteRVs, RV =/= ok],
    case BadDeletes of
        [] -> ok;
        _ ->
            ?log_error("Deleting some old copies of vbucket failed: ~p", [BadDeletes])
    end,
{'fraction_non_alphanumeric': 0.13057467155241756, 'fraction_numerical': 0.008579855214943248, 'mean_word_length': 2.5992280476037313, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 53, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5076442', 'n_tokens_mistral': 3570, 'n_tokens_neox': 3406, 'n_words': 820}