}

    fn machine_assignment_for(&self, machine_eid: usize, channel_id: ChannelId) -> Option<bool> {

        let test = move |bitset: &BitSet| bitset.test(machine_eid);

        self.ecs

            .assignments

            .get(&(channel_id, true))

            .map(test)

            .or_else(move || self.ecs.assignments.get(&(channel_id, false)).map(test))

    }

    fn feed_msg(&mut self, payload_eid: usize, ekey: usize) {

        // 1. identify the component who:

        //    * is blocked on this ekey,

        //    * and has a predicate at least as strict as that of this payload

        let mut slice_builder = vec![];

        let ekey_bitset =

            self.ecs.ekeys.get_mut(&ekey).expect("Payload sets this => cannot be empty");

        slice_builder.push(ekey_bitset.as_slice());

        for bitset in self.ecs.assignments.values() {

            // it doesn't matter which assignment! just that this payload sets it too

            if bitset.test(payload_eid) {

                slice_builder.push(bitset.as_slice());

            }

        }

        let chunk_iter =

            InAllExceptIter::new(slice_builder.as_slice(), self.ecs.payloads.as_slice());

        let mut iter = BitChunkIter::new(chunk_iter);

        if let Some(machine_eid) = iter.next() {

            // TODO is it possible for there to be 2+ iterations? I'm thinking No

            // RUN THIS MACHINE

            ekey_bitset.unset(machine_eid);

            self.ecs.to_run_w.set(machine_eid);

        }

    }

}

struct InAllExceptIter<'a> {

    next_chunk_index: usize,

    in_all: &'a [&'a [u32]],

    except: &'a [u32],

}

impl<'a> InAllExceptIter<'a> {

    fn new(in_all: &'a [&'a [u32]], except: &'a [u32]) -> Self {

        Self { in_all, except, next_chunk_index: 0 }

    }

}

impl<'a> Iterator for InAllExceptIter<'a> {

    type Item = u32;

    fn next(&mut self) -> Option<Self::Item> {

        let i = self.next_chunk_index;

        self.next_chunk_index += 1;

        let init = self.except.get(i).map(move |&x| !x).or(Some(1));

        self.in_all.iter().fold(init, move |folding, slice| {

            let a = folding?;

            let b = slice.get(i).copied().unwrap_or(0);

            Some(a & !b)

        })

    }

}

struct InNoneExceptIter<'a> {

    next_chunk_index: usize,

    in_none: &'a [&'a [u32]],

    except: &'a [u32],

}

impl<'a> InNoneExceptIter<'a> {

    fn new(in_none: &'a [&'a [u32]], except: &'a [u32]) -> Self {

        Self { in_none, except, next_chunk_index: 0 }

    }

}

impl<'a> Iterator for InNoneExceptIter<'a> {

    type Item = u32;

    fn next(&mut self) -> Option<Self::Item> {

        let i = self.next_chunk_index;

        self.next_chunk_index += 1;

        let init = self.except.get(i).copied()?;

        Some(self.in_none.iter().fold(init, move |folding, slice| {

            let a = folding;

            let b = slice.get(i).copied().unwrap_or(0);

            a & !b

        }))

    }

}

/*

The idea is we have a set of component machines that fork whenever they reflect on the oracle to make concrete their predicates.

their speculative execution procedure BLOCKS whenever they reflect on the contents of a message that has not yet arrived.

the promise is, therefore, never to forget about these blocked machines.

the only event that unblocks a machine

operations needed:

1. FORK

given a component and a predicate,

create and retain a clone of the component, and the predicate, with one additional assignment

2. GET

when running a machine with {state S, predicate P}, it may try to get a message at K.

IF there exists a payload at K with predicate P2 s.t. P2 >= P, feed S the message and continue.

ELSE list (S,P,K) as BLOCKED and...

for all payloads X at K with predicate P2 s.t. P2 < P, fork S to create S2. store it with predicate P2 and feed it X and continue

2. RECV

when receiving a payload at key K with predicate P,

IF there exists a payload at K with predicate P2 where P2 >= P, discard the new one and continue.

ELSE if there exists a payload at K with predicate P2 where P2 < P, assert their contents are identical, overwrite P2 with P try feed this payload to any BLOCKED machines

ELSE insert this payload with P and K as a new payload, and feed it to any compatible machines blocked on K

====================

EXTREME approach:

1. entities: {states} U {payloads}

2. ecs: {}

==================

*/

