caminos-lib 0.2.0

/*!

A Traffic defines the way load is generated by the servers.

see [`new_traffic`](fn.new_traffic.html) for documentation on the configuration syntax of predefined traffics.

*/


use crate::config_parser::ConfigurationValue;
use std::boxed::Box;
use std::cell::{RefCell};
use crate::{Message,Plugs};
use ::rand::{Rng,StdRng};
use crate::pattern::{Pattern,new_pattern,PatternBuilderArgument};
use std::rc::Rc;
use std::collections::{BTreeSet,BTreeMap,VecDeque};
use crate::topology::Topology;
use quantifiable_derive::Quantifiable;//the derive macro
use crate::quantify::Quantifiable;
//use std::mem::{size_of};
use std::fmt::Debug;

///Possible errors when trying to generate a message with a `Traffic`.
#[derive(Debug)]
pub enum TrafficError
{
	///The traffic tried to send a message outside the network range.
	OriginOutsideTraffic,
	///A server has generated a message to itself. Not necessarily an error.
	SelfMessage,
}

///A traffic to be offered to a network. Each server may generate and consume messages.
///Each should call `should_generate` every cycle unless it is unable to store more messages.
pub trait Traffic : Quantifiable + Debug
{
	///Returns a new message following the indications of the traffic
	fn generate_message(&mut self, origin:usize, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> Result<Rc<Message>,TrafficError>;
	///Get its probability of generating per cycle
	fn probability_per_cycle(&self, server:usize) -> f32;
	///If the message was generated by the traffic updates itself and returns true
	fn try_consume(&mut self, message: Rc<Message>, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> bool;
	///Indicates if the traffic is not going to generate any more messages.
	fn is_finished(&self) -> bool;
	///Returns true if a server should generate a message this cycle
	fn should_generate(&self, server:usize, rng: &RefCell<StdRng>) -> bool
	{
		let p=self.probability_per_cycle(server);
		let r=rng.borrow_mut().gen_range(0f32,1f32);
		r<p
	}
}

#[non_exhaustive]
#[derive(Debug)]
pub struct TrafficBuilderArgument<'a>
{
	///A ConfigurationValue::Object defining the traffic.
	pub cv: &'a ConfigurationValue,
	///The user defined plugs. In case the traffic needs to create elements.
	pub plugs: &'a Plugs,
	///The topology of the network that is gonna to receive the traffic.
	pub topology: &'a Box<dyn Topology>,
	///The random number generator to use.
	pub rng: &'a RefCell<StdRng>,
}

/**Build a new traffic.

## Base traffics.

### Homogeneous traffic
Is a traffic where all servers behave equally and uniform in time. Some `pattern` is generated
by `servers` number of involved servers along the whole simulation. Each server tries to use its link toward the network a `load`
fraction of the cycles. The generated messages has a size in phits of `message_size`. The generation is the typical Bernoulli process.
```
HomogeneousTraffic{
	pattern:Uniform,
	servers:1000,
	load: 0.9,
	message_size: 16,
}
```

### Burst
In the Burst traffic each of the involved `servers` has a initial list of `messages_per_server` messages to emit. When all the messages
are consumed the simulation is requested to end.
```
Burst{
	pattern:Uniform,
	servers:1000,
	messages_per_server:200,
	message_size: 16,
}
```

### Reactive

A Reactive traffic is composed of an `action_traffic` generated normally, whose packets, when consumed create a response by the `reaction_traffic`.
If both subtraffics are requesting to end and there is no pending message the reactive traffic also requests to end.
```
Reactive{
	action_traffic:HomogeneousTraffic{...},
	reaction_traffic:HomogeneousTraffic{...},
}
```

## Operations

### TrafficSum

Generates several traffic at once, if the total load allows it.
```
TrafficSum{
	list: [HomogeneousTraffic{...},... ],
}
```

### ShiftedTraffic

A ShiftedTraffic shifts a given traffic a certain amount of servers. Yu should really check if some pattern transformation fit your purpose, since it will be simpler.
```
ShiftedTraffic{
	traffic: HomogeneousTraffic{...},
	shift: 50,
}
```

### ProductTraffic

A ProductTraffic divides the servers into blocks. Each group generates traffic following the `block_traffic`, but instead of having the destination in the same block it is selected a destination by using the `global_pattern` of the block. Blocks of interest are
* The servers attached to a router. Then if the global_pattern is a permutation, all the servers will comunicate with servers attached to the same router. This can stress the network a lot more than a permutation of servers.
* All servers in a group of a dragonfly. If the global_pattern is a permutation, there is only a global link between groups, and Shortest routing is used, then all the packets generated in a group will try by the same global link. Other global links being unused.
Note there is also a product at pattern level, which may be easier to use.

```
ProductTraffic{
	block_size: 10,
	block_traffic: HomogeneousTraffic{...},
	global_pattern: RandomPermutation,
}
```

### SubRangeTraffic

A SubRangeTraffic makes servers outise the range to not generate traffic.
```
SubRangeTraffic{
	start: 100,
	end: 200,
	traffic: HomogeneousTraffic{...},
}
```
*/
pub fn new_traffic(arg:TrafficBuilderArgument) -> Box<dyn Traffic>
{
	if let &ConfigurationValue::Object(ref cv_name, ref _cv_pairs)=arg.cv
	{
		match arg.plugs.traffics.get(cv_name)
		{
			Some(builder) => return builder(arg),
			_ => (),
		};
		match cv_name.as_ref()
		{
			"HomogeneousTraffic" => Box::new(Homogeneous::new(arg)),
			"TrafficSum" => Box::new(Sum::new(arg)),
			"ShiftedTraffic" => Box::new(Shifted::new(arg)),
			"ProductTraffic" => Box::new(ProductTraffic::new(arg)),
			"SubRangeTraffic" => Box::new(SubRangeTraffic::new(arg)),
			"Burst" => Box::new(Burst::new(arg)),
			"Reactive" => Box::new(Reactive::new(arg)),
			_ => panic!("Unknown traffic {}",cv_name),
		}
	}
	else
	{
		panic!("Trying to create a traffic from a non-Object");
	}
}

///Traffic in which all messages have same size, follow the same pattern, and there is no change with time.
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Homogeneous
{
	///Number of servers applying this traffic.
	servers: usize,
	///The pattern of the communication.
	pattern: Box<dyn Pattern>,
	///The size of each sent message.
	message_size: usize,
	///The load offered to the network. Proportion of the cycles that should be injecting phits.
	load: f32,
	///Set of generated messages.
	generated_messages: BTreeSet<*const Message>,
}

//impl Quantifiable for Homogeneous
//{
//	fn total_memory(&self) -> usize
//	{
//		return size_of::<Self>() + self.pattern.total_memory() + self.generated_messages.total_memory();
//	}
//	fn print_memory_breakdown(&self)
//	{
//		unimplemented!();
//	}
//	fn forecast_total_memory(&self) -> usize
//	{
//		unimplemented!();
//	}
//}

impl Traffic for Homogeneous
{
	fn generate_message(&mut self, origin:usize, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> Result<Rc<Message>,TrafficError>
	{
		if origin>=self.servers
		{
			//panic!("origin {} does not belong to the traffic",origin);
			return Err(TrafficError::OriginOutsideTraffic);
		}
		let destination=self.pattern.get_destination(origin,topology,rng);
		if origin==destination
		{
			return Err(TrafficError::SelfMessage);
		}
		let message=Rc::new(Message{
			origin: origin,
			destination,
			size:self.message_size,
			creation_cycle: cycle,
		});
		self.generated_messages.insert(message.as_ref() as *const Message);
		Ok(message)
	}
	fn probability_per_cycle(&self, _server:usize) -> f32
	{
		let r=self.load/self.message_size as f32;
		//println!("load={} r={} size={}",self.load,r,self.message_size);
		if r>1.0
		{
			1.0
		}
		else
		{
			r
		}
	}
	fn try_consume(&mut self, message: Rc<Message>, _cycle:usize, _topology:&Box<dyn Topology>, _rng: &RefCell<StdRng>) -> bool
	{
		let message_ptr=message.as_ref() as *const Message;
		self.generated_messages.remove(&message_ptr)
	}
	fn is_finished(&self) -> bool
	{
		false
	}
}

impl Homogeneous
{
	pub fn new(arg:TrafficBuilderArgument) -> Homogeneous
	{
		let mut servers=None;
		let mut load=None;
		let mut pattern=None;
		let mut message_size=None;
		if let &ConfigurationValue::Object(ref cv_name, ref cv_pairs)=arg.cv
		{
			if cv_name!="HomogeneousTraffic"
			{
				panic!("A Homogeneous must be created from a `HomogeneousTraffic` object not `{}`",cv_name);
			}
			for &(ref name,ref value) in cv_pairs
			{
				match name.as_ref()
				{
					"pattern" => pattern=Some(new_pattern(PatternBuilderArgument{cv:value,plugs:arg.plugs})),
					"servers" => match value
					{
						&ConfigurationValue::Number(f) => servers=Some(f as usize),
						_ => panic!("bad value for servers"),
					}
					"load" => match value
					{
						&ConfigurationValue::Number(f) => load=Some(f as f32),
						_ => panic!("bad value for load ({:?})",value),
					}
					"message_size" => match value
					{
						&ConfigurationValue::Number(f) => message_size=Some(f as usize),
						_ => panic!("bad value for message_size"),
					}
					_ => panic!("Nothing to do with field {} in HomogeneousTraffic",name),
				}
			}
		}
		else
		{
			panic!("Trying to create a Homogeneous from a non-Object");
		}
		let servers=servers.expect("There were no servers");
		let message_size=message_size.expect("There were no message_size");
		let load=load.expect("There were no load");
		let mut pattern=pattern.expect("There were no pattern");
		pattern.initialize(servers, servers, arg.topology, arg.rng);
		Homogeneous{
			servers,
			pattern,
			message_size,
			load,
			generated_messages: BTreeSet::new(),
		}
	}
}

///Traffic which is the sum of a list of oter traffics.
///While it will clearly work when the sum of the generation rates is at most 1, it should behave nicely enough otherwise.
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Sum
{
	///List of traffic summands
	list: Vec<Box<dyn Traffic>>,
}

//impl Quantifiable for Sum
//{
//	fn total_memory(&self) -> usize
//	{
//		unimplemented!();
//	}
//	fn print_memory_breakdown(&self)
//	{
//		unimplemented!();
//	}
//	fn forecast_total_memory(&self) -> usize
//	{
//		unimplemented!();
//	}
//}

impl Traffic for Sum
{
	fn generate_message(&mut self, origin:usize, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> Result<Rc<Message>,TrafficError>
	{
		let probs:Vec<f32> =self.list.iter().map(|t|t.probability_per_cycle(origin)).collect();
		let mut r=rng.borrow_mut().gen_range(0f32,probs.iter().sum());
		for i in 0..self.list.len()
		{
			if r<probs[i]
			{
				return self.list[i].generate_message(origin,cycle,topology,rng);
			}
			else
			{
				r-=probs[i];
			}
		}
		panic!("failed probability");
	}
	//fn should_generate(&self, rng: &RefCell<StdRng>) -> bool
	//{
	//	let r=rng.borrow_mut().gen_range(0f32,1f32);
	//	r<=self.list.iter().map(|t|t.probability_per_cycle()).sum()
	//}
	fn probability_per_cycle(&self,server:usize) -> f32
	{
		self.list.iter().map(|t|t.probability_per_cycle(server)).sum()
	}
	fn try_consume(&mut self, message: Rc<Message>, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> bool
	{
		for traffic in self.list.iter_mut()
		{
			if traffic.try_consume(message.clone(),cycle,topology,rng)
			{
				return true;
			}
		}
		return false;
	}
	fn is_finished(&self) -> bool
	{
		for traffic in self.list.iter()
		{
			if !traffic.is_finished()
			{
				return false;
			}
		}
		return true;
	}
}

impl Sum
{
	pub fn new(arg:TrafficBuilderArgument) -> Sum
	{
		let mut list=None;
		if let &ConfigurationValue::Object(ref cv_name, ref cv_pairs)=arg.cv
		{
			if cv_name!="TrafficSum"
			{
				panic!("A Sum must be created from a `TrafficSum` object not `{}`",cv_name);
			}
			for &(ref name,ref value) in cv_pairs
			{
				match name.as_ref()
				{
					//"pattern" => pattern=Some(new_pattern(value,plugs)),
					"list" => match value
					{
						&ConfigurationValue::Array(ref a) => list=Some(a.iter().map(|v|new_traffic(TrafficBuilderArgument{cv:v,..arg})).collect()),
						_ => panic!("bad value for list"),
					}
					_ => panic!("Nothing to do with field {} in TrafficSum",name),
				}
			}
		}
		else
		{
			panic!("Trying to create a Sum from a non-Object");
		}
		let list=list.expect("There were no list");
		Sum{
			list
		}
	}
}

///Traffic which is another shifted by some amount of servers
///First check whether a transformation at the `Pattern` level is enough.
///The server `index+shit` will be seen as just `index` by the inner traffic.
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Shifted
{
	///The amount of the shift in servers
	shift: usize,
	///The traffic that is being shifted
	traffic: Box<dyn Traffic>,
	///Set of generated messages.
	generated_messages: BTreeMap<*const Message,Rc<Message>>,
}

//impl Quantifiable for Shifted
//{
//	fn total_memory(&self) -> usize
//	{
//		unimplemented!();
//	}
//	fn print_memory_breakdown(&self)
//	{
//		unimplemented!();
//	}
//	fn forecast_total_memory(&self) -> usize
//	{
//		unimplemented!();
//	}
//}

impl Traffic for Shifted
{
	fn generate_message(&mut self, origin:usize, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> Result<Rc<Message>,TrafficError>
	{
		if origin<self.shift
		{
			return Err(TrafficError::OriginOutsideTraffic);
		}
		//let mut message=self.traffic.generate_message(origin-self.shift,rng)?;
		//message.origin=origin;
		//message.destination+=self.shift;
		//Ok(message)
		let inner_message=self.traffic.generate_message(origin-self.shift,cycle,topology,rng)?;
		let outer_message=Rc::new(Message{
			origin,
			destination:inner_message.destination+self.shift,
			size:inner_message.size,
			creation_cycle: cycle,
		});
		self.generated_messages.insert(outer_message.as_ref() as *const Message,inner_message);
		Ok(outer_message)
	}
	fn probability_per_cycle(&self,server:usize) -> f32
	{
		self.traffic.probability_per_cycle(server)
	}
	fn try_consume(&mut self, message: Rc<Message>, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> bool
	{
		let message_ptr=message.as_ref() as *const Message;
		let outer_message=match self.generated_messages.remove(&message_ptr)
		{
			None => return false,
			Some(m) => m,
		};
		if !self.traffic.try_consume(outer_message,cycle,topology,rng)
		{
			panic!("Shifted traffic consumed a message but its child did not.");
		}
		true
	}
	fn is_finished(&self) -> bool
	{
		self.traffic.is_finished()
	}
}

impl Shifted
{
	pub fn new(arg:TrafficBuilderArgument) -> Shifted
	{
		let mut shift=None;
		let mut traffic=None;
		if let &ConfigurationValue::Object(ref cv_name, ref cv_pairs)=arg.cv
		{
			if cv_name!="ShiftedTraffic"
			{
				panic!("A Shifted must be created from a `ShiftedTraffic` object not `{}`",cv_name);
			}
			for &(ref name,ref value) in cv_pairs
			{
				match name.as_ref()
				{
					"traffic" => traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,..arg})),
					"shift" => match value
					{
						&ConfigurationValue::Number(f) => shift=Some(f as usize),
						_ => panic!("bad value for shift"),
					}
					_ => panic!("Nothing to do with field {} in ShiftedTraffic",name),
				}
			}
		}
		else
		{
			panic!("Trying to create a Shifted from a non-Object");
		}
		let shift=shift.expect("There were no shift");
		let traffic=traffic.expect("There were no traffic");
		Shifted{
			shift,
			traffic,
			generated_messages: BTreeMap::new(),
		}
	}
}

///Divides the network in blocks and use a `Traffic` inside blocks applying a global `Pattern` among blocks.
///First check whether a transformation at the `Pattern` level is enough; specially see the `Product` pattern.
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct ProductTraffic
{
	block_size: usize,
	block_traffic: Box<dyn Traffic>,
	global_pattern: Box<dyn Pattern>,
	// ///The amount of the shift in servers
	// shift: usize,
	// ///The traffic that is being shifted
	// traffic: Box<dyn Traffic>,
	///Set of generated messages.
	generated_messages: BTreeMap<*const Message,Rc<Message>>,
}

impl Traffic for ProductTraffic
{
	fn generate_message(&mut self, origin:usize, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> Result<Rc<Message>,TrafficError>
	{
		let local=origin % self.block_size;
		let global=origin / self.block_size;
		//let local_dest=self.block_pattern.get_destination(local,topology,rng);
		let global_dest=self.global_pattern.get_destination(global,topology,rng);
		//global_dest*self.block_size+local_dest
		let inner_message=self.block_traffic.generate_message(local,cycle,topology,rng)?;
		let outer_message=Rc::new(Message{
			origin,
			destination:global_dest*self.block_size+inner_message.destination,
			size:inner_message.size,
			creation_cycle: cycle,
		});
		self.generated_messages.insert(outer_message.as_ref() as *const Message,inner_message);
		Ok(outer_message)
	}
	fn probability_per_cycle(&self,server:usize) -> f32
	{
		self.block_traffic.probability_per_cycle(server)
	}
	fn try_consume(&mut self, message: Rc<Message>, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> bool
	{
		let message_ptr=message.as_ref() as *const Message;
		let outer_message=match self.generated_messages.remove(&message_ptr)
		{
			None => return false,
			Some(m) => m,
		};
		if !self.block_traffic.try_consume(outer_message,cycle,topology,rng)
		{
			panic!("ProductTraffic traffic consumed a message but its child did not.");
		}
		true
	}
	fn is_finished(&self) -> bool
	{
		self.block_traffic.is_finished()
	}
}

impl ProductTraffic
{
	pub fn new(arg:TrafficBuilderArgument) -> ProductTraffic
	{
		let mut block_size=None;
		let mut block_traffic=None;
		let mut global_pattern=None;
		if let &ConfigurationValue::Object(ref cv_name, ref cv_pairs)=arg.cv
		{
			if cv_name!="ProductTraffic"
			{
				panic!("A ProductTraffic must be created from a `ProductTraffic` object not `{}`",cv_name);
			}
			for &(ref name,ref value) in cv_pairs
			{
				match name.as_ref()
				{
					"block_traffic" => block_traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,..arg})),
					"global_pattern" => global_pattern=Some(new_pattern(PatternBuilderArgument{cv:value,plugs:arg.plugs})),
					"block_size" => match value
					{
						&ConfigurationValue::Number(f) => block_size=Some(f as usize),
						_ => panic!("bad value for block_size"),
					}
					_ => panic!("Nothing to do with field {} in ShiftedTraffic",name),
				}
			}
		}
		else
		{
			panic!("Trying to create a ProductTraffic from a non-Object");
		}
		let block_size=block_size.expect("There were no block_size");
		let block_traffic=block_traffic.expect("There were no block_traffic");
		let mut global_pattern=global_pattern.expect("There were no global_pattern");
		let global_size=arg.topology.num_servers()/block_size;
		global_pattern.initialize(global_size,global_size,arg.topology,arg.rng);
		ProductTraffic{
			block_size,
			block_traffic,
			global_pattern,
			generated_messages: BTreeMap::new(),
		}
	}
}

///Only allow servers in range will generate messages. The messages can go out of the given range.
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct SubRangeTraffic
{
	///The first element actually in the traffic.
	start: usize,
	///The next to the last element actually in the traffic.
	end: usize,
	///The traffic that is being filtered.
	traffic: Box<dyn Traffic>,
	// /Set of generated messages.
	//generated_messages: BTreeMap<*const Message,Rc<Message>>,
}

impl Traffic for SubRangeTraffic
{
	fn generate_message(&mut self, origin:usize, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> Result<Rc<Message>,TrafficError>
	{
		if origin<self.start || origin>=self.end
		{
			return Err(TrafficError::OriginOutsideTraffic);
		}
		self.traffic.generate_message(origin,cycle,topology,rng)
	}
	fn probability_per_cycle(&self,server:usize) -> f32
	{
		self.traffic.probability_per_cycle(server)
	}
	fn try_consume(&mut self, message: Rc<Message>, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> bool
	{
		self.traffic.try_consume(message,cycle,topology,rng)
	}
	fn is_finished(&self) -> bool
	{
		self.traffic.is_finished()
	}
}

impl SubRangeTraffic
{
	pub fn new(arg:TrafficBuilderArgument) -> SubRangeTraffic
	{
		let mut start=None;
		let mut end=None;
		let mut traffic=None;
		if let &ConfigurationValue::Object(ref cv_name, ref cv_pairs)=arg.cv
		{
			if cv_name!="SubRangeTraffic"
			{
				panic!("A SubRangeTraffic must be created from a `SubRangeTraffic` object not `{}`",cv_name);
			}
			for &(ref name,ref value) in cv_pairs
			{
				match name.as_ref()
				{
					"traffic" => traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,..arg})),
					"start" => match value
					{
						&ConfigurationValue::Number(f) => start=Some(f as usize),
						_ => panic!("bad value for start"),
					}
					"end" => match value
					{
						&ConfigurationValue::Number(f) => end=Some(f as usize),
						_ => panic!("bad value for end"),
					}
					_ => panic!("Nothing to do with field {} in SubRangeTraffic",name),
				}
			}
		}
		else
		{
			panic!("Trying to create a SubRangeTraffic from a non-Object");
		}
		let start=start.expect("There were no start");
		let end=end.expect("There were no end");
		let traffic=traffic.expect("There were no traffic");
		SubRangeTraffic{
			start,
			end,
			traffic,
			//generated_messages: BTreeMap::new(),
		}
	}
}

///Initialize an amount of messages to send from each server.
///The traffic will be considered complete when all servers have generated their messages and all of them have been consumed.
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Burst
{
	///Number of servers applying this traffic.
	servers: usize,
	///The pattern of the communication.
	pattern: Box<dyn Pattern>,
	///The size of each sent message.
	message_size: usize,
	///The number of messages each server has pending to sent.
	pending_messages: Vec<usize>,
	///Set of generated messages.
	generated_messages: BTreeSet<*const Message>,
}

//impl Quantifiable for Burst
//{
//	fn total_memory(&self) -> usize
//	{
//		unimplemented!();
//	}
//	fn print_memory_breakdown(&self)
//	{
//		unimplemented!();
//	}
//	fn forecast_total_memory(&self) -> usize
//	{
//		unimplemented!();
//	}
//}

impl Traffic for Burst
{
	fn generate_message(&mut self, origin:usize, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> Result<Rc<Message>,TrafficError>
	{
		if origin>=self.servers
		{
			//panic!("origin {} does not belong to the traffic",origin);
			return Err(TrafficError::OriginOutsideTraffic);
		}
		self.pending_messages[origin]-=1;
		let destination=self.pattern.get_destination(origin,topology,rng);
		if origin==destination
		{
			return Err(TrafficError::SelfMessage);
		}
		let message=Rc::new(Message{
			origin: origin,
			destination,
			size:self.message_size,
			creation_cycle: cycle,
		});
		self.generated_messages.insert(message.as_ref() as *const Message);
		Ok(message)
	}
	fn probability_per_cycle(&self, server:usize) -> f32
	{
		if self.pending_messages[server]>0
		{
			1.0
		}
		else
		{
			0.0
		}
	}
	fn try_consume(&mut self, message: Rc<Message>, _cycle:usize, _topology:&Box<dyn Topology>, _rng: &RefCell<StdRng>) -> bool
	{
		let message_ptr=message.as_ref() as *const Message;
		self.generated_messages.remove(&message_ptr)
	}
	fn is_finished(&self) -> bool
	{
		if self.generated_messages.len()>0
		{
			return false;
		}
		for &pm in self.pending_messages.iter()
		{
			if pm>0
			{
				return false;
			}
		}
		true
	}
}

impl Burst
{
	pub fn new(arg:TrafficBuilderArgument) -> Burst
	{
		let mut servers=None;
		let mut messages_per_server=None;
		let mut pattern=None;
		let mut message_size=None;
		if let &ConfigurationValue::Object(ref cv_name, ref cv_pairs)=arg.cv
		{
			if cv_name!="Burst"
			{
				panic!("A Burst must be created from a `Burst` object not `{}`",cv_name);
			}
			for &(ref name,ref value) in cv_pairs
			{
				match name.as_ref()
				{
					"pattern" => pattern=Some(new_pattern(PatternBuilderArgument{cv:value,plugs:arg.plugs})),
					"servers" => match value
					{
						&ConfigurationValue::Number(f) => servers=Some(f as usize),
						_ => panic!("bad value for servers"),
					}
					"messages_per_server" => match value
					{
						&ConfigurationValue::Number(f) => messages_per_server=Some(f as usize),
						_ => panic!("bad value for messages_per_server ({:?})",value),
					}
					"message_size" => match value
					{
						&ConfigurationValue::Number(f) => message_size=Some(f as usize),
						_ => panic!("bad value for message_size"),
					}
					_ => panic!("Nothing to do with field {} in Burst",name),
				}
			}
		}
		else
		{
			panic!("Trying to create a Burst from a non-Object");
		}
		let servers=servers.expect("There were no servers");
		let message_size=message_size.expect("There were no message_size");
		let messages_per_server=messages_per_server.expect("There were no messages_per_server");
		let mut pattern=pattern.expect("There were no pattern");
		pattern.initialize(servers, servers, arg.topology, arg.rng);
		Burst{
			servers,
			pattern,
			message_size,
			pending_messages:vec![messages_per_server;servers],
			generated_messages: BTreeSet::new(),
		}
	}
}


///Has a major traffic `action_traffic` generated normally. When a message from this `action_traffic` is consumed, the `reaction_traffic` is requested for a message. This reaction message will be generated by the server that consumed the action message. The destination of the reaction message is independent of the origin of the action message.
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Reactive
{
	action_traffic: Box<dyn Traffic>,
	reaction_traffic: Box<dyn Traffic>,
	pending_messages: Vec<VecDeque<Rc<Message>>>,
}


//impl Quantifiable for Reactive
//{
//	fn total_memory(&self) -> usize
//	{
//		unimplemented!();
//	}
//	fn print_memory_breakdown(&self)
//	{
//		unimplemented!();
//	}
//	fn forecast_total_memory(&self) -> usize
//	{
//		unimplemented!();
//	}
//}
impl Traffic for Reactive
{
	fn generate_message(&mut self, origin:usize, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> Result<Rc<Message>,TrafficError>
	{
		if origin<self.pending_messages.len()
		{
			if let Some(message)=self.pending_messages[origin].pop_front()
			{
				return Ok(message);
			}
		}
		return self.action_traffic.generate_message(origin,cycle,topology,rng);
	}
	fn probability_per_cycle(&self, server:usize) -> f32
	{
		if server<self.pending_messages.len()
		{
			if self.pending_messages[server].len()>0
			{
				return 1.0;
			}
		}
		return self.action_traffic.probability_per_cycle(server);
	}
	fn try_consume(&mut self, message: Rc<Message>, cycle:usize, topology:&Box<dyn Topology>, rng: &RefCell<StdRng>) -> bool
	{
		if self.action_traffic.try_consume(message.clone(),cycle,topology,rng)
		{
			if self.reaction_traffic.should_generate(message.origin,rng)
			{
				match self.reaction_traffic.generate_message(message.origin,cycle,topology,rng)
				{
					Ok(response_message) =>
					{
						if self.pending_messages.len()<message.origin+1
						{
							self.pending_messages.resize(message.origin+1,VecDeque::new());
						}
						self.pending_messages[message.origin].push_back(response_message);
					},
					//Err(TrafficError::OriginOutsideTraffic) => (),
					Err(error) => panic!("An error happened when generating response traffic: {:?}",error),
				};
			}
			return true;
		}
		self.reaction_traffic.try_consume(message,cycle,topology,rng)
	}
	fn is_finished(&self) -> bool
	{
		if !self.action_traffic.is_finished() || !self.reaction_traffic.is_finished()
		{
			return false;
		}
		for pm in self.pending_messages.iter()
		{
			if pm.len()>0
			{
				return false;
			}
		}
		return true;
	}
}

impl Reactive
{
	pub fn new(arg:TrafficBuilderArgument) -> Reactive
	{
		let mut action_traffic=None;
		let mut reaction_traffic=None;
		if let &ConfigurationValue::Object(ref cv_name, ref cv_pairs)=arg.cv
		{
			if cv_name!="Reactive"
			{
				panic!("A Reactive must be created from a `Reactive` object not `{}`",cv_name);
			}
			for &(ref name,ref value) in cv_pairs
			{
				match name.as_ref()
				{
					"action_traffic" => action_traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,..arg})),
					"reaction_traffic" => reaction_traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,..arg})),
					_ => panic!("Nothing to do with field {} in Reactive",name),
				}
			}
		}
		else
		{
			panic!("Trying to create a Reactive from a non-Object");
		}
		let action_traffic=action_traffic.expect("There were no action_traffic");
		let reaction_traffic=reaction_traffic.expect("There were no reaction_traffic");
		Reactive{
			action_traffic,
			reaction_traffic,
			pending_messages:vec![],
		}
	}
}