if port_info.state.is_blocked() {

            return CompScheduling::Sleep;

        }

    }

    sched_ctx.info(&format!("Component starting exit (reason: {:?})", exec_state.exit_reason));

    exec_state.mode = CompMode::BusyExit;

    let exit_inside_sync = exec_state.exit_reason.is_in_sync();

    // If exiting while inside sync mode, report to the leader of the current

    // round that we've failed.

    if exit_inside_sync {

        let decision = consensus.notify_sync_end_failure(sched_ctx, comp_ctx);

        default_handle_sync_decision(sched_ctx, exec_state, comp_ctx, decision, consensus);

    }

    // Iterating over ports by index to work around borrowing rules

    for port_index in 0..comp_ctx.num_ports() {

        let port = comp_ctx.get_port_by_index_mut(port_index);

        if port.state.is_closed() || port.state.is_set(PortStateFlag::Transmitted) || port.close_at_sync_end {

            // Already closed, or in the process of being closed

            continue;

        }

        // Mark as closed

        let port_id = port.self_id;

        port.state.set(PortStateFlag::Closed);

        // Notify peer of closing

        let port_handle = comp_ctx.get_port_handle(port_id);

        let (peer, message) = control.initiate_port_closing(port_handle, exit_inside_sync, comp_ctx);

        let peer_info = comp_ctx.get_peer(peer);

        peer_info.handle.send_message_logged(sched_ctx, Message::Control(message), true);

    }

    return CompScheduling::Immediate; // to check if we can shut down immediately

}

/// Handles a component waiting until all peers are notified that it is quitting

/// (i.e. after calling `default_handle_start_exit`).

#[must_use]

pub(crate) fn default_handle_busy_exit(

    exec_state: &mut CompExecState, control: &ControlLayer,

    sched_ctx: &SchedulerCtx

) -> CompScheduling {

    debug_assert_eq!(exec_state.mode, CompMode::BusyExit);

    if control.has_acks_remaining() {

        sched_ctx.info("Component busy exiting, still has `Ack`s remaining");

        return CompScheduling::Sleep;

    } else {

        sched_ctx.info("Component busy exiting, now shutting down");

        exec_state.mode = CompMode::Exit;

        return CompScheduling::Exit;

    }

}

/// Handles a potential synchronous round decision. If there was a decision then

/// the `Some(success)` value indicates whether the round succeeded or not.

/// Might also end up changing the `ExecState`.

///

/// Might be called in two cases:

/// 1. The component is in regular execution mode, at the end of a sync round,

///     and is waiting for a solution to the round.

/// 2. The component has encountered an error during a sync round and is

///     exiting, hence is waiting for a "Failure" message from the leader.

pub(crate) fn default_handle_sync_decision(

    sched_ctx: &SchedulerCtx, exec_state: &mut CompExecState, comp_ctx: &mut CompCtx,

    decision: SyncRoundDecision, consensus: &mut Consensus

) -> Option<bool> {

    let success = match decision {

        SyncRoundDecision::None => return None,

        SyncRoundDecision::Solution => true,

        SyncRoundDecision::Failure => false,

    };

    debug_assert!(

        exec_state.mode == CompMode::SyncEnd || (

            exec_state.mode.is_busy_exiting() && exec_state.exit_reason.is_error()

        ) || (

            exec_state.mode.is_in_sync_block() && decision == SyncRoundDecision::Failure

        )

    );

    sched_ctx.info(&format!("Handling decision {:?} (in mode: {:?})", decision, exec_state.mode));

    consensus.notify_sync_decision(decision);

    if success {

        // We cannot get a success message if the component has encountered an

        // error.

        for port_index in 0..comp_ctx.num_ports() {

            let port_info = comp_ctx.get_port_by_index_mut(port_index);

            if port_info.close_at_sync_end {

                port_info.state.set(PortStateFlag::Closed);

            }

            port_info.state.clear(PortStateFlag::Received);

        }

        debug_assert_eq!(exec_state.mode, CompMode::SyncEnd);

        exec_state.mode = CompMode::NonSync;

        return Some(true);

    } else {

        // We may get failure both in all possible cases. But we should only

        // modify the execution state if we're not already in exit mode

        if !exec_state.mode.is_busy_exiting() {

            sched_ctx.error("failed synchronous round, initiating exit");

            exec_state.set_as_start_exit(ExitReason::ErrorNonSync);

        }

        return Some(false);

    }

}

