glacierimpl.java

pastry的java实现的2.0b版

    bulkRefreshPatchInterval = p.getLong("p2p_glacier_bulkRefreshPatchInterval");
    bulkRefreshPatchRetries = p.getInt("p2p_glacier_bulkRefreshPatchRetries");

    bucketTokensPerSecond = p.getLong("p2p_glacier_bucketTokensPerSecond");
    bucketMaxBurstSize = p.getLong("p2p_glacier_bucketMaxBurstSize");

    jitterRange = p.getDouble("p2p_glacier_jitterRange");

    statisticsReportInterval = p.getLong("p2p_glacier_statisticsReportInterval");

    maxActiveRestores = p.getInt("p2p_glacier_maxActiveRestores");


    this.fragmentStorage = fragmentStorageArg;
    this.neighborStorage = neighborStorageArg;
    this.trashStorage = null;
    this.policy = policyArg;
    this.node = nodeArg;
    this.instance = instanceArg;
    this.endpoint = node.buildEndpoint(this, instance);
    this.endpoint.setDeserializer(
      new MessageDeserializer() {

        public Message deserialize(InputBuffer buf, short type, byte priority,
                                   NodeHandle sender) throws IOException {
          switch (type) {
            case GlacierDataMessage.TYPE:
              return GlacierDataMessage.build(buf, endpoint);
            case GlacierFetchMessage.TYPE:
              return GlacierFetchMessage.build(buf, endpoint);
            case GlacierNeighborRequestMessage.TYPE:
              return GlacierNeighborRequestMessage.build(buf, endpoint);
            case GlacierNeighborResponseMessage.TYPE:
              return GlacierNeighborResponseMessage.build(buf, endpoint);
            case GlacierQueryMessage.TYPE:
              return GlacierQueryMessage.build(buf, endpoint);
            case GlacierRangeForwardMessage.TYPE:
              return GlacierRangeForwardMessage.build(buf, endpoint);
            case GlacierRangeQueryMessage.TYPE:
              return GlacierRangeQueryMessage.build(buf, endpoint);
            case GlacierRangeResponseMessage.TYPE:
              return GlacierRangeResponseMessage.build(buf, endpoint);
            case GlacierRefreshCompleteMessage.TYPE:
              return GlacierRefreshCompleteMessage.build(buf, endpoint);
            case GlacierRefreshPatchMessage.TYPE:
              return GlacierRefreshPatchMessage.build(buf, endpoint);
            case GlacierRefreshProbeMessage.TYPE:
              return GlacierRefreshProbeMessage.build(buf, endpoint);
            case GlacierRefreshResponseMessage.TYPE:
              return GlacierRefreshResponseMessage.build(buf, endpoint);
            case GlacierResponseMessage.TYPE:
              return GlacierResponseMessage.build(buf, endpoint);
            case GlacierSyncMessage.TYPE:
              return GlacierSyncMessage.build(buf, endpoint);
          }
          throw new IllegalArgumentException("Unknown type:" + type);
        }

      });
    this.contentDeserializer = new JavaPastContentDeserializer();
    this.contentHandleDeserializer = new JavaPastContentHandleDeserializer();
    this.numFragments = numFragmentsArg;
    this.numSurvivors = numSurvivorsArg;
    this.factory = factoryArg;
    this.responsibleRange = null;
    this.nextUID = 0;
    this.continuations = new Hashtable();
    this.pendingTraffic = new Hashtable();
    this.timer = null;
    this.nextContinuationTimeout = -1;
    this.statistics = new GlacierStatistics(tagMax, environment);
    this.listeners = new Vector();
    this.numActiveRestores = new int[1];
    this.numActiveRestores[0] = 0;
    this.currentFragmentRequestTimeout = fragmentRequestTimeoutDefault;
    this.tokenBucket = 0;
    this.bucketLastUpdated = environment.getTimeSource().currentTimeMillis();
    determineResponsibleRange();
    endpoint.register();
  }

  /**
   * Gets the HashInput attribute of the GlacierImpl object
   *
   * @param vkey DESCRIBE THE PARAMETER
   * @param expiration DESCRIBE THE PARAMETER
   * @return The HashInput value
   */
  private byte[] getHashInput(VersionKey vkey, long expiration) {
    byte[] a = vkey.toByteArray();
    byte[] b = new byte[a.length + 8];
    for (int i = 0; i < a.length; i++) {
      b[i] = a[i];
    }

    b[a.length + 0] = (byte) (0xFF & (expiration >> 56));
    b[a.length + 1] = (byte) (0xFF & (expiration >> 48));
    b[a.length + 2] = (byte) (0xFF & (expiration >> 40));
    b[a.length + 3] = (byte) (0xFF & (expiration >> 32));
    b[a.length + 4] = (byte) (0xFF & (expiration >> 24));
    b[a.length + 5] = (byte) (0xFF & (expiration >> 16));
    b[a.length + 6] = (byte) (0xFF & (expiration >> 8));
    b[a.length + 7] = (byte) (0xFF & (expiration));

    return b;
  }

  /**
   * Gets the UID attribute of the GlacierImpl object
   *
   * @return The UID value
   */
  protected int getUID() {
    return nextUID++;
  }

  /**
   * Determines the point in the ring where a particular fragment should be
   * stored.
   *
   * @param objectKey Key of the original object (from PAST)
   * @param fragmentNr Fragment number (0..n-1)
   * @param version DESCRIBE THE PARAMETER
   * @return The location of the fragment
   */
  private Id getFragmentLocation(Id objectKey, int fragmentNr, long version) {
    double totalOffset = (((float) fragmentNr) / ((float) numFragments)) + version * (1.0 / 2.7182821);
    return objectKey.addToId(factory.buildIdDistance(getDistance(totalOffset - Math.floor(totalOffset))));
  }

  /**
   * Gets the FragmentLocation attribute of the GlacierImpl object
   *
   * @param fkey DESCRIBE THE PARAMETER
   * @return The FragmentLocation value
   */
  private Id getFragmentLocation(FragmentKey fkey) {
    return getFragmentLocation(
      fkey.getVersionKey().getId(),
      fkey.getFragmentID(),
      fkey.getVersionKey().getVersion()
      );
  }

  /**
   * Gets the NeighborRanges attribute of the GlacierImpl object
   *
   * @return The NeighborRanges value
   */
  public Id[][] getNeighborRanges() {
    Iterator iter = neighborStorage.scan().getIterator();
    Vector ccwIDs = new Vector();
    Vector cwIDs = new Vector();
    Id myID = getLocalNodeHandle().getId();

    while (iter.hasNext()) {
      Id thisNeighbor = (Id) iter.next();
      if (myID.clockwise(thisNeighbor)) {
        cwIDs.add(thisNeighbor);
      } else {
        ccwIDs.add(thisNeighbor);
      }
    }

    for (int j = 0; j < 2; j++) {
      Vector v = (j == 0) ? cwIDs : ccwIDs;
      boolean madeProgress = true;
      while (madeProgress) {
        madeProgress = false;
        for (int i = 0; i < (v.size() - 1); i++) {
          if (((Id) v.elementAt(i + 1)).clockwise((Id) v.elementAt(i))) {
            Object h = v.elementAt(i);
            v.setElementAt(v.elementAt(i + 1), i);
            v.setElementAt(h, i + 1);
            madeProgress = true;
          }
        }
      }
    }

    Vector allIDs = new Vector();
    allIDs.addAll(ccwIDs);
    allIDs.add(myID);
    allIDs.addAll(cwIDs);

    Id[][] result = new Id[allIDs.size()][3];
    for (int i = 0; i < allIDs.size(); i++) {
      Id currentElement = (Id) allIDs.elementAt(i);
      Id cwId;
      Id ccwId;

      if (i > 0) {
        Id previousElement = (Id) allIDs.elementAt(i - 1);
        ccwId = previousElement.addToId(previousElement.distanceFromId(currentElement).shiftDistance(1, 0));
      } else {
        ccwId = currentElement;
      }

      if (i < (allIDs.size() - 1)) {
        Id nextElement = (Id) allIDs.elementAt(i + 1);
        cwId = currentElement.addToId(currentElement.distanceFromId(nextElement).shiftDistance(1, 0));
      } else {
        cwId = currentElement;
      }

      result[i][0] = ccwId;
      result[i][1] = currentElement;
      result[i][2] = cwId;
    }

    return result;
  }

  /**
   * Gets the ReplicationFactor attribute of the GlacierImpl object
   *
   * @return The ReplicationFactor value
   */
  public int getReplicationFactor() {
    return 1;
  }

  /**
   * Gets the LocalNodeHandle attribute of the GlacierImpl object
   *
   * @return The LocalNodeHandle value
   */
  public NodeHandle getLocalNodeHandle() {
    return endpoint.getLocalNodeHandle();
  }

  /**
   * Gets the TrashSize attribute of the GlacierImpl object
   *
   * @return The TrashSize value
   */
  public long getTrashSize() {
    if (trashStorage == null) {
      return 0;
    }

    return trashStorage.getStorage().getTotalSize();
  }

  /**
   * Gets the Environment attribute of the GlacierImpl object
   *
   * @return The Environment value
   */
  public Environment getEnvironment() {
    return environment;
  }

  /**
   * Gets the Instance attribute of the GlacierImpl object
   *
   * @return The Instance value
   */
  public String getInstance() {
    return instance;
  }

  /**
   * Sets the Trashcan attribute of the GlacierImpl object
   *
   * @param trashStorage The new Trashcan value
   */
  public void setTrashcan(StorageManager trashStorage) {
    this.trashStorage = trashStorage;
  }

  /**
   * Schedule a timer event
   *
   * @param timeoutMsec Length of the delay (in milliseconds)
   */
  private void setTimer(int timeoutMsec) {
    timer = endpoint.scheduleMessage(new GlacierTimeoutMessage(0, getLocalNodeHandle()), timeoutMsec);
  }

  /**
   * Sets the SyncInterval attribute of the GlacierImpl object
   *
   * @param syncIntervalSec The new SyncInterval value
   */
  public void setSyncInterval(int syncIntervalSec) {
    this.syncInterval = syncIntervalSec * SECONDS;
  }

  /**
   * Sets the SyncMaxFragments attribute of the GlacierImpl object
   *
   * @param syncMaxFragments The new SyncMaxFragments value
   */
  public void setSyncMaxFragments(int syncMaxFragments) {
    this.syncMaxFragments = syncMaxFragments;
  }

  /**
   * Sets the RateLimit attribute of the GlacierImpl object
   *
   * @param rps The new RateLimit value
   */
  public void setRateLimit(int rps) {
    this.rateLimitedRequestsPerSecond = rps;
  }

  /**
   * Sets the NeighborTimeout attribute of the GlacierImpl object
   *
   * @param neighborTimeoutMin The new NeighborTimeout value
   */
  public void setNeighborTimeout(long neighborTimeoutMin) {
    this.neighborTimeout = neighborTimeoutMin * MINUTES;
  }

  /**
   * Sets the BandwidthLimit attribute of the GlacierImpl object
   *
   * @param bytesPerSecond The new BandwidthLimit value
   * @param maxBurst The new BandwidthLimit value
   */
  public void setBandwidthLimit(long bytesPerSecond, long maxBurst) {
    this.bucketTokensPerSecond = bytesPerSecond;
    this.bucketMaxBurstSize = maxBurst;
  }

  /**
   * Sets the ContentDeserializer attribute of the GlacierImpl object
   *
   * @param deserializer The new ContentDeserializer value
   */
  public void setContentDeserializer(PastContentDeserializer deserializer) {
    contentDeserializer = deserializer;
  }

  /**
   * Sets the ContentHandleDeserializer attribute of the GlacierImpl object
   *
   * @param deserializer The new ContentHandleDeserializer value
   */
  public void setContentHandleDeserializer(PastContentHandleDeserializer deserializer) {
    contentHandleDeserializer = deserializer;
  }

  /**