/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You 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.
 */
package org.apache.geode.internal.cache;

import static org.apache.geode.internal.offheap.annotations.OffHeapIdentifier.TX_ENTRY_STATE;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.DataSerializer;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.cache.CacheRuntimeException;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.CommitConflictException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryDestroyedException;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.EntryNotFoundException;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.lang.StringUtils;
import org.apache.geode.internal.offheap.OffHeapHelper;
import org.apache.geode.internal.offheap.Releasable;
import org.apache.geode.internal.offheap.StoredObject;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.offheap.annotations.Unretained;
import org.apache.geode.internal.serialization.DataSerializableFixedID;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.pdx.PdxSerializationException;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * TXEntryState is the entity that tracks transactional changes, except for those tracked by
 * {@link TXEntryUserAttrState}, to an entry.
 *
 * @since GemFire 4.0
 */
public class TXEntryState implements Releasable {
  private static final Logger logger = LogService.getLogger();

  /**
   * This field is final except for when it is nulled out during cleanup
   */
  @Retained(TX_ENTRY_STATE)
  private Object originalVersionId;
  private final Object originalValue;

  /**
   * Serial number that is set each time this entry is modified. Used to order the events in a
   * TransactionEvent.
   */
  protected int modSerialNum;
  /**
   * Used to remember the event id to use on the farSide for this entry. See bug 39434.
   *
   * @since GemFire 5.7
   */
  private int farSideEventOffset = -1;
  /**
   * Used to remember the event id to use on the nearSide for this entry. See bug 39434.
   *
   * @since GemFire 5.7
   */
  private int nearSideEventOffset = -1;

  private Object pendingValue;

  private byte[] serializedPendingValue;

  private EntryEventImpl pendingCallback;

  /**
   * Remember the callback argument for listener invocation
   */
  private Object callBackArgument;

  private byte op;

  /**
   * destroy field remembers the strongest destroy op perfomed on this entry
   */
  private byte destroy; // DESTROY_NONE, DESTROY_LOCAL, DESTROY_DISTRIBUTED

  // ORDER of the following is important to the implementation!
  private static final byte DESTROY_NONE = 0;

  private static final byte DESTROY_LOCAL = 1;

  private static final byte DESTROY_DISTRIBUTED = 2;

  // ORDER of the following is important to the implementation!
  private static final byte OP_NULL = 0;

  private static final byte OP_L_DESTROY = 1;

  private static final byte OP_CREATE_LD = 2;

  private static final byte OP_LLOAD_CREATE_LD = 3;

  private static final byte OP_NLOAD_CREATE_LD = 4;

  private static final byte OP_PUT_LD = 5;

  private static final byte OP_LLOAD_PUT_LD = 6;

  private static final byte OP_NLOAD_PUT_LD = 7;

  private static final byte OP_D_INVALIDATE_LD = 8;

  private static final byte OP_D_DESTROY = 9;

  private static final byte OP_L_INVALIDATE = 10;

  private static final byte OP_PUT_LI = 11;

  private static final byte OP_LLOAD_PUT_LI = 12;

  private static final byte OP_NLOAD_PUT_LI = 13;

  private static final byte OP_D_INVALIDATE = 14;

  private static final byte OP_CREATE_LI = 15;

  private static final byte OP_LLOAD_CREATE_LI = 16;

  private static final byte OP_NLOAD_CREATE_LI = 17;

  private static final byte OP_CREATE = 18;

  private static final byte OP_SEARCH_CREATE = 19;

  private static final byte OP_LLOAD_CREATE = 20;

  private static final byte OP_NLOAD_CREATE = 21;

  private static final byte OP_LOCAL_CREATE = 22;

  private static final byte OP_PUT = 23;

  static byte getOpCreate() {
    return OP_CREATE;
  }

  static byte getOpSearchCreate() {
    return OP_SEARCH_CREATE;
  }

  static byte getOpLloadCreate() {
    return OP_LLOAD_CREATE;
  }

  static byte getOpNloadCreate() {
    return OP_NLOAD_CREATE;
  }

  static byte getOpPut() {
    return OP_PUT;
  }

