<s> from . base import BaseServer , LOGGER <EOL> from . . resources2 import WorkerResource <EOL> from . . networkaddress import getNetworkAddress <EOL> from . . amqp import amqp as AMQP <EOL> from . . resources import InterfaceResource , ExposedResource <EOL> from MySQLdb . cursors import DictCursor <EOL> from twisted . internet import reactor , protocol , task <EOL> from twisted . enterprise import adbapi <EOL> from twisted . web import server <EOL> from twisted . protocols . memcache import MemCacheProtocol , DEFAULT_PORT <EOL> from twisted . internet . defer import Deferred , DeferredList , maybeDeferred , inlineCallbacks <EOL> from twisted . internet . threads import deferToThread <EOL> from uuid import UUID , uuid4 <EOL> import pprint <EOL> import simplejson <EOL> PRETTYPRINTER = pprint . PrettyPrinter ( indent = <NUM_LIT:4> ) <EOL> class WorkerServer ( BaseServer ) : <EOL> public_ip = None <EOL> local_ip = None <EOL> network_information = { } <EOL> simultaneous_jobs = <NUM_LIT:50> <EOL> jobs_complete = <NUM_LIT:0> <EOL> job_queue = [ ] <EOL> job_queue_a = job_queue . append <EOL> jobsloop = None <EOL> pending_dequeue = False <EOL> def __init__ ( self , <EOL> aws_access_key_id , <EOL> aws_secret_access_key , <EOL> aws_s3_http_cache_bucket = None , <EOL> aws_s3_storage_bucket = None , <EOL> mysql_username = None , <EOL> mysql_password = None , <EOL> mysql_host = None , <EOL> mysql_database = None , <EOL> amqp_host = None , <EOL> amqp_username = None , <EOL> amqp_password = None , <EOL> amqp_vhost = None , <EOL> amqp_queue = None , <EOL> amqp_exchange = None , <EOL> memcached_host = None , <EOL> scheduler_server_group = '<STR_LIT>' , <EOL> schedulerserver_port = <NUM_LIT> , <EOL> service_mapping = None , <EOL> service_args_mapping = None , <EOL> amqp_port = <NUM_LIT> , <EOL> amqp_prefetch_count = <NUM_LIT:200> , <EOL> mysql_port = <NUM_LIT> , <EOL> memcached_port = <NUM_LIT> , <EOL> max_simultaneous_requests = <NUM_LIT:100> , <EOL> max_requests_per_host_per_second = <NUM_LIT:0> , <EOL> max_simultaneous_requests_per_host = <NUM_LIT:0> , <EOL> port = <NUM_LIT> , <EOL> log_file = '<STR_LIT>' , <EOL> log_directory = None , <EOL> log_level = "<STR_LIT>" ) : <EOL> self . network_information [ "<STR_LIT:port>" ] = port <EOL> self . mysql = adbapi . ConnectionPool ( <EOL> "<STR_LIT>" , <EOL> db = mysql_database , <EOL> port = mysql_port , <EOL> user = mysql_username , <EOL> passwd = mysql_password , <EOL> host = mysql_host , <EOL> cp_reconnect = True , <EOL> cursorclass = DictCursor ) <EOL> self . memcached_host = memcached_host <EOL> self . memcached_port = memcached_port <EOL> self . memc_ClientCreator = protocol . ClientCreator ( <EOL> reactor , MemCacheProtocol ) <EOL> self . scheduler_server_group = scheduler_server_group <EOL> self . schedulerserver_port = schedulerserver_port <EOL> self . service_mapping = service_mapping <EOL> self . service_args_mapping = service_args_mapping <EOL> resource = WorkerResource ( self ) <EOL> self . site_port = reactor . listenTCP ( port , server . Site ( resource ) ) <EOL> self . amqp_host = amqp_host <EOL> self . amqp_vhost = amqp_vhost <EOL> self . amqp_port = amqp_port <EOL> self . amqp_username = amqp_username <EOL> self . amqp_password = amqp_password <EOL> self . amqp_queue = amqp_queue <EOL> self . amqp_exchange = amqp_exchange <EOL> self . amqp_prefetch_count = amqp_prefetch_count <EOL> BaseServer . __init__ ( <EOL> self , <EOL> aws_access_key_id = aws_access_key_id , <EOL> aws_secret_access_key = aws_secret_access_key , <EOL> aws_s3_http_cache_bucket = aws_s3_http_cache_bucket , <EOL> aws_s3_storage_bucket = aws_s3_storage_bucket , <EOL> scheduler_server_group = scheduler_server_group , <EOL> max_simultaneous_requests = max_simultaneous_requests , <EOL> max_requests_per_host_per_second = max_requests_per_host_per_second , <EOL> max_simultaneous_requests_per_host = max_simultaneous_requests_per_host , <EOL> log_file = log_file , <EOL> log_directory = log_directory , <EOL> log_level = log_level ) <EOL> def start ( self ) : <EOL> reactor . callWhenRunning ( self . _start ) <EOL> return self . start_deferred <EOL> @ inlineCallbacks <EOL> def _start ( self ) : <EOL> yield self . getNetworkAddress ( ) <EOL> self . memc = yield self . memc_ClientCreator . connectTCP ( self . memcached_host , self . memcached_port ) <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> self . conn = yield AMQP . createClient ( <EOL> self . amqp_host , <EOL> self . amqp_vhost , <EOL> self . amqp_port ) <EOL> self . auth = yield self . conn . authenticate ( <EOL> self . amqp_username , <EOL> self . amqp_password ) <EOL> self . chan = yield self . conn . channel ( <NUM_LIT:2> ) <EOL> yield self . chan . channel_open ( ) <EOL> yield self . chan . basic_qos ( prefetch_count = self . amqp_prefetch_count ) <EOL> yield self . chan . queue_declare ( <EOL> queue = self . amqp_queue , <EOL> durable = False , <EOL> exclusive = False , <EOL> auto_delete = False ) <EOL> yield self . chan . exchange_declare ( <EOL> exchange = self . amqp_exchange , <EOL> type = "<STR_LIT>" , <EOL> durable = False , <EOL> auto_delete = False ) <EOL> yield self . chan . queue_bind ( <EOL> queue = self . amqp_queue , <EOL> exchange = self . amqp_exchange ) <EOL> yield self . chan . basic_consume ( queue = self . amqp_queue , <EOL> no_ack = False , <EOL> consumer_tag = "<STR_LIT>" ) <EOL> self . queue = yield self . conn . queue ( "<STR_LIT>" ) <EOL> yield BaseServer . start ( self ) <EOL> self . jobsloop = task . LoopingCall ( self . executeJobs ) <EOL> self . jobsloop . start ( <NUM_LIT> ) <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> self . dequeue ( ) <EOL> @ inlineCallbacks <EOL> def shutdown ( self ) : <EOL> LOGGER . debug ( "<STR_LIT>" ) <EOL> try : <EOL> self . jobsloop . cancel ( ) <EOL> except : <EOL> pass <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> yield self . chan . channel_close ( ) <EOL> chan0 = yield self . conn . channel ( <NUM_LIT:0> ) <EOL> yield chan0 . connection_close ( ) <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> yield self . mysql . close ( ) <EOL> @ inlineCallbacks <EOL> def reconnectMemcache ( self ) : <EOL> self . memc = yield self . memc_ClientCreator . connectTCP ( self . memcached_host , self . memcached_port ) <EOL> def dequeue ( self ) : <EOL> LOGGER . debug ( '<STR_LIT>' % ( self . pending_dequeue , self . jobs_complete , len ( self . job_queue ) , len ( self . active_jobs ) ) ) <EOL> if len ( self . job_queue ) <= self . amqp_prefetch_count and not self . pending_dequeue : <EOL> self . pending_dequeue = True <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> d = self . queue . get ( ) <EOL> d . addCallback ( self . _dequeueCallback ) <EOL> d . addErrback ( self . _dequeueErrback ) <EOL> else : <EOL> reactor . callLater ( <NUM_LIT:1> , self . dequeue ) <EOL> def _dequeueErrback ( self , error ) : <EOL> LOGGER . error ( '<STR_LIT>' % error ) <EOL> self . pending_dequeue = False <EOL> reactor . callLater ( <NUM_LIT:0> , self . dequeue ) <EOL> return error <EOL> def _dequeueCallback ( self , msg ) : <EOL> if msg . delivery_tag : <EOL> LOGGER . debug ( '<STR_LIT>' % msg . delivery_tag ) <EOL> d = self . chan . basic_ack ( msg . delivery_tag ) <EOL> d . addCallback ( self . _dequeueCallback2 , msg ) <EOL> d . addErrback ( self . basicAckErrback ) <EOL> return d <EOL> else : <EOL> self . _dequeueCallback2 ( data = True , msg = msg ) <EOL> def _dequeueCallback2 ( self , data , msg ) : <EOL> LOGGER . debug ( '<STR_LIT>' % repr ( msg ) ) <EOL> uuid = UUID ( bytes = msg . content . body ) . hex <EOL> d = self . getJob ( uuid , msg . delivery_tag ) <EOL> d . addCallback ( self . _dequeueCallback3 , msg ) <EOL> d . addErrback ( self . _dequeueErrback ) <EOL> def _dequeueCallback3 ( self , job , msg ) : <EOL> if job is not None : <EOL> if job [ '<STR_LIT>' ] in self . functions : <EOL> LOGGER . debug ( '<STR_LIT>' ) <EOL> job [ '<STR_LIT>' ] = self . functions [ job [ '<STR_LIT>' ] ] <EOL> if not job . has_key ( '<STR_LIT>' ) : <EOL> job [ '<STR_LIT>' ] = self . mapKwargs ( job ) <EOL> if not job . has_key ( '<STR_LIT>' ) : <EOL> job [ '<STR_LIT>' ] = msg . delivery_tag <EOL> job [ "<STR_LIT>" ] = None <EOL> if self . functions [ job [ '<STR_LIT>' ] ] [ "<STR_LIT>" ] : <EOL> d = self . getReservationFastCache ( job [ '<STR_LIT>' ] ) <EOL> d . addCallback ( self . _dequeueCallback4 , job ) <EOL> self . job_queue_a ( job ) <EOL> else : <EOL> LOGGER . error ( "<STR_LIT>" % job [ '<STR_LIT>' ] ) <EOL> self . pending_dequeue = False <EOL> reactor . callLater ( <NUM_LIT:0> , self . dequeue ) <EOL> def _dequeueCallback4 ( self , data , job ) : <EOL> job [ "<STR_LIT>" ] = data <EOL> def executeJobs ( self ) : <EOL> while len ( self . job_queue ) > <NUM_LIT:0> and len ( self . active_jobs ) < self . simultaneous_jobs : <EOL> job = self . job_queue . pop ( <NUM_LIT:0> ) <EOL> exposed_function = job [ "<STR_LIT>" ] <EOL> kwargs = job [ "<STR_LIT>" ] <EOL> function_name = job [ "<STR_LIT>" ] <EOL> if job . has_key ( '<STR_LIT>' ) : <EOL> uuid = job [ "<STR_LIT>" ] <EOL> else : <EOL> uuid = UUID ( bytes = msg . content . body ) . hex <EOL> d = self . callExposedFunction ( <EOL> exposed_function [ "<STR_LIT>" ] , <EOL> kwargs , <EOL> function_name , <EOL> reservation_fast_cache = job [ "<STR_LIT>" ] , <EOL> uuid = uuid ) <EOL> d . addCallback ( self . _executeJobCallback , job ) <EOL> d . addErrback ( self . workerErrback , '<STR_LIT>' , job [ '<STR_LIT>' ] ) <EOL> def _executeJobCallback ( self , data , job ) : <EOL> self . jobs_complete += <NUM_LIT:1> <EOL> LOGGER . debug ( '<STR_LIT>' % ( self . jobs_complete , len ( self . job_queue ) , len ( self . active_jobs ) ) ) <EOL> if job . has_key ( '<STR_LIT>' ) : <EOL> del ( job [ '<STR_LIT>' ] ) <EOL> d = self . memc . set ( job [ '<STR_LIT>' ] , simplejson . dumps ( job ) , <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT:7> ) <EOL> d . addCallback ( self . _executeJobCallback2 ) <EOL> d . addErrback ( self . workerErrback , '<STR_LIT>' , job [ '<STR_LIT>' ] ) <EOL> return d <EOL> def _executeJobCallback2 ( self , data ) : <EOL> return <EOL> def workerErrback ( self , error , function_name = '<STR_LIT>' , delivery_tag = None ) : <EOL> LOGGER . error ( '<STR_LIT>' % ( function_name , str ( error ) ) ) <EOL> LOGGER . debug ( '<STR_LIT>' % ( len ( self . job_queue ) , len ( self . active_jobs ) ) ) <EOL> LOGGER . debug ( '<STR_LIT>' % repr ( self . active_jobs ) ) <EOL> self . pending_dequeue = False <EOL> if '<STR_LIT>' in str ( error ) : <EOL> LOGGER . info ( '<STR_LIT>' ) <EOL> d = self . reconnectMemcache ( ) <EOL> d . addCallback ( self . _reconnectMemcacheCallback ) <EOL> return d <EOL> else : <EOL> return error <EOL> def _reconnectMemcacheCallback ( self , data ) : <EOL> return <EOL> def _basicAckCallback ( self , data ) : <EOL> return <EOL> def basicAckErrback ( self , error ) : <EOL> LOGGER . error ( '<STR_LIT>' % ( error ) ) <EOL> return <EOL> def getJob ( self , uuid , delivery_tag ) : <EOL> d = self . memc . get ( uuid ) <EOL> d . addCallback ( self . _getJobCallback , uuid , delivery_tag ) <EOL> d . addErrback ( self . workerErrback , '<STR_LIT>' , delivery_tag ) <EOL> return d <EOL> def _getJobCallback ( self , account , uuid , delivery_tag ) : <EOL> job = account [ <NUM_LIT:1> ] <EOL> if not job : <EOL> LOGGER . debug ( '<STR_LIT>' % uuid ) <EOL> sql = "<STR_LIT>" % uuid <EOL> d = self . mysql . runQuery ( sql ) <EOL> d . addCallback ( self . getAccountMySQL , uuid , delivery_tag ) <EOL> d . addErrback ( self . workerErrback , '<STR_LIT>' , delivery_tag ) <EOL> return d <EOL> else : <EOL> LOGGER . debug ( '<STR_LIT>' % uuid ) <EOL> return simplejson . loads ( job ) <EOL> def getAccountMySQL ( self , spider_info , uuid , delivery_tag ) : <EOL> if spider_info : <EOL> account_type = spider_info [ <NUM_LIT:0> ] [ '<STR_LIT:type>' ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] <EOL> sql = "<STR_LIT>" % ( account_type . lower ( ) , spider_info [ <NUM_LIT:0> ] [ '<STR_LIT>' ] ) <EOL> d = self . mysql . runQuery ( sql ) <EOL> d . addCallback ( self . createJob , spider_info , uuid , delivery_tag ) <EOL> d . addErrback ( self . workerErrback , '<STR_LIT>' , delivery_tag ) <EOL> return d <EOL> LOGGER . debug ( '<STR_LIT>' % uuid ) <EOL> return None <EOL> def createJob ( self , account_info , spider_info , uuid , delivery_tag ) : <EOL> job = { } <EOL> account = account_info [ <NUM_LIT:0> ] <EOL> function_name = spider_info [ <NUM_LIT:0> ] [ '<STR_LIT:type>' ] <EOL> job [ '<STR_LIT:type>' ] = function_name . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] <EOL> if self . service_mapping and self . service_mapping . has_key ( function_name ) : <EOL> LOGGER . debug ( '<STR_LIT>' % ( function_name , self . service_mapping [ function_name ] ) ) <EOL> function_name = self . service_mapping [ function_name ] <EOL> job [ '<STR_LIT>' ] = function_name <EOL> job [ '<STR_LIT>' ] = uuid <EOL> job [ '<STR_LIT>' ] = account <EOL> job [ '<STR_LIT>' ] = delivery_tag <EOL> return job <EOL> def mapKwargs ( self , job ) : <EOL> kwargs = { } <EOL> service_name = job [ '<STR_LIT>' ] . split ( '<STR_LIT:/>' ) [ <NUM_LIT:0> ] <EOL> if service_name in self . service_args_mapping : <EOL> for key in self . service_args_mapping [ service_name ] : <EOL> if key in job [ '<STR_LIT>' ] : <EOL> job [ '<STR_LIT>' ] [ self . service_args_mapping [ service_name ] [ key ] ] = job [ '<STR_LIT>' ] [ key ] <EOL> for arg in job [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> if arg in job : <EOL> kwargs [ arg ] = job [ arg ] <EOL> elif arg in job [ '<STR_LIT>' ] : <EOL> kwargs [ arg ] = job [ '<STR_LIT>' ] [ arg ] <EOL> for arg in job [ '<STR_LIT>' ] [ '<STR_LIT>' ] : <EOL> if arg in job [ '<STR_LIT>' ] : <EOL> kwargs [ arg ] = job [ '<STR_LIT>' ] [ arg ] <EOL> LOGGER . debug ( '<STR_LIT>' % ( job [ '<STR_LIT>' ] , repr ( kwargs ) ) ) <EOL> return kwargs <EOL> def getNetworkAddress ( self ) : <EOL> d = getNetworkAddress ( ) <EOL> d . addCallback ( self . _getNetworkAddressCallback ) <EOL> d . addErrback ( self . _getNetworkAddressErrback ) <EOL> return d <EOL> def _getNetworkAddressCallback ( self , data ) : <EOL> if "<STR_LIT>" in data : <EOL> self . public_ip = data [ "<STR_LIT>" ] <EOL> self . network_information [ "<STR_LIT>" ] = self . public_ip <EOL> if "<STR_LIT>" in data : <EOL> self . local_ip = data [ "<STR_LIT>" ] <EOL> self . network_information [ "<STR_LIT>" ] = self . local_ip <EOL> def _getNetworkAddressErrback ( self , error ) : <EOL> message = "<STR_LIT>" <EOL> LOGGER . error ( message ) <EOL> raise Exception ( message ) <EOL> def getReservationFastCache ( self , uuid ) : <EOL> d = self . memc . get ( "<STR_LIT>" % uuid ) <EOL> d . addCallback ( self . _getReservationFastCacheCallback , uuid ) <EOL> d . addErrback ( self . _getReservationFastCacheErrback , uuid ) <EOL> return d <EOL> def _getReservationFastCacheCallback ( self , data , uuid ) : <EOL> flags , value = data <EOL> if value : <EOL> LOGGER . debug ( "<STR_LIT>" % uuid ) <EOL> return value <EOL> else : <EOL> LOGGER . debug ( "<STR_LIT>" % uuid ) <EOL> return None <EOL> def _getReservationFastCacheErrback ( self , error , uuid ) : <EOL> LOGGER . debug ( "<STR_LIT>" % uuid ) <EOL> return None <EOL> def setReservationFastCache ( self , uuid , data ) : <EOL> if not isinstance ( data , str ) : <EOL> raise Exception ( "<STR_LIT>" ) <EOL> if uuid is None : <EOL> return None <EOL> d = self . memc . set ( "<STR_LIT>" % uuid , data , <NUM_LIT> * <NUM_LIT> * <NUM_LIT> * <NUM_LIT:7> ) <EOL> d . addCallback ( self . _setReservationFastCacheCallback , uuid ) <EOL> d . addErrback ( self . _setReservationFastCacheErrback ) <EOL> def _setReservationFastCacheCallback ( self , data , uuid ) : <EOL> LOGGER . debug ( "<STR_LIT>" % uuid ) <EOL> def _setReservationFastCacheErrback ( self , error ) : <EOL> LOGGER . error ( str ( error ) ) </s>
