/*
 * 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.distributed.internal;

import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Set;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.TimeoutException;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.internal.deadlock.MessageDependencyMonitor;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.cache.versions.ConcurrentCacheModificationException;
import org.apache.geode.internal.serialization.DSFIDNotFoundException;
import org.apache.geode.internal.serialization.Version;
import org.apache.geode.internal.serialization.Versioning;
import org.apache.geode.internal.util.Breadcrumbs;
import org.apache.geode.internal.util.concurrent.StoppableCountDownLatch;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * This class processes responses to {@link DistributionMessage}s. It handles a the generic case of
 * simply waiting for responses from all members. It is intended to be subclassed for special cases.
 *
 * <P>
 *
 * Note that, unlike the reply processors in versions previous to GemFire 2.1, this reply processor
 * is kept entirely in JOM.
 *
 * <p>
 *
 * Recommended usage pattern in subclass...
 *
 * <pre>
 *
 * public void process(DistributionMessage msg) {
 *   try {
 *     ...custom code for subclass goes here...
 *   }
 *   finally {
 *     super.process(msg);
 *   }
 * }
 *
 * </pre>
 *
 * The above usage pattern causes the waitForReplies latch to not be released until after the
 * message has been processed. In addition, it is guaranteed to be released even if the custom
 * subclass code throws a runtime exception.
 *
 * <p>
 *
 * @see MessageWithReply
 * @since GemFire 2.1
 */
public class ReplyProcessor21 implements MembershipListener {

  private static final Logger logger = LogService.getLogger();

  public static final boolean THROW_EXCEPTION_ON_TIMEOUT =
      Boolean.getBoolean("ack-threshold-exception");

  /**
   * the ratio by which ack-severe-alert-threshold is lowered when waiting for a BucketRegion
   * operation
   */
  public static final double PR_SEVERE_ALERT_RATIO;

  /** All live reply processors in this VM */
  @MakeNotStatic
  protected static final ProcessorKeeper21 keeper = new ProcessorKeeper21();

  //////////////////// Instance Methods ////////////////////

  /**
   * The members that haven't replied yet
   *
   * Concurrency: protected by synchronization of itself
   */
  protected final InternalDistributedMember[] members;

  /**
   * Set to true in preWait, set to false in postWait. Used to avoid removing membership listener in
   * Runnable in postWait if we've called waitForReplies again.
   */
  protected volatile boolean waiting = false;

  /**
   * An <code>Exception</code> that occurred when processing a reply
   * <p>
   * Since this is set by the executor and read by the initiating thread, this is a volatile.
   *
   * @see ReplyMessage#getException
   */
  protected volatile ReplyException exception;

  /** Have we heard back from everyone? */
  private volatile boolean done;

  private boolean keeperCleanedUp;

  /** Have we been aborted due to shutdown? */
  protected volatile boolean shutdown;

  /** Semaphore used for wait/notify */
  private final StoppableCountDownLatch latch;

  /** The id of this processor */
  protected int processorId;

  /**
   * Used to get the ack wait threshold (which might change at runtime), etc.
   */
  protected final InternalDistributedSystem system;

  /** the distribution manager - if null, get the manager from the system */
  protected final DistributionManager dmgr;

  /** Start time for replyWait stat, in nanos */
  long statStart;

  /** Start time for ack-wait-threshold, in millis */
  private final long initTime;

  /**
   * whether this reply processor should perform severe-alert processing for the message being ack'd
   */
  private boolean severeAlertEnabled;

  /**
   * whether the severe-alert timeout has been reset. This can happen if a member we're waiting for
   * is waiting on a suspect member, for instance.
   */
  private volatile boolean severeAlertTimerReset;

  /**
   * whether this reply processor should shorten severe-alert processing due to another vm waiting
   * on this one. This is a thread-local so that lower level comm layers can tell that the interval
   * should be shortened
   */
  private static final ThreadLocal<Boolean> severeAlertShorten =
      ThreadLocal.withInitial(() -> Boolean.FALSE);

