<commit_before># Copyright (C) 2011, Quentin "mefyl" Hocquet
#
# This software is provided "as is" without warranty of any kind,
# either expressed or implied, including but not limited to the
# implied warranties of fitness for a particular purpose.
#
# See the LICENSE file for more information.

import greenlet
import sys
import threading
import time
import traceback
import types
import drake.debug
import drake.threadpool

class Frozen:
  pass

class CoroutineDone(Exception):
  pass

class CoroutineFrozen(Exception):
  pass

class Terminate(Exception):
  pass

class OrderedSet:

  def __init__(self):
    self.__set = set()
    self.__list = []

  def add(self, v):
    self.__set.add(v)
    self.__list.append(v)

  def remove(self, v):
    self.__set.remove(v)

  def __iter__(self):
    remove = []
    i = 0
    while i < len(self.__list):
      elt = self.__list[i]
      if elt in self.__set:
        yield elt
        i += 1
      else:
        del self.__list[i]

  def __bool__(self):
    return bool(self.__set)

class Scheduler:

  __instance = None
  __pool = None

  @classmethod
  def scheduler(self):
    return Scheduler.__instance

  def __init__(self):
    self.reset()
    Scheduler.__instance = self
    self.__lock = threading.Condition()
    self.__scheduled = []

  def reset(self):
    self.__coroutines = OrderedSet()
    self.__coroutines_frozen = set()
    self.__running = False
    self.__exception = None
    self.__traceback = None

  def running(self):
    return self.__running

  def debug(self, msg):
    drake.debug.debug(msg, drake.debug.DEBUG_SCHED)


  def add(self, coro):
    self.debug('new coroutine: %s' % (coro.name))
    self.__coroutines.add(coro)

  def run(self):

    assert not self.__running

    self.__running = True
    self.die = False
    Scheduler.__pool = drake.threadpool.ThreadPool()

    try:
      while True:
        if self.__exception is not None:
          self.debug('pending exception %s, dying' % (self.__exception,))
          # FIXME: terminate everyone
          e = self.__exception
          self.__exception = None
          raise e
        if self.__coroutines:
          with self.__lock:
            for f in self.__scheduled:
              f()
            self.__scheduled = []
        elif not self.__coroutines_frozen:
          self.debug('no more coroutine, dying')
          break
        else:
          while not self.__coroutines:
            with self.__lock:
              if not self.__scheduled:
                self.__lock.wait()
              assert self.__scheduled
              for f in self.__scheduled:
                f()
              self.__scheduled = []

        running = list(self.__coroutines)
        for coro in running:
          self.__step(coro)
    finally:
      Scheduler.__pool.stop()
      Scheduler.__pool = None
      self.__running = False

  def __step(self, coro):
    if (coro.done):
      return
    self.debug('step %s' % coro)
    triggered = None
    try:
      coro.step()
    except Exception as e:
      parent = coro._Coroutine__parent
      if parent is not None:
        parent.throw(e, coro._Coroutine__traceback)
        triggered = parent
      else:
        self.__exception = e.with_traceback(coro._Coroutine__traceback)
    if coro.done:
      self.debug('%s ended' % coro)
      self.__coroutines.remove(coro)
    elif coro.frozen:
      self.debug('%s froze' % coro)
      self.__coroutines_frozen.add(coro)
      self.__coroutines.remove(coro)
    if triggered is not None:
      self.__step(triggered)

  def unfreeze(self, coro):
    self.debug('%s wakes up' % coro)
    self.__coroutines.add(coro)
    self.__coroutines_frozen.remove(coro)

  def run_one(self, coro):
    assert not self.running()
    self.reset()
    try:
      prev = self.__local.coroutine
      self.__local.coroutine = coro
      while coro.routine:
        coro._Coroutine__step()
    finally:
      self.__local.coroutine = prev

  def schedule(self, f):
    with self.__lock:
      self.__scheduled.append(f)
      self.__lock.notify()

class Waitable:
  def __init__(self):
    self.__waiting = set()

  def __wait(self, coro):
    self.__waiting.add(coro)
    return True

  def __unwait(self, coro):
    self.__waiting.remove(coro)

  def __wake_all(self):
    for coro in self.__waiting:
      coro._Coroutine__unwait(self)
    self.__waiting.clear()

  def __wake_one(self):
    if self.__waiting:
      self.__wake(next(iter(self.__waiting)))

  def __wake(self, coro):
    coro._Coroutine__unwait(self)
    self.__waiting.remove(coro)

class Signal(Waitable):

  def signal(self):
    self._Waitable__wake_all()

class classproperty:
  def __init__(self, f):
    self.__f = f
  def __get__(self, instance, owner):
    return self.__f(instance or owner)

class Coroutine(Waitable):

  __current = None<commit_msg>Sched: Waitable.wake_one: return whether a coroutine was waked up.<commit_after># Copyright (C) 2011, Quentin "mefyl" Hocquet
