Method: Bolt::Executor#queue_execute

Defined in:
lib/bolt/executor.rb

#queue_execute(targets) ⇒ Object

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.



129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
 
# File 'lib/bolt/executor.rb', line 129

def queue_execute(targets)
  if @warn_concurrency && targets.length > @concurrency
    @warn_concurrency = false
    msg = "The ulimit is low, which might cause file limit issues. Default concurrency has been set to "\
          "'#{@concurrency}' to mitigate those issues, which might 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."
    Bolt::Logger.warn("low_ulimit", msg)
  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
      # Pass this argument through to avoid retaining a reference to a
      # local variable that will change on the next iteration of the loop.
      @pool.post(batch_promises) do |result_promises|
        Thread.current[:name] ||= Thread.current.name
        results = yield transport, batch
        Array(results).each do |result|
          result_promises[result.target].set(result)
        end
      # NotImplementedError can be thrown if the transport is not implemented improperly
      rescue StandardError, NotImplementedError => e
        result_promises.each do |target, promise|
          # 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.
          @logger.warn(e)
          promise.set(Bolt::Result.from_exception(target, e))
        end
      ensure
        # 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.
        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