<commit_before>#
#    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 contextlib
import uuid

from oslo.config import cfg
from oslo import messaging
from oslo.utils import excutils

from senlin.common import exception
from senlin.common.i18n import _LI
from senlin.common.i18n import _LW
from senlin.common import messaging as rpc_messaging
from senlin.db import api as db_api
from senlin.openstack.common import log as logging

cfg.CONF.import_opt('engine_life_check_timeout', 'senlin.common.config')

LOG = logging.getLogger(__name__)


class BaseLock(object):
    def __init__(self, context, target, engine_id):
        self.context = context
        self.target = target
        self.engine_id = engine_id
        self.listener = None

    @staticmethod
    def engine_alive(context, engine_id):
        client = rpc_messaging.get_rpc_client(version='1.0', topic=engine_id)
        client_context = client.prepare(
            timeout=cfg.CONF.engine_life_check_timeout)
        try:
            return client_context.call(context, 'listening')
        except messaging.MessagingTimeout:
            return False

    @staticmethod
    def generate_engine_id():
        return str(uuid.uuid4())

    def try_acquire(self):
        """
        Try to acquire a lock for target, but don't raise an ActionInProgress
        exception or try to steal lock.
        """
        return db_api.senlin_lock_create(self.target.uuid, self.engine_id)

    def acquire(self, retry=True):
        """
        Acquire a lock on the target.

        :param retry: When True, retry if lock was released while stealing.
        :type retry: boolean
        """
        lock_engine_id = db_api.senlin_lock_create(self.target.uuid,
                                                  self.engine_id)
        if lock_engine_id is None:
            LOG.debug("Engine %(engine)s acquired lock on target "
                      "%(target)s" % {'engine': self.engine_id,
                                     'target': self.target.uuid})
            return

        if lock_engine_id == self.engine_id or \
           self.engine_alive(self.context, lock_engine_id):
            LOG.debug("Lock on target %(target)s is owned by engine "
                      "%(engine)s" % {'target': self.target.uuid,
                                      'engine': lock_engine_id})
            raise exception.ActionInProgress(target_name=self.target.name,
                                             action=self.target.status)
        else:
            LOG.info(_LI("Stale lock detected on target %(target)s.  Engine "
                         "%(target)s will attempt to steal the lock"),
                     {'target': self.target.uuid, 'engine': self.engine_id})

            result = db_api.senlin_lock_steal(self.target.uuid, lock_engine_id,
                                             self.engine_id)

            if result is None:
                LOG.info(_LI("Engine %(engine)s successfully stole the lock "
                             "on target %(target)s"),
                         {'engine': self.engine_id,
                          'target': self.target.uuid})
                return
            elif result is True:
                if retry:
                    LOG.info(_LI("The lock on target %(target)s was released "
                                 "while engine %(engine)s was stealing it. "
                                 "Trying again"), {'target': self.target.uuid,
                                                   'engine': self.engine_id})
                    return self.acquire(retry=False)
            else:
                new_lock_engine_id = result
                LOG.info(_LI("Failed to steal lock on target %(target)s. "
                             "Engine %(engine)s stole the lock first"),
                         {'target': self.target.uuid,
                          'engine': new_lock_engine_id})

            raise exception.ActionInProgress(
                target_name=self.target.name, action=self.target.status)

    def release(self, target_id):
        """Release a target lock."""
        # Only the engine that owns the lock will be releasing it.
        result = db_api.senlin_lock_release(target_id, self.engine_id)
        if result is True:
            LOG.warn(_LW("Lock was already released on target %s!"), target_id)
        else:
            LOG.debug("Engine %(engine)s released lock on target "
                      "%(target)s" % {'engine': self.engine_id,
                                     'target': target_id})

    @contextlib.contextmanager
    def thread_lock(self, target_id):
        """
        Acquire a lock and release it only if there is an exception.  The
        release method still needs to be scheduled to be run at the
        end of the thread using the Thread.link method.
        """
        try:
            self.acquire()
            yield
        except:  # noqa
            with excutils.save_and_reraise_exception():
                self.release(target_id)

    @contextlib.contextmanager
    def try_thread_lock(self, target_id):
        """
        Similar to thread_lock, but acquire the lock using try_acquire
        and only release it upon any exception after a successful
        acquisition.
        """
        result = None
        try:
            result = self.try_acquire()
            yield result
        except:  # noqa
            if result is None:  # Lock was successfully acquired
                with excutils.save_and_reraise_exception():
                    self.release(target_id)
            raise

class ClusterLock(BaseLock):
    def __init__(self, context, cluster, engine_id):
        super(ClusterLock, self).__init__()
        self.context = context
        self.target = cluster
        self.engine_id = engine_id
        self.listener = None<commit_msg>Initial version -- Lock hierarchy in Senlin engine<commit_after>#
#    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 contextlib
import uuid

from oslo.config import cfg
from oslo import messaging
from oslo.utils import excutils

from senlin.common import exception
from senlin.common.i18n import _LI
from senlin.common.i18n import _LW
from senlin.common import messaging as rpc_messaging
from senlin.db import api as db_api
from senlin.openstack.common import log as logging

cfg.CONF.import_opt('engine_life_check_timeout', 'senlin.common.config')

