[package]

name = "reowolf_rs"

version = "0.1.1"

authors = [

	"Christopher Esterhuyse <christopher.esterhuyse@gmail.com>",

	"Hans-Dieter Hiep <hdh@cwi.nl>"

]

edition = "2018"

[dependencies]

# hibitset = "0.6.2"

# runtime stuff

derive_more = "0.99.2"

getrandom = "0.1.14" # tiny crate. used to guess controller-id

take_mut = "0.2.2"

maplit = "1.0.2" # convenience macros

indexmap = "1.3.0" # hashsets/hashmaps with efficient arbitrary element removal

# network stuff

integer-encoding = "1.0.7"

byteorder = "1.3.2"

mio = "0.6.21" # migrate to mio 0.7.0 when it stabilizes. It's much better.

mio-extras = "2.0.6"

# protocol stuff

id-arena = "2.2.1"

backtrace = "0.3"

[dev-dependencies]

test-generator = "0.3.0"

crossbeam-utils = "0.7.0"

[lib]

crate-type = ["cdylib"]

[features]

default = ["ffi"]

ffi = [] # no feature dependencies

use super::vec_storage::VecStorage;

use crate::common::*;

use crate::runtime::endpoint::EndpointExt;

use crate::runtime::endpoint::EndpointInfo;

use crate::runtime::endpoint::{Endpoint, Msg, SetupMsg};

use crate::runtime::errors::EndpointErr;

use crate::runtime::errors::MessengerRecvErr;

use crate::runtime::errors::PollDeadlineErr;

use crate::runtime::MessengerState;

use crate::runtime::Messengerlike;

use crate::runtime::ReceivedMsg;

use crate::runtime::{ProtocolD, ProtocolS};

use std::net::SocketAddr;

use std::sync::Arc;

pub enum Coupling {

    Active,

    Passive,

}

#[derive(Debug)]

struct Family {

    parent: Option<Port>,

    children: HashSet<Port>,

}

pub struct Binding {

    pub coupling: Coupling,

    pub polarity: Polarity,

    pub addr: SocketAddr,

}

pub struct InPort(Port); // InPort and OutPort are AFFINE (exposed to Rust API)

pub struct OutPort(Port);

impl From<InPort> for Port {

    fn from(x: InPort) -> Self {

        x.0

    }

}

impl From<OutPort> for Port {

    fn from(x: OutPort) -> Self {

        x.0

    }

}

#[derive(Default, Debug)]

struct ChannelIndexStream {

                parent, controller_id

            ),

        }

        // 3. broadcast leader announcement (except to parent: confirm they are your parent)

        //    in this loop, every node sends 1 message to each neighbor

        let msg_for_non_parents = S(LeaderAnnounce { leader: my_leader });

        for n in neighbors.clone() {

            let msg =

                if Some(n) == parent { S(YouAreMyParent) } else { msg_for_non_parents.clone() };

            messenger.send(n, msg).map_err(|e| EndpointErr(n, e))?;

        }

        // await 1 message from all non-parents

        for n in neighbors.clone() {

            if Some(n) != parent {

                awaiting.insert(n);

            }

        }

        let mut children = HashSet::default();

        messenger.undelay_all();

        while !awaiting.is_empty() {

            let ReceivedMsg { recipient, msg } = messenger.recv_until(deadline)?.ok_or(Timeout)?;

            let recipient = recipient;

            match msg {

                S(YouAreMyParent) => {

                    assert!(awaiting.remove(&recipient));

                    children.insert(recipient);

                }

                S(SetupMsg::LeaderAnnounce { leader }) => {

                    assert!(awaiting.remove(&recipient));

                    assert!(leader == my_leader);

                    assert!(Some(recipient) != parent);

                    // they wouldn't send me this if they considered me their parent

                }

                _ => messenger.delay(ReceivedMsg { recipient, msg }),

            }

        }

        let family = Family { parent, children };

        // done!

        Ok(Connected {

            components: Default::default(),

            controller_id,

            channel_index_stream,

            endpoint_exts,

            native_ports,

            family,

        })

    }

    /////////

    pub fn connect_using_id(

        &mut self,

        controller_id: ControllerId,

        timeout: Option<Duration>,

    ) -> Result<Connected, ConnectErr> {

        // 1. try and create a connection from these bindings with self immutable.

        let connected = self.new_connected(controller_id, timeout)?;

        // 2. success! drain self and return

        self.bindings.clear();

        Ok(connected)

    }

    pub fn connect(&mut self, timeout: Option<Duration>) -> Result<Connected, ConnectErr> {

        self.connect_using_id(Self::random_controller_id(), timeout)

    }

}

