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

            let sending_port_instruction = comp_ctx.get_port(sending_port_handle).last_instruction;

            return Err((

                sending_port_instruction,

                String::from("Cannot transmit one of the ports in this message, as it is used in this sync round")

            ));

        }

        if transmit_port_info.state.is_set(PortStateFlag::Transmitted) {

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

            let sending_port_instruction = comp_ctx.get_port(sending_port_handle).last_instruction;

            return Err((

                sending_port_instruction,

                String::from("Cannot transmit one of the ports in this message, as that port is already transmitted")

            ));

        }

        // Set the flag for transmission

        transmit_port_info.state.set(PortStateFlag::Transmitted);

        // Block the peer of the port

        let message = control.create_port_transfer_message(unblock_put_control_id, comp_ctx.id, peer_port_id);

        let peer_handle = comp_ctx.get_peer_handle(peer_comp_id);

        let peer_info = comp_ctx.get_peer(peer_handle);

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

    }

    // We've set up the protocol, once all the PPC's are blocked we are supposed

    // to transfer the message to the recipient. So store it temporarily

    exec_state.mode = CompMode::PutPortsBlockedAwaitingAcks;

    return Ok(());

}

/// Performs the transmission of a data message that contains ports. These were

/// all stored in the component's execution state by the

/// `prepare_send_message_with_ports` function. Port must be ready to send!

fn perform_send_message_with_ports_to_receiver(

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

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

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

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

    // Find all ports again

    let mut transmit_ports = Vec::new();

    find_ports_in_value_group(&exec_state.mode_value, &mut transmit_ports);

    // Retrieve the port over which we're going to send the message

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

    let port_info = comp_ctx.get_port(port_handle);

    if !port_info.state.is_open() {

        return Err((

            port_info.last_instruction,

            String::from("cannot send over this port, as it is closed")

        ));

    }

    debug_assert!(!port_info.state.is_blocked_due_to_port_change()); // caller should have checked this

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

    // Change state back to its default

    exec_state.mode = CompMode::Sync;

    let message_value = exec_state.mode_value.take();

    exec_state.mode_port = PortId::new_invalid();

    // Annotate the data message

    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

    for (port_locations, transmit_port_id) in transmit_ports {

        let transmit_port_handle = comp_ctx.get_port_handle(transmit_port_id);

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

        transmit_port_state.clear(PortStateFlag::Transmitted);

        annotated_message.ports.push(TransmittedPort{

            locations: port_locations,

            messages: transmit_messages,

            peer_comp: transmit_port_info.peer_comp_id,

            peer_port: transmit_port_info.peer_port_id,

            kind: transmit_port_info.kind,

            state: transmit_port_state

        });

        comp_ctx.change_port_peer(sched_ctx, transmit_port_handle, None);

    }

    // And finally, send the message to the peer

    let peer_info = comp_ctx.get_peer(peer_handle);

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

    return Ok(());

}

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

fn default_handle_ack(

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

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

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

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

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

            AckAction::SendMessage(target_comp, message) => {

                // FIX @NoDirectHandle

                let mut handle = sched_ctx.runtime.get_component_public(target_comp);

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

                let _should_remove = handle.decrement_users();

                debug_assert!(_should_remove.is_none());

            },

            AckAction::ScheduleComponent(to_schedule) => {

                // FIX @NoDirectHandle

                let mut handle = sched_ctx.runtime.get_component_public(to_schedule);

                // Note that the component is intentionally not

                // sleeping, so we just wake it up

                debug_assert!(!handle.sleeping.load(std::sync::atomic::Ordering::Acquire));

                let key = unsafe { to_schedule.upgrade() };

                sched_ctx.runtime.enqueue_work(key);

                let _should_remove = handle.decrement_users();

                debug_assert!(_should_remove.is_none());

            },

            AckAction::UnblockPutWithPorts => {

                // Send the message (containing ports) stored in the component

                // execution state to the recipient

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

                exec_state.mode = CompMode::PutPortsBlockedSendingPort;

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

                // Little bit of a hack, we didn't really unblock the sending

                // port, but this will mesh nicely with waiting for the sending

                // port to become unblocked.

                default_handle_recently_unblocked_port(

                    exec_state, control, consensus, port_handle, sched_ctx,

                    comp_ctx, inbox_main, inbox_backup

                )?;

            },

            AckAction::None => {}

        }

        match new_to_ack {

            Some(new_to_ack) => to_ack = new_to_ack,

            None => break,

        }

    }

    return Ok(());

}

