<commit_before>/*
 *  @author <NAME>
 *  
 *  Copyright 2011 University of Zurich
 *      
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *         http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *  
 */

package com.signalcollect.implementations.coordinator

import com.signalcollect.interfaces._
import com.signalcollect.configuration._
import com.signalcollect.implementations.messaging._
import com.signalcollect.implementations.graph.DefaultGraphApi
import com.signalcollect.api.factory._

import java.lang.reflect.Method
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong

import scala.collection.parallel.mutable.ParArray
import scala.collection.JavaConversions._

class WorkerApi(config: Configuration, logger: MessageRecipient[LogMessage]) extends MessageRecipient[Any] with DefaultGraphApi with Logging {

  override def toString = "WorkerApi"

  // initialize workers array
  protected val workers = new Array[Any](config.numberOfWorkers)

  /**
   * Individual worker creation
   * It creates workers based on configuration provided, whether local or distributed
   * Worker still needs to be initialized before using it
   *
   * @param: workerId is the id of the worker
   * @return: the worker should be cast to the desired worker (Local or Akka)
   */
  def createWorker(workerId: Int, workerConfiguration: WorkerConfiguration): Any = {

    val workerFactory = config.workerConfiguration.workerFactory

    // create the worker
    val worker = workerFactory.createInstance(workerId, workerConfiguration, config.numberOfWorkers, this, mapper)

    // put it to the array of workers
    workers(workerId) = worker

    worker

  }

  protected lazy val workerProxies: Array[Worker] = createWorkerProxies
  protected lazy val workerProxyMessageBuses: Array[MessageBus[Any]] = createWorkerProxyMessageBuses
  protected lazy val parallelWorkerProxies = workerProxies.par
  protected lazy val mapper = new DefaultVertexToWorkerMapper(config.numberOfWorkers)
  protected lazy val messageBus = config.workerConfiguration.messageBusFactory.createInstance(config.numberOfWorkers, mapper)
  protected lazy val workerStatusMap = new ConcurrentHashMap[Int, WorkerStatus]()
  protected val messagesReceived = new AtomicLong(0l)
  protected val statusMonitor = new Object

  var signalSteps = 0l
  var collectSteps = 0l

  protected def createWorkerProxyMessageBuses: Array[MessageBus[Any]] = {
    val workerProxyMessageBuses = new Array[MessageBus[Any]](config.numberOfWorkers)
    for (workerId <- 0 until config.numberOfWorkers) {
      val proxyMessageBus = config.workerConfiguration.messageBusFactory.createInstance(config.numberOfWorkers, mapper)
      proxyMessageBus.registerCoordinator(this)
      workerProxyMessageBuses(workerId) = proxyMessageBus
    }
    workerProxyMessageBuses
  }

  protected def createWorkerProxies: Array[Worker] = {
    val workerProxies = new Array[Worker](config.numberOfWorkers)
    for (workerId <- 0 until config.numberOfWorkers) {
      val workerProxy = WorkerProxy.create(workerId, workerProxyMessageBuses(workerId))
      workerProxies(workerId) = workerProxy
    }
    workerProxies
  }

  var isInitialized = false

  def initialize {
    if (!isInitialized) {
      Thread.currentThread.setName("Coordinator")
      messageBus.registerCoordinator(this)
      workerProxyMessageBuses foreach (_.registerCoordinator(this))
      for (workerId <- 0 until config.numberOfWorkers) {
        messageBus.registerWorker(workerId, workers(workerId))
        workerProxyMessageBuses foreach (_.registerWorker(workerId, workers(workerId)))
      }
      for (workerId <- 0 until config.numberOfWorkers) {
        parallelWorkerProxies foreach (_.registerWorker(workerId, workers(workerId)))
      }
      isInitialized = true
    }
  }

  def getWorkerStatistics: List[WorkerStatistics] = {
    parallelWorkerProxies.map(_.getWorkerStatistics).toList
  }

