<commit_before>use std::net::{SocketAddr, TcpStream, TcpListener};
use std::io::{Read, Write, ErrorKind, BufReader};
use std::thread;
use std::time::Duration;
use super::inbox::Inbox;
use super::id::{ID, broadcast_machine_id};
use super::type_registry::ShortTypeId;
use super::messaging::{Message, Packet};
use byteorder::{LittleEndian, WriteBytesExt, ByteOrder};
use compact::Compact;

/// Represents all networking environment and networking state
/// of an `ActorSystem`
pub struct Networking {
    /// The machine index of this machine within the network of peers
    pub machine_id: u8,
    /// The current network turn this machine is in. Used to keep track
    /// if this machine lags behind or runs fast compared to its peers
    pub n_turns: usize,
    network: Vec<SocketAddr>,
    network_connections: Vec<Option<Connection>>,
}

impl Networking {
    /// Create network environment based on this machines id/index
    /// and all peer addresses (including this machine)
    pub fn new(machine_id: u8, network: Vec<SocketAddr>) -> Networking {
        Networking {
            machine_id,
            n_turns: 0,
            network_connections: (0..network.len()).into_iter().map(|_| None).collect(),
            network,
        }
    }

    /// Connect to all peers in the network
    pub fn connect(&mut self) {
        let listener = TcpListener::bind(self.network[self.machine_id as usize]).unwrap();

        // first wait for all smaller machine_ids to connect
        for (machine_id, _address) in self.network.iter().enumerate() {
            if machine_id < self.machine_id as usize {
                self.network_connections[machine_id] =
                    Some(Connection::new(listener.accept().unwrap().0))
            }
        }

        thread::sleep(Duration::from_secs(2));

        // then try to connecto to all larger machine_ids
        for (machine_id, address) in self.network.iter().enumerate() {
            if machine_id > self.machine_id as usize {
                self.network_connections[machine_id] =
                    Some(Connection::new(TcpStream::connect(address).unwrap()))
            }
        }

        println!("All mapped");
    }

    /// Finish the current networking turn and wait for peers which lag behind
    /// based on their turn number. This is the main backpressure mechanism.
    pub fn finish_turn(&mut self, inboxes: &mut [Option<Inbox>]) {
        let mut should_sleep = None;

        for maybe_connection in &mut self.network_connections {
            if let Some(Connection { n_turns, .. }) = *maybe_connection {
                if n_turns + 120 < self.n_turns {
                    // ~2s difference
                    should_sleep = Some((
                        Duration::from_millis(
                            ((self.n_turns - 120 - n_turns) / 10) as u64,
                        ),
                        n_turns,
                    ));
                }
            }
        }

        if let Some((duration, other_n_turns)) = should_sleep {
            // println!(
            //     "Sleeping to let other machine catch up ({} vs. {} turns)",
            //     other_n_turns,
            //     self.n_turns
            // );
            // Try to process extra messages if we are ahead
            self.send_and_receive(inboxes);
            self.send_and_receive(inboxes);
            self.send_and_receive(inboxes);
            ::std::thread::sleep(duration);
        };

        self.n_turns += 1;

        for maybe_connection in &mut self.network_connections {
            if let Some(ref mut connection) = *maybe_connection {
                // write turn end, use 0 to distinguish from actual packet
                connection.write_queue.write_u64::<LittleEndian>(0).unwrap();
                connection
                    .write_queue
                    .write_u64::<LittleEndian>(self.n_turns as u64)
                    .unwrap();
                connection.n_turns_since_own_turn = 0;
            }
        }
    }

    /// Send queued outbound messages and take incoming queued messages
    /// and forward them to their local target recipient(s)
    pub fn send_and_receive(&mut self, inboxes: &mut [Option<Inbox>]) {
        for maybe_connection in &mut self.network_connections {
            if let Some(ref mut connection) = *maybe_connection {
                connection.try_send();
                connection.try_receive(inboxes)
            }
        }
    }

