<commit_before>%% @author Couchbase, Inc <<EMAIL>>
%% @copyright 2011 Couchbase, 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_single_vbucket_mover).

-export([spawn_mover/5, mover/6]).

-include("ns_common.hrl").

spawn_mover(Node, Bucket, VBucket,
            OldChain, NewChain) ->
    Parent = self(),
    Node1 =
        case Node of
            undefined ->
                node();
            _ ->
                ensure_module_loaded(Node),
                Node
        end,

    spawn_link(Node1,
               ns_single_vbucket_mover, mover,
               [Parent, Node, Bucket, VBucket, OldChain, NewChain]).

mover(Parent, Node, Bucket, VBucket,
      OldChain, [NewNode|_] = NewChain) ->
    process_flag(trap_exit, true),
    %% We do a no-op here rather than filtering these out so that the
    %% replication update will still work properly.
    if
        Node =:= undefined ->
            %% this handles case of missing vbucket (like after failing over
            %% more nodes then replica count)
            ok = ns_memcached:set_vbucket(NewNode, Bucket, VBucket, active),
            ok;
        Node /= NewNode ->
            run_mover(Bucket, VBucket, Node, NewNode, 2);
        true ->
            ok
    end,
    Parent ! {move_done, {Node, VBucket, OldChain, NewChain}}.

run_mover(Bucket, V, N1, N2, Tries) ->
    case {ns_memcached:get_vbucket(N1, Bucket, V),
          ns_memcached:get_vbucket(N2, Bucket, V)} of
        {{ok, active}, {memcached_error, not_my_vbucket, _}} ->
            %% Standard starting state
            ok = ns_memcached:set_vbucket(N2, Bucket, V, replica),
            {ok, _Pid} = ns_vbm_sup:spawn_mover(Bucket, V, N1, N2),
            wait_for_mover(Bucket, V, N1, N2, Tries);
        {{ok, dead}, {ok, active}} ->
            %% Standard ending state
            ok;
        {{memcached_error, not_my_vbucket, _}, {ok, active}} ->
            %% This generally shouldn't happen, but it's an OK final state.
            ?rebalance_warning("Weird: vbucket ~p missing from source node ~p but "
                               "active on destination node ~p.", [V, N1, N2]),
            ok;
        {{ok, active}, {ok, S}} when S /= active ->
            %% This better have been a replica, a failed previous
            %% attempt to migrate, or loaded from a valid copy or this
            %% will result in inconsistent data!
            if S /= replica ->
                    ok = ns_memcached:set_vbucket(N2, Bucket, V, replica);
               true ->
                    ok
            end,
            {ok, _Pid} = ns_vbm_sup:spawn_mover(Bucket, V, N1, N2),
            wait_for_mover(Bucket, V, N1, N2, Tries);
        {{ok, dead}, {ok, pending}} ->
            %% This is a strange state to end up in - the source
            %% shouldn't close until the destination has acknowledged
            %% the last message, at which point the state should be
            %% active.
            ?rebalance_warning("Weird: vbucket ~p in pending state on node ~p.",
                               [V, N2]),
            ok = ns_memcached:set_vbucket(N1, Bucket, V, active),
            ok = ns_memcached:set_vbucket(N2, Bucket, V, replica),
            {ok, _Pid} = ns_vbm_sup:spawn_mover(Bucket, V, N1, N2),
            wait_for_mover(Bucket, V, N1, N2, Tries)
    end.

wait_for_mover(Bucket, V, N1, N2, Tries) ->
    receive
        {'EXIT', _Pid, normal} ->
            case {ns_memcached:get_vbucket(N1, Bucket, V),
                  ns_memcached:get_vbucket(N2, Bucket, V)} of
                {{ok, dead}, {ok, active}} ->
                    ok;
                E ->
                    exit({wrong_state_after_transfer, E, V})
            end;
        {'EXIT', _Pid, stopped} ->
            exit(stopped);
        {'EXIT', _Pid, Reason} ->
            case Tries of
                0 ->
                    exit({mover_failed, Reason});
                _ ->
                    ?rebalance_warning("Got unexpected exit reason from mover:~n~p",
                                       [Reason]),
                    run_mover(Bucket, V, N1, N2, Tries-1)
            end;
        Msg ->
            ?rebalance_warning("Mover parent got unexpected message:~n"
                               "~p", [Msg]),
            wait_for_mover(Bucket, V, N1, N2, Tries)
    end.

%% Auxiliary functions

%% Check whether ?MODULE module exists on remote node.
check_module(Node) ->
    Loaded = rpc:call(Node, erlang, module_loaded, [?MODULE]),
    TryLoad =
        fun () ->
                case rpc:call(Node, code, load_file, [?MODULE]) of
                    {module, ?MODULE} ->
                        true;
                    {error, _What} ->
                        false
                end
        end,
    Loaded orelse TryLoad().

%% Loads ?MODULE on a remote node.
upload_module(Node) ->
    {?MODULE, Binary, Filename} = code:get_object_code(?MODULE),
    %% Filename is used only for informational purposes so it does not
    %% matter whether the path exists on the remote node at all.
    {module, ?MODULE} = rpc:call(Node, code, load_binary,
                                 [?MODULE, Filename, Binary]).

