kazurator 0.2.0

Example of Interop with Curator

See the example directory.

Inter Process Mutex

To start, let’s take a look at how we could implement a simple shared (across process) mutex:

from kazoo.client import KazooClient
from kazurator import Mutex

def main():
    client = KazooClient(hosts="YOUR_ZK_CONNECT_STRING_HERE")
    client.start()

    mutex = Mutex(client, "/some/path")

    with mutex:
        # do your thread-safe thing here

    client.stop()

This example assumes you want a single thread to be in the critical region. In order to support simultaneous multi-threaded access, you can set the max_leases kwarg to a higher number. For example:

mutex = Mutex(client, "/some/path", max_leases=2)  # 2 thread at a time

Also, if you’d rather not use the content management protocol, you can call acquire and release directly.

Inter Process Read Write Lock

In some cases, you’ll need to support an unlimited number of read locks, but only a single write lock. For example, suppose you were processing some HDFS paths by altering the format and replacing the data (totally hypothetical of course :smile:).

You’d want any consumers of the data to acquire a read lock. This would prevent the altering process from acquiring a write lock until the consumer(s) are finished. Similarly, the consumers wouldn’t be able to acquire read locks until the altering process removes the write lock.

Consumers will block until the lock is available, or timeout after the specified timeout (default is 1s), at which point a kazoo.LockTimeout is raised.

from kazoo.client import KazooClient
from kazurator import ReadWriteLock

def main():
    client = KazooClient(hosts="YOUR_ZK_CONNECT_STRING_HERE")
    client.start()

    # can optionally supply `timeout` kwarg as well
    lock = ReadWriteLock(client, "/some/path")

    with lock.write_lock:  # block until write_lock is available
        # do your thread-safe thing here

    with lock.read_lock:  # block until write_lock is gone
        # do your thread-safe thing here

    client.stop()