if !self.consensus.handle_new_data_message(&self.tree, &message, ctx, &mut branches) {

            // Old message, so drop it

            return;

        }

            self.consensus.notify_of_received_message(receiving_branch_id, &message.sync_header, &message.data_header, &message.content);

                        if self.consensus.branch_can_receive(branch_id, &message.sync_header, &message.data_header, &message.content) {

                            self.consensus.notify_of_received_message(receiving_branch_id, &message.sync_header, &message.data_header, &message.content);

struct Peer {

    id: ConnectorId,

    encountered_this_round: bool,

    expected_sync_round: u32,

}

    peers: Vec<Peer>,

    sync_round: u32,

            peers: Vec::new(),

            sync_round: 0,

                            sync_round: self.sync_round

        self.sync_round += 1;

        for peer in self.peers.iter_mut() {

            peer.encountered_this_round = false;

            peer.expected_sync_round += 1;

        }

    pub fn handle_new_data_message(&mut self, exec_tree: &ExecTree, message: &DataMessage, ctx: &mut ComponentCtx, target_ids: &mut Vec<BranchId>) -> bool {

        self.handle_received_data_header(exec_tree, &message.sync_header, &message.data_header, &message.content, target_ids);

        return self.handle_received_sync_header(&message.sync_header, ctx)

        if !self.handle_received_sync_header(&message.sync_header, ctx) {

            return None;

        }

    pub fn notify_of_received_message(&mut self, branch_id: BranchId, sync_header: &SyncHeader, data_header: &DataHeader, content: &DataContent) {

        debug_assert!(self.branch_can_receive(branch_id, sync_header, data_header, content));

    pub fn branch_can_receive(&self, branch_id: BranchId, sync_header: &SyncHeader, data_header: &DataHeader, content: &DataContent) -> bool {

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

            if sync_header.sync_round < peer.expected_sync_round {

                return false;

            }

        }

    fn handle_received_data_header(&self, exec_tree: &ExecTree, sync_header: &SyncHeader, data_header: &DataHeader, content: &DataContent, target_ids: &mut Vec<BranchId>) {

                if self.branch_can_receive(branch.id, sync_header, data_header, content) {

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

        if !self.handle_peer(sync_header) {

            // We can drop this package

            return false;

        }

            for peer in self.peers.iter() {

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

                    target_component_id: peer.id,

        return true;

    }

    /// Handles a (potentially new) peer. Returns `false` if the provided sync

    /// number is different then the expected one.

    fn handle_peer(&mut self, sync_header: &SyncHeader) -> bool {

        let position = self.peers.iter().position(|v| v.id == sync_header.sending_component_id);

        match position {

            Some(index) => {

                let entry = &mut self.peers[index];

                entry.encountered_this_round = true;

                // TODO: Proper handling of potential overflow

                if sync_header.sync_round >= entry.expected_sync_round {

                    entry.expected_sync_round = sync_header.sync_round;

                    return true;

                } else {

                    return false;

                }

            },

            None => {

                self.peers.push(Peer{

                    id: sync_header.sending_component_id,

                    encountered_this_round: true,

                    expected_sync_round: sync_header.sync_round,

                });

                return true;

            }

        }

            sync_round: self.sync_round,

// TODO: Remove all debug derives

#[derive(Debug)]

#[derive(Debug)]

#[derive(Debug)]

#[derive(Debug)]

struct CheckEntry {

    component_index: usize,         // component index in combiner's vector

    solution_index: usize,          // solution entry in the above component entry

    parent_entry_index: usize,      // parent that caused the creation of this checking entry

    match_index_in_parent: usize,   // index in the matches array of the parent

    solution_index_in_parent: usize,// index in the solution array of the match entry in the parent

}

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

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

        // Construct initial entry on stack

        let mut stack = Vec::with_capacity(self.local.len());

        stack.push(CheckEntry{

            component_index: initial_component_index,

            solution_index: initial_solution_index,

            parent_entry_index: 0,

            match_index_in_parent: 0,

            solution_index_in_parent: 0,

        });

        'check_last_stack: loop {

            let cur_index = stack.len() - 1;

            let cur_entry = &stack[cur_index];

            // Check if the current component is matching with all other entries

            let mut all_match = true;

            'check_against_existing: for prev_index in 0..cur_index {

                let prev_entry = &stack[prev_index];

                let prev_component = &self.local[prev_entry.component_index];

                let prev_solution = &prev_component.solutions[prev_entry.solution_index];

                for prev_matching_component in &prev_solution.matches {

                    if prev_matching_component.target_index == cur_entry.component_index {

                        // Previous entry has shared ports with the current

                        // entry, so see if we have a composable pair of

                        // solutions.

                        if !prev_matching_component.match_indices.contains(&cur_entry.solution_index) {

                            all_match = false;

                            break 'check_against_existing;

                }

            }

            if all_match {

                // All components matched until now.

                if stack.len() == self.local.len() {

                    // We have found a global solution

                    break 'check_last_stack;

                }

                // Not all components found yet, look for a new one that has not

                // yet been added yet.

                for (parent_index, parent_entry) in stack.iter().enumerate() {

                    let parent_component = &self.local[parent_entry.component_index];

                    let parent_solution = &parent_component.solutions[parent_entry.solution_index];

                    for (peer_index, peer_component) in parent_solution.matches.iter().enumerate() {

                        if peer_component.match_indices.is_empty() {

                            continue;

                        }

                        let already_added = stack.iter().any(|v| v.component_index == peer_component.target_index);

                        if !already_added {

                            // New component to try

                            stack.push(CheckEntry{

                                component_index: peer_component.target_index,

                                solution_index: peer_component.match_indices[0],

                                parent_entry_index: parent_index,

                                match_index_in_parent: peer_index,

                                solution_index_in_parent: 0,

                            });

                            continue 'check_last_stack;

                        }

                // Cannot find a peer to add. This is possible if, for example,

                // we have a component A which has the only connection to

                // component B. And B has sent a local solution saying it is

                // finished, but the last data message has not yet arrived at A.

                // In any case, we just exit the if statement and handle not

                // being able to find a new connector as being forced to try a

                // new permutation of possible local solutions.

            }

            // Either the currently considered local solution is inconsistent

            // with other local solutions, or we cannot find a new component to

            // add. This is where we perform backtracking as long as needed to

            // try a new solution.

            while stack.len() > 1 {

                // Check if our parent has another solution we can try

                let cur_index = stack.len() - 1;

                let cur_entry = &stack[cur_index];

                let parent_entry = &stack[cur_entry.parent_entry_index];

                let parent_component = &self.local[parent_entry.component_index];

                let parent_solution = &parent_component.solutions[parent_entry.solution_index];

                let match_component = &parent_solution.matches[cur_entry.match_index_in_parent];

                debug_assert!(match_component.target_index == cur_entry.component_index);

                let new_solution_index_in_parent = cur_entry.solution_index_in_parent + 1;

                if new_solution_index_in_parent < match_component.match_indices.len() {

                    // We can still try a new one

                    let new_solution_index = match_component.match_indices[new_solution_index_in_parent];

                    let cur_entry = &mut stack[cur_index];

                    cur_entry.solution_index_in_parent = new_solution_index_in_parent;

                    cur_entry.solution_index = new_solution_index;

                    continue 'check_last_stack;

                } else {

                    // We're out of options here. So pop an entry, then in

                    // the next iteration of this backtracking loop we try

                    // to increment that solution

                    stack.pop();

                }

            // Stack length is 1, hence we're back at our initial solution.

            // Since that doesn't yield a global solution, we simply:

            return None;

        // Constructing the representation of the global solution

        debug_assert_eq!(stack.len(), self.local.len());

        let mut final_branches = Vec::with_capacity(stack.len());

        for entry in &stack {

            let component = &self.local[entry.component_index];

            let solution = &component.solutions[entry.solution_index];

        for entry in &stack {

            let component = &self.local[entry.component_index];

        for entry in &stack {

            let component = &self.local[entry.component_index];

            let solution = &component.solutions[entry.solution_index];

    pub sync_round: u32,

const NUM_THREADS: u32 = 3;     // number of threads in runtime

const NUM_INSTANCES: u32 = 5;   // number of test instances constructed

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

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

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

}

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

            synchronous {

                put(req, initial_value);

                auto result = get(resp);

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

            }

            loop_index += 1;

        }

    }

    primitive middle(

        in<u32> req_in, out<u32> req_forward,

        in<u32> resp_in, out<u32> resp_forward,

        u32 num_loops

    ) {

        u32 loop_index = 0;

        while (loop_index < num_loops) {

            synchronous {

                auto req = get(req_in);

                put(req_forward, req + 1);

                auto resp = get(resp_in);

                put(resp_forward, resp + 1);

            }

            loop_index += 1;

        }

    }

    primitive end(in<u32> req_in, out<u32> resp_out, u32 num_loops) {

        u32 loop_index = 0;

        while (loop_index < num_loops) {

            synchronous {

                auto req = get(req_in);

                put(resp_out, req);

            }

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

    }

    ";

    let thing = TestTimer::new("conga_line_request");

    run_test_in_runtime(CODE, |api| {

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

            Value::UInt32(5),

            Value::UInt32(NUM_LOOPS)

        ]));

    });