sourceroutenode.java

DTNSim2 is a simulator for Delay-Tolerant Networks (DTNs) written in Java. It is based on Sushant Ja

package implementations;

import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;

import simulator.Contact;
import simulator.InputLine;
import simulator.InputReader;
import simulator.Message;
import simulator.Node;
import simulator.Stats;

import util.CommandStatus;
import util.Verbose;

public class SourceRouteNode extends Node
{
	private static final Message.Header MSG_ROUTE_DATA_KEY = new Message.Header();

	public static final int ALGO_MED = 0;

	public static final int ALGO_ED = 1;

	public static final int ALGO_HOP_COUNT = 2;

	protected int nodeAlgo = ALGO_MED;

	protected boolean blockingNode = false;

	protected boolean perContactRouting = false;

	protected boolean allowShortcuts = false;

	protected boolean accurateSendPreds = false;

	protected int usedBufferSize = 0;

	protected int bufferSize = 2000000;

	protected int msgDataLength = 10;

	protected static double simulationTime = -1;

	protected static double autoSimTime = -1;

	protected ArrayList<Message> messageQueue = new ArrayList<Message>();

	private HashSet<Contact> idleContacts = new HashSet<Contact>();

	protected NetworkDijkstra lastRouting = null;

	protected static HashMap<Contact, ContactSchedule> contactSchedules = new HashMap<Contact, ContactSchedule>();

	/**
	 * Hack function to allow this to be used from junit: we need to clear the
	 * contacts between runs. The contactSchedules should be stored in the
	 * Network or the Contact, or in some other class. This would make this
	 * reset() not needed.
	 */
	public static void reset()
	{
		contactSchedules = new HashMap<Contact, ContactSchedule>();
	}

	public SourceRouteNode()
	{
	}

	public SourceRouteNode(SourceRouteNode org)
	{
		super(org);

		usedBufferSize = 0;
		bufferSize = org.bufferSize;
		nodeAlgo = org.nodeAlgo;
		blockingNode = org.blockingNode;
		msgDataLength = org.msgDataLength;
		allowShortcuts = org.allowShortcuts;
		perContactRouting = org.perContactRouting;
		accurateSendPreds = org.accurateSendPreds;
	}

	public void defaultLoad()
	{
		RouteField rf = new RouteField();

		network.stats().registerFieldInterpreter("%msg_route", rf);
		network.stats().registerFieldInterpreter("%msg_route_len", rf);
	}

	public void defaultUnload()
	{
		network.stats().unregisterFieldInterpreter("%msg_route");
		network.stats().unregisterFieldInterpreter("%msg_route_len");
	}

	public void setNodeAlgo(int a)
	{
		assert a == ALGO_ED || a == ALGO_MED || a == ALGO_HOP_COUNT;
		nodeAlgo = a;
	}

	public void setNodeBlocking(boolean b)
	{
		blockingNode = b;
	}

	public boolean willAcceptMessageSize(int len)
	{
		if (bufferSize >= 0 && len + usedBufferSize > bufferSize)
			return false;

		return true;
	}

	/**
	 * Used to iterate over messages available to be sent over the specified
	 * contact. It is called each time a contact becomes idle. For each message,
	 * shouldDeliver will be called until a message returns true, or all the
	 * messages return false. The message that returns true will be sent over
	 * the contact.
	 */
	protected Iterator<Message> messagesForContact(Contact contact)
	{
		return messageQueue.iterator();
	}

