glacierimpl.java

pastry的java实现的2.0b版

/*************************************************************************

"FreePastry" Peer-to-Peer Application Development Substrate 

Copyright 2002, Rice University. All rights reserved.

Redistribution and use in source and binary forms, with or without
modification, are permitted provided that the following conditions are
met:

- Redistributions of source code must retain the above copyright
notice, this list of conditions and the following disclaimer.

- Redistributions in binary form must reproduce the above copyright
notice, this list of conditions and the following disclaimer in the
documentation and/or other materials provided with the distribution.

- Neither  the name  of Rice  University (RICE) nor  the names  of its
contributors may be  used to endorse or promote  products derived from
this software without specific prior written permission.

This software is provided by RICE and the contributors on an "as is"
basis, without any representations or warranties of any kind, express
or implied including, but not limited to, representations or
warranties of non-infringement, merchantability or fitness for a
particular purpose. In no event shall RICE or contributors be liable
for any direct, indirect, incidental, special, exemplary, or
consequential damages (including, but not limited to, procurement of
substitute goods or services; loss of use, data, or profits; or
business interruption) however caused and on any theory of liability,
whether in contract, strict liability, or tort (including negligence
or otherwise) arising in any way out of the use of this software, even
if advised of the possibility of such damage.

********************************************************************************/
package rice.p2p.glacier.v2;

import java.io.*;
import java.security.*;
import java.util.*;

import rice.Continuation;
import rice.Executable;
import rice.environment.Environment;
import rice.environment.logging.Logger;
import rice.environment.params.Parameters;
import rice.p2p.commonapi.*;
import rice.p2p.commonapi.rawserialization.*;
import rice.p2p.glacier.*;
import rice.p2p.glacier.v2.messaging.*;
import rice.p2p.past.Past;
import rice.p2p.past.PastContent;
import rice.p2p.past.PastContentHandle;
import rice.p2p.past.gc.GCPast;
import rice.p2p.past.gc.GCPastContent;
import rice.p2p.past.rawserialization.*;
import rice.p2p.util.DebugCommandHandler;
import rice.persistence.Storage;
import rice.persistence.StorageManager;
import rice.persistence.PersistentStorage;

/**
 * DESCRIBE THE CLASS
 *
 * @version $Id: pretty.settings 2305 2005-03-11 20:22:33Z jeffh $
 * @author jeffh
 */
public class GlacierImpl implements Glacier, GCPast, VersioningPast, Application, DebugCommandHandler {
  /**
   * DESCRIBE THE FIELD
   */
  protected final StorageManager fragmentStorage;
  /**
   * DESCRIBE THE FIELD
   */
  protected final StorageManager neighborStorage;
  /**
   * DESCRIBE THE FIELD
   */
  protected final GlacierPolicy policy;
  /**
   * DESCRIBE THE FIELD
   */
  protected final Node node;
  /**
   * DESCRIBE THE FIELD
   */
  protected final int numFragments;
  /**
   * DESCRIBE THE FIELD
   */
  protected final String instance;
  /**
   * DESCRIBE THE FIELD
   */
  protected final int numSurvivors;
  /**
   * DESCRIBE THE FIELD
   */
  protected final Endpoint endpoint;
  /**
   * DESCRIBE THE FIELD
   */
  protected final IdFactory factory;
  /**
   * DESCRIBE THE FIELD
   */
  protected final Hashtable continuations;
  /**
   * DESCRIBE THE FIELD
   */
  protected final Hashtable pendingTraffic;
  /**
   * DESCRIBE THE FIELD
   */
  protected StorageManager trashStorage;
  /**
   * DESCRIBE THE FIELD
   */
  protected long nextContinuationTimeout;
  /**
   * DESCRIBE THE FIELD
   */
  protected IdRange responsibleRange;
  /**
   * DESCRIBE THE FIELD
   */
  protected int nextUID;
  /**
   * DESCRIBE THE FIELD
   */
  protected CancellableTask timer;
  /**
   * DESCRIBE THE FIELD
   */
  protected GlacierStatistics statistics;
  /**
   * DESCRIBE THE FIELD
   */
  protected Vector listeners;
  /**
   * DESCRIBE THE FIELD
   */
  protected long currentFragmentRequestTimeout;
  /**
   * DESCRIBE THE FIELD
   */
  protected long tokenBucket;
  /**
   * DESCRIBE THE FIELD
   */
  protected long bucketLastUpdated;
  /**
   * DESCRIBE THE FIELD
   */
  protected long bucketMin;
  /**
   * DESCRIBE THE FIELD
   */
  protected long bucketMax;
  /**
   * DESCRIBE THE FIELD
   */
  protected long bucketConsumed;

