<s> '''<STR_LIT>''' <EOL> from ConfigParser import ConfigParser <EOL> import cPickle as pickle <EOL> import functools <EOL> from interruptingcow import timeout <EOL> import logging <EOL> from multiprocessing import Process <EOL> from optparse import OptionParser <EOL> import os <EOL> from random import randint <EOL> import signal <EOL> from laikaboss . objectmodel import ScanResult , ScanObject , QuitScanException <EOL> import sys <EOL> import syslog <EOL> import time <EOL> import traceback <EOL> import zlib <EOL> import zmq <EOL> import json <EOL> import base64 <EOL> from distutils . util import strtobool <EOL> SHUTDOWN_GRACE_TIMEOUT_DEFAULT = <NUM_LIT:30> <EOL> LRU_READY = "<STR_LIT>" <EOL> LRU_RESULT_READY = "<STR_LIT>" <EOL> LRU_RESULT_QUIT = "<STR_LIT>" <EOL> LRU_QUIT = "<STR_LIT>" <EOL> REQ_TYPE_PICKLE = '<STR_LIT:1>' <EOL> REQ_TYPE_PICKLE_ZLIB = '<STR_LIT:2>' <EOL> REQ_TYPE_JSON = '<STR_LIT:3>' <EOL> REQ_TYPE_JSON_ZLIB = '<STR_LIT:4>' <EOL> class ResultEncoder ( json . JSONEncoder ) : <EOL> def default ( self , obj ) : <EOL> if isinstance ( obj , ScanObject ) : <EOL> newdict = obj . __dict__ . copy ( ) <EOL> del newdict [ '<STR_LIT>' ] <EOL> return newdict <EOL> if isinstance ( obj , ScanResult ) : <EOL> res = { } <EOL> res [ '<STR_LIT>' ] = obj . rootUID <EOL> res [ '<STR_LIT:source>' ] = obj . source <EOL> res [ '<STR_LIT>' ] = obj . level <EOL> res [ '<STR_LIT>' ] = obj . startTime <EOL> tmpFiles = { } <EOL> for uid , sO in obj . files . iteritems ( ) : <EOL> tmpFiles [ str ( uid ) ] = sO <EOL> res [ '<STR_LIT>' ] = tmpFiles <EOL> return res <EOL> return json . JSONEncoder . default ( self , obj ) <EOL> CONFIGS = { } <EOL> DEFAULT_CONFIGS = { <EOL> '<STR_LIT>' : '<STR_LIT:4>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:False>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT:False>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' <EOL> } <EOL> def log_debug ( message ) : <EOL> '''<STR_LIT>''' <EOL> syslog . syslog ( syslog . LOG_DEBUG , "<STR_LIT>" % ( os . getpid ( ) , message ) ) <EOL> def get_option ( option , default = '<STR_LIT>' ) : <EOL> '''<STR_LIT>''' <EOL> value = default <EOL> if option in CONFIGS : <EOL> value = CONFIGS [ option ] <EOL> elif option in DEFAULT_CONFIGS : <EOL> value = DEFAULT_CONFIGS [ option ] <EOL> return value <EOL> def shutdown_handler ( proc , signum , frame ) : <EOL> '''<STR_LIT>''' <EOL> logging . debug ( "<STR_LIT>" , signum ) <EOL> proc . shutdown ( ) <EOL> class AsyncBroker ( Process ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , broker_backend_address , broker_frontend_address ) : <EOL> '''<STR_LIT>''' <EOL> super ( AsyncBroker , self ) . __init__ ( ) <EOL> self . broker_backend_address = broker_backend_address <EOL> self . broker_frontend_address = broker_frontend_address <EOL> self . keep_running = True <EOL> def shutdown ( self ) : <EOL> '''<STR_LIT>''' <EOL> logging . debug ( "<STR_LIT>" ) <EOL> self . keep_running = False <EOL> def run ( self ) : <EOL> '''<STR_LIT>''' <EOL> logging . debug ( "<STR_LIT>" ) <EOL> self . keep_running = True <EOL> signal . signal ( signal . SIGTERM , functools . partial ( shutdown_handler , self ) ) <EOL> signal . signal ( signal . SIGINT , functools . partial ( shutdown_handler , self ) ) <EOL> context = zmq . Context ( <NUM_LIT:1> ) <EOL> backend = context . socket ( zmq . ROUTER ) <EOL> backend . bind ( self . broker_backend_address ) <EOL> backend_poller = zmq . Poller ( ) <EOL> backend_poller . register ( backend , zmq . POLLIN ) <EOL> frontend = context . socket ( zmq . PULL ) <EOL> frontend . bind ( self . broker_frontend_address ) <EOL> frontend_poller = zmq . Poller ( ) <EOL> frontend_poller . register ( frontend , zmq . POLLIN ) <EOL> frontend_poller . register ( backend , zmq . POLLIN ) <EOL> available_workers = [ ] <EOL> while self . keep_running : <EOL> logging . debug ( "<STR_LIT>" , <EOL> str ( available_workers ) ) <EOL> try : <EOL> if available_workers : <EOL> msgs = dict ( frontend_poller . poll ( ) ) <EOL> else : <EOL> msgs = dict ( backend_poller . poll ( ) ) <EOL> if msgs . get ( frontend ) == zmq . POLLIN : <EOL> msg = frontend . recv_multipart ( ) <EOL> worker_id = available_workers . pop ( <NUM_LIT:0> ) <EOL> backend . send_multipart ( [ worker_id , '<STR_LIT>' , worker_id , '<STR_LIT>' ] + msg ) <EOL> if msgs . get ( backend ) == zmq . POLLIN : <EOL> msg = backend . recv_multipart ( ) <EOL> worker_id = msg [ <NUM_LIT:0> ] <EOL> status = msg [ <NUM_LIT:2> ] <EOL> if status == LRU_READY or status == LRU_RESULT_READY : <EOL> logging . debug ( "<STR_LIT>" , worker_id ) <EOL> if worker_id not in available_workers : <EOL> available_workers . append ( worker_id ) <EOL> elif status == LRU_RESULT_QUIT or status == LRU_QUIT : <EOL> logging . debug ( "<STR_LIT>" , worker_id ) <EOL> try : <EOL> available_workers . remove ( worker_id ) <EOL> except ValueError : <EOL> pass <EOL> else : <EOL> logging . warn ( "<STR_LIT>" ) <EOL> except zmq . ZMQError as zmqerror : <EOL> if "<STR_LIT>" not in str ( zmqerror ) : <EOL> logging . exception ( "<STR_LIT>" ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> class SyncBroker ( Process ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , broker_backend_address , broker_frontend_address , <EOL> shutdown_grace_timeout = SHUTDOWN_GRACE_TIMEOUT_DEFAULT ) : <EOL> '''<STR_LIT>''' <EOL> super ( SyncBroker , self ) . __init__ ( ) <EOL> self . broker_backend_address = broker_backend_address <EOL> self . broker_frontend_address = broker_frontend_address <EOL> self . shutdown_grace_timeout = shutdown_grace_timeout <EOL> self . keep_running = True <EOL> def shutdown ( self ) : <EOL> '''<STR_LIT>''' <EOL> logging . debug ( "<STR_LIT>" ) <EOL> self . keep_running = False <EOL> def run ( self ) : <EOL> '''<STR_LIT>''' <EOL> logging . debug ( "<STR_LIT>" ) <EOL> self . keep_running = True <EOL> signal . signal ( signal . SIGTERM , functools . partial ( shutdown_handler , self ) ) <EOL> signal . signal ( signal . SIGINT , functools . partial ( shutdown_handler , self ) ) <EOL> context = zmq . Context ( <NUM_LIT:1> ) <EOL> backend = context . socket ( zmq . ROUTER ) <EOL> backend . bind ( self . broker_backend_address ) <EOL> backend_poller = zmq . Poller ( ) <EOL> backend_poller . register ( backend , zmq . POLLIN ) <EOL> frontend = context . socket ( zmq . ROUTER ) <EOL> frontend . bind ( self . broker_frontend_address ) <EOL> frontend_poller = zmq . Poller ( ) <EOL> frontend_poller . register ( frontend , zmq . POLLIN ) <EOL> frontend_poller . register ( backend , zmq . POLLIN ) <EOL> available_workers = [ ] <EOL> working_workers = [ ] <EOL> while self . keep_running : <EOL> logging . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , str ( available_workers ) , str ( working_workers ) ) <EOL> try : <EOL> if available_workers : <EOL> msgs = dict ( frontend_poller . poll ( ) ) <EOL> else : <EOL> msgs = dict ( backend_poller . poll ( ) ) <EOL> if msgs . get ( frontend ) == zmq . POLLIN : <EOL> msg = frontend . recv_multipart ( ) <EOL> worker_id = available_workers . pop ( <NUM_LIT:0> ) <EOL> backend . send_multipart ( [ worker_id , '<STR_LIT>' ] + msg ) <EOL> working_workers . append ( worker_id ) <EOL> if msgs . get ( backend ) == zmq . POLLIN : <EOL> msg = backend . recv_multipart ( ) <EOL> worker_id = msg [ <NUM_LIT:0> ] <EOL> status = msg [ <NUM_LIT:2> ] <EOL> if status == LRU_READY : <EOL> logging . debug ( "<STR_LIT>" , worker_id ) <EOL> if ( worker_id not in available_workers and <EOL> worker_id not in working_workers ) : <EOL> available_workers . append ( worker_id ) <EOL> elif status == LRU_RESULT_READY : <EOL> logging . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , worker_id ) <EOL> try : <EOL> working_workers . remove ( worker_id ) <EOL> except ValueError : <EOL> pass <EOL> frontend . send_multipart ( msg [ <NUM_LIT:4> : ] ) <EOL> if ( worker_id not in available_workers and <EOL> worker_id not in working_workers ) : <EOL> available_workers . append ( worker_id ) <EOL> elif status == LRU_RESULT_QUIT : <EOL> logging . debug ( "<STR_LIT>" <EOL> "<STR_LIT>" , worker_id ) <EOL> try : <EOL> working_workers . remove ( worker_id ) <EOL> except ValueError : <EOL> pass <EOL> frontend . send_multipart ( msg [ <NUM_LIT:4> : ] ) <EOL> elif status == LRU_QUIT : <EOL> logging . debug ( "<STR_LIT>" , worker_id ) <EOL> try : <EOL> available_workers . remove ( worker_id ) <EOL> except ValueError : <EOL> pass <EOL> else : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> except zmq . ZMQError as zmqerror : <EOL> if "<STR_LIT>" not in str ( zmqerror ) : <EOL> logging . exception ( "<STR_LIT>" ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> poll_timeout = ( self . shutdown_grace_timeout / <NUM_LIT:3> ) * <NUM_LIT:1000> or <NUM_LIT:1> <EOL> start_time = time . time ( ) <EOL> while ( working_workers and <EOL> ( time . time ( ) - start_time < self . shutdown_grace_timeout ) ) : <EOL> logging . debug ( "<STR_LIT>" <EOL> "<STR_LIT:%s>" , str ( working_workers ) ) <EOL> msgs = dict ( backend_poller . poll ( poll_timeout ) ) <EOL> if msgs . get ( backend ) == zmq . POLLIN : <EOL> msg = backend . recv_multipart ( ) <EOL> worker_id = msg [ <NUM_LIT:0> ] <EOL> status = msg [ <NUM_LIT:2> ] <EOL> if status == LRU_RESULT_READY or status == LRU_RESULT_QUIT : <EOL> logging . debug ( "<STR_LIT>" , <EOL> worker_id ) <EOL> try : <EOL> working_workers . remove ( worker_id ) <EOL> except ValueError : <EOL> pass <EOL> frontend . send_multipart ( msg [ <NUM_LIT:4> : ] ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> class Worker ( Process ) : <EOL> '''<STR_LIT>''' <EOL> def __init__ ( self , config_location , broker_address , max_scan_items , ttl , <EOL> logresult = False , <EOL> poll_timeout = <NUM_LIT> , <EOL> shutdown_grace_timeout = SHUTDOWN_GRACE_TIMEOUT_DEFAULT ) : <EOL> '''<STR_LIT>''' <EOL> super ( Worker , self ) . __init__ ( ) <EOL> self . config_location = config_location <EOL> self . max_scan_items = max_scan_items <EOL> self . ttl = ttl <EOL> self . shutdown_grace_timeout = shutdown_grace_timeout <EOL> self . keep_running = False <EOL> self . broker_address = broker_address <EOL> self . identity = "<STR_LIT>" % ( randint ( <NUM_LIT:0> , <NUM_LIT> ) , randint ( <NUM_LIT:0> , <NUM_LIT> ) ) <EOL> self . broker = None <EOL> self . broker_poller = zmq . Poller ( ) <EOL> self . poll_timeout = poll_timeout * <NUM_LIT:1000> <EOL> self . logresult = logresult <EOL> def perform_scan ( self , poll_timeout ) : <EOL> '''<STR_LIT>''' <EOL> from laikaboss . dispatch import Dispatch <EOL> from laikaboss . objectmodel import ScanResult , ExternalObject , ExternalVars <EOL> from laikaboss . util import log_result <EOL> try : <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> tasks = dict ( self . broker_poller . poll ( poll_timeout ) ) <EOL> if tasks . get ( self . broker ) == zmq . POLLIN : <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> task = self . broker . recv_multipart ( ) <EOL> client_id = task [ <NUM_LIT:1> ] <EOL> if len ( task ) == <NUM_LIT:6> : <EOL> request_type = task [ <NUM_LIT:3> ] <EOL> request = task [ <NUM_LIT:5> ] <EOL> if request_type in [ REQ_TYPE_PICKLE , REQ_TYPE_PICKLE_ZLIB ] : <EOL> if request_type == REQ_TYPE_PICKLE_ZLIB : <EOL> externalObject = pickle . loads ( zlib . decompress ( request ) ) <EOL> else : <EOL> externalObject = pickle . loads ( request ) <EOL> elif request_type in [ REQ_TYPE_JSON , REQ_TYPE_JSON_ZLIB ] : <EOL> if request_type == REQ_TYPE_JSON_ZLIB : <EOL> jsonRequest = json . loads ( zlib . decompress ( request ) ) <EOL> else : <EOL> jsonRequest = json . loads ( request ) <EOL> if not '<STR_LIT>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> else : <EOL> try : <EOL> jsonRequest [ '<STR_LIT>' ] = base64 . b64decode ( jsonRequest [ '<STR_LIT>' ] ) <EOL> except : <EOL> jsonRequest [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if not '<STR_LIT:filename>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT:filename>' ] = '<STR_LIT>' <EOL> if not '<STR_LIT>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if not '<STR_LIT>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if not '<STR_LIT>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT>' ] = [ ] <EOL> if not '<STR_LIT>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if not '<STR_LIT:source>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT:source>' ] = '<STR_LIT>' <EOL> if not '<STR_LIT>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> if not '<STR_LIT>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT>' ] = { } <EOL> if not '<STR_LIT>' in jsonRequest : <EOL> jsonRequest [ '<STR_LIT>' ] = <NUM_LIT:2> <EOL> externalVars = ExternalVars ( filename = jsonRequest [ '<STR_LIT:filename>' ] , <EOL> ephID = jsonRequest [ '<STR_LIT>' ] , <EOL> uniqID = jsonRequest [ '<STR_LIT>' ] , <EOL> contentType = jsonRequest [ '<STR_LIT>' ] , <EOL> timestamp = jsonRequest [ '<STR_LIT>' ] , <EOL> source = jsonRequest [ '<STR_LIT:source>' ] , <EOL> origRootUID = jsonRequest [ '<STR_LIT>' ] , <EOL> extMetaData = jsonRequest [ '<STR_LIT>' ] ) <EOL> externalObject = ExternalObject ( buffer = jsonRequest [ '<STR_LIT>' ] , <EOL> level = jsonRequest [ '<STR_LIT>' ] , <EOL> externalVars = externalVars ) <EOL> else : <EOL> return [ client_id , '<STR_LIT>' , '<STR_LIT>' ] <EOL> result = ScanResult ( <EOL> source = externalObject . externalVars . source , <EOL> level = externalObject . level ) <EOL> result . startTime = time . time ( ) <EOL> try : <EOL> Dispatch ( externalObject . buffer , result , <NUM_LIT:0> , <EOL> externalVars = externalObject . externalVars ) <EOL> except QuitScanException : <EOL> raise <EOL> except : <EOL> exc_type , exc_value , exc_traceback = sys . exc_info ( ) <EOL> log_debug ( <EOL> "<STR_LIT>" % ( <EOL> externalObject . externalVars . filename , <EOL> repr ( traceback . format_exception ( <EOL> exc_type , exc_value , exc_traceback ) ) ) ) <EOL> if self . logresult : <EOL> log_result ( result ) <EOL> if request_type == REQ_TYPE_PICKLE_ZLIB : <EOL> result = zlib . compress ( <EOL> pickle . dumps ( result , pickle . HIGHEST_PROTOCOL ) ) <EOL> elif request_type == REQ_TYPE_PICKLE : <EOL> result = pickle . dumps ( result , pickle . HIGHEST_PROTOCOL ) <EOL> elif request_type == REQ_TYPE_JSON_ZLIB : <EOL> result = zlib . compress ( <EOL> json . dumps ( result , cls = ResultEncoder ) ) <EOL> elif request_type == REQ_TYPE_JSON : <EOL> result = json . dumps ( result , cls = ResultEncoder ) <EOL> return [ client_id , '<STR_LIT>' , result ] <EOL> else : <EOL> return [ client_id , '<STR_LIT>' , '<STR_LIT>' ] <EOL> except zmq . ZMQError as zmqerror : <EOL> if "<STR_LIT>" not in str ( zmqerror ) : <EOL> logging . exception ( "<STR_LIT>" , self . identity ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> return None <EOL> def shutdown ( self ) : <EOL> '''<STR_LIT>''' <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> self . keep_running = False <EOL> raise QuitScanException ( ) <EOL> def run ( self ) : <EOL> '''<STR_LIT>''' <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> from laikaboss import config <EOL> from laikaboss . dispatch import close_modules <EOL> from laikaboss . util import init_logging <EOL> logging . debug ( "<STR_LIT>" , self . config_location ) <EOL> config . init ( path = self . config_location ) <EOL> init_logging ( ) <EOL> log_debug ( "<STR_LIT>" % ( self . identity , time . time ( ) ) ) <EOL> self . keep_running = True <EOL> perform_grace_check = False <EOL> signal . signal ( signal . SIGTERM , functools . partial ( shutdown_handler , self ) ) <EOL> signal . signal ( signal . SIGINT , functools . partial ( shutdown_handler , self ) ) <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> context = zmq . Context ( <NUM_LIT:1> ) <EOL> self . broker = context . socket ( zmq . DEALER ) <EOL> self . broker . setsockopt ( zmq . IDENTITY , self . identity ) <EOL> self . broker . connect ( self . broker_address ) <EOL> self . broker_poller . register ( self . broker , zmq . POLLIN ) <EOL> self . broker . send_multipart ( [ '<STR_LIT>' , LRU_READY ] ) <EOL> counter = <NUM_LIT:0> <EOL> start_time = time . time ( ) + randint ( <NUM_LIT:1> , <NUM_LIT> ) <EOL> while self . keep_running : <EOL> try : <EOL> result = self . perform_scan ( self . poll_timeout ) <EOL> if result : <EOL> counter += <NUM_LIT:1> <EOL> should_quit = ( <EOL> counter >= self . max_scan_items or <EOL> ( ( time . time ( ) - start_time ) / <NUM_LIT> ) >= self . ttl or <EOL> not self . keep_running ) <EOL> status = LRU_QUIT <EOL> if result : <EOL> if should_quit : <EOL> status = LRU_RESULT_QUIT <EOL> else : <EOL> status = LRU_RESULT_READY <EOL> else : <EOL> if should_quit : <EOL> status = LRU_QUIT <EOL> perform_grace_check = True <EOL> else : <EOL> status = LRU_READY <EOL> if result : <EOL> reply = [ '<STR_LIT>' , status , '<STR_LIT>' ] + result <EOL> else : <EOL> reply = [ '<STR_LIT>' , status ] <EOL> tracker = self . broker . send_multipart ( reply , copy = False , track = True ) <EOL> while not tracker . done and result : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> if should_quit : <EOL> self . keep_running = False <EOL> except zmq . ZMQError as zmqerror : <EOL> if "<STR_LIT>" not in str ( zmqerror ) : <EOL> logging . exception ( "<STR_LIT>" , self . identity ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> except QuitScanException : <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> break <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> if perform_grace_check : <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> try : <EOL> result = self . perform_scan ( self . poll_timeout ) <EOL> if result : <EOL> reply = [ '<STR_LIT>' , LRU_RESULT_QUIT , '<STR_LIT>' ] + result <EOL> tracker = self . broker . send_multipart ( reply , copy = False , track = True ) <EOL> while not tracker . done : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> except zmq . ZMQError as zmqerror : <EOL> if "<STR_LIT>" not in str ( zmqerror ) : <EOL> logging . exception ( "<STR_LIT>" , self . identity ) <EOL> else : <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> try : <EOL> with timeout ( self . shutdown_grace_timeout , exception = QuitScanException ) : <EOL> close_modules ( ) <EOL> except QuitScanException : <EOL> logging . debug ( "<STR_LIT>" , <EOL> self . identity ) <EOL> log_debug ( "<STR_LIT>" % ( <EOL> self . identity , counter , time . time ( ) - start_time ) ) <EOL> logging . debug ( "<STR_LIT>" , self . identity ) <EOL> KEEP_RUNNING = True <EOL> def main ( ) : <EOL> '''<STR_LIT>''' <EOL> parser = OptionParser ( usage = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store_true>" , default = False , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:string>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT:-c>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:string>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:string>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:string>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:string>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store_true>" , default = False , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:string>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:int>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:int>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:int>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store_true>" , default = False , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> parser . add_option ( "<STR_LIT>" , "<STR_LIT>" , <EOL> action = "<STR_LIT:store>" , type = "<STR_LIT:int>" , <EOL> dest = "<STR_LIT>" , <EOL> help = "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> ( options , _ ) = parser . parse_args ( ) <EOL> config_location = '<STR_LIT>' <EOL> if options . laikad_config_path : <EOL> config_location = options . laikad_config_path <EOL> if not os . path . exists ( options . laikad_config_path ) : <EOL> print "<STR_LIT>" <EOL> return <NUM_LIT:1> <EOL> elif os . path . exists ( DEFAULT_CONFIGS [ '<STR_LIT>' ] ) : <EOL> config_location = DEFAULT_CONFIGS [ '<STR_LIT>' ] <EOL> elif os . path . exists ( DEFAULT_CONFIGS [ '<STR_LIT>' ] ) : <EOL> config_location = DEFAULT_CONFIGS [ '<STR_LIT>' ] <EOL> else : <EOL> print '<STR_LIT>' % ( DEFAULT_CONFIGS [ '<STR_LIT>' ] , DEFAULT_CONFIGS [ '<STR_LIT>' ] ) <EOL> return <NUM_LIT:1> <EOL> config_parser = ConfigParser ( ) <EOL> config_parser . read ( config_location ) <EOL> for section in config_parser . sections ( ) : <EOL> CONFIGS . update ( dict ( config_parser . items ( section ) ) ) <EOL> if options . laikaboss_config_path : <EOL> laikaboss_config_path = options . laikaboss_config_path <EOL> logging . debug ( "<STR_LIT>" % options . laikaboss_config_path ) <EOL> if not os . path . exists ( options . laikaboss_config_path ) : <EOL> print "<STR_LIT>" <EOL> return <NUM_LIT:1> <EOL> elif os . path . exists ( get_option ( '<STR_LIT>' ) ) : <EOL> laikaboss_config_path = get_option ( '<STR_LIT>' ) <EOL> elif os . path . exists ( DEFAULT_CONFIGS [ '<STR_LIT>' ] ) : <EOL> laikaboss_config_path = DEFAULT_CONFIGS [ '<STR_LIT>' ] <EOL> elif os . path . exists ( DEFAULT_CONFIGS [ '<STR_LIT>' ] ) : <EOL> laikaboss_config_path = DEFAULT_CONFIGS [ '<STR_LIT>' ] <EOL> else : <EOL> print '<STR_LIT>' % ( DEFAULT_CONFIGS [ '<STR_LIT>' ] , DEFAULT_CONFIGS [ '<STR_LIT>' ] ) <EOL> return <NUM_LIT:1> <EOL> if options . num_procs : <EOL> num_procs = options . num_procs <EOL> else : <EOL> num_procs = int ( get_option ( '<STR_LIT>' ) ) <EOL> if options . ttl : <EOL> ttl = options . ttl <EOL> else : <EOL> ttl = int ( get_option ( '<STR_LIT>' ) ) <EOL> if options . time_ttl : <EOL> time_ttl = options . time_ttl <EOL> else : <EOL> time_ttl = int ( get_option ( '<STR_LIT>' ) ) <EOL> if options . broker_backend_address : <EOL> broker_backend_address = options . broker_backend_address <EOL> else : <EOL> broker_backend_address = get_option ( '<STR_LIT>' ) <EOL> if options . broker_frontend_address : <EOL> broker_frontend_address = options . broker_frontend_address <EOL> else : <EOL> broker_frontend_address = get_option ( '<STR_LIT>' ) <EOL> if options . worker_connect_address : <EOL> worker_connect_address = options . worker_connect_address <EOL> else : <EOL> worker_connect_address = get_option ( '<STR_LIT>' ) <EOL> if options . gracetimeout : <EOL> gracetimeout = options . gracetimeout <EOL> else : <EOL> gracetimeout = int ( get_option ( '<STR_LIT>' ) ) <EOL> if options . run_async : <EOL> async = True <EOL> else : <EOL> async = strtobool ( get_option ( '<STR_LIT>' ) ) <EOL> logresult = strtobool ( get_option ( '<STR_LIT>' ) ) <EOL> runas_uid = None <EOL> runas_gid = None <EOL> if options . runas_uid : <EOL> from pwd import getpwnam <EOL> runas_uid = getpwnam ( options . runas_uid ) . pw_uid <EOL> runas_gid = getpwnam ( options . runas_uid ) . pw_gid <EOL> if options . debug : <EOL> logging . basicConfig ( level = logging . DEBUG ) <EOL> try : <EOL> if runas_uid : <EOL> os . setgid ( runas_gid ) <EOL> os . setuid ( runas_uid ) <EOL> except OSError : <EOL> print "<STR_LIT>" % runas_uid <EOL> def shutdown ( signum , frame ) : <EOL> '''<STR_LIT>''' <EOL> logging . debug ( "<STR_LIT>" ) <EOL> global KEEP_RUNNING <EOL> KEEP_RUNNING = False <EOL> signal . signal ( signal . SIGTERM , shutdown ) <EOL> signal . signal ( signal . SIGINT , shutdown ) <EOL> broker_proc = None <EOL> if not options . no_broker : <EOL> if async : <EOL> broker_proc = AsyncBroker ( broker_backend_address , broker_frontend_address ) <EOL> else : <EOL> broker_proc = SyncBroker ( broker_backend_address , broker_frontend_address , gracetimeout ) <EOL> broker_proc . start ( ) <EOL> workers = [ ] <EOL> for _ in range ( num_procs ) : <EOL> worker_proc = Worker ( laikaboss_config_path , worker_connect_address , ttl , <EOL> time_ttl , logresult , int ( get_option ( '<STR_LIT>' ) ) , gracetimeout ) <EOL> worker_proc . start ( ) <EOL> workers . append ( worker_proc ) <EOL> while KEEP_RUNNING : <EOL> if not options . no_broker and not broker_proc . is_alive ( ) : <EOL> if async : <EOL> broker_proc = AsyncBroker ( broker_backend_address , broker_frontend_address ) <EOL> else : <EOL> broker_proc = SyncBroker ( broker_backend_address , broker_frontend_address , <EOL> gracetimeout ) <EOL> broker_proc . start ( ) <EOL> dead_workers = [ ] <EOL> for worker_proc in workers : <EOL> if not worker_proc . is_alive ( ) : <EOL> dead_workers . append ( worker_proc ) <EOL> for worker_proc in dead_workers : <EOL> workers . remove ( worker_proc ) <EOL> new_proc = Worker ( laikaboss_config_path , worker_connect_address , ttl , time_ttl , <EOL> logresult , int ( get_option ( '<STR_LIT>' ) ) , gracetimeout ) <EOL> new_proc . start ( ) <EOL> workers . append ( new_proc ) <EOL> worker_proc . join ( ) <EOL> time . sleep ( <NUM_LIT:5> ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> logging . info ( "<STR_LIT>" , gracetimeout ) <EOL> time . sleep ( gracetimeout ) <EOL> logging . info ( "<STR_LIT>" ) <EOL> for worker_proc in workers : <EOL> if worker_proc . is_alive ( ) : <EOL> os . kill ( worker_proc . pid , signal . SIGKILL ) <EOL> for worker_proc in workers : <EOL> worker_proc . join ( ) <EOL> if not options . no_broker : <EOL> if broker_proc . is_alive ( ) : <EOL> os . kill ( broker_proc . pid , signal . SIGKILL ) <EOL> broker_proc . join ( ) <EOL> logging . debug ( "<STR_LIT>" ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
