<commit_before>
from __future__ import absolute_import, with_statement

import functools
import logging
import signal
import threading
import time

import pyuv

from collections import deque
from twisted.internet.base import _SignalReactorMixin
from twisted.internet.posixbase import PosixReactorBase
from twisted.internet.interfaces import IReactorFDSet, IDelayedCall, IReactorTime
from twisted.python import failure, log
from twisted.internet import error
from zope.interface import implements

from .util import SocketPair


class UVWaker(object):
    def __init__(self, reactor):
        self._async = pyuv.Async(reactor._loop, lambda x: None)
        self._async.unref()
    def wakeUp(self):
        self._async.send()


class UVDelayedCall(object):
    implements(IDelayedCall)

    def __init__(self, reactor, seconds, f, *args, **kw):
        self._reactor = reactor
        self._func = functools.partial(f, *args, **kw)
        self._time = self._reactor.seconds() + seconds
        self._timer = pyuv.Timer(self._reactor._loop)
        self._timer.start(self._called, seconds, 0.0)

    def _called(self, handle):
        self._timer.stop()
        self._reactor._removeDelayedCall(self)
        try:
            self._func()
        except Exception:
            logging.error("_called caught exception", exc_info=True)

    def getTime(self):
        return self._time

    def cancel(self):
        self._timer.stop()
        self._reactor._removeDelayedCall(self)

    def delay(self, seconds):
        self._timer.stop()
        self._time += seconds
        self._timer.start(self._called, self._time-self._reactor.seconds(), 0.0)

    def reset(self, seconds):
        self._timer.stop()
        self._time = self._reactor.seconds() + seconds
        self._timer.start(self._called, seconds, 0.0)

    def active(self):
        return self._timer.active


class UVReactor(PosixReactorBase):
    implements(IReactorTime, IReactorFDSet)

    def __init__(self):
        self._loop = pyuv.Loop()
        self._async_handle = pyuv.Async(self._loop, self._async_cb)
        self._async_handle_lock = threading.Lock()
        self._async_callbacks = deque()
        self._readers = {}  # map of reader objects to fd
        self._writers = {}  # map of writer objects to fd
        self._fds = {}      # map of fd to a (reader, writer) tuple
        self._delayedCalls = {}<commit_msg>Need to use "implementer" decorator<commit_after>
from __future__ import absolute_import, with_statement

import functools
import logging
import signal
import threading
import time

import pyuv

from collections import deque
from twisted.internet.base import _SignalReactorMixin
from twisted.internet.posixbase import PosixReactorBase
from twisted.internet.interfaces import IReactorFDSet, IDelayedCall, IReactorTime
from twisted.python import failure, log
from twisted.internet import error
from zope.interface import implementer

from .util import SocketPair


class UVWaker(object):
    def __init__(self, reactor):
        self._async = pyuv.Async(reactor._loop, lambda x: None)
        self._async.unref()
    def wakeUp(self):
        self._async.send()

@implementer(IDelayedCall)
class UVDelayedCall(object):
    def __init__(self, reactor, seconds, f, *args, **kw):
        self._reactor = reactor
        self._func = functools.partial(f, *args, **kw)
        self._time = self._reactor.seconds() + seconds
        self._timer = pyuv.Timer(self._reactor._loop)
        self._timer.start(self._called, seconds, 0.0)

    def _called(self, handle):
        self._timer.stop()
        self._reactor._removeDelayedCall(self)
        try:
            self._func()
        except Exception:
            logging.error("_called caught exception", exc_info=True)

    def getTime(self):
        return self._time

    def cancel(self):
        self._timer.stop()
        self._reactor._removeDelayedCall(self)

    def delay(self, seconds):
        self._timer.stop()
        self._time += seconds
        self._timer.start(self._called, self._time-self._reactor.seconds(), 0.0)

    def reset(self, seconds):
        self._timer.stop()
        self._time = self._reactor.seconds() + seconds
        self._timer.start(self._called, seconds, 0.0)

    def active(self):
        return self._timer.active

@implementer(IReactorTime, IReactorFDSet)
class UVReactor(PosixReactorBase):
    def __init__(self):
        self._loop = pyuv.Loop()
        self._async_handle = pyuv.Async(self._loop, self._async_cb)
        self._async_handle_lock = threading.Lock()
        self._async_callbacks = deque()
        self._readers = {}  # map of reader objects to fd
        self._writers = {}  # map of writer objects to fd
        self._fds = {}      # map of fd to a (reader, writer) tuple
        self._delayedCalls = {}
{'fraction_non_alphanumeric': 0.09977135730617336, 'fraction_numerical': 0.002494283932654334, 'mean_word_length': 2.960493827160494, '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': '629731', 'n_tokens_mistral': 1462, 'n_tokens_neox': 1432, 'n_words': 352}