  /**
   * whether the next replyProcessor for the current thread should perform severe-alert processing
   */
  private static final ThreadLocal<Boolean> forceSevereAlertProcessing =
      ThreadLocal.withInitial(() -> Boolean.FALSE);

  ////////////////////// Static Methods /////////////////////

  static {
    String str = System
        .getProperty(GeodeGlossary.GEMFIRE_PREFIX + "ack-severe-alert-reduction-ratio", ".80");
    double ratio;
    try {
      ratio = Double.parseDouble(str);
    } catch (NumberFormatException e) {
      System.err.println(
          "Unable to parse gemfire.ack-severe-alert-reduction-ratio setting of \"" + str + "\"");
      ratio = 0.80;
    }
    PR_SEVERE_ALERT_RATIO = ratio;
  }

  /**
   * Returns the <code>ReplyProcessor</code> with the given id, or <code>null</code> if it no longer
   * exists.
   *
   * @param processorId The id of the processor to get
   *
   * @see #getProcessorId
   */
  public static ReplyProcessor21 getProcessor(int processorId) {
    return (ReplyProcessor21) keeper.retrieve(processorId);
  }

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

  /**
   * Creates a new <code>ReplyProcessor</code> that wants replies from a single member of a
   * distributed system.
   *
   * @param system the DistributedSystem connection
   * @param member the member this processor wants a reply from
   */
  public ReplyProcessor21(InternalDistributedSystem system, InternalDistributedMember member) {
    this(system, Collections.singleton(member));
  }

  /**
   * Creates a new <code>ReplyProcessor</code> that wants replies from a single member of a
   * distributed system.
   *
   * @param system the DistributedSystem connection
   * @param member the member this processor wants a reply from
   * @param cancelCriterion optional CancelCriterion to use; will use the DistributionManager if
   *        null
   */
  public ReplyProcessor21(InternalDistributedSystem system, InternalDistributedMember member,
      CancelCriterion cancelCriterion) {
    this(system, Collections.singleton(member), cancelCriterion);
  }

  /**
   * Creates a new <code>ReplyProcessor</code> that wants replies from a single member of a
   * distributed system.
   *
   * @param dm the DistributionManager to use for messaging and membership
   * @param member the member this processor wants a reply from
   */
  public ReplyProcessor21(DistributionManager dm, InternalDistributedMember member) {
    this(dm, Collections.singleton(member));
  }

  /**
   * Creates a new <code>ReplyProcessor</code> that wants replies from some number of members of a
   * distributed system. Call this method with
   * {@link ClusterDistributionManager#getDistributionManagerIds} if you want replies from all DMs
   * including the one hosted in this VM.
   *
   * @param dm the DistributionManager to use for messaging and membership
   * @param initMembers the Set of members this processor wants replies from
   */
  public ReplyProcessor21(DistributionManager dm, Collection initMembers) {
    this(dm, initMembers, null);
  }

  /**
   * Creates a new <code>ReplyProcessor</code> that wants replies from some number of members of a
   * distributed system. Call this method with
   * {@link ClusterDistributionManager#getDistributionManagerIds} if you want replies from all DMs
   * including the one hosted in this VM.
   *
   * @param dm the DistributionManager to use for messaging and membership
   * @param initMembers the Set of members this processor wants replies from
   * @param cancelCriterion optional CancelCriterion to use; will use the dm if null
   */
  public ReplyProcessor21(DistributionManager dm, Collection initMembers,
      CancelCriterion cancelCriterion) {
    this(dm, dm.getSystem(), initMembers, cancelCriterion);
  }

  /**
   * Creates a new <code>ReplyProcessor</code> that wants replies from some number of members of a
   * distributed system. Call this method with
   * {@link ClusterDistributionManager#getDistributionManagerIds} if you want replies from all DMs
   * including the one hosted in this VM.
   *
   * @param system the DistributedSystem connection
   * @param initMembers the Set of members this processor wants replies from
   */
  public ReplyProcessor21(InternalDistributedSystem system, Collection initMembers) {
    this(system.getDistributionManager(), system, initMembers, null);
  }

