<commit_before>"""
Elliptics is a fault tolerant distributed key/value storage.
See http://reverbrain.com/elliptics and
https://github.com/reverbrain/elliptics for additional info.

Docs: http://doc.reverbrain.com/
Deployment guide: http://doc.reverbrain.com/elliptics:server-tutorial
Packages: http://repo.reverbrain.com/
"""

import itertools
import logging

import cache_lru

from . import Storage

import elliptics


DEFAULT_NAMESPACE = "DOCKER"
logger = logging.getLogger(__name__)


class EllipticsStorage(Storage):

    def __init__(self, config):
        cfg = elliptics.Config()
        # The parameter which sets the time to wait for the operation complete
        cfg.config.wait_timeout = config.get("elliptics_wait_timeout", 60)
        # The parameter which sets the timeout for pinging node
        cfg.config.check_timeout = config.get("elliptics_check_timeout", 60)
        # Number of IO threads in processing pool
        cfg.config.io_thread_num = config.get("elliptics_io_thread_num", 2)
        # Number of threads in network processing pool
        cfg.config.net_thread_num = config.get("elliptics_net_thread_num", 2)
        # Number of IO threads in processing pool dedicated to nonblocking ops
        nblock_iothreads = config.get("elliptics_nonblocking_io_thread_num", 2)
        cfg.config.nonblocking_io_thread_num = nblock_iothreads
        self.groups = config.get('elliptics_groups', [])
        if len(self.groups) == 0:
            raise ValueError("Specify groups")

        # loglevel of elliptics logger
        elliptics_log_level = config.get('elliptics_verbosity', 0)

        # path to logfile
        elliptics_log_file = config.get('elliptics_logfile', '/dev/stderr')
        log = elliptics.Logger(elliptics_log_file, elliptics_log_level)
        self._elliptics_node = elliptics.Node(log, cfg)

        self.namespace = config.get('elliptics_namespace', DEFAULT_NAMESPACE)
        logger.info("Using namespace %s", self.namespace)

        at_least_one = False
        for host, port in config.get('elliptics_nodes').iteritems():
            try:
                self._elliptics_node.add_remote(host, port, config.get('elliptics_addr_family', 2))
                at_least_one = True
            except Exception as err:
                logger.error("Failed to add remote %s:%d %s", host, port, err)

        if not at_least_one:
            raise Exception("Unable to connect to Elliptics")

    @property
    def _session(self):
        session = elliptics.Session(self._elliptics_node)
        session.groups = self.groups
        session.set_namespace(self.namespace)
        session.exceptions_policy = elliptics.exceptions_policy.no_exceptions
        return session<commit_msg>Pep8 fix<commit_after>"""
Elliptics is a fault tolerant distributed key/value storage.
See http://reverbrain.com/elliptics and
https://github.com/reverbrain/elliptics for additional info.

Docs: http://doc.reverbrain.com/
Deployment guide: http://doc.reverbrain.com/elliptics:server-tutorial
Packages: http://repo.reverbrain.com/
"""

import itertools
import logging

import cache_lru

from . import Storage

import elliptics


DEFAULT_NAMESPACE = "DOCKER"
logger = logging.getLogger(__name__)


class EllipticsStorage(Storage):

    def __init__(self, config):
        cfg = elliptics.Config()
        # The parameter which sets the time to wait for the operation complete
        cfg.config.wait_timeout = config.get("elliptics_wait_timeout", 60)
        # The parameter which sets the timeout for pinging node
        cfg.config.check_timeout = config.get("elliptics_check_timeout", 60)
        # Number of IO threads in processing pool
        cfg.config.io_thread_num = config.get("elliptics_io_thread_num", 2)
        # Number of threads in network processing pool
        cfg.config.net_thread_num = config.get("elliptics_net_thread_num", 2)
        # Number of IO threads in processing pool dedicated to nonblocking ops
        nblock_iothreads = config.get("elliptics_nonblocking_io_thread_num", 2)
        cfg.config.nonblocking_io_thread_num = nblock_iothreads
        self.groups = config.get('elliptics_groups', [])
        if len(self.groups) == 0:
            raise ValueError("Specify groups")

        # loglevel of elliptics logger
        elliptics_log_level = config.get('elliptics_verbosity', 0)

        # path to logfile
        elliptics_log_file = config.get('elliptics_logfile', '/dev/stderr')
        log = elliptics.Logger(elliptics_log_file, elliptics_log_level)
        self._elliptics_node = elliptics.Node(log, cfg)

        self.namespace = config.get('elliptics_namespace', DEFAULT_NAMESPACE)
        logger.info("Using namespace %s", self.namespace)

        at_least_one = False
        for host, port in config.get('elliptics_nodes').iteritems():
            try:
                self._elliptics_node.add_remote(host, port,
                                                config.get(
                                                    'elliptics_addr_family',
                                                    2))
                at_least_one = True
            except Exception as err:
                logger.error("Failed to add remote %s:%d %s", host, port, err)

        if not at_least_one:
            raise Exception("Unable to connect to Elliptics")

    @property
    def _session(self):
        session = elliptics.Session(self._elliptics_node)
        session.groups = self.groups
        session.set_namespace(self.namespace)
        session.exceptions_policy = elliptics.exceptions_policy.no_exceptions
        return session
{'fraction_non_alphanumeric': 0.0877318116975749, 'fraction_numerical': 0.003744650499286733, 'mean_word_length': 2.8078750848608283, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 6, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5605992', 'n_tokens_mistral': 1690, 'n_tokens_neox': 1589, 'n_words': 381}