caminos-lib 0.6.3

/*!

A [Traffic] defines the way their tasks generate load. In the root traffic of a simulation there should be as many tasks as servers. Traffics with other number of tasks can be combined into a such main traffic.

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

*/

use std::boxed::Box;
use std::rc::Rc;
use std::collections::{BTreeSet,BTreeMap,VecDeque};
//use std::mem::{size_of};
use std::fmt::Debug;

use ::rand::{Rng,rngs::StdRng};

use crate::match_object_panic;
use crate::config_parser::ConfigurationValue;
use crate::{Message,Plugs};
use crate::pattern::{Pattern,new_pattern,PatternBuilderArgument};
use crate::topology::Topology;
use crate::event::Time;
use quantifiable_derive::Quantifiable;//the derive macro
use crate::quantify::Quantifiable;

///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 task has generated a message to itself. Not necessarily an error.
	SelfMessage,
}

#[derive(Debug)]
pub enum TaskTrafficState
{
	///The task is currently generating traffic.
	Generating,
	///The task is currently waiting to receive some message from others.
	///If the task is known to not going to generate any more traffic it should be a `FinishedGenerating` state instead.
	WaitingData,
	///The task is not going to generate traffic nor change state until the `cycle`.
	WaitingCycle{cycle:Time},
	///The task is not generating traffic for some other reasons.
	UnspecifiedWait,
	///This task will not generate more traffic, but perhaps it will consume.
	FinishedGenerating,
	///This task has nothing else to do within this `Traffic`.
	Finished,
}

///A traffic to be offered to a network. Each task 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:Time, topology:&dyn Topology, rng: &mut StdRng) -> Result<Rc<Message>,TrafficError>;
	///Get its probability of generating per cycle
	fn probability_per_cycle(&self, task:usize) -> f32;
	///If the message was generated by the traffic updates itself and returns true
	///The argument `task` is the one consuming the message.
	fn try_consume(&mut self, task:usize, message: Rc<Message>, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> bool;
	///Indicates if the traffic is not going to generate any more messages.
	///Should be true if and only if the state of all tasks is `Finished`.
	fn is_finished(&self) -> bool;
	///Returns true if a task should generate a message this cycle
	///Should coincide with having the `Generating` state for deterministic traffics.
	fn should_generate(&self, task:usize, _cycle:Time, rng: &mut StdRng) -> bool
	{
		let p=self.probability_per_cycle(task);
		let r=rng.gen_range(0f32..1f32);
		r<p
	}
	///Indicates the state of the task within the traffic.
	fn task_state(&self, task:usize, cycle:Time) -> TaskTrafficState;

	/// Indicates the number of tasks in the traffic.
	/// A task is a process that generates traffic.
	fn number_tasks(&self) -> usize;
}

#[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 dyn Topology,
	///The random number generator to use.
	pub rng: &'a mut StdRng,
}