  /**
   * Creates a new <code>ReplyProcessor</code> that wants replies from some number of members of a
   * distributed system. Call this method with
   * {@link ClusterDistributionManager#getDistributionManagerIds} if you want replies from all DMs
   * including the one hosted in this VM.
   *
   * @param system the DistributedSystem connection
   * @param initMembers the Set of members this processor wants replies from
   * @param cancelCriterion optional CancelCriterion to use; will use the DistributedSystem's
   *        DistributionManager if null
   */
  public ReplyProcessor21(InternalDistributedSystem system, Collection initMembers,
      CancelCriterion cancelCriterion) {
    this(system.getDistributionManager(), system, initMembers, cancelCriterion);
  }

  /**
   * Construct new ReplyProcessor21.
   *
   * @param dm the DistributionManager to use for messaging and membership
   * @param system the DistributedSystem connection
   * @param initMembers the collection of members this processor wants replies from
   * @param cancelCriterion optional CancelCriterion to use; will use the dm if null
   */
  private ReplyProcessor21(DistributionManager dm, InternalDistributedSystem system,
      Collection initMembers, CancelCriterion cancelCriterion) {

    this(dm, system, initMembers, cancelCriterion, true);
  }

  /**
   * Construct new ReplyProcessor21.
   *
   * @param dm the DistributionManager to use for messaging and membership
   * @param system the DistributedSystem connection
   * @param initMembers the collection of members this processor wants replies from
   * @param cancelCriterion optional CancelCriterion to use; will use the dm if null
   */
  protected ReplyProcessor21(DistributionManager dm, InternalDistributedSystem system,
      Collection initMembers, CancelCriterion cancelCriterion, boolean register) {
    if (!allowReplyFromSender()) {
      Assert.assertTrue(initMembers != null, "null initMembers");
      Assert.assertTrue(system != null, "null system");
      if (dm != null) {
        Assert.assertTrue(!initMembers.contains(dm.getId()),
            "dm present in initMembers but reply from sender is not allowed");
      }
    }
    this.system = system;
    dmgr = dm;
    if (cancelCriterion == null) {
      cancelCriterion = dm.getCancelCriterion();
    }
    latch = new StoppableCountDownLatch(cancelCriterion, 1);
    int sz = initMembers.size();
    members = new InternalDistributedMember[sz];
    if (sz > 0) {
      int i = 0;
      for (Iterator it = initMembers.iterator(); it.hasNext(); i++) {
        members[i] = (InternalDistributedMember) it.next();
      }
    }
    done = false;
    shutdown = false;
    exception = null;
    if (register) {
      register();
    }
    keeperCleanedUp = false;
    initTime = System.currentTimeMillis();
  }

  protected int register() {
    processorId = keeper.put(this);
    return processorId;
  }

  ///////////////////// Instance Methods /////////////////////

  /**
   * get the distribution manager for this processor. If the distributed system has a distribution
   * manager, it is used. Otherwise, we expect a distribution manager has been set with
   * setDistributionManager and we'll use that
   */
  public DistributionManager getDistributionManager() {
    try {
      DistributionManager result = system.getDistributionManager();
      if (result == null) {
        result = dmgr;
        Assert.assertTrue(result != null, "null DistributionManager");
      }
      return result;
    } catch (IllegalStateException ex) {
      // fix for bug 35000
      system.getCancelCriterion().checkCancelInProgress(null);
      throw new DistributedSystemDisconnectedException(ex.getMessage());
    }
  }


  /**
   * Override and return true if processor should wait for reply from sender.
   * <p>
   * NOTE: the communication layer does not reliably support sending a message to oneself, so other
   * means must be used to execute the message in this VM. Typically you would set the sender of the
   * message and then invoke its process() method in another thread.
   */
  protected boolean allowReplyFromSender() {
    return false;
  }

