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

        let _new_inbox_index = inbox_main.len();

        if !received_port.messages.is_empty() {

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

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

        } else {

            inbox_main.push(None);

        }

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

        );

        debug_assert_eq!(_new_inbox_index, comp_ctx.get_port_index(new_port_handle));

        comp_ctx.change_port_peer(sched_ctx, new_port_handle, Some(received_port.peer_comp));

        let new_port = comp_ctx.get_port(new_port_handle);

        // Add the port tho the consensus

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

        for message_location in received_port.locations.iter().copied() {

            let value = message.content.get_value_mut(message_location);

            match value {

                Value::Input(_) => {

                    debug_assert_eq!(new_port.kind, PortKind::Getter);

                    *value = Value::Input(port_id_to_eval(new_port.self_id));

                },

                Value::Output(_) => {

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

                    *value = Value::Output(port_id_to_eval(new_port.self_id));

                },

                _ => unreachable!(),

            }

        }

        // Let the peer know that the port can now be used

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

        let peer_info = comp_ctx.get_peer(peer_handle);

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

            id: ControlId::new_invalid(),

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

    }

}

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

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

        instantiator_ctx.remove_port(pair.instantiator_handle);

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

}

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;

}

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

            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{

            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(

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

        }

    }

}

enum ListenerSyncState {

    AwaitingCmd,

    FinishSyncThenQuit,

}

pub struct ComponentTcpListener {

    // Properties for the tcp socket

    socket_state: ListenerSocketState,

    sync_state: ListenerSyncState,

    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

    // Information about union tags

    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;

                ) {

                    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 comonent ComponentTcpListener (mode: {:?})", self.exec_state.mode));

        match self.exec_state.mode {

            CompMode::BlockedSelect

                => unreachable!(),

            CompMode::PutPortsBlockedTransferredPorts |

            CompMode::PutPortsBlockedAwaitingAcks |

            CompMode::PutPortsBlockedSendingPort |

            CompMode::NewComponentBlocked

                => return CompScheduling::Sleep,

            CompMode::NonSync => {

                match &self.socket_state {

                    ListenerSocketState::Connected(_socket) => {

                                component::default_handle_sync_start(

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

                                );

                        }

                        return CompScheduling::Immediate;

                    },

                    ListenerSocketState::Error => {

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

                        return CompScheduling::Immediate;

                    }

                }

            },

            CompMode::Sync => {

                match self.sync_state {

                    ListenerSyncState::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, _) = message.content.values[0].as_union();

                                if tag_value == self.input_union_accept_tag {

                                } else if tag_value == self.input_union_shutdown_tag {

                                    self.sync_state = ListenerSyncState::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;

                            }

                        }

                    },

                        let socket = self.socket_state.get_socket();

                        match socket.accept() {

                            Ok(client) => {

                                let client = client.unwrap();

                            },

                            Err(err) => {

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

                                    return CompScheduling::Sleep;

                                } else {

                                }

                            }

                        }

                    },

                    ListenerSyncState::FinishSyncThenQuit => {

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

                        return CompScheduling::Requeue;

                    }

                }

            },

            CompMode::BlockedGet => {

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

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

        debug_assert_eq!(arguments.values.len(), 4);

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

        let socket = SocketTcpListener::new(ip_address, port);

        debug_assert_eq!(self.mode, Mode::SyncBusy);

        self.mode = Mode::SyncAwaitingSolution;

        let local_solution = self.generate_local_solution(comp_ctx, true);

        let decision = self.handle_local_solution(sched_ctx, comp_ctx, comp_ctx.id, local_solution, true);

        return decision;

    }

    /// Notifies that a decision has been reached. Note that the caller should

    /// still take the appropriate actions based on the decision it is supplying

    /// to the consensus layer.

    pub(crate) fn notify_sync_decision(&mut self, _decision: SyncRoundDecision) {

        // Reset everything for the next round

        debug_assert_eq!(self.mode, Mode::SyncAwaitingSolution);

        self.mode = Mode::NonSync;

        self.round_index = self.round_index.wrapping_add(1);

        for port in self.ports.iter_mut() {

            port.mapping = None;

        }

        self.solution.clear();

    }

        debug_assert_eq!(_expected_index, self.ports.len());

        let port_info = comp_ctx.get_port(port_handle);

        self.ports.push(PortAnnotation{

            self_comp_id: comp_ctx.id,

            self_port_id: port_info.self_id,

            peer_comp_id: port_info.peer_comp_id,

            peer_port_id: port_info.peer_port_id,

            peer_discovered: false,

            mapping: None,

            kind: port_info.kind,

        });

    }

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

    // Handling inbound and outbound messages

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

    /// Prepares a set of values to be sent of a channel.

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

            socket_handle,

            is_blocking: SOCKET_BLOCKING,

        });

    }

    pub fn accept(&self) -> Result<Result<SocketTcpClient, SocketError>, IoError> {

        let (mut address, mut address_size) = create_sockaddr_in_empty();

        let address_pointer = &mut address as *mut sockaddr_in;

        let socket_handle = unsafe { accept(self.socket_handle, address_pointer.cast(), &mut address_size) };

        if socket_handle < 0 {

            return Err(IoError::last_os_error());

        }

        return Ok(SocketTcpClient::new_from_handle(socket_handle));

    }

}

impl Drop for SocketTcpListener {

    fn drop(&mut self) {

        debug_assert!(self.socket_handle >= 0);

        unsafe{ close(self.socket_handle) };

    }

}

/// Raw socket receiver. Essentially a listener that accepts a single connection

struct SocketRawRx {

    listen_handle: c_int,

    accepted_handle: c_int,

}

impl SocketRawRx {

    pub fn new(ip: Option<Ipv4Addr>, port: u16) -> Result<Self, SocketError> {

        let ip = ip.unwrap_or(Ipv4Addr::UNSPECIFIED); // unspecified is the same as INADDR_ANY

        let address = unsafe{ in_addr{

            s_addr: std::mem::transmute(ip.octets()),

        }};

        let socket_address = sockaddr_in{

            sin_family: libc::AF_INET as libc::sa_family_t,

            sin_port: htons(port),

            sin_addr: address,

            sin_zero: [0; 8],

        };

        unsafe {

            let socket_handle = create_and_bind_socket(libc::SOCK_RAW, 0, IpAddr::V4(ip), port)?;