<s> import zc . zk <EOL> import zookeeper <EOL> import multiprocessing <EOL> import sys <EOL> import traceback <EOL> import pettingzoo . utils <EOL> import logging <EOL> from abc import ABCMeta , abstractmethod <EOL> from pettingzoo . deleted import Deleted <EOL> from pettingzoo . utils import get_logger <EOL> PREFIX = "<STR_LIT>" <EOL> logger = logging . getLogger ( '<STR_LIT>' ) <EOL> class LeaderQueue ( object ) : <EOL> def __init__ ( self , connection , path ) : <EOL> self . connection = connection <EOL> self . path = path <EOL> self . deletion_handlers = { } <EOL> self . connection . create_recursive ( <EOL> self . path + PREFIX , "<STR_LIT>" , acl = zc . zk . OPEN_ACL_UNSAFE ) <EOL> self . lock = multiprocessing . RLock ( ) <EOL> self . candidate_by_predecessor = { } <EOL> self . counter_by_candidate = { } <EOL> def remove_candidate ( self , candidate ) : <EOL> '''<STR_LIT>''' <EOL> with self . lock : <EOL> del_id = self . counter_by_candidate . get ( candidate , None ) <EOL> get_logger ( ) . info ( "<STR_LIT>" % ( del_id ) ) <EOL> if del_id == None : <EOL> return False <EOL> try : <EOL> self . connection . delete ( id_to_item_path ( self . path , del_id ) ) <EOL> self . _handle_remove ( del_id ) <EOL> with self . lock : <EOL> del self . counter_by_candidate [ candidate ] <EOL> return True <EOL> except zookeeper . NoNodeException : <EOL> return False <EOL> def add_candidate ( self , candidate , meta_data = None ) : <EOL> '''<STR_LIT>''' <EOL> if self . has_candidate ( candidate ) : <EOL> get_logger ( ) . warning ( <EOL> "<STR_LIT>" ) <EOL> return False <EOL> else : <EOL> flags = zookeeper . EPHEMERAL | zookeeper . SEQUENCE <EOL> newpath = self . connection . create ( <EOL> self . path + PREFIX + PREFIX , <EOL> meta_data , <EOL> zc . zk . OPEN_ACL_UNSAFE , <EOL> flags ) <EOL> counter = pettingzoo . utils . counter_value ( newpath ) <EOL> get_logger ( ) . info ( "<STR_LIT>" % ( newpath ) ) <EOL> try : <EOL> self . _create_deletion_handlers ( counter ) <EOL> except : <EOL> exc_class , exc , tback = sys . exc_info ( ) <EOL> sys . stderr . write ( str ( exc_class ) + "<STR_LIT:\n>" ) <EOL> traceback . print_tb ( tback ) <EOL> raise exc_class , exc , tback <EOL> with self . lock : <EOL> self . counter_by_candidate [ candidate ] = counter <EOL> self . _handle_add ( counter , candidate ) <EOL> return True <EOL> def has_candidate ( self , candidate ) : <EOL> '''<STR_LIT>''' <EOL> with self . lock : <EOL> return self . counter_by_candidate . has_key ( candidate ) <EOL> def _process_deleted ( self , node ) : <EOL> """<STR_LIT>""" <EOL> del_id = pettingzoo . utils . counter_value ( node . path ) <EOL> get_logger ( ) . debug ( "<STR_LIT>" % ( del_id ) ) <EOL> self . _handle_remove ( del_id ) <EOL> with self . lock : <EOL> del self . deletion_handlers [ del_id ] <EOL> def _handle_remove ( self , del_id ) : <EOL> '''<STR_LIT>''' <EOL> with self . lock : <EOL> candidate = self . candidate_by_predecessor . get ( del_id , None ) <EOL> if del_id == None : <EOL> get_logger ( ) . warning ( <EOL> "<STR_LIT>" % <EOL> ( del_id ) ) <EOL> elif candidate == None : <EOL> get_logger ( ) . debug ( <EOL> "<STR_LIT>" <EOL> + "<STR_LIT>" % ( del_id ) ) <EOL> else : <EOL> self . _update_predecessor_dict ( del_id , candidate ) <EOL> del self . candidate_by_predecessor [ del_id ] <EOL> return None <EOL> def _create_deletion_handlers ( self , counter ) : <EOL> '''<STR_LIT>''' <EOL> with self . lock : <EOL> if self . deletion_handlers . get ( counter , None ) == None : <EOL> path = id_to_item_path ( self . path , counter ) <EOL> delete_watch = Deleted ( <EOL> self . connection , path , [ self . _process_deleted ] ) <EOL> self . deletion_handlers [ counter ] = delete_watch <EOL> return None <EOL> def _handle_add ( self , counter , candidate ) : <EOL> '''<STR_LIT>''' <EOL> if counter == None : <EOL> get_logger ( ) . warning ( <EOL> "<STR_LIT>" % ( counter ) ) <EOL> else : <EOL> self . _update_predecessor_dict ( counter , candidate ) <EOL> return None <EOL> def _update_predecessor_dict ( self , counter , candidate ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> children = self . connection . children ( self . path + PREFIX ) <EOL> pred_id = pettingzoo . utils . min_predecessor ( children , counter ) <EOL> with self . lock : <EOL> self . candidate_by_predecessor [ pred_id ] = candidate <EOL> if pred_id == - <NUM_LIT:1> : <EOL> candidate . on_elected ( ) <EOL> else : <EOL> self . _create_deletion_handlers ( pred_id ) <EOL> except : <EOL> exc_class , exc , tback = sys . exc_info ( ) <EOL> sys . stderr . write ( str ( exc_class ) + "<STR_LIT:\n>" ) <EOL> traceback . print_tb ( tback ) <EOL> raise exc_class , exc , tback <EOL> return None <EOL> class Candidate : <EOL> __metaclass__ = ABCMeta <EOL> @ abstractmethod <EOL> def on_elected ( self ) : <EOL> '''<STR_LIT>''' <EOL> def id_to_item_path ( path , item_id ) : <EOL> """<STR_LIT>""" <EOL> return pettingzoo . utils . counter_path ( path + PREFIX + PREFIX , item_id ) </s>
