fn is_integer_literal_start(c: u8) -> bool {

    return c >= b'0' && c <= b'9';

}

fn maybe_number_remaining(c: u8) -> bool {

    // Note: hex range includes the possible binary indicator 'b' and 'B';

    return

        (c == b'o' || c == b'O' || c == b'x' || c == b'X') ||

            (c >= b'0' && c <= b'9') || (c >= b'A' && c <= b'F') || (c >= b'a' && c <= b'f') ||

            c == b'_';

}

            debug_assert!(receiving_branch.awaiting_port == message.data_header.target_port);

            receiving_branch.awaiting_port = PortIdLocal::new_invalid();

            receiving_branch.prepared = PreparedStatement::PerformedGet(message.content.clone());

            self.consensus.notify_of_received_message(receiving_branch_id, &message, ctx);

            // 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;

    }

    pub fn handle_new_sync_port_message(&mut self, message: SyncPortMessage, ctx: &mut ComponentCtx) {

        self.consensus.handle_new_sync_port_message(message, ctx);

    }

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

                        return ConnectorScheduling::Immediate;

                    },

                    Err(_) => {

                        // We don't own the port

                        let pd = &sched_ctx.runtime.protocol_description;

                        let eval_error = branch.code_state.new_error_at_expr(

                            &pd.modules, &pd.heap,

                            String::from("attempted to 'put' on port that is no longer owned")

                        );

                        self.eval_error = Some(eval_error);

                        self.mode = Mode::SyncError;

                        if let Some(conclusion) = self.consensus.notify_of_fatal_branch(branch_id, comp_ctx) {

                            return self.enter_non_sync_mode(conclusion, comp_ctx);

                        }

                    }

                }

            },

            _ => unreachable!("unexpected run result {:?} in sync mode", run_result),

        }

        // If here then the run result did not require a particular action. We

        // return whether we have more active branches to run or not.

        if self.tree.queue_is_empty(QueueKind::Runnable) {

            return ConnectorScheduling::NotNow;

        self.branch_markers.push(new_branch_id);

    }

    /// Notifies the consensus algorithm that a particular branch has

    /// encountered an unrecoverable error.

    pub fn notify_of_fatal_branch(&mut self, failed_branch_id: BranchId, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        debug_assert!(self.is_in_sync());

        // Check for trivial case, where branch has not yet communicated within

        // the consensus algorithm

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

        if branch.channel_mapping.iter().all(|v| v.registered_id.is_none()) {

            return Some(RoundConclusion::Failure);

        }

        // We're not in the trivial case: since we've communicated we need to

        // let everyone know that this round is probably not going to end well.

        return self.initiate_sync_failure(ctx);

    }

    /// Notifies the consensus algorithm that a branch has reached the end of

    /// the sync block. A final check for consistency will be performed that the

    /// caller has to handle. Note that

    pub fn notify_of_finished_branch(&self, branch_id: BranchId) -> Consistency {

    }

    /// Notifies the consensus algorithm about the chosen branch to commit to

    /// memory (may be the invalid "start" branch)

    pub fn end_sync(&mut self, branch_id: BranchId, _final_ports: &mut Vec<ComponentPortChange>) {

        debug_assert!(self.is_in_sync());

        // 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;

        self.conclusion = None;

        self.ack_remaining = 0;

        // And modify persistent storage

        self.sync_round += 1;

        for peer in self.peers.iter_mut() {

            peer.encountered_this_round = false;

            peer.expected_sync_round += 1;

        }

    }

    // --- Handling messages

    /// Prepares a message for sending. Caller should have made sure that

    /// sending the message is consistent with the speculative state.

    pub fn handle_message_to_send(&mut self, branch_id: BranchId, source_port_id: PortIdLocal, content: &ValueGroup, ctx: &mut ComponentCtx) -> (SyncHeader, DataHeader) {

        debug_assert!(self.is_in_sync());

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

        let port_info = ctx.get_port_by_id(source_port_id).unwrap();

        if cfg!(debug_assertions) {

                return self.send_to_leader_or_handle_as_leader(message.content, ctx);

            },

            SyncCompContent::GlobalSolution(solution) => {

                // Found a global solution

                debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader

                let (_, branch_id, _) = solution.component_branches.iter()

                    .find(|(component_id, _, _)| *component_id == ctx.id)

                    .unwrap();

                return Some(RoundConclusion::Success(*branch_id));

            },

            SyncCompContent::GlobalFailure => {

                // Global failure of round, send Ack to leader

                debug_assert_ne!(self.highest_connector_id, ctx.id); // not the leader

                let _result = self.send_to_leader_or_handle_as_leader(SyncCompContent::AckFailure, ctx);

                debug_assert!(_result.is_none());

                return Some(RoundConclusion::Failure);

            },

            SyncCompContent::Notification => {

                // We were just interested in the sync header we handled above

                return None;

            }

        }

    }

                }

            }

        }

        return true;

    }

    // --- 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 {

            // Sender has higher component ID. So should be the target of our

            // messages. We should also let all of our peers know

            self.highest_connector_id = sync_header.highest_component_id;

            for peer in self.peers.iter() {

                if peer.id == sync_header.sending_component_id || !peer.encountered_this_round {

                    // Don't need to send it to this one

                    if encountered.push(owner_b) {

                        let message = SyncCompMessage{

                            sync_header: self.create_sync_header(ctx),

                            target_component_id: owner_b,

                            content: SyncCompContent::GlobalFailure,

                        };

                        ctx.submit_message(Message::SyncComp(message)).unwrap();

                    }

                }

            }

        }

        self.conclusion = Some(RoundConclusion::Failure);

        if encountered.is_empty() {

            // We don't have to wait on Acks

            return Some(RoundConclusion::Failure);

        } else {

            self.ack_remaining = encountered.len() as u32;

            return None;

        }

    }

    fn initiate_sync_failure(&mut self, ctx: &mut ComponentCtx) -> Option<RoundConclusion> {

        debug_assert!(self.is_in_sync());

        let maybe_already = self.send_to_leader_or_handle_as_leader(SyncCompContent::Presence(ComponentPresence{

            component_id: ctx.id,

            channels: channel_presence,

        }), ctx);

        if self.handled_wave {

            // Someone (or us) has already initiated a sync failure.

            return maybe_already;

        }

        let maybe_conclusion = self.send_to_leader_or_handle_as_leader(SyncCompContent::LocalFailure, ctx);

        debug_assert!(if maybe_already.is_some() { maybe_conclusion.is_some() } else { true });

        // Initiate a discovery wave so peers can do the same

        self.handled_wave = true;

        for mapping in &self.branch_annotations[0].channel_mapping {

            let channel_id = mapping.channel_id;

            let port_info = ctx.get_port_by_channel_id(channel_id).unwrap();

            let message = SyncPortMessage{

                sync_header: self.create_sync_header(ctx),

                source_port: port_info.self_id,

                target_port: port_info.peer_id,

                content: SyncPortContent::NotificationWave,

            };

            }

            if !found {

                self.presence.push(ChannelPresence{

                    owner_a: component_id,

                    owner_b: None,

                    id: entry.channel_id,

                    state: if entry.is_closed { PresenceState::Closed } else { PresenceState::OnePresent },

                });

            }

        }

        return self.check_for_global_failure();

    }

    fn mark_failure_and_check_for_global_failure(&mut self) -> bool {

        self.failure_reported = true;

        return self.check_for_global_failure();

    }

    /// Checks if, starting at the provided local solution, a global solution

    /// can be formed.

    // TODO: At some point, check if divide and conquer is faster?

    fn check_for_global_solution(&self, initial_component_index: usize, initial_solution_index: usize) -> Option<GlobalSolution> {

                        final_branch_id: matched.final_branch_id,

                        port_mapping: matched.channel_mapping,

                    };

                    let maybe_solution = self.add_solution_and_check_for_global_solution(local_solution);

                    if let Some(global_solution) = maybe_solution {

                        return Some(LeaderConclusion::Solution(global_solution));

                    }

                }

            }

        }

        // Handle channel presence

        if self.presence.is_empty() {

            // Trivial case

            self.presence = combiner.presence;

        } else {

            for presence in combiner.presence {

                match self.presence.iter_mut().find(|v| v.id == presence.id) {

                    Some(entry) => {

                        // Combine entries. Take first that has Closed, then

                        // check first that has both, then check if they are

                        // combinable

                        if entry.state == PresenceState::Closed {

                            // Do nothing

                        } else if presence.state == PresenceState::Closed {

                            entry.owner_a = presence.owner_a;

                            entry.owner_b = presence.owner_b;

                        } else {

                            // Both have one presence, combine into both present

                            debug_assert!(entry.state == PresenceState::OnePresent && presence.state == PresenceState::OnePresent);

                            entry.owner_b = Some(presence.owner_a);

                            entry.state = PresenceState::BothPresent;

                        }

                    },

                    None => {

                        self.presence.push(presence);

                    }

                }

            }

            // After adding everything we might have immediately found a solution

            if self.check_for_global_failure() {

                return Some(LeaderConclusion::Failure);

            }

        }

        return None;

    }

    fn clear(&mut self) {

        self.local.clear();

        self.presence.clear();

        self.failure_reported = false;

    }

            unsafe {

                let target = &mut **self.entries.get_mut(index);

                std::ptr::write(&mut target.connector as *mut _, connector);

                let _old_num_users = target.num_users.fetch_add(1, Ordering::SeqCst);

                debug_assert_eq!(_old_num_users, 0);

                let generation = target.generation.load(Ordering::Acquire);

                key = ConnectorKey{ index: index as u32, generation };

                target.connector.ctx.id = key.downcast();

            }

        }

        return key;

    }

    /// Destroys a connector. Caller should make sure it is not scheduled for

    /// execution. Otherwise one experiences "bad stuff" (tm).

    fn destroy(&mut self, key: ConnectorKey) {

        unsafe {

            let target = self.entries.get_mut(key.index as usize);

            (**target).generation.fetch_add(1, Ordering::SeqCst);

            std::ptr::drop_in_place(*target);

            // Note: but not deallocating!

        }

        self.free.push(key.index as usize);

    }

}

