So You Want to Build a Language VM - Part 27 - Clustering: Part 1

- DATA_ROOT_DIR:
    help: Root directory where the Iridium VM should store its data. Defaults to /var/lib/iridium.
    required: false
    takes_value: true
    long: data-root-dir
- NODE_ALIAS:
    help: An alias that can be used to refer to a running VM across a network
    required: false
    takes_value: true
    long: node-alias

let alias = matches.value_of("NODE_ALIAS").unwrap_or("");

pub struct VM {
    /// Array that simulates having hardware registers
    pub registers: [i32; 32],
    /// Array that simulates having floating point hardware registers
    pub float_registers: [f64; 32],
    /// Program counter that tracks which byte is being executed
    pc: usize,
    /// The bytecode of the program being run
    pub program: Vec<u8>,
    /// Used for heap memory
    heap: Vec<u8>,
    /// Used to represent the stack
    stack: Vec<u8>,
    /// Contains the remainder of modulo division ops
    remainder: usize,
    /// Contains the result of the last comparison operation
    equal_flag: bool,
    /// Loop counter field, used with the `LOOP` instruction
    loop_counter: usize,
    /// Contains the read-only section data
    ro_data: Vec<u8>,
    /// Is a unique, randomly generated UUID for identifying this VM
    id: Uuid,
    /// An alias that can be specified by the user and used to refer to the Node
    alias: Option<String>,
    /// Keeps a list of events for a particular VM
    events: Vec<VMEvent>,
    /// Number of logical cores the system reports
    pub logical_cores: usize,
}

pub fn with_alias(mut self, alias: String) -> Self {
    if alias == "" {
        self.alias = Some(alias)
    } else {
        self.alias = None
    }
    self
}

let alias = matches.value_of("NODE_ALIAS").unwrap_or("");

pub fn with_cluster_bind(mut self, server_addr: String, server_port: String) -> Self {
    self.server_addr = Some(server_addr);
    self.server_port = Some(server_port);
    self
}

- SERVER_LISTEN_PORT:
    help: Which port Iridium should listen for remote connections on from other Iridium VMs. Defaults to 2254.
    required: false
    takes_value: true
    long: server-bind-port
    short: P
- SERVER_LISTEN_HOST:
    help: Which address Iridium should listen for remote connections on from other Iridium VMs. Defaults to "127.0.0.1".
    required: false
    takes_value: true
    long: server-bind-host
    short: H

let server_addr = matches.value_of("SERVER_LISTEN_HOST").unwrap_or("127.0.0.1");
let server_port = matches.value_of("SERVER_LISTEN_PORT").unwrap_or("2254");

// Server address that the VM will bind to for server-to-server communications
server_addr: Option<String>,
// Port the server will bind to for server-to-server communications
server_port: Option<String>

let mut vm = VM::new();

let mut vm = VM::new().with_alias(alias.to_string()).with_cluster_bind(server_addr.into(), server_port.into());

let mut repl = REPL::new(vm);

pub fn new(vm: VM) -> REPL

use std::net::{TcpListener, SocketAddr};
use std::thread;
use cluster::client::ClusterClient;

pub fn listen(addr: SocketAddr) {
    info!("Initializing Cluster server...");
    let listener =
        TcpListener::bind(addr).unwrap();
    for stream in listener.incoming() {
        info!("New Node connected!");
        let stream = stream.unwrap();
        thread::spawn(|| {
            let mut client = ClusterClient::new(stream);
            client.run();
        });
    }
}

use std::io::{BufRead, Write};
use std::io::{BufReader, BufWriter};
use std::net::TcpStream;
use std::thread;
use std::sync::mpsc::channel;
use std::sync::mpsc::{Receiver, Sender};

/// This represents another Iridium VM that has joined this server. On their end, we would be the ClusterClient.
/// The main difference between this and `Client` is that we don't create a `REPL`, as we don't need one.
pub struct ClusterClient {
    // Wrap the stream in a BufReader to make it easier to read
    reader: BufReader<TcpStream>,
    // Wrap the stream in a BufWriter to make it easier to write
    writer: BufWriter<TcpStream>,
    // These are standard mpsc channels.
    // We'll start a thread that watches for messages coming in on on this channel from other parts of *our* application
    // to be sent out to the ClusterClient
    rx: Option<Receiver<String>>,
    //If something wants to send something to this client, they can clone the `tx` channel.
    tx: Option<Sender<String>>,
    raw_stream: TcpStream,
}

impl ClusterClient {
    /// Creates and returns a new ClusterClient that sets up the stream clones and mpsc channels
    pub fn new(stream: TcpStream) -> ClusterClient {
        // TODO: Handle this better
        let reader = stream.try_clone().unwrap();
        let writer = stream.try_clone().unwrap();
        let (tx, rx) = channel();
        ClusterClient {
            reader: BufReader::new(reader),
            writer: BufWriter::new(writer),
            raw_stream: stream,
            tx: Some(tx),
            rx: Some(rx)
        }
    }

    // Writes a message as bytes to the connected ClusterClient
    fn w(&mut self, msg: &str) -> bool {
        match self.writer.write_all(msg.as_bytes()) {
            Ok(_) => match self.writer.flush() {
                Ok(_) => true,
                Err(e) => {
                    println!("Error flushing to client: {}", e);
                    false
                }
            },
            Err(e) => {
                println!("Error writing to client: {}", e);
                false
            }
        }
    }

    // This is a background loop that watches for incoming messages on the mpsc channel. When it receives one, it sends it out to the ClusterClient.
    fn recv_loop(&mut self) {
        // We take the rx channel so that we can move ownership into the thread and loop on it
        let chan = self.rx.take().unwrap();
        let mut writer = self.raw_stream.try_clone().unwrap();
        let _t = thread::spawn(move || {
            loop {
                match chan.recv() {
                    Ok(msg) => {
                        match writer.write_all(msg.as_bytes()) {
                            Ok(_) => {}
                            Err(_e) => {}
                        };
                        match writer.flush() {
                            Ok(_) => {}
                            Err(_e) => {}
                        }
                    }
                    Err(_e) => {}
                }
            }
        });
    }

    /// Main function to call to set up the ClusterClient
    pub fn run(&mut self) {
        // Starts the recv_loop in a background thread
        self.recv_loop();
        let mut buf = String::new();
        // Loop over the incoming data, waiting for data.
        loop {
            match self.reader.read_line(&mut buf) {
                Ok(_) => {
                    buf.trim_right();
                }
                Err(e) => {
                    println!("Error receiving: {:#?}", e);
                }
            }
        }
    }
}