/**Build a new traffic.

## Base traffics.

### Homogeneous
[Homogeneous] is a traffic where all tasks behave equally and uniform in time. Some `pattern` is generated
by `tasks` number of involved tasks along the whole simulation. Each task 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.

Example configuration.
```ignore
HomogeneousTraffic{
	pattern:Uniform,
	tasks:1000,
	load: 0.9,
	message_size: 16,
}
```

### Burst
In the [Burst] traffic each of the involved `tasks` has a initial list of `messages_per_task` messages to emit. When all the messages
are consumed the simulation is requested to end.
```ignore
Burst{
	pattern:Uniform,
	tasks:1000,
	messages_per_task: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.
```ignore
Reactive{
	action_traffic:HomogeneousTraffic{...},
	reaction_traffic:HomogeneousTraffic{...},
}
```

## Operations

### TrafficSum

[TrafficSum](Sum) generates several traffic at once. Each task generates load for all the traffics, if the total load allows it.
```ignore
TrafficSum{
	list: [HomogeneousTraffic{...},... ],
}
```

### ShiftedTraffic

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

### ProductTraffic

A [ProductTraffic] divides the tasks 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 tasks attached to a router. Then if the global_pattern is a permutation, all the tasks will comunicate with tasks attached to the same router. This can stress the network a lot more than a permutation of tasks.
* All tasks 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.

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

### SubRangeTraffic

A [SubRangeTraffic] makes tasks outside the range to not generate traffic.
```ignore
SubRangeTraffic{
	start: 100,
	end: 200,
	traffic: HomogeneousTraffic{...},
}
```

### TimeSequenced

[TimeSequenced] defines a sequence of traffics with the given finalization times.

```ignore
TimeSequenced{
	traffics: [HomogeneousTraffic{...}, HomogeneousTraffic{...}],
	times: [2000, 15000],
}
```

### Sequence

Defines a [Sequence] of traffics. When one is completed the next starts.

```ignore
Sequence{
	traffics: [Burst{...}, Burst{...}],
}
```

## Meta traffics

### TrafficMap

A [TrafficMap] applies a map over the tasks of a traffic. This can be used to shuffle the tasks in a application, as in the following example.

```ignore
TrafficMap{
	tasks: 1000,
	application: HomogeneousTraffic{...},
	map: RandomPermutation,
}
```

A [TrafficMap] also can map the set of tasks into a greater set. This is, a small application can be seen as a large one in which many tasks do nothing. This is useful to combine several traffics into one. See its documentation for more details.

*/
pub fn new_traffic(arg:TrafficBuilderArgument) -> Box<dyn Traffic>
{
	if let &ConfigurationValue::Object(ref cv_name, ref _cv_pairs)=arg.cv
	{
		if let Some(builder) = arg.plugs.traffics.get(cv_name)
		{
			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)),
			"MultimodalBurst" => Box::new(MultimodalBurst::new(arg)),
			"Reactive" => Box::new(Reactive::new(arg)),
			"TimeSequenced" => Box::new(TimeSequenced::new(arg)),
			"Sequence" => Box::new(Sequence::new(arg)),
			"BoundedDifference" => Box::new(BoundedDifference::new(arg)),
			"TrafficMap" => Box::new(TrafficMap::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.

```ignore
HomogeneousTraffic{
	pattern:Uniform,
	tasks:1000,
	load: 0.9,
	message_size: 16,
}
```
**/
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Homogeneous
{
	///Number of tasks applying this traffic.
	tasks: 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 Traffic for Homogeneous
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> Result<Rc<Message>,TrafficError>
	{
		if origin>=self.tasks
		{
			//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,
			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, _task: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, _task:usize, message: Rc<Message>, _cycle:Time, _topology:&dyn Topology, _rng: &mut StdRng) -> bool
	{
		let message_ptr=message.as_ref() as *const Message;
		self.generated_messages.remove(&message_ptr)
	}
	fn is_finished(&self) -> bool
	{
		false
	}
	fn task_state(&self, _task:usize, _cycle:Time) -> TaskTrafficState
	{
		TaskTrafficState::Generating
	}

	fn number_tasks(&self) -> usize {
		self.tasks
	}
}

impl Homogeneous
{
	pub fn new(arg:TrafficBuilderArgument) -> Homogeneous
	{
		let mut tasks=None;
		let mut load=None;
		let mut pattern=None;
		let mut message_size=None;
		match_object_panic!(arg.cv,"HomogeneousTraffic",value,
			"pattern" => pattern=Some(new_pattern(PatternBuilderArgument{cv:value,plugs:arg.plugs})),
			"tasks" | "servers" => tasks=Some(value.as_f64().expect("bad value for tasks") as usize),
			"load" => load=Some(value.as_f64().expect("bad value for load") as f32),
			"message_size" => message_size=Some(value.as_f64().expect("bad value for message_size") as usize),
		);
		let tasks=tasks.expect("There were no tasks");
		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(tasks, tasks, arg.topology, arg.rng);
		Homogeneous{
			tasks,
			pattern,
			message_size,
			load,
			generated_messages: BTreeSet::new(),
		}
	}
}

/**
Traffic which is the sum of a list of other traffics.
While it will clearly work when the sum of the generation rates is at most 1, it should behave nicely enough otherwise.

All the subtraffics in `list` must give the same value for `number_tasks`, which is also used for TrafficSum. At least one such subtraffic must be provided.

```ignore
TrafficSum{
	list: [HomogeneousTraffic{...},... ],
}
```
**/
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Sum
{
	///List of traffic summands
	list: Vec<Box<dyn Traffic>>,
}

impl Traffic for Sum
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> Result<Rc<Message>,TrafficError>
	{
		let probs:Vec<f32> =self.list.iter().map(|t|t.probability_per_cycle(origin)).collect();
		//let mut r=rng.gen_range(0f32,probs.iter().sum());//rand-0.4
		let mut r=rng.gen_range(0f32..probs.iter().sum());//rand-0.8
		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: &mut StdRng) -> bool
	//{
	//	let r=rng.gen_range(0f32,1f32);
	//	r<=self.list.iter().map(|t|t.probability_per_cycle()).sum()
	//}
	fn probability_per_cycle(&self,task:usize) -> f32
	{
		self.list.iter().map(|t|t.probability_per_cycle(task)).sum()
	}
	fn try_consume(&mut self, task:usize, message: Rc<Message>, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> bool
	{
		for traffic in self.list.iter_mut()
		{
			if traffic.try_consume(task,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;
	}
	fn task_state(&self, task:usize, cycle:Time) -> TaskTrafficState
	{
		use TaskTrafficState::*;
		//let states = self.list.iter().map(|t|t.server_state(server,cycle)).collect();
		let mut state = Finished;
		for traffic in self.list.iter()
		{
			match traffic.task_state(task,cycle)
			{
				Finished => (),
				Generating => return Generating,
				FinishedGenerating => state = FinishedGenerating,
				_ => state = UnspecifiedWait,
			}
		}
		state
	}

	fn number_tasks(&self) -> usize {
		// all traffics have the same number of tasks
		self.list[0].number_tasks()
	}
}

impl Sum
{
	pub fn new(mut arg:TrafficBuilderArgument) -> Sum
	{
		let mut list : Option<Vec<_>> =None;
		match_object_panic!(arg.cv,"TrafficSum",value,
			"list" => list = Some(value.as_array().expect("bad value for list").iter()
				.map(|v|new_traffic(TrafficBuilderArgument{cv:v,rng:&mut arg.rng,..arg})).collect()),
		);
		let list=list.expect("There were no list");
		assert!( !list.is_empty() , "cannot sum 0 traffics" );
		let size = list[0].number_tasks();
		for traffic in list.iter().skip(1)
		{
			assert_eq!( traffic.number_tasks(), size , "In SumTraffic all sub-traffics must involve the same number of tasks." );
		}
		Sum{
			list
		}
	}
}

/**
Traffic which is another shifted by some amount of tasks.
First check whether a transformation at the `Pattern` level is enough.
The task `index+shift` will be seen as just `index` by the inner traffic.
```ignore
ShiftedTraffic{
	traffic: HomogeneousTraffic{...},
	shift: 50,
}
```
**/
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Shifted
{
	///The amount of the shift in tasks.
	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 Shifted
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut 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,task:usize) -> f32
	{
		self.traffic.probability_per_cycle(task-self.shift)
	}
	fn try_consume(&mut self, task:usize, message: Rc<Message>, cycle:Time, topology:&dyn Topology, rng: &mut 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(task,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()
	}
	fn task_state(&self, task:usize, cycle:Time) -> TaskTrafficState
	{
		self.traffic.task_state(task-self.shift,cycle)
	}

	fn number_tasks(&self) -> usize {
		// TODO: think if this is correct.
		self.traffic.number_tasks()
	}
}

impl Shifted
{
	pub fn new(mut arg:TrafficBuilderArgument) -> Shifted
	{
		let mut shift=None;
		let mut traffic=None;
		match_object_panic!(arg.cv,"ShiftedTraffic",value,
			"traffic" => traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,rng:&mut arg.rng,..arg})),
			"shift" => shift=Some(value.as_f64().expect("bad value for shift") as usize),
		);
		let shift=shift.expect("There were no shift");
		let traffic=traffic.expect("There were no traffic");
		Shifted{
			shift,
			traffic,
			generated_messages: BTreeMap::new(),
		}
	}
}

/**
The tasks in a ProductTraffic are grouped in blocks of size `block_size`. The traffic each block generates follows the underlying `block_traffic` [Traffic],
but with the group of destination being indicated by the `global_pattern`.
First check whether a transformation at the [Pattern] level is enough; specially see the [crate::pattern::ProductPattern] pattern.

```ignore
ProductTraffic{
	block_size: 10,
	block_traffic: HomogeneousTraffic{...},
	global_pattern: RandomPermutation,
}
```
**/
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct ProductTraffic
{
	block_size: usize,
	block_traffic: Box<dyn Traffic>,
	global_pattern: Box<dyn Pattern>,
	global_size: usize,
	///Set of generated messages.
	generated_messages: BTreeMap<*const Message,Rc<Message>>,
}

impl Traffic for ProductTraffic
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut 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,task:usize) -> f32
	{
		let local=task % self.block_size;
		self.block_traffic.probability_per_cycle(local)
	}
	fn try_consume(&mut self, task:usize, message: Rc<Message>, cycle:Time, topology:&dyn Topology, rng: &mut 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(task,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()
	}
	fn task_state(&self, task:usize, cycle:Time) -> TaskTrafficState
	{
		let local=task % self.block_size;
		self.block_traffic.task_state(local,cycle)
	}

	fn number_tasks(&self) -> usize {
		self.block_traffic.number_tasks() * self.global_size
	}
}

impl ProductTraffic
{
	pub fn new(mut arg:TrafficBuilderArgument) -> ProductTraffic
	{
		let mut block_size=None;
		let mut block_traffic=None;
		let mut global_pattern=None;
		match_object_panic!(arg.cv,"ProductTraffic",value,
			"block_traffic" => block_traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,rng:&mut arg.rng,..arg})),
			"global_pattern" => global_pattern=Some(new_pattern(PatternBuilderArgument{cv:value,plugs:arg.plugs})),
			"block_size" => block_size=Some(value.as_f64().expect("bad value for block_size") as usize),
		);
		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");
		// TODO: should receive a `global_size` argument. When missing, fall back to use topology size.
		// TODO: Also check for divisibility.
		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,
			global_size,
			generated_messages: BTreeMap::new(),
		}
	}
}