  def receive(message: Any) {
    this.synchronized {
      if (!message.isInstanceOf[LogMessage]) {
        messagesReceived.incrementAndGet
      }
      message match {
        case r: WorkerReply =>
          workerProxies(r.workerId).receive(message)
        case ws: WorkerStatus =>
          statusMonitor.synchronized {
            if (!workerStatusMap.contains(ws.workerId) || workerStatusMap.get(ws.workerId).messagesSent < ws.messagesSent) {
              workerStatusMap.put(ws.workerId, ws)
              statusMonitor.notifyAll
            }

          }
        case l: LogMessage =>
          logger.receive(l)
        case other => println("Received unknown message: " + other)
      }
    }
  }

  def messagesSentByWorkers = workerStatusMap.values.foldLeft(0l)(_ + _.messagesSent) + workerStatusMap.size // the status message that was sent was not yet counted by the worker
  def messagesSentByWorkerProxies = workerProxyMessageBuses.foldLeft(0l)(_ + _.messagesSent)
  def messagesSentByCoordinator = messageBus.messagesSent

  def messagesReceivedByWorkers = workerStatusMap.values.foldLeft(0l)(_ + _.messagesReceived)
  def messagesReceivedByCoordinator = messagesReceived.get

  def totalMessagesSent: Long = messagesSentByWorkers + messagesSentByWorkerProxies + messagesSentByCoordinator
  def totalMessagesReceived: Long = messagesReceivedByWorkers + messagesReceivedByCoordinator

  def idle: Boolean = workerStatusMap.values.forall(_.isIdle) && totalMessagesSent == totalMessagesReceived

  def awaitIdle: Long = {
    val startTime = System.nanoTime
    statusMonitor.synchronized {
      while (!idle) {
        statusMonitor.wait(10)
      }
    }
    val stopTime = System.nanoTime
    stopTime - startTime
  }

  def paused: Boolean = workerStatusMap.values.forall(_.isPaused)

  def awaitPaused {
    statusMonitor.synchronized {
      while (!paused) {
        statusMonitor.wait(10)
      }
    }
  }

  def reachedMinInboxSize: Boolean = {
    workerStatusMap.isEmpty || workerStatusMap.values.forall(!_.isOverstrained)
  }

  def awaitMessageProcessing {
    statusMonitor.synchronized {
      while (!reachedMinInboxSize) {
        statusMonitor.wait(10)
      }
    }
  }

  override def addEdge(edge: Edge[_, _]) {
    super.addEdge(edge)
    if(config.workerConfiguration.messageInboxLimits.isDefined) {
    	awaitMessageProcessing      
    }
  }

  override def addVertex(vertex: Vertex[_, _]) {
    super.addVertex(vertex)
    if(config.workerConfiguration.messageInboxLimits.isDefined) {
    	awaitMessageProcessing      
    }
  }

  override def sendSignalToVertex(signal: Any, targetId: Any, sourceId: Any = EXTERNAL) {
    super.sendSignalToVertex(signal, targetId, sourceId)
    if(config.workerConfiguration.messageInboxLimits.isDefined) {
    	awaitMessageProcessing      
    }
  }

  def started: Boolean = workerStatusMap.values.forall(!_.isPaused)

  def awaitStarted {
    statusMonitor.synchronized {
      while (!started) {
        statusMonitor.wait(10)
      }
    }
  }

  def signalStep = {
    signalSteps += 1
    parallelWorkerProxies foreach (_.signalStep)
  }

  def collectStep: Boolean = {
    collectSteps += 1
    parallelWorkerProxies.map(_.collectStep).fold(true)(_ && _)
  }

