/// Returns the next branch ID of a branch (assumed to be in a particular

    /// queue.

    pub fn get_queue_next(&self, branch_id: BranchId) -> Option<BranchId> {

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

        if branch.next_in_queue.is_valid() {

            return Some(branch.next_in_queue);

        } else {

            return None;

        }

    }

    // --- Preparing and finishing a speculative round

    /// Starts a synchronous round by cloning the non-sync branch and marking it

    /// as the root of the speculative tree. The id of this root sync branch is

    /// returned.

    pub fn start_sync(&mut self) -> BranchId {

        debug_assert!(!self.is_in_sync());

        let sync_branch = Branch::new_sync(1, &self.branches[0]);

        let sync_branch_id = sync_branch.id;

        self.branches.push(sync_branch);

        return sync_branch_id;

        debug_assert!(index.is_valid());

        return &self.branches[index.index as usize];

    }

}

impl IndexMut<BranchId> for ExecTree {

    fn index_mut(&mut self, index: BranchId) -> &mut Self::Output {

        debug_assert!(index.is_valid());

        return &mut self.branches[index.index as usize];

    }

}

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

// FakeTree

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

/// Generic fake branch. This is supposed to be used in conjunction with the

/// fake tree. The purpose is to have a branching-like tree to use in

/// combination with a consensus algorithm in places where we don't have PDL

/// code.

pub(crate) struct FakeBranch {

    pub id: BranchId,

    pub parent_id: BranchId,

    pub sync_state: SpeculativeState,

