NOTE: the queue has to be thread safe.
425765
protected void doPrepare(Map<String, Object> topoConf, TopologyContext context, OutputCollector
      collector, Collection<Event<Tuple>> queue) {
    Objects.requireNonNull(topoConf);
    Objects.requireNonNull(context);
    Objects.requireNonNull(collector);
    Objects.requireNonNull(queue);
    this.windowedOutputCollector = new WindowedOutputCollector(collector);
    bolt.prepare(topoConf, context, windowedOutputCollector);
    this.listener = newWindowLifecycleListener();
    this.windowManager = initWindowManager(listener, topoConf, context, queue);
    start();
    LOG.info(String.format("Initialized window manager %s", windowManager));
  }
576194
void setStrongMatchWaitLock() {
        synchronized (reqQueueLockObject) {
            for (ServerRequest req : queue) {
                if (req != null) {
                    if (req instanceof ServerRequestInitSession) {
                        req.addProcessWaitLock(ServerRequest.PROCESS_WAIT_LOCK.STRONG_MATCH_PENDING_WAIT_LOCK);
                    }
                }
            }
        }
    }
1212803
@Override
    public void flushQueueBlocking() {
        flushQueue();

        val context =((CudaContext) AtomicAllocator.getInstance().getDeviceContext().getContext());

        context.syncSpecialStream();
        context.syncOldStream();
    }
495301
private void initializeQueue() {
        this.queue.clear();
        for (Map.Entry<String, NodeT> entry: nodeTable.entrySet()) {
            if (!entry.getValue().hasDependencies()) {
                this.queue.add(entry.getKey());
            }
        }
        if (queue.isEmpty()) {
            throw new IllegalStateException("Detected circular dependency");
        }
    }
593204
private Queue<T> loadMorselQueue() {
        Queue<T> morselQueue = createQueue();

        //load morsels from state;
        Set<T> morsels = new LinkedHashSet<>(this.stateManager.getMorsels());

        morselQueue.addAll(morsels);

        if (morselQueue.isEmpty()) {
            loadMorselQueueFromSource(morselQueue);
        }

        return morselQueue;
    }
1214210
private static void queueTask(long threadId, Runnable toRun){
    assert control.isHeldByCurrentThread();
    assert threadId != currentThread;
    //(get queue)
    if(!threadedLogQueue.containsKey(threadId)){
      threadedLogQueue.put(threadId, new LinkedList<Runnable>());
    }
    Queue<Runnable> threadLogQueue = threadedLogQueue.get(threadId);
    //(add to queue)
    threadLogQueue.offer( toRun );
    //(register this thread as waiting)
    if(!threadsWaiting.contains(threadId)){
      threadsWaiting.offer(threadId);
      assert threadedLogQueue.get(threadId) != null && !threadedLogQueue.get(threadId).isEmpty();
    }
  }
1267485
def flush
      mutex.synchronize do
        if thread && thread.alive?
          queue.push(marker)
          marker.wait(mutex)
        end
      end
    end
201719
void initializeReplQueues() throws SafeModeException {
    writeLock();
    try {
      if (isPopulatingReplQueues()) {
        NameNode.stateChangeLog.info("STATE* Safe mode is already OFF."
            + " Replication queues are initialized");
        return;
      }
      safeMode.initializeReplicationQueues();
    } finally {
      writeUnlock();
    }
  }
114501
boolean enqueue(Reference<? extends T> reference) {
        synchronized (lock) {
            if (enqueueLocked(reference)) {
                lock.notifyAll();
                return true;
            }
            return false;
        }
    }
472338
protected void initQueue(T queue, QueueSpec spec) throws Exception {
        queue.setObserver(defaultObserver);
        queue.init();
    }
861859
def iterqueue(queue, expected):
    """"""
    while expected > 0:
        for item in iter(queue.get, EXIT):
            yield item
        expected -= 1
550850
private void requirePending() {
		if (pending != null) {
			return;
		}

		long exclusionComparable = Long.MIN_VALUE;
		while (nInclusionsRemaining != 0 && !queue.isEmpty()) {
			//find a candidate that is not excluded
			HeapElement inclusion = null;
			do {
				HeapElement candidate = queue.poll();
				if (candidate.inclusion) {
					if (exclusionComparable != candidate.comparable()) {
						inclusion = candidate;
						break;
					}
				} else {
					exclusionComparable = candidate.comparable();
				}
				reattach(candidate);
				if (nInclusionsRemaining == 0) {
					return;
				}
			} while (!queue.isEmpty());
			if (inclusion == null) {
				return;
			}
			long inclusionComparable = inclusion.comparable();

			/*
			 * Check for any following exclusions and for duplicates. We could
			 * change the sort order so that exclusions always preceded
			 * inclusions, but that would be less efficient and would make the
			 * ordering different than the comparable value.
			 */
			boolean excluded = exclusionComparable == inclusionComparable;
			while (!queue.isEmpty() && queue.peek().comparable() == inclusionComparable) {
				HeapElement match = queue.poll();
				excluded |= !match.inclusion;
				reattach(match);
				if (nInclusionsRemaining == 0) {
					return;
				}
			}
			if (!excluded) {
				pending = inclusion;
				return;
			}
			reattach(inclusion);
		}
	}
1129793
public synchronized void lock()
  {
    if (tc.isEntryEnabled())
      SibTr.entry(tc, "lock", this);

    boolean interrupted = false;

    // Attempt to get a lock on the mutex.
    // if we fail, then that is because the lock
    // must be held exclusively.
    while (!tryLock())
      try
      {
        // Wait for 1 second then try again.
        if (tc.isDebugEnabled())
          SibTr.debug(tc, "Waiting for lock");
        wait(1000);
      }
      catch (InterruptedException e)
      {
        // No FFDC code needed
        interrupted = true;
      }

    if (interrupted)
      Thread.currentThread().interrupt();

    if (tc.isEntryEnabled())
      SibTr.exit(tc, "lock");

  }
673243
public static <T> T queueSpinTake(BlockingQueue<T> queue) throws InterruptedException {
        if (QUEUE_SPIN_MICROSECONDS > 0) {
            T retval = null;
            long nanos = -1;
            for (;;) {
                if ((retval = queue.poll()) != null) return retval;
                if (nanos == -1) {
                    nanos = System.nanoTime();
                } else if (System.nanoTime() - nanos > QUEUE_SPIN_MICROSECONDS) {
                    return queue.take();
                }
            }
        } else {
            return queue.take();
        }
    }
1046684
async def queue_declare(self):
         
        await self.channel.queue_declare(
                self.queue,
                durable=self.durable,
                exclusive=self.exclusive,
                no_wait=self.no_wait
            )
397930
private void ensureAvail(int n) {
    if (pos + n >= buf.length) {
      int newSize = Math.max(pos + n, buf.length * 2);
      buf = Arrays.copyOf(buf, newSize);
    }
  }
1368682
@Override
	protected void consume() {
		E obj = null;
		try {
			obj = queue.take();
		} catch (InterruptedException e) {
			//thread.isInterrupted();
			return;
		}
		process(obj);
	}