// Make sure directory exists

        let path = Path::new(&base_path);

        if !path.exists() {

        }

        // Try to load all standard library files. We might need a more unified

use crate::protocol::*;

use crate::runtime2::*;

use crate::runtime2::communication::*;

use super::{CompCtx, CompPDL, CompId};

use super::component_context::*;

pub(crate) fn default_send_data_message(

    exec_state: &mut CompExecState, transmitting_port_id: PortId,

    port_instruction: PortInstruction, value: ValueGroup,

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

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

        let mut ports = Vec::new();

        find_ports_in_value_group(&value, &mut ports);

        if !ports.is_empty() {

        }

    }

}

/// of full buffers (hence, will use the control layer to make sure the peer

/// will become unblocked).

pub(crate) fn default_handle_received_data_message(

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

    comp_ctx: &mut CompCtx, sched_ctx: &SchedulerCtx, control: &mut ControlLayer

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

    let port_handle = comp_ctx.get_port_handle(targeted_port);

    let port_index = comp_ctx.get_port_index(port_handle);

    // If we received any ports, add them to the port tracking and inbox struct.

    // Then notify the peers that they can continue sending to this port, but

    // now at a new address.

    for received_port in &mut message.ports {

        // Transfer messages to main/backup inbox

        if !received_port.messages.is_empty() {

            inbox_main.push(Some(received_port.messages.remove(0)));

        }

            received_port.peer_comp, received_port.peer_port,

            received_port.kind, new_port_state

        );

        let new_port = comp_ctx.get_port(new_port_handle);

        // Replace all references to the port in the received message

            let value = match message_location {

                ValueId::Heap(heap_pos, heap_index) => &mut message.content.regions[heap_pos as usize][heap_index as usize],

                ValueId::Stack(stack_index) => &mut message.content.values[stack_index as usize],

            id: ControlId::new_invalid(),

            sender_comp_id: comp_ctx.id,

            target_port_id: Some(new_port.peer_port_id),

        }), true);

    }

            // One more message, place it in the slot

            let message = inbox_backup.remove(message_index);

            debug_assert!(comp_ctx.get_port(port_handle).state.is_blocked()); // since we're removing another message from the backup

            return Ok(());

        }

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

    match message.content {

        ControlMessageContent::Ack => {

        },

        ControlMessageContent::BlockPort => {

            // One of our messages was accepted, but the port should be

                // The two components (sender and this component) are closing

                // the channel at the same time. So we don't care about the

                // content of the `ClosePort` message.

            } else {

                // Respond to the message

                let port_info = comp_ctx.get_port(port_handle);

            let port_handle = comp_ctx.get_port_handle(port_to_change);

            let port_info = comp_ctx.get_port(port_handle);

            debug_assert!(port_info.state.is_set(PortStateFlag::BlockedDueToPeerChange));

            let port_info = comp_ctx.get_port_mut(port_handle);

            port_info.peer_port_id = new_port_id;

            if port_info.close_at_sync_end {

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

            }

        }

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

        exec_state.mode = CompMode::NonSync;

    // And further enhance the message by adding data about the ports that are

    // being transferred

        let transmit_port_info = comp_ctx.get_port(transmit_port_handle);

        let transmit_messages = take_port_messages(comp_ctx, transmit_port_id, inbox_main, inbox_backup);

/// Handles an `Ack` for the control layer.

fn default_handle_ack(

    exec_state: &mut CompExecState, control: &mut ControlLayer, control_id: ControlId,

) {

    // Since an `Ack` may cause another one, handle them in a loop

    let mut to_ack = control_id;

    // Clear the ports, then scan all the available values

    ports.clear();

        find_port_in_value(value_group, value, ValueId::Stack(value_index as u32), ports);

    }

}

        return Channel{ putter_id, getter_id };

    }

    pub(crate) fn add_port(&mut self, peer_comp_id: CompId, peer_port_id: PortId, kind: PortKind, state: PortState) -> LocalPortHandle {

        let self_id = PortId(self.take_port_id());

        self.ports.push(Port{

        return LocalPortHandle(self_id);

    }

    pub(crate) fn remove_port(&mut self, port_handle: LocalPortHandle) -> Port {

        let port_index = self.must_get_port_index(port_handle);

        let port = self.ports.remove(port_index);

            Message::Control(message) => {

                if let Err(location_and_message) = component::default_handle_control_message(

                    &mut self.exec_state, &mut self.control, &mut self.consensus,

                ) {

                    component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);

                }

        sched_ctx.info(&format!("Running component ComponentTcpClient (mode: {:?}, sync state: {:?})", self.exec_state.mode, self.sync_state));

        match self.exec_state.mode {

                // Not possible: we never enter this state

                unreachable!();

            },

                            Ok(num_received) => {

                                self.byte_buffer.resize(num_received, 0);

                                let message_content = self.bytes_to_data_message_content(&self.byte_buffer);

                                if let Err(location_and_message) = send_result {

                                    component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);

                                    return CompScheduling::Immediate;

    // Should be same length as the number of ports. Corresponding indices imply

    // message is intended for that port.

    pub inbox_main: InboxMain,

}

impl Component for CompPDL {

            Message::Control(message) => {

                if let Err(location_and_message) = component::default_handle_control_message(

                    &mut self.exec_state, &mut self.control, &mut self.consensus,

                ) {

                    self.handle_generic_component_error(sched_ctx, location_and_message);

                }

            CompMode::NonSync | CompMode::Sync => {

                // continue and run PDL code

            },

                return CompScheduling::Sleep;

            }

            CompMode::StartExit => return component::default_handle_start_exit(

                let send_result = component::default_send_data_message(

                    &mut self.exec_state, target_port_id,

                    PortInstruction::SourceLocation(expr_id), value,

                );

                if let Err(location_and_message) = send_result {

                    self.handle_generic_component_error(sched_ctx, location_and_message);

        let mut opened_port_id_pairs = Vec::new();

        let mut closed_port_id_pairs = Vec::new();

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

        // Take all the ports ID that are in the `args` (and currently belong to

        // the creator component) and translate them into new IDs that are

                        let peer_pair = &opened_port_id_pairs[created_peer_port_index];

                        created_port_info.peer_port_id = peer_pair.created_id;

                        created_port_info.peer_comp_id = reservation.id();

                    },

                    None => {

                        // Peer port remains with creator component.

        // to message exchanges between remote peers.

        let total_num_ports = opened_port_id_pairs.len() + closed_port_id_pairs.len();

        let component = component::create_component(&sched_ctx.runtime.protocol, definition_id, type_id, arguments, total_num_ports);

            reservation, component, created_ctx, false,

        );

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

use rand::RngCore;

use crate::protocol::eval::{ValueGroup, Value};

use super::*;

use super::control_layer::*;

use super::consensus::*;

    did_perform_send: bool, // when in sync mode

    control: ControlLayer,

    consensus: Consensus,

}

impl Component for ComponentRandomU32 {

            Message::Control(message) => {

                if let Err(location_and_message) = component::default_handle_control_message(

                    &mut self.exec_state, &mut self.control, &mut self.consensus,

                ) {

                    component::default_handle_error_for_builtin(&mut self.exec_state, sched_ctx, location_and_message);

                }

        sched_ctx.info(&format!("Running component ComponentRandomU32 (mode: {:?})", self.exec_state.mode));

        match self.exec_state.mode {

                // impossible for this component, no input ports and no select

                // blocks

                unreachable!();

                    let send_result = component::default_send_data_message(

                        &mut self.exec_state, self.output_port_id,

                        PortInstruction::NoSource, value_group,

                    );

                    if let Err(location_and_message) = send_result {

            did_perform_send: false,

            control: ControlLayer::default(),

            consensus: Consensus::new(),

        }

    }

}

        let data_header = self.create_data_header_and_update_mapping(port_info);

        let sync_header = self.create_sync_header(comp_ctx);

    }

    /// Handles the arrival of a new data message (needs to be called for every

/// Represents an unrecoverable runtime error that is reported to the user (for

/// debugging purposes). Basically a human-readable message with its source

    }

    pub(crate) fn run(&mut self) {

        use crate::runtime2::communication::Message;

        const NUM_EVENTS: usize = 256;

        loop {

            // Retrieve events first (because the PollingClient will first

            // register at epoll, and then push a command into the queue).

            // Then handle everything in the command queue.

            while let Some(command) = self.queue.pop() {

    pub exiting: bool,

}

// TODO: Do something about the `num_handles` thing. This needs to be a bit more

//  "foolproof" to lighten the mental burden of using the `num_handles`

//  variable.

}

impl CompHandle {

            target: public,

            id,

            #[cfg(debug_assertions)] decremented: false,

        };

        handle.increment_users();

        return handle;

    }

            module_name, routine_name,

            ValueGroup::new_stack(Vec::new())

        )?;

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

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

        self.inner.enqueue_work(key);

        return Ok(())

    // Creating/destroying components

        self.increment_active_components();

        let reservation = self.components.reserve();

        return CompReserved{ reservation };

    }

        &self, reserved: CompReserved,

        component: Box<dyn Component>, mut context: CompCtx, initially_sleeping: bool,

    ) -> (CompKey, &mut RuntimeComp) {

        let inbox_queue = QueueDynMpsc::new(16);

        let inbox_producer = inbox_queue.producer();

            public: CompPublic{

                sleeping: AtomicBool::new(initially_sleeping),

                num_handles: AtomicU32::new(1), // the component itself acts like a handle

            exiting: false,

        };

        let index = self.components.submit(reserved.reservation, component);

        let component = self.components.get_mut(index);

        return (CompKey(index), component);

    }

                CompScheduling::Sleep => { self.mark_component_as_sleeping(comp_key, component); },

                CompScheduling::Exit => {

                    component.component.on_shutdown(&scheduler_ctx);

                }

            }

        }

    /// Marks the component as exiting by removing the reference it holds to

    /// itself. Afterward the component will enter "normal" sleeping mode (if it

    /// has not yet been destroyed)

        // If we didn't yet decrement our reference count, do so now

        if !component.exiting {

            component.exiting = true;

                sched_ctx.runtime.destroy_component(comp_key);

                return;

            }

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

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

mod error_handling;

const LOG_LEVEL: LogLevel = LogLevel::Debug;

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

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

    let prompt = rt.inner.protocol.new_component(

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

    ).expect("create prompt");

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

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

    rt.inner.enqueue_work(key);

}

    }

}