import sys, os, time from twisted.python import util from twisted.internet import defer, task from twisted.spread import pb from index import * from indexserver import IndexServer, Syncer, chordToArpdPort from chord import * import chord_types from connectioncache import ConnectionCache from utils import * waitFor = defer.waitForDeferred REPL_OFFER_ALL_INTERVAL = 60 # Time in seconds to wait between # sending a replication offer to # all nearby nodes REPL_CHURN_CHECK_INTERVAL = 1 # Time in seconds to wait between # checking for a change in # predecessor/successor lists. DEBUG = True DEBUG_VERBOSE = False class SimpleSyncer(Syncer): @typechecked(object) def __init__(self): pass @typechecked(object, IndexServer) def initialize(self, indexServer): self.indexServer = indexServer self.lastPredSuccList = [] # For detecting changes self.replOfferTask = task.LoopingCall( self.__replOfferToAllSuccsAndPreds) self.replOfferTask.start(REPL_OFFER_ALL_INTERVAL) self.replCheckTask = task.LoopingCall(self.__replCheckForChurn) self.replCheckTask.start(REPL_CHURN_CHECK_INTERVAL) @defer.deferredGenerator def __replicatedIndexes(self, k, numReplicas=None): """Return a list of all indexes we have that should be replicated on the index server with the specified key, assuming that each index is replicated numReplicas times. This requires that the server is in our predecessor or successor list -- usually this shouldn't be a problem because this is where we find servers to synchronize with anyway.""" if numReplicas == None: numReplicas = self.indexServer.numReplicas d = waitFor(self.indexServer.replicatedKeyspace(k, numReplicas)) yield d (a, b) = d.getResult() yield self.indexServer.indexesBetween(a,b) return @defer.deferredGenerator def __replGetMissingBlocks(self, hostport, blockIDs): """Given a list of blocks (as a list of (hash, expire)) available from the node at hostport, identify the ones that this node *should* have but *does not* currently have, and send a request to that node for those blocks.""" # XXX We can't easily figure out what blocks we should have, # since we're only exchanging a hash of the metadata. We'll # just have to trust the other node when they say that we # should have a block. if DEBUG_VERBOSE: print "REPL: getting missing blocks from", hostport requestBlocks = [] # Blocks to request for blockID, blockExpire in blockIDs: # Do we have the block? if self.indexServer.indexes.getByID(blockID) != None: if DEBUG_VERBOSE: print "Already have block", blockID self.indexServer.indexes.refreshExpire(blockID, blockExpire) else: if DEBUG_VERBOSE: print "Don't have block", blockID requestBlocks.append(blockID) if len(requestBlocks) == 0: if DEBUG: print "REPL: don't have any missing blocks from", hostport return if DEBUG: print "REPL: getting", len(requestBlocks), \ "missing blocks from", hostport # Connect to the server and request the blocks d = waitFor(self.indexServer.connectionCache.connect(hostport[0], chordToArpdPort(hostport[1]))) yield d remoteIndex = d.getResult() d = waitFor(remoteIndex.callRemote("replRequestBlocks", requestBlocks)) yield d blocks = d.getResult() for block, expireTime in blocks: d = waitFor(self.indexServer.addBlockToIndexes(block, expireTime)) yield d return @defer.deferredGenerator @typechecked(object, ChordNode) def __replSendOfferBlocks(self, chordNode): """Send a 'offer blocks' request to the specified Chord node with all the metadata records that we have that should be replicated on that node.""" # Find the indexes d = waitFor(self.__replicatedIndexes(chordNode.chordID)) yield d indexes = d.getResult() # Build the list of metadata records entries = set() for ind in indexes: entries.update(ind.entries) lst = [] for md in entries: mdID = hash(md) mdExpire = self.indexServer.indexes.getByID(mdID).expire lst.append((mdID, mdExpire)) if len(lst) == 0: if DEBUG: print "REPL: No blocks to offer to", chordNode return if DEBUG: print "REPL: Offering", len(lst), "blocks to", chordNode # Send it to the appropriate node d = waitFor(self.indexServer.connectionCache.connect(chordNode.host, chordToArpdPort(chordNode.port))) yield d remoteRoot = d.getResult() myHostPort = (self.indexServer.chord.myChordNode().host, self.indexServer.chord.myChordNode().port) d = waitFor(remoteRoot.callRemote("replOfferBlocks", myHostPort, lst)) yield d yield None return @defer.deferredGenerator def __replOfferToAllSuccsAndPreds(self): """Initiate the replication sequence with all nodes in the predecessor/successor list by sending them a list of blocks we have available for them.""" # Build a list of nodes we know about. d = waitFor(self.indexServer.chord.getPredList(self.indexServer.chord.myVnode())) yield d nodes = list(reversed(d.getResult())) d = waitFor(self.indexServer.chord.getSuccList(self.indexServer.chord.myVnode())) yield d nodes += d.getResult() nodes = uniquify(nodes) nodes = [x for x in nodes if x.chordID != self.indexServer.chord.myVnode()] # Remove myself if DEBUG: print "REPLICATING: offering to", [str(x.chordID) for x in nodes] # Offer to all in parallel try: dfds = [self.__replSendOfferBlocks(x) for x in nodes] for dfd in dfds: dfd.addErrback(self.__replErrWarning) d = waitFor(defer.DeferredList(dfds)) yield d except Exception, e: print "Replication error, ignoring: ", str(e) return def __replErrWarning(self, err): print "Ignoring replication error:", err @defer.deferredGenerator def __replCheckForChurn(self): """Check whether the predcessor/successor lists have changed, and if so send block offers.""" d = waitFor(self.indexServer.chord.getPredList(self.indexServer.chord.myVnode())) yield d nodes = list(reversed(d.getResult())) d = waitFor(self.indexServer.chord.getSuccList(self.indexServer.chord.myVnode())) yield d nodes += d.getResult() nodes = uniquify(nodes) nodes = [x for x in nodes if x.chordID != self.indexServer.chord.myVnode()] # Remove myself if nodes != self.lastPredSuccList: self.lastPredSuccList = nodes print "Predecessor/successor lists have changed" d = waitFor(self.__replOfferToAllSuccsAndPreds()) yield d return @typechecked(object, tuple, list) def handleOfferBlocks(self, hostport, blocks): """Handle a 'offer blocks' request --- that is, receive a list of metadata records that some remote node thinks this node should have. This RPC returns nothing interesting, but this should trigger a request to obtain any of the missing blocks.""" if DEBUG: print "REPL: Recieved offer of", len(blocks), \ "blocks from", hostport # print "Blocks are:", blocks self.__replGetMissingBlocks(hostport, blocks) return None @typechecked(object, list) def handleRequestBlocks(self, blockList): """Handle a 'request blocks' request --- that is, return a list of Metadata blocks and expire times corresponding to the IDs sent by the remote node.""" if DEBUG: print "REPL: received request for", len(blockList), "blocks" return [(self.indexServer.indexes.getByID(x).metadata, self.indexServer.indexes.getByID(x).expire) for x in blockList]