anchor.diff:              %% In the future, we will handle this using msg_delay for
               %% outstanding txids to trigger a delete of the items.
       end,
  -    State;
  +    NewState;
   
   %% generated by on(register_TP) via msg_delay to not increase memory
   %% footprint
  
positive.code: on({tx_tm_rtm_delete, TxId, Decision} = Msg, State) ->
      ?TRACE("tx_tm_rtm:on({delete, ~p, ~p}) in ~p ~n",
             [TxId, Decision, state_get_role(State)]),
      %% TODO: use ok as error code?!
      %% {ok, TxState} = my_get_tx_entry(TxId, State),
      {ErrCode, TxState} = my_get_tx_entry(TxId, State),
      %% inform RTMs on delete
      Role = state_get_role(State),
      case {ErrCode, Role} of
          {ok, tx_tm} ->
              RTMs = tx_state:get_rtms(TxState),
              send_to_rtms(RTMs, fun(_X) -> Msg end),
              %% inform used learner to delete paxosids.
              AllPaxIds =
                  [ begin
                        {ok, ItemState} = my_get_item_entry(ItemId, State),
                        [ PaxId || {PaxId, _RTLog, _TP}
                          <- tx_item_state:get_paxosids_rtlogs_tps(ItemState) ]
                    end || {_TLogEntry, ItemId} <- tx_state:get_tlog_txitemids(TxState) ],
              %% We could delete immediately, but we still miss the
              %% minority of learner_decides, which would re-create the
              %% id in the learner, which then would have to be deleted
              %% separately, so we give the minority a second to arrive
              %% and then send the delete request.
              %% learner:stop_paxosids(GLLearner, lists:flatten(AllPaxIds)),
              GLLearner = state_get_gllearner(State),
              msg_delay:send_local(1, comm:make_local(GLLearner),
                                   {learner_deleteids, lists:flatten(AllPaxIds)}),
              DeleteIt = true;
          {ok, _} ->
              %% the test my_trigger_delete was passed, at least by the TM
              %% RTMs only wait for all tp register messages, to not miss them
              %% record, that every TP was informed and all paxids decided
  
              %% TODO: instead subscribing directly do a reference
              %% counting in the status field state_get_subs /
              %% state_set_subs
              fd:unsubscribe(tx_state:get_tm(TxState)),
              TmpTxState = tx_state:set_numinformed(
                             TxState, tx_state:get_numids(TxState) *
                                 config:read(replication_factor)),
              Tmp2TxState = tx_state:set_numpaxdecided(
                             TmpTxState, tx_state:get_numids(TxState) *
                                  config:read(replication_factor)),
              Tmp3TxState = tx_state:set_decided(Tmp2TxState, Decision),
              _ = my_set_entry(Tmp3TxState, State),
              DeleteIt = tx_state:all_tps_registered(TxState),
              %% inform used acceptors to delete paxosids.
              AllPaxIds =
                  [ begin
                    {ok, ItemState} = my_get_item_entry(ItemId, State),
                    [ PaxId || {PaxId, _RTlog, _TP}
                      <- tx_item_state:get_paxosids_rtlogs_tps(ItemState) ]
                end || {_TLogEntry, ItemId} <- tx_state:get_tlog_txitemids(TxState) ],
              LAcceptor = state_get_lacceptor(State),
  %%            msg_delay:send_local((config:read(tx_timeout) * 2) div 1000, LAcceptor,
  %%                                 {acceptor_deleteids, lists:flatten(AllPaxIds)});
              comm:send_local(LAcceptor,
                                 {acceptor_deleteids, lists:flatten(AllPaxIds)});
          {new, _} ->
              %% already deleted
              DeleteIt = false;
          {uninitialized, _} ->
              DeleteIt = false %% will be deleted when msg_delay triggers it
      end,
      case DeleteIt of
          false ->
              %% @TODO if we are a rtm, we still wait for register TPs
  
negative.code: state_to_on(State) ->
      {A, B, C, D, E, F, _G} = State,
      {A, B, C, D, E, F}.
  
  -spec state_subscribe(state(), comm:mypid()) -> state().
  state_subscribe(State, Pid) ->
      Subs = state_get_subs(State),
      NewSubs = case lists:keyfind(Pid, 1, Subs) of
                    false -> fd:subscribe(Pid),
                             [{Pid, 1} | Subs];
                    Tuple ->
                        NewVal = setelement(2, Tuple, element(2, Tuple) + 1),
                        lists:keyreplace(Pid, 1, Subs, NewVal)
                end,
      state_set_subs(State, NewSubs).
  
  -spec state_unsubscribe(state(), comm:mypid()) -> state().
  state_unsubscribe(State, Pid) ->
      Subs = state_get_subs(State),
      NewSubs = case lists:keyfind(Pid, 1, Subs) of
                    false -> Subs;
                    Tuple ->
                        case element(2, Tuple) of
                            1 ->
                                %% delay the actual unsubscribe for better perf.?
                                fd:unsubscribe(element(1, Tuple)),
                                lists:keydelete(Pid, 1, Subs);
                            Num ->
                                NewVal = setelement(2, Tuple, Num - 1),
                                lists:keyreplace(Pid, 1, Subs, NewVal)
                        end
                end,
      state_set_subs(State, NewSubs).
  
  %% @doc Checks whether config parameters for tx_tm_rtm exist and are
  %%      valid.
  