<commit_before>import asyncio
import fcntl
import os
import ctypes
import errno
import io


class AsyncFile:
    DEFAULT_BLOCK_SIZE = 8192

    def __init__(self, loop=None, filename=None,
                 fileobj=None, mode='rb'):
        if (filename is None and fileobj is None) or \
                (filename is not None and fileobj is not None):
            raise RuntimeError('Confilicting arguments')

        if filename is not None:
            if 'b' not in mode:
                raise RuntimeError('Only binary mode is supported')
            fileobj = open(filename, mode=mode)
        elif 'b' not in fileobj.mode:
            raise RuntimeError('Only binary mode is supported')

        fl = fcntl.fcntl(fileobj, fcntl.F_GETFL)
        if fcntl.fcntl(fileobj, fcntl.F_SETFL, fl | os.O_NONBLOCK) != 0:
            if filename is not None:
                fileobj.close()
            errcode = ctypes.get_errno()
            raise OSError((errcode, errno.errorcode[errcode]))

        self.fileobj = fileobj

        if loop is None:
            loop = asyncio.get_event_loop()
        self.loop = loop
        self.rbuffer = bytearray()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def seek(self, offset, whence=None):
        if whence is None:
            return self.fileobj.seek(offset)
        else:
            return self.fileobj.seek(offset, whence)

    def tell(self):
        return self.fileobj.tell()

    def read_ready(self, future, n, total):
        try:
            res = self.fileobj.read(n)
        except (BlockingIOError, InterruptedError):
            self.read_handle = \
                self.loop.call_soon(self.read_ready, future, n, total)
            return
        except Exception as exc:
            future.set_exception(exc)
            return

        if not res:     # EOF
            future.set_result(bytes(self.rbuffer))
            return

        self.rbuffer.extend(res)

        if total > 0:
            more_to_go = total - len(self.rbuffer)
            if more_to_go <= 0:  # enough
                res, self.rbuffer = self.rbuffer[:n], self.rbuffer[n:]
                future.set_result(bytes(res))
            else:
                more_to_go = min(self.DEFAULT_BLOCK_SIZE, more_to_go)
                self.read_handle = \
                    self.loop.call_soon(self.read_ready, future,
                                        more_to_go, total)
        else:   # total < 0
            self.read_handle = \
                self.loop.call_soon(self.read_ready, future,
                                    self.DEFAULT_BLOCK_SIZE, total)

    @asyncio.coroutine
    def read(self, n=-1):
        future = asyncio.Future(loop=self.loop)

        if n == 0:
            future.set_result(b'')
        else:
            try:
                res = self.fileobj.read(n)
            except (BlockingIOError, InterruptedError):
                if n < 0:
                    self.rbuffer.clear()
                    self.read_handle = \
                        self.loop.call_soon(self.read_ready, future,
                                            self.DEFAULT_BLOCK_SIZE, n)
                else:
                    self.rbuffer.clear()
                    read_block_size = min(self.DEFAULT_BLOCK_SIZE, n)
                    self.read_handle = \
                        self.loop.call_soon(self.read_ready, future,
                                            read_block_size, n)
            except Exception as exc:
                future.set_exception(exc)

            future.set_result(res)

        return future

    def write_ready(self, future, data, written):
        try:
            res = self.fileobj.write(data)
        except (BlockingIOError, InterruptedError):
            self.write_handle = \
                self.loop.call_soon(self.write_ready, future, data, written)
            return
        except Exception as exc:
            future.set_exception(exc)
            return

        if res < len(data):
            data = data[res:]
            self.write_handle = \
                self.loop.call_soon(self.write_ready, future,
                                    data, written + res)
        else:
            future.set_result(written + res)

    @asyncio.coroutine
    def write(self, data):
        future = asyncio.Future(loop=self.loop)

        if len(data) == 0:
            future.set_result(0)
        else:
            try:
                res = self.fileobj.write(data)
            except (BlockingIOError, InterruptedError):
                self.write_handle = \
                    self.loop.call_soon(self.write_ready, future, data, 0)
            except Exception as exc:
                future.set_exception(exc)

            future.set_result(res)

        return future

    def close(self):
        self.fileobj.close()
        if hasattr(self, 'read_handle'):
            self.read_handle.cancel()
        if hasattr(self, 'write_handle'):
            self.write_handle.cancel()


class BufferedMixin:
    def init_buffer(self):
        self._buffer = []

    def flush_buffer(self):
        buffered = b''.join(self._buffer)
        self._buffer = []
        return buffered

    def read_from_buffer(self, n):
        if n < 0:
            self._buffer.reverse()
            buffered = b''.join(self._buffer)
            self._buffer = []
            return (buffered, n)
        else:
            buffered = bytearray()
            while n > 0 and len(self._buffer) > 0:
                data = self._buffer.pop()
                if len(data) > n:
                    self._buffer.append(data[n:])
                    buffered.extend(data[0:n])
                    n = 0
                else:
                    buffered.extend(data)
                    n -= len(data)<commit_msg>Wait for IO in AsyncFile, instead of polling<commit_after>import asyncio
import fcntl
import os
import ctypes
import errno
import io