  static byte getOpSearchPut() {
    return OP_SEARCH_PUT;
  }

  static byte getOpLloadPut() {
    return OP_LLOAD_PUT;
  }

  static byte getOpNloadPut() {
    return OP_NLOAD_PUT;
  }

  private static final byte OP_SEARCH_PUT = 24;

  private static final byte OP_LLOAD_PUT = 25;

  private static final byte OP_NLOAD_PUT = 26;

  /**
   * System property to be set when read conflicts should be detected. Benefits of read conflict
   * detection are at: https://wiki.gemstone.com/display/PR/Read+conflict+detection
   */
  private static final boolean DETECT_READ_CONFLICTS =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "detectReadConflicts");

  // TODO: optimize footprint by having this field on a subclass
  // that is only created by TXRegionState when it knows its region needs refCounts.
  /**
   * A reference to the RegionEntry, in committed state, that this tx entry has referenced. Note:
   * this field is only needed if the committed region has eviction or expiration enabled. In both
   * those cases the tx needs to do reference counting.
   */
  private final RegionEntry refCountEntry;

  /**
   * Set to true once this entry has been written by a tx operation.
   */
  private boolean dirty;

  /**
   * Set to true if this operation is result of a bulk op. We use this boolean to determine op type
   * rather than extending the operation algebra.
   */
  private boolean bulkOp;

  private FilterRoutingInfo filterRoutingInfo = null;
  private Set<InternalDistributedMember> adjunctRecipients = null;

  /**
   * versionTag to be sent to other members. This is propogated up from AbstractRegionMap and then
   * used while building TXCommitMessage
   */
  private VersionTag versionTag = null;

  /**
   * tailKey (used by wan) to be sent to other members. This is propagated up from AbstractRegionMap
   * and then used while building TXCommitMessage
   */
  private long tailKey = -1;

  /**
   * versionTag that is fetched from remote members, if this member's data policy is not REPLICATE
   */
  private VersionTag remoteVersionTag = null;

  /**
   * Next region version generated on the primary
   */
  private final long nextRegionVersion = -1;

  /*
   * For Distributed Transaction. THis value is set when applying commit
   */
  private transient DistTxThinEntryState distTxThinEntryState;

  /**
   * Use this system property if you need to display/log string values in conflict messages
   */
  private static final boolean VERBOSE_CONFLICT_STRING =
      Boolean.getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "verboseConflictString");

  /**
   * This constructor is used to create a singleton used by LocalRegion to signal that noop
   * invalidate op has been performed. The instance returned by this constructor is just a marker;
   * it is not good for anything else.
   */
  protected TXEntryState() {
    op = OP_NULL;
    originalVersionId = Token.REMOVED_PHASE1;
    originalValue = Token.REMOVED_PHASE1;
    refCountEntry = null;
  }

  private TXRegionState txRegionState = null;

  /**
   * This constructor is used when creating an entry
   */
  protected TXEntryState(RegionEntry re, Object pvId, Object pv, TXRegionState txRegionState,
      boolean isDistributed) {
    Object vId = pvId;
    if (vId == null) {
      vId = Token.REMOVED_PHASE1;
    }
    if (pv == null) {
      pv = Token.REMOVED_PHASE1;
    }
    op = OP_NULL;
    pendingValue = pv;
    originalVersionId = vId;
    originalValue = pv;
    if (txRegionState.needsRefCounts()) {
      refCountEntry = re;
      if (re != null) {
        re.incRefCount();
      }
    } else {
      refCountEntry = null;
    }
    this.txRegionState = txRegionState;
    if (isDistributed) {
      distTxThinEntryState = new DistTxThinEntryState();
    }
  }

  public byte[] getSerializedPendingValue() {
    return serializedPendingValue;
  }

  private void setSerializedPendingValue(byte[] serializedPendingValue) {
    this.serializedPendingValue = serializedPendingValue;
  }

  void serializePendingValue() {
    Object pendingValue = getPendingValue();
    if (Token.isInvalidOrRemoved(pendingValue)) {
      // token
      return;
    }
    if (pendingValue instanceof byte[]) {
      // byte[]
      return;
    }

    // CachedDeserialized, Object or PDX
    if (pendingValue instanceof CachedDeserializable) {
      CachedDeserializable cachedDeserializable = (CachedDeserializable) pendingValue;
      setSerializedPendingValue(cachedDeserializable.getSerializedValue());
    } else {
      setSerializedPendingValue(EntryEventImpl.serialize(pendingValue));
    }
  }

  public TXRegionState getTXRegionState() {
    return txRegionState;
  }

  public Object getOriginalVersionId() {
    return originalVersionId;
  }

  public Object getOriginalValue() {
    return originalValue;
  }

  public Object getPendingValue() {
    return pendingValue;
  }

  public Object getCallbackArgument() {
    return callBackArgument;
  }

  /**
   * Gets the pending value for near side operations. Special cases local destroy and local
   * invalidate to fix bug 34387.
   */
  @Unretained
  public Object getNearSidePendingValue() {
    if (isOpLocalDestroy()) {
      return null;
    } else if (isOpLocalInvalidate()) {
      return Token.LOCAL_INVALID;
    } else {
      return getPendingValue();
    }
  }

  void setPendingValue(Object pv) {
    pendingValue = pv;
  }

  void setCallbackArgument(Object callbackArgument) {
    callBackArgument = callbackArgument;
  }

  /**
   * Fetches the current modification serial number from the txState and puts it in this entry.
   *
   * @param modNum the next modified serial number gotten from TXState
   *
   * @since GemFire 5.0
   */
  public void updateForWrite(final int modNum) {
    dirty = true;
    modSerialNum = modNum;
  }

  /**
   * Returns true if this entry has been written; false if only read
   *
   * @since GemFire 5.1
   */
  public boolean isDirty() {
    return DETECT_READ_CONFLICTS || dirty;
  }

  /**
   * Return true if entry has an operation.
   *
   * @since GemFire 5.5
   */
  public boolean hasOp() {
    return op != OP_NULL;
  }

  /**
   * Returns true if the transaction state has this entry existing "locally". Returns false if the
   * transaction is going to remove this entry.
   */
  public boolean existsLocally() {
    if (op == OP_NULL) {
      return !Token.isRemoved(getOriginalValue());
    } else {
      return op > OP_D_DESTROY;
    }
  }

  private boolean isOpLocalDestroy() {
    return op >= OP_L_DESTROY && op <= OP_D_INVALIDATE_LD;
  }

  private boolean isOpLocalInvalidate() {
    return op >= OP_L_INVALIDATE && op <= OP_NLOAD_CREATE_LI
        && op != OP_D_INVALIDATE;
  }

  /**
   * Return true if this transaction has completely invalidated or destroyed the value of this entry
   * in the entire distributed system. Return false if a netsearch should be done.
   */
  public boolean noValueInSystem() {
    if (op == OP_D_DESTROY || op == OP_D_INVALIDATE_LD || op == OP_D_INVALIDATE) {
      return true;
    } else if (getNearSidePendingValue() == Token.INVALID) {
      // Note that we are not interested in LOCAL_INVALID
      return (op >= OP_CREATE_LD && op != OP_L_INVALIDATE && op != OP_SEARCH_CREATE
          && op != OP_LOCAL_CREATE && op != OP_SEARCH_PUT);
    } else {
      return false;
    }
  }

  /**
   * Returns true if the transaction state has this entry existing locally and has a valid local
   * value. Returns false if the transaction is going to remove this entry or its local value is
   * invalid.
   */
  public boolean isLocallyValid(boolean isProxy) {
    if (op == OP_NULL) {
      if (isProxy) {
        // If it is a proxy that consider it locally valid
        // since we don't have any local committed state
        return true;
      } else {
        return !Token.isInvalidOrRemoved(getOriginalValue());
      }
    } else {
      return op >= OP_CREATE && !Token.isInvalid(getNearSidePendingValue());
    }
  }

  @Retained
  public Object getValueInVM(Object key) throws EntryNotFoundException {
    if (!existsLocally()) {
      throw new EntryNotFoundException(String.valueOf(key));
    }
    return getNearSidePendingValue();
  }

  /**
   * @return the value, or null if the value does not exist in the cache, Token.INVALID or
   *         Token.LOCAL_INVALID if the value is invalid
   */
  public Object getValue(Object key, Region r, boolean preferCD) {
    if (!existsLocally()) {
      return null;
    }
    Object v = getNearSidePendingValue();
    if (v instanceof CachedDeserializable && !preferCD) {
      // The only reason we would is if we do a read
      // in a transaction and the initial op that created the TXEntryState
      // did not call getDeserializedValue.
      // If a write op is done then the pendingValue does not belong to
      // the cache yet so we do not need the RegionEntry.
      // If we do need the RegionEntry then TXEntryState
      // will need to be changed to remember the RegionEntry it is created with.
      v = ((CachedDeserializable) v).getDeserializedValue(r, refCountEntry);
    }

    return v;
  }

  private boolean isOpCreate() {
    return op >= OP_CREATE_LI && op <= OP_LOCAL_CREATE;
  }

  boolean isOpCreateEvent() {
    return isOpCreate();
  }

  private boolean isOpPut() {
    return op >= OP_PUT;
  }

  protected boolean isOpPutEvent() {
    return isOpPut();
  }

  private boolean isOpInvalidate() {
    // Note that OP_CREATE_LI, OP_LLOAD_CREATE_LI, and OP_NLOAD_CREATE_LI
    // do not return true here because they are actually creates
    // with a value of LOCAL_INVALID locally and some other value remotely.
    return op <= OP_D_INVALIDATE && op >= OP_L_INVALIDATE;
  }

  boolean isOpInvalidateEvent() {
    return isOpInvalidate();
  }

  private boolean isOpDestroy() {
    return op <= OP_D_DESTROY && op >= OP_L_DESTROY;
  }

  boolean isOpDestroyEvent(InternalRegion r) {
    // Note that if the region is a proxy then we go ahead and distributed
    // the destroy because we can't eliminate it based on committed state
    return isOpDestroy()
        && (r.isProxy() || (getOriginalValue() != null && !Token.isRemoved(getOriginalValue())));
  }

  /**
   * Returns true if this operation has an event for the tx listener
   *
   * @since GemFire 5.0
   */
  boolean isOpAnyEvent(InternalRegion r) {
    return isOpPutEvent() || isOpCreateEvent() || isOpInvalidateEvent() || isOpDestroyEvent(r);
  }

  boolean isOpSearch() {
    return op == OP_SEARCH_CREATE || op == OP_SEARCH_PUT;
  }

  String opToString() {
    return opToString(op);
  }

  private String opToString(byte opCode) {
    switch (opCode) {
      case OP_NULL:
        return "OP_NULL";
      case OP_L_DESTROY:
        return "OP_L_DESTROY";
      case OP_CREATE_LD:
        return "OP_CREATE_LD";
      case OP_LLOAD_CREATE_LD:
        return "OP_LLOAD_CREATE_LD";
      case OP_NLOAD_CREATE_LD:
        return "OP_NLOAD_CREATE_LD";
      case OP_PUT_LD:
        return "OP_PUT_LD";
      case OP_LLOAD_PUT_LD:
        return "OP_LLOAD_PUT_LD";
      case OP_NLOAD_PUT_LD:
        return "OP_NLOAD_PUT_LD";
      case OP_D_INVALIDATE_LD:
        return "OP_D_INVALIDATE_LD";
      case OP_D_DESTROY:
        return "OP_D_DESTROY";
      case OP_L_INVALIDATE:
        return "OP_L_INVALIDATE";
      case OP_PUT_LI:
        return "OP_PUT_LI";
      case OP_LLOAD_PUT_LI:
        return "OP_LLOAD_PUT_LI";
      case OP_NLOAD_PUT_LI:
        return "OP_NLOAD_PUT_LI";
      case OP_D_INVALIDATE:
        return "OP_D_INVALIDATE";
      case OP_CREATE_LI:
        return "OP_CREATE_LI";
      case OP_LLOAD_CREATE_LI:
        return "OP_LLOAD_CREATE_LI";
      case OP_NLOAD_CREATE_LI:
        return "OP_NLOAD_CREATE_LI";
      case OP_CREATE:
        return "OP_CREATE";
      case OP_SEARCH_CREATE:
        return "OP_SEARCH_CREATE";
      case OP_LLOAD_CREATE:
        return "OP_LLOAD_CREATE";
      case OP_NLOAD_CREATE:
        return "OP_NLOAD_CREATE";
      case OP_LOCAL_CREATE:
        return "OP_LOCAL_CREATE";
      case OP_PUT:
        return "OP_PUT";
      case OP_SEARCH_PUT:
        return "OP_SEARCH_PUT";
      case OP_LLOAD_PUT:
        return "OP_LLOAD_PUT";
      case OP_NLOAD_PUT:
        return "OP_NLOAD_PUT";
      default:
        return "<unhandled op " + opCode + " >";
    }
  }

  /**
   * Returns an Operation instance that matches what the transactional operation done on this entry
   * in the cache the the transaction was performed in.
   */
  protected Operation getNearSideOperation() {
    switch (op) {
      case OP_NULL:
        return null;
      case OP_L_DESTROY:
        return Operation.LOCAL_DESTROY;
      case OP_CREATE_LD:
        return Operation.LOCAL_DESTROY;
      case OP_LLOAD_CREATE_LD:
        return Operation.LOCAL_DESTROY;
      case OP_NLOAD_CREATE_LD:
        return Operation.LOCAL_DESTROY;
      case OP_PUT_LD:
        return Operation.LOCAL_DESTROY;
      case OP_LLOAD_PUT_LD:
        return Operation.LOCAL_DESTROY;
      case OP_NLOAD_PUT_LD:
        return Operation.LOCAL_DESTROY;
      case OP_D_INVALIDATE_LD:
        return Operation.LOCAL_DESTROY;
      case OP_D_DESTROY:
        return getDestroyOperation();
      case OP_L_INVALIDATE:
        return Operation.LOCAL_INVALIDATE;
      case OP_PUT_LI:
        return Operation.LOCAL_INVALIDATE;
      case OP_LLOAD_PUT_LI:
        return Operation.LOCAL_INVALIDATE;
      case OP_NLOAD_PUT_LI:
        return Operation.LOCAL_INVALIDATE;
      case OP_D_INVALIDATE:
        return Operation.INVALIDATE;
      case OP_CREATE_LI:
        return getCreateOperation();
      case OP_LLOAD_CREATE_LI:
        return getCreateOperation();
      case OP_NLOAD_CREATE_LI:
        return getCreateOperation();
      case OP_CREATE:
        return getCreateOperation();
      case OP_SEARCH_CREATE:
        return Operation.SEARCH_CREATE;
      case OP_LLOAD_CREATE:
        return Operation.LOCAL_LOAD_CREATE;
      case OP_NLOAD_CREATE:
        return Operation.NET_LOAD_CREATE;
      case OP_LOCAL_CREATE:
        return getCreateOperation();
      case OP_PUT:
        return getUpdateOperation();
      case OP_SEARCH_PUT:
        return Operation.SEARCH_UPDATE;
      case OP_LLOAD_PUT:
        return Operation.LOCAL_LOAD_UPDATE;
      case OP_NLOAD_PUT:
        return Operation.NET_LOAD_CREATE;
      default:
        throw new IllegalStateException(
            String.format("<unhandled op %s >", op));
    }
  }

  /**
   * @return true when the operation is the result of a bulk op
   */
  private boolean isBulkOp() {
    return bulkOp;
  }

  /**
   * Calculate and return the event offset based on the sequence id on TXState.
   *
   * @since GemFire 5.7
   */
  private static int generateEventOffset(TXState txState) {
    long seqId = EventID.reserveSequenceId();
    return (int) (seqId - txState.getBaseSequenceId());
  }

  /**
   * Generate offsets for different eventIds; one for nearside and one for farside for the ops for
   * this entry.
   *
   * @since GemFire 5.7
   */
  private void generateBothEventOffsets(TXState txState) {
    assert farSideEventOffset == -1;
    farSideEventOffset = generateEventOffset(txState);
    generateNearSideEventOffset(txState);
  }

  /**
   * Generate the offset for an eventId that will be used for both a farside and nearside op for
   * this entry.
   *
   * @since GemFire 5.7
   */
  private void generateSharedEventOffset(TXState txState) {
    assert farSideEventOffset == -1;
    generateNearSideEventOffset(txState);
    farSideEventOffset = nearSideEventOffset;
  }

  /**
   * Generate the offset for an eventId that will be used for the nearside op for this entry. No
   * farside op will be done.
   *
   * @since GemFire 5.7
   */
  private void generateNearSideOnlyEventOffset(TXState txState) {
    generateNearSideEventOffset(txState);
  }

  private void generateNearSideEventOffset(TXState txState) {
    assert nearSideEventOffset == -1;
    nearSideEventOffset = generateEventOffset(txState);
  }

  private int getFarSideEventOffset() {
    assert nearSideEventOffset != -1;
    return nearSideEventOffset;
  }

  private static EventID createEventID(TXState txState, int offset) {
    return new EventID(txState.getBaseMembershipId(), txState.getBaseThreadId(),
        txState.getBaseSequenceId() + offset);
  }

  /**
   * Calculate (if farside has not already done so) and return then eventID to use for near side op
   * applications.
   *
   * @since GemFire 5.7
   */
  private EventID getNearSideEventId(TXState txState) {
    assert nearSideEventOffset != -1;
    return createEventID(txState, nearSideEventOffset);
  }

  /**
   * Calculate and return the event offset for this entry's farSide operation.
   *
   * @since GemFire 5.7
   */
  void generateEventOffsets(TXState txState) {
    switch (op) {
      case OP_NULL:
        // no eventIds needed
        break;
      case OP_L_DESTROY:
        generateNearSideOnlyEventOffset(txState);
        break;
      case OP_CREATE_LD:
        generateBothEventOffsets(txState);
        break;
      case OP_LLOAD_CREATE_LD:
        generateBothEventOffsets(txState);
        break;
      case OP_NLOAD_CREATE_LD:
        generateBothEventOffsets(txState);
        break;
      case OP_PUT_LD:
        generateBothEventOffsets(txState);
        break;
      case OP_LLOAD_PUT_LD:
        generateBothEventOffsets(txState);
        break;
      case OP_NLOAD_PUT_LD:
        generateBothEventOffsets(txState);
        break;
      case OP_D_INVALIDATE_LD:
        generateBothEventOffsets(txState);
        break;
      case OP_D_DESTROY:
        generateSharedEventOffset(txState);
        break;
      case OP_L_INVALIDATE:
        generateNearSideOnlyEventOffset(txState);
        break;
      case OP_PUT_LI:
        generateBothEventOffsets(txState);
        break;
      case OP_LLOAD_PUT_LI:
        generateBothEventOffsets(txState);
        break;
      case OP_NLOAD_PUT_LI:
        generateBothEventOffsets(txState);
        break;
      case OP_D_INVALIDATE:
        generateSharedEventOffset(txState);
        break;
      case OP_CREATE_LI:
        generateBothEventOffsets(txState);
        break;
      case OP_LLOAD_CREATE_LI:
        generateBothEventOffsets(txState);
        break;
      case OP_NLOAD_CREATE_LI:
        generateBothEventOffsets(txState);
        break;
      case OP_CREATE:
        generateSharedEventOffset(txState);
        break;
      case OP_SEARCH_CREATE:
        generateNearSideOnlyEventOffset(txState);
        break;
      case OP_LLOAD_CREATE:
        generateSharedEventOffset(txState);
        break;
      case OP_NLOAD_CREATE:
        generateSharedEventOffset(txState);
        break;
      case OP_LOCAL_CREATE:
        generateNearSideOnlyEventOffset(txState);
        break;
      case OP_PUT:
        generateSharedEventOffset(txState);
        break;
      case OP_SEARCH_PUT:
        generateNearSideOnlyEventOffset(txState);
        break;
      case OP_LLOAD_PUT:
        generateSharedEventOffset(txState);
        break;
      case OP_NLOAD_PUT:
        generateSharedEventOffset(txState);
        break;
      default:
        throw new IllegalStateException("<unhandled op " + op + " >");
    }
  }

  /**
   * Gets the operation code for the operation done on this entry in caches remote from the
   * originator of the tx (i.e. the "far side").
   *
   * @return null if no far side operation
   */
  private Operation getFarSideOperation() {
    switch (op) {
      case OP_NULL:
        return null;
      case OP_L_DESTROY:
        return null;
      case OP_CREATE_LD:
        return getCreateOperation();
      case OP_LLOAD_CREATE_LD:
        return Operation.LOCAL_LOAD_CREATE;
      case OP_NLOAD_CREATE_LD:
        return Operation.NET_LOAD_CREATE;
      case OP_PUT_LD:
        return getUpdateOperation();
      case OP_LLOAD_PUT_LD:
        return Operation.LOCAL_LOAD_UPDATE;
      case OP_NLOAD_PUT_LD:
        return Operation.NET_LOAD_UPDATE;
      case OP_D_INVALIDATE_LD:
        return Operation.INVALIDATE;
      case OP_D_DESTROY:
        return getDestroyOperation();
      case OP_L_INVALIDATE:
        return null;
      case OP_PUT_LI:
        return getUpdateOperation();
      case OP_LLOAD_PUT_LI:
        return Operation.LOCAL_LOAD_UPDATE;
      case OP_NLOAD_PUT_LI:
        return Operation.NET_LOAD_UPDATE;
      case OP_D_INVALIDATE:
        return Operation.INVALIDATE;
      case OP_CREATE_LI:
        return getCreateOperation();
      case OP_LLOAD_CREATE_LI:
        return Operation.LOCAL_LOAD_CREATE;
      case OP_NLOAD_CREATE_LI:
        return Operation.NET_LOAD_CREATE;
      case OP_CREATE:
        return getCreateOperation();
      case OP_SEARCH_CREATE:
        return null;
      case OP_LLOAD_CREATE:
        return Operation.LOCAL_LOAD_CREATE;
      case OP_NLOAD_CREATE:
        return Operation.NET_LOAD_CREATE;
      case OP_LOCAL_CREATE:
        return getCreateOperation();
      case OP_PUT:
        return getUpdateOperation();
      case OP_SEARCH_PUT:
        return null;
      case OP_LLOAD_PUT:
        return Operation.LOCAL_LOAD_UPDATE;
      case OP_NLOAD_PUT:
        return Operation.NET_LOAD_UPDATE;
      default:
        throw new IllegalStateException(
            String.format("<unhandled op %s >", op));
    }
  }

  @Retained
  EntryEvent getEvent(InternalRegion r, Object key, TXState txs) {
    InternalRegion eventRegion = r;
    if (r.isUsedForPartitionedRegionBucket()) {
      eventRegion = r.getPartitionedRegion();
    }
    @Retained
    EntryEventImpl result = new TxEntryEventImpl(eventRegion, key);
    boolean returnedResult = false;
    try {
      if (destroy == DESTROY_NONE || isOpDestroy()) {
        result.setOldValue(getOriginalValue());
      }
      result.setOriginRemote(txs.isOriginRemoteForEvents());
      result.setTransactionId(txs.getTransactionId());
      returnedResult = true;
      return result;
    } finally {
      if (!returnedResult) {
        result.release();
      }
    }
  }

  /**
   * @return true if invalidate was done
   */
  public boolean invalidate(EntryEventImpl event) throws EntryNotFoundException {
    if (event.isLocalInvalid()) {
      performOp(adviseOp(OP_L_INVALIDATE, event), event);
    } else {
      performOp(adviseOp(OP_D_INVALIDATE, event), event);
    }

    return true;
  }

  /**
   * @return true if destroy was done
   */
  public boolean destroy(EntryEventImpl event, boolean cacheWrite, boolean originRemote)
      throws CacheWriterException, EntryNotFoundException, TimeoutException {
    InternalRegion internalRegion = event.getRegion();
    CacheWriter cWriter = internalRegion.basicGetWriter();
    byte advisedOp = adviseOp((cacheWrite ? OP_D_DESTROY : OP_L_DESTROY), event);
    if (cWriter != null && cacheWrite && !event.inhibitAllNotifications()) {
      boolean oldOriginRemote = event.isOriginRemote();
      if (event.hasClientOrigin() || originRemote) {
        event.setOriginRemote(true);
      }
      cWriter.beforeDestroy(event);
      event.setOriginRemote(oldOriginRemote);
    }

    if (advisedOp != OP_NULL) {
      bulkOp = event.getOperation().isRemoveAll();
      if (cacheWrite) {
        performOp(advisedOp, event);
        destroy = DESTROY_DISTRIBUTED;
      } else {
        performOp(advisedOp, event);
        if (destroy != DESTROY_DISTRIBUTED) {
          destroy = DESTROY_LOCAL;
        }
      }
    }

    return true;
  }

  /**
   * @param event the event object for this operation, with the exception that the oldValue
   *        parameter is not yet filled in. The oldValue will be filled in by this operation.
   *
   * @param ifNew true if this operation must not overwrite an existing key
   * @param originRemote value for cacheWriter's isOriginRemote flag
   * @return true if put was done; otherwise returns false
   */
  public boolean basicPut(EntryEventImpl event, boolean ifNew, boolean originRemote)
      throws CacheWriterException, TimeoutException {
    byte putOp = OP_CREATE;
    boolean doingCreate = true;
    if (!ifNew) {
      if (existsLocally()) {
        putOp = OP_PUT;
        doingCreate = false;
      }
    }
    if (doingCreate) {
      event.makeCreate();
    } else {
      event.makeUpdate();
    }
    if (event.isNetSearch()) {
      putOp += (byte) (OP_SEARCH_PUT - OP_PUT);
    } else if (event.isLocalLoad()) {
      putOp += (byte) (OP_LLOAD_PUT - OP_PUT);
    } else if (event.isNetLoad()) {
      putOp += (byte) (OP_NLOAD_PUT - OP_PUT);
    }
    bulkOp = event.getOperation().isPutAll();
    byte advisedOp = adviseOp(putOp, event);

    InternalRegion internalRegion = event.getRegion();
    if (!event.isNetSearch()) {
      CacheWriter cWriter = internalRegion.basicGetWriter();
      boolean oldOriginRemote = event.isOriginRemote();
      if (event.hasClientOrigin() || originRemote) {
        event.setOriginRemote(true);
      }
      if (cWriter != null && event.isGenerateCallbacks() && !event.inhibitAllNotifications()) {
        if (doingCreate) {
          cWriter.beforeCreate(event);
        } else {
          cWriter.beforeUpdate(event);
        }
      }
      event.setOriginRemote(oldOriginRemote);
    }

    performOp(advisedOp, event);

    if (putOp == OP_CREATE) {
      fetchRemoteVersionTag(event);
    }
    return true;
  }

  /**
   * We will try to establish TXState on members with dataPolicy REPLICATE, this is done for the
   * first region to be involved in a transaction. For subsequent region if the dataPolicy is not
   * REPLICATE, we fetch the VersionTag from replicate members.
   */
  private void fetchRemoteVersionTag(EntryEventImpl event) {
    if (event.getRegion() instanceof DistributedRegion) {
      DistributedRegion dr = (DistributedRegion) event.getRegion();
      if (dr.getDataPolicy() == DataPolicy.NORMAL || dr.getDataPolicy() == DataPolicy.PRELOADED) {
        VersionTag tag = null;
        try {
          tag = dr.fetchRemoteVersionTag(event.getKey());
        } catch (EntryNotFoundException e) {
          // ignore
        }
        if (tag != null) {
          setRemoteVersionTag(tag);
        }
      }
    }
  }

  /**
   * Perform operation algebra
   *
   * @return false if operation was not done
   */
  private byte adviseOp(byte requestedOpCode, EntryEventImpl event) {
    { // Set event old value
      Object oldVal;
      if (op == OP_NULL) {
        oldVal = getOriginalValue();
      } else {
        oldVal = getNearSidePendingValue();
      }

      Region region = event.getRegion();
      boolean needOldValue = region instanceof HARegion // fix for bug 37909
          || region instanceof BucketRegion;
      event.setTXEntryOldValue(oldVal, needOldValue);
    }

