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

}

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

                )?;

            },

            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,

}

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,

}

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

        self.input_union_shutdown_tag = cmd_type.get_variant_tag_value(b"Shutdown").unwrap();

        let conn_type = pd.find_type(b"std.internet", b"TcpConnection")

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

        let conn_type = conn_type.as_struct();

        assert_eq!(conn_type.get_num_struct_fields(), 2);

                        match self.sync_state {

                            ListenerSyncState::AwaitingCmd => {

                                component::default_handle_sync_start(

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

                                );

                            },

                            ListenerSyncState::AcceptCommandReceived |

                            ListenerSyncState::AcceptChannelGenerated => unreachable!(),

                            ListenerSyncState::AcceptGenerateComponent => {

                                // Now that we're outside the sync round, create the tcp client

                                // component

                                let pending = self.pending_component.take().unwrap();

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

                                );

                            },

                            ListenerSyncState::FinishSyncThenQuit => {

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

                            },

                        }

                        return CompScheduling::Immediate;

                    },

                    ListenerSocketState::Error => {

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

                        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;

                            }

                        }

                    },

                    ListenerSyncState::AcceptCommandReceived => {

                        let socket = self.socket_state.get_socket();

                        match socket.accept() {

                                // Create the channels (and the inbox entries, to stay consistent

                                // with the expectations from the `component` module's functions)

                                let cmd_channel = comp_ctx.create_channel();

                                let data_channel = comp_ctx.create_channel();

                                let port_ids = [

                                    cmd_channel.putter_id, cmd_channel.getter_id,

                                    data_channel.putter_id, data_channel.getter_id,

                                ];

                                for port_id in port_ids {

                                    let expected_port_index = self.inbox_main.len();

                                    let port_handle = comp_ctx.get_port_handle(port_id);

                                    self.inbox_main.push(None);

                                    self.consensus.notify_of_new_port(expected_port_index, port_handle, comp_ctx);

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

                                ) {

                                    component::default_handle_error_for_builtin(

                                        &mut self.exec_state, sched_ctx, location_and_message

                                    );

                                }

                                // Prepare for finishing the consensus round, and once finished,

                                // create the tcp client component

                                self.sync_state = ListenerSyncState::AcceptChannelGenerated;

                                debug_assert!(self.pending_component.is_none());

                                self.pending_component = Some(PendingComponent{

                                    cmd_rx: cmd_channel.getter_id,

                                    data_tx: data_channel.putter_id

                                });

                                return CompScheduling::Requeue;

                            },

                            Err(err) => {

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

                                    return CompScheduling::Sleep;

                                } else {

                                    todo!("handle listener.accept error {:?}", err)

                                }

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

        }

        return Self {

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

            sync_state: ListenerSyncState::AwaitingCmd,

            pending_component: None,

            poll_ticket: None,

            inbox_main: vec![None, None],

            inbox_backup: InboxBackup::new(),

            pdl_input_port_id: input_port,

            pdl_output_port_id: output_port,

            input_union_accept_tag: -1,

            input_union_shutdown_tag: -1,

            output_struct_tx_index: 0,

            output_struct_rx_index: 0,

            exec_state: CompExecState::new(),

            control: ControlLayer::default(),

            consensus: Consensus::new(),

        }

    }

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

                    handle.send_message(&self.runtime, Message::Poll, true);

                }

            }

        }

    }

    #[inline]

    fn user_data_as_key(data: UserData) -> u64 {

        return data.0;

    }

    fn log(&self, message: &str) {

            println!("[polling] {}", message);

        }

    }

}

// bit convoluted, but it works

pub(crate) struct PollingThreadHandle {

    // requires Option, because:

    queue: Option<QueueDynProducer<PollCmd>>, // destructor needs to be called

    handle: Option<thread::JoinHandle<()>>, // we need to call `join`

}

impl SocketTcpClient {

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

        let socket_handle = create_and_connect_socket(

            libc::SOCK_STREAM, libc::IPPROTO_TCP, ip, port

        )?;

        if !set_socket_blocking(socket_handle, SOCKET_BLOCKING) {

            unsafe{ libc::close(socket_handle); }

            return Err(SocketError::Modifying);

        }

        return Ok(SocketTcpClient{

            socket_handle,

            is_blocking: SOCKET_BLOCKING,

        })

    }

        if !set_socket_blocking(socket_handle, SOCKET_BLOCKING) {

            unsafe{ libc::close(socket_handle); }

            return Err(SocketError::Modifying);

        }

        return Ok(SocketTcpClient{

            socket_handle,

            is_blocking: SOCKET_BLOCKING,

        })

    }

    pub fn send(&self, message: &[u8]) -> Result<usize, IoError> {

            libc::recv(self.socket_handle, message_pointer, buffer.len(), 0)

        };

        if result < 0 {

            return Err(IoError::last_os_error());

        }

        return Ok(result as usize);

    }

}

