<commit_before>#
# Copyright 2014 Red Hat, Inc.
# Copyright 2015 Hewlett-Packard Development Company, L.P.
#
# Author: <NAME> <<EMAIL>>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import math
import time
import uuid

from oslo_config import cfg
from oslo_log import log
import tooz.coordination

from designate.i18n import _LI
from designate.i18n import _LW
from designate.i18n import _LE


LOG = log.getLogger(__name__)

OPTS = [
    cfg.StrOpt('backend_url',
               help='The backend URL to use for distributed coordination. If '
                    'unset services that need coordination will function as '
                    'a standalone service.'),
    cfg.FloatOpt('heartbeat_interval',
                 default=1.0,
                 help='Number of seconds between heartbeats for distributed '
                      'coordination.'),
    cfg.FloatOpt('run_watchers_interval',
                 default=10.0,
                 help='Number of seconds between checks to see if group '
                      'membership has changed')

]
cfg.CONF.register_opts(OPTS, group='coordination')

CONF = cfg.CONF


class CoordinationMixin(object):
    def __init__(self, *args, **kwargs):
        super(CoordinationMixin, self).__init__(*args, **kwargs)

        self._coordinator = None

    def start(self):
        self._coordination_id = ":".join([CONF.host, str(uuid.uuid4())])

        if CONF.coordination.backend_url is not None:
            backend_url = cfg.CONF.coordination.backend_url
            self._coordinator = tooz.coordination.get_coordinator(
                backend_url, self._coordination_id)
            self._coordination_started = False

            self.tg.add_timer(cfg.CONF.coordination.heartbeat_interval,
                              self._coordinator_heartbeat)
            self.tg.add_timer(cfg.CONF.coordination.run_watchers_interval,
                              self._coordinator_run_watchers)

        else:
            msg = _LW("No coordination backend configured, distributed "
                      "coordination functionality will be disabled. "
                      "Please configure a coordination backend.")
            LOG.warning(msg)

        super(CoordinationMixin, self).start()

        if self._coordinator is not None:
            while not self._coordination_started:
                try:
                    self._coordinator.start()
                    self._coordinator.create_group(self.service_name)
                    self._coordinator.join_group(self.service_name)
                    self._coordination_started = True

                except Exception:
                    LOG.warning(_LW("Failed to start Coordinator:"),
                             exc_info=True)
                    time.sleep(15)

    def stop(self):
        if self._coordinator is not None:
            self._coordination_started = False

            self._coordinator.leave_group(self.service_name)
            self._coordinator.stop()

        super(CoordinationMixin, self).stop()

        self._coordinator = None

    def _coordinator_heartbeat(self):
        if not self._coordination_started:
            return

        try:
            self._coordinator.heartbeat()
        except tooz.coordination.ToozError:
            LOG.exception(_LE('Error sending a heartbeat to coordination '
                          'backend.'))

    def _coordinator_run_watchers(self):
        if not self._coordination_started:
            return

        self._coordinator.run_watchers()


class Partitioner(object):
    def __init__(self, coordinator, group_id, my_id, partitions):
        self._coordinator = coordinator
        self._group_id = group_id
        self._my_id = my_id
        self._partitions = partitions

        self._started = False
        self._my_partitions = None
        self._callbacks = []

    def _warn_no_backend(self):
        LOG.warning(_LW('No coordination backend configured, assuming we are '
                        'the only worker. Please configure a coordination '
                        'backend'))

    def _get_members(self, group_id):
        get_members_req = self._coordinator.get_members(group_id)
        try:
            return get_members_req.get()
        except tooz.ToozError:
            self.join_group(group_id)

    def _on_group_change(self, event):
        LOG.debug("Received member change %s" % event)
        members, self._my_partitions = self._update_partitions()

        self._run_callbacks(members, event)

    def _partition(self, members, me, partitions):<commit_msg>Partitoner references invalid exception class<commit_after>#
# Copyright 2014 Red Hat, Inc.
# Copyright 2015 Hewlett-Packard Development Company, L.P.
#
# Author: <NAME> <<EMAIL>>
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.

