#NEWLINE# @ safe #NEWLINE# def resourceOffers( self , driver , offers ) : #NEWLINE# #INDENT# tpn = self . options . task_per_node #NEWLINE# random . shuffle ( offers ) #NEWLINE# self . last_offer_time = time . time ( ) #NEWLINE# for offer in offers : #NEWLINE# #INDENT# attrs = self . getAttributes ( offer ) #NEWLINE# if ( self . options . group and ( attrs . get ( 'group' , 'None' ) not in self . options . group ) ) : #NEWLINE# #INDENT# driver . launchTasks ( offer . id , [ ] , REFUSE_FILTER ) #NEWLINE# continue #NEWLINE# #UNINDENT# ( cpus , mem ) = self . getResource ( offer ) #NEWLINE# logger . debug ( 'got resource offer %s: cpus:%s, mem:%s at %s' , offer . id . value , cpus , mem , offer . hostname ) #NEWLINE# sid = offer . slave_id . value #NEWLINE# tasks = [ ] #NEWLINE# while ( self . total_tasks and ( ( cpus + 0.0001 ) >= self . cpus ) and ( mem >= self . mem ) and ( ( tpn == 0 ) or ( ( tpn > 0 ) and ( len ( self . slaveTasks . get ( sid , set ( ) ) ) < tpn ) ) ) ) : #NEWLINE# #INDENT# logger . debug ( 'Accepting slot on slave %s (%s)' , offer . slave_id . value , offer . hostname ) #NEWLINE# t = self . total_tasks . pop ( ) #NEWLINE# task = self . create_task ( offer , t , cpus ) #NEWLINE# tasks . append ( task ) #NEWLINE# t . state = mesos_pb2 . TASK_STARTING #NEWLINE# t . state_time = time . time ( ) #NEWLINE# self . task_launched [ t . id ] = tpn #NEWLINE# self . slaveTasks . setdefault ( sid , set ( ) ) . add ( t . id ) #NEWLINE# cpus -= self . cpus #NEWLINE# mem -= self . mem #NEWLINE# if ( not self . total_tasks ) : #NEWLINE# #INDENT# break #NEWLINE# #UNINDENT# #UNINDENT# logger . debug ( 'dispatch %d tasks to slave %s' , len ( tasks ) , offer . hostname ) #NEWLINE# driver . launchTasks ( offer . id , tasks , REFUSE_FILTER )