Author: mrogers
Date: 2006-10-31 11:35:46 +0000 (Tue, 31 Oct 2006)
New Revision: 10751
Added:
trunk/apps/load-balancing-sims/phase7/sim/CongestionWindow.java
trunk/apps/load-balancing-sims/phase7/sim/DeadlineQueue.java
trunk/apps/load-balancing-sims/phase7/sim/Event.java
trunk/apps/load-balancing-sims/phase7/sim/EventTarget.java
trunk/apps/load-balancing-sims/phase7/sim/LruCache.java
trunk/apps/load-balancing-sims/phase7/sim/LruMap.java
trunk/apps/load-balancing-sims/phase7/sim/NetworkInterface.java
trunk/apps/load-balancing-sims/phase7/sim/Node.java
trunk/apps/load-balancing-sims/phase7/sim/Packet.java
trunk/apps/load-balancing-sims/phase7/sim/Peer.java
trunk/apps/load-balancing-sims/phase7/sim/Sim.java
trunk/apps/load-balancing-sims/phase7/sim/TokenBucket.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/
trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkInsertHandler.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkRequestHandler.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/MessageHandler.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/RequestHandler.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/SskInsertHandler.java
trunk/apps/load-balancing-sims/phase7/sim/handlers/SskRequestHandler.java
trunk/apps/load-balancing-sims/phase7/sim/messages/
trunk/apps/load-balancing-sims/phase7/sim/messages/Accepted.java
trunk/apps/load-balancing-sims/phase7/sim/messages/Ack.java
trunk/apps/load-balancing-sims/phase7/sim/messages/Block.java
trunk/apps/load-balancing-sims/phase7/sim/messages/ChkDataFound.java
trunk/apps/load-balancing-sims/phase7/sim/messages/ChkInsert.java
trunk/apps/load-balancing-sims/phase7/sim/messages/ChkRequest.java
trunk/apps/load-balancing-sims/phase7/sim/messages/DataInsert.java
trunk/apps/load-balancing-sims/phase7/sim/messages/DataNotFound.java
trunk/apps/load-balancing-sims/phase7/sim/messages/InsertReply.java
trunk/apps/load-balancing-sims/phase7/sim/messages/Message.java
trunk/apps/load-balancing-sims/phase7/sim/messages/RejectedLoop.java
trunk/apps/load-balancing-sims/phase7/sim/messages/RouteNotFound.java
trunk/apps/load-balancing-sims/phase7/sim/messages/Search.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskAccepted.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskDataFound.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskInsert.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskPubKey.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskRequest.java
trunk/apps/load-balancing-sims/phase7/sim/messages/TransfersCompleted.java
Removed:
trunk/apps/load-balancing-sims/phase7/CongestionWindow.java
trunk/apps/load-balancing-sims/phase7/DeadlineQueue.java
trunk/apps/load-balancing-sims/phase7/Event.java
trunk/apps/load-balancing-sims/phase7/EventTarget.java
trunk/apps/load-balancing-sims/phase7/LruCache.java
trunk/apps/load-balancing-sims/phase7/LruMap.java
trunk/apps/load-balancing-sims/phase7/NetworkInterface.java
trunk/apps/load-balancing-sims/phase7/Node.java
trunk/apps/load-balancing-sims/phase7/Packet.java
trunk/apps/load-balancing-sims/phase7/Peer.java
trunk/apps/load-balancing-sims/phase7/Sim.java
trunk/apps/load-balancing-sims/phase7/TokenBucket.java
trunk/apps/load-balancing-sims/phase7/handlers/
trunk/apps/load-balancing-sims/phase7/messages/
trunk/apps/load-balancing-sims/phase7/sim/messages/Accepted.java
trunk/apps/load-balancing-sims/phase7/sim/messages/Ack.java
trunk/apps/load-balancing-sims/phase7/sim/messages/Block.java
trunk/apps/load-balancing-sims/phase7/sim/messages/ChkDataFound.java
trunk/apps/load-balancing-sims/phase7/sim/messages/ChkInsert.java
trunk/apps/load-balancing-sims/phase7/sim/messages/ChkRequest.java
trunk/apps/load-balancing-sims/phase7/sim/messages/DataInsert.java
trunk/apps/load-balancing-sims/phase7/sim/messages/DataNotFound.java
trunk/apps/load-balancing-sims/phase7/sim/messages/InsertReply.java
trunk/apps/load-balancing-sims/phase7/sim/messages/Message.java
trunk/apps/load-balancing-sims/phase7/sim/messages/RejectedLoop.java
trunk/apps/load-balancing-sims/phase7/sim/messages/RouteNotFound.java
trunk/apps/load-balancing-sims/phase7/sim/messages/Search.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskAccepted.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskDataFound.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskInsert.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskPubKey.java
trunk/apps/load-balancing-sims/phase7/sim/messages/SskRequest.java
trunk/apps/load-balancing-sims/phase7/sim/messages/TransfersCompleted.java
Modified:
trunk/apps/load-balancing-sims/phase7/sim/Network.java
Log:
Putting things in packages
Deleted: trunk/apps/load-balancing-sims/phase7/CongestionWindow.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/CongestionWindow.java 2006-10-31
11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/CongestionWindow.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -1,76 +0,0 @@
-// AIMD congestion control
-
-package sim;
-
-class CongestionWindow
-{
- public final static int MIN_CWIND = Packet.MAX_SIZE; // Min window size
- public final static int MAX_CWIND = 1000000; // Max window size
- public final static double ALPHA = 0.3125; // AIMD increase parameter
- public final static double BETA = 0.875; // AIMD decrease parameter
- public final static double GAMMA = 3.0; // Slow start divisor
-
- private double cwind = MIN_CWIND; // Size of window in bytes
- private int inflight = 0; // Bytes sent but not acked
- private boolean slowStart = true; // Are we in the slow start phase?
- private Peer peer; // The owner
-
- public CongestionWindow (Peer peer)
- {
- this.peer = peer;
- }
-
- public void reset()
- {
- peer.log ("congestion window decreased to " + MIN_CWIND);
- cwind = MIN_CWIND;
- peer.log ("returning to slow start");
- slowStart = true;
- }
-
- public int available()
- {
- return (int) cwind - inflight;
- }
-
- // Put bytes in flight
- public void bytesSent (int bytes)
- {
- inflight += bytes;
- peer.log (inflight + " bytes in flight");
- }
-
- // Take bytes out of flight
- public void bytesAcked (int bytes)
- {
- inflight -= bytes;
- peer.log (inflight + " bytes in flight");
- // Increase the window
- if (slowStart) cwind += bytes / GAMMA;
- else cwind += bytes * bytes * ALPHA / cwind;
- if (cwind > MAX_CWIND) cwind = MAX_CWIND;
- peer.log ("congestion window increased to " + cwind);
- }
-
- // Decrease the window when a packet is fast retransmitted
- public void fastRetransmission (double now)
- {
- peer.log (inflight + " bytes in flight");
- cwind *= BETA;
- if (cwind < MIN_CWIND) cwind = MIN_CWIND;
- peer.log ("congestion window decreased to " + cwind);
- // The slow start phase ends when the first packet is lost
- if (slowStart) {
- peer.log ("leaving slow start");
- slowStart = false;
- }
- }
-
- // Decrease the window when a packet is retransmitted due to a timeout
- public void timeout (double now)
- {
- peer.log (inflight + " bytes in flight");
- if (slowStart) fastRetransmission (now); // Leave slow start
- else reset(); // Reset the window and return to slow start
- }
-}
Deleted: trunk/apps/load-balancing-sims/phase7/DeadlineQueue.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/DeadlineQueue.java 2006-10-31
11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/DeadlineQueue.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -1,36 +0,0 @@
-// A queue storing outgoing messages and their coalescing deadlines
-
-package sim;
-import java.util.LinkedList;
-import sim.messages.Message;
-
-class DeadlineQueue<MESSAGE extends Message>
-{
- public int size = 0; // Size in bytes
- private LinkedList<MESSAGE> messages = new LinkedList<MESSAGE>();
-
- public void add (MESSAGE m)
- {
- size += m.size();
- messages.add (m);
- }
-
- public int headSize()
- {
- if (messages.isEmpty()) return 0;
- else return messages.peek().size();
- }
-
- public double deadline()
- {
- if (messages.isEmpty()) return Double.POSITIVE_INFINITY;
- else return messages.peek().deadline;
- }
-
- public MESSAGE pop()
- {
- MESSAGE m = messages.poll();
- if (m != null) size -= m.size();
- return m;
- }
-}
Deleted: trunk/apps/load-balancing-sims/phase7/Event.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/Event.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/Event.java 2006-10-31 11:35:46 UTC
(rev 10751)
@@ -1,99 +0,0 @@
-package sim;
-import java.util.TreeSet; // Gotta love the collections framework...
-
-class Event implements Comparable
-{
- // Static variables and methods for the event queue
-
- private static TreeSet<Event> queue = new TreeSet<Event>();
- private static double clockTime = 0.0;
- private static int nextId = 0;
- public static double duration = Double.POSITIVE_INFINITY;
-
- public static void reset()
- {
- queue.clear();
- clockTime = 0.0;
- nextId = 0;
- duration = Double.POSITIVE_INFINITY;
- }
-
- public static void schedule (EventTarget target, double time,
- int type, Object data)
- {
- queue.add (new Event (target, time + clockTime, type, data));
- }
-
- public static boolean nextEvent()
- {
- try {
- Event e = queue.first();
- queue.remove (e);
- // Update the clock
- clockTime = e.time;
- // Quit if the simulation's alloted time has run out
- if (clockTime > duration) return false;
- // Pass the packet to the target's callback method
- e.target.handleEvent (e.type, e.data);
- return true;
- }
- catch (java.util.NoSuchElementException x) {
- // No more events to dispatch
- return false;
- }
- }
-
- public static double time()
- {
- return clockTime;
- }
-
- public static void log (String message)
- {
- System.out.print (clockTime + " " + message + "\n");
- }
-
- // Run until the duration expires or there are no more events to process
- public static void run()
- {
- while (nextEvent()) {}
- }
-
- // Non-static variables and methods for individual events
-
- private EventTarget target;
- private double time;
- private int id;
- private int type;
- private Object data;
-
- public Event (EventTarget target, double time, int type, Object data)
- {
- this.target = target;
- this.time = time;
- this.type = type;
- this.data = data;
- id = nextId++;
- }
-
- // Must be consistent with compareTo()
- public boolean equals (Object o)
- {
- Event e = (Event) o;
- if (e.time == time && e.id == id) return true;
- return false;
- }
-
- // Must be consistent with equals()
- public int compareTo (Object o)
- {
- Event e = (Event) o;
- // Sort events by time (order of occurrence)
- if (e.time > time) return -1;
- if (e.time < time) return 1;
- // Break ties by ID (order of scheduling)
- if (e.id > id) return -1;
- if (e.id < id) return 1;
- return 0;
- }
-}
Deleted: trunk/apps/load-balancing-sims/phase7/EventTarget.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/EventTarget.java 2006-10-31
11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/EventTarget.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -1,6 +0,0 @@
-package sim;
-
-interface EventTarget
-{
- public void handleEvent (int type, Object data);
-}
Deleted: trunk/apps/load-balancing-sims/phase7/LruCache.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/LruCache.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/LruCache.java 2006-10-31 11:35:46 UTC
(rev 10751)
@@ -1,47 +0,0 @@
-// Limited-capacity LRU cache
-
-package sim;
-import java.util.LinkedHashSet;
-
-class LruCache<Key>
-{
- public final int capacity;
- private LinkedHashSet<Key> set;
-
- public LruCache (int capacity)
- {
- this.capacity = capacity;
- set = new LinkedHashSet<Key> (capacity);
- }
-
- public boolean get (Key key)
- {
- log ("searching cache for key " + key);
- if (set.remove (key)) {
- set.add (key); // Move the key to the fresh end
- return true;
- }
- return false;
- }
-
- public void put (Key key)
- {
- if (set.remove (key))
- log ("key " + key + " already in cache");
- else {
- log ("adding key " + key + " to cache");
- if (set.size() == capacity) {
- // Discard the oldest element
- Key oldest = set.iterator().next();
- log ("discarding key " + oldest);
- set.remove (oldest);
- }
- }
- set.add (key); // Add or move the key to the fresh end
- }
-
- private void log (String message)
- {
- // Event.log (message);
- }
-}
Deleted: trunk/apps/load-balancing-sims/phase7/LruMap.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/LruMap.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/LruMap.java 2006-10-31 11:35:46 UTC
(rev 10751)
@@ -1,60 +0,0 @@
-// Limited-capacity LRU cache that stores a value for each key
-
-package sim;
-import java.util.LinkedHashSet;
-import java.util.HashMap;
-
-class LruMap<Key,Value>
-{
- public final int capacity;
- private LinkedHashSet<Key> set;
- private HashMap<Key,Value> map;
-
- public LruMap (int capacity)
- {
- this.capacity = capacity;
- set = new LinkedHashSet<Key> (capacity);
- map = new HashMap<Key,Value> (capacity);
- }
-
- public Value get (Key key)
- {
- log ("searching cache for key " + key);
- Value value = map.get (key);
- if (value != null) {
- // Move the key to the fresh end
- set.remove (key);
- set.add (key);
- }
- return value;
- }
-
- // Return the existing value (which is not replaced), or null
- public Value put (Key key, Value value)
- {
- Value existing = map.get (key);
- if (existing == null) {
- log ("adding key " + key + " to cache");
- if (set.size() == capacity) {
- // Discard the oldest element
- Key oldest = set.iterator().next();
- log ("discarding key " + oldest);
- set.remove (oldest);
- }
- map.put (key, value);
- return value;
- }
- else {
- log ("key " + key + " already in cache");
- // Move the key to the fresh end
- set.remove (key);
- set.add (key);
- return existing;
- }
- }
-
- private void log (String message)
- {
- // Event.log (message);
- }
-}
Deleted: trunk/apps/load-balancing-sims/phase7/NetworkInterface.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/NetworkInterface.java 2006-10-31
11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/NetworkInterface.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -1,129 +0,0 @@
-// A node's low-level network interface
-
-package sim;
-import java.util.LinkedList;
-
-class NetworkInterface implements EventTarget
-{
- public int address; // Represents an IP address and port
- private Node node; // The owner of this network interface
- private double txSpeed, rxSpeed; // Bytes per second
-
- private LinkedList<Packet> txQueue; // Queue of outgoing packets
- private LinkedList<Packet> rxQueue; // Queue of incoming packets
- private int txQueueSize, rxQueueSize; // Limited-size drop-tail queues
- private int txQueueMaxSize, rxQueueMaxSize; // Bytes
-
- public NetworkInterface (Node node, double txSpeed, double rxSpeed)
- {
- this.node = node;
- this.txSpeed = txSpeed;
- this.rxSpeed = rxSpeed;
- txQueue = new LinkedList<Packet>();
- rxQueue = new LinkedList<Packet>();
- txQueueSize = rxQueueSize = 0; // Bytes currently queued
- txQueueMaxSize = 10000;
- rxQueueMaxSize = 20000;
- // Attach the interface to the network
- address = Network.register (this);
- }
-
- // Called by Peer
- public void send (Packet p, int dest, double latency)
- {
- p.src = address;
- p.dest = dest;
- p.latency = latency;
- if (txQueueSize + p.size > txQueueMaxSize) {
- log ("no room in txQueue, " + p + " lost");
- return;
- }
- txQueue.add (p);
- txQueueSize += p.size;
- log (txQueueSize + " bytes in txQueue");
- // If there are no other packets in the queue, start to transmit
- if (txQueue.size() == 1) txStart (p);
- }
-
- // Event callbacks
-
- // Add a packet to the rx queue
- private void rxQueueAdd (Packet p)
- {
- if (rxQueueSize + p.size > rxQueueMaxSize) {
- log ("no room in rxQueue, " + p + " lost");
- return;
- }
- rxQueue.add (p);
- rxQueueSize += p.size;
- log (rxQueueSize + " bytes in rxQueue");
- // If there are no other packets in the queue, start to receive
- if (rxQueue.size() == 1) rxStart (p);
- }
-
- // Start receiving a packet
- private void rxStart (Packet p)
- {
- log ("starting to receive " + p);
- // Delay depends on rx speed
- Event.schedule (this, p.size / rxSpeed, RX_END, p);
- }
-
- // Finish receiving a packet, pass it to the node
- private void rxEnd (Packet p)
- {
- log ("finished receiving " + p);
- node.handlePacket (p);
- rxQueueSize -= p.size;
- rxQueue.remove (p);
- // If there's another packet waiting, start to receive it
- if (!rxQueue.isEmpty()) rxStart (rxQueue.peek());
- }
-
- // Start transmitting a packet
- private void txStart (Packet p)
- {
- log ("starting to transmit " + p);
- // Delay depends on tx speed
- Event.schedule (this, p.size / txSpeed, TX_END, p);
- }
-
- // Finish transmitting a packet
- private void txEnd (Packet p)
- {
- log ("finished transmitting " + p);
- Network.deliver (p);
- txQueueSize -= p.size;
- txQueue.remove (p);
- // If there's another packet waiting, start to transmit it
- if (!txQueue.isEmpty()) txStart (txQueue.peek());
- }
-
- private void log (String message)
- {
- // Event.log (address + " " + message);
- }
-
- // EventTarget interface
- public void handleEvent (int type, Object data)
- {
- switch (type) {
- case RX_QUEUE:
- rxQueueAdd ((Packet) data);
- break;
-
- case RX_END:
- rxEnd ((Packet) data);
- break;
-
- case TX_END:
- txEnd ((Packet) data);
- break;
- }
- }
-
- // Each EventTarget class has its own event codes
- public final static int RX_QUEUE = 1;
- private final static int RX_END = 2;
- private final static int TX_END = 3;
-}
Deleted: trunk/apps/load-balancing-sims/phase7/Node.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/Node.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/Node.java 2006-10-31 11:35:46 UTC
(rev 10751)
@@ -1,443 +0,0 @@
-package sim;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.ArrayList;
-import java.util.Collections;
-import sim.handlers.*;
-import sim.messages.*;
-
-class Node implements EventTarget
-{
- // Token bucket bandwidth limiter
- public final static int BUCKET_RATE = 30000; // Bytes per second
- public final static int BUCKET_SIZE = 60000; // Burst size in bytes
-
- public double location; // Routing location
- public NetworkInterface net;
- private HashMap<Integer,Peer> peers; // Look up a peer by its address
- private HashSet<Integer> recentlySeenRequests; // Request IDs
- private HashMap<Integer,MessageHandler> messageHandlers; // By ID
- private LruCache<Integer> chkStore;
- private LruCache<Integer> chkCache;
- private LruMap<Integer,Integer> sskStore; // SSKs can collide
- private LruMap<Integer,Integer> sskCache;
- private LruCache<Integer> pubKeyCache; // SSK public keys
- private boolean decrementMaxHtl = false;
- private boolean decrementMinHtl = false;
- public TokenBucket bandwidth; // Bandwidth limiter
- private boolean timerRunning = false; // Is the retx timer running?
-
- public Node (double txSpeed, double rxSpeed)
- {
- this (Math.random(), txSpeed, rxSpeed);
- }
-
- public Node (double location, double txSpeed, double rxSpeed)
- {
- this.location = location;
- net = new NetworkInterface (this, txSpeed, rxSpeed);
- peers = new HashMap<Integer,Peer>();
- recentlySeenRequests = new HashSet<Integer>();
- messageHandlers = new HashMap<Integer,MessageHandler>();
- chkStore = new LruCache<Integer> (10);
- chkCache = new LruCache<Integer> (10);
- sskStore = new LruMap<Integer,Integer> (10);
- sskCache = new LruMap<Integer,Integer> (10);
- pubKeyCache = new LruCache<Integer> (10);
- if (Math.random() < 0.5) decrementMaxHtl = true;
- if (Math.random() < 0.25) decrementMinHtl = true;
- bandwidth = new TokenBucket (BUCKET_RATE, BUCKET_SIZE);
- }
-
- // Return true if a connection was added, false if already connected
- public boolean connect (Node n, double latency)
- {
- if (n == this) return false;
- if (peers.containsKey (n.net.address)) return false;
- // log ("adding peer " + n.net.address);
- Peer p = new Peer (this, n.net.address, n.location, latency);
- peers.put (n.net.address, p);
- return true;
- }
-
- public boolean connectBothWays (Node n, double latency)
- {
- if (connect (n, latency)) return n.connect (this, latency);
- else return false;
- }
-
- // Calculate the circular distance between two locations
- public static double distance (double a, double b)
- {
- if (a > b) return Math.min (a - b, b - a + 1.0);
- else return Math.min (b - a, a - b + 1.0);
- }
-
- // Convert an integer key to a routing location
- public static double keyToLocation (int key)
- {
- return key / (double) Integer.MAX_VALUE;
- }
-
- // Convert a routing location to an integer key
- public static int locationToKey (double location)
- {
- return (int) (location * Integer.MAX_VALUE);
- }
-
- // Return true if this node is as close to the target as any peer
- private boolean closerThanPeers (double target)
- {
- double bestDist = Double.POSITIVE_INFINITY;
- for (Peer peer : peers.values()) {
- double dist = distance (target, peer.location);
- if (dist < bestDist) bestDist = dist;
- }
- return distance (target, location) <= bestDist;
- }
-
- // Decrement a request or insert's hops to live
- public int decrementHtl (int htl)
- {
- if ((htl == Search.MAX_HTL && !decrementMaxHtl)
- || (htl == 1 && !decrementMinHtl)) return htl;
- else return htl - 1;
- }
-
- // Add a CHK to the cache
- public void cacheChk (int key)
- {
- log ("key " + key + " added to CHK cache");
- chkCache.put (key);
- }
-
- // Consider adding a CHK to the store
- public void storeChk (int key)
- {
- if (closerThanPeers (keyToLocation (key))) {
- log ("key " + key + " added to CHK store");
- chkStore.put (key);
- }
- else log ("key " + key + " not added to CHK store");
- }
-
- // Retrieve an SSK from the cache or the store
- public Integer fetchSsk (int key)
- {
- Integer data = sskStore.get (key);
- if (data == null) return sskCache.get (key);
- else return data;
- }
-
- // Add an SSK to the cache
- public void cacheSsk (int key, int value)
- {
- log ("key " + key + " added to SSK cache");
- sskCache.put (key, value);
- }
-
- // Consider adding an SSK to the store
- public void storeSsk (int key, int value)
- {
- if (closerThanPeers (keyToLocation (key))) {
- log ("key " + key + " added to SSK store");
- sskStore.put (key, value);
- }
- else log ("key " + key + " not added to SSK store");
- }
-
- // Add a public key to the cache
- public void cachePubKey (int key)
- {
- log ("public key " + key + " added to cache");
- pubKeyCache.put (key);
- }
-
- // Called by Peer
- public void startTimer()
- {
- if (timerRunning) return;
- // log ("starting retransmission/coalescing timer");
- Event.schedule (this, Peer.MAX_DELAY, CHECK_TIMEOUTS, null);
- timerRunning = true;
- }
-
- // Called by NetworkInterface
- public void handlePacket (Packet packet)
- {
- Peer peer = peers.get (packet.src);
- if (peer == null) log ("received packet from unknown peer");
- else peer.handlePacket (packet);
- }
-
- // Called by Peer
- public void handleMessage (Message m, Peer src)
- {
- if (src != null) log ("received " + m + " from " + src);
- if (m instanceof ChkRequest)
- handleChkRequest ((ChkRequest) m, src);
- else if (m instanceof ChkInsert)
- handleChkInsert ((ChkInsert) m, src);
- else if (m instanceof SskRequest)
- handleSskRequest ((SskRequest) m, src);
- else if (m instanceof SskInsert)
- handleSskInsert ((SskInsert) m, src);
- else {
- MessageHandler mh = messageHandlers.get (m.id);
- if (mh == null) log ("no message handler for " + m);
- else mh.handleMessage (m, src);
- }
- }
-
- private void handleChkRequest (ChkRequest r, Peer prev)
- {
- if (!recentlySeenRequests.add (r.id)) {
- log ("rejecting recently seen " + r);
- prev.sendMessage (new RejectedLoop (r.id));
- // Don't forward the same search back to prev
- MessageHandler mh = messageHandlers.get (r.id);
- if (mh != null) mh.removeNextHop (prev);
- return;
- }
- // Accept the search
- if (prev != null) {
- log ("accepting " + r);
- prev.sendMessage (new Accepted (r.id));
- }
- // If the data is in the store, return it
- if (chkStore.get (r.key)) {
- log ("key " + r.key + " found in CHK store");
- if (prev == null) log (r + " succeeded locally");
- else {
- prev.sendMessage (new ChkDataFound (r.id));
- for (int i = 0; i < 32; i++)
- prev.sendMessage (new Block (r.id, i));
- }
- return;
- }
- log ("key " + r.key + " not found in CHK store");
- // If the data is in the cache, return it
- if (chkCache.get (r.key)) {
- log ("key " + r.key + " found in CHK cache");
- if (prev == null) log (r + " succeeded locally");
- else {
- prev.sendMessage (new ChkDataFound (r.id));
- for (int i = 0; i < 32; i++)
- prev.sendMessage (new Block (r.id, i));
- }
- return;
- }
- log ("key " + r.key + " not found in CHK cache");
- // Store the request handler and forward the search
- ChkRequestHandler rh = new ChkRequestHandler (r, this, prev);
- messageHandlers.put (r.id, rh);
- rh.start();
- }
-
- private void handleChkInsert (ChkInsert i, Peer prev)
- {
- if (!recentlySeenRequests.add (i.id)) {
- log ("rejecting recently seen " + i);
- prev.sendMessage (new RejectedLoop (i.id));
- // Don't forward the same search back to prev
- MessageHandler mh = messageHandlers.get (i.id);
- if (mh != null) mh.removeNextHop (prev);
- return;
- }
- // Accept the search
- if (prev != null) {
- log ("accepting " + i);
- prev.sendMessage (new Accepted (i.id));
- }
- // Store the insert handler and wait for a DataInsert
- ChkInsertHandler ih = new ChkInsertHandler (i, this, prev);
- messageHandlers.put (i.id, ih);
- ih.start();
- }
-
- private void handleSskRequest (SskRequest r, Peer prev)
- {
- if (!recentlySeenRequests.add (r.id)) {
- log ("rejecting recently seen " + r);
- prev.sendMessage (new RejectedLoop (r.id));
- // Don't forward the same search back to prev
- MessageHandler mh = messageHandlers.get (r.id);
- if (mh != null) mh.removeNextHop (prev);
- return;
- }
- // Look up the public key
- boolean pub = pubKeyCache.get (r.key);
- if (pub) log ("public key " + r.key + " found in cache");
- else log ("public key " + r.key + " not found in cache");
- // Accept the search
- if (prev != null) {
- log ("accepting " + r);
- prev.sendMessage (new Accepted (r.id));
- }
- // If the data is in the store, return it
- Integer data = sskStore.get (r.key);
- if (pub && data != null) {
- log ("key " + r.key + " found in SSK store");
- if (prev == null) log (r + " succeeded locally");
- else {
- prev.sendMessage (new SskDataFound (r.id,data));
- if (r.needPubKey)
- prev.sendMessage
- (new SskPubKey (r.id, r.key));
- }
- return;
- }
- log ("key " + r.key + " not found in SSK store");
- // If the data is in the cache, return it
- data = sskCache.get (r.key);
- if (pub && data != null) {
- log ("key " + r.key + " found in SSK cache");
- if (prev == null) log (r + " succeeded locally");
- else {
- prev.sendMessage (new SskDataFound (r.id,data));
- if (r.needPubKey)
- prev.sendMessage
- (new SskPubKey (r.id, r.key));
- }
- return;
- }
- log ("key " + r.key + " not found in SSK cache");
- // Store the request handler and forward the search
- SskRequestHandler rh = new SskRequestHandler (r,this,prev,!pub);
- messageHandlers.put (r.id, rh);
- rh.start();
- }
-
- private void handleSskInsert (SskInsert i, Peer prev)
- {
- if (!recentlySeenRequests.add (i.id)) {
- log ("rejecting recently seen " + i);
- prev.sendMessage (new RejectedLoop (i.id));
- // Don't forward the same search back to prev
- MessageHandler mh = messageHandlers.get (i.id);
- if (mh != null) mh.removeNextHop (prev);
- return;
- }
- // Look up the public key
- boolean pub = pubKeyCache.get (i.key);
- if (pub) log ("public key " + i.key + " found in cache");
- else log ("public key " + i.key + " not found in cache");
- // Accept the search
- if (prev != null) {
- log ("accepting " + i);
- prev.sendMessage (new SskAccepted (i.id, !pub));
- }
- // Store the insert handler and possibly wait for the pub key
- SskInsertHandler ih = new SskInsertHandler (i,this,prev,!pub);
- messageHandlers.put (i.id, ih);
- ih.start();
- }
-
- public void removeMessageHandler (int id)
- {
- MessageHandler mh = messageHandlers.remove (id);
- if (mh == null) log ("no message handler to remove for " + id);
- else log ("removing message handler for " + id);
- }
-
- // Return the list of peers in a random order
- public ArrayList<Peer> peers()
- {
- ArrayList<Peer> copy = new ArrayList<Peer> (peers.values());
- Collections.shuffle (copy);
- return copy;
- }
-
- public void log (String message)
- {
- Event.log (net.address + " " + message);
- }
-
- // Event callbacks
-
- private void generateChkRequest (int key)
- {
- ChkRequest cr = new ChkRequest (key, location);
- log ("generating " + cr);
- handleChkRequest (cr, null);
- }
-
- private void generateChkInsert (int key)
- {
- ChkInsert ci = new ChkInsert (key, location);
- log ("generating " + ci);
- handleChkInsert (ci, null);
- handleMessage (new DataInsert (ci.id), null);
- for (int i = 0; i < 32; i++)
- handleMessage (new Block (ci.id, i), null);
- }
-
- private void generateSskRequest (int key)
- {
- SskRequest sr = new SskRequest (key, location, true);
- log ("generating " + sr);
- handleSskRequest (sr, null);
- }
-
- private void generateSskInsert (int key, int value)
- {
- SskInsert si = new SskInsert (key, value, location);
- log ("generating " + si);
- pubKeyCache.put (key);
- handleSskInsert (si, null);
- }
-
- private void checkTimeouts()
- {
- // Check the peers in a random order each time
- double deadline = Double.POSITIVE_INFINITY;
- for (Peer p : peers())
- deadline = Math.min (deadline, p.checkTimeouts());
- if (deadline == Double.POSITIVE_INFINITY) {
- // log ("stopping retransmission/coalescing timer");
- timerRunning = false;
- }
- else {
- double sleep = Math.max (deadline - Event.time(), 0.0);
- // log ("sleeping for " + sleep + " seconds");
- Event.schedule (this, sleep, CHECK_TIMEOUTS, null);
- }
- }
-
- // EventTarget interface
- public void handleEvent (int type, Object data)
- {
- switch (type) {
- case GENERATE_CHK_REQUEST:
- generateChkRequest ((Integer) data);
- break;
-
- case GENERATE_CHK_INSERT:
- generateChkInsert ((Integer) data);
- break;
-
- case GENERATE_SSK_REQUEST:
- generateSskRequest ((Integer) data);
- break;
-
- case GENERATE_SSK_INSERT:
- generateSskInsert ((Integer) data, 0);
- break;
-
- case GENERATE_SSK_COLLISION:
- generateSskInsert ((Integer) data, 1);
-
- case CHECK_TIMEOUTS:
- checkTimeouts();
- break;
- }
- }
-
- // Each EventTarget class has its own event codes
- public final static int GENERATE_CHK_REQUEST = 1;
- public final static int GENERATE_CHK_INSERT = 2;
- public final static int GENERATE_SSK_REQUEST = 3;
- public final static int GENERATE_SSK_INSERT = 4;
- public final static int GENERATE_SSK_COLLISION = 5;
- private final static int CHECK_TIMEOUTS = 6;
-}
Deleted: trunk/apps/load-balancing-sims/phase7/Packet.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/Packet.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/Packet.java 2006-10-31 11:35:46 UTC
(rev 10751)
@@ -1,42 +0,0 @@
-// A low-level packet (as opposed to a high-level message)
-
-package sim;
-import java.util.ArrayList;
-import sim.messages.Message;
-import sim.messages.Ack;
-
-class Packet
-{
- public final static int HEADER_SIZE = 70; // Including IP & UDP headers
- public final static int ACK_SIZE = 4; // Size of an ack in bytes
- public final static int MAX_SIZE = 1450; // MTU including headers
- public final static int SENSIBLE_PAYLOAD = 1000; // Coalescing
-
- public int src, dest; // Network addresses
- public int size = HEADER_SIZE; // Size in bytes, including headers
- public int seq = -1; // Data sequence number (-1 if no data)
- public ArrayList<Ack> acks = null;
- public ArrayList<Message> messages = null;
-
- public double sent; // Time at which the packet was (re) transmitted
- public double latency; // Link latency (stored here for convenience)
-
- public void addAck (Ack a)
- {
- if (acks == null) acks = new ArrayList<Ack>();
- acks.add (a);
- size += a.size();
- }
-
- public void addMessage (Message m)
- {
- if (messages == null) messages = new ArrayList<Message>();
- messages.add (m);
- size += m.size();
- }
-
- public String toString()
- {
- return new String ("packet " + src + ":" + dest + ":" + seq);
- }
-}
Deleted: trunk/apps/load-balancing-sims/phase7/Peer.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/Peer.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/Peer.java 2006-10-31 11:35:46 UTC
(rev 10751)
@@ -1,320 +0,0 @@
-package sim;
-import java.util.LinkedList;
-import java.util.Iterator;
-import java.util.HashSet;
-import sim.messages.*;
-
-class Peer
-{
- private Node node; // The local node
- public int address; // The remote node's address
- public double location; // The remote node's routing location
- private double latency; // The latency of the connection in seconds
-
- // Retransmission parameters
- public final static double RTO = 4.0; // Retransmission timeout in RTTs
- public final static double FRTO = 1.5; // Fast retx timeout in RTTs
- public final static double RTT_DECAY = 0.9; // Exp moving average
- public final static double LINK_IDLE = 8.0; // RTTs without transmitting
-
- // Coalescing
- public final static double MIN_SLEEP = 0.01; // Poll the b/w limiter
- public final static double MAX_DELAY = 0.1; // Coalescing delay in secs
-
- // Out-of-order delivery with eventual detection of missing packets
- public final static int SEQ_RANGE = 1000;
-
- // Sender state
- private double rtt = 5.0; // Estimated round-trip time in seconds
- private int txSeq = 0; // Sequence number of next outgoing data packet
- private int txMaxSeq = SEQ_RANGE - 1; // Highest sequence number
- private LinkedList<Packet> txBuffer; // Retransmission buffer
- private DeadlineQueue<Ack> ackQueue; // Outgoing acks
- private DeadlineQueue<Message> searchQueue; // Outgoing search messages
- private DeadlineQueue<Message> transferQueue; // Outgoing transfers
- private CongestionWindow window; // AIMD congestion window
- private double lastTransmission = 0.0; // Clock time
- private boolean tgif = false; // "Transfers go in first" toggle
-
- // Receiver state
- private HashSet<Integer> rxDupe; // Detect duplicates by sequence number
- private int rxSeq = 0; // Sequence number of next in-order incoming pkt
-
- public Peer (Node node, int address, double location, double latency)
- {
- this.node = node;
- this.address = address;
- this.location = location;
- this.latency = latency;
- txBuffer = new LinkedList<Packet>();
- ackQueue = new DeadlineQueue<Ack>();
- searchQueue = new DeadlineQueue<Message>();
- transferQueue = new DeadlineQueue<Message>();
- window = new CongestionWindow (this);
- rxDupe = new HashSet<Integer>();
- }
-
- // Queue a message for transmission
- public void sendMessage (Message m)
- {
- m.deadline = Event.time() + MAX_DELAY;
- if (m instanceof Block || m instanceof DataInsert
- || m instanceof ChkDataFound) {
- log (m + " added to transfer queue");
- transferQueue.add (m);
- }
- else {
- log (m + " added to search queue");
- searchQueue.add (m);
- }
- // Start the node's timer if necessary
- node.startTimer();
- // Send as many packets as possible
- while (send());
- }
-
- // Queue an ack for transmission
- private void sendAck (int seq)
- {
- log ("ack " + seq + " added to ack queue");
- ackQueue.add (new Ack (seq, Event.time() + MAX_DELAY));
- // Start the node's timer if necessary
- node.startTimer();
- // Send as many packets as possible
- while (send());
- }
-
- // Try to send a packet, return true if a packet was sent
- private boolean send()
- {
- if (ackQueue.size + searchQueue.size + transferQueue.size == 0){
- log ("nothing to send");
- return false;
- }
- log (ackQueue.size + " bytes of acks in queue");
- log (searchQueue.size + " bytes of searches in queue");
- log (transferQueue.size + " bytes of transfers in queue");
-
- // Return to slow start when the link is idle
- double now = Event.time();
- if (now - lastTransmission > LINK_IDLE * rtt) window.reset();
- lastTransmission = now;
-
- // Delay small packets for coalescing
- if (now < deadline (now)) {
- int payload = searchQueue.size + transferQueue.size;
- log ("delaying transmission of " + payload + " bytes");
- return false;
- }
-
- Packet p = new Packet();
-
- // Put all waiting acks in the packet
- while (ackQueue.size > 0) p.addAck (ackQueue.pop());
-
- // Don't send sequence number n+SEQ_RANGE until sequence
- // number n has been acked - this limits the number of
- // sequence numbers the receiver must store for replay
- // detection. We must still be allowed to send acks,
- // otherwise the connection could deadlock.
-
- if (txSeq > txMaxSeq)
- log ("waiting for ack " + (txMaxSeq - SEQ_RANGE + 1));
- else if (window.available() <= 0)
- log ("no room in congestion window for messages");
- else if (node.bandwidth.available() <= 0)
- log ("no bandwidth available for messages");
- else pack (p); // OK to send data
-
- // Don't send empty packets
- if (p.acks == null && p.messages == null) return false;
-
- // If the packet contains data, buffer it for retransmission
- if (p.messages != null) {
- p.seq = txSeq++;
- p.sent = now;
- txBuffer.add (p);
- window.bytesSent (p.size);
- }
-
- // Send the packet
- log ("sending packet " + p.seq + ", " + p.size + " bytes");
- node.net.send (p, address, latency);
- node.bandwidth.remove (p.size);
- return true;
- }
-
- // Called by Node when a packet arrives
- public void handlePacket (Packet p)
- {
- if (p.acks != null) for (Ack a : p.acks) handleAck (a);
- if (p.messages != null) handleData (p);
- }
-
- private void handleData (Packet p)
- {
- log ("received packet " + p.seq + ", " + p.size + " bytes");
- if (p.seq < rxSeq || rxDupe.contains (p.seq)) {
- log ("duplicate packet");
- sendAck (p.seq); // Original ack may have been lost
- }
- else if (p.seq == rxSeq) {
- log ("packet in order");
- // Find the sequence number of the next missing packet
- int was = rxSeq;
- while (rxDupe.remove (++rxSeq));
- log ("rxSeq was " + was + ", now " + rxSeq);
- // Deliver the packet
- unpack (p);
- sendAck (p.seq);
- }
- else if (p.seq < rxSeq + SEQ_RANGE) {
- log ("packet out of order - expected " + rxSeq);
- if (rxDupe.add (p.seq)) unpack (p);
- else log ("duplicate packet");
- sendAck (p.seq); // Original ack may have been lost
- }
- // This indicates a misbehaving sender - discard the packet
- else log ("warning: received " + p.seq + " before " + rxSeq);
- }
-
- private void handleAck (Ack a)
- {
- int seq = a.id;
- log ("received ack " + seq);
- double now = Event.time();
- Iterator<Packet> i = txBuffer.iterator();
- while (i.hasNext()) {
- Packet p = i.next();
- double age = now - p.sent;
- // Explicit ack
- if (p.seq == seq) {
- log ("packet " + p.seq + " acknowledged");
- i.remove();
- // Update the congestion window
- window.bytesAcked (p.size);
- // Update the average round-trip time
- rtt = rtt * RTT_DECAY + age * (1.0 - RTT_DECAY);
- log ("round-trip time " + age);
- log ("average round-trip time " + rtt);
- break;
- }
- // Fast retransmission
- if (p.seq < seq && age > FRTO * rtt + MAX_DELAY) {
- p.sent = now;
- log ("fast retransmitting packet " + p.seq);
- node.net.send (p, address, latency);
- window.fastRetransmission (now);
- }
- }
- // Recalculate the maximum sequence number
- if (txBuffer.isEmpty()) txMaxSeq = txSeq + SEQ_RANGE - 1;
- else txMaxSeq = txBuffer.peek().seq + SEQ_RANGE - 1;
- log ("maximum sequence number " + txMaxSeq);
- // Send as many packets as possible
- while (send());
- }
-
- // Add messages to a packet
- private void pack (Packet p)
- {
- // Alternate between giving searches and transfers priority
- if (tgif) {
- // Transfers go in first
- while (transferQueue.size > 0
- && p.size + transferQueue.headSize() <= Packet.MAX_SIZE)
- p.addMessage (transferQueue.pop());
- // Fill any remaining space with searches
- while (searchQueue.size > 0
- && p.size + searchQueue.headSize() <= Packet.MAX_SIZE)
- p.addMessage (searchQueue.pop());
- tgif = false;
- }
- else {
- // Searches go in first
- while (searchQueue.size > 0
- && p.size + searchQueue.headSize() <= Packet.MAX_SIZE)
- p.addMessage (searchQueue.pop());
- // Fill any remaining space with transfers
- while (transferQueue.size > 0
- && p.size + transferQueue.headSize() <= Packet.MAX_SIZE)
- p.addMessage (transferQueue.pop());
- tgif = true;
- }
- }
-
- // Remove messages from a packet and deliver them to the node
- private void unpack (Packet p)
- {
- if (p.messages == null) return;
- for (Message m : p.messages) node.handleMessage (m, this);
- }
-
- // Called by Node, returns the next coalescing or retx deadline
- public double checkTimeouts()
- {
- log ("checking timeouts");
- // Send as many packets as possible
- while (send());
-
- log (txBuffer.size() + " packets in flight");
- double now = Event.time();
- if (txBuffer.isEmpty()) return deadline (now);
-
- for (Packet p : txBuffer) {
- if (now - p.sent > RTO * rtt + MAX_DELAY) {
- // Retransmission timeout
- log ("retransmitting packet " + p.seq);
- p.sent = now;
- node.net.send (p, address, latency);
- window.timeout (now);
- }
- }
-
- // Sleep for up to MAX_DELAY seconds until the next deadline
- return Math.min (now + MAX_DELAY, deadline (now));
- }
-
- // Work out when the first ack or search or transfer needs to be sent
- private double deadline (double now)
- {
- return Math.min (ackQueue.deadline(), dataDeadline (now));
- }
-
- // Work out when the first search or transfer needs to be sent
- private double dataDeadline (double now)
- {
- // If there's no data waiting, use the ack deadline
- if (searchQueue.size + transferQueue.size == 0)
- return Double.POSITIVE_INFINITY;
-
- double deadline = Math.min (searchQueue.deadline(),
- transferQueue.deadline());
-
- // Delay small packets until the coalescing deadline
- if (searchQueue.size + transferQueue.size
- < Packet.SENSIBLE_PAYLOAD)
- return deadline;
-
- // If there's not enough room in the window, wait for an ack
- if (window.available() <= 0)
- return Double.POSITIVE_INFINITY;
-
- // If there's not enough bandwidth, try again shortly
- if (node.bandwidth.available() <= 0)
- return Math.max (now + MIN_SLEEP, deadline);
-
- // Send a packet immediately
- return now;
- }
-
- public void log (String message)
- {
- Event.log (node.net.address + ":" + address + " " + message);
- }
-
- public String toString()
- {
- return Integer.toString (address);
- }
-}
Deleted: trunk/apps/load-balancing-sims/phase7/Sim.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/Sim.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/Sim.java 2006-10-31 11:35:46 UTC
(rev 10751)
@@ -1,70 +0,0 @@
-package sim;
-
-class Sim
-{
- private final int NODES = 100; // Number of nodes
- private final int DEGREE = 4; // Average degree
- private final double SPEED = 20000; // Bytes per second
- private final double LATENCY = 0.1; // Seconds
- private Node[] nodes;
-
- public Sim()
- {
- Network.reorder = true;
- Network.lossRate = 0.001;
-
- // Create the nodes
- nodes = new Node[NODES];
- for (int i = 0; i < NODES; i++)
- nodes[i] = new Node (1.0 / NODES * i, SPEED, SPEED);
- // Connect the nodes
- makeKleinbergNetwork();
-
- int key = Node.locationToKey (Math.random());
- Event.schedule (nodes[0], 0.0,
- Node.GENERATE_CHK_INSERT, key);
- Event.schedule (nodes[NODES/4], 30.0,
- Node.GENERATE_CHK_REQUEST, key);
- Event.schedule (nodes[NODES/2], 60.0,
- Node.GENERATE_CHK_INSERT, key);
- Event.schedule (nodes[3*NODES/4], 90.0,
- Node.GENERATE_CHK_REQUEST, key);
-
- // Run the simulation
- Event.run();
- }
-
- // Return the lattice distance between a and b
- private int latticeDistance (int a, int b)
- {
- if (a > b) return Math.min (a - b, b - a + NODES);
- else return Math.min (b - a, a - b + NODES);
- }
-
- private void makeKleinbergNetwork()
- {
- // Calculate the normalising constant
- double norm = 0.0;
- for (int i = 1; i < NODES; i++)
- norm += 1.0 / latticeDistance (0, i);
-
- // Add DEGREE shortcuts per node, randomly with replacement
- for (int i = 0; i < NODES; i++) {
- for (int j = 0; j < i; j++) {
- double p = 1.0 / latticeDistance (i, j) / norm;
- for (int k = 0; k < DEGREE; k++) {
- if (Math.random() < p) {
- nodes[i].connectBothWays
- (nodes[j], LATENCY);
- break;
- }
- }
- }
- }
- }
-
- public static void main (String[] args)
- {
- new Sim();
- }
-}
Deleted: trunk/apps/load-balancing-sims/phase7/TokenBucket.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/TokenBucket.java 2006-10-31
11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/TokenBucket.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -1,31 +0,0 @@
-package sim;
-
-class TokenBucket
-{
- private double tokens, rate, size, lastUpdated;
-
- public TokenBucket (double rate, double size)
- {
- tokens = size;
- this.rate = rate;
- this.size = size;
- lastUpdated = 0.0; // Clock time
- }
-
- public int available()
- {
- double now = Event.time();
- double elapsed = now - lastUpdated;
- lastUpdated = now;
- tokens += elapsed * rate;
- if (tokens > size) tokens = size;
- // Event.log (tokens + " tokens available");
- return (int) tokens;
- }
-
- public void remove (int t)
- {
- tokens -= t; // Counter can go negative
- // Event.log (t + " tokens removed, " + tokens + " available");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/CongestionWindow.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/CongestionWindow.java)
Copied: trunk/apps/load-balancing-sims/phase7/sim/DeadlineQueue.java (from rev
10750, trunk/apps/load-balancing-sims/phase7/DeadlineQueue.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/DeadlineQueue.java 2006-10-31
11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/DeadlineQueue.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -0,0 +1,36 @@
+// A queue storing outgoing messages and their coalescing deadlines
+
+package sim;
+import sim.messages.Message;
+import java.util.LinkedList;
+
+class DeadlineQueue<MESSAGE extends Message>
+{
+ public int size = 0; // Size in bytes
+ private LinkedList<MESSAGE> messages = new LinkedList<MESSAGE>();
+
+ public void add (MESSAGE m)
+ {
+ size += m.size();
+ messages.add (m);
+ }
+
+ public int headSize()
+ {
+ if (messages.isEmpty()) return 0;
+ else return messages.peek().size();
+ }
+
+ public double deadline()
+ {
+ if (messages.isEmpty()) return Double.POSITIVE_INFINITY;
+ else return messages.peek().deadline;
+ }
+
+ public MESSAGE pop()
+ {
+ MESSAGE m = messages.poll();
+ if (m != null) size -= m.size();
+ return m;
+ }
+}
Copied: trunk/apps/load-balancing-sims/phase7/sim/Event.java (from rev 10750,
trunk/apps/load-balancing-sims/phase7/Event.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/Event.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/Event.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -0,0 +1,99 @@
+package sim;
+import java.util.TreeSet; // Gotta love the collections framework...
+
+public class Event implements Comparable
+{
+ // Static variables and methods for the event queue
+
+ private static TreeSet<Event> queue = new TreeSet<Event>();
+ private static double clockTime = 0.0;
+ private static int nextId = 0;
+ public static double duration = Double.POSITIVE_INFINITY;
+
+ public static void reset()
+ {
+ queue.clear();
+ clockTime = 0.0;
+ nextId = 0;
+ duration = Double.POSITIVE_INFINITY;
+ }
+
+ public static void schedule (EventTarget target, double time,
+ int type, Object data)
+ {
+ queue.add (new Event (target, time + clockTime, type, data));
+ }
+
+ public static boolean nextEvent()
+ {
+ try {
+ Event e = queue.first();
+ queue.remove (e);
+ // Update the clock
+ clockTime = e.time;
+ // Quit if the simulation's alloted time has run out
+ if (clockTime > duration) return false;
+ // Pass the packet to the target's callback method
+ e.target.handleEvent (e.type, e.data);
+ return true;
+ }
+ catch (java.util.NoSuchElementException x) {
+ // No more events to dispatch
+ return false;
+ }
+ }
+
+ public static double time()
+ {
+ return clockTime;
+ }
+
+ public static void log (String message)
+ {
+ System.out.print (clockTime + " " + message + "\n");
+ }
+
+ // Run until the duration expires or there are no more events to process
+ public static void run()
+ {
+ while (nextEvent()) {}
+ }
+
+ // Non-static variables and methods for individual events
+
+ private EventTarget target;
+ private double time;
+ private int id;
+ private int type;
+ private Object data;
+
+ public Event (EventTarget target, double time, int type, Object data)
+ {
+ this.target = target;
+ this.time = time;
+ this.type = type;
+ this.data = data;
+ id = nextId++;
+ }
+
+ // Must be consistent with compareTo()
+ public boolean equals (Object o)
+ {
+ Event e = (Event) o;
+ if (e.time == time && e.id == id) return true;
+ return false;
+ }
+
+ // Must be consistent with equals()
+ public int compareTo (Object o)
+ {
+ Event e = (Event) o;
+ // Sort events by time (order of occurrence)
+ if (e.time > time) return -1;
+ if (e.time < time) return 1;
+ // Break ties by ID (order of scheduling)
+ if (e.id > id) return -1;
+ if (e.id < id) return 1;
+ return 0;
+ }
+}
Copied: trunk/apps/load-balancing-sims/phase7/sim/EventTarget.java (from rev
10750, trunk/apps/load-balancing-sims/phase7/EventTarget.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/EventTarget.java 2006-10-31
11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/EventTarget.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -0,0 +1,6 @@
+package sim;
+
+public interface EventTarget
+{
+ public void handleEvent (int type, Object data);
+}
Copied: trunk/apps/load-balancing-sims/phase7/sim/LruCache.java (from rev
10750, trunk/apps/load-balancing-sims/phase7/LruCache.java)
Copied: trunk/apps/load-balancing-sims/phase7/sim/LruMap.java (from rev 10750,
trunk/apps/load-balancing-sims/phase7/LruMap.java)
Modified: trunk/apps/load-balancing-sims/phase7/sim/Network.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/sim/Network.java 2006-10-31
11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/Network.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -1,3 +1,4 @@
+package sim;
import java.util.HashMap;
class Network
Copied: trunk/apps/load-balancing-sims/phase7/sim/NetworkInterface.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/NetworkInterface.java)
Copied: trunk/apps/load-balancing-sims/phase7/sim/Node.java (from rev 10750,
trunk/apps/load-balancing-sims/phase7/Node.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/Node.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/Node.java 2006-10-31 11:35:46 UTC
(rev 10751)
@@ -0,0 +1,443 @@
+package sim;
+import sim.handlers.*;
+import sim.messages.*;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.ArrayList;
+import java.util.Collections;
+
+public class Node implements EventTarget
+{
+ // Token bucket bandwidth limiter
+ public final static int BUCKET_RATE = 30000; // Bytes per second
+ public final static int BUCKET_SIZE = 60000; // Burst size in bytes
+
+ public double location; // Routing location
+ public NetworkInterface net;
+ private HashMap<Integer,Peer> peers; // Look up a peer by its address
+ private HashSet<Integer> recentlySeenRequests; // Request IDs
+ private HashMap<Integer,MessageHandler> messageHandlers; // By ID
+ private LruCache<Integer> chkStore;
+ private LruCache<Integer> chkCache;
+ private LruMap<Integer,Integer> sskStore; // SSKs can collide
+ private LruMap<Integer,Integer> sskCache;
+ private LruCache<Integer> pubKeyCache; // SSK public keys
+ private boolean decrementMaxHtl = false;
+ private boolean decrementMinHtl = false;
+ public TokenBucket bandwidth; // Bandwidth limiter
+ private boolean timerRunning = false; // Is the retx timer running?
+
+ public Node (double txSpeed, double rxSpeed)
+ {
+ this (Math.random(), txSpeed, rxSpeed);
+ }
+
+ public Node (double location, double txSpeed, double rxSpeed)
+ {
+ this.location = location;
+ net = new NetworkInterface (this, txSpeed, rxSpeed);
+ peers = new HashMap<Integer,Peer>();
+ recentlySeenRequests = new HashSet<Integer>();
+ messageHandlers = new HashMap<Integer,MessageHandler>();
+ chkStore = new LruCache<Integer> (10);
+ chkCache = new LruCache<Integer> (10);
+ sskStore = new LruMap<Integer,Integer> (10);
+ sskCache = new LruMap<Integer,Integer> (10);
+ pubKeyCache = new LruCache<Integer> (10);
+ if (Math.random() < 0.5) decrementMaxHtl = true;
+ if (Math.random() < 0.25) decrementMinHtl = true;
+ bandwidth = new TokenBucket (BUCKET_RATE, BUCKET_SIZE);
+ }
+
+ // Return true if a connection was added, false if already connected
+ public boolean connect (Node n, double latency)
+ {
+ if (n == this) return false;
+ if (peers.containsKey (n.net.address)) return false;
+ // log ("adding peer " + n.net.address);
+ Peer p = new Peer (this, n.net.address, n.location, latency);
+ peers.put (n.net.address, p);
+ return true;
+ }
+
+ public boolean connectBothWays (Node n, double latency)
+ {
+ if (connect (n, latency)) return n.connect (this, latency);
+ else return false;
+ }
+
+ // Calculate the circular distance between two locations
+ public static double distance (double a, double b)
+ {
+ if (a > b) return Math.min (a - b, b - a + 1.0);
+ else return Math.min (b - a, a - b + 1.0);
+ }
+
+ // Convert an integer key to a routing location
+ public static double keyToLocation (int key)
+ {
+ return key / (double) Integer.MAX_VALUE;
+ }
+
+ // Convert a routing location to an integer key
+ public static int locationToKey (double location)
+ {
+ return (int) (location * Integer.MAX_VALUE);
+ }
+
+ // Return true if this node is as close to the target as any peer
+ private boolean closerThanPeers (double target)
+ {
+ double bestDist = Double.POSITIVE_INFINITY;
+ for (Peer peer : peers.values()) {
+ double dist = distance (target, peer.location);
+ if (dist < bestDist) bestDist = dist;
+ }
+ return distance (target, location) <= bestDist;
+ }
+
+ // Decrement a request or insert's hops to live
+ public int decrementHtl (int htl)
+ {
+ if ((htl == Search.MAX_HTL && !decrementMaxHtl)
+ || (htl == 1 && !decrementMinHtl)) return htl;
+ else return htl - 1;
+ }
+
+ // Add a CHK to the cache
+ public void cacheChk (int key)
+ {
+ log ("key " + key + " added to CHK cache");
+ chkCache.put (key);
+ }
+
+ // Consider adding a CHK to the store
+ public void storeChk (int key)
+ {
+ if (closerThanPeers (keyToLocation (key))) {
+ log ("key " + key + " added to CHK store");
+ chkStore.put (key);
+ }
+ else log ("key " + key + " not added to CHK store");
+ }
+
+ // Retrieve an SSK from the cache or the store
+ public Integer fetchSsk (int key)
+ {
+ Integer data = sskStore.get (key);
+ if (data == null) return sskCache.get (key);
+ else return data;
+ }
+
+ // Add an SSK to the cache
+ public void cacheSsk (int key, int value)
+ {
+ log ("key " + key + " added to SSK cache");
+ sskCache.put (key, value);
+ }
+
+ // Consider adding an SSK to the store
+ public void storeSsk (int key, int value)
+ {
+ if (closerThanPeers (keyToLocation (key))) {
+ log ("key " + key + " added to SSK store");
+ sskStore.put (key, value);
+ }
+ else log ("key " + key + " not added to SSK store");
+ }
+
+ // Add a public key to the cache
+ public void cachePubKey (int key)
+ {
+ log ("public key " + key + " added to cache");
+ pubKeyCache.put (key);
+ }
+
+ // Called by Peer
+ public void startTimer()
+ {
+ if (timerRunning) return;
+ // log ("starting retransmission/coalescing timer");
+ Event.schedule (this, Peer.MAX_DELAY, CHECK_TIMEOUTS, null);
+ timerRunning = true;
+ }
+
+ // Called by NetworkInterface
+ public void handlePacket (Packet packet)
+ {
+ Peer peer = peers.get (packet.src);
+ if (peer == null) log ("received packet from unknown peer");
+ else peer.handlePacket (packet);
+ }
+
+ // Called by Peer
+ public void handleMessage (Message m, Peer src)
+ {
+ if (src != null) log ("received " + m + " from " + src);
+ if (m instanceof ChkRequest)
+ handleChkRequest ((ChkRequest) m, src);
+ else if (m instanceof ChkInsert)
+ handleChkInsert ((ChkInsert) m, src);
+ else if (m instanceof SskRequest)
+ handleSskRequest ((SskRequest) m, src);
+ else if (m instanceof SskInsert)
+ handleSskInsert ((SskInsert) m, src);
+ else {
+ MessageHandler mh = messageHandlers.get (m.id);
+ if (mh == null) log ("no message handler for " + m);
+ else mh.handleMessage (m, src);
+ }
+ }
+
+ private void handleChkRequest (ChkRequest r, Peer prev)
+ {
+ if (!recentlySeenRequests.add (r.id)) {
+ log ("rejecting recently seen " + r);
+ prev.sendMessage (new RejectedLoop (r.id));
+ // Don't forward the same search back to prev
+ MessageHandler mh = messageHandlers.get (r.id);
+ if (mh != null) mh.removeNextHop (prev);
+ return;
+ }
+ // Accept the search
+ if (prev != null) {
+ log ("accepting " + r);
+ prev.sendMessage (new Accepted (r.id));
+ }
+ // If the data is in the store, return it
+ if (chkStore.get (r.key)) {
+ log ("key " + r.key + " found in CHK store");
+ if (prev == null) log (r + " succeeded locally");
+ else {
+ prev.sendMessage (new ChkDataFound (r.id));
+ for (int i = 0; i < 32; i++)
+ prev.sendMessage (new Block (r.id, i));
+ }
+ return;
+ }
+ log ("key " + r.key + " not found in CHK store");
+ // If the data is in the cache, return it
+ if (chkCache.get (r.key)) {
+ log ("key " + r.key + " found in CHK cache");
+ if (prev == null) log (r + " succeeded locally");
+ else {
+ prev.sendMessage (new ChkDataFound (r.id));
+ for (int i = 0; i < 32; i++)
+ prev.sendMessage (new Block (r.id, i));
+ }
+ return;
+ }
+ log ("key " + r.key + " not found in CHK cache");
+ // Store the request handler and forward the search
+ ChkRequestHandler rh = new ChkRequestHandler (r, this, prev);
+ messageHandlers.put (r.id, rh);
+ rh.start();
+ }
+
+ private void handleChkInsert (ChkInsert i, Peer prev)
+ {
+ if (!recentlySeenRequests.add (i.id)) {
+ log ("rejecting recently seen " + i);
+ prev.sendMessage (new RejectedLoop (i.id));
+ // Don't forward the same search back to prev
+ MessageHandler mh = messageHandlers.get (i.id);
+ if (mh != null) mh.removeNextHop (prev);
+ return;
+ }
+ // Accept the search
+ if (prev != null) {
+ log ("accepting " + i);
+ prev.sendMessage (new Accepted (i.id));
+ }
+ // Store the insert handler and wait for a DataInsert
+ ChkInsertHandler ih = new ChkInsertHandler (i, this, prev);
+ messageHandlers.put (i.id, ih);
+ ih.start();
+ }
+
+ private void handleSskRequest (SskRequest r, Peer prev)
+ {
+ if (!recentlySeenRequests.add (r.id)) {
+ log ("rejecting recently seen " + r);
+ prev.sendMessage (new RejectedLoop (r.id));
+ // Don't forward the same search back to prev
+ MessageHandler mh = messageHandlers.get (r.id);
+ if (mh != null) mh.removeNextHop (prev);
+ return;
+ }
+ // Look up the public key
+ boolean pub = pubKeyCache.get (r.key);
+ if (pub) log ("public key " + r.key + " found in cache");
+ else log ("public key " + r.key + " not found in cache");
+ // Accept the search
+ if (prev != null) {
+ log ("accepting " + r);
+ prev.sendMessage (new Accepted (r.id));
+ }
+ // If the data is in the store, return it
+ Integer data = sskStore.get (r.key);
+ if (pub && data != null) {
+ log ("key " + r.key + " found in SSK store");
+ if (prev == null) log (r + " succeeded locally");
+ else {
+ prev.sendMessage (new SskDataFound (r.id,data));
+ if (r.needPubKey)
+ prev.sendMessage
+ (new SskPubKey (r.id, r.key));
+ }
+ return;
+ }
+ log ("key " + r.key + " not found in SSK store");
+ // If the data is in the cache, return it
+ data = sskCache.get (r.key);
+ if (pub && data != null) {
+ log ("key " + r.key + " found in SSK cache");
+ if (prev == null) log (r + " succeeded locally");
+ else {
+ prev.sendMessage (new SskDataFound (r.id,data));
+ if (r.needPubKey)
+ prev.sendMessage
+ (new SskPubKey (r.id, r.key));
+ }
+ return;
+ }
+ log ("key " + r.key + " not found in SSK cache");
+ // Store the request handler and forward the search
+ SskRequestHandler rh = new SskRequestHandler (r,this,prev,!pub);
+ messageHandlers.put (r.id, rh);
+ rh.start();
+ }
+
+ private void handleSskInsert (SskInsert i, Peer prev)
+ {
+ if (!recentlySeenRequests.add (i.id)) {
+ log ("rejecting recently seen " + i);
+ prev.sendMessage (new RejectedLoop (i.id));
+ // Don't forward the same search back to prev
+ MessageHandler mh = messageHandlers.get (i.id);
+ if (mh != null) mh.removeNextHop (prev);
+ return;
+ }
+ // Look up the public key
+ boolean pub = pubKeyCache.get (i.key);
+ if (pub) log ("public key " + i.key + " found in cache");
+ else log ("public key " + i.key + " not found in cache");
+ // Accept the search
+ if (prev != null) {
+ log ("accepting " + i);
+ prev.sendMessage (new SskAccepted (i.id, !pub));
+ }
+ // Store the insert handler and possibly wait for the pub key
+ SskInsertHandler ih = new SskInsertHandler (i,this,prev,!pub);
+ messageHandlers.put (i.id, ih);
+ ih.start();
+ }
+
+ public void removeMessageHandler (int id)
+ {
+ MessageHandler mh = messageHandlers.remove (id);
+ if (mh == null) log ("no message handler to remove for " + id);
+ else log ("removing message handler for " + id);
+ }
+
+ // Return the list of peers in a random order
+ public ArrayList<Peer> peers()
+ {
+ ArrayList<Peer> copy = new ArrayList<Peer> (peers.values());
+ Collections.shuffle (copy);
+ return copy;
+ }
+
+ public void log (String message)
+ {
+ Event.log (net.address + " " + message);
+ }
+
+ // Event callbacks
+
+ private void generateChkRequest (int key)
+ {
+ ChkRequest cr = new ChkRequest (key, location);
+ log ("generating " + cr);
+ handleChkRequest (cr, null);
+ }
+
+ private void generateChkInsert (int key)
+ {
+ ChkInsert ci = new ChkInsert (key, location);
+ log ("generating " + ci);
+ handleChkInsert (ci, null);
+ handleMessage (new DataInsert (ci.id), null);
+ for (int i = 0; i < 32; i++)
+ handleMessage (new Block (ci.id, i), null);
+ }
+
+ private void generateSskRequest (int key)
+ {
+ SskRequest sr = new SskRequest (key, location, true);
+ log ("generating " + sr);
+ handleSskRequest (sr, null);
+ }
+
+ private void generateSskInsert (int key, int value)
+ {
+ SskInsert si = new SskInsert (key, value, location);
+ log ("generating " + si);
+ pubKeyCache.put (key);
+ handleSskInsert (si, null);
+ }
+
+ private void checkTimeouts()
+ {
+ // Check the peers in a random order each time
+ double deadline = Double.POSITIVE_INFINITY;
+ for (Peer p : peers())
+ deadline = Math.min (deadline, p.checkTimeouts());
+ if (deadline == Double.POSITIVE_INFINITY) {
+ // log ("stopping retransmission/coalescing timer");
+ timerRunning = false;
+ }
+ else {
+ double sleep = Math.max (deadline - Event.time(), 0.0);
+ // log ("sleeping for " + sleep + " seconds");
+ Event.schedule (this, sleep, CHECK_TIMEOUTS, null);
+ }
+ }
+
+ // EventTarget interface
+ public void handleEvent (int type, Object data)
+ {
+ switch (type) {
+ case GENERATE_CHK_REQUEST:
+ generateChkRequest ((Integer) data);
+ break;
+
+ case GENERATE_CHK_INSERT:
+ generateChkInsert ((Integer) data);
+ break;
+
+ case GENERATE_SSK_REQUEST:
+ generateSskRequest ((Integer) data);
+ break;
+
+ case GENERATE_SSK_INSERT:
+ generateSskInsert ((Integer) data, 0);
+ break;
+
+ case GENERATE_SSK_COLLISION:
+ generateSskInsert ((Integer) data, 1);
+
+ case CHECK_TIMEOUTS:
+ checkTimeouts();
+ break;
+ }
+ }
+
+ // Each EventTarget class has its own event codes
+ public final static int GENERATE_CHK_REQUEST = 1;
+ public final static int GENERATE_CHK_INSERT = 2;
+ public final static int GENERATE_SSK_REQUEST = 3;
+ public final static int GENERATE_SSK_INSERT = 4;
+ public final static int GENERATE_SSK_COLLISION = 5;
+ private final static int CHECK_TIMEOUTS = 6;
+}
Copied: trunk/apps/load-balancing-sims/phase7/sim/Packet.java (from rev 10750,
trunk/apps/load-balancing-sims/phase7/Packet.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/Packet.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/Packet.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -0,0 +1,42 @@
+// A low-level packet (as opposed to a high-level message)
+
+package sim;
+import sim.messages.Ack;
+import sim.messages.Message;
+import java.util.ArrayList;
+
+class Packet
+{
+ public final static int HEADER_SIZE = 70; // Including IP & UDP headers
+ public final static int ACK_SIZE = 4; // Size of an ack in bytes
+ public final static int MAX_SIZE = 1450; // MTU including headers
+ public final static int SENSIBLE_PAYLOAD = 1000; // Coalescing
+
+ public int src, dest; // Network addresses
+ public int size = HEADER_SIZE; // Size in bytes, including headers
+ public int seq = -1; // Data sequence number (-1 if no data)
+ public ArrayList<Ack> acks = null;
+ public ArrayList<Message> messages = null;
+
+ public double sent; // Time at which the packet was (re) transmitted
+ public double latency; // Link latency (stored here for convenience)
+
+ public void addAck (Ack a)
+ {
+ if (acks == null) acks = new ArrayList<Ack>();
+ acks.add (a);
+ size += a.size();
+ }
+
+ public void addMessage (Message m)
+ {
+ if (messages == null) messages = new ArrayList<Message>();
+ messages.add (m);
+ size += m.size();
+ }
+
+ public String toString()
+ {
+ return new String ("packet " + src + ":" + dest + ":" + seq);
+ }
+}
Copied: trunk/apps/load-balancing-sims/phase7/sim/Peer.java (from rev 10750,
trunk/apps/load-balancing-sims/phase7/Peer.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/Peer.java 2006-10-31 11:14:10 UTC
(rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/Peer.java 2006-10-31 11:35:46 UTC
(rev 10751)
@@ -0,0 +1,320 @@
+package sim;
+import sim.messages.*;
+import java.util.LinkedList;
+import java.util.Iterator;
+import java.util.HashSet;
+
+public class Peer
+{
+ private Node node; // The local node
+ public int address; // The remote node's address
+ public double location; // The remote node's routing location
+ private double latency; // The latency of the connection in seconds
+
+ // Retransmission parameters
+ public final static double RTO = 4.0; // Retransmission timeout in RTTs
+ public final static double FRTO = 1.5; // Fast retx timeout in RTTs
+ public final static double RTT_DECAY = 0.9; // Exp moving average
+ public final static double LINK_IDLE = 8.0; // RTTs without transmitting
+
+ // Coalescing
+ public final static double MIN_SLEEP = 0.01; // Poll the b/w limiter
+ public final static double MAX_DELAY = 0.1; // Coalescing delay in secs
+
+ // Out-of-order delivery with eventual detection of missing packets
+ public final static int SEQ_RANGE = 1000;
+
+ // Sender state
+ private double rtt = 5.0; // Estimated round-trip time in seconds
+ private int txSeq = 0; // Sequence number of next outgoing data packet
+ private int txMaxSeq = SEQ_RANGE - 1; // Highest sequence number
+ private LinkedList<Packet> txBuffer; // Retransmission buffer
+ private DeadlineQueue<Ack> ackQueue; // Outgoing acks
+ private DeadlineQueue<Message> searchQueue; // Outgoing search messages
+ private DeadlineQueue<Message> transferQueue; // Outgoing transfers
+ private CongestionWindow window; // AIMD congestion window
+ private double lastTransmission = 0.0; // Clock time
+ private boolean tgif = false; // "Transfers go in first" toggle
+
+ // Receiver state
+ private HashSet<Integer> rxDupe; // Detect duplicates by sequence number
+ private int rxSeq = 0; // Sequence number of next in-order incoming pkt
+
+ public Peer (Node node, int address, double location, double latency)
+ {
+ this.node = node;
+ this.address = address;
+ this.location = location;
+ this.latency = latency;
+ txBuffer = new LinkedList<Packet>();
+ ackQueue = new DeadlineQueue<Ack>();
+ searchQueue = new DeadlineQueue<Message>();
+ transferQueue = new DeadlineQueue<Message>();
+ window = new CongestionWindow (this);
+ rxDupe = new HashSet<Integer>();
+ }
+
+ // Queue a message for transmission
+ public void sendMessage (Message m)
+ {
+ m.deadline = Event.time() + MAX_DELAY;
+ if (m instanceof Block || m instanceof DataInsert
+ || m instanceof ChkDataFound) {
+ log (m + " added to transfer queue");
+ transferQueue.add (m);
+ }
+ else {
+ log (m + " added to search queue");
+ searchQueue.add (m);
+ }
+ // Start the node's timer if necessary
+ node.startTimer();
+ // Send as many packets as possible
+ while (send());
+ }
+
+ // Queue an ack for transmission
+ private void sendAck (int seq)
+ {
+ log ("ack " + seq + " added to ack queue");
+ ackQueue.add (new Ack (seq, Event.time() + MAX_DELAY));
+ // Start the node's timer if necessary
+ node.startTimer();
+ // Send as many packets as possible
+ while (send());
+ }
+
+ // Try to send a packet, return true if a packet was sent
+ private boolean send()
+ {
+ if (ackQueue.size + searchQueue.size + transferQueue.size == 0){
+ log ("nothing to send");
+ return false;
+ }
+ log (ackQueue.size + " bytes of acks in queue");
+ log (searchQueue.size + " bytes of searches in queue");
+ log (transferQueue.size + " bytes of transfers in queue");
+
+ // Return to slow start when the link is idle
+ double now = Event.time();
+ if (now - lastTransmission > LINK_IDLE * rtt) window.reset();
+ lastTransmission = now;
+
+ // Delay small packets for coalescing
+ if (now < deadline (now)) {
+ int payload = searchQueue.size + transferQueue.size;
+ log ("delaying transmission of " + payload + " bytes");
+ return false;
+ }
+
+ Packet p = new Packet();
+
+ // Put all waiting acks in the packet
+ while (ackQueue.size > 0) p.addAck (ackQueue.pop());
+
+ // Don't send sequence number n+SEQ_RANGE until sequence
+ // number n has been acked - this limits the number of
+ // sequence numbers the receiver must store for replay
+ // detection. We must still be allowed to send acks,
+ // otherwise the connection could deadlock.
+
+ if (txSeq > txMaxSeq)
+ log ("waiting for ack " + (txMaxSeq - SEQ_RANGE + 1));
+ else if (window.available() <= 0)
+ log ("no room in congestion window for messages");
+ else if (node.bandwidth.available() <= 0)
+ log ("no bandwidth available for messages");
+ else pack (p); // OK to send data
+
+ // Don't send empty packets
+ if (p.acks == null && p.messages == null) return false;
+
+ // If the packet contains data, buffer it for retransmission
+ if (p.messages != null) {
+ p.seq = txSeq++;
+ p.sent = now;
+ txBuffer.add (p);
+ window.bytesSent (p.size);
+ }
+
+ // Send the packet
+ log ("sending packet " + p.seq + ", " + p.size + " bytes");
+ node.net.send (p, address, latency);
+ node.bandwidth.remove (p.size);
+ return true;
+ }
+
+ // Called by Node when a packet arrives
+ public void handlePacket (Packet p)
+ {
+ if (p.acks != null) for (Ack a : p.acks) handleAck (a);
+ if (p.messages != null) handleData (p);
+ }
+
+ private void handleData (Packet p)
+ {
+ log ("received packet " + p.seq + ", " + p.size + " bytes");
+ if (p.seq < rxSeq || rxDupe.contains (p.seq)) {
+ log ("duplicate packet");
+ sendAck (p.seq); // Original ack may have been lost
+ }
+ else if (p.seq == rxSeq) {
+ log ("packet in order");
+ // Find the sequence number of the next missing packet
+ int was = rxSeq;
+ while (rxDupe.remove (++rxSeq));
+ log ("rxSeq was " + was + ", now " + rxSeq);
+ // Deliver the packet
+ unpack (p);
+ sendAck (p.seq);
+ }
+ else if (p.seq < rxSeq + SEQ_RANGE) {
+ log ("packet out of order - expected " + rxSeq);
+ if (rxDupe.add (p.seq)) unpack (p);
+ else log ("duplicate packet");
+ sendAck (p.seq); // Original ack may have been lost
+ }
+ // This indicates a misbehaving sender - discard the packet
+ else log ("warning: received " + p.seq + " before " + rxSeq);
+ }
+
+ private void handleAck (Ack a)
+ {
+ int seq = a.id;
+ log ("received ack " + seq);
+ double now = Event.time();
+ Iterator<Packet> i = txBuffer.iterator();
+ while (i.hasNext()) {
+ Packet p = i.next();
+ double age = now - p.sent;
+ // Explicit ack
+ if (p.seq == seq) {
+ log ("packet " + p.seq + " acknowledged");
+ i.remove();
+ // Update the congestion window
+ window.bytesAcked (p.size);
+ // Update the average round-trip time
+ rtt = rtt * RTT_DECAY + age * (1.0 - RTT_DECAY);
+ log ("round-trip time " + age);
+ log ("average round-trip time " + rtt);
+ break;
+ }
+ // Fast retransmission
+ if (p.seq < seq && age > FRTO * rtt + MAX_DELAY) {
+ p.sent = now;
+ log ("fast retransmitting packet " + p.seq);
+ node.net.send (p, address, latency);
+ window.fastRetransmission (now);
+ }
+ }
+ // Recalculate the maximum sequence number
+ if (txBuffer.isEmpty()) txMaxSeq = txSeq + SEQ_RANGE - 1;
+ else txMaxSeq = txBuffer.peek().seq + SEQ_RANGE - 1;
+ log ("maximum sequence number " + txMaxSeq);
+ // Send as many packets as possible
+ while (send());
+ }
+
+ // Add messages to a packet
+ private void pack (Packet p)
+ {
+ // Alternate between giving searches and transfers priority
+ if (tgif) {
+ // Transfers go in first
+ while (transferQueue.size > 0
+ && p.size + transferQueue.headSize() <= Packet.MAX_SIZE)
+ p.addMessage (transferQueue.pop());
+ // Fill any remaining space with searches
+ while (searchQueue.size > 0
+ && p.size + searchQueue.headSize() <= Packet.MAX_SIZE)
+ p.addMessage (searchQueue.pop());
+ tgif = false;
+ }
+ else {
+ // Searches go in first
+ while (searchQueue.size > 0
+ && p.size + searchQueue.headSize() <= Packet.MAX_SIZE)
+ p.addMessage (searchQueue.pop());
+ // Fill any remaining space with transfers
+ while (transferQueue.size > 0
+ && p.size + transferQueue.headSize() <= Packet.MAX_SIZE)
+ p.addMessage (transferQueue.pop());
+ tgif = true;
+ }
+ }
+
+ // Remove messages from a packet and deliver them to the node
+ private void unpack (Packet p)
+ {
+ if (p.messages == null) return;
+ for (Message m : p.messages) node.handleMessage (m, this);
+ }
+
+ // Called by Node, returns the next coalescing or retx deadline
+ public double checkTimeouts()
+ {
+ log ("checking timeouts");
+ // Send as many packets as possible
+ while (send());
+
+ log (txBuffer.size() + " packets in flight");
+ double now = Event.time();
+ if (txBuffer.isEmpty()) return deadline (now);
+
+ for (Packet p : txBuffer) {
+ if (now - p.sent > RTO * rtt + MAX_DELAY) {
+ // Retransmission timeout
+ log ("retransmitting packet " + p.seq);
+ p.sent = now;
+ node.net.send (p, address, latency);
+ window.timeout (now);
+ }
+ }
+
+ // Sleep for up to MAX_DELAY seconds until the next deadline
+ return Math.min (now + MAX_DELAY, deadline (now));
+ }
+
+ // Work out when the first ack or search or transfer needs to be sent
+ private double deadline (double now)
+ {
+ return Math.min (ackQueue.deadline(), dataDeadline (now));
+ }
+
+ // Work out when the first search or transfer needs to be sent
+ private double dataDeadline (double now)
+ {
+ // If there's no data waiting, use the ack deadline
+ if (searchQueue.size + transferQueue.size == 0)
+ return Double.POSITIVE_INFINITY;
+
+ double deadline = Math.min (searchQueue.deadline(),
+ transferQueue.deadline());
+
+ // Delay small packets until the coalescing deadline
+ if (searchQueue.size + transferQueue.size
+ < Packet.SENSIBLE_PAYLOAD)
+ return deadline;
+
+ // If there's not enough room in the window, wait for an ack
+ if (window.available() <= 0)
+ return Double.POSITIVE_INFINITY;
+
+ // If there's not enough bandwidth, try again shortly
+ if (node.bandwidth.available() <= 0)
+ return Math.max (now + MIN_SLEEP, deadline);
+
+ // Send a packet immediately
+ return now;
+ }
+
+ public void log (String message)
+ {
+ Event.log (node.net.address + ":" + address + " " + message);
+ }
+
+ public String toString()
+ {
+ return Integer.toString (address);
+ }
+}
Copied: trunk/apps/load-balancing-sims/phase7/sim/Sim.java (from rev 10750,
trunk/apps/load-balancing-sims/phase7/Sim.java)
Copied: trunk/apps/load-balancing-sims/phase7/sim/TokenBucket.java (from rev
10750, trunk/apps/load-balancing-sims/phase7/TokenBucket.java)
Copied: trunk/apps/load-balancing-sims/phase7/sim/handlers (from rev 10749,
trunk/apps/load-balancing-sims/phase7/handlers)
Copied:
trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkInsertHandler.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/handlers/ChkInsertHandler.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/handlers/ChkInsertHandler.java
2006-10-31 11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkInsertHandler.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -0,0 +1,270 @@
+// The state of a CHK insert as stored at each node along the path
+
+package sim.handlers;
+import sim.*;
+import sim.messages.*;
+import java.util.HashSet;
+
+public class ChkInsertHandler extends MessageHandler implements EventTarget
+{
+ private int inState = STARTED; // State of incoming transfer
+ private HashSet<Peer> receivers; // Peers that should receive data
+ private Block[] blocks; // Store incoming blocks for forwarding
+ private int blocksReceived = 0;
+
+ public ChkInsertHandler (ChkInsert i, Node node, Peer prev)
+ {
+ super (i, node, prev);
+ receivers = new HashSet<Peer>();
+ blocks = new Block[32];
+ }
+
+ public void start()
+ {
+ // Wait 10 seconds for the incoming transfer to start
+ Event.schedule (this, 10.0, DATA_TIMEOUT, null);
+ }
+
+ public void handleMessage (Message m, Peer src)
+ {
+ if (src == prev) {
+ if (m instanceof DataInsert)
+ handleDataInsert ((DataInsert) m);
+ else if (m instanceof Block)
+ handleBlock ((Block) m);
+ else node.log ("unexpected type for " + m);
+ }
+ else if (src == next) {
+ if (m instanceof Accepted)
+ handleAccepted ((Accepted) m);
+ else if (m instanceof RejectedLoop)
+ handleRejectedLoop ((RejectedLoop) m);
+ else if (m instanceof RouteNotFound)
+ handleRouteNotFound ((RouteNotFound) m);
+ else if (m instanceof InsertReply)
+ handleInsertReply ((InsertReply) m);
+ else if (m instanceof TransfersCompleted)
+ handleCompleted ((TransfersCompleted) m, src);
+ else node.log ("unexpected type for " + m);
+ }
+ else if (receivers.contains (src)) {
+ if (m instanceof TransfersCompleted)
+ handleCompleted ((TransfersCompleted) m, src);
+ else node.log ("unexpected type for " + m);
+ }
+ else node.log ("unexpected source for " + m);
+ }
+
+ private void handleDataInsert (DataInsert di)
+ {
+ if (inState != STARTED) node.log (di + " out of order");
+ inState = TRANSFERRING;
+ // Start the search
+ forwardSearch();
+ // Wait for transfer to complete (FIXME: check real timeout)
+ Event.schedule (this, 120.0, TRANSFER_IN_TIMEOUT, null);
+ }
+
+ private void handleBlock (Block b)
+ {
+ if (inState != TRANSFERRING) node.log (b + " out of order");
+ if (blocks[b.index] != null) return; // Ignore duplicates
+ blocks[b.index] = b;
+ blocksReceived++;
+ // Forward the block to all receivers
+ for (Peer p : receivers) p.sendMessage (b);
+ // If the transfer is complete, consider finishing
+ if (blocksReceived == 32) {
+ inState = COMPLETED;
+ considerFinishing();
+ }
+ }
+
+ private void handleCompleted (TransfersCompleted tc, Peer src)
+ {
+ receivers.remove (src);
+ considerFinishing();
+ }
+
+ private void handleAccepted (Accepted a)
+ {
+ if (searchState != SENT) node.log (a + " out of order");
+ searchState = ACCEPTED;
+ // Wait 120 seconds for a reply to the search
+ Event.schedule (this, 120.0, SEARCH_TIMEOUT, next);
+ // Add the next hop to the list of receivers
+ receivers.add (next);
+ next.sendMessage (new DataInsert (id));
+ // Send all previously received blocks
+ for (int i = 0; i < 32; i++)
+ if (blocks[i] != null) next.sendMessage (blocks[i]);
+ // Wait for TransfersCompleted (FIXME: check real timeout)
+ Event.schedule (this, 240.0, TRANSFER_OUT_TIMEOUT, next);
+ }
+
+ private void handleRejectedLoop (RejectedLoop rl)
+ {
+ if (searchState != SENT) node.log (rl + " out of order");
+ forwardSearch();
+ }
+
+ private void handleRouteNotFound (RouteNotFound rnf)
+ {
+ if (searchState != ACCEPTED) node.log (rnf + " out of order");
+ if (rnf.htl < htl) htl = rnf.htl;
+ // Use the remaining htl to try another peer
+ forwardSearch();
+ }
+
+ private void handleInsertReply (InsertReply ir)
+ {
+ if (searchState != ACCEPTED) node.log (ir + " out of order");
+ if (prev == null) node.log (this + " succeeded");
+ else prev.sendMessage (ir); // Forward the message
+ searchState = COMPLETED;
+ considerFinishing();
+ }
+
+ public void forwardSearch()
+ {
+ next = null;
+ // If the search has run out of hops, send InsertReply
+ if (htl == 0) {
+ node.log (this + " has no hops left");
+ if (prev == null) node.log (this + " succeeded");
+ else prev.sendMessage (new InsertReply (id));
+ searchState = COMPLETED;
+ considerFinishing();
+ return;
+ }
+ // Forward the search to the closest remaining peer
+ next = closestPeer();
+ if (next == null) {
+ node.log ("route not found for " + this);
+ if (prev == null) node.log (this + " failed");
+ else prev.sendMessage (new RouteNotFound (id, htl));
+ searchState = COMPLETED;
+ considerFinishing();
+ return;
+ }
+ // Decrement the htl if the next node is not the closest so far
+ double target = Node.keyToLocation (key);
+ if (Node.distance (target, next.location)
+ > Node.distance (target, closest))
+ htl = node.decrementHtl (htl);
+ node.log (this + " has htl " + htl);
+ node.log ("forwarding " + this + " to " + next.address);
+ next.sendMessage (makeSearchMessage());
+ nexts.remove (next);
+ searchState = SENT;
+ // Wait 10 seconds for the next hop to accept the search
+ Event.schedule (this, 10.0, ACCEPTED_TIMEOUT, next);
+ }
+
+ private void considerFinishing()
+ {
+ // An insert finishes when the search, the incoming transfer
+ // and all outgoing transfers are complete
+ if (searchState == COMPLETED && inState == COMPLETED
+ && receivers.isEmpty()) finish();
+ }
+
+ private void finish()
+ {
+ inState = COMPLETED;
+ searchState = COMPLETED;
+ node.cacheChk (key);
+ node.storeChk (key);
+ if (prev == null) node.log (this + " completed");
+ else prev.sendMessage (new TransfersCompleted (id));
+ node.removeMessageHandler (id);
+ }
+
+ protected Search makeSearchMessage()
+ {
+ return new ChkInsert (id, key, closest, htl);
+ }
+
+ public String toString()
+ {
+ return new String ("CHK insert (" + id + "," + key + ")");
+ }
+
+ // Event callbacks
+
+ private void acceptedTimeout (Peer p)
+ {
+ if (p != next) return; // We've already moved on to another peer
+ if (searchState != SENT) return;
+ node.log (this + " accepted timeout waiting for " + p);
+ forwardSearch(); // Try another peer
+ }
+
+ private void searchTimeout (Peer p)
+ {
+ if (p != next) return; // We've already moved on to another peer
+ if (searchState != ACCEPTED) return;
+ node.log (this + " search timeout waiting for " + p);
+ if (prev == null) node.log (this + " failed");
+ searchState = COMPLETED;
+ considerFinishing();
+ }
+
+ private void dataTimeout()
+ {
+ if (inState != STARTED) return;
+ node.log (this + " data timeout waiting for " + prev);
+ if (prev == null) node.log (this + " failed");
+ else prev.sendMessage (new TransfersCompleted (id));
+ finish();
+ }
+
+ private void transferInTimeout()
+ {
+ if (inState != TRANSFERRING) return;
+ node.log (this + " transfer timeout receiving from " + prev);
+ if (prev == null) node.log (this + " failed");
+ else prev.sendMessage (new TransfersCompleted (id));
+ finish();
+ }
+
+ private void transferOutTimeout (Peer p)
+ {
+ if (!receivers.remove (p)) return;
+ node.log (this + " transfer timeout sending to " + p);
+ considerFinishing();
+ }
+
+ // EventTarget interface
+ public void handleEvent (int type, Object data)
+ {
+ switch (type) {
+ case ACCEPTED_TIMEOUT:
+ acceptedTimeout ((Peer) data);
+ break;
+
+ case SEARCH_TIMEOUT:
+ searchTimeout ((Peer) data);
+ break;
+
+ case DATA_TIMEOUT:
+ dataTimeout();
+ break;
+
+ case TRANSFER_IN_TIMEOUT:
+ transferInTimeout();
+ break;
+
+ case TRANSFER_OUT_TIMEOUT:
+ transferOutTimeout ((Peer) data);
+ break;
+ }
+ }
+
+ // Each EventTarget class has its own event codes
+ private final static int ACCEPTED_TIMEOUT = 1;
+ private final static int SEARCH_TIMEOUT = 2;
+ private final static int DATA_TIMEOUT = 3;
+ private final static int TRANSFER_IN_TIMEOUT = 4;
+ private final static int TRANSFER_OUT_TIMEOUT = 5;
+}
Copied:
trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkRequestHandler.java (from
rev 10750,
trunk/apps/load-balancing-sims/phase7/handlers/ChkRequestHandler.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/handlers/ChkRequestHandler.java
2006-10-31 11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/ChkRequestHandler.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -0,0 +1,76 @@
+// The state of a CHK request as stored at each node along the path
+
+package sim.handlers;
+import sim.*;
+import sim.messages.*;
+
+public class ChkRequestHandler extends RequestHandler
+{
+ private boolean[] received; // Keep track of received blocks
+ private int blocksReceived = 0;
+
+ public ChkRequestHandler (ChkRequest r, Node node, Peer prev)
+ {
+ super (r, node, prev);
+ received = new boolean[32];
+ }
+
+ public void handleMessage (Message m, Peer src)
+ {
+ if (src != next) {
+ node.log ("unexpected source for " + m);
+ return;
+ }
+ if (m instanceof Accepted)
+ handleAccepted ((Accepted) m);
+ else if (m instanceof RejectedLoop)
+ handleRejectedLoop ((RejectedLoop) m);
+ else if (m instanceof RouteNotFound)
+ handleRouteNotFound ((RouteNotFound) m);
+ else if (m instanceof DataNotFound)
+ handleDataNotFound ((DataNotFound) m);
+ else if (m instanceof ChkDataFound)
+ handleChkDataFound ((ChkDataFound) m);
+ else if (m instanceof Block)
+ handleBlock ((Block) m);
+ else node.log ("unexpected type for " + m);
+ }
+
+ private void handleChkDataFound (ChkDataFound df)
+ {
+ if (searchState != ACCEPTED) node.log (df + " out of order");
+ searchState = TRANSFERRING;
+ if (prev != null) prev.sendMessage (df); // Forward the message
+ // Wait for the transfer to complete (FIXME: check real timeout)
+ Event.schedule (this, 120.0, TRANSFER_TIMEOUT, next);
+ }
+
+ private void handleBlock (Block b)
+ {
+ if (searchState != TRANSFERRING) node.log (b + " out of order");
+ if (received[b.index]) return; // Ignore duplicates
+ received[b.index] = true;
+ blocksReceived++;
+ // Forward the block
+ if (prev != null) {
+ node.log ("forwarding " + b);
+ prev.sendMessage (b);
+ }
+ // If the transfer is complete, cache the data
+ if (blocksReceived == 32) {
+ node.cacheChk (key);
+ if (prev == null) node.log (this + " succeeded");
+ finish();
+ }
+ }
+
+ protected Search makeSearchMessage()
+ {
+ return new ChkRequest (id, key, closest, htl);
+ }
+
+ public String toString()
+ {
+ return new String ("CHK request (" + id + "," + key + ")");
+ }
+}
Copied: trunk/apps/load-balancing-sims/phase7/sim/handlers/MessageHandler.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/handlers/MessageHandler.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/handlers/MessageHandler.java
2006-10-31 11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/MessageHandler.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -0,0 +1,75 @@
+// The state of a search as stored at each node along the path
+
+package sim.handlers;
+import sim.Node;
+import sim.Peer;
+import sim.messages.Search;
+import sim.messages.Message;
+import java.util.LinkedList;
+
+public abstract class MessageHandler
+{
+ // State machine
+ protected final static int STARTED = 0;
+ protected final static int SENT = 1;
+ protected final static int ACCEPTED = 2;
+ protected final static int TRANSFERRING = 3;
+ protected final static int COMPLETED = 4;
+
+ protected final int id; // The unique ID of the request or insert
+ protected final int key; // The target of the search
+ protected double closest; // The closest location seen so far
+ protected int htl; // Hops to live for backtracking
+
+ protected Node node; // The owner of this MessageHandler
+ protected Peer prev; // The previous hop of the search
+ protected Peer next = null; // The (current) next hop of the search
+ protected LinkedList<Peer> nexts; // Candidates for the next hop
+ protected int searchState = STARTED; // The state of the search
+
+ public MessageHandler (Search s, Node node, Peer prev)
+ {
+ id = s.id;
+ key = s.key;
+ closest = s.closest;
+ htl = s.htl;
+ this.node = node;
+ this.prev = prev;
+ nexts = new LinkedList<Peer> (node.peers());
+ nexts.remove (prev);
+ // If this is the closest location seen so far, reset htl
+ double target = Node.keyToLocation (key);
+ if (Node.distance (target, node.location)
+ < Node.distance (target, closest)) {
+ node.log ("resetting htl of " + this); // FIXME
+ closest = node.location;
+ htl = Search.MAX_HTL;
+ }
+ }
+
+ // Remove a peer from the list of candidates for the next hop
+ public void removeNextHop (Peer p)
+ {
+ nexts.remove (p);
+ }
+
+ // Find the closest remaining peer
+ protected Peer closestPeer ()
+ {
+ double keyLoc = Node.keyToLocation (key);
+ double closestDist = Double.POSITIVE_INFINITY;
+ Peer closestPeer = null;
+ for (Peer peer : nexts) {
+ double dist = Node.distance (keyLoc, peer.location);
+ if (dist < closestDist) {
+ closestDist = dist;
+ closestPeer = peer;
+ }
+ }
+ return closestPeer; // Null if the list was empty
+ }
+
+ public abstract void handleMessage (Message m, Peer src);
+
+ protected abstract Search makeSearchMessage();
+}
Copied: trunk/apps/load-balancing-sims/phase7/sim/handlers/RequestHandler.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/handlers/RequestHandler.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/handlers/RequestHandler.java
2006-10-31 11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/RequestHandler.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -0,0 +1,139 @@
+// The parent class of ChkRequestHandler and SskRequestHandler
+
+package sim.handlers;
+import sim.*;
+import sim.messages.*;
+
+public abstract class RequestHandler extends MessageHandler
+ implements EventTarget
+{
+ public RequestHandler (Search s, Node node, Peer prev)
+ {
+ super (s, node, prev);
+ }
+
+ public void start()
+ {
+ forwardSearch();
+ }
+
+ protected void handleAccepted (Accepted a)
+ {
+ if (searchState != SENT) node.log (a + " out of order");
+ searchState = ACCEPTED;
+ // Wait 60 seconds for a reply to the search
+ Event.schedule (this, 60.0, SEARCH_TIMEOUT, next);
+ }
+
+ protected void handleRejectedLoop (RejectedLoop rl)
+ {
+ if (searchState != SENT) node.log (rl + " out of order");
+ forwardSearch();
+ }
+
+ protected void handleRouteNotFound (RouteNotFound rnf)
+ {
+ if (searchState != ACCEPTED) node.log (rnf + " out of order");
+ if (rnf.htl < htl) htl = rnf.htl;
+ // Use the remaining htl to try another peer
+ forwardSearch();
+ }
+
+ protected void handleDataNotFound (DataNotFound dnf)
+ {
+ if (searchState != ACCEPTED) node.log (dnf + " out of order");
+ if (prev == null) node.log (this + " failed");
+ else prev.sendMessage (dnf); // Forward the message
+ finish();
+ }
+
+ protected void forwardSearch()
+ {
+ next = null;
+ // If the search has run out of hops, send DataNotFound
+ if (htl == 0) {
+ node.log ("data not found for " + this);
+ if (prev == null) node.log (this + " failed");
+ else prev.sendMessage (new DataNotFound (id));
+ finish();
+ return;
+ }
+ // Forward the search to the closest remaining peer
+ next = closestPeer();
+ if (next == null) {
+ node.log ("route not found for " + this);
+ if (prev == null) node.log (this + " failed");
+ else prev.sendMessage (new RouteNotFound (id, htl));
+ finish();
+ return;
+ }
+ // Decrement the htl if the next node is not the closest so far
+ double target = Node.keyToLocation (key);
+ if (Node.distance (target, next.location)
+ > Node.distance (target, closest))
+ htl = node.decrementHtl (htl);
+ node.log (this + " has htl " + htl);
+ node.log ("forwarding " + this + " to " + next.address);
+ next.sendMessage (makeSearchMessage());
+ nexts.remove (next);
+ searchState = SENT;
+ // Wait 5 seconds for the next hop to accept the search
+ Event.schedule (this, 5.0, ACCEPTED_TIMEOUT, next);
+ }
+
+ protected void finish()
+ {
+ searchState = COMPLETED;
+ node.removeMessageHandler (id);
+ }
+
+ // Event callbacks
+
+ protected void acceptedTimeout (Peer p)
+ {
+ if (p != next) return; // We've already moved on to another peer
+ if (searchState != SENT) return;
+ node.log (this + " accepted timeout waiting for " + p);
+ forwardSearch(); // Try another peer
+ }
+
+ protected void searchTimeout (Peer p)
+ {
+ if (p != next) return; // We've already moved on to another peer
+ if (searchState != ACCEPTED) return;
+ node.log (this + " search timeout waiting for " + p);
+ if (prev == null) node.log (this + " failed");
+ finish();
+ }
+
+ protected void transferTimeout (Peer p)
+ {
+ if (searchState != TRANSFERRING) return;
+ node.log (this + " transfer timeout receiving from " + p);
+ if (prev == null) node.log (this + " failed");
+ finish();
+ }
+
+ // EventTarget interface
+ public void handleEvent (int type, Object data)
+ {
+ switch (type) {
+ case ACCEPTED_TIMEOUT:
+ acceptedTimeout ((Peer) data);
+ break;
+
+ case SEARCH_TIMEOUT:
+ searchTimeout ((Peer) data);
+ break;
+
+ case TRANSFER_TIMEOUT:
+ transferTimeout ((Peer) data);
+ break;
+ }
+ }
+
+ // Each EventTarget class has its own event codes
+ protected final static int ACCEPTED_TIMEOUT = 1;
+ protected final static int SEARCH_TIMEOUT = 2;
+ protected final static int TRANSFER_TIMEOUT = 3;
+}
Copied:
trunk/apps/load-balancing-sims/phase7/sim/handlers/SskInsertHandler.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/handlers/SskInsertHandler.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/handlers/SskInsertHandler.java
2006-10-31 11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/SskInsertHandler.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -0,0 +1,220 @@
+// The state of an SSK insert as stored at each node along the path
+
+package sim.handlers;
+import sim.*;
+import sim.messages.*;
+import java.util.HashSet;
+
+public class SskInsertHandler extends MessageHandler implements EventTarget
+{
+ private int searchState = STARTED; // searchState of search
+ private SskPubKey pubKey = null;
+ private int data; // The data being inserted
+
+ public SskInsertHandler (SskInsert i, Node node,
+ Peer prev, boolean needPubKey)
+ {
+ super (i, node, prev);
+ data = i.data;
+ if (!needPubKey) pubKey = new SskPubKey (id, key);
+ }
+
+ public void start()
+ {
+ if (pubKey == null) {
+ // Wait 10 seconds for the previous hop to send the key
+ Event.schedule (this, 10.0, KEY_TIMEOUT, null);
+ }
+ else {
+ checkCollision();
+ forwardSearch();
+ }
+ }
+
+ // Check whether an older version of the data is already stored
+ private void checkCollision()
+ {
+ Integer old = node.fetchSsk (key);
+ if (old != null && old != data) {
+ node.log (this + " collided");
+ if (prev == null) node.log (this + " collided locally");
+ else prev.sendMessage (new SskDataFound (id, old));
+ // Continue inserting the old data
+ data = old;
+ return;
+ }
+ }
+
+ public void handleMessage (Message m, Peer src)
+ {
+ if (src == prev) {
+ if (m instanceof SskPubKey)
+ handleSskPubKey ((SskPubKey) m);
+ else node.log ("unexpected type for " + m);
+ }
+ else if (src == next) {
+ if (m instanceof SskAccepted)
+ handleSskAccepted ((SskAccepted) m);
+ else if (m instanceof RejectedLoop)
+ handleRejectedLoop ((RejectedLoop) m);
+ else if (m instanceof RouteNotFound)
+ handleRouteNotFound ((RouteNotFound) m);
+ else if (m instanceof SskDataFound)
+ handleCollision ((SskDataFound) m);
+ else if (m instanceof InsertReply)
+ handleInsertReply ((InsertReply) m);
+ else node.log ("unexpected type for " + m);
+ }
+ else node.log ("unexpected source for " + m);
+ }
+
+ private void handleSskPubKey (SskPubKey pk)
+ {
+ if (searchState != STARTED) node.log (pk + " out of order");
+ pubKey = pk;
+ checkCollision();
+ forwardSearch();
+ }
+
+ private void handleSskAccepted (SskAccepted sa)
+ {
+ if (searchState != SENT) node.log (sa + " out of order");
+ searchState = ACCEPTED;
+ // Wait 60 seconds for a reply to the search
+ Event.schedule (this, 60.0, SEARCH_TIMEOUT, next);
+ // Send the public key if requested
+ if (sa.needPubKey) next.sendMessage (pubKey);
+ }
+
+ private void handleRejectedLoop (RejectedLoop rl)
+ {
+ if (searchState != SENT) node.log (rl + " out of order");
+ forwardSearch();
+ }
+
+ private void handleRouteNotFound (RouteNotFound rnf)
+ {
+ if (searchState != ACCEPTED) node.log (rnf + " out of order");
+ if (rnf.htl < htl) htl = rnf.htl;
+ // Use the remaining htl to try another peer
+ forwardSearch();
+ }
+
+ private void handleCollision (SskDataFound sdf)
+ {
+ if (searchState != ACCEPTED) node.log (sdf + " out of order");
+ if (prev == null) node.log (this + " collided");
+ else prev.sendMessage (sdf); // Forward the message
+ data = sdf.data; // Is this safe?
+ }
+
+ private void handleInsertReply (InsertReply ir)
+ {
+ if (searchState != ACCEPTED) node.log (ir + " out of order");
+ if (prev == null) node.log (this + " succeeded");
+ else prev.sendMessage (ir); // Forward the message
+ finish();
+ }
+
+ public void forwardSearch()
+ {
+ next = null;
+ // If the search has run out of hops, send InsertReply
+ if (htl == 0) {
+ node.log (this + " has no hops left");
+ if (prev == null) node.log (this + " succeeded");
+ else prev.sendMessage (new InsertReply (id));
+ finish();
+ return;
+ }
+ // Forward the search to the closest remaining peer
+ next = closestPeer();
+ if (next == null) {
+ node.log ("route not found for " + this);
+ if (prev == null) node.log (this + " failed");
+ else prev.sendMessage (new RouteNotFound (id, htl));
+ finish();
+ return;
+ }
+ // Decrement the htl if the next node is not the closest so far
+ double target = Node.keyToLocation (key);
+ if (Node.distance (target, next.location)
+ > Node.distance (target, closest))
+ htl = node.decrementHtl (htl);
+ node.log (this + " has htl " + htl);
+ node.log ("forwarding " + this + " to " + next.address);
+ next.sendMessage (makeSearchMessage());
+ nexts.remove (next);
+ searchState = SENT;
+ // Wait 10 seconds for the next hop to accept the search
+ Event.schedule (this, 10.0, ACCEPTED_TIMEOUT, next);
+ }
+
+ private void finish()
+ {
+ searchState = COMPLETED;
+ node.cachePubKey (key);
+ node.cacheSsk (key, data);
+ node.storeSsk (key, data);
+ node.removeMessageHandler (id);
+ }
+
+ protected Search makeSearchMessage()
+ {
+ return new SskInsert (id, key, data, closest, htl);
+ }
+
+ public String toString()
+ {
+ return new String ("SSK insert (" +id+ "," +key+ "," +data+")");
+ }
+
+ // Event callbacks
+
+ private void keyTimeout()
+ {
+ if (searchState != STARTED) return;
+ node.log (this + " key timeout waiting for " + prev);
+ finish();
+ }
+
+ private void acceptedTimeout (Peer p)
+ {
+ if (p != next) return; // We've already moved on to another peer
+ if (searchState != SENT) return;
+ node.log (this + " accepted timeout waiting for " + p);
+ forwardSearch(); // Try another peer
+ }
+
+ private void searchTimeout (Peer p)
+ {
+ if (p != next) return; // We've already moved on to another peer
+ if (searchState != ACCEPTED) return;
+ node.log (this + " search timeout waiting for " + p);
+ if (prev == null) node.log (this + " failed");
+ finish();
+ }
+
+ // EventTarget interface
+ public void handleEvent (int type, Object data)
+ {
+ switch (type) {
+ case KEY_TIMEOUT:
+ keyTimeout();
+ break;
+
+ case ACCEPTED_TIMEOUT:
+ acceptedTimeout ((Peer) data);
+ break;
+
+ case SEARCH_TIMEOUT:
+ searchTimeout ((Peer) data);
+ break;
+ }
+ }
+
+ // Each EventTarget class has its own event codes
+ private final static int KEY_TIMEOUT = 1;
+ private final static int ACCEPTED_TIMEOUT = 2;
+ private final static int SEARCH_TIMEOUT = 3;
+}
Copied:
trunk/apps/load-balancing-sims/phase7/sim/handlers/SskRequestHandler.java (from
rev 10750,
trunk/apps/load-balancing-sims/phase7/handlers/SskRequestHandler.java)
===================================================================
--- trunk/apps/load-balancing-sims/phase7/handlers/SskRequestHandler.java
2006-10-31 11:14:10 UTC (rev 10750)
+++ trunk/apps/load-balancing-sims/phase7/sim/handlers/SskRequestHandler.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -0,0 +1,81 @@
+// The state of an SSK request as stored at each node along the path
+
+package sim.handlers;
+import sim.*;
+import sim.messages.*;
+
+public class SskRequestHandler extends RequestHandler
+{
+ private boolean needPubKey; // Ask the next hop for the public key?
+ private SskPubKey pubKey = null;
+ private SskDataFound dataFound = null;
+
+ public SskRequestHandler (SskRequest r, Node node,
+ Peer prev, boolean needPubKey)
+ {
+ super (r, node, prev);
+ this.needPubKey = needPubKey;
+ if (!needPubKey) pubKey = new SskPubKey (id, key);
+ }
+
+ public void handleMessage (Message m, Peer src)
+ {
+ if (src != next) {
+ node.log ("unexpected source for " + m);
+ return;
+ }
+ if (m instanceof Accepted)
+ handleAccepted ((Accepted) m);
+ else if (m instanceof RejectedLoop)
+ handleRejectedLoop ((RejectedLoop) m);
+ else if (m instanceof RouteNotFound)
+ handleRouteNotFound ((RouteNotFound) m);
+ else if (m instanceof DataNotFound)
+ handleDataNotFound ((DataNotFound) m);
+ else if (m instanceof SskDataFound)
+ handleSskDataFound ((SskDataFound) m);
+ else if (m instanceof SskPubKey)
+ handleSskPubKey ((SskPubKey) m);
+ else node.log ("unexpected type for " + m);
+ }
+
+ private void handleSskDataFound (SskDataFound df)
+ {
+ if (searchState != ACCEPTED) node.log (df + " out of order");
+ dataFound = df;
+ if (pubKey == null) return; // Keep waiting
+ if (prev == null) node.log (this + " succeeded");
+ else {
+ prev.sendMessage (dataFound);
+ if (needPubKey) prev.sendMessage (pubKey);
+ }
+ node.cachePubKey (key);
+ node.cacheSsk (key, dataFound.data);
+ finish();
+ }
+
+ private void handleSskPubKey (SskPubKey pk)
+ {
+ if (searchState != ACCEPTED) node.log (pk + " out of order");
+ pubKey = pk;
+ if (dataFound == null) return; // Keep waiting
+ if (prev == null) node.log (this + " succeeded");
+ else {
+ prev.sendMessage (dataFound);
+ if (needPubKey) prev.sendMessage (pubKey);
+ }
+ node.cachePubKey (key);
+ node.cacheSsk (key, dataFound.data);
+ finish();
+ }
+
+ protected Search makeSearchMessage()
+ {
+ return new SskRequest (id, key, closest, htl, pubKey == null);
+ }
+
+ public String toString()
+ {
+ return new String ("SSK request (" + id + "," + key + ")");
+ }
+}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages (from rev 10749,
trunk/apps/load-balancing-sims/phase7/messages)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/Accepted.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/Accepted.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/Accepted.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,14 +0,0 @@
-package messages;
-
-public class Accepted extends Message
-{
- public Accepted (int id)
- {
- this.id = id;
- }
-
- public String toString()
- {
- return new String ("accepted (" + id + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/Accepted.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/messages/Accepted.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/Ack.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/Ack.java 2006-10-31
11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/Ack.java 2006-10-31
11:35:46 UTC (rev 10751)
@@ -1,15 +0,0 @@
-package messages;
-
-public class Ack extends Message
-{
- public Ack (int seq, double deadline)
- {
- id = seq; // Space-saving hack
- this.deadline = deadline;
- }
-
- public int size()
- {
- return ACK_SIZE;
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/Ack.java (from rev
10750, trunk/apps/load-balancing-sims/phase7/messages/Ack.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/Block.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/Block.java 2006-10-31
11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/Block.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,24 +0,0 @@
-// A single block of a multi-block transfer (currently 32 blocks per transfer)
-
-package messages;
-
-public class Block extends Message
-{
- public final int index; // Index of this block from 0-31
-
- public Block (int id, int index)
- {
- this.id = id;
- this.index = index;
- }
-
- public int size()
- {
- return HEADER_SIZE + DATA_SIZE;
- }
-
- public String toString()
- {
- return new String ("block (" + id + "," + index + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/Block.java (from rev
10750, trunk/apps/load-balancing-sims/phase7/messages/Block.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/ChkDataFound.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/ChkDataFound.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/ChkDataFound.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,14 +0,0 @@
-package messages;
-
-public class ChkDataFound extends Message
-{
- public ChkDataFound (int id)
- {
- this.id = id;
- }
-
- public String toString()
- {
- return new String ("CHK data found (" + id + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/ChkDataFound.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/messages/ChkDataFound.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/ChkInsert.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/ChkInsert.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/ChkInsert.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,21 +0,0 @@
-package messages;
-
-public class ChkInsert extends Search
-{
- // Start a new insert
- public ChkInsert (int key, double location)
- {
- super (key, location);
- }
-
- // Forward an insert
- public ChkInsert (int id, int key, double closest, int htl)
- {
- super (id, key, closest, htl);
- }
-
- public String toString()
- {
- return new String ("CHK insert (" + id + "," + key + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/ChkInsert.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/messages/ChkInsert.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/ChkRequest.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/ChkRequest.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/ChkRequest.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,21 +0,0 @@
-package messages;
-
-public class ChkRequest extends Search
-{
- // Start a new request
- public ChkRequest (int key, double location)
- {
- super (key, location);
- }
-
- // Forward a request
- public ChkRequest (int id, int key, double closest, int htl)
- {
- super (id, key, closest, htl);
- }
-
- public String toString()
- {
- return new String ("CHK request (" + id + "," + key + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/ChkRequest.java
(from rev 10750, trunk/apps/load-balancing-sims/phase7/messages/ChkRequest.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/DataInsert.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/DataInsert.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/DataInsert.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,14 +0,0 @@
-package messages;
-
-public class DataInsert extends Message
-{
- public DataInsert (int id)
- {
- this.id = id;
- }
-
- public String toString()
- {
- return new String ("data insert (" + id + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/DataInsert.java
(from rev 10750, trunk/apps/load-balancing-sims/phase7/messages/DataInsert.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/DataNotFound.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/DataNotFound.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/DataNotFound.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,14 +0,0 @@
-package messages;
-
-public class DataNotFound extends Message
-{
- public DataNotFound (int id)
- {
- this.id = id;
- }
-
- public String toString()
- {
- return new String ("data not found (" + id + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/DataNotFound.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/messages/DataNotFound.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/InsertReply.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/InsertReply.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/InsertReply.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,14 +0,0 @@
-package messages;
-
-public class InsertReply extends Message
-{
- public InsertReply (int id)
- {
- this.id = id;
- }
-
- public String toString()
- {
- return new String ("insert reply (" + id + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/InsertReply.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/messages/InsertReply.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/Message.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/Message.java 2006-10-31
11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/Message.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,23 +0,0 @@
-// A high-level message (as opposed to a low-level packet)
-
-package messages;
-
-public class Message
-{
- public final static int HEADER_SIZE = 12; // Bytes, including unique ID
- public final static int KEY_SIZE = 32; // Size of a routing key, bytes
- public final static int PUB_KEY_SIZE = 1024; // Size of a pub key, bytes
- public final static int DATA_SIZE = 1024; // Size of a data block, bytes
- public final static int ACK_SIZE = 4; // Size of a sequence num, bytes
-
- public static int nextId = 0; // Each request and insert has a unique ID
-
- public int id; // Unique request ID
- public double deadline = 0.0; // Coalescing, stored here for convenience
-
- // Override this
- public int size()
- {
- return HEADER_SIZE;
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/Message.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/messages/Message.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/RejectedLoop.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/RejectedLoop.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/RejectedLoop.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,14 +0,0 @@
-package messages;
-
-public class RejectedLoop extends Message
-{
- public RejectedLoop (int id)
- {
- this.id = id;
- }
-
- public String toString()
- {
- return new String ("rejected loop (" + id + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/RejectedLoop.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/messages/RejectedLoop.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/RouteNotFound.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/RouteNotFound.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/RouteNotFound.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,17 +0,0 @@
-package messages;
-
-public class RouteNotFound extends Message
-{
- public int htl; // Hops to live for backtracking
-
- public RouteNotFound (int id, int htl)
- {
- this.id = id;
- this.htl = htl;
- }
-
- public String toString()
- {
- return new String ("route not found (" + id + "," + htl + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/RouteNotFound.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/messages/RouteNotFound.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/Search.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/Search.java 2006-10-31
11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/Search.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,38 +0,0 @@
-package messages;
-
-public class Search extends Message
-{
- public final static int MAX_HTL = 10; // Maximum amount of backtracking
-
- public final int key; // The target of the search
- public double closest; // The closest location seen so far
- public int htl; // Hops to live for backtracking
-
- // Start a new search
- public Search (int key, double location)
- {
- id = Message.nextId++;
- this.key = key;
- closest = location;
- htl = MAX_HTL;
- }
-
- // Forward a search
- public Search (int id, int key, double closest, int htl)
- {
- this.id = id;
- this.key = key;
- this.closest = closest;
- this.htl = htl;
- }
-
- public int size()
- {
- return HEADER_SIZE + KEY_SIZE;
- }
-
- public String toString()
- {
- return new String ("search (" +id+ "," +key+ "," +htl+ ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/Search.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/messages/Search.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/SskAccepted.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/SskAccepted.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/SskAccepted.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,17 +0,0 @@
-package messages;
-
-public class SskAccepted extends Message
-{
- public final boolean needPubKey;
-
- public SskAccepted (int id, boolean needPubKey)
- {
- this.id = id;
- this.needPubKey = needPubKey;
- }
-
- public String toString()
- {
- return new String ("SSK accepted (" +id+ "," +needPubKey+ ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/SskAccepted.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/messages/SskAccepted.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/SskDataFound.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/SskDataFound.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/SskDataFound.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,22 +0,0 @@
-package messages;
-
-public class SskDataFound extends Message
-{
- public final int data;
-
- public SskDataFound (int id, int data)
- {
- this.id = id;
- this.data = data;
- }
-
- public int size()
- {
- return HEADER_SIZE + DATA_SIZE;
- }
-
- public String toString()
- {
- return new String ("SSK data found (" + id + "," + data + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/SskDataFound.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/messages/SskDataFound.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/SskInsert.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/SskInsert.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/SskInsert.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,30 +0,0 @@
-package messages;
-
-public class SskInsert extends Search
-{
- public final int data;
-
- // Start a new insert
- public SskInsert (int key, int data, double location)
- {
- super (key, location);
- this.data = data;
- }
-
- // Forward an insert
- public SskInsert (int id, int key, int data, double closest, int htl)
- {
- super (id, key, closest, htl);
- this.data = data;
- }
-
- public int size()
- {
- return HEADER_SIZE + KEY_SIZE + DATA_SIZE;
- }
-
- public String toString()
- {
- return new String ("SSK insert (" +id+ "," +key+ "," +data+")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/SskInsert.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/messages/SskInsert.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/SskPubKey.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/SskPubKey.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/SskPubKey.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,22 +0,0 @@
-package messages;
-
-public class SskPubKey extends Message
-{
- public final int key;
-
- public SskPubKey (int id, int key)
- {
- this.id = id;
- this.key = key;
- }
-
- public int size()
- {
- return HEADER_SIZE + PUB_KEY_SIZE;
- }
-
- public String toString()
- {
- return new String ("SSK public key (" + id + "," + key + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/SskPubKey.java (from
rev 10750, trunk/apps/load-balancing-sims/phase7/messages/SskPubKey.java)
Deleted: trunk/apps/load-balancing-sims/phase7/sim/messages/SskRequest.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/SskRequest.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/SskRequest.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,27 +0,0 @@
-package messages;
-
-public class SskRequest extends Search
-{
- public final boolean needPubKey;
-
- // Start a new request
- public SskRequest (int key, double location, boolean needPubKey)
- {
- super (key, location);
- this.needPubKey = needPubKey;
- }
-
- // Forward a request
- public SskRequest (int id, int key, double closest,
- int htl, boolean needPubKey)
- {
- super (id, key, closest, htl);
- this.needPubKey = needPubKey;
- }
-
- public String toString()
- {
- return new String ("SSK request (" + id + "," + key
- + "," + needPubKey + ")");
- }
-}
Copied: trunk/apps/load-balancing-sims/phase7/sim/messages/SskRequest.java
(from rev 10750, trunk/apps/load-balancing-sims/phase7/messages/SskRequest.java)
Deleted:
trunk/apps/load-balancing-sims/phase7/sim/messages/TransfersCompleted.java
===================================================================
--- trunk/apps/load-balancing-sims/phase7/messages/TransfersCompleted.java
2006-10-31 11:01:12 UTC (rev 10749)
+++ trunk/apps/load-balancing-sims/phase7/sim/messages/TransfersCompleted.java
2006-10-31 11:35:46 UTC (rev 10751)
@@ -1,14 +0,0 @@
-package messages;
-
-public class TransfersCompleted extends Message
-{
- public TransfersCompleted (int id)
- {
- this.id = id;
- }
-
- public String toString()
- {
- return new String ("transfers completed (" + id + ")");
- }
-}
Copied:
trunk/apps/load-balancing-sims/phase7/sim/messages/TransfersCompleted.java
(from rev 10750,
trunk/apps/load-balancing-sims/phase7/messages/TransfersCompleted.java)