/**
Only allowed tasks in range will generate messages. The messages can go out of the given range.

```ignore
SubRangeTraffic{
	start: 100,
	end: 200,
	traffic: HomogeneousTraffic{...},
}
```
**/
#[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:Time, topology:&dyn Topology, rng: &mut 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,task:usize) -> f32
	{
		self.traffic.probability_per_cycle(task)
	}
	fn try_consume(&mut self, task:usize, message: Rc<Message>, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> bool
	{
		self.traffic.try_consume(task,message,cycle,topology,rng)
	}
	fn is_finished(&self) -> bool
	{
		self.traffic.is_finished()
	}
	fn task_state(&self, task:usize, cycle:Time) -> TaskTrafficState
	{
		self.traffic.task_state(task,cycle)
	}

	fn number_tasks(&self) -> usize {
		self.traffic.number_tasks()
	}
}

impl SubRangeTraffic
{
	pub fn new(mut arg:TrafficBuilderArgument) -> SubRangeTraffic
	{
		let mut start=None;
		let mut end=None;
		let mut traffic=None;
		match_object_panic!(arg.cv,"SubRangeTraffic",value,
			"traffic" => traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,rng:&mut arg.rng,..arg})),
			"start" => start=Some(value.as_f64().expect("bad value for start") as usize),
			"end" => end=Some(value.as_f64().expect("bad value for end") as usize),
		);
		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 task.
The traffic will be considered complete when all tasks have generated their messages and all of them have been consumed.

```ignore
Burst{
	pattern:Uniform,
	tasks:1000,
	messages_per_task:200,
	message_size: 16,
}
```
**/
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Burst
{
	///Number of tasks applying this traffic.
	tasks: usize,
	///The pattern of the communication.
	pattern: Box<dyn Pattern>,
	///The size of each sent message.
	message_size: usize,
	///The number of messages each task has pending to sent.
	pending_messages: Vec<usize>,
	///Set of generated messages.
	generated_messages: BTreeSet<*const Message>,
}

impl Traffic for Burst
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> Result<Rc<Message>,TrafficError>
	{
		if origin>=self.tasks
		{
			//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,
			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, task:usize) -> f32
	{
		if self.pending_messages[task]>0
		{
			1.0
		}
		else
		{
			0.0
		}
	}
	fn try_consume(&mut self, _task:usize, message: Rc<Message>, _cycle:Time, _topology:&dyn Topology, _rng: &mut 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.is_empty()
		{
			return false;
		}
		for &pm in self.pending_messages.iter()
		{
			if pm>0
			{
				return false;
			}
		}
		true
	}
	fn task_state(&self, task:usize, _cycle:Time) -> TaskTrafficState
	{
		if self.pending_messages[task]>0 {
			TaskTrafficState::Generating
		} else {
			//We do not know whether someone is sending us data.
			//if self.is_finished() { TaskTrafficState::Finished } else { TaskTrafficState::UnspecifiedWait }
			// Sometimes it could be Finished, but it is not worth computing...
			TaskTrafficState::FinishedGenerating
		}
	}

	fn number_tasks(&self) -> usize {
		self.tasks
	}
}

