enable multiple pw games

backend/Cargo.toml

@@ -8,9 +8,11 @@ edition = "2018"
 
 [dependencies]
 mozaic = { git = "https://github.com/ZeusWPI/MOZAICP" }
-tokio = "0.1.22"
 rand = { version = "0.6.5", default-features = true }
-futures = "0.1.28"
+
+async-std = { version = "1.5.0", features = ["attributes"] }
+futures = { version = "0.3.1", features = ["executor", "thread-pool"] }
+
 serde = "1.0.100"
 serde_derive = "1.0.100"
 serde_json = "1.0"

backend/src/main.rs

@@ -3,138 +3,124 @@ extern crate serde;
 extern crate serde_derive;
 extern crate serde_json;
 
+extern crate async_std;
 extern crate futures;
 extern crate mozaic;
 extern crate rand;
-extern crate tokio;
 
 extern crate tracing;
 extern crate tracing_futures;
 extern crate tracing_subscriber;
 
-use mozaic::errors;
+use tracing_subscriber::{EnvFilter, FmtSubscriber};
-use mozaic::messaging::types::*;
+
-use std::env;
 use std::net::SocketAddr;
+use std::{env, time};
 
-use tracing::{span, Level};
+use mozaic::modules::types::*;
-use tracing_futures::Instrument;
+use mozaic::modules::{game, StepLock};
-use tracing_subscriber::{fmt, EnvFilter};
 
-use mozaic::modules::{game, Aggregator, Steplock};
+use futures::executor::ThreadPool;
+use futures::future::FutureExt;
+
+use mozaic::graph;
+use mozaic::modules::*;
 
 mod planetwars;
 
 // Load the config and start the game.