class AsyncFile:
    DEFAULT_BLOCK_SIZE = 8192

    def __init__(self, loop=None, filename=None,
                 fileobj=None, mode='rb'):
        if (filename is None and fileobj is None) or \
                (filename is not None and fileobj is not None):
            raise RuntimeError('Confilicting arguments')

        if filename is not None:
            if 'b' not in mode:
                raise RuntimeError('Only binary mode is supported')
            fileobj = open(filename, mode=mode)
        elif 'b' not in fileobj.mode:
            raise RuntimeError('Only binary mode is supported')

        fl = fcntl.fcntl(fileobj, fcntl.F_GETFL)
        if fcntl.fcntl(fileobj, fcntl.F_SETFL, fl | os.O_NONBLOCK) != 0:
            if filename is not None:
                fileobj.close()
            errcode = ctypes.get_errno()
            raise OSError((errcode, errno.errorcode[errcode]))

        self.fileobj = fileobj

        if loop is None:
            loop = asyncio.get_event_loop()
        self.loop = loop
        self.rbuffer = bytearray()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_value, traceback):
        self.close()

    def seek(self, offset, whence=None):
        if whence is None:
            return self.fileobj.seek(offset)
        else:
            return self.fileobj.seek(offset, whence)

    def tell(self):
        return self.fileobj.tell()

    def read_ready(self, future, n, total):
        try:
            res = self.fileobj.read(n)
        except (BlockingIOError, InterruptedError):
            self.loop.add_reader(self.fileobj.fileno(),
                                 self.read_ready,
                                 future, n, total)
            return
        except Exception as exc:
            future.set_exception(exc)
            return

        if not res:     # EOF
            future.set_result(bytes(self.rbuffer))
            return

        self.rbuffer.extend(res)

        if total > 0:
            more_to_go = total - len(self.rbuffer)
            if more_to_go <= 0:  # enough
                res, self.rbuffer = self.rbuffer[:n], self.rbuffer[n:]
                future.set_result(bytes(res))
            else:
                more_to_go = min(self.DEFAULT_BLOCK_SIZE, more_to_go)
                self.loop.add_reader(self.fileobj.fileno(),
                                     self.read_ready,
                                     future, more_to_go, total)
        else:   # total < 0
            self.loop.add_reader(self.fileobj.fileno(),
                                 self.read_ready,
                                 future, self.DEFAULT_BLOCK_SIZE, total)

    @asyncio.coroutine
    def read(self, n=-1):
        future = asyncio.Future(loop=self.loop)

        if n == 0:
            future.set_result(b'')
        else:
            try:
                res = self.fileobj.read(n)
            except (BlockingIOError, InterruptedError):
                if n < 0:
                    self.rbuffer.clear()
                    self.loop.add_reader(self.fileobj.fileno(),
                                         self.read_ready,
                                         future, self.DEFAULT_BLOCK_SIZE, n)
                else:
                    self.rbuffer.clear()
                    read_block_size = min(self.DEFAULT_BLOCK_SIZE, n)
                    self.loop.add_reader(self.fileobj.fileno(),
                                         self.read_ready,
                                         future, read_block_size, n)
            except Exception as exc:
                future.set_exception(exc)

            future.set_result(res)

        return future

    def write_ready(self, future, data, written):
        try:
            res = self.fileobj.write(data)
        except (BlockingIOError, InterruptedError):
            self.loop.add_writer(self.fileobj.fileno(),
                                 self.write_ready,
                                 future, data, written)
            return
        except Exception as exc:
            future.set_exception(exc)
            return

        if res < len(data):
            data = data[res:]
            self.loop.add_writer(self.fileobj.fileno(),
                                 self.write_ready,
                                 future, data, written + res)
        else:
            future.set_result(written + res)

    @asyncio.coroutine
    def write(self, data):
        future = asyncio.Future(loop=self.loop)

        if len(data) == 0:
            future.set_result(0)
        else:
            try:
                res = self.fileobj.write(data)
            except (BlockingIOError, InterruptedError):
                self.loop.add_writer(self.fileobj.fileno(),
                                     self.write_ready,
                                     future, data, 0)
            except Exception as exc:
                future.set_exception(exc)

            future.set_result(res)

        return future

    def close(self):
        self.loop.remove_reader(self.fileobj.fileno())
        self.loop.remove_writer(self.fileobj.fileno())
        self.fileobj.close()


class BufferedMixin:
    def init_buffer(self):
        self._buffer = []

    def flush_buffer(self):
        buffered = b''.join(self._buffer)
        self._buffer = []
        return buffered

    def read_from_buffer(self, n):
        if n < 0:
            self._buffer.reverse()
            buffered = b''.join(self._buffer)
            self._buffer = []
            return (buffered, n)
        else:
            buffered = bytearray()
            while n > 0 and len(self._buffer) > 0:
                data = self._buffer.pop()
                if len(data) > n:
                    self._buffer.append(data[n:])
                    buffered.extend(data[0:n])
                    n = 0
                else:
                    buffered.extend(data)
                    n -= len(data)
{'fraction_non_alphanumeric': 0.09005016722408027, 'fraction_numerical': 0.003177257525083612, 'mean_word_length': 1.3912435025989605, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 27, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2577185', 'n_tokens_mistral': 3392, 'n_tokens_neox': 3199, 'n_words': 728}