return &mut self.branches[0];

    }

    /// Returns the branch ID of the first branch in a particular queue.

    pub fn get_queue_first(&self, kind: QueueKind) -> Option<BranchId> {

        let queue = &self.queues[kind.as_index()];

        if queue.first.is_valid() {

            return Some(queue.first);

        } else {

            return None;

        }

    }

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

    }

    /// Creates a new speculative branch based on the provided one. The index to

    /// retrieve this new branch will be returned.

    pub fn fork_branch(&mut self, parent_branch_id: BranchId) -> BranchId {

        debug_assert!(self.is_in_sync());

        let parent_branch = &self[parent_branch_id];

        let new_branch = Branch::new_sync(self.branches.len() as u32, parent_branch);

        let new_branch_id = new_branch.id;

        self.branches.push(new_branch);

        return new_branch_id;

        // Clear out all the queues

        for queue_idx in 0..NUM_QUEUES {

            self.queues[queue_idx] = BranchQueue::new();

        }

    }

}

impl Index<BranchId> for ExecTree {

    type Output = Branch;

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

        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,

    pub awaiting_port: PortIdLocal,

    pub next_in_queue: BranchId,

    pub inbox: HashMap<PortIdLocal, ValueGroup>,

}

impl BranchListItem for FakeBranch {

    #[inline] fn get_id(&self) -> BranchId { return self.id; }

    #[inline] fn set_next_id(&mut self, id: BranchId) { self.next_in_queue = id; }

    #[inline] fn get_next_id(&self) -> BranchId { return self.next_in_queue; }

}

impl FakeBranch {

struct ConnectorRunContext<'a> {

    branch_id: BranchId,

    consensus: &'a Consensus,

    prepared: PreparedStatement,

}

impl<'a> RunContext for ConnectorRunContext<'a>{

    fn performed_put(&mut self, _port: PortId) -> bool {

        return match self.prepared.take() {

            PreparedStatement::None => false,

            PreparedStatement::PerformedPut => true,

            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() {

            PreparedStatement::None => None,

            PreparedStatement::ForkedExecution(path) => Some(path),

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

        };

    }

}

impl Connector for ConnectorPDL {

    fn run(&mut self, sched_ctx: SchedulerCtx, comp_ctx: &mut ComponentCtx) -> ConnectorScheduling {

        if let Some(scheduling) = self.handle_new_messages(comp_ctx) {

            return scheduling;

        }

impl Consensus {

    pub fn new() -> Self {

        return Self {

            highest_connector_id: ConnectorId::new_invalid(),

            branch_annotations: Vec::new(),

            branch_markers: Vec::new(),

            encountered_ports: VecSet::new(),

            solution_combiner: SolutionCombiner::new(),

            handled_wave: false,

            conclusion: None,

            ack_remaining: 0,

            peers: Vec::new(),

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

                .map(|v| ChannelAnnotation {

                    channel_id: v.channel_id,

                    registered_id: None,

                    expected_firing: None,

                })

                .collect(),

            cur_marker: BranchMarker::new_invalid(),

        });

        self.branch_markers.push(BranchId::new_invalid());

        self.highest_connector_id = ctx.id;

                        // the silent ports (in scope) is actually closed

                        return self.notify_of_fatal_branch(branch_id, ctx);

                    }

                }

            }

