peerview.java

JXTA&#8482 is a set of open, generalized peer-to-peer (P2P) protocols that allow any networked devi

/*
 * Copyright (c) 2002-2007 Sun Micro//Systems, Inc.  All rights reserved.
 *  
 *  The Sun Project JXTA(TM) Software License
 *  
 *  Redistribution and use in source and binary forms, with or without 
 *  modification, are permitted provided that the following conditions are met:
 *  
 *  1. Redistributions of source code must retain the above copyright notice,
 *     this list of conditions and the following disclaimer.
 *  
 *  2. 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.
 *  
 *  3. The end-user documentation included with the redistribution, if any, must 
 *     include the following acknowledgment: "This product includes software 
 *     developed by Sun Microsystems, Inc. for JXTA(TM) technology." 
 *     Alternately, this acknowledgment may appear in the software itself, if 
 *     and wherever such third-party acknowledgments normally appear.
 *  
 *  4. The names "Sun", "Sun Microsystems, Inc.", "JXTA" and "Project JXTA" must 
 *     not be used to endorse or promote products derived from this software 
 *     without prior written permission. For written permission, please contact 
 *     Project JXTA at http://www.jxta.org.
 *  
 *  5. Products derived from this software may not be called "JXTA", nor may 
 *     "JXTA" appear in their name, without prior written permission of Sun.
 *  
 *  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESSED OR IMPLIED WARRANTIES,
 *  INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND 
 *  FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL SUN 
 *  MICROSYSTEMS OR ITS 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.
 *  
 *  JXTA is a registered trademark of Sun Microsystems, Inc. in the United 
 *  States and other countries.
 *  
 *  Please see the license information page at :
 *  <http://www.jxta.org/project/www/license.html> for instructions on use of 
 *  the license in source files.
 *  
 *  ====================================================================
 *  
 *  This software consists of voluntary contributions made by many individuals 
 *  on behalf of Project JXTA. For more information on Project JXTA, please see 
 *  http://www.jxta.org.
 *  
 *  This license is based on the BSD license adopted by the Apache Foundation. 
 */
package net.jxta.impl.rendezvous.rpv;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.NoSuchElementException;
import java.util.Random;
import java.util.Set;
import java.util.SortedSet;
import java.util.Timer;
import java.util.TimerTask;
import java.util.TreeSet;
import java.util.Vector;
import java.util.logging.Level;
import java.util.logging.Logger;

import net.jxta.discovery.DiscoveryService;
import net.jxta.document.Advertisement;
import net.jxta.document.AdvertisementFactory;
import net.jxta.document.MimeMediaType;
import net.jxta.document.StructuredDocumentFactory;
import net.jxta.document.XMLDocument;
import net.jxta.endpoint.EndpointAddress;
import net.jxta.endpoint.EndpointListener;
import net.jxta.endpoint.EndpointService;
import net.jxta.endpoint.Message;
import net.jxta.endpoint.MessageElement;
import net.jxta.endpoint.Messenger;
import net.jxta.endpoint.StringMessageElement;
import net.jxta.endpoint.TextDocumentMessageElement;
import net.jxta.id.ID;
import net.jxta.id.IDFactory;
import net.jxta.logging.Logging;
import net.jxta.peer.PeerID;
import net.jxta.peergroup.PeerGroup;
import net.jxta.pipe.InputPipe;
import net.jxta.pipe.OutputPipe;
import net.jxta.pipe.PipeID;
import net.jxta.pipe.PipeMsgEvent;
import net.jxta.pipe.PipeMsgListener;
import net.jxta.pipe.PipeService;
import net.jxta.protocol.ConfigParams;
import net.jxta.protocol.PeerAdvertisement;
import net.jxta.protocol.PipeAdvertisement;
import net.jxta.protocol.RdvAdvertisement;
import net.jxta.protocol.RouteAdvertisement;
import net.jxta.rendezvous.RendezvousEvent;
import net.jxta.rendezvous.RendezvousListener;

import net.jxta.impl.endpoint.EndpointUtils;
import net.jxta.impl.endpoint.relay.RelayReferralSeedingManager;
import net.jxta.impl.protocol.RdvConfigAdv;
import net.jxta.impl.rendezvous.RendezVousServiceImpl;
import net.jxta.impl.util.SeedingManager;
import net.jxta.impl.util.TimeUtils;
import net.jxta.impl.util.URISeedingManager;

