<commit_before>/**
 * @author hamiltont
 *
 */
package clasp.core

import java.util.UUID
import scala.collection.immutable.Map
import scala.collection.mutable.ListBuffer
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Promise
import scala.concurrent.duration.DurationInt
import scala.concurrent.future
import scala.language.postfixOps
import scala.sys.process.Process
import scala.sys.process.ProcessLogger
import scala.sys.process.stringToProcess
import scala.util.Failure
import scala.util.Success
import scala.util.Try
import org.slf4j.LoggerFactory
import EmulatorActor._
import EmulatorLogger._
import EmulatorManager._
import akka.actor.Actor
import akka.actor.ActorIdentity
import akka.actor.ActorLogging
import akka.actor.ActorRef
import akka.actor.Cancellable
import akka.actor.Identify
import akka.actor.Props
import akka.actor.actorRef2Scala
import akka.pattern.ask
import clasp.ClaspConf
import clasp.Emulator
import clasp.core.remoting.ChannelServer
import clasp.core.remoting.ClaspJson.Ack
import clasp.core.remoting.WebSocketChannelManager._
import clasp.core.sdktools.EmulatorOptions
import clasp.core.sdktools.avd
import clasp.core.sdktools.sdk
import clasp.utils.ActorLifecycleLogging
import clasp.utils.ActorLifecycleLogging
import clasp.utils.ActorStack
import clasp.utils.ActorStack
import clasp.utils.Slf4jLoggingStack
import clasp.utils.Slf4jLoggingStack
import clasp.utils.Slf4jLoggingStack
import spray.json._
import clasp.core.remoting.ClaspJson._ 

object EmulatorManager {
  case class EmulatorReady(emu: EmulatorDescription, bootTime: Long)
  case class EmulatorFailedBoot(actor: ActorRef)
  case class EmulatorCrashed(emu: EmulatorDescription)

  // Used to send commands to manager
  case class ListEmulators()
  case class LaunchEmulator()
  case class GetEmulatorOptions(uuid: String)

