multiple AI remotes - wip

modules/ai/src/main/Ai.scala

@@ -8,6 +8,4 @@ trait Ai {
   def play(game: Game, pgn: String, initialFen: Option[String], level: Int): Fu[(Game, Move)]
 
   def analyse(pgn: String, initialFen: Option[String]): Fu[AnalysisMaker]
-
-  def load: Fu[Option[Int]]
 }

modules/ai/src/main/Env.scala

@@ -1,5 +1,7 @@
 package lila.ai
 
+import scala.collection.JavaConversions._
+
 import akka.actor._
 import akka.pattern.pipe
 import com.typesafe.config.Config
@@ -14,9 +16,10 @@ final class Env(
     val EngineName = config getString "engine"
     val IsServer = config getBoolean "server"
     val IsClient = config getBoolean "client"
-    val StockfishPlayUrl = config getString "stockfish.play.url"
-    val StockfishAnalyseUrl = config getString "stockfish.analyse.url"
-    val StockfishLoadUrl = config getString "stockfish.load.url"
+    val StockfishRemotes = config getStringList "stockfish.remotes" toList
+    val StockfishPlayRoute = config getString "stockfish.play.route"
+    val StockfishAnalyseRoute = config getString "stockfish.analyse.route"
+    val StockfishLoadRoute = config getString "stockfish.load.route"
     val StockfishQueueName = config getString "stockfish.queue.name"
     val StockfishQueueDispatcher = config getString "stockfish.queue.dispatcher"
     val StockfishAnalyseTimeout = config duration "stockfish.analyse.timeout"
@@ -34,10 +37,10 @@ final class Env(
     analyseTimeout = config duration "stockfish.analyse.timeout",
     debug = config getBoolean "stockfish.debug")
 
-  def ai: () ⇒ Fu[Ai] = () ⇒ (EngineName, IsClient) match {
-    case ("stockfish", true)  ⇒ stockfishClient or stockfishAi
-    case ("stockfish", false) ⇒ fuccess(stockfishAi)
-    case _                    ⇒ fuccess(stupidAi)
+  lazy val ai: Ai = (EngineName, IsClient) match {
+    case ("stockfish", true)  ⇒ stockfishClient
+    case ("stockfish", false) ⇒ stockfishAi
+    case _                    ⇒ stupidAi
   }
 
   def isServer = IsServer
@@ -47,20 +50,26 @@ final class Env(
     def receive = {
       case lila.hub.actorApi.ai.GetLoad ⇒ IsClient.fold(
         stockfishClient.load pipeTo sender,
-        sender ! none
+        sender ! Nil
       )
       case lila.hub.actorApi.ai.Analyse(id, pgn, fen) ⇒
-        ai() flatMap { _.analyse(pgn, fen) } map { _(id) } pipeTo sender
+        ai.analyse(pgn, fen) map { _(id) } pipeTo sender
     }
   }), name = ActorName)
 
   private lazy val stockfishAi = new stockfish.Ai(stockfishServer)
 
   private lazy val stockfishClient = new stockfish.Client(
-    playUrl = StockfishPlayUrl,
-    analyseUrl = StockfishAnalyseUrl,
-    loadUrl = StockfishLoadUrl,
-    system = system)
+    dispatcher = system.actorOf(
+      Props(new stockfish.remote.Dispatcher(
+        urls = StockfishRemotes,
+        router = stockfish.remote.Router(
+          playRoute = StockfishPlayRoute,
+          analyseRoute = StockfishAnalyseRoute,
+          loadRoute = StockfishLoadRoute) _,
+        scheduler = system.scheduler
+      )), name = "stockfish-dispatcher"),
+    fallback = stockfishAi)
 
   lazy val stockfishServer = new stockfish.Server(
     queue = stockfishQueue,

modules/ai/src/main/StupidAi.scala

@@ -13,6 +13,4 @@ private[ai] final class StupidAi extends Ai {
 
   def analyse(pgn: String, initialFen: Option[String]) =
     throw new RuntimeException("Stupid analysis is not implemented")
-
-  def load = fuccess(0.some)
 }

modules/ai/src/main/stockfish/Ai.scala

@@ -16,8 +16,6 @@ final class Ai(server: Server) extends lila.ai.Ai {
   def analyse(pgn: String, initialFen: Option[String]): Fu[AnalysisMaker] =
     server.analyse(pgn, initialFen)
 
-  def load: Fu[Option[Int]] = server.load map(_.some)
-
   private def withValidSituation[A](game: Game)(op: ⇒ Fu[A]): Fu[A] =
     if (game.situation playable true) op
     else fufail("[ai stockfish] invalid game situation: " + game.situation)

modules/ai/src/main/stockfish/Client.scala

@@ -3,74 +3,37 @@ package stockfish
 
 import scala.concurrent.duration._
 
-import actorApi.monitor._
 import akka.actor._
 import akka.pattern.{ ask, pipe }
 import chess.format.UciMove
 import chess.{ Game, Move }
 import play.api.libs.concurrent._
 import play.api.Play.current
+import remote.actorApi._
 
 import lila.analyse.AnalysisMaker
-import lila.common.ws.WS
 import lila.hub.actorApi.ai.GetLoad
 
 final class Client(
-    val playUrl: String,
-    analyseUrl: String,
-    loadUrl: String,
-    system: ActorSystem) extends lila.ai.Ai {
+    dispatcher: ActorRef,
+    fallback: lila.ai.Ai) extends lila.ai.Ai {
+
+  private implicit val timeout = makeTimeout minutes 60
 
   def play(game: Game, pgn: String, initialFen: Option[String], level: Int): Fu[(Game, Move)] =
-    fetchMove(pgn, ~initialFen, level) flatMap { Stockfish.applyMove(game, pgn, _) }
+    dispatcher ? Play(pgn, ~initialFen, level) mapTo manifest[String] flatMap {
+      Stockfish.applyMove(game, pgn, _)
+    } recoverWith {
+      case e: Exception ⇒ fallback.play(game, pgn, initialFen, level)
+    }
 
   def analyse(pgn: String, initialFen: Option[String]): Fu[AnalysisMaker] =
-    fetchAnalyse(pgn, ~initialFen) flatMap { str ⇒
+    dispatcher ? Analyse(pgn, ~initialFen) mapTo manifest[String] flatMap { str ⇒
       (AnalysisMaker(str, true) toValid "Can't read analysis results").future
+    } recoverWith {
+      case e: Exception ⇒ fallback.analyse(pgn, initialFen)
     }
 
-  def load: Fu[Option[Int]] = {
-    import makeTimeout.short
-    actor ? GetLoad mapTo manifest[Option[Int]]
-  }
-
-  private val actor = system.actorOf(Props(new Actor {
-
-    private var load = none[Int]
-
-    def receive = {
-      case GetLoad   ⇒ sender ! load
-      case CalculateLoad ⇒ scala.concurrent.Future {
-        try {
-          load = fetchLoad await makeTimeout.seconds(1)
-        }
-        catch {
-          case e: Exception ⇒ {
-            logwarn("[stockfish client calculate load] " + e.getMessage)
-            load = none
-          }
-        }
-      }
-    }
-  }))
-  system.scheduler.schedule(1.millis, 1.second, actor, CalculateLoad)
-
-  def or(fallback: lila.ai.Ai): Fu[lila.ai.Ai] = 
-    load map(_.isDefined) map { _.fold(this, fallback) }
-
-  private def fetchMove(pgn: String, initialFen: String, level: Int): Fu[String] =
-    WS.url(playUrl).withQueryString(
-      "pgn" -> pgn,
-      "initialFen" -> initialFen,
-      "level" -> level.toString
-    ).get() map (_.body)
-
-  private def fetchAnalyse(pgn: String, initialFen: String): Fu[String] =
-    WS.url(analyseUrl).withQueryString(
-      "pgn" -> pgn,
-      "initialFen" -> initialFen
-    ).get() map (_.body)
-
-  private def fetchLoad: Fu[Option[Int]] =
-    WS.url(loadUrl).get() map (_.body) map parseIntOption
+  def load: Fu[List[Option[Int]]] =
+    dispatcher ? GetLoad mapTo manifest[List[Option[Int]]]
 }

modules/ai/src/main/stockfish/Monitor.scala

@@ -6,7 +6,7 @@ import scala.concurrent.duration._
 import akka.actor._
 import akka.pattern.{ ask, pipe }
 
-import actorApi.monitor._
+import actorApi._
 import lila.hub.actorApi.ai.GetLoad
 import makeTimeout.short
 

modules/ai/src/main/stockfish/Queue.scala

@@ -8,7 +8,7 @@ import akka.actor._
 import akka.pattern.{ ask, pipe }
 import akka.util.Timeout
 
-import actorApi._, monitor._
+import actorApi._
 import lila.analyse.{ AnalysisMaker, Info }
 import lila.hub.actorApi.ai.GetLoad
 

modules/ai/src/main/stockfish/actorApi.scala

@@ -4,10 +4,7 @@ package actorApi
 
 import akka.actor.ActorRef
 
-package monitor {
-  case class AddTime(time: Int)
-  case object CalculateLoad
-}
+case class AddTime(time: Int)
 
 sealed trait State
 case object Starting extends State

modules/ai/src/main/stockfish/remote/Connection.scala

@@ -0,0 +1,45 @@
+package lila.ai
+package stockfish
+package remote
+
+import scala.concurrent.duration._
+
+import akka.actor._
+import akka.pattern.{ ask, pipe }
+
+import actorApi._
+import lila.common.ws.WS
+import lila.hub.actorApi.ai.GetLoad
+
+private[ai] final class Connection(router: Router) extends Actor {
+
+  private var load = none[Int]
+
+  def receive = {
+
+    case GetLoad ⇒ sender ! load
+
+    case CalculateLoad ⇒ scala.concurrent.Future {
+      try {
+        load = WS.url(router.load).get() map (_.body) map parseIntOption await makeTimeout.seconds(1)
+      }
+      catch {
+        case e: Exception ⇒ {
+          // logwarn("[stockfish client calculate load] " + e.getMessage)
+          load = none
+        }
+      }
+    }
+    case Play(pgn, fen, level) ⇒ WS.url(router.play).withQueryString(
+      "pgn" -> pgn,
+      "initialFen" -> fen,
+      "level" -> level.toString
+    ).get() map (_.body) pipeTo sender
+
+    case Analyse(pgn: String, fen: String) ⇒ WS.url(router.analyse).withQueryString(
+      "pgn" -> pgn,
+      "initialFen" -> fen
+    ).get() map (_.body) pipeTo sender
+  }
+
+}

modules/ai/src/main/stockfish/remote/Dispatcher.scala

@@ -0,0 +1,55 @@
+package lila.ai
+package stockfish
+package remote
+
+import scala.concurrent.duration._
+
+import akka.actor._
+import akka.pattern.{ ask, pipe }
+
+import actorApi._
+import lila.hub.actorApi.ai.GetLoad
+
+private[ai] final class Dispatcher(
+    urls: List[String],
+    router: String ⇒ Router,
+    scheduler: Scheduler) extends Actor {
+
+  private lazy val connections: List[ActorRef] = urls map { url ⇒
+    context.actorOf(
+      Props(new Connection(router(url))),
+      name = urlToActorName(url)
+    ) ~ { actor ⇒
+        scheduler.schedule(200.millis, 1.second, actor, CalculateLoad)
+      }
+  }
+
+  def receive = {
+    case GetLoad    ⇒ loaded map { _ map { _._2 } } pipeTo sender
+    case x: Play    ⇒ forward(x)
+    case x: Analyse ⇒ forward(x)
+  }
+
+  private def forward(x: Any) = lessLoadedConnection.effectFold(
+    e ⇒ sender ! Status.Failure(e),
+    c ⇒ c forward x)
+
+  private def loaded: Fu[List[(ActorRef, Option[Int])]] = {
+    import makeTimeout.short
+    connections map { c ⇒
+      c ? GetLoad mapTo manifest[Option[Int]] map { l ⇒ List(c -> l) }
+    }
+  }.suml
+
+  private def lessLoadedConnection: Fu[ActorRef] = loaded map { xs ⇒
+    (xs collect {
+      case (a, Some(l)) ⇒ a -> l
+    }).sortBy(x ⇒ x._2).headOption.map(_._1)
+  } flatten s"[stockfish] No available remote found"
+
+  private val urlRegex = """^https?://([^\/]+)/.+$""".r
+  private def urlToActorName(url: String) = url match {
+    case urlRegex(domain) ⇒ domain
+    case _                ⇒ ornicar.scalalib.Random nextString 8
+  }
+}

modules/ai/src/main/stockfish/remote/Router.scala

@@ -0,0 +1,25 @@
+package lila.ai
+package stockfish
+package remote
+
+private[ai] final class Router(
+    url: String,
+    playRoute: String,
+    analyseRoute: String,
+    loadRoute: String) {
+
+  val play = make(playRoute)
+  val analyse = make(analyseRoute)
+  val load = make(loadRoute)
+
+  private def make(route: String) = route.replace("{remote}", url)
+}
+
+private[ai] object Router {
+
+  def apply(
+    playRoute: String,
+    analyseRoute: String,
+    loadRoute: String)(url: String): Router =
+    new Router(url, playRoute, analyseRoute, loadRoute)
+}

modules/ai/src/main/stockfish/remote/actorApi.scala

@@ -0,0 +1,8 @@
+package lila.ai.stockfish.remote.actorApi
+
+import lila.game.Game
+
+case object CalculateLoad
+
+case class Play(pgn: String, fen: String, level: Int) 
+case class Analyse(pgn: String, fen: String) 

modules/monitor/src/main/Reporting.scala

@@ -38,7 +38,7 @@ private[monitor] final class Reporting(
   var mps = 0
   var cpu = 0
   var mongoStatus = MongoStatus.default
-  var aiLoad = none[Int]
+  var aiLoads = List[Option[Int]]()
 
   var displays = 0
 
@@ -66,7 +66,7 @@ private[monitor] final class Reporting(
     case Update ⇒ {
       val before = nowMillis
       MongoStatus(db.db)(mongoStatus) zip
-        (hub.actor.ai ? lila.hub.actorApi.ai.GetLoad).mapTo[Option[Int]] zip
+        (hub.actor.ai ? lila.hub.actorApi.ai.GetLoad).mapTo[List[Option[Int]]] zip
         (hub.socket.site ? GetNbMembers).mapTo[Int] zip
         (hub.socket.lobby ? GetNbMembers).mapTo[Int] zip
         (hub.socket.round ? Size).mapTo[Int] zip
@@ -86,7 +86,7 @@ private[monitor] final class Reporting(
               rps = rpsProvider.rps
               mps = mpsProvider.rps
               cpu = ((cpuStats.getCpuUsage() * 1000).round / 10.0).toInt
-              aiLoad = aiL
+              aiLoads = aiL
               socket ! MonitorData(monitorData)
             }
           }
@@ -98,9 +98,11 @@ private[monitor] final class Reporting(
     nbGames,
     game.nbHubs,
     loadAvg.toString,
-    ~aiLoad
+    aiLoadString
   ) mkString " "
 
+  private def aiLoadString = aiLoads.map(_.fold("!!")(_.toString)) mkString ","
+
   private def monitorData = List(
     "users" -> allMembers,
     "lobby" -> lobby.nbMembers,
@@ -116,7 +118,7 @@ private[monitor] final class Reporting(
     "dbConn" -> mongoStatus.connection,
     "dbQps" -> mongoStatus.qps,
     "dbLock" -> math.round(mongoStatus.lock * 10) / 10d,
-    "ai" -> ~aiLoad
+    "ai" -> aiLoadString
   ) map {
       case (name, value) ⇒ value + ":" + name
     }

modules/round/src/main/Env.scala

@@ -16,7 +16,7 @@ final class Env(
     flood: lila.security.Flood,
     db: lila.db.Env,
     hub: lila.hub.Env,
-    ai: () ⇒ Fu[lila.ai.Ai],
+    ai: lila.ai.Ai,
     getUsername: String ⇒ Fu[Option[String]],
     getUsernameOrAnon: String ⇒ Fu[String],
     i18nKeys: lila.i18n.I18nKeys,
@@ -87,7 +87,7 @@ final class Env(
     timeline = hub.actor.gameTimeline)
 
   private lazy val player: Player = new Player(
-    ai = ai,
+    engine = ai,
     notifyMove = notifyMove,
     finisher = finisher,
     cheatDetector = cheatDetector,

modules/round/src/main/Player.scala

@@ -10,7 +10,7 @@ import lila.game.{ Game, GameRepo, PgnRepo, Pov, Progress }
 import lila.hub.actorApi.map.Tell
 
 private[round] final class Player(
-    ai: () ⇒ Fu[Ai],
+    engine: Ai,
     notifyMove: (String, String, Option[String]) ⇒ Unit,
     finisher: Finisher,
     cheatDetector: CheatDetector,
@@ -54,7 +54,7 @@ private[round] final class Player(
       (game.variant.exotic ?? { GameRepo initialFen game.id }) zip
         (PgnRepo get game.id) flatMap {
           case (fen, pgn) ⇒
-            ai() flatMap { _.play(game.toChess, pgn, fen, ~game.aiLevel) } flatMap {
+            engine.play(game.toChess, pgn, fen, ~game.aiLevel) flatMap {
               case (newChessGame, move) ⇒ {
                 val (progress, pgn2) = game.update(newChessGame, move)
                 (GameRepo save progress) >>

modules/setup/src/main/Env.scala

@@ -11,7 +11,7 @@ final class Env(
     db: lila.db.Env,
     hub: lila.hub.Env,
     messenger: lila.round.Messenger,
-    ai: () ⇒ Fu[lila.ai.Ai],
+    ai: lila.ai.Ai,
     system: ActorSystem) {
 
   private val FriendMemoTtl = config duration "friend.memo.ttl"
@@ -29,7 +29,7 @@ final class Env(
     friendConfigMemo = friendConfigMemo,
     timeline = hub.actor.gameTimeline,
     router = hub.actor.router,
-    getAi = ai)
+    engine = ai)
 
   lazy val friendJoiner = new FriendJoiner(
     messenger = messenger,

modules/setup/src/main/Processor.scala

@@ -6,7 +6,6 @@ import chess.{ Game ⇒ ChessGame, Board, Color ⇒ ChessColor }
 import makeTimeout.short
 import play.api.libs.json.{ Json, JsObject }
 
-import lila.ai.Ai
 import lila.db.api._
 import lila.game.tube.gameTube
 import lila.game.{ Game, GameRepo, PgnRepo, Pov }
@@ -22,7 +21,7 @@ private[setup] final class Processor(
     friendConfigMemo: FriendConfigMemo,
     timeline: ActorSelection,
     router: ActorSelection,
-    getAi: () ⇒ Fu[Ai]) {
+    engine: lila.ai.Ai) {
 
   def filter(config: FilterConfig)(implicit ctx: Context): Funit =
     saveConfig(_ withFilter config)
@@ -36,7 +35,7 @@ private[setup] final class Processor(
       game.player.isHuman.fold(fuccess(pov), for {
         initialFen ← game.variant.exotic ?? (GameRepo initialFen game.id)
         pgnString ← PgnRepo get game.id
-        aiResult ← getAi() flatMap { _.play(game.toChess, pgnString, initialFen, ~game.aiLevel) }
+        aiResult ← engine.play(game.toChess, pgnString, initialFen, ~game.aiLevel)
         (newChessGame, move) = aiResult
         (progress, pgn) = game.update(newChessGame, move)
         _ ← (GameRepo save progress) >> PgnRepo.save(game.id, pgn)