impl Burst
{
	pub fn new(arg:TrafficBuilderArgument) -> Burst
	{
		let mut tasks=None;
		let mut messages_per_task=None;
		let mut pattern=None;
		let mut message_size=None;
		match_object_panic!(arg.cv,"Burst",value,
			"pattern" => pattern=Some(new_pattern(PatternBuilderArgument{cv:value,plugs:arg.plugs})),
			"tasks" | "servers" => tasks=Some(value.as_f64().expect("bad value for tasks") as usize),
			"messages_per_task" | "messages_per_server" => messages_per_task=Some(value.as_f64().expect("bad value for messages_per_task") as usize),
			"message_size" => message_size=Some(value.as_f64().expect("bad value for message_size") as usize),
		);
		let tasks=tasks.expect("There were no tasks");
		let message_size=message_size.expect("There were no message_size");
		let messages_per_task=messages_per_task.expect("There were no messages_per_task");
		let mut pattern=pattern.expect("There were no pattern");
		pattern.initialize(tasks, tasks, arg.topology, arg.rng);
		Burst{
			tasks,
			pattern,
			message_size,
			pending_messages:vec![messages_per_task;tasks],
			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 task that consumed the action message. The destination of the reaction message is independent of the origin of the action message. The two traffics must involve the same number of tasks.
**/
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Reactive
{
	action_traffic: Box<dyn Traffic>,
	reaction_traffic: Box<dyn Traffic>,
	pending_messages: Vec<VecDeque<Rc<Message>>>,
}


impl Traffic for Reactive
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut 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, task:usize) -> f32
	{
		if task<self.pending_messages.len() && !self.pending_messages[task].is_empty()
		{
			return 1.0;
		}
		return self.action_traffic.probability_per_cycle(task);
	}
	fn try_consume(&mut self, task:usize, message: Rc<Message>, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> bool
	{
		if self.action_traffic.try_consume(task,message.clone(),cycle,topology,rng)
		{
			if self.reaction_traffic.should_generate(message.origin,cycle,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(task,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.is_empty()
			{
				return false;
			}
		}
		return true;
	}
	fn task_state(&self, task:usize, cycle:Time) -> TaskTrafficState
	{
		use TaskTrafficState::*;
		let action_state = self.action_traffic.task_state(task,cycle);
		if let Finished = action_state
		{
			return Finished
		}
		let reaction_state = self.reaction_traffic.task_state(task,cycle);
		if let Finished = reaction_state
		{
			return Finished
		}
		if self.is_finished() { Finished } else { UnspecifiedWait }
	}

	fn number_tasks(&self) -> usize {
		// Both traffics have the same number of tasks
		self.action_traffic.number_tasks()
	}
}

impl Reactive
{
	pub fn new(mut arg:TrafficBuilderArgument) -> Reactive
	{
		let mut action_traffic=None;
		let mut reaction_traffic=None;
		match_object_panic!(arg.cv,"Reactive",value,
			"action_traffic" => action_traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,rng:&mut arg.rng,..arg})),
			"reaction_traffic" => reaction_traffic=Some(new_traffic(TrafficBuilderArgument{cv:value,rng:&mut arg.rng,..arg})),
		);
		let action_traffic=action_traffic.expect("There were no action_traffic");
		let reaction_traffic=reaction_traffic.expect("There were no reaction_traffic");
		assert_eq!( action_traffic.number_tasks() , reaction_traffic.number_tasks(), "In Reactive both subtraffics should involve the same number of tasks." );
		Reactive{
			action_traffic,
			reaction_traffic,
			pending_messages:vec![],
		}
	}
}



/**
Selects the traffic from a sequence depending on current cycle. This traffics is useful to make sequences of traffics that do no end by themselves.

All the subtraffics in `traffics` must give the same value for `number_tasks`, which is also used for TimeSequenced. At least one such subtraffic must be provided.

```ignore
TimeSequenced{
	traffics: [HomogeneousTraffic{...}, HomogeneousTraffic{...}],
	times: [2000, 15000],
}
```
**/
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct TimeSequenced
{
	///List of applicable traffics.
	traffics: Vec<Box<dyn Traffic>>,
	///End time of each traffic. Counting from the end of the previous one.
	times: Vec<Time>,
}

impl Traffic for TimeSequenced
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> Result<Rc<Message>,TrafficError>
	{
		let mut offset = cycle;
		let mut traffic_index = 0;
		while traffic_index<self.traffics.len() && offset >= self.times[traffic_index]
		{
			offset -= self.times[traffic_index];
			traffic_index += 1;
		}
		assert!(traffic_index<self.traffics.len());
		self.traffics[traffic_index].generate_message(origin,cycle,topology,rng)
	}
	fn probability_per_cycle(&self,_task:usize) -> f32
	{
		//Can we do better here?
		1.0
	}
	fn try_consume(&mut self, task:usize, message: Rc<Message>, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> bool
	{
		for traffic in self.traffics.iter_mut()
		{
			if traffic.try_consume(task,message.clone(),cycle,topology,rng)
			{
				return true;
			}
		}
		return false;
	}
	fn is_finished(&self) -> bool
	{
		//This is a bit silly for a time sequence
		for traffic in self.traffics.iter()
		{
			if !traffic.is_finished()
			{
				return false;
			}
		}
		return true;
	}
	fn should_generate(&self, task:usize, cycle:Time, rng: &mut StdRng) -> bool
	{
		let mut offset = cycle;
		let mut traffic_index = 0;
		while traffic_index<self.traffics.len() && offset >= self.times[traffic_index]
		{
			offset -= self.times[traffic_index];
			traffic_index += 1;
		}
		if traffic_index<self.traffics.len(){
			self.traffics[traffic_index].should_generate(task,cycle,rng)
		} else {
			false
		}
	}
	fn task_state(&self, task:usize, cycle:Time) -> TaskTrafficState
	{
		let mut offset = cycle;
		let mut traffic_index = 0;
		while traffic_index<self.traffics.len() && offset >= self.times[traffic_index]
		{
			offset -= self.times[traffic_index];
			traffic_index += 1;
		}
		if traffic_index == self.traffics.len()
		{
			return TaskTrafficState::Finished;
		}
		let state = self.traffics[traffic_index].task_state(task,cycle);
		if let TaskTrafficState::Finished = state {
			TaskTrafficState::WaitingCycle { cycle:self.times[traffic_index] }
		} else {
			state
		}
	}

	fn number_tasks(&self) -> usize {
		// each traffic has the same number of tasks
		self.traffics[0].number_tasks()
	}
}

impl TimeSequenced
{
	pub fn new(mut arg:TrafficBuilderArgument) -> TimeSequenced
	{
		let mut traffics : Option<Vec<_>> =None;
		let mut times=None;
		match_object_panic!(arg.cv,"TimeSequenced",value,
			"traffics" => traffics = Some(value.as_array().expect("bad value for traffics").iter()
				.map(|v|new_traffic(TrafficBuilderArgument{cv:v,rng:&mut arg.rng,..arg})).collect()),
			"times" => times = Some(value.as_array()
				.expect("bad value for times").iter()
				.map(|v|v.as_time().expect("bad value in times")).collect()),
		);
		let traffics=traffics.expect("There were no traffics");
		assert!( !traffics.is_empty() , "Cannot make a TimeSequenced of 0 traffics." );
		let size = traffics[0].number_tasks();
		for traffic in traffics.iter().skip(1)
		{
			assert_eq!( traffic.number_tasks(), size , "In TimeSequenced all sub-traffics must involve the same number of tasks." );
		}
		let times=times.expect("There were no times");
		TimeSequenced{
			traffics,
			times,
		}
	}
}

/**
A sequence of traffics. When a traffic declares itself to be finished moves to the next.

All the subtraffics in `traffics` must give the same value for `number_tasks`, which is also used for Sequence. At least one such subtraffic must be provided.

```ignore
Sequence{
	traffics: [Burst{...}, Burst{...}],
}
```
**/
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct Sequence
{
	///List of applicable traffics.
	traffics: Vec<Box<dyn Traffic>>,
	//How many times to apply the whole traffic period. default to 1.
	//period_limit: usize,
	///The traffic which is currently in use.
	current_traffic: usize,
	//The period number, starting at 0. The whole traffic finishes before `current_period` reaching `period_limit`.
	//current_period: usize,
}

impl Traffic for Sequence
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> Result<Rc<Message>,TrafficError>
	{
		while self.traffics[self.current_traffic].is_finished()
		{
			self.current_traffic += 1;
			//self.current_traffic = (self.current_traffic + 1) % self.traffics.len();
		}
		assert!(self.current_traffic<self.traffics.len());
		self.traffics[self.current_traffic].generate_message(origin,cycle,topology,rng)
	}
	fn probability_per_cycle(&self,task:usize) -> f32
	{
		self.traffics[self.current_traffic].probability_per_cycle(task)
	}
	fn try_consume(&mut self, task:usize, message: Rc<Message>, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> bool
	{
		for traffic in self.traffics.iter_mut()
		{
			if traffic.try_consume(task,message.clone(),cycle,topology,rng)
			{
				while self.current_traffic < self.traffics.len() && self.traffics[self.current_traffic].is_finished()
				{
					//self.current_traffic = (self.current_traffic + 1) % self.traffics.len();
					self.current_traffic += 1;
				}
				return true;
			}
		}
		return false;
	}
	fn is_finished(&self) -> bool
	{
		//return current_period == period_limit;
		return self.current_traffic>=self.traffics.len() || (self.current_traffic==self.traffics.len()-1 && self.traffics[self.current_traffic].is_finished())
	}
	fn should_generate(&self, task:usize, cycle:Time, rng: &mut StdRng) -> bool
	{
		if self.current_traffic>=self.traffics.len()
		{
			false
		} else {
			self.traffics[self.current_traffic].should_generate(task,cycle,rng)
		}
	}
	fn task_state(&self, task:usize, cycle:Time) -> TaskTrafficState
	{
		use TaskTrafficState::*;
		if self.current_traffic>=self.traffics.len()
		{
			Finished
		} else {
			let state = self.traffics[self.current_traffic].task_state(task,cycle);
			if let Finished=state{
				UnspecifiedWait
			} else {
				state
			}
			//In the last traffic we could try to check for FinishedGenerating
		}
	}

	fn number_tasks(&self) -> usize {
		// every traffic has the same number of tasks
		self.traffics[0].number_tasks()
	}
}

impl Sequence
{
	pub fn new(arg:TrafficBuilderArgument) -> Sequence
	{
		let mut traffics_args =None;
		let mut period_number=1usize;
		match_object_panic!(arg.cv,"Sequence",value,
			"traffics" => traffics_args = Some(value.as_array().expect("bad value for traffics")),
			"period_number" => period_number=value.as_f64().expect("bad value for period_number") as usize,
		);
		let traffics_args=traffics_args.expect("There were no traffics");
		let TrafficBuilderArgument{plugs,topology,rng, ..} = arg;
		let traffics : Vec<_> = (0..period_number).flat_map(|_ip| traffics_args.iter().map(
			|v|new_traffic(TrafficBuilderArgument{cv:v,plugs,topology,rng:&mut *rng})
		).collect::<Vec<_>>() ).collect();
		//let mut traffics = Vec::with_capacity(period_number*traffics_args.len());
		//for _ip in 0..period_number
		//{
		//	for v in traffics_args
		//	{
		//		//traffics.push( new_traffic(TrafficBuilderArgument{cv:v,..arg}) );
		//		traffics.push( new_traffic(TrafficBuilderArgument{cv:v,plugs,topology,rng}) );
		//	}
		//}
		assert!( !traffics.is_empty() , "Cannot make a Sequence of 0 traffics." );
		let size = traffics[0].number_tasks();
		for traffic in traffics.iter().skip(1)
		{
			assert_eq!( traffic.number_tasks(), size , "In Sequence all sub-traffics must involve the same number of tasks." );
		}
		Sequence{
			traffics,
			current_traffic:0,
			//current_period:0,
		}
	}
}

/// Like the `Burst` pattern, but generating messages from different patterns and with different message sizes.
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct MultimodalBurst
{
	///Number of tasks applying this traffic.
	tasks: usize,
	/// For each kind of message `provenance` we have
	/// `(pattern,total_messages,message_size,step_size)`
	/// a Pattern deciding the destination of the message
	/// a usize with the total number of messages of this kind that each task must generate
	/// a usize with the size of each message size.
	/// a usize with the number of messages to send of this kind before switching to the next one.
	provenance: Vec< (Box<dyn Pattern>,usize,usize,usize) >,
	///For each task and kind we track `pending[task][kind]=(total_remaining,step_remaining)`.
	///where `total_remaining` is the total number of messages of this kind that this task has yet to send.
	///and `step_remaining` is the number of messages that the task will send before switch to the next kind.
	pending: Vec<Vec<(usize,usize)>>,
	///For each task we track which provenance kind is the next one.
	///If for the annotated provenance there is not anything else to send then use the next one.
	next_provenance: Vec<usize>,
	///Set of generated messages.
	generated_messages: BTreeSet<*const Message>,
}

impl Traffic for MultimodalBurst
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> Result<Rc<Message>,TrafficError>
	{
		if origin>=self.tasks
		{
			//panic!("origin {} does not belong to the traffic",origin);
			return Err(TrafficError::OriginOutsideTraffic);
		}
		let pending = &mut self.pending[origin];
		// Determine the kind to use.
		let mut provenance_index = self.next_provenance[origin];
		loop
		{
			let (ref mut total_remaining, ref mut step_remaining) = pending[provenance_index];
			if *total_remaining > 0
			{
				*step_remaining -=1;
				*total_remaining -=1;
				if *step_remaining == 0
				{
					//When the whole step is performed advance `next_provenance`.
					let (ref _pattern, _total_messages, _message_size, step_size) = self.provenance[provenance_index];
					*step_remaining = step_size;
					self.next_provenance[origin] = (provenance_index+1) % pending.len();
				}
				break;
			}
			provenance_index = (provenance_index+1) % pending.len();
		}
		// Build the message
		let (ref pattern,_total_messages,message_size,_step_size) = self.provenance[provenance_index];
		let destination=pattern.get_destination(origin,topology,rng);
		if origin==destination
		{
			return Err(TrafficError::SelfMessage);
		}
		let message = Rc::new(Message{
			origin,
			destination,
			size:message_size,
			creation_cycle: cycle,
		});
		self.generated_messages.insert(message.as_ref() as *const Message);
		Ok(message)
	}
	fn probability_per_cycle(&self, task:usize) -> f32
	{
		for (total_remaining,_step_remaining) in self.pending[task].iter()
		{
			if *total_remaining > 0
			{
				return 1.0;
			}
		}
		0.0
	}
	fn try_consume(&mut self, _task:usize, message: Rc<Message>, _cycle:Time, _topology:&dyn Topology, _rng: &mut 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.is_empty()
		{
			return false;
		}
		for task_pending in self.pending.iter()
		{
			for (total_remaining, _step_remaining) in task_pending.iter()
			{
				if *total_remaining > 0
				{
					return false;
				}
			}
		}
		true
	}
	fn task_state(&self, task:usize, _cycle:Time) -> TaskTrafficState
	{
		if self.pending[task].iter().any(|(total_remaining,_step_remaining)| *total_remaining > 0 ) {
			TaskTrafficState::Generating
		} else {
			//We do not know whether someone is sending us data.
			//if self.is_finished() { TaskTrafficState::Finished } else { TaskTrafficState::UnspecifiedWait }
			// Sometimes it could be Finished, but it is not worth computing...
			TaskTrafficState::FinishedGenerating
		}
	}

