<commit_before>from multiprocessing.queues import Empty

import pulsar
from pulsar import system

    
class IOQueue(system.EpollProxy):
    '''The polling mechanism for a task queue. No select or epoll performed here, simply
return task from the queue if available.
This is an interface for using the same IOLoop class of other workers.'''
    def __init__(self, queue):
        super(IOQueue,self).__init__()
        self._queue = queue
        self._fd = id(queue)
        self._empty = []
    
    def fileno(self):
        return self._fd
    
    def poll(self, timeout = 0):
        try:
            req = self._queue.get(timeout = timeout)
            return {self._fd:req}
        except Empty:
            return self._empty


class TaskScheduler(pulsar.WorkerMonitor):
    
    def _addtask(self, caller, task_name, targs, tkwargs, ack = True, **kwargs):
        try:
            request = self.app.make_request(task_name, targs, tkwargs, **kwargs)
            self.task_queue.put((None,request))
            if ack:
                return request.todict()
        except Exception as e:
            return e
    
    def actor_addtask(self, caller, task_name, targs, tkwargs, ack=True, **kwargs):
        return self._addtask(caller, task_name, targs, tkwargs, ack = True, **kwargs)
        
    def actor_addtask_noack(self, caller, task_name, targs, tkwargs, ack=False, **kwargs):
        return self._addtask(caller, task_name, targs, tkwargs, ack = False, **kwargs)
    actor_addtask_noack.ack = False
    
    def actor_task_finished(self, caller, response):
        self.app.task_finished(response)
    actor_task_finished.ack = False
    
    def actor_get_task(self, caller, id):
        return self.app.get_task(id)
    actor_get_task.ack = True
        

class Worker(pulsar.Worker):
    '''A Task worker on a subprocess'''
    _class_code = 'TaskQueue'


    

<commit_msg>Lucas change for exposing a function to return the list of registered task handlers<commit_after>from multiprocessing.queues import Empty

import pulsar
from pulsar import system

    
class IOQueue(system.EpollProxy):
    '''The polling mechanism for a task queue. No select or epoll performed here, simply
return task from the queue if available.
This is an interface for using the same IOLoop class of other workers.'''
    def __init__(self, queue):
        super(IOQueue,self).__init__()
        self._queue = queue
        self._fd = id(queue)
        self._empty = []
    
    def fileno(self):
        return self._fd
    
    def poll(self, timeout = 0):
        try:
            req = self._queue.get(timeout = timeout)
            return {self._fd:req}
        except Empty:
            return self._empty


class TaskScheduler(pulsar.WorkerMonitor):
    
    def _addtask(self, caller, task_name, targs, tkwargs, ack = True, **kwargs):
        try:
            request = self.app.make_request(task_name, targs, tkwargs, **kwargs)
            self.task_queue.put((None,request))
            if ack:
                return request.todict()
        except Exception as e:
            return e
    
    def actor_tasks_list(self, caller):
        return self.app.tasks_list()
    
    def actor_addtask(self, caller, task_name, targs, tkwargs, ack=True, **kwargs):
        return self._addtask(caller, task_name, targs, tkwargs, ack = True, **kwargs)
        
    def actor_addtask_noack(self, caller, task_name, targs, tkwargs, ack=False, **kwargs):
        return self._addtask(caller, task_name, targs, tkwargs, ack = False, **kwargs)
    actor_addtask_noack.ack = False
    
    def actor_task_finished(self, caller, response):
        self.app.task_finished(response)
    actor_task_finished.ack = False
    
    def actor_get_task(self, caller, id):
        return self.app.get_task(id)
    actor_get_task.ack = True
        

class Worker(pulsar.Worker):
    '''A Task worker on a subprocess'''
    _class_code = 'TaskQueue'


    


{'fraction_non_alphanumeric': 0.09432048681541583, 'fraction_numerical': 0.0005070993914807302, 'mean_word_length': 2.5863636363636364, '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': '1753122', 'n_tokens_mistral': 1215, 'n_tokens_neox': 1193, 'n_words': 338}