<commit_before>from datetime import datetime, timedelta
from queue import Queue
from threading import Thread
import logging
import time
import warnings

from bndl.execute.profile import CpuProfiling, MemoryProfiling
from bndl.execute.scheduler import Scheduler
from bndl.execute.worker import current_worker
from bndl.util import plugins
from bndl.util.conf import Config
from bndl.util.exceptions import catch
from bndl.util.lifecycle import Lifecycle


logger = logging.getLogger(__name__)


def _num_connected():
    worker = current_worker()
    return sum(1 for w in worker.peers.filter(node_type='worker') if w.is_connected)


class ExecutionContext(Lifecycle):
    instances = set()

    def __init__(self, node, config=Config()):
        # Make sure the BNDL plugins are loaded
        plugins.load_plugins()

        super().__init__()
        self._node = node
        self.conf = config
        self.jobs = []
        self.signal_start()
        self.instances.add(self)


    @property
    def node(self):
        if self._node is None:
            with catch(RuntimeError):
                self._node = current_worker()
        return self._node


    def execute(self, job, workers=None, order_results=True, concurrency=None, attempts=None):
        assert self.running, 'context is not running'

        if workers is None:
            self.await_workers()
            workers = self.workers[:]

        if not job.tasks:
            return

        self.jobs.append(job)

        done = Queue()
        scheduler = Scheduler(self, job.tasks, done.put, workers, concurrency, attempts)
        scheduler_driver = Thread(target=scheduler.run,
                                  name='bndl-scheduler-%s' % (job.id),
                                  daemon=True)
        job.signal_start()
        scheduler_driver.start()

        try:
            if order_results:
                yield from self._execute_ordered(job, done)
            else:
                yield from self._execute_unordered(job, done)
            scheduler_driver.join()
        except KeyboardInterrupt:
            scheduler.abort()
            raise
        except GeneratorExit:
            scheduler.abort()
        finally:
            scheduler_driver.join()
            for task in job.tasks:
                task.release()
            job.signal_stop()


    def _execute_ordered(self, job, done):
        # keep a dict with results which are 'early' and the task id
        task_ids = iter(task.id for task in job.tasks)
        next_taskid = next(task_ids)
        early = {}

        for task in self._execute_unordered(job, done):
            if task.failed:
                raise task.exception()
            elif task.id != next_taskid:
                early[task.id] = task
            else:
                early.pop(task.id, None)
                yield task
                try:
                    # possibly yield tasks which are next
                    next_taskid = next(task_ids)
                    while early:
                        if next_taskid in early:
                            yield early.pop(next_taskid)
                            next_taskid = next(task_ids)
                        else:
                            break
                except StopIteration:
                    break

        assert not early, 'results for tasks %r not yet yielded' % early


    def _execute_unordered(self, job, done):
        seen = set()
        for task in iter(done.get, None):
            if isinstance(task, Exception):
                raise task
            elif task not in seen:
                seen.add(task)
                yield task


    def await_workers(self, worker_count=None, connect_timeout=5, stable_timeout=60):
        '''
        await_workers waits for workers to be available. If not in
        connect_timeout a RuntimeError is raised. Once a worker is found, at
        most stable_timeout seconds will be waited for the cluster to settle.
        That is, until no new workers are discovered / the worker count is
        stable.

        :param worker_count: int or None
            The expected worker count. When connected to exactly worker_count
            workers, this method will return faster.
        :param connect_timeout: int or float
            Maximum time in seconds waited until the first worker is discovered.
        :param stable_timeout: int or float
            Maximum time in seconds waited until no more workers are discovered.
        '''
        if not isinstance(connect_timeout, timedelta):
            connect_timeout = timedelta(seconds=connect_timeout)
        if not isinstance(stable_timeout, timedelta):<commit_msg>also raise from unordered collection<commit_after>from datetime import datetime, timedelta
from queue import Queue
from threading import Thread
import logging
import time
import warnings