/**
 * This class models a Rendezvous Peer View (RPV):
 * ordered collection of all other Rendezvous Peers visible to
 * this peer.
 * <p/>
 * Presently this class implements a random "diffusion" algorithm
 * where each Peer periodically selects a randomly selected peer advertisement
 * from its view and sends it over to a randomly selected peer from its view.
 * Over time, this causes every peer to learn about every other peer, resulting
 * in a "converged" peer view.
 * <p/>
 * This diffusion process is bootstrapped by every peer sending their
 * own peer advertisements to some well-known, stable, "seed" peers on
 * startup.
 */
public final class PeerView implements EndpointListener, RendezvousListener {

    /**
     * Logger
     */
    private static final transient Logger LOG = Logger.getLogger(PeerView.class.getName());

    /**
     * Our service name
     */
    static final String SERVICE_NAME = "PeerView";

    /**
     * Namespace used for rdv message elements.
     */
    static final String MESSAGE_NAMESPACE = "jxta";

    /**
     * Element name of outgoing messages. Note that the element contains a
     * RdvAvertisement and <emphasis>not</emphasis> a Peer Advertisement.
     */
    static final String MESSAGE_ELEMENT_NAME = "PeerView.PeerAdv";

    /**
     * Element name of responses. Note that the element contains a
     * RdvAvertisement and <emphasis>not</emphasis> a Peer Advertisement.
     */
    static final String RESPONSE_ELEMENT_NAME = "PeerView.PeerAdv.Response";

    /**
     * Message element name for PeerView "Cached" Message Element
     */
    static final String CACHED_RADV_ELEMENT_NAME = "PeerView.Cached";

    /**
     * Optional message element that specifies by its presence in a peerview
     * message that the referenced peer is not the provider of the
     * RdvAdvertisement and the advertisement is a "hint" or referral from the
     * responding peer.
     * <p/>
     * In practice, when sending its own RdvAdvertisement, a peer does not
     * include this element, but when sending another peer's RdvAdvertisement,
     * this element is included.
     */
    static final MessageElement CACHED_RADV_ELEMENT = new StringMessageElement(CACHED_RADV_ELEMENT_NAME, Boolean.TRUE.toString(), null);

    /**
     * Message element name that specifies the route advertisement of the
     * source of the message.
     */
    static final String SRCROUTEADV_ELEMENT_NAME = "PeerView.SrcRouteAdv";

    /**
     * Message element name for PeerView "Edge" Message Element
     */
    static final String EDGE_ELEMENT_NAME = "PeerView.EdgePeer";

    /**
     * Optional message element that specifies by its presence in a peerview
     * message that the referenced peer is an edge peer and not a member of the
     * peerview.
     */
    static final MessageElement EDGE_ELEMENT = new StringMessageElement(EDGE_ELEMENT_NAME, Boolean.TRUE.toString(), null);

    /**
     * Message element name for PeerView "Failure" Message Element
     */
    static final String FAILURE_ELEMENT_NAME = "PeerView.Failure";

    /**
     * Optional message element that specifies by its presence in a peerview
     * message that the referenced peer has either failed or is quitting. If the
     * "cached" element is also set then the error is being reported by a third
     * party.
     */
    static final MessageElement FAILURE_ELEMENT = new StringMessageElement(FAILURE_ELEMENT_NAME, Boolean.TRUE.toString(), null);

    /**
     * This is the interval between adv exchange in seconds. This is
     * the main tunable runtime parameter for the diffusion
     * process. An interval that is too low will improve view
     * consistency at the expense of gratuitous network traffic. On
     * the other hand, an interval that is too high will cause the
     * view to become inconsistent. It is desirable to err on the side
     * of extra traffic.
     */
    private static final long DEFAULT_SEEDING_PERIOD = 5 * TimeUtils.ASECOND;

    private static final long WATCHDOG_PERIOD = 30 * TimeUtils.ASECOND;
    private static final long WATCHDOG_GRACE_DELAY = 5 * TimeUtils.AMINUTE;

