return None;

    }

    pub fn handle_new_data_message(&mut self, ticket: MessageTicket, ctx: &mut ComponentCtx) {

        // Go through all branches that are awaiting new messages and see if

        // there is one that can receive this message.

            // Message should not be handled now

            return;

        }

        let message = ctx.read_message_using_ticket(ticket).as_data();

        let mut iter_id = self.tree.get_queue_first(QueueKind::AwaitingMessage);

            // And prepare the branch for running

            self.tree.push_into_queue(QueueKind::Runnable, receiving_branch_id);

        }

    }

    pub fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) -> Option<ConnectorScheduling> {

        if let Some(round_conclusion) = self.consensus.handle_new_sync_comp_message(message, ctx) {

            return Some(self.enter_non_sync_mode(round_conclusion, ctx));

        }

        return None;

    }

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

pub(crate) enum Consistency {

    Valid,

    Inconsistent,

}

pub(crate) enum MessageOrigin {

    Past,

    Present,

    Future

}

        // TODO: Handle sending and receiving ports

        // Set final ports

        let branch = &self.branch_annotations[branch_id.index as usize];

        // Clear out internal storage to defaults

        self.highest_connector_id = ConnectorId::new_invalid();

        self.branch_annotations.clear();

        self.branch_markers.clear();

        self.encountered_ports.clear();

        self.solution_combiner.clear();

        self.handled_wave = false;

    // --- Internal helpers

    fn handle_received_sync_header(&mut self, sync_header: SyncHeader, ctx: &mut ComponentCtx) -> MessageOrigin {

        debug_assert!(sync_header.sending_component_id != ctx.id); // not sending to ourselves

        let origin = self.handle_peer(&sync_header);

        if origin != MessageOrigin::Present {

            // We do not have to handle it now

            return origin;

        }

        if sync_header.highest_component_id > self.highest_connector_id {

                // Run the main behaviour of the connector, depending on its

                // current state.

                if scheduled.shutting_down {

                    // Nothing to do. But we're stil waiting for all our pending

                    // control messages to be answered.

                    self.debug_conn(connector_id, &format!("Shutting down, {} Acks remaining", scheduled.router.num_pending_acks()));

                    if scheduled.router.num_pending_acks() == 0 {

                        // We're actually done, we can safely destroy the

                        // currently running connector

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    } else {

                            );

                            self.debug_conn(connector_id, &format!("Sending message [ exit ] \n --- {:?}", message));

                            self.runtime.send_message(port.peer_connector, Message::Control(message));

                        }

                    }

                    if scheduled.router.num_pending_acks() == 0 {

                        // All ports (if any) already closed

                        self.runtime.destroy_component(connector_key);

                        continue 'thread_loop;

                    }

                    continue;

                }

            }

            // If here, then we should handle the message

            self.debug_conn(connector_id, " ... Handling the message");

                    scheduled.ctx.inbox.insert_new(message);

                }

            }

        }

    }

        }

    }

    /// Handles changes to the context that were made by the component. This is

    /// the way (due to Rust's borrowing rules) that we bubble up changes in the

    /// component's state that the scheduler needs to know about (e.g. a message

        if scheduled.ctx.changed_in_sync {

            if scheduled.ctx.is_in_sync {

                // Just entered sync region

            } else {

                // Just left sync region. So prepare inbox for the next sync

                // round

            }

            scheduled.ctx.changed_in_sync = false; // reset flag

        }

    }

/// A structure that contains inbox messages. Some messages are left inside and

/// continuously re-read. Others are taken out, but may potentially be put back

/// for later reading. Later reading in this case implies that they are put back

/// for reading in the next sync round.

struct Inbox {

    messages: RawVec<Message>,

    next_delay_idx: u32,

    start_read_idx: u32,

    next_read_idx: u32,

    generation: u32,

}

    generation: u32,

}

impl Inbox {

    fn new() -> Self {

        return Inbox {

            messages: RawVec::new(),

            next_delay_idx: 0,

            start_read_idx: 0,

            next_read_idx: 0,

            generation: 0,

        }

    }

    fn insert_new(&mut self, message: Message) {

        assert!(self.messages.len() < u32::MAX as usize); // TODO: @Size

        self.messages.push(message);

    }

    fn get_next_message_ticket(&mut self) -> Option<MessageTicket> {

        let cur_read_idx = self.next_read_idx as usize;

        if cur_read_idx >= self.messages.len() {

            return None;

        }

        self.generation += 1;

            generation: self.generation

        });

    }

    fn read_message_using_ticket(&self, ticket: MessageTicket) -> &Message {

        debug_assert_eq!(self.generation, ticket.generation);

        return unsafe{ &*self.messages.get(ticket.index as usize) }

    }

    fn take_message_using_ticket(&mut self, ticket: MessageTicket) -> Message {

        debug_assert_eq!(self.generation, ticket.generation);

        unsafe {

            let take_idx = ticket.index as usize;

            let val = std::ptr::read(self.messages.get(take_idx));

            // Move messages to the right, clearing up space in the

            // front.

        }

    }

    fn put_back_message(&mut self, message: Message) {

        // We have space in front of the array because we've taken out a message

        // before.

        debug_assert!(self.next_delay_idx < self.start_read_idx);

        unsafe {

            // Write to front of the array

            std::ptr::write(self.messages.get_mut(self.next_delay_idx as usize), message);

            self.next_delay_idx += 1;

        }

    }

    fn get_read_data_messages(&self, match_port_id: PortIdLocal) -> MessagesIter {

        return MessagesIter{

            messages: self.messages.as_slice(),

            next_index: self.start_read_idx as usize,

            max_index: self.next_read_idx as usize,

            match_port_id

        };

    }

        // Deallocate everything that was read

        self.destroy_range(self.start_read_idx, self.next_read_idx);

        self.generation += 1;

        // Join up all remaining values with the delayed ones in the front

        let num_to_move = self.messages.len() - self.next_read_idx as usize;

        self.start_read_idx = 0;

        self.next_read_idx = 0;

        self.messages.len = new_len;

    }

    fn transfer_messages_for_port(&mut self, port: PortIdLocal, new_inbox: &mut Inbox) {

        let mut idx = 0;

        while idx < self.messages.len() {

            let message = unsafe{ &*self.messages.get(idx) };

            if let Some(target_port) = message.target_port() {

                if target_port == port {

                    // Transfer port

                    unsafe {

                        let message = std::ptr::read(message as *const _);

                        }

                        self.messages.len -= 1;

                        new_inbox.insert_new(message);

                    }

                }

            }

        }

    }

    #[inline]

    fn destroy_range(&mut self, start_idx: u32, end_idx: u32) {

        for idx in (start_idx as usize)..(end_idx as usize) {

                let msg = self.messages.get_mut(idx);

                std::ptr::drop_in_place(msg);

            }

        }

    }

}

// -----------------------------------------------------------------------------

// Control messages

// -----------------------------------------------------------------------------

struct ControlEntry {

// Generic testing constants, use when appropriate to simplify stress-testing

// pub(crate) const NUM_THREADS: u32 = 3;     // number of threads in runtime

// pub(crate) const NUM_INSTANCES: u32 = 7;   // number of test instances constructed

// pub(crate) const NUM_LOOPS: u32 = 8;       // number of loops within a single test (not used by all tests)

fn create_runtime(pdl: &str) -> Runtime {

    let protocol = ProtocolDescription::parse(pdl.as_bytes()).expect("parse pdl");

    let runtime = Runtime::new(NUM_THREADS, protocol);