            target_mapping.push((

                channel_id,

                port.registered_id.unwrap_or(BranchMarker::new_invalid())

            ));

        }

        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;

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

        }

        println!("DEBUG: ***** Peers post round are:\n{:#?}", &self.peers)

    }

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

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

        };

        // Update port mapping

        for mapping in &mut branch.channel_mapping {

            if mapping.channel_id == port_info.channel_id {

                mapping.expected_firing = Some(true);

                mapping.registered_id = Some(branch.cur_marker);

            }

        }

        // Update branch marker

        let new_marker = BranchMarker::new(self.branch_markers.len() as u32);

        branch.cur_marker = new_marker;

        // currently stored one.

        debug_assert_eq!(message.in_response_to_sync_round, self.sync_round);

        match message.content {

            SyncControlContent::ChannelIsClosed(_) => {

                return self.initiate_sync_failure(ctx);

            }

        }

    }

    pub fn notify_of_received_message(&mut self, branch_id: BranchId, message: &DataMessage, ctx: &ComponentCtx) {

        debug_assert!(self.branch_can_receive(branch_id, message));

        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

    /// match then the branch can receive the message.

    pub fn branch_can_receive(&self, branch_id: BranchId, message: &DataMessage) -> bool {

        if let Some(peer) = self.peers.iter().find(|v| v.id == message.sync_header.sending_component_id) {

            if message.sync_header.sync_round < peer.expected_sync_round {

                return false;

            }

        }

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

        for expected in &message.data_header.expected_mapping {

            // If we own the port, then we have an entry in the

            // annotation, check if the current mapping matches

            peer_id: putter_id,

            channel_id,

            kind: PortKind::Getter,

            state: PortState::Open,

            peer_connector: creating_connector,

        };

        let putter_port = Port{

            self_id: putter_id,

            peer_id: getter_id,

            channel_id,

            kind: PortKind::Putter,

            state: PortState::Open,

            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

                // (destroyed, at a different generation number, busy being

                // destroyed, etc.) we cannot send the message and will not

                // modify the component

                return false;

            }

            cur_num_users = old_num_users;

        }

        // We incremented the counter. But we might still be at the wrong

        // generation number. The generation number is a monotonically

        // increasing value. Since it only increases when someone gets the

        let job_queue = Arc::new(Mutex::new(VecDeque::with_capacity(32)));

        let connector = ConnectorApplication {

            sync_done: sync_done.clone(),

            job_queue: job_queue.clone(),

            is_in_sync: false,

            sync_desc: Vec::new(),

            tree: FakeTree::new(),

            consensus: Consensus::new(),

            last_finished_handled: None,

            branch_extra: vec![0],

        };

        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;

        }

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

        while let Some(branch_id) = iter_id {

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

            iter_id = self.tree.get_queue_next(branch_id);

            let branch = &self.tree[branch_id];

            if branch.awaiting_port != message.data_header.target_port { continue; }

            if !self.consensus.branch_can_receive(branch_id, &message) { continue; }

            // This branch can receive, so fork and given it the message

    }

    pub(crate) fn handle_new_sync_comp_message(&mut self, message: SyncCompMessage, ctx: &mut ComponentCtx) {

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

            self.collapse_sync_to_conclusion(conclusion, ctx);

        }

    }

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

        self.consensus.handle_new_sync_port_message(message, ctx);

    }

    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 {

            // We still have instructions to perform

            let cur_instruction = &self.sync_desc[instruction_idx];

            self.branch_extra[branch_id.index as usize] += 1;

            match &cur_instruction {

                ApplicationSyncAction::Put(port_id, content) => {

                    let port_id = *port_id;

                    let (sync_header, data_header) = self.consensus.handle_message_to_send(branch_id, port_id, &content, comp_ctx);

                    let message = Message::Data(DataMessage {

                        sync_header,

                        data_header,

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

        debug_assert!(!self.is_in_sync);

        // In non-sync mode the application component doesn't really do anything

        // except performing jobs submitted from the API. This is the only

        // case where we expect to be woken up.

        // Note that we have to communicate to the scheduler when we've received

        // ports or created components (hence: given away ports) *before* we

        // enter a sync round.

        let mut queue = self.job_queue.lock().unwrap();

        while let Some(job) = queue.pop_front() {

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

                    self.branch_extra.push(0); // set first branch to first instruction

                    return ConnectorScheduling::Immediate;

                },

                ApplicationJob::Shutdown => {

                    debug_assert!(queue.is_empty());

                    return ConnectorScheduling::Exit;

                }

            }

        }

                let new_component_idx = match removed_entry.variant {

                    ControlVariant::ChangedPort(message) => {

                        self.position(message.new_component_entry_id).unwrap()

                    },

                    _ => return None,

                };

                // Decrement counter, if 0, then schedule component

                let new_component_entry = self.active[new_component_idx].variant.as_new_component_mut();

                new_component_entry.num_acks_pending -= 1;

                if new_component_entry.num_acks_pending != 0 {

                    return None;

                }

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

        let generated_id = self.id_counter;

        let (new_id, _) = self.id_counter.overflowing_add(1);

        self.id_counter = new_id;

        return generated_id;

    }

    #[inline]

    fn position(&self, id: u32) -> Option<usize> {

        return self.active.iter().position(|v| v.id == id);

    }

}

// basics.rs

//

// The most basic of testing: sending a message, receiving a message, etc.

use super::*;

#[test]

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;

        while (index < loops) {

            sync {

                put(sender, true);

            }

            index += 1;

        }

    }

    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");

    });

}

#[test]

fn test_combined_put_and_get() {

    const CODE: &'static str = "

    primitive put_then_get(out<bool> output, in<bool> input, u32 num_loops) {

        u32 index = 0;

        while (index < num_loops) {

            sync {

                put(output, true);

                auto value = get(input);

                put(vals, 0b01000000);

            }

            index += 1;

        }

    }

    primitive getter_dynamic(in<u8> vals, u32 num_loops) {

        u32 loop_index = 0;

        while (loop_index < num_loops) {

            sync {

                u32 recv_index = 0;

                u8 expected = 1;

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

    })

}

            // print(\"ending loop\");

            loop_index += 1;

        }

    }

    composite constructor(u32 num_edges, u32 num_loops) {

        auto requests = {};

        auto responses = {};

        u32 edge_index = 0;

        while (edge_index < num_edges) {

            channel req_put -> req_get;

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

        u32 loop_index = 0;

        u32 initial_value = 1337;

        while (loop_index < num_loops) {

            sync {

                put(req, initial_value);

                auto result = get(resp);

                assert(result == initial_value + num_nodes * 2);

            }

            loop_index += 1;

        }

    }

    composite constructor(u32 num_nodes, u32 num_loops) {

        channel initial_req -> req_in;

        channel resp_out -> final_resp;

        new start(initial_req, final_resp, num_nodes, num_loops);

        in<u32> last_req_in = req_in;

        out<u32> last_resp_out = resp_out;

        u32 node = 0;

        while (node < num_nodes) {

            channel new_req_fw -> new_req_in;

            channel new_resp_out -> new_resp_in;

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

    }

    ";