<commit_before>import json
import logging
import os
import threading
import time

from pysyncobj import SyncObj, SyncObjConf, replicated, FAIL_REASON
from pysyncobj.dns_resolver import globalDnsResolver
from pysyncobj.node import TCPNode
from pysyncobj.transport import TCPTransport, CONNECTION_STATE
from pysyncobj.utility import TcpUtility

from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
from ..utils import validate_directory

logger = logging.getLogger(__name__)


class _TCPTransport(TCPTransport):

    def __init__(self, syncObj, selfNode, otherNodes):
        super(_TCPTransport, self).__init__(syncObj, selfNode, otherNodes)
        self.setOnUtilityMessageCallback('members', syncObj.getMembers)

    def _connectIfNecessarySingle(self, node):
        try:
            return super(_TCPTransport, self)._connectIfNecessarySingle(node)
        except Exception as e:
            logger.debug('Connection to %s failed: %r', node, e)
            return False


def resolve_host(self):
    return globalDnsResolver().resolve(self.host)


setattr(TCPNode, 'ip', property(resolve_host))


class SyncObjUtility(object):

    def __init__(self, otherNodes, conf, retry_timeout=10):
        self._nodes = otherNodes
        self._utility = TcpUtility(conf.password, retry_timeout/max(1, len(otherNodes)))

    def executeCommand(self, command):
        try:
            return self._utility.executeCommand(self.__node, command)
        except Exception:
            return None

    def getMembers(self):
        for self.__node in self._nodes:
            response = self.executeCommand(['members'])
            if response:
                return [member['addr'] for member in response]


class DynMemberSyncObj(SyncObj):

    def __init__(self, selfAddress, partnerAddrs, conf, retry_timeout=10):
        self.__early_apply_local_log = selfAddress is not None
        self.applied_local_log = False

        utility = SyncObjUtility(partnerAddrs, conf, retry_timeout)
        members = utility.getMembers()
        add_self = members and selfAddress not in members

        partnerAddrs = [member for member in (members or partnerAddrs) if member != selfAddress]

        super(DynMemberSyncObj, self).__init__(selfAddress, partnerAddrs, conf, transportClass=_TCPTransport)

        if add_self:
            thread = threading.Thread(target=utility.executeCommand, args=(['add', selfAddress],))
            thread.daemon = True
            thread.start()

    def getMembers(self, args, callback):
        callback([{'addr': node.id, 'leader': node == self._getLeader(), 'status': CONNECTION_STATE.CONNECTED
                   if self.isNodeConnected(node) else CONNECTION_STATE.DISCONNECTED} for node in self.otherNodes] +
                 [{'addr': self.selfNode.id, 'leader': self._isLeader(), 'status': CONNECTION_STATE.CONNECTED}], None)

    def _onTick(self, timeToWait=0.0):
        super(DynMemberSyncObj, self)._onTick(timeToWait)

        # The SyncObj calls onReady callback only when cluster got the leader and is ready for writes.
        # In some cases for us it is safe to "signal" the Raft object when the local log is fully applied.
        # We are using the `applied_local_log` property for that, but not calling the callback function.
        if self.__early_apply_local_log and not self.applied_local_log and self.raftLastApplied == self.raftCommitIndex:
            self.applied_local_log = True