from bndl.execute.profile import CpuProfiling, MemoryProfiling
from bndl.execute.scheduler import Scheduler
from bndl.execute.worker import current_worker
from bndl.util import plugins
from bndl.util.conf import Config
from bndl.util.exceptions import catch
from bndl.util.lifecycle import Lifecycle


logger = logging.getLogger(__name__)


def _num_connected():
    worker = current_worker()
    return sum(1 for w in worker.peers.filter(node_type='worker') if w.is_connected)


class ExecutionContext(Lifecycle):
    instances = set()

    def __init__(self, node, config=Config()):
        # Make sure the BNDL plugins are loaded
        plugins.load_plugins()

        super().__init__()
        self._node = node
        self.conf = config
        self.jobs = []
        self.signal_start()
        self.instances.add(self)


    @property
    def node(self):
        if self._node is None:
            with catch(RuntimeError):
                self._node = current_worker()
        return self._node


    def execute(self, job, workers=None, order_results=True, concurrency=None, attempts=None):
        assert self.running, 'context is not running'

        if workers is None:
            self.await_workers()
            workers = self.workers[:]

        if not job.tasks:
            return

        self.jobs.append(job)

        done = Queue()
        scheduler = Scheduler(self, job.tasks, done.put, workers, concurrency, attempts)
        scheduler_driver = Thread(target=scheduler.run,
                                  name='bndl-scheduler-%s' % (job.id),
                                  daemon=True)
        job.signal_start()
        scheduler_driver.start()

        try:
            if order_results:
                yield from self._execute_ordered(job, done)
            else:
                yield from self._execute_unordered(job, done)
            scheduler_driver.join()
        except KeyboardInterrupt:
            scheduler.abort()
            raise
        except GeneratorExit:
            scheduler.abort()
        finally:
            scheduler_driver.join()
            for task in job.tasks:
                task.release()
            job.signal_stop()


    def _execute_ordered(self, job, done):
        # keep a dict with results which are 'early' and the task id
        task_ids = iter(task.id for task in job.tasks)
        next_taskid = next(task_ids)
        early = {}

        for task in self._execute_unordered(job, done):
            if task.id != next_taskid:
                early[task.id] = task
            else:
                early.pop(task.id, None)
                yield task
                try:
                    # possibly yield tasks which are next
                    next_taskid = next(task_ids)
                    while early:
                        if next_taskid in early:
                            yield early.pop(next_taskid)
                            next_taskid = next(task_ids)
                        else:
                            break
                except StopIteration:
                    break

        assert not early, 'results for tasks %r not yet yielded' % early


    def _execute_unordered(self, job, done):
        seen = set()
        for task in iter(done.get, None):
            if isinstance(task, Exception):
                raise task
            elif task.failed:
                raise task.exception()
            elif task not in seen:
                seen.add(task)
                yield task


    def await_workers(self, worker_count=None, connect_timeout=5, stable_timeout=60):
        '''
        await_workers waits for workers to be available. If not in
        connect_timeout a RuntimeError is raised. Once a worker is found, at
        most stable_timeout seconds will be waited for the cluster to settle.
        That is, until no new workers are discovered / the worker count is
        stable.

        :param worker_count: int or None
            The expected worker count. When connected to exactly worker_count
            workers, this method will return faster.
        :param connect_timeout: int or float
            Maximum time in seconds waited until the first worker is discovered.
        :param stable_timeout: int or float
            Maximum time in seconds waited until no more workers are discovered.
        '''
        if not isinstance(connect_timeout, timedelta):
            connect_timeout = timedelta(seconds=connect_timeout)
        if not isinstance(stable_timeout, timedelta):
{'fraction_non_alphanumeric': 0.06823254330959719, 'fraction_numerical': 0.0008502497608672548, 'mean_word_length': 1.8317785133915137, '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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1036859', 'n_tokens_mistral': 2450, 'n_tokens_neox': 2343, 'n_words': 769}