glacierimpl.java

pastry的java实现的2.0b版

              final VersionKey thisVKey = (VersionKey) iter.next();

              int localFragmentID = 0;
              int queriesHere = 0;
              for (int i = 0; i < numFragments; i++) {
                FragmentKey keyHere = new FragmentKey(thisVKey, i);
                if (fragments.isMemberId(keyHere)) {
                  localFragmentID = i;
                  break;
                } else if (responsibleRange.containsId(getFragmentLocation(keyHere))) {
                  queriesHere++;
                }
              }

              if (logger.level <= Logger.FINE) {
                logger.log("Local scan: Fetching manifest for " + thisVKey + " (" + queriesHere + " pending queries)");
              }
              queriesSent += queriesHere;

              fragmentStorage.getObject(new FragmentKey(thisVKey, localFragmentID),
                new Continuation() {
                  public void receiveResult(Object o) {
                    if (o instanceof FragmentAndManifest) {
                      final Manifest thisManifest = ((FragmentAndManifest) o).manifest;

                      for (int i = 0; i < numFragments; i++) {
                        final FragmentKey thisKey = new FragmentKey(thisVKey, i);
                        if (responsibleRange.containsId(getFragmentLocation(thisKey))) {
                          if (!fragments.isMemberId(thisKey)) {
                            if (logger.level <= Logger.FINE) {
                              logger.log("Local scan: Sending query for " + thisKey);
                            }
                            final long tStart = environment.getTimeSource().currentTimeMillis();
                            rateLimitedRetrieveFragment(thisKey, thisManifest, tagLocalScan,
                              new GlacierContinuation() {
                                public long getTimeout() {
                                  return tStart + overallRestoreTimeout;
                                }

                                public String toString() {
                                  return "Local scan: Fetch fragment: " + thisKey;
                                }

                                public void receiveResult(Object o) {
                                  if (o instanceof Fragment) {
                                    if (logger.level <= Logger.INFO) {
                                      logger.log("Local scan: Received fragment " + thisKey + " (from primary) matches existing manifest, storing...");
                                    }

                                    FragmentAndManifest fam = new FragmentAndManifest((Fragment) o, thisManifest);

                                    fragmentStorage.store(thisKey, new FragmentMetadata(thisManifest.getExpiration(), 0, environment.getTimeSource().currentTimeMillis()), fam,
                                      new Continuation() {
                                        public void receiveResult(Object o) {
                                          if (logger.level <= Logger.FINE) {
                                            logger.log("Local scan: Recovered fragment stored OK");
                                          }
                                        }

                                        public void receiveException(Exception e) {
                                          if (logger.level <= Logger.WARNING) {
                                            logger.log("Local scan: receiveException(" + e + ") while storing a fragment with existing manifest (key=" + thisKey + ")");
                                          }
                                        }
                                      }
                                      );
                                  } else {
                                    if (logger.level <= Logger.WARNING) {
                                      logger.log("Local scan: FS received something other than a fragment: " + o);
                                    }
                                  }
                                }

                                public void receiveException(Exception e) {
                                  if (logger.level <= Logger.WARNING) {
                                    logger.logException("Local scan: Exception while recovering synced fragment " + thisKey + ": ", e);
                                  }
                                  terminate();
                                }

                                public void timeoutExpired() {
                                  if (logger.level <= Logger.WARNING) {
                                    logger.log("Local scan: Timeout while fetching synced fragment " + thisKey + " -- aborted");
                                  }
                                  terminate();
                                }
                              });
                          }
                        }
                      }
                    } else {
                      if (logger.level <= Logger.WARNING) {
                        logger.log("Local scan: Cannot retrieve " + thisVKey + " from local store, received o=" + o);
                      }
                    }
                  }

                  public void receiveException(Exception e) {
                    if (logger.level <= Logger.WARNING) {
                      logger.logException("Local scan: Cannot retrieve " + thisVKey + " from local store, exception e=", e);
                    }
                  }
                });
            }

            if (logger.level <= Logger.INFO) {
              logger.log(queriesSent + " queries sent after local scan");
            }
          } else {
            if (logger.level <= Logger.INFO) {
              logger.log("Local scan completed; no missing fragments");
            }
          }
        }
      });

    /*
     *  Traffic shaper
     */
    addContinuation(
      new GlacierContinuation() {
        long nextTimeout;

        public long getTimeout() {
          return nextTimeout;
        }

        public String toString() {
          return "Traffic shaper";
        }

        public void init() {
          nextTimeout = environment.getTimeSource().currentTimeMillis() + rateLimitedCheckInterval;
        }

        public void receiveResult(Object o) {
          if (logger.level <= Logger.WARNING) {
            logger.log("TS received object: " + o);
          }
        }

        public void receiveException(Exception e) {
          if (logger.level <= Logger.WARNING) {
            logger.logException("TS received exception: ", e);
          }
        }

        public void timeoutExpired() {
          /*
           *  Use relative timeout to avoid backlog!
           */
          nextTimeout = environment.getTimeSource().currentTimeMillis() + (1 * SECONDS);

          if (pendingTraffic.isEmpty()) {
            if (logger.level <= Logger.FINE) {
              logger.log("Traffic shaper: Idle");
            }
            nextTimeout += rateLimitedCheckInterval;
            return;
          }

          int numCurrentRestores = 0;
          synchronized (numActiveRestores) {
            numCurrentRestores = numActiveRestores[0];
          }

          if (logger.level <= Logger.INFO) {
            logger.log("Traffic shaper: " + pendingTraffic.size() + " jobs waiting (" + numCurrentRestores + " active jobs, " + tokenBucket + " tokens)");
          }

          updateTokenBucket();
          if ((numCurrentRestores < maxActiveRestores) && (tokenBucket > 0)) {
            for (int i = 0; i < rateLimitedRequestsPerSecond; i++) {
              if (!pendingTraffic.isEmpty()) {
                Enumeration keys = pendingTraffic.keys();
                Object thisKey = (Object) keys.nextElement();
                if (logger.level <= Logger.FINE) {
                  logger.log("Sending request " + thisKey);
                }
                Continuation c = (Continuation) pendingTraffic.remove(thisKey);
                c.receiveResult(new Boolean(true));
              }
            }
          }
        }
      });

    /*
     *  Statistics
     */
    addContinuation(
      new GlacierContinuation() {
        long nextTimeout;

        public long getTimeout() {
          return nextTimeout;
        }

        public String toString() {
          return "Statistics";
        }

        public void init() {
          nextTimeout = environment.getTimeSource().currentTimeMillis() + statisticsReportInterval;
        }

        public void receiveResult(Object o) {
          if (logger.level <= Logger.WARNING) {
            logger.log("STAT received object: " + o);
          }
        }

        public void receiveException(Exception e) {
          if (logger.level <= Logger.WARNING) {
            logger.logException("STAT received exception: ", e);
          }
        }

        public void timeoutExpired() {
          nextTimeout += statisticsReportInterval;

          if (!listeners.isEmpty()) {
            statistics.pendingRequests = pendingTraffic.size();
            statistics.numNeighbors = neighborStorage.scan().numElements();
            statistics.numFragments = fragmentStorage.scan().numElements();
            statistics.numContinuations = continuations.size();
//          statistics.numObjectsInTrash = (trashStorage == null) ? 0 : trashStorage.scan().numElements();
            statistics.responsibleRange = responsibleRange;
            statistics.activeFetches = numActiveRestores[0];
            statistics.bucketMin = bucketMin;
            statistics.bucketMax = bucketMax;
            statistics.bucketConsumed = bucketConsumed;
            statistics.bucketTokensPerSecond = bucketTokensPerSecond;
            statistics.bucketMaxBurstSize = bucketMaxBurstSize;
            bucketMin = tokenBucket;
            bucketMax = tokenBucket;
            bucketConsumed = 0;

            Storage storageF = fragmentStorage.getStorage();
            if (storageF instanceof PersistentStorage) {
              statistics.fragmentStorageSize = ((PersistentStorage) storageF).getTotalSize();
            }

            Storage storageT = (trashStorage == null) ? null : trashStorage.getStorage();
            if (storageT instanceof PersistentStorage) {
              statistics.trashStorageSize = ((PersistentStorage) storageT).getTotalSize();
            }

            if (logStatistics) {
              statistics.dump(environment.getLogManager().getLogger(GlacierStatistics.class, instance));
            }

            Enumeration enumeration = listeners.elements();
            while (enumeration.hasMoreElements()) {
              GlacierStatisticsListener gsl = (GlacierStatisticsListener) enumeration.nextElement();
              gsl.receiveStatistics(statistics);
            }
          }

          statistics = new GlacierStatistics(tagMax, environment);
        }
      });
  }

  /**
   * DESCRIBE THE METHOD
   */
  protected void updateTokenBucket() {
    final long now = environment.getTimeSource().currentTimeMillis();
    final long contentsBefore = tokenBucket;

    while (bucketLastUpdated < now) {
      bucketLastUpdated += SECONDS / 10;
      tokenBucket += bucketTokensPerSecond / 10;
      if (tokenBucket > bucketMaxBurstSize) {
        tokenBucket = bucketMaxBurstSize;
      }
    }

    if (bucketMax < tokenBucket) {
      bucketMax = tokenBucket;
    }

    if (logger.level <= Logger.FINE) {
      logger.log("Token bucket contains " + tokenBucket + " tokens (added " + (tokenBucket - contentsBefore) + ")");
    }
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param basis DESCRIBE THE PARAMETER
   * @return DESCRIBE THE RETURN VALUE
   */
  private long jitterTerm(long basis) {
    return (long) ((1 - jitterRange) * basis) + environment.getRandomSource().nextInt((int) (2 * jitterRange * basis));
  }

  /**
   * DESCRIBE THE METHOD
   *
   * @param fkey DESCRIBE THE PARAMETER
   * @param command DESCRIBE THE PARAMETER
   */
  private void deleteFragment(final Id fkey, final Continuation command) {
    if (trashStorage != null) {
      if (logger.level <= Logger.INFO) {
        logger.log("Moving fragment " + fkey.toStringFull() + " to trash");
      }
      fragmentStorage.getObject(fkey,
        new Continuation() {
          public void receiveResult(Object o) {
            if (logger.level <= Logger.FINE) {
              logger.log("Fragment " + fkey.toStringFull() + " retrieved, storing in trash");
            }
            if (o != null) {
              trashStorage.store(fkey, null, (Serializable) o,
                new Continuation() {
                  public void receiveResult(Object o) {
                    if (logger.level <= Logger.FINE) {
                      logger.log("Deleting fragment " + fkey.toStringFull());
                    }
                    fragmentStorage.unstore(fkey, command);
                  }

                  public void receiveException(Exception e) {
                    if (logger.level <= Logger.WARNING) {
                      logger.logException("Cannot store in trash: " + fkey.toStringFull() + ", e=", e);
                    }
                    command.receiveException(e);
                  }
                });
            } else {
              receiveException(new GlacierException("Move to trash: Fragment " + fkey + " does not exist?!?"));
            }
          }

          public void receiveException(Exception e) {
            if (logger.level <= Logger.WARNING) {
              logger.logException("Cannot retrieve fragment " + fkey + " for deletion: e=", e);
            }
            command.receiveException(new GlacierException("Cannot retrieve fragment " + fkey + " for deletion"));
          }
        });