/// Special component creation function. This function assumes that the

/// transferred ports are NOT blocked, and that the channels to whom the ports

/// belong are fully owned by the creating component. This will be checked in

/// debug mode.

pub(crate) fn special_create_component(

    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, instantiator_ctx: &mut CompCtx,

    control: &mut ControlLayer, inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup,

    component_instance: Box<dyn Component>, component_ports: Vec<PortId>,

) {

    debug_assert_eq!(exec_state.mode, CompMode::NonSync);

    let reservation = sched_ctx.runtime.start_create_component();

    let mut created_ctx = CompCtx::new(&reservation);

    let mut port_pairs = Vec::new();

    // Retrieve ports

    for instantiator_port_id in component_ports.iter() {

        let instantiator_port_id = *instantiator_port_id;

        let instantiator_port_handle = instantiator_ctx.get_port_handle(instantiator_port_id);

        let instantiator_port = instantiator_ctx.get_port(instantiator_port_handle);

        // Check if conditions for calling this function are valid

        debug_assert!(!instantiator_port.state.is_blocked_due_to_port_change());

        debug_assert_eq!(instantiator_port.peer_comp_id, instantiator_ctx.id);

        // Create port at new component

        let created_port_handle = created_ctx.add_port(

            instantiator_port.peer_comp_id, instantiator_port.peer_port_id,

            instantiator_port.kind, instantiator_port.state

        );

        let created_port = created_ctx.get_port(created_port_handle);

        let created_port_id = created_port.self_id;

        // Store in port pairs

        let is_open = instantiator_port.state.is_open();

        port_pairs.push(PortPair{

            instantiator_id: instantiator_port_id,

            instantiator_handle: instantiator_port_handle,

            created_id: created_port_id,

            created_handle: created_port_handle,

            is_open,

        });

    }

    // Set peer of the port for the new component

    for pair in port_pairs.iter() {

        let instantiator_port_info = instantiator_ctx.get_port(pair.instantiator_handle);

        let created_port_info = created_ctx.get_port_mut(pair.created_handle);

        // Note: we checked above (in debug mode) that the peer of the port is

        // owned by the creator as well, now check if the peer is transferred

        // as well.

        let created_port_peer_index = port_pairs.iter()

            .position(|v| v.instantiator_id == instantiator_port_info.peer_port_id);

        match created_port_peer_index {

            Some(created_port_peer_index) => {

                // Both ends of the channel are moving to the new component

                let peer_pair = &port_pairs[created_port_peer_index];

                created_port_info.peer_port_id = peer_pair.created_id;

                created_port_info.peer_comp_id = reservation.id();

            },

            None => {

                created_port_info.peer_comp_id = instantiator_ctx.id;

                if pair.is_open {

                    created_ctx.change_port_peer(sched_ctx, pair.created_handle, Some(instantiator_ctx.id));

                }

            }

        }

    }

    // Store component in runtime storage and retrieve component fields in their

    // name memory location

    let (created_key, created_runtime_component) = sched_ctx.runtime.finish_create_component(

        reservation, component_instance, created_ctx, false

    );

    let created_ctx = &mut created_runtime_component.ctx;

    let created_component = &mut created_runtime_component.component;

    created_component.on_creation(created_key.downgrade(), sched_ctx);

    // Transfer messages and link instantiator to created component

    for pair in port_pairs.iter() {

        instantiator_ctx.change_port_peer(sched_ctx, pair.instantiator_handle, None);

        transfer_messages(inbox_main, inbox_backup, pair, instantiator_ctx, created_ctx, created_component.as_mut());

        let created_port_info = created_ctx.get_port(pair.created_handle);

        if pair.is_open && created_port_info.peer_comp_id == instantiator_ctx.id {

            // Set up channel between instantiator component port, and its peer,

            // which is owned by the new component

            let instantiator_port_handle = instantiator_ctx.get_port_handle(created_port_info.peer_port_id);

            let instantiator_port_info = instantiator_ctx.get_port_mut(instantiator_port_handle);

            instantiator_port_info.peer_port_id = created_port_info.self_id;

            instantiator_ctx.change_port_peer(sched_ctx, instantiator_port_handle, Some(created_ctx.id));

        }

    }

    // By definition we did not have any remote peers for the transferred ports,

    // so we can schedule the new component immediately

    sched_ctx.runtime.enqueue_work(created_key);

}

