<commit_before>package leo
package datastructures
package blackboard.scheduler

import java.util.concurrent.atomic.AtomicInteger

import leo.datastructures.blackboard._

import scala.collection.immutable.TreeMap
import scala.collection.mutable
import java.util.concurrent.{Executors, RejectedExecutionException, ThreadFactory, TimeUnit, Future}

import leo.datastructures.blackboard.impl.SZSDataStore
import leo.datastructures.context.Context
import leo.modules.SZSException


/**
 * Singleton Scheduler
 */
object Scheduler {

  private[scheduler] lazy val s : Scheduler = {val s = new SchedulerImpl(n); s.start(); s}
  private lazy val n : Int = try {Configuration.THREADCOUNT} catch { case _ : Exception => Configuration.DEFAULT_THREADCOUNT}

  /**
   * Creates a Scheduler for 5 Threads or a get for the singleton,
   * if the scheduler already exists.
    *
    * @return
   */
  def apply() : Scheduler = {
    s
  }
}


/**
 * Scheduler Interface
 */

trait Scheduler {

  def isTerminated : Boolean

  /**
   * Terminate all working processes.
   */
  def killAll() : Unit

  /**
   * Tells the Scheduler to continue to Work
   */
  def signal() : Unit

  /**
   * Pauses the execution of the scheduler.
   */
  def pause() : Unit

  /**
   * Performs one round of auction
   */
  def step() : Unit

  def clear() : Unit

  protected[scheduler] def start()

  def openTasks : Int

  def submitIndependent(r : Runnable) : Future[_]
}





// TODO IF GROWS MOVE TO IMPL PACKAGE
/**
 * <p>
 * Central Object for coordinating the ThreadPool responsible for
 * executing the Agents
 * </p>
  *
  * @author <NAME>
 * @since 5/15/14
 */
