Add machine section for global state.

Add networking instructions.
Add functions with different set of arguments.

Author: erthalion

src/lib.rs

@@ -3,6 +3,7 @@ use serde::Deserialize;
 use std::{fmt::Display, net::Ipv4Addr};
 use syscalls::Sysno;
 
+pub mod machine;
 pub mod script;
 pub mod worker;
 

src/machine.rs

@@ -0,0 +1,71 @@
+use crate::script::ast::MachineInstruction;
+
+use log::{debug, trace};
+use std::{
+    io::{prelude::*, BufReader},
+    net::TcpListener,
+    thread,
+};
+
+#[derive(Debug)]
+pub enum MachineError {
+    Internal,
+}
+
+fn start_server(addr: String, target_port: u16) -> Result<(), MachineError> {
+    debug!("Starting server at {:?}:{:?}", addr, target_port);
+
+    let listener = TcpListener::bind((addr, target_port)).unwrap();
+
+    for stream in listener.incoming() {
+        let mut stream = stream.unwrap();
+
+        // As a simplest solution to keep a connection open, spawn a
+        // thread.  It's not the best one though, as we waste resources.
+        // For the purpose of only keeping connections open we could e.g.
+        // spawn only two threads, where the first one receives connections
+        // and adds streams into the list of active, and the second iterates
+        // through streams and replies. This way the connections will have
+        // high latency, but for the purpose of networking workload it
+        // doesn't matter.
+        thread::spawn(move || loop {
+            let mut buf_reader = BufReader::new(&stream);
+            let mut buffer = String::new();
+
+            match buf_reader.read_line(&mut buffer) {
+                Ok(0) => {
+                    // EOF, exit
+                    trace!("EOF");
+                    return;
+                }
+                Ok(_n) => {
+                    trace!("Received {:?}", buffer);
+
+                    let response = "hello\n";
+                    match stream.write_all(response.as_bytes()) {
+                        Ok(_) => {
+                            // Response is sent, handle the next one
+                        }
+                        Err(e) => {
+                            trace!("ERROR: sending response, {}", e);
+                            break;
+                        }
+                    }
+                }
+                Err(e) => {
+                    trace!("ERROR: reading a line, {}", e)
+                }
+            }
+        });
+    }
+
+    Ok(())
+}
+
+pub fn apply(instr: MachineInstruction) -> Result<(), MachineError> {
+    match instr {
+        MachineInstruction::Server { port } => {
+            start_server("127.0.0.1".to_string(), port)
+        }
+    }
+}

src/main.rs

@@ -22,6 +22,7 @@ use core_affinity::CoreId;
 use docopt::Docopt;
 use fork::{fork, Fork};
 use itertools::iproduct;
+use itertools::{Either, Itertools};
 use nix::errno::Errno;
 use nix::sys::signal::{kill, Signal};
 use nix::sys::wait::waitpid;
@@ -30,6 +31,7 @@ use serde::Deserialize;
 use std::time::SystemTime;
 use std::{thread, time};
 
