<s> import os <EOL> import sys <EOL> import time <EOL> import signal <EOL> import struct <EOL> import fcntl <EOL> import socket <EOL> import shutil <EOL> import re <EOL> import subprocess <EOL> import beanstalkc <EOL> import syslog <EOL> import pprint <EOL> import random <EOL> import shlex <EOL> import traceback <EOL> from pwd import getpwnam <EOL> from grp import getgrnam <EOL> import multiprocessing <EOL> from lxml import etree <EOL> import msdb <EOL> import libIntent <EOL> import logging <EOL> sys . path . append ( '<STR_LIT>' ) <EOL> from config import * <EOL> logger = None <EOL> class Consumer ( multiprocessing . Process ) : <EOL> def __init__ ( self , task_queue , result_queue ) : <EOL> multiprocessing . Process . __init__ ( self ) <EOL> self . task_queue = task_queue <EOL> self . result_queue = result_queue <EOL> def run ( self ) : <EOL> proc_name = self . name <EOL> while True : <EOL> next_task = self . task_queue . get ( ) <EOL> if next_task is None : <EOL> logger . info ( '<STR_LIT>' % proc_name ) <EOL> break <EOL> logger . info ( '<STR_LIT>' % ( proc_name , next_task ) ) <EOL> answer = next_task ( ) <EOL> self . result_queue . put ( answer ) <EOL> return <EOL> class Task ( object ) : <EOL> def __init__ ( self , a , b ) : <EOL> self . a = a <EOL> self . b = b <EOL> def __call__ ( self ) : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> return '<STR_LIT>' % ( self . a , self . b , self . a * self . b ) <EOL> def __str__ ( self ) : <EOL> return '<STR_LIT>' % ( self . a , self . b ) <EOL> class ActivePool ( object ) : <EOL> def __init__ ( self ) : <EOL> super ( ActivePool , self ) . __init__ ( ) <EOL> self . mgr = multiprocessing . Manager ( ) <EOL> self . active = self . mgr . list ( ) <EOL> self . lock = multiprocessing . Lock ( ) <EOL> def makeActive ( self , name ) : <EOL> with self . lock : <EOL> self . active . append ( name ) <EOL> def makeInactive ( self , name ) : <EOL> with self . lock : <EOL> self . active . remove ( name ) <EOL> def __str__ ( self ) : <EOL> with self . lock : <EOL> return str ( self . active ) <EOL> __DEBUG__ = True <EOL> NUMWORKERS = <NUM_LIT:5> <EOL> NUMACTIVEWORKERS = <NUM_LIT:5> <EOL> ADBPORT = <NUM_LIT> <EOL> PSIZE = <NUM_LIT> <EOL> ADBTRIES = <NUM_LIT:10> <EOL> VM_POLL_REST = <NUM_LIT:7> <EOL> VMSDCARD = <NUM_LIT:100> <EOL> if not os . path . isfile ( SDKPATH + "<STR_LIT>" ) : <EOL> logger . info ( "<STR_LIT>" % SDKPATH ) <EOL> if not os . path . isfile ( ADBPATH + "<STR_LIT>" ) : <EOL> logger . info ( "<STR_LIT>" % ADBPATH ) <EOL> beanstalk = None <EOL> ADBFAILCOUNT = None <EOL> class AVM ( object ) : <EOL> def __init__ ( self , name , path , state , version ) : <EOL> self . name = name <EOL> self . path = path <EOL> self . state = state <EOL> self . version = version <EOL> def setVersion ( self , v ) : <EOL> self . state = v <EOL> def setState ( self , state ) : <EOL> self . state = state <EOL> syslog . openlog ( "<STR_LIT>" , syslog . LOG_PID , syslog . LOG_DAEMON ) <EOL> flagTerminate = False <EOL> def sigint_handler ( signal , frame ) : <EOL> global flagTerminate <EOL> syslog . syslog ( '<STR_LIT>' % len ( assignments ) ) <EOL> flagTerminate = True <EOL> def formatExceptionInfo ( maxTBlevel = <NUM_LIT:5> ) : <EOL> cla , exc , trbk = sys . exc_info ( ) <EOL> excName = cla . __name__ <EOL> try : <EOL> excArgs = exc . __dict__ [ "<STR_LIT:args>" ] <EOL> except KeyError : <EOL> excArgs = "<STR_LIT>" <EOL> excTb = traceback . format_tb ( trbk , maxTBlevel ) <EOL> return ( excName , excArgs , excTb ) <EOL> def get_VMs ( VMs ) : <EOL> for item in os . listdir ( VMBASEDIR ) : <EOL> if os . path . isdir ( os . path . join ( VMBASEDIR , item ) ) : <EOL> m = re . match ( "<STR_LIT>" , item ) <EOL> if m : <EOL> logger . info ( item ) <EOL> logger . info ( m . group ( <NUM_LIT:1> ) ) <EOL> d = { "<STR_LIT:name>" : m . group ( <NUM_LIT:0> ) , "<STR_LIT:path>" : os . path . join ( VMBASEDIR , item ) , "<STR_LIT:state>" : "<STR_LIT>" , "<STR_LIT:version>" : m . group ( <NUM_LIT:1> ) } <EOL> VMs [ m . group ( <NUM_LIT:0> ) ] = AVM ( ** d ) <EOL> def list_VMs ( VMs ) : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> for i in VMs . keys ( ) : <EOL> logger . debug ( "<STR_LIT>" % ( VMs [ i ] . name , VMs [ i ] . path , VMs [ i ] . state , VMs [ i ] . version ) ) <EOL> logger . debug ( "<STR_LIT>" ) <EOL> for i in VMs . keys ( ) : <EOL> if VMs [ i ] . state != "<STR_LIT>" : <EOL> logger . info ( "<STR_LIT>" % ( VMs [ i ] . name , VMs [ i ] . path , VMs [ i ] . state , VMs [ i ] . version ) ) <EOL> def kick_adb ( ) : <EOL> name = multiprocessing . current_process ( ) . name <EOL> ADBKICK = "<STR_LIT>" <EOL> cmd = ADBKICK % ( ADBPATH ) <EOL> logger . info ( name + "<STR_LIT::>" + cmd ) <EOL> try : <EOL> args = shlex . split ( cmd ) <EOL> pKick = subprocess . Popen ( args ) <EOL> pKick . terminate ( ) <EOL> except : <EOL> logger . info ( name + "<STR_LIT::>" + "<STR_LIT>" , sys . exc_info ( ) [ <NUM_LIT:0> ] ) <EOL> def kick_adb_maybe ( adbl , fcnt ) : <EOL> with adbl : <EOL> fcnt . value += <NUM_LIT:1> <EOL> if fcnt . value > <NUM_LIT:20> : <EOL> fcnt . value = <NUM_LIT:0> <EOL> kick_adb ( ) <EOL> def find_key ( dic , val ) : <EOL> """<STR_LIT>""" <EOL> return [ k for k , v in dic . iteritems ( ) if v == val ] [ <NUM_LIT:0> ] <EOL> def error_log ( msg ) : <EOL> logmsg = time . strftime ( "<STR_LIT>" , time . gmtime ( ) ) + '<STR_LIT>' + msg <EOL> logger . info ( logmsg ) <EOL> syslog . syslog ( logmsg ) <EOL> def updateStatus ( workers , alive , dead ) : <EOL> f = open ( '<STR_LIT>' , '<STR_LIT:w>' ) <EOL> f . write ( "<STR_LIT>" % ( workers , alive , dead ) ) <EOL> f . close ( ) <EOL> def runInVM ( m , job , VMs , adbport , vmlock , adbl , results , fcnt ) : <EOL> name = multiprocessing . current_process ( ) . name <EOL> formatter = logging . Formatter ( fmt = LOGFORMAT , datefmt = LOGDATEFORMAT ) <EOL> pathForFile = MALWAREJOBSBASE + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT>" <EOL> print "<STR_LIT>" % pathForFile <EOL> handler = logging . FileHandler ( pathForFile , mode = '<STR_LIT:w>' ) <EOL> handler . setFormatter ( formatter ) <EOL> vlogger = logger . getChild ( "<STR_LIT>" + name + "<STR_LIT>" ) <EOL> vlogger . setLevel ( logging . DEBUG ) <EOL> vlogger . addHandler ( handler ) <EOL> vlogger . info ( "<STR_LIT>" + pathForFile ) <EOL> rval = True <EOL> if VMs [ m ] . state in [ "<STR_LIT>" ] : <EOL> with vmlock : <EOL> VMlocal = VMs [ m ] ; <EOL> VMlocal . state = "<STR_LIT>" <EOL> VMs [ m ] = VMlocal ; <EOL> t1 = time . time ( ) <EOL> EMUSTART = "<STR_LIT>" <EOL> EMUWAIT = "<STR_LIT>" <EOL> EMULOGCAT = "<STR_LIT>" <EOL> EMUINSTALL = "<STR_LIT>" <EOL> EMUINTENT = "<STR_LIT>" <EOL> EMULAUNCH = "<STR_LIT>" <EOL> EMUKILL = "<STR_LIT>" <EOL> EMUMONKEY = "<STR_LIT>" <EOL> permissions_filename = job [ '<STR_LIT>' ] + "<STR_LIT>" <EOL> image_used = VMs [ m ] . name <EOL> start_time = time . strftime ( "<STR_LIT>" , time . gmtime ( ) ) <EOL> msdb . updateStartRun ( job [ '<STR_LIT>' ] , image_used , start_time , permissions_filename , MSVERSION ) <EOL> vlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" + VMs [ m ] . name ) <EOL> cmd = EMUSTART % ( SDKPATH , VMs [ m ] . name , PSIZE , str ( adbport ) , MALWAREJOBSBASE + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT>" ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> pEmu = None <EOL> try : <EOL> args = shlex . split ( cmd ) <EOL> pEmu = subprocess . Popen ( args ) <EOL> except : <EOL> vlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" , sys . exc_info ( ) [ <NUM_LIT:0> ] ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> cmd = EMUWAIT % ( ADBPATH , "<STR_LIT>" + str ( adbport ) ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> try : <EOL> ret = subprocess . check_call ( [ cmd ] , shell = True ) <EOL> vlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" % ret ) <EOL> except : <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" % cmd ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> vlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" + str ( adbport ) + "<STR_LIT>" ) <EOL> time . sleep ( VM_POLL_REST * <NUM_LIT:3> ) <EOL> with vmlock : <EOL> VMlocal = VMs [ m ] ; <EOL> VMlocal . state = "<STR_LIT>" <EOL> VMs [ m ] = VMlocal ; <EOL> cmd = EMULOGCAT % ( SDKPATH , "<STR_LIT>" + str ( adbport ) ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> pLogCat = None <EOL> fLogCat = None <EOL> try : <EOL> args = shlex . split ( cmd ) <EOL> vlogger . info ( args ) <EOL> fLogCat = open ( MALWAREJOBSBASE + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT>" , "<STR_LIT:w>" ) <EOL> fNull = open ( "<STR_LIT>" ) <EOL> except : <EOL> vlogger . error ( name + "<STR_LIT>" + cmd ) <EOL> vlogger . error ( formatExceptionInfo ( ) ) <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" , sys . exc_info ( ) [ <NUM_LIT:0> ] ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> cmd = EMUINSTALL % ( ADBPATH , "<STR_LIT>" + str ( adbport ) , "<STR_LIT:'>" + job [ '<STR_LIT>' ] + "<STR_LIT:'>" ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> try : <EOL> ret = subprocess . check_output ( [ cmd ] , shell = True ) <EOL> vlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" % ret ) <EOL> install_attempts = ADBTRIES <EOL> while "<STR_LIT>" not in ret and install_attempts > <NUM_LIT:0> : <EOL> kick_adb_maybe ( adbl , fcnt ) <EOL> vlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" % ret ) <EOL> time . sleep ( VM_POLL_REST ) <EOL> ret = subprocess . check_output ( [ cmd ] , shell = True ) <EOL> install_attempts -= <NUM_LIT:1> <EOL> if install_attempts <= <NUM_LIT:0> : <EOL> syslog . syslog ( VMs [ m ] . name + "<STR_LIT::>" + "<STR_LIT>" % ret ) <EOL> vlogger . error ( VMs [ m ] . name + "<STR_LIT::>" + "<STR_LIT>" % ret ) <EOL> except : <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" % cmd ) <EOL> vlogger . error ( formatExceptionInfo ( ) ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> time . sleep ( VM_POLL_REST * <NUM_LIT:3> ) <EOL> doc = etree . parse ( MALWAREJOBSBASE + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT>" ) <EOL> package = doc . find ( '<STR_LIT>' ) <EOL> li = libIntent . libIntent ( "<STR_LIT:localhost>" , adbport ) <EOL> for rint in doc . findall ( '<STR_LIT>' ) : <EOL> vlogger . info ( name + "<STR_LIT>" + str ( rint . text ) ) <EOL> li . handleRIntent ( str ( rint . text ) ) <EOL> for perm in doc . findall ( '<STR_LIT>' ) : <EOL> vlogger . info ( name + "<STR_LIT>" + str ( perm . text ) ) <EOL> li . handlePermission ( str ( perm . text ) ) <EOL> for action in doc . findall ( '<STR_LIT:action>' ) : <EOL> vlogger . info ( name + "<STR_LIT>" + str ( action . text ) ) <EOL> li . handleAction ( str ( action . text ) ) <EOL> cmd = EMUINTENT % ( ADBPATH , "<STR_LIT>" + str ( adbport ) , str ( action . text ) ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> try : <EOL> ret = subprocess . check_call ( [ cmd ] , shell = True ) <EOL> except : <EOL> syslog . syslog ( "<STR_LIT>" % ( job [ '<STR_LIT>' ] , str ( action . text ) ) ) <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" % cmd ) <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" % ret ) <EOL> li . sendCall ( ) <EOL> time . sleep ( VM_POLL_REST ) <EOL> li . endAllCalls ( ) <EOL> for activity in doc . findall ( '<STR_LIT>' ) : <EOL> for action in doc . findall ( '<STR_LIT:action>' ) : <EOL> vlogger . info ( name + "<STR_LIT>" + "<STR_LIT>" + str ( activity . text ) ) <EOL> cmd = EMULAUNCH % ( ADBPATH , "<STR_LIT>" + str ( adbport ) , str ( action . text ) , str ( package . text ) , str ( activity . text ) ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> try : <EOL> ret = subprocess . check_call ( [ cmd ] , shell = True ) <EOL> except : <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" % cmd ) <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" % ret ) <EOL> time . sleep ( VM_POLL_REST * <NUM_LIT:5> ) <EOL> cmd = EMUMONKEY % ( ADBPATH , "<STR_LIT>" + str ( adbport ) , str ( package . text ) , random . randint ( <NUM_LIT:1> , <NUM_LIT:100> ) ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> try : <EOL> ret = subprocess . check_output ( [ cmd ] , shell = True ) <EOL> vlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" % ret ) <EOL> except : <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" % cmd ) <EOL> syslog . syslog ( VMs [ m ] . name + "<STR_LIT::>" + "<STR_LIT>" % cmd ) <EOL> vlogger . error ( formatExceptionInfo ( ) ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> cmd = EMUKILL % ( ADBPATH , "<STR_LIT>" + str ( adbport ) ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> try : <EOL> ret = subprocess . check_call ( [ cmd ] , shell = True ) <EOL> except : <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> if pEmu . poll ( ) is not None : <EOL> vlogger . info ( name + "<STR_LIT>" + "<STR_LIT>" + str ( pEmu . poll ( ) ) ) <EOL> try : <EOL> pEmu . terminate ( ) <EOL> except : <EOL> vlogger . error ( name + "<STR_LIT>" + "<STR_LIT>" ) <EOL> fLogCat . flush ( ) <EOL> fLogCat . close ( ) <EOL> fNull . close ( ) <EOL> time . sleep ( VM_POLL_REST ) <EOL> vlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" + VMs [ m ] . name ) <EOL> with vmlock : <EOL> VMlocal = VMs [ m ] ; <EOL> VMlocal . state = "<STR_LIT>" <EOL> VMs [ m ] = VMlocal ; <EOL> t2 = time . time ( ) <EOL> complete_time = time . strftime ( "<STR_LIT>" , time . gmtime ( ) ) <EOL> results_file = job [ '<STR_LIT>' ] + "<STR_LIT>" <EOL> msdb . updateFinishRun ( job [ '<STR_LIT>' ] , complete_time , results_file ) <EOL> originalpcap = MALWAREJOBSBASE + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT>" <EOL> tmppcap = MALWAREJOBSBASE + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT>" <EOL> finalpcap = MALWAREJOBSBASE + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT:/>" + job [ '<STR_LIT>' ] + "<STR_LIT>" <EOL> cmd = "<STR_LIT>" % ( originalpcap , tmppcap , str ( adbport ) ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> try : <EOL> ret = subprocess . check_call ( [ cmd ] , shell = True ) <EOL> except : <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> cmd = "<STR_LIT>" % ( tmppcap , finalpcap , str ( int ( adbport ) + <NUM_LIT:1> ) ) <EOL> vlogger . info ( name + "<STR_LIT::>" + cmd ) <EOL> try : <EOL> ret = subprocess . check_call ( [ cmd ] , shell = True ) <EOL> except : <EOL> vlogger . error ( name + "<STR_LIT::>" + "<STR_LIT>" ) <EOL> sys . exit ( <NUM_LIT:1> ) <EOL> vlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" % str ( t2 - t1 ) ) <EOL> results . put ( t2 - t1 ) <EOL> elif VMs [ m ] . state in [ "<STR_LIT>" ] : <EOL> vlogger . info ( name + "<STR_LIT::>" + VMs [ m ] . name + "<STR_LIT>" ) <EOL> rval = False <EOL> elif VMs [ m ] . state in [ "<STR_LIT>" ] : <EOL> vlogger . info ( name + "<STR_LIT::>" + VMs [ m ] . name + "<STR_LIT>" ) <EOL> rval = False <EOL> return rval <EOL> def findCompatibleVM ( job , VMs , vml ) : <EOL> rval = None <EOL> with vml : <EOL> for m in VMs . keys ( ) : <EOL> targetsdk = int ( job [ '<STR_LIT:target>' ] ) <EOL> if targetsdk == <NUM_LIT:0> : <EOL> targetsdk = random . randint ( MINSDKVER , MAXSDKVER ) <EOL> syslog . syslog ( "<STR_LIT>" % ( job [ '<STR_LIT>' ] , targetsdk ) ) <EOL> if VMs [ m ] . state in [ "<STR_LIT>" ] and int ( VMs [ m ] . version ) == targetsdk : <EOL> if not flagTerminate : <EOL> logger . info ( VMs [ m ] . name + "<STR_LIT>" + str ( m ) + "<STR_LIT>" ) <EOL> VMlocal = VMs [ m ] ; <EOL> VMlocal . setState ( "<STR_LIT>" ) <EOL> VMs [ m ] = VMlocal ; <EOL> rval = m <EOL> break <EOL> elif VMs [ m ] . state in [ "<STR_LIT>" ] : <EOL> v = <NUM_LIT:1> <EOL> return rval <EOL> def rebeanstalk ( ) : <EOL> global beanstalk <EOL> if not beanstalk : <EOL> error_log ( '<STR_LIT>' ) <EOL> beanstalk = beanstalkc . Connection ( host = BSHOST , port = BSPORT ) <EOL> beanstalk . watch ( BSMSBOXQ ) <EOL> beanstalk . use ( BSMSBOXQ ) <EOL> beanstalk . ignore ( '<STR_LIT:default>' ) <EOL> def worker ( s , pool , q , str1 , VMs , vml , adbport , count , adbl , results , fcnt ) : <EOL> global beanstalk <EOL> name = multiprocessing . current_process ( ) . name <EOL> formatter = logging . Formatter ( fmt = LOGFORMAT , datefmt = LOGDATEFORMAT ) <EOL> handler = logging . FileHandler ( "<STR_LIT>" ) <EOL> handler . setFormatter ( formatter ) <EOL> wlogger = logger . getChild ( "<STR_LIT>" + name ) <EOL> wlogger . setLevel ( logging . DEBUG ) <EOL> wlogger . addHandler ( handler ) <EOL> with s : <EOL> pool . makeActive ( name ) <EOL> while True : <EOL> content = q . get ( ) <EOL> q . put ( content ) <EOL> if content == "<STR_LIT>" : <EOL> wlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" ) <EOL> pool . makeInactive ( name ) <EOL> sys . exit ( <NUM_LIT:0> ) <EOL> else : <EOL> wlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" ) <EOL> rebeanstalk ( ) <EOL> if BSMSBOXQ not in beanstalk . tubes ( ) : <EOL> wlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" ) <EOL> else : <EOL> arejobs = beanstalk . peek_ready ( ) <EOL> if arejobs is None : <EOL> wlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" ) <EOL> time . sleep ( VM_POLL_REST ) <EOL> else : <EOL> beansalkjob = None <EOL> try : <EOL> beanstalkjob = beanstalk . reserve ( ) <EOL> except : <EOL> wlogger . info ( name + "<STR_LIT>" + "<STR_LIT>" ) <EOL> job = None <EOL> if beanstalkjob : <EOL> try : <EOL> job = eval ( beanstalkjob . body ) <EOL> except : <EOL> error_log ( '<STR_LIT>' ) <EOL> beanstalkjob . delete ( ) <EOL> wlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" ) <EOL> continue <EOL> if job : <EOL> m = findCompatibleVM ( job , VMs , vml ) <EOL> if m is not None : <EOL> wlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" + str ( m ) + "<STR_LIT>" + job [ '<STR_LIT>' ] + "<STR_LIT:(>" + job [ '<STR_LIT:target>' ] + "<STR_LIT:)>" ) <EOL> beanstalkjob . delete ( ) <EOL> if runInVM ( m , job , VMs , adbport , vml , adbl , results , fcnt ) : <EOL> rebeanstalk ( ) <EOL> count . value += <NUM_LIT:1> <EOL> if flagTerminate : <EOL> wlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" ) <EOL> sys . exit ( <NUM_LIT:0> ) <EOL> else : <EOL> continue <EOL> else : <EOL> beanstalkjob . release ( delay = BSDELAY ) <EOL> wlogger . info ( name + "<STR_LIT::>" + "<STR_LIT>" + job [ '<STR_LIT:target>' ] ) <EOL> def setupLogger ( ) : <EOL> formatter = logging . Formatter ( fmt = LOGFORMAT , datefmt = LOGDATEFORMAT ) <EOL> handler = logging . FileHandler ( LOGFILE ) <EOL> handler . setFormatter ( formatter ) <EOL> logger = logging . getLogger ( ) <EOL> logger . setLevel ( logging . DEBUG ) <EOL> logger . addHandler ( handler ) <EOL> return logger <EOL> def main ( ) : <EOL> global flagTerminate <EOL> global beanstalk <EOL> global logger <EOL> logger = setupLogger ( ) <EOL> console = logging . StreamHandler ( ) <EOL> console . setLevel ( logging . INFO ) <EOL> formatter = logging . Formatter ( '<STR_LIT>' ) <EOL> console . setFormatter ( formatter ) <EOL> logging . getLogger ( ) . addHandler ( console ) <EOL> random . seed ( ) <EOL> pool = ActivePool ( ) <EOL> assignments = pool . mgr . dict ( ) <EOL> count = pool . mgr . Value ( '<STR_LIT:i>' , <NUM_LIT:0> ) <EOL> ADBFAILCOUNT = pool . mgr . Value ( '<STR_LIT:i>' , <NUM_LIT:0> ) <EOL> VMs = pool . mgr . dict ( ) <EOL> s = multiprocessing . Semaphore ( NUMACTIVEWORKERS ) <EOL> q = multiprocessing . Queue ( ) <EOL> r = multiprocessing . Queue ( ) <EOL> results = [ ] <EOL> al = multiprocessing . RLock ( ) <EOL> vml = multiprocessing . RLock ( ) <EOL> cl = multiprocessing . RLock ( ) <EOL> get_VMs ( VMs ) <EOL> list_VMs ( VMs ) <EOL> for i in range ( <NUM_LIT:30> ) : <EOL> q . put ( "<STR_LIT>" + str ( i ) ) <EOL> port = ADBPORT <EOL> jobs = [ <EOL> multiprocessing . Process ( target = worker , name = str ( i ) , args = ( s , pool , q , "<STR_LIT>" , VMs , vml , str ( port + ( i * <NUM_LIT:2> ) ) , count , al , r , ADBFAILCOUNT ) ) <EOL> for i in range ( NUMWORKERS ) <EOL> ] <EOL> for j in jobs : <EOL> logger . info ( "<STR_LIT>" % j . name ) <EOL> j . start ( ) <EOL> rebeanstalk ( ) <EOL> while True : <EOL> logger . info ( "<STR_LIT>" ) <EOL> logger . info ( "<STR_LIT>" % count . value ) <EOL> try : <EOL> thing = r . get ( block = False ) <EOL> results . append ( thing ) <EOL> except : <EOL> logger . warning ( "<STR_LIT>" ) <EOL> if len ( results ) > <NUM_LIT:0> : <EOL> logger . info ( "<STR_LIT>" % str ( sum ( results ) / float ( len ( results ) ) ) ) <EOL> try : <EOL> if BSMSBOXQ not in beanstalk . tubes ( ) : <EOL> logger . info ( "<STR_LIT>" ) <EOL> else : <EOL> if beanstalk . peek_ready ( ) is not None : <EOL> logger . info ( str ( beanstalk . stats_tube ( BSMSBOXQ ) [ '<STR_LIT>' ] ) + "<STR_LIT>" ) <EOL> else : <EOL> logger . info ( "<STR_LIT>" ) <EOL> if beanstalk . peek_reserved ( ) is not None : <EOL> logger . info ( str ( beanstalk . stats_tube ( BSMSBOXQ ) [ '<STR_LIT>' ] ) + "<STR_LIT>" ) <EOL> else : <EOL> logger . info ( "<STR_LIT>" ) <EOL> if beanstalk . peek_delayed ( ) is not None : <EOL> logger . info ( str ( beanstalk . stats_tube ( BSMSBOXQ ) [ '<STR_LIT>' ] ) + "<STR_LIT>" ) <EOL> else : <EOL> logger . info ( "<STR_LIT>" ) <EOL> except KeyError : <EOL> logger . warning ( "<STR_LIT>" ) <EOL> except : <EOL> logger . warning ( "<STR_LIT>" ) <EOL> logger . info ( "<STR_LIT>" + str ( q . qsize ( ) ) + "<STR_LIT>" ) <EOL> logger . info ( "<STR_LIT>" + str ( ADBFAILCOUNT . value ) + "<STR_LIT>" ) <EOL> logger . debug ( "<STR_LIT>" + str ( NUMWORKERS ) + "<STR_LIT>" ) <EOL> walive = <NUM_LIT:0> <EOL> wdead = <NUM_LIT:0> <EOL> for j in jobs : <EOL> if j . is_alive ( ) : <EOL> logger . debug ( "<STR_LIT:U+0020>" + str ( j . pid ) ) <EOL> walive += <NUM_LIT:1> <EOL> else : <EOL> logger . debug ( "<STR_LIT>" ) <EOL> wdead += <NUM_LIT:1> <EOL> logger . info ( "<STR_LIT>" % ( walive , wdead ) ) <EOL> updateStatus ( NUMWORKERS , walive , wdead ) <EOL> list_VMs ( VMs ) <EOL> time . sleep ( VM_POLL_REST + <NUM_LIT:10> ) <EOL> sys . exit ( <NUM_LIT:0> ) <EOL> signal . signal ( signal . SIGINT , sigint_handler ) <EOL> syslog . syslog ( '<STR_LIT>' ) <EOL> while True : <EOL> get_VMs ( VMs ) <EOL> list_VMs ( VMs ) <EOL> if not beanstalk : <EOL> error_log ( '<STR_LIT>' ) <EOL> beanstalk = beanstalkc . Connection ( host = BSHOST , port = BSPORT ) <EOL> beanstalk . watch ( BSMSBOXQ ) <EOL> beanstalk . use ( BSMSBOXQ ) <EOL> beanstalk . ignore ( '<STR_LIT:default>' ) <EOL> beanstalkjob = beanstalk . reserve ( ) <EOL> if beanstalkjob : <EOL> try : <EOL> job = eval ( beanstalkjob . body ) <EOL> except : <EOL> error_log ( '<STR_LIT>' ) <EOL> beanstalkjob . delete ( ) <EOL> return False <EOL> m = findCompatibleVM ( job , VMs , vml ) <EOL> if m is not None : <EOL> logger . info ( "<STR_LIT>" + str ( m ) ) <EOL> sys . exit ( <NUM_LIT:0> ) <EOL> else : <EOL> beanstalkjob . release ( delay = <NUM_LIT> ) <EOL> logger . info ( "<STR_LIT>" + job [ '<STR_LIT:target>' ] ) <EOL> if flagTerminate and len ( assignments ) == <NUM_LIT:0> : <EOL> syslog . syslog ( '<STR_LIT>' ) <EOL> break <EOL> time . sleep ( VM_POLL_REST ) <EOL> sys . exit ( <NUM_LIT:0> ) <EOL> syslog . syslog ( '<STR_LIT>' ) <EOL> return <NUM_LIT:0> <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> main ( ) </s>
