import socket import threading import math import xctl from synchronizer import Synchronizer, ISynchronizable from codec import QemuEncoder, QemuDecoder import utils DEBUG_SYNC = False DEBUG_SEND_CMDS = False DEBUG_RECV_CMDS = False DEBUG_SEND_CMDS_VERBOSE = False DEBUG_RECV_CMDS_VERBOSE = False DEBUG_SNAPSHOTS = False DEFAULT_LATENCY = 0.1 DEFAULT_LOSS_RATE = 0.0 DEFAULT_SEND_BW = 1048576 # in bits/sec DEFAULT_SEND_QUEUE = 1048576 # in bits def secToNsec(secs): return long(secs*1000000000) def nsecToSec(nsecs): return float(nsecs)/1000000000 SNAPSHOT_PERIOD_NSECS = secToNsec(2) class PacketInfo(object): next_uid = 0 def __init__(self, sendTime, queueTime, receiveTime, src, dst, desc): self.sendTime = sendTime self.queueTime = queueTime self.receiveTime = receiveTime self.src = src self.dst = dst self.desc = desc self.uid = PacketInfo.next_uid PacketInfo.next_uid += 1 class IRemoteSlavedResponseHandler(object): def __init__(self): pass def onLocalReady(self, slaved, nsecs, log): pass def onSnapshotTaken(self, slaved, ids): pass class RemoteSlaved(xctl.IPollable, xctl.IChannelHandler): __slots__ = ["__id", "__channel", "__handlers"] def __init__(self, myid, addr): self.__id = myid self.__sock = socket.socket() self.__addr = addr self.__sock.connect(addr) self.__channel = xctl.Channel(self.__sock) self.__channel.registerHandler(self) self.__handlers = [] if DEBUG_RECV_CMDS: self.registerHandler( RemoteSlavedLoggingHandler(`self.getID()`+": ")) def getID(self): return self.__id def getChannel(self): return self.__channel def getHost(self): return self.__addr[0] def getPort(self): return self.__addr[1] def getNetPort(self): return self.__addr[1]+1 # Assume next port def registerHandler(self, handler): assert isinstance(handler, IRemoteSlavedResponseHandler) self.__handlers.append(handler) def unregisterHandler(self, handler): self.__handlers.remove(handler) def onPacket(self, chan, packet): if packet == None: method = "Close" args = () else: cmd = packet.cmd dec = QemuDecoder(packet.args) if cmd == "lrdy": method = "LocalReady" nsecs = dec.get_be64() logCount = dec.get_be32() log = [] for i in range(logCount): sendTime = dec.get_be64() queueTime = dec.get_be64() deliveryTime = dec.get_be64() srcMac = utils.MacAddr(dec.get_string()) dstMac = utils.MacAddr(dec.get_string()) desc = dec.get_string() pi = PacketInfo(sendTime, queueTime, deliveryTime, srcMac, dstMac, desc) log.append(pi) args = (nsecs, log) elif cmd == "snap": method = "SnapshotTaken" nids = dec.get_be32() ids = [dec.get_string() for _ in range(nids)] args = (ids,) else: raise RuntimeError, "Unknown command received %s" % `cmd` for handler in self.__handlers: func = getattr(handler, "on" + method) func(self, *args) if packet is None: # Close the channel self.__channel = None def cmdInit(self, quantum): if DEBUG_SEND_CMDS: print "%s: master->slave initial configuration" % self.__id enc = QemuEncoder() enc.put_be32(quantum) self.__channel.sendPacket(xctl.Packet("init", str(enc))) def cmdNode(self, nodeSpec): if DEBUG_SEND_CMDS: print "%s: master->slave node config %d" % (self.__id, nodeSpec.nodeID) enc = QemuEncoder() enc.put_be32(nodeSpec.nodeID) enc.put_byte(nodeSpec.ipOctet) enc.put_string(nodeSpec.hda) enc.put_string(nodeSpec.kernel) enc.put_string(nodeSpec.append) enc.put_string(nodeSpec.cdrom) enc.put_be32(nodeSpec.cpuSpeed) self.__channel.sendPacket(xctl.Packet("node", str(enc))) def cmdMacMap(self, octet, addr): host = addr[0] port = addr[1] if DEBUG_SEND_CMDS: print "%s: master->slave MAC %d at %s:%d" % (self.__id, octet, host, port) enc = QemuEncoder() enc.put_byte(octet) enc.put_string(host) enc.put_be16(port) self.__channel.sendPacket(xctl.Packet("macm", str(enc))) def cmdLatency(self, srcOctet, dstOctet, latency): if DEBUG_SEND_CMDS: print "%s: master->slave set latency %d -> %d = %d" % (self.__id, srcOctet, dstOctet, latency) enc = QemuEncoder() enc.put_byte(srcOctet) enc.put_byte(dstOctet) enc.put_be64(latency) self.__channel.sendPacket(xctl.Packet("ltnc", str(enc))) def cmdLossRate(self, srcOctet, dstOctet, rate): if DEBUG_SEND_CMDS: print "%s: master->slave set loss rate %d -> %d = %f" % (self.__id, srcOctet, dstOctet, rate) enc = QemuEncoder() enc.put_byte(srcOctet) enc.put_byte(dstOctet) enc.put_be32(int(rate*1e6)) self.__channel.sendPacket(xctl.Packet("loss", str(enc))) def cmdSendBottleneck(self, srcOctet, bw, qlen): if DEBUG_SEND_CMDS: print ("%s: master->slave set send bottleneck %d = " + \ "bw=%d, qlen=%d") % (self.__id, srcOctet, bw, qlen) enc = QemuEncoder() enc.put_byte(srcOctet) enc.put_be32(bw) enc.put_be32(qlen) self.__channel.sendPacket(xctl.Packet("snbn", str(enc))) def cmdStart(self): if DEBUG_SEND_CMDS: print "%s: master->slave start execution" % self.__id self.__channel.sendPacket(xctl.Packet("strt")) def cmdSynchronizerProceed(self, n, isSnapshot): if DEBUG_SEND_CMDS_VERBOSE: print "%s: master->slave sync proceed (wait for %d)" % (self.__id, n) enc = QemuEncoder() enc.put_be32(n) if isSnapshot: enc.put_byte(1) else: enc.put_byte(0) self.__channel.sendPacket(xctl.Packet("sypr", str(enc))) def cmdRollback(self, ids): if DEBUG_SEND_CMDS: print "%s: master->slave rollback(%s)" % (self.__id, `ids`) enc = QemuEncoder() enc.put_be32(len(ids)) for i in ids: enc.put_string(i) self.__channel.sendPacket(xctl.Packet("roll", str(enc))) def cmdDropPacket(self, st, rt, src, dst): if DEBUG_SEND_CMDS: print "%s: master->slave drop packet" % (self.__id) enc = QemuEncoder() enc.put_be64(st) enc.put_be64(rt) enc.put_string(str(src)) enc.put_string(str(dst)) self.__channel.sendPacket(xctl.Packet("drpp", str(enc))) class RemoteSlavedLoggingHandler(IRemoteSlavedResponseHandler): def __init__(self, prefix = ""): self.prefix = prefix def onLocalReady(self, slaved, nsecs, log): if DEBUG_RECV_CMDS_VERBOSE: self.log("LocalReady(%d, %d packets)" % (nsecs, len(log))) def onSnapshotTaken(self, slaved, ids): if DEBUG_RECV_CMDS: self.log("SnapshotTaken([%d ids])" % len(ids)) def log(self, msg): print "%smaster<-slave %s" % (self.prefix, msg) class RemoteSlavedHandler(IRemoteSlavedResponseHandler, ISynchronizable): def __init__(self, slaved, ripcord, quantum, nodes): self.__slaved = slaved self.__ripcord = ripcord self.__quantum = quantum self.__nodes = nodes self.__synchronizer = None self.__curNsecs = 0 self.__snapshotOnNext = False def getSlaved(self): return self.__slaved def setSynchronizer(self, sync): self.__synchronizer = sync sync.join(self) def getTimeNsecs(self): return self.__curNsecs def initNodes(self): self.__slaved.cmdInit(self.__quantum) for node in self.__nodes: self.__slaved.cmdNode(node) for octet in self.__ripcord.macAddrMap.iterkeys(): self.__slaved.cmdMacMap(octet, self.__ripcord.getMacMapping(octet)) def sendLatency(self, srcOctet, dstOctet, latency): self.__slaved.cmdLatency(srcOctet, dstOctet, secToNsec(latency)) def sendLossRate(self, srcOctet, dstOctet, rate): self.__slaved.cmdLossRate(srcOctet, dstOctet, rate) def sendSendBottleneck(self, srcOctet, bw, qlen): self.__slaved.cmdSendBottleneck(srcOctet, bw, qlen) def start(self): self.__slaved.cmdStart() def dropPacket(self, st, rt, src, dst): self.__slaved.cmdDropPacket(st, rt, src, dst) def onLocalReady(self, slaved, nsecs, log): #print ("LocalReady(%d, %d packets)" % (nsecs, len(log))) if DEBUG_SYNC: print "Got local ready from %d @ %d" % (slaved.getID(), nsecs) self.__curNsecs = nsecs self.__ripcord.addReceivedPackets(log) self.__synchronizer.reached(self) if DEBUG_SYNC: self.__synchronizer.dump() def onSyncProceed(self): # Figure out how many packets the slave should expect log = self.__ripcord.getReceivedPackets() n = 0 for pi in log: srcOctet = pi.src.toTuple()[-1] dstOctet = pi.dst.toTuple()[-1] if pi.dst.isMulticast() or \ (self.__ripcord.macAddrMap[dstOctet] == self): # Also require non-local origin if (self.__ripcord.macAddrMap[srcOctet] != self): n += 1 if self.__snapshotOnNext: self.__snapshotOnNext = False self.__slaved.cmdSynchronizerProceed(n, True) else: self.__slaved.cmdSynchronizerProceed(n, False) def snapshotOnNext(self): self.__snapshotOnNext = True def onSnapshotTaken(self, slaved, ids): if DEBUG_SNAPSHOTS: print "Snapshot taken on slaved %d" % slaved.getID() print ids self.__ripcord.snapshotTaken(self, ids) def rollback(self, ids): self.__slaved.cmdRollback(ids) class CanopyNodeSpec: def __init__(self, ipOctet, cdrom, hda = "../rootimage/canopy.img", kernel = "../kernel/bzImage-2.4.32", append = "root=/dev/hda1" + " ide2=noprobe ide3=noprobe ide4=noprobe ide5=noprobe", cpuSpeed = 30000000): self.nodeID = 0 self.ipOctet = ipOctet self.hda = hda self.kernel = kernel self.append = append self.cdrom = cdrom self.cpuSpeed = cpuSpeed self.assignedSlave = None # XXX Kludge self.macAddr = utils.MacAddr("52:54:00:12:43:%2.2x" % self.ipOctet) self.ipAddr = "192.168.0.%d" % self.ipOctet class Canopy: """A Canopy represents the configuration of an emulation environment: the slave nodes that operate it, the virtual nodes that exist inside it, and the network that connects the nodes.""" def __init__(self, quantum = 100): self.quantum = quantum self.slaves = [] self.nodes = [] self.nextID = 0 self.latencies = {} # Maps spec for # (sender, receiver) to latency (s) self.losses = {} # Maps (s,r) specs to packet # loss fraction self.sendBottlenecks = {} # Map spec to (bandwidth, # queue length) def addSlave(self, host, port): if host == "localhost" or host == "127.0.0.1": print "************************************************************" print "WARNING: RUNNING SLAVES ON LOCALHOST IS PROBABLY A BAD IDEA!" print "************************************************************" self.slaves.append((host,port)) def addNode(self, spec): spec.nodeID = self.nextID self.nodes.append(spec) print "Created node id %d (IP 192.168.0.%d)" % (self.nextID, spec.ipOctet) for x in self.nodes: self.latencies[spec, x] = DEFAULT_LATENCY self.latencies[x, spec] = DEFAULT_LATENCY self.losses[spec, x] = DEFAULT_LOSS_RATE self.losses[x, spec] = DEFAULT_LOSS_RATE self.sendBottlenecks[x] = (DEFAULT_SEND_BW, DEFAULT_SEND_QUEUE) self.nextID += 1 def assignSlaves(self): for i, x in enumerate(self.nodes): x.assignedSlave = i % len(self.slaves) def go(self): self.assignSlaves() r = Ripcord(self) return r class Ripcord(ISynchronizable, xctl.IPollIdler): """A Ripcord provides control over a Canopy.""" def __init__(self, canopy): self.__canopy = canopy self.__handlers = [] self.__pl = xctl.PollLoop() self.macAddrMap = {} # Maps last octet of MAC addr # to handler self.__receivedPackets = [] # Global log for this quantum self.__pl.registerIdler(self, 10) self.__clearReceivedOnNext = False self.__packets = {} # Maps uid to PacketInfo # Create slaveds and handlers for i, x in enumerate(self.__canopy.slaves): slaved = RemoteSlaved(i, x) assignedNodes = [] for node in self.__canopy.nodes: if node.assignedSlave == i: assignedNodes.append(node) handler = RemoteSlavedHandler(slaved, self, self.__canopy.quantum, assignedNodes) self.__handlers.append(handler) slaved.registerHandler(handler) self.__pl.register(slaved) for node in assignedNodes: self.macAddrMap[node.ipOctet] = handler # Tie everything to the global synchronizer self.__sync = Synchronizer() self.__sync.join(self, remainder = True) for handler in self.__handlers: handler.setSynchronizer(self.__sync) self.__sync.start() # Setup nodes for handler in self.__handlers: handler.initNodes() # Setup network self.sendNetInfo() # Start 'em up for handler in self.__handlers: handler.start() # Initialize debugger state self.__running = False self.__inRemainder = False self.__breakpointNS = None self.__lastSnapshotNS = None self.__snapshots = {} # Create the master poll thread self.__masterThread = threading.Thread(name = "MasterLoop", target = xctl.mainLoop, args = (self.__pl,)) self.__masterThread.setDaemon(True) self.__threadCommander = utils.ThreadCommander(self.__masterThread) self.__masterThread.start() def sendNetInfo(self): for pair, latency in self.__canopy.latencies.iteritems(): srcHandler = self.macAddrMap[pair[0].ipOctet] srcHandler.sendLatency(pair[0].ipOctet, pair[1].ipOctet, latency) for pair, loss in self.__canopy.losses.iteritems(): srcHandler = self.macAddrMap[pair[0].ipOctet] srcHandler.sendLossRate(pair[0].ipOctet, pair[1].ipOctet, loss) for src, bottleneck in self.__canopy.sendBottlenecks.iteritems(): srcHandler.sendSendBottleneck(src.ipOctet, bottleneck[0], bottleneck[1]) def setLatency(self, srcOctet, dstOctet, latency): srcSpec = [x for x in self.__canopy.nodes if x.ipOctet == srcOctet][0] dstSpec = [x for x in self.__canopy.nodes if x.ipOctet == dstOctet][0] self.__canopy.latencies[srcSpec,dstSpec] = latency # XXX Go ahead and resend all latencies to everyone. Obviously # this isn't necessary, but it shouldn't happen enough to # worry about. self.sendNetInfo() def setLossRate(self, srcOctet, dstOctet, rate): srcSpec = [x for x in self.__canopy.nodes if x.ipOctet == srcOctet][0] dstSpec = [x for x in self.__canopy.nodes if x.ipOctet == dstOctet][0] self.__canopy.losses[srcSpec,dstSpec] = rate self.sendNetInfo() def setSendBottleneck(self, srcOctet, bw, queuelen): srcSpec = [x for x in self.__canopy.nodes if x.ipOctet == srcOctet][0] self.__canopy.sendBottlenecks[srcSpec] = (bw, queuelen) self.sendNetInfo() def getMacMapping(self, octet): handler = self.macAddrMap[octet] return (handler.getSlaved().getHost(), handler.getSlaved().getNetPort()) def __maybeClearReceived(self): if self.__clearReceivedOnNext: self.__receivedPackets = [] self.__clearReceivedOnNext = False def addReceivedPackets(self, log): self.__maybeClearReceived() self.__receivedPackets += log def getReceivedPackets(self): return self.__receivedPackets def onSyncProceed(self): self.__maybeClearReceived() self.__clearReceivedOnNext = True for p in self.__receivedPackets: self.__packets[p.uid] = p # if len(self.__receivedPackets) > 0: # print "Received packets this quantum: " # for x in self.__receivedPackets: # print " send=%d receive=%d src=%s dst=%s" % (x.sendTime, # x.receiveTime, # str(x.src), # str(x.dst)) def onSyncRemainder(self): # Hit breakpoint? if (self.__breakpointNS is not None and self.__getTimeNsecs() >= self.__breakpointNS): print "Reached breakpoint at %g seconds" % \ nsecToSec(self.__breakpointNS) self.__running = False self.__breakpointNS = None self.__threadCommander.putResponse(None) # Time to snapshot? if (self.__lastSnapshotNS is None or (self.__getTimeNsecs() - self.__lastSnapshotNS >= SNAPSHOT_PERIOD_NSECS)): if DEBUG_SNAPSHOTS: print "Time to snapshot" self.__lastSnapshotNS = self.__getTimeNsecs() for handler in self.__handlers: handler.snapshotOnNext() # Officially enter remainder region and process a new command self.__inRemainder = True self.__handleCommand() # If running, leave the remainder region if self.__running: self.__inRemainder = False # Prevent onPollIdle from doubling-up # __sync.reached call self.__sync.reached(self) def snapshotTaken(self, slavedHandler, ids): snapshot = self.__snapshots.get(self.__lastSnapshotNS, None) if snapshot is None: snapshot = {} self.__snapshots[self.__lastSnapshotNS] = snapshot snapshot[slavedHandler] = ids if DEBUG_SNAPSHOTS: print "Current snapshots:" print self.__snapshots def canPollIdle(self): return self.__inRemainder def onPollIdle(self): if self.__inRemainder: self.__handleCommand() if self.__running: self.__inRemainder = False self.__sync.reached(self) def __handleCommand(self): cmd = self.__threadCommander.getCommand() if cmd is None: return cmd() def run(self): return self.__threadCommander.doCommand(self.__run) def __run(self): print "Running..." self.__running = True self.__threadCommander.putResponse(None) def stop(self): return self.__threadCommander.doCommand(self.__stop) def __stop(self): print "Stopping at %g secs..." % nsecToSec(self.__getTimeNsecs()) self.__running = False # Must be done in the remainder self.__threadCommander.putResponse(None) def step(self, secs): self.__threadCommander.doCommand(lambda : self.__step(secToNsec(secs))) def __step(self, nsecs): print "Stepping by %g seconds..." % nsecToSec(nsecs) target = long(self.__getTimeNsecs() + nsecs) self.__stepTo(target) def stepTo(self, asecs): self.__threadCommander.doCommand(lambda : self.__stepTo(secToNsec(asecs))) def __stepTo(self, ansecs): if ansecs < self.__getTimeNsecs(): print "That time is in the past (use rollback instead?)" self.__threadCommander.putResponse(None) else: print "Stepping to %g seconds..." % nsecToSec(ansecs) self.__running = True self.__breakpointNS = long(ansecs) def rollback(self, secs): self.__threadCommander.doCommand(lambda : self.__rollback(secToNsec(secs))) def __rollback(self, nsecs): print "Rolling back by %g seconds..." % nsecToSec(nsecs) target = long(self.__getTimeNsecs() - nsecs) self.__rollbackTo(target) def rollbackTo(self, asecs): self.__threadCommander.doCommand(lambda : self.__rollbackTo(secToNsec(asecs))) def __rollbackTo(self, ansecs): if ansecs >= self.__getTimeNsecs(): print "That time is in the future (use step instead?)" self.__threadCommander.putResponse(None) else: print "Rolling back to %g seconds..." % nsecToSec(ansecs) # Find the preceeding rollback point predecessorNS = None for snapshotNS in self.__snapshots.keys(): if snapshotNS <= ansecs: if predecessorNS is None or snapshotNS > predecessorNS: predecessorNS = snapshotNS # Set the breakpoint to roll forward self.__breakpointNS = ansecs # Update __lastSnapshotNS self.__lastSnapshotNS = predecessorNS print "(Specifically, rolling back to %d ns, then forward to %d ns)" % \ (predecessorNS, ansecs) # Perform the VM load snapshot = self.__snapshots[predecessorNS] for handler, ids in snapshot.items(): handler.rollback(ids) self.__sync.reset() # Discard packets that haven't been sent yet for k,v in self.__packets.items(): if v.sendTime >= predecessorNS: del self.__packets[k] # Start running so the BP can be reached self.__running = True def __packetsBetween(self, start, end): p = [packet for packet in self.__packets.itervalues() if packet.receiveTime >= start and packet.sendTime <= end] return p def listPackets(self): # Return a list ns = self.__getTimeNsecs() return self.__packetsBetween(ns, ns) def showPackets(self): # Pretty-print listPackets print "%4s %9s %9s %9s %17s %17s" % ("Num", "SendTime", "QTime", "RecvTime", "Source", "Destination") for x in self.listPackets(): print "%4d %9.2f %9.2f %9.2f %17s -> %17s" % (x.uid, nsecToSec(x.sendTime), nsecToSec(x.queueTime), nsecToSec(x.receiveTime), str(x.src), str(x.dst)) def drawPackets(self, startTime = -5, endTime = 0): import postscript, math reload(postscript) # XXX now = nsecToSec(self.__getTimeNsecs()) if startTime < 0: startTime = now + startTime if endTime <= 0: endTime = now + endTime endNS = secToNsec(math.floor(endTime)) startNS = secToNsec(math.ceil(startTime)) p = self.__packetsBetween(startNS, endNS) postscript.drawPackets(startNS, endNS, self.__canopy.nodes, p) def dropPacket(self, n): self.__threadCommander.doCommand(lambda : self._dropPacket(n)) def _dropPacket(self, n): p = self.__packets[n] if p.dst.isMulticast(): hList = self.__handlers else: hList = [self.macAddrMap[p.dst.toTuple()[-1]]] for h in hList: h.dropPacket(p.sendTime, p.receiveTime, p.src, p.dst) del self.__packets[n] self.__threadCommander.putResponse(None) def delayPacket(self, n, bysecs): pass def showInfo(self): self.__threadCommander.doCommand(self.__showInfo) def __showInfo(self): # Print list of nodes (IP, MAC), current time print "Running:", self.__running print "Last snapshot at:", if self.__lastSnapshotNS is None: print "(never)" else: print "%g secs" % nsecToSec(self.__lastSnapshotNS) print "Breakpoint:", if self.__breakpointNS is None: print "Not set" else: print "%g secs" % nsecToSec(self.__breakpointNS) print "Nodes:" for node in self.__canopy.nodes: print " %s (%s)" % (node.ipAddr, node.macAddr) print " Image: %s" % node.cdrom print "Time: %g secs" % nsecToSec(self.__getTimeNsecs()) self.__threadCommander.putResponse(None) def now(self): return self.__threadCommander.doCommand(self.__now) def __now(self): self.__threadCommander.putResponse(nsecToSec(self.__getTimeNsecs())) def __getTimeNsecs(self): return self.__handlers[0].getTimeNsecs() def main(): global r c = Canopy(100) c.addSlave("awakening.mit.edu", 19013) c.addSlave("awakening.mit.edu", 19015) #c.addSlave("18.239.6.147", 19007) #c.addSlave("localhost", 19007) c.addNode(CanopyNodeSpec(1, "../../experimental/austin/linux-test/bbc-2.1.iso")) c.addNode(CanopyNodeSpec(2, "../../experimental/austin/linux-test/bbc-2.1.iso")) # c.addNode(CanopyNodeSpec(3, "../rootimage/foo.img", # "../kernel/bzImage-2.4.32", # "root=/dev/hda1" + # " ide2=noprobe ide3=noprobe ide4=noprobe ide5=noprobe", # "../../experimental/austin/linux-test/bbc-2.1.iso", # 30000000)) # c.addNode(CanopyNodeSpec(4, "../rootimage/foo.img", # "../kernel/bzImage-2.4.32", # "root=/dev/hda1" + # " ide2=noprobe ide3=noprobe ide4=noprobe ide5=noprobe", # "../../experimental/austin/linux-test/bbc-2.1.iso", # 30000000)) r = c.go() if __name__ == "__main__": main()