%% Ensure that ?MODULE is loaded on remote node.
ensure_module_loaded(Node) ->
    case check_module(Node) of
        true ->
            ?rebalance_info("Remote node ~p already has a copy of ~p module.",
                            [Node, ?MODULE]),
            ok;
        false ->
            ?rebalance_info("Remote node ~p does not have a copy of ~p module. "
                            "Uploading it.",
                            [Node, ?MODULE]),
            upload_module(Node),
            ok
    end.<commit_msg>Revert "Ensure that ns_single_vbucket_mover is loaded on the remote node."<commit_after>%% @author Couchbase, Inc <<EMAIL>>
%% @copyright 2011 Couchbase, 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_single_vbucket_mover).

-export([spawn_mover/5, mover/6]).

-include("ns_common.hrl").

spawn_mover(Node, Bucket, VBucket,
            OldChain, NewChain) ->
    Parent = self(),
    Node1 =
        case Node of
            undefined ->
                node();
            _ ->
                Node
        end,

    spawn_link(Node1,
               ns_single_vbucket_mover, mover,
               [Parent, Node, Bucket, VBucket, OldChain, NewChain]).

mover(Parent, Node, Bucket, VBucket,
      OldChain, [NewNode|_] = NewChain) ->
    process_flag(trap_exit, true),
    %% We do a no-op here rather than filtering these out so that the
    %% replication update will still work properly.
    if
        Node =:= undefined ->
            %% this handles case of missing vbucket (like after failing over
            %% more nodes then replica count)
            ok = ns_memcached:set_vbucket(NewNode, Bucket, VBucket, active),
            ok;
        Node /= NewNode ->
            run_mover(Bucket, VBucket, Node, NewNode, 2);
        true ->
            ok
    end,
    Parent ! {move_done, {Node, VBucket, OldChain, NewChain}}.

run_mover(Bucket, V, N1, N2, Tries) ->
    case {ns_memcached:get_vbucket(N1, Bucket, V),
          ns_memcached:get_vbucket(N2, Bucket, V)} of
        {{ok, active}, {memcached_error, not_my_vbucket, _}} ->
            %% Standard starting state
            ok = ns_memcached:set_vbucket(N2, Bucket, V, replica),
            {ok, _Pid} = ns_vbm_sup:spawn_mover(Bucket, V, N1, N2),
            wait_for_mover(Bucket, V, N1, N2, Tries);
        {{ok, dead}, {ok, active}} ->
            %% Standard ending state
            ok;
        {{memcached_error, not_my_vbucket, _}, {ok, active}} ->
            %% This generally shouldn't happen, but it's an OK final state.
            ?rebalance_warning("Weird: vbucket ~p missing from source node ~p but "
                               "active on destination node ~p.", [V, N1, N2]),
            ok;
        {{ok, active}, {ok, S}} when S /= active ->
            %% This better have been a replica, a failed previous
            %% attempt to migrate, or loaded from a valid copy or this
            %% will result in inconsistent data!
            if S /= replica ->
                    ok = ns_memcached:set_vbucket(N2, Bucket, V, replica);
               true ->
                    ok
            end,
            {ok, _Pid} = ns_vbm_sup:spawn_mover(Bucket, V, N1, N2),
            wait_for_mover(Bucket, V, N1, N2, Tries);
        {{ok, dead}, {ok, pending}} ->
            %% This is a strange state to end up in - the source
            %% shouldn't close until the destination has acknowledged
            %% the last message, at which point the state should be
            %% active.
            ?rebalance_warning("Weird: vbucket ~p in pending state on node ~p.",
                               [V, N2]),
            ok = ns_memcached:set_vbucket(N1, Bucket, V, active),
            ok = ns_memcached:set_vbucket(N2, Bucket, V, replica),
            {ok, _Pid} = ns_vbm_sup:spawn_mover(Bucket, V, N1, N2),
            wait_for_mover(Bucket, V, N1, N2, Tries)
    end.

wait_for_mover(Bucket, V, N1, N2, Tries) ->
    receive
        {'EXIT', _Pid, normal} ->
            case {ns_memcached:get_vbucket(N1, Bucket, V),
                  ns_memcached:get_vbucket(N2, Bucket, V)} of
                {{ok, dead}, {ok, active}} ->
                    ok;
                E ->
                    exit({wrong_state_after_transfer, E, V})
            end;
        {'EXIT', _Pid, stopped} ->
            exit(stopped);
        {'EXIT', _Pid, Reason} ->
            case Tries of
                0 ->
                    exit({mover_failed, Reason});
                _ ->
                    ?rebalance_warning("Got unexpected exit reason from mover:~n~p",
                                       [Reason]),
                    run_mover(Bucket, V, N1, N2, Tries-1)
            end;
        Msg ->
            ?rebalance_warning("Mover parent got unexpected message:~n"
                               "~p", [Msg]),
            wait_for_mover(Bucket, V, N1, N2, Tries)
    end.
{'fraction_non_alphanumeric': 0.11172078798236934, 'fraction_numerical': 0.008275613924619951, 'mean_word_length': 1.8161094224924013, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 63, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2991373', 'n_tokens_mistral': 3369, 'n_tokens_neox': 3251, 'n_words': 1022}