lib/bolt/executor.rb
# Starts executing the given block on a list of nodes in parallel, one thread per "batch".
#
# This is the main driver of execution on a list of targets. It first
# groups targets by transport, then divides each group into batches as
# defined by the transport. Yields each batch, along with the corresponding
# transport, to the block in turn and returns an array of result promises.
Starts executing the given block on a list of nodes in parallel, one thread per "batch".
This is the main driver of execution on a list of targets. It first
groups targets by transport, then divides each group into batches as
defined by the transport. Yields each batch, along with the corresponding
transport, to the block in turn and returns an array of result promises.
Starts executing the given block on a list of nodes in parallel one thread per " batch " . This is the main driver of execution on a list of targets . It first groups targets by transport then divides each group into batches as defined by the transport . Yields each batch along with the corresponding transport to the block in turn and returns an array of result promises .
def queue_execute(targets)
      if @warn_concurrency && targets.length > @concurrency
        @warn_concurrency = false
        @logger.warn("The ulimit is low, which may cause file limit issues. Default concurrency has been set to "\
                     "'#{@concurrency}' to mitigate those issues, which may cause Bolt to run slow. "\
                     "Disable this warning by configuring ulimit using 'ulimit -n <limit>' in your shell "\
                     "configuration, or by configuring Bolt's concurrency. "\
                     "See https://puppet.com/docs/bolt/latest/bolt_known_issues.html for details.")
      end
      targets.group_by(&:transport).flat_map do |protocol, protocol_targets|
        transport = transport(protocol)
        report_transport(transport, protocol_targets.count)
        transport.batches(protocol_targets).flat_map do |batch|
          batch_promises = Array(batch).each_with_object({}) do |target, h|
            h[target] = Concurrent::Promise.new(executor: :immediate)
          end
          @pool.post(batch_promises) do |result_promises|
            results = yield transport, batch
            Array(results).each do |result|
              result_promises[result.target].set(result)
            end
          rescue StandardError, NotImplementedError => e
            result_promises.each do |target, promise|
              @logger.warn(e)
              promise.set(Bolt::Result.from_exception(target, e))
            end
          ensure
            result_promises.each do |target, promise|
              next if promise.fulfilled?
              error = $ERROR_INFO || Bolt::Error.new("No result was returned for #{target.uri}",
                                                     "puppetlabs.bolt/missing-result-error")
              promise.set(Bolt::Result.from_exception(target, error))
            end
          end
          batch_promises.values
        end
      end
    end
def queue_execute ( targets ) if @warn_concurrency && targets . length > @concurrency @warn_concurrency = false @logger . warn ( "The ulimit is low, which may cause file limit issues. Default concurrency has been set to " "'#{@concurrency}' to mitigate those issues, which may cause Bolt to run slow. " "Disable this warning by configuring ulimit using 'ulimit -n <limit>' in your shell " "configuration, or by configuring Bolt's concurrency. " "See https://puppet.com/docs/bolt/latest/bolt_known_issues.html for details." ) end targets . group_by ( & :transport ) . flat_map do | protocol , protocol_targets | transport = transport ( protocol ) report_transport ( transport , protocol_targets . count ) transport . batches ( protocol_targets ) . flat_map do | batch | batch_promises = Array ( batch ) . each_with_object ( { } ) do | target , h | h [ target ] = Concurrent :: Promise . new ( executor : :immediate ) end @pool . post ( batch_promises ) do | result_promises | results = yield transport , batch Array ( results ) . each do | result | result_promises [ result . target ] . set ( result ) end rescue StandardError , NotImplementedError => e result_promises . each do | target , promise | @logger . warn ( e ) promise . set ( Bolt :: Result . from_exception ( target , e ) ) end ensure result_promises . each do | target , promise | next if promise . fulfilled? error = $ERROR_INFO || Bolt :: Error . new ( "No result was returned for #{target.uri}" , "puppetlabs.bolt/missing-result-error" ) promise . set ( Bolt :: Result . from_exception ( target , error ) ) end end batch_promises . values end end end
Starts executing the given block on a list of nodes in parallel, one thread per "batch".
Starts executing the given block on a list of nodes in parallel one thread per " batch " .
# Pass this argument through to avoid retaining a reference to a # local variable that will change on the next iteration of the loop. # NotImplementedError can be thrown if the transport is not implemented improperly # If an exception happens while running, the result won't be logged # by the CLI. Log a warning, as this is probably a problem with the transport. # If batch_* commands are used from the Base transport, then exceptions # normally shouldn't reach here. # Make absolutely sure every promise gets a result to avoid a # deadlock. Use whatever exception is causing this block to # execute, or generate one if we somehow got here without an # exception and some promise is still missing a result.
targets