<s> import next . utils <EOL> import hashlib <EOL> import bisect <EOL> from datetime import datetime , timedelta <EOL> import celery <EOL> from next . broker . celery_app import tasks as tasks <EOL> from next . broker . celery_app . celery_broker import app <EOL> import next . constants <EOL> import redis <EOL> import json <EOL> import time <EOL> import next . utils as utils <EOL> class JobBroker : <EOL> def __init__ ( self ) : <EOL> self . num_replicas = <NUM_LIT:5> <EOL> self . r = redis . StrictRedis ( host = next . constants . RABBITREDIS_HOSTNAME , port = next . constants . RABBITREDIS_PORT , db = <NUM_LIT:0> ) <EOL> def applyAsync ( self , app_id , exp_uid , task_name , args , ignore_result = False ) : <EOL> """<STR_LIT>""" <EOL> submit_timestamp = utils . datetimeNow ( '<STR_LIT:string>' ) <EOL> domain = self . __get_domain_for_job ( app_id + "<STR_LIT:_>" + exp_uid ) <EOL> result = tasks . apply . apply_async ( args = [ app_id , exp_uid , task_name , args , submit_timestamp ] , exchange = '<STR_LIT>' + domain , routing_key = '<STR_LIT>' + domain ) <EOL> if ignore_result : <EOL> return True <EOL> else : <EOL> return result . get ( interval = <NUM_LIT> ) <EOL> def applySyncByNamespace ( self , app_id , exp_uid , task_name , args , namespace = None , ignore_result = False , time_limit = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> submit_timestamp = utils . datetimeNow ( '<STR_LIT:string>' ) <EOL> if namespace == None : <EOL> namespace = exp_uid <EOL> domain = self . __get_domain_for_job ( app_id + "<STR_LIT:_>" + exp_uid ) <EOL> num_queues = next . constants . CELERY_SYNC_WORKER_COUNT <EOL> try : <EOL> namespace_cnt = int ( self . r . get ( namespace + "<STR_LIT>" ) ) <EOL> except : <EOL> pipe = self . r . pipeline ( True ) <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> pipe . watch ( namespace + "<STR_LIT>" , "<STR_LIT>" ) <EOL> if not pipe . exists ( namespace + "<STR_LIT>" ) : <EOL> if not pipe . exists ( '<STR_LIT>' ) : <EOL> namespace_counter = <NUM_LIT:0> <EOL> else : <EOL> namespace_counter = pipe . get ( '<STR_LIT>' ) <EOL> pipe . multi ( ) <EOL> pipe . set ( namespace + "<STR_LIT>" , int ( namespace_counter ) + <NUM_LIT:1> ) <EOL> pipe . set ( '<STR_LIT>' , int ( namespace_counter ) + <NUM_LIT:1> ) <EOL> pipe . execute ( ) <EOL> else : <EOL> pipe . unwatch ( ) <EOL> break <EOL> except redis . exceptions . WatchError : <EOL> continue <EOL> finally : <EOL> pipe . reset ( ) <EOL> namespace_cnt = int ( self . r . get ( namespace + "<STR_LIT>" ) ) <EOL> queue_number = ( namespace_cnt % num_queues ) + <NUM_LIT:1> <EOL> queue_name = '<STR_LIT>' + str ( queue_number ) + '<STR_LIT:@>' + domain <EOL> job_uid = utils . getNewUID ( ) <EOL> if time_limit == <NUM_LIT:0> : <EOL> soft_time_limit = None <EOL> hard_time_limit = None <EOL> else : <EOL> soft_time_limit = time_limit <EOL> hard_time_limit = time_limit + <NUM_LIT> <EOL> result = tasks . apply_sync_by_namespace . apply_async ( args = [ app_id , exp_uid , task_name , args , namespace , job_uid , submit_timestamp , time_limit ] , queue = queue_name , soft_time_limit = soft_time_limit , time_limit = hard_time_limit ) <EOL> if ignore_result : <EOL> return True <EOL> else : <EOL> return result . get ( interval = <NUM_LIT> ) <EOL> def __get_domain_for_job ( self , job_id ) : <EOL> """<STR_LIT>""" <EOL> while not self . r . exists ( '<STR_LIT>' ) : <EOL> print "<STR_LIT>" <EOL> time . sleep ( <NUM_LIT> ) <EOL> replicated_domains_hash = json . loads ( self . r . get ( '<STR_LIT>' ) ) <EOL> h = self . hash ( job_id ) <EOL> if h > replicated_domains_hash [ - <NUM_LIT:1> ] [ <NUM_LIT:1> ] : return replicated_domains_hash [ <NUM_LIT:0> ] [ <NUM_LIT:0> ] <EOL> hash_values = map ( lambda x : x [ <NUM_LIT:1> ] , replicated_domains_hash ) <EOL> index = bisect . bisect_left ( hash_values , h ) <EOL> return replicated_domains_hash [ index ] [ <NUM_LIT:0> ] <EOL> def refresh_domain_hashes ( self ) : <EOL> worker_pings = None <EOL> while worker_pings == None : <EOL> try : <EOL> worker_pings = app . control . inspect ( ) . ping ( ) <EOL> except : <EOL> worker_pings = None <EOL> worker_names = worker_pings . keys ( ) <EOL> domain_names_with_dups = [ item . split ( '<STR_LIT:@>' ) [ <NUM_LIT:1> ] for item in worker_names ] <EOL> domain_names = list ( set ( domain_names_with_dups ) ) <EOL> timestamp = utils . datetime2str ( utils . datetimeNow ( ) ) <EOL> print "<STR_LIT>" % ( timestamp , str ( domain_names ) ) <EOL> replicated_domains_hash = [ ] <EOL> for domain in domain_names : <EOL> for i in range ( self . num_replicas ) : <EOL> replicated_domains_hash . append ( ( domain , self . hash ( domain + '<STR_LIT>' + str ( i ) ) , i ) ) <EOL> replicated_domains_hash = sorted ( replicated_domains_hash , key = lambda x : x [ <NUM_LIT:1> ] ) <EOL> self . r . set ( '<STR_LIT>' , json . dumps ( replicated_domains_hash ) ) <EOL> def hash ( self , key ) : <EOL> """<STR_LIT>""" <EOL> return ( int ( hashlib . md5 ( key ) . hexdigest ( ) , <NUM_LIT:16> ) % <NUM_LIT> ) / <NUM_LIT> </s>