	fn number_tasks(&self) -> usize {
		self.tasks
	}
}

impl MultimodalBurst
{
	pub fn new(arg:TrafficBuilderArgument) -> MultimodalBurst
	{
		let mut tasks=None;
		let mut provenance : Option<Vec<(_,_,_,_)>> = None;
		match_object_panic!(arg.cv,"MultimodalBurst",value,
			"tasks" | "servers" => tasks=Some(value.as_f64().expect("bad value for tasks") as usize),
			"provenance" => match value
			{
				&ConfigurationValue::Array(ref a) => provenance=Some(a.iter().map(|pcv|{
					let mut messages_per_task=None;
					let mut pattern=None;
					let mut message_size=None;
					let mut step_size=None;
					match_object_panic!(pcv,"Provenance",pvalue,
						"pattern" => pattern=Some(new_pattern(PatternBuilderArgument{cv:pvalue,plugs:arg.plugs})),
						"messages_per_task" | "messages_per_server" | "total_messages" =>
							messages_per_task=Some(pvalue.as_f64().expect("bad value for messages_per_task") as usize),
						"message_size" => message_size=Some(pvalue.as_f64().expect("bad value for message_size") as usize),
						"step_size" => step_size=Some(pvalue.as_f64().expect("bad value for step_size") as usize),
					);
					let pattern=pattern.expect("There were no pattern");
					let messages_per_task=messages_per_task.expect("There were no messages_per_task");
					let message_size=message_size.expect("There were no message_size");
					let step_size=step_size.expect("There were no step_size");
					(pattern,messages_per_task,message_size,step_size)
				}).collect()),
				_ => panic!("bad value for provenance"),
			}
		);
		let tasks=tasks.expect("There were no tasks");
		let mut provenance=provenance.expect("There were no provenance");
		for (pattern,_total_messages,_message_size,_step_size) in provenance.iter_mut()
		{
			pattern.initialize(tasks, tasks, arg.topology, arg.rng);
		}
		let each_pending = provenance.iter().map(|(_pattern,total_messages,_message_size,step_size)|(*total_messages,*step_size)).collect();
		MultimodalBurst{
			tasks,
			provenance,
			pending: vec![each_pending;tasks],
			next_provenance:vec![0;tasks],
			generated_messages: BTreeSet::new(),
		}
	}
}