  // TODO move to task manager
  case class QueueEmulatorTask(function: Emulator => Map[String, Serializable], promise: Promise[Map[String, Serializable]])
  case class TaskSuccess(taskId: String, data: Map[String, Serializable], emulator: EmulatorDescription)
  case class TaskFailure(taskId: String, reason: Throwable, emulator: EmulatorDescription)
}
class EmulatorManager(val nodeManager: ActorRef, val conf: ClaspConf)
  extends Actor
  with ActorLifecycleLogging
  with ActorStack
  with Slf4jLoggingStack {
  lazy val log = LoggerFactory.getLogger(getClass())
  import log.{ error, debug, info, trace }

  val emulators = ListBuffer[EmulatorDescription]()

  // Tasks that have been delivered but not fulfilled
  val outstandingTasks: scala.collection.mutable.Map[String, Promise[Map[String, Serializable]]] = scala.collection.mutable.Map()

  // Contains tasks that need to be delivered to workers
  val undeliveredTasks: scala.collection.mutable.Queue[EmulatorTask] = scala.collection.mutable.Queue()

  def sendTask(to: EmulatorDescription) = {
    if (undeliveredTasks.length != 0) {
      info(s"Dequeuing task for $to")
      // Only works because the remote system has the same classpath loaded, 
      // so all anonymous functions exist on both systems. If we want this to 
      // work in a dynamic manner we would have to serialize the class that's 
      // associated with the anonymous function, send that
      // across the wire & load it.
      // See http://doc.akka.io/docs/akka/snapshot/scala/serialization.html#serialization-scala
      // See http://www.scala-lang.org/node/10566
      // Alternatively, just copy all *.class files
      to.actor ! undeliveredTasks.dequeue
    } else
      error("Emulator came online, but no work was available.")
  }

  def wrappedReceive = {
    case EmulatorReady(emulator) => {
      emulators += emulator
      info(s"Emulator ready: ${emulator}")
      info(s"${emulators.length} emulators awake")
      sendTask(emulator)
    }
    case GetEmulatorOptions(uuid) => {
      val matched = emulators.filter(description => description.uuid.equals(uuid))
      if (matched.isEmpty) { // TODO

      } else
        matched.head.actor ! GetOptions(sender)
    }
    case EmulatorFailedBoot(actor) => {
      info(s"Emulator failed to boot: $actor")
    }
    case EmulatorCrashed(emulator) => {
      info(s"Emulator crashed: $emulator")
      emulators -= emulator
    }
    case _: ListEmulators => sender ! emulators.toList
    case QueueEmulatorTask(task, promise) => {
      // Note that this is not strictly threadsafe.<commit_msg>Create cheap "logging" inside emulator manager by registering channel :-P<commit_after>/**
 * @author hamiltont
 *
 */
package clasp.core

import java.util.UUID
import scala.collection.immutable.Map
import scala.collection.mutable.ListBuffer
import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.Promise
import scala.concurrent.duration.DurationInt
import scala.concurrent.future
import scala.language.postfixOps
import scala.sys.process.Process
import scala.sys.process.ProcessLogger
import scala.sys.process.stringToProcess
import scala.util.Failure
import scala.util.Success
import scala.util.Try
import org.slf4j.LoggerFactory
import EmulatorActor._
import EmulatorLogger._
import EmulatorManager._
import akka.actor.Actor
import akka.actor.ActorIdentity
import akka.actor.ActorLogging
import akka.actor.ActorRef
import akka.actor.Cancellable
import akka.actor.Identify
import akka.actor.Props
import akka.actor.actorRef2Scala
import akka.pattern.ask
import clasp.ClaspConf
import clasp.Emulator
import clasp.core.remoting.ChannelServer
import clasp.core.remoting.ClaspJson.Ack
import clasp.core.remoting.WebSocketChannelManager._
import clasp.core.sdktools.EmulatorOptions
import clasp.core.sdktools.avd
import clasp.core.sdktools.sdk
import clasp.utils.ActorLifecycleLogging
import clasp.utils.ActorLifecycleLogging
import clasp.utils.ActorStack
import clasp.utils.ActorStack
import clasp.utils.Slf4jLoggingStack
import clasp.utils.Slf4jLoggingStack
import clasp.utils.Slf4jLoggingStack
import spray.json._
import clasp.core.remoting.ClaspJson._ 

object EmulatorManager {
  case class EmulatorReady(emu: EmulatorDescription, bootTime: Long)
  case class EmulatorFailedBoot(actor: ActorRef)
  case class EmulatorCrashed(emu: EmulatorDescription)

  // Used to send commands to manager
  case class ListEmulators()
  case class LaunchEmulator()
  case class GetEmulatorOptions(uuid: String)

  // TODO move to task manager
  case class QueueEmulatorTask(function: Emulator => Map[String, Serializable], promise: Promise[Map[String, Serializable]])
  case class TaskSuccess(taskId: String, data: Map[String, Serializable], emulator: EmulatorDescription)
  case class TaskFailure(taskId: String, reason: Throwable, emulator: EmulatorDescription)
}
class EmulatorManager(val nodeManager: ActorRef, val conf: ClaspConf)
  extends Actor
  with ActorLifecycleLogging
  with ActorStack
  with Slf4jLoggingStack 
  with ChannelServer
   {
  lazy val log = LoggerFactory.getLogger(getClass())
  import log.{ error, debug, info, trace }

  val emulators = ListBuffer[EmulatorDescription]()
  
  val chanName = "/emulatormanager"
  val channelManager = Some(context.actorFor(s"akka.tcp://clasp@${conf.ip()}:2552/user/channelManager"))
  channelRegister(chanName)(self, channelManager)

  // Tasks that have been delivered but not fulfilled
  val outstandingTasks: scala.collection.mutable.Map[String, Promise[Map[String, Serializable]]] = scala.collection.mutable.Map()

  // Contains tasks that need to be delivered to workers
  val undeliveredTasks: scala.collection.mutable.Queue[EmulatorTask] = scala.collection.mutable.Queue()

  def sendTask(to: EmulatorDescription) = {
    if (undeliveredTasks.length != 0) {
      info(s"Dequeuing task for $to")
      // Only works because the remote system has the same classpath loaded, 
      // so all anonymous functions exist on both systems. If we want this to 
      // work in a dynamic manner we would have to serialize the class that's 
      // associated with the anonymous function, send that
      // across the wire & load it.
      // See http://doc.akka.io/docs/akka/snapshot/scala/serialization.html#serialization-scala
      // See http://www.scala-lang.org/node/10566
      // Alternatively, just copy all *.class files
      to.actor ! undeliveredTasks.dequeue
    } else
      error("Emulator came online, but no work was available.")
  }

  def wrappedReceive = {
    case EmulatorReady(emulator, time) => {
      // Create a record of emulator boot time
      val o = JsObject("emulators" -> JsNumber(emulators.length), "boottime" -> JsNumber(time))
      channelManager.get ! Message(chanName, createMessageString(o), self)

      emulators += emulator
      info(s"Emulator ready: ${emulator}")
      info(s"${emulators.length} emulators awake")
      // sendTask(emulator)
      
    }
    case GetEmulatorOptions(uuid) => {
      val matched = emulators.filter(description => description.uuid.equals(uuid))
      if (matched.isEmpty) { // TODO

      } else
        matched.head.actor ! GetOptions(sender)
    }
    case EmulatorFailedBoot(actor) => {
      info(s"Emulator failed to boot: $actor")
    }
    case EmulatorCrashed(emulator) => {
      info(s"Emulator crashed: $emulator")
      emulators -= emulator
    }
    case _: ListEmulators => sender ! emulators.toList
    case QueueEmulatorTask(task, promise) => {
      // Note that this is not strictly threadsafe.
{'fraction_non_alphanumeric': 0.0907258064516129, 'fraction_numerical': 0.0029711375212224107, 'mean_word_length': 4.289001122334455, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '93489', 'n_tokens_mistral': 2842, 'n_tokens_neox': 2739, 'n_words': 724}