(path, false)

        };

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

        // way to do this in the future (i.e. a "std" package, containing all

        // of the modules)

        let mut file_path = PathBuf::new();

use std::fmt::{Display as FmtDisplay, Result as FmtResult, Formatter};

use crate::protocol::eval::{Prompt, EvalError, ValueGroup, Value, ValueId, PortId as EvalPortId};

use crate::protocol::*;

use crate::runtime2::*;

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

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

use super::component_context::*;

use super::component_random::*;

use super::component_internet::*;

use super::control_layer::*;

/// unblocked if so instructed by the receiving component. The returned

/// scheduling value must be used.

#[must_use]

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 port_handle = comp_ctx.get_port_handle(transmitting_port_id);

    let port_info = comp_ctx.get_port_mut(port_handle);

    port_info.last_instruction = port_instruction;

        return Ok(CompScheduling::Sleep);

    } else {

        // Check if there are any ports that are being transmitted

        let mut ports = Vec::new();

        find_ports_in_value_group(&value, &mut ports);

        if !ports.is_empty() {

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

            return Ok(CompScheduling::Immediate);

        }

    }

pub(crate) enum IncomingData {

    PlacedInSlot,

    SlotFull(DataMessage),

}

/// Default handling that has been received through a `get`. Will check if any

/// more messages are waiting, and if the corresponding port was blocked because

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

        }

        inbox_backup.extend(received_port.messages.drain(..));

        // Create a new port locally

        let mut new_port_state = received_port.state;

        new_port_state.set(PortStateFlag::Received);

        let new_port_handle = comp_ctx.add_port(

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

            };

            match value {

        let peer_info = comp_ctx.get_peer(peer_handle);

        peer_info.handle.send_message_logged(sched_ctx, Message::Control(ControlMessage{

            id: ControlId::new_invalid(),

            sender_comp_id: comp_ctx.id,

            target_port_id: Some(new_port.peer_port_id),

        }), true);

    }

    // Modify last-known location where port instruction was retrieved

    let port_info = comp_ctx.get_port(port_handle);

    debug_assert_ne!(port_info.last_instruction, PortInstruction::None); // set by caller

    for message_index in 0..inbox_backup.len() {

        let message = &inbox_backup[message_index];

        if message.data_header.target_port == targeted_port {

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

        }

    }

    // Did not have any more messages, so if we were blocked, then we need to

    exec_state: &mut CompExecState, control: &mut ControlLayer, consensus: &mut Consensus,

    message: ControlMessage, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

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

    match message.content {

        ControlMessageContent::Ack => {

        },

        ControlMessageContent::BlockPort => {

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

            // blocked.

            let port_to_block = message.target_port_id.unwrap();

            let port_handle = comp_ctx.get_port_handle(port_to_block);

            // port ourselves, meaning that the `ClosePort` messages got

            // sent to one another.

            if let Some(control_id) = control.has_close_port_entry(port_handle, comp_ctx) {

                // 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 last_instruction = port_info.last_instruction;

                let port_has_had_message = port_info.received_message_for_sync;

                default_send_ack(message.id, peer_handle, sched_ctx, comp_ctx);

        },

        ControlMessageContent::PortPeerChangedUnblock(new_port_id, new_comp_id) => {

            let port_to_change = message.target_port_id.unwrap();

            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;

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

            comp_ctx.change_port_peer(sched_ctx, port_handle, Some(new_comp_id));

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

            }

        }

        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

    // Annotate the data message

    let message_value = exec_state.mode_value.take();

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

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

        let mut transmit_port_state = transmit_port_info.state;

        debug_assert!(transmit_port_state.is_set(PortStateFlag::Transmitted));

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

}

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

    loop {

        let (action, new_to_ack) = control.handle_ack(to_ack, sched_ctx, comp_ctx);

        match action {

            _ => {}, // values we don't care about

        }

    }

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

    }

}

/// Goes through the inbox of a component and takes out all the messages that

