<commit_before># -*- coding: utf-8 -*-
"""
    celery.worker.autoreload
    ~~~~~~~~~~~~~~~~~~~~~~~~

    This module implements automatic module reloading
"""
from __future__ import absolute_import
from __future__ import with_statement

import errno
import hashlib
import os
import select
import sys
import time

from collections import defaultdict

from ..abstract import StartStopComponent
from ..utils.threads import bgThread, Event

try:
    import pyinotify
    _ProcessEvent = pyinotify.ProcessEvent
except ImportError:
    pyinotify = None        # noqa
    _ProcessEvent = object  # noqa


class WorkerComponent(StartStopComponent):
    name = "worker.autoreloader"
    requires = ("pool", )

    def __init__(self, w, autoreload=None, **kwargs):
        self.enabled = w.autoreload = autoreload
        w.autoreloader = None

    def create(self, w):
        w.autoreloader = self.instantiate(w.autoreloader_cls,
                                          controller=w,
                                          logger=w.logger)
        return w.autoreloader


def file_hash(filename, algorithm="md5"):
    hobj = hashlib.new(algorithm)
    with open(filename, "rb") as f:
        for chunk in iter(lambda: f.read(2 ** 20), ''):
            hobj.update(chunk)
    return hobj.digest()


class BaseMonitor(object):

    def __init__(self, files, on_change=None, shutdown_event=None,
            interval=0.5):
        self.files = files
        self.interval = interval
        self._on_change = on_change
        self.modify_times = defaultdict(int)
        self.shutdown_event = shutdown_event or Event()

    def start(self):
        raise NotImplementedError("Subclass responsibility")

    def stop(self):
        pass

    def on_change(self, modified):
        if self._on_change:
            return self._on_change(modified)


class StatMonitor(BaseMonitor):
    """File change monitor based on the ``stat`` system call."""

    def _mtimes(self):
        return ((f, self._mtime(f)) for f in self.files)

    def _maybe_modified(self, f, mt):
        return mt is not None and self.modify_times[f] != mt

    def start(self):
        while not self.shutdown_event.is_set():
            modified = dict((f, mt) for f, mt in self._mtimes()
                                if self._maybe_modified(f, mt))
            if modified:
                self.on_change(modified.keys())
                self.modify_times.update(modified)
            time.sleep(self.interval)

    @staticmethod
    def _mtime(path):
        try:
            return os.stat(path).st_mtime
        except Exception:
            pass


class KQueueMonitor(BaseMonitor):
    """File change monitor based on BSD kernel event notifications"""

    def __init__(self, *args, **kwargs):
        assert hasattr(select, "kqueue")
        super(KQueueMonitor, self).__init__(*args, **kwargs)
        self.filemap = dict((f, None) for f in self.files)

    def start(self):
        self._kq = select.kqueue()
        kevents = []
        for f in self.filemap:
            self.filemap[f] = fd = os.open(f, os.O_RDONLY)

            ev = select.kevent(fd,
                    filter=select.KQ_FILTER_VNODE,
                    flags=select.KQ_EV_ADD |
                            select.KQ_EV_ENABLE |
                            select.KQ_EV_CLEAR,
                    fflags=select.KQ_NOTE_WRITE |
                            select.KQ_NOTE_EXTEND)
            kevents.append(ev)

        events = self._kq.control(kevents, 0)
        while not self.shutdown_event.is_set():
            events = self._kq.control(kevents, 1)
            fds = [e.ident for e in events]
            modified = [k for k, v in self.filemap.iteritems()
                                        if v in fds]
            self.on_change(modified)

    def stop(self):
        self._kq.close()
        for fd in filter(None, self.filemap.values()):
            try:
                os.close(fd)
            except OSError, exc:
                if exc != errno.EBADF:
                    raise
            self.filemap[fd] = None
        self.filemap.clear()


class InotifyMonitor(_ProcessEvent):
    """File change monitor based on Linux kernel `inotify` subsystem"""

    def __init__(self, modules, on_change=None, **kwargs):
        assert pyinotify
        self._modules = modules
        self._on_change = on_change

    def start(self):
        try:
            self._wm = pyinotify.WatchManager()
            self._notifier = pyinotify.Notifier(self._wm)
            for m in self._modules:
                self._wm.add_watch(m, pyinotify.IN_MODIFY)
            self._notifier.loop()
        finally:
            self.close()

    def close(self):
        self._notifier.stop()
        self._wm.close()

    def process_IN_MODIFY(self, event):
        self.on_change(event.pathname)

    def on_change(self, modified):<commit_msg>Imporves InotifyMonitor event handling<commit_after># -*- coding: utf-8 -*-
"""
    celery.worker.autoreload
    ~~~~~~~~~~~~~~~~~~~~~~~~

    This module implements automatic module reloading
"""
from __future__ import absolute_import
from __future__ import with_statement

