ef3f8d221a5afefb01d655448ce7e3f537399810,src/python/pants/engine/exp/engine.py,LocalMultiprocessEngine,reduce,#LocalMultiprocessEngine#Any#Any#,446

Before Change

    // we limit pending submissions to the number of pool workers.  This gives us a bounded number
    // of items to check for submission readiness, ie: we"ll never need to scan more than the pool
    // size, commonly the number of cores, no matter the size and shape of the execution graph.
    for promise, plan in execution_graph.walk():
      pending_submission[promise] = plan
      submit_satisfied_pending()
      // Saturate the pool before we block on a result for maximum processing parallelism.

After Change

    // 1. Whenever we don"t have enough work to saturate the pool, request more.
    // 2. Whenever the pool is not saturated, submit currently pending work.
    for step_batch in self._scheduler.schedule(build_request):
      if not step_batch:
        // A batch should only be empty if all dependency work is currently blocked/running.
        if not in_flight and not pending_submission:
          raise Exception("Scheduler provided an empty batch while no work is in progress!")
      else:
        // Submit and wait for work for as long as we"re able to keep the pool saturated.
        pending_submission.update(step_batch)
        while submit_until(self._pool_size) > 0:
          await_one()
      // Await at least one entry per scheduling loop.
      submit_until(0)
      if in_flight:
        await_one()

Instances