  /**
   * The first time a reply is received that contains an exception, the ReplyProcessor will save
   * that exception to be passed along to the waiting client thread. By default, any exception
   * encountered after that will be logged. Subclasses can reimplement this method to disable
   * logging of multiple exceptions.
   */
  protected boolean logMultipleExceptions() {
    return true;
  }


  /**
   * Makes note of a reply from a member. If all members have replied, the waiting thread is
   * signaled. This method can be overridden to provide customized functionality, however the
   * overriden method should always invoke <code>super.process()</code>.
   */
  public void process(DistributionMessage msg) {
    process(msg, true);
  }

  protected void process(DistributionMessage msg, boolean warn) {
    if (logger.isDebugEnabled()) {
      logger.debug("{} got process({}) from {}", this, msg, msg.getSender());
    }
    if (msg instanceof ReplyMessage) {
      ReplyException ex = ((ReplyMessage) msg).getException();
      if (ex != null) {
        if (ex.getCause() instanceof DSFIDNotFoundException) {
          processException(msg, (DSFIDNotFoundException) ex.getCause());
        } else {
          processException(msg, ex);
        }
      }
    }

    final InternalDistributedMember sender = msg.getSender();
    if (!removeMember(sender, false) && warn) {
      // if the member hasn't left the system, something is wrong
      final DistributionManager dm = getDistributionManager(); // fix for bug 33253
      Set ids = getDistributionManagerIds();
      if (ids == null || ids.contains(sender)) {
        List viewMembers = dm.getViewMembers();
        if (system.getConfig().getMcastPort() == 0 // could be using multicast & will get responses
                                                   // from everyone
            && (viewMembers == null || viewMembers.contains(sender))) {
          logger.warn(
              "Received reply from member {} but was not expecting one. More than one reply may have been received. The reply that was not expected is: {}",
              new Object[] {sender, msg});
        }
      }
    }
    checkIfDone();
  }


  protected synchronized void processException(DistributionMessage msg, ReplyException ex) {
    processException(ex);
  }

  protected synchronized void processException(ReplyException ex) {
    if (exception == null) { // only keep first exception
      exception = ex;

    } else if (logMultipleExceptions()) {
      if (!(ex.getCause() instanceof ConcurrentCacheModificationException)) {
        logger.fatal(
            "Exception received in ReplyMessage. Only one exception is passed back to caller. This exception is logged only.",
            ex);
      }
    }
  }

  /**
   * Handle a {@link DSFIDNotFoundException} indicating a message type is not implemented on another
   * server (for example due to different product version). Default implementation logs the
   * exception as severe and moves on.
   *
   * Rationale for default handling: New operations can have caused changes to other newer versioned
   * GFE JVMs that cannot be reverted. So ignoring exceptions is a conservative way considering such
   * scenarios. It will be upto individual messages to handle differently by overriding the above
   * method.
   */
  protected synchronized void processException(DistributionMessage msg, DSFIDNotFoundException ex) {
    final short versionOrdinal = ex.getProductVersionOrdinal();
    final Version anyVersion = Versioning.getVersion(versionOrdinal);
    logger.fatal(String.format(
        "Exception received due to missing DSFID %s on remote node %s running version %s.",
        ex.getUnknownDSFID(), msg.getSender(), anyVersion), ex);
  }

  @Override
  public void memberJoined(DistributionManager distributionManager, InternalDistributedMember id) {
    // Nothing to do - member wasn't sent the operation, anyway.
  }

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

  @Override
  public void memberSuspect(DistributionManager distributionManager, InternalDistributedMember id,
      InternalDistributedMember whoSuspected, String reason) {
    if (isSevereAlertProcessingEnabled()) {
      // if we're waiting for the member that initiated suspicion, we don't
      // want to be hasty about kicking it out of the distributed system
      synchronized (members) {
        int cells = members.length;
        for (InternalDistributedMember e : members) {
          if (e != null && e.equals(whoSuspected)) {
            severeAlertTimerReset = true;
          }
        }
      } // synchronized
    }
  }

  @Override
  public void memberDeparted(DistributionManager distributionManager,
      final InternalDistributedMember id, final boolean crashed) {
    removeMember(id, true);
    checkIfDone();
  }

