EXPECT THE self.lock TO BE HAD, WAITS FOR self.queue TO HAVE A LITTLE SPACE
185134
def _wait_for_queue_space(self, timeout=DEFAULT_WAIT_TIME):
        
        wait_time = 5

        (DEBUG and len(self.queue) > 1 * 1000 * 1000) and Log.warning("Queue {{name}} has over a million items")

        now = time()
        if timeout != None:
            time_to_stop_waiting = now + timeout
        else:
            time_to_stop_waiting = now + DEFAULT_WAIT_TIME

        if self.next_warning < now:
            self.next_warning = now + wait_time

        while not self.closed and len(self.queue) >= self.max:
            if now > time_to_stop_waiting:
                Log.error(THREAD_TIMEOUT)

            if self.silent:
                self.lock.wait(Till(till=time_to_stop_waiting))
            else:
                self.lock.wait(Till(seconds=wait_time))
                if len(self.queue) >= self.max:
                    now = time()
                    if self.next_warning < now:
                        self.next_warning = now + wait_time
                        Log.alert(
                            "Queue by name of {{name|quote}} is full with ({{num}} items), thread(s) have been waiting {{wait_time}} sec",
                            name=self.name,
                            num=len(self.queue),
                            wait_time=wait_time
                        )
90579
async def wait(self, container = None):
        '''
        
        '''
        if hasattr(self, '_result'):
            if hasattr(self, '_exception'):
                raise self._exception
            else:
                return self._result
        else:
            ev = await FutureEvent.createMatcher(self)
            if hasattr(ev, 'exception'):
                raise ev.exception
            else:
                return ev.result
880702
async def release(self, *args, **kwargs):
        '''
        
        '''
        return await _maybe_await(self.lock.release(*args, **kwargs))
1278567
void waitForResuming() throws IOException
   {
      if (isSuspended.get())
      {
         try
         {
            latcher.get().await();
         }
         catch (InterruptedException e)
         {
            throw new IOException(e);
         }
      }
   }
1022548
def wait_closed(self):
        
        if self._closed:
            return
        if not self._closing:
            raise RuntimeError(
                ".wait_closed() should be called "
                "after .close()"
            )

        while self._free:
            conn = self._free.popleft()
            if not conn.closed:
                yield from conn.close()
            else:
                # pragma: no cover
                pass
        with (yield from self._cond):
            while self.size > self.freesize:
                yield from self._cond.wait()
        self._used.clear()
        self._closed = True
951984
def wait(self):
        

        self.log.debug("acquiring wait lock to wait for completion")
        # using the lock in a with-context blocks, which is what we want if
        # we're running wait()
        with self._wait_lock:
            self.log.debug("got wait lock")
            witnessed_end = False

            if self.exit_code is None:
                self.log.debug("exit code not set, waiting on pid")
                pid, exit_code = no_interrupt(os.waitpid, self.pid, 0) # blocks
                self.exit_code = handle_process_exit_code(exit_code)
                witnessed_end = True

            else:
                self.log.debug("exit code already set (%d), no need to wait",
                        self.exit_code)

            self._quit_threads.set()

            # we may not have a thread for stdin, if the pipe has been connected
            # via _piped="direct"
            if self._input_thread:
                self._input_thread.join()

            # wait, then signal to our output thread that the child process is
            # done, and we should have finished reading all the stdout/stderr
            # data that we can by now
            timer = threading.Timer(2.0, self._stop_output_event.set)
            timer.start()

            # wait for our stdout and stderr streamreaders to finish reading and
            # aggregating the process output
            self._output_thread.join()
            timer.cancel()

            self._background_thread.join()

            if witnessed_end:
                self._process_just_ended()

            return self.exit_code
1201034
public static void wait(Object lock, long timeout) {
	try {
	    synchronized (lock) {
		lock.wait(timeout);
	    }
	}
	catch (InterruptedException e) {}
    }
349815
def expect(self, relist, timeout=None, cleanup=None):
        
        return self._waitfor(relist, timeout, True, cleanup=cleanup)
222164
def _ensure_queue(self):
        """"""
        if self._queue_path not in self._client.kv:
            self._client.kv[self._queue_path] = None
958512
def prepare_queues(queues, lock):
    """"""
    for queue in queues:
        queue._pebble_lock = lock
        with queue.mutex:
            queue._pebble_old_method = queue._put
            queue._put = MethodType(new_method, queue)
640079
private void enqueue(E x) {
        // assert lock.getHoldCount() == 1;
        // assert items[putIndex] == null;
        final Object[] items = this.items;
        items[putIndex] = x;
        if (++putIndex == items.length) putIndex = 0;
        count++;
        notEmpty.signal();
    }
1266242
def wait(self):
        
        ret = os.read(self.event_fd, 64 / 8)
        return struct.unpack('Q', ret)
591437
async function take(key) {
    const previousLocks = LOCKS_MAP.get(key) || [];
    let locksLength = previousLocks.length;

    log('debug', `Taking the lock on ${key} (queue length was ${locksLength})`);

    let release;

    const newLock = {
      releasePromise: new Promise(async (resolve, reject) => {
        release = resolve;

        if (LOCK_TIMEOUT !== Infinity) {
          await delay.create(LOCK_TIMEOUT);

          reject(new YError('E_LOCK_TIMEOUT'));
        }
      }),
      release,
    };

    previousLocks.push(newLock);
    locksLength++;

    if (locksLength > 1) {
      await previousLocks[locksLength - 1].releasePromise;
    } else {
      LOCKS_MAP.set(key, previousLocks);
    }
  }
1257328
void invokeAndWait(final Runnable r) {
        final CountDownLatch latch = new CountDownLatch(1);
        queue.postRunnable(() -> {
            try {
                r.run();
            } finally {
                latch.countDown();
            }
        });
        try {
            latch.await();
        } catch (InterruptedException e) { }
    }
795552
def lock(self, resource, label='', expire=60, patience=60):
        
        queue = Queue(client=self.client, resource=resource)
        with queue.draw(label=label, expire=expire) as number:
            queue.wait(number=number, patience=patience)
            yield
        queue.close()
1125005
def ready(self):
        """"""
        if self.__timeout < time.time():
            self.cancel()
        return self.__queue.full() and not self.__queue.empty()
624719
def wait_for(self, timeout=None):
        
        from broqer.op import OnEmitFuture  # due circular dependency
        return self | OnEmitFuture(timeout=timeout)