/// Puts the component in an execution state where the specified component will

/// end up being created. The component goes through state changes (driven by

/// incoming control messages) to make sure that all of the ports that are going

/// to be transferred are not in a blocked state. Once finished the component

/// returns to the `NonSync` mode.

pub(crate) fn default_start_create_component(

    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    control: &mut ControlLayer, inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup,

    definition_id: ProcedureDefinitionId, type_id: TypeId, arguments: ValueGroup

) {

    debug_assert_eq!(exec_state.mode, CompMode::NonSync);

    let mut transferred_ports = Vec::new();

    find_ports_in_value_group(&arguments, &mut transferred_ports);

    // Set execution state as waiting until we can create the component. If we

    // can do so right away, then we will.

    exec_state.set_as_create_component_blocked(definition_id, type_id, arguments);

    if ports_not_blocked(comp_ctx, &transferred_ports) {

        perform_create_component(exec_state, sched_ctx, comp_ctx, control, inbox_main, inbox_backup);

    }

}

/// Actually creates a component (and assumes that the caller made sure that

/// none of the ports are involved in a blocking operation).

pub(crate) fn perform_create_component(

    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx, instantiator_ctx: &mut CompCtx,

    control: &mut ControlLayer, inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

) {

    // Retrieve ports from the arguments

    debug_assert_eq!(exec_state.mode, CompMode::NewComponentBlocked);

    let (procedure_id, procedure_type_id) = exec_state.mode_component;

    let mut arguments = exec_state.mode_value.take();

    let mut ports = Vec::new();

    find_ports_in_value_group(&arguments, &mut ports);

    debug_assert!(ports_not_blocked(instantiator_ctx, &ports));

    // Reserve a location for the new component

    let reservation = sched_ctx.runtime.start_create_component();

    let mut created_ctx = CompCtx::new(&reservation);

    let mut port_pairs = Vec::with_capacity(ports.len());

    // Go over all the ports that will be transferred. Since the ports will get

    // a new ID in the new component, we will take care of that here.

    for (port_location, instantiator_port_id) in &ports {

        // Retrieve port information from instantiator

        let instantiator_port_id = *instantiator_port_id;

        let instantiator_port_handle = instantiator_ctx.get_port_handle(instantiator_port_id);

        let instantiator_port = instantiator_ctx.get_port(instantiator_port_handle);

        // Create port at created component

        let created_port_handle = created_ctx.add_port(

            instantiator_port.peer_comp_id, instantiator_port.peer_port_id,

            instantiator_port.kind, instantiator_port.state

        );

        let created_port = created_ctx.get_port(created_port_handle);

        let created_port_id = created_port.self_id;

        // Modify port ID in the arguments to the new component and store them

        // for later access

        let is_open = instantiator_port.state.is_open();

        port_pairs.push(PortPair{

            instantiator_id: instantiator_port_id,

            instantiator_handle: instantiator_port_handle,

            created_id: created_port_id,

            created_handle: created_port_handle,

            is_open,

        });

        for location in port_location.iter().copied() {

            let value = arguments.get_value_mut(location);

            match value {

                Value::Input(id) => *id = port_id_to_eval(created_port_id),

                Value::Output(id) => *id = port_id_to_eval(created_port_id),

                _ => unreachable!(),

            }

        }

    }

    // For each of the ports in the newly created component we set the peer to

    // the correct value. We will not yet change the peer on the instantiator's

    // ports (as we haven't yet stored the new component in the runtime's

    // component storage)

    let mut created_component_has_remote_peers = false;

    for pair in port_pairs.iter() {

        let instantiator_port_info = instantiator_ctx.get_port(pair.instantiator_handle);

        let created_port_info = created_ctx.get_port_mut(pair.created_handle);

        if created_port_info.peer_comp_id == instantiator_ctx.id {

            // The peer of the created component's port seems to be the

            // instantiator.

            let created_port_peer_index = port_pairs.iter()

                .position(|v| v.instantiator_id == instantiator_port_info.peer_port_id);

            match created_port_peer_index {

                Some(created_port_peer_index) => {

                    // However, the peer port is also moved to the new

                    // component, so the complete channel is owned by the new

                    // component.

                    let peer_pair = &port_pairs[created_port_peer_index];

                    created_port_info.peer_port_id = peer_pair.created_id;

                    created_port_info.peer_comp_id = reservation.id();

                },

                None => {

                    // Peer port remains with instantiator. However, we cannot

                    // set the peer on the instantiator yet, because the new

                    // component has not yet been stored in the runtime's

                    // component storage. So we do this later

                    created_port_info.peer_comp_id = instantiator_ctx.id;

                    if pair.is_open {

                        created_ctx.change_port_peer(sched_ctx, pair.created_handle, Some(instantiator_ctx.id));

                    }

                }

            }

        } else {

            // Peer is a different component

            if pair.is_open {

                // And the port is still open, so we need to notify the peer

                let peer_handle = instantiator_ctx.get_peer_handle(created_port_info.peer_comp_id);

                let peer_info = instantiator_ctx.get_peer(peer_handle);

                created_ctx.change_port_peer(sched_ctx, pair.created_handle, Some(peer_info.id));

                created_component_has_remote_peers = true;

            }

        }

    }

    // Now we store the new component into the runtime's component storage using

    // the reservation.

    let component = create_component(

        &sched_ctx.runtime.protocol, procedure_id, procedure_type_id,

        arguments, port_pairs.len()

    );

    let (created_key, created_runtime_component) = sched_ctx.runtime.finish_create_component(

        reservation, component, created_ctx, false

    );

    let created_ctx = &mut created_runtime_component.ctx;

    let created_component = &mut created_runtime_component.component;

    created_component.on_creation(created_key.downgrade(), sched_ctx);

    // We now pass along the messages that the instantiator component still has

    // that belong to the new component. At the same time we'll take care of

    // setting the correct peer of the instantiator component

    for pair in port_pairs.iter() {

        // Transferring the messages and removing the port from the

        // instantiator component

        instantiator_ctx.change_port_peer(sched_ctx, pair.instantiator_handle, None);

        transfer_messages(inbox_main, inbox_backup, pair, instantiator_ctx, created_ctx, created_component.as_mut());

        // Here we take care of the case where the instantiator previously owned

        // both ends of the channel, but has transferred one port to the new

        // component (hence creating a channel between the instantiator

        // component and the new component).

        let created_port_info = created_ctx.get_port(pair.created_handle);

        if pair.is_open && created_port_info.peer_comp_id == instantiator_ctx.id {

            // Note: the port we're receiving here belongs to the instantiator

            // and is NOT in the "port_pairs" array.

            let instantiator_port_handle = instantiator_ctx.get_port_handle(created_port_info.peer_port_id);

            let instantiator_port_info = instantiator_ctx.get_port_mut(instantiator_port_handle);

            instantiator_port_info.peer_port_id = created_port_info.self_id;

            instantiator_ctx.change_port_peer(sched_ctx, instantiator_port_handle, Some(created_ctx.id));

        }

    }

    // Finally: if we did move ports around whose peers are different

    // components, then we'll initiate the appropriate protocol to notify them.

    if created_component_has_remote_peers {

        let schedule_entry_id = control.add_schedule_entry(created_ctx.id);

        for pair in &port_pairs {

            let port_info = created_ctx.get_port(pair.created_handle);

            if pair.is_open && port_info.peer_comp_id != instantiator_ctx.id && port_info.peer_comp_id != created_ctx.id {

                // Peer component is not the instantiator, and it is not the

                // new component itself

                let message = control.add_reroute_entry(

                    instantiator_ctx.id, port_info.peer_port_id, port_info.peer_comp_id,

                    pair.instantiator_id, pair.created_id, created_ctx.id,

                    schedule_entry_id

                );

                let peer_handle = created_ctx.get_peer_handle(port_info.peer_comp_id);

                let peer_info = created_ctx.get_peer(peer_handle);

                peer_info.handle.send_message_logged(sched_ctx, message, true);

            }

        }

    } else {

        // We can schedule the component immediately, we do not have to wait

        // for any peers: there are none.

        sched_ctx.runtime.enqueue_work(created_key);

    }

    exec_state.mode = CompMode::NonSync;

    exec_state.mode_component = (ProcedureDefinitionId::new_invalid(), TypeId::new_invalid());

}