/// are targeted at a specific port

            associated_with_peer: false,

        });

        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{

            self_id, peer_comp_id, peer_port_id, kind, state,

            last_instruction: PortInstruction::None,

            close_at_sync_end: false,

            received_message_for_sync: false,

            associated_with_peer: false,

        });

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

        dbg_code!(assert!(!port.associated_with_peer));

        return port;

    }

                let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);

                component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);

            },

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

                }

            },

            Message::Poll => {

                sched_ctx.info("Received polling event");

    }

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {

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

            },

            CompMode::NonSync => {

                // When in non-sync mode

                match &mut self.socket_state {

                        let socket = self.socket_state.get_socket();

                        self.byte_buffer.resize(BUFFER_SIZE, 0);

                        match socket.receive(&mut self.byte_buffer) {

                            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;

                                } else {

                                    let scheduling = send_result.unwrap();

                                    self.sync_state = SyncState::AwaitingCmd;

    pub exec_ctx: ExecCtx,

    // TODO: Temporary field, simulates future plans of having one storage place

    //  reserved per port.

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

    fn on_creation(&mut self, _id: CompId, _sched_ctx: &SchedulerCtx) {

        // Intentionally empty

    }

            Message::Data(message) => {

                self.handle_incoming_data_message(sched_ctx, comp_ctx, message);

            },

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

                }

            },

            Message::Sync(message) => {

                self.handle_incoming_sync_message(sched_ctx, comp_ctx, message);

        // Depending on the mode don't do anything at all, take some special

        // actions, or fall through and run the PDL code.

        match self.exec_state.mode {

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

                // continue and run PDL code

            },

                return CompScheduling::Sleep;

            }

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

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

            ),

            CompMode::BusyExit => return component::default_handle_busy_exit(

                // Send the message

                let target_port_id = port_id_from_eval(port_id);

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

                    return CompScheduling::Immediate;

                } else {

                    // When `run` is called again (potentially after becoming

            created_handle: LocalPortHandle,

            created_id: PortId,

        }

        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

        // associated with the component we're about to create

        let mut arg_iter = ValueGroupPortIter::new(&mut arguments);

        while let Some(port_reference) = arg_iter.next() {

                    Some(created_peer_port_index) => {

                        // Addendum to the above comment: but that port is also

                        // moving to the new component

                        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.

                        created_port_info.peer_comp_id = creator_ctx.id;

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

                    }

        // We'll now actually turn our reservation for a new component into an

        // actual component. Note that we initialize it as "not sleeping" as

        // its initial scheduling might be performed based on `Ack`s in response

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

        // Now modify the creator's ports: remove every transferred port and

        // potentially remove the peer component.

use rand::prelude as random;

use rand::RngCore;

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

use super::*;

use super::control_layer::*;

use super::consensus::*;

/// TODO: Temporary component to figure out what to do with custom components.

///     This component sends random numbers between two u32 limits

pub struct ComponentRandomU32 {

    generator: random::ThreadRng,

    // Generic state-tracking

    exec_state: CompExecState,

    did_perform_send: bool, // when in sync mode

    control: ControlLayer,

    consensus: Consensus,

}

impl Component for ComponentRandomU32 {

    fn on_creation(&mut self, _id: CompId, _sched_ctx: &SchedulerCtx) {}

    fn on_shutdown(&mut self, sched_ctx: &SchedulerCtx) {}

                let decision = self.consensus.receive_sync_message(sched_ctx, comp_ctx, message);

                component::default_handle_sync_decision(sched_ctx, &mut self.exec_state, comp_ctx, decision, &mut self.consensus);

            },

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

                }

            },

            Message::Poll => unreachable!(),

        }

    }

    fn run(&mut self, sched_ctx: &mut SchedulerCtx, comp_ctx: &mut CompCtx) -> CompScheduling {

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

            }

            CompMode::NonSync => {

                // If in non-sync mode then we check if the arguments make sense

                    random += self.random_minimum;

                    let value_group = ValueGroup::new_stack(vec![Value::UInt32(random)]);

                    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 {

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

                        return CompScheduling::Immediate

                    } else {

            max_num_sends: num_sends,

            generator: random::thread_rng(),

            exec_state: CompExecState::new(),

            did_perform_send: false,

            control: ControlLayer::default(),

            consensus: Consensus::new(),

        }

    }

}

    pub(crate) fn annotate_data_message(&mut self, comp_ctx: &CompCtx, port_info: &Port, content: ValueGroup) -> DataMessage {

        debug_assert_eq!(self.mode, Mode::SyncBusy); // can only send between sync start and sync end

        debug_assert!(self.ports.iter().any(|v| v.self_port_id == port_info.self_id));

        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

    /// new data message, even though it might not end up being received). This

    /// is used to determine peers of `get`ter ports.

    // TODO: The use of this function is rather ugly. Find a more robust

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

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

/// location. The error is chainable.

pub struct RtError {

    file: &'static str,

        };

        return Ok((thread_handle, client_factory));

    }

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

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

        const NUM_EVENTS: usize = 256;

        const EPOLL_DURATION: time::Duration = time::Duration::from_millis(250);

        // @performance: Lot of improvements possible here, a HashMap is likely

        let mut lookup = HashMap::with_capacity(64);

        self.log("Starting polling thread");

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

                match command {

                    PollCmd::Register(handle, user_data) => {

                        self.log(&format!("Registering component {:?} as {}", handle.id(), user_data.0));

    pub component: Box<dyn Component>,

    pub ctx: CompCtx,

    pub inbox: QueueDynMpsc<Message>,

    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.

pub(crate) struct CompPublic {

    pub sleeping: AtomicBool,

    pub num_handles: AtomicU32, // manually modified (!)

    target: *const CompPublic,

    id: CompId,

    #[cfg(debug_assertions)] decremented: bool,

}

impl CompHandle {

            target: public,

            id,

            #[cfg(debug_assertions)] decremented: false,

        };

        handle.increment_users();

        return handle;

    }

    pub(crate) fn send_message(&self, runtime: &RuntimeInner, message: Message, try_wake_up: bool) {

        self.inbox.push(message);

    pub fn create_component(&self, module_name: &[u8], routine_name: &[u8]) -> Result<(), ComponentCreationError> {

        use crate::protocol::eval::ValueGroup;

        let prompt = self.inner.protocol.new_component(

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

    }

}

        lock.push_back(key);

        self.work_condvar.notify_one();

    }

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

                inbox: inbox_producer,

            },

            component,

            ctx: context,

            inbox: inbox_queue,

            exiting: false,

        };

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

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

        return (CompKey(index), component);

    }

    pub(crate) fn get_component(&self, key: CompKey) -> &mut RuntimeComp {

        let component = self.components.get_mut(key.0);

        return component;

            match new_scheduling {

                CompScheduling::Immediate => unreachable!(),

                CompScheduling::Requeue => { self.runtime.enqueue_work(comp_key); },

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

                CompScheduling::Exit => {

                    component.component.on_shutdown(&scheduler_ctx);

                }

            }

        }

    }

    // local utilities

        }

    }

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

            }

        }

        // Enter "regular" sleeping mode