<commit_before>from __future__ import print_function

import codecs
import os

from leapp.compat import string_types
from leapp.libraries.stdlib.eventloop import POLL_HUP, POLL_IN, POLL_OUT, POLL_PRI, EventLoop

STDIN = 0
STDOUT = 1
STDERR = 2


def _multiplex(ep, read_fds, callback_raw, callback_linebuffered,
               encoding='utf-8', write=None, timeout=1, buffer_size=80):
    # Register the file descriptors (stdout + stderr) with the epoll object
    # so that we'll get notifications when data are ready to read
    for fd in read_fds:
        ep.register(fd, POLL_IN | POLL_PRI)

    # Register a write file descriptor
    if write:
        ep.register(write[0], POLL_OUT)

    # Offset into the `write[1]` buffer where we should continue writing to stdin
    offset = 0

    # We need to keep track of which file descriptors have already been drained
    # because when running under `pytest` it seems that all `epoll` events are
    # received twice so using solely `ep.unregister(fd)` will not work
    hupped = set()
    # Total number of 'hupped' file descriptors we expect
    num_expected = len(read_fds) + (1 if write else 0)
    # Set up file-descriptor specific buffers where we'll buffer the output
    buf = {fd: bytes() for fd in read_fds}
    if encoding:
        linebufs = {fd: '' for fd in read_fds}
        decoders = {fd: codecs.getincrementaldecoder(encoding)() for fd in read_fds}

    def _get_fd_type(fd):
        """
        File descriptors passed via `read_fds` are always representing [stdout, stderr],
        since arrays start at index 0, we need to add 1 to get the real symbolic value
        `STDOUT` or `STDERR`.
        """
        return read_fds.index(fd) + 1

    while not ep.closed and len(hupped) != num_expected:
        events = ep.poll(timeout)
        for fd, event in events:
            if event == POLL_HUP:
                hupped.add(fd)
                ep.unregister(fd)
            if event & (POLL_IN | POLL_PRI) != 0:
                fd_type = _get_fd_type(fd)
                read = os.read(fd, buffer_size)
                callback_raw((fd, fd_type), read)
                if encoding:
                    linebufs[fd] += decoders[fd].decode(read)
                    while '\n' in linebufs[fd]:
                        pre, post = linebufs[fd].split('\n', 1)
                        linebufs[fd] = post
                        callback_linebuffered((fd, fd_type), pre)
                buf[fd] += read
            elif event == POLL_OUT:
                # Write data to pipe, `os.write` returns the number of bytes written,
                # thus we need to offset
                wfd, data = write
                if fd in hupped:
                    continue
                offset += os.write(fd, data[offset:])
                if offset == len(data):
                    os.close(fd)
                    hupped.add(fd)
                    ep.unregister(fd)

    # Process leftovers from line buffering
    if encoding:
        for (fd, lb) in linebufs.items():
            if lb:
                # [stdout, stderr] is relayed, stdout=1 a stderr=2
                # as the field starting indexed is 0, so the +1 needs to be added<commit_msg>Fix the call function from the stdlib library on Python 3.9<commit_after>from __future__ import print_function

import codecs
import os

from leapp.compat import string_types
from leapp.libraries.stdlib.eventloop import POLL_HUP, POLL_IN, POLL_OUT, POLL_PRI, EventLoop

STDIN = 0
STDOUT = 1
STDERR = 2


def _multiplex(ep, read_fds, callback_raw, callback_linebuffered,
               encoding='utf-8', write=None, timeout=1, buffer_size=80):
    # Register the file descriptors (stdout + stderr) with the epoll object
    # so that we'll get notifications when data are ready to read
    for fd in read_fds:
        ep.register(fd, POLL_IN | POLL_PRI)

    # Register a write file descriptor
    if write:
        ep.register(write[0], POLL_OUT)

    # Offset into the `write[1]` buffer where we should continue writing to stdin
    offset = 0

    # We need to keep track of which file descriptors have already been drained
    # because when running under `pytest` it seems that all `epoll` events are
    # received twice so using solely `ep.unregister(fd)` will not work
    hupped = set()
    # Total number of 'hupped' file descriptors we expect
    num_expected = len(read_fds) + (1 if write else 0)
    # Set up file-descriptor specific buffers where we'll buffer the output
    buf = {fd: bytes() for fd in read_fds}
    if encoding:
        linebufs = {fd: '' for fd in read_fds}
        decoders = {fd: codecs.getincrementaldecoder(encoding)() for fd in read_fds}

    def _get_fd_type(fd):
        """
        File descriptors passed via `read_fds` are always representing [stdout, stderr],
        since arrays start at index 0, we need to add 1 to get the real symbolic value
        `STDOUT` or `STDERR`.
        """
        return read_fds.index(fd) + 1

    while not ep.closed and len(hupped) != num_expected:
        events = ep.poll(timeout)
        for fd, event in events:
            if event == POLL_HUP:
                hupped.add(fd)
                ep.unregister(fd)
            if event & (POLL_IN | POLL_PRI) != 0:
                fd_type = _get_fd_type(fd)
                read = os.read(fd, buffer_size)
                callback_raw((fd, fd_type), read)
                if encoding:
                    linebufs[fd] += decoders[fd].decode(read)
                    while '\n' in linebufs[fd]:
                        pre, post = linebufs[fd].split('\n', 1)
                        linebufs[fd] = post
                        callback_linebuffered((fd, fd_type), pre)
                buf[fd] += read
            elif event == POLL_OUT:
                # Write data to pipe, `os.write` returns the number of bytes written,
                # thus we need to offset
                wfd, data = write
                if fd in hupped:
                    continue
                offset += os.write(fd, data[offset:])
                if offset == len(data):
                    hupped.add(fd)
                    ep.unregister(fd)
                    os.close(fd)

    # Process leftovers from line buffering
    if encoding:
        for (fd, lb) in linebufs.items():
            if lb:
                # [stdout, stderr] is relayed, stdout=1 a stderr=2
                # as the field starting indexed is 0, so the +1 needs to be added
{'fraction_non_alphanumeric': 0.08323038912909203, 'fraction_numerical': 0.0067943174799258805, 'mean_word_length': 1.9629460201280877, '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': '2205308', 'n_tokens_mistral': 1995, 'n_tokens_neox': 1861, 'n_words': 659}