///In this traffic each task has a limited amount of data that can send over the amount it has received.
///For example, with `bound=1` after a task sends a message it must wait to receive one.
///And if received `x` messages then it may generate `x+bound` before having to wait.
///All messages have same size, follow the same pattern.
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct BoundedDifference
{
	///Number of tasks applying this traffic.
	tasks: 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,
	///The number of messages each task may generate over the amount it has received.
	bound: usize,
	///Set of generated messages.
	generated_messages: BTreeSet<*const Message>,
	///The number of messages each task is currently allowed to generate until they consume more.
	///It is initialized to `bound`.
	allowance: Vec<usize>,
}

impl Traffic for BoundedDifference
{
	fn generate_message(&mut self, origin:usize, cycle:Time, topology:&dyn Topology, rng: &mut StdRng) -> Result<Rc<Message>,TrafficError>
	{
		if origin>=self.tasks
		{
			//panic!("origin {} does not belong to the traffic",origin);
			return Err(TrafficError::OriginOutsideTraffic);
		}
		assert!(self.allowance[origin]>0,"Origin {} has no allowance to send more messages.",origin);
		let destination=self.pattern.get_destination(origin,topology,rng);
		if origin==destination
		{
			return Err(TrafficError::SelfMessage);
		}
		self.allowance[origin]-=1;
		let message=Rc::new(Message{
			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, task:usize) -> f32
	{
		if self.allowance[task]>0
		{
			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
			}
		} else { 0f32 }
	}
	fn try_consume(&mut self, task:usize, message: Rc<Message>, _cycle:Time, _topology:&dyn Topology, _rng: &mut StdRng) -> bool
	{
		let message_ptr=message.as_ref() as *const Message;
		self.allowance[task]+=1;
		self.generated_messages.remove(&message_ptr)
	}
	fn is_finished(&self) -> bool
	{
		false
	}
	fn task_state(&self, task:usize, _cycle:Time) -> TaskTrafficState
	{
		if self.allowance[task]>0 {
			TaskTrafficState::Generating
		} else {
			TaskTrafficState::WaitingData
		}
	}

	fn number_tasks(&self) -> usize {
		self.tasks
	}
}

impl BoundedDifference
{
	pub fn new(arg:TrafficBuilderArgument) -> BoundedDifference
	{
		let mut tasks=None;
		let mut load=None;
		let mut pattern=None;
		let mut message_size=None;
		let mut bound=None;
		match_object_panic!(arg.cv,"BoundedDifference",value,
			"pattern" => pattern=Some(new_pattern(PatternBuilderArgument{cv:value,plugs:arg.plugs})),
			"tasks" | "servers" => tasks=Some(value.as_f64().expect("bad value for tasks") as usize),
			"load" => load=Some(value.as_f64().expect("bad value for load") as f32),
			"message_size" => message_size=Some(value.as_f64().expect("bad value for message_size") as usize),
			"bound" => bound=Some(value.as_f64().expect("bad value for bound") as usize),
		);
		let tasks=tasks.expect("There were no tasks");
		let message_size=message_size.expect("There were no message_size");
		let bound=bound.expect("There were no bound");
		let load=load.expect("There were no load");
		let mut pattern=pattern.expect("There were no pattern");
		pattern.initialize(tasks, tasks, arg.topology, arg.rng);
		BoundedDifference{
			tasks,
			pattern,
			message_size,
			load,
			bound,
			generated_messages: BTreeSet::new(),
			allowance: vec![bound;tasks],
		}
	}
}


/**
Applies a map over the tasks of a traffic. The source and destination sets may differ. A simple example is to shuffle the tasks in a application, as in the following configuration.

```ignore
TrafficMap{
	tasks: 1000,
	application: HomogeneousTraffic{...},
	map: RandomPermutation,
}
```

TrafficMap also gives the possibility of seeing a small application as a large, helping in the composition of large applications.
The following example uses TrafficMap together with [TrafficSum](Sum),
[CartesianEmbedding](crate::pattern::CartesianEmbedding), [Composition](crate::pattern::Composition),
and [CartesianTransform](crate::pattern::CartesianTransform) to divide the network into
two regions, each employing a different kind of traffic.
```ignore
TrafficSum
{
	list: [
		TrafficMap{
			tasks: 150,
			map: CartesianEmbedding{
				source_sides: [3,5,5],
				destination_sides: [3,10,5],
			},
			application: HomogeneousTraffic{
				pattern: Uniform,
				tasks: 75,
				load: 1.0,
				message_size: 16,
			},
		},
		TrafficMap{
			tasks: 150,
			map: Composition{patterns:[
				CartesianEmbedding{
					source_sides: [3,5,5],
					destination_sides: [3,10,5],
				},
				CartesianTransform{
					sides: [3,10,5],
					shift: [0,5,0],
				},
			]},
			application: Burst{
				pattern: Uniform,
				tasks: 75,
				message_size: 16,
				messages_per_task: 100,
			},
		},
	],
},
```

The `map` is computed once when the traffic is created. Thus, it is recommended for the [Pattern] indicated by `map` to be idempotent.

Currently, the `map` is required to be injective. This is, two tasks must not be mapped into a single one. This restriction could be lifted in the future.
**/
#[derive(Quantifiable)]
#[derive(Debug)]
pub struct TrafficMap
{
	// src_machine -> src_app -> dst_app -> dst_machine

