gistfile1.txt

from hacheck_utils import hadown
from hacheck_utils import haup
from sleeper import Sleeper
from socket import gethostname

def wrap_operation(name, inner_op, wait, post, lock_delay):
    """ Uses a distributed semaphore to wrap an operation, where an operation
    is an arbitrary function.

    Args:
        name: the name uniquely identifying this operation. The name is the
        unit of lock contention throttling the speed of the operation.
        inner_op: the operation to be wrapped.
        dectest: the workhorse operation. Decrements and tests the counter.
        It must block until it is safe to proceed with inner_op. dectest will
        be called under mutual exclusion from all other operations of the same
        `name`.
        increment: increments the counter. Mutual exclusion is NOT provided
        when calling `increment`.
        lock_delay: the number of seconds to wait for events that occurred
        before we held the lock to become visible to us. Useful if dectest has
        some propagation delay.

    """
    with zk.lock("{}/lock".format((name,))):
        sleep(lock_delay)
        semaphore.wait()

    operation()
    semaphore.post()


class SmartstackSemaphore(object):
    def __init__(self, name, minimum, zk, node_ids):
        self.name = name
        self.minimum = minimum
        self.zk = zk
        self.node_ids = ids

    def _decrement(self):
        hadown(self.name)


    def _increment(self):
        haup(self.name)


    def _verify(self, sleep_delay, timeout):
        """Verifies in Zookeeper that the nodes in question are actually
        absent.

        This prevents situations where we decrement, but Nerve (due to
        configuration mishap or some other reason) does not remove our
        nodes from Zookeeper.
        """

        sleeper = Sleeper(timeout)
        for i in self.ids:
            while True:
                if self.zk.get("{}/{}".format(self.name, i)):
                    sleeper.use(sleep_delay)


    def _inner_wait(self, sleep_delay, timeout):
        """
        Uses Smartstack registration data to test if a named service meets a
        given minimum replication level, and blocks until the test passes.
        """

        sleeper = Sleeper(timeout)
        while True:
            children = zk_client.get_children(name)
            if len(children) > minimum:
                break

            sleeper.use(sleep_delay)


    def wait(self,
            wait_delay=1, wait_timeout=180,
            verify_delay=0.25, verify_timeout=4):
        self._inner_wait(wait_delay, wait_timeout)
        self._decrement()
        try:
            self._verify(verify_delay, verify_timeout)
        except:
            self._increment()
            raise


    def post(self):
        self._increment()


def sayhello():
    print('Hi')


def do_cluster_sayhello(zk, name, minimum=5, sayhello):
    op_name = '/db/{}/sayhello'.format(name)

    # This hostname logic is horribly wrong, but hopefully illustrative
    hostname = gethostname()

    semaphore = SmartstackSemaphore(name, minimum, zk, (hostname,))
    wrap_operation(op_name, semaphore.wait, semaphore.post, 10)