Block until all jobs in the ThreadPool are finished. Beware that this can
    make the program run into a deadlock if another thread adds new jobs to the
    pool!

    # Raises
    Timeout: If the timeout is exceeded.
181272
def wait(self, timeout=None):
    

    if not self.__running:
      raise RuntimeError("ThreadPool ain't running")
    self.__queue.wait(timeout)
1049462
async def block(self) -> None:
        """"""
        await asyncio.gather(
            *itertools.chain(
                (sp.wait_done() for sp in self._running_set),
                (sp.wait_done() for sp in self._pending_set)))
631697
def shutdown(force=false)
      status(:closed)
      args = []
      args.push(:force) if force
      @logger.info("Pool is now shutting down #{force ? 'using force' : ''}")
      @queue.clear if force
      @queue.wait_empty
      while(t = @threads.pop) do
        t.stop(*args)
      end
      unless(force)
        flush
        @threads.each{|x|x.join}
      end
      nil
    end
579435
def Stop(self, join_timeout=600):
    """"""
    if not self.started:
      logging.warning("Tried to stop a thread pool that was not running.")
      return

    # Remove all workers from the pool.
    workers = list(itervalues(self._workers))
    self._workers = {}
    self._workers_ro_copy = {}

    # Send a stop message to all the workers. We need to be careful here to not
    # send messages while we are still counting. If threads that haven't been
    # counted yet pick up a message and exit, the count will be off and the
    # shutdown process will deadlock.
    stop_messages_needed = 0
    for worker in workers:
      if worker.isAlive():
        stop_messages_needed += 1

    for _ in range(stop_messages_needed):
      self._queue.put(STOP_MESSAGE)

    self.started = False
    self.Join()

    # Wait for the threads to all exit now.
    for worker in workers:
      worker.join(join_timeout)
      if worker.isAlive():
        raise RuntimeError("Threadpool worker did not finish in time.")
681514
def wait_until_exit(self):
        

        if self._timeout is None:
            raise Exception("Thread will never exit. Use stop or specify timeout when starting it!")

        self._thread.join()
        self.stop()
923793
def wait(self, timeout=None):
    

    def cond(self):
      return self.__state not in (Job.PENDING, Job.RUNNING) or self.__cancelled
    if not wait_for_condition(self, cond, timeout):
      raise Job.Timeout
    return self.result
1140393
def _RemoveFromPool(self):
    
    with self.pool.lock:

      # Pool is shutting down, we can't interfere.
      if not self.pool.started:
        return False

      # Keep a minimum number of threads in the pool.
      if len(self.pool) <= self.pool.min_threads:
        return False

      # Remove us from our pool.
      self.pool._RemoveWorker(self.name)  # pylint: disable=protected-access

      return True
647559
private void endBlocking(boolean completed) throws AsynchronousCloseException {
    synchronized (blockingThreads) {
      blockingThreads.remove(Thread.currentThread());
    }
    end(completed);
  }
1212520
def wait(self, timeout=None):
    

    tbegin = _get_timeout_begin(timeout)
    with self._lock:
      while self._queue or self._running:
        remainder = _get_timeout_remainder(tbegin, timeout)
        if remainder is not None and remainder <= 0.0:
          return False  # timeout
        self._lock.wait(remainder)
      if self._shutdown:
        for worker in self._workers:
          worker.join()
    return True
143748
def _wait_threads(self):
        
        # Note that you need two loops, since you can't say which
        # thread will get each sentinel
        for t in self._threads:
            self._to_fetch.put(None)    # sentinel
        for t in self._threads:
            t.join()
        self._threads = []
1368214
def run_queue(self, pool):
        
        def callback(result):
            """Note: mark_done, at a minimum, must happen here or dbt will
            deadlock during ephemeral result error handling!
            """
            self._handle_result(result)
            self.job_queue.mark_done(result.node.unique_id)

        while not self.job_queue.empty():
            node = self.job_queue.get()
            self._raise_set_error()
            runner = self.get_runner(node)
            # we finally know what we're running! Make sure we haven't decided
            # to skip it due to upstream failures
            if runner.node.unique_id in self._skipped_children:
                cause = self._skipped_children.pop(runner.node.unique_id)
                runner.do_skip(cause=cause)
            args = (runner,)
            self._submit(pool, args, callback)

        # block on completion
        self.job_queue.join()
        # if an error got set during join(), raise it.
        self._raise_set_error()

        return
97500
@Override
	protected void doStop() throws Exception {
		for (ThreadPoolExecutor pool: threadPools.values()){
			pool.shutdown();
		}

		for (ThreadPoolExecutor pool: threadPools.values()){
			shutdownAndAwaitTermination(pool, shutdownWaitSeconds);
		}

	}
165866
public void lock() {
		if (cancel != null)
			return;
		if (error != null)
			return;
		Thread t = Thread.currentThread();
		if (t == lockingThread) {
			lockedTimes++;
			return;
		}
		BlockedThreadHandler blockedHandler = null;
		do {
			synchronized (this) {
				if (lockingThread == null) {
					lockingThread = t;
					lockedTimes = 1;
					return;
				}
				if (blockedHandler == null)
					blockedHandler = Threading.getBlockedThreadHandler(t);
				if (blockedHandler == null) {
					try { this.wait(0); }
					catch (InterruptedException e) { /* ignore */ }
					continue;
				}
			}
			blockedHandler.blocked(this, 0);
		} while (true);
	}
882844
def stop(self, timeout=None):
        
        with self._lock:
            if self._thread:
                self._queue.put_nowait(self._terminator)
                self._thread.join(timeout=timeout)
                self._thread = None
                self._thread_for_pid = None
1146188
def stop &block
      if block_given? then
        @stopped = true
        @suspended = false
        # Wait till something is put on the dead queue...
        # This stops us from acquiring the mutex until after @thread
        # has processed @stopped set to true.
        sig = @dead.deq
        # The cron thread should be dead now, or wrapping up (with the
        # acquired mutex)... 
        @mutex.synchronize {
          while @thread.alive?
            sleep 0.2
          end
          block.call(self)
        }
      end
    end
671317
def shutdown(self):
		
		with self.lock:
			self.pool.inqueue.put(StopIteration)   # Stop the pool workers
			self.waitqueue.put(StopIteration)      # Stop the input_feeder
			_iterqueue(self.waitqueue) >> item[-1] # Exhaust the waitqueue
			self.closed = True
		self.join()
12959
public boolean waitFor() throws InterruptedException {
    try {
      // [START ]
      Job completedJob = job.waitFor();
      if (completedJob == null) {
        // job no longer exists
      } else if (completedJob.getStatus().getError() != null) {
        // job failed, handle error
      } else {
        // job completed successfully
      }
      // [END ]
    } catch (BigQueryException e) {
      // Timeouts shouldn't happen without a timeout option.
      if (e.getCause() instanceof PollException) {
        return false;
      }
      throw e;
    }
    return true;
  }