  def startComputation {
    parallelWorkerProxies.foreach(_.startComputation)
    awaitStarted
  }<commit_msg>Fixes the check for most recent worker state<commit_after>/*
 *  @author <NAME>
 *  
 *  Copyright 2011 University of Zurich
 *      
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *         http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 *  
 */

package com.signalcollect.implementations.coordinator

import com.signalcollect.interfaces._
import com.signalcollect.configuration._
import com.signalcollect.implementations.messaging._
import com.signalcollect.implementations.graph.DefaultGraphApi
import com.signalcollect.api.factory._

import java.lang.reflect.Method
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.atomic.AtomicLong

import scala.collection.parallel.mutable.ParArray
import scala.collection.JavaConversions._

class WorkerApi(config: Configuration, logger: MessageRecipient[LogMessage]) extends MessageRecipient[Any] with DefaultGraphApi with Logging {

  override def toString = "WorkerApi"

  // initialize workers array
  protected val workers = new Array[Any](config.numberOfWorkers)

  /**
   * Individual worker creation
   * It creates workers based on configuration provided, whether local or distributed
   * Worker still needs to be initialized before using it
   *
   * @param: workerId is the id of the worker
   * @return: the worker should be cast to the desired worker (Local or Akka)
   */
  def createWorker(workerId: Int, workerConfiguration: WorkerConfiguration): Any = {

    val workerFactory = config.workerConfiguration.workerFactory

    // create the worker
    val worker = workerFactory.createInstance(workerId, workerConfiguration, config.numberOfWorkers, this, mapper)

    // put it to the array of workers
    workers(workerId) = worker

    worker

  }

  protected lazy val workerProxies: Array[Worker] = createWorkerProxies
  protected lazy val workerProxyMessageBuses: Array[MessageBus[Any]] = createWorkerProxyMessageBuses
  protected lazy val parallelWorkerProxies = workerProxies.par
  protected lazy val mapper = new DefaultVertexToWorkerMapper(config.numberOfWorkers)
  protected lazy val messageBus = config.workerConfiguration.messageBusFactory.createInstance(config.numberOfWorkers, mapper)
  protected lazy val workerStatusMap = new ConcurrentHashMap[Int, WorkerStatus]()
  protected val messagesReceived = new AtomicLong(0l)
  protected val statusMonitor = new Object

  var signalSteps = 0l
  var collectSteps = 0l

  protected def createWorkerProxyMessageBuses: Array[MessageBus[Any]] = {
    val workerProxyMessageBuses = new Array[MessageBus[Any]](config.numberOfWorkers)
    for (workerId <- 0 until config.numberOfWorkers) {
      val proxyMessageBus = config.workerConfiguration.messageBusFactory.createInstance(config.numberOfWorkers, mapper)
      proxyMessageBus.registerCoordinator(this)
      workerProxyMessageBuses(workerId) = proxyMessageBus
    }
    workerProxyMessageBuses
  }

  protected def createWorkerProxies: Array[Worker] = {
    val workerProxies = new Array[Worker](config.numberOfWorkers)
    for (workerId <- 0 until config.numberOfWorkers) {
      val workerProxy = WorkerProxy.create(workerId, workerProxyMessageBuses(workerId))
      workerProxies(workerId) = workerProxy
    }
    workerProxies
  }

  var isInitialized = false

  def initialize {
    if (!isInitialized) {
      Thread.currentThread.setName("Coordinator")
      messageBus.registerCoordinator(this)
      workerProxyMessageBuses foreach (_.registerCoordinator(this))
      for (workerId <- 0 until config.numberOfWorkers) {
        messageBus.registerWorker(workerId, workers(workerId))
        workerProxyMessageBuses foreach (_.registerWorker(workerId, workers(workerId)))
      }
      for (workerId <- 0 until config.numberOfWorkers) {
        parallelWorkerProxies foreach (_.registerWorker(workerId, workers(workerId)))
      }
      isInitialized = true
    }
  }

  def getWorkerStatistics: List[WorkerStatistics] = {
    parallelWorkerProxies.map(_.getWorkerStatistics).toList
  }

