<commit_before>from observer import Observer
from concurrency import Atomic


class Sink(object):
  """Base class for implementation of query operators, providing
  a lightweight sink that can be disposed to mute the outgoing observer."""

  def __init__(self, observer, cancel):
    super(Sink, self).__init__()
    self.observer = observer
    self.cancel = Atomic(cancel)

  def dispose(self):
    self.observer = Observer.noop

    cancel = self.cancel.exchange(None)

    if cancel != None:
      cancel.dispose()

  class Forewarder(Observer):
    def __init__(self, foreward):
      self.foreward = foreward

    def onNext(self, value):
      self.foreward.observer.onNext(value)

    def onError(self, exception):
      self.foreward.observer.onError(exception)
      self.foreward.dispose()

    def onCompleted(self):
      self.foreward.observer.onCompleted()
      self.foreward.dispose()

  def getForewarder(self):
    return self.Forewarder(self)<commit_msg>added tailrecursivesink<commit_after>from observer import Observer
from concurrency import Atomic
from dispoable import AsyncLock, Disposable, CompositDisposable, SerialDisposable, SingleAssignmentDisposable
from scheduler import Scheduler


class Sink(object):
  """Base class for implementation of query operators, providing
  a lightweight sink that can be disposed to mute the outgoing observer."""

  def __init__(self, observer, cancel):
    super(Sink, self).__init__()
    self.observer = observer
    self.cancel = Atomic(cancel)

  def dispose(self):
    self.observer = Observer.noop

    cancel = self.cancel.exchange(None)

    if cancel != None:
      cancel.dispose()

  class Forewarder(Observer):
    def __init__(self, foreward):
      self.foreward = foreward

    def onNext(self, value):
      self.foreward.observer.onNext(value)

    def onError(self, exception):
      self.foreward.observer.onError(exception)
      self.foreward.dispose()

    def onCompleted(self):
      self.foreward.observer.onCompleted()
      self.foreward.dispose()

  def getForewarder(self):
    return self.Forewarder(self)


class TailRecursiveSink(Sink):
  def __init__(self, observer, cancel):
    super(TailRecursiveSink, self).__init__(observer, cancel)

  @staticmethod
  def unpack(source):
    while hasattr(source, 'eval'):
      source = source.eval()

    return source

  def run(self, sources):
    self.isDisposed = False
    self.subscription = SerialDisposable()
    self.gate = AsyncLock()
    self.stack = []
    self.length = []

    self.stack.append(iter(sources))
    self.length.append(len(sources))

    def scheduled(schedule):
      self.recurse = schedule
      self.gate.wait(self.moveNext)

    cancel = Scheduler.tailRecursion.schedule(scheduled)

    return CompositDisposable(
      self.subscription,
      cancel,
      Disposable.create(lambda: self.gate.wait(self.dispose))
    )

  def extract(self, obj):
    raise NotImplementedError()

  def moveNext(self):
    hasCurrent = False
    current = None

    while True:
      if self.stack.count == 0:
        break

      if self.isDisposed:
        return

      e = self.stack[-1]
      l = self.length[-1]

      try:
        current = next(e)
        hasCurrent = True
      except StopIteration:
        pass
      except Exception as e:
        self.observer.onError(e)
        self.dispose()
        return

      if not hasCurrent:
        self.stack.pop()
        self.length.pop()
      else:
        r = l - 1
        self.length[-1] = r

        try:
          current = TailRecursiveSink.unpack(current)
        except Exception as e:
          self.observer.onError(e)
          self.dispose()
          return

        # Tail recursive case; drop the current frame.
        if r == 0:
          self.stack.pop()
          self.length.pop()

        # Flattening of nested sequences. Prevents stack overflow in observers.
        nextSeq = self.extract(current)

        if nextSeq != None:
          self.stack.append(iter(nextSeq))
          self.length.append(len(nextSeq))

          hasCurrent = False

      if hasCurrent:
        break

    if not hasCurrent:
      self.done()
      return

    d = SingleAssignmentDisposable()
    self.subscription.disposable = d
    d.disposable = current.subscribeSafe(self)

  def dispose(self):
    self.stack.clear()
    self.length.clear()
    self.isDisposed = True

  def done(self):
    self.observer.onCompleted()
    self.dispose()

{'fraction_non_alphanumeric': 0.09832026875699888, 'fraction_numerical': 0.0013437849944008958, 'mean_word_length': 2.703150912106136, 'pattern_counts': {'":': 0, '<': 3, '<?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': '5163250', 'n_tokens_mistral': 1381, 'n_tokens_neox': 1290, 'n_words': 319}