Merge branch 'bot-api' into next

This commit is contained in:
Ilion Beyst 2022-07-04 20:16:42 +02:00
commit 268e080ec1
18 changed files with 493 additions and 32 deletions

View file

@ -3,6 +3,6 @@
members = [
"planetwars-rules",
"planetwars-matchrunner",
"planetwars-cli",
"planetwars-server",
"planetwars-client",
]

View file

@ -0,0 +1,18 @@
[package]
name = "planetwars-client"
version = "0.0.0"
edition = "2021"
# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
[dependencies]
tokio = { version = "1.15", features = ["full"] }
tokio-stream = "0.1.9"
prost = "0.10"
tonic = "0.7.2"
serde = { version = "1.0", features = ["derive"] }
toml = "0.5"
planetwars-matchrunner = { path = "../planetwars-matchrunner" }
[build-dependencies]
tonic-build = "0.7.2"

View file

@ -0,0 +1,9 @@
extern crate tonic_build;
fn main() -> Result<(), Box<dyn std::error::Error>> {
tonic_build::configure()
.build_server(false)
.build_client(true)
.compile(&["../proto/bot_api.proto"], &["../proto"])?;
Ok(())
}

View file

@ -0,0 +1,2 @@
name = "simplebot"
command = ["python", "../simplebot/simplebot.py"]

View file

@ -0,0 +1,72 @@
pub mod pb {
tonic::include_proto!("grpc.planetwars.bot_api");
}
use pb::bot_api_service_client::BotApiServiceClient;
use planetwars_matchrunner::bot_runner::Bot;
use serde::Deserialize;
use std::{path::PathBuf, time::Duration};
use tokio::sync::mpsc;
use tokio_stream::wrappers::UnboundedReceiverStream;
use tonic::{metadata::MetadataValue, transport::Channel, Request, Status};
#[derive(Deserialize)]
struct BotConfig {
#[allow(dead_code)]
name: String,
command: Vec<String>,
}
#[tokio::main]
async fn main() {
let content = std::fs::read_to_string("simplebot.toml").unwrap();
let bot_config: BotConfig = toml::from_str(&content).unwrap();
let channel = Channel::from_static("http://localhost:50051")
.connect()
.await
.unwrap();
let created_match = create_match(channel.clone()).await.unwrap();
run_player(bot_config, created_match.player_key, channel).await;
tokio::time::sleep(Duration::from_secs(1)).await;
}
async fn create_match(channel: Channel) -> Result<pb::CreatedMatch, Status> {
let mut client = BotApiServiceClient::new(channel);
let res = client
.create_match(Request::new(pb::MatchRequest {
opponent_name: "simplebot".to_string(),
}))
.await;
res.map(|response| response.into_inner())
}
async fn run_player(bot_config: BotConfig, player_key: String, channel: Channel) {
let mut client = BotApiServiceClient::with_interceptor(channel, |mut req: Request<()>| {
let player_key: MetadataValue<_> = player_key.parse().unwrap();
req.metadata_mut().insert("player_key", player_key);
Ok(req)
});
let mut bot_process = Bot {
working_dir: PathBuf::from("."),
argv: bot_config.command,
}
.spawn_process();
let (tx, rx) = mpsc::unbounded_channel();
let mut stream = client
.connect_bot(UnboundedReceiverStream::new(rx))
.await
.unwrap()
.into_inner();
while let Some(message) = stream.message().await.unwrap() {
let moves = bot_process.communicate(&message.content).await.unwrap();
tx.send(pb::PlayerRequestResponse {
request_id: message.request_id,
content: moves.as_bytes().to_vec(),
})
.unwrap();
}
}

View file

@ -8,6 +8,7 @@ edition = "2021"
[dependencies]
futures = "0.3"
tokio = { version = "1.15", features = ["full"] }
tokio-stream = "0.1.9"
hyper = "0.14"
axum = { version = "0.5", features = ["json", "headers", "multipart"] }
diesel = { version = "1.4.4", features = ["postgres", "chrono"] }
@ -29,9 +30,14 @@ config = { version = "0.12", features = ["toml"] }
thiserror = "1.0.31"
sha2 = "0.10"
tokio-util = { version="0.7.3", features=["io"] }
prost = "0.10"
tonic = "0.7.2"
# TODO: remove me
shlex = "1.1"
[build-dependencies]
tonic-build = "0.7.2"
[dev-dependencies]
parking_lot = "0.11"

View file