  private final long SECONDS = 1000;
  private final long MINUTES = 60 * SECONDS;
  private final long HOURS = 60 * MINUTES;
  private final long DAYS = 24 * HOURS;
  private final long WEEKS = 7 * DAYS;

  private final boolean logStatistics;
  private final boolean faultInjectionEnabled;

  private final long insertTimeout;
  private final double minFragmentsAfterInsert;

  private final long refreshTimeout;

  private final long expireNeighborsDelayAfterJoin;
  private final long expireNeighborsInterval;
  private long neighborTimeout;

  private final long syncDelayAfterJoin;
  private final long syncMinRemainingLifetime;
  private final long syncMinQuietTime;
  private final int syncBloomFilterNumHashes;
  private final int syncBloomFilterBitsPerKey;
  private final int syncPartnersPerTrial;
  private long syncInterval;
  private final long syncRetryInterval;
  private int syncMaxFragments;

//  private final int fragmentRequestMaxAttempts = 3;
  private final int fragmentRequestMaxAttempts;
  private final long fragmentRequestTimeoutDefault;
  private final long fragmentRequestTimeoutMin;
  private final long fragmentRequestTimeoutMax;
  private final long fragmentRequestTimeoutDecrement;

  private final long manifestRequestTimeout;
  private final long manifestRequestInitialBurst;
  private final long manifestRequestRetryBurst;
  private final int manifestAggregationFactor;

  private final long overallRestoreTimeout;

  private final long handoffDelayAfterJoin;
  private final long handoffInterval;
  private final int handoffMaxFragments;

  private final long garbageCollectionInterval;
  private final int garbageCollectionMaxFragmentsPerRun;

  private final long localScanInterval;
  private final int localScanMaxFragmentsPerRun;

  private final double restoreMaxRequestFactor;
  private final int restoreMaxBoosts;

  private final long rateLimitedCheckInterval;
  private int rateLimitedRequestsPerSecond;

  private final boolean enableBulkRefresh;
  private final long bulkRefreshProbeInterval;
  private final double bulkRefreshMaxProbeFactor;
  private final long bulkRefreshManifestInterval;
  private final int bulkRefreshManifestAggregationFactor;
  private final int bulkRefreshPatchAggregationFactor;
  private final long bulkRefreshPatchInterval;
  private final int bulkRefreshPatchRetries;

  private long bucketTokensPerSecond;
  private long bucketMaxBurstSize;

  private final double jitterRange;

  private final long statisticsReportInterval;

  private final int maxActiveRestores;
  private int[] numActiveRestores;

  private final char tagNeighbor = 1;
  private final char tagSync = 2;
  private final char tagSyncManifests = 3;
  private final char tagSyncFetch = 4;
  private final char tagHandoff = 5;
  private final char tagDebug = 6;
  private final char tagRefresh = 7;
  private final char tagInsert = 8;
  private final char tagLookupHandles = 9;
  private final char tagLookup = 10;
  private final char tagFetch = 11;
  private final char tagLocalScan = 12;
  private final char tagMax = 13;

  private Environment environment;
  /**
   * DESCRIBE THE FIELD
   */
  protected Logger logger;

  /**
   * DESCRIBE THE FIELD
   */
  protected PastContentDeserializer contentDeserializer;
  /**
   * DESCRIBE THE FIELD
   */
  protected PastContentHandleDeserializer contentHandleDeserializer;