/// Little helper for sending the most common kind of `Ack`

fn default_send_ack(

    causer_of_ack_id: ControlId, peer_handle: LocalPeerHandle,

    sched_ctx: &SchedulerCtx, comp_ctx: &CompCtx

) {

    let peer_info = comp_ctx.get_peer(peer_handle);

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

        id: causer_of_ack_id,

        sender_comp_id: comp_ctx.id,

        target_port_id: None,

        content: ControlMessageContent::Ack

    }), true);

}

/// Handles the unblocking of a putter port. In case there is a pending message

/// on that port then it will be sent. There are two reasons for calling this

/// function: either a port was blocked (i.e. the Blocked state flag was

/// cleared), or the component is ready to send a message containing ports

/// (stored in the execution state). In this latter case we might still have

/// a blocked port.

fn default_handle_recently_unblocked_port(

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

    port_handle: LocalPortHandle, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

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

    let port_info = comp_ctx.get_port_mut(port_handle);

    let port_id = port_info.self_id;

    if port_info.state.is_blocked() {

        // Port is still blocked. We wait until the next control message where

        // we unblock the port.

        return Ok(());

    }

    if exec_state.is_blocked_on_put_without_ports(port_id) {

        // Annotate the message that we're going to send

        let port_info = comp_ctx.get_port(port_handle); // for immutable access

        debug_assert_eq!(port_info.kind, PortKind::Putter);

        let to_send = exec_state.mode_value.take();

        let to_send = consensus.annotate_data_message(comp_ctx, port_info, to_send);

        // Retrieve peer to send the message

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

        let peer_info = comp_ctx.get_peer(peer_handle);

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

        // Return to the regular execution mode

        exec_state.mode = CompMode::Sync;

        exec_state.mode_port = PortId::new_invalid();

    } else if exec_state.mode == CompMode::PutPortsBlockedTransferredPorts {

        // We are waiting until all of the transferred ports become unblocked,

        // check so here.

        let mut transfer_ports = Vec::new();

        find_ports_in_value_group(&exec_state.mode_value, &mut transfer_ports);

        if ports_not_blocked(comp_ctx, &transfer_ports) {

            perform_send_message_with_ports_notify_peers(

                exec_state, comp_ctx, sched_ctx, control, transfer_ports

            )?;

        }

    } else if exec_state.mode == CompMode::PutPortsBlockedSendingPort && exec_state.mode_port == port_id {

        // We checked above that the port became unblocked, so we can send the

        // message

        perform_send_message_with_ports_to_receiver(

            exec_state, sched_ctx, comp_ctx, consensus, inbox_main, inbox_backup

        )?;

    } else if exec_state.is_blocked_on_create_component() {

        let mut ports = Vec::new();

        find_ports_in_value_group(&exec_state.mode_value, &mut ports);

        if ports_not_blocked(comp_ctx, &ports) {

            perform_create_component(

                exec_state, sched_ctx, comp_ctx, control, inbox_main, inbox_backup

            );

        }

    }

    return Ok(());

}

#[inline]

pub(crate) fn port_id_from_eval(port_id: EvalPortId) -> PortId {

    return PortId(port_id.id);

}

#[inline]

pub(crate) fn port_id_to_eval(port_id: PortId) -> EvalPortId {

    return EvalPortId{ id: port_id.0 };

}

// TODO: Optimize double vec

type EncounteredPorts = Vec<(Vec<ValueId>, PortId)>;

/// Recursively goes through the value group, attempting to find ports.

/// Duplicates will only be added once.