  def receive(message: Any) {
    this.synchronized {
      if (!message.isInstanceOf[LogMessage]) {
        messagesReceived.incrementAndGet
      }
      message match {
        case r: WorkerReply =>
          workerProxies(r.workerId).receive(message)
        case ws: WorkerStatus =>
          statusMonitor.synchronized {
            if (!workerStatusMap.keySet.contains(ws.workerId) || workerStatusMap.get(ws.workerId).messagesSent < ws.messagesSent) {
              workerStatusMap.put(ws.workerId, ws)
              statusMonitor.notifyAll
            }

          }
        case l: LogMessage =>
          logger.receive(l)
        case other => println("Received unknown message: " + other)
      }
    }
  }

  def messagesSentByWorkers = workerStatusMap.values.foldLeft(0l)(_ + _.messagesSent) + workerStatusMap.size // the status message that was sent was not yet counted by the worker
  def messagesSentByWorkerProxies = workerProxyMessageBuses.foldLeft(0l)(_ + _.messagesSent)
  def messagesSentByCoordinator = messageBus.messagesSent

  def messagesReceivedByWorkers = workerStatusMap.values.foldLeft(0l)(_ + _.messagesReceived)
  def messagesReceivedByCoordinator = messagesReceived.get

  def totalMessagesSent: Long = messagesSentByWorkers + messagesSentByWorkerProxies + messagesSentByCoordinator
  def totalMessagesReceived: Long = messagesReceivedByWorkers + messagesReceivedByCoordinator

  def idle: Boolean = workerStatusMap.values.forall(_.isIdle) && totalMessagesSent == totalMessagesReceived

  def awaitIdle: Long = {
    val startTime = System.nanoTime
    statusMonitor.synchronized {
      while (!idle) {
        statusMonitor.wait(10)
      }
    }
    val stopTime = System.nanoTime
    stopTime - startTime
  }

  def paused: Boolean = workerStatusMap.values.forall(_.isPaused)

  def awaitPaused {
    statusMonitor.synchronized {
      while (!paused) {
        statusMonitor.wait(10)
      }
    }
  }

  def reachedMinInboxSize: Boolean = workerStatusMap.isEmpty || workerStatusMap.values.forall(!_.isOverstrained)

  def awaitMessageProcessing {
    statusMonitor.synchronized {
      while (!reachedMinInboxSize) {
        statusMonitor.wait(10)
      }
    }
  }

  override def addEdge(edge: Edge[_, _]) {
    super.addEdge(edge)
    if (config.workerConfiguration.messageInboxLimits.isDefined) {
      awaitMessageProcessing
    }
  }

  override def addVertex(vertex: Vertex[_, _]) {
    super.addVertex(vertex)
    if (config.workerConfiguration.messageInboxLimits.isDefined) {
      awaitMessageProcessing
    }
  }

  override def sendSignalToVertex(signal: Any, targetId: Any, sourceId: Any = EXTERNAL) {
    super.sendSignalToVertex(signal, targetId, sourceId)
    if (config.workerConfiguration.messageInboxLimits.isDefined) {
      awaitMessageProcessing
    }
  }

  def started: Boolean = workerStatusMap.values.forall(!_.isPaused)

  def awaitStarted {
    statusMonitor.synchronized {
      while (!started) {
        statusMonitor.wait(10)
      }
    }
  }

  def signalStep = {
    signalSteps += 1
    parallelWorkerProxies foreach (_.signalStep)
  }

  def collectStep: Boolean = {
    collectSteps += 1
    parallelWorkerProxies.map(_.collectStep).fold(true)(_ && _)
  }

  def startComputation {
    parallelWorkerProxies.foreach(_.startComputation)
    awaitStarted
  }
{'fraction_non_alphanumeric': 0.08055805580558056, 'fraction_numerical': 0.0036003600360036, 'mean_word_length': 3.767085504137297, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 13, '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': '1568170', 'n_tokens_mistral': 4412, 'n_tokens_neox': 4308, 'n_words': 1028}