#NEWLINE# def try_to_start_mpi( self , command , tasks , items ) : #NEWLINE# #INDENT# if self . p : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# self . p . kill ( ) #NEWLINE# #UNINDENT# except : #NEWLINE# #INDENT# pass #NEWLINE# #UNINDENT# #UNINDENT# hosts = ',' . join ( ( ( '%s:%d' % ( hostname , slots ) ) for ( hostname , slots ) in items ) ) #NEWLINE# logger . debug ( 'choosed hosts: %s' , hosts ) #NEWLINE# info = subprocess . check_output ( [ 'mpirun' , '--version' ] ) #NEWLINE# for line in info . splitlines ( ) : #NEWLINE# #INDENT# if ( ( 'Launchers available' in line ) and ( ' none ' in line ) ) : #NEWLINE# #INDENT# cmd = ( [ 'mpirun' , '-prepend-rank' , '-launcher' , 'none' , '-hosts' , hosts , '-np' , str ( tasks ) ] + command ) #NEWLINE# break #NEWLINE# #UNINDENT# #UNINDENT# else : #NEWLINE# #INDENT# cmd = ( [ 'mpirun' , '-prepend-rank' , '-launcher' , 'manual' , '-rmk' , 'user' , '-hosts' , hosts , '-np' , str ( tasks ) ] + command ) #NEWLINE# #UNINDENT# self . p = p = subprocess . Popen ( cmd , bufsize = 0 , stdout = subprocess . PIPE ) #NEWLINE# slaves = [ ] #NEWLINE# prefix = 'HYDRA_LAUNCH: ' #NEWLINE# while True : #NEWLINE# #INDENT# line = p . stdout . readline ( ) #NEWLINE# if ( not line ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# if line . startswith ( prefix ) : #NEWLINE# #INDENT# slaves . append ( line [ len ( prefix ) : ( - 1 ) ] . strip ( ) ) #NEWLINE# #UNINDENT# if ( line == 'HYDRA_LAUNCH_END\n' ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# if ( len ( slaves ) != len ( items ) ) : #NEWLINE# #INDENT# logger . error ( 'hosts: %s, slaves: %s' , items , slaves ) #NEWLINE# raise Exception ( 'slaves not match with hosts' ) #NEWLINE# #UNINDENT# def output( f ) : #NEWLINE# #INDENT# while True : #NEWLINE# #INDENT# line = f . readline ( ) #NEWLINE# if ( not line ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# sys . stdout . write ( line ) #NEWLINE# #UNINDENT# #UNINDENT# self . tout = t = threading . Thread ( target = output , args = [ p . stdout ] ) #NEWLINE# t . deamon = True #NEWLINE# t . start ( ) #NEWLINE# return dict ( zip ( ( hostname for ( hostname , slots ) in items ) , slaves ) )