  /**
   * Wait for all expected acks to be returned or an exception to come in. This method will return
   * whether a) we have received replies from all desired members or b) the necessary conditions
   * have been met so that we don't need to wait anymore.
   *
   * @throws InternalGemFireException if ack-threshold was exceeded and system property
   *         "ack-threshold-exception" is set to true
   * @throws InterruptedException thrown if the wait is interrupted
   * @see #canStopWaiting()
   */
  public void waitForReplies() throws InterruptedException, ReplyException {
    boolean result = waitForReplies(0);
    Assert.assertTrue(result, "failed but no exception thrown");
  }


  /**
   * Registers this processor as a membership listener and returns a set of the current members.
   *
   * @return a Set of the current members
   * @since GemFire 5.7
   */
  protected Set addListenerAndGetMembers() {
    return getDistributionManager().addMembershipListenerAndGetDistributionManagerIds(this);
  }

  /**
   * Unregisters this processor as a membership listener
   *
   * @since GemFire 5.7
   */
  protected void removeListener() {
    try {
      getDistributionManager().removeMembershipListener(this);
    } catch (DistributedSystemDisconnectedException e) {
      // ignore
    }
  }

  /**
   * Returns the set of members that this processor should care about.
   *
   * @return a Set of the current members
   * @since GemFire 5.7
   */
  protected Set getDistributionManagerIds() {
    return getDistributionManager().getDistributionManagerIds();
  }

  protected void preWait() {
    waiting = true;
    DistributionManager mgr = getDistributionManager();
    statStart = mgr.getStats().startReplyWait();
    synchronized (members) {
      Set activeMembers = addListenerAndGetMembers();
      processActiveMembers(activeMembers);
    }
  }

  /**
   * perform initial membership processing while under synchronization of this.members
   *
   * @param activeMembers the DM's current membership set
   */
  protected void processActiveMembers(Set activeMembers) {
    for (final InternalDistributedMember member : members) {
      if (member != null) {
        if (!activeMembers.contains(member)) {
          memberDeparted(getDistributionManager(), member, false);
        }
      }
    }
  }

  private void postWait() {
    waiting = false;
    removeListener();
    final DistributionManager mgr = getDistributionManager();
    mgr.getStats().endReplyWait(statStart, initTime);
    mgr.getCancelCriterion().checkCancelInProgress(null);
  }


  /**
   * Wait a given number of milliseconds for the expected acks to be received. If <code>msecs</code>
   * milliseconds pass before all acknowlegdements are received, <code>false</code> is returned.
   *
   * @param msecs the number of milliseconds to wait for replies
   * @throws InterruptedException if interrupted while waiting on latch
   * @throws ReplyException an exception passed back in reply
   * @throws InternalGemFireException if ack-threshold was exceeded and system property
   *         "ack-threshold-exception" is set to true
   * @throws IllegalStateException if the processor is not registered to receive messages
   *
   * @return Whether or not we received all of the replies in the given amount of time.
   */
  public boolean waitForReplies(long msecs) throws InterruptedException, ReplyException {
    return waitForReplies(msecs, getLatch(), true);
  }

  public boolean waitForReplies(long msecs, StoppableCountDownLatch latch, boolean doCleanUp)
      throws InterruptedException, ReplyException {
    if (keeperCleanedUp) {
      throw new IllegalStateException(
          "This reply processor has already been removed from the processor keeper");
    }
    boolean result = true;
    boolean interrupted = Thread.interrupted();
    MessageDependencyMonitor.waitingForReply(this);
    try {
      // do the interrupted check inside the try so that cleanup is called
      if (interrupted) {
        throw new InterruptedException();
      }
      if (stillWaiting()) {
        preWait();
        try {
          result = basicWait(msecs, latch);
        } catch (InterruptedException e) {
          interrupted = true;
        } finally {
          if (doCleanUp) {
            postWait();
          }
        }
      }
      if (exception != null) {
        throw exception;
      }
    } finally {
      if (doCleanUp) {
        try {
          cleanup();
        } finally {
          if (interrupted) {
            throw new InterruptedException();
          }
        }
      }
      MessageDependencyMonitor.doneWaiting(this);
    }
    return result;
  }

