<commit_before>package lila.fishnet

import scala.concurrent.duration._

final private class Monitor(
    repo: FishnetRepo,
    cacheApi: lila.memo.CacheApi
)(implicit
    ec: scala.concurrent.ExecutionContext,
    system: akka.actor.ActorSystem
) {

  val statusCache = cacheApi.unit[Monitor.Status] {
    _.refreshAfterWrite(1 minute)
      .buildAsyncFuture { _ =>
        repo.status.compute
      }
  }

  private val monBy = lila.mon.fishnet.analysis.by

  private def sumOf[A](items: List[A])(f: A => Option[Int]) =
    items.foldLeft(0) {
      case (acc, a) => acc + f(a).getOrElse(0)
    }

  private[fishnet] def analysis(
      work: Work.Analysis,
      client: Client,
      result: JsonApi.Request.CompleteAnalysis
  ) = {
    Monitor.success(work, client)

    val threads = result.stockfish.options.threadsInt
    val userId  = client.userId.value

    result.stockfish.options.hashInt foreach { monBy.hash(userId).update(_) }
    result.stockfish.options.threadsInt foreach { monBy.threads(userId).update(_) }

    monBy.totalSecond(userId).increment(sumOf(result.evaluations)(_.time) * threads.|(1) / 1000)
    monBy
      .totalMeganode(userId)
      .increment(sumOf(result.evaluations) { eval =>
        eval.nodes ifFalse eval.mateFound
      } / 1000000)

    val metaMovesSample = sample(result.evaluations.drop(6).filterNot(_.mateFound), 100)
    def avgOf(f: JsonApi.Request.Evaluation => Option[Int]): Option[Int] = {
      val (sum, nb) = metaMovesSample.foldLeft(0 -> 0) {
        case ((sum, nb), move) =>
          f(move).fold(sum -> nb) { v =>
            (sum + v, nb + 1)
          }
      }
      (nb > 0) option (sum / nb)
    }
    avgOf(_.time) foreach { monBy.movetime(userId).record(_) }
    avgOf(_.nodes) foreach { monBy.node(userId).record(_) }
    avgOf(_.cappedNps) foreach { monBy.nps(userId).record(_) }
    avgOf(_.depth) foreach { monBy.depth(userId).record(_) }
    avgOf(_.pv.size.some) foreach { monBy.pvSize(userId).record(_) }

    val significantPvSizes =
      result.evaluations.filterNot(_.mateFound).filterNot(_.deadDraw).map(_.pv.size)

    monBy.pv(userId, isLong = false).increment(significantPvSizes.count(_ < 3))
    monBy.pv(userId, isLong = true).increment(significantPvSizes.count(_ >= 6))
  }

  private def sample[A](elems: List[A], n: Int) =
    if (elems.sizeIs <= n) elems else lila.common.ThreadLocalRandom shuffle elems take n

  private def monitorClients(): Funit =
    repo.allRecentClients map { clients =>
      import lila.mon.fishnet.client._

      status(true).update(clients.count(_.enabled))
      status(false).update(clients.count(_.disabled))

      val instances = clients.flatMap(_.instance)

      instances.map(_.version.value).groupBy(identity).view.mapValues(_.size) foreach {
        case (v, nb) => version(v).update(nb)
      }
      instances.map(_.engines.stockfish.name).groupBy(identity).view.mapValues(_.size) foreach {
        case (s, nb) => stockfish(s).update(nb)
      }
      instances.map(_.python.value).groupBy(identity).view.mapValues(_.size) foreach {
        case (s, nb) => python(s).update(nb)
      }
    }

  private def monitorStatus(): Funit =
    statusCache.get {} map { c =>
      lila.mon.fishnet.work("queued", "system").update(c.system.queued)
      lila.mon.fishnet.work("queued", "user").update(c.user.queued)
      lila.mon.fishnet.work("acquired", "system").update(c.system.acquired)
      lila.mon.fishnet.work("acquired", "user").update(c.user.acquired)
      lila.mon.fishnet.oldest("system").update(c.system.oldest)
      lila.mon.fishnet.oldest("user").update(c.user.oldest)
    }

  system.scheduler.scheduleWithFixedDelay(1 minute, 1 minute) { () =><commit_msg>codegolf: using groupMapReduce<commit_after>package lila.fishnet

import scala.concurrent.duration._

final private class Monitor(
    repo: FishnetRepo,
    cacheApi: lila.memo.CacheApi
)(implicit
    ec: scala.concurrent.ExecutionContext,
    system: akka.actor.ActorSystem
) {

  val statusCache = cacheApi.unit[Monitor.Status] {
    _.refreshAfterWrite(1 minute)
      .buildAsyncFuture { _ =>
        repo.status.compute
      }
  }

  private val monBy = lila.mon.fishnet.analysis.by

  private def sumOf[A](items: List[A])(f: A => Option[Int]) =
    items.foldLeft(0) {
      case (acc, a) => acc + f(a).getOrElse(0)
    }

  private[fishnet] def analysis(
      work: Work.Analysis,
      client: Client,
      result: JsonApi.Request.CompleteAnalysis
  ) = {
    Monitor.success(work, client)

    val threads = result.stockfish.options.threadsInt
    val userId  = client.userId.value

    result.stockfish.options.hashInt foreach { monBy.hash(userId).update(_) }
    result.stockfish.options.threadsInt foreach { monBy.threads(userId).update(_) }

    monBy.totalSecond(userId).increment(sumOf(result.evaluations)(_.time) * threads.|(1) / 1000)
    monBy
      .totalMeganode(userId)
      .increment(sumOf(result.evaluations) { eval =>
        eval.nodes ifFalse eval.mateFound
      } / 1000000)

    val metaMovesSample = sample(result.evaluations.drop(6).filterNot(_.mateFound), 100)
    def avgOf(f: JsonApi.Request.Evaluation => Option[Int]): Option[Int] = {
      val (sum, nb) = metaMovesSample.foldLeft(0 -> 0) {
        case ((sum, nb), move) =>
          f(move).fold(sum -> nb) { v =>
            (sum + v, nb + 1)
          }
      }
      (nb > 0) option (sum / nb)
    }
    avgOf(_.time) foreach { monBy.movetime(userId).record(_) }
    avgOf(_.nodes) foreach { monBy.node(userId).record(_) }
    avgOf(_.cappedNps) foreach { monBy.nps(userId).record(_) }
    avgOf(_.depth) foreach { monBy.depth(userId).record(_) }
    avgOf(_.pv.size.some) foreach { monBy.pvSize(userId).record(_) }

    val significantPvSizes =
      result.evaluations.filterNot(_.mateFound).filterNot(_.deadDraw).map(_.pv.size)

    monBy.pv(userId, isLong = false).increment(significantPvSizes.count(_ < 3))
    monBy.pv(userId, isLong = true).increment(significantPvSizes.count(_ >= 6))
  }

  private def sample[A](elems: List[A], n: Int) =
    if (elems.sizeIs <= n) elems else lila.common.ThreadLocalRandom shuffle elems take n

  private def monitorClients(): Funit =
    repo.allRecentClients map { clients =>
      import lila.mon.fishnet.client._

      status(true).update(clients.count(_.enabled))
      status(false).update(clients.count(_.disabled))

      val instances = clients.flatMap(_.instance)

      instances.groupMapReduce(_.version.value)(_ => 1)(_ + _) foreach {
        case (v, nb) => version(v).update(nb)
      }
      instances.groupMapReduce(_.engines.stockfish.name)(_ => 1)(_ + _) foreach {
        case (s, nb) => stockfish(s).update(nb)
      }
      instances.groupMapReduce(_.python.value)(_ => 1)(_ + _) foreach {
        case (s, nb) => python(s).update(nb)
      }
    }

  private def monitorStatus(): Funit =
    statusCache.get {} map { c =>
      lila.mon.fishnet.work("queued", "system").update(c.system.queued)
      lila.mon.fishnet.work("queued", "user").update(c.user.queued)
      lila.mon.fishnet.work("acquired", "system").update(c.system.acquired)
      lila.mon.fishnet.work("acquired", "user").update(c.user.acquired)
      lila.mon.fishnet.oldest("system").update(c.system.oldest)
      lila.mon.fishnet.oldest("user").update(c.user.oldest)
    }

  system.scheduler.scheduleWithFixedDelay(1 minute, 1 minute) { () =>
{'fraction_non_alphanumeric': 0.15107133170599404, 'fraction_numerical': 0.007729861676159479, 'mean_word_length': 3.192723138146674, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 40, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5922000', 'n_tokens_mistral': 2594, 'n_tokens_neox': 2518, 'n_words': 403}