gt:         if (pendingRequest != null) {
context: /*
   * ################################################################
   *
   * ProActive Parallel Suite(TM): The Java(TM) library for
   *    Parallel, Distributed, Multi-Core Computing for
   *    Enterprise Grids & Clouds
   *
   * Copyright (C) 1997-2012 INRIA/University of
   *                 Nice-Sophia Antipolis/ActiveEon
   * Contact: proactive@ow2.org or contact@activeeon.com
   *
   * This library is free software; you can redistribute it and/or
   * modify it under the terms of the GNU Affero General Public License
   * as published by the Free Software Foundation; version 3 of
   * the License.
   *
   * This library is distributed in the hope that it will be useful,
   * but WITHOUT ANY WARRANTY; without even the implied warranty of
   * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
   * Affero General Public License for more details.
   *
   * You should have received a copy of the GNU Affero General Public License
   * along with this library; if not, write to the Free Software
   * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307
   * USA
   *
   * If needed, contact us to obtain a release under GPL Version 2 or 3
   * or a different license than the AGPL.
   *
   *  Initial developer(s):               The ProActive Team
   *                        http://proactive.inria.fr/team_members.htm
   *  Contributor(s):
   *
   * ################################################################
   * $$PROACTIVE_INITIAL_DEV$$
   */
  package org.objectweb.proactive.core.body.ft.protocols.cic.managers;
  
  import java.io.IOException;
  import java.rmi.RemoteException;
  import java.util.Enumeration;
  import java.util.Hashtable;
  import java.util.Iterator;
  import java.util.List;
  import java.util.ListIterator;
  import java.util.Vector;
  
  import org.apache.log4j.Logger;
  import org.objectweb.proactive.core.ProActiveException;
  import org.objectweb.proactive.core.UniqueID;
  import org.objectweb.proactive.core.body.AbstractBody;
  import org.objectweb.proactive.core.body.UniversalBody;
  import org.objectweb.proactive.core.body.ft.checkpointing.Checkpoint;
  import org.objectweb.proactive.core.body.ft.checkpointing.CheckpointInfo;
  import org.objectweb.proactive.core.body.ft.exception.ProtocolErrorException;
  import org.objectweb.proactive.core.body.ft.internalmsg.FTMessage;
  import org.objectweb.proactive.core.body.ft.internalmsg.GlobalStateCompletion;
  import org.objectweb.proactive.core.body.ft.internalmsg.OutputCommit;
  import org.objectweb.proactive.core.body.ft.message.HistoryUpdater;
  import org.objectweb.proactive.core.body.ft.message.MessageInfo;
  import org.objectweb.proactive.core.body.ft.message.ReplyLog;
  import org.objectweb.proactive.core.body.ft.message.RequestLog;
  import org.objectweb.proactive.core.body.ft.protocols.FTManager;
  import org.objectweb.proactive.core.body.ft.protocols.cic.infos.CheckpointInfoCIC;
  import org.objectweb.proactive.core.body.ft.protocols.cic.infos.MessageInfoCIC;
  import org.objectweb.proactive.core.body.ft.servers.recovery.RecoveryProcess;
  import org.objectweb.proactive.core.body.future.MethodCallResult;
  import org.objectweb.proactive.core.body.message.Message;
  import org.objectweb.proactive.core.body.reply.Reply;
  import org.objectweb.proactive.core.body.reply.ReplyImpl;
  import org.objectweb.proactive.core.body.request.AwaitedRequest;
  import org.objectweb.proactive.core.body.request.BlockingRequestQueue;
  import org.objectweb.proactive.core.body.request.BlockingRequestQueueImpl;
  import org.objectweb.proactive.core.body.request.Request;
  import org.objectweb.proactive.core.body.request.RequestImpl;
  import org.objectweb.proactive.core.mop.Utils;
  import org.objectweb.proactive.core.security.exceptions.CommunicationForbiddenException;
  import org.objectweb.proactive.core.security.exceptions.RenegotiateSessionException;
  import org.objectweb.proactive.core.util.CircularArrayList;
  import org.objectweb.proactive.core.util.MutableLong;
  import org.objectweb.proactive.core.util.log.Loggers;
  import org.objectweb.proactive.core.util.log.ProActiveLogger;
  
  
  /**
   * This class implements a Communication Induced Checkpointing protocol for ProActive.
   * This FTManager is linked to each fault-tolerant active object.
   * @author The ProActive Team
   * @since ProActive 2.2
   */
  public class FTManagerCIC extends org.objectweb.proactive.core.body.ft.protocols.FTManager {
  
      private static final long serialVersionUID = 54L;
  
      /**
       *
       */
  
      /** Value returned by an object if the recieved message must be send again */
      public static final int RESEND_MESSAGE = -3;
  
      /** Value returned by an object if the sender of the received message must recover asap */
      public static final int RECOVER = -4;
  
      //logger
      protected static Logger logger = ProActiveLogger.getLogger(Loggers.FAULT_TOLERANCE_CIC);
  
      // local runtime
      // private static final Runtime runtime = Runtime.getRuntime();
      // FT Values
      private int incarnation;
      private int lastRecovery; // index of the last ckpt used for recovery
      private int checkpointIndex; //index of the latest perfomred checkpoint
      private long checkpointTimer;
      private int nextMax;
  
      // private int nextMin;
      private int historyIndex; // index of the latest closed history
  
      // logged messages
      private Hashtable<Integer, Vector<RequestLog>> requestToResend;
      private int latestRequestLog;
      private Hashtable<Integer, Vector<ReplyLog>> replyToResend;
      private int latestReplyLog;
  
      // awaited request in owner request queue
      private Vector<AwaitedRequest> awaitedRequests;
  
      // pool of MessageInfo
      private MessageInfoCIC forSentRequest;
      private MessageInfoCIC forSentReply;
  
      // history
      private Vector<UniqueID> history;
  
      // cannot lock hisotry itself, because it is modified in synchronized blocks !
      private final Character historyLock = Character.valueOf('l');
  
      // protocol for output commit
      private long deliveredRequestsCounter;
      private MutableLong lastServedRequestIndex; // reference to localVectorClock[this.ownerID]
      private Hashtable<UniqueID, MutableLong> localVectorClock; // ids <-> lastServedRequest(local view)
      private long historyBaseIndex;
      private long lastCommitedIndex;
      private boolean completingCheckpoint; // true if the latest checkpoint is still not completed with its minimal history
  
      // ****** static toggle for OC [TEST] *******
      private final static boolean isOCEnable = false;
  
      @Override
      public int init(AbstractBody owner) throws ProActiveException {
          super.init(owner);
          this.incarnation = 1;
          this.checkpointIndex = 0;
          this.historyIndex = 0;
          this.nextMax = 1;
          //this.nextMin = 0;
          this.lastRecovery = 0;
          this.checkpointTimer = System.currentTimeMillis();
          this.requestToResend = new Hashtable<Integer, Vector<RequestLog>>();
          this.latestRequestLog = 0;
          this.replyToResend = new Hashtable<Integer, Vector<ReplyLog>>();
          this.latestReplyLog = 0;
          this.history = new Vector<UniqueID>();
          this.awaitedRequests = new Vector<AwaitedRequest>();
          this.forSentRequest = new MessageInfoCIC();
          this.forSentReply = new MessageInfoCIC();
          this.deliveredRequestsCounter = -1;
          this.lastCommitedIndex = -1;
          this.lastServedRequestIndex = new MutableLong(0);
          this.historyBaseIndex = 0;
          this.completingCheckpoint = false;
          this.localVectorClock = new Hashtable<UniqueID, MutableLong>();
          this.localVectorClock.put(this.ownerID, this.lastServedRequestIndex);
          logger.info(" CIC fault-tolerance is enabled for body " + this.ownerID);
          return 0;
      }
  
      @Override
      public int onReceiveReply(Reply reply) {
          reply.setFTManager(this);
          return this.incarnationTest(reply);
      }
  
      @Override
      public int onReceiveRequest(Request request) {
          request.setFTManager(this);
          return this.incarnationTest(request);
      }
  
      /*
       * This method test if the message m has to be take into account by the receiver;
       * If not, set the ignore tag in m.
       */
      private int incarnationTest(Message m) {
          if (this.isSignificant(m)) {
              MessageInfoCIC mi = (MessageInfoCIC) m.getMessageInfo();
              int localInt = this.incarnation;
              int inc = mi.incarnation;
              if (inc > localInt) {
                  // this body will recover
                  m.setIgnoreIt(true);
                  return FTManagerCIC.RESEND_MESSAGE;
              } else if (inc < localInt) {
                  // force the sender to recover and ignore this message
                  m.setIgnoreIt(true);
                  return FTManagerCIC.RECOVER;
              }
          }
          return 0; //This value is not returned to the sender
      }
  
      @Override
      public synchronized int onDeliverReply(Reply reply) {
          int currentCheckpointIndex = this.checkpointIndex;
          if (this.isSignificant(reply)) {
              MessageInfoCIC mi = (MessageInfoCIC) reply.getMessageInfo();
  
              // history closure
              this.updateHistory(mi.historyIndex);
              // udpate checkpoint index
              if (mi.checkpointIndex > currentCheckpointIndex) {
                  this.nextMax = Math.max(this.nextMax, mi.checkpointIndex);
              }
          }
          return currentCheckpointIndex;
      }
  
      @Override
      public synchronized int onDeliverRequest(Request request) {
          int currentCheckpointIndex = this.checkpointIndex;
  
          //System.out.println(""+ this.ownerID + " receive " + request);
          if (this.isSignificant(request)) {
              //System.out.println(""+ this.ownerID + " receive significant " + request);
              MessageInfoCIC mi = (MessageInfoCIC) request.getMessageInfo();
  
              // history closure
              this.updateHistory(mi.historyIndex);
              //is there any corresponding awaited request ?
              if (!(this.updateAwaitedRequests(request))) {
                  if (FTManagerCIC.isOCEnable || this.completingCheckpoint) {
                      synchronized (historyLock) {
                          // if no, an awaited request is added to history
                          history.add(request.getSourceBodyID());
                      }
  
                      // inc the historized index
                      this.deliveredRequestsCounter++;
                      // manage local vector clock only if needed
                      if (FTManagerCIC.isOCEnable) {
                          // set the position in history of the request
                          mi.positionInHistory = this.deliveredRequestsCounter;
                          // update local vector clock
                          this.updateLocalVectorClock(mi.vectorClock);
                      }
                  }
              } else {
                  //this request must be then ignored...
                  request.setIgnoreIt(true);
              }
  
              // udpate checkpoint index
              int ckptIndex = mi.checkpointIndex;
              if (ckptIndex > currentCheckpointIndex) {
                  this.nextMax = Math.max(this.nextMax, ckptIndex);
                  // mark the request that is orphan; it will be changed in awaited req in next ckpt
                  // oprhan du indexCkpt+1 a mi.ckptIndex compris
                  mi.isOrphanFor = (char) ckptIndex;
                  //System.out.println("" + this.ownerID + " will have orphans in checkpoint " + ckptIndex);
              }
          }
          return currentCheckpointIndex;
      }
  
      /*
       * Close and commit the current history if needed.
       */
      private void updateHistory(int index) {
          if (index > this.historyIndex) {
              // commit minimal history to the server
              this.commitHistories(this.checkpointIndex, this.deliveredRequestsCounter, true, true);
              if (this.completingCheckpoint) {
                  this.completingCheckpoint = false;
              }
          }
      }
  
      /*
       * Return true if the message m is significant for the protocol, i.e
       * if it's not sent by a non_ft body nor a half body
       */
      private boolean isSignificant(Message m) {
          return ((m.getMessageInfo() != null) && (!m.getMessageInfo().isFromHalfBody()));
      }
  
      /*
       * Update the local vector clock regarding the paramater.
       * if local[i]<param[i] or local[i] doesn't exist, then local[i]=param[i];
       */
      private void updateLocalVectorClock(Hashtable<UniqueID, MutableLong> vectorClock) {
          Enumeration<UniqueID> ids = vectorClock.keys();
          MutableLong localClock;
          MutableLong senderClock = null;
          UniqueID id = null;
          while (ids.hasMoreElements()) {
              id = ids.nextElement();
              localClock = (this.localVectorClock.get(id));
              senderClock = (vectorClock.get(id));
              if (localClock == null) {
                  // there is no clock for the AO id
                  this.localVectorClock.put(id, new MutableLong(senderClock.getValue()));
              } else if (localClock.isLessThan(senderClock)) {
                  // local clock is not uptodate
                  localClock.setValue(senderClock.getValue());
              }
          }
      }
  
      @Override
      public synchronized int onSendReplyBefore(Reply reply) {
          // set message info values
          this.forSentReply.checkpointIndex = (char) this.checkpointIndex;
          this.forSentReply.historyIndex = (char) this.historyIndex;
          this.forSentReply.incarnation = (char) this.incarnation;
          this.forSentReply.lastRecovery = (char) this.lastRecovery;
          this.forSentReply.isOrphanFor = Character.MAX_VALUE;
          this.forSentReply.fromHalfBody = false;
          this.forSentReply.vectorClock = null;
          reply.setMessageInfo(this.forSentReply);
  
          // output commit
          if (FTManagerCIC.isOCEnable && this.isOutputCommit(reply)) {
              try {
                  if (logger.isDebugEnabled()) {
                      logger.debug(this.ownerID + " is output commiting for reply " + reply);
                  }
                  this.storage.outputCommit(this.forSentReply);
              } catch (RemoteException e) {
                  logger.error("**ERROR** Cannot perform output commit");
                  e.printStackTrace();
              }
          }
  
          return 0;
      }
  
      @Override
      public synchronized int onSendRequestBefore(Request request) {
          // set message info values
          this.forSentRequest.checkpointIndex = (char) this.checkpointIndex;
          this.forSentRequest.historyIndex = (char) this.historyIndex;
          this.forSentRequest.incarnation = (char) this.incarnation;
          this.forSentRequest.lastRecovery = (char) this.lastRecovery;
          this.forSentRequest.isOrphanFor = Character.MAX_VALUE;
          this.forSentRequest.fromHalfBody = false;
          if (FTManagerCIC.isOCEnable) {
              this.forSentRequest.vectorClock = this.localVectorClock;
          }
          request.setMessageInfo(this.forSentRequest);
  
          // output commit
          if (FTManagerCIC.isOCEnable && this.isOutputCommit(request)) {
              try {
                  if (logger.isDebugEnabled()) {
                      logger.debug(this.ownerID + " is output commiting for request " + request);
                  }
                  this.storage.outputCommit(this.forSentRequest);
              } catch (RemoteException e) {
                  logger.error("**ERROR** Cannot perform output commit");
                  e.printStackTrace();
              }
          }
          return 0;
      }
  
      /*
       * Return true if the sending of the paramter message is an output commit
       * ** TEST IMPLEMENTATION **
       */
      private boolean isOutputCommit(Message m) {
          if (FTManagerCIC.isOCEnable) {
              Request r = (Request) m;
              if (r.getMethodName().equals("logEvent")) {
                  return true;
              } else {
                  return false;
              }
          } else {
              return false;
          }
      }
  
      @Override
      public synchronized int onSendReplyAfter(Reply reply, int rdvValue, UniversalBody destination) {
          // if return value is RESEND, receiver have to recover --> resend the message
          if (rdvValue == FTManagerCIC.RESEND_MESSAGE) {
              try {
                  reply.setIgnoreIt(false);
                  Thread.sleep(FTManager.TIME_TO_RESEND);
                  int rdvValueBis = sendReply(reply, destination);
                  return this.onSendReplyAfter(reply, rdvValueBis, destination);
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }
          }
          int currentCheckpointIndex = this.checkpointIndex;
  
          // update checkpoint index
          if (rdvValue > currentCheckpointIndex) {
              this.nextMax = Math.max(this.nextMax, rdvValue);
              // log this in-transit message
              this.extendReplyLog(rdvValue);
              // must make a deep copy of result !
              try {
                  Reply toLog = null;
  
                  //try {
                  //    toLog = new ReplyImpl(reply.getSourceBodyID(),
                  //            reply.getSequenceNumber(), reply.getMethodName(),
                  //            (MethodCallResult) Utils.makeDeepCopy(reply.getResult()),
                  //            owner.getProActiveSecurityManager());
                  //} catch (SecurityNotAvailableException e1) {
                  toLog = new ReplyImpl(reply.getSourceBodyID(), reply.getSequenceNumber(), reply
                          .getMethodName(), (MethodCallResult) Utils.makeDeepCopy(reply.getResult()), null);
                  //}
                  ReplyLog log = new ReplyLog(toLog, destination.getRemoteAdapter());
                  for (int i = currentCheckpointIndex + 1; i <= rdvValue; i++) {
                      (this.replyToResend.get(new Integer(i))).add(log);
                  }
              } catch (IOException e) {
                  e.printStackTrace();
              }
          }
          return 0;
      }
  
      @Override
      public synchronized int onSendRequestAfter(Request request, int rdvValue, UniversalBody destination)
              throws RenegotiateSessionException, CommunicationForbiddenException {
          //	if return value is RESEDN, receiver have to recover --> resend the message
          if (rdvValue == FTManagerCIC.RESEND_MESSAGE) {
              try {
                  request.resetSendCounter();
                  request.setIgnoreIt(false);
                  Thread.sleep(FTManager.TIME_TO_RESEND);
                  int rdvValueBis = sendRequest(request, destination);
                  return this.onSendRequestAfter(request, rdvValueBis, destination);
              } catch (RenegotiateSessionException e1) {
                  throw e1;
              } catch (InterruptedException e) {
                  e.printStackTrace();
              }
          }
  
          int currentCheckpointIndex = this.checkpointIndex;
  
          // update checkpoint index
          if (rdvValue > currentCheckpointIndex) {
              this.nextMax = Math.max(this.nextMax, rdvValue);
              // log this in-transit message in the rdvValue-currentIndex next checkpoints
              this.extendRequestLog(rdvValue);
              try {
                  //must make deep copy of paramteres
                  request.getMethodCall().makeDeepCopyOfArguments();
                  //must reset the send counter (this request has not been forwarded)
                  request.resetSendCounter();
                  RequestLog log = new RequestLog(request, destination.getRemoteAdapter());
                  for (int i = currentCheckpointIndex + 1; i <= rdvValue; i++) {
                      //System.out.println(""+this.ownerID + " logs a request for " + destination.getID());
                      this.requestToResend.get(new Integer(i)).add(log);
                  }
              } catch (IOException e) {
                  e.printStackTrace();
              }
          }
          return 0;
      }
  
      @Override
      public int onServeRequestBefore(Request request) {
          // checkpoint if needed
          while (this.haveToCheckpoint()) {
              this.checkpoint(request);
          }
  
          // update the last served request index only if needed
          if (FTManagerCIC.isOCEnable) {
              MessageInfo mi = request.getMessageInfo();
              if (mi != null) {
                  long requestIndex = ((MessageInfoCIC) (mi)).positionInHistory;
                  if (this.lastServedRequestIndex.getValue() < requestIndex) {
                      this.lastServedRequestIndex.setValue(requestIndex);
                  }
              }
          }
          return 0;
      }
  
      @Override
      public int onServeRequestAfter(Request request) {
          return 0;
      }
  
      // Active Object is created but not started
      @Override
      public int beforeRestartAfterRecovery(CheckpointInfo ci, int inc) {
          CheckpointInfoCIC cic = (CheckpointInfoCIC) ci;
          BlockingRequestQueue queue = (owner).getRequestQueue();
          int index = cic.checkpointIndex;
  
          //	reinit ft values
          this.history = new Vector<UniqueID>();
          this.completingCheckpoint = false;
          this.lastCommitedIndex = cic.lastCommitedIndex;
          // historized requests are supposed to be "already received"
          this.deliveredRequestsCounter = cic.lastCommitedIndex; //cic.lastRcvdRequestIndex;
          // new history then begin at the end of the history of the checkpoint
  
          this.historyBaseIndex = cic.lastCommitedIndex + 1; //;cic.lastRcvdRequestIndex+1;
  
          // HERE, we need a proof that running in "histo mode" is equivalent that
          // running in normal mode from the end of the histo.
          this.awaitedRequests = new Vector<AwaitedRequest>();
          this.replyToResend = new Hashtable<Integer, Vector<ReplyLog>>();
          this.requestToResend = new Hashtable<Integer, Vector<RequestLog>>();
          this.checkpointIndex = index;
          this.nextMax = index;
          this.checkpointTimer = System.currentTimeMillis();
          this.historyIndex = index;
          this.lastRecovery = index;
          this.incarnation = inc;
  
          //add pending request to reuqestQueue
          Request pendingRequest = cic.pendingRequest;
  
          //pending request could be null with OOSPMD synchronization
  