anchor.diff: +%% @author Couchbase <info@couchbase.com>
  +%% @copyright 2014 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.
  +%%
  +%% @doc compaction daemon for 3.0. should be renamed to compaction_daemon after
  +%% compatibility with pre 3.0 nodes will be dropped
  +%%
   -module(compaction_new_daemon).
   
  --behaviour(gen_fsm).
  +-behaviour(gen_server).
   
   -include("ns_common.hrl").
   -include("couch_db.hrl").
   
   %% API
  --export([start_link/0,
  -         inhibit_view_compaction/3, uninhibit_view_compaction/3]).
  -
  -%% gen_fsm callbacks
  --export([init/1, handle_event/3, handle_sync_event/4, handle_info/3,
  -         code_change/4, terminate/3]).
  -
  --record(state, {buckets_to_compact :: [binary()],
  -                compactor_pid,
  -                compactor_config,
  -                scheduler,
  -
  -                %% bucket for which view compaction is inhibited (note
  -                %% it's also set when we're running 'uninhibit' views
  -                %% compaction as 'final stage')
  -                view_compaction_inhibited_bucket :: undefined | binary(),
  -                %% if view compaction is inhibited this field will
  -                %% hold monitor ref of the guy who requested it. If
  -                %% that guy dies, inhibition will be canceled
  -                view_compaction_inhibited_ref :: (undefined | reference()),
  -                %% when we have 'uninhibit' request, we're starting
  -                %% our compaction with 'forced' view compaction asap
  -                %%
  -                %% requested indicates we have this request
  -                view_compaction_uninhibit_requested = false :: boolean(),
  -                %% started indicates we have started such compaction
  -                %% and waiting for it to complete
  -                view_compaction_uninhibit_started = false :: boolean(),
  -                %% this is From to reply to when such compaction is
  -                %% done
  -                view_compaction_uninhibit_requested_waiter,
  -
  +-export([start_link/0]).
  +
  +%% gen_server callbacks
  +-export([init/1, handle_call/3, handle_cast/2,
  +         handle_info/2, terminate/2, code_change/3]).
  +
  +-record(compaction_state, {buckets_to_compact :: [binary()],
  +                           compactor_pid :: pid(),
  +                           scheduler,
  +                           compactor_fun :: fun(),
  +                           compactor_config}).
  +
  +-record(state, {kv_compaction :: #compaction_state{},
  +                views_compaction :: #compaction_state{},
                   %% mapping from compactor pid to #forced_compaction{}
                   running_forced_compactions,
                   %% reverse mapping from #forced_compaction{} to compactor pid
  
positive.code:                                            OriginalTarget),
                  register_forced_compaction(Pid, Compaction, State)
          end,
      {reply, ok, StateName, NewState};
  handle_sync_event({cancel_forced_bucket_compaction, Bucket}, _From,
                    StateName, State) ->
      Compaction = #forced_compaction{type=bucket, name=Bucket},
      {reply, ok, StateName, maybe_cancel_compaction(Compaction, State)};
  handle_sync_event({cancel_forced_db_compaction, Bucket}, _From,
                    StateName, State) ->
      Compaction = #forced_compaction{type=db, name=Bucket},
      {reply, ok, StateName, maybe_cancel_compaction(Compaction, State)};
  handle_sync_event({cancel_forced_view_compaction, Bucket, DDocId},
                    _From, StateName, State) ->
      Name = <<Bucket/binary, $/, DDocId/binary>>,
      Compaction = #forced_compaction{type=view, name=Name},
      {reply, ok, StateName, maybe_cancel_compaction(Compaction, State)};
  handle_sync_event({inhibit_view_compaction, BinBucketName, Pid}, From, StateName,
                    #state{view_compaction_inhibited_ref = OldRef,
                           view_compaction_uninhibit_requested = UninhibitRequested,
                           buckets_to_compact = BucketsToCompact} = State) ->
      case OldRef =/= undefined orelse UninhibitRequested of
          true ->
              gen_server:reply(From, nack),
              {next_state, StateName, State};
          _ ->
              MRef = erlang:monitor(process, Pid),
              NewState = State#state{
                           view_compaction_inhibited_bucket = BinBucketName,
                           view_compaction_inhibited_ref = MRef,
                           view_compaction_uninhibit_requested = false},
              case StateName =:= compacting andalso hd(BucketsToCompact) =:= BinBucketName of
                  true ->
                      ?log_info("Killing currently running compactor to handle inhibit_view_compaction request for ~s", [BinBucketName]),
                      erlang:exit(NewState#state.compactor_pid, shutdown);
                  false ->
                      ok
              end,
              {reply, {ok, MRef}, StateName, NewState}
      end;
  handle_sync_event({uninhibit_view_compaction, BinBucketName, MRef}, From, StateName,
                    #state{view_compaction_inhibited_bucket = ABinBucketName,
                           view_compaction_inhibited_ref = AnMRef,
                           view_compaction_uninhibit_requested = false} = State)
    when AnMRef =:= MRef andalso ABinBucketName =:= BinBucketName ->
      State1 = case StateName of
                   compacting ->
                       ?log_info("Killing current compactor to speed up uninhibit_view_compaction: ~p", [State]),
                       erlang:exit(State#state.compactor_pid, shutdown),
                       State;
                   idle ->
                       schedule_immediate_compaction(State)
               end,
      erlang:demonitor(State#state.view_compaction_inhibited_ref),
      {next_state, StateName, State1#state{view_compaction_uninhibit_requested_waiter = From,
                                           view_compaction_inhibited_ref = undefined,
                                           view_compaction_uninhibit_started = false,
                                           view_compaction_uninhibit_requested = true}};
  handle_sync_event({uninhibit_view_compaction, _BinBucketName, _MRef} = Msg, _From, StateName, State) ->
      ?log_debug("Sending back nack on uninhibit_view_compaction: ~p, state: ~p", [Msg, State]),
      {reply, nack, StateName, State};
  handle_sync_event(Event, _From, StateName, State) ->
      ?log_warning("Got unexpected sync event ~p (when in ~p):~n~p",
                   [Event, StateName, State]),
      {reply, ok, StateName, State}.
  
  handle_info(compact, idle,
              #state{buckets_to_compact=Buckets0,
                     scheduler = Scheduler} = State) ->
      undefined = State#state.compactor_pid,
  
      Buckets =
          case Buckets0 of
              [] ->
                  lists:map(fun list_to_binary/1,
                            ns_bucket:node_bucket_names_of_type(node(), membase));
              _ ->
                  Buckets0
          end,
  
      case Buckets of
          [] ->
              ?log_debug("No buckets to compact. Rescheduling compaction."),
              {next_state, idle, State#state{scheduler =
                                                 compaction_scheduler:schedule_next(Scheduler)}};
          _ ->
              ?log_debug("Starting compaction for the following buckets: ~n~p",
                         [Buckets]),
              NewState1 = State#state{buckets_to_compact=Buckets,
                                      scheduler=
                                          compaction_scheduler:start_now(Scheduler)},
              NewState = compact_next_bucket(NewState1),
              {next_state, compacting, NewState}
      end;
  handle_info({'DOWN', MRef, _, _, _}, StateName, #state{view_compaction_inhibited_ref=MRef,
                                                         view_compaction_inhibited_bucket=BinBucket}=State) ->
      ?log_debug("Looks like vbucket mover inhibiting view compaction for for bucket \"~s\" is dead."
                 " Canceling inhibition", [BinBucket]),
      {next_state, StateName, State#state{view_compaction_inhibited_ref = undefined,
                                          view_compaction_inhibited_bucket = undefined}};
  handle_info({'EXIT', Compactor, Reason}, compacting,
              #state{compactor_pid=Compactor,
                     buckets_to_compact=Buckets,
                     scheduler=Scheduler} = State) ->
      true = Buckets =/= [],
  
      case Reason of
          normal ->
              ok;
          shutdown ->
              ?log_debug("Compactor ~p exited with reason ~p", [Compactor, Reason]);
          _ ->
              ?log_error("Compactor ~p exited unexpectedly: ~p. "
                         "Moving to the next bucket.", [Compactor, Reason])
      end,
  
      {DoneBucket, NewBuckets} =
          case Reason of
              shutdown ->
                  %% this can happen either on config change or if compaction
                  %% has been stopped due to end of allowed time period; in the
                  %% latter case there's no need to compact bucket again; but it
                  %% won't hurt: compaction shall be terminated rightaway with
                  %% reason normal
                  {undefined, Buckets};
              _ ->
                  {hd(Buckets), tl(Buckets)}
          end,
  
      NewState0 = case (DoneBucket =:= State#state.view_compaction_inhibited_bucket
                        andalso State#state.view_compaction_uninhibit_started) of
                      true ->
                          gen_server:reply(State#state.view_compaction_uninhibit_requested_waiter, ok),
                          State#state{view_compaction_inhibited_bucket = undefined,
                                      view_compaction_uninhibit_requested_waiter = undefined,
                                      view_compaction_uninhibit_requested = false,
                                      view_compaction_uninhibit_started = false};
                      _ ->
                          State
                  end,
  
      NewState1 = NewState0#state{compactor_pid=undefined,
                                  compactor_config=undefined,
                                  buckets_to_compact=NewBuckets},
      case NewBuckets of
          [] ->
              ?log_debug("Finished compaction iteration."),
              {next_state, idle, NewState1#state{scheduler =
                                                     compaction_scheduler:schedule_next(Scheduler)}};
          _ ->
              {next_state, compacting, compact_next_bucket(NewState1)}
      end;
  handle_info({'EXIT', Pid, Reason}, StateName,
              #state{running_forced_compactions=Compactions} = State) ->
      case dict:find(Pid, Compactions) of
          {ok, #forced_compaction{type=Type, name=Name}} ->
  
negative.code:                 lists:member(node(), ns_bucket:bucket_nodes(Config));
          not_present ->
              false
      end.
  
  exit_if_no_bucket(BucketName) ->
      case bucket_exists(BucketName) of
          true ->
              ok;
          false ->
              exit(normal)
      end.
  
  -spec register_forced_compaction(pid(), #forced_compaction{},
  