socketsourceroutemanager.java

pastry的java实现的2.0b版

    // the remote address of this manager
    /**
     * DESCRIBE THE FIELD
     */
    protected SocketNodeHandle address;

    /**
     * the current best route to this remote address if best == null, we are
     * already in a CheckDead, which means we are searching for a path
     */
    protected SourceRoute best;

    /**
     * the queue of messages waiting for a route of SocketBuffer
     */
    protected LinkedList queue;

    // the queue of appSockets waiting for a connection
    /**
     * DESCRIBE THE FIELD
     */
    protected LinkedList pendingAppSockets;

    // the list of known route -> managers for this manager
    /**
     * DESCRIBE THE FIELD
     */
    protected HashMap routes;

    // the current liveness of this address
    /**
     * DESCRIBE THE FIELD
     */
    protected int liveness;

    // the last time this address was pinged
    /**
     * DESCRIBE THE FIELD
     */
    protected long updated;

    /**
     * Constructor, given an address and whether or not it should attempt to
     * find the best route
     *
     * @param address The address
     * @param search Whether or not the manager should try and find a route
     */
    public AddressManager(SocketNodeHandle address, boolean search) {
      this.address = address;
      this.queue = new LinkedList();
      this.pendingAppSockets = new LinkedList();
      this.routes = new HashMap();
      this.liveness = SocketNodeHandle.LIVENESS_SUSPECTED;
      this.updated = 0L;

      if (logger.level <= Logger.FINE) {
        logger.log("(SSRM) ADDRESS MANAGER CREATED AT " + localAddress + " FOR " + address);
      }

      if (search) {
        getRouteManager(SourceRoute.build(address.eaddress)).checkLiveness();
        this.updated = spn.getEnvironment().getTimeSource().currentTimeMillis();
      }
    }

    /**
     * Method which returns the route manager for the given route
     *
     * @param route The route
     * @return THe manager
     * @throws IllegalArgumentException if route is null
     */
    protected SourceRouteManager getRouteManager(SourceRoute route) {
      if (route == null) {
        throw new IllegalArgumentException("route is null in " + toString());
      }

      SourceRouteManager result = (SourceRouteManager) routes.get(route);

      if (result == null) {
        result = new SourceRouteManager(route);
        routes.put(route, result);
      }

      return result;
    }

    /**
     * Method which returns the last cached liveness value for the given
     * address. If there is no cached value, then true is returned.
     *
     * @return The Alive value
     */
    public int getLiveness() {
      return liveness;
    }

    /**
     * Internal method which marks this address as being alive. If we were dead
     * before, it sends an update out to the observers. best must be non-null
     *
     * @throws IllegalStateException if best is null.
     */
    protected void setAlive() {
      if (best == null) {
        throw new IllegalStateException("best is null in " + toString());
      }

      // we can now send any pending messages
      while (!queue.isEmpty()) {
        getRouteManager(best).send((SocketBuffer) queue.removeFirst());
      }
      // we can now send any pending messages
      while (!pendingAppSockets.isEmpty()) {
        PendingAppSocket pas = (PendingAppSocket) pendingAppSockets.removeFirst();
        getRouteManager(best).connect(pas.appAddress, pas.receiver, 0);
      }
      if (queue.isEmpty() && pendingAppSockets.isEmpty()) {
        hardLinks.remove(this);
      }

      switch (liveness) {
        case SocketNodeHandle.LIVENESS_DEAD:
          liveness = SocketNodeHandle.LIVENESS_ALIVE;
          if (address != null) {
            address.update(SocketNodeHandle.DECLARED_LIVE);
          }
          if (logger.level <= Logger.FINE) {
            logger.log("COUNT: " + localAddress + " Found address " + address + " to be alive again.");
          }
          break;
        case SocketNodeHandle.LIVENESS_SUSPECTED:
          liveness = SocketNodeHandle.LIVENESS_ALIVE;
          if (logger.level <= Logger.FINE) {
            logger.log("COUNT: " + localAddress + " Found address " + address + " to be unsuspected.");
          }
          break;
        case SocketNodeHandle.LIVENESS_DEAD_FOREVER:
          if (logger.level <= Logger.WARNING) {
            logger.log("ERROR: Found dead-forever handle to " + address + " to be alive again!");
          }
          break;
      }
    }

    /**
     * Internal method which marks this address as being suspected.
     */
    protected void setSuspected() {
      switch (liveness) {
        case SocketNodeHandle.LIVENESS_ALIVE:
          liveness = SocketNodeHandle.LIVENESS_SUSPECTED;
          if (logger.level <= Logger.FINE) {
            logger.log("COUNT: " + spn.getEnvironment().getTimeSource().currentTimeMillis() + " " + localAddress + " Found address " + address + " to be suspected.");
          }
          break;
        case SocketNodeHandle.LIVENESS_DEAD:
          if (logger.level <= Logger.WARNING) {
            logger.log("ERROR: Found node handle " + address + " to be suspected from dead - should not happen!");
          }
          break;
        case SocketNodeHandle.LIVENESS_DEAD_FOREVER:
          if (logger.level <= Logger.WARNING) {
            logger.log("ERROR: Found node handle " + address + " to be suspected from dead forever - should never ever happen!");
          }
          break;
      }

      // and finally we can now reroute any route messages
      Object[] array = queue.toArray();

      for (int i = 0; i < array.length; i++) {
        SocketBuffer sb = (SocketBuffer) array[i];
        if (sb.isRouteMessage()) {
          if (sb.getOptions().multipleHopsAllowed() && sb.getOptions().rerouteIfSuspected()) {
            if (logger.level <= Logger.FINE) {
              logger.log("REROUTE: Rerouting message " + sb + " due to suspected next hop " + address);
            }
            reroute(address.eaddress, sb);
            queue.remove(sb);
          }
        }
      }
      if (queue.isEmpty() && pendingAppSockets.isEmpty()) {
        hardLinks.remove(this);
      }
    }

    /**
     * Internal method which marks this address as being dead. If we were alive
     * or suspected before, it sends an update out to the observers.
     */
    protected void setDead() {
      switch (liveness) {
        case SocketNodeHandle.LIVENESS_DEAD:
          return;
        case SocketNodeHandle.LIVENESS_DEAD_FOREVER:
          if (logger.level <= Logger.WARNING) {
            logger.log("ERROR: Found node handle " + address + " to be dead from dead forever - should not happen!");
          }
          break;
        default:
          this.best = null;
          this.liveness = SocketNodeHandle.LIVENESS_DEAD;
          if (address != null) {
            address.update(SocketNodeHandle.DECLARED_DEAD);
          }
          manager.declaredDead(address.eaddress);
          if (logger.level <= Logger.FINE) {
            logger.log("COUNT: " + localAddress + " Found address " + address + " to be dead.");
          }
          break;
      }

      purgeQueue();
    }

    /**
     * Internal method which marks this address as being dead. If we were alive
     * or suspected before, it sends an update out to the observers.
     */
    protected void setDeadForever() {
      switch (liveness) {
        case SocketNodeHandle.LIVENESS_DEAD_FOREVER:
          return;
        case SocketNodeHandle.LIVENESS_DEAD:
          this.liveness = SocketNodeHandle.LIVENESS_DEAD_FOREVER;
          if (logger.level <= Logger.FINE) {
            logger.log("COUNT: " + localAddress + " Found address " + address + " to be dead forever.");
          }
          break;
        default:
          this.best = null;
          this.liveness = SocketNodeHandle.LIVENESS_DEAD_FOREVER;
          if (address != null) {
            address.update(SocketNodeHandle.DECLARED_DEAD);
          }
          if (logger.level <= Logger.FINE) {
            logger.log("COUNT: " + localAddress + " Found address " + address + " to be dead forever.");
          }
          break;
      }
      purgeQueue();
    }

    /**
     * Method which returns the last cached proximity value for the given
     * address. If there is no cached value, then DEFAULT_PROXIMITY is returned.
     *
     * @return The ping value to the remote address
     */
    public int proximity() {
      if (best == null) {
        return SocketNodeHandle.DEFAULT_PROXIMITY;
      } else {
        return getRouteManager(best).proximity();
      }
    }

    /**
     * This method should be called when a known route is declared alive.
     *
     * @param route The now-live route
     */
    protected synchronized void markAlive(SourceRoute route) {
      getRouteManager(route).markAlive();

      // first, we check and see if we have no best route (this can happen if the best just died)
      if (best == null) {
        if (logger.level <= Logger.FINE) {
          logger.log("(SSRM) No previous best route existed to " + address + " route " + route + " is now the best");
        }
        best = route;
      }

      // now, we check if the route is (a) shorter, or (b) the same length but quicker
      // if se, we switch our best route to that one
      if ((best.getNumHops() > route.getNumHops()) ||
        ((best.getNumHops() == route.getNumHops()) &&
        (getRouteManager(best).proximity() > getRouteManager(route).proximity()))) {
        if (logger.level <= Logger.FINE) {
          logger.log("(SSRM) Route " + route + " is better than previous best route " + best + " - replacing");
        }

        best = route;

        if (address != null) {
          address.update(SocketNodeHandle.PROXIMITY_CHANGED);
        }
      }

      // finally, mark this address as alive
      setAlive();
    }

    /**
     * This method should be called when a known route is declared suspected.
     *
     * @param route The now-suspected route
     */
    protected synchronized void markSuspected(SourceRoute route) {
      getRouteManager(route).markSuspected();

      // mark this address as suspected, if this is currently the best route
      if (((best == null) || (best.equals(route))) &&
      // because we set the best == null when we didn't have a route
      (liveness < SocketNodeHandle.LIVENESS_DEAD)) {
        // don't promote the node
        setSuspected();
      }
    }

    /**
     * This method should be called when a known route is declared dead.
     *
     * @param route The now-dead route
     */
    protected synchronized void markDead(SourceRoute route) {
      getRouteManager(route).markDead();

      // if we're already dead, who cares
      if (liveness >= SocketNodeHandle.LIVENESS_DEAD) {
        return;
      }

      // if this route was the best, or if we have no best, we need to
      // look for alternate routes - if all alternates are now dead,
      // we mark ourselves as dead
      if ((best == null) || (route.equals(best))) {
        best = null;

        SourceRoute[] routes = getAllRoutes(route.getLastHop());
        boolean found = false;

        for (int i = 0; i < routes.length; i++) {
          if (getRouteManager(routes[i]).checkLiveness()) {
            found = true;
          }
        }

        if (!found) {
          setDead();
        }
      }
    }

    /**
     * This method should be called when a known node is declared dead - this is
     * ONLY called when a new epoch of that node is detected. Note that this
     * method is silent - no checks are done. Caveat emptor.
     *
     */
    protected synchronized void markDeadForever() {
      this.best = null;