<commit_before>
from random import choice
from os import chdir, getcwd

from .config import DaemonConfig


def repo_status(daemon_config=None):
    """Return a dict of repo_config -> [((pipeline_name, version), number_outstanding_jobs)]."""
    daemon_config = daemon_config if daemon_config else DaemonConfig.load_from_yaml()
    original_dir = getcwd()
    for repo_config in daemon_config.list_repos():
        try:
            chdir(repo_config.repo_path)
            yield repo_config, _status_one_repo(daemon_config, repo_config)
        except Exception:
            raise
        finally:
            chdir(original_dir)


class _SnakemakeInfoGrabber:

    def __init__(self):
        self.num_outstanding_jobs = -1

    def handle_msg(self, msg):
        if msg['level'] != 'run_info':
            return
        msg = msg['msg']
        line_list = msg.split('\n')[2:]  # first two lines are cruft
        line_list = [el.strip().split('\t') for el in line_list]
        job_list = [
            (tkns[1], int(tkns[0]))
            for tkns in line_list if len(tkns) == 2
        ]
        self.num_outstanding_jobs = 0
        for rulename, count in job_list:<commit_msg>multiprocessing status<commit_after>
from random import choice
from os import chdir, getcwd
from multiprocessing import Pool, TimeoutError

from .config import DaemonConfig


def repo_status(daemon_config=None, timeout=3600):
    """Return a dict of repo_config -> [((pipeline_name, version), number_outstanding_jobs)]."""
    daemon_config = daemon_config if daemon_config else DaemonConfig.load_from_yaml()
    original_dir = getcwd()
    pool = Pool(daemon_config.total_jobs)

    def handle_one_repo(repo_config):
        try:
            chdir(repo_config.repo_path)
            return repo_config, list(_status_one_repo(daemon_config, repo_config))
        except Exception:
            raise
        finally:
            chdir(original_dir)

    for result in pool.apply_async(handle_one_repo, daemon_config.list_repos()):
        try:
            yield result.get(timeout)
        except TimeoutError:
            pass


class _SnakemakeInfoGrabber:

    def __init__(self):
        self.num_outstanding_jobs = -1

    def handle_msg(self, msg):
        if msg['level'] != 'run_info':
            return
        msg = msg['msg']
        line_list = msg.split('\n')[2:]  # first two lines are cruft
        line_list = [el.strip().split('\t') for el in line_list]
        job_list = [
            (tkns[1], int(tkns[0]))
            for tkns in line_list if len(tkns) == 2
        ]
        self.num_outstanding_jobs = 0
        for rulename, count in job_list:
{'fraction_non_alphanumeric': 0.09622641509433963, 'fraction_numerical': 0.0060377358490566035, 'mean_word_length': 2.6068027210884352, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '2611890', 'n_tokens_mistral': 848, 'n_tokens_neox': 828, 'n_words': 201}