#[derive(Debug)]

pub struct Connected {

    native_ports: HashSet<Port>,

    controller_id: ControllerId,

    channel_index_stream: ChannelIndexStream,

    endpoint_exts: VecStorage<EndpointExt>,

    components: VecStorage<Component>,

    family: Family,

}

impl Connected {

    pub fn new_component(

        &mut self,

        protocol: &Arc<ProtocolD>,

        identifier: &Arc<[u8]>,

        moved_port_list: &[Port],

    ) -> Result<(), MainComponentErr> {

        //////////////////////////////////////////

        // 1. try and create a new component (without mutating self)

        use MainComponentErr::*;

        let moved_port_set = {

            let mut set: HashSet<Port> = Default::default();

            for &port in moved_port_list.iter() {

                if !self.native_ports.contains(&port) {

                    return Err(CannotMovePort(port));

                }

                if !set.insert(port) {

                    return Err(DuplicateMovedPort(port));

                }

            }

            set

        };

        // moved_port_set is disjoint to native_ports

        let expected_polarities = protocol.component_polarities(identifier)?;

        if moved_port_list.len() != expected_polarities.len() {

            return Err(WrongNumberOfParamaters { expected: expected_polarities.len() });

        }

        // correct polarity list

        for (param_index, (&port, &expected_polarity)) in

            moved_port_list.iter().zip(expected_polarities.iter()).enumerate()

        {

            let polarity =

                self.endpoint_exts.get_occupied(port.0).ok_or(UnknownPort(port))?.info.polarity;

            if polarity != expected_polarity {

                return Err(WrongPortPolarity { param_index, port });

            }

        }

        let state = protocol.new_main_component(identifier, &moved_port_list);

        let component = Component {

            port_set: moved_port_set,

            protocol: protocol.clone(),

            identifier: identifier.clone(),

            state,

        };

        //////////////////////////////

        // success! mutate self and return Ok

        self.native_ports.retain(|e| !component.port_set.contains(e));

    let mut c = c.connect(None).unwrap();

    let (mem_out, mem_in) = c.new_channel();

    let mut inbuf = [0u8; 64];

    let identifier: Arc<[u8]> = b"sync".to_vec().into();

    c.new_component(&proto_0, &identifier, &[net_in.into(), mem_out.into()]).unwrap();

    let mut ops = [

        PortOpRs::In { msg_range: None, port: &mem_in },

        PortOpRs::Out { msg: b"hey", port: &net_out, optional: false },

        PortOpRs::Out { msg: b"hi?", port: &net_out, optional: true },

        PortOpRs::Out { msg: b"yo!", port: &net_out, optional: false },

    ];

    c.sync_set(&mut inbuf, &mut ops).unwrap();

    c.sync_subsets(&mut inbuf, &mut ops, &[bitslice! {0,1,2}]).unwrap();

}

#[repr(C)]

pub struct PortOp {

    msgptr: *mut u8, // read if OUT, field written if IN, will point into buf

    msglen: usize,   // read if OUT, written if IN, won't exceed buf

    port: Port,

    optional: bool, // no meaning if

}

pub enum PortOpRs<'a> {