	/** Delivers the message if this is the next source routed contact to use. */
	protected boolean shouldDeliver(Message msg, Contact ct)
	{
		// If we find a contact which leads directly to final destination of
		// this message - we should deliver it ignoring computed path!
		if (allowShortcuts && msg.getDestNode() == ct.getDest())
			return true;
		
		if (perContactRouting)
		{
			// Deliver this message if the route includes this contact
			ArrayList<Contact> route = computeRouting(msg, network.getCurrentTime());
			
			// TODO: If route == null, it means the message is no longer deliverable and should get dropped
			// For now we will just ignore it and keep it in the buffer doing nothing
			if (route != null && route.get(0) == ct)
			{
				// Put this route in the message
				msg.storeData(MSG_ROUTE_DATA_KEY, route);
				// Send the message out on this contact
				return true;
			}
			
			return false;
		}

		// ~ System.out.println( "shouldDeliver " + msg + " on " + ct );
		ArrayList<Contact> sourceRoute = (ArrayList<Contact>) msg.getData(MSG_ROUTE_DATA_KEY);
		
		if (sourceRoute.get(0) != ct)
		{
			return false;
		}

		if (blockingNode)
		{
			// Blocking algorithms will check if node they want to send the
			// message to is able to accept that data. (Only if that node is not
			// a final destination of this message)
			if (msg.getDestNode() != ct.getDest()
					&& !((SourceRouteNode) ct.getDest()).willAcceptMessageSize(msg.getLength()))
			{
				return false;
			}
		}

		return true;
	}

	public boolean sendNext(Contact contact)
	{
		super.contactIdle(contact);

		for (Iterator<Message> i = messagesForContact(contact); i.hasNext();)
		{
			Message msg = (Message) i.next();

			if (shouldDeliver(msg, contact))
			{
				// We should deliver this message: go do it

				if (contact.sendMessage(msg))
				{
					// Do not allow this message to be sent by other contacts
					boolean success = messageQueue.remove(msg);
					assert success;

					if (msg.getSourceNode() == this)
					{
						if (nodeAlgo == ALGO_ED && accurateSendPreds)
						{
							computeRouting(msg, network.getCurrentTime());
						}

						assert lastRouting != null;

						msg.storeData(Stats.MSG_SEND_PRED_DELAY, lastRouting.getCost((SourceRouteNode) msg
								.getDestNode()));
						msg.storeData(Stats.MSG_SEND_TIME, new Double(network.getCurrentTime()));
					}

					return true;
				}

				// If we reach this code, it means that we lost that message -
				// it is not in a buffer and there will be no messageSent event!
				assert false;
			}
		}

		if (network.vShouldLog(Verbose.DEBUG4))
			network.vprint("NOP - Nothing left to send in " + this);

		return false;
	}

	public void contactIdle(Contact contact)
	{
		super.contactIdle(contact);

		if (!sendNext(contact))
			idleContacts.add(contact);
	}

	public void contactDown(Contact contact)
	{
		super.contactDown(contact);
		
		// Remove this contact from the idle list, if it was there
		idleContacts.remove(contact);
		
		if (perContactRouting && nodeAlgo == ALGO_MED)
		{
			// Reset the route cache when a link goes down
			lastRouting = null;
		}
	}

	private void checkIdleContacts()
	{
		Iterator<Contact> it = idleContacts.iterator();

		while (it.hasNext())
		{
			Contact c = it.next();

			if (sendNext(c))
				it.remove();
		}
	}

	public void checkContactsToNode(Node dest)
	{
		Iterator<Contact> it = idleContacts.iterator();

		while (it.hasNext())
		{
			Contact c = it.next();

			if (c.getDest() == dest)
			{
				if (sendNext(c))
					it.remove();

				return;
			}
		}
	}

	/**
	 * Decides if a message can be accepted for further processing. The message
	 * has just arrived from the specified contact. This implementation accepts
	 * delivery if there is buffer space.
	 */
	protected boolean acceptMessage(Message msg, Contact source, int size)
	{
		if (size < msg.getLength())
			return false;

		// Verify that this node exists in the source route
		@SuppressWarnings("unchecked")
		ArrayList<Contact> sourceRoute = (ArrayList<Contact>) msg.getData(MSG_ROUTE_DATA_KEY);

		// The next hop should be the same as the last hop
		// ~ System.out.println( sourceRoute.get(0) + " != " + lastHop );
		assert (sourceRoute != null);
		assert (sourceRoute.get(0) == source);
		assert (sourceRoute.get(0).getDest() == this);

		sourceRoute.remove(0);

		// This node should be the source for the next hop
		assert (sourceRoute.get(0).getSource() == this);

		// At this point, we now have to find storage for the message
		// Try to add the message to the forwarding buffer
		if (tryStoreMessage(msg, true))
		{
			// It was added!
			// See if we can send this message out any of the idle contacts
			checkIdleContacts();
			return true;
		}

		statMsgEntry.setup(msg, network.getCurrentTime(), Stats.MSG_DROPPED_OOBUF);
		network.stats().notify(statMsgEntry);

		if (network.vShouldLog(Verbose.ERR))
			network.vprint("DROPPED " + msg + " in " + this + " - out of buffer!");

		return false;
	}

