<commit_before>"""Base implementation of event loop.

The event loop can be broken up into a multiplexer (the part
responsible for notifying us of IO events) and the event loop proper,
which wraps a multiplexer 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 heapq
import inspect
import logging
import os
import socket
import subprocess
import time
import traceback
import sys

from . import coroutines
from . import events
from . import futures
from . import tasks
from .coroutines import coroutine
from .log import logger


__all__ = ['BaseEventLoop', 'Server']


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


def _format_handle(handle):
    cb = handle._callback
    if inspect.ismethod(cb) and isinstance(cb.__self__, tasks.Task):
        # format the task
        return repr(cb.__self__)
    else:
        return str(handle)


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


def _check_resolved_address(sock, address):
    # Ensure that the address is already resolved to avoid the trap of hanging
    # the entire event loop when the address requires doing a DNS lookup.
    family = sock.family
    if family == socket.AF_INET:
        host, port = address
    elif family == socket.AF_INET6:
        host, port = address[:2]
    else:
        return

    type_mask = 0
    if hasattr(socket, 'SOCK_NONBLOCK'):
        type_mask |= socket.SOCK_NONBLOCK
    if hasattr(socket, 'SOCK_CLOEXEC'):
        type_mask |= socket.SOCK_CLOEXEC
    # Use getaddrinfo(AI_NUMERICHOST) to ensure that the address is
    # already resolved.
    try:
        socket.getaddrinfo(host, port,
                           family=family,
                           type=(sock.type & ~type_mask),
                           proto=sock.proto,
                           flags=socket.AI_NUMERICHOST)
    except socket.gaierror as err:
        raise ValueError("address must be resolved (IP address), got %r: %s"
                         % (address, err))

def _raise_stop_error(*args):
    raise _StopError


class Server(events.AbstractServer):

    def __init__(self, loop, sockets):
        self.loop = loop
        self.sockets = sockets
        self.active_count = 0
        self.waiters = []

    def attach(self, transport):
        assert self.sockets is not None
        self.active_count += 1

    def detach(self, transport):
        assert self.active_count > 0
        self.active_count -= 1
        if self.active_count == 0 and self.sockets is None:
            self._wakeup()

    def close(self):
        sockets = self.sockets
        if sockets is not None:
            self.sockets = None
            for sock in sockets:
                self.loop._stop_serving(sock)
            if self.active_count == 0:
                self._wakeup()

    def _wakeup(self):
        waiters = self.waiters
        self.waiters = None
        for waiter in waiters:
            if not waiter.done():
                waiter.set_result(waiter)

    @coroutine
    def wait_closed(self):
        if self.sockets is None or self.waiters is None:
            return
        waiter = futures.Future(loop=self.loop)
        self.waiters.append(waiter)
        yield from waiter


class BaseEventLoop(events.AbstractEventLoop):

    def __init__(self):
        self._closed = False
        self._ready = collections.deque()
        self._scheduled = []
        self._default_executor = None
        self._internal_fds = 0
        self._running = False
        self._clock_resolution = time.get_clock_info('monotonic').resolution
        self._exception_handler = None
        self._debug = (not sys.flags.ignore_environment
                       and bool(os.environ.get('PYTHONASYNCIODEBUG')))
        # In debug mode, if the execution of a callback or a step of a task
        # exceed this duration in seconds, the slow callback/task is logged.
        self.slow_callback_duration = 0.1

    def __repr__(self):
        return ('<%s running=%s closed=%s debug=%s>'
                % (self.__class__.__name__, self.is_running(),
                   self.is_closed(), self.get_debug()))

    def create_task(self, coro):
        """Schedule a coroutine object.

        Return a task object."""
        return tasks.Task(coro, loop=self)

    def _make_socket_transport(self, sock, protocol, waiter=None, *,<commit_msg>Fix create_task(): truncate the traceback to hide the call to create_task()<commit_after>"""Base implementation of event loop.