@ -0,0 +1,9 @@
extern crate tonic_build;
fn main() -> Result<(), Box<dyn std::error::Error>> {
tonic_build::configure()
.build_server(true)
.build_client(false)
.compile(&["../proto/bot_api.proto"], &["../proto"])?;
Ok(())
}

View file

@ -0,0 +1 @@
ALTER TABLE match_players ALTER COLUMN code_bundle_id SET NOT NULL;

View file

@ -0,0 +1 @@
ALTER TABLE match_players ALTER COLUMN code_bundle_id DROP NOT NULL;

View file

@ -25,7 +25,7 @@ pub struct NewMatchPlayer {
/// player id within the match
pub player_id: i32,
/// id of the bot behind this player
pub code_bundle_id: i32,
pub code_bundle_id: Option<i32>,
}
#[derive(Queryable, Identifiable)]
@ -44,11 +44,11 @@ pub struct MatchBase {
pub struct MatchPlayer {
pub match_id: i32,
pub player_id: i32,
pub code_bundle_id: i32,
pub code_bundle_id: Option<i32>,
}
pub struct MatchPlayerData {
pub code_bundle_id: i32,
pub code_bundle_id: Option<i32>,
}
pub fn create_match(
@ -92,7 +92,10 @@ pub fn list_matches(conn: &PgConnection) -> QueryResult<Vec<FullMatchData>> {
let matches = matches::table.get_results::<MatchBase>(conn)?;
let match_players = MatchPlayer::belonging_to(&matches)
.inner_join(code_bundles::table)
.left_join(
code_bundles::table
.on(match_players::code_bundle_id.eq(code_bundles::id.nullable())),
)
.left_join(bots::table.on(code_bundles::bot_id.eq(bots::id.nullable())))
.load::<FullMatchPlayerData>(conn)?
.grouped_by(&matches);
@ -120,7 +123,7 @@ pub struct FullMatchData {
// #[primary_key(base.match_id, base::player_id)]
pub struct FullMatchPlayerData {
pub base: MatchPlayer,
pub code_bundle: CodeBundle,
pub code_bundle: Option<CodeBundle>,
pub bot: Option<Bot>,
}
@ -142,7 +145,10 @@ pub fn find_match(id: i32, conn: &PgConnection) -> QueryResult<FullMatchData> {
let match_base = matches::table.find(id).get_result::<MatchBase>(conn)?;
let match_players = MatchPlayer::belonging_to(&match_base)
.inner_join(code_bundles::table)
.left_join(
code_bundles::table
.on(match_players::code_bundle_id.eq(code_bundles::id.nullable())),
)
.left_join(bots::table.on(code_bundles::bot_id.eq(bots::id.nullable())))
.load::<FullMatchPlayerData>(conn)?;

View file

@ -0,0 +1,272 @@
pub mod pb {
tonic::include_proto!("grpc.planetwars.bot_api");
}
use std::collections::HashMap;
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use runner::match_context::{EventBus, PlayerHandle, RequestError, RequestMessage};
use runner::match_log::MatchLogger;
use tokio::sync::{mpsc, oneshot};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tonic;
use tonic::transport::Server;
use tonic::{Request, Response, Status, Streaming};
use planetwars_matchrunner as runner;
use crate::db;
use crate::util::gen_alphanumeric;
use crate::ConnectionPool;
use super::matches::{MatchPlayer, RunMatch};
pub struct BotApiServer {
conn_pool: ConnectionPool,
router: PlayerRouter,
}
/// Routes players to their handler
#[derive(Clone)]
struct PlayerRouter {
routing_table: Arc<Mutex<HashMap<String, SyncThingData>>>,
}
impl PlayerRouter {
pub fn new() -> Self {
PlayerRouter {
routing_table: Arc::new(Mutex::new(HashMap::new())),
}
}
}
impl Default for PlayerRouter {
fn default() -> Self {
Self::new()
}
}
// TODO: implement a way to expire entries
impl PlayerRouter {
fn put(&self, player_key: String, entry: SyncThingData) {
let mut routing_table = self.routing_table.lock().unwrap();
routing_table.insert(player_key, entry);
}
fn take(&self, player_key: &str) -> Option<SyncThingData> {
// TODO: this design does not allow for reconnects. Is this desired?
let mut routing_table = self.routing_table.lock().unwrap();
routing_table.remove(player_key)
}
}
#[tonic::async_trait]
impl pb::bot_api_service_server::BotApiService for BotApiServer {
type ConnectBotStream = UnboundedReceiverStream<Result<pb::PlayerRequest, Status>>;
async fn connect_bot(
&self,
req: Request<Streaming<pb::PlayerRequestResponse>>,
) -> Result<Response<Self::ConnectBotStream>, Status> {
// TODO: clean up errors
let player_key = req
.metadata()
.get("player_key")
.ok_or_else(|| Status::unauthenticated("no player_key provided"))?;
let player_key_str = player_key
.to_str()
.map_err(|_| Status::invalid_argument("unreadable string"))?;
let sync_data = self
.router
.take(player_key_str)
.ok_or_else(|| Status::not_found("player_key not found"))?;
let stream = req.into_inner();
sync_data.tx.send(stream).unwrap();
Ok(Response::new(UnboundedReceiverStream::new(
sync_data.server_messages,
)))
}
async fn create_match(
&self,
req: Request<pb::MatchRequest>,
) -> Result<Response<pb::CreatedMatch>, Status> {
// TODO: unify with matchrunner module
let conn = self.conn_pool.get().await.unwrap();
let match_request = req.get_ref();
let opponent = db::bots::find_bot_by_name(&match_request.opponent_name, &conn)
.map_err(|_| Status::not_found("opponent not found"))?;
let opponent_code_bundle = db::bots::active_code_bundle(opponent.id, &conn)
.map_err(|_| Status::not_found("opponent has no code"))?;
let player_key = gen_alphanumeric(32);
let remote_bot_spec = Box::new(RemoteBotSpec {
player_key: player_key.clone(),
router: self.router.clone(),
});
let mut run_match = RunMatch::from_players(vec![
MatchPlayer::from_bot_spec(remote_bot_spec),
MatchPlayer::from_code_bundle(&opponent_code_bundle),
]);
let created_match = run_match
.store_in_database(&conn)
.expect("failed to save match");
run_match.spawn(self.conn_pool.clone());
Ok(Response::new(pb::CreatedMatch {
match_id: created_match.base.id,
player_key,
}))
}
}
// TODO: please rename me
struct SyncThingData {
tx: oneshot::Sender<Streaming<pb::PlayerRequestResponse>>,
server_messages: mpsc::UnboundedReceiver<Result<pb::PlayerRequest, Status>>,
}
struct RemoteBotSpec {
player_key: String,
router: PlayerRouter,
}
#[tonic::async_trait]
impl runner::BotSpec for RemoteBotSpec {
async fn run_bot(
&self,
player_id: u32,
event_bus: Arc<Mutex<EventBus>>,
_match_logger: MatchLogger,
) -> Box<dyn PlayerHandle> {
let (tx, rx) = oneshot::channel();
let (server_msg_snd, server_msg_recv) = mpsc::unbounded_channel();
self.router.put(
self.player_key.clone(),
SyncThingData {
tx,
server_messages: server_msg_recv,
},
);
let fut = tokio::time::timeout(Duration::from_secs(10), rx);
match fut.await {
Ok(Ok(client_messages)) => {
// let client_messages = rx.await.unwrap();
tokio::spawn(handle_bot_messages(
player_id,
event_bus.clone(),
client_messages,
));
}
_ => {
// ensure router cleanup
self.router.take(&self.player_key);
}
};
// If the player did not connect, the receiving half of `sender`
// will be dropped here, resulting in a time-out for every turn.
// This is fine for now, but
// TODO: provide a formal mechanism for player startup failure
Box::new(RemoteBotHandle {
sender: server_msg_snd,
player_id,
event_bus,
})
}
}
async fn handle_bot_messages(
player_id: u32,
event_bus: Arc<Mutex<EventBus>>,
mut messages: Streaming<pb::PlayerRequestResponse>,
) {
while let Some(message) = messages.message().await.unwrap() {
let request_id = (player_id, message.request_id as u32);
event_bus
.lock()
.unwrap()
.resolve_request(request_id, Ok(message.content));
}
}
struct RemoteBotHandle {
sender: mpsc::UnboundedSender<Result<pb::PlayerRequest, Status>>,
player_id: u32,
event_bus: Arc<Mutex<EventBus>>,
}
impl PlayerHandle for RemoteBotHandle {
fn send_request(&mut self, r: RequestMessage) {
let res = self.sender.send(Ok(pb::PlayerRequest {
request_id: r.request_id as i32,
content: r.content,
}));
match res {
Ok(()) => {
// schedule a timeout. See comments at method implementation
tokio::spawn(schedule_timeout(
(self.player_id, r.request_id),
r.timeout,
self.event_bus.clone(),
));
}
Err(_send_error) => {
// cannot contact the remote bot anymore;
// directly mark all requests as timed out.
// TODO: create a dedicated error type for this.
// should it be logged?
println!("send error: {:?}", _send_error);
self.event_bus
.lock()
.unwrap()
.resolve_request((self.player_id, r.request_id), Err(RequestError::Timeout));
}
}
}
}
// TODO: this will spawn a task for every request, which might not be ideal.
// Some alternatives:
// - create a single task that manages all time-outs.
// - intersperse timeouts with incoming client messages
// - push timeouts upwards, into the matchrunner logic (before we hit the playerhandle).
// This was initially not done to allow timer start to be delayed until the message actually arrived
// with the player. Is this still needed, or is there a different way to do this?
//
async fn schedule_timeout(
request_id: (u32, u32),
duration: Duration,
event_bus: Arc<Mutex<EventBus>>,
) {
tokio::time::sleep(duration).await;
event_bus
.lock()
.unwrap()
.resolve_request(request_id, Err(RequestError::Timeout));
}
pub async fn run_bot_api(pool: ConnectionPool) {
let router = PlayerRouter::new();
let server = BotApiServer {
router,
conn_pool: pool.clone(),
};
let addr = SocketAddr::from(([127, 0, 0, 1], 50051));
Server::builder()
.add_service(pb::bot_api_service_server::BotApiServiceServer::new(server))
.serve(addr)
.await
.unwrap()
}

View file

@ -16,32 +16,54 @@ use crate::{
const PYTHON_IMAGE: &str = "python:3.10-slim-buster";
pub struct RunMatch<'a> {
pub struct RunMatch {
log_file_name: String,
player_code_bundles: Vec<&'a db::bots::CodeBundle>,
players: Vec<MatchPlayer>,
match_id: Option<i32>,
}
impl<'a> RunMatch<'a> {
pub fn from_players(player_code_bundles: Vec<&'a db::bots::CodeBundle>) -> Self {
pub struct MatchPlayer {
bot_spec: Box<dyn BotSpec>,
// meta that will be passed on to database
code_bundle_id: Option<i32>,
}
impl MatchPlayer {
pub fn from_code_bundle(code_bundle: &db::bots::CodeBundle) -> Self {
MatchPlayer {
bot_spec: code_bundle_to_botspec(code_bundle),
code_bundle_id: Some(code_bundle.id),
}
}
pub fn from_bot_spec(bot_spec: Box<dyn BotSpec>) -> Self {
MatchPlayer {
bot_spec,
code_bundle_id: None,
}
}
}
impl RunMatch {
pub fn from_players(players: Vec<MatchPlayer>) -> Self {
let log_file_name = format!("{}.log", gen_alphanumeric(16));
RunMatch {
log_file_name,
player_code_bundles,
players,
match_id: None,
}
}
pub fn runner_config(&self) -> runner::MatchConfig {
pub fn into_runner_config(self) -> runner::MatchConfig {
runner::MatchConfig {
map_path: PathBuf::from(MAPS_DIR).join("hex.json"),
map_name: "hex".to_string(),
log_path: PathBuf::from(MATCHES_DIR).join(&self.log_file_name),
players: self
.player_code_bundles
.iter()
.map(|b| runner::MatchPlayer {
bot_spec: code_bundle_to_botspec(b),
.players
.into_iter()
.map(|player| runner::MatchPlayer {
bot_spec: player.bot_spec,
})
.collect(),
}
@ -56,10 +78,10 @@ impl<'a> RunMatch<'a> {
log_path: &self.log_file_name,
};
let new_match_players = self
.player_code_bundles
.players
.iter()
.map(|b| db::matches::MatchPlayerData {
code_bundle_id: b.id,
.map(|p| db::matches::MatchPlayerData {
code_bundle_id: p.code_bundle_id,
})
.collect::<Vec<_>>();
@ -70,7 +92,7 @@ impl<'a> RunMatch<'a> {
pub fn spawn(self, pool: ConnectionPool) -> JoinHandle<MatchOutcome> {
let match_id = self.match_id.expect("match must be saved before running");
let runner_config = self.runner_config();
let runner_config = self.into_runner_config();
tokio::spawn(run_match_task(pool, runner_config, match_id))
}
}

View file

@ -1,5 +1,6 @@
// This module implements general domain logic, not directly
// tied to the database or API layers.
pub mod bot_api;
pub mod bots;
pub mod matches;
pub mod ranking;

View file

@ -1,8 +1,8 @@
use crate::{db::bots::Bot, DbPool};
use crate::db;
use crate::modules::matches::RunMatch;
use diesel::{PgConnection, QueryResult};
use crate::modules::matches::{MatchPlayer, RunMatch};
use rand::seq::SliceRandom;
use std::collections::HashMap;
use std::mem;
@ -44,9 +44,12 @@ async fn play_ranking_match(selected_bots: Vec<Bot>, db_pool: DbPool) {
code_bundles.push(code_bundle);
}
let code_bundle_refs = code_bundles.iter().collect::<Vec<_>>();
let players = code_bundles
.iter()
.map(MatchPlayer::from_code_bundle)
.collect::<Vec<_>>();
let mut run_match = RunMatch::from_players(code_bundle_refs);
let mut run_match = RunMatch::from_players(players);
run_match
.store_in_database(&db_conn)
.expect("could not store match in db");

View file

@ -1,7 +1,7 @@
use crate::db;
use crate::db::matches::{FullMatchData, FullMatchPlayerData};
use crate::modules::bots::save_code_bundle;
use crate::modules::matches::RunMatch;
use crate::modules::matches::{MatchPlayer, RunMatch};
use crate::ConnectionPool;
use axum::extract::Extension;
use axum::Json;
@ -46,7 +46,10 @@ pub async fn submit_bot(
// TODO: can we recover from this?
.expect("could not save bot code");
let mut run_match = RunMatch::from_players(vec![&player_code_bundle, &opponent_code_bundle]);
let mut run_match = RunMatch::from_players(vec![
MatchPlayer::from_code_bundle(&player_code_bundle),
MatchPlayer::from_code_bundle(&opponent_code_bundle),
]);
let match_data = run_match
.store_in_database(&conn)
.expect("failed to save match");
@ -58,12 +61,12 @@ pub async fn submit_bot(
match_players: vec![
FullMatchPlayerData {
base: match_data.match_players[0].clone(),
code_bundle: player_code_bundle,
code_bundle: Some(player_code_bundle),
bot: None,
},
FullMatchPlayerData {
base: match_data.match_players[1].clone(),
code_bundle: opponent_code_bundle,
code_bundle: Some(opponent_code_bundle),
bot: Some(opponent),
},
],

View file

@ -61,7 +61,7 @@ pub async fn play_match(
});
bot_ids.push(matches::MatchPlayerData {
code_bundle_id: code_bundle.id,
code_bundle_id: Some(code_bundle.id),
});
}
@ -107,7 +107,7 @@ pub struct ApiMatch {
#[derive(Serialize, Deserialize)]
pub struct ApiMatchPlayer {
code_bundle_id: i32,
code_bundle_id: Option<i32>,
bot_id: Option<i32>,
bot_name: Option<String>,
}
@ -127,7 +127,7 @@ pub fn match_data_to_api(data: matches::FullMatchData) -> ApiMatch {
.match_players
.iter()
.map(|_p| ApiMatchPlayer {
code_bundle_id: _p.code_bundle.id,
code_bundle_id: _p.code_bundle.as_ref().map(|cb| cb.id),
bot_id: _p.bot.as_ref().map(|b| b.id),
bot_name: _p.bot.as_ref().map(|b| b.name.clone()),
})

View file

@ -31,7 +31,7 @@ table! {
match_players (match_id, player_id) {
match_id -> Int4,
player_id -> Int4,
code_bundle_id -> Int4,
code_bundle_id -> Nullable<Int4>,
}
}

36
proto/bot_api.proto Normal file
View file

@ -0,0 +1,36 @@
syntax = "proto3";
package grpc.planetwars.bot_api;
message Hello {
string hello_message = 1;
}
message HelloResponse {
string response = 1;
}
message PlayerRequest {
int32 request_id = 1;
bytes content = 2;
}
message PlayerRequestResponse {
int32 request_id = 1;
bytes content = 2;
}
message MatchRequest {
string opponent_name = 1;
}
message CreatedMatch {
int32 match_id = 1;
string player_key = 2;
}
service BotApiService {
rpc CreateMatch(MatchRequest) returns (CreatedMatch);
// server sends requests to the player, player responds
rpc ConnectBot(stream PlayerRequestResponse) returns (stream PlayerRequest);
}