<commit_before>package mpservice

import akka.actor.{ Actor, Props, UnboundedStash, ActorLogging, Inbox, ActorRef }
import scala.concurrent.duration.DurationInt
import akka.routing.FromConfig
import akka.actor.ActorSystem
import java.math.BigInteger
import scala.collection._
import com.typesafe.config.ConfigFactory
import scala.util.Try
import scala.util.Success
import com.squareup.jnagmp.Gmp

// public api
case class ModPow(base: BigInteger, pow: BigInteger, mod: BigInteger)
trait ModPowService {
  def compute(work: Array[ModPow]): Array[BigInteger]
}

// implementation
case class Work(requestId: Int, workId: Int, work: Array[ModPow])
case class WorkReply(requestId: Int, workId: Int, result: Array[BigInteger])
case class ModPowArray(modpows: Array[ModPow])
case class ModPowArrayResult(result: Array[BigInteger])
case class RequestData(client: ActorRef, length: Int, results: mutable.ArrayBuffer[WorkReply], sent: Long = System.currentTimeMillis)

object SequentialModPowService extends ModPowService {
  def compute(work: Array[ModPow]): Array[BigInteger] = work.map(x => x.base.modPow(x.pow, x.mod))
}
object GmpParallelModPowService extends ModPowService {
  def compute(work: Array[ModPow]): Array[BigInteger] = {
    work.par.map(x => Gmp.modPowInsecure(x.base, x.pow, x.mod)).seq.toArray
  }
}
object ParallelModPowService extends ModPowService {
  def compute(work: Array[ModPow]): Array[BigInteger] = work.par.map(x => x.base.modPow(x.pow, x.mod)).seq.toArray
}

class AkkaModPowService(system: ActorSystem, modPowService: ActorRef) extends ModPowService {
  val inbox = Inbox.create(system)

  def compute(work: Array[ModPow]): Array[BigInteger] = {
    val before = System.currentTimeMillis
    inbox.send(modPowService, ModPowArray(work))
    Try(inbox.receive(1000.seconds)) match {
      case Success(ModPowArrayResult(answer)) => answer
      // FIXME
      case _ => throw new Exception()
    }
  }
}

object AkkaModPowService extends ModPowService {
  val config = ConfigFactory.load()
  val system = ActorSystem("ClusterSystem", config)
  val maxChunkSize = config.getInt("master.max-chunk-size")
  val sendDelay = config.getInt("master.send-delay-ms")
  val minChunks = config.getInt("master.min-chunk")
  val useGmp = config.getBoolean("worker.use-gmp")

  val serviceActor = system.actorOf(ModPowServiceActor.props(minChunks, maxChunkSize, sendDelay, useGmp), name = "ModPowService")
  val service = new AkkaModPowService(system, serviceActor)

  def compute(work: Array[ModPow]): Array[BigInteger] = service.compute(work)
  def shutdown = system.shutdown
}

object MPService extends ModPowService {
  val service = AkkaModPowService

  def compute(work: Array[ModPow]): Array[BigInteger] = service.compute(work)
  def shutdown = service.shutdown
}

class ModPowServiceActor(val minChunks: Int, val maxChunkSize: Int, val sendDelay: Int, val useGmp: Boolean) extends Actor with ActorLogging {
  
  val workerRouter = context.actorOf(WorkerActor.props(useGmp).withRouter(FromConfig()), name = "workerRouter")

  // actor state
  var requestId = 0
  val requests = mutable.Map[Int, RequestData]()

  def receive: Receive = {
    
    case ModPowArray(modpows) => {
      requestId = requestId + 1
      val size = math.min(math.max(modpows.length / minChunks, 1), maxChunkSize)
      val chunks = modpows.sliding(size, size).toArray
      requests.put(requestId, RequestData(sender, chunks.length, mutable.ArrayBuffer()))
      println(s"request with ${modpows.length} units, splitting into ${chunks.length} chunks")
      chunks.indices.foreach { i =>
        val work = Work(requestId, i, chunks(i))
        // println(s"Sending chunk $work")
        
        Thread sleep sendDelay
        workerRouter ! work
      }
    }
    
    case w: WorkReply => {
      val requestData = requests.get(w.requestId).get
      requestData.results += w
      val diff = System.currentTimeMillis - requestData.sent
      println(s"${w.requestId} $diff")
      if(requestData.results.length == requestData.length) {
        requests.remove(w.requestId)
        val sorted = requestData.results.sortWith(_.workId < _.workId)
        requestData.client ! ModPowArrayResult(sorted.flatMap(_.result).toArray)
      }
    }
  }
}

class WorkerActor(val useGmp: Boolean) extends Actor with ActorLogging {
  val service = if(useGmp) GmpParallelModPowService else ParallelModPowService

