#NEWLINE# def batch_futures( self , session , statement_generator ) : #NEWLINE# #INDENT# concurrency = 10 #NEWLINE# futures = Queue ( maxsize = futures ) #NEWLINE# number_of_timeouts = 0 #NEWLINE# for ( i , statement ) in enumerate ( statement_generator ) : #NEWLINE# #INDENT# if ( ( i > 0 ) and ( ( i % ( concurrency - 1 ) ) == 0 ) ) : #NEWLINE# #INDENT# while True : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# futures . get_nowait ( ) . result ( ) #NEWLINE# #UNINDENT# except ( OperationTimedOut , WriteTimeout ) : #NEWLINE# #INDENT# ( ex_type , ex , tb ) = sys . exc_info ( ) #NEWLINE# number_of_timeouts += 1 #NEWLINE# log . warn ( '{0}: {1} Backtrace: {2}' . format ( ex_type . __name__ , ex , traceback . extract_tb ( tb ) ) ) #NEWLINE# del tb #NEWLINE# time . sleep ( 1 ) #NEWLINE# #UNINDENT# except Empty : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# #UNINDENT# future = session . execute_async ( statement ) #NEWLINE# futures . put_nowait ( future ) #NEWLINE# #UNINDENT# while True : #NEWLINE# #INDENT# try : #NEWLINE# #INDENT# futures . get_nowait ( ) . result ( ) #NEWLINE# #UNINDENT# except ( OperationTimedOut , WriteTimeout ) : #NEWLINE# #INDENT# ( ex_type , ex , tb ) = sys . exc_info ( ) #NEWLINE# number_of_timeouts += 1 #NEWLINE# log . warn ( '{0}: {1} Backtrace: {2}' . format ( ex_type . __name__ , ex , traceback . extract_tb ( tb ) ) ) #NEWLINE# del tb #NEWLINE# time . sleep ( 1 ) #NEWLINE# #UNINDENT# except Empty : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# return number_of_timeouts