<commit_before>import contextlib
import errno
import fcntl
import os
import socket
import stat
import weakref
try:
    from cStringIO import StringIO
except ImportError: #pragma: no cover
    from StringIO import StringIO

from greenhouse import utils
from greenhouse._state import state


__all__ = ["Socket", "File", "monkeypatch", "unmonkeypatch", "pipe"]


_socket = socket.socket
_open = __builtins__['open']
_file = __builtins__['file']

SOCKET_CLOSED = set((errno.ECONNRESET, errno.ENOTCONN, errno.ESHUTDOWN))

def monkeypatch():
    """replace functions in the standard library socket module
    with their non-blocking greenhouse equivalents"""
    socket.socket = Socket
    __builtins__['open'] = __builtins__['file'] = File

def unmonkeypatch():
    "undo a call to monkeypatch()"
    socket.socket = _socket
    __builtins__['open'] = _open
    __builtins__['file'] = _file

<EMAIL>
class Socket(object):
    def __init__(self, *args, **kwargs):
        # wrap a basic socket or build our own
        self._sock = kwargs.pop('fromsock', None)
        if isinstance(self._sock, Socket):
            self._sock = self._sock._sock
        if not self._sock:
            self._sock = _socket(*args, **kwargs)

        # copy over attributes
        self.family = self._sock.family
        self.type = self._sock.type
        self.proto = self._sock.proto
        self._fileno = self._sock.fileno()

        # make the underlying socket non-blocking
        self.setblocking(False)

        # create events
        self._readable = utils.Event()
        self._writable = utils.Event()

        # some more housekeeping
        self._timeout = None
        self._closed = False

        # make sure we have a state.poller
        if not hasattr(state, 'poller'):
            import greenhouse.poller #pragma: no cover

        # allow for lookup by fileno
        state.descriptormap[self._fileno].append(weakref.ref(self))

    def __del__(self):
        try:
            state.poller.unregister(self)
        except:
            pass

    @contextlib.contextmanager
    def _registered(self):
        try:
            state.poller.register(self)
        except (IOError, OSError), error: #pragma: no cover
            if error.args and error.args[0] in errno.errorcode:
                raise socket.error(*error.args)
            raise

        yield

        try:
            state.poller.unregister(self)
        except (IOError, OSError), error: #pragma: no cover
            if error.args and error.args[0] in errno.errorcode:
                raise socket.error(*error.args)
            raise

    def accept(self):
        with self._registered():
            while 1:
                try:
                    client, addr = self._sock.accept()
                except socket.error, err:
                    if err[0] in (errno.EAGAIN, errno.EWOULDBLOCK):
                        self._readable.wait()
                        continue
                    else:
                        raise #pragma: no cover
                return type(self)(fromsock=client), addr

    def bind(self, *args, **kwargs):
        return self._sock.bind(*args, **kwargs)

    def close(self):
        self._closed = True
        return self._sock.close()

    def connect(self, address):
        with self._registered():
            while True:
                err = self.connect_ex(address)
                if err in (errno.EINPROGRESS, errno.EALREADY,
                        errno.EWOULDBLOCK):
                    self._writable.wait()
                    continue
                if err not in (0, errno.EISCONN): #pragma: no cover
                    raise socket.error(err, errno.errorcode[err])
                return

    def connect_ex(self, address):
        return self._sock.connect_ex(address)

    def dup(self):
        return type(self)(fromsock=self._sock.dup())

    def fileno(self):
        return self._fileno

    def getpeername(self):
        return self._sock.getpeername()

    def getsockname(self):
        return self._sock.getsockname()

    def getsockopt(self, *args):
        return self._sock.getsockopt(*args)

    def gettimeout(self):
        return self._timeout

    def listen(self, backlog):
        return self._sock.listen(backlog)

    def makefile(self, mode='r', bufsize=None):
        return socket._fileobject(self)

    def recv(self, nbytes):
        with self._registered():
            while 1:
                if self._closed:
                    raise socket.error(errno.EBADF, "Bad file descriptor")
                try:
                    return self._sock.recv(nbytes)
                except socket.error, e:
                    if e[0] == errno.EWOULDBLOCK: #pragma: no cover
                        self._readable.wait()
                        continue
                    if e[0] in SOCKET_CLOSED:
                        self._closed = True
                        return ''
                    raise #pragma: no cover

    def recv_into(self, buffer, nbytes):
        with self._registered():
            self._readable.wait()
            return self._sock.recv_into(buffer, nbytes)

    def recvfrom(self, nbytes):
        with self._registered():
            self._readable.wait()
            return self._sock.recvfrom(nbytes)

    def recvfrom_into(self, buffer, nbytes):
        with self._registered():
            self._readable.wait()
            return self._sock.recvfrom_into(buffer, nbytes)

    def send(self, data):
        with self._registered():
            try:
                return self._sock.send(data)
            except socket.error, err: #pragma: no cover
                if err[0] in (errno.EWOULDBLOCK, errno.ENOTCONN):
                    return 0
                raise

    def sendall(self, data):
        sent = self.send(data)
        while sent < len(data): #pragma: no cover
            self._writable.wait()
            sent += self.send(data[sent:])

    def sendto(self, *args):
        try:
            return self._sock.sendto(*args)
        except socket.error, err: #pragma: no cover<commit_msg>when we register socket descriptors with pollers, mask the events we don't need<commit_after>import contextlib
import errno
import fcntl
import os
import socket
import stat
import weakref
try:
    from cStringIO import StringIO
except ImportError: #pragma: no cover
    from StringIO import StringIO

from greenhouse import utils
from greenhouse._state import state


