use cellular_raza_concepts::IndexError;
use itertools::Itertools;
use std::{
collections::{BTreeMap, BTreeSet},
marker::PhantomData,
sync::atomic::AtomicBool,
};
#[cfg(feature = "tracing")]
use tracing::instrument;
use super::errors::SimulationError;
pub struct BarrierSync {
barrier: hurdles::Barrier,
got_error: std::sync::Arc<AtomicBool>,
}
#[cfg_attr(feature = "tracing", instrument(skip_all))]
pub fn validate_map<I>(map: &std::collections::BTreeMap<I, BTreeSet<I>>) -> bool
where
I: Eq + core::hash::Hash + Clone + Ord,
{
for (index, neighbors) in map.iter() {
if neighbors.iter().any(|n| match map.get(n) {
Some(reverse_neighbors) => !reverse_neighbors.contains(index),
None => true,
}) {
return false;
}
}
true
}
pub struct UDGraph<I>(pub(crate) Vec<(I, I)>);
impl<I> UDGraph<I> {
pub fn new() -> Self {
Self(Vec::new())
}
pub fn push(&mut self, new_connection: (I, I)) -> Option<(I, I)>
where
I: PartialEq,
{
if new_connection.0 == new_connection.1 {
return Some(new_connection);
}
if self
.0
.iter()
.any(|connection| connection == &new_connection)
{
return Some(new_connection);
}
self.0.push(new_connection);
None
}
pub fn extend<J>(&mut self, new_connections: J) -> Vec<(I, I)>
where
I: PartialEq,
J: IntoIterator<Item = (I, I)>,
{
new_connections
.into_iter()
.filter_map(|new_connection| self.push(new_connection))
.collect()
}
pub fn clear(&mut self) {
self.0.clear()
}
pub fn drain<R>(&mut self, range: R) -> std::vec::Drain<'_, (I, I)>
where
R: core::ops::RangeBounds<usize>,
{
self.0.drain(range)
}
pub fn nodes(&self) -> Vec<&I>
where
I: Clone + Eq + core::hash::Hash,
{
self.0
.iter()
.map(|(c1, c2)| [c1, c2].into_iter())
.flatten()
.unique()
.collect()
}
}
impl<I> IntoIterator for UDGraph<I> {
type Item = (I, I);
type IntoIter = std::vec::IntoIter<(I, I)>;
fn into_iter(self) -> std::vec::IntoIter<(I, I)> {
self.0.into_iter()
}
}
impl<I> UDGraph<I>
where
I: core::hash::Hash + Clone + Eq + Ord,
{
pub fn to_btree(self) -> BTreeMap<I, BTreeSet<I>> {
let mut map: BTreeMap<_, BTreeSet<I>> = self
.0
.iter()
.map(|x| {
[
(x.0.clone(), BTreeSet::new()),
(x.1.clone(), BTreeSet::new()),
]
.into_iter()
})
.flatten()
.collect();
self.0.iter().for_each(|(c1, c2)| {
map.entry(c1.clone()).and_modify(|v| {
v.insert(c2.clone());
});
map.entry(c2.clone()).and_modify(|v| {
v.insert(c1.clone());
});
});
map
}
}
impl<I> core::ops::Deref for UDGraph<I> {
type Target = Vec<(I, I)>;
fn deref(&self) -> &Self::Target {
&self.0
}
}
impl<I> From<UDGraph<I>> for PhantomData<I> {
#[allow(unused)]
fn from(value: UDGraph<I>) -> Self {
PhantomData
}
}
pub trait BuildFromGraph<I>
where
Self: Sized,
I: Clone + Eq + core::hash::Hash + Ord,
{
fn build_from_graph(graph: UDGraph<I>) -> Result<BTreeMap<I, Self>, IndexError>;
}
impl<I> BuildFromGraph<I> for PhantomData<I>
where
Self: Sized,
I: Clone + Eq + core::hash::Hash + Ord,
{
fn build_from_graph(graph: UDGraph<I>) -> Result<BTreeMap<I, Self>, IndexError> {
Ok(graph
.into_iter()
.map(|(key, _)| (key, PhantomData::<I>))
.collect())
}
}
pub trait FromMap<I>
where
Self: Sized,
{
fn from_map(map: &BTreeMap<I, BTreeSet<I>>) -> Result<BTreeMap<I, Self>, IndexError>
where
I: Eq + core::hash::Hash + Clone + Ord;
}
impl<I> FromMap<I> for PhantomData<I> {
fn from_map(map: &BTreeMap<I, BTreeSet<I>>) -> Result<BTreeMap<I, Self>, IndexError>
where
I: Eq + core::hash::Hash + Clone + Ord,
{
Ok(map
.into_iter()
.map(|(key, _)| (key.clone(), PhantomData::<I>))
.collect())
}
}
pub trait SyncSubDomains {
fn sync(&mut self) -> Result<(), SimulationError>;
fn store_error(
&mut self,
maybe_error: Result<(), SimulationError>,
) -> Result<bool, SimulationError>;
}
impl<I> FromMap<I> for BarrierSync {
fn from_map(map: &BTreeMap<I, BTreeSet<I>>) -> Result<BTreeMap<I, Self>, IndexError>
where
I: Eq + core::hash::Hash + Clone + Ord,
{
let barrier = hurdles::Barrier::new(map.len());
let got_error = std::sync::Arc::new(AtomicBool::new(false));
let res = map
.keys()
.map(|i| {
(
i.clone(),
Self {
barrier: barrier.clone(),
got_error: std::sync::Arc::clone(&got_error),
},
)
})
.collect();
Ok(res)
}
}
impl<I> BuildFromGraph<I> for BarrierSync
where
I: Clone + Eq + core::hash::Hash + std::cmp::Ord,
{
fn build_from_graph(graph: UDGraph<I>) -> Result<BTreeMap<I, Self>, IndexError> {
let barrier = hurdles::Barrier::new(graph.len());
let got_error = std::sync::Arc::new(AtomicBool::new(false));
let res = graph
.nodes()
.into_iter()
.map(|key| {
(
key.clone(),
Self {
barrier: barrier.clone(),
got_error: std::sync::Arc::clone(&got_error),
},
)
})
.collect();
Ok(res)
}
}
impl SyncSubDomains for BarrierSync {
fn sync(&mut self) -> Result<(), SimulationError> {
self.barrier.wait();
match self.got_error.load(std::sync::atomic::Ordering::Relaxed) {
true => Err(SimulationError::OtherThreadError(format!(
"Another thread returned an error. Winding down."
))),
false => Ok(()),
}
}
fn store_error(
&mut self,
maybe_error: Result<(), SimulationError>,
) -> Result<bool, SimulationError> {
match maybe_error {
Ok(_) => Ok(false),
Err(SimulationError::OtherThreadError(_)) => Ok(true),
Err(x) => {
self.got_error
.store(true, std::sync::atomic::Ordering::Relaxed);
self.barrier.wait();
Err(x)
}
}
}
}
pub trait Communicator<I, T>
where
Self: Sized,
{
fn send(&mut self, receiver: &I, message: T) -> Result<(), SimulationError>;
fn receive(&mut self) -> Vec<T>;
}
#[derive(Clone)]
pub struct ChannelComm<I, T> {
senders: std::collections::BTreeMap<I, crossbeam_channel::Sender<T>>,
receiver: crossbeam_channel::Receiver<T>,
}
impl<T, I> FromMap<I> for ChannelComm<I, T>
where
I: Ord,
{
fn from_map(map: &BTreeMap<I, BTreeSet<I>>) -> Result<BTreeMap<I, Self>, IndexError>
where
I: Clone + core::hash::Hash + Eq,
{
let channels: BTreeMap<_, _> = map
.keys()
.into_iter()
.map(|sender_key| {
let (s, r) = crossbeam_channel::unbounded::<T>();
(sender_key, (s, r))
})
.collect();
let mut comms = BTreeMap::new();
for key in map.keys().into_iter() {
let senders = map
.get(&key)
.ok_or(IndexError("Network of communicators could not be constructed due to incorrect entries in map".into()))?
.clone()
.into_iter()
.map(|connected_key| (connected_key.clone(), channels[&connected_key].0.clone())).collect();
let comm = ChannelComm {
senders,
receiver: channels[&key].1.clone(),
};
comms.insert(key.clone(), comm);
}
Ok(comms)
}
}
#[cfg(test)]
mod test_channel_comm {
use itertools::Itertools;
use super::*;
#[test]
fn test_from_map() -> Result<(), IndexError> {
let map = BTreeMap::from([
(0_usize, BTreeSet::from([3, 1])),
(1_usize, BTreeSet::from([0, 2])),
(2_usize, BTreeSet::from([1, 3])),
(3_usize, BTreeSet::from([2, 0])),
]);
assert!(validate_map(&map));
let channel_comms = ChannelComm::<usize, ()>::from_map(&map)?;
assert_eq!(channel_comms.len(), 4);
for i in 0..4 {
assert!(channel_comms.keys().contains(&i));
}
for i in 0..4 {
let higher = (i + 1) % 4;
let lower = (i + 5) % 4;
assert!(channel_comms[&i].senders.contains_key(&lower));
assert!(channel_comms[&i].senders.contains_key(&higher));
}
Ok(())
}
#[test]
fn test_from_map_2() -> Result<(), Box<dyn std::error::Error>> {
let map = BTreeMap::from([
(0, BTreeSet::from([1, 2, 3])),
(1, BTreeSet::from([0, 2, 3])),
(2, BTreeSet::from([0, 1, 3])),
(3, BTreeSet::from([0, 1, 2])),
]);
assert!(validate_map(&map));
let channel_comms = ChannelComm::<usize, Option<()>>::from_map(&map)?;
for i in 0..4 {
assert!(channel_comms.keys().contains(&i));
}
for i in 0..4 {
for j in 0..4 {
let contains = channel_comms[&i].senders.keys().contains(&j);
assert_eq!(contains, i != j);
}
}
Ok(())
}
fn from_map_n_line(n: usize) -> Result<(), Box<dyn std::error::Error>> {
if n <= 1 {
return Ok(());
}
let mut map =
BTreeMap::from([(0, BTreeSet::from([1, n])), (n, BTreeSet::from([n - 1, 0]))]);
for i in 1..n {
map.insert(i, BTreeSet::from([i - 1, i + 1]));
}
assert!(validate_map(&map));
let mut channel_comms = ChannelComm::<usize, usize>::from_map(&map)?;
channel_comms
.iter_mut()
.map(|(key, comm)| {
let recv_key = (key + 1) % (n + 1);
comm.send(&recv_key, *key)
})
.collect::<Result<Vec<_>, _>>()?;
channel_comms.iter_mut().for_each(|(key, comm)| {
let results = comm.receive();
assert!(results.len() > 0);
if key > &0 {
assert_eq!(results, vec![key - 1]);
} else {
assert_eq!(results, vec![n]);
}
});
Ok(())
}
#[test]
fn from_map_lines() {
for i in 2..100 {
from_map_n_line(i).unwrap();
}
}
#[test]
fn test_send() -> Result<(), Box<dyn std::error::Error>> {
let map = BTreeMap::from([
(1_usize, BTreeSet::from([2])),
(2_usize, BTreeSet::from([1])),
]);
let mut channel_comms = ChannelComm::<usize, bool>::from_map(&map)?;
channel_comms.get_mut(&1).unwrap().send(&2, true)?;
channel_comms.get_mut(&2).unwrap().send(&1, false)?;
Ok(())
}
#[test]
fn test_empty_receive() -> Result<(), Box<dyn std::error::Error>> {
let map = BTreeMap::from([
(1_usize, BTreeSet::from([2])),
(2_usize, BTreeSet::from([1])),
]);
let mut channel_comms = ChannelComm::<usize, f64>::from_map(&map)?;
for (_, comm) in channel_comms.iter_mut() {
let received_elements = comm.receive().into_iter().collect::<Vec<_>>();
assert_eq!(received_elements.len(), 0);
}
Ok(())
}
#[test]
fn test_send_receive() -> Result<(), Box<dyn std::error::Error>> {
let map = BTreeMap::from([
(0, BTreeSet::from([1, 2])),
(1, BTreeSet::from([0, 2])),
(2, BTreeSet::from([0, 1])),
]);
let mut channel_comms = ChannelComm::from_map(&map)?;
for (index, comm) in channel_comms.iter_mut() {
let next_index = (index + 1) % map.len();
comm.send(&next_index, next_index as f64)?;
}
for (index, comm) in channel_comms.iter_mut() {
let received_elements = comm.receive().into_iter().collect::<Vec<_>>();
assert_eq!(received_elements, vec![*index as f64]);
}
Ok(())
}
#[test]
fn test_send_plain_voxel() -> Result<(), Box<dyn std::error::Error>> {
use crate::backend::chili::SubDomainPlainIndex;
let map = BTreeMap::from([
(
SubDomainPlainIndex(0),
BTreeSet::from([SubDomainPlainIndex(1), SubDomainPlainIndex(2)]),
),
(
SubDomainPlainIndex(1),
BTreeSet::from([SubDomainPlainIndex(0), SubDomainPlainIndex(2)]),
),
(
SubDomainPlainIndex(2),
BTreeSet::from([SubDomainPlainIndex(0), SubDomainPlainIndex(1)]),
),
]);
let mut channel_comms = ChannelComm::from_map(&map)?;
for (index, comm) in channel_comms.iter_mut() {
let index = index.0;
let next_index = SubDomainPlainIndex((index + 1) % map.len());
comm.send(&next_index, next_index)?;
}
for (index, comm) in channel_comms.iter_mut() {
let received_elements = comm.receive().into_iter().collect::<Vec<_>>();
assert_eq!(received_elements, vec![*index]);
}
Ok(())
}
}
impl<I, T> Communicator<I, T> for ChannelComm<I, T>
where
I: core::hash::Hash + Eq + Ord,
{
fn receive(&mut self) -> Vec<T> {
self.receiver.try_iter().collect()
}
fn send(&mut self, receiver: &I, message: T) -> Result<(), SimulationError> {
let sender = self
.senders
.get(&receiver)
.ok_or(super::IndexError(format!(
"could not find specified receiver"
)))?;
sender.send(message)?;
Ok(())
}
}
#[doc(hidden)]
#[allow(unused)]
mod test_derive_communicator {
fn default() {}
fn two_communicators_explicit() {}
fn two_communicators_generic_one() {}
}
#[doc(hidden)]
#[allow(unused)]
mod test_derive_from_map {
fn default() {}
fn generic_index() {}
fn where_clause() {}
fn where_clause_on_index() {}
}
#[doc(hidden)]
#[allow(unused)]
mod test_build_communicator {
macro_rules! test_build_communicator(
(
name:$func_name:ident,
aspects:[$($asp:ident),*]
) => {
#[doc = stringify!($($asp),*)]
#[doc = concat!($(
concat!("test_aspect!(", stringify!($asp), ");")
,)*)]
#[allow(non_snake_case)]
fn $func_name () {}
};
);
cellular_raza_core_proc_macro::run_test_for_aspects!(
test: test_build_communicator,
aspects: [Mechanics, Interaction, ReactionsContact]
);
fn without_path() {}
}
#[cfg(test)]
mod test_sync {
use super::*;
use std::sync::*;
fn test_single_map<S>(map: BTreeMap<usize, BTreeSet<usize>>)
where
S: 'static + SyncSubDomains + FromMap<usize> + Send + Sync,
{
let n_iterations = 1_000;
let n_threads = map.len();
let iteration_counter =
Arc::new(Mutex::new(Vec::from_iter((0..n_threads).map(|_| 0_usize))));
let syncers = S::from_map(&map).unwrap();
let handles = syncers
.into_iter()
.map(|(n_thread, mut syncer)| {
let iteration_counter_thread = Arc::clone(&iteration_counter);
std::thread::spawn(move || {
for n_iteration in 0..n_iterations {
syncer.sync().unwrap();
iteration_counter_thread.lock().unwrap()[n_thread] += 1;
syncer.sync().unwrap();
let current_value = iteration_counter_thread.lock().unwrap().clone();
assert_eq!(current_value, vec![n_iteration + 1; n_threads]);
}
})
})
.collect::<Vec<_>>();
for handle in handles.into_iter() {
handle.join().unwrap();
}
}
fn test_multiple_maps<S>()
where
S: 'static + SyncSubDomains + FromMap<usize> + Send + Sync,
{
let map0 = BTreeMap::from_iter([(0, BTreeSet::from([1])), (1, BTreeSet::from([0]))]);
test_single_map::<S>(map0);
let map1 = BTreeMap::from_iter([
(0, BTreeSet::from([1, 2])),
(1, BTreeSet::from([0, 2])),
(2, BTreeSet::from([0, 1])),
]);
test_single_map::<S>(map1);
let map2 = BTreeMap::from_iter([
(0, BTreeSet::from([1, 2, 3])),
(1, BTreeSet::from([0, 2, 3])),
(2, BTreeSet::from([0, 1, 3])),
(3, BTreeSet::from([0, 1, 2])),
]);
test_single_map::<S>(map2);
let map3 = BTreeMap::from_iter([
(0, BTreeSet::from([1, 2])),
(1, BTreeSet::from([0, 3])),
(2, BTreeSet::from([0, 3])),
(3, BTreeSet::from([1, 2])),
]);
test_single_map::<S>(map3);
let map4 = BTreeMap::from_iter([
(0, BTreeSet::from([1])),
(1, BTreeSet::from([2])),
(2, BTreeSet::from([3])),
(3, BTreeSet::from([4])),
(4, BTreeSet::from([0])),
]);
test_single_map::<BarrierSync>(map4);
}
#[test]
fn barrier_sync() {
test_multiple_maps::<BarrierSync>();
}
}