impl Drop for ConnectorStore {

    fn drop(&mut self) {

        // Everything in the freelist already had its destructor called, so only

        // has to be deallocated

        for free_idx in self.free.iter().copied() {

            unsafe {

                let memory = self.entries.get_mut(free_idx);

                let layout = std::alloc::Layout::for_value(&**memory);

            // the exact same atomic compare-and-swap at this point in time)

            let should_wake_up_again = connector.public.sleeping

                .compare_exchange(true, false, Ordering::SeqCst, Ordering::Acquire)

                .is_ok();

            if should_wake_up_again {

                self.runtime.push_work(connector_key)

            }

        }

    }

    fn debug(&self, message: &str) {

    }

    fn debug_conn(&self, conn: ConnectorId, message: &str) {

    }

}

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

// ComponentCtx

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

enum ComponentStateChange {

    CreatedComponent(ConnectorPDL, Vec<PortIdLocal>),

    CreatedPort(Port),

    ChangedPort(ComponentPortChange),

}

    pub(crate) fn submit_message(&mut self, contents: Message) -> Result<(), ()> {

        debug_assert!(self.is_in_sync);

        if let Some(port_id) = contents.source_port() {

            let port_info = self.get_port_by_id(port_id);

            let is_valid = match port_info {

                Some(port_info) => {

                    port_info.state == PortState::Open

                },

                None => false,

            };

            if !is_valid {

                // We don't own the port

                return Err(());

            }

        }

        self.outbox.push_back(contents);

        return Ok(());

    }

    /// Notify that component just finished a sync block. Like

    /// `notify_sync_start`: drop out of the `Component::Run` function.

    pub(crate) fn notify_sync_end(&mut self, changed_ports: &[ComponentPortChange]) {

        debug_assert!(self.is_in_sync);