// Drain drains all jobs in the queue before returning. Note that if jobs are added faster than we can process them, this function wouldn't return.
180067
func (wp *WorkerPool) Drain() {
	wg := sync.WaitGroup{}
	for _, w := range wp.workers {
		wg.Add(1)
		go func(w *worker) {
			w.drain()
			wg.Done()
		}(w)
	}
	wg.Wait()
}
793941
func (logger *Logger) Flush() {
	b := &backgroundFlusher{
		logger:     logger,
		updateChan: make(chan struct{}),
	}
	transitChannel <- b
	<-b.updateChan
}
682818
func (d *Disque) Fetch(queueName string, timeout time.Duration) (job *Job, err error) {
	var jobs []*Job
	if jobs, err = d.FetchMultiple(queueName, 1, timeout); err == nil {
		if len(jobs) > 0 {
			job = jobs[0]
		}
	}
	return
}
219410
func (cl *ChannelLogger) DrainForever(logger log.Logger, errCh channels.Channel) {
	// logLine could be nil if channel was closed while waiting for next line
	for logLine := cl.WaitReadLogLine(); logLine != nil; logLine = cl.WaitReadLogLine() {
		err := logger.Log(logLine...)
		if err != nil && errCh != nil {
			errCh.In() <- err
		}
	}
}
934978
func (b *Buffer) Dequeue() (interface{}, error) {
	var ln int
	var item interface{}

	b.rw.Lock()
	{
		if ln = len(b.buffer); ln > 0 {
			item = b.buffer[0]
			b.buffer = b.buffer[1:]
		}
	}
	b.rw.Unlock()

	if ln == 0 {
		return nil, ErrBufferEmpty
	}

	return item, nil
}
961582
func (s *SQS) Purge() error {

	params := &sqs.PurgeQueueInput{
		QueueUrl: s.queueURL,
	}
	_, err := s.client().PurgeQueue(params)
	return err
}
1109162
func (l *BytesBufferLogger) Flush() {
	l.mux.Lock()
	defer l.mux.Unlock()
	l.buf.Reset()
}
645566
func (n *NodeDrainer) handleJobAllocDrain(req *DrainRequest) {
	index, err := n.batchDrainAllocs(req.Allocs)
	req.Resp.Respond(index, err)
}
631601
func (w *BufferedWriter) Flush() error {
	if w.err == nil {
		w.err = w.bw.Flush()
	}

	return w.err
}
1155695
func (ew *EventWorker) Flush() []ssf.SSFSample {
	ew.mutex.Lock()

	retsamples := ew.samples
	// these slices will be allocated again at append time
	ew.samples = nil

	ew.mutex.Unlock()
	if len(retsamples) != 0 {
		ew.stats.Count("worker.other_samples_flushed_total", int64(len(retsamples)), nil, 1.0)
	}
	return retsamples
}
1329455
func (c *Client) Queues() ([]*Queue, error) {
	conn := c.pool.Get()
	defer conn.Close()

	key := redisKeyKnownJobs(c.namespace)
	jobNames, err := redis.Strings(conn.Do("SMEMBERS", key))
	if err != nil {
		return nil, err
	}
	sort.Strings(jobNames)

	for _, jobName := range jobNames {
		conn.Send("LLEN", redisKeyJobs(c.namespace, jobName))
	}

	if err := conn.Flush(); err != nil {
		logError("client.queues.flush", err)
		return nil, err
	}

	queues := make([]*Queue, 0, len(jobNames))

	for _, jobName := range jobNames {
		count, err := redis.Int64(conn.Receive())
		if err != nil {
			logError("client.queues.receive", err)
			return nil, err
		}

		queue := &Queue{
			JobName: jobName,
			Count:   count,
		}

		queues = append(queues, queue)
	}

	for _, s := range queues {
		if s.Count > 0 {
			conn.Send("LINDEX", redisKeyJobs(c.namespace, s.JobName), -1)
		}
	}

	if err := conn.Flush(); err != nil {
		logError("client.queues.flush2", err)
		return nil, err
	}

	now := nowEpochSeconds()

	for _, s := range queues {
		if s.Count > 0 {
			b, err := redis.Bytes(conn.Receive())
			if err != nil {
				logError("client.queues.receive2", err)
				return nil, err
			}

			job, err := newJob(b, nil, nil)
			if err != nil {
				logError("client.queues.new_job", err)
			}
			s.Latency = now - job.EnqueuedAt
		}
	}

	return queues, nil
}
295719
func (c *Client) Enqueue(j *Job) error {
	return execEnqueue(j, c.pool)
}
80947
func (f *FileWriter) Flush() error {
	if f.closed {
		return io.ErrClosedPipe
	}

	if f.blockWriter != nil {
		return f.blockWriter.Flush()
	}

	return nil
}
682786
func (c *Channel) flush() error {
	var msgBuf bytes.Buffer

	if len(c.memoryMsgChan) > 0 || len(c.inFlightMessages) > 0 || len(c.deferredMessages) > 0 {
		c.ctx.nsqd.logf(LOG_INFO, "CHANNEL(%s): flushing %d memory %d in-flight %d deferred messages to backend",
			c.name, len(c.memoryMsgChan), len(c.inFlightMessages), len(c.deferredMessages))
	}

	for {
		select {
		case msg := <-c.memoryMsgChan:
			err := writeMessageToBackend(&msgBuf, msg, c.backend)
			if err != nil {
				c.ctx.nsqd.logf(LOG_ERROR, "failed to write message to backend - %s", err)
			}
		default:
			goto finish
		}
	}

finish:
	c.inFlightMutex.Lock()
	for _, msg := range c.inFlightMessages {
		err := writeMessageToBackend(&msgBuf, msg, c.backend)
		if err != nil {
			c.ctx.nsqd.logf(LOG_ERROR, "failed to write message to backend - %s", err)
		}
	}
	c.inFlightMutex.Unlock()

	c.deferredMutex.Lock()
	for _, item := range c.deferredMessages {
		msg := item.Value.(*Message)
		err := writeMessageToBackend(&msgBuf, msg, c.backend)
		if err != nil {
			c.ctx.nsqd.logf(LOG_ERROR, "failed to write message to backend - %s", err)
		}
	}
	c.deferredMutex.Unlock()

	return nil
}
330299
func (q *PlanQueue) Flush() {
	q.l.Lock()
	defer q.l.Unlock()

	// Error out all the futures
	for _, pending := range q.ready {
		pending.respond(nil, planQueueFlushed)
	}

	// Reset the broker
	q.stats.Depth = 0
	q.ready = make([]*pendingPlan, 0, 16)

	// Unblock any waiters
	select {
	case q.waitCh <- struct{}{}:
	default:
	}
}
1000709
func (t *T) DrainAll() error {
	for i := range t.shards {
		t.shards[i].closeConns(true, true)
	}
	for i := range t.shards {
		t.shards[i].wait()
	}
	return nil
}
1289178
func (tp *BytePool) Close() {
	tp.Drain()
	if tp.drainTicker != nil {
		tp.drainTicker.Stop()
		tp.drainTicker = nil
	}
}