fn as_slice(&self) -> &[u32] {

        self.0.as_slice()

    }

    fn iter(&self) -> impl Iterator<Item = u32> + '_ {

        self.0.iter().copied()

    }

    fn is_empty(&self) -> bool {

        // relies on the invariant: no trailing zero u32's

        self.0.is_empty()

    }

    fn clear(&mut self) {

        self.0.clear();

    }

    fn set_ones_until(&mut self, mut end: usize) {

        self.0.clear();

        loop {

            if end >= 32 {

                // full 32 bits of 1

                self.0.push(!0u32);

            } else {

                if end > 0 {

                    // #end ones, with a (32-end) prefix of zeroes

                    self.0.push(!0u32 >> (32 - end));

                }

                return;

            }

        }

    }

#[derive(Default)]

struct Ecs {

    entities: Vec<Entity>,

    assignments: HashMap<(ChannelId, bool), BitSet>,

    ekeys: HashMap<Key, BitSet>,

    csb: ComponentStatusBits,

}

#[derive(Default)]

struct ComponentStatusBits {

    inconsistent: BitSet,

    blocked: BitSet,

    sync_ended: BitSet,

    to_run_r: BitSet, // read from and drained while...

    to_run_w: BitSet, // .. written to and populated.

}

impl ComponentStatusBits {

    fn clear_all(&mut self) {

        self.blocked.clear();

        self.inconsistent.clear();

        self.to_run_r.clear();

        self.to_run_w.clear();

        self.sync_ended.clear();

    }

}

struct Msg {

    assignments: Vec<(ChannelId, bool)>, // invariant: no two elements have same ChannelId value

    payload: Payload,

}

impl Ecs {

    fn round(&mut self) {

        // 1. at the start of the round we throw away all assignments.

        //    we are going to shift entities around, so all bitsets need to be cleared anyway.

        self.assignments.clear();

        self.csb.clear_all();

        self.ekeys.clear();

        // 2. We discard all payloads; they are all stale now.

        //    All components are now contiguous in the vector.

        self.entities.retain(|entity| if let Entity::State(_) = entity { true } else { false });

        // 3. initially, all the components need a chance to run in MONO mode

        self.csb.to_run_r.set_ones_until(self.entities.len());

        // 4. INVARIANT established:

        //    for all State variants in self.entities,

        //        exactly one bit throughout the fields of csb is set.

        // 5. Run all machines in (csb.to_run_r U csb.to_run_w).

        //    Single, logical set is broken into readable / writable parts to allow concurrent reads / writes safely.

        while !self.csb.to_run_r.is_empty() {

            for _entity_index in self.csb.to_run_r.iter() {

                // TODO run and possbibly manipulate self.to_run_w

            }

            self.csb.to_run_r.clear();

            std::mem::swap(&mut self.csb.to_run_r, &mut self.csb.to_run_w);

        }

        assert!(self.csb.to_run_w.is_empty());

        #[allow(unreachable_code)] // DEBUG

        'recv_loop: loop {

            let ekey: Key = todo!();

            let msg: Msg = todo!();

            // 1. check if this message is redundant, i.e., there is already an equivalent payload with predicate >= this one.

            //    ie. starting from all payloads

            // 2. try and find a payload whose predicate subsumes this one

            if let Some(_entity_index) = self.ekeys.get(&ekey).map(|ekey_bitset| {

                let mut slice_builder = vec![];

                for &(channel_id, boolean) in msg.assignments.iter() {

                    if let Some(bitset) = self.assignments.get(&(channel_id, !boolean)) {

                        slice_builder.push(bitset.as_slice());

                    }

                }

                NoStricterPayloadIter {

                    next_chunk_index: 0,

                    in_here: ekey_bitset.as_slice(),

                    but_in_none_of: slice_builder.as_slice(),

                }

                .next()

            }) {

                continue 'recv_loop;

            }

            // receive incoming messages

        }

    }

}

struct NoStricterPayloadIter<'a> {

    next_chunk_index: usize,

    in_here: &'a [u32],

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

}

impl<'a> Iterator for NoStricterPayloadIter<'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.in_here.get(i).copied();

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

            let a = folding?;

            let b = slice.get(i).copied()?;

            Some(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. flags: {}

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