            taken => unreachable!("prepared statement is '{:?}' during 'performed_put()'", taken)

        };

    }

    fn performed_get(&mut self, _port: PortId) -> Option<ValueGroup> {

        return match self.prepared.take() {

            PreparedStatement::None => None,

            PreparedStatement::PerformedGet(value) => Some(value),

            taken => unreachable!("prepared statement is '{:?}' during 'performed_get()'", taken),

        };

    }

    }

    fn created_channel(&mut self) -> Option<(Value, Value)> {

        return match self.prepared.take() {

            PreparedStatement::None => None,

            PreparedStatement::CreatedChannel(ports) => Some(ports),

            taken => unreachable!("prepared statement is '{:?}' during 'created_channel()'", taken),

        };

    }

    fn performed_fork(&mut self) -> Option<bool> {

        return match self.prepared.take() {

            sync_round: 0,

            workspace_ports: Vec::new(),

        }

    }

    // --- Controlling sync round and branches

    /// Returns whether the consensus algorithm is running in sync mode

    pub fn is_in_sync(&self) -> bool {

        return !self.branch_annotations.is_empty();

    }

    /// Sets up the consensus algorithm for a new synchronous round. The

    /// provided ports should be the ports the component owns at the start of

    /// the sync round.

    pub fn start_sync(&mut self, ctx: &ComponentCtx) {

        debug_assert!(!self.highest_connector_id.is_valid());

        debug_assert!(self.branch_annotations.is_empty());

        debug_assert!(self.solution_combiner.local.is_empty());

        // We'll use the first "branch" (the non-sync one) to store our ports,

        // this allows cloning if we created a new branch.

        self.branch_annotations.push(BranchAnnotation{

            channel_mapping: ctx.get_ports().iter()

        let local_solution = LocalSolution{

            component: ctx.id,

            sync_round_number: self.sync_round,

            final_branch_id: branch_id,

            port_mapping: target_mapping,

        };

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

        return maybe_conclusion;

    }

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

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

        debug_assert!(self.is_in_sync());

        // TODO: Handle sending and receiving ports

        // Set final ports

        // Clear out internal storage to defaults

        println!("DEBUG: ***** Incrementing sync round stuff");

        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;

            // Check for consistent mapping

            let port = branch.channel_mapping.iter()

                .find(|v| v.channel_id == port_info.channel_id)

                .unwrap();

            debug_assert!(port.expected_firing == None || port.expected_firing == Some(true));

        }

        // Check for ports that are being sent

        debug_assert!(self.workspace_ports.is_empty());

        find_ports_in_value_group(content, &mut self.workspace_ports);

        if !self.workspace_ports.is_empty() {

            todo!("handle sending ports");

        }

        // Construct data header

        let data_header = DataHeader{

            expected_mapping: branch.channel_mapping.iter()

                .filter(|v| v.registered_id.is_some() || v.channel_id == port_info.channel_id)

                .copied()

                .collect(),

            sending_port: port_info.self_id,

            target_port: port_info.peer_id,

            new_mapping: branch.cur_marker,

        };

        let target_port = ctx.get_port_by_id(message.data_header.target_port).unwrap();

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

        for mapping in &mut branch.channel_mapping {

            if mapping.channel_id == target_port.channel_id {

                // Found the port in which the message should be inserted

                mapping.registered_id = Some(message.data_header.new_mapping);

                // Check for sent ports

                debug_assert!(self.workspace_ports.is_empty());

                find_ports_in_value_group(&message.content, &mut self.workspace_ports);

                if !self.workspace_ports.is_empty() {

                    todo!("handle received ports");

                }

                return;

            }

        }

        // If here, then the branch didn't actually own the port? Means the

        // caller made a mistake

        unreachable!("incorrect notify_of_received_message");

    }

    /// Matches the mapping between the branch and the data message. If they

            peer_connector: creating_connector,

        };

        return (getter_port, putter_port);

    }

    /// Sends a message directly (without going through the port) to a

    /// component. This is slightly less efficient then sending over a port, but

    /// might be preferable for some algorithms. If the component was sleeping

    /// then it is scheduled for execution.

    pub(crate) fn send_message_maybe_destroyed(&self, target_id: ConnectorId, message: Message) -> bool {

        let target = {

            lock.get(target_id.index)

        };

        // Do a CAS on the number of users. Most common case the component is

        // alive and we're the only one sending the message. Note that if we

        // finish this block, we're sure that no-one has set the `num_users`

        // value to 0. This is essential! When at 0, the component is added to

        // the freelist and the generation counter will be incremented.

        let mut cur_num_users = 1;

        while let Err(old_num_users) = target.num_users.compare_exchange(cur_num_users, cur_num_users + 1, Ordering::SeqCst, Ordering::Acquire) {

            if old_num_users == 0 {

                // Cannot send message. Whatever the component state is

        let interface = ApplicationInterface::new(sync_done, job_queue, runtime);

        return (connector, interface);

    }

    fn handle_new_messages(&mut self, comp_ctx: &mut ComponentCtx) {

        while let Some(ticket) = comp_ctx.get_next_message_ticket() {

            let message = comp_ctx.read_message_using_ticket(ticket);

            if let Message::Data(_) = message {

                self.handle_new_data_message(ticket, comp_ctx)

            } else {

                match comp_ctx.take_message_using_ticket(ticket) {

                    Message::SyncComp(message) => self.handle_new_sync_comp_message(message, comp_ctx),

                    Message::SyncPort(message) => self.handle_new_sync_port_message(message, comp_ctx),

                    Message::Control(_) => unreachable!("control message in native API component"),

                }

            }

        }

    }

    pub(crate) 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.

        if !self.consensus.handle_new_data_message(ticket, ctx) {

            // Old message, so drop it

            return;

    fn run_in_sync_mode(&mut self, _sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {

        debug_assert!(self.is_in_sync);

        self.handle_new_messages(comp_ctx);

        let branch_id = self.tree.pop_from_queue(QueueKind::Runnable);

        if branch_id.is_none() {

            return ConnectorScheduling::NotNow;

        }

        let branch_id = branch_id.unwrap();

        let branch = &mut self.tree[branch_id];

        if instruction_idx >= self.sync_desc.len() {

            // Performed last instruction, so this branch is officially at the

            // end of the synchronous interaction.

            let consistency = self.consensus.notify_of_finished_branch(branch_id);

            if consistency == Consistency::Valid {

                branch.sync_state = SpeculativeState::ReachedSyncEnd;

                self.tree.push_into_queue(QueueKind::FinishedSync, branch_id);

            } else {

                branch.sync_state = SpeculativeState::Inconsistent;

            }

        } else {

            match job {

                ApplicationJob::NewChannel((endpoint_a, endpoint_b)) => {

                    comp_ctx.push_port(endpoint_a);

                    comp_ctx.push_port(endpoint_b);

                    return ConnectorScheduling::Immediate;

                }

                ApplicationJob::NewConnector(connector, initial_ports) => {

                    comp_ctx.push_component(connector, initial_ports);

                    return ConnectorScheduling::Later;

                },

                    // Entering sync mode

                    comp_ctx.notify_sync_start();

                    self.sync_desc = description;

                    self.is_in_sync = true;

                    debug_assert!(self.last_finished_handled.is_none());

                    debug_assert!(self.branch_extra.len() == 1);

                    let first_branch_id = self.tree.start_sync();

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

                    debug_assert!(first_branch_id.index == 1);

                    self.consensus.start_sync(comp_ctx);

                    self.consensus.notify_of_new_branch(BranchId::new_invalid(), first_branch_id);

                }

                // Return component key for scheduling

                let new_component_entry = self.active.remove(new_component_idx);

                let new_component_entry = match new_component_entry.variant {

                    ControlVariant::NewComponent(entry) => entry,

                    _ => unreachable!(),

                };

                return Some(new_component_entry.component_key);

            },

            None => {

            },

        }

    }

    /// Retrieves the number of responses we still expect to receive from our

    /// peers

    #[inline]

    pub fn num_pending_acks(&self) -> usize {

        return self.active.len();

    }

    fn take_id(&mut self) -> u32 {

fn test_doing_nothing() {

    // If this thing does not get into an infinite loop, (hence: the runtime

    // exits), then the test works

    const CODE: &'static str ="

    primitive silent_willy(u32 loops) {

        u32 index = 0;

        while (index < loops) {

            sync { index += 1; }

        }

    }

    ";

    run_test_in_runtime(CODE, |api| {

        api.create_connector("", "silent_willy", ValueGroup::new_stack(vec![

            Value::UInt32(NUM_LOOPS),

        ])).expect("create component");

    });

}

#[test]

fn test_single_put_and_get() {

    const CODE: &'static str = "

    primitive putter(out<bool> sender, u32 loops) {

        u32 index = 0;

    primitive getter(in<bool> receiver, u32 loops) {

        u32 index = 0;

        while (index < loops) {

            sync {

                auto result = get(receiver);

                assert(result);

            }

            index += 1;

        }

    }

    ";

    run_test_in_runtime(CODE, |api| {

        let channel = api.create_channel().unwrap();

        api.create_connector("", "putter", ValueGroup::new_stack(vec![

            Value::Output(PortId::new(channel.putter_id.index)),

            Value::UInt32(NUM_LOOPS)

        ])).expect("create putter");

        api.create_connector("", "getter", ValueGroup::new_stack(vec![

            Value::Input(PortId::new(channel.getter_id.index)),

            Value::UInt32(NUM_LOOPS)

        ])).expect("create getter");

                while (recv_index < 4) {

                    auto gotten = get(vals);

                    assert(gotten == expected);

                    expected <<= 2;

                    recv_index += 1;

                }

            }

            loop_index += 1;

        }

    }

    ";

    run_test_in_runtime(CODE, |api| {

        let channel = api.create_channel().unwrap();

        api.create_connector("", "putter_static", ValueGroup::new_stack(vec![

            Value::Output(PortId::new(channel.putter_id.index)),

            Value::UInt32(NUM_LOOPS),

        ])).unwrap();

        api.create_connector("", "getter_dynamic", ValueGroup::new_stack(vec![

            Value::Input(PortId::new(channel.getter_id.index)),

            Value::UInt32(NUM_LOOPS),

        ])).unwrap();

    })

}

            channel resp_put -> resp_get;

            new edge(req_get, resp_put, num_loops);

            requests @= { req_put };

            responses @= { resp_get };

            edge_index += 1;

        }

        new center(requests, responses, num_loops);

    }

    ";

    run_test_in_runtime(CODE, |api| {

        api.create_connector("", "constructor", ValueGroup::new_stack(vec![

            Value::UInt32(5),

            Value::UInt32(NUM_LOOPS),

        ])).expect("create connector");

    });

}

#[test]

fn test_conga_line_request() {

    const CODE: &'static str = "

    primitive start(out<u32> req, in<u32> resp, u32 num_nodes, u32 num_loops) {

            new middle(last_req_in, new_req_fw, new_resp_in, last_resp_out, num_loops);

            last_req_in = new_req_in;

            last_resp_out = new_resp_out;

            node += 1;

        }

        new end(last_req_in, last_resp_out, num_loops);

    }

    ";

    run_test_in_runtime(CODE, |api| {

        api.create_connector("", "constructor", ValueGroup::new_stack(vec![

            Value::UInt32(1),

            Value::UInt32(NUM_LOOPS)

        ])).expect("create connector");

    });

}