tahoe-lafs/src/allmydata/introducer.py

import re
from base64 import b32encode, b32decode
from zope.interface import implements
from twisted.application import service
from twisted.python import log
from foolscap import Referenceable
from allmydata import node
from allmydata.interfaces import RIIntroducer, RIIntroducerClient
from allmydata.util import observer

class IntroducerNode(node.Node):
    PORTNUMFILE = "introducer.port"
    NODETYPE = "introducer"
    ENCODING_PARAMETERS_FILE = "encoding_parameters"
    DEFAULT_K, DEFAULT_DESIRED, DEFAULT_N = 3, 7, 10

    def tub_ready(self):
        k, desired, n = self.DEFAULT_K, self.DEFAULT_DESIRED, self.DEFAULT_N
        data = self.get_config("encoding_parameters")
        if data is not None:
            k,desired,n = data.split()
            k = int(k); desired = int(desired); n = int(n)
        introducerservice = IntroducerService(self.basedir, (k, desired, n))
        self.add_service(introducerservice)
        self.introducer_url = self.tub.registerReference(introducerservice, "introducer")
        self.log(" introducer is at %s" % self.introducer_url)
        self.write_config("introducer.furl", self.introducer_url + "\n")

class IntroducerService(service.MultiService, Referenceable):
    implements(RIIntroducer)
    name = "introducer"

    def __init__(self, basedir=".", encoding_parameters=None):
        service.MultiService.__init__(self)
        self.introducer_url = None
        self.nodes = set()
        self.furls = set()
        self._encoding_parameters = encoding_parameters

    def remote_hello(self, node, furl):
        log.msg("introducer: new contact at %s, node is %s" % (furl, node))
        def _remove():
            log.msg(" introducer: removing %s %s" % (node, furl))
            self.nodes.remove(node)
            self.furls.remove(furl)
        node.notifyOnDisconnect(_remove)
        self.furls.add(furl)
        node.callRemote("new_peers", self.furls)
        if self._encoding_parameters is not None:
            node.callRemote("set_encoding_parameters",
                            self._encoding_parameters)
        for othernode in self.nodes:
            othernode.callRemote("new_peers", set([furl]))
        self.nodes.add(node)

class IntroducerClient(service.Service, Referenceable):
    implements(RIIntroducerClient)

    def __init__(self, tub, introducer_furl, my_furl):
        self.tub = tub
        self.introducer_furl = introducer_furl
        self.my_furl = my_furl

        self.connections = {} # k: nodeid, v: ref
        self.reconnectors = {} # k: FURL, v: reconnector
        self._connected = False

        self.connection_observers = observer.ObserverList()
        self.encoding_parameters = None

        # The N'th element of _observers_of_enough_peers is None if nobody has
        # asked to be informed when N peers become connected, it is a
        # OneShotObserverList if someone has asked to be informed, and that
        # list is fired when N peers next become connected (or immediately if
        # N peers are already connected when someone asks), and the N'th
        # element is replaced by None when the number of connected peers falls
        # below N.  _observers_of_enough_peers is always just long enough to
        # hold the highest-numbered N that anyone is interested in (i.e.,
        # there are never trailing Nones in _observers_of_enough_peers).
        self._observers_of_enough_peers = []

    def startService(self):
        service.Service.startService(self)
        self.introducer_reconnector = self.tub.connectTo(self.introducer_furl,
                                                         self._got_introducer)
        def connect_failed(failure):
            self.log("\n\nInitial Introducer connection failed: "
                     "perhaps it's down\n")
            self.log(str(failure))
        d = self.tub.getReference(self.introducer_furl)
        d.addErrback(connect_failed)

    def log(self, msg):
        self.parent.log(msg)

    def remote_new_peers(self, furls):
        for furl in furls:
            self._new_peer(furl)

    def remote_set_encoding_parameters(self, parameters):
        self.encoding_parameters = parameters

    def stopService(self):
        service.Service.stopService(self)
        self.introducer_reconnector.stopConnecting()
        for reconnector in self.reconnectors.itervalues():
            reconnector.stopConnecting()

    def _new_peer(self, furl):
        if furl in self.reconnectors:
            return
        # TODO: rather than using the TubID as a nodeid, we should use
        # something else. The thing that requires the least additional
        # mappings is to use the foolscap "identifier" (the last component of
        # the furl), since these are unguessable. Before we can do that,
        # though, we need a way to conveniently make these identifiers
        # persist from one run of the client program to the next. Also, using
        # the foolscap identifier would mean that anyone who knows the name
        # of the node also has all the secrets they need to contact and use
        # them, which may or may not be what we want.
        m = re.match(r'pb://(\w+)@', furl)
        assert m
        nodeid = b32decode(m.group(1).upper())
        def _got_peer(rref):
            self.log("connected to %s" % b32encode(nodeid).lower()[:8])
            self.connection_observers.notify(nodeid, rref)
            self.connections[nodeid] = rref
            if len(self._observers_of_enough_peers) > len(self.connections):
                osol = self._observers_of_enough_peers[len(self.connections)]
                if osol:
                    osol.fire(None)
            def _lost():
                # TODO: notifyOnDisconnect uses eventually(), but connects do
                # not. Could this cause a problem?
                del self.connections[nodeid]
                if len(self._observers_of_enough_peers) > len(self.connections):
                    self._observers_of_enough_peers[len(self.connections)] = None
                    while self._observers_of_enough_peers and (not self._observers_of_enough_peers[-1]):
                        self._observers_of_enough_peers.pop()
                for numpeers in self._observers_of_enough_peers:
                    if len(self.connections) == (numpeers-1):
                        # We know that this observer list must have been
                        # fired, since we had enough peers before this one was
                        # lost.
                        del self._observers_of_enough_peers[numpeers]

            rref.notifyOnDisconnect(_lost)
        self.log("connecting to %s" % b32encode(nodeid).lower()[:8])
        self.reconnectors[furl] = self.tub.connectTo(furl, _got_peer)

    def _got_introducer(self, introducer):
        self.log("introducing ourselves: %s, %s" % (self, self.my_furl[6:13]))
        self._connected = True
        d = introducer.callRemote("hello",
                                  node=self,
                                  furl=self.my_furl)
        introducer.notifyOnDisconnect(self._disconnected)

    def _disconnected(self):
        self.log("bummer, we've lost our connection to the introducer")
        self._connected = False

    def notify_on_new_connection(self, cb):
        """Register a callback that will be fired (with nodeid, rref) when
        a new connection is established."""
        self.connection_observers.subscribe(cb)

    def connected_to_introducer(self):
        return self._connected

    def get_all_peerids(self):
        return self.connections.iterkeys()

    def get_all_peers(self):
        return self.connections.iteritems()

    def when_enough_peers(self, numpeers):
        """
        I return a deferred that fires the next time that at least numpeers
        are connected, or fires immediately if numpeers are currently
        available.
        """
        self._observers_of_enough_peers.extend([None]*(numpeers+1-len(self._observers_of_enough_peers)))
        if not self._observers_of_enough_peers[numpeers]:
            self._observers_of_enough_peers[numpeers] = observer.OneShotObserverList()
            if len(self.connections) >= numpeers:
                self._observers_of_enough_peers[numpeers].fire(self)
        return self._observers_of_enough_peers[numpeers].when_fired()