  /**
   * basicWait occurs after preWait and before postWait. Attempts to acquire the latch are made.
   *
   * @param msecs the number of milliseconds to wait for replies
   * @return whether or not we received all of the replies in the given amount of time
   */
  private boolean basicWait(long msecs, StoppableCountDownLatch latch)
      throws InterruptedException, ReplyException {
    if (Thread.interrupted()) {
      throw new InterruptedException();
    }

    if (stillWaiting()) {
      long timeout = getAckWaitThreshold() * 1000L;
      long timeSoFar = System.currentTimeMillis() - initTime;
      final long severeAlertTimeout = getAckSevereAlertThresholdMS();
      // only start SUSPECT processing if severe alerts are enabled
      final boolean doSuspectProcessing =
          isSevereAlertProcessingEnabled() && (severeAlertTimeout > 0);
      if (timeout <= 0) {
        timeout = Long.MAX_VALUE;
      }
      if (msecs == 0) {
        boolean timedOut = false;
        if (timeout <= timeSoFar + 1) {
          timedOut = !latch.await(10);
        }
        if (timedOut || !latch.await(timeout - timeSoFar - 1)) {
          dmgr.getCancelCriterion().checkCancelInProgress(null);

          timeout(doSuspectProcessing, false);

          // If ack-severe-alert-threshold has been set, we now
          // wait for that period of time and then force the non-responding
          // members from the system. Then we wait indefinitely
          if (doSuspectProcessing) {
            boolean wasNotUnlatched;
            do {
              severeAlertTimerReset = false; // retry if this gets set by suspect processing
                                             // (splitbrain requirement)
              wasNotUnlatched = !latch.await(severeAlertTimeout);
            } while (wasNotUnlatched && severeAlertTimerReset);
            if (wasNotUnlatched) {
              dmgr.getCancelCriterion().checkCancelInProgress(null);
              timeout(false, true);

              long suspectProcessingErrorAlertTimeout = severeAlertTimeout * 3;
              if (!latch.await(suspectProcessingErrorAlertTimeout)) {
                long now = System.currentTimeMillis();
                long totalTimeElapsed = now - initTime;

                String waitingOnMembers;
                synchronized (members) {
                  waitingOnMembers = Arrays.toString(members);
                }
                logger.fatal("An additional " + suspectProcessingErrorAlertTimeout
                    + " milliseconds have elapsed while waiting for replies. Total of "
                    + totalTimeElapsed + " milliseconds elapsed (init time:" + initTime
                    + ", now: " + now + ") Waiting for members: " + waitingOnMembers);

                // for consistency, we must now wait indefinitely for a membership view
                // that ejects the removed members
                latch.await();
              }
            }
          } else {
            latch.await();
          }
          // Give an info message since timeout gave a warning.
          logger.info("{} wait for replies completed", shortName());
        }
      } else if (msecs > timeout) {
        if (!latch.await(timeout)) {
          timeout(doSuspectProcessing, false);
          // after timeout alert, wait remaining time
          if (!latch.await(msecs - timeout)) {
            logger.info("wait for replies timing out after {} seconds",
                msecs / 1000);
            return false;
          }
          // Give an info message since timeout gave a warning.
          logger.info("{} wait for replies completed", shortName());
        }
      } else {
        if (!latch.await(msecs)) {
          return false;
        }
      }
    }
    Assert.assertTrue(latch != this.latch || !stillWaiting(), this);
    if (stopBecauseOfExceptions()) {
      throw exception;
    }
    return true;
  }

