<s> from __future__ import print_function , division , absolute_import <EOL> import logging <EOL> import os <EOL> from subprocess import Popen , CalledProcessError <EOL> import time <EOL> from six . moves . queue import Queue <EOL> class AsyncProcess ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( <EOL> self , <EOL> args , <EOL> suppress_stderr = False , <EOL> redirect_stdout_file = None ) : <EOL> assert len ( args ) > <NUM_LIT:0> <EOL> self . cmd = args [ <NUM_LIT:0> ] <EOL> self . args = args <EOL> self . redirect_stdout_file = redirect_stdout_file <EOL> with open ( os . devnull , '<STR_LIT:w>' ) as devnull : <EOL> stdout = redirect_stdout_file if redirect_stdout_file else devnull <EOL> stderr = devnull if suppress_stderr else None <EOL> self . process = Popen ( args , stdout = stdout , stderr = stderr ) <EOL> def poll ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . process . poll ( ) <EOL> def wait ( self ) : <EOL> ret_code = self . process . wait ( ) <EOL> logging . info ( <EOL> "<STR_LIT>" , <EOL> self . cmd , <EOL> ret_code ) <EOL> if ret_code : <EOL> raise CalledProcessError ( ret_code , self . cmd ) <EOL> return ret_code <EOL> def run_command ( args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> assert len ( args ) > <NUM_LIT:0> <EOL> cmd = args [ <NUM_LIT:0> ] <EOL> start_time = time . time ( ) <EOL> process = AsyncProcess ( args , ** kwargs ) <EOL> process . wait ( ) <EOL> elapsed_time = time . time ( ) - start_time <EOL> logging . info ( "<STR_LIT>" , cmd , elapsed_time ) <EOL> def run_multiple_commands_redirect_stdout ( <EOL> multiple_args_dict , <EOL> print_commands = True , <EOL> process_limit = <NUM_LIT:0> , <EOL> polling_freq = <NUM_LIT:1> , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> assert len ( multiple_args_dict ) > <NUM_LIT:0> <EOL> assert all ( len ( args ) > <NUM_LIT:0> for args in multiple_args_dict . values ( ) ) <EOL> assert all ( hasattr ( f , '<STR_LIT:name>' ) for f in multiple_args_dict . keys ( ) ) <EOL> start_time = time . time ( ) <EOL> processes = Queue ( maxsize = process_limit ) <EOL> def add_to_queue ( process ) : <EOL> if print_commands : <EOL> print ( "<STR_LIT:U+0020>" . join ( process . args ) , "<STR_LIT:>>" , <EOL> process . redirect_stdout_file . name ) <EOL> processes . put ( process ) <EOL> for f , args in multiple_args_dict . items ( ) : <EOL> p = AsyncProcess ( <EOL> args , <EOL> redirect_stdout_file = f , <EOL> ** kwargs ) <EOL> if not processes . full ( ) : <EOL> add_to_queue ( p ) <EOL> else : <EOL> while processes . full ( ) : <EOL> to_remove = [ ] <EOL> for possibly_done in processes . queue : <EOL> if possibly_done . poll ( ) is not None : <EOL> possibly_done . wait ( ) <EOL> to_remove . append ( possibly_done ) <EOL> if to_remove : <EOL> for process_to_remove in to_remove : <EOL> processes . queue . remove ( process_to_remove ) <EOL> break <EOL> time . sleep ( polling_freq ) <EOL> add_to_queue ( p ) <EOL> while not processes . empty ( ) : <EOL> processes . get ( ) . wait ( ) <EOL> elapsed_time = time . time ( ) - start_time <EOL> logging . info ( <EOL> "<STR_LIT>" , <EOL> len ( multiple_args_dict ) , <EOL> elapsed_time ) </s>