#
# This software is provided "as is" without warranty of any kind,
# either expressed or implied, including but not limited to the
# implied warranties of fitness for a particular purpose.
#
# See the LICENSE file for more information.

import greenlet
import sys
import threading
import time
import traceback
import types
import drake.debug
import drake.threadpool

class Frozen:
  pass

class CoroutineDone(Exception):
  pass

class CoroutineFrozen(Exception):
  pass

class Terminate(Exception):
  pass

class OrderedSet:

  def __init__(self):
    self.__set = set()
    self.__list = []

  def add(self, v):
    self.__set.add(v)
    self.__list.append(v)

  def remove(self, v):
    self.__set.remove(v)

  def __iter__(self):
    remove = []
    i = 0
    while i < len(self.__list):
      elt = self.__list[i]
      if elt in self.__set:
        yield elt
        i += 1
      else:
        del self.__list[i]

  def __bool__(self):
    return bool(self.__set)

class Scheduler:

  __instance = None
  __pool = None

  @classmethod
  def scheduler(self):
    return Scheduler.__instance

  def __init__(self):
    self.reset()
    Scheduler.__instance = self
    self.__lock = threading.Condition()
    self.__scheduled = []

  def reset(self):
    self.__coroutines = OrderedSet()
    self.__coroutines_frozen = set()
    self.__running = False
    self.__exception = None
    self.__traceback = None

  def running(self):
    return self.__running

  def debug(self, msg):
    drake.debug.debug(msg, drake.debug.DEBUG_SCHED)


  def add(self, coro):
    self.debug('new coroutine: %s' % (coro.name))
    self.__coroutines.add(coro)

  def run(self):

    assert not self.__running

    self.__running = True
    self.die = False
    Scheduler.__pool = drake.threadpool.ThreadPool()

    try:
      while True:
        if self.__exception is not None:
          self.debug('pending exception %s, dying' % (self.__exception,))
          # FIXME: terminate everyone
          e = self.__exception
          self.__exception = None
          raise e
        if self.__coroutines:
          with self.__lock:
            for f in self.__scheduled:
              f()
            self.__scheduled = []
        elif not self.__coroutines_frozen:
          self.debug('no more coroutine, dying')
          break
        else:
          while not self.__coroutines:
            with self.__lock:
              if not self.__scheduled:
                self.__lock.wait()
              assert self.__scheduled
              for f in self.__scheduled:
                f()
              self.__scheduled = []

        running = list(self.__coroutines)
        for coro in running:
          self.__step(coro)
    finally:
      Scheduler.__pool.stop()
      Scheduler.__pool = None
      self.__running = False

  def __step(self, coro):
    if (coro.done):
      return
    self.debug('step %s' % coro)
    triggered = None
    try:
      coro.step()
    except Exception as e:
      parent = coro._Coroutine__parent
      if parent is not None:
        parent.throw(e, coro._Coroutine__traceback)
        triggered = parent
      else:
        self.__exception = e.with_traceback(coro._Coroutine__traceback)
    if coro.done:
      self.debug('%s ended' % coro)
      self.__coroutines.remove(coro)
    elif coro.frozen:
      self.debug('%s froze' % coro)
      self.__coroutines_frozen.add(coro)
      self.__coroutines.remove(coro)
    if triggered is not None:
      self.__step(triggered)

  def unfreeze(self, coro):
    self.debug('%s wakes up' % coro)
    self.__coroutines.add(coro)
    self.__coroutines_frozen.remove(coro)

  def run_one(self, coro):
    assert not self.running()
    self.reset()
    try:
      prev = self.__local.coroutine
      self.__local.coroutine = coro
      while coro.routine:
        coro._Coroutine__step()
    finally:
      self.__local.coroutine = prev

  def schedule(self, f):
    with self.__lock:
      self.__scheduled.append(f)
      self.__lock.notify()

class Waitable:
  def __init__(self):
    self.__waiting = set()

  def __wait(self, coro):
    self.__waiting.add(coro)
    return True

  def __unwait(self, coro):
    self.__waiting.remove(coro)

  def __wake_all(self):
    for coro in self.__waiting:
      coro._Coroutine__unwait(self)
    self.__waiting.clear()

  def __wake_one(self):
    if self.__waiting:
      self.__wake(next(iter(self.__waiting)))
      return True
    else:
      return False

  def __wake(self, coro):
    coro._Coroutine__unwait(self)
    self.__waiting.remove(coro)

class Signal(Waitable):

  def signal(self):
    self._Waitable__wake_all()

class classproperty:
  def __init__(self, f):
    self.__f = f
  def __get__(self, instance, owner):
    return self.__f(instance or owner)

class Coroutine(Waitable):

  __current = None
{'fraction_non_alphanumeric': 0.09891792568395263, 'fraction_numerical': 0.0012249897917517355, 'mean_word_length': 2.4691926345609065, 'pattern_counts': {'":': 0, '<': 5, '<?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': '873581', 'n_tokens_mistral': 3175, 'n_tokens_neox': 3131, 'n_words': 780}