diff options
Diffstat (limited to 'planetwars-matchrunner')
| -rw-r--r-- | planetwars-matchrunner/Cargo.toml | 3 | ||||
| -rw-r--r-- | planetwars-matchrunner/src/bin/testmatch.rs | 162 | ||||
| -rw-r--r-- | planetwars-matchrunner/src/docker_runner.rs | 188 | ||||
| -rw-r--r-- | planetwars-matchrunner/src/lib.rs | 31 |
4 files changed, 230 insertions, 154 deletions
diff --git a/planetwars-matchrunner/Cargo.toml b/planetwars-matchrunner/Cargo.toml index c4cb00d..b041d61 100644 --- a/planetwars-matchrunner/Cargo.toml +++ b/planetwars-matchrunner/Cargo.toml @@ -19,4 +19,5 @@ serde_json = "1.0" planetwars-rules = { path = "../planetwars-rules" } chrono = { version = "0.4", features = ["serde"] } bollard = { git = "https://github.com/antoinert/bollard" } -bytes = "1.1"
\ No newline at end of file +bytes = "1.1" +async-trait = "0.1"
\ No newline at end of file diff --git a/planetwars-matchrunner/src/bin/testmatch.rs b/planetwars-matchrunner/src/bin/testmatch.rs index a072b9f..db160cf 100644 --- a/planetwars-matchrunner/src/bin/testmatch.rs +++ b/planetwars-matchrunner/src/bin/testmatch.rs @@ -1,28 +1,6 @@ -extern crate planetwars_matchrunner; -extern crate tokio; +use std::{env, path::PathBuf}; -use std::collections::HashMap; -use std::io::{self, Write}; -use std::path::{Path, PathBuf}; -use std::pin::Pin; -use std::sync::{Arc, Mutex}; - -use bollard::container::{self, AttachContainerOptions, AttachContainerResults, LogOutput}; -use bollard::exec::StartExecResults; -use bollard::Docker; -use bytes::Bytes; -use futures::{Stream, StreamExt}; -use planetwars_matchrunner::{ - match_context::{EventBus, MatchCtx, PlayerHandle, RequestMessage}, - pw_match, MatchConfig, MatchMeta, PlayerInfo, -}; -use planetwars_rules::protocol as proto; -use std::env; -use tokio::io::{AsyncWrite, AsyncWriteExt}; -use tokio::sync::mpsc; - -const IMAGE: &'static str = "python:3.10.1-slim-buster"; -// const IMAGE: &'static str = "simplebot:latest"; +use planetwars_matchrunner::{docker_runner::DockerBotSpec, run_match, MatchConfig, MatchPlayer}; #[tokio::main] async fn main() { @@ -32,134 +10,30 @@ async fn main() { _run_match(map_path).await; } +const IMAGE: &'static str = "python:3.10-slim-buster"; + async fn _run_match(map_path: String) { - let docker = Docker::connect_with_socket_defaults().unwrap(); let code_dir_path = PathBuf::from("../simplebot"); - let params = BotParams { - image: IMAGE, - code_path: &code_dir_path, - argv: vec!["python", "simplebot.py"], + let bot_spec = DockerBotSpec { + image: IMAGE.to_string(), + code_path: code_dir_path, + argv: vec!["python".to_string(), "simplebot.py".to_string()], }; - let mut process = spawn_docker_process(&docker, params).await.unwrap(); - let state = proto::State { - planets: vec![ - proto::Planet { + run_match(MatchConfig { + map_path: PathBuf::from(map_path), + map_name: "hex".to_string(), + log_path: PathBuf::from("match.log"), + players: vec![ + MatchPlayer { name: "a".to_string(), - owner: Some(1), - ship_count: 100, - x: -1.0, - y: 0.0, + bot_spec: Box::new(bot_spec.clone()), }, - proto::Planet { + MatchPlayer { name: "b".to_string(), - owner: Some(2), - ship_count: 100, - x: 1.0, - y: 0.0, + bot_spec: Box::new(bot_spec.clone()), }, ], - expeditions: vec![], - }; - - let serialized = serde_json::to_vec(&state).unwrap(); - let out = process.communicate(&serialized).await.unwrap(); - - print!("got output: {}", String::from_utf8(out.to_vec()).unwrap()); -} - -pub struct BotParams<'a> { - pub image: &'a str, - pub code_path: &'a Path, - pub argv: Vec<&'a str>, -} - -async fn spawn_docker_process( - docker: &Docker, - params: BotParams<'_>, -) -> Result<ContainerProcess, bollard::errors::Error> { - let bot_code_dir = std::fs::canonicalize(params.code_path).unwrap(); - let code_dir_str = bot_code_dir.as_os_str().to_str().unwrap(); - - let config = container::Config { - image: Some(params.image), - host_config: Some(bollard::models::HostConfig { - binds: Some(vec![format!("{}:{}", code_dir_str, "/workdir")]), - ..Default::default() - }), - working_dir: Some("/workdir"), - cmd: Some(params.argv), - attach_stdin: Some(true), - attach_stdout: Some(true), - attach_stderr: Some(true), - open_stdin: Some(true), - ..Default::default() - }; - - let response = docker.create_container::<&str, &str>(None, config).await?; - let container_id = response.id; - - docker - .start_container::<String>(&container_id, None) - .await?; - - let AttachContainerResults { output, input } = docker - .attach_container( - &container_id, - Some(AttachContainerOptions::<String> { - stdout: Some(true), - stderr: Some(true), - stdin: Some(true), - stream: Some(true), - logs: Some(true), - ..Default::default() - }), - ) - .await?; - - Ok(ContainerProcess { - stdin: input, - output, }) -} - -pub struct ContainerProcess { - stdin: Pin<Box<dyn AsyncWrite + Send>>, - output: Pin<Box<dyn Stream<Item = Result<LogOutput, bollard::errors::Error>>>>, -} - -impl ContainerProcess { - pub async fn communicate(&mut self, input: &[u8]) -> io::Result<Bytes> { - self.write_line(input).await?; - self.read_line().await - } - - async fn write_line(&mut self, bytes: &[u8]) -> io::Result<()> { - self.stdin.write_all(bytes).await?; - self.stdin.write_u8(b'\n').await?; - self.stdin.flush().await?; - Ok(()) - } - - async fn read_line(&mut self) -> io::Result<Bytes> { - while let Some(item) = self.output.next().await { - let log_output = item.expect("failed to get log output"); - match log_output { - LogOutput::StdOut { message } => { - // TODO: this is not correct (buffering and such) - return Ok(message); - } - LogOutput::StdErr { message } => { - // TODO - println!("{}", String::from_utf8_lossy(&message)); - } - _ => (), - } - } - - Err(io::Error::new( - io::ErrorKind::UnexpectedEof, - "no response received", - )) - } + .await; } diff --git a/planetwars-matchrunner/src/docker_runner.rs b/planetwars-matchrunner/src/docker_runner.rs new file mode 100644 index 0000000..84101ef --- /dev/null +++ b/planetwars-matchrunner/src/docker_runner.rs @@ -0,0 +1,188 @@ +use std::io; +use std::path::PathBuf; +use std::pin::Pin; +use std::sync::{Arc, Mutex}; + +use async_trait::async_trait; +use bollard::container::{self, AttachContainerOptions, AttachContainerResults, LogOutput}; +use bollard::Docker; +use bytes::Bytes; +use futures::{Stream, StreamExt}; +use tokio::io::{AsyncWrite, AsyncWriteExt}; +use tokio::sync::mpsc; +use tokio::time::timeout; + +use crate::match_context::{EventBus, PlayerHandle, RequestError, RequestMessage}; +use crate::BotSpec; + +#[derive(Clone, Debug)] +pub struct DockerBotSpec { + pub image: String, + pub code_path: PathBuf, + pub argv: Vec<String>, +} + +#[async_trait] +impl BotSpec for DockerBotSpec { + async fn run_bot( + &self, + player_id: u32, + event_bus: Arc<Mutex<EventBus>>, + ) -> Box<dyn PlayerHandle> { + let (handle, runner) = create_docker_bot(player_id, event_bus); + let process = spawn_docker_process(self).await.unwrap(); + tokio::spawn(runner.run(process)); + return Box::new(handle); + } +} + +async fn spawn_docker_process( + params: &DockerBotSpec, +) -> Result<ContainerProcess, bollard::errors::Error> { + let docker = Docker::connect_with_socket_defaults()?; + let bot_code_dir = std::fs::canonicalize(¶ms.code_path).unwrap(); + let code_dir_str = bot_code_dir.as_os_str().to_str().unwrap(); + + let config = container::Config { + image: Some(params.image.clone()), + host_config: Some(bollard::models::HostConfig { + binds: Some(vec![format!("{}:{}", code_dir_str, "/workdir")]), + ..Default::default() + }), + working_dir: Some("/workdir".to_string()), + cmd: Some(params.argv.clone()), + attach_stdin: Some(true), + attach_stdout: Some(true), + attach_stderr: Some(true), + open_stdin: Some(true), + ..Default::default() + }; + + let response = docker + .create_container::<&str, String>(None, config) + .await?; + let container_id = response.id; + + docker + .start_container::<String>(&container_id, None) + .await?; + + let AttachContainerResults { output, input } = docker + .attach_container( + &container_id, + Some(AttachContainerOptions::<String> { + stdout: Some(true), + stderr: Some(true), + stdin: Some(true), + stream: Some(true), + logs: Some(true), + ..Default::default() + }), + ) + .await?; + + Ok(ContainerProcess { + stdin: input, + output, + }) +} + +pub struct ContainerProcess { + stdin: Pin<Box<dyn AsyncWrite + Send>>, + output: Pin<Box<dyn Stream<Item = Result<LogOutput, bollard::errors::Error>> + Send>>, +} + +impl ContainerProcess { + pub async fn communicate(&mut self, input: &[u8]) -> io::Result<Bytes> { + self.write_line(input).await?; + self.read_line().await + } + + async fn write_line(&mut self, bytes: &[u8]) -> io::Result<()> { + self.stdin.write_all(bytes).await?; + self.stdin.write_u8(b'\n').await?; + self.stdin.flush().await?; + Ok(()) + } + + async fn read_line(&mut self) -> io::Result<Bytes> { + while let Some(item) = self.output.next().await { + let log_output = item.expect("failed to get log output"); + match log_output { + LogOutput::StdOut { message } => { + // TODO: this is not correct (buffering and such) + return Ok(message); + } + LogOutput::StdErr { message } => { + // TODO + println!("{}", String::from_utf8_lossy(&message)); + } + _ => (), + } + } + + Err(io::Error::new( + io::ErrorKind::UnexpectedEof, + "no response received", + )) + } +} + +fn create_docker_bot( + player_id: u32, + event_bus: Arc<Mutex<EventBus>>, +) -> (DockerBotHandle, DockerBotRunner) { + let (tx, rx) = mpsc::unbounded_channel(); + let bot_handle = DockerBotHandle { tx }; + let bot_runner = DockerBotRunner { + player_id, + event_bus, + rx, + }; + (bot_handle, bot_runner) +} + +pub struct DockerBotHandle { + tx: mpsc::UnboundedSender<RequestMessage>, +} + +impl PlayerHandle for DockerBotHandle { + fn send_request(&mut self, r: RequestMessage) { + self.tx + .send(r) + .expect("failed to send message to local bot"); + } + + fn send_info(&mut self, _msg: String) { + // TODO: log this somewhere + // drop info message + } +} + +pub struct DockerBotRunner { + event_bus: Arc<Mutex<EventBus>>, + rx: mpsc::UnboundedReceiver<RequestMessage>, + player_id: u32, +} + +impl DockerBotRunner { + pub async fn run(mut self, mut process: ContainerProcess) { + while let Some(request) = self.rx.recv().await { + let resp_fut = process.communicate(&request.content); + let result = timeout(request.timeout, resp_fut) + .await + // TODO: how can this failure be handled cleanly? + .expect("process read failed"); + let result = match result { + Ok(line) => Ok(line.to_vec()), + Err(_elapsed) => Err(RequestError::Timeout), + }; + let request_id = (self.player_id, request.request_id); + + self.event_bus + .lock() + .unwrap() + .resolve_request(request_id, result); + } + } +} diff --git a/planetwars-matchrunner/src/lib.rs b/planetwars-matchrunner/src/lib.rs index 1f23551..2e4200c 100644 --- a/planetwars-matchrunner/src/lib.rs +++ b/planetwars-matchrunner/src/lib.rs @@ -1,4 +1,5 @@ pub mod bot_runner; +pub mod docker_runner; pub mod match_context; pub mod pw_match; @@ -8,6 +9,8 @@ use std::{ sync::{Arc, Mutex}, }; +use async_trait::async_trait; +use futures::{stream::FuturesOrdered, FutureExt, StreamExt}; use match_context::MatchCtx; use planetwars_rules::PwConfig; use serde::{Deserialize, Serialize}; @@ -35,8 +38,16 @@ pub struct PlayerInfo { pub struct MatchPlayer { pub name: String, - pub path: PathBuf, - pub argv: Vec<String>, + pub bot_spec: Box<dyn BotSpec>, +} + +#[async_trait] +pub trait BotSpec { + async fn run_bot( + &self, + player_id: u32, + event_bus: Arc<Mutex<EventBus>>, + ) -> Box<dyn PlayerHandle>; } pub async fn run_match(config: MatchConfig) { @@ -48,20 +59,22 @@ pub async fn run_match(config: MatchConfig) { let event_bus = Arc::new(Mutex::new(EventBus::new())); // start bots + // TODO: what happens when a bot fails? let players = config .players .iter() .enumerate() .map(|(player_id, player)| { let player_id = (player_id + 1) as u32; - let bot = bot_runner::Bot { - working_dir: player.path.clone(), - argv: player.argv.clone(), - }; - let handle = bot_runner::run_local_bot(player_id, event_bus.clone(), bot); - (player_id, Box::new(handle) as Box<dyn PlayerHandle>) + player + .bot_spec + .run_bot(player_id, event_bus.clone()) + .map(move |handle| (player_id, handle)) }) - .collect(); + .collect::<FuturesOrdered<_>>() + // await all results + .collect() + .await; let mut log_file = std::fs::File::create(config.log_path).expect("could not create log file"); // assemble the math meta struct |