class KVStoreTTL(DynMemberSyncObj):

    def __init__(self, on_ready, on_set, on_delete, **config):
        self.__thread = None
        self.__on_set = on_set
        self.__on_delete = on_delete
        self.__limb = {}
        self.__retry_timeout = None

        self_addr = config.get('self_addr')
        partner_addrs = set(config.get('partner_addrs', []))
        if config.get('patronictl'):
            if self_addr:
                partner_addrs.add(self_addr)
            self_addr = None

        # Create raft data_dir if necessary
        raft_data_dir = config.get('data_dir', '')
        if raft_data_dir != '':
            validate_directory(raft_data_dir)

        file_template = (self_addr or '')
        file_template = file_template.replace(':', '_') if os.name == 'nt' else file_template
        file_template = os.path.join(raft_data_dir, file_template)
        conf = SyncObjConf(password=config.get('password'), autoTick=False, appendEntriesUseBatch=False,
                           bindAddress=config.get('bind_addr'), dnsFailCacheTime=(config.get('loop_wait') or 10),
                           dnsCacheTime=(config.get('ttl') or 30), commandsWaitLeader=config.get('commandsWaitLeader'),
                           fullDumpFile=(file_template + '.dump' if self_addr else None),
                           journalFile=(file_template + '.journal' if self_addr else None),
                           onReady=on_ready, dynamicMembershipChange=True)

        retry_timeout = int(config.get('retry_timeout') or 10)
        super(KVStoreTTL, self).__init__(self_addr, partner_addrs, conf, retry_timeout)
        self.__data = {}

    @staticmethod
    def __check_requirements(old_value, **kwargs):
        return ('prevExist' not in kwargs or bool(kwargs['prevExist']) == bool(old_value)) and \
            ('prevValue' not in kwargs or old_value and old_value['value'] == kwargs['prevValue']) and \
            (not kwargs.get('prevIndex') or old_value and old_value['index'] == kwargs['prevIndex'])

    def set_retry_timeout(self, retry_timeout):
        self.__retry_timeout = retry_timeout

    def retry(self, func, *args, **kwargs):
        event = threading.Event()
        ret = {'result': None, 'error': -1}<commit_msg>Make sure self.__retry_timeout is set (#2440)<commit_after>import json
import logging
import os
import threading
import time

from pysyncobj import SyncObj, SyncObjConf, replicated, FAIL_REASON
from pysyncobj.dns_resolver import globalDnsResolver
from pysyncobj.node import TCPNode
from pysyncobj.transport import TCPTransport, CONNECTION_STATE
from pysyncobj.utility import TcpUtility

from . import AbstractDCS, ClusterConfig, Cluster, Failover, Leader, Member, SyncState, TimelineHistory
from ..utils import validate_directory

logger = logging.getLogger(__name__)


class _TCPTransport(TCPTransport):

    def __init__(self, syncObj, selfNode, otherNodes):
        super(_TCPTransport, self).__init__(syncObj, selfNode, otherNodes)
        self.setOnUtilityMessageCallback('members', syncObj.getMembers)

    def _connectIfNecessarySingle(self, node):
        try:
            return super(_TCPTransport, self)._connectIfNecessarySingle(node)
        except Exception as e:
            logger.debug('Connection to %s failed: %r', node, e)
            return False


def resolve_host(self):
    return globalDnsResolver().resolve(self.host)


setattr(TCPNode, 'ip', property(resolve_host))


class SyncObjUtility(object):

    def __init__(self, otherNodes, conf, retry_timeout=10):
        self._nodes = otherNodes
        self._utility = TcpUtility(conf.password, retry_timeout/max(1, len(otherNodes)))

    def executeCommand(self, command):
        try:
            return self._utility.executeCommand(self.__node, command)
        except Exception:
            return None

    def getMembers(self):
        for self.__node in self._nodes:
            response = self.executeCommand(['members'])
            if response:
                return [member['addr'] for member in response]


class DynMemberSyncObj(SyncObj):

    def __init__(self, selfAddress, partnerAddrs, conf, retry_timeout=10):
        self.__early_apply_local_log = selfAddress is not None
        self.applied_local_log = False

        utility = SyncObjUtility(partnerAddrs, conf, retry_timeout)
        members = utility.getMembers()
        add_self = members and selfAddress not in members

        partnerAddrs = [member for member in (members or partnerAddrs) if member != selfAddress]

        super(DynMemberSyncObj, self).__init__(selfAddress, partnerAddrs, conf, transportClass=_TCPTransport)

        if add_self:
            thread = threading.Thread(target=utility.executeCommand, args=(['add', selfAddress],))
            thread.daemon = True
            thread.start()

    def getMembers(self, args, callback):
        callback([{'addr': node.id, 'leader': node == self._getLeader(), 'status': CONNECTION_STATE.CONNECTED
                   if self.isNodeConnected(node) else CONNECTION_STATE.DISCONNECTED} for node in self.otherNodes] +
                 [{'addr': self.selfNode.id, 'leader': self._isLeader(), 'status': CONNECTION_STATE.CONNECTED}], None)

    def _onTick(self, timeToWait=0.0):
        super(DynMemberSyncObj, self)._onTick(timeToWait)

        # The SyncObj calls onReady callback only when cluster got the leader and is ready for writes.
        # In some cases for us it is safe to "signal" the Raft object when the local log is fully applied.
        # We are using the `applied_local_log` property for that, but not calling the callback function.
        if self.__early_apply_local_log and not self.applied_local_log and self.raftLastApplied == self.raftCommitIndex:
            self.applied_local_log = True


class KVStoreTTL(DynMemberSyncObj):

    def __init__(self, on_ready, on_set, on_delete, **config):
        self.__thread = None
        self.__on_set = on_set
        self.__on_delete = on_delete
        self.__limb = {}
        self.set_retry_timeout(int(config.get('retry_timeout') or 10))

        self_addr = config.get('self_addr')
        partner_addrs = set(config.get('partner_addrs', []))
        if config.get('patronictl'):
            if self_addr:
                partner_addrs.add(self_addr)
            self_addr = None

        # Create raft data_dir if necessary
        raft_data_dir = config.get('data_dir', '')
        if raft_data_dir != '':
            validate_directory(raft_data_dir)

        file_template = (self_addr or '')
        file_template = file_template.replace(':', '_') if os.name == 'nt' else file_template
        file_template = os.path.join(raft_data_dir, file_template)
        conf = SyncObjConf(password=config.get('password'), autoTick=False, appendEntriesUseBatch=False,
                           bindAddress=config.get('bind_addr'), dnsFailCacheTime=(config.get('loop_wait') or 10),
                           dnsCacheTime=(config.get('ttl') or 30), commandsWaitLeader=config.get('commandsWaitLeader'),
                           fullDumpFile=(file_template + '.dump' if self_addr else None),
                           journalFile=(file_template + '.journal' if self_addr else None),
                           onReady=on_ready, dynamicMembershipChange=True)

        super(KVStoreTTL, self).__init__(self_addr, partner_addrs, conf, self.__retry_timeout)
        self.__data = {}

    @staticmethod
    def __check_requirements(old_value, **kwargs):
        return ('prevExist' not in kwargs or bool(kwargs['prevExist']) == bool(old_value)) and \
            ('prevValue' not in kwargs or old_value and old_value['value'] == kwargs['prevValue']) and \
            (not kwargs.get('prevIndex') or old_value and old_value['index'] == kwargs['prevIndex'])

    def set_retry_timeout(self, retry_timeout):
        self.__retry_timeout = retry_timeout

    def retry(self, func, *args, **kwargs):
        event = threading.Event()
        ret = {'result': None, 'error': -1}
{'fraction_non_alphanumeric': 0.09847627340008706, 'fraction_numerical': 0.002786242925555072, 'mean_word_length': 3.1301690039554115, 'pattern_counts': {'":': 0, '<': 3, '<?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': '7277690', 'n_tokens_mistral': 3335, 'n_tokens_neox': 3258, 'n_words': 817}