/*
 * 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.cache.LocalRegion.InitializationLevel.BEFORE_INITIAL_IMAGE;

import java.io.DataInput;
import java.io.DataOutput;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Set;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.DataSerializer;
import org.apache.geode.GemFireException;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.CacheEvent;
import org.apache.geode.cache.CacheLoader;
import org.apache.geode.cache.CacheLoaderException;
import org.apache.geode.cache.CacheWriter;
import org.apache.geode.cache.CacheWriterException;
import org.apache.geode.cache.DataPolicy;
import org.apache.geode.cache.EntryEvent;
import org.apache.geode.cache.LoaderHelper;
import org.apache.geode.cache.Operation;
import org.apache.geode.cache.RegionAttributes;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.RegionEvent;
import org.apache.geode.cache.Scope;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.cache.util.ObjectSizer;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.internal.ClusterDistributionManager;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.HighPriorityDistributionMessage;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.PooledDistributionMessage;
import org.apache.geode.distributed.internal.ProcessorKeeper21;
import org.apache.geode.distributed.internal.ReplyProcessor21;
import org.apache.geode.distributed.internal.SerialDistributionMessage;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.cache.LocalRegion.InitializationLevel;
import org.apache.geode.internal.cache.versions.DiskVersionTag;
import org.apache.geode.internal.cache.versions.VersionStamp;
import org.apache.geode.internal.cache.versions.VersionTag;
import org.apache.geode.internal.offheap.Releasable;
import org.apache.geode.internal.offheap.annotations.Released;
import org.apache.geode.internal.offheap.annotations.Retained;
import org.apache.geode.internal.serialization.DeserializationContext;
import org.apache.geode.internal.serialization.SerializationContext;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Implementation for distributed search, load and write operations in the GemFire system. Provides
 * an API for doing netSearch, netLoad, netSearchAndLoad and netWrite. The class uses the
 * DistributionAdvisor to route requests to the appropriate members. It also uses the
 * DistributionAdvisor to get region scope and applies rules based on the scope. It makes uses of
 * intelligent acceptors that allow netSearch to happen as a one phase operation at all
 * times.netLoad happens as a one phase operation in all cases except where the scope is GLOBAL At
 * the receiving end, the request is converted into an appropriate message whose process method
 * responds to the request.
 */
public class SearchLoadAndWriteProcessor implements MembershipListener {
  private static final Logger logger = LogService.getLogger();

  public static final int SMALL_BLOB_SIZE =
      Integer.getInteger("DistributionManager.OptimizedUpdateByteLimit", 2000);

