lib/bolt/fiber_executor.rb
# Visit each PlanFuture registered with the FiberExecutor and resume it.
# Fibers will yield themselves back, either if they kicked off a
# long-running process or if the current long-running process hasn't
# completed. If the Fiber finishes after being resumed, store the result in
# the PlanFuture and remove the PlanFuture from the FiberExecutor.
#
Visit each PlanFuture registered with the FiberExecutor and resume it.
Fibers will yield themselves back, either if they kicked off a
long-running process or if the current long-running process hasn't
completed. If the Fiber finishes after being resumed, store the result in
the PlanFuture and remove the PlanFuture from the FiberExecutor.
Visit each PlanFuture registered with the FiberExecutor and resume it . Fibers will yield themselves back either if they kicked off a long - running process or if the current long - running process hasn ' t completed . If the Fiber finishes after being resumed store the result in the PlanFuture and remove the PlanFuture from the FiberExecutor .
def round_robin
      active_futures.each do |future|
        @logger.trace("Checking future '#{future.name}'")
        if future.alive?
          @logger.trace("Resuming future '#{future.name}'")
          Puppet.override(global_scope: future.scope) { future.resume }
        end
        next if future.alive?
        @logger.trace("Cleaning up future '#{future.name}'")
        unless active_futures.map(&:id).include?(0) || future.state == "done"
          Bolt::Logger.warn('errored_futures', "Error in future '#{future.name}': #{future.value}")
        end
        finished_futures.push(active_futures.delete(future))
      end
      return unless active_futures.all? { |f| %i[returned_immediately unfinished].include?(f.value) }
      @logger.trace("Nothing can be resumed. Rechecking in 0.5 seconds.")
      sleep(0.5)
    end
def round_robin active_futures . each do | future | @logger . trace ( "Checking future '#{future.name}'" ) if future . alive? @logger . trace ( "Resuming future '#{future.name}'" ) Puppet . override ( global_scope : future . scope ) { future . resume } end next if future . alive? @logger . trace ( "Cleaning up future '#{future.name}'" ) unless active_futures . map ( & :id ) . include? ( 0 ) || future . state == "done" Bolt :: Logger . warn ( 'errored_futures' , "Error in future '#{future.name}': #{future.value}" ) end finished_futures . push ( active_futures . delete ( future ) ) end return unless active_futures . all? { | f | %i[ returned_immediately unfinished ] . include? ( f . value ) } @logger . trace ( "Nothing can be resumed. Rechecking in 0.5 seconds." ) sleep ( 0.5 ) end
Visit each PlanFuture registered with the FiberExecutor and resume it.
Visit each PlanFuture registered with the FiberExecutor and resume it .
# If the Fiber is still running and can be resumed, then resume it. # Override Puppet's global_scope to prevent ephemerals in other scopes # from being popped off in the wrong order due to race conditions. # This primarily happens when running executor functions from custom # Puppet language functions, but may happen elsewhere. # Once we've restarted the Fiber, check to see if it's finished again # and cleanup if it has. # If the future errored and the main plan has already exited, log the # error at warn level. # Remove the PlanFuture from the FiberExecutor. # If the Fiber immediately returned or if the Fiber is blocking on a # `wait` call, Bolt should pause for long enough that something can # execute before checking again. This mitigates CPU # thrashing.