<s> """<STR_LIT>""" <EOL> __author__ = "<STR_LIT>" <EOL> __email__ = "<STR_LIT>" <EOL> __copyright__ = "<STR_LIT>" <EOL> __contributors__ = [ ] <EOL> __maintainer__ = "<STR_LIT>" <EOL> __license__ = "<STR_LIT>" <EOL> __url__ = "<STR_LIT>" <EOL> __status__ = "<STR_LIT>" <EOL> import os <EOL> import sys <EOL> import time <EOL> import stat <EOL> import socket <EOL> import multiprocessing <EOL> import threading <EOL> import subprocess <EOL> import traceback <EOL> import logging <EOL> import marshal <EOL> import tempfile <EOL> import shutil <EOL> import glob <EOL> import functools <EOL> import inspect <EOL> import cPickle as pickle <EOL> import cStringIO as io <EOL> try : <EOL> import psutil <EOL> except ImportError : <EOL> psutil = None <EOL> from dispy import _JobReply , DispyJob , DispyNodeAvailInfo , _Function , _Compute , _XferFile , _node_ipaddr , _dispy_version , auth_code , num_min , _same_file , MsgTimeout <EOL> import asyncoro <EOL> from asyncoro import Coro , AsynCoro , AsyncSocket , serialize , unserialize <EOL> __version__ = _dispy_version <EOL> __all__ = [ ] <EOL> MaxFileSize = <NUM_LIT:0> <EOL> def dispy_provisional_result ( result , timeout = MsgTimeout ) : <EOL> """<STR_LIT>""" <EOL> dispy_job_reply = __dispy_job_info . job_reply <EOL> dispy_job_reply . status = DispyJob . ProvisionalResult <EOL> dispy_job_reply . result = result <EOL> dispy_job_reply . end_time = time . time ( ) <EOL> sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> sock = AsyncSocket ( sock , blocking = True , keyfile = __dispy_job_keyfile , <EOL> certfile = __dispy_job_certfile ) <EOL> sock . settimeout ( timeout ) <EOL> try : <EOL> sock . connect ( __dispy_job_info . reply_addr ) <EOL> sock . send_msg ( '<STR_LIT>' + serialize ( dispy_job_reply ) ) <EOL> ack = sock . recv_msg ( ) <EOL> assert ack == '<STR_LIT>' <EOL> except : <EOL> print ( "<STR_LIT>" % ( str ( result ) , traceback . format_exc ( ) ) ) <EOL> return - <NUM_LIT:1> <EOL> else : <EOL> return <NUM_LIT:0> <EOL> finally : <EOL> sock . close ( ) <EOL> def dispy_send_file ( path , timeout = MsgTimeout ) : <EOL> """<STR_LIT>""" <EOL> path = os . path . expanduser ( path ) <EOL> xf = _XferFile ( path , os . stat ( path ) ) <EOL> if MaxFileSize and xf . stat_buf . st_size > MaxFileSize : <EOL> return - <NUM_LIT:1> <EOL> xf . name = os . path . splitdrive ( path ) [ <NUM_LIT:1> ] <EOL> if xf . name . startswith ( os . sep ) : <EOL> xf . name = xf . name [ len ( os . sep ) : ] <EOL> dispy_job_reply = __dispy_job_info . job_reply <EOL> sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> sock = AsyncSocket ( sock , blocking = True , <EOL> keyfile = __dispy_job_keyfile , certfile = __dispy_job_certfile ) <EOL> sock . settimeout ( timeout ) <EOL> try : <EOL> sock . connect ( __dispy_job_info . reply_addr ) <EOL> sock . send_msg ( '<STR_LIT>' . encode ( ) + serialize ( xf ) ) <EOL> sock . send_msg ( serialize ( dispy_job_reply ) ) <EOL> recvd = sock . recv_msg ( ) <EOL> recvd = unserialize ( recvd ) <EOL> fd = open ( path , '<STR_LIT:rb>' ) <EOL> sent = <NUM_LIT:0> <EOL> while sent == recvd : <EOL> data = fd . read ( <NUM_LIT> ) <EOL> if not data : <EOL> break <EOL> sock . sendall ( data ) <EOL> sent += len ( data ) <EOL> recvd = sock . recv_msg ( ) <EOL> recvd = unserialize ( recvd ) <EOL> fd . close ( ) <EOL> assert recvd == xf . stat_buf . st_size <EOL> except : <EOL> print ( '<STR_LIT>' % ( path , traceback . format_exc ( ) ) ) <EOL> return - <NUM_LIT:1> <EOL> else : <EOL> return <NUM_LIT:0> <EOL> finally : <EOL> sock . close ( ) <EOL> class _DispyJobInfo ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , job_reply , reply_addr , compute , xfer_files ) : <EOL> self . job_reply = job_reply <EOL> self . reply_addr = reply_addr <EOL> self . compute_id = compute . id <EOL> self . compute_dest_path = compute . dest_path <EOL> self . xfer_files = xfer_files <EOL> self . compute_auth = compute . auth <EOL> self . proc = None <EOL> def _dispy_job_func ( __dispy_job_info , __dispy_job_certfile , __dispy_job_keyfile , <EOL> __dispy_job_name , __dispy_job_args , __dispy_job_kwargs , <EOL> __dispy_job_code , __dispy_job_globals , __dispy_path , __dispy_reply_Q ) : <EOL> """<STR_LIT>""" <EOL> os . chdir ( __dispy_path ) <EOL> sys . stdout = io . StringIO ( ) <EOL> sys . stderr = io . StringIO ( ) <EOL> __dispy_job_reply = __dispy_job_info . job_reply <EOL> globals ( ) . update ( __dispy_job_globals ) <EOL> try : <EOL> exec ( marshal . loads ( __dispy_job_code [ <NUM_LIT:0> ] ) ) in globals ( ) <EOL> if __dispy_job_code [ <NUM_LIT:1> ] : <EOL> exec ( __dispy_job_code [ <NUM_LIT:1> ] ) in globals ( ) <EOL> __dispy_job_args = unserialize ( __dispy_job_args ) <EOL> __dispy_job_kwargs = unserialize ( __dispy_job_kwargs ) <EOL> globals ( ) . update ( locals ( ) ) <EOL> exec ( '<STR_LIT>' % <EOL> __dispy_job_name ) in globals ( ) <EOL> __dispy_job_reply . status = DispyJob . Finished <EOL> except : <EOL> __dispy_job_reply . exception = traceback . format_exc ( ) <EOL> __dispy_job_reply . status = DispyJob . Terminated <EOL> __dispy_job_reply . stdout = sys . stdout . getvalue ( ) <EOL> __dispy_job_reply . stderr = sys . stderr . getvalue ( ) <EOL> __dispy_job_reply . end_time = time . time ( ) <EOL> __dispy_job_info . proc = None <EOL> __dispy_reply_Q . put ( __dispy_job_reply ) <EOL> class _DispyNode ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , cpus , ip_addr = None , ext_ip_addr = None , node_port = None , <EOL> name = '<STR_LIT>' , scheduler_node = None , scheduler_port = None , <EOL> dest_path_prefix = '<STR_LIT>' , clean = False , secret = '<STR_LIT>' , keyfile = None , certfile = None , <EOL> zombie_interval = <NUM_LIT> , service_start = None , service_stop = None , service_end = None , <EOL> serve = - <NUM_LIT:1> , daemon = False ) : <EOL> assert <NUM_LIT:0> < cpus <= multiprocessing . cpu_count ( ) <EOL> self . num_cpus = cpus <EOL> if name : <EOL> self . name = name <EOL> else : <EOL> self . name = socket . gethostname ( ) <EOL> if ip_addr : <EOL> ip_addr = _node_ipaddr ( ip_addr ) <EOL> if not ip_addr : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> else : <EOL> ip_addr = socket . gethostbyname ( socket . gethostname ( ) ) <EOL> if ip_addr . startswith ( '<STR_LIT>' ) : <EOL> _dispy_logger . warning ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' , ip_addr ) <EOL> if ext_ip_addr : <EOL> ext_ip_addr = _node_ipaddr ( ext_ip_addr ) <EOL> if not ext_ip_addr : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> else : <EOL> ext_ip_addr = ip_addr <EOL> if not self . name : <EOL> try : <EOL> self . name = socket . gethostbyaddr ( ext_ip_addr ) [ <NUM_LIT:0> ] <EOL> except : <EOL> self . name = '<STR_LIT>' <EOL> if node_port is None : <EOL> node_port = <NUM_LIT> <EOL> self . ext_ip_addr = ext_ip_addr <EOL> self . pulse_interval = None <EOL> self . keyfile = keyfile <EOL> self . certfile = certfile <EOL> if self . keyfile : <EOL> self . keyfile = os . path . abspath ( self . keyfile ) <EOL> if self . certfile : <EOL> self . certfile = os . path . abspath ( self . certfile ) <EOL> self . asyncoro = AsynCoro ( ) <EOL> self . tcp_sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) , <EOL> keyfile = keyfile , certfile = certfile ) <EOL> self . tcp_sock . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> self . tcp_sock . bind ( ( ip_addr , node_port ) ) <EOL> self . address = self . tcp_sock . getsockname ( ) <EOL> self . port = self . address [ <NUM_LIT:1> ] <EOL> self . tcp_sock . listen ( <NUM_LIT:30> ) <EOL> if not dest_path_prefix : <EOL> dest_path_prefix = os . path . join ( tempfile . gettempdir ( ) , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . dest_path_prefix = os . path . abspath ( dest_path_prefix . strip ( ) ) . rstrip ( os . sep ) <EOL> if clean : <EOL> shutil . rmtree ( self . dest_path_prefix , ignore_errors = True ) <EOL> if not os . path . isdir ( self . dest_path_prefix ) : <EOL> os . makedirs ( self . dest_path_prefix ) <EOL> os . chmod ( self . dest_path_prefix , stat . S_IRUSR | stat . S_IWUSR | stat . S_IXUSR ) <EOL> self . avail_cpus = self . num_cpus <EOL> self . computations = { } <EOL> self . job_infos = { } <EOL> self . terminate = False <EOL> self . sign = os . urandom ( <NUM_LIT:10> ) . encode ( '<STR_LIT>' ) <EOL> self . secret = secret <EOL> self . auth = auth_code ( self . secret , self . sign ) <EOL> self . zombie_interval = <NUM_LIT> * zombie_interval <EOL> if not scheduler_port : <EOL> scheduler_port = <NUM_LIT> <EOL> self . scheduler = { '<STR_LIT>' : None , '<STR_LIT:port>' : scheduler_port , '<STR_LIT>' : set ( ) } <EOL> self . cpu_time = <NUM_LIT:0> <EOL> self . num_jobs = <NUM_LIT:0> <EOL> self . num_computations = <NUM_LIT:0> <EOL> fd = open ( os . path . join ( self . dest_path_prefix , '<STR_LIT>' ) , '<STR_LIT:wb>' ) <EOL> config = { <EOL> '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT:port>' : self . port , '<STR_LIT>' : self . avail_cpus , <EOL> '<STR_LIT>' : self . sign , '<STR_LIT>' : self . secret , '<STR_LIT>' : self . auth <EOL> } <EOL> pickle . dump ( config , fd ) <EOL> fd . close ( ) <EOL> sys . path . insert ( <NUM_LIT:0> , '<STR_LIT:.>' ) <EOL> proc = multiprocessing . Process ( target = functools . partial ( int ) , args = ( <NUM_LIT> , ) ) <EOL> proc . start ( ) <EOL> proc . join ( ) <EOL> self . thread_lock = threading . Lock ( ) <EOL> self . udp_sock = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) <EOL> self . udp_sock . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> self . udp_sock . bind ( ( '<STR_LIT>' , self . port ) ) <EOL> _dispy_logger . info ( '<STR_LIT>' , <EOL> self . num_cpus , self . ext_ip_addr , self . port ) <EOL> _dispy_logger . debug ( '<STR_LIT>' , self . address [ <NUM_LIT:0> ] , self . address [ <NUM_LIT:1> ] ) <EOL> self . udp_sock = AsyncSocket ( self . udp_sock ) <EOL> self . reply_Q = multiprocessing . Queue ( ) <EOL> self . reply_Q_thread = threading . Thread ( target = self . __reply_Q ) <EOL> self . reply_Q_thread . start ( ) <EOL> self . serve = serve <EOL> self . timer_coro = Coro ( self . timer_task ) <EOL> self . service_start = self . service_stop = self . service_end = None <EOL> if isinstance ( service_start , time . struct_time ) and ( isinstance ( service_stop , time . struct_time ) or <EOL> isinstance ( service_end , time . struct_time ) ) : <EOL> self . service_start = ( service_start . tm_hour , service_start . tm_min ) <EOL> if isinstance ( service_stop , time . struct_time ) : <EOL> self . service_stop = ( service_stop . tm_hour , service_stop . tm_min ) <EOL> if isinstance ( service_end , time . struct_time ) : <EOL> self . service_end = ( service_end . tm_hour , service_end . tm_min ) <EOL> Coro ( self . service_schedule ) <EOL> self . __init_code = '<STR_LIT>' . join ( inspect . getsource ( dispy_provisional_result ) ) <EOL> self . __init_code += '<STR_LIT>' . join ( inspect . getsource ( dispy_send_file ) ) <EOL> self . __init_modules = dict ( sys . modules ) <EOL> if os . name == '<STR_LIT>' : <EOL> self . __init_globals = dict ( globals ( ) ) <EOL> self . tcp_coro = Coro ( self . tcp_server ) <EOL> self . udp_coro = Coro ( self . udp_server , _node_ipaddr ( scheduler_node ) , scheduler_port ) <EOL> if not daemon : <EOL> Coro ( self . read_stdin ) <EOL> def broadcast_ping_msg ( self , coro = None ) : <EOL> if ( self . scheduler [ '<STR_LIT>' ] or self . job_infos or not self . avail_cpus or <EOL> not self . service_available ( ) ) : <EOL> raise StopIteration <EOL> sock = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) <EOL> sock . setsockopt ( socket . SOL_SOCKET , socket . SO_BROADCAST , <NUM_LIT:1> ) <EOL> sock = AsyncSocket ( sock ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> ping_msg = { '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT:port>' : self . port , '<STR_LIT>' : self . sign , <EOL> '<STR_LIT:version>' : _dispy_version , '<STR_LIT>' : None } <EOL> try : <EOL> yield sock . sendto ( '<STR_LIT>' . encode ( ) + serialize ( ping_msg ) , <EOL> ( '<STR_LIT>' , self . scheduler [ '<STR_LIT:port>' ] ) ) <EOL> except : <EOL> _dispy_logger . debug ( traceback . format_exc ( ) ) <EOL> pass <EOL> sock . close ( ) <EOL> def send_pong_msg ( self , info , addr , coro = None ) : <EOL> if ( self . scheduler [ '<STR_LIT>' ] or self . job_infos or not self . num_cpus or <EOL> not self . service_available ( ) ) : <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> self . avail_cpus , self . num_cpus , addr [ <NUM_LIT:0> ] ) <EOL> raise StopIteration <EOL> try : <EOL> scheduler_ip_addrs = info [ '<STR_LIT>' ] <EOL> if not info . get ( '<STR_LIT>' , None ) : <EOL> scheduler_ip_addrs . append ( addr [ <NUM_LIT:0> ] ) <EOL> scheduler_port = info [ '<STR_LIT:port>' ] <EOL> except : <EOL> _dispy_logger . debug ( traceback . format_exc ( ) ) <EOL> raise StopIteration <EOL> if info . get ( '<STR_LIT>' , None ) : <EOL> pong_msg = { '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT:port>' : self . port , '<STR_LIT>' : self . sign , <EOL> '<STR_LIT:version>' : _dispy_version , '<STR_LIT:name>' : self . name , '<STR_LIT>' : self . avail_cpus , <EOL> '<STR_LIT>' : auth_code ( self . secret , info [ '<STR_LIT>' ] ) } <EOL> if psutil : <EOL> pong_msg [ '<STR_LIT>' ] = DispyNodeAvailInfo ( <EOL> <NUM_LIT> - psutil . cpu_percent ( ) , psutil . virtual_memory ( ) . available , <EOL> psutil . disk_usage ( self . dest_path_prefix ) . free , <EOL> <NUM_LIT> - psutil . swap_memory ( ) . percent ) <EOL> else : <EOL> pong_msg [ '<STR_LIT>' ] = None <EOL> for scheduler_ip_addr in scheduler_ip_addrs : <EOL> addr = ( scheduler_ip_addr , scheduler_port ) <EOL> pong_msg [ '<STR_LIT>' ] = scheduler_ip_addr <EOL> sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) , <EOL> keyfile = self . keyfile , certfile = self . certfile ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> try : <EOL> yield sock . connect ( addr ) <EOL> yield sock . send_msg ( '<STR_LIT>' . encode ( ) + serialize ( pong_msg ) ) <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' , addr [ <NUM_LIT:0> ] , addr [ <NUM_LIT:1> ] ) <EOL> finally : <EOL> sock . close ( ) <EOL> else : <EOL> sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> ping_msg = { '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT:port>' : self . port , '<STR_LIT>' : self . sign , <EOL> '<STR_LIT:version>' : _dispy_version } <EOL> for scheduler_ip_addr in scheduler_ip_addrs : <EOL> addr = ( scheduler_ip_addr , scheduler_port ) <EOL> ping_msg [ '<STR_LIT>' ] = scheduler_ip_addr <EOL> try : <EOL> yield sock . sendto ( '<STR_LIT>' . encode ( ) + serialize ( ping_msg ) , addr ) <EOL> except : <EOL> _dispy_logger . debug ( traceback . format_exc ( ) ) <EOL> pass <EOL> sock . close ( ) <EOL> def udp_server ( self , scheduler_ip , scheduler_port , coro = None ) : <EOL> coro . set_daemon ( ) <EOL> yield self . broadcast_ping_msg ( coro = coro ) <EOL> ping_msg = { '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT:port>' : self . port , '<STR_LIT>' : self . sign , <EOL> '<STR_LIT:version>' : _dispy_version } <EOL> def send_ping_msg ( self , info , coro = None ) : <EOL> sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> addr = ( info [ '<STR_LIT>' ] , info [ '<STR_LIT:port>' ] ) <EOL> info . update ( ping_msg ) <EOL> info [ '<STR_LIT>' ] = addr [ <NUM_LIT:0> ] <EOL> try : <EOL> yield sock . sendto ( '<STR_LIT>' . encode ( ) + serialize ( info ) , addr ) <EOL> except : <EOL> pass <EOL> finally : <EOL> sock . close ( ) <EOL> sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) , <EOL> keyfile = self . keyfile , certfile = self . certfile ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> try : <EOL> yield sock . connect ( addr ) <EOL> yield sock . send_msg ( '<STR_LIT>' . encode ( ) + serialize ( info ) ) <EOL> except : <EOL> pass <EOL> finally : <EOL> sock . close ( ) <EOL> if scheduler_ip : <EOL> Coro ( send_ping_msg , self , { '<STR_LIT>' : scheduler_ip , '<STR_LIT:port>' : scheduler_port } ) <EOL> while True : <EOL> msg , addr = yield self . udp_sock . recvfrom ( <NUM_LIT:1000> ) <EOL> if msg . startswith ( '<STR_LIT>' ) : <EOL> try : <EOL> info = unserialize ( msg [ len ( '<STR_LIT>' ) : ] ) <EOL> if info [ '<STR_LIT:version>' ] != _dispy_version : <EOL> _dispy_logger . warning ( '<STR_LIT>' , addr [ <NUM_LIT:0> ] ) <EOL> continue <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' , addr [ <NUM_LIT:0> ] , addr [ <NUM_LIT:1> ] ) <EOL> continue <EOL> Coro ( self . send_pong_msg , info , addr ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> try : <EOL> info = unserialize ( msg [ len ( '<STR_LIT>' ) : ] ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , addr [ <NUM_LIT:0> ] ) <EOL> else : <EOL> if info [ '<STR_LIT>' ] == self . scheduler [ '<STR_LIT>' ] : <EOL> now = time . time ( ) <EOL> for compute in self . computations . itervalues ( ) : <EOL> compute . last_pulse = now <EOL> else : <EOL> _dispy_logger . warning ( '<STR_LIT>' , addr [ <NUM_LIT:0> ] ) <EOL> def tcp_server ( self ) : <EOL> while <NUM_LIT:1> : <EOL> try : <EOL> conn , addr = yield self . tcp_sock . accept ( ) <EOL> except GeneratorExit : <EOL> break <EOL> except : <EOL> _dispy_logger . debug ( traceback . format_exc ( ) ) <EOL> continue <EOL> Coro ( self . tcp_serve_task , conn , addr ) <EOL> def tcp_serve_task ( self , conn , addr , coro = None ) : <EOL> def job_request_task ( msg ) : <EOL> try : <EOL> _job = unserialize ( msg ) <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' , addr [ <NUM_LIT:0> ] ) <EOL> raise StopIteration <EOL> compute = self . computations . get ( _job . compute_id , None ) <EOL> if compute is not None : <EOL> if compute . scheduler_ip_addr != self . scheduler [ '<STR_LIT>' ] or compute . scheduler_port != self . scheduler [ '<STR_LIT:port>' ] or compute . auth not in self . scheduler [ '<STR_LIT>' ] : <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> compute . scheduler_ip_addr , compute . scheduler_port , <EOL> self . scheduler [ '<STR_LIT>' ] , self . scheduler [ '<STR_LIT:port>' ] ) <EOL> compute = None <EOL> if self . avail_cpus == <NUM_LIT:0> : <EOL> try : <EOL> yield conn . send_msg ( '<STR_LIT>' . encode ( ) ) <EOL> except : <EOL> pass <EOL> raise StopIteration <EOL> elif compute is None : <EOL> _dispy_logger . warning ( '<STR_LIT>' , _job . compute_id ) <EOL> try : <EOL> yield conn . send_msg ( ( '<STR_LIT>' % <EOL> _job . compute_id ) . encode ( ) ) <EOL> except : <EOL> pass <EOL> raise StopIteration <EOL> for xf in _job . xfer_files : <EOL> if MaxFileSize and xf . stat_buf . st_size > MaxFileSize : <EOL> try : <EOL> yield conn . send_msg ( '<STR_LIT>' . encode ( ) ) <EOL> except : <EOL> pass <EOL> raise StopIteration <EOL> reply_addr = ( compute . scheduler_ip_addr , compute . job_result_port ) <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> _job . uid , addr [ <NUM_LIT:0> ] , compute . scheduler_ip_addr ) <EOL> if compute . type == _Compute . func_type : <EOL> reply = _JobReply ( _job , self . ext_ip_addr ) <EOL> reply . start_time = time . time ( ) <EOL> job_info = _DispyJobInfo ( reply , reply_addr , compute , _job . xfer_files ) <EOL> args = ( job_info , self . certfile , self . keyfile , compute . name , <EOL> _job . args , _job . kwargs , ( compute . code , _job . code ) , <EOL> compute . globals , compute . dest_path , self . reply_Q ) <EOL> try : <EOL> yield conn . send_msg ( '<STR_LIT>' ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , str ( addr ) ) <EOL> raise StopIteration <EOL> proc = multiprocessing . Process ( target = _dispy_job_func , args = args ) <EOL> self . avail_cpus -= <NUM_LIT:1> <EOL> compute . pending_jobs += <NUM_LIT:1> <EOL> self . thread_lock . acquire ( ) <EOL> self . job_infos [ _job . uid ] = job_info <EOL> self . thread_lock . release ( ) <EOL> try : <EOL> proc . start ( ) <EOL> except : <EOL> job_info . job_reply . status = DispyJob . Terminated <EOL> job_info . job_reply . exception = traceback . format_exc ( ) <EOL> job_info . job_reply . end_time = time . time ( ) <EOL> job_info . proc = None <EOL> self . reply_Q . put ( job_info . job_reply ) <EOL> else : <EOL> job_info . proc = proc <EOL> job_info . job_reply . status = DispyJob . Running <EOL> raise StopIteration <EOL> elif compute . type == _Compute . prog_type : <EOL> try : <EOL> yield conn . send_msg ( '<STR_LIT>' ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , str ( addr ) ) <EOL> raise StopIteration <EOL> reply = _JobReply ( _job , self . ext_ip_addr ) <EOL> reply . start_time = time . time ( ) <EOL> job_info = _DispyJobInfo ( reply , reply_addr , compute , _job . xfer_files ) <EOL> job_info . job_reply . status = DispyJob . Running <EOL> self . thread_lock . acquire ( ) <EOL> self . job_infos [ _job . uid ] = job_info <EOL> self . thread_lock . release ( ) <EOL> self . avail_cpus -= <NUM_LIT:1> <EOL> compute . pending_jobs += <NUM_LIT:1> <EOL> prog_thread = threading . Thread ( target = self . __job_program , args = ( _job , job_info ) ) <EOL> prog_thread . start ( ) <EOL> raise StopIteration <EOL> else : <EOL> try : <EOL> yield conn . send_msg ( ( '<STR_LIT>' % <EOL> compute . type ) . encode ( ) ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , str ( addr ) ) <EOL> def add_computation_task ( msg ) : <EOL> try : <EOL> compute = unserialize ( msg ) <EOL> except : <EOL> try : <EOL> yield conn . send_msg ( ( '<STR_LIT>' ) . encode ( ) ) <EOL> except : <EOL> pass <EOL> raise StopIteration <EOL> if not ( ( self . scheduler [ '<STR_LIT>' ] is None and not self . scheduler [ '<STR_LIT>' ] ) or <EOL> ( self . scheduler [ '<STR_LIT>' ] == compute . scheduler_ip_addr and <EOL> self . scheduler [ '<STR_LIT:port>' ] == compute . scheduler_port and <EOL> self . service_available ( ) ) ) : <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> compute . scheduler_ip_addr , self . scheduler [ '<STR_LIT>' ] , <EOL> self . avail_cpus , self . num_cpus ) <EOL> try : <EOL> yield conn . send_msg ( ( '<STR_LIT>' ) . encode ( ) ) <EOL> except : <EOL> pass <EOL> raise StopIteration <EOL> if MaxFileSize : <EOL> for xf in compute . xfer_files : <EOL> if xf . stat_buf . st_size > MaxFileSize : <EOL> try : <EOL> yield conn . send_msg ( ( '<STR_LIT>' % <EOL> ( xf . name , MaxFileSize ) ) . encode ( ) ) <EOL> except : <EOL> pass <EOL> raise StopIteration <EOL> compute . xfer_files = set ( ) <EOL> dest = os . path . join ( self . dest_path_prefix , compute . scheduler_ip_addr ) <EOL> if not os . path . isdir ( dest ) : <EOL> try : <EOL> os . mkdir ( dest ) <EOL> except : <EOL> yield conn . send_msg ( ( '<STR_LIT>' ) . encode ( ) ) <EOL> raise StopIteration <EOL> if compute . dest_path and isinstance ( compute . dest_path , str ) : <EOL> if not compute . dest_path . startswith ( os . sep ) : <EOL> compute . dest_path = os . path . join ( dest , compute . dest_path ) <EOL> if not os . path . isdir ( compute . dest_path ) : <EOL> try : <EOL> os . makedirs ( compute . dest_path ) <EOL> except : <EOL> try : <EOL> yield conn . send_msg ( ( '<STR_LIT>' ) . encode ( ) ) <EOL> except : <EOL> pass <EOL> raise StopIteration <EOL> else : <EOL> compute . dest_path = tempfile . mkdtemp ( prefix = compute . name + '<STR_LIT:_>' , dir = dest ) <EOL> os . chmod ( compute . dest_path , stat . S_IRUSR | stat . S_IWUSR | stat . S_IXUSR ) <EOL> if compute . id in self . computations : <EOL> _dispy_logger . warning ( '<STR_LIT>' , <EOL> compute . name , compute . id ) <EOL> setattr ( compute , '<STR_LIT>' , time . time ( ) ) <EOL> setattr ( compute , '<STR_LIT>' , <NUM_LIT:0> ) <EOL> setattr ( compute , '<STR_LIT>' , <NUM_LIT:0> ) <EOL> setattr ( compute , '<STR_LIT>' , False ) <EOL> setattr ( compute , '<STR_LIT>' , { } ) <EOL> setattr ( compute , '<STR_LIT>' , set ( sys . modules . iterkeys ( ) ) ) <EOL> setattr ( compute , '<STR_LIT>' , { } ) <EOL> if compute . code : <EOL> try : <EOL> code = compute . code <EOL> code += self . __init_code <EOL> code = compile ( code , '<STR_LIT>' , '<STR_LIT>' ) <EOL> except : <EOL> if os . path . isdir ( compute . dest_path ) : <EOL> os . rmdir ( compute . dest_path ) <EOL> try : <EOL> yield conn . send_msg ( ( '<STR_LIT>' % <EOL> ( self . ext_ip_addr , compute . name ) ) . encode ( ) ) <EOL> except : <EOL> pass <EOL> raise StopIteration <EOL> compute . code = marshal . dumps ( code ) <EOL> if compute . type == _Compute . prog_type : <EOL> compute . name = os . path . join ( compute . dest_path , os . path . basename ( compute . name ) ) <EOL> if not ( ( self . scheduler [ '<STR_LIT>' ] is None ) or <EOL> ( self . scheduler [ '<STR_LIT>' ] == compute . scheduler_ip_addr and <EOL> self . scheduler [ '<STR_LIT:port>' ] == compute . scheduler_port ) ) : <EOL> if os . path . isdir ( compute . dest_path ) : <EOL> try : <EOL> os . rmdir ( compute . dest_path ) <EOL> yield conn . send_msg ( serialize ( - <NUM_LIT:1> ) ) <EOL> except : <EOL> pass <EOL> raise StopIteration <EOL> self . computations [ compute . id ] = compute <EOL> self . scheduler [ '<STR_LIT>' ] = compute . scheduler_ip_addr <EOL> self . scheduler [ '<STR_LIT:port>' ] = compute . scheduler_port <EOL> self . scheduler [ '<STR_LIT>' ] . add ( compute . auth ) <EOL> compute_save = os . path . join ( self . dest_path_prefix , '<STR_LIT>' % ( compute . id , compute . auth ) ) <EOL> fd = open ( compute_save , '<STR_LIT:wb>' ) <EOL> pickle . dump ( compute , fd ) <EOL> fd . close ( ) <EOL> if os . name == '<STR_LIT>' : <EOL> compute . globals = { } <EOL> else : <EOL> for var in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , <EOL> '<STR_LIT>' , '<STR_LIT>' ) : <EOL> compute . globals [ var ] = globals ( ) [ var ] <EOL> compute . globals . update ( self . __init_modules ) <EOL> compute . globals [ '<STR_LIT>' ] = None <EOL> try : <EOL> yield conn . send_msg ( serialize ( self . avail_cpus ) ) <EOL> except : <EOL> del self . computations [ compute . id ] <EOL> compute . globals = { } <EOL> self . scheduler [ '<STR_LIT>' ] = None <EOL> self . scheduler [ '<STR_LIT>' ] . discard ( compute . auth ) <EOL> os . remove ( compute_save ) <EOL> if os . path . isdir ( compute . dest_path ) : <EOL> try : <EOL> os . rmdir ( compute . dest_path ) <EOL> except : <EOL> pass <EOL> else : <EOL> self . pulse_interval = num_min ( self . pulse_interval , compute . pulse_interval ) <EOL> if not self . pulse_interval : <EOL> self . pulse_interval = <NUM_LIT:10> * <NUM_LIT> <EOL> if self . zombie_interval : <EOL> self . pulse_interval = num_min ( self . pulse_interval , self . zombie_interval / <NUM_LIT> ) <EOL> self . timer_coro . resume ( True ) <EOL> def xfer_file_task ( msg ) : <EOL> try : <EOL> xf = unserialize ( msg ) <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' , addr [ <NUM_LIT:0> ] ) <EOL> raise StopIteration <EOL> compute = self . computations . get ( xf . compute_id , None ) <EOL> if not compute or ( MaxFileSize and xf . stat_buf . st_size > MaxFileSize ) : <EOL> _dispy_logger . error ( '<STR_LIT>' , xf . name ) <EOL> yield conn . send_msg ( serialize ( - <NUM_LIT:1> ) ) <EOL> raise StopIteration <EOL> tgt = os . path . join ( compute . dest_path , os . path . basename ( xf . name ) ) <EOL> if os . path . isfile ( tgt ) and _same_file ( tgt , xf ) : <EOL> if tgt in compute . file_uses : <EOL> compute . file_uses [ tgt ] += <NUM_LIT:1> <EOL> else : <EOL> compute . file_uses [ tgt ] = <NUM_LIT:2> <EOL> yield conn . send_msg ( serialize ( xf . stat_buf . st_size ) ) <EOL> else : <EOL> try : <EOL> fd = open ( tgt , '<STR_LIT:wb>' ) <EOL> recvd = <NUM_LIT:0> <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> xf . name , tgt , xf . stat_buf . st_size ) <EOL> while recvd < xf . stat_buf . st_size : <EOL> yield conn . send_msg ( serialize ( recvd ) ) <EOL> data = yield conn . recvall ( min ( xf . stat_buf . st_size - recvd , <NUM_LIT> ) ) <EOL> if not data : <EOL> break <EOL> fd . write ( data ) <EOL> recvd += len ( data ) <EOL> yield conn . send_msg ( serialize ( recvd ) ) <EOL> fd . close ( ) <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> tgt , recvd , xf . stat_buf . st_size ) <EOL> assert recvd == xf . stat_buf . st_size <EOL> os . utime ( tgt , ( xf . stat_buf . st_atime , xf . stat_buf . st_mtime ) ) <EOL> os . chmod ( tgt , stat . S_IMODE ( xf . stat_buf . st_mode ) ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , <EOL> xf . name , traceback . format_exc ( ) ) <EOL> os . remove ( tgt ) <EOL> else : <EOL> if tgt in compute . file_uses : <EOL> compute . file_uses [ tgt ] += <NUM_LIT:1> <EOL> else : <EOL> compute . file_uses [ tgt ] = <NUM_LIT:1> <EOL> raise StopIteration <EOL> def setup_computation ( msg ) : <EOL> try : <EOL> compute_id = unserialize ( msg ) <EOL> compute = self . computations [ compute_id ] <EOL> assert isinstance ( compute . setup , _Function ) <EOL> os . chdir ( compute . dest_path ) <EOL> localvars = { '<STR_LIT>' : compute . setup . args , <EOL> '<STR_LIT>' : compute . setup . kwargs } <EOL> if os . name == '<STR_LIT>' : <EOL> globalvars = globals ( ) <EOL> else : <EOL> globalvars = compute . globals <EOL> exec ( marshal . loads ( compute . code ) ) in globalvars , localvars <EOL> exec ( '<STR_LIT>' % <EOL> compute . setup . name ) in globalvars , localvars <EOL> if os . name == '<STR_LIT>' : <EOL> compute . globals . update ( { var : globals ( ) [ var ] for var in globals ( ) <EOL> if var not in self . __init_globals } ) <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' ) <EOL> resp = traceback . format_exc ( ) . encode ( ) <EOL> else : <EOL> resp = '<STR_LIT>' <EOL> if resp != '<STR_LIT>' : <EOL> if not compute . cleanup : <EOL> compute . cleanup = True <EOL> compute . zombie = True <EOL> self . cleanup_computation ( compute ) <EOL> yield conn . send_msg ( resp ) <EOL> def terminate_job_task ( compute , job_info ) : <EOL> if not job_info . proc : <EOL> raise StopIteration <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> job_info . job_reply . uid , compute . name ) <EOL> job_info . proc . terminate ( ) <EOL> if isinstance ( job_info . proc , multiprocessing . Process ) : <EOL> for x in range ( <NUM_LIT:20> ) : <EOL> if job_info . proc . is_alive ( ) : <EOL> yield coro . sleep ( <NUM_LIT:0.1> ) <EOL> else : <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> compute . name , job_info . job_reply . uid ) <EOL> break <EOL> else : <EOL> _dispy_logger . warning ( '<STR_LIT>' , compute . name ) <EOL> raise StopIteration <EOL> else : <EOL> assert isinstance ( job_info . proc , subprocess . Popen ) <EOL> for x in range ( <NUM_LIT:20> ) : <EOL> rc = job_info . proc . poll ( ) <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> compute . name , job_info . job_reply . uid , rc ) <EOL> if rc is not None : <EOL> break <EOL> if x == <NUM_LIT:10> : <EOL> _dispy_logger . debug ( '<STR_LIT>' , job_info . job_reply . uid ) <EOL> job_info . proc . kill ( ) <EOL> yield coro . sleep ( <NUM_LIT:0.1> ) <EOL> else : <EOL> _dispy_logger . warning ( '<STR_LIT>' , compute . name ) <EOL> raise StopIteration <EOL> job_info . job_reply . end_time = time . time ( ) <EOL> job_info . proc = None <EOL> self . thread_lock . acquire ( ) <EOL> if self . job_infos . get ( job_info . job_reply . uid , None ) == job_info : <EOL> job_info . job_reply . status = DispyJob . Terminated <EOL> self . reply_Q . put ( job_info . job_reply ) <EOL> self . thread_lock . release ( ) <EOL> def retrieve_job_task ( msg ) : <EOL> def send_reply ( reply ) : <EOL> try : <EOL> yield conn . send_msg ( serialize ( reply ) ) <EOL> except : <EOL> raise StopIteration ( - <NUM_LIT:1> ) <EOL> raise StopIteration ( <NUM_LIT:0> ) <EOL> try : <EOL> req = unserialize ( msg ) <EOL> uid = req [ '<STR_LIT>' ] <EOL> compute_id = req [ '<STR_LIT>' ] <EOL> auth = req [ '<STR_LIT>' ] <EOL> job_hash = req [ '<STR_LIT>' ] <EOL> except : <EOL> yield send_reply ( None ) <EOL> raise StopIteration <EOL> pkl_path = os . path . join ( self . dest_path_prefix , '<STR_LIT>' % ( compute_id , auth ) ) <EOL> compute = self . computations . get ( compute_id , None ) <EOL> if not compute : <EOL> fd = open ( pkl_path , '<STR_LIT:rb>' ) <EOL> compute = pickle . load ( fd ) <EOL> fd . close ( ) <EOL> if not compute or compute . auth != auth : <EOL> yield send_reply ( None ) <EOL> raise StopIteration <EOL> info_file = os . path . join ( compute . dest_path , '<STR_LIT>' % uid ) <EOL> if not os . path . isfile ( info_file ) : <EOL> yield send_reply ( None ) <EOL> raise StopIteration <EOL> try : <EOL> fd = open ( info_file , '<STR_LIT:rb>' ) <EOL> job_reply = pickle . load ( fd ) <EOL> fd . close ( ) <EOL> assert job_reply . hash == job_hash <EOL> except : <EOL> yield send_reply ( None ) <EOL> raise StopIteration <EOL> try : <EOL> yield conn . send_msg ( serialize ( job_reply ) ) <EOL> ack = yield conn . recv_msg ( ) <EOL> assert ack == '<STR_LIT>' <EOL> compute . pending_results -= <NUM_LIT:1> <EOL> fd = open ( pkl_path , '<STR_LIT:wb>' ) <EOL> pickle . dump ( compute , fd ) <EOL> fd . close ( ) <EOL> except : <EOL> pass <EOL> else : <EOL> try : <EOL> os . remove ( info_file ) <EOL> except : <EOL> pass <EOL> if compute . pending_results == <NUM_LIT:0> : <EOL> self . cleanup_computation ( compute ) <EOL> try : <EOL> req = yield conn . recvall ( len ( self . auth ) ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' ) <EOL> conn . close ( ) <EOL> raise StopIteration <EOL> msg = yield conn . recv_msg ( ) <EOL> if req != self . auth : <EOL> if msg . startswith ( '<STR_LIT>' ) : <EOL> pass <EOL> else : <EOL> _dispy_logger . warning ( '<STR_LIT>' ) <EOL> conn . close ( ) <EOL> raise StopIteration <EOL> if not msg : <EOL> conn . close ( ) <EOL> raise StopIteration <EOL> if msg . startswith ( '<STR_LIT>' ) : <EOL> msg = msg [ len ( '<STR_LIT>' ) : ] <EOL> yield job_request_task ( msg ) <EOL> conn . close ( ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> msg = msg [ len ( '<STR_LIT>' ) : ] <EOL> yield add_computation_task ( msg ) <EOL> conn . close ( ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> msg = msg [ len ( '<STR_LIT>' ) : ] <EOL> yield xfer_file_task ( msg ) <EOL> conn . close ( ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> msg = msg [ len ( '<STR_LIT>' ) : ] <EOL> yield setup_computation ( msg ) <EOL> conn . close ( ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> msg = msg [ len ( '<STR_LIT>' ) : ] <EOL> try : <EOL> info = unserialize ( msg ) <EOL> compute_id = info [ '<STR_LIT>' ] <EOL> auth = info [ '<STR_LIT>' ] <EOL> terminate_pending = info . get ( '<STR_LIT>' , False ) <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' , traceback . format_exc ( ) ) <EOL> else : <EOL> compute = self . computations . get ( compute_id , None ) <EOL> if compute is None or compute . auth != auth : <EOL> _dispy_logger . warning ( '<STR_LIT>' , compute_id ) <EOL> else : <EOL> compute . zombie = True <EOL> if terminate_pending : <EOL> self . thread_lock . acquire ( ) <EOL> job_infos = [ job_info for job_info in self . job_infos . values ( ) <EOL> if job_info . compute_id == compute_id ] <EOL> self . thread_lock . release ( ) <EOL> for job_info in job_infos : <EOL> yield terminate_job_task ( compute , job_info ) <EOL> self . cleanup_computation ( compute ) <EOL> yield conn . send_msg ( '<STR_LIT>' ) <EOL> conn . close ( ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> msg = msg [ len ( '<STR_LIT>' ) : ] <EOL> try : <EOL> _job = unserialize ( msg ) <EOL> compute = self . computations [ _job . compute_id ] <EOL> self . thread_lock . acquire ( ) <EOL> job_info = self . job_infos . get ( _job . uid , None ) <EOL> self . thread_lock . release ( ) <EOL> assert job_info is not None <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> addr [ <NUM_LIT:0> ] , compute . scheduler_ip_addr ) <EOL> else : <EOL> yield terminate_job_task ( compute , job_info ) <EOL> conn . close ( ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> msg = msg [ len ( '<STR_LIT>' ) : ] <EOL> try : <EOL> info = unserialize ( msg ) <EOL> compute_id = info [ '<STR_LIT>' ] <EOL> auth = info [ '<STR_LIT>' ] <EOL> except : <EOL> reply = <NUM_LIT:0> <EOL> else : <EOL> compute = self . computations . get ( compute_id , None ) <EOL> if compute is None or compute . auth != auth : <EOL> try : <EOL> fd = open ( os . path . join ( self . dest_path_prefix , <EOL> '<STR_LIT>' % ( compute_id , auth ) ) , '<STR_LIT:rb>' ) <EOL> compute = pickle . load ( fd ) <EOL> fd . close ( ) <EOL> except : <EOL> pass <EOL> if compute is None : <EOL> reply = <NUM_LIT:0> <EOL> else : <EOL> reply = compute . pending_results + compute . pending_jobs <EOL> yield conn . send_msg ( serialize ( reply ) ) <EOL> conn . close ( ) <EOL> if reply > <NUM_LIT:0> : <EOL> yield self . resend_job_results ( compute , coro = coro ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> try : <EOL> info = unserialize ( msg [ len ( '<STR_LIT>' ) : ] ) <EOL> if ( info [ '<STR_LIT:version>' ] == _dispy_version and <EOL> not self . scheduler [ '<STR_LIT>' ] and not self . job_infos ) : <EOL> Coro ( self . send_pong_msg , info , addr ) <EOL> except : <EOL> _dispy_logger . debug ( traceback . format_exc ( ) ) <EOL> conn . close ( ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> msg = msg [ len ( '<STR_LIT>' ) : ] <EOL> reply = { '<STR_LIT>' : [ ] , '<STR_LIT>' : <NUM_LIT:0> } <EOL> try : <EOL> info = unserialize ( msg ) <EOL> compute_id = info [ '<STR_LIT>' ] <EOL> auth = info [ '<STR_LIT>' ] <EOL> except : <EOL> pass <EOL> else : <EOL> compute = self . computations . get ( compute_id , None ) <EOL> if compute is None or compute . auth != auth : <EOL> fd = open ( os . path . join ( self . dest_path_prefix , <EOL> '<STR_LIT>' % ( compute_id , auth ) ) , '<STR_LIT:rb>' ) <EOL> compute = pickle . load ( fd ) <EOL> fd . close ( ) <EOL> if compute is not None : <EOL> done = [ ] <EOL> if compute . pending_results : <EOL> for result_file in glob . glob ( os . path . join ( compute . dest_path , <EOL> '<STR_LIT>' ) ) : <EOL> result_file = os . path . basename ( result_file ) <EOL> try : <EOL> uid = int ( result_file [ len ( '<STR_LIT>' ) : ] ) <EOL> except : <EOL> pass <EOL> else : <EOL> done . append ( uid ) <EOL> if len ( done ) > <NUM_LIT:50> : <EOL> break <EOL> reply [ '<STR_LIT>' ] = done <EOL> reply [ '<STR_LIT>' ] = compute . pending_jobs <EOL> yield conn . send_msg ( serialize ( reply ) ) <EOL> conn . close ( ) <EOL> elif msg . startswith ( '<STR_LIT>' ) : <EOL> msg = msg [ len ( '<STR_LIT>' ) : ] <EOL> yield retrieve_job_task ( msg ) <EOL> conn . close ( ) <EOL> else : <EOL> _dispy_logger . warning ( '<STR_LIT>' , <EOL> msg [ : min ( <NUM_LIT:10> , len ( msg ) ) ] , addr [ <NUM_LIT:0> ] ) <EOL> resp = ( '<STR_LIT>' % ( msg [ : min ( <NUM_LIT:10> , len ( msg ) ) ] ) ) . encode ( ) <EOL> try : <EOL> yield conn . send_msg ( resp ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , str ( addr ) ) <EOL> conn . close ( ) <EOL> def resend_job_results ( self , compute , coro = None ) : <EOL> if not os . path . isdir ( compute . dest_path ) : <EOL> raise StopIteration <EOL> result_files = [ f for f in os . listdir ( compute . dest_path ) <EOL> if f . startswith ( '<STR_LIT>' ) ] <EOL> result_files = result_files [ : min ( len ( result_files ) , <NUM_LIT:64> ) ] <EOL> for result_file in result_files : <EOL> result_file = os . path . join ( compute . dest_path , result_file ) <EOL> try : <EOL> fd = open ( result_file , '<STR_LIT:rb>' ) <EOL> job_result = pickle . load ( fd ) <EOL> fd . close ( ) <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' , result_file ) <EOL> continue <EOL> job_info = _DispyJobInfo ( job_result , ( compute . scheduler_ip_addr , <EOL> compute . job_result_port ) , compute , [ ] ) <EOL> status = yield self . _send_job_reply ( job_info , resending = True ) <EOL> if status : <EOL> break <EOL> def timer_task ( self , coro = None ) : <EOL> coro . set_daemon ( ) <EOL> last_pulse_time = last_zombie_time = time . time ( ) <EOL> while True : <EOL> reset = yield coro . suspend ( self . pulse_interval ) <EOL> if reset : <EOL> continue <EOL> now = time . time ( ) <EOL> if self . pulse_interval and ( now - last_pulse_time ) >= self . pulse_interval : <EOL> if self . scheduler [ '<STR_LIT>' ] : <EOL> last_pulse_time = now <EOL> sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_DGRAM ) ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> info = { '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT:port>' : self . port , <EOL> '<STR_LIT>' : self . num_cpus - self . avail_cpus , <EOL> '<STR_LIT>' : self . scheduler [ '<STR_LIT>' ] } <EOL> if psutil : <EOL> info [ '<STR_LIT>' ] = DispyNodeAvailInfo ( <EOL> <NUM_LIT> - psutil . cpu_percent ( ) , psutil . virtual_memory ( ) . available , <EOL> psutil . disk_usage ( self . dest_path_prefix ) . free , <EOL> <NUM_LIT> - psutil . swap_memory ( ) . percent ) <EOL> else : <EOL> info [ '<STR_LIT>' ] = None <EOL> yield sock . sendto ( '<STR_LIT>' + serialize ( info ) , <EOL> ( self . scheduler [ '<STR_LIT>' ] , self . scheduler [ '<STR_LIT:port>' ] ) ) <EOL> sock . close ( ) <EOL> resend = [ compute for compute in self . computations . itervalues ( ) <EOL> if compute . pending_results and not compute . zombie ] <EOL> for compute in resend : <EOL> Coro ( self . resend_job_results , compute ) <EOL> if self . zombie_interval and ( now - last_zombie_time ) >= self . zombie_interval : <EOL> last_zombie_time = now <EOL> for compute in self . computations . itervalues ( ) : <EOL> if ( now - compute . last_pulse ) > self . zombie_interval : <EOL> _dispy_logger . warning ( '<STR_LIT>' , compute . name ) <EOL> compute . zombie = True <EOL> zombies = [ compute for compute in self . computations . itervalues ( ) <EOL> if compute . zombie and compute . pending_jobs == <NUM_LIT:0> ] <EOL> for compute in zombies : <EOL> _dispy_logger . warning ( '<STR_LIT>' , compute . name ) <EOL> self . cleanup_computation ( compute ) <EOL> for compute in zombies : <EOL> sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) , <EOL> keyfile = self . keyfile , certfile = self . certfile ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> _dispy_logger . debug ( '<STR_LIT>' , compute . scheduler_ip_addr ) <EOL> info = { '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT:port>' : self . port , '<STR_LIT>' : self . sign } <EOL> try : <EOL> yield sock . connect ( ( compute . scheduler_ip_addr , compute . scheduler_port ) ) <EOL> yield sock . send_msg ( '<STR_LIT>' . encode ( ) + serialize ( info ) ) <EOL> except : <EOL> pass <EOL> finally : <EOL> sock . close ( ) <EOL> if ( not self . scheduler [ '<STR_LIT>' ] and not self . job_infos and self . avail_cpus > <NUM_LIT:0> ) : <EOL> self . pulse_interval = None <EOL> yield self . broadcast_ping_msg ( coro = coro ) <EOL> def service_available ( self ) : <EOL> if self . serve == <NUM_LIT:0> : <EOL> return False <EOL> if not self . service_start or not self . service_end : <EOL> return True <EOL> now = time . localtime ( ) <EOL> if self . service_stop : <EOL> end = self . service_stop <EOL> else : <EOL> end = self . service_end <EOL> if self . service_start < end : <EOL> if self . service_start <= ( now . tm_hour , now . tm_min ) < end : <EOL> return True <EOL> else : <EOL> if ( now . tm_hour , now . tm_min ) >= self . service_start or ( now . tm_hour , now . tm_min ) < end : <EOL> return True <EOL> return False <EOL> def service_schedule ( self , coro = None ) : <EOL> coro . set_daemon ( ) <EOL> while True : <EOL> yield coro . sleep ( <NUM_LIT> ) <EOL> if self . service_available ( ) : <EOL> yield self . broadcast_ping_msg ( coro = coro ) <EOL> else : <EOL> if self . scheduler [ '<STR_LIT>' ] : <EOL> now = time . localtime ( ) <EOL> if self . service_end and ( now . tm_hour , now . tm_min ) > self . service_end : <EOL> _dispy_logger . debug ( '<STR_LIT>' ) <EOL> self . shutdown ( quit = False ) <EOL> else : <EOL> _dispy_logger . debug ( '<STR_LIT>' ) <EOL> sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) , <EOL> keyfile = self . keyfile , certfile = self . certfile ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> try : <EOL> yield sock . connect ( ( self . scheduler [ '<STR_LIT>' ] , self . scheduler [ '<STR_LIT:port>' ] ) ) <EOL> info = { '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT>' : self . sign , '<STR_LIT>' : <NUM_LIT:0> } <EOL> yield sock . send_msg ( '<STR_LIT>' . encode ( ) + serialize ( info ) ) <EOL> except : <EOL> pass <EOL> finally : <EOL> sock . close ( ) <EOL> def __job_program ( self , _job , job_info ) : <EOL> compute = self . computations [ _job . compute_id ] <EOL> if compute . name . endswith ( '<STR_LIT>' ) : <EOL> program = [ sys . executable , compute . name ] <EOL> else : <EOL> program = [ compute . name ] <EOL> args = unserialize ( _job . args ) <EOL> program . extend ( args ) <EOL> reply = job_info . job_reply <EOL> try : <EOL> os . chdir ( compute . dest_path ) <EOL> env = { } <EOL> env . update ( os . environ ) <EOL> env [ '<STR_LIT>' ] = compute . dest_path + os . pathsep + env [ '<STR_LIT>' ] <EOL> job_info . proc = subprocess . Popen ( program , stdout = subprocess . PIPE , <EOL> stderr = subprocess . PIPE , env = env ) <EOL> assert isinstance ( job_info . proc , subprocess . Popen ) <EOL> reply . stdout , reply . stderr = job_info . proc . communicate ( ) <EOL> reply . result = job_info . proc . returncode <EOL> reply . status = DispyJob . Finished <EOL> except : <EOL> reply . exception = traceback . format_exc ( ) <EOL> reply . status = DispyJob . Terminated <EOL> reply . end_time = time . time ( ) <EOL> job_info . proc = None <EOL> self . reply_Q . put ( reply ) <EOL> def __reply_Q ( self ) : <EOL> while True : <EOL> job_reply = self . reply_Q . get ( ) <EOL> if job_reply is None : <EOL> break <EOL> self . thread_lock . acquire ( ) <EOL> job_info = self . job_infos . get ( job_reply . uid , None ) <EOL> if job_info is not None : <EOL> job_info . job_reply = job_reply <EOL> self . thread_lock . release ( ) <EOL> if job_info is not None : <EOL> self . num_jobs += <NUM_LIT:1> <EOL> self . cpu_time += ( job_reply . end_time - job_reply . start_time ) <EOL> if job_info . proc is not None : <EOL> if isinstance ( job_info . proc , multiprocessing . Process ) : <EOL> job_info . proc . join ( <NUM_LIT:2> ) <EOL> else : <EOL> job_info . proc . wait ( ) <EOL> Coro ( self . _send_job_reply , job_info , resending = False ) <EOL> compute = self . computations . get ( job_info . compute_id , None ) <EOL> if not compute : <EOL> continue <EOL> for xf in job_info . xfer_files : <EOL> path = os . path . join ( compute . dest_path , os . path . basename ( xf . name ) ) <EOL> try : <EOL> compute . file_uses [ path ] -= <NUM_LIT:1> <EOL> if compute . file_uses [ path ] == <NUM_LIT:0> : <EOL> compute . file_uses . pop ( path ) <EOL> os . remove ( path ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , path ) <EOL> continue <EOL> def _send_job_reply ( self , job_info , resending = False , coro = None ) : <EOL> """<STR_LIT>""" <EOL> job_reply = job_info . job_reply <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> job_reply . uid , job_reply . status , str ( job_info . reply_addr ) ) <EOL> compute = self . computations . get ( job_info . compute_id , None ) <EOL> if not resending : <EOL> self . thread_lock . acquire ( ) <EOL> assert self . job_infos . pop ( job_reply . uid , None ) is not None <EOL> self . thread_lock . release ( ) <EOL> self . avail_cpus += <NUM_LIT:1> <EOL> assert self . avail_cpus <= self . num_cpus <EOL> if compute : <EOL> compute . pending_jobs -= <NUM_LIT:1> <EOL> sock = socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) <EOL> sock = AsyncSocket ( sock , keyfile = self . keyfile , certfile = self . certfile ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> try : <EOL> yield sock . connect ( job_info . reply_addr ) <EOL> yield sock . send_msg ( '<STR_LIT>' + serialize ( job_reply ) ) <EOL> ack = yield sock . recv_msg ( ) <EOL> assert ack == '<STR_LIT>' <EOL> except : <EOL> status = - <NUM_LIT:1> <EOL> if not resending : <EOL> f = os . path . join ( job_info . compute_dest_path , '<STR_LIT>' % job_reply . uid ) <EOL> _dispy_logger . error ( '<STR_LIT>' , <EOL> job_reply . uid , str ( job_info . reply_addr ) , f ) <EOL> try : <EOL> fd = open ( f , '<STR_LIT:wb>' ) <EOL> pickle . dump ( job_reply , fd ) <EOL> fd . close ( ) <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' , job_reply . uid ) <EOL> else : <EOL> if compute is not None : <EOL> compute . pending_results += <NUM_LIT:1> <EOL> else : <EOL> status = <NUM_LIT:0> <EOL> if compute : <EOL> compute . last_pulse = time . time ( ) <EOL> if resending : <EOL> compute . pending_results -= <NUM_LIT:1> <EOL> elif compute . pending_results : <EOL> Coro ( self . resend_job_results , compute ) <EOL> if resending : <EOL> f = os . path . join ( job_info . compute_dest_path , <EOL> '<STR_LIT>' % job_reply . uid ) <EOL> if os . path . isfile ( f ) : <EOL> try : <EOL> os . remove ( f ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , f ) <EOL> if compute is None : <EOL> fd = open ( os . path . join ( self . dest_path_prefix , <EOL> '<STR_LIT>' % ( job_info . compute_id , job_info . compute_auth ) ) , <EOL> '<STR_LIT:rb>' ) <EOL> compute = pickle . load ( fd ) <EOL> fd . close ( ) <EOL> if compute : <EOL> compute . pending_results -= <NUM_LIT:1> <EOL> finally : <EOL> sock . close ( ) <EOL> if compute and compute . pending_jobs == <NUM_LIT:0> and compute . zombie : <EOL> self . cleanup_computation ( compute ) <EOL> raise StopIteration ( status ) <EOL> def cleanup_computation ( self , compute ) : <EOL> if not compute . zombie or compute . pending_jobs > <NUM_LIT:0> : <EOL> return <EOL> if compute . pending_jobs != <NUM_LIT:0> : <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> compute . name , compute . id , compute . pending_jobs ) <EOL> if self . computations . pop ( compute . id , None ) is None : <EOL> _dispy_logger . warning ( '<STR_LIT>' , compute . id ) <EOL> return <EOL> self . num_computations += <NUM_LIT:1> <EOL> file_uses , compute . file_uses = compute . file_uses , { } <EOL> globalvars , compute . globals = compute . globals , { } <EOL> pkl_path = os . path . join ( self . dest_path_prefix , '<STR_LIT>' % ( compute . id , compute . auth ) ) <EOL> if compute . pending_results == <NUM_LIT:0> : <EOL> try : <EOL> os . remove ( pkl_path ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , pkl_path ) <EOL> else : <EOL> fd = open ( pkl_path , '<STR_LIT:wb>' ) <EOL> pickle . dump ( compute , fd ) <EOL> fd . close ( ) <EOL> self . scheduler [ '<STR_LIT>' ] . discard ( compute . auth ) <EOL> if ( ( not self . computations ) and ( not self . scheduler [ '<STR_LIT>' ] ) and <EOL> compute . scheduler_ip_addr == self . scheduler [ '<STR_LIT>' ] and <EOL> compute . scheduler_port == self . scheduler [ '<STR_LIT:port>' ] ) : <EOL> self . scheduler [ '<STR_LIT>' ] = None <EOL> self . pulse_interval = None <EOL> self . timer_coro . resume ( None ) <EOL> if self . serve > <NUM_LIT:0> : <EOL> self . serve -= <NUM_LIT:1> <EOL> Coro ( self . broadcast_ping_msg ) <EOL> if compute . cleanup is False : <EOL> if self . serve == <NUM_LIT:0> : <EOL> self . shutdown ( quit = True ) <EOL> return <EOL> os . chdir ( self . dest_path_prefix ) <EOL> if isinstance ( compute . cleanup , _Function ) : <EOL> try : <EOL> localvars = { '<STR_LIT>' : compute . cleanup . args , <EOL> '<STR_LIT>' : compute . cleanup . kwargs } <EOL> if os . name == '<STR_LIT>' : <EOL> globalvars = globals ( ) <EOL> exec ( marshal . loads ( compute . code ) ) in globalvars , localvars <EOL> exec ( '<STR_LIT>' % <EOL> compute . cleanup . name ) in globalvars , localvars <EOL> except : <EOL> _dispy_logger . debug ( '<STR_LIT>' , compute . cleanup . name ) <EOL> _dispy_logger . debug ( traceback . format_exc ( ) ) <EOL> if os . name == '<STR_LIT>' : <EOL> for var in globals ( ) . keys ( ) : <EOL> if var not in self . __init_globals : <EOL> _dispy_logger . debug ( '<STR_LIT>' , <EOL> var , compute . name , compute . scheduler_ip_addr ) <EOL> globals ( ) . pop ( var , None ) <EOL> for var , value in self . __init_globals . iteritems ( ) : <EOL> if var in ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) : <EOL> continue <EOL> if value != globals ( ) . get ( var , None ) : <EOL> _dispy_logger . warning ( '<STR_LIT>' , <EOL> var , compute . name , compute . scheduler_ip_addr ) <EOL> globals ( ) [ var ] = value <EOL> for module in sys . modules . keys ( ) : <EOL> if module not in compute . ante_modules : <EOL> sys . modules . pop ( module , None ) <EOL> sys . modules . update ( self . __init_modules ) <EOL> for path in os . listdir ( compute . dest_path ) : <EOL> path = os . path . join ( compute . dest_path , path ) <EOL> if file_uses . get ( path , <NUM_LIT:1> ) == <NUM_LIT:1> : <EOL> try : <EOL> if os . path . isfile ( path ) or os . path . islink ( path ) : <EOL> os . remove ( path ) <EOL> elif os . path . isdir ( path ) : <EOL> shutil . rmtree ( path , ignore_errors = True ) <EOL> else : <EOL> os . remove ( path ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , path ) <EOL> if os . path . isdir ( compute . dest_path ) and compute . dest_path . startswith ( self . dest_path_prefix ) and len ( os . listdir ( compute . dest_path ) ) == <NUM_LIT:0> : <EOL> try : <EOL> os . rmdir ( compute . dest_path ) <EOL> except : <EOL> _dispy_logger . warning ( '<STR_LIT>' , compute . dest_path ) <EOL> else : <EOL> _dispy_logger . debug ( '<STR_LIT>' , compute . dest_path ) <EOL> if self . serve == <NUM_LIT:0> : <EOL> self . shutdown ( quit = True ) <EOL> def shutdown ( self , quit = True ) : <EOL> def _shutdown ( self , quit , coro = None ) : <EOL> self . thread_lock . acquire ( ) <EOL> job_infos , self . job_infos = self . job_infos , { } <EOL> if quit and self . reply_Q : <EOL> self . reply_Q . put ( None ) <EOL> self . scheduler [ '<STR_LIT>' ] = None <EOL> self . scheduler [ '<STR_LIT>' ] = set ( ) <EOL> self . avail_cpus += len ( job_infos ) <EOL> if self . avail_cpus != self . num_cpus : <EOL> _dispy_logger . warning ( '<STR_LIT>' , self . avail_cpus , self . num_cpus ) <EOL> self . thread_lock . release ( ) <EOL> for uid , job_info in job_infos . iteritems ( ) : <EOL> job_info . proc . terminate ( ) <EOL> _dispy_logger . debug ( '<STR_LIT>' , uid ) <EOL> if isinstance ( job_info . proc , multiprocessing . Process ) : <EOL> job_info . proc . join ( <NUM_LIT:2> ) <EOL> else : <EOL> job_info . proc . wait ( ) <EOL> for cid , compute in self . computations . items ( ) : <EOL> sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) , <EOL> keyfile = self . keyfile , certfile = self . certfile ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> _dispy_logger . debug ( '<STR_LIT>' , compute . scheduler_ip_addr ) <EOL> info = { '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT:port>' : self . port , '<STR_LIT>' : self . sign } <EOL> try : <EOL> yield sock . connect ( ( compute . scheduler_ip_addr , compute . scheduler_port ) ) <EOL> yield sock . send_msg ( '<STR_LIT>' . encode ( ) + serialize ( info ) ) <EOL> except : <EOL> pass <EOL> sock . close ( ) <EOL> compute . pending_jobs = <NUM_LIT:0> <EOL> compute . zombie = True <EOL> self . cleanup_computation ( compute ) <EOL> if quit : <EOL> self . tcp_coro . terminate ( ) <EOL> self . sign = '<STR_LIT>' <EOL> if self . sign : <EOL> Coro ( _shutdown , self , quit ) <EOL> def read_stdin ( self , coro = None ) : <EOL> coro . set_daemon ( ) <EOL> thread_pool = asyncoro . AsyncThreadPool ( <NUM_LIT:1> ) <EOL> while True : <EOL> sys . stdout . write ( '<STR_LIT>' <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> sys . stdout . flush ( ) <EOL> try : <EOL> cmd = yield thread_pool . async_task ( raw_input ) <EOL> except : <EOL> continue <EOL> cmd = cmd . strip ( ) . lower ( ) <EOL> if cmd in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> break <EOL> elif cmd in ( '<STR_LIT>' , '<STR_LIT:start>' , '<STR_LIT>' ) : <EOL> if cmd == '<STR_LIT>' : <EOL> cpus = <NUM_LIT:0> <EOL> elif cmd == '<STR_LIT:start>' : <EOL> cpus = self . num_cpus <EOL> elif cmd == '<STR_LIT>' : <EOL> cpus = multiprocessing . cpu_count ( ) <EOL> sys . stdout . write ( '<STR_LIT>' % <EOL> ( cpus - <NUM_LIT:1> , cpus ) ) <EOL> sys . stdout . flush ( ) <EOL> try : <EOL> cpus = yield thread_pool . async_task ( raw_input ) <EOL> cpus = int ( cpus ) <EOL> if cpus >= <NUM_LIT:0> : <EOL> assert cpus <= multiprocessing . cpu_count ( ) <EOL> else : <EOL> cpus += multiprocessing . cpu_count ( ) <EOL> assert cpus >= <NUM_LIT:0> <EOL> except : <EOL> print ( '<STR_LIT>' ) <EOL> continue <EOL> self . num_cpus = cpus <EOL> self . avail_cpus = cpus - len ( self . job_infos ) <EOL> if self . scheduler [ '<STR_LIT>' ] : <EOL> sock = AsyncSocket ( socket . socket ( socket . AF_INET , socket . SOCK_STREAM ) , <EOL> keyfile = self . keyfile , certfile = self . certfile ) <EOL> sock . settimeout ( MsgTimeout ) <EOL> try : <EOL> yield sock . connect ( ( self . scheduler [ '<STR_LIT>' ] , self . scheduler [ '<STR_LIT:port>' ] ) ) <EOL> info = { '<STR_LIT>' : self . ext_ip_addr , '<STR_LIT>' : self . sign , '<STR_LIT>' : cpus } <EOL> yield sock . send_msg ( '<STR_LIT>' . encode ( ) + serialize ( info ) ) <EOL> except : <EOL> pass <EOL> finally : <EOL> sock . close ( ) <EOL> else : <EOL> if self . num_cpus > <NUM_LIT:0> : <EOL> Coro ( self . broadcast_ping_msg ) <EOL> else : <EOL> print ( '<STR_LIT>' % <EOL> ( self . avail_cpus + len ( self . job_infos ) , <EOL> '<STR_LIT>' % self . serivce_start if self . service_start else '<STR_LIT>' , <EOL> '<STR_LIT>' % self . service_end if self . service_end else '<STR_LIT>' , <EOL> '<STR_LIT>' % self . serve if self . serve > <NUM_LIT:0> else '<STR_LIT>' ) ) <EOL> print ( '<STR_LIT>' % <EOL> ( self . num_computations , self . num_jobs , self . cpu_time ) ) <EOL> print ( '<STR_LIT>' ) <EOL> for i , compute in enumerate ( self . computations . itervalues ( ) , start = <NUM_LIT:1> ) : <EOL> print ( '<STR_LIT>' % <EOL> ( i , compute . name , compute . scheduler_ip_addr , compute . pending_jobs ) ) <EOL> print ( '<STR_LIT>' ) <EOL> self . shutdown ( quit = True ) <EOL> if __name__ == '<STR_LIT:__main__>' : <EOL> import argparse <EOL> import re <EOL> _dispy_logger = asyncoro . Logger ( '<STR_LIT>' ) <EOL> _dispy_logger . info ( '<STR_LIT>' , _dispy_version ) <EOL> parser = argparse . ArgumentParser ( ) <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 = False , action = '<STR_LIT:store_true>' , <EOL> help = '<STR_LIT>' ) <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>' , action = '<STR_LIT:store_true>' , dest = '<STR_LIT>' , default = False , <EOL> help = '<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:name>' , type = str , 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>' , type = int , default = <NUM_LIT> , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = str ( MaxFileSize ) , type = str , <EOL> help = '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = <NUM_LIT> , type = float , <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> '<STR_LIT>' ) <EOL> parser . add_argument ( '<STR_LIT>' , dest = '<STR_LIT>' , default = None , <EOL> help = '<STR_LIT>' <EOL> '<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 = MsgTimeout , type = float , <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>' , 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> _dispy_config = vars ( parser . parse_args ( sys . argv [ <NUM_LIT:1> : ] ) ) <EOL> del parser <EOL> if _dispy_config [ '<STR_LIT>' ] and os . path . isfile ( _dispy_config [ '<STR_LIT>' ] ) : <EOL> import ConfigParser <EOL> cfgp = ConfigParser . ConfigParser ( ) <EOL> cfgp . read ( _dispy_config [ '<STR_LIT>' ] ) <EOL> _dispy_config , cfgp = dict ( cfgp . items ( '<STR_LIT>' ) ) , _dispy_config <EOL> _dispy_config . update ( cfgp ) <EOL> del cfgp <EOL> if _dispy_config [ '<STR_LIT>' ] : <EOL> import ConfigParser <EOL> cfgp = ConfigParser . ConfigParser ( _dispy_config ) <EOL> if _dispy_config [ '<STR_LIT>' ] : <EOL> cfgfd = open ( _dispy_config [ '<STR_LIT>' ] , '<STR_LIT:w>' ) <EOL> _dispy_config . pop ( '<STR_LIT>' ) <EOL> else : <EOL> cfgfd = sys . stdout <EOL> _dispy_config . pop ( '<STR_LIT>' ) <EOL> cfgp . write ( cfgfd ) <EOL> exit ( <NUM_LIT:0> ) <EOL> _dispy_config . pop ( '<STR_LIT>' , None ) <EOL> _dispy_config . pop ( '<STR_LIT>' , None ) <EOL> if _dispy_config [ '<STR_LIT>' ] : <EOL> _dispy_logger . setLevel ( logging . DEBUG ) <EOL> asyncoro . logger . setLevel ( logging . DEBUG ) <EOL> else : <EOL> _dispy_logger . setLevel ( logging . INFO ) <EOL> del _dispy_config [ '<STR_LIT>' ] <EOL> cpus = multiprocessing . cpu_count ( ) <EOL> if _dispy_config [ '<STR_LIT>' ] : <EOL> if _dispy_config [ '<STR_LIT>' ] > <NUM_LIT:0> : <EOL> if _dispy_config [ '<STR_LIT>' ] > cpus : <EOL> raise Exception ( '<STR_LIT>' % cpus ) <EOL> else : <EOL> if _dispy_config [ '<STR_LIT>' ] <= - cpus : <EOL> raise Exception ( '<STR_LIT>' % cpus ) <EOL> cpus += _dispy_config [ '<STR_LIT>' ] <EOL> _dispy_config [ '<STR_LIT>' ] = cpus <EOL> else : <EOL> _dispy_config [ '<STR_LIT>' ] = cpus <EOL> del cpus <EOL> if _dispy_config [ '<STR_LIT>' ] : <EOL> _dispy_config [ '<STR_LIT>' ] = float ( _dispy_config [ '<STR_LIT>' ] ) <EOL> if _dispy_config [ '<STR_LIT>' ] < <NUM_LIT:1> : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> MsgTimeout = _dispy_config [ '<STR_LIT>' ] <EOL> del _dispy_config [ '<STR_LIT>' ] <EOL> m = re . match ( r'<STR_LIT>' , _dispy_config [ '<STR_LIT>' ] ) <EOL> if m : <EOL> MaxFileSize = int ( m . group ( <NUM_LIT:1> ) ) <EOL> if m . group ( <NUM_LIT:2> ) : <EOL> m = m . group ( <NUM_LIT:2> ) . lower ( ) <EOL> if m == '<STR_LIT:k>' : <EOL> MaxFileSize *= <NUM_LIT> <EOL> elif m == '<STR_LIT:m>' : <EOL> MaxFileSize *= <NUM_LIT> ** <NUM_LIT:2> <EOL> elif m == '<STR_LIT:g>' : <EOL> MaxFileSize *= <NUM_LIT> ** <NUM_LIT:3> <EOL> elif m == '<STR_LIT:t>' : <EOL> MaxFileSize *= <NUM_LIT> ** <NUM_LIT:4> <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> else : <EOL> raise Exception ( '<STR_LIT>' ) <EOL> del m <EOL> del _dispy_config [ '<STR_LIT>' ] <EOL> if _dispy_config [ '<STR_LIT>' ] : <EOL> _dispy_config [ '<STR_LIT>' ] = time . strptime ( _dispy_config [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> if _dispy_config [ '<STR_LIT>' ] : <EOL> _dispy_config [ '<STR_LIT>' ] = time . strptime ( _dispy_config [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> if _dispy_config [ '<STR_LIT>' ] : <EOL> _dispy_config [ '<STR_LIT>' ] = time . strptime ( _dispy_config [ '<STR_LIT>' ] , '<STR_LIT>' ) <EOL> try : <EOL> if os . getpgrp ( ) != os . tcgetpgrp ( sys . stdin . fileno ( ) ) : <EOL> _dispy_config [ '<STR_LIT>' ] = True <EOL> except : <EOL> pass <EOL> if psutil : <EOL> psutil . cpu_percent ( <NUM_LIT:0.1> ) <EOL> else : <EOL> print ( '<STR_LIT>' ) <EOL> print ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> _dispy_node = None <EOL> _dispy_node = _DispyNode ( ** _dispy_config ) <EOL> del _dispy_config <EOL> _dispy_node . asyncoro . finish ( ) </s>