The event loop can be broken up into a multiplexer (the part
responsible for notifying us of IO events) and the event loop proper,
which wraps a multiplexer 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 heapq
import inspect
import logging
import os
import socket
import subprocess
import time
import traceback
import sys

from . import coroutines
from . import events
from . import futures
from . import tasks
from .coroutines import coroutine
from .log import logger


__all__ = ['BaseEventLoop', 'Server']


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


def _format_handle(handle):
    cb = handle._callback
    if inspect.ismethod(cb) and isinstance(cb.__self__, tasks.Task):
        # format the task
        return repr(cb.__self__)
    else:
        return str(handle)


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


def _check_resolved_address(sock, address):
    # Ensure that the address is already resolved to avoid the trap of hanging
    # the entire event loop when the address requires doing a DNS lookup.
    family = sock.family
    if family == socket.AF_INET:
        host, port = address
    elif family == socket.AF_INET6:
        host, port = address[:2]
    else:
        return

    type_mask = 0
    if hasattr(socket, 'SOCK_NONBLOCK'):
        type_mask |= socket.SOCK_NONBLOCK
    if hasattr(socket, 'SOCK_CLOEXEC'):
        type_mask |= socket.SOCK_CLOEXEC
    # Use getaddrinfo(AI_NUMERICHOST) to ensure that the address is
    # already resolved.
    try:
        socket.getaddrinfo(host, port,
                           family=family,
                           type=(sock.type & ~type_mask),
                           proto=sock.proto,
                           flags=socket.AI_NUMERICHOST)
    except socket.gaierror as err:
        raise ValueError("address must be resolved (IP address), got %r: %s"
                         % (address, err))

def _raise_stop_error(*args):
    raise _StopError


class Server(events.AbstractServer):

    def __init__(self, loop, sockets):
        self.loop = loop
        self.sockets = sockets
        self.active_count = 0
        self.waiters = []

    def attach(self, transport):
        assert self.sockets is not None
        self.active_count += 1

    def detach(self, transport):
        assert self.active_count > 0
        self.active_count -= 1
        if self.active_count == 0 and self.sockets is None:
            self._wakeup()

    def close(self):
        sockets = self.sockets
        if sockets is not None:
            self.sockets = None
            for sock in sockets:
                self.loop._stop_serving(sock)
            if self.active_count == 0:
                self._wakeup()

    def _wakeup(self):
        waiters = self.waiters
        self.waiters = None
        for waiter in waiters:
            if not waiter.done():
                waiter.set_result(waiter)

    @coroutine
    def wait_closed(self):
        if self.sockets is None or self.waiters is None:
            return
        waiter = futures.Future(loop=self.loop)
        self.waiters.append(waiter)
        yield from waiter


class BaseEventLoop(events.AbstractEventLoop):

    def __init__(self):
        self._closed = False
        self._ready = collections.deque()
        self._scheduled = []
        self._default_executor = None
        self._internal_fds = 0
        self._running = False
        self._clock_resolution = time.get_clock_info('monotonic').resolution
        self._exception_handler = None
        self._debug = (not sys.flags.ignore_environment
                       and bool(os.environ.get('PYTHONASYNCIODEBUG')))
        # In debug mode, if the execution of a callback or a step of a task
        # exceed this duration in seconds, the slow callback/task is logged.
        self.slow_callback_duration = 0.1

    def __repr__(self):
        return ('<%s running=%s closed=%s debug=%s>'
                % (self.__class__.__name__, self.is_running(),
                   self.is_closed(), self.get_debug()))

    def create_task(self, coro):
        """Schedule a coroutine object.

        Return a task object."""
        task = tasks.Task(coro, loop=self)
        if task._source_traceback:
            del task._source_traceback[-1]
        return task

    def _make_socket_transport(self, sock, protocol, waiter=None, *,
{'fraction_non_alphanumeric': 0.07668836452400325, 'fraction_numerical': 0.0027461350691619204, 'mean_word_length': 2.4369101712687873, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, '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': '4701309', 'n_tokens_mistral': 2764, 'n_tokens_neox': 2641, 'n_words': 918}