<commit_before># -*- coding: utf-8 -
#
# This file is part of restkit released under the MIT license. 
# See the NOTICE for more information.

from __future__ import with_statement

from collections import Counter, deque
import logging
import signal
import socket
import threading
import time

from ..sock import close

log = logging.getLogger(__name__)

class Manager(object):

    def __init__(self, max_conn=10, timeout=300):
        self.max_conn = max_conn
        self.timeout = timeout

        self.sockets = dict()
        self.active_sockets = dict()
        self._lock = self.get_lock()
        self.connections_count = Counter()

        if timeout and timeout is not None:
            self.start()

    def get_lock(self):
        return threading.RLock()

    def murder_connections(self, *args):
        self._lock.acquire()
        try:
            active_sockets = self.active_sockets.copy()
            for fno, (sock, t0) in active_sockets.items():
                diff = time.time() - t0
                if diff <= self.timeout:
                    continue
                close(sock)
                del self.active_sockets[fno]
        finally:
            self._lock.release()
       
    def start(self):
        signal.signal(signal.SIGALRM, self.murder_connections)
        signal.alarm(self.timeout)

    def all_connections_count(self, n=None):
        """ return all counts per address registered. if n is specified,
        it will return the n most commons """
        return self.connections_count.most_common(n)

    def connection_count(self, addr, ssl):
        """ get connections count for an address """
        return self.connections_count[(addr, ssl)]

    def find_socket(self, addr, ssl=False):
        """ find a socket from a its address in the pool and return if
        there is one available, else, return None """

        self._lock.acquire()
        try:
            key = (addr, ssl)
            try:
                socks = self.sockets[key]
                while True:
                    sock = socks.pop()
                    if sock.fileno() in self.active_sockets:
                        del self.active_sockets[sock.fileno()]
                        break
                self.sockets[key] = socks
                self.connections_count[key] -= 1
                log.debug("get connection from manager")
                return sock
            except (IndexError, KeyError,):
                return None
        finally:
            self._lock.release()

    def store_socket(self, sock, addr, ssl=False):
        """ store a socket in the pool to reuse it across threads """
        self._lock.acquire()
        try:
            key = (addr, ssl)
            try:
                socks = self.sockets[key]
            except KeyError:
                socks = deque()

            if len(socks) < self.max_conn:
                # add connection to the pool
                try:
                    self.active_sockets[sock.fileno()] = (sock, time.time())
                except (socket.error, AttributeError,):
                    # socket has been closed
                    log.info("socket closed")
                    return

                socks.appendleft(sock)
                self.sockets[key] = socks
                self.connections_count[key] += 1

                log.debug("put connection in manager %s" %
                        self.all_connections_count())
            else:
                # close connection if we have enough connections in the
                # pool.
                close(sock)
        finally:
            self._lock.release()
                <commit_msg>preserver compatibility with old pythons.<commit_after># -*- coding: utf-8 -
#
# This file is part of restkit released under the MIT license. 
# See the NOTICE for more information.

from __future__ import with_statement

from collections import deque
import logging
import signal
import socket
import threading
import time

from ..sock import close

log = logging.getLogger(__name__)

class Manager(object):

    def __init__(self, max_conn=10, timeout=300):
        self.max_conn = max_conn
        self.timeout = timeout

        self.sockets = dict()
        self.active_sockets = dict()
        self.connections_count = dict()
        self._lock = self.get_lock()

        if timeout and timeout is not None:
            self.start()

    def get_lock(self):
        return threading.RLock()

    def murder_connections(self, *args):
        self._lock.acquire()
        try:
            active_sockets = self.active_sockets.copy()
            for fno, (sock, t0) in active_sockets.items():
                diff = time.time() - t0
                if diff <= self.timeout:
                    continue
                close(sock)
                del self.active_sockets[fno]
        finally:
            self._lock.release()
       
    def start(self):
        signal.signal(signal.SIGALRM, self.murder_connections)
        signal.alarm(self.timeout)

    def all_connections_count(self):
        """ return all counts per address registered. """
        return self.connections_count.items()

    def connection_count(self, addr, ssl):
        """ get connections count for an address """
        self._lock.acquire()
        try:
            return self.connections_count[(addr, ssl)]
        finally:
            self._lock.release()

        return self.connections_count[(addr, ssl)]

    def find_socket(self, addr, ssl=False):
        """ find a socket from a its address in the pool and return if
        there is one available, else, return None """

        self._lock.acquire()
        try:
            key = (addr, ssl)
            try:
                socks = self.sockets[key]
                while True:
                    sock = socks.pop()
                    if sock.fileno() in self.active_sockets:
                        del self.active_sockets[sock.fileno()]
                        break
                self.sockets[key] = socks
                self.connections_count[key] -= 1
                log.debug("get connection from manager")
                return sock
            except (IndexError, KeyError,):
                return None
        finally:
            self._lock.release()

    def store_socket(self, sock, addr, ssl=False):
        """ store a socket in the pool to reuse it across threads """
        self._lock.acquire()
        try:
            key = (addr, ssl)
            try:
                socks = self.sockets[key]
            except KeyError:
                socks = deque()

            if len(socks) < self.max_conn:
                # add connection to the pool
                try:
                    self.active_sockets[sock.fileno()] = (sock, time.time())
                except (socket.error, AttributeError,):
                    # socket has been closed
                    log.info("socket closed")
                    return

                socks.appendleft(sock)
                self.sockets[key] = socks
                
                try:
                    self.connections_count[key] += 1
                except KeyError:
                    self.connections_count[key] = 1 

                log.debug("put connection in manager %s" %
                        self.all_connections_count())
            else:
                # close connection if we have enough connections in the
                # pool.
                close(sock)
        finally:
            self._lock.release()
                
{'fraction_non_alphanumeric': 0.08650013287270794, 'fraction_numerical': 0.0027903268668615466, 'mean_word_length': 1.5300840336134454, 'pattern_counts': {'":': 0, '<': 7, '<?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': '5194338', 'n_tokens_mistral': 2018, 'n_tokens_neox': 1900, 'n_words': 553}