	/**
	 * Returns true when a message was added to the local buffer, false if it
	 * could not be added.
	 */
	protected boolean tryStoreMessage(Message msg, boolean modifyBufferSize)
	{
		if (bufferSize < 0 || msg.getLength() + usedBufferSize <= bufferSize)
		{
			if (modifyBufferSize)
			{
				usedBufferSize += msg.getLength();

				assert bufferSize < 0 || usedBufferSize <= bufferSize;
			}

			messageQueue.add(msg);

			return true;
		}
		else
		{
			assert modifyBufferSize == true;

			return false;
		}
	}

	/**
	 * Called when a contact is done sending a message. If the message was
	 * completely sent, it is not added to the queue again.
	 */
	public void messageSent(Message msg, Contact ct, int size)
	{
		// The message was *not* sent successfully: try to put it back in the
		// queue
		if (size < msg.getLength())
		{
			// We don't want it to modify buffer size. The size for this message
			// is "reserved".
			tryStoreMessage(msg, false);
		}
		else
		{
			// We modify buffer size now!
			usedBufferSize -= msg.getLength();

			assert usedBufferSize >= 0;

			if (blockingNode)
			{
				Iterator<Contact> it = getContacts();

				while (it.hasNext())
				{
					((SourceRouteNode) it.next().getDest()).checkContactsToNode(this);
				}
			}
		}
	}

	public boolean sendNewMessage(int dataLength, Node destNode)
	{
		Message msg = new Message(network.getNextMessageId(), network.getCurrentTime(), this, destNode, dataLength);

		ArrayList<Contact> sourceRoute = computeRouting(msg, network.getCurrentTime());

		statMsgEntry.setup(msg, network.getCurrentTime(), Stats.MSG_CREATED);
		network.stats().notify(statMsgEntry);

		// Drop message when there is no route
		if (sourceRoute == null || sourceRoute.size() < 1)
		{
			if (network.vShouldLog(Verbose.ERR))
				network.vprint("ERR_SENDING " + msg + " from " + this + " to " + destNode + " - no Route Found!");
			return false;
		}

		if (network.vShouldLog(Verbose.NOTIFY))
			network.vprint("ROUTE for " + msg + " (" + msg.getSourceNode() + " => " + msg.getDestNode() + ") found: "
					+ sourceRoute + "; D: " + lastRouting);

		// Store the route with the message
		msg.storeData(MSG_ROUTE_DATA_KEY, sourceRoute);

		assert lastRouting != null;

		msg.storeData(Stats.MSG_CREATE_PRED_DELAY, lastRouting.getCost((SourceRouteNode) msg.getDestNode()));

		// At this point, we now have to find storage for the message
		// Try to add the message to the forwarding buffer
		if (tryStoreMessage(msg, true))
		{
			// It was added!
			// See if we can send this message out any of the idle contacts
			checkIdleContacts();
			return true;
		}

		return false;
	}

	/** Returns a string version of the route. */
	protected static String routeToString(ArrayList<Contact> route)
	{
		if (route.size() == 0)
			return "[ empty route ]";

		StringBuilder buffer = new StringBuilder(route.get(0).getSource().toString());

		for (Contact c : route)
		{
			buffer.append(", ").append(c.getDest()).append(" (").append(c.getName()).append(")");
		}

		return buffer.toString();
	}

