<commit_before>import asyncio
from concurrent.futures import ThreadPoolExecutor

import janus

from lightbus.utilities.async_tools import block, cancel, make_exception_checker, get_event_loop

import logging


# Logging setup


class AsyncioLoggingFilter(logging.Filter):
    def filter(self, record):
        task = asyncio.Task.current_task()

        record.task = f'[task {id(task)}]' if task else '[NOLOOP]'
        return True


logger = logging.getLogger(__name__)
logger.addFilter(AsyncioLoggingFilter())
logging.getLogger('asyncio').setLevel(logging.WARNING)


logging.basicConfig(level=logging.DEBUG, format="%(msecs)f %(threadName)s %(task)s %(msg)s")

thread_pool_executor = ThreadPoolExecutor(thread_name_prefix="dispatch")


# Utilities


async def execute_in_thread(fn):
    def new_thread_setup():
        # Make sure each thread has an event loop available, even if it
        # isn't actually running (stops janus.Queue barfing)
        get_event_loop()
        return fn()

    # Run the given blocking fn in a thread
    return await asyncio.get_event_loop().run_in_executor(
        executor=thread_pool_executor,
        func=new_thread_setup
    )


def run_in_main_thread():
    """Decorator to ensure pass any method invocations into the main thread"""

    def decorator(fn):
        def wrapper(*args, **kwargs):
            # Assume the first arg is the 'self', i.e. the bus client
            bus = args[0]

            # We'll provide a queue as the return path for results
            result_queue = janus.Queue()

            # Enqueue the function, it's arguments, and our return path queue
            logger.debug("Adding callable to queue")
            bus._call_queue.sync_q.put((fn, args, kwargs, result_queue))

            # Wait for a return value on the result queue
            logger.debug("Awaiting execution completion")
            result = result_queue.sync_q.get()

            # Cleanup
            bus._call_queue.sync_q.join()  # Needed?
            result_queue.close()

            return result
        return wrapper
    return decorator


# Simplified bus client


class Bus(object):

    def __init__(self):
        self._call_queue = janus.Queue()

    def open(self):
        logger.debug("Bus is opening")

        # Start a background task running to handling incoming calls on the _perform_calls queue
        self._perform_calls_task = asyncio.ensure_future(self._perform_calls())

        # Housekeeping for error handling and logging
        self._perform_calls_task.add_done_callback(make_exception_checker())
        self._perform_calls_task.name = 'PerformCallsTask'

    @run_in_main_thread()
    async def call_rpc(self, name):<commit_msg>Experiment cleanup<commit_after>import asyncio
from concurrent.futures import ThreadPoolExecutor

import janus

from lightbus.utilities.async_tools import block, cancel, make_exception_checker, get_event_loop

import logging


# Logging setup


class AsyncioLoggingFilter(logging.Filter):
    def filter(self, record):
        task = asyncio.Task.current_task()

        record.task = f'[task {id(task)}]' if task else '[NOLOOP         ]'
        return True


logger = logging.getLogger(__name__)
logger.addFilter(AsyncioLoggingFilter())
logging.getLogger('asyncio').setLevel(logging.WARNING)


logging.basicConfig(level=logging.DEBUG, format="%(msecs)f %(threadName)s %(task)s %(msg)s")

thread_pool_executor = ThreadPoolExecutor(thread_name_prefix="dispatch")


# Utilities


async def execute_in_thread(fn):
    def new_thread_setup():
        # Make sure each thread has an event loop available, even if it
        # isn't actually running (stops janus.Queue barfing)
        get_event_loop()
        return fn()

    # Run the given blocking fn in a thread
    return await asyncio.get_event_loop().run_in_executor(
        executor=thread_pool_executor,
        func=new_thread_setup
    )


def run_in_main_thread():
    """Decorator to ensure pass any method invocations into the main thread"""

    def decorator(fn):
        def wrapper(*args, **kwargs):
            # Assume the first arg is the 'self', i.e. the bus client
            bus = args[0]

            # We'll provide a queue as the return path for results
            result_queue = janus.Queue()

            # Enqueue the function, it's arguments, and our return path queue
            logger.debug("Adding callable to queue")
            bus._call_queue.sync_q.put((fn, args, kwargs, result_queue))

            # Wait for a return value on the result queue
            logger.debug("Awaiting execution completion")
            result = result_queue.sync_q.get()

            # Cleanup
            bus._call_queue.sync_q.join()  # Needed?
            result_queue.close()

            return result
        return wrapper
    return decorator


# Simplified bus client


class Bus(object):

    def __init__(self):
        self._call_queue = janus.Queue()

    def open(self):
        logger.debug("Bus is opening")

        # Start a background task running to handling incoming calls on the _perform_calls queue
        self._perform_calls_task = asyncio.ensure_future(self._perform_calls())

        # Housekeeping for error handling
        self._perform_calls_task.add_done_callback(make_exception_checker())

    @run_in_main_thread()
    async def call_rpc(self, name):
{'fraction_non_alphanumeric': 0.08595935110945366, 'fraction_numerical': 0.00037292560134253216, 'mean_word_length': 2.985141158989599, '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': '6707460', 'n_tokens_mistral': 1565, 'n_tokens_neox': 1513, 'n_words': 424}