<s> """<STR_LIT>""" <EOL> import sys <EOL> try : <EOL> from time import monotonic as now <EOL> except ImportError : <EOL> from time import time as now <EOL> import uuid <EOL> import six <EOL> from kazoo . retry import ( <EOL> KazooRetry , <EOL> RetryFailedError , <EOL> ForceRetryError <EOL> ) <EOL> from kazoo . exceptions import CancelledError <EOL> from kazoo . exceptions import KazooException <EOL> from kazoo . exceptions import LockTimeout <EOL> from kazoo . exceptions import NoNodeError <EOL> from kazoo . protocol . states import KazooState <EOL> class _Watch ( object ) : <EOL> def __init__ ( self , duration = None ) : <EOL> self . duration = duration <EOL> self . started_at = None <EOL> def start ( self ) : <EOL> self . started_at = now ( ) <EOL> def leftover ( self ) : <EOL> if self . duration is None : <EOL> return None <EOL> else : <EOL> elapsed = now ( ) - self . started_at <EOL> return max ( <NUM_LIT:0> , self . duration - elapsed ) <EOL> class Lock ( object ) : <EOL> """<STR_LIT>""" <EOL> _NODE_NAME = '<STR_LIT>' <EOL> def __init__ ( self , client , path , identifier = None ) : <EOL> """<STR_LIT>""" <EOL> self . client = client <EOL> self . path = path <EOL> self . data = str ( identifier or "<STR_LIT>" ) . encode ( '<STR_LIT:utf-8>' ) <EOL> self . node = None <EOL> self . wake_event = client . handler . event_object ( ) <EOL> self . prefix = uuid . uuid4 ( ) . hex + self . _NODE_NAME <EOL> self . create_path = self . path + "<STR_LIT:/>" + self . prefix <EOL> self . create_tried = False <EOL> self . is_acquired = False <EOL> self . assured_path = False <EOL> self . cancelled = False <EOL> self . _retry = KazooRetry ( max_tries = None , <EOL> sleep_func = client . handler . sleep_func ) <EOL> self . _lock = client . handler . lock_object ( ) <EOL> def _ensure_path ( self ) : <EOL> self . client . ensure_path ( self . path ) <EOL> self . assured_path = True <EOL> def cancel ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cancelled = True <EOL> self . wake_event . set ( ) <EOL> def acquire ( self , blocking = True , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> def _acquire_lock ( ) : <EOL> got_it = self . _lock . acquire ( False ) <EOL> if not got_it : <EOL> raise ForceRetryError ( ) <EOL> return True <EOL> retry = self . _retry . copy ( ) <EOL> retry . deadline = timeout <EOL> locked = self . _lock . acquire ( False ) <EOL> if not locked and not blocking : <EOL> return False <EOL> if not locked : <EOL> try : <EOL> locked = retry ( _acquire_lock ) <EOL> except RetryFailedError : <EOL> return False <EOL> already_acquired = self . is_acquired <EOL> try : <EOL> gotten = False <EOL> try : <EOL> gotten = retry ( self . _inner_acquire , <EOL> blocking = blocking , timeout = timeout ) <EOL> except RetryFailedError : <EOL> pass <EOL> except KazooException : <EOL> exc_info = sys . exc_info ( ) <EOL> if not already_acquired : <EOL> self . _best_effort_cleanup ( ) <EOL> self . cancelled = False <EOL> six . reraise ( exc_info [ <NUM_LIT:0> ] , exc_info [ <NUM_LIT:1> ] , exc_info [ <NUM_LIT:2> ] ) <EOL> if gotten : <EOL> self . is_acquired = gotten <EOL> if not gotten and not already_acquired : <EOL> self . _best_effort_cleanup ( ) <EOL> return gotten <EOL> finally : <EOL> self . _lock . release ( ) <EOL> def _watch_session ( self , state ) : <EOL> self . wake_event . set ( ) <EOL> return True <EOL> def _inner_acquire ( self , blocking , timeout ) : <EOL> if self . is_acquired : <EOL> if not blocking : <EOL> return False <EOL> raise ForceRetryError ( ) <EOL> if not self . assured_path : <EOL> self . _ensure_path ( ) <EOL> node = None <EOL> if self . create_tried : <EOL> node = self . _find_node ( ) <EOL> else : <EOL> self . create_tried = True <EOL> if not node : <EOL> node = self . client . create ( self . create_path , self . data , <EOL> ephemeral = True , sequence = True ) <EOL> node = node [ len ( self . path ) + <NUM_LIT:1> : ] <EOL> self . node = node <EOL> while True : <EOL> self . wake_event . clear ( ) <EOL> if self . cancelled : <EOL> raise CancelledError ( ) <EOL> children = self . _get_sorted_children ( ) <EOL> try : <EOL> our_index = children . index ( node ) <EOL> except ValueError : <EOL> raise ForceRetryError ( ) <EOL> if self . acquired_lock ( children , our_index ) : <EOL> return True <EOL> if not blocking : <EOL> return False <EOL> predecessor = self . path + "<STR_LIT:/>" + children [ our_index - <NUM_LIT:1> ] <EOL> self . client . add_listener ( self . _watch_session ) <EOL> try : <EOL> if self . client . exists ( predecessor , self . _watch_predecessor ) : <EOL> self . wake_event . wait ( timeout ) <EOL> if not self . wake_event . isSet ( ) : <EOL> raise LockTimeout ( "<STR_LIT>" <EOL> "<STR_LIT>" % ( self . path , timeout ) ) <EOL> finally : <EOL> self . client . remove_listener ( self . _watch_session ) <EOL> def acquired_lock ( self , children , index ) : <EOL> return index == <NUM_LIT:0> <EOL> def _watch_predecessor ( self , event ) : <EOL> self . wake_event . set ( ) <EOL> def _get_sorted_children ( self ) : <EOL> children = self . client . get_children ( self . path ) <EOL> lockname = self . _NODE_NAME <EOL> children . sort ( key = lambda c : c [ c . find ( lockname ) + len ( lockname ) : ] ) <EOL> return children <EOL> def _find_node ( self ) : <EOL> children = self . client . get_children ( self . path ) <EOL> for child in children : <EOL> if child . startswith ( self . prefix ) : <EOL> return child <EOL> return None <EOL> def _delete_node ( self , node ) : <EOL> self . client . delete ( self . path + "<STR_LIT:/>" + node ) <EOL> def _best_effort_cleanup ( self ) : <EOL> try : <EOL> node = self . node or self . _find_node ( ) <EOL> if node : <EOL> self . _delete_node ( node ) <EOL> except KazooException : <EOL> pass <EOL> def release ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . client . retry ( self . _inner_release ) <EOL> def _inner_release ( self ) : <EOL> if not self . is_acquired : <EOL> return False <EOL> try : <EOL> self . _delete_node ( self . node ) <EOL> except NoNodeError : <EOL> pass <EOL> self . is_acquired = False <EOL> self . node = None <EOL> return True <EOL> def contenders ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . assured_path : <EOL> self . _ensure_path ( ) <EOL> children = self . _get_sorted_children ( ) <EOL> contenders = [ ] <EOL> for child in children : <EOL> try : <EOL> data , stat = self . client . get ( self . path + "<STR_LIT:/>" + child ) <EOL> contenders . append ( data . decode ( '<STR_LIT:utf-8>' ) ) <EOL> except NoNodeError : <EOL> pass <EOL> return contenders <EOL> def __enter__ ( self ) : <EOL> self . acquire ( ) <EOL> def __exit__ ( self , exc_type , exc_value , traceback ) : <EOL> self . release ( ) <EOL> class Semaphore ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , client , path , identifier = None , max_leases = <NUM_LIT:1> ) : <EOL> """<STR_LIT>""" <EOL> self . client = client <EOL> self . path = path <EOL> self . data = str ( identifier or "<STR_LIT>" ) . encode ( '<STR_LIT:utf-8>' ) <EOL> self . max_leases = max_leases <EOL> self . wake_event = client . handler . event_object ( ) <EOL> self . create_path = self . path + "<STR_LIT:/>" + uuid . uuid4 ( ) . hex <EOL> self . lock_path = path + '<STR_LIT:->' + '<STR_LIT>' <EOL> self . is_acquired = False <EOL> self . assured_path = False <EOL> self . cancelled = False <EOL> self . _session_expired = False <EOL> def _ensure_path ( self ) : <EOL> result = self . client . ensure_path ( self . path ) <EOL> self . assured_path = True <EOL> if result is True : <EOL> data , _ = self . client . get ( self . path ) <EOL> try : <EOL> leases = int ( data . decode ( '<STR_LIT:utf-8>' ) ) <EOL> except ( ValueError , TypeError ) : <EOL> pass <EOL> else : <EOL> if leases != self . max_leases : <EOL> raise ValueError ( <EOL> "<STR_LIT>" % <EOL> ( leases , self . max_leases ) <EOL> ) <EOL> else : <EOL> self . client . set ( self . path , str ( self . max_leases ) . encode ( '<STR_LIT:utf-8>' ) ) <EOL> def cancel ( self ) : <EOL> """<STR_LIT>""" <EOL> self . cancelled = True <EOL> self . wake_event . set ( ) <EOL> def acquire ( self , blocking = True , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> self . cancelled = False <EOL> try : <EOL> self . is_acquired = self . client . retry ( <EOL> self . _inner_acquire , blocking = blocking , timeout = timeout ) <EOL> except KazooException : <EOL> self . _best_effort_cleanup ( ) <EOL> self . cancelled = False <EOL> raise <EOL> return self . is_acquired <EOL> def _inner_acquire ( self , blocking , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> self . _session_expired = False <EOL> self . client . add_listener ( self . _watch_session ) <EOL> if not self . assured_path : <EOL> self . _ensure_path ( ) <EOL> if self . client . exists ( self . create_path ) : <EOL> return True <EOL> w = _Watch ( duration = timeout ) <EOL> w . start ( ) <EOL> lock = self . client . Lock ( self . lock_path , self . data ) <EOL> gotten = lock . acquire ( blocking = blocking , timeout = w . leftover ( ) ) <EOL> if not gotten : <EOL> return False <EOL> try : <EOL> while True : <EOL> self . wake_event . clear ( ) <EOL> if self . _get_lease ( ) : <EOL> return True <EOL> if blocking : <EOL> self . wake_event . wait ( w . leftover ( ) ) <EOL> if not self . wake_event . isSet ( ) : <EOL> raise LockTimeout ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" % ( self . path , timeout ) ) <EOL> else : <EOL> return False <EOL> finally : <EOL> lock . release ( ) <EOL> def _watch_lease_change ( self , event ) : <EOL> self . wake_event . set ( ) <EOL> def _get_lease ( self , data = None ) : <EOL> if self . _session_expired : <EOL> raise ForceRetryError ( "<STR_LIT>" ) <EOL> if self . cancelled : <EOL> raise CancelledError ( "<STR_LIT>" ) <EOL> children = self . client . get_children ( self . path , <EOL> self . _watch_lease_change ) <EOL> if len ( children ) < self . max_leases : <EOL> self . client . create ( self . create_path , self . data , ephemeral = True ) <EOL> if self . client . exists ( self . create_path ) : <EOL> self . is_acquired = True <EOL> else : <EOL> self . is_acquired = False <EOL> return self . is_acquired <EOL> def _watch_session ( self , state ) : <EOL> if state == KazooState . LOST : <EOL> self . _session_expired = True <EOL> self . wake_event . set ( ) <EOL> return True <EOL> def _best_effort_cleanup ( self ) : <EOL> try : <EOL> self . client . delete ( self . create_path ) <EOL> except KazooException : <EOL> pass <EOL> def release ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . client . retry ( self . _inner_release ) <EOL> def _inner_release ( self ) : <EOL> if not self . is_acquired : <EOL> return False <EOL> try : <EOL> self . client . delete ( self . create_path ) <EOL> except NoNodeError : <EOL> pass <EOL> self . is_acquired = False <EOL> return True <EOL> def lease_holders ( self ) : <EOL> """<STR_LIT>""" <EOL> if not self . client . exists ( self . path ) : <EOL> return [ ] <EOL> children = self . client . get_children ( self . path ) <EOL> lease_holders = [ ] <EOL> for child in children : <EOL> try : <EOL> data , stat = self . client . get ( self . path + "<STR_LIT:/>" + child ) <EOL> lease_holders . append ( data . decode ( '<STR_LIT:utf-8>' ) ) <EOL> except NoNodeError : <EOL> pass <EOL> return lease_holders <EOL> def __enter__ ( self ) : <EOL> self . acquire ( ) <EOL> def __exit__ ( self , exc_type , exc_value , traceback ) : <EOL> self . release ( ) </s>