	/// Maps the origin of the traffic.
	/// (source_machine -> source_app)
	from_machine_to_app: Vec<Option<usize>>,

	/// Maps the destination of the traffic.
	/// (destination_app -> destination_machine)
	from_app_to_machine: Vec<usize>,

	/// The traffic to be mapped.
	/// (source_app -> destination_app)
	application: Box<dyn Traffic>,

	/// The number of tasks in the traffic.
	number_tasks: usize,

	/// The map to be applied to the traffic.
	map: Box<dyn Pattern>,

	///Set of generated messages.
	generated_messages: BTreeMap<*const Message,Rc<Message>>,
}

impl Traffic for TrafficMap
{
	fn generate_message(&mut self, origin: usize, cycle: Time, topology: &dyn Topology, rng: &mut StdRng) -> Result<Rc<Message>, TrafficError>
	{
		// the machine origin of the message
		if origin >= self.from_machine_to_app.len()
		{
			return Err(TrafficError::OriginOutsideTraffic);
		}

		// Get the origin of the message (the app) from the base map
		let app_origin = self.from_machine_to_app[origin].expect("There was no origin for the message");

		// generate the message from the application
		let app_message = self.application.generate_message(app_origin, cycle, topology, rng)?;

		let app_destination = app_message.destination;

		// build the message
		let message = Rc::new(Message{
			origin,
			destination: self.from_app_to_machine[app_destination], // get the destination of the message (the machine) from the base map
			size: app_message.size,
			creation_cycle: app_message.creation_cycle,
		});
		self.generated_messages.insert(message.as_ref() as *const Message, app_message);
		Ok(message)
	}