    private static final long DEFAULT_BOOTSTRAP_KICK_INTERVAL = 3 * TimeUtils.ASECOND;

    private static final int MIN_BOOTLEVEL = 0;
    private static final int BOOTLEVEL_INCREMENT = 1;
    private static final int MAX_BOOTLEVEL = 6;

    /**
     * DEFAULT_SEEDING_RDVPEERS
     * <p/>
     * This value is the maximum number of rendezvous peers that will be
     * send our own advertisement at boot time.
     */
    //private static final int DEFAULT_SEEDING_RDVPEERS = 5;

    private final PeerGroup group;
    
    /**
     *  The group in which our propagate pipe will run.
     */
    private final PeerGroup advertisingGroup;
    private final RendezVousServiceImpl rdvService;
    private final EndpointService endpoint;

    /**
     * The name of this PeerView.
     * <p/>
     * FIXME 20040623 bondolo This should be a CodatID.
     */
    private final String name;

    /**
     * Delay in relative milliseconds to apply before contacting seeding rdvs.
     * 0 is supposed to be reserved by RdvConfig to mean "use the default".
     * However, it is in fact a valid value and also the one we want for the default.
     * The only problem with that is that it is not possible to configure this value
     * explicitly, should it one day not be the default. The issue is actually in RdvConfig.
     */
    private long seedingRdvConnDelay = 0;

    private final boolean useOnlySeeds;

    private final SeedingManager seedingManager;

    /**
     * If the peerview is smaller than this we will try harder to find
     * additional peerview members.
     */
    private int minHappyPeerView = 4;

    /**
     * A single timer is used to periodically kick each PeerView
     * into activity. For the Random PeerView, this activity consists
     * of selecting a PeerViewElement at random from its view and
     * sending it across to a randomly-selected peer from its view.
     * <p/>
     * FIXME 20021121 lomax
     * <p/>
     * The original idea of using a common timer in order to save threads IS a
     * very good idea. However, limitations, and actually, bugs, in java.util.Timer
     * creates the following problems when using a static Timer:
     * <p/>
     * <ul>
     * <li>Memory Leak: Canceling a TimerTask does not remove it from the
     * execution queue of the Timer until the Timer is canceled or the
     * TimerTask is fired. Since most of the TimerTasks are inner classes
     * this can mean that the PeerView is held around for a long time.</li>
     * <p/>
     * <li>java.util.Timer is not only not real-time (which is more or less fine
     * for the PeerView, but it sequentially invokes tasks (only one Thread
     * per Timer). As a result, tasks that takes a long time to run delays
     * other tasks.</li>
     * </ul>
     * <p/>
     * The PeerView would function better with a better Timer, but JDK does
     * not provide a standard timer that would fulfill the needs of the
     * PeerView. Maybe we should implement a JXTA Timer, since lots of the JXTA
     * services, by being very asynchronous, rely on the same kind of timer
     * semantics as the PeerView. Until then, creating a Timer per instance of
     * the PeerView (i.e. per instance of joined PeerGroup) is the best
     * workaround.
     */
    private final Timer timer;

    /**
     * A random number generator.
     */
    private final static Random random = new Random();

    /**
     * List of scheduled tasks
     */
    private final Set<TimerTask> scheduledTasks = Collections.synchronizedSet(new HashSet<TimerTask>());

    /**
     * Describes the frequency and amount of effort we will spend updating
     * the peerview.
     */
    private int bootLevel = MIN_BOOTLEVEL;

    /**
     * Earliest absolute time in milliseconds at which we will allow a reseed
     * to take place.
     */
    private long earliestReseed = 0L;

    private final String uniqueGroupId;

    /**
     * Listeners for PeerView Events.
     */
    private final Set<PeerViewListener> rpvListeners = Collections.synchronizedSet(new HashSet<PeerViewListener>());

    /**
     * Used for querying for pves.
     */
    private InputPipe wirePipeInputPipe = null;

    /**
     * Used for querying for pves.
     */
    private OutputPipe wirePipeOutputPipe = null;

    /**
     * Used for notifications about pve failures.
     */
    private InputPipe localGroupWirePipeInputPipe = null;

    /**
     * Used for notifications about pve failures.
     */