<commit_before>"""
Connection pooling for Cassandra connections.

"""

import time, threading, random

import connection
import queue as pool_queue
from logging.pool_logger import PoolLogger
from util import as_interface
from cassandra.ttypes import TimedOutException, UnavailableException
from thrift import Thrift

import threading
import socket

_BASE_BACKOFF = 0.01

__all__ = ['QueuePool', 'ConnectionPool', 'PoolListener',
           'ConnectionWrapper', 'AllServersUnavailable',
           'MaximumRetryException', 'NoConnectionAvailable',
           'InvalidRequestError']

class NoServerAvailable(Exception):
    """Raised if all servers are currently marked dead."""
    pass


class AbstractPool(object):
    """An abstract base class for all other pools."""

    def __init__(self, keyspace, server_list, credentials, timeout,
                 logging_name, use_threadlocal, listeners, framed_transport):
        """An abstract base class for all other pools."""

        if logging_name:
            self.logging_name = self._orig_logging_name = logging_name
        else:
            self._orig_logging_name = None
            self.logging_name = id(self)

        self._pool_threadlocal = use_threadlocal
        self.keyspace = keyspace
        self.credentials = credentials
        self.timeout = timeout
        self.framed_transport = framed_transport
        if use_threadlocal:
            self._tlocal = threading.local()

        # Listener groups
        self.listeners = []
        self._on_connect = []
        self._on_checkout = []
        self._on_checkin = []
        self._on_dispose = []
        self._on_recycle = []
        self._on_failure = []
        self._on_server_list = []
        self._on_pool_recreate = []
        self._on_pool_dispose = []
        self._on_pool_max = []

        self.add_listener(PoolLogger())

        for l in listeners:
            self.add_listener(l)

        self.set_server_list(server_list)

    def set_server_list(self, server_list):
        """
        Sets the server list that the pool will make connections to.

        `server_list` should be sequence of servers in the form 'host:port' that
        the pool will connect to.  The list will be randomly permuted before
        being used. `server_list` may also be a function that returns the
        sequence of servers.

        """

        if callable(server_list):
            self.server_list = list(server_list())
        else:
            self.server_list = list(server_list)

        assert len(self.server_list) > 0

        # Randomly permute the array (trust me, it's uniformly random)
        n = len(self.server_list)
        for i in range(0, n):
            j = random.randint(i, n-1)
            temp = self.server_list[j]
            self.server_list[j] = self.server_list[i]
            self.server_list[i] = temp

        self._list_position = 0
        self._notify_on_server_list(self.server_list)

    def _get_next_server(self):
        """
        Gets the next 'localhost:port' combination from the list of
        servers and increments the position. This is not thread-safe,
        but client-side load-balancing isn't so important that this is
        a problem.
        """
        server = self.server_list[self._list_position % len(self.server_list)]
        self._list_position += 1
        return server

    def _create_connection(self):
        """Creates a ConnectionWrapper, which opens a
        pycassa.connection.Connection."""
        failure_count = 0
        while failure_count < 2 * len(self.server_list):
            try:
                server = self._get_next_server()
                wrapper = self._get_new_wrapper(server)
                return wrapper
            except (Thrift.TException, socket.error, IOError, EOFError), exc:
                self._notify_on_failure(exc, server)
                failure_count += 1
        raise AllServersUnavailable('An attempt was made to connect to each of the servers '
                'twice, but none of the attempts succeeded.')

    def _get_new_wrapper(self, server):
        raise NotImplementedError()

    def recreate(self):
        """Return a new instance with identical creation arguments."""

        raise NotImplementedError()

    def dispose(self):
        """ Dispose of this pool.  """
        raise NotImplementedError()

    def get(self):
        raise NotImplementedError()

    def return_conn(self, conn):
        raise NotImplementedError()

    def status(self):
        raise NotImplementedError()

    def add_listener(self, listener):
        """
        Add a :class:`PoolListener`-like object to this pool.

        `listener` may be an object that implements some or all of
        :class:`PoolListener`, or a dictionary of callables containing implementations
        of some or all of the named methods in :class:`PoolListener`.

        """

        listener = as_interface(listener,
            methods=('connection_created', 'connection_checked_out',
                     'connection_checked_in', 'connection_disposed',
                     'connection_recycled', 'connection_failed',
                     'obtained_server_list', 'pool_recreated',
                     'pool_disposed', 'pool_at_max'))

        self.listeners.append(listener)
        if hasattr(listener, 'connection_created'):
            self._on_connect.append(listener)
        if hasattr(listener, 'connection_checked_out'):
            self._on_checkout.append(listener)
        if hasattr(listener, 'connection_checked_in'):
            self._on_checkin.append(listener)
        if hasattr(listener, 'connection_disposed'):
            self._on_dispose.append(listener)
        if hasattr(listener, 'connection_recycled'):
            self._on_recycle.append(listener)
        if hasattr(listener, 'connection_failed'):
            self._on_failure.append(listener)
        if hasattr(listener, 'obtained_server_list'):
            self._on_server_list.append(listener)
        if hasattr(listener, 'pool_recreated'):
            self._on_pool_recreate.append(listener)
        if hasattr(listener, 'pool_disposed'):
            self._on_pool_dispose.append(listener)
        if hasattr(listener, 'pool_at_max'):
            self._on_pool_max.append(listener)

    def _notify_on_pool_recreate(self):
        if self._on_pool_recreate:
            dic = self._get_dic()
            dic['level'] = 'info'
            for l in self._on_pool_recreate:
                l.pool_recreated(dic)

    def _notify_on_pool_dispose(self):
        if self._on_pool_dispose:
            dic = self._get_dic()
            dic['level'] = 'info'
            for l in self._on_pool_dispose:
                l.pool_disposed(dic)

    def _notify_on_pool_max(self, pool_max):
        if self._on_pool_max:
            dic = self._get_dic()
            dic['level'] = 'info'
            dic['pool_max'] = pool_max
            for l in self._on_pool_max:
                l.pool_at_max(dic)

    def _notify_on_dispose(self, conn_record, msg="", error=None):
        if self._on_dispose:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['connection'] = conn_record
            if msg:
                dic['message'] = msg
            if error:
                dic['error'] = error
                dic['level'] = 'warn'
            for l in self._on_dispose:
                l.connection_disposed(dic)

    def _notify_on_server_list(self, server_list):
        dic = self._get_dic()
        dic['level'] = 'debug'
        dic['server_list'] = server_list
        if self._on_server_list:
            for l in self._on_server_list:
                l.obtained_server_list(dic)

    def _notify_on_recycle(self, old_conn, new_conn):
        if self._on_recycle:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['old_conn'] = old_conn
            dic['new_conn'] = new_conn
        for l in self._on_recycle:
            l.connection_recycled(dic)

    def _notify_on_connect(self, conn_record, msg="", error=None):
        if self._on_connect:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['connection'] = conn_record
            if msg:
                dic['message'] = msg
            if error:
                dic['error'] = error
                dic['level'] = 'warn'
            for l in self._on_connect:
                l.connection_created(dic)

    def _notify_on_checkin(self, conn_record):
        if self._on_checkin:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['connection'] = conn_record
            for l in self._on_checkin:
                l.connection_checked_in(dic)

    def _notify_on_checkout(self, conn_record):
        if self._on_checkout:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['connection'] = conn_record
            for l in self._on_checkout:
                l.connection_checked_out(dic)

    def _notify_on_failure(self, error, server, connection=None):
        if self._on_failure:
            dic = self._get_dic()
            dic['level'] = 'info'
            dic['error'] = error
            dic['server'] = server
            dic['connection'] = connection
            for l in self._on_failure:
                l.connection_failed(dic)

    def _get_dic(self):
        return {'pool_type': self.__class__.__name__,
                'pool_id': self.logging_name}


