<commit_before>"""UNIX event loop and related classes.

The event loop can be broken up into a selector (the part responsible
for telling us when file descriptors are ready) and the event loop
proper, which wraps a selector with functionality for scheduling
callbacks, immediately or at a given time in the future.

Whenever a public API takes a callback, subsequent positional
arguments will be passed to the callback if/when it is called.  This
avoids the proliferation of trivial lambdas implementing closures.
Keyword arguments for the callback are not supported; this is a
conscious design decision, leaving the door open for keyword arguments
to modify the meaning of the API call itself.
"""

import collections
import concurrent.futures
import errno
import heapq
import logging
import select
import socket
try:
    import ssl
except ImportError:
    ssl = None
import sys
import threading
import time

try:
    import signal
except ImportError:
    signal = None

from . import events
from . import futures
from . import protocols
from . import selectors
from . import tasks
from . import transports

try:
    from socket import socketpair
except ImportError:
    assert sys.platform == 'win32'
    from .winsocketpair import socketpair

# Errno values indicating the connection was disconnected.
_DISCONNECTED = frozenset((errno.ECONNRESET,
                           errno.ENOTCONN,
                           errno.ESHUTDOWN,
                           errno.ECONNABORTED,
                           errno.EPIPE,
                           errno.EBADF,
                           ))

# Errno values indicating the socket isn't ready for I/O just yet.
_TRYAGAIN = frozenset((errno.EAGAIN, errno.EWOULDBLOCK, errno.EINPROGRESS))
if sys.platform == 'win32':
    _TRYAGAIN = frozenset(list(_TRYAGAIN) + [errno.WSAEWOULDBLOCK])

# Argument for default thread pool executor creation.
_MAX_WORKERS = 5


class _StopError(BaseException):
    """Raised to stop the event loop."""


def _raise_stop_error():
    raise _StopError


class UnixEventLoop(events.EventLoop):
    """Unix event loop.

    See events.EventLoop for API specification.
    """

    def __init__(self, selector=None):
        super().__init__()
        if selector is None:
            # pick the best selector class for the platform
            selector = selectors.Selector()
            logging.info('Using selector: %s', selector.__class__.__name__)
        self._selector = selector
        self._ready = collections.deque()
        self._scheduled = []
        self._everytime = []
        self._default_executor = None
        self._signal_handlers = {}
        self._make_self_pipe()

    def close(self):
        if self._selector is not None:
            self._selector.close()
            self._selector = None

    def _make_self_pipe(self):
        # A self-socket, really. :-)
        self._ssock, self._csock = socketpair()
        self._ssock.setblocking(False)
        self._csock.setblocking(False)
        self.add_reader(self._ssock.fileno(), self._read_from_self)

    def _read_from_self(self):
        try:<commit_msg>Close ssock/csock FDs upon event loop close().<commit_after>"""UNIX event loop and related classes.

The event loop can be broken up into a selector (the part responsible
for telling us when file descriptors are ready) and the event loop
proper, which wraps a selector with functionality for scheduling
callbacks, immediately or at a given time in the future.

Whenever a public API takes a callback, subsequent positional
arguments will be passed to the callback if/when it is called.  This
avoids the proliferation of trivial lambdas implementing closures.
Keyword arguments for the callback are not supported; this is a
conscious design decision, leaving the door open for keyword arguments
to modify the meaning of the API call itself.
"""

import collections
import concurrent.futures
import errno
import heapq
import logging
import select
import socket
try:
    import ssl
except ImportError:
    ssl = None
import sys
import threading
import time

try:
    import signal
except ImportError:
    signal = None

from . import events
from . import futures
from . import protocols
from . import selectors
from . import tasks
from . import transports

try:
    from socket import socketpair
except ImportError:
    assert sys.platform == 'win32'
    from .winsocketpair import socketpair

# Errno values indicating the connection was disconnected.
_DISCONNECTED = frozenset((errno.ECONNRESET,
                           errno.ENOTCONN,
                           errno.ESHUTDOWN,
                           errno.ECONNABORTED,
                           errno.EPIPE,
                           errno.EBADF,
                           ))

# Errno values indicating the socket isn't ready for I/O just yet.
_TRYAGAIN = frozenset((errno.EAGAIN, errno.EWOULDBLOCK, errno.EINPROGRESS))
if sys.platform == 'win32':
    _TRYAGAIN = frozenset(list(_TRYAGAIN) + [errno.WSAEWOULDBLOCK])

# Argument for default thread pool executor creation.
_MAX_WORKERS = 5


class _StopError(BaseException):
    """Raised to stop the event loop."""


def _raise_stop_error():
    raise _StopError


class UnixEventLoop(events.EventLoop):
    """Unix event loop.

    See events.EventLoop for API specification.
    """

    def __init__(self, selector=None):
        super().__init__()
        if selector is None:
            # pick the best selector class for the platform
            selector = selectors.Selector()
            logging.debug('Using selector: %s', selector.__class__.__name__)
        self._selector = selector
        self._ready = collections.deque()
        self._scheduled = []
        self._everytime = []
        self._default_executor = None
        self._signal_handlers = {}
        self._make_self_pipe()

    def close(self):
        if self._selector is not None:
            self._selector.close()
            self._selector = None
        self._ssock.close()
        self._csock.close()

    def _make_self_pipe(self):
        # A self-socket, really. :-)
        self._ssock, self._csock = socketpair()
        self._ssock.setblocking(False)
        self._csock.setblocking(False)
        self.add_reader(self._ssock.fileno(), self._read_from_self)

    def _read_from_self(self):
        try:
{'fraction_non_alphanumeric': 0.07088607594936709, 'fraction_numerical': 0.0015822784810126582, 'mean_word_length': 2.7073313782991204, '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': '7446909', 'n_tokens_mistral': 1760, 'n_tokens_neox': 1670, 'n_words': 603}