import math
import time
import uuid

from oslo_config import cfg
from oslo_log import log
import tooz.coordination

from designate.i18n import _LI
from designate.i18n import _LW
from designate.i18n import _LE


LOG = log.getLogger(__name__)

OPTS = [
    cfg.StrOpt('backend_url',
               help='The backend URL to use for distributed coordination. If '
                    'unset services that need coordination will function as '
                    'a standalone service.'),
    cfg.FloatOpt('heartbeat_interval',
                 default=1.0,
                 help='Number of seconds between heartbeats for distributed '
                      'coordination.'),
    cfg.FloatOpt('run_watchers_interval',
                 default=10.0,
                 help='Number of seconds between checks to see if group '
                      'membership has changed')

]
cfg.CONF.register_opts(OPTS, group='coordination')

CONF = cfg.CONF


class CoordinationMixin(object):
    def __init__(self, *args, **kwargs):
        super(CoordinationMixin, self).__init__(*args, **kwargs)

        self._coordinator = None

    def start(self):
        self._coordination_id = ":".join([CONF.host, str(uuid.uuid4())])

        if CONF.coordination.backend_url is not None:
            backend_url = cfg.CONF.coordination.backend_url
            self._coordinator = tooz.coordination.get_coordinator(
                backend_url, self._coordination_id)
            self._coordination_started = False

            self.tg.add_timer(cfg.CONF.coordination.heartbeat_interval,
                              self._coordinator_heartbeat)
            self.tg.add_timer(cfg.CONF.coordination.run_watchers_interval,
                              self._coordinator_run_watchers)

        else:
            msg = _LW("No coordination backend configured, distributed "
                      "coordination functionality will be disabled. "
                      "Please configure a coordination backend.")
            LOG.warning(msg)

        super(CoordinationMixin, self).start()

        if self._coordinator is not None:
            while not self._coordination_started:
                try:
                    self._coordinator.start()
                    self._coordinator.create_group(self.service_name)
                    self._coordinator.join_group(self.service_name)
                    self._coordination_started = True

                except Exception:
                    LOG.warning(_LW("Failed to start Coordinator:"),
                             exc_info=True)
                    time.sleep(15)

    def stop(self):
        if self._coordinator is not None:
            self._coordination_started = False

            self._coordinator.leave_group(self.service_name)
            self._coordinator.stop()

        super(CoordinationMixin, self).stop()

        self._coordinator = None

    def _coordinator_heartbeat(self):
        if not self._coordination_started:
            return

        try:
            self._coordinator.heartbeat()
        except tooz.coordination.ToozError:
            LOG.exception(_LE('Error sending a heartbeat to coordination '
                          'backend.'))

    def _coordinator_run_watchers(self):
        if not self._coordination_started:
            return

        self._coordinator.run_watchers()


class Partitioner(object):
    def __init__(self, coordinator, group_id, my_id, partitions):
        self._coordinator = coordinator
        self._group_id = group_id
        self._my_id = my_id
        self._partitions = partitions

        self._started = False
        self._my_partitions = None
        self._callbacks = []

    def _warn_no_backend(self):
        LOG.warning(_LW('No coordination backend configured, assuming we are '
                        'the only worker. Please configure a coordination '
                        'backend'))

    def _get_members(self, group_id):
        get_members_req = self._coordinator.get_members(group_id)
        try:
            return get_members_req.get()

        except tooz.coordination.GroupNotCreated:
            LOG.error(_LE('Attempting to partition over a non-existent group: '
                          '%s'), self._group_id)
            raise

    def _on_group_change(self, event):
        LOG.debug("Received member change %s" % event)
        members, self._my_partitions = self._update_partitions()

        self._run_callbacks(members, event)

    def _partition(self, members, me, partitions):
{'fraction_non_alphanumeric': 0.07916949645871738, 'fraction_numerical': 0.00504511497040846, 'mean_word_length': 2.202236719478099, 'pattern_counts': {'":': 2, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1846396', 'n_tokens_mistral': 2795, 'n_tokens_neox': 2625, 'n_words': 720}