  def receive: Receive = {
    case Work(requestId, workId, modpows) => {
      // println(s"received request length ${modpows.length} at actor $this")
      val before = System.currentTimeMillis
      val result = service.compute(modpows).seq.toArray
      val diff = (System.currentTimeMillis - before)
      println(s"$requestId $diff")
      sender ! WorkReply(requestId, workId, result)
    }<commit_msg>timings<commit_after>package mpservice

import akka.actor.{ Actor, Props, UnboundedStash, ActorLogging, Inbox, ActorRef }
import scala.concurrent.duration.DurationInt
import akka.routing.FromConfig
import akka.actor.ActorSystem
import java.math.BigInteger
import scala.collection._
import com.typesafe.config.ConfigFactory
import scala.util.Try
import scala.util.Success
import com.squareup.jnagmp.Gmp

// public api
case class ModPow(base: BigInteger, pow: BigInteger, mod: BigInteger)
trait ModPowService {
  def compute(work: Array[ModPow]): Array[BigInteger]
}

// implementation
case class Work(requestId: Int, workId: Int, work: Array[ModPow])
case class WorkReply(requestId: Int, workId: Int, result: Array[BigInteger])
case class ModPowArray(modpows: Array[ModPow])
case class ModPowArrayResult(result: Array[BigInteger])
case class RequestData(client: ActorRef, length: Int, results: mutable.ArrayBuffer[WorkReply], sent: Long = System.currentTimeMillis)

object SequentialModPowService extends ModPowService {
  def compute(work: Array[ModPow]): Array[BigInteger] = work.map(x => x.base.modPow(x.pow, x.mod))
}
object GmpParallelModPowService extends ModPowService {
  def compute(work: Array[ModPow]): Array[BigInteger] = {
    work.par.map(x => Gmp.modPowInsecure(x.base, x.pow, x.mod)).seq.toArray
  }
}
object ParallelModPowService extends ModPowService {
  def compute(work: Array[ModPow]): Array[BigInteger] = work.par.map(x => x.base.modPow(x.pow, x.mod)).seq.toArray
}

class AkkaModPowService(system: ActorSystem, modPowService: ActorRef) extends ModPowService {
  val inbox = Inbox.create(system)

  def compute(work: Array[ModPow]): Array[BigInteger] = {
    val before = System.currentTimeMillis
    inbox.send(modPowService, ModPowArray(work))
    Try(inbox.receive(1000.seconds)) match {
      case Success(ModPowArrayResult(answer)) => answer
      // FIXME
      case _ => throw new Exception()
    }
  }
}

object AkkaModPowService extends ModPowService {
  val config = ConfigFactory.load()
  val system = ActorSystem("ClusterSystem", config)
  val maxChunkSize = config.getInt("master.max-chunk-size")
  val sendDelay = config.getInt("master.send-delay-ms")
  val minChunks = config.getInt("master.min-chunk")
  val useGmp = config.getBoolean("worker.use-gmp")

  val serviceActor = system.actorOf(ModPowServiceActor.props(minChunks, maxChunkSize, sendDelay, useGmp), name = "ModPowService")
  val service = new AkkaModPowService(system, serviceActor)

  def compute(work: Array[ModPow]): Array[BigInteger] = service.compute(work)
  def shutdown = system.shutdown
}

object MPService extends ModPowService {
  val service = AkkaModPowService

  def compute(work: Array[ModPow]): Array[BigInteger] = service.compute(work)
  def shutdown = service.shutdown
}

class ModPowServiceActor(val minChunks: Int, val maxChunkSize: Int, val sendDelay: Int, val useGmp: Boolean) extends Actor with ActorLogging {
  
  val workerRouter = context.actorOf(WorkerActor.props(useGmp).withRouter(FromConfig()), name = "workerRouter")

  // actor state
  var requestId = 0
  val requests = mutable.Map[Int, RequestData]()

  def receive: Receive = {
    
    case ModPowArray(modpows) => {
      requestId = requestId + 1
      val size = math.min(math.max(modpows.length / minChunks, 1), maxChunkSize)
      val chunks = modpows.sliding(size, size).toArray
      requests.put(requestId, RequestData(sender, chunks.length, mutable.ArrayBuffer()))
      println(s"request with ${modpows.length} units, splitting into ${chunks.length} chunks")
      chunks.indices.foreach { i =>
        val work = Work(requestId, i, chunks(i))
        // println(s"Sending chunk $work")
        
        Thread sleep sendDelay
        workerRouter ! work
      }
    }
    
    case w: WorkReply => {
      val requestData = requests.get(w.requestId).get
      requestData.results += w
      val diff = System.currentTimeMillis - requestData.sent
      println(s"${w.workId} $diff")
      if(requestData.results.length == requestData.length) {
        requests.remove(w.requestId)
        val sorted = requestData.results.sortWith(_.workId < _.workId)
        requestData.client ! ModPowArrayResult(sorted.flatMap(_.result).toArray)
      }
    }
  }
}

class WorkerActor(val useGmp: Boolean) extends Actor with ActorLogging {
  val service = if(useGmp) GmpParallelModPowService else ParallelModPowService

  def receive: Receive = {
    case Work(requestId, workId, modpows) => {
      // println(s"received request length ${modpows.length} at actor $this")
      val before = System.currentTimeMillis
      val result = service.compute(modpows).seq.toArray
      val diff = (System.currentTimeMillis - before)
      println(s"$workId $diff")
      sender ! WorkReply(requestId, workId, result)
    }
{'fraction_non_alphanumeric': 0.11270833333333333, 'fraction_numerical': 0.0014583333333333334, 'mean_word_length': 4.319113573407202, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 21, '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': '1865047', 'n_tokens_mistral': 3112, 'n_tokens_neox': 2978, 'n_words': 690}