  /**
   * Wait a given number of milliseconds for the expected acks to be received. If <code>msecs</code>
   * milliseconds pass before all acknowlegdements are received, <code>false</code> is returned.
   * <p>
   * Thread interruptions will be ignored while waiting. If interruption occurred while in this
   * method, the current thread's interrupt flag will be true, but InterruptedException will not be
   * thrown.
   *
   * @param p_msecs the number of milliseconds to wait for replies, zero will be interpreted as
   *        Long.MAX_VALUE
   *
   * @throws ReplyException an exception passed back in reply
   *
   * @throws InternalGemFireException if ack-threshold was exceeded and system property
   *         "ack-threshold-exception" is set to true
   * @throws IllegalStateException if the processor is not registered to receive replies
   */
  public boolean waitForRepliesUninterruptibly(long p_msecs) throws ReplyException {
    return waitForRepliesUninterruptibly(p_msecs, getLatch(), true);
  }

  public boolean waitForRepliesUninterruptibly(long p_msecs, StoppableCountDownLatch latch,
      boolean doCleanUp) throws ReplyException {
    if (keeperCleanedUp) {
      throw new IllegalStateException(
          "This reply processor has already been removed from the processor keeper");
    }
    long msecs = p_msecs; // don't overwrite parameter
    boolean result = true;
    MessageDependencyMonitor.waitingForReply(this);
    try {
      if (stillWaiting()) {
        preWait();
        try {
          while (true) {
            // cancellation check
            dmgr.getCancelCriterion().checkCancelInProgress(null);

            long startWaitTime = System.currentTimeMillis();
            boolean interrupted = Thread.interrupted();
            try {
              result = basicWait(msecs, latch);
              break;
            } catch (InterruptedException e) {
              interrupted = true; // keep looping
              dmgr.getCancelCriterion().checkCancelInProgress(e);
              if (msecs > 0) {
                final long interruptTime = System.currentTimeMillis();
                msecs -= interruptTime - startWaitTime;
                if (msecs <= 0) {
                  msecs = 1;
                  result = false;
                  break;
                }
              }
            } finally {
              if (interrupted) {
                Thread.currentThread().interrupt();
              }
            }
          } // while
        } finally {
          if (doCleanUp) {
            postWait();
          }
        }
      } // stillWaiting
      if (exception != null) {
        throw exception;
      }
    } finally {
      if (doCleanUp) {
        cleanup();
      }
      MessageDependencyMonitor.doneWaiting(this);
    }
    return result;
  }

  /**
   * Used to cleanup resources allocated by the processor after we are done using it.
   *
   * @since GemFire 5.1
   */
  public void cleanup() {
    if (!keeperCleanedUp) {
      keeperCleanedUp = true;
      keeper.remove(getProcessorId());
    }
  }

  /**
   * Wait for the expected acks to be received.
   * <p>
   * Thread interruptions will be ignored while waiting. If interruption occurred while in this
   * method, the current thread's interrupt flag will be true, but InterruptedException will not be
   * thrown.
   *
   * @throws ReplyException an exception passed back in reply
   *
   * @throws InternalGemFireException if ack-threshold was exceeded and system property
   *         "ack-threshold-exception" is set to true
   */
  public void waitForRepliesUninterruptibly() throws ReplyException {
    waitForRepliesUninterruptibly(0);
  }

  /**
   * Returns the id of this reply processor. This id is often sent in messages, so that reply
   * messages know which processor to notify.
   *
   * @see #getProcessor
   */
  public int getProcessorId() {
    return processorId;
  }

  /**
   * Returns whether or not this reply processor can stop waiting for replies. This method can be
   * overridden to allow the waiter to be notified before all responses have been received. This is
   * useful when we are waiting for a value from any member, or if we can stop waiting if a remote
   * exception occurred. By default, this method returns <code>false</code>.
   */
  protected boolean canStopWaiting() {
    return false;
  }

  /**
   * We're still waiting if there is any member still left in the set and an exception hasn't been
   * returned from anyone yet.
   *
   * @return true if we are still waiting for a response
   */
  protected boolean stillWaiting() {
    if (shutdown) {
      // Create the exception here, so that the call stack reflects the
      // failed computation. If you set the exception in onShutdown,
      // the resulting stack is not of interest.