    In { msg_range: Option<Range<usize>>, port: &'a InPort },

    Out { msg: &'a [u8], port: &'a OutPort, optional: bool },

}

unsafe fn c_sync_set(

    connected: &mut Connected,

    inbuflen: usize,

    inbufptr: *mut u8,

    opslen: usize,

    opsptr: *mut PortOp,

) -> i32 {

    let buf = as_mut_slice(inbuflen, inbufptr);

    let ops = as_mut_slice(opslen, opsptr);

    let (subset_index, wrote) = sync_inner(connected, buf);

    assert_eq!(0, subset_index);

    for op in ops {

        if let Some(range) = wrote.get(&op.port) {

            op.msgptr = inbufptr.add(range.start);

            op.msglen = range.end - range.start;

        }

    }

    0

}

unsafe fn c_sync_subset(

    connected: &mut Connected,

    inbuflen: usize,

    inbufptr: *mut u8,

    opslen: usize,

    opsptr: *mut PortOp,

    subsetslen: usize,

    subsetsptr: *const *const usize,

) -> i32 {

    let buf: &mut [u8] = as_mut_slice(inbuflen, inbufptr);

    let ops: &mut [PortOp] = as_mut_slice(opslen, opsptr);

    let subsets: &[*const usize] = as_const_slice(subsetslen, subsetsptr);

    let subsetlen = usizes_for_bits(opslen);

    // don't yet know subsetptr; which subset fires unknown!

    let (subset_index, wrote) = sync_inner(connected, buf);

    let subsetptr: *const usize = subsets[subset_index];

    let subset: &[usize] = as_const_slice(subsetlen, subsetptr);

    for index in BitChunkIter::new(subset.iter().copied()) {

        let op = &mut ops[index as usize];

        if let Some(range) = wrote.get(&op.port) {

            op.msgptr = inbufptr.add(range.start);

            op.msglen = range.end - range.start;

        }

    }

    subset_index as i32

}

// dummy fn for the actual synchronous round

fn sync_inner<'c, 'b>(

    _connected: &'c mut Connected,

    _buf: &'b mut [u8],

) -> (usize, &'b HashMap<Port, Range<usize>>) {

    todo!()

}

unsafe fn as_mut_slice<'a, T>(len: usize, ptr: *mut T) -> &'a mut [T] {

    std::slice::from_raw_parts_mut(ptr, len)

}

unsafe fn as_const_slice<'a, T>(len: usize, ptr: *const T) -> &'a [T] {

    std::slice::from_raw_parts(ptr, len)

}

#[test]

fn api_connecting() {

    let addrs: [SocketAddr; 3] = [

        "127.0.0.1:8888".parse().unwrap(),

        "127.0.0.1:8889".parse().unwrap(),

        "127.0.0.1:8890".parse().unwrap(),

    ];

    const TIMEOUT: Option<Duration> = Some(Duration::from_secs(1));

    let handles = vec![

        std::thread::spawn(move || {

            let mut connecting = Connecting::default();

        }),

        std::thread::spawn(move || {

            let mut connecting = Connecting::default();

            let _b: InPort = connecting.bind(Coupling::Passive, addrs[1]);

            let _c: InPort = connecting.bind(Coupling::Active, addrs[2]);

        }),

        std::thread::spawn(move || {

            let mut connecting = Connecting::default();

            let _a: OutPort = connecting.bind(Coupling::Passive, addrs[2]);

        }),

    ];

    for h in handles {

        h.join().unwrap();

    }

}

            // n is [32,16,8,4,2,1] on 64-bit machine

            // this loop is unrolled with release optimizations

            let n_least_significant_mask = (1 << n) - 1;

            if chunk & n_least_significant_mask == 0 {

                // no 1 set within 0..n least significant bits.

                self.next_bit_index += n;

                chunk >>= n;

            }