LOG = logging.getLogger(__name__)


class BaseLock(object):
    def __init__(self, context, target, engine_id):
        self.context = context
        self.target = target
        self.target_type = 'target'
        self.engine_id = engine_id
        self.listener = None

    @staticmethod
    def engine_alive(context, engine_id):
        client = rpc_messaging.get_rpc_client(version='1.0', topic=engine_id)
        client_context = client.prepare(
            timeout=cfg.CONF.engine_life_check_timeout)
        try:
            return client_context.call(context, 'listening')
        except messaging.MessagingTimeout:
            return False

    @staticmethod
    def generate_engine_id():
        return str(uuid.uuid4())

    def try_acquire(self):
        """
        Try to acquire a lock for target, but don't raise an ActionInProgress
        exception or try to steal lock.
        """
        return db_api.senlin_lock_create(self.target.uuid, self.engine_id)

    def acquire(self, retry=True):
        """
        Acquire a lock on the target.

        :param retry: When True, retry if lock was released while stealing.
        :type retry: boolean
        """
        lock_engine_id = db_api.senlin_lock_create(self.target.uuid,
                                                  self.engine_id)
        if lock_engine_id is None:
            LOG.debug("Engine %(engine)s acquired lock on %(target_type)s "
                      "%(target)s" % {'engine': self.engine_id,
                                      'target_type': self.target_type,
                                      'target': self.target.uuid})
            return

        if lock_engine_id == self.engine_id or \
           self.engine_alive(self.context, lock_engine_id):
            LOG.debug("Lock on %(target_type)s %(target)s is owned by engine "
                      "%(engine)s" % {'target_type': self.target_type,
                                      'target': self.target.uuid,
                                      'engine': lock_engine_id})
            raise exception.ActionInProgress(target_name=self.target.name,
                                             action=self.target.status)
        else:
            LOG.info(_LI("Stale lock detected on %(target_type)s %(target)s.  Engine "
                         "%(target)s will attempt to steal the lock"),
                     {'target_type': self.target_type, 
                      'target': self.target.uuid,
                      'engine': self.engine_id})

            result = db_api.senlin_lock_steal(self.target.uuid, lock_engine_id,
                                             self.engine_id)

            if result is None:
                LOG.info(_LI("Engine %(engine)s successfully stole the lock "
                             "on %(target_type)s %(target)s"),
                         {'engine': self.engine_id,
                          'target_type': self.target_type,
                          'target': self.target.uuid})
                return
            elif result is True:
                if retry:
                    LOG.info(_LI("The lock on %(target_type)s %(target)s was released "
                                 "while engine %(engine)s was stealing it. "
                                 "Trying again"), {'target_type': self.target_type,
                                                   'target': self.target.uuid,
                                                   'engine': self.engine_id})
                    return self.acquire(retry=False)
            else:
                new_lock_engine_id = result
                LOG.info(_LI("Failed to steal lock on %(target_type)s %(target)s. "
                             "Engine %(engine)s stole the lock first"),
                         {'target_type': self.target_type,
                          'target': self.target.uuid,
                          'engine': new_lock_engine_id})

            raise exception.ActionInProgress(
                target_name=self.target.name, action=self.target.status)

    def release(self, target_id):
        """Release a target lock."""
        # Only the engine that owns the lock will be releasing it.
        result = db_api.senlin_lock_release(target_id, self.engine_id)
        if result is True:
            LOG.warn(_LW("Lock was already released on %(target_type) %(target)s!"),
                     {'target_type': self.target_type,
                      'target': target_id)
        else:
            LOG.debug("Engine %(engine)s released lock on %(target_type)s "
                      "%(target)s" % {'engine': self.engine_id,
                                      'target_type': self.target_type,
                                      'target': target_id})

    @contextlib.contextmanager
    def thread_lock(self, target_id):
        """
        Acquire a lock and release it only if there is an exception.  The
        release method still needs to be scheduled to be run at the
        end of the thread using the Thread.link method.
        """
        try:
            self.acquire()
            yield
        except:  # noqa
            with excutils.save_and_reraise_exception():
                self.release(target_id)

    @contextlib.contextmanager
    def try_thread_lock(self, target_id):
        """
        Similar to thread_lock, but acquire the lock using try_acquire
        and only release it upon any exception after a successful
        acquisition.
        """
        result = None
        try:
            result = self.try_acquire()
            yield result
        except:  # noqa
            if result is None:  # Lock was successfully acquired
                with excutils.save_and_reraise_exception():
                    self.release(target_id)
            raise

class ClusterLock(BaseLock):
    def __init__(self, context, cluster, engine_id):
        super(ClusterLock, self).__init__(context, cluster, engine_id)
        self.target_type = 'cluster'


class NodeLock(BaseLock):
    def __init__(self, context, node, engine_id):
        super(ClusterLock, self).__init__(context, cluster, engine_id)
        self.target_type = 'node'
{'fraction_non_alphanumeric': 0.08349243648112792, 'fraction_numerical': 0.0016155088852988692, 'mean_word_length': 1.7850715746421268, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 33, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7363850', 'n_tokens_mistral': 3558, 'n_tokens_neox': 3462, 'n_words': 1029}