	fn probability_per_cycle(&self, task: usize) -> f32
	{
		// The probability of a task is the same as the probability of the task in the application
		let task_app = self.from_machine_to_app[task];

		task_app.map(|app| {
			// get the probability of the task in the application
			self.application.probability_per_cycle(app)
		}).unwrap_or(0.0) // if the task_app has no origin, it has no probability
	}

	fn should_generate(&self, task: usize, cycle: Time, rng: &mut StdRng) -> bool {
		let task_app = self.from_machine_to_app[task];

		task_app.map(|app| {
			self.application.should_generate(app, cycle, rng)
		}).unwrap_or(false)
	}


	fn try_consume(&mut self, task: usize, message: Rc<Message>, cycle: Time, topology: &dyn Topology, rng: &mut StdRng) -> bool
	{
		// TODO: Maybe we want to return a Result instead of a bool
		let message_ptr = message.as_ref() as *const Message;
		let app_message = match self.generated_messages.remove(&message_ptr)
		{
			Some(app_message) => app_message,
			None => return false,
		};

		let task_app = self.from_machine_to_app[task].expect("There was no origin for the message");

		// try to consume the message in the application
		self.application.try_consume(task_app, app_message, cycle, topology, rng)
	}

	fn is_finished(&self) -> bool
	{
		self.application.is_finished()
	}

	fn task_state(&self, task: usize, cycle: Time) -> TaskTrafficState
	{
		let task_app = self.from_machine_to_app[task];

		task_app.map(|app| {
			// get the state of the task in the application
			self.application.task_state(app, cycle)
		}).unwrap_or(TaskTrafficState::Finished) // if the task has no origin, it is finished

	}

	fn number_tasks(&self) -> usize {
		self.number_tasks
	}
}


impl TrafficMap
{
	pub fn new(mut arg:TrafficBuilderArgument) -> TrafficMap
	{
		let mut application = None;
		let mut map = None;
		let mut number_tasks = None;
		match_object_panic!(arg.cv,"TrafficMap",value,
			"tasks" => number_tasks=Some(value.as_f64().expect("bad value for tasks") as usize),
			"application" => application = Some(new_traffic(TrafficBuilderArgument{cv:value,rng:&mut arg.rng,..arg})), //traffic of the application
			"map" => map = Some(new_pattern(PatternBuilderArgument{cv:value,plugs:arg.plugs})), //map of the application over the machine
		);

		let number_tasks = number_tasks.expect("There were no tasks in configuration of TrafficMap.");
		let application = application.expect("There were no application in configuration of TrafficMap.");
		let mut map = map.expect("There were no map in configuration of TrafficMap.");

		let app_tasks = application.number_tasks();
		map.initialize(app_tasks, number_tasks, arg.topology, arg.rng);

		let from_app_to_machine: Vec<_> = (0..app_tasks).map(|inner_origin| {
			map.get_destination(inner_origin, arg.topology, arg.rng)
		}).collect();

		// from_machine_to_app is the inverse of from_app_to_machine
		let mut from_machine_to_app = vec![None; number_tasks];
		for i in 0..app_tasks
		{
			from_machine_to_app[from_app_to_machine[i]] = Some(i);
		}

		TrafficMap
		{
			application,
			from_machine_to_app,
			from_app_to_machine,
			number_tasks,
			map,
			generated_messages: BTreeMap::new(),
		}
	}
}