<commit_before>package susuru.core.resource

import org.apache.log4j.Logger

import scala.collection.mutable
import scala.concurrent.{Future, Promise}

/**
 * Stock resources and matching on demand.
 * @param source Function to get recent resources.
 * @param update Function to get limit count and reset time.
 * @tparam T Resource type
 * @tparam R Result type with limit count and reset time.
 */
class Susuru[T, R](source: () => Map[Id, T], update: R => (Int, EpochTimeMillis)) extends Thread {
  private val logger = Logger.getLogger(this.getClass.getName)
  /** Margin for time window. Unit: millisecond */
  private val margin = 1000
  private var killed = false

  val supplies: mutable.Map[Id, Supply[T]] = mutable.Map.empty[Id, Supply[T]]
  val demands: mutable.Queue[Demand[T, R]] = mutable.Queue.empty[Demand[T, R]]

  def kill(): Unit = {
    killed = true
    demands.synchronized(demands.notify())
  }

  override def run(): Unit = {
    try {
      source().foreach { case (id, t) => supplies.put(id, new Supply[T](t, 1, Long.MaxValue)) }
      while (!killed) {
        matching()
        supplies.map(_._2.until).min - System.currentTimeMillis() match {
          case n if 0 < n =>
            logger.debug(s"Susuru will wait: demands.synchronized(demands.wait($n))")
            demands.synchronized(demands.wait(n))
            logger.debug(s"Susuru was notified: demands.synchronized(demands.wait($n))")
          case _ =>
            logger.debug("Susuru will wait: demands.synchronized(demands.wait())")
            demands.synchronized(demands.wait())
            logger.debug("Susuru was notified: demands.synchronized(demands.wait())")
        }
      }<commit_msg>Check if demands exists before susuru wait<commit_after>package susuru.core.resource

import org.apache.log4j.Logger

import scala.collection.mutable
import scala.concurrent.{Future, Promise}

/**
 * Stock resources and matching on demand.
 * @param source Function to get recent resources.
 * @param update Function to get limit count and reset time.
 * @tparam T Resource type
 * @tparam R Result type with limit count and reset time.
 */
class Susuru[T, R](source: () => Map[Id, T], update: R => (Int, EpochTimeMillis)) extends Thread {
  private val logger = Logger.getLogger(this.getClass.getName)
  /** Margin for time window. Unit: millisecond */
  private val margin = 1000
  private var killed = false

  val supplies: mutable.Map[Id, Supply[T]] = mutable.Map.empty[Id, Supply[T]]
  val demands: mutable.Queue[Demand[T, R]] = mutable.Queue.empty[Demand[T, R]]

  def kill(): Unit = {
    killed = true
    demands.synchronized(demands.notify())
  }

  override def run(): Unit = {
    try {
      source().foreach { case (id, t) => supplies.put(id, new Supply[T](t, 1, Long.MaxValue)) }
      while (!killed) {
        matching()
        demands.synchronized {
          supplies.map(_._2.until).min - System.currentTimeMillis() match {
            case _ if 0 < demands.length =>
              logger.debug(s"Susuru will not wait: demands exists.")
            case n if 0 < n =>
              logger.debug(s"Susuru will wait: demands.synchronized(demands.wait($n))")
              demands.wait(n)
              logger.debug(s"Susuru was notified: demands.synchronized(demands.wait($n))")
            case _ =>
              logger.debug("Susuru will wait: demands.synchronized(demands.wait())")
              demands.wait()
              logger.debug("Susuru was notified: demands.synchronized(demands.wait())")
          }
        }
      }
{'fraction_non_alphanumeric': 0.12919605077574048, 'fraction_numerical': 0.004795486600846262, 'mean_word_length': 2.970884658454647, 'pattern_counts': {'":': 0, '<': 6, '<?xml version=': 0, '>': 14, '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': '4548195', 'n_tokens_mistral': 1087, 'n_tokens_neox': 1066, 'n_words': 291}