            n /= 2;

        }

        // least significant bit of chunk is 1. Item to return is known.

        // assert(chunk & 1 == 1)

        // prepare our state for the next time Self::next is called.

        // Overwrite self.cached such that its shifted state is retained,

        // and jump over the bit whose index we are about to return.

        self.next_bit_index += 1;

        self.cached = chunk >> 1;

        // returned index is usize_bits() smaller than self.next_bit_index because we use an

        // off-by-usize_bits() encoding to avoid having to cache an Option<usize>.

        Some(self.next_bit_index - 1 - usize_bits())

    }

}

/*  --properties-->

     ___ ___ ___ ___

    |___|___|___|___|

  | |___|___|___|___|

  | |___|___|___|___|

  | |___|___|___|___|

  |

  V

 entity chunks (groups of size usize_bits())

*/

// TODO newtypes Entity and Property

#[derive(Debug, Copy, Clone, Eq, PartialEq)]

struct Pair {

    entity: u32,

    property: u32,

}

impl From<[u32; 2]> for Pair {

    fn from([entity, property]: [u32; 2]) -> Self {

        Pair { entity, property }

    }

}

    buffer: *mut usize,

    bounds: Pair,

    layout: Layout, // layout of the currently-allocated buffer

}

impl Drop for BitMatrix {

    fn drop(&mut self) {

        unsafe {

            // ?

            std::alloc::dealloc(self.buffer as *mut u8, self.layout);

        }

    }

}

impl Debug for BitMatrix {

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

                write!(f, "|")?;

                    };

                }

            }

        }

    }

}

impl BitMatrix {

    #[inline]

    const fn row_of(entity: usize) -> usize {

        entity / usize_bits()

    }

    #[inline]

    const fn row_chunks(property_bound: usize) -> usize {

        property_bound

    }

    #[inline]

    const fn column_chunks(entity_bound: usize) -> usize {

        usizes_for_bits(entity_bound + 1)

    }

    #[inline]

    fn offsets_unchecked(&self, at: Pair) -> [usize; 2] {

        let o_in = at.entity as usize % usize_bits();

        let row = Self::row_of(at.entity as usize);

        let row_chunks = self.bounds.property as usize;

        let o_of = row * row_chunks + at.property as usize;

        [o_of, o_in]

    }

    // returns a u32 which has bits 000...000111...111

    // for the last JAGGED chunk given the column size

    // if the last chunk is not jagged (when entity_bound % 32 == 0)

    // None is returned,

    // otherwise Some(x) is returned such that x & chunk would mask out

    // the bits NOT in 0..entity_bound

    fn last_row_chunk_mask(entity_bound: u32) -> Option<usize> {

        let zero_prefix_len = entity_bound as usize % usize_bits();

        if zero_prefix_len == 0 {

            None

        } else {

            Some(!0 >> (usize_bits() - zero_prefix_len))

        }

    }

    fn assert_within_bounds(&self, at: Pair) {

        assert!(at.entity < self.bounds.entity);

        assert!(at.property < self.bounds.property);

    }

    fn layout_for(total_chunks: usize) -> std::alloc::Layout {

        unsafe {

            // this layout is ALWAYS valid:

            // 1. size is always nonzero

            // 2. size is always a multiple of 4 and 4-aligned

            Layout::from_size_align_unchecked(usize_bytes() * total_chunks.max(1), usize_bytes())

        BitChunkIter::new(self.0.iter().copied()).map(|x| x as usize)

    }

    fn first(&self) -> Option<usize> {

        self.iter().next()

    }

}

// A T-type arena which:

// 1. does not check for the ABA problem

// 2. imposes the object keys on the user

// 3. allows the reservation of a space (getting the key) to precede the value being provided.

// 4. checks for user error

//

// Data contains values in one of three states:

// 1. occupied: ininitialized. will be dropped.

// 2. vacant: uninitialized. may be reused implicitly. won't be dropped.

// 2. reserved: uninitialized. may be occupied implicitly. won't be dropped.

//

// element access is O(1)

// removal is O(1) amortized.

// insertion is O(N) with a small constant factor,

//    doing at worst one linear probe through N/(word_size) contiguous words

//

// invariant A: data elements are inititalized <=> occupied bit is set

// invariant B: occupied and vacant have an empty intersection

// invariant C: (vacant U occupied) subset of (0..data.len)

// invariant D: last element of data is not in VACANT state

// invariant E: number of allocated bits in vacant and occupied >= data.len()

// invariant F: vacant_bit_count == vacant.iter().count()

pub struct VecStorage<T> {

