glacierimpl.java

pastry的java实现的2.0b版

   * DESCRIBE THE METHOD
   */
  public void startup() {

    /*
     *  Neighbor requests
     */
    addContinuation(
      new GlacierContinuation() {
        long nextTimeout;

        public long getTimeout() {
          return nextTimeout;
        }

        public String toString() {
          return "Neighbor continuation";
        }

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

          NodeHandleSet leafSet = endpoint.neighborSet(999);
          NodeHandle localHandle = getLocalNodeHandle();
          NodeHandle cwExtreme = localHandle;
          NodeHandle ccwExtreme = localHandle;

          for (int i = 0; i < leafSet.size(); i++) {
            NodeHandle thisHandle = leafSet.getHandle(i);
            if (localHandle.getId().clockwise(thisHandle.getId())) {
              if (cwExtreme.getId().clockwise(thisHandle.getId())) {
                cwExtreme = thisHandle;
              }
            } else {
              if (ccwExtreme.getId().clockwise(thisHandle.getId())) {
                ccwExtreme = thisHandle;
              }
            }
          }

          IdRange leafRange = factory.buildIdRange(ccwExtreme.getId(), cwExtreme.getId());

          for (int k = 0; k < leafSet.size(); k++) {
            if (!leafSet.getHandle(k).getId().equals(getLocalNodeHandle().getId())) {
              neighborSeen(leafSet.getHandle(k).getId(), environment.getTimeSource().currentTimeMillis());
              if (logger.level <= Logger.INFO) {
                logger.log("Asking " + leafSet.getHandle(k).getId() + " about neighbors in " + leafRange);
              }
              sendMessage(
                null,
                new GlacierNeighborRequestMessage(getMyUID(), leafRange, getLocalNodeHandle(), leafSet.getHandle(k).getId(), tagNeighbor),
                leafSet.getHandle(k)
                );
            }
          }
        }

        public void receiveResult(Object o) {
          if (o instanceof GlacierNeighborResponseMessage) {
            final GlacierNeighborResponseMessage gnrm = (GlacierNeighborResponseMessage) o;
            if (logger.level <= Logger.FINE) {
              logger.log("NeighborResponse from " + gnrm.getSource() + " with " + gnrm.numNeighbors() + " neighbors");
            }
            for (int i = 0; i < gnrm.numNeighbors(); i++) {
              neighborSeen(gnrm.getNeighbor(i), gnrm.getLastSeen(i));
            }
          } else {
            if (logger.level <= Logger.WARNING) {
              logger.log("Unknown response in neighbor continuation: " + o + " -- discarded");
            }
          }
        }

        public void receiveException(Exception e) {
          if (logger.level <= Logger.WARNING) {
            logger.logException("Exception in neighbor continuation: ", e);
          }
          terminate();
        }

        public void timeoutExpired() {
          nextTimeout = environment.getTimeSource().currentTimeMillis() + expireNeighborsInterval;

          final long earliestAcceptableDate = environment.getTimeSource().currentTimeMillis() - neighborTimeout;
          IdSet allNeighbors = neighborStorage.scan();
          Iterator iter = allNeighbors.getIterator();
          NodeHandleSet leafSet = endpoint.neighborSet(999);

          if (logger.level <= Logger.INFO) {
            logger.log("Checking neighborhood for expired certificates...");
          }

          while (iter.hasNext()) {
            final Id thisNeighbor = (Id) iter.next();
            if (leafSet.memberHandle(thisNeighbor)) {
              if (logger.level <= Logger.FINE) {
                logger.log("CNE: Refreshing current neighbor: " + thisNeighbor);
              }
              neighborSeen(thisNeighbor, environment.getTimeSource().currentTimeMillis());
            } else {
              if (logger.level <= Logger.FINE) {
                logger.log("CNE: Retrieving " + thisNeighbor);
              }
              neighborStorage.getObject(thisNeighbor,
                new Continuation() {
                  public void receiveResult(Object o) {
                    if (o == null) {
                      if (logger.level <= Logger.WARNING) {
                        logger.log("CNE: Cannot retrieve neighbor " + thisNeighbor);
                      }
                      return;
                    }

                    long lastSeen = ((Long) o).longValue();
                    if (lastSeen < earliestAcceptableDate) {
                      if (logger.level <= Logger.INFO) {
                        logger.log("CNE: Removing expired neighbor " + thisNeighbor + " (" + lastSeen + "<" + earliestAcceptableDate + ")");
                      }
                      neighborStorage.unstore(thisNeighbor,
                        new Continuation() {
                          public void receiveResult(Object o) {
                            if (logger.level <= Logger.FINE) {
                              logger.log("CNE unstore successful: " + thisNeighbor + ", returned " + o);
                            }
                          }

                          public void receiveException(Exception e) {
                            if (logger.level <= Logger.WARNING) {
                              logger.log("CNE unstore failed: " + thisNeighbor + ", returned " + e);
                            }
                          }
                        });
                    } else {
                      if (logger.level <= Logger.INFO) {
                        logger.log("CNE: Neighbor " + thisNeighbor + " still active, last seen " + lastSeen);
                      }
                    }
                  }

                  public void receiveException(Exception e) {
                    if (logger.level <= Logger.WARNING) {
                      logger.log("CNE: Exception while retrieving neighbor " + thisNeighbor + ", e=" + e);
                    }
                  }
                });
            }
          }

          determineResponsibleRange();
        }
      });

    /*
     *  Sync
     */
    addContinuation(
      new GlacierContinuation() {
        long nextTimeout;
        int offset;

        public long getTimeout() {
          return nextTimeout;
        }

        public String toString() {
          return "Sync continuation";
        }

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

        public void receiveResult(Object o) {
          if (o instanceof GlacierRangeResponseMessage) {
            final GlacierRangeResponseMessage grrm = (GlacierRangeResponseMessage) o;

            Id ccwId = getFragmentLocation(grrm.getCommonRange().getCCWId(), numFragments - offset, 0);
            Id cwId = getFragmentLocation(grrm.getCommonRange().getCWId(), numFragments - offset, 0);
            final IdRange originalRange = factory.buildIdRange(ccwId, cwId);

            if (logger.level <= Logger.INFO) {
              logger.log("Range response (offset: " + offset + "): " + grrm.getCommonRange() + ", original=" + originalRange);
            }

            final IdSet keySet = fragmentStorage.scan();
            endpoint.process(
              new Executable() {
                public Object execute() {
                  BloomFilter bv = new BloomFilter((2 * keySet.numElements() + 5) * syncBloomFilterBitsPerKey, syncBloomFilterNumHashes, environment.getRandomSource());
                  Iterator iter = keySet.getIterator();

                  while (iter.hasNext()) {
                    FragmentKey fkey = (FragmentKey) iter.next();
                    Id thisPos = getFragmentLocation(fkey);
                    if (originalRange.containsId(thisPos)) {
                      FragmentMetadata metadata = (FragmentMetadata) fragmentStorage.getMetadata(fkey);
                      if (metadata != null) {
                        long currentExp = metadata.getCurrentExpiration();
                        long prevExp = metadata.getPreviousExpiration();
                        if (logger.level <= Logger.FINER) {
                          logger.log(" - Adding " + fkey + " as " + fkey.getVersionKey().getId() + ", ecur=" + currentExp + ", eprev=" + prevExp);
                        }
                        bv.add(getHashInput(fkey.getVersionKey(), currentExp));
                        bv.add(getHashInput(fkey.getVersionKey(), prevExp));
                      } else {
                        if (logger.level <= Logger.WARNING) {
                          logger.log("SYNC Cannot read metadata of object " + fkey.toStringFull() + ", storage returned null");
                        }
                      }
                    }
                  }

                  return bv;
                }
              },
              new Continuation() {
                public void receiveResult(Object o) {
                  if (o instanceof BloomFilter) {
                    BloomFilter bv = (BloomFilter) o;
                    if (logger.level <= Logger.FINE) {
                      logger.log("Got " + bv);
                    }
                    if (logger.level <= Logger.INFO) {
                      logger.log(keySet.numElements() + " keys added, sending sync request...");
                    }

                    sendMessage(
                      null,
                      new GlacierSyncMessage(getUID(), grrm.getCommonRange(), offset, bv, getLocalNodeHandle(), grrm.getSource().getId(), tagSync),
                      grrm.getSource()
                      );
                  } else {
                    if (logger.level <= Logger.WARNING) {
                      logger.log("While processing range response: Result is of unknown type: " + o + " -- discarding request");
                    }
                  }
                }

                public void receiveException(Exception e) {
                  if (logger.level <= Logger.WARNING) {
                    logger.logException("Exception while processing range response: " + e + " -- discarding request", e);
                  }
                }
              });
          } else {
            if (logger.level <= Logger.WARNING) {
              logger.log("Unknown result in sync continuation: " + o + " -- discarded");
            }
          }
        }

        public void receiveException(Exception e) {
          if (logger.level <= Logger.WARNING) {
            logger.logException("Exception in sync continuation: ", e);
          }
          terminate();
        }

        public void timeoutExpired() {
          if (numActiveRestores[0] > 0) {
            if (logger.level <= Logger.INFO) {
              logger.log("Sync postponed; " + numActiveRestores[0] + " fetches pending");
            }
            nextTimeout = environment.getTimeSource().currentTimeMillis() + jitterTerm(syncRetryInterval);
          } else {
            nextTimeout = environment.getTimeSource().currentTimeMillis() + jitterTerm(syncInterval);
            offset = 1 + environment.getRandomSource().nextInt(numFragments - 1);

            Id dest = getFragmentLocation(getLocalNodeHandle().getId(), offset, 0);
            Id ccwId = getFragmentLocation(responsibleRange.getCCWId(), offset, 0);
            Id cwId = getFragmentLocation(responsibleRange.getCWId(), offset, 0);
            IdRange requestedRange = factory.buildIdRange(ccwId, cwId);

            if (logger.level <= Logger.INFO) {
              logger.log("Sending range query for (" + requestedRange + ") to " + dest);
            }
            sendMessage(
              dest,
              new GlacierRangeQueryMessage(getMyUID(), requestedRange, getLocalNodeHandle(), dest, tagSync),
              null
              );
          }
        }
      });

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

        public long getTimeout() {
          return nextTimeout;
        }

        public String toString() {
          return "Handoff continuation";
        }

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

        public void receiveResult(Object o) {
          if (o instanceof GlacierResponseMessage) {
            final GlacierResponseMessage grm = (GlacierResponseMessage) o;
            if (logger.level <= Logger.FINE) {
              logger.log("Received handoff response from " + grm.getSource().getId() + " with " + grm.numKeys() + " keys");
            }
            for (int i = 0; i < grm.numKeys(); i++) {
              final FragmentKey thisKey = grm.getKey(i);
              if (grm.getAuthoritative(i)) {
                if (grm.getHaveIt(i)) {
                  Id thisPos = getFragmentLocation(thisKey);
                  if (!responsibleRange.containsId(thisPos)) {
                    if (logger.level <= Logger.FINE) {
                      logger.log("Deleting fragment " + thisKey);
                    }
                    deleteFragment(thisKey,
                      new Continuation() {
                        public void receiveResult(Object o) {
                          if (logger.level <= Logger.FINE) {
                            logger.log("Handed off fragment deleted: " + thisKey + " (o=" + o + ")");
                          }
                        }

                        public void receiveException(Exception e) {
                          if (logger.level <= Logger.WARNING) {
                            logger.logException("Delete failed during handoff: " + thisKey + ", returned ", e);
                          }
                        }
                      });
                  } else {
                    if (logger.level <= Logger.WARNING) {
                      logger.log("Handoff response for " + thisKey + ", for which I am still responsible (attack?) -- ignored");
                    }
                  }
                } else {
                  fragmentStorage.getObject(thisKey,
                    new Continuation() {
                      public void receiveResult(Object o) {
                        if (o != null) {
                          if (logger.level <= Logger.INFO) {
                            logger.log("Fragment " + thisKey + " found (" + o + "), handing off...");
                          }