/// Invariant: every component is either:

///        in to_run = (to_run_r U to_run_w)

///     or in ONE of the ekeys (which means it is blocked by a get on that ekey)

///     or in sync_ended (because they reached the end of their sync block)

///     or in inconsistent (because they are inconsistent)

#[derive(Default)]

struct Ecs {

    component_info: Vec<(Arc<Protocol>, HashSet<ChannelId>)>,

    entities: Vec<Entity>,

    round_solution: Vec<(ChannelId, bool)>, // encodes an ASSIGNMENT

    ekey_channel_ids: Vec<ChannelId>,       // all channel Ids for local keys

    flags: EntityFlags,

}

#[derive(Default)]

struct EntityFlags {

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

    payloads: BitSet,

    ekeys: HashMap<Key, BitSet>,

    inconsistent: BitSet,

    sync_ended: BitSet,

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

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

}

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

        self.flags.assignments.clear();

        self.flags.payloads.clear();

        self.flags.ekeys.clear();

        self.flags.inconsistent.clear();

        self.flags.to_run_r.clear();

        self.flags.to_run_w.clear();

        self.flags.sync_ended.clear();

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

        //    All machines are contiguous in the vector

        self.entities

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

        self.flags.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.flags.to_run_r.is_empty() {

            for _eid in self.flags.to_run_r.iter() {

        assert!(self.flags.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 is the same or more general than this one

            //    if it exists, drop the message; it is uninteresting.

            let ekey_bitset = self.flags.ekeys.get(&ekey);

                let mut slice_builder = vec![];

                // collect CONFLICTING assignments into slice_builder

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

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

                        slice_builder.push(bitset.as_slice());

                    }

                }

                let chunk_iter =

                    InNoneExceptIter::new(slice_builder.as_slice(), ekey_bitset.as_slice());

                BitChunkIter::new(chunk_iter).next()

            }) {

                // _eid is a payload whose predicate is at least as general

                // drop this message!

                continue 'recv_loop;

            }

            // 3. insert this payload as an entity, overwriting an existing LESS GENERAL payload if it exists.

                let mut slice_builder = vec![];

                slice_builder.push(ekey_bitset.as_slice());

                for assignment in msg.assignments.iter() {

                    if let Some(bitset) = self.flags.assignments.get(assignment) {

                        slice_builder.push(bitset.as_slice());

                    }

                }

                let chunk_iter = AndChunkIter::new(slice_builder.as_slice());

                BitChunkIter::new(chunk_iter).next()

            }) {

                // overwrite this entity index.

                eid

                    bitset.set(eid);

                }

                self.flags.payloads.set(eid);

                eid

            };

            self.feed_msg(payload_eid, ekey);

        }

    }

    fn run_poly_p(&mut self, machine_eid: usize) {

        match self.entities.get_mut(machine_eid) {

                // TODO run the machine

                // 1. make the assignment of this machine concrete WRT its ports

                let component_info = self.component_info.get(*component_index).unwrap();

                for &channel_id in component_info.1.iter() {

                    let test = self

                        .flags

                        .assignments

                        .get(&(channel_id, true))

                        .unwrap_or(false);

                    if !test {

                        // TRUE assignment wasn't set

                        // so set FALSE assignment (no effect if already set)

                        self.flags

                            .assignments

                            .entry((channel_id, false))

                            .or_default()

                            .set(machine_eid);

                    }

                }

                // 2. this machine becomes solved

            }

            // SOLUTION COMPLETE!

            return;

        }

    }

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

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

        self.flags

            .assignments

            .get(&(channel_id, true))

            .map(test)

    }

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

        // 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.flags.ekeys.get_mut(&ekey).expect("Payload sets this => cannot be empty");

        slice_builder.push(ekey_bitset.as_slice());

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

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

    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 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()?;

            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