#[inline]

pub(crate) fn default_handle_exit(_exec_state: &CompExecState) -> CompScheduling {

    debug_assert_eq!(_exec_state.mode, CompMode::Exit);

    return CompScheduling::Exit;

}

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

// Internal messaging/state utilities

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

struct PortPair {

    instantiator_id: PortId,

    instantiator_handle: LocalPortHandle,

    created_id: PortId,

    created_handle: LocalPortHandle,

    is_open: bool,

}

pub(crate) fn ports_not_blocked(comp_ctx: &CompCtx, ports: &EncounteredPorts) -> bool {

    for (_port_locations, port_id) in ports {

        let port_handle = comp_ctx.get_port_handle(*port_id);

        let port_info = comp_ctx.get_port(port_handle);

        if port_info.state.is_blocked_due_to_port_change() {

            return false;

        }

    }

    return true;

}

fn transfer_messages(

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup, port_pair: &PortPair,

    instantiator_ctx: &mut CompCtx, created_ctx: &mut CompCtx, created_component: &mut dyn Component

) {

    let instantiator_port_index = instantiator_ctx.get_port_index(port_pair.instantiator_handle);

    if let Some(mut message) = inbox_main.remove(instantiator_port_index) {

        message.data_header.target_port = port_pair.created_id;

        created_component.adopt_message(created_ctx, message);

    }

    let mut message_index = 0;

    while message_index < inbox_backup.len() {

        let message = &inbox_backup[message_index];

        if message.data_header.target_port == port_pair.instantiator_id {

            // Transfer the message

            let mut message = inbox_backup.remove(message_index);

            message.data_header.target_port = port_pair.created_id;

            created_component.adopt_message(created_ctx, message);

        } else {

            // Message does not belong to the port pair that we're

            // transferring to the new component.

            message_index += 1;

        }

    }

}