__all__ = ["Socket", "File", "monkeypatch", "unmonkeypatch", "pipe"]


_socket = socket.socket
_open = __builtins__['open']
_file = __builtins__['file']

SOCKET_CLOSED = set((errno.ECONNRESET, errno.ENOTCONN, errno.ESHUTDOWN))

def monkeypatch():
    """replace functions in the standard library socket module
    with their non-blocking greenhouse equivalents"""
    socket.socket = Socket
    __builtins__['open'] = __builtins__['file'] = File

def unmonkeypatch():
    "undo a call to monkeypatch()"
    socket.socket = _socket
    __builtins__['open'] = _open
    __builtins__['file'] = _file

#<EMAIL>debugger
class Socket(object):
    def __init__(self, *args, **kwargs):
        # wrap a basic socket or build our own
        self._sock = kwargs.pop('fromsock', None)
        if isinstance(self._sock, Socket):
            self._sock = self._sock._sock
        if not self._sock:
            self._sock = _socket(*args, **kwargs)

        # copy over attributes
        self.family = self._sock.family
        self.type = self._sock.type
        self.proto = self._sock.proto
        self._fileno = self._sock.fileno()

        # make the underlying socket non-blocking
        self.setblocking(False)

        # create events
        self._readable = utils.Event()
        self._writable = utils.Event()

        # some more housekeeping
        self._timeout = None
        self._closed = False

        # make sure we have a state.poller
        if not hasattr(state, 'poller'):
            import greenhouse.poller #pragma: no cover

        # allow for lookup by fileno
        state.descriptormap[self._fileno].append(weakref.ref(self))

    def __del__(self):
        try:
            state.poller.unregister(self)
        except:
            pass

    @contextlib.contextmanager
    def _registered(self, events=None):
        poller = state.poller
        mask = None
        if events:
            mask = 0
            if 'r' in events:
                mask |= poller.INMASK
            if 'w' in events:
                mask |= poller.OUTMASK
            if 'e' in events:
                mask |= poller.ERRMASK
        try:
            poller.register(self, mask)
        except (IOError, OSError), error: #pragma: no cover
            if error.args and error.args[0] in errno.errorcode:
                raise socket.error(*error.args)
            raise

        yield

        try:
            poller.unregister(self)
        except (IOError, OSError), error: #pragma: no cover
            if error.args and error.args[0] in errno.errorcode:
                raise socket.error(*error.args)
            raise

    def accept(self):
        with self._registered('r'):
            while 1:
                try:
                    client, addr = self._sock.accept()
                except socket.error, err:
                    if err[0] in (errno.EAGAIN, errno.EWOULDBLOCK):
                        self._readable.wait()
                        continue
                    else:
                        raise #pragma: no cover
                return type(self)(fromsock=client), addr

    def bind(self, *args, **kwargs):
        return self._sock.bind(*args, **kwargs)

    def close(self):
        self._closed = True
        return self._sock.close()

    def connect(self, address):
        with self._registered('w'):
            while True:
                err = self.connect_ex(address)
                if err in (errno.EINPROGRESS, errno.EALREADY,
                        errno.EWOULDBLOCK):
                    self._writable.wait()
                    continue
                if err not in (0, errno.EISCONN): #pragma: no cover
                    raise socket.error(err, errno.errorcode[err])
                return

    def connect_ex(self, address):
        return self._sock.connect_ex(address)

    def dup(self):
        return type(self)(fromsock=self._sock.dup())

    def fileno(self):
        return self._fileno

    def getpeername(self):
        return self._sock.getpeername()

    def getsockname(self):
        return self._sock.getsockname()

    def getsockopt(self, *args):
        return self._sock.getsockopt(*args)

    def gettimeout(self):
        return self._timeout

    def listen(self, backlog):
        return self._sock.listen(backlog)

    def makefile(self, mode='r', bufsize=None):
        return socket._fileobject(self)

    def recv(self, nbytes):
        with self._registered('r'):
            while 1:
                if self._closed:
                    raise socket.error(errno.EBADF, "Bad file descriptor")
                try:
                    return self._sock.recv(nbytes)
                except socket.error, e:
                    if e[0] == errno.EWOULDBLOCK: #pragma: no cover
                        self._readable.wait()
                        continue
                    if e[0] in SOCKET_CLOSED:
                        self._closed = True
                        return ''
                    raise #pragma: no cover

    def recv_into(self, buffer, nbytes):
        with self._registered('r'):
            self._readable.wait()
            return self._sock.recv_into(buffer, nbytes)

    def recvfrom(self, nbytes):
        with self._registered('r'):
            self._readable.wait()
            return self._sock.recvfrom(nbytes)

    def recvfrom_into(self, buffer, nbytes):
        with self._registered('r'):
            self._readable.wait()
            return self._sock.recvfrom_into(buffer, nbytes)

    def send(self, data):
        try:
            return self._sock.send(data)
        except socket.error, err: #pragma: no cover
            if err[0] in (errno.EWOULDBLOCK, errno.ENOTCONN):
                return 0
            raise

    def sendall(self, data):
        with self._registered('w'):
            sent = self.send(data)
            while sent < len(data): #pragma: no cover
                self._writable.wait()
                sent += self.send(data[sent:])

    def sendto(self, *args):
        try:
            return self._sock.sendto(*args)
        except socket.error, err: #pragma: no cover
{'fraction_non_alphanumeric': 0.09752801844050553, 'fraction_numerical': 0.0016691836896908037, 'mean_word_length': 1.8003561095036724, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 5, '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': '606404', 'n_tokens_mistral': 3620, 'n_tokens_neox': 3388, 'n_words': 915}