import errno
import hashlib
import os
import select
import sys
import time

from collections import defaultdict

from ..abstract import StartStopComponent
from ..utils.threads import bgThread, Event

try:
    import pyinotify
    _ProcessEvent = pyinotify.ProcessEvent
except ImportError:
    pyinotify = None        # noqa
    _ProcessEvent = object  # noqa


class WorkerComponent(StartStopComponent):
    name = "worker.autoreloader"
    requires = ("pool", )

    def __init__(self, w, autoreload=None, **kwargs):
        self.enabled = w.autoreload = autoreload
        w.autoreloader = None

    def create(self, w):
        w.autoreloader = self.instantiate(w.autoreloader_cls,
                                          controller=w,
                                          logger=w.logger)
        return w.autoreloader


def file_hash(filename, algorithm="md5"):
    hobj = hashlib.new(algorithm)
    with open(filename, "rb") as f:
        for chunk in iter(lambda: f.read(2 ** 20), ''):
            hobj.update(chunk)
    return hobj.digest()


class BaseMonitor(object):

    def __init__(self, files, on_change=None, shutdown_event=None,
            interval=0.5):
        self.files = files
        self.interval = interval
        self._on_change = on_change
        self.modify_times = defaultdict(int)
        self.shutdown_event = shutdown_event or Event()

    def start(self):
        raise NotImplementedError("Subclass responsibility")

    def stop(self):
        pass

    def on_change(self, modified):
        if self._on_change:
            return self._on_change(modified)


class StatMonitor(BaseMonitor):
    """File change monitor based on the ``stat`` system call."""

    def _mtimes(self):
        return ((f, self._mtime(f)) for f in self.files)

    def _maybe_modified(self, f, mt):
        return mt is not None and self.modify_times[f] != mt

    def start(self):
        while not self.shutdown_event.is_set():
            modified = dict((f, mt) for f, mt in self._mtimes()
                                if self._maybe_modified(f, mt))
            if modified:
                self.on_change(modified.keys())
                self.modify_times.update(modified)
            time.sleep(self.interval)

    @staticmethod
    def _mtime(path):
        try:
            return os.stat(path).st_mtime
        except Exception:
            pass


class KQueueMonitor(BaseMonitor):
    """File change monitor based on BSD kernel event notifications"""

    def __init__(self, *args, **kwargs):
        assert hasattr(select, "kqueue")
        super(KQueueMonitor, self).__init__(*args, **kwargs)
        self.filemap = dict((f, None) for f in self.files)

    def start(self):
        self._kq = select.kqueue()
        kevents = []
        for f in self.filemap:
            self.filemap[f] = fd = os.open(f, os.O_RDONLY)

            ev = select.kevent(fd,
                    filter=select.KQ_FILTER_VNODE,
                    flags=select.KQ_EV_ADD |
                            select.KQ_EV_ENABLE |
                            select.KQ_EV_CLEAR,
                    fflags=select.KQ_NOTE_WRITE |
                            select.KQ_NOTE_EXTEND)
            kevents.append(ev)

        events = self._kq.control(kevents, 0)
        while not self.shutdown_event.is_set():
            events = self._kq.control(kevents, 1)
            fds = [e.ident for e in events]
            modified = [k for k, v in self.filemap.iteritems()
                                        if v in fds]
            self.on_change(modified)

    def stop(self):
        self._kq.close()
        for fd in filter(None, self.filemap.values()):
            try:
                os.close(fd)
            except OSError, exc:
                if exc != errno.EBADF:
                    raise
            self.filemap[fd] = None
        self.filemap.clear()


class InotifyMonitor(_ProcessEvent):
    """File change monitor based on Linux kernel `inotify` subsystem"""

    def __init__(self, modules, on_change=None, **kwargs):
        assert pyinotify
        self._modules = modules
        self._on_change = on_change
        self._wm = None
        self._notifier = None

    def start(self):
        try:
            self._wm = pyinotify.WatchManager()
            self._notifier = pyinotify.Notifier(self._wm, self)
            for m in self._modules:
                self._wm.add_watch(m, pyinotify.IN_MODIFY|pyinotify.IN_ATTRIB)
            self._notifier.loop()
        finally:
            if self._wm:
                self._wm.close()
                # Notifier.close is called at the end of Notifier.loop
                self._wm = self._notifier = None

    def stop(self):
        pass

    def process_(self, event):
        self.on_change([event.path])

    process_IN_ATTRIB = process_IN_MODIFY = process_

    def on_change(self, modified):
{'fraction_non_alphanumeric': 0.0999003984063745, 'fraction_numerical': 0.0017928286852589642, 'mean_word_length': 1.9593280282935455, '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': 14, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4793898', 'n_tokens_mistral': 2970, 'n_tokens_neox': 2855, 'n_words': 708}