/// Performs the default action of printing the provided error, and then putting

/// the component in the state where it will shut down. Only to be used for

/// builtin components: their error message construction is simpler (and more

/// common) as they don't have any source code.

pub(crate) fn default_handle_error_for_builtin(

    exec_state: &mut CompExecState, sched_ctx: &SchedulerCtx,

    location_and_message: (PortInstruction, String)

) {

    let (_location, message) = location_and_message;

    sched_ctx.error(&message);

    let exit_reason = if exec_state.mode.is_in_sync_block() {

        ExitReason::ErrorInSync

    } else {

        ExitReason::ErrorNonSync

    };

    exec_state.set_as_start_exit(exit_reason);

}

/// Sends a message without any transmitted ports. Does not check if sending

/// is actually valid.

fn send_message_without_ports(

    sending_port_handle: LocalPortHandle, value: ValueGroup,

    comp_ctx: &CompCtx, sched_ctx: &SchedulerCtx, consensus: &mut Consensus,

) {

    let port_info = comp_ctx.get_port(sending_port_handle);

    debug_assert!(port_info.state.can_send());

    let peer_handle = comp_ctx.get_peer_handle(port_info.peer_comp_id);

    let peer_info = comp_ctx.get_peer(peer_handle);

    let annotated_message = consensus.annotate_data_message(comp_ctx, port_info, value);

    peer_info.handle.send_message_logged(sched_ctx, Message::Data(annotated_message), true);

}