pub(crate) fn find_ports_in_value_group(value_group: &ValueGroup, ports: &mut EncounteredPorts) {

    // Helper to check a value for a port and recurse if needed.

    fn find_port_in_value(group: &ValueGroup, value: &Value, value_location: ValueId, ports: &mut EncounteredPorts) {

        match value {

            Value::Input(port_id) | Value::Output(port_id) => {

                // This is an actual port

                let cur_port = PortId(port_id.id);

                for prev_port in ports.iter_mut() {

                    if prev_port.1 == cur_port {

                        // Already added

                        prev_port.0.push(value_location);

                        return;

                    }

                }

                ports.push((vec![value_location], cur_port));

            },

            Value::Array(heap_pos) |

            Value::Message(heap_pos) |

            Value::String(heap_pos) |

            Value::Struct(heap_pos) |

            Value::Union(_, heap_pos) => {

                // Reference to some dynamic thing which might contain ports,

                // so recurse

                let heap_region = &group.regions[*heap_pos as usize];

                for (value_index, embedded_value) in heap_region.iter().enumerate() {

                    let value_location = ValueId::Heap(*heap_pos, value_index as u32);

                    find_port_in_value(group, embedded_value, value_location, ports);

                }

            },

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

        }

    }

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

    ports.clear();

    for (value_index, value) in value_group.values.iter().enumerate() {

        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

pub(crate) fn take_port_messages(

    comp_ctx: &CompCtx, port_id: PortId,

    inbox_main: &mut InboxMain, inbox_backup: &mut InboxBackup

) -> Vec<DataMessage> {

    let mut messages = Vec::new();

    let port_handle = comp_ctx.get_port_handle(port_id);

    let port_index = comp_ctx.get_port_index(port_handle);

    if let Some(message) = inbox_main[port_index].take() {

        messages.push(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_id {

            let message = inbox_backup.remove(message_index);

            messages.push(message);

        } else {

            message_index += 1;

        }

    }

    return messages;

}

        match message {

            Message::Data(message) => {

                self.handle_incoming_data_message(sched_ctx, comp_ctx, message);

            },

            Message::Sync(message) => {

                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,

                    message, sched_ctx, comp_ctx, &mut self.inbox_main, &mut self.inbox_backup

                ) {

                    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 {

            CompMode::BlockedSelect |

            CompMode::PutPortsBlockedTransferredPorts |

            CompMode::PutPortsBlockedAwaitingAcks |

            CompMode::PutPortsBlockedSendingPort |

            CompMode::NewComponentBlocked => {

                // Not possible: we never enter this state

                unreachable!();

            },

            CompMode::NonSync => {

                // When in non-sync mode

                match &self.socket_state {

                    ClientSocketState::Connected(_socket) => {

                        if self.sync_state == ClientSyncState::FinishSyncThenQuit {

                            // Previous request was to let the component shut down

                            self.exec_state.set_as_start_exit(ExitReason::Termination);

                        } else {

                            // Reset for a new request

                            self.sync_state = ClientSyncState::AwaitingCmd;

                            component::default_handle_sync_start(

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

                            );

                        }

                        return CompScheduling::Immediate;

                    },

                    ClientSocketState::Error => {

                        // Could potentially send an error message to the

                        // connected component.

                        self.exec_state.set_as_start_exit(ExitReason::ErrorNonSync);

                        return CompScheduling::Immediate;

                    }

                }

            },

            CompMode::Sync => {

                // When in sync mode: wait for a command to come in

                match self.sync_state {

                    ClientSyncState::AwaitingCmd => {

                        match component::default_attempt_get(

                            &mut self.exec_state, self.pdl_input_port_id, PortInstruction::NoSource,

                            &mut self.inbox_main, &mut self.inbox_backup, sched_ctx, comp_ctx,

                            &mut self.control, &mut self.consensus

                        ) {

                            GetResult::Received(message) => {

                                let (tag_value, embedded_heap_pos) = message.content.values[0].as_union();

                                if tag_value == self.input_union_send_tag_value {

                                    // Retrieve bytes from the message

                                    self.byte_buffer.clear();

                                    let union_content = &message.content.regions[embedded_heap_pos as usize];

                                    debug_assert_eq!(union_content.len(), 1);

                                    let array_heap_pos = union_content[0].as_array();

                                    let array_values = &message.content.regions[array_heap_pos as usize];

                                    self.byte_buffer.reserve(array_values.len());

                                    for value in array_values {

                                        self.byte_buffer.push(value.as_uint8());

                                    }

                                    self.sync_state = ClientSyncState::Putting;

                                } else if tag_value == self.input_union_receive_tag_value {

                                    // Component requires a `recv`

                                    self.sync_state = ClientSyncState::Getting;

                                } else if tag_value == self.input_union_finish_tag_value {

                                    // Component requires us to end the sync round

                                    self.sync_state = ClientSyncState::FinishSync;

                                } else if tag_value == self.input_union_shutdown_tag_value {

                                    // Component wants to close the connection

                                    self.sync_state = ClientSyncState::FinishSyncThenQuit;

                                } else {

                                    unreachable!("got tag_value {}", tag_value)

                                }

                                return CompScheduling::Immediate;

                            },

                            GetResult::NoMessage => {

                                return CompScheduling::Sleep;

                            },

                            GetResult::Error(location_and_message) => {

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

                                return CompScheduling::Immediate;

                            }

                        }

                    },

                    ClientSyncState::Putting => {

                        // We're supposed to send a user-supplied message fully

                        // over the socket. But we might end up blocking. In

                        // that case the component goes to sleep until it is

                        // polled.

                        let socket = self.socket_state.get_socket();

                        while !self.byte_buffer.is_empty() {

                            match socket.send(&self.byte_buffer) {

                                Ok(bytes_sent) => {

                                    self.byte_buffer.drain(..bytes_sent);

                                },

                                Err(err) => {

                                    if err.kind() == IoErrorKind::WouldBlock {

                                        return CompScheduling::Sleep; // wait until notified

                                    } else {

                                        todo!("handle socket.send error {:?}", err)

                                    }

                                }

                            }

                        }

                        // If here then we're done putting the data, we can

                        // finish the sync round

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

                        return CompScheduling::Requeue;

                    },

                    ClientSyncState::Getting => {

                        // We're going to try and receive a single message. If

                        // this causes us to end up blocking the component

                        // goes to sleep until it is polled.

                        const BUFFER_SIZE: usize = 1024; // TODO: Move to config

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

                                let send_result = component::default_send_data_message(

                                    &mut self.exec_state, self.pdl_output_port_id, PortInstruction::NoSource,

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

                                );

                                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 = ClientSyncState::AwaitingCmd;

                                    return scheduling;

                                }

                            },

                            Err(err) => {

                                if err.kind() == IoErrorKind::WouldBlock {

                                    return CompScheduling::Sleep; // wait until polled

                                } else {

                                    todo!("handle socket.receive error {:?}", err)

                                }

                            }

                        }

                    },

                    ClientSyncState::FinishSync | ClientSyncState::FinishSyncThenQuit => {

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

                        return CompScheduling::Requeue;

                    },

                }

            },

            CompMode::BlockedGet => {

                // Entered when awaiting a new command

                debug_assert_eq!(self.sync_state, ClientSyncState::AwaitingCmd);

                return CompScheduling::Sleep;

            },

            CompMode::SyncEnd | CompMode::BlockedPut =>

                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(&mut self.exec_state, &mut self.control, sched_ctx),

            CompMode::Exit =>

                return component::default_handle_exit(&self.exec_state),

        }

    }

}

