06e62bd1f00e130d76ada31b932062c5531cd717,src/python/pants/engine/exp/engine.py,Executor,_submit_all_satisfied,#Executor#,154

Before Change

      for index, (product_type, plan) in enumerate(self._waiting):
        if all(promise in self._products_by_promise for promise in plan.promises):
          self._waiting.pop(index)
          func, args, kwargs = plan.bind({promise: self._products_by_promise[promise]
                                          for promise in plan.promises})
          execute_plan = functools.partial(_execute_plan, func, product_type, plan.subjects)
          self._pool.apply_async(execute_plan, args=args, kwds=kwargs, callback=self._results.put)

    def _gather_one(self):

After Change

      for index, (promise, plan) in enumerate(self._waiting):
        if all(pr in self._products_by_promise for pr in plan.promises):
          self._waiting.pop(index)
          inputs = self.collect_inputs(self._products_by_promise, promise, plan)
          if isinstance(inputs, FailedToProduce):
            // Short circuit plan execution since we don"t have all the inputs it needs.
            self._results.put((promise, plan.subjects, inputs))
          else:
            func, args, kwargs = plan.bind(inputs)

            // A no-arg callable that, when executed, produces the promised product.
            executable = functools.partial(func, *args, **kwargs)

            // A wrapper of executable that handles failing slow as needed.
            maybe_fail_slow_executor = functools.partial(maybe_fail_slow,
                                                         executable,
                                                         promise,
                                                         plan,
                                                         self._fail_slow)

            // A picklable execution that returns the promise and subjects in addition to the
            // product produced by executable.  We need this triple to feed the consume side of the
            // _results queue.
            execute_plan = functools.partial(_execute_plan,
                                             maybe_fail_slow_executor,
                                             promise,
                                             plan.subjects)

            self._pool.apply_async(execute_plan, callback=self._results.put)

    def _gather_one(self):
      promise, subjects, product = self._results.get()
      for subject in subjects:
        self._products_by_promise[promise.rebind(subject)] = product

Instances