protected[scheduler] class SchedulerImpl (numberOfThreads : Int) extends Scheduler {
  import leo.agents._

  private var exe = Executors.newFixedThreadPool(numberOfThreads, MyThreadFactory)
  private val s : SchedulerRun = new SchedulerRun()
  private val w : Writer = new Writer()

  private var sT : Thread = null
  private var sW : Thread = null

  protected val curExec : mutable.Set[Task] = new mutable.HashSet[Task] with mutable.SynchronizedSet[Task]

  def openTasks : Int = synchronized(curExec.size)

  override def isTerminated : Boolean = endFlag

  override def submitIndependent(r : Runnable) : Future[_] = {
    exe.submit(r)
  }

  def signal() : Unit = s.synchronized{
    pauseFlag = false
    s.notifyAll()
  }

  def step() : Unit = s.synchronized{s.notifyAll()}

//  def toWork(a : Agent) : Unit = exe.submit(new Runnable {
//    override def run(): Unit = if (a.guard()) a.apply()
//  })

  def killAll() : Unit = s.synchronized{
    endFlag = true
    pauseFlag = false
    exe.shutdown()
    exe.shutdownNow()
    if(!exe.awaitTermination(30, TimeUnit.MILLISECONDS)) {
      MyThreadFactory.killAll()
    }
    AgentWork.executingAgents() foreach(_.kill())
    Blackboard().filterAll(a => a.filter(DoneEvent()))
    curExec.clear()
    AgentWork.clear()
    ExecTask.put(ExitResult,ExitTask, null)   // For the writer to exit, if he is waiting for a result
    sT.interrupt()
    s.notifyAll()
    curExec.clear()
//    Scheduler.s = null
  }


  var pauseFlag = true
  var endFlag = false

  def pause() : Unit = {s.synchronized(pauseFlag = true)
//    println("Scheduler paused.")
  }

  def clear() : Unit = {
    pause()
    Blackboard().forceCheck()
    curExec.clear()
    AgentWork.executingAgents() foreach(_.kill())
    AgentWork.clear()
  }

  protected[scheduler] def start() {
//    println("Scheduler started.")
    sT = new Thread(s)
    sT.start()      // Start Scheduler
    sW = new Thread(w)
    sW.start()      // Start writer
  }

  /**
   * Takes Tasks from the Queue and Executes it.
   */
  private class SchedulerRun extends Runnable {
    override def run(): Unit = while (true) {
      // Check Status TODO Try catch
      this.synchronized {
        if (pauseFlag) {
          // If is paused wait
          Out.trace("Scheduler paused.")
          try {
            this.wait()
          } catch {
            case e : InterruptedException => Out.info("Scheduler interrupted. Quiting now."); return
          }
          Out.trace("Scheduler is commencing.")
        }
        if (endFlag) return // If is ended quit
      }
<commit_msg>Talkative scheduler termination<commit_after>package leo
package datastructures
package blackboard.scheduler

import java.util.concurrent.atomic.AtomicInteger

import leo.datastructures.blackboard._

import scala.collection.immutable.TreeMap
import scala.collection.mutable
import java.util.concurrent.{Executors, RejectedExecutionException, ThreadFactory, TimeUnit, Future}

import leo.datastructures.blackboard.impl.SZSDataStore
import leo.datastructures.context.Context
import leo.modules.SZSException


/**
 * Singleton Scheduler
 */
object Scheduler {

  private[scheduler] lazy val s : Scheduler = {val s = new SchedulerImpl(n); s.start(); s}
  private lazy val n : Int = try {Configuration.THREADCOUNT} catch { case _ : Exception => Configuration.DEFAULT_THREADCOUNT}

  /**
   * Creates a Scheduler for 5 Threads or a get for the singleton,
   * if the scheduler already exists.
    *
    * @return
   */
  def apply() : Scheduler = {
    s
  }
}


/**
 * Scheduler Interface
 */

trait Scheduler {

  def isTerminated : Boolean

  /**
   * Terminate all working processes.
   */
  def killAll() : Unit

  /**
   * Tells the Scheduler to continue to Work
   */
  def signal() : Unit

  /**
   * Pauses the execution of the scheduler.
   */
  def pause() : Unit

  /**
   * Performs one round of auction
   */
  def step() : Unit

  def clear() : Unit

  protected[scheduler] def start()

  def openTasks : Int

  def submitIndependent(r : Runnable) : Future[_]
}





// TODO IF GROWS MOVE TO IMPL PACKAGE
/**
 * <p>
 * Central Object for coordinating the ThreadPool responsible for
 * executing the Agents
 * </p>
  *
  * @author <NAME>
 * @since 5/15/14
 */
protected[scheduler] class SchedulerImpl (numberOfThreads : Int) extends Scheduler {
  import leo.agents._

  private var exe = Executors.newFixedThreadPool(numberOfThreads, MyThreadFactory)
  private val s : SchedulerRun = new SchedulerRun()
  private val w : Writer = new Writer()

  private var sT : Thread = null
  private var sW : Thread = null

  protected val curExec : mutable.Set[Task] = new mutable.HashSet[Task] with mutable.SynchronizedSet[Task]

  def openTasks : Int = synchronized(curExec.size)

  override def isTerminated : Boolean = endFlag

  override def submitIndependent(r : Runnable) : Future[_] = {
    exe.submit(r)
  }

  def signal() : Unit = s.synchronized{
    pauseFlag = false
    s.notifyAll()
  }

  def step() : Unit = s.synchronized{s.notifyAll()}

//  def toWork(a : Agent) : Unit = exe.submit(new Runnable {
//    override def run(): Unit = if (a.guard()) a.apply()
//  })

  def killAll() : Unit = s.synchronized{
    endFlag = true
    pauseFlag = false
    exe.shutdown()
    exe.shutdownNow()
    if(!exe.awaitTermination(30, TimeUnit.MILLISECONDS)) {
      MyThreadFactory.killAll()
    }
    AgentWork.executingAgents() foreach(_.kill())
    Blackboard().filterAll(a => a.filter(DoneEvent()))
    curExec.clear()
    AgentWork.clear()
    ExecTask.put(ExitResult,ExitTask, null)   // For the writer to exit, if he is waiting for a result
    sT.interrupt()
    s.notifyAll()
    curExec.clear()
//    Scheduler.s = null
  }


  var pauseFlag = true
  var endFlag = false

  def pause() : Unit = {s.synchronized(pauseFlag = true)
//    println("Scheduler paused.")
  }

  def clear() : Unit = {
    pause()
    Blackboard().forceCheck()
    curExec.clear()
    AgentWork.executingAgents() foreach(_.kill())
    AgentWork.clear()
  }

  protected[scheduler] def start() {
//    println("Scheduler started.")
    sT = new Thread(s)
    sT.start()      // Start Scheduler
    sW = new Thread(w)
    sW.start()      // Start writer
  }

  /**
   * Takes Tasks from the Queue and Executes it.
   */
  private class SchedulerRun extends Runnable {
    override def run(): Unit = while (true) {
      // Check Status TODO Try catch
      this.synchronized {
        if (pauseFlag) {
          // If is paused wait
          Out.trace("Scheduler paused.")
          try {
            this.wait()
          } catch {
            case e : InterruptedException => Out.trace("Scheduler interrupted. Quiting now."); return
          }
          Out.trace("Scheduler is commencing.")
        }
        if (endFlag) return // If is ended quit
      }

{'fraction_non_alphanumeric': 0.11448129352059483, 'fraction_numerical': 0.0018883512333293993, 'mean_word_length': 2.843083900226757, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 15, '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': '3229435', 'n_tokens_mistral': 2731, 'n_tokens_neox': 2612, 'n_words': 734}