MaxPeerCountChurnGenerator.java

/*
 * Copyright (c) 2005-2010 KOM – Multimedia Communications Lab
 *
 * This file is part of PeerfactSim.KOM.
 * 
 * PeerfactSim.KOM is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * any later version.
 * 
 * PeerfactSim.KOM is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with PeerfactSim.KOM.  If not, see <http://www.gnu.org/licenses/>.
 *
 */

package de.tud.kom.p2psim.impl.churn;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedList;
import java.util.List;
import java.util.PriorityQueue;

import de.tud.kom.p2psim.api.common.SimHost;
import de.tud.kom.p2psim.api.network.SimNetInterface;
import de.tud.kom.p2psim.api.network.SimNetworkComponent;
import de.tud.kom.p2psim.impl.scenario.DefaultConfigurator;
import de.tud.kom.p2psim.impl.simengine.Simulator;
import de.tud.kom.p2psim.impl.util.oracle.GlobalOracle;
import de.tud.kom.p2psim.impl.util.toolkits.CollectionHelpers;
import de.tud.kom.p2psim.impl.util.toolkits.Predicates;
import de.tudarmstadt.maki.simonstrator.api.Event;
import de.tudarmstadt.maki.simonstrator.api.EventHandler;
import de.tudarmstadt.maki.simonstrator.api.Time;
import de.tudarmstadt.maki.simonstrator.api.component.GlobalComponent;
import de.tudarmstadt.maki.simonstrator.api.util.XMLConfigurableConstructor;

/**
 * ChurnModel that follows a defined "trace" in a csv file. The trace must
 * consist of the following parameters per line:
 * 
 * startTime, intervalLength numberOfClients
 * 
 * startTime - the time when the specified number of nodes is to be set.
 * 
 * intervalLength - the time in which the specified number of nodes (startTime)
 * must be achieved by the model. Those two together also form the inter
 * join/leaving rate.
 * 
 * numberOfClients - the number of clients to be achieved.
 * 
 * The model checks for a minimum intervalLength of 10 minutes.
 * 
 * 
 * @author Nils Richerzhagen
 * @version 1.0, Nov 24, 2015
 */