impl ComponentTcpClient {

    pub(crate) fn new(arguments: ValueGroup) -> Self {

        // Parsing arguments

            let (ip_address, port) = ip_addr_and_port_from_args(&arguments, 0, 1);

            let input_port = component::port_id_from_eval(arguments.values[2].as_input());

            let output_port = component::port_id_from_eval(arguments.values[3].as_output());

        if let Err(socket) = socket {

            todo!("friendly error reporting: failed to open socket (reason: {:?})", socket);

        }

        return Self{

            socket_state: ClientSocketState::Connected(socket.unwrap()),

            sync_state: ClientSyncState::AwaitingCmd,

            poll_ticket: None,

            inbox_main: vec![None, None],

            inbox_backup: Vec::new(),

            input_union_send_tag_value: -1,

            input_union_receive_tag_value: -1,

            input_union_finish_tag_value: -1,

            input_union_shutdown_tag_value: -1,

            pdl_input_port_id: input_port,

            pdl_output_port_id: output_port,

            exec_state: CompExecState::new(),

            control: ControlLayer::default(),

            consensus: Consensus::new(),

            byte_buffer: Vec::new(),

        }

    }

    pub(crate) fn new_with_existing_connection(socket: SocketTcpClient, input_port: PortId, output_port: PortId) -> Self {

        return Self{

            socket_state: ClientSocketState::Connected(socket),

            sync_state: ClientSyncState::AwaitingCmd,

            poll_ticket: None,

            inbox_main: vec![None, None],

            inbox_backup: Vec::new(),

            input_union_send_tag_value: -1,

            input_union_receive_tag_value: -1,

            input_union_finish_tag_value: -1,

            input_union_shutdown_tag_value: -1,

            pdl_input_port_id: input_port,

            pdl_output_port_id: output_port,

            exec_state: CompExecState::new(),

            control: ControlLayer::default(),

            consensus: Consensus::new(),

            byte_buffer: Vec::new(),

        }

    }