	private NetworkDijkstra.NetworkGraph networkGraph()
	{
		if (nodeAlgo == ALGO_ED)
		{
			return new EDGraph();
		}
		else if (nodeAlgo == ALGO_HOP_COUNT)
		{
			assert ! perContactRouting;
			return new HopCountGraph();
		}
		else
		{
			assert nodeAlgo == ALGO_MED;
			if (perContactRouting)
			{
				return new MEDPerContactGraph(this);
			}
			else
			{
				return new MEDGraph();
			}
		}
	}

	/**
	 * Returns a path to use for routing msg at time forTime.
	 */
	public ArrayList<Contact> computeRouting(Message msg, double forTime)
	{
		if (lastRouting != null)
		{
			if (nodeAlgo == ALGO_ED && (lastRouting.getStartTime() != forTime || lastRouting.context().getDataLength() != msg.getDataLength()))
			{
				/**
				 * ED algorithms keep the same routing until time or message length changes
				 */
				lastRouting = null;
			}
			// Hop count metric must be recomputed if the time or the
			// destination has changed
			else if (nodeAlgo == ALGO_HOP_COUNT
					&& (lastRouting.getStartTime() != forTime || lastRouting.context().getDestNode() != msg
							.getDestNode()))
			{
				lastRouting = null;
			}
		}

		if (lastRouting == null)
		{
			/**
			 * If there is no routing we need to calculate this. MED algorithm
			 * never changes its predictions, so we just need to do routing
			 * once.
			 */
			lastRouting = new NetworkDijkstra(networkGraph(), this, forTime, msg);
		}

		return lastRouting.routeTo((SourceRouteNode) msg.getDestNode());
	}

	public void scheduleContactUp(Contact c, double time)
	{
		ContactSchedule cs = contactSchedules.get(c);

		if (cs == null)
		{
			cs = new ContactSchedule(c);

			contactSchedules.put(c, cs);
		}

		cs.scheduleUp(time);
	}

	public void scheduleContactDown(Contact c, double time)
	{
		ContactSchedule cs = contactSchedules.get(c);

		// It means that that contact was not scheduled to go up before -
		// something is wrong!
		assert cs != null;

		cs.scheduleDown(time);
	}

	public CommandStatus parseCommandPart(ArrayList<String> part, String path)
	{
		String param = part.get(0);
		CommandStatus ok = super.parseCommandPart(part, path);

		if (ok != null)
			return ok;

		ok = new CommandStatus(CommandStatus.COMMAND_OK);

		if (part.size() != 2)
			return null;

		if (param.equals("send"))
		{
			sendNewMessage(msgDataLength, network.getNode(part.get(1)));
			return ok;
		}

		int val;

		try
		{
			if (param.equals("buffer_size"))
			{
				val = Integer.parseInt(part.get(1));

				if (val < -1)
				{
					return new CommandStatus("Max buffer size can not be < -1!");
				}

				bufferSize = val;

				return ok;
			}
			else if (param.equals("msg_size"))
			{
				val = Integer.parseInt(part.get(1));

				if (val < 1)
				{
					return new CommandStatus("Message Size can not be < 1!");
				}

				msgDataLength = val;

				return ok;
			}
			else if (param.equals("sim_time") || param.equals("simulation_time"))
			{
				double f = Double.parseDouble(part.get(1));

				if (f < 1)
				{
					return new CommandStatus("Simulation Time can not be < 1!");
				}

				simulationTime = f;

				return ok;
			}
			else if (param.equals("allow_shortcuts"))
			{
				allowShortcuts = Boolean.parseBoolean(part.get(1));

				return ok;
			}
			else if (param.equals("accurate_send_preds"))
			{
				accurateSendPreds = Boolean.parseBoolean(part.get(1));

				return ok;
			}
			else if (param.equals("contact_routing"))
			{
				perContactRouting = Boolean.parseBoolean(part.get(1));
				if (perContactRouting)
				{
					allowShortcuts = true;
				}

				return ok;
			}
			else if (param.equals("algo") || param.equals("algorithm"))
			{
				String alg = part.get(1).toLowerCase();

				if (alg.equals("ed"))
				{
					nodeAlgo = ALGO_ED;
				}
				else if (alg.equals("med"))