<s> """<STR_LIT>""" <EOL> __author__ = "<STR_LIT>" <EOL> __copyright__ = "<STR_LIT>" <EOL> __license__ = "<STR_LIT>" <EOL> __url__ = "<STR_LIT>" <EOL> def _discoro_proc ( ) : <EOL> """<STR_LIT>""" <EOL> import os <EOL> import shutil <EOL> import traceback <EOL> import sys <EOL> import time <EOL> try : <EOL> import psutil <EOL> except : <EOL> psutil = None <EOL> import asyncoro . disasyncoro as asyncoro <EOL> from asyncoro import Coro <EOL> from asyncoro . discoro import MinPulseInterval , MaxPulseInterval , DiscoroNodeInfo , DiscoroNodeAvailInfo <EOL> class Nonlocals ( object ) : <EOL> def __init__ ( self , ** kwargs ) : <EOL> self . __dict__ . update ( kwargs ) <EOL> _discoro_coro = asyncoro . AsynCoro . cur_coro ( ) <EOL> _discoro_config = yield _discoro_coro . receive ( ) <EOL> assert _discoro_config [ '<STR_LIT>' ] == '<STR_LIT>' <EOL> _discoro_timer_coro = _discoro_config [ '<STR_LIT>' ] <EOL> yield asyncoro . AsynCoro . instance ( ) . peer ( _discoro_timer_coro . location ) <EOL> if _discoro_config [ '<STR_LIT>' ] > <NUM_LIT:0> : <EOL> MinPulseInterval = _discoro_config [ '<STR_LIT>' ] <EOL> if _discoro_config [ '<STR_LIT>' ] > <NUM_LIT:0> : <EOL> MaxPulseInterval = _discoro_config [ '<STR_LIT>' ] <EOL> _discoro_coro . register ( '<STR_LIT>' ) <EOL> _discoro_name = asyncoro . AsynCoro . instance ( ) . name <EOL> asyncoro . AsynCoro . instance ( ) . dest_path = os . path . join ( '<STR_LIT>' , _discoro_name ) <EOL> _discoro_dest_path = asyncoro . AsynCoro . instance ( ) . dest_path <EOL> _discoro_pid_path = os . path . join ( _discoro_dest_path , '<STR_LIT:..>' , '<STR_LIT>' % _discoro_name ) <EOL> _discoro_pid_path = os . path . normpath ( _discoro_pid_path ) <EOL> if os . path . exists ( _discoro_pid_path ) : <EOL> with open ( _discoro_pid_path , '<STR_LIT:r>' ) as _discoro_req : <EOL> _discoro_var = _discoro_req . read ( ) <EOL> _discoro_var = int ( _discoro_var ) <EOL> if not _discoro_config [ '<STR_LIT>' ] : <EOL> print ( '<STR_LIT>' <EOL> '<STR_LIT>' % <EOL> ( _discoro_var , _discoro_pid_path ) ) <EOL> _discoro_var = os . getpid ( ) <EOL> import signal <EOL> try : <EOL> os . kill ( _discoro_var , signal . SIGTERM ) <EOL> except : <EOL> pass <EOL> else : <EOL> time . sleep ( <NUM_LIT:0.1> ) <EOL> try : <EOL> if os . waitpid ( _discoro_var , os . WNOHANG ) [ <NUM_LIT:0> ] != _discoro_var : <EOL> asyncoro . logger . warning ( '<STR_LIT>' , _discoro_var ) <EOL> except : <EOL> pass <EOL> del signal <EOL> if os . path . isdir ( _discoro_dest_path ) : <EOL> shutil . rmtree ( _discoro_dest_path ) <EOL> os . makedirs ( _discoro_dest_path ) <EOL> os . chdir ( _discoro_dest_path ) <EOL> sys . path . insert ( <NUM_LIT:0> , _discoro_dest_path ) <EOL> with open ( _discoro_pid_path , '<STR_LIT:w>' ) as _discoro_var : <EOL> _discoro_var . write ( '<STR_LIT:%s>' % os . getpid ( ) ) <EOL> def _discoro_peer ( peer , coro = None ) : <EOL> yield asyncoro . AsynCoro . instance ( ) . peer ( peer ) <EOL> for _discoro_var in _discoro_config . pop ( '<STR_LIT>' , [ ] ) : <EOL> Coro ( _discoro_peer , _discoro_var ) <EOL> del _discoro_peer <EOL> for _discoro_var in [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] : <EOL> del _discoro_config [ _discoro_var ] <EOL> asyncoro . logger . info ( '<STR_LIT>' <EOL> '<STR_LIT>' , <EOL> _discoro_name , _discoro_coro . location , _discoro_dest_path ) <EOL> _discoro_req = _discoro_client = _discoro_auth = _discoro_msg = None <EOL> _discoro_pulse_coro = _discoro_peer_status = _discoro_zombie_timeout = None <EOL> _discoro_monitor_coro = _discoro_monitor_proc = _discoro_schedule_coro = None <EOL> _discoro_computation = _discoro_func = _discoro_var = None <EOL> _discoro_job_coros = set ( ) <EOL> _discoro_nonlocals = Nonlocals ( busy_time = time . time ( ) ) <EOL> _discoro_globals = { } <EOL> _discoro_locals = { } <EOL> _discoro_modules = dict ( sys . modules ) <EOL> _discoro_globals . update ( globals ( ) ) <EOL> _discoro_locals . update ( locals ( ) ) <EOL> def _discoro_peer_status ( coro = None ) : <EOL> coro . set_daemon ( ) <EOL> while <NUM_LIT:1> : <EOL> status = yield coro . receive ( ) <EOL> if isinstance ( status , asyncoro . PeerStatus ) and status . status == asyncoro . PeerStatus . Offline and _discoro_pulse_coro and _discoro_pulse_coro . location == status . location : <EOL> asyncoro . logger . debug ( '<STR_LIT>' , <EOL> status . location , _discoro_computation . _auth ) <EOL> _discoro_coro . send ( { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _discoro_config [ '<STR_LIT>' ] } ) <EOL> def _discoro_monitor_proc ( coro = None ) : <EOL> coro . set_daemon ( ) <EOL> while <NUM_LIT:1> : <EOL> msg = yield coro . receive ( timeout = _discoro_zombie_timeout ) <EOL> if isinstance ( msg , asyncoro . MonitorException ) : <EOL> _discoro_nonlocals . busy_time = time . time ( ) <EOL> asyncoro . logger . debug ( '<STR_LIT>' , msg . args [ <NUM_LIT:0> ] ) <EOL> _discoro_job_coros . discard ( msg . args [ <NUM_LIT:0> ] ) <EOL> elif ( ( not _discoro_job_coros ) and _discoro_zombie_timeout and <EOL> ( ( time . time ( ) - _discoro_nonlocals . busy_time ) > _discoro_zombie_timeout ) ) : <EOL> asyncoro . logger . debug ( '<STR_LIT>' , <EOL> coro . location , _discoro_computation . _auth ) <EOL> _discoro_coro . send ( { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _discoro_config [ '<STR_LIT>' ] } ) <EOL> def _discoro_schedule_proc ( client , job_coro , coro = None ) : <EOL> if ( yield client . deliver ( job_coro , timeout = <NUM_LIT:5> ) ) == <NUM_LIT:1> : <EOL> Coro . _asyncoro . _add ( job_coro ) <EOL> else : <EOL> _discoro_job_coros . discard ( job_coro ) <EOL> _discoro_monitor_coro = Coro ( _discoro_monitor_proc ) <EOL> asyncoro . AsynCoro . instance ( ) . peer_status ( Coro ( _discoro_peer_status ) ) <EOL> while <NUM_LIT:1> : <EOL> _discoro_msg = yield _discoro_coro . receive ( ) <EOL> if not isinstance ( _discoro_msg , dict ) : <EOL> continue <EOL> _discoro_req = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if _discoro_req == '<STR_LIT>' : <EOL> _discoro_client = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> _discoro_auth = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> _discoro_func = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if not isinstance ( _discoro_client , Coro ) or not _discoro_computation or _discoro_auth != _discoro_computation . _auth : <EOL> asyncoro . logger . warning ( '<STR_LIT>' , type ( _discoro_func ) ) <EOL> if isinstance ( _discoro_client , Coro ) : <EOL> _discoro_client . send ( None ) <EOL> continue <EOL> try : <EOL> _discoro_func = asyncoro . unserialize ( _discoro_func ) <EOL> if _discoro_func . code : <EOL> exec ( _discoro_func . code ) in globals ( ) <EOL> job_coro = Coro ( globals ( ) [ _discoro_func . name ] , <EOL> * ( _discoro_func . args ) , ** ( _discoro_func . kwargs ) ) <EOL> except : <EOL> asyncoro . logger . debug ( '<STR_LIT>' ) <EOL> job_coro = ( sys . exc_info ( ) [ <NUM_LIT:0> ] , getattr ( _discoro_func , '<STR_LIT:name>' , _discoro_func ) , <EOL> traceback . format_exc ( ) ) <EOL> _discoro_client . send ( job_coro ) <EOL> else : <EOL> asyncoro . logger . debug ( '<STR_LIT>' , job_coro ) <EOL> _discoro_job_coros . add ( job_coro ) <EOL> job_coro . notify ( _discoro_monitor_coro ) <EOL> _discoro_var = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if isinstance ( _discoro_var , Coro ) : <EOL> job_coro . notify ( _discoro_var ) <EOL> if Coro . _asyncoro . _remove ( job_coro ) == <NUM_LIT:0> : <EOL> Coro ( _discoro_schedule_proc , _discoro_client , job_coro ) <EOL> _discoro_nonlocals . busy_time = time . time ( ) <EOL> del job_coro <EOL> elif _discoro_req == '<STR_LIT>' : <EOL> _discoro_client = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> _discoro_pulse_coro = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if not isinstance ( _discoro_client , Coro ) or not isinstance ( _discoro_pulse_coro , Coro ) : <EOL> continue <EOL> if _discoro_computation is not None : <EOL> asyncoro . logger . debug ( '<STR_LIT>' ) <EOL> _discoro_client . send ( - <NUM_LIT:1> ) <EOL> continue <EOL> os . chdir ( _discoro_dest_path ) <EOL> try : <EOL> _discoro_computation = _discoro_msg [ '<STR_LIT>' ] <EOL> exec ( '<STR_LIT>' ) in globals ( ) <EOL> if _discoro_computation . _code : <EOL> exec ( _discoro_computation . _code ) in globals ( ) <EOL> except : <EOL> _discoro_computation = None <EOL> asyncoro . logger . warning ( '<STR_LIT>' ) <EOL> asyncoro . logger . debug ( traceback . format_exc ( ) ) <EOL> _discoro_client . send ( - <NUM_LIT:1> ) <EOL> continue <EOL> if isinstance ( _discoro_computation . pulse_interval , int ) and MinPulseInterval <= _discoro_computation . pulse_interval <= MaxPulseInterval : <EOL> _discoro_computation . pulse_interval = _discoro_computation . pulse_interval <EOL> else : <EOL> _discoro_computation . pulse_interval = MinPulseInterval <EOL> _discoro_timer_coro . send ( { '<STR_LIT>' : _discoro_pulse_coro , <EOL> '<STR_LIT>' : _discoro_computation . pulse_interval , <EOL> '<STR_LIT>' : _discoro_dest_path } ) <EOL> _discoro_nonlocals . busy_time = time . time ( ) <EOL> _discoro_zombie_timeout = _discoro_computation . zombie_period <EOL> asyncoro . logger . debug ( '<STR_LIT>' , <EOL> _discoro_computation . _auth , _discoro_msg [ '<STR_LIT>' ] . location , <EOL> _discoro_zombie_timeout ) <EOL> if not _discoro_zombie_timeout : <EOL> _discoro_zombie_timeout = None <EOL> _discoro_monitor_coro . send ( None ) <EOL> _discoro_client . send ( <NUM_LIT:0> ) <EOL> elif _discoro_req == '<STR_LIT>' : <EOL> _discoro_auth = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if not _discoro_computation or ( _discoro_auth != _discoro_computation . _auth and <EOL> _discoro_auth != _discoro_config [ '<STR_LIT>' ] ) : <EOL> continue <EOL> asyncoro . logger . debug ( '<STR_LIT>' , <EOL> _discoro_coro . location , _discoro_computation . _auth ) <EOL> if _discoro_auth != _discoro_computation . _auth and _discoro_pulse_coro : <EOL> _discoro_pulse_coro . send ( { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:location>' : _discoro_coro . location } ) <EOL> for _discoro_var in _discoro_job_coros : <EOL> _discoro_var . terminate ( ) <EOL> _discoro_job_coros . clear ( ) <EOL> for _discoro_var in list ( globals ( ) ) : <EOL> if _discoro_var not in _discoro_globals : <EOL> globals ( ) . pop ( _discoro_var , None ) <EOL> globals ( ) . update ( _discoro_globals ) <EOL> for _discoro_var in sys . modules . keys ( ) : <EOL> if _discoro_var not in _discoro_modules : <EOL> sys . modules . pop ( _discoro_var , None ) <EOL> sys . modules . update ( _discoro_modules ) <EOL> for _discoro_var in os . listdir ( _discoro_dest_path ) : <EOL> _discoro_var = os . path . join ( _discoro_dest_path , _discoro_var ) <EOL> if os . path . isdir ( _discoro_var ) and not os . path . islink ( _discoro_var ) : <EOL> shutil . rmtree ( _discoro_var , ignore_errors = True ) <EOL> else : <EOL> os . remove ( _discoro_var ) <EOL> if not os . path . isdir ( _discoro_dest_path ) : <EOL> try : <EOL> os . remove ( _discoro_dest_path ) <EOL> except : <EOL> pass <EOL> os . makedirs ( _discoro_dest_path ) <EOL> if not os . path . isfile ( _discoro_pid_path ) : <EOL> try : <EOL> if os . path . islink ( _discoro_pid_path ) : <EOL> os . remove ( _discoro_pid_path ) <EOL> else : <EOL> shutil . rmtree ( _discoro_pid_path ) <EOL> with open ( _discoro_pid_path , '<STR_LIT:w>' ) as _discoro_var : <EOL> _discoro_var . write ( '<STR_LIT:%s>' % os . getpid ( ) ) <EOL> except : <EOL> asyncoro . logger . warning ( '<STR_LIT>' , _discoro_pid_path ) <EOL> os . chdir ( _discoro_dest_path ) <EOL> asyncoro . AsynCoro . instance ( ) . dest_path = _discoro_dest_path <EOL> _discoro_computation = _discoro_client = _discoro_pulse_coro = None <EOL> if _discoro_config [ '<STR_LIT>' ] > <NUM_LIT:0> : <EOL> _discoro_config [ '<STR_LIT>' ] -= <NUM_LIT:1> <EOL> if _discoro_config [ '<STR_LIT>' ] == <NUM_LIT:0> : <EOL> break <EOL> _discoro_timer_coro . send ( { '<STR_LIT>' : _discoro_pulse_coro , '<STR_LIT>' : None , <EOL> '<STR_LIT>' : '<STR_LIT>' } ) <EOL> _discoro_zombie_timeout = None <EOL> elif _discoro_req == '<STR_LIT>' : <EOL> if psutil : <EOL> _discoro_var = DiscoroNodeAvailInfo ( _discoro_coro . location . addr , <EOL> <NUM_LIT> - psutil . cpu_percent ( ) , <EOL> psutil . virtual_memory ( ) . available , <EOL> psutil . disk_usage ( _discoro_dest_path ) . free , <EOL> <NUM_LIT> - psutil . swap_memory ( ) . percent ) <EOL> else : <EOL> _discoro_var = None <EOL> _discoro_var = DiscoroNodeInfo ( _discoro_name [ : _discoro_name . rfind ( '<STR_LIT:->' ) ] , <EOL> _discoro_coro . location . addr , _discoro_var ) <EOL> _discoro_client = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if isinstance ( _discoro_client , Coro ) : <EOL> _discoro_client . send ( _discoro_var ) <EOL> elif _discoro_req == '<STR_LIT:status>' : <EOL> if _discoro_msg . get ( '<STR_LIT>' , None ) != _discoro_config [ '<STR_LIT>' ] : <EOL> asyncoro . logger . debug ( '<STR_LIT>' , _discoro_msg . get ( '<STR_LIT>' ) ) <EOL> continue <EOL> if _discoro_pulse_coro : <EOL> print ( '<STR_LIT>' % <EOL> ( _discoro_name , _discoro_coro . location , len ( _discoro_job_coros ) , <EOL> _discoro_pulse_coro . location ) ) <EOL> else : <EOL> print ( '<STR_LIT>' % <EOL> ( _discoro_name , _discoro_coro . location ) ) <EOL> elif _discoro_req == '<STR_LIT>' : <EOL> if _discoro_msg . get ( '<STR_LIT>' , None ) != _discoro_config [ '<STR_LIT>' ] : <EOL> asyncoro . logger . debug ( '<STR_LIT>' , _discoro_msg . get ( '<STR_LIT>' ) ) <EOL> continue <EOL> if _discoro_pulse_coro : <EOL> _discoro_pulse_coro . send ( { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:location>' : _discoro_coro . location } ) <EOL> break <EOL> elif _discoro_req == '<STR_LIT>' : <EOL> if _discoro_msg . get ( '<STR_LIT>' , None ) != _discoro_config [ '<STR_LIT>' ] : <EOL> asyncoro . logger . debug ( '<STR_LIT>' , _discoro_msg . get ( '<STR_LIT>' ) ) <EOL> continue <EOL> if _discoro_pulse_coro : <EOL> _discoro_pulse_coro . send ( { '<STR_LIT:status>' : '<STR_LIT>' , <EOL> '<STR_LIT:location>' : _discoro_coro . location } ) <EOL> if _discoro_computation : <EOL> msg = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _discoro_computation . _auth } <EOL> _discoro_config [ '<STR_LIT>' ] = <NUM_LIT:1> <EOL> _discoro_coro . send ( msg ) <EOL> else : <EOL> break <EOL> else : <EOL> asyncoro . logger . warning ( '<STR_LIT>' , _discoro_req ) <EOL> _discoro_client = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if not isinstance ( _discoro_client , Coro ) : <EOL> continue <EOL> _discoro_client . send ( - <NUM_LIT:1> ) <EOL> while _discoro_job_coros : <EOL> _discoro_msg = yield _discoro_coro . receive ( ) <EOL> if not isinstance ( _discoro_msg , dict ) : <EOL> continue <EOL> _discoro_req = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if _discoro_req == '<STR_LIT>' : <EOL> _discoro_auth = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if not _discoro_computation or _discoro_auth != _discoro_computation . _auth : <EOL> continue <EOL> asyncoro . logger . debug ( '<STR_LIT>' , <EOL> _discoro_coro . location , _discoro_computation . _auth ) <EOL> for _discoro_var in list ( globals ( ) ) : <EOL> if _discoro_var not in _discoro_globals : <EOL> globals ( ) . pop ( _discoro_var , None ) <EOL> globals ( ) . update ( _discoro_globals ) <EOL> break <EOL> else : <EOL> asyncoro . logger . warning ( '<STR_LIT>' , _discoro_req ) <EOL> _discoro_client = _discoro_msg . get ( '<STR_LIT>' , None ) <EOL> if not isinstance ( _discoro_client , Coro ) : <EOL> continue <EOL> _discoro_client . send ( - <NUM_LIT:1> ) <EOL> for _discoro_var in os . listdir ( _discoro_dest_path ) : <EOL> _discoro_var = os . path . join ( _discoro_dest_path , _discoro_var ) <EOL> if os . path . isdir ( _discoro_var ) and not os . path . islink ( _discoro_var ) : <EOL> shutil . rmtree ( _discoro_var , ignore_errors = True ) <EOL> else : <EOL> os . remove ( _discoro_var ) <EOL> if os . path . isfile ( _discoro_pid_path ) : <EOL> os . remove ( _discoro_pid_path ) <EOL> _discoro_config [ '<STR_LIT>' ] . put ( { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _discoro_config [ '<STR_LIT>' ] } ) <EOL> asyncoro . logger . debug ( '<STR_LIT>' , <EOL> _discoro_name , _discoro_coro . location ) <EOL> def _discoro_process ( _discoro_config , _discoro_server_id , _discoro_mp_queue , _discoro_auth ) : <EOL> import os <EOL> import logging <EOL> import asyncoro . disasyncoro as asyncoro <EOL> if _discoro_config [ '<STR_LIT>' ] : <EOL> asyncoro . logger . setLevel ( logging . DEBUG ) <EOL> else : <EOL> asyncoro . logger . setLevel ( logging . INFO ) <EOL> del _discoro_config [ '<STR_LIT>' ] <EOL> _discoro_config_msg = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT:id>' : _discoro_server_id , <EOL> '<STR_LIT>' : _discoro_config . pop ( '<STR_LIT>' , False ) , <EOL> '<STR_LIT>' : _discoro_config . pop ( '<STR_LIT>' , - <NUM_LIT:1> ) , <EOL> '<STR_LIT>' : _discoro_config . pop ( '<STR_LIT>' , [ ] ) , <EOL> '<STR_LIT>' : _discoro_config . pop ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : _discoro_config . pop ( '<STR_LIT>' ) , <EOL> '<STR_LIT>' : _discoro_auth , '<STR_LIT>' : _discoro_mp_queue } <EOL> _discoro_scheduler = asyncoro . AsynCoro ( ** _discoro_config ) <EOL> _discoro_coro = asyncoro . Coro ( _discoro_proc ) <EOL> for _discoro_var in globals ( ) . keys ( ) : <EOL> if _discoro_var . startswith ( '<STR_LIT>' ) : <EOL> globals ( ) . pop ( _discoro_var ) <EOL> del logging , os , _discoro_config , _discoro_var <EOL> req_queue , _discoro_mp_queue = _discoro_mp_queue , None <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> req = req_queue . get ( ) <EOL> except KeyboardInterrupt : <EOL> req = { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _discoro_auth } <EOL> req_queue . put ( req ) <EOL> asyncoro . logger . debug ( '<STR_LIT>' , asyncoro . AsynCoro . instance ( ) . _location ) <EOL> if not isinstance ( req , dict ) or req . get ( '<STR_LIT>' ) != _discoro_auth : <EOL> asyncoro . logger . warning ( '<STR_LIT>' , type ( req ) ) <EOL> continue <EOL> cmd = req . get ( '<STR_LIT>' ) <EOL> if cmd == '<STR_LIT:status>' or cmd == '<STR_LIT>' : <EOL> _discoro_coro . send ( req ) <EOL> elif cmd == '<STR_LIT:start>' : <EOL> _discoro_config_msg [ '<STR_LIT>' ] = req . get ( '<STR_LIT>' , None ) <EOL> _discoro_coro . send ( _discoro_config_msg ) <EOL> del _discoro_config_msg <EOL> elif cmd == '<STR_LIT>' or cmd == '<STR_LIT>' : <EOL> _discoro_coro . send ( req ) <EOL> break <EOL> _discoro_scheduler . finish ( ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> """<STR_LIT>""" <EOL> import sys <EOL> import time <EOL> import argparse <EOL> import multiprocessing <EOL> import socket <EOL> import os <EOL> import collections <EOL> import hashlib <EOL> import logging <EOL> try : <EOL> import readline <EOL> except : <EOL> pass <EOL> import asyncoro . disasyncoro as asyncoro <EOL> try : <EOL> import psutil <EOL> except ImportError : <EOL> print ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> psutil = None <EOL> else : <EOL> psutil . cpu_percent ( <NUM_LIT:0.1> ) <EOL> parser = argparse . ArgumentParser ( ) <EOL> parser . add_argument ( '<STR_LIT:-c>' , '<STR_LIT>' , dest = '<STR_LIT>' , type = int , default = <NUM_LIT:0> , <EOL> help = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , dest = '<STR_LIT>' , default = None , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = None , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , dest = '<STR_LIT>' , type = int , default = <NUM_LIT> , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , action = '<STR_LIT>' , default = [ ] , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , dest = '<STR_LIT:name>' , default = None , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = None , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = None , type = int , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , dest = '<STR_LIT>' , default = '<STR_LIT>' , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = None , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = None , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = - <NUM_LIT:1> , type = int , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = <NUM_LIT:0> , type = int , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = <NUM_LIT:0> , type = int , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , action = '<STR_LIT:store_true>' , dest = '<STR_LIT>' , default = False , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , action = '<STR_LIT:store_true>' , dest = '<STR_LIT>' , default = False , <EOL> help = '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , action = '<STR_LIT:store_true>' , dest = '<STR_LIT>' , <EOL> default = False , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , action = '<STR_LIT>' , default = [ ] , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , '<STR_LIT>' , action = '<STR_LIT:store_true>' , dest = '<STR_LIT>' , default = False , <EOL> help = '<STR_LIT>' ) <EOL> _discoro_config = vars ( parser . parse_args ( sys . argv [ <NUM_LIT:1> : ] ) ) <EOL> if _discoro_config [ '<STR_LIT>' ] and _discoro_config [ '<STR_LIT>' ] < <NUM_LIT:1> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> if ( _discoro_config [ '<STR_LIT>' ] and _discoro_config [ '<STR_LIT>' ] and <EOL> _discoro_config [ '<STR_LIT>' ] < _discoro_config [ '<STR_LIT>' ] ) : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> _discoro_cpus = multiprocessing . cpu_count ( ) <EOL> if _discoro_config [ '<STR_LIT>' ] > <NUM_LIT:0> : <EOL> if _discoro_config [ '<STR_LIT>' ] > _discoro_cpus : <EOL> raise Exception ( '<STR_LIT>' % _discoro_cpus ) <EOL> _discoro_cpus = _discoro_config [ '<STR_LIT>' ] <EOL> elif _discoro_config [ '<STR_LIT>' ] < <NUM_LIT:0> : <EOL> if - _discoro_config [ '<STR_LIT>' ] >= _discoro_cpus : <EOL> raise Exception ( '<STR_LIT>' % _discoro_cpus ) <EOL> _discoro_cpus += _discoro_config [ '<STR_LIT>' ] <EOL> del _discoro_config [ '<STR_LIT>' ] <EOL> _discoro_tcp_ports = set ( ) <EOL> tcp_port = tcp_ports = None <EOL> for tcp_port in _discoro_config . pop ( '<STR_LIT>' , [ ] ) : <EOL> tcp_ports = tcp_port . split ( '<STR_LIT:->' ) <EOL> if len ( tcp_ports ) == <NUM_LIT:1> : <EOL> _discoro_tcp_ports . add ( int ( tcp_ports [ <NUM_LIT:0> ] ) ) <EOL> elif len ( tcp_ports ) == <NUM_LIT:2> : <EOL> _discoro_tcp_ports = _discoro_tcp_ports . union ( range ( int ( tcp_ports [ <NUM_LIT:0> ] ) , <EOL> int ( tcp_ports [ <NUM_LIT:1> ] ) + <NUM_LIT:1> ) ) <EOL> else : <EOL> raise Exception ( '<STR_LIT>' % tcp_ports ) <EOL> _discoro_tcp_ports = sorted ( _discoro_tcp_ports ) <EOL> if _discoro_tcp_ports : <EOL> for tcp_port in range ( _discoro_tcp_ports [ - <NUM_LIT:1> ] + <NUM_LIT:1> , <EOL> _discoro_tcp_ports [ - <NUM_LIT:1> ] + <NUM_LIT:1> + <EOL> ( _discoro_cpus + <NUM_LIT:1> ) - len ( _discoro_tcp_ports ) ) : <EOL> _discoro_tcp_ports . append ( tcp_port ) <EOL> else : <EOL> _discoro_tcp_ports = [ <NUM_LIT:0> ] * ( _discoro_cpus + <NUM_LIT:1> ) <EOL> _discoro_peers , _discoro_config [ '<STR_LIT>' ] = _discoro_config [ '<STR_LIT>' ] , [ ] <EOL> peer = None <EOL> for peer in _discoro_peers : <EOL> peer = peer . split ( '<STR_LIT::>' ) <EOL> if len ( peer ) != <NUM_LIT:2> : <EOL> raise Exception ( '<STR_LIT>' % '<STR_LIT::>' . join ( peer ) ) <EOL> _discoro_config [ '<STR_LIT>' ] . append ( asyncoro . Location ( peer [ <NUM_LIT:0> ] , peer [ <NUM_LIT:1> ] ) ) <EOL> del peer <EOL> _discoro_peers = _discoro_config [ '<STR_LIT>' ] <EOL> _discoro_name = _discoro_config [ '<STR_LIT:name>' ] <EOL> if not _discoro_name : <EOL> _discoro_name = socket . gethostname ( ) <EOL> if not _discoro_name : <EOL> _discoro_name = '<STR_LIT>' <EOL> _discoro_daemon = _discoro_config . pop ( '<STR_LIT>' , False ) <EOL> if not _discoro_daemon : <EOL> try : <EOL> if os . getpgrp ( ) != os . tcgetpgrp ( sys . stdin . fileno ( ) ) : <EOL> _discoro_daemon = True <EOL> except : <EOL> pass <EOL> _discoro_auth = hashlib . sha1 ( os . urandom ( <NUM_LIT:10> ) . encode ( '<STR_LIT>' ) ) . hexdigest ( ) <EOL> _discoro_server_infos = [ ] <EOL> _discoro_ServerInfo = collections . namedtuple ( '<STR_LIT>' , [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> _discoro_mp_queue = None <EOL> for _discoro_server_id in range ( <NUM_LIT:1> , _discoro_cpus + <NUM_LIT:1> ) : <EOL> _discoro_config [ '<STR_LIT:name>' ] = '<STR_LIT>' % ( _discoro_name , _discoro_server_id ) <EOL> _discoro_config [ '<STR_LIT>' ] = _discoro_tcp_ports [ _discoro_server_id ] <EOL> _discoro_mp_queue = multiprocessing . Queue ( ) <EOL> _discoro_server_info = _discoro_ServerInfo ( <EOL> multiprocessing . Process ( target = _discoro_process , <EOL> args = ( dict ( _discoro_config ) , _discoro_server_id , <EOL> _discoro_mp_queue , _discoro_auth ) ) , <EOL> _discoro_mp_queue ) <EOL> _discoro_server_infos . append ( _discoro_server_info ) <EOL> _discoro_server_info . Proc . start ( ) <EOL> def _discoro_timer_proc ( coro = None ) : <EOL> from asyncoro . discoro import DiscoroNodeAvailInfo <EOL> coro . set_daemon ( ) <EOL> last_pulse = last_proc_check = time . time ( ) <EOL> interval = pulse_coro = None <EOL> for peer in _discoro_peers : <EOL> yield asyncoro . AsynCoro . instance ( ) . peer ( peer ) <EOL> while <NUM_LIT:1> : <EOL> msg = yield coro . receive ( timeout = interval ) <EOL> now = time . time ( ) <EOL> if msg : <EOL> pulse_coro = msg . get ( '<STR_LIT>' , None ) <EOL> interval = msg . get ( '<STR_LIT>' , None ) <EOL> disk_path = msg . get ( '<STR_LIT>' , '<STR_LIT:.>' ) <EOL> if not isinstance ( pulse_coro , asyncoro . Coro ) : <EOL> interval = None <EOL> continue <EOL> last_pulse = now <EOL> continue <EOL> if not pulse_coro : <EOL> continue <EOL> msg = { '<STR_LIT:location>' : coro . location } <EOL> if psutil : <EOL> msg [ '<STR_LIT>' ] = DiscoroNodeAvailInfo ( <EOL> coro . location . addr , <NUM_LIT> - psutil . cpu_percent ( ) , <EOL> psutil . virtual_memory ( ) . available , psutil . disk_usage ( disk_path ) . free , <EOL> <NUM_LIT> - psutil . swap_memory ( ) . percent ) <EOL> if ( yield pulse_coro . deliver ( msg , timeout = <NUM_LIT:2> ) ) == <NUM_LIT:1> : <EOL> last_pulse = now <EOL> elif ( now - last_pulse ) > ( <NUM_LIT:5> * interval ) : <EOL> asyncoro . logger . warning ( '<STR_LIT>' ) <EOL> for _discoro_server_info in _discoro_server_infos : <EOL> _discoro_server_info . Queue . put ( { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _discoro_auth } ) <EOL> if ( now - last_proc_check ) > ( <NUM_LIT:3> * interval ) : <EOL> last_proc_check = now <EOL> for _discoro_server_info in _discoro_server_infos : <EOL> if not _discoro_server_info . Proc . is_alive ( ) : <EOL> asyncoro . logger . warning ( '<STR_LIT>' , <EOL> _discoro_server_info . Proc . pid , <EOL> _discoro_server_info . Proc . exitcode ) <EOL> _discoro_server_id = <NUM_LIT:0> <EOL> _discoro_config [ '<STR_LIT:name>' ] = '<STR_LIT>' % ( _discoro_name , _discoro_server_id ) <EOL> _discoro_config [ '<STR_LIT>' ] = _discoro_tcp_ports [ _discoro_server_id ] <EOL> _discoro_config . pop ( '<STR_LIT>' , False ) <EOL> _discoro_config . pop ( '<STR_LIT>' , - <NUM_LIT:1> ) <EOL> _discoro_config . pop ( '<STR_LIT>' , [ ] ) <EOL> _discoro_config . pop ( '<STR_LIT>' ) <EOL> _discoro_config . pop ( '<STR_LIT>' ) <EOL> _discoro_config [ '<STR_LIT>' ] = False <EOL> if _discoro_config [ '<STR_LIT>' ] : <EOL> asyncoro . logger . setLevel ( logging . DEBUG ) <EOL> else : <EOL> asyncoro . logger . setLevel ( logging . INFO ) <EOL> del _discoro_config [ '<STR_LIT>' ] <EOL> _discoro_scheduler = asyncoro . AsynCoro ( ** _discoro_config ) <EOL> _discoro_timer_coro = asyncoro . Coro ( _discoro_timer_proc ) <EOL> for _discoro_server_info in _discoro_server_infos : <EOL> _discoro_server_info . Queue . put ( { '<STR_LIT>' : '<STR_LIT:start>' , '<STR_LIT>' : _discoro_auth , <EOL> '<STR_LIT>' : _discoro_timer_coro } ) <EOL> del _discoro_mp_queue , _discoro_tcp_ports , tcp_port , tcp_ports , _discoro_config <EOL> del parser , argparse , multiprocessing , socket , collections , os <EOL> if not _discoro_daemon : <EOL> def _discoro_cmd_reader ( coro = None ) : <EOL> coro . set_daemon ( ) <EOL> async_threads = asyncoro . AsyncThreadPool ( <NUM_LIT:1> ) <EOL> while <NUM_LIT:1> : <EOL> yield coro . sleep ( <NUM_LIT> ) <EOL> try : <EOL> _discoro_cmd = yield async_threads . async_task ( <EOL> raw_input , <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> except : <EOL> _discoro_cmd = '<STR_LIT>' <EOL> else : <EOL> _discoro_cmd = _discoro_cmd . strip ( ) . lower ( ) <EOL> print ( '<STR_LIT>' ) <EOL> if _discoro_cmd == '<STR_LIT:status>' or _discoro_cmd == '<STR_LIT>' : <EOL> for _discoro_server_info in _discoro_server_infos : <EOL> _discoro_server_info . Queue . put ( { '<STR_LIT>' : _discoro_cmd , '<STR_LIT>' : _discoro_auth } ) <EOL> elif _discoro_cmd in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> for _discoro_server_info in _discoro_server_infos : <EOL> _discoro_server_info . Queue . put ( { '<STR_LIT>' : _discoro_cmd , '<STR_LIT>' : _discoro_auth } ) <EOL> break <EOL> asyncoro . Coro ( _discoro_cmd_reader ) <EOL> else : <EOL> _discoro_cmd = None <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> for _discoro_server_info in _discoro_server_infos : <EOL> if _discoro_server_info . Proc . is_alive ( ) : <EOL> _discoro_server_info . Proc . join ( ) <EOL> break <EOL> except : <EOL> for _discoro_server_info in _discoro_server_infos : <EOL> _discoro_server_info . Queue . put ( { '<STR_LIT>' : '<STR_LIT>' , '<STR_LIT>' : _discoro_auth } ) <EOL> exit ( <NUM_LIT:0> ) </s>