    // Handles incoming data from the PDL side (hence, going into the socket)

    fn handle_incoming_data_message(&mut self, sched_ctx: &SchedulerCtx, comp_ctx: &mut CompCtx, message: DataMessage) {

        if self.exec_state.mode.is_in_sync_block() {

            self.consensus.handle_incoming_data_message(comp_ctx, &message);

        }

        match component::default_handle_incoming_data_message(

            &mut self.exec_state, &mut self.inbox_main, comp_ctx, message, sched_ctx, &mut self.control

        ) {

            IncomingData::PlacedInSlot => {},

            IncomingData::SlotFull(message) => {

                self.inbox_backup.push(message);

            }

        }

    }

    fn data_message_to_bytes(&self, message: DataMessage, bytes: &mut Vec<u8>) {

        debug_assert_eq!(message.data_header.target_port, self.pdl_input_port_id);

        debug_assert_eq!(message.content.values.len(), 1);

        if let Value::Array(array_pos) = message.content.values[0] {

            let region = &message.content.regions[array_pos as usize];

            bytes.reserve(region.len());

            for value in region {

                bytes.push(value.as_uint8());

            }

        } else {

            unreachable!();

        }

    }

    fn bytes_to_data_message_content(&self, buffer: &[u8]) -> ValueGroup {

        // Turn bytes into silly executor-style array

        let mut values = Vec::with_capacity(buffer.len());

        for byte in buffer.iter().copied() {

            values.push(Value::UInt8(byte));

        }

        // Put in a value group

        let mut value_group = ValueGroup::default();

        value_group.regions.push(values);

        value_group.values.push(Value::Array(0));

        return value_group;

    }

}

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

// ComponentTcpListener

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

enum ListenerSocketState {

    Connected(SocketTcpListener),

    Error,

}

impl ListenerSocketState {

    fn get_socket(&self) -> &SocketTcpListener {

        match self {

            ListenerSocketState::Connected(v) => return v,

            ListenerSocketState::Error => unreachable!(),

        }

    }

}

struct PendingComponent {

    cmd_rx: PortId,

    data_tx: PortId,

}

enum ListenerSyncState {

    AwaitingCmd,

    AcceptCommandReceived, // just received `Accept` command

    AcceptChannelGenerated, // created channel, waiting to end the sync round

    AcceptGenerateComponent, // sync ended, back in non-sync, now generate component

    FinishSyncThenQuit,

}

pub struct ComponentTcpListener {

    // Properties for the tcp socket

    socket_state: ListenerSocketState,

    sync_state: ListenerSyncState,

    pending_component: Option<PendingComponent>,

    poll_ticket: Option<PollTicket>,

    inbox_main: InboxMain,

    inbox_backup: InboxBackup,

    pdl_input_port_id: PortId, // input port, receives commands

    pdl_output_port_id: PortId, // output port, sends connections

    // Type information extracted from protocol

    tcp_client_definition: (ProcedureDefinitionId, TypeId),

    input_union_accept_tag: i64,

    input_union_shutdown_tag: i64,

    output_struct_rx_index: usize,

    output_struct_tx_index: usize,

    // Generic component state

    exec_state: CompExecState,

    control: ControlLayer,

    consensus: Consensus,

}

impl Component for ComponentTcpListener {

    fn on_creation(&mut self, id: CompId, sched_ctx: &SchedulerCtx) {

        // Retrieve type information for the message with ports we're going to send

        let pd = &sched_ctx.runtime.protocol;

        let cmd_type = pd.find_type(b"std.internet", b"ListenerCmd")

            .expect("'ListenerCmd' type in the 'std.internet' module");

        let cmd_type = cmd_type.as_union();

        self.input_union_accept_tag = cmd_type.get_variant_tag_value(b"Accept").unwrap();