/// Prepares sending a message that contains ports. Only once a particular

/// protocol has completed (where we notify all the peers that the ports will

/// be transferred) will we actually send the message to the recipient.

fn start_send_message_with_ports(

    sending_port_id: PortId, sending_port_instruction: PortInstruction, value: ValueGroup,

    exec_state: &mut CompExecState, comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx,

    control: &mut ControlLayer

) -> Result<(), (PortInstruction, String)> {

    debug_assert_eq!(exec_state.mode, CompMode::Sync); // busy in sync, trying to send

    // Retrieve ports we're going to transfer

    let sending_port_handle = comp_ctx.get_port_handle(sending_port_id);

    let sending_port_info = comp_ctx.get_port_mut(sending_port_handle);

    sending_port_info.last_instruction = sending_port_instruction;

    let mut transmit_ports = Vec::new();

    find_ports_in_value_group(&value, &mut transmit_ports);

    debug_assert!(!transmit_ports.is_empty()); // required from caller

    // Enter the state where we'll wait until all transferred ports are not

    // blocked.

    exec_state.set_as_blocked_put_with_ports(sending_port_id, value);

    if ports_not_blocked(comp_ctx, &transmit_ports) {

        // Ports are not blocked, so we can send them right away.

        perform_send_message_with_ports_notify_peers(

            exec_state, comp_ctx, sched_ctx, control, transmit_ports

        )?;

    } // else: wait until they become unblocked

    return Ok(())

}