impl Debug for FlagMatrix {

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

        for r in 0..self.dims[0] {

            write!(f, "|")?;

            for c in 0..self.dims[1] {

                write!(

                    f,

                    "{}",

                    match self.test([r, c]) {

                        false => '0',

                        true => '1',

                    }

                )?;

            }

            write!(f, "|\n")?;

        }

        Ok(())

    }

}

struct FlagMatrix {

    bytes: *mut u32,

    u32s_total: usize,

    u32s_per_row: usize,

    dims: [usize; 2],

}

#[inline(always)]

fn ceiling_to_mul_32(value: usize) -> usize {

    (value + 31) & !31

}

impl Drop for FlagMatrix {

    fn drop(&mut self) {

        let layout = Self::layout_for(self.u32s_total);

        unsafe {

            //?

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

        }

    }

}

impl FlagMatrix {

    fn get_dims(&self) -> &[usize; 2] {

        &self.dims

    }

    fn reshape(&mut self, new_dims: [usize; 2]) {

        dbg!(self.u32s_total, self.u32s_per_row);

        // 1. calc new u32s_per_row

        let new_u32s_per_row = match ceiling_to_mul_32(new_dims[1]) / 32 {

            min if min > self.u32s_per_row => Some(min * 2),

            _ => None,

        };

        // 2. calc new u32s_total

        let new_u32s_total = match new_u32s_per_row.unwrap_or(self.u32s_per_row) * new_dims[0] {

            min if min > self.u32s_total => Some(min * 2),

            _ => None,

        };

        dbg!(new_u32s_per_row, new_u32s_total);

        match [new_u32s_per_row, new_u32s_total] {

            [None, None] => { /* do nothing */ }

            [None, Some(new_u32s_total)] => {

                assert!(new_u32s_total > self.u32s_total);

                // realloc only!

                dbg!("REALLOC ONLY");

                println!("BEFORE {:?}", self.bytes);

                let old_layout = Self::layout_for(self.u32s_total);

                let new_layout = Self::layout_for(new_u32s_total);

                let new_bytes = unsafe {

                    let new_bytes = std::alloc::alloc(new_layout) as *mut u32;

                    // copy the previous total

                    self.bytes.copy_to_nonoverlapping(new_bytes, self.u32s_total);

                    // and zero the remainder

                    // drop the previous buffer

                    std::alloc::dealloc(self.bytes as *mut u8, old_layout);

                    new_bytes

                };

                self.bytes = new_bytes;

                println!("AFTER {:?}", self.bytes);

                self.u32s_total = new_u32s_total;

            }

            [Some(new_u32s_per_row), None] => {

                assert!(new_u32s_per_row > self.u32s_per_row);

                // shift only!

                dbg!("SHIFT ONLY");

                for r in (0..self.dims[0]).rev() {

                    // iterate in REVERSE order because new row[n] may overwrite old row[n+m]

                    unsafe {

                        // copy the used prefix

                        src.copy_to(dest, self.u32s_per_row);

                        // and zero the remainder

                    }

                }

                self.u32s_per_row = new_u32s_per_row;

            }

            [Some(new_u32s_per_row), Some(new_u32s_total)] => {

                assert!(new_u32s_total > self.u32s_total);

                assert!(new_u32s_per_row > self.u32s_per_row);

                // alloc AND shift!

                dbg!("BOTH");

                let old_layout = Self::layout_for(self.u32s_total);

                let new_layout = Self::layout_for(new_u32s_total);

                let new_bytes = unsafe { std::alloc::alloc(new_layout) as *mut u32 };

                for r in 0..self.dims[0] {

                    // iterate forwards over rows!

                    unsafe {

                        // copy the used prefix

                        src.copy_to_nonoverlapping(dest, self.u32s_per_row);

                        // and zero the remainder

                }

                }

                unsafe { std::alloc::dealloc(self.bytes as *mut u8, old_layout) };

                self.u32s_per_row = new_u32s_per_row;

                self.bytes = new_bytes;

                self.u32s_total = new_u32s_total;

            }

        }

        self.dims = new_dims;

    }

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

            // this layout is ALWAYS valid:

            // 1. size is always nonzero

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

    }

    fn new(dims: [usize; 2], extra_dim_space: [usize; 2]) -> Self {

        let u32s_per_row = ceiling_to_mul_32(dims[1] + extra_dim_space[1]) / 32; // HALF dead columns

        let u32s_total = u32s_per_row * (dims[0] + extra_dim_space[0]);

        let layout = Self::layout_for(u32s_total);

        let bytes = unsafe {

            // allocate

            let bytes = std::alloc::alloc(layout) as *mut u32;

            // and zero

            bytes.write_bytes(0u8, u32s_total);

            bytes

        };

        Self { bytes, u32s_total, u32s_per_row, dims }

    }

    fn assert_within_bounds(&self, at: [usize; 2]) {

        assert!(at[0] < self.dims[0]);

        assert!(at[1] < self.dims[1]);

    }

    #[inline(always)]

        (self.u32s_per_row * at[0]) + at[1] / 32

    }

    #[inline(always)]

        let in_chunk = at[1] % 32;

        [of_chunk, in_chunk]

    }

    fn set(&mut self, at: [usize; 2]) {

        self.assert_within_bounds(at);

    }

    fn unset(&mut self, at: [usize; 2]) {

        self.assert_within_bounds(at);

    }

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

        self.assert_within_bounds(at);

    }

}

#[derive(Debug, Copy, Clone)]

enum ColumnCombinator<'a> {

    Row(usize),

    True,

    False,

    And(&'a ColumnCombinator<'a>, &'a ColumnCombinator<'a>),

    Or(&'a ColumnCombinator<'a>, &'a ColumnCombinator<'a>),

    Not(&'a ColumnCombinator<'a>),

}

struct FlaggedColumnIter<'a> {

    flag_matrix: &'a FlagMatrix,

    next_column_chunk: usize,

    combinator: &'a ColumnCombinator<'a>,

}

impl<'a> FlaggedColumnIter<'a> {

    fn new(flag_matrix: &'a FlagMatrix, combinator: &'a ColumnCombinator<'a>) -> Self {

        Self { flag_matrix, combinator, next_column_chunk: 0 }

    }

    /// #Safety: bounds on self.next_column_chunk have been checked with self.flag_matrix

    /// retrieves the column chunk at self.next_column_chunk

    unsafe fn combine(&self, c: &ColumnCombinator) -> u32 {

        use ColumnCombinator as Cc;

        match c {

            Cc::Row(row) => self.flag_matrix.copy_chunk_unchecked(*row, self.next_column_chunk),

            Cc::False => 0u32,

            Cc::True => !0u32,

            Cc::And(a, b) => self.combine(a) & self.combine(b),

            Cc::Or(a, b) => self.combine(a) | self.combine(b),

            Cc::Not(a) => !self.combine(a),

        }

    }

}

impl<'a> Iterator for FlaggedColumnIter<'a> {

    type Item = u32;

    fn next(&mut self) -> Option<Self::Item> {

        struct CombineCtx<'a> {

            flag_matrix: &'a FlagMatrix,

            nth_col_chunk: usize,

        }

        if self.next_column_chunk >= self.flag_matrix.u32s_per_row {

            None

        } else {

            let x = unsafe { self.combine(self.combinator) };

            self.next_column_chunk += 1;

            Some(x)

        }

    }

}

struct ColumnIter<'a> {

    bit_chunk_iter: BitChunkIter<FlaggedColumnIter<'a>>,

}

impl<'a> ColumnIter<'a> {

    fn new(m: &'a FlagMatrix, combinator: &'a ColumnCombinator) -> Self {

        let iter = FlaggedColumnIter::new(m, combinator);

        let bit_chunk_iter = BitChunkIter::new(iter);

        Self { bit_chunk_iter }

    }

}

impl<'a> Iterator for ColumnIter<'a> {

    type Item = usize;

    fn next(&mut self) -> Option<Self::Item> {

        let v: Option<usize> = self.bit_chunk_iter.next();

        v.filter(|&x| x < self.bit_chunk_iter.chunk_iter.flag_matrix.dims[1])

    }

}

#[test]

fn matrix() {

    let mut m = FlagMatrix::new([5, 5], [0, 0]);

    for i in 0..5 {

    }

    println!("{:?}", &m);

    m.reshape([6, 40]);

    println!("{:?}", &m);

}