<commit_before>package spark

import scala.collection.mutable.HashMap

import scala.actors._
import scala.actors.Actor._
import scala.actors.remote._
import scala.collection.mutable.HashSet

sealed trait UpdatedProgressSharerMessage
case class SlaveLocation(slaveHost: String, slavePort: Int) extends UpdatedProgressSharerMessage
case class UpdatedProgressMessageToMaster(message: UpdatedProgressMasterMessage[_,_]) extends UpdatedProgressSharerMessage
case class UpdatedProgressDiffToSlave(diff: UpdatedProgressDiff[_,_]) extends UpdatedProgressSharerMessage
case object StopUpdatedProgressSharer extends UpdatedProgressSharerMessage

class UpdatedProgressMasterSenderActor(_serverUris: HashMap[String, Int]) extends DaemonActor with Logging {
    var diffsToSend = new HashMap[Long, UpdatedProgressDiff[_,_]]
    var serverUris = _serverUris

    def act() {
        val port = System.getProperty("spark.master.port").toInt
        RemoteActor.alive(port)
        RemoteActor.register('UpdatedProgressMasterSender, self)
        logInfo("Registered actor on port " + port)

        while(true) {
          diffsToSend.synchronized {
            for(diff <- diffsToSend.values) {
              // todo: jperla: this sets up and tears down a TCP connection, slow!
              // improve this by using keeping connections around, or using UDP or something
              // iterate over serveruris, and send messages
              serverUris.synchronized {
                for ((host,port) <- serverUris) {
                    var slave = RemoteActor.select(Node(host, port), 'UpdatedProgressSharerSlave)
                    slave ! UpdatedProgressDiffToSlave(diff)
                    slave = null
                }
              }
            }
            // sent all diffs, so clear
            diffsToSend.clear()<commit_msg>better synchronization<commit_after>package spark

import scala.collection.mutable.HashMap

import scala.actors._
import scala.actors.Actor._
import scala.actors.remote._
import scala.collection.mutable.HashSet

sealed trait UpdatedProgressSharerMessage
case class SlaveLocation(slaveHost: String, slavePort: Int) extends UpdatedProgressSharerMessage
case class UpdatedProgressMessageToMaster(message: UpdatedProgressMasterMessage[_,_]) extends UpdatedProgressSharerMessage
case class UpdatedProgressDiffToSlave(diff: UpdatedProgressDiff[_,_]) extends UpdatedProgressSharerMessage
case object StopUpdatedProgressSharer extends UpdatedProgressSharerMessage

class UpdatedProgressMasterSenderActor(_serverUris: HashMap[String, Int]) extends DaemonActor with Logging {
    var diffsToSend = new HashMap[Long, UpdatedProgressDiff[_,_]]
    var serverUris = _serverUris

    var diffsWillSend = null : HashMap[Long, UpdatedProgressDiff[_,_]]

    def act() {
        val port = System.getProperty("spark.master.port").toInt
        RemoteActor.alive(port)
        RemoteActor.register('UpdatedProgressMasterSender, self)
        logInfo("Registered actor on port " + port)

        while(true) {
          var diffsCopy = diffsToSend
          var serverUrisCopy = serverUris

          diffsToSend.synchronized {
            diffsCopy = diffsToSend.clone()
            // will send all diffs, so clear
            diffsToSend.clear()
          }

          serverUris.synchronized {
            serverUrisCopy = serverUris.clone()
          }

          for(diff <- diffsCopy.values) {
            // todo: jperla: this sets up and tears down a TCP connection, slow!
            // improve this by using keeping connections around, or using UDP or something
            // iterate over serveruris, and send messages
              for ((host,port) <- serverUrisCopy) {
                  var slave = RemoteActor.select(Node(host, port), 'UpdatedProgressSharerSlave)
                  slave ! UpdatedProgressDiffToSlave(diff)
                  slave = null
              }
{'fraction_non_alphanumeric': 0.07086007702182286, 'fraction_numerical': 0.0, 'mean_word_length': 3.0331262939958594, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 3, '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': '7482723', 'n_tokens_mistral': 1011, 'n_tokens_neox': 977, 'n_words': 260}