-
Notifications
You must be signed in to change notification settings - Fork 25
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
7 changed files
with
320 additions
and
279 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,46 @@ | ||
| from .base import BaseAPI | ||
|
|
||
|
|
||
| class LockAPI(BaseAPI): | ||
| def lock(self, name, lease=0): | ||
| """ | ||
| Lock acquires a distributed shared lock on a given named lock. | ||
| On success, it will return a unique key that exists so long as | ||
| the lock is held by the caller. This key can be used in | ||
| conjunction with transactions to safely ensure updates to etcd | ||
| only occur while holding lock ownership. The lock is held until | ||
| Unlock is called on the key or the lease associate with the | ||
| owner expires. | ||
| :type name: str | ||
| :param name: name is the identifier for the distributed shared lock to be acquired. | ||
| :type lease: int | ||
| :param lease: lease is the lease ID to associate with the key in the key-value store. A lease | ||
| value of 0 indicates no lease. | ||
| """ | ||
|
|
||
| method = '/v3beta/lock/lock' | ||
| data = { | ||
| "name": name, | ||
| "lease": lease | ||
| } | ||
| return self.call_rpc(method, data=data) | ||
|
|
||
| def unlock(self, key): | ||
| """ | ||
| Unlock takes a key returned by Lock and releases the hold on | ||
| lock. The next Lock caller waiting for the lock will then be | ||
| woken up and given ownership of the lock. | ||
| :type key: str | ||
| :param key: key is the lock ownership key granted by Lock. | ||
| :type lease: int | ||
| :param lease: lease is the lease ID to associate with the key in the key-value store. A lease | ||
| value of 0 indicates no lease. | ||
| """ | ||
|
|
||
| method = '/v3beta/lock/unlock' | ||
| data = { | ||
| "key": key, | ||
| } | ||
| return self.call_rpc(method, data=data) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| import logging | ||
| import os | ||
| import six | ||
| import socket | ||
| import tempfile | ||
| import threading | ||
| import uuid | ||
|
|
||
| from .watch import EventType | ||
| from ..errors import ErrLeaseNotFound | ||
|
|
||
| log = logging.getLogger('etcd3.Lock') | ||
|
|
||
|
|
||
| class EtcdLockError(Exception): | ||
| pass | ||
|
|
||
|
|
||
| class EtcdLockAcquireTimeout(Exception): | ||
| pass | ||
|
|
||
|
|
||
| class ApiLock(object): # TODO: maybe we could improve the performance by reduce some HTTP requests | ||
| """ | ||
| Based on etcd lock API | ||
| """ | ||
|
|
||
| DEFAULT_LOCK_TTL = 60 | ||
|
|
||
| HOST = 'host' | ||
| PROCESS = 'process' | ||
| THREAD = 'thread' | ||
|
|
||
| def __init__(self, client, lock_name, lock_ttl=DEFAULT_LOCK_TTL): | ||
| """ | ||
| :type client: BaseClient | ||
| :param client: instance of etcd.Client | ||
| :type lock_name: str | ||
| :param lock_name: the name of the lock | ||
| :type lock_ttl: int | ||
| :param lock_ttl: ttl of the lock, default is 60s | ||
| """ | ||
| self.client = client | ||
| self.name = lock_name | ||
| self.lock_ttl = lock_ttl | ||
| self.is_taken = False # if the lock is taken by someone | ||
| self.lease = None | ||
| self.lock_key = None | ||
| log.debug("Initiating lock %s", self.lock_name) | ||
|
|
||
| @property | ||
| def is_acquired(self): | ||
| """ | ||
| if the lock is acquired | ||
| """ | ||
| if not self.lease or self.lease.ttl < 0: | ||
| return False | ||
| return True | ||
|
|
||
| acquired = is_acquired | ||
|
|
||
| def acquire(self, lock_ttl=None): | ||
| """ | ||
| Acquire the lock. | ||
| :type lock_ttl: int | ||
| :param lock_ttl: The duration of the lock we acquired, set to None for eternal locks | ||
| :type timeout: int | ||
| :param timeout: The time to wait before giving up on getting a lock | ||
| """ | ||
|
|
||
| if self.lease and self.lock_key and self.lease.alive(): | ||
| return True | ||
| else not is_acquired: | ||
| self.lease = client.Lease(lock_ttl or self.lock_ttl) | ||
| self.lock_key = self.client.lock(self.name, self.lease) | ||
| return self.lock_key | ||
|
|
||
| def wait(self, locker=None, timeout=None): | ||
| """ | ||
| Wait until the lock is lock is able to acquire | ||
| :param locker: kv of the lock | ||
| :param timeout: wait timeout | ||
| """ | ||
| locker = locker or self._get_locker() | ||
| if not locker: | ||
| return | ||
| self.watcher = watcher = self.client.Watcher(key=locker.key, max_retries=0) | ||
| return watcher.watch_once(lambda e: e.type == EventType.DELETE or e.value == self.uuid, timeout=timeout) | ||
|
|
||
| def release(self): | ||
| """ | ||
| Release the lock | ||
| """ | ||
| if self.reentrant: | ||
| n = self.decr_holder() | ||
| if n is not None and n == 0: | ||
| self.lease.revoke() | ||
| self.lease = None | ||
| self.is_taken = False | ||
| else: | ||
| self.lease.cancel_keepalive(join=False) | ||
| self.lease = None | ||
| self.is_taken = True | ||
| else: | ||
| self.lease.revoke() | ||
| self.lease = None | ||
| self.is_taken = False | ||
| log.debug("Lock released (lock_key: %s, value: %s)" % (self.lock_key, self.uuid)) | ||
|
|
||
| def __enter__(self): | ||
| """ | ||
| You can use the lock as a contextmanager | ||
| """ | ||
| self.acquire() | ||
| return self | ||
|
|
||
| def __exit__(self, type, value, traceback): | ||
| self.release() | ||
| return False |
Oops, something went wrong.