  static final long RETRY_TIME =
      Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "search-retry-interval", 2000);

  private volatile InternalDistributedMember selectedNode;
  private boolean selectedNodeDead = false;
  private int timeout;
  private boolean netSearchDone = false;
  private boolean netLoadDone = false;
  private long lastModified = 0;
  protected int processorId;
  private Object aCallbackArgument;
  private String regionName;
  private Object key;
  protected LocalRegion region;
  private Object result = null;
  private boolean isSerialized = false; // is result serialized?
  private CacheDistributionAdvisor advisor = null;
  protected Exception remoteException = null;
  public DistributionManager distributionManager = null;
  private volatile boolean requestInProgress = false;
  private boolean remoteGetInProgress = false;
  private volatile boolean authorative = false;
  /** remoteLoadInProgress is volatile to make sure response threads see the current value */
  private volatile boolean remoteLoadInProgress = false;
  @MakeNotStatic
  private static final ProcessorKeeper21 processorKeeper = new ProcessorKeeper21(false);
  // private static Set availableAcceptHelperSet = new HashSet();
  /** The members that haven't replied yet */
  // changed pendingResponders to synchronized Set to fix bug 38972
  private final Set pendingResponders = Collections.synchronizedSet(new HashSet());
  private List responseQueue = null;
  private boolean netWriteSucceeded = false;
  private int remainingTimeout = 0;
  private long startTimeSnapShot = 0;
  private long endTimeSnapShot = 0;

  private boolean netSearch = false;
  private boolean netLoad = false;
  private boolean attemptedLocalLoad = false; // added for bug 39738
  private boolean localLoad = false;
  private boolean localWrite = false;
  private boolean netWrite = false;

  private final Object membersLock = new Object();

  private ArrayList<InternalDistributedMember> departedMembers;

  private Lock lock = null; // if non-null, then needs to be unlocked in release

  static final int NETSEARCH = 0;
  static final int NETLOAD = 1;
  static final int NETWRITE = 2;

  static final int BEFORECREATE = 0;
  static final int BEFOREDESTROY = 1;
  static final int BEFOREUPDATE = 2;
  static final int BEFOREREGIONDESTROY = 3;
  static final int BEFOREREGIONCLEAR = 4;


  /************** Public Methods ************************/

  Object doNetSearch() throws TimeoutException {

    resetResults();
    RegionAttributes attrs = region.getAttributes();
    requestInProgress = true;
    Scope scope = attrs.getScope();
    Assert.assertTrue(scope != Scope.LOCAL);
    netSearchForBlob();
    requestInProgress = false;
    return result;
  }



  void doSearchAndLoad(EntryEventImpl event, TXStateInterface txState, Object localValue,
      boolean preferCD) throws CacheLoaderException, TimeoutException {
    requestInProgress = true;
    RegionAttributes attrs = region.getAttributes();
    Scope scope = attrs.getScope();
    CacheLoader loader = region.basicGetLoader();
    if (scope.isLocal()) {
      Object obj = doLocalLoad(loader, false, preferCD);
      event.setNewValue(obj);
    } else {
      searchAndLoad(event, txState, localValue, preferCD);
    }
    requestInProgress = false;
    if (netSearch) {
      if (event.getOperation().isCreate()) {
        event.setOperation(Operation.SEARCH_CREATE);
      } else {
        event.setOperation(Operation.SEARCH_UPDATE);
      }
    } else if (netLoad) {
      if (event.getOperation().isCreate()) {
        event.setOperation(Operation.NET_LOAD_CREATE);
      } else {
        event.setOperation(Operation.NET_LOAD_UPDATE);
      }
    } else if (localLoad) {
      if (event.getOperation().isCreate()) {
        event.setOperation(Operation.LOCAL_LOAD_CREATE);
      } else {
        event.setOperation(Operation.LOCAL_LOAD_UPDATE);
      }
    }
  }

  /** return true if a CacheWriter was actually invoked */
  boolean doNetWrite(CacheEvent event, Set netWriteRecipients, CacheWriter localWriter, int paction)
      throws CacheWriterException, TimeoutException {

    int action = paction;
    requestInProgress = true;
    Scope scope = region.getScope();
    if (localWriter != null) {
      doLocalWrite(localWriter, event, action);
      requestInProgress = false;
      return true;
    }
    if (scope == Scope.LOCAL && (region.getPartitionAttributes() == null)) {
      return false;
    }
    @Released
    CacheEvent listenerEvent = getEventForListener(event);
    try {
      if (action == BEFOREUPDATE && listenerEvent.getOperation().isCreate()) {
        action = BEFORECREATE;
      }
      boolean cacheWrote = netWrite(listenerEvent, action, netWriteRecipients);
      requestInProgress = false;
      return cacheWrote;
    } finally {
      if (event != listenerEvent) {
        if (listenerEvent instanceof EntryEventImpl) {
          ((Releasable) listenerEvent).release();
        }
      }
    }
  }

  @Override
  public void memberJoined(DistributionManager distributionManager, InternalDistributedMember id) {
    // Ignore - if they just joined, they don't have what we want
  }

  @Override
  public void memberSuspect(DistributionManager distributionManager, InternalDistributedMember id,
      InternalDistributedMember whoSuspected, String reason) {}

  @Override
  public void quorumLost(DistributionManager distributionManager,
      Set<InternalDistributedMember> failures, List<InternalDistributedMember> remaining) {}

  @Override
  public void memberDeparted(DistributionManager distributionManager,
      final InternalDistributedMember id, final boolean crashed) {

    synchronized (membersLock) {
      pendingResponders.remove(id);
    }
    synchronized (this) {
      if (id.equals(selectedNode) && (requestInProgress) && (remoteGetInProgress)) {
        if (departedMembers == null) {
          departedMembers = new ArrayList<>();
        }
        departedMembers.add(id);
        selectedNode = null;
        selectedNodeDead = true;
        computeRemainingTimeout();
        if (logger.isDebugEnabled()) {
          logger.debug("{}: processing loss of member {}", this, id);
        }
        lastNotifySpot = 3;
        notifyAll(); // signal the waiter; we are not done; but we need the waiter to call
                     // sendNetSearchRequest
      }
      if (responseQueue != null) {
        responseQueue.remove(id);
      }
      checkIfDone();
    }
  }

  int getProcessorId() {
    return processorId;
  }

  synchronized void checkIfDone() {
    if (!remoteGetInProgress && pendingResponders.isEmpty()) {
      // Synchronize in case a different response/reply is still
      // in progress, and it's the one thats got the goods (bug 28741)
      signalDone();
    }


  }

  synchronized void signalDone() {
    requestInProgress = false;
    lastNotifySpot = 1;
    notifyAll();
  }

  synchronized void signalTimedOut() {
    lastNotifySpot = 2;
    notifyAll();
  }

  private volatile int lastNotifySpot = 0;

  private VersionTag versionTag;

  synchronized void nackResponseComplete() {
    /*
     * if (this.requestInProgress && currentHelper != null && optimizer != null &&
     * optimizer.allRepliesReceived(currentHelper.nackServerChannel)) { this.requestInProgress =
     * false; signalDone(); }
     */

  }

  static ProcessorKeeper21 getProcessorKeeper() {
    return processorKeeper;
  }

  boolean isNetSearch() {
    return netSearch;
  }

  boolean isNetLoad() {
    return netLoad;
  }

  boolean isLocalLoad() {
    return localLoad;
  }

  boolean isNetWrite() {
    return netWrite;
  }

  boolean isLocalWrite() {
    return localWrite;
  }

  long getLastModified() {
    return lastModified;
  }

  boolean resultIsSerialized() {
    return isSerialized;
  }

  static SearchLoadAndWriteProcessor getProcessor() {
    SearchLoadAndWriteProcessor processor = new SearchLoadAndWriteProcessor();
    processor.processorId = getProcessorKeeper().put(processor);
    return processor;
  }

  void release() {
    try {
      if (lock != null) {
        try {
          lock.unlock();
        } catch (CancelException ignore) {
        }
        lock = null;
      }
    } finally {
      try {
        if (advisor != null) {
          advisor.removeMembershipListener(this);
        }
      } catch (IllegalArgumentException ignore) {
      } finally {
        getProcessorKeeper().remove(processorId);
      }
    }
  }

  void remove() {
    getProcessorKeeper().remove(processorId);
  }

  void initialize(LocalRegion theRegion, Object theKey, Object theCallbackArg) {
    region = theRegion;
    regionName = theRegion.getFullPath();
    key = theKey;
    aCallbackArgument = theCallbackArg;
    RegionAttributes attrs = theRegion.getAttributes();
    Scope scope = attrs.getScope();
    if (scope.isDistributed()) {
      advisor = ((CacheDistributionAdvisee) region).getCacheDistributionAdvisor();
      distributionManager = theRegion.getDistributionManager();
      timeout = getSearchTimeout();
      advisor.addMembershipListener(this);
    }
  }

  void setKey(Object key) {
    this.key = key;
  }

  protected void setSelectedNode(InternalDistributedMember selectedNode) {
    this.selectedNode = selectedNode;
    selectedNodeDead = false;
  }

  protected int getTimeout() {
    return timeout;
  }

  protected Object getKey() {
    return key;
  }

  InternalDistributedMember getSelectedNode() {
    return selectedNode;
  }

  /**
   * Even though SearchLoadAndWriteProcessor may be in invoked in the context of a local region,
   * most of the services it provides are relevant to distribution only. The 3 services it provides
   * are netSearch, netLoad, netWrite
   */
  private SearchLoadAndWriteProcessor() {
    resetResults();
    pendingResponders.clear();
    attemptedLocalLoad = false;
    netSearchDone = false;
    isSerialized = false;
    result = null;
    key = null;
    requestInProgress = false;
    netWriteSucceeded = false;
    remoteGetInProgress = false;
    responseQueue = null;
  }

  /**
   * If we have a local cache loader and the region is not global, then invoke the loader If the
   * region is local, or the result is non-null, then return whatever the loader returned do a
   * netSearch amongst selected peers if netSearch returns a blob, deserialize the blob and return
   * that as the result netSearch failed, so all we can do at this point is do a load return result
   * from load
   *
   */
  private void searchAndLoad(EntryEventImpl event, TXStateInterface txState, Object localValue,
      boolean preferCD) throws CacheLoaderException, TimeoutException {

    RegionAttributes attrs = region.getAttributes();
    Scope scope = attrs.getScope();
    DataPolicy dataPolicy = attrs.getDataPolicy();

    if (txState != null) {
      TXEntryState tx = txState.txReadEntry(event.getKeyInfo(), region, false,
          true/* create txEntry is absent */);
      if (tx != null) {
        if (tx.noValueInSystem()) {
          // If the tx view has it invalid or destroyed everywhere
          // then don't do a netsearch. We want to see the
          // transactional view.
          load(event, preferCD);
          return;
        }
      }
    }

    // if mirrored then we can optimize by skipping netsearch in some cases,
    // and if also skip netSearch if we find an INVALID token since we
    // know it was distributed. (Otherwise it would be a LOCAL_INVALID token)
    {
      if (localValue == Token.INVALID || dataPolicy.withReplication()) {
        load(event, preferCD);
        return;
      }
    }

    Object obj = null;
    if (!scope.isGlobal()) {
      // copy into local var to prevent race condition
      CacheLoader loader = region.basicGetLoader();
      if (loader != null) {
        obj = doLocalLoad(loader, true, preferCD);
        Assert.assertTrue(obj != Token.INVALID && obj != Token.LOCAL_INVALID);
        event.setNewValue(obj);
        isSerialized = false;
        result = obj;
        return;
      }
      if (scope.isLocal()) {
        return;
      }
    }
    netSearchForBlob();
    if (result != null) {
      Assert.assertTrue(result != Token.INVALID && result != Token.LOCAL_INVALID);
      if (isSerialized) {
        event.setSerializedNewValue((byte[]) result);
      } else {
        event.setNewValue(result);
      }
      event.setVersionTag(versionTag);
      return;
    }

    load(event, preferCD);
  }

  /** perform a net-search, setting this.result to the object found in the search */
  private void netSearchForBlob() throws TimeoutException {
    if (netSearchDone) {
      return;
    }
    netSearchDone = true;
    CachePerfStats stats = region.getCachePerfStats();
    long start = 0;
    Set sendSet = null;

    result = null;
    RegionAttributes attrs = region.getAttributes();
    // Object aCallbackArgument = null;
    requestInProgress = true;
    selectedNodeDead = false;
    initRemainingTimeout();
    start = stats.startNetsearch();
    try {
      List<InternalDistributedMember> replicates =
          new ArrayList(advisor.adviseInitializedReplicates());
      if (replicates.size() > 1) {
        Collections.shuffle(replicates);
      }
      for (InternalDistributedMember replicate : replicates) {
        synchronized (pendingResponders) {
          pendingResponders.clear();
        }

        synchronized (this) {
          requestInProgress = true;
          remoteGetInProgress = true;
          setSelectedNode(replicate);
          lastNotifySpot = 0;

          sendValueRequest(replicate);
          waitForObject2(remainingTimeout);

          if (authorative) {
            if (result != null) {
              netSearch = true;
            }
            return;
          } else {
            // clear anything that might have been set by our query.
            selectedNode = null;
            selectedNodeDead = false;
            lastNotifySpot = 0;
            result = null;
          }
        }
      }
      synchronized (membersLock) {
        Set recipients = advisor.adviseNetSearch();
        if (recipients.isEmpty()) {
          return;
        }
        ArrayList list = new ArrayList(recipients);
        Collections.shuffle(list);
        sendSet = new HashSet(list);
        synchronized (pendingResponders) {
          pendingResponders.clear();
          pendingResponders.addAll(list);
        }
      }

      boolean useMulticast = region.getMulticastEnabled() && (region instanceof DistributedRegion)
          && region.getSystem().getConfig().getMcastPort() != 0;

      // moved outside the sync to fix bug 39458
      QueryMessage.sendMessage(this, regionName, key, useMulticast, sendSet,
          remainingTimeout, attrs.getEntryTimeToLive().getTimeout(),
          attrs.getEntryIdleTimeout().getTimeout());

      synchronized (this) {
        // moved this send back into sync to fix bug 37132
        // QueryMessage.sendMessage(this, this.regionName,this.key,useMulticast,
        // sendSet,this.remainingTimeout ,
        // attrs.getEntryTimeToLive().getTimeout(),
        // attrs.getEntryIdleTimeout().getTimeout());
        boolean done = false;
        do {
          waitForObject2(remainingTimeout);
          if (selectedNodeDead && remoteGetInProgress) {
            sendNetSearchRequest();
          } else {
            done = true;
          }
        } while (!done);

        if (result != null) {
          netSearch = true;
        }
        return;
      }
    } finally {
      stats.endNetsearch(start);
    }
  }

  private void load(EntryEventImpl event, boolean preferCD)
      throws CacheLoaderException, TimeoutException {
    Object obj = null;
    RegionAttributes attrs = region.getAttributes();
    Scope scope = attrs.getScope();
    CacheLoader loader = region.basicGetLoader();
    Assert.assertTrue(scope.isDistributed());

    if ((loader != null) && (!scope.isGlobal())) {
      obj = doLocalLoad(loader, false, preferCD);
      event.setNewValue(obj);
      Assert.assertTrue(obj != Token.INVALID && obj != Token.LOCAL_INVALID);
      return;
    }

    if (scope.isGlobal()) {
      Assert.assertTrue(lock == null);
      Set loadCandidatesSet = advisor.adviseNetLoad();
      if ((loader == null) && (loadCandidatesSet.isEmpty())) {
        // no one has a data Loader. No point getting a lock
        return;
      }

      lock = region.getDistributedLock(key);
      boolean locked = false;
      try {
        final CancelCriterion stopper = region.getCancelCriterion();
        for (;;) {
          stopper.checkCancelInProgress(null);
          boolean interrupted = Thread.interrupted();
          try {
            locked = lock.tryLock(region.getCache().getLockTimeout(), TimeUnit.SECONDS);
            if (!locked) {
              throw new TimeoutException(
                  String.format("Timed out locking %s before load",
                      key));
            }
            break;
          } catch (InterruptedException ignore) {
            interrupted = true;
            region.getCancelCriterion().checkCancelInProgress(null);
            // continue;
          } finally {
            if (interrupted) {
              Thread.currentThread().interrupt();
            }
          }
        } // for
        if (loader == null) {
          localLoad = false;
          if (scope.isDistributed()) {
            isSerialized = false;
            obj = doNetLoad();
            Assert.assertTrue(obj != Token.INVALID && obj != Token.LOCAL_INVALID);
            if (isSerialized && obj != null) {
              event.setSerializedNewValue((byte[]) obj);
            } else {
              event.setNewValue(obj);
            }
          }
        } else {
          obj = doLocalLoad(loader, false, preferCD);
          Assert.assertTrue(obj != Token.INVALID && obj != Token.LOCAL_INVALID);
          event.setNewValue(obj);
        }
        return;
      } finally {
        // The lock will not actually be released until release is
        // called on this processor
        if (!locked) {
          lock = null;
        }
      }
    }
    if (scope.isDistributed()) {
      // long start = System.currentTimeMillis();
      obj = doNetLoad();
      if (isSerialized && obj != null) {
        event.setSerializedNewValue((byte[]) obj);
      } else {
        Assert.assertTrue(obj != Token.INVALID && obj != Token.LOCAL_INVALID);
        event.setNewValue(obj);
      }
      // long end = System.currentTimeMillis();
    }
  }

  Object doNetLoad() throws CacheLoaderException, TimeoutException {
    if (netLoadDone) {
      return null;
    }
    netLoadDone = true;
    if (advisor != null) {
      Set loadCandidatesSet = advisor.adviseNetLoad();
      if (loadCandidatesSet.isEmpty()) {
        return null;
      }
      CachePerfStats stats = region.getCachePerfStats();
      long start = stats.startNetload();
      ArrayList list = new ArrayList(loadCandidatesSet);
      Collections.shuffle(list);
      InternalDistributedMember[] loadCandidates =
          (InternalDistributedMember[]) list.toArray(new InternalDistributedMember[0]);
      initRemainingTimeout();

      RegionAttributes attrs = region.getAttributes();
      int index = 0;
      boolean stayInLoop = false; // never set to true!
      remoteLoadInProgress = true;
      try {
        do { // the only time this loop repeats is when continue is called
          InternalDistributedMember next = loadCandidates[index++];
          setSelectedNode(next);
          lastNotifySpot = 0;
          requestInProgress = true;
          if (remainingTimeout <= 0) { // @todo this looks wrong; why not a timeout exception?
            break;
          }
          remoteException = null;
          NetLoadRequestMessage.sendMessage(this, regionName, key, aCallbackArgument,
              next, remainingTimeout, attrs.getEntryTimeToLive().getTimeout(),
              attrs.getEntryIdleTimeout().getTimeout());
          waitForObject2(remainingTimeout);
          if (remoteException == null) {
            if (!requestInProgress) {
              // note even if result is null we are done; see bug 39738
              localLoad = false;
              if (result != null) {
                netLoad = true;
              }
              return result;
            } else {
              // Why does the following test for selectedNodeDead?
              // Seems like this will cause us to quit trying netLoad
              // even if we don't have a result yet and have not tried everyone.
              if ((selectedNodeDead) && (index < loadCandidates.length)) {
                continue;
              }
              // otherwise we are done
            }
          } else {
            Throwable cause;
            if (remoteException instanceof TryAgainException) {
              if (index < loadCandidates.length) {
                continue;
              } else {
                break;
              }
            }
            if (remoteException instanceof CacheLoaderException) {
              cause = remoteException.getCause();
            } else {
              cause = remoteException;
            }
            throw new CacheLoaderException(
                String.format("While invoking a remote netLoad: %s",
                    cause),
                cause);
          }

        } while (stayInLoop);
      } finally {
        stats.endNetload(start);
      }

    }
    return null;
  }

  /**
   * This exception is just used in the class to tell the caller that it should try again.
   * InternalGemFireException used to be used for this which seems dangerous.
   */
  private static class TryAgainException extends GemFireException {

    ////////////////////// Constructors //////////////////////

    public TryAgainException() {
      super();
    }

    /**
     * Creates a new <code>TryAgainException</code>.
     */
    public TryAgainException(String message) {
      super(message);
    }
  }


  private Object doLocalLoad(CacheLoader loader, boolean netSearchAllowed, boolean preferCD)
      throws CacheLoaderException {
    Object obj = null;
    if (loader != null && !attemptedLocalLoad) {
      attemptedLocalLoad = true;
      CachePerfStats stats = region.getCachePerfStats();
      LoaderHelper loaderHelper = region.loaderHelperFactory.createLoaderHelper(key,
          aCallbackArgument, netSearchAllowed, true /* netLoadAllowed */, this);
      long statStart = stats.startLoad();
      try {
        obj = loader.load(loaderHelper);
        obj = region.getCache().convertPdxInstanceIfNeeded(obj, preferCD);
      } finally {
        stats.endLoad(statStart);
      }
      if (obj != null) {
        localLoad = true;
      }
    }
    return obj;
  }

  /**
   * Returns an event for listener notification. The event's operation may be altered to conform to
   * the ConcurrentMap implementation specification. If the returned value is not == to the event
   * parameter then the caller is responsible for releasing it.
   *
   * @param event the original event
   * @return the original event or a new event having a change in operation
   */
  @Retained
  private CacheEvent getEventForListener(CacheEvent event) {
    Operation op = event.getOperation();
    if (!op.isEntry()) {
      return event;
    } else {
      EntryEventImpl r = (EntryEventImpl) event;
      @Retained
      EntryEventImpl result = r;
      if (r.isSingleHop()) {
        // fix for bug #46130 - origin remote incorrect for one-hop operation in receiver
        result = new EntryEventImpl(r);
        result.setOriginRemote(true);
        // if this is from a non-replicate and it's not in a tx it should be seen as a Create
        // because that's what the sender would use in notifying listeners. bug #46955
        if (result.getOperation().isUpdate() && (result.getTransactionId() == null)) {
          result.makeCreate();
        }
      }
      if (op == Operation.REPLACE) {
        if (result == r) {
          result = new EntryEventImpl(r);
        }
        result.setOperation(Operation.UPDATE);
      } else if (op == Operation.PUT_IF_ABSENT) {
        if (result == r) {
          result = new EntryEventImpl(r);
        }
        result.setOperation(Operation.CREATE);
      } else if (op == Operation.REMOVE) {
        if (result == r) {
          result = new EntryEventImpl(r);
        }
        result.setOperation(Operation.DESTROY);
      }
      return result;
    }
  }

  private boolean doLocalWrite(CacheWriter writer, CacheEvent pevent, int paction)
      throws CacheWriterException {
    // Return if the inhibit all notifications flag is set
    if (pevent instanceof EntryEventImpl) {
      if (((EntryEventImpl) pevent).inhibitAllNotifications()) {
        if (logger.isDebugEnabled()) {
          logger.debug("Notification inhibited for key {}", pevent);
        }
        return false;
      }
    }
    @Released
    CacheEvent event = getEventForListener(pevent);

    int action = paction;
    if (event.getOperation().isCreate() && action == BEFOREUPDATE) {
      action = BEFORECREATE;
    }
    if (event instanceof EntryEventImpl) {
      ((EntryEventImpl) event).setReadOldValueFromDisk(true);
    }
    try {
      switch (action) {
        case BEFORECREATE:
          writer.beforeCreate((EntryEvent) event);
          break;
        case BEFOREDESTROY:
          writer.beforeDestroy((EntryEvent) event);
          break;
        case BEFOREUPDATE:
          writer.beforeUpdate((EntryEvent) event);
          break;
        case BEFOREREGIONDESTROY:
          writer.beforeRegionDestroy((RegionEvent) event);
          break;
        case BEFOREREGIONCLEAR:
          writer.beforeRegionClear((RegionEvent) event);
          break;
        default:
          break;

      }
    } finally {
      if (event instanceof EntryEventImpl) {
        ((EntryEventImpl) event).setReadOldValueFromDisk(false);
      }
      if (event != pevent) {
        if (event instanceof EntryEventImpl) {
          ((Releasable) event).release();
        }
      }
    }
    localWrite = true;
    return true;

  }

  /** Return true if cache writer was invoked */
  private boolean netWrite(CacheEvent event, int action, Set writeCandidateSet)
      throws CacheWriterException, TimeoutException {
    if (writeCandidateSet == null || writeCandidateSet.isEmpty()) {
      return false;
    }
    ArrayList list = new ArrayList(writeCandidateSet);
    Collections.shuffle(list);
    InternalDistributedMember[] writeCandidates =
        (InternalDistributedMember[]) list.toArray(new InternalDistributedMember[0]);
    initRemainingTimeout();
    int index = 0;
    do {
      InternalDistributedMember next = writeCandidates[index++];
      Set set = new HashSet();
      set.add(next);
      netWriteSucceeded = false;
      requestInProgress = true;
      remoteException = null;
      NetWriteRequestMessage.sendMessage(this, regionName, remainingTimeout, event, set,
          action);
      if (remainingTimeout <= 0) { // @todo: should this throw a timeout exception?
        break;
      }
      waitForObject2(remainingTimeout);
      if (netWriteSucceeded) {
        netWrite = true;
        break;
      }
      if (remoteException != null) {
        Throwable cause;
        if (remoteException instanceof TryAgainException) {
          if (index < writeCandidates.length) {
            continue;
          } else {
            break;
          }
        }
        if (remoteException instanceof CacheWriterException
            && remoteException.getCause() != null) {
          cause = remoteException.getCause();
        } else {
          cause = remoteException;
        }
        throw new CacheWriterException(
            String.format("While invoking a remote netWrite: %s",
                cause),
            cause);
      }
    } while (index < writeCandidates.length);

