total_token.java

来自「JGRoups源码」· Java 代码 · 共 1,160 行 · 第 1/3 页

      {
         int queueSize = newMessagesQueue.size();

         if (queueSize <= 0)
         {
            return 0;
         }

         else if (queueSize > allowedCount)
         {
            sendList = new ArrayList(newMessagesQueue.subList(0, allowedCount));
            newMessagesQueue.removeAll(sendList);
         }

         else
         {
            sendList = new ArrayList();
            sendList.addAll(newMessagesQueue);
            newMessagesQueue.clear();
         }
      }

      long tokenSeq = token.getHighestSequence();

      for (Iterator iterator = sendList.iterator(); iterator.hasNext();)
      {
         Message m = (Message) iterator.next();
         m.setSrc(localAddress);
         m.setDest(null); // mcast address
         m.putHeader(getName(), new TotalTokenHeader(++tokenSeq));
         receivedMessagesQueue.put(new Long(tokenSeq), m);
         passDown(new Event(Event.MSG, m));
      }

      if (token.getHighestSequence() == token.getAllReceivedUpto())
      {
         token.setAllReceivedUpto(tokenSeq);
      }
      token.setHighestSequence(tokenSeq);
      return sendList.size();
   }


   /**
    * TokenTransmitter thread runs this method after receiving token.
    * Thread is possibly blocked if up-handler thread is currently running
    * through this stack i.e delivering an Event. Up-hanlder thread will
    * notify blocked TokenTransmitter thread when it has delivered current
    * Event so TokenTransmitter can proceed.
    * TokenTransmitter thread in turn notifies possibly blocked up-handler thread
    * when token has left the stack. Thus TokenTransmitter and up-hadler thread
    * coordinate their access to shared variables(receivedMessageQueue and myAru).
    *
    * tokenReceived method and subsequent methods called from tokenReceived represent
    * in some parts the totaly ordered algorithm presented in Amir's paper (see class
    * header for link)
    *
    *
    *
    */
   private void tokenReceived(RingToken token)
   {


       if(log.isInfoEnabled()) log.info(token.toString());
       if(log.isDebugEnabled()) log.debug(getState());


       flowControl.setBacklog(newMessagesQueue.size());
      flowControl.updateWindow(token);


      blockSenderIfRequired();
      unBlockSenderIfAcceptable();


      long tokensAru = 0;
      int broadcastCount = 0;
      int rebroadcastCount = 0;
      synchronized (mutex)
      {
         if (!tokenSeen)
         {
            long lastRoundAru = token.getHighestSequence() - token.getLastRoundBroadcastCount();
            if (myAru < token.getAllReceivedUpto())
            {
               myAru = lastRoundAru;
            }
            //if(log.isInfoEnabled()) log.info("TOTAL_TOKEN.tokenReceived()", "tokenSeen " + myAru);
            tokenSeen = true;
         }

         if (token.getType() == RingToken.RECOVERY)
         {
            highestSeenSeq = token.getHighestSequence();
            if (highestSeenSeq == myAru)
            {
               if(log.isInfoEnabled()) log.info("member node recovered");
               token.addRecoveredMember(localAddress);
            }
         }

         updateTokenRtR(token);

         int allowedToBroadcast = flowControl.getAllowedToBroadcast(token);
         rebroadcastCount = rebroadcastMessages(token);
         allowedToBroadcast -= rebroadcastCount;


          if(log.isInfoEnabled()) log.info("myAllReceivedUpto" + myAru);
          if(log.isInfoEnabled()) log.info("allowedToBroadcast" + allowedToBroadcast);
          if(log.isInfoEnabled()) log.info("newMessagesQueue.size()" + newMessagesQueue.size());

          tokensAru = token.getAllReceivedUpto();

         if (myAru < tokensAru ||
                 localAddress.equals(token.getAruId()) ||
                 token.getAruId() == null)
         {
            token.setAllReceivedUpto(myAru);
            if (token.getAllReceivedUpto() == token.getHighestSequence())
            {
               token.setAruId(null);
            }
            else
            {
               token.setAruId(localAddress);
            }
         }
         if (allowedToBroadcast > 0 && token.getType() == RingToken.OPERATIONAL)
         {
            broadcastCount = broadcastMessages(allowedToBroadcast, token);
         }

         if (tokensAru > lastRoundTokensAru)
         {
            removeStableMessages(receivedMessagesQueue, lastRoundTokensAru);
         }

      } //end synchronized

      //give CPU some breath
      Util.sleep(sleepTime);

      token.incrementTokenSequence();
      token.addLastRoundBroadcastCount(broadcastCount - lastRoundTransmitCount);
      token.addBacklog(flowControl.getBacklogDifference());
      flowControl.setPreviousBacklog();
      lastRoundTransmitCount = broadcastCount;
      lastRoundRebroadcastCount = rebroadcastCount;
      lastRoundTokensAru = tokensAru;
   }

   /**
    *
    * Rebroadcasts messages specified in token's retransmission
    * request list if those messages are available in this stack.
    * Returns number of rebroadcasted messages.
    *
    */
   private int rebroadcastMessages(RingToken token)
   {
      int rebroadCastCount = 0;
      Collection rexmitRequests = token.getRetransmissionRequests();
      if (rexmitRequests.size() > 0)
      {
         Collection rbl = getRebroadcastList(rexmitRequests);
         rebroadCastCount = rbl.size();
         if (rebroadCastCount > 0)
         {

             if(log.isInfoEnabled()) log.info("rebroadcasting " + rbl);

             Long s = null;
            for (Iterator iterator = rbl.iterator(); iterator.hasNext();)
            {
               s = (Long) iterator.next();
               Message m = (Message) receivedMessagesQueue.get(s);
               passDown(new Event(Event.MSG, m));
            }
         }
      }
      return rebroadCastCount;
   }


   private void invalidateOnTokenloss()
   {
      lastRoundTransmitCount = 0;
      flowControl.invalidate();
   }

   /**
    * Checks if the down pending queue's (newMessagesQueue) size is
    * greater than blockSendingBacklogThreshold specified in the properties.
    * If it is, client's sending thread is effectively blocked until
    * down pending queue's size drops below unblockSendingBacklogThreshold.
    *
    *
    */
   private void blockSenderIfRequired()
   {
      if (!senderBlocked && flowControl.getBacklog() > blockSendingBacklogThreshold)
      {
          synchronized(block_sending) {
              senderBlocked = true;
              while(senderBlocked) {
                  try {
                      block_sending.wait();
                  }
                  catch(InterruptedException e) {
                  }
              }
          }
      }
   }

   /**
    * Checks if the down pending queue's (newMessagesQueue) size is
    * smaller than unblockSendingBacklogThreshold specified in the properties.
    * If it is, client's sending thread is effectively unblocked enabling
    * new messages to be queued for transmission.
    *
    *
    */
   private void unBlockSenderIfAcceptable()
   {
      if (senderBlocked && flowControl.getBacklog() < unblockSendingBacklogThreshold)
      {
          synchronized(block_sending) {
              senderBlocked = false;
              block_sending.notifyAll();
          }
      }

   }

   /**
    * Removes messages determined to be stable(i.e seen by all members)
    * from the specified queue.  If the client also clears all reference
    * to these messages (in application space) they become eligible for garabge
    * collection.
    *
    *
    */

   private void removeStableMessages(TreeMap m, long upToSeq)
   {

      if (m.size() > 0)
      {
         long first = ((Long) m.firstKey()).longValue();
         if (first > upToSeq)
         {
            upToSeq = first;
         }


          if(log.isDebugEnabled()) log.debug("cutting queue first key " + m.firstKey()
                      + " cut at " + upToSeq + " last key " + m.lastKey());
          SortedMap stable = m.headMap(new Long(upToSeq));
         stable.clear();
      }
   }

   /**
    * Determines a subset of message sequence numbers
    * available for retransmission from this stack.
    *
    */
   private Collection getRebroadcastList(Collection rtr)
   {
      ArrayList rebroadcastList = new ArrayList(rtr);
      rebroadcastList.retainAll(receivedMessagesQueue.keySet());
      rtr.removeAll(rebroadcastList);
      Collections.sort(rebroadcastList);
      return rebroadcastList;
   }

   /**
    * TokenTransimitter thread transmits the token to the next member
    * in the logical ring as well as it receives token from the previous
    * member in the ring. Smoothed ring roundtrip time is computed
    * in order to detect token loss.  If the timeout expires AND this
    * stack has received SUSPECT message, recovery protocol is invoked.
    * See recover method for details.
    *
    */
   private class TokenTransmitter extends Thread
   {
      long rtt = 0;
      long timer;
      double srtt = 1000; //1 second to start
      final double a = 0.09;
      final int timeoutFactor = 10;
      volatile boolean running = false;

      private TokenTransmitter()
      {
         super(Util.getGlobalThreadGroup(), "TokenTransmitter");
         resetTimeout();
         running = true;
      }

      private void shutDown()
      {
          running = false;
      }

      private void recalculateTimeout()
      {
         long now = System.currentTimeMillis();
         if (timer > 0)
         {
            rtt = now - timer;
            srtt = (1 - a) * srtt + a * rtt;
         }
      }

      private double getTimeout()
      {
         return srtt * timeoutFactor;
      }

      private void resetTimeout()
      {
         timer = System.currentTimeMillis();
      }

      private boolean isRecoveryCompleted(RingToken token)
      {
          return liveMembersInRecovery.equals(token.getRecoveredMembers());
      }

      public void run()
      {
         while (running)
         {
            RingToken token = null;
            int timeout = 0;

             if(node == null) {
                 // sleep some time, then retry
                 Util.sleep(500);
                 continue;
             }

            try
            {
               timeout = (int) getTimeout();

                  if(log.isInfoEnabled()) log.info("timeout(ms)=" + timeout);

               token = (RingToken) node.receiveToken(timeout);

               if (token.getType() == RingToken.OPERATIONAL &&
                       state == RECOVERY_STATE)
               {
                  state = OPERATIONAL_STATE;
               }

               tokenReceived(token);
               recalculateTimeout();

               if (token.getType() == RingToken.RECOVERY &&
                       isRecoveryCompleted(token))
               {

                     if(log.isInfoEnabled()) log.info("all members recovered, injecting operational token");
                  token.setType(RingToken.OPERATIONAL);
               }
               node.passToken(token);
               resetTimeout();
            }
            catch (TokenLostException tle)
            {
               invalidateOnTokenloss();
               state = RECOVERY_STATE;
               recover();
            }
         }
      }
   }
}