anchor.diff:  -export([start_link/2, start_link/1]).
   
   %% Callbacks
  --export([init/1, code_change/4, handle_event/3, handle_info/3,
  -         handle_sync_event/4, terminate/3]).
  +-export([init/1,
  +         code_change/4,
  +         handle_event/3,
  +         handle_info/3,
  +         handle_sync_event/4,
  +         terminate/3]).
   
   %% States
  --export([execute_op/3, finish_op/3, prepare/2,
  -         receive_prepared/2, committing/3, receive_committed/2, abort/2,
  +-export([execute_op/3,
  +         finish_op/3,
  +         prepare/2,
  +         receive_prepared/2,
  +         committing/3,
  +         receive_committed/2,
  +         abort/2,
            reply_to_client/2]).
   
   %%---------------------------------------------------------------------
  
positive.code:                     {reply, {ok, ReadResult}, execute_op, SD0}
              end;
          update ->
              {Key, Type, Param}=Args,
              Preflist = log_utilities:get_preflist_from_key(Key),
              IndexNode = hd(Preflist),
              case clocksi_downstream:generate_downstream_op(Transaction, IndexNode, Key, Type, Param) of
                  {ok, DownstreamRecord} ->
                      case clocksi_vnode:update_data_item(IndexNode, Transaction,
                                                  Key, Type, DownstreamRecord) of
                          ok ->
                              case lists:member(IndexNode, Updated_partitions) of
                                  false ->
                                      New_updated_partitions=
                                          [IndexNode | Updated_partitions],
                                      {reply, ok, execute_op,
                                      SD0#state
                                      {updated_partitions= New_updated_partitions}};
                                  true->
                                      {reply, ok, execute_op, SD0}
                              end;
  
negative.code: 
  %% @doc after receiving all prepare_times, send the commit message to all
  %%       updated partitions, and go to the "receive_committed" state.
  committing(commit, Sender, SD0=#state{transaction=Transaction,
                                        updated_partitions=UpdatedPartitions,
                                        commit_time=CommitTime}) ->
      NumToAck = length(UpdatedPartitions),
      case NumToAck of
          0 ->
              {next_state, reply_to_client, SD0#state{state=committed, from=Sender}, 0};
          _ ->
              clocksi_vnode:commit(UpdatedPartitions, Transaction, CommitTime),
              {next_state, receive_committed, SD0#state{num_to_ack=NumToAck, from=Sender, state=committing}}
      end.
  
  %% @doc the fsm waits for acks indicating that each partition has successfully
  %%	committed the tx and finishes operation.
  %%      Should we retry sending the committed message if we don't receive a
  %%      reply from every partition?
  %%      What delivery guarantees does sending messages provide?
  receive_committed(committed, S0=#state{num_to_ack=NumToAck}) ->
      case NumToAck of
          1 ->
              {next_state, reply_to_client, S0#state{state=committed}, 0};
          _ ->
              {next_state, receive_committed, S0#state{num_to_ack=NumToAck-1}}
      end.
  
  %% @doc When an error occurs or an updated partition
  %%      does not pass the certification check, the transaction aborts.
  abort(timeout, SD0=#state{transaction=Transaction,
                            updated_partitions=UpdatedPartitions}) ->
      clocksi_vnode:abort(UpdatedPartitions, Transaction),
      {next_state, reply_to_client, SD0#state{state=aborted},0};
  
  abort(abort, SD0=#state{transaction=Transaction,
                          updated_partitions=UpdatedPartitions}) ->
      clocksi_vnode:abort(UpdatedPartitions, Transaction),
      {next_state, reply_to_client, SD0#state{state=aborted},0}.
  
  %% @doc when the transaction has committed or aborted,
  %%       a reply is sent to the client that started the transaction.
  reply_to_client(timeout, SD=#state{from=From, transaction=Transaction, state=TxState, commit_time=CommitTime}) ->
      case undefined =/= From of
          true ->
              TxId = Transaction#transaction.txn_id,
              Reply = case TxState of
                  committed ->
                      DcId = dc_utilities:get_my_dc_id(),
                      CausalClock = vectorclock:set_clock_of_dc(DcId, CommitTime, Transaction#transaction.vec_snapshot_time),
                      {ok, {TxId, CausalClock}};
                  aborted ->
                      {aborted, TxId};
                  Reason ->
                      {TxId, Reason}
              end,
              gen_fsm:reply(From, Reply);
          false ->
              ok
      end,
      {stop, normal, SD}.
  
  %% =============================================================================
  
  