    /// Enqueue a new (potentially) outbound packet
    pub fn enqueue<M: Message>(&mut self, message_type_id: ShortTypeId, mut packet: Packet<M>) {
        let packet_size = Compact::total_size_bytes(&packet);
        let total_size = ::std::mem::size_of::<ShortTypeId>() + packet_size;
        let machine_id = packet.recipient_id.machine;

        let mut connections: Vec<&mut Connection> = if machine_id == broadcast_machine_id() {
            self.network_connections
                .iter_mut()
                .filter_map(|maybe_connection| maybe_connection.as_mut())
                .collect()
        } else {
            vec![<commit_msg>Make a single instance work again<commit_after>use std::net::{SocketAddr, TcpStream, TcpListener};
use std::io::{Read, Write, ErrorKind, BufReader};
use std::thread;
use std::time::Duration;
use super::inbox::Inbox;
use super::id::{ID, broadcast_machine_id};
use super::type_registry::ShortTypeId;
use super::messaging::{Message, Packet};
use byteorder::{LittleEndian, WriteBytesExt, ByteOrder};
use compact::Compact;

/// Represents all networking environment and networking state
/// of an `ActorSystem`
pub struct Networking {
    /// The machine index of this machine within the network of peers
    pub machine_id: u8,
    /// The current network turn this machine is in. Used to keep track
    /// if this machine lags behind or runs fast compared to its peers
    pub n_turns: usize,
    network: Vec<SocketAddr>,
    network_connections: Vec<Option<Connection>>,
}

impl Networking {
    /// Create network environment based on this machines id/index
    /// and all peer addresses (including this machine)
    pub fn new(machine_id: u8, network: Vec<SocketAddr>) -> Networking {
        Networking {
            machine_id,
            n_turns: 0,
            network_connections: (0..network.len()).into_iter().map(|_| None).collect(),
            network,
        }
    }

    /// Connect to all peers in the network
    pub fn connect(&mut self) {
        let listener = TcpListener::bind(self.network[self.machine_id as usize]).unwrap();

        // first wait for all smaller machine_ids to connect
        for (machine_id, _address) in self.network.iter().enumerate() {
            if machine_id < self.machine_id as usize {
                self.network_connections[machine_id] =
                    Some(Connection::new(listener.accept().unwrap().0))
            }
        }

        thread::sleep(Duration::from_secs(2));

        // then try to connecto to all larger machine_ids
        for (machine_id, address) in self.network.iter().enumerate() {
            if machine_id > self.machine_id as usize {
                self.network_connections[machine_id] =
                    Some(Connection::new(TcpStream::connect(address).unwrap()))
            }
        }

        println!("All mapped");
    }

    /// Finish the current networking turn and wait for peers which lag behind
    /// based on their turn number. This is the main backpressure mechanism.
    pub fn finish_turn(&mut self, inboxes: &mut [Option<Inbox>]) {
        let mut should_sleep = None;

        for maybe_connection in &mut self.network_connections {
            if let Some(Connection { n_turns, .. }) = *maybe_connection {
                if n_turns + 120 < self.n_turns {
                    // ~2s difference
                    should_sleep = Some((
                        Duration::from_millis(
                            ((self.n_turns - 120 - n_turns) / 10) as u64,
                        ),
                        n_turns,
                    ));
                }
            }
        }

        if let Some((duration, other_n_turns)) = should_sleep {
            // println!(
            //     "Sleeping to let other machine catch up ({} vs. {} turns)",
            //     other_n_turns,
            //     self.n_turns
            // );
            // Try to process extra messages if we are ahead
            self.send_and_receive(inboxes);
            self.send_and_receive(inboxes);
            self.send_and_receive(inboxes);
            ::std::thread::sleep(duration);
        };

        self.n_turns += 1;

        for maybe_connection in &mut self.network_connections {
            if let Some(ref mut connection) = *maybe_connection {
                // write turn end, use 0 to distinguish from actual packet
                connection.write_queue.write_u64::<LittleEndian>(0).unwrap();
                connection
                    .write_queue
                    .write_u64::<LittleEndian>(self.n_turns as u64)
                    .unwrap();
                connection.n_turns_since_own_turn = 0;
            }
        }
    }

    /// Send queued outbound messages and take incoming queued messages
    /// and forward them to their local target recipient(s)
    pub fn send_and_receive(&mut self, inboxes: &mut [Option<Inbox>]) {
        for maybe_connection in &mut self.network_connections {
            if let Some(ref mut connection) = *maybe_connection {
                connection.try_send();
                connection.try_receive(inboxes)
            }
        }
    }

    /// Enqueue a new (potentially) outbound packet
    pub fn enqueue<M: Message>(&mut self, message_type_id: ShortTypeId, mut packet: Packet<M>) {
        if self.network.len() == 1 {
            return;
        }

        let packet_size = Compact::total_size_bytes(&packet);
        let total_size = ::std::mem::size_of::<ShortTypeId>() + packet_size;
        let machine_id = packet.recipient_id.machine;

        let mut connections: Vec<&mut Connection> = if machine_id == broadcast_machine_id() {
            self.network_connections
                .iter_mut()
                .filter_map(|maybe_connection| maybe_connection.as_mut())
                .collect()
        } else {
            vec![
{'fraction_non_alphanumeric': 0.10377176015473888, 'fraction_numerical': 0.005319148936170213, 'mean_word_length': 1.9939200926462073, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3923040', 'n_tokens_mistral': 2877, 'n_tokens_neox': 2758, 'n_words': 764}