+use berserker::machine::apply;
 use berserker::script::{ast::Node, parser::parse_instructions};
 use berserker::{
     worker::new_script_worker, worker::new_worker, WorkloadConfig,
@@ -58,15 +60,35 @@ fn run_script(script_path: String) -> Vec<Option<i32>> {
         parse_instructions(&std::fs::read_to_string(script_path).unwrap())
             .unwrap();
 
-    ast.iter().for_each(|node| {
+    let (machine, works): (Vec<_>, Vec<_>) =
+        ast.iter().partition_map(|node| match node {
+            Node::Work { .. } => Either::Right(node),
+            Node::Machine { .. } => Either::Left(node),
+        });
+
+    if let Some(m) = machine.into_iter().next() {
+        let Node::Machine { m_instructions } = m.clone() else {
+            unreachable!()
+        };
+
+        for instr in m_instructions {
+            debug!("INSTR {:?}", instr);
+            thread::spawn(move || apply(instr.clone()));
+        }
+    };
+
+    works.into_iter().for_each(|node| {
         debug!("AST NODE: {:?}", node);
 
         let Node::Work {
             name: _,
             args,
             instructions: _,
             dist: _,
-        } = node;
+        } = node
+        else {
+            unreachable!()
+        };
 
         let workers: u32 = args
             .get("workers")

src/script/ast.rs

@@ -14,13 +14,21 @@ pub enum Instruction {
     Debug { text: String },
 }
 
+#[derive(Debug, Clone)]
+pub enum MachineInstruction {
+    Server { port: u16 },
+}
+
 #[derive(Debug, Clone)]
 pub enum Dist {
     Exp { rate: f64 },
 }
 
 #[derive(Debug, Clone)]
 pub enum Node {
+    Machine {
+        m_instructions: Vec<MachineInstruction>,
+    },
     Work {
         name: String,
         args: HashMap<String, String>,

src/script/grammar.peg

@@ -6,7 +6,7 @@ ident = @{ident_char ~ (ASCII_DIGIT | ident_char)*}
 
 constant = {
     "\"" ~ value ~ "\""
-  | ASCII_DIGIT
+  | ASCII_DIGIT*
 }
 
 arg = {
@@ -15,7 +15,7 @@ arg = {
 }
 
 args = {"(" ~ (arg ~ ("," ~ arg)* ~ ","?)? ~ ")"}
-value = {(ASCII_ALPHANUMERIC| "." | " " | "/")*}
+value = {(ASCII_ALPHANUMERIC| "." | " " | "/" | ":")*}
 
 param = {ident ~ "=" ~ value}
 params = {"(" ~ (param ~ ("," ~ param)* ~ ","?)? ~ ")"}
@@ -24,13 +24,15 @@ task = { "task" }
 network = { "network" }
 port = { "port" }
 open = { "open" }
+ping = { "ping" }
 debug = { "debug" }
 
 funcName = {
     task
   | network
   | port
   | open
+  | ping
   | debug
 }
 
@@ -65,6 +67,20 @@ function = {
     ~ dist?
 }
 
+server = { "server" }
+load = { "load" }
+memory = { "memory" }
+
+mInstrName = {
+    server
+  | load
+  | memory
+}
+
+machineInstruction = {mInstrName ~ args? ~ ";"}
+
+machine = {"machine" ~ "{" ~ machineInstruction* ~ "}"}
+
 file = {
-    SOI ~ (expr*) ~ EOI
+    SOI ~ machine? ~ (expr*) ~ EOI
 }

src/script/parser.rs

@@ -2,7 +2,7 @@ use log::debug;
 use pest::{self, error::Error, Parser};
 use std::collections::HashMap;
 
-use crate::script::ast::{Arg, Dist, Instruction, Node};
+use crate::script::ast::{Arg, Dist, Instruction, MachineInstruction, Node};
 
 // ANCHOR: parser
 #[derive(pest_derive::Parser)]
@@ -18,9 +18,17 @@ pub fn parse_instructions(source: &str) -> Result<Vec<Node>, Error<Rule>> {
     pest::set_error_detail(true);
     let mut ast = vec![];
     let pairs = InstructionParser::parse(Rule::file, source)?;
+    let expr_rules = [Rule::expr, Rule::machine];
+
     for pair in pairs {
-        if let Rule::file = pair.as_rule() {
-            ast.push(build_ast_from_expr(pair.into_inner().next().unwrap()));
+        if pair.as_rule() != Rule::file {
+            continue;
+        }
+
+        for i in pair.into_inner() {
+            if expr_rules.contains(&i.as_rule()) {
+                ast.push(build_ast_from_expr(i));
+            }
         }
     }
     debug!("AST {:?}", ast);
@@ -30,6 +38,9 @@ pub fn parse_instructions(source: &str) -> Result<Vec<Node>, Error<Rule>> {
 fn build_ast_from_expr(pair: pest::iterators::Pair<Rule>) -> Node {
     match pair.as_rule() {
         Rule::expr => build_ast_from_expr(pair.into_inner().next().unwrap()),
+        Rule::machine => Node::Machine {
+            m_instructions: build_ast_from_minstr(pair.into_inner()),
+        },
         Rule::function => {
             let mut inner = pair.into_inner();
             let mut work = inner.next().unwrap().into_inner();
@@ -65,6 +76,49 @@ fn build_ast_from_expr(pair: pest::iterators::Pair<Rule>) -> Node {
     }
 }
 
+fn build_ast_from_minstr(
+    pair: pest::iterators::Pairs<Rule>,
+) -> Vec<MachineInstruction> {
+    let mut instr = vec![] as Vec<MachineInstruction>;
+
+    for i in pair {
+        let mut inner = i.into_inner();
+        let name = inner.next().unwrap();
+        match name.into_inner().next().unwrap().as_rule() {
+            Rule::server => {
+                let port: u16 = inner
+                    .next()
+                    .unwrap()
+                    .into_inner()
+                    .next()
+                    .unwrap()
+                    .as_span()
+                    .as_str()
+                    .to_string()
+                    .parse()
+                    .unwrap();
+                instr.push(MachineInstruction::Server { port });
+            }
+            unknown => panic!("Unknown machine instruction: {unknown:?}"),
+        }
+
+        //debug!("PAIR {:?}", inner.next().unwrap().as_span());
+        //debug!("PAIR {:?}", inner.next().unwrap().into_inner().next().unwrap().as_span());
+        //match i.as_rule() {
+        //Rule::mInstrName => {
+        //debug!("INSTR {:?}", i.into_inner());
+        ////let mut instrs = i.into_inner().next().unwrap().into_inner();
+        ////let name = instrs.next().unwrap().as_span().as_str().to_string();
+
+        //instr.push(MachineInstruction::Server { port: 80 });
+        //}
+        //unknown => panic!("Unknown machine instruction: {unknown:?}"),
+        //}
+    }
+
+    instr
+}
+
 fn build_ast_from_instr(pair: pest::iterators::Pair<Rule>) -> Vec<Instruction> {
     let mut instr = vec![] as Vec<Instruction>;