splitstreamscribepolicy.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.splitstream;

import java.util.*;

import rice.p2p.commonapi.*;
import rice.p2p.scribe.*;
import rice.p2p.scribe.messaging.*;

import rice.pastry.routing.RoutingTable;

/**
 * This class represents SplitStream's policy for Scribe, which only allows
 * children according to the bandwidth manager and makes anycasts first traverse
 * all nodes who have the stripe in question as their primary stripe, and then
 * the nodes who do not.
 *
 * @version $Id: SplitStreamScribePolicy.java 2661 2005-07-15 16:06:14Z jeffh $
 * @author Alan Mislove
 * @author Atul Singh
 */
public class SplitStreamScribePolicy implements ScribePolicy {

  /**
   * The default maximum number of children per channel
   */
  public final int DEFAULT_MAXIMUM_CHILDREN;
  // = 24;

  /**
   * A reference to this policy's splitstream object
   */
  protected SplitStream splitStream;

  /**
   * A reference to this policy's scribe object
   */
  protected Scribe scribe;

  /**
   * A mapping from channelId -> maximum children
   */
  protected Hashtable policy;

  /**
   * Constructor which takes a splitStream object
   *
   * @param splitStream The local splitstream
   * @param scribe DESCRIBE THE PARAMETER
   */
  public SplitStreamScribePolicy(Scribe scribe, SplitStream splitStream) {
    DEFAULT_MAXIMUM_CHILDREN = scribe.getEnvironment().getParameters().getInt("p2p_splitStream_policy_default_maximum_children");
    this.scribe = scribe;
    this.splitStream = splitStream;
    this.policy = new Hashtable();
  }

  /**
   * Gets the max bandwidth for a channel.
   *
   * @param id The id to get the max bandwidth of
   * @return The amount of bandwidth used
   */
  public int getMaxChildren(ChannelId id) {
    Integer max = (Integer) policy.get(id);

    if (max == null) {
      return DEFAULT_MAXIMUM_CHILDREN;
    } else {
      return max.intValue();
    }
  }

  /**
   * Returns the Channel which contains the stripe cooresponding to the provided
   * topic.
   *
   * @param topic The topic in question
   * @return The channel which contains a cooresponding stripe
   */
  private Channel getChannel(Topic topic) {
    Channel[] channels = splitStream.getChannels();

    for (int i = 0; i < channels.length; i++) {
      Channel channel = channels[i];
      Stripe[] stripes = channel.getStripes();

      for (int j = 0; j < stripes.length; j++) {
        Stripe stripe = stripes[j];

        if (stripe.getStripeId().getId().equals(topic.getId())) {
          return channel;
        }
      }
    }

    return null;
  }

  /**
   * Returns the total number of children for the given channel
   *
   * @param channel The channel to get the children for
   * @return The total number of children for that channel
   */
  public int getTotalChildren(Channel channel) {
    int total = 0;
    Stripe[] stripes = channel.getStripes();

    for (int j = 0; j < stripes.length; j++) {
      total += scribe.getChildren(new Topic(stripes[j].getStripeId().getId())).length;
    }

    return total;
  }

  /**
   * Adjust the max bandwidth for this channel.
   *
   * @param id The id to get the max bandwidth of
   * @param children The new maximum bandwidth for this channel
   */
  public void setMaxChildren(ChannelId id, int children) {
    policy.put(id, new Integer(children));
  }

  /**
   * This method implements the "locating parent" algorithm of SplitStream.
   * Whenever a node receives subscription request, it executes following
   * algorithm : 1) Checks if it has available capacity, if yes, take it if no,
   * go to step 2 2) If subscription is for primary stripe, if no, dont take it,
   * return false if yes, need to drop someone first check if our children list
   * for this topic is non-zero if yes, find a child which shares less prefix
   * match than the new subscriber if found one, send it a drop message and
   * accept newe subscriber, return true if no, dont accept the new subscriber,
   * return false if no, look for children in other stripes which we can drop we
   * select a stripe such that it is not a) primary stripe b) we are not the
   * root for the stripe (may happen in very small networks) c) we have non-zero
   * children for this stripe if found such a stripe, we drop a random child
   * from that stripe and accept the new subscriber, return true. 3) Checks if
   * one of our child for a stripe sent us a request message for dropping it and
   * taking the new subscriber (STAGE_3), then we drop the child and return true
   * to accept the subscriber.
   *
   * @param message The subscribe message in question
   * @param children The list of children who are currently subscribed to this
   *      topic
   * @param clients The list of clients are are currently subscribed to this
   *      topic
   * @return Whether or not this child should be allowed add.
   */
  public boolean allowSubscribe(SubscribeMessage message, ScribeClient[] clients, NodeHandle[] children) {

    Channel channel = getChannel(message.getTopic());
    NodeHandle newChild = (NodeHandle) message.getSubscriber();

    if (channel == null) {
      // If the channel is null, return false *unless you are root*, in which case you
      // return true.  This is a weird case.  We have to accept here if we are the root, even
      // if we are not attached to the channel (it's a bootstrapping issue)
      return scribe.isRoot(message.getTopic());
    }

    /*
     *  do not accept self - wierd case, should not happen
     */
    if (message.getSubscriber().getId().equals(channel.getLocalId())) {
      return false;
    }

    /*
     *  first see if we are in the 3rd stage of algorithm for locating parent.
     */
    ScribeContent content = message.getContent();

    /*
     *  this occurs if we are in the third stage of locating a parent
     */
    if (content != null && (content instanceof SplitStreamSubscribeContent)) {
      int stage = ((SplitStreamSubscribeContent) content).getStage();
      if (stage == SplitStreamSubscribeContent.STAGE_FINAL) {
        List list = Arrays.asList(children);

        if (!list.contains(message.getSource())) {
          return false;
        } else {
          this.scribe.removeChild(message.getTopic(), message.getSource());
          return true;
        }
      }
    }

    /*
     *  see if we can accept
     */
    if (getTotalChildren(channel) < getMaxChildren(channel.getChannelId())) {
      return true;
    } else {
      /*
       *  check if non-primary stripe
       */
      if ((!message.getTopic().getId().equals(channel.getPrimaryStripe().getStripeId().getId())) &&
        (!scribe.isRoot(message.getTopic()))) {
        return false;
      } else {
        /*
         *  find a victim child
         */
        if (children.length > 0) {
          NodeHandle victimChild = freeBandwidth(channel, newChild, message.getTopic().getId());

          /*
           *  make sure victim is not subscriber
           */
          if (victimChild.getId().equals(newChild.getId())) {
            return false;
          } else {
            scribe.removeChild(new Topic(message.getTopic().getId()), victimChild);
            return true;
          }
        } else {
          /*
           *  we must accept, because this is primary stripe
           */
          Vector res = freeBandwidthUltimate(message.getTopic().getId());
          if (res != null) {