  /**
   * Constructor for GlacierImpl.
   *
   * @param nodeArg DESCRIBE THE PARAMETER
   * @param fragmentStorageArg DESCRIBE THE PARAMETER
   * @param neighborStorageArg DESCRIBE THE PARAMETER
   * @param numFragmentsArg DESCRIBE THE PARAMETER
   * @param numSurvivorsArg DESCRIBE THE PARAMETER
   * @param factoryArg DESCRIBE THE PARAMETER
   * @param instanceArg DESCRIBE THE PARAMETER
   * @param policyArg DESCRIBE THE PARAMETER
   */
  public GlacierImpl(Node nodeArg, StorageManager fragmentStorageArg, StorageManager neighborStorageArg, int numFragmentsArg, int numSurvivorsArg, IdFactory factoryArg, String instanceArg, GlacierPolicy policyArg) {
    this.environment = nodeArg.getEnvironment();
    this.logger = environment.getLogManager().getLogger(GlacierImpl.class, instanceArg);

    Parameters p = environment.getParameters();

    logStatistics = p.getBoolean("p2p_glacier_logStatistics");
    faultInjectionEnabled = p.getBoolean("p2p_glacier_faultInjectionEnabled");

    insertTimeout = p.getLong("p2p_glacier_insertTimeout");
    minFragmentsAfterInsert = p.getDouble("p2p_glacier_minFragmentsAfterInsert");

    refreshTimeout = p.getLong("p2p_glacier_refreshTimeout");

    expireNeighborsDelayAfterJoin = p.getLong("p2p_glacier_expireNeighborsDelayAfterJoin");
    expireNeighborsInterval = p.getLong("p2p_glacier_expireNeighborsInterval");
    neighborTimeout = p.getLong("p2p_glacier_neighborTimeout");

    syncDelayAfterJoin = p.getLong("p2p_glacier_syncDelayAfterJoin");
    syncMinRemainingLifetime = p.getLong("p2p_glacier_syncMinRemainingLifetime");
    syncMinQuietTime = p.getLong("p2p_glacier_syncMinQuietTime");
    syncBloomFilterNumHashes = p.getInt("p2p_glacier_syncBloomFilterNumHashes");
    syncBloomFilterBitsPerKey = p.getInt("p2p_glacier_syncBloomFilterBitsPerKey");
    syncPartnersPerTrial = p.getInt("p2p_glacier_syncPartnersPerTrial");
    syncInterval = p.getLong("p2p_glacier_syncInterval");
    syncRetryInterval = p.getLong("p2p_glacier_syncRetryInterval");
    syncMaxFragments = p.getInt("p2p_glacier_syncMaxFragments");

    fragmentRequestMaxAttempts = p.getInt("p2p_glacier_fragmentRequestMaxAttempts");
    fragmentRequestTimeoutDefault = p.getLong("p2p_glacier_fragmentRequestTimeoutDefault");
    fragmentRequestTimeoutMin = p.getLong("p2p_glacier_fragmentRequestTimeoutMin");
    fragmentRequestTimeoutMax = p.getLong("p2p_glacier_fragmentRequestTimeoutMax");
    fragmentRequestTimeoutDecrement = p.getLong("p2p_glacier_fragmentRequestTimeoutDecrement");

    manifestRequestTimeout = p.getLong("p2p_glacier_manifestRequestTimeout");
    manifestRequestInitialBurst = p.getLong("p2p_glacier_manifestRequestInitialBurst");
    manifestRequestRetryBurst = p.getLong("p2p_glacier_manifestRequestRetryBurst");
    manifestAggregationFactor = p.getInt("p2p_glacier_manifestAggregationFactor");

    overallRestoreTimeout = p.getLong("p2p_glacier_overallRestoreTimeout");

    handoffDelayAfterJoin = p.getLong("p2p_glacier_handoffDelayAfterJoin");
    handoffInterval = p.getLong("p2p_glacier_handoffInterval");
    handoffMaxFragments = p.getInt("p2p_glacier_handoffMaxFragments");

    garbageCollectionInterval = p.getLong("p2p_glacier_garbageCollectionInterval");
    garbageCollectionMaxFragmentsPerRun = p.getInt("p2p_glacier_garbageCollectionMaxFragmentsPerRun");

    localScanInterval = p.getLong("p2p_glacier_localScanInterval");
    localScanMaxFragmentsPerRun = p.getInt("p2p_glacier_localScanMaxFragmentsPerRun");

    restoreMaxRequestFactor = p.getDouble("p2p_glacier_restoreMaxRequestFactor");
    restoreMaxBoosts = p.getInt("p2p_glacier_restoreMaxBoosts");

    rateLimitedCheckInterval = p.getLong("p2p_glacier_rateLimitedCheckInterval");
    rateLimitedRequestsPerSecond = p.getInt("p2p_glacier_rateLimitedRequestsPerSecond");

    enableBulkRefresh = p.getBoolean("p2p_glacier_enableBulkRefresh");
    bulkRefreshProbeInterval = p.getLong("p2p_glacier_bulkRefreshProbeInterval");
    bulkRefreshMaxProbeFactor = p.getDouble("p2p_glacier_bulkRefreshMaxProbeFactor");
    bulkRefreshManifestInterval = p.getLong("p2p_glacier_bulkRefreshManifestInterval");
    bulkRefreshManifestAggregationFactor = p.getInt("p2p_glacier_bulkRefreshManifestAggregationFactor");
    bulkRefreshPatchAggregationFactor = p.getInt("p2p_glacier_bulkRefreshPatchAggregationFactor");