public class MaxPeerCountChurnGenerator
		implements EventHandler, GlobalComponent {

	private static final int _PeerCountEvent = 1001;

	private static final int _CHURN_START = 1002;

	private static final int _CHURN_EVENT = 1003;

	private static final int _CHURN_NOCHURN_HOSTS = 1004;

	private final String commentsDelimiter = "#";

	private final String SEP = ",";

	private final long _minBurstLength = 10 * Time.MINUTE;

	/**
	 * {@link ChurnInfo} from the csv file.
	 */
	private LinkedList<ChurnInfo> churnInfos = new LinkedList<ChurnInfo>();

	private PriorityQueue<HostSessionInfo> onlineHostsSortedByOnlineTime;

	private PriorityQueue<HostSessionInfo> offlineHostsSortedByOfflineTime;

	/**
	 * Comparator used to sort client infos by offline time
	 */
	private static final Comparator<HostSessionInfo> COMP_TIME = new Comparator<MaxPeerCountChurnGenerator.HostSessionInfo>() {
		@Override
		public int compare(HostSessionInfo o1, HostSessionInfo o2) {
			return ((Long) o1.timestamp).compareTo(o2.timestamp);
		}
	};

	@XMLConfigurableConstructor({ "file" })
	public MaxPeerCountChurnGenerator(String file) {
		parseTrace(file);
	}

	/**
	 * Called by the configurator.
	 * 
	 * @param churnStart
	 */
	public void setChurnStart(long churnStart) {
		// Event.scheduleWithDelay(churnStart, this, null, _CHURN_START);
		Event.scheduleImmediately(this, null, _CHURN_START);
	}

	public void initialize() {
		for (ChurnInfo churnInfo : churnInfos) {
			Event.scheduleWithDelay(churnInfo.getStartTime(), this, churnInfo,
					_PeerCountEvent);
		}
	}

	/**
	 * Start adapting on new churn rate, when next churnInfo is valid.
	 */
	private void configureMaxPeerCount(ChurnInfo currentChurnInfo) {
		long currentTime = Simulator.getCurrentTime();

		assert currentChurnInfo
				.getStartTime() == currentTime : "The ChurnInfo to use is scheduled for the exact time, thus it should be the same.";

		/*
		 * Wanted number > current number. --> need nodes = go online
		 */
		if (currentChurnInfo
				.getNumberOfClients() >= onlineHostsSortedByOnlineTime.size()) {
			int count = currentChurnInfo.getNumberOfClients()
					- onlineHostsSortedByOnlineTime.size();

			for (int i = 0; i < count; i++) {
				/*
				 * Schedule the required number of hosts for going online
				 * churnEvent. Get oldest entry in sortedOffline list and put
				 * online.
				 */
				HostSessionInfo hostSessionInfo = offlineHostsSortedByOfflineTime
						.poll();
				assert hostSessionInfo != null : "HostSessionInfo shouldn't be null - means to few hosts were configured.";

				ChurnEvent churnEvent = new ChurnEvent(hostSessionInfo.host,
						true);
				long currentJoin = i
						* (currentChurnInfo.getBurstLength() / count);

				Event.scheduleWithDelay(currentJoin, this, churnEvent,
						_CHURN_EVENT);
			}
		}
		/*
		 * Wanted number < currentNumber --> remove nodes = goOffline
		 */
		else if (currentChurnInfo
				.getNumberOfClients() < onlineHostsSortedByOnlineTime.size()) {
			int count = onlineHostsSortedByOnlineTime.size()
					- currentChurnInfo.getNumberOfClients();

			for (int i = 0; i < count; i++) {
				/*
				 * Schedule the required number of hosts for going offline
				 * churnEvent. Get oldest entry in sortedOnline list and put
				 * offline.
				 */
				HostSessionInfo hostSessionInfo = onlineHostsSortedByOnlineTime
						.poll();
				assert hostSessionInfo != null : "HostSessionInfo shouldn't be null - means no hosts were online.";

				ChurnEvent churnEvent = new ChurnEvent(hostSessionInfo.host,
						false);
				long currentLeave = i
						* (currentChurnInfo.getBurstLength() / count);
				Event.scheduleWithDelay(currentLeave, this, churnEvent,
						_CHURN_EVENT);
			}
		} else {
			throw new AssertionError();
		}
	}

	@Override
	public void eventOccurred(Object content, int type) {
		if (type == _PeerCountEvent) {
			ChurnInfo churnInfo = (ChurnInfo) content;
			configureMaxPeerCount(churnInfo);
		} else if (type == _CHURN_EVENT) {
			long currentTime = Simulator.getCurrentTime();
			ChurnEvent churnEvent = (ChurnEvent) content;
			SimNetworkComponent net = churnEvent.host.getNetworkComponent();
			if (churnEvent.goOnline) {
				for (SimNetInterface netI : net.getSimNetworkInterfaces()) {
					if (netI.isOffline()) {
						netI.goOnline();
					}
				}
				onlineHostsSortedByOnlineTime
						.add(new HostSessionInfo(churnEvent.host, currentTime));
			} else {
				for (SimNetInterface netI : net.getSimNetworkInterfaces()) {
					if (netI.isOnline()) {
						netI.goOffline();
					}
				}
				offlineHostsSortedByOfflineTime
						.add(new HostSessionInfo(churnEvent.host, currentTime));
			}
		} else if (type == _CHURN_START) {
			initialize();

			List<SimHost> hosts = new ArrayList<SimHost>(this.filterHosts());
			this.prepare(hosts);

			offlineHostsSortedByOfflineTime = new PriorityQueue<HostSessionInfo>(
					(int) Math.ceil(hosts.size() / 10.0), COMP_TIME);
			onlineHostsSortedByOnlineTime = new PriorityQueue<HostSessionInfo>(
					(int) Math.ceil(hosts.size() / 10.0), COMP_TIME);

			long currentTime = Simulator.getCurrentTime();
			for (SimHost host : hosts) {
				offlineHostsSortedByOfflineTime
						.add(new HostSessionInfo(host, currentTime));
			}
		} else if (type == _CHURN_NOCHURN_HOSTS) {
			// Send no-churn-hosts online immediately
			List<SimHost> nochurnhosts = (List<SimHost>) content;

			for (SimHost host : nochurnhosts) {
				for (SimNetInterface netI : host.getNetworkComponent()
						.getSimNetworkInterfaces()) {
					if (netI.isOffline()) {
						netI.goOnline();
					}
				}
			}
		}
	}

	/**
	 * Send hosts offline! Should be used to start with churnHosts in offline
	 * state.
	 * 
	 * @param hosts
	 */
	private void prepare(List<SimHost> hosts) {
		for (SimHost host : hosts) {
			for (SimNetInterface netI : host.getNetworkComponent()
					.getSimNetworkInterfaces()) {
				if (netI.isOnline()) {
					netI.goOffline();
				}
			}
		}
	}

	/**
	 * Gets all hosts and takes all churn affected hosts. Schedules the non
	 * affected host to go online immediately.
	 * 
	 * @return
	 */
	private List<SimHost> filterHosts() {
		List<SimHost> tmp = GlobalOracle.getHosts();
		List<SimHost> filteredHosts = new LinkedList<SimHost>();

		CollectionHelpers.filter(tmp, filteredHosts,
				Predicates.IS_CHURN_AFFECTED);
		List<SimHost> noChurn = new LinkedList<SimHost>();
		noChurn.addAll(tmp);
		noChurn.removeAll(filteredHosts);
		Event.scheduleImmediately(this, noChurn, _CHURN_NOCHURN_HOSTS);
		return filteredHosts;
	}

	/**
	 * Reads the file given by the configuration and parses the churn events.
	 * 
	 * @param filename
	 */
	private void parseTrace(String filename) {
		System.out.println("==============================");
		System.out.println("Reading trace from " + filename);

		/*
		 * This parser works for the following csv file structure.
		 * 
		 * startTime, intervalLength numberOfClients
		 * 
		 */
		BufferedReader csv = null;
		boolean entrySuccessfullyRead = false;

		try {
			csv = new BufferedReader(new FileReader(filename));

			long previousEndTime = 0;

			while (csv.ready()) {
				String line = csv.readLine();
				if (line.length() == 0 || line.startsWith(commentsDelimiter))
					continue;

				if (line.indexOf(SEP) > -1) {
					String[] parts = line.split(SEP);

					if (parts.length == 3) {
						try {

							long startTime = DefaultConfigurator.parseNumber(
									parts[0].replaceAll("\\s+", ""),
									Long.class);
							long burstLength = DefaultConfigurator.parseNumber(
									parts[1].replaceAll("\\s+", ""),
									Long.class);
							int numberOfClients = DefaultConfigurator
									.parseNumber(
											parts[2].replaceAll("\\s+", ""),
											Integer.class);

							// Insanity Checks
							assert startTime >= previousEndTime : "Start time for next fluctuation must be greater than previous end time.";

							assert burstLength >= _minBurstLength : "The minimal length of the burst must be at least 10m.";

							previousEndTime = startTime + burstLength;

							churnInfos.add(new ChurnInfo(startTime, burstLength,
									numberOfClients));

							entrySuccessfullyRead = true;

						} catch (NumberFormatException e) {
							// Ignore leading comments
							if (entrySuccessfullyRead) {
								// System.err.println("CSV ParseError " +
								// line);
							}
						}
					} else {
						throw new AssertionError("To many/few columns in CSV.");
					}
				}
			}
		} catch (Exception e) {
			System.err.println("Could not open " + filename);
			throw new RuntimeException("Could not open " + filename);
		} finally {
			if (csv != null) {
				try {
					csv.close();
				} catch (IOException e) {
					//
				}
			}
		}
	}

	/**
	 * 
	 * @author Nils Richerzhagen
	 * @version 1.0, Nov 25, 2015
	 */
	private class HostSessionInfo {

		public final SimHost host;

		public final long timestamp;

		public HostSessionInfo(SimHost host, long timestamp) {
			this.host = host;
			this.timestamp = timestamp;
		}
	}

	private class ChurnEvent {

		public final SimHost host;

		public final boolean goOnline;

		ChurnEvent(SimHost host, boolean goOnline) {
			this.host = host;
			this.goOnline = goOnline;
		}
	}

	/**
	 * Churn Info for the fluctuation intervals.
	 * 
	 * @author Nils Richerzhagen
	 */
	private class ChurnInfo {

		/**
		 * The time the burst starts.
		 */
		private long startTime;

		/**
		 * The time the burst takes.
		 */
		private long burstLength;

		/**
		 * The max number of nodes that join during that burst.
		 */
		private int numberOfClients;

		public ChurnInfo(long startTime, long burstLength,
				int numberOfClients) {
			this.startTime = startTime;
			this.burstLength = burstLength;
			this.numberOfClients = numberOfClients;
		}

		public long getStartTime() {
			return startTime;
		}

		public int getNumberOfClients() {
			return numberOfClients;
		}

		public long getBurstLength() {
			return burstLength;
		}
	}
}