fn perform_send_message_with_ports_notify_peers(

    exec_state: &mut CompExecState, comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx,

    control: &mut ControlLayer, transmit_ports: EncounteredPorts

) -> Result<(), (PortInstruction, String)> {

    // Check we're in the correct state in debug mode

    debug_assert_eq!(exec_state.mode, CompMode::PutPortsBlockedTransferredPorts);

    debug_assert!(ports_not_blocked(comp_ctx, &transmit_ports));

    // Set up the final Ack that triggers us to send our final message

    let unblock_put_control_id = control.add_unblock_put_with_ports_entry();

    for (_, port_id) in &transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(*port_id);

        let transmit_port_info = comp_ctx.get_port_mut(transmit_port_handle);

        let peer_comp_id = transmit_port_info.peer_comp_id;

        let peer_port_id = transmit_port_info.peer_port_id;

        // Note: we checked earlier that we are currently in sync mode. Now we

        // will check if we've already used the port we're about to transmit.

        if !transmit_port_info.last_instruction.is_none() {

            let sending_port_handle = comp_ctx.get_port_handle(exec_state.mode_port);

use super::*;

// silly test to make sure that the PDL will never be an issue when doing TCP

// stuff with the actual components

#[test]

fn test_stdlib_file() {

    compile_and_create_component("

    import std.internet as inet;

    comp fake_listener_once(out<inet::TcpConnection> tx) {

        channel cmd_tx -> cmd_rx;

        channel data_tx -> data_rx;

        new fake_socket(cmd_rx, data_tx);

        sync put(tx, inet::TcpConnection{

            tx: cmd_tx,

            rx: data_rx,

        });

    }

        auto to_send = {};

        auto shutdown = false;

        while (!shutdown) {

            auto keep_going = true;

            sync {

                while (keep_going) {

                    auto cmd = get(cmds);

                        to_send = data;

                        keep_going = false;

                        put(tx, to_send);

                        keep_going = false;

                        keep_going = false;

                        shutdown = true;

                    }

                }

            }

        }

    }

    comp fake_client(inet::TcpConnection conn) {

        sync {

            auto val = get(conn.rx);

            while (val[0] != 1 || val[1] != 3 || val[2] != 3 || val[3] != 7) {

                print(\"this is going very wrong\");

            }

        }

    }

    comp constructor() {

        channel conn_tx -> conn_rx;

        new fake_listener_once(conn_tx);

        // Same crap as before:

        channel cmd_tx -> unused_cmd_rx;

        channel unused_data_tx -> data_rx;

        auto connection = inet::TcpConnection{ tx: cmd_tx, rx: data_rx };

        sync {

            connection = get(conn_rx);

        }

        new fake_client(connection);

    }

    ", "constructor", no_args());

}

use crate::protocol::*;

use crate::protocol::eval::*;

use crate::runtime2::runtime::*;

use crate::runtime2::component::{CompCtx, CompPDL};

mod messaging;

mod error_handling;

mod transfer_ports;

mod internet;

const LOG_LEVEL: LogLevel = LogLevel::Debug;

const NUM_THREADS: u32 = 1;

pub(crate) fn compile_and_create_component(source: &str, routine_name: &str, args: ValueGroup) {

    let protocol = ProtocolDescription::parse(source.as_bytes())

        .expect("successful compilation");

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

        .expect("successful runtime startup");

    create_component(&runtime, "", routine_name, args);

}

pub(crate) fn create_component(rt: &Runtime, module_name: &str, routine_name: &str, args: ValueGroup) {

    let prompt = rt.inner.protocol.new_component(

        module_name.as_bytes(), routine_name.as_bytes(), args

    ).expect("create prompt");

    let reserved = rt.inner.start_create_component();

    let ctx = CompCtx::new(&reserved);

    let component = Box::new(CompPDL::new(prompt, 0));

    let (key, _) = rt.inner.finish_create_component(reserved, component, ctx, false);

    rt.inner.enqueue_work(key);

}

pub(crate) fn no_args() -> ValueGroup { ValueGroup::new_stack(Vec::new()) }

#[test]

fn test_component_creation() {

    let pd = ProtocolDescription::parse(b"

    comp nothing_at_all() {

        s32 a = 5;

        auto b = 5 + a;

    }

    ").expect("compilation");

    let rt = Runtime::new(1, LOG_LEVEL, pd).unwrap();

    for _i in 0..20 {

        create_component(&rt, "", "nothing_at_all", no_args());

    }

}

#[test]

fn test_simple_select() {

    let pd = ProtocolDescription::parse(b"

    func infinite_assert<T>(T val, T expected) -> () {

        while (val != expected) { print(\"nope!\"); }

        return ();

    }

    comp receiver(in<u32> in_a, in<u32> in_b, u32 num_sends) {

        auto num_from_a = 0;

        auto num_from_b = 0;

        while (num_from_a + num_from_b < 2 * num_sends) {

            sync select {

                auto v = get(in_a) -> {

                    print(\"got something from A\");

                    auto _ = infinite_assert(v, num_from_a);

                    num_from_a += 1;

                }

                auto v = get(in_b) -> {

                    print(\"got something from B\");

                    auto _ = infinite_assert(v, num_from_b);

                    num_from_b += 1;

                }

            }

        }

    }

    comp sender(out<u32> tx, u32 num_sends) {

        auto index = 0;

        while (index < num_sends) {

            sync {

                put(tx, index);

                index += 1;

            }

        }

    }

    comp constructor() {

        auto num_sends = 1;

        channel tx_a -> rx_a;

        channel tx_b -> rx_b;

        new sender(tx_a, num_sends);

        new receiver(rx_a, rx_b, num_sends);

        new sender(tx_b, num_sends);

    }

    ").expect("compilation");

    let rt = Runtime::new(3, LOG_LEVEL, pd).unwrap();

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_unguarded_select() {

    let pd = ProtocolDescription::parse(b"

    comp constructor_outside_select() {

        u32 index = 0;

        while (index < 5) {

            sync select { auto v = () -> print(\"hello\"); }

            index += 1;

        }

    }

    comp constructor_inside_select() {

        u32 index = 0;

        while (index < 5) {

            sync select { auto v = () -> index += 1; }

        }

    }

    ").expect("compilation");

    let rt = Runtime::new(3, LOG_LEVEL, pd).unwrap();

    create_component(&rt, "", "constructor_outside_select", no_args());

    create_component(&rt, "", "constructor_inside_select", no_args());

}

#[test]

fn test_empty_select() {

    let pd = ProtocolDescription::parse(b"

    comp constructor() {

        u32 index = 0;

        while (index < 5) {

            sync select {}

            index += 1;

        }

    }

    ").expect("compilation");

    let rt = Runtime::new(3, LOG_LEVEL, pd).unwrap();

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_random_u32_temporary_thingo() {

    let pd = ProtocolDescription::parse(b"

    import std.random::random_u32;

    comp random_taker(in<u32> generator, u32 num_values) {

        auto i = 0;

        while (i < num_values) {

            sync {

                auto a = get(generator);

            }

            i += 1;

        }

    }

    comp constructor() {

        channel tx -> rx;

        auto num_values = 25;

        new random_u32(tx, 1, 100, num_values);

        new random_taker(rx, num_values);

    }

    ").expect("compilation");

    let rt = Runtime::new(1, LOG_LEVEL, pd).unwrap();

    create_component(&rt, "", "constructor", no_args());

}

#[test]

fn test_tcp_socket_http_request() {

    let _pd = ProtocolDescription::parse(b"

    import std.internet::*;

        print(\"*** TCPSocket: Sending request\");

        sync {

        }

        print(\"*** TCPSocket: Receiving response\");

        auto buffer = {};

        auto done_receiving = false;

        sync while (!done_receiving) {

            auto data = get(data_rx);

            buffer @= data;

            // Completely crap detection of end-of-document. But here we go, we

            // try to detect the trailing </html>. Proper way would be to parse

            // for 'content-length' or 'content-encoding'

            s32 index = 0;

            s32 partial_length = cast(length(data) - 7);

            while (index < partial_length) {

                // No string conversion yet, so check byte buffer one byte at

                // a time.

                auto c1 = data[index];

                if (c1 == cast('<')) {

                    auto c2 = data[index + 1];

                    auto c3 = data[index + 2];

                    auto c4 = data[index + 3];

                    auto c5 = data[index + 4];

                    auto c6 = data[index + 5];

                    auto c7 = data[index + 6];

                    if ( // i.e. if (data[index..] == '</html>'

                        c2 == cast('/') && c3 == cast('h') && c4 == cast('t') &&

                        c5 == cast('m') && c6 == cast('l') && c7 == cast('>')

                    ) {

                        print(\"*** TCPSocket: Detected </html>\");

                        done_receiving = true;

                    }

                }

                index += 1;

            }

        }

        print(\"*** TCPSocket: Requesting shutdown\");

        sync {

        }

    }

    comp main() {

        channel cmd_tx -> cmd_rx;

        channel data_tx -> data_rx;

        new tcp_client({142, 250, 179, 163}, 80, cmd_rx, data_tx); // port 80 of google

        new requester(cmd_tx, data_rx);