class ConnectionWrapper(connection.Connection):
    """
    A wrapper class for :class:`Connection`s that adds pooling functionality.

    """

    # These mark the state of the connection so that we can
    # check to see that they are not returned, checked out,
    # or disposed twice (or from the wrong state).
    _IN_QUEUE = 0
    _CHECKED_OUT = 1
    _DISPOSED = 2

    def __init__(self, pool, max_retries, *args, **kwargs):
        """
        Creates a wrapper for a :class:`pycassa.connection.Connection`
        object, adding pooling related functionality while still allowing
        access to the thrift API calls.

        These should not be created directly, only obtained through
        Pool's :meth:`~.AbstractPool.get()` method.

        """
        self._pool = pool
        self._retry_count = 0
        self._max_retries = max_retries
        self._lock = threading.Lock()
        self.info = {}
        self.starttime = time.time()
        self.operation_count = 0
        self._state = ConnectionWrapper._CHECKED_OUT
        super(ConnectionWrapper, self).__init__(*args, **kwargs)
        self._pool._notify_on_connect(self)

        # For testing purposes only
        self._should_fail = False
        self._original_meth = self.send_batch_mutate

    def return_to_pool(self):
        """
        Returns this to the pool.

        This has the same effect as calling :meth:`AbstractPool.return_conn()`
        on the wrapper.

        """
        self._pool.return_conn(self)

    def _checkin(self):
        try:
            self._lock.acquire()
            if self._state == ConnectionWrapper._IN_QUEUE:
                raise InvalidRequestError("A connection has been returned to "
                        "the connection pool twice.")
            elif self._state == ConnectionWrapper._DISPOSED:
                raise InvalidRequestError("A disposed connection has been returned "
                        "to the connection pool.")
            self._state = ConnectionWrapper._IN_QUEUE
        finally:
            self._lock.release()

    def _checkout(self):
        try:
            self._lock.acquire()
            if self._state != ConnectionWrapper._IN_QUEUE:
                raise InvalidRequestError("A connection has been checked "
                        "out twice.")
            self._state = ConnectionWrapper._CHECKED_OUT
        finally:
            self._lock.release()

    def _is_in_queue_or_disposed(self):
        try:
            self._lock.acquire()
            ret = self._state == ConnectionWrapper._IN_QUEUE or \
                  self._state == ConnectionWrapper._DISPOSED
        finally:
            self._lock.release()
        return ret

    def _dispose_wrapper(self, reason=None):
        try:
            self._lock.acquire()
            if self._state == ConnectionWrapper._DISPOSED:
                raise InvalidRequestError("A connection has been disposed twice.")
            self._state = ConnectionWrapper._DISPOSED
        finally:
            self._lock.release()

        self.close()
        self._pool._notify_on_dispose(self, msg=reason)

    def _replace(self, new_conn_wrapper):
        """
        Get another wrapper from the pool and replace our own contents
        with its contents.

        """
        self.transport = new_conn_wrapper.transport
        self._iprot = new_conn_wrapper._iprot
        self._oprot = new_conn_wrapper._oprot
        self._lock = new_conn_wrapper._lock
        self.info = new_conn_wrapper.info
        self.starttime = new_conn_wrapper.starttime
        self.operation_count = new_conn_wrapper.operation_count
        self._state = ConnectionWrapper._CHECKED_OUT
        self._should_fail = new_conn_wrapper._should_fail

    def _retry(f):
        def new_f(self, *args, **kwargs):
            self.operation_count += 1
            try:
                if kwargs.pop('reset', False):
                    if hasattr(self._pool, '_replace_wrapper'):
                        self._pool._replace_wrapper() # puts a new wrapper in the queue
                    self._replace(self._pool.get()) # swaps out transport
                result = getattr(super(ConnectionWrapper, self), f.__name__)(*args, **kwargs)
                self._retry_count = 0 # reset the count after a success
                return result
            except (TimedOutException, UnavailableException, Thrift.TException,
                    socket.error, IOError, EOFError), exc:
                self._pool._notify_on_failure(exc, server=self.server, connection=self)

                self._retry_count += 1
                if self._max_retries != -1 and self._retry_count > self._max_retries:
                    self.close()
                    self._pool._clear_current()
                    raise MaximumRetryException('Retried %d times. Last failure was %s' %
                                                (self._retry_count, exc))

                # Exponential backoff
                time.sleep(_BASE_BACKOFF * (2 ** self._retry_count))

                self.close()
                self._pool._clear_current()
		kwargs['reset'] = True
                return new_f(self, *args, **kwargs)

        new_f.__name__ = f.__name__
        return new_f

    def _fail_once(self, *args, **kwargs):
        if self._should_fail:
            self._should_fail = False
            raise TimedOutException
        else:
            return self._original_meth(*args, **kwargs)

    @_retry<commit_msg>fix issue #52<commit_after>"""