    data: Vec<MaybeUninit<T>>,

    occupied: Bitvec,

    vacant: Bitvec,

    occupied_bit_count: usize,

}

impl<T> Default for VecStorage<T> {

    fn default() -> Self {

        Self {

            data: Default::default(),

            vacant: Default::default(),

            occupied: Default::default(),

            occupied_bit_count: 0,

        }

    }

}

impl<T: Debug> Debug for VecStorage<T> {

    fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

        enum FmtT<'a, T> {

        };

        impl<T: Debug> Debug for FmtT<'_, T> {

            fn fmt(&self, f: &mut Formatter) -> std::fmt::Result {

                match self {

                }

            }

        }

        let iter = (0..self.data.len()).map(|i| unsafe {

            // 1. data bounds are checked by construction of i.

            // 2. occupied index => valid data is read.

            // 3. bitset bounds are ensured by invariant E.

            if self.occupied.contains(i) {

            } else if self.vacant.contains(i) {

            } else {

            }

        });

        f.debug_list().entries(iter).finish()

    }

}

impl<T> Drop for VecStorage<T> {

    fn drop(&mut self) {

        self.clear();

    }

}

impl<T> VecStorage<T> {

    // ASSUMES that i in 0..self.data.len()

    unsafe fn get_occupied_unchecked(&self, i: usize) -> Option<&T> {

        if self.occupied.contains(i) {

            // 2. Invariant A => reading valid ata

            Some(&*self.data.get_unchecked(i).as_ptr())

        }

    }

    //////////////

    pub fn len(&self) -> usize {

        self.occupied_bit_count

    }

    pub fn with_reserved_range(range_end: usize) -> Self {

        let mut data = Vec::with_capacity(range_end);

        unsafe {

            // data is uninitialized, as intended

            data.set_len(range_end);

        }

        let bitset_len = (range_end + (usize_bits() - 1)) / usize_bits();

        let chunk_iter = std::iter::repeat(0usize).take(bitset_len);

        Self {

            data,

            vacant: Bitvec(chunk_iter.clone().collect()),

            occupied: Bitvec(chunk_iter.collect()),

            occupied_bit_count: 0,

        }

    }

    pub fn clear(&mut self) {

        for i in 0..self.data.len() {

            // SAFE: bitvec bounds ensured by invariant E

            if unsafe { self.occupied.contains(i) } {

                // invariant A: this element is OCCUPIED

                unsafe {

                    // 1. by construction, i is in bounds

                    // 2. i is occupied => initialized data is being dropped

                    drop(self.data.get_unchecked_mut(i).as_ptr().read());

                }

            }

        }

        self.vacant.0.clear();

        self.occupied.0.clear();

        self.occupied_bit_count = 0;

    }

    pub fn iter(&self) -> impl Iterator<Item = &T> {

        (0..self.data.len()).filter_map(move |i| unsafe { self.get_occupied_unchecked(i) })

    }

    pub fn iter_mut(&mut self) -> impl Iterator<Item = &mut T> {

        (0..self.data.len()).filter_map(move |i| unsafe {

            // SAFE: bitvec bounds ensured by invariant E

            if self.occupied.contains(i) {

                // Invariant A => reading valid data

                Some(&mut *self.data.get_unchecked_mut(i).as_mut_ptr())

            } else {

                None