impl Drop for SocketTcpClient {

    fn drop(&mut self) {

        debug_assert!(self.socket_handle >= 0);

        unsafe{ close(self.socket_handle) };

    }

}

impl AsFileDescriptor for SocketTcpClient {

    fn as_file_descriptor(&self) -> FileDescriptor {

        return self.socket_handle;

    }

}

/// TCP listener. Yielding new connections

            return Err(SocketError::Modifying);

        }

        // Listen

        unsafe {

            let result = listen(socket_handle, libc::SOMAXCONN);

            if result < 0 {

                unsafe{ libc::close(socket_handle); }

                return Err(SocketError::Listening);

            }

        }

        return Ok(SocketTcpListener{

            socket_handle,

            is_blocking: SOCKET_BLOCKING,

        });

    }

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

        }

    }

}

impl Drop for SocketTcpListener {

    fn drop(&mut self) {

        debug_assert!(self.socket_handle >= 0);

        unsafe{ close(self.socket_handle) };

    }

}

impl AsFileDescriptor for SocketTcpListener {

    fn as_file_descriptor(&self) -> FileDescriptor {

        return self.socket_handle;

    }

}

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

        new fake_client(connection);

    }

    ", "constructor", no_args());

}

#[test]

fn test_tcp_listener_and_client() {

    compile_and_create_component("

    import std.internet::*;

    func listen_port() -> u16 {

    }

        // Start tcp listener

        channel listen_cmd_tx -> listen_cmd_rx;

        channel listen_conn_tx -> listen_conn_rx;

        new tcp_listener({}, listen_port(), listen_cmd_rx, listen_conn_tx);

        // Fake channels such that we can create a dummy connection variable

        channel client_cmd_tx -> unused_client_cmd_rx;

        channel unused_client_data_tx -> client_data_rx;

        auto new_connection = TcpConnection{

            tx: client_cmd_tx,

            rx: client_data_rx,

        };

                put(listen_cmd_tx, ListenerCmd::Accept);

                new_connection = get(listen_conn_rx);

            }

            // We have a new connection, spawn an 'echoer' for it

            print(\"server: spawning an echo'ing component\");

            new echo_machine(new_connection);

            connection_counter += 1;

        }

        // Shut down the listener

        print(\"server: shutting down listener\");

    }

    // Waits for a single TCP byte (to simplify potentially having to

    // concatenate requests) and echos it

    comp echo_machine(TcpConnection conn) {

        auto data_to_echo = {};

        // Wait for a message

        sync {

            print(\"echo: receiving data\");

            put(conn.tx, ClientCmd::Receive);

            data_to_echo = get(conn.rx);

            put(conn.tx, ClientCmd::Finish);

        }

        // Echo the message

        print(\"echo: sending back data\");

        sync put(conn.tx, ClientCmd::Send(data_to_echo));

        // Ask the tcp connection to shut down

        print(\"echo: shutting down\");

        sync put(conn.tx, ClientCmd::Shutdown);

    }

        channel cmd_tx -> cmd_rx;

        channel data_tx -> data_rx;

        new tcp_client({127, 0, 0, 1}, listen_port(), cmd_rx, data_tx);

        // Send the message

        print(\"requester: sending bytes\");

        sync put(cmd_tx, ClientCmd::Send({ byte_to_send }));

        // Receive the echo'd byte

        auto received_byte = byte_to_send + 1;

        sync {

            print(\"requester: receiving echo response\");

            received_byte = get(data_rx)[0];

            put(cmd_tx, ClientCmd::Finish);

        }

        // Silly check, as always

        while (byte_to_send != received_byte) {

            print(\"requester: Oh no! The echo is an otherworldly distorter\");

        }

        // Shut down the TCP connection

        print(\"requester: shutting down TCP component\");

        sync put(cmd_tx, ClientCmd::Shutdown);

    }

    comp constructor() {

        auto num_connections = 1;

        auto connection_index = 0;

        while (connection_index < num_connections) {

        }

    }

    ", "constructor", no_args());

}

mod messaging;

mod error_handling;

mod transfer_ports;

mod internet;

const LOG_LEVEL: LogLevel = LogLevel::Debug;

const NUM_THREADS: u32 = 1;

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

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

        .expect("successful compilation");

        .expect("successful runtime startup");

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

}

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

    let prompt = rt.inner.protocol.new_component(

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

    ).expect("create prompt");

    let reserved = rt.inner.start_create_component();

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

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

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