Connection pooling for Cassandra connections.

"""

import time, threading, random

import connection
import queue as pool_queue
from logging.pool_logger import PoolLogger
from util import as_interface
from cassandra.ttypes import TimedOutException, UnavailableException
from thrift import Thrift

import threading
import socket

_BASE_BACKOFF = 0.01

__all__ = ['QueuePool', 'ConnectionPool', 'PoolListener',
           'ConnectionWrapper', 'AllServersUnavailable',
           'MaximumRetryException', 'NoConnectionAvailable',
           'InvalidRequestError']

class NoServerAvailable(Exception):
    """Raised if all servers are currently marked dead."""
    pass


class AbstractPool(object):
    """An abstract base class for all other pools."""

    def __init__(self, keyspace, server_list, credentials, timeout,
                 logging_name, use_threadlocal, listeners, framed_transport):
        """An abstract base class for all other pools."""

        if logging_name:
            self.logging_name = self._orig_logging_name = logging_name
        else:
            self._orig_logging_name = None
            self.logging_name = id(self)

        self._pool_threadlocal = use_threadlocal
        self.keyspace = keyspace
        self.credentials = credentials
        self.timeout = timeout
        self.framed_transport = framed_transport
        if use_threadlocal:
            self._tlocal = threading.local()

        # Listener groups
        self.listeners = []
        self._on_connect = []
        self._on_checkout = []
        self._on_checkin = []
        self._on_dispose = []
        self._on_recycle = []
        self._on_failure = []
        self._on_server_list = []
        self._on_pool_recreate = []
        self._on_pool_dispose = []
        self._on_pool_max = []

        self.add_listener(PoolLogger())

        for l in listeners:
            self.add_listener(l)

        self.set_server_list(server_list)

    def set_server_list(self, server_list):
        """
        Sets the server list that the pool will make connections to.

        `server_list` should be sequence of servers in the form 'host:port' that
        the pool will connect to.  The list will be randomly permuted before
        being used. `server_list` may also be a function that returns the
        sequence of servers.

        """

        if callable(server_list):
            self.server_list = list(server_list())
        else:
            self.server_list = list(server_list)

        assert len(self.server_list) > 0

        # Randomly permute the array (trust me, it's uniformly random)
        n = len(self.server_list)
        for i in range(0, n):
            j = random.randint(i, n-1)
            temp = self.server_list[j]
            self.server_list[j] = self.server_list[i]
            self.server_list[i] = temp

        self._list_position = 0
        self._notify_on_server_list(self.server_list)

    def _get_next_server(self):
        """
        Gets the next 'localhost:port' combination from the list of
        servers and increments the position. This is not thread-safe,
        but client-side load-balancing isn't so important that this is
        a problem.
        """
        server = self.server_list[self._list_position % len(self.server_list)]
        self._list_position += 1
        return server

    def _create_connection(self):
        """Creates a ConnectionWrapper, which opens a
        pycassa.connection.Connection."""
        failure_count = 0
        while failure_count < 2 * len(self.server_list):
            try:
                server = self._get_next_server()
                wrapper = self._get_new_wrapper(server)
                return wrapper
            except (Thrift.TException, socket.error, IOError, EOFError), exc:
                self._notify_on_failure(exc, server)
                failure_count += 1
        raise AllServersUnavailable('An attempt was made to connect to each of the servers '
                'twice, but none of the attempts succeeded.')

    def _get_new_wrapper(self, server):
        raise NotImplementedError()

    def recreate(self):
        """Return a new instance with identical creation arguments."""

        raise NotImplementedError()

    def dispose(self):
        """ Dispose of this pool.  """
        raise NotImplementedError()

    def get(self):
        raise NotImplementedError()

    def return_conn(self, conn):
        raise NotImplementedError()

    def status(self):
        raise NotImplementedError()

    def add_listener(self, listener):
        """
        Add a :class:`PoolListener`-like object to this pool.

        `listener` may be an object that implements some or all of
        :class:`PoolListener`, or a dictionary of callables containing implementations
        of some or all of the named methods in :class:`PoolListener`.

        """

        listener = as_interface(listener,
            methods=('connection_created', 'connection_checked_out',
                     'connection_checked_in', 'connection_disposed',
                     'connection_recycled', 'connection_failed',
                     'obtained_server_list', 'pool_recreated',
                     'pool_disposed', 'pool_at_max'))

        self.listeners.append(listener)
        if hasattr(listener, 'connection_created'):
            self._on_connect.append(listener)
        if hasattr(listener, 'connection_checked_out'):
            self._on_checkout.append(listener)
        if hasattr(listener, 'connection_checked_in'):
            self._on_checkin.append(listener)
        if hasattr(listener, 'connection_disposed'):
            self._on_dispose.append(listener)
        if hasattr(listener, 'connection_recycled'):
            self._on_recycle.append(listener)
        if hasattr(listener, 'connection_failed'):
            self._on_failure.append(listener)
        if hasattr(listener, 'obtained_server_list'):
            self._on_server_list.append(listener)
        if hasattr(listener, 'pool_recreated'):
            self._on_pool_recreate.append(listener)
        if hasattr(listener, 'pool_disposed'):
            self._on_pool_dispose.append(listener)
        if hasattr(listener, 'pool_at_max'):
            self._on_pool_max.append(listener)

    def _notify_on_pool_recreate(self):
        if self._on_pool_recreate:
            dic = self._get_dic()
            dic['level'] = 'info'
            for l in self._on_pool_recreate:
                l.pool_recreated(dic)

    def _notify_on_pool_dispose(self):
        if self._on_pool_dispose:
            dic = self._get_dic()
            dic['level'] = 'info'
            for l in self._on_pool_dispose:
                l.pool_disposed(dic)

    def _notify_on_pool_max(self, pool_max):
        if self._on_pool_max:
            dic = self._get_dic()
            dic['level'] = 'info'
            dic['pool_max'] = pool_max
            for l in self._on_pool_max:
                l.pool_at_max(dic)

    def _notify_on_dispose(self, conn_record, msg="", error=None):
        if self._on_dispose:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['connection'] = conn_record
            if msg:
                dic['message'] = msg
            if error:
                dic['error'] = error
                dic['level'] = 'warn'
            for l in self._on_dispose:
                l.connection_disposed(dic)

    def _notify_on_server_list(self, server_list):
        dic = self._get_dic()
        dic['level'] = 'debug'
        dic['server_list'] = server_list
        if self._on_server_list:
            for l in self._on_server_list:
                l.obtained_server_list(dic)

    def _notify_on_recycle(self, old_conn, new_conn):
        if self._on_recycle:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['old_conn'] = old_conn
            dic['new_conn'] = new_conn
        for l in self._on_recycle:
            l.connection_recycled(dic)

    def _notify_on_connect(self, conn_record, msg="", error=None):
        if self._on_connect:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['connection'] = conn_record
            if msg:
                dic['message'] = msg
            if error:
                dic['error'] = error
                dic['level'] = 'warn'
            for l in self._on_connect:
                l.connection_created(dic)

    def _notify_on_checkin(self, conn_record):
        if self._on_checkin:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['connection'] = conn_record
            for l in self._on_checkin:
                l.connection_checked_in(dic)

    def _notify_on_checkout(self, conn_record):
        if self._on_checkout:
            dic = self._get_dic()
            dic['level'] = 'debug'
            dic['connection'] = conn_record
            for l in self._on_checkout:
                l.connection_checked_out(dic)

    def _notify_on_failure(self, error, server, connection=None):
        if self._on_failure:
            dic = self._get_dic()
            dic['level'] = 'info'
            dic['error'] = error
            dic['server'] = server
            dic['connection'] = connection
            for l in self._on_failure:
                l.connection_failed(dic)

    def _get_dic(self):
        return {'pool_type': self.__class__.__name__,
                'pool_id': self.logging_name}


class ConnectionWrapper(connection.Connection):
    """
    A wrapper class for :class:`Connection`s that adds pooling functionality.

    """

    # These mark the state of the connection so that we can
    # check to see that they are not returned, checked out,
    # or disposed twice (or from the wrong state).
    _IN_QUEUE = 0
    _CHECKED_OUT = 1
    _DISPOSED = 2

    def __init__(self, pool, max_retries, *args, **kwargs):
        """
        Creates a wrapper for a :class:`pycassa.connection.Connection`
        object, adding pooling related functionality while still allowing
        access to the thrift API calls.

        These should not be created directly, only obtained through
        Pool's :meth:`~.AbstractPool.get()` method.

        """
        self._pool = pool
        self._retry_count = 0
        self._max_retries = max_retries
        self._lock = threading.Lock()
        self.info = {}
        self.starttime = time.time()
        self.operation_count = 0
        self._state = ConnectionWrapper._CHECKED_OUT
        super(ConnectionWrapper, self).__init__(*args, **kwargs)
        self._pool._notify_on_connect(self)

        # For testing purposes only
        self._should_fail = False
        self._original_meth = self.send_batch_mutate

    def return_to_pool(self):
        """
        Returns this to the pool.

        This has the same effect as calling :meth:`AbstractPool.return_conn()`
        on the wrapper.

        """
        self._pool.return_conn(self)

    def _checkin(self):
        try:
            self._lock.acquire()
            if self._state == ConnectionWrapper._IN_QUEUE:
                raise InvalidRequestError("A connection has been returned to "
                        "the connection pool twice.")
            elif self._state == ConnectionWrapper._DISPOSED:
                raise InvalidRequestError("A disposed connection has been returned "
                        "to the connection pool.")
            self._state = ConnectionWrapper._IN_QUEUE
        finally:
            self._lock.release()

    def _checkout(self):
        try:
            self._lock.acquire()
            if self._state != ConnectionWrapper._IN_QUEUE:
                raise InvalidRequestError("A connection has been checked "
                        "out twice.")
            self._state = ConnectionWrapper._CHECKED_OUT
        finally:
            self._lock.release()

    def _is_in_queue_or_disposed(self):
        try:
            self._lock.acquire()
            ret = self._state == ConnectionWrapper._IN_QUEUE or \
                  self._state == ConnectionWrapper._DISPOSED
        finally:
            self._lock.release()
        return ret

    def _dispose_wrapper(self, reason=None):
        try:
            self._lock.acquire()
            if self._state == ConnectionWrapper._DISPOSED:
                raise InvalidRequestError("A connection has been disposed twice.")
            self._state = ConnectionWrapper._DISPOSED
        finally:
            self._lock.release()

        self.close()
        self._pool._notify_on_dispose(self, msg=reason)

    def _replace(self, new_conn_wrapper):
        """
        Get another wrapper from the pool and replace our own contents
        with its contents.

        """
        self.transport = new_conn_wrapper.transport
        self._iprot = new_conn_wrapper._iprot
        self._oprot = new_conn_wrapper._oprot
        self._lock = new_conn_wrapper._lock
        self.info = new_conn_wrapper.info
        self.starttime = new_conn_wrapper.starttime
        self.operation_count = new_conn_wrapper.operation_count
        self._state = ConnectionWrapper._CHECKED_OUT
        self._should_fail = new_conn_wrapper._should_fail

    def _retry(f):
        def new_f(self, *args, **kwargs):
            self.operation_count += 1
            try:
                if kwargs.pop('reset', False):
                    if hasattr(self._pool, '_replace_wrapper'):
                        self._pool._replace_wrapper() # puts a new wrapper in the queue
                    self._replace(self._pool.get()) # swaps out transport
                result = getattr(super(ConnectionWrapper, self), f.__name__)(*args, **kwargs)
                self._retry_count = 0 # reset the count after a success
                return result
            except (TimedOutException, UnavailableException, Thrift.TException,
                    socket.error, IOError, EOFError), exc:
                self._pool._notify_on_failure(exc, server=self.server, connection=self)

                self._retry_count += 1
                if self._max_retries != -1 and self._retry_count > self._max_retries:
                    self.close()
                    self._pool._clear_current()
                    raise MaximumRetryException('Retried %d times. Last failure was %s' %
                                                (self._retry_count, exc))

                # Exponential backoff
                time.sleep(_BASE_BACKOFF * (2 ** self._retry_count))

                self.close()
                self._pool._clear_current()
		
		kwargs['reset'] = True
                return new_f(self, *args, **kwargs)

        new_f.__name__ = f.__name__
        return new_f

    def _fail_once(self, *args, **kwargs):
        if self._should_fail:
            self._should_fail = False
            raise TimedOutException
        else:
            return self._original_meth(*args, **kwargs)

    @_retry
{'fraction_non_alphanumeric': 0.08944824137356226, 'fraction_numerical': 0.001466911151858643, 'mean_word_length': 2.028063799717343, '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': 4, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2706175', 'n_tokens_mistral': 8374, 'n_tokens_neox': 8139, 'n_words': 1989}