-fn main() {
+#[async_std::main]
+async fn main() {
     let args: Vec<String> = env::args().collect();
     let name = args[0].clone();
-    match run(args) {
+    match run(args).await {
         None => print_info(&name),
         _ => {}
     };
 }
 
-use errors::Consumable;
+fn build_builder(
-use mozaic::modules::util;
+    pool: ThreadPool,
-use mozaic::modules::ConnectionManager;
+    number_of_clients: u64,
-use mozaic::runtime::Broker;
+    max_turns: u64,
-use rand::Rng;
+    map: &str,
-use std::collections::HashMap;
+    location: &str,
+) -> game::Builder<planetwars::PlanetWarsGame> {
+    let config = planetwars::Config {
+        map_file: map.to_string(),
+        max_turns: max_turns,
+    };
 
-fn print_info(name: &str) {
+    let game =
-    println!(
+        planetwars::PlanetWarsGame::new(config.create_game(number_of_clients as usize), location);
-        "Usage: {} map_location [number_of_clients [output [max_turns]]]",
+
-        name
+    let players: Vec<PlayerId> = (0..number_of_clients).collect();
-    );
+
+    game::Builder::new(players.clone(), game).with_step_lock(
+        StepLock::new(players.clone(), pool.clone())
+            .with_timeout(std::time::Duration::from_secs(1)),
+    )
 }
 
-pub fn run(args: Vec<String>) -> Option<()> {
+async fn run(args: Vec<String>) -> Option<()> {
-    let subscriber = fmt::Subscriber::builder()
+    let fut = graph::set_default();
-        .with_env_filter(EnvFilter::try_from_default_env().unwrap_or(EnvFilter::from("info")))
+
-        .without_time()
+    let sub = FmtSubscriber::builder()
-        .inherit_fields(true)
+        .with_env_filter(EnvFilter::from_default_env())
         .finish();
-    let _ = tracing::subscriber::set_global_default(subscriber);
+    tracing::subscriber::set_global_default(sub).unwrap();
 
     let addr = "127.0.0.1:9142".parse::<SocketAddr>().unwrap();
 
-    let manager_id: ReactorId = rand::thread_rng().gen();
-    let welcomer_id: ReactorId = rand::thread_rng().gen();
-    let aggregator_id: ReactorId = rand::thread_rng().gen();
-    let steplock_id: ReactorId = rand::thread_rng().gen();
-
     let map = args.get(1)?;
     let number_of_clients = args
         .get(2)
         .map(|x| x.parse().expect("Client number should be a number"))
         .unwrap_or(1);
     let location = args.get(3).map(|x| x.as_str()).unwrap_or("game.json");
-    let max_turns = args
+    let max_turns: u64 = args
         .get(4)
         .map(|x| x.parse().expect("Max turns should be a number"))
         .unwrap_or(500);
 
-    let ids: HashMap<util::Identifier, util::PlayerId> = (0..number_of_clients)
+    let pool = ThreadPool::new().ok()?;
-        .map(|x| (rand::thread_rng().gen::<u64>().into(), x.into()))
+    pool.spawn_ok(fut.map(|_| ()));
-        .collect();
 
-    let config = planetwars::Config {
+    let (gmb, handle) = game::Manager::builder(pool.clone());
-        map_file: map.to_string(),
+    let ep = TcpEndpoint::new(addr, pool.clone());
-        max_turns: max_turns,
+    let gmb = gmb.add_endpoint(ep, "TCP endpoint");
-    };
+    let mut gm = gmb.build();
-    let game =
-        planetwars::PlanetWarsGame::new(config.create_game(number_of_clients as usize), location);
 
-    println!("Tokens:");
+    let game_builder = build_builder(pool.clone(), number_of_clients, max_turns, map, location);
-    let keys: Vec<u64> = ids.keys().map(|&x| x.into()).collect();
+    std::thread::sleep(time::Duration::from_millis(3000));
-    for key in keys {
+
-        println!("key {}", key);
+    let mut current_game = gm.start_game(game_builder).await.unwrap();
+
+    loop {
+        match gm.get_state(current_game).await {
+            None => {
+                println!("Game finished, let's play a new one");
+                let game_builder =
+                    build_builder(pool.clone(), number_of_clients, max_turns, map, location);
+                current_game = gm.start_game(game_builder).await.unwrap();
+            }
+            Some(state) => {
+                println!("{:?}", state);
+            }
+        }
+        std::thread::sleep(time::Duration::from_millis(3000));
     }
 
-    tokio::run(futures::lazy(move || {
+    handle.await;
-        mozaic::graph::set_graph(mozaic::graph::Graph::new());
-        let mut broker = Broker::new().unwrap();
 
-        broker
+    std::thread::sleep(time::Duration::from_millis(100));
-            .spawn(
-                welcomer_id.clone(),
-                game::GameReactor::params(steplock_id.clone(), Box::new(game)),
-                "Main",
-            )
-            .display();
-        broker
-            .spawn(
-                steplock_id.clone(),
-                Steplock::new(
-                    broker.clone(),
-                    ids.values().cloned().collect(),
-                    welcomer_id.clone(),
-                    aggregator_id.clone(),
-                )
-                .with_timeout(5000)
-                .params(),
-                "Steplock",
-            )
-            .display();
-        broker
-            .spawn(
-                aggregator_id.clone(),
-                Aggregator::params(manager_id.clone(), steplock_id.clone()),
-                "Aggregator",
-            )
-            .display();
-        broker
-            .spawn(
-                manager_id.clone(),
-                ConnectionManager::params(broker.clone(), ids, aggregator_id.clone(), addr),
-                "Connection Manager",
-            )
-            .display();
-
-        Ok(())
-    }).instrument(span!(Level::TRACE, "main")),
-);
 
     Some(())
 }
+
+fn print_info(name: &str) {
+    println!(
+        "Usage: {} map_location [number_of_clients [output [max_turns]]]",
+        name
+    );
+}

backend/src/planetwars/mod.rs

@@ -1,5 +1,5 @@
-
 use mozaic::modules::game;
+use mozaic::modules::types::{Data, HostMsg, PlayerMsg};
 
 use serde_json;
 
@@ -9,38 +9,52 @@ use std::fs::File;
 use std::io::Write;
 
 mod pw_config;
-mod pw_serializer;
-mod pw_rules;
 mod pw_protocol;
-use pw_protocol::{ self as proto, CommandError };
+mod pw_rules;
-use pw_rules::Dispatch;
+mod pw_serializer;
 pub use pw_config::Config;
+use pw_protocol::{self as proto, CommandError};
+use pw_rules::Dispatch;
 
 pub struct PlanetWarsGame {
     state: pw_rules::PlanetWars,
     planet_map: HashMap<String, usize>,
-    log_file: File
+    log_file: File,
 }
 
 impl PlanetWarsGame {
-
     pub fn new(state: pw_rules::PlanetWars, location: &str) -> Self {
-        let planet_map = state.planets.iter().map(|p| (p.name.clone(), p.id)).collect();
+        let planet_map = state
+            .planets
+            .iter()
+            .map(|p| (p.name.clone(), p.id))
+            .collect();
         let file = File::create(location).unwrap();
 
         Self {
-            state, planet_map,
+            state,
+            planet_map,
             log_file: file,
         }
     }
 
-    fn dispatch_state(&mut self, were_alive: Vec<usize>, updates: &mut Vec<game::Update>, ) {
+    fn dispatch_state(&mut self, were_alive: Vec<usize>, updates: &mut Vec<HostMsg>) {
         let state = pw_serializer::serialize(&self.state);
-        write!(self.log_file, "{}\n", serde_json::to_string(&state).unwrap()).unwrap();
+        write!(
+            self.log_file,
+            "{}\n",
+            serde_json::to_string(&state).unwrap()
+        )
+        .unwrap();
 
         // println!("{}", serde_json::to_string(&state).unwrap());
 
-        for player in self.state.players.iter().filter(|p| were_alive.contains(&p.id)) {
+        for player in self
+            .state
+            .players
+            .iter()
+            .filter(|p| were_alive.contains(&p.id))
+        {
             let state = pw_serializer::serialize_rotated(&self.state, player.id);
             let state = if player.alive && !self.state.is_finished() {
                 proto::ServerMessage::GameState(state)
@@ -48,64 +62,80 @@ impl PlanetWarsGame {
                 proto::ServerMessage::FinalState(state)
             };
 
-            updates.push(
+            updates.push(HostMsg::Data(
-                game::Update::Player((player.id as u64).into(), serde_json::to_vec(&state).unwrap())
+                Data {
-            );
+                    value: serde_json::to_string(&state).unwrap(),
+                },
+                Some(player.id as u64),
+            ));
 
             if !player.alive || self.state.is_finished() {
                 println!("Kicking player {}", player.id);
-                updates.push(game::Update::Kick((player.id as u64).into()));
+                updates.push(HostMsg::Kick(player.id as u64));
             }
         }
     }
 
-    fn execute_commands<'a>(&mut self, turns: Vec<game::PlayerTurn<'a>>, updates: &mut Vec<game::Update>) {
+    fn execute_commands<'a>(&mut self, turns: Vec<PlayerMsg>, updates: &mut Vec<HostMsg>) {
-        for (player_id, command) in turns.into_iter() {
+        for PlayerMsg { id, data } in turns.into_iter() {
-            let player_num: usize = (*player_id).try_into().unwrap();
+            let player_num: usize = (id).try_into().unwrap();
-            let action = proto::ServerMessage::PlayerAction(self.execute_action(player_num, command));
+            let action = proto::ServerMessage::PlayerAction(self.execute_action(player_num, data));
-            let serialized_action = serde_json::to_vec(&action).unwrap();
+            let serialized_action = serde_json::to_string(&action).unwrap();
-            updates.push(game::Update::Player(player_id, serialized_action));
+            updates.push(HostMsg::Data(
+                Data {
+                    value: serialized_action,
+                },
+                Some(id),
+            ));
         }
     }
 
-    fn execute_action<'a>(&mut self, player_num: usize, turn: game::Turn<'a>) -> proto::PlayerAction {
+    fn execute_action<'a>(&mut self, player_num: usize, turn: Option<Data>) -> proto::PlayerAction {
         let turn = match turn {
-            game::Turn::Timeout => return proto::PlayerAction::Timeout,
+            None => return proto::PlayerAction::Timeout,
-            game::Turn::Action(bytes) => bytes,
+            Some(turn) => turn.value,
         };
 
-        let action: proto::Action = match serde_json::from_slice(&turn) {
+        let action: proto::Action = match serde_json::from_str(&turn) {
             Err(err) => return proto::PlayerAction::ParseError(err.to_string()),
             Ok(action) => action,
         };
 
-        let commands = action.commands.into_iter().map(|command| {
+        let commands = action
-            match self.check_valid_command(player_num, &command) {
+            .commands
-                Ok(dispatch) => {
+            .into_iter()
-                    self.state.dispatch(&dispatch);
+            .map(
-                    proto::PlayerCommand {
+                |command| match self.check_valid_command(player_num, &command) {
-                        command,
+                    Ok(dispatch) => {
-                        error: None,
+                        self.state.dispatch(&dispatch);
+                        proto::PlayerCommand {
+                            command,
+                            error: None,
+                        }
                     }
-                },
+                    Err(error) => proto::PlayerCommand {
-                Err(error) => {
-                    proto::PlayerCommand {
                         command,
                         error: Some(error),
-                    }
+                    },
-                }
+                },
-            }
+            )
-        }).collect();
+            .collect();
 
         return proto::PlayerAction::Commands(commands);
     }
 
-    fn check_valid_command(&self, player_num: usize, mv: &proto::Command) -> Result<Dispatch, CommandError> {
+    fn check_valid_command(
-        let origin_id = *self.planet_map
+        &self,
+        player_num: usize,
+        mv: &proto::Command,
+    ) -> Result<Dispatch, CommandError> {
+        let origin_id = *self
+            .planet_map
             .get(&mv.origin)
             .ok_or(CommandError::OriginDoesNotExist)?;
 
-        let target_id = *self.planet_map
+        let target_id = *self
+            .planet_map
             .get(&mv.destination)
             .ok_or(CommandError::DestinationDoesNotExist)?;
 
@@ -129,8 +159,14 @@ impl PlanetWarsGame {
     }
 }
 
-impl game::GameController for PlanetWarsGame {
+impl game::Controller for PlanetWarsGame {
-    fn step<'a>(&mut self, turns: Vec<game::PlayerTurn<'a>>) -> Vec<game::Update> {
+    fn start(&mut self) -> Vec<HostMsg> {
+        let mut updates = Vec::new();
+        self.dispatch_state(self.state.living_players(), &mut updates);
+        updates
+    }
+
+    fn step(&mut self, turns: Vec<PlayerMsg>) -> Vec<HostMsg> {
         let mut updates = Vec::new();
 
         let alive = self.state.living_players();
@@ -143,4 +179,8 @@ impl game::GameController for PlanetWarsGame {
 
         updates
     }
+
+    fn is_done(&mut self) -> bool {
+        self.state.is_finished()
+    }
 }

backend/src/planetwars/pw_config.rs

@@ -1,6 +1,6 @@
 use std::fs::File;
-use std::io::Read;
 use std::io;
+use std::io::Read;
 
 use serde_json;
 
@@ -17,7 +17,10 @@ impl Config {
     pub fn create_game(&self, clients: usize) -> PlanetWars {
         let planets = self.load_map(clients);
         let players = (0..clients)
-            .map(|client_id| Player { id: client_id, alive: true })
+            .map(|client_id| Player {
+                id: client_id,
+                alive: true,
+            })
             .collect();
 
         PlanetWars {
@@ -33,7 +36,8 @@ impl Config {
     fn load_map(&self, num_players: usize) -> Vec<Planet> {
         let map = self.read_map().expect("[PLANET_WARS] reading map failed");
 
-        return map.planets
+        return map
+            .planets
             .into_iter()
             .enumerate()
             .map(|(num, planet)| {
@@ -62,7 +66,8 @@ impl Config {
                     y: planet.y,
                     fleets: fleets,
                 };
-            }).collect();
+            })
+            .collect();
     }
 
     fn read_map(&self) -> io::Result<Map> {

backend/src/planetwars/pw_rules.rs

@@ -1,4 +1,3 @@
-
 /// The planet wars game rules.
 pub struct PlanetWars {
     pub players: Vec<Player>,
@@ -49,16 +48,12 @@ pub struct Dispatch {
 }
 
 impl PlanetWars {
-
     pub fn dispatch(&mut self, dispatch: &Dispatch) {
-        let distance = self.planets[dispatch.origin].distance(
+        let distance = self.planets[dispatch.origin].distance(&self.planets[dispatch.target]);
-            &self.planets[dispatch.target]
-        );
 
         let origin = &mut self.planets[dispatch.origin];
         origin.fleets[0].ship_count -= dispatch.ship_count;
 
-
         let expedition = Expedition {
             id: self.expedition_num,
             origin: dispatch.origin,
@@ -136,17 +131,13 @@ impl PlanetWars {
     }
 
     pub fn living_players(&self) -> Vec<usize> {
-        self.players.iter().filter_map(|p| {
+        self.players
-            if p.alive {
+            .iter()
-                Some(p.id)
+            .filter_map(|p| if p.alive { Some(p.id) } else { None })
-            } else {
+            .collect()
-                None
-            }
-        }).collect()
     }
 }
 
-
 impl Planet {
     pub fn owner(&self) -> Option<usize> {
         self.fleets.first().and_then(|f| f.owner)
@@ -176,7 +167,8 @@ impl Planet {
         // winner.ship_count -= second_largest.ship_count, but this does not
         // allow for simple customizations (such as changing combat balance).
 
-        self.fleets.sort_by(|a, b| a.ship_count.cmp(&b.ship_count).reverse());
+        self.fleets
+            .sort_by(|a, b| a.ship_count.cmp(&b.ship_count).reverse());
         while self.fleets.len() > 1 {
             let fleet = self.fleets.pop().unwrap();
             // destroy some ships

backend/src/planetwars/pw_serializer.rs

@@ -1,6 +1,5 @@
-
-use super::pw_rules::{PlanetWars, Planet, Expedition};
 use super::pw_protocol as proto;
+use super::pw_rules::{Expedition, Planet, PlanetWars};
 
 /// Serialize given gamestate
 pub fn serialize(state: &PlanetWars) -> proto::State {
@@ -28,12 +27,14 @@ impl<'a> Serializer<'a> {
 
     fn serialize_state(&self) -> proto::State {
         proto::State {
-            planets: self.state
+            planets: self
+                .state
                 .planets
                 .iter()
                 .map(|planet| self.serialize_planet(planet))
                 .collect(),
-            expeditions: self.state
+            expeditions: self
+                .state
                 .expeditions
                 .iter()
                 .map(|exp| self.serialize_expedition(exp))

client/bot.py

@@ -0,0 +1,58 @@
+#! /usr/bin/env python
+
+import socket, sys, subprocess, argparse, io, threading, json
+
+
+def execute(cmd):
+    popen = subprocess.Popen(cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE, universal_newlines=True)
+    yield(popen.stdin)
+
+    for stdout_line in iter(popen.stdout.readline, "\n"):
+        yield stdout_line
+    popen.stdout.close()
+    return_code = popen.wait()
+    if return_code:
+        raise subprocess.CalledProcessError(return_code, cmd)
+
+def connect(host, port, id):
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.connect((host, port))
+    s.sendall(f"{id}\n".encode("utf8"))
+    return s
+
+def handle_input(it, socket):
+    for line in it:
+        socket.sendall(line.encode('utf8'))
+
+def main():
+    parser = argparse.ArgumentParser(description='Run MOZAIC bot')
+    parser.add_argument('--id', '-i', required=True,
+                        help='The bot\'s ID')
+    parser.add_argument('--host', default="localhost",
+                        help='What host to connect to')
+    parser.add_argument('--port', '-p', default=6666, type=int,
+                        help='What port to connect to')
+    parser.add_argument('arguments', nargs=argparse.REMAINDER,
+                        help='How to run the bot')
+    args = parser.parse_args()
+
+    sock = connect(args.host, args.port, args.id)
+    f = sock.makefile("rw")
+
+    it = execute(args.arguments)
+    stdin = next(it)
+
+    threading.Thread(target=lambda: handle_input(it, sock), daemon=True).start()
+
+    line = f.readline()
+    while line:
+        content = json.loads(line)
+        if content["type"] == "game_state":
+            stdin.write(json.dumps(content["content"])+"\n")
+            stdin.flush()
+        line = f.readline()
+
+    print(content)
+
+if __name__ == "__main__":
+    main()

client/run2.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+python bot.py -p 9142 -i $1 python simple.py