// Would be nice to switch to this model http://redis.io/commands/rpoplpush#pattern-reliable-queue
764618
func (mq *RedisMQ) Reserve(ctx context.Context) (*models.Task, error) {

	conn := mq.pool.Get()
	defer conn.Close()
	var job models.Task
	var resp []byte
	var err error
	for i := 2; i >= 0; i-- {
		resp, err = redis.Bytes(conn.Do("RPOP", fmt.Sprintf("%s%d", mq.queueName, i)))
		if mq.checkNilResponse(err) {
			if i == 0 {
				// Out of queues!
				return nil, nil
			}

			// No valid job on this queue, try lower priority queue.
			continue
		} else if err != nil {
			// Some other error!
			return nil, err
		}

		// We got a valid high priority job.
		break
	}

	if err != nil {
		return nil, err
	}
	err = json.Unmarshal(resp, &job)
	if err != nil {
		return nil, err
	}

	response, err := redis.Int64(conn.Do("INCR", mq.queueName+"_incr"))
	if err != nil {
		return nil, err
	}
	reservationId := strconv.FormatInt(response, 10)
	_, err = conn.Do("ZADD", "timeout:", time.Now().Add(time.Minute).Unix(), reservationId)
	if err != nil {
		return nil, err
	}
	_, err = conn.Do("HSET", "timeout", reservationId, resp)
	if err != nil {
		return nil, err
	}

	// Map from job.ID -> reservation ID
	_, err = conn.Do("HSET", "reservations", job.ID, reservationId)
	if err != nil {
		return nil, err
	}

	_, log := common.LoggerWithFields(ctx, logrus.Fields{"call_id": job.ID})
	log.Println("Reserved")

	return &job, nil
}
83128
func (w *WorkerConfig) scheduler() {
	pollSets := []string{w.nsKey("retry"), w.nsKey("schedule")}

	for _ = range time.Tick(w.PollInterval) {
		w.RLock() // don't let quitHandler() stop us in the middle of a run
		conn := w.RedisPool.Get()
		now := fmt.Sprintf("%f", timeFloat(time.Now()))
		for _, set := range pollSets {
			conn.Send("MULTI")
			conn.Send("ZRANGEBYSCORE", set, "-inf", now)
			conn.Send("ZREMRANGEBYSCORE", set, "-inf", now)
			res, err := redis.Values(conn.Do("EXEC"))
			if err != nil {
				w.handleError(err)
				continue
			}

			for _, msg := range res[0].([]interface{}) {
				parsedMsg := &struct {
					Queue string `json:"queue"`
				}{}
				msgBytes := msg.([]byte)
				err := json.Unmarshal(msgBytes, parsedMsg)
				if err != nil {
					w.handleError(err)
					continue
				}
				if _, err = conn.Do("RPUSH", w.nsKey("queue:"+parsedMsg.Queue), msgBytes); err != nil {
					w.handleError(err)
				}
			}
		}
		conn.Close()
		w.RUnlock()
	}
}
12977
func (q *prioQueue) Pop() int {
	n := q.Len() - 1
	q.swap(0, n)
	q.down(0, n)

	v := q.heap[n]
	q.index[v] = -1
	q.heap = q.heap[:n]
	return v
}
900817
func (p *PriorityQueue) Pop() (interface{}, error) {
	if len(*p.itemHeap) == 0 {
		return nil, errors.New("empty queue")
	}

	item := heap.Pop(p.itemHeap).(*item)
	delete(p.lookup, item.value)
	return item.value, nil
}
878313
func (queue *Queue) RequeueFailed() error {
	l := queue.GetFailedLength()
	// TODO implement this in lua
	for l > 0 {
		err := queue.redisClient.RPopLPush(queueFailedKey(queue.Name), queueInputKey(queue.Name)).Err()
		if err != nil {
			return err
		}
		queue.incrRate(queueInputRateKey(queue.Name), 1)
		l--
	}
	return nil
}
1304031
func (pq *txPriorityQueue) Pop() interface{} {
	n := len(pq.items)
	item := pq.items[n-1]
	pq.items[n-1] = nil
	pq.items = pq.items[0 : n-1]
	return item
}
563286
func (keap *KamEvapiPool) PopKamEvapi() (*KamEvapi, error) {
	if keap == nil {
		return nil, errors.New("UNCONFIGURED_KAMAILIO_POOL")
	}
	if len(keap.conns) != 0 { // Select directly if available, so we avoid randomness of selection
		KamEvapi := <-keap.conns
		return KamEvapi, nil
	}
	var KamEvapi *KamEvapi
	var err error
	select { // No KamEvapi available in the pool, wait for first one showing up
	case KamEvapi = <-keap.conns:
	case <-keap.allowedConns:
		KamEvapi, err = NewKamEvapi(keap.kamAddr, keap.connId, keap.reconnects, nil, keap.logger)
		if err != nil {
			return nil, err
		}
		return KamEvapi, nil
	}
	return KamEvapi, nil
}
334526
func (c *asyncClient) Brpoplpush(arg0 string, arg1 string, timeout int) (result FutureBytesArray, err Error) {
	arg0bytes := []byte(arg0)
	arg1bytes := []byte(arg1)
	arg2bytes := []byte(fmt.Sprint(timeout))

	var resp *PendingResponse
	resp, err = c.conn.QueueRequest(&BRPOPLPUSH, [][]byte{arg0bytes, arg1bytes, arg2bytes})
	if err == nil {
		result = resp.future.(FutureBytesArray)
	}
	return result, err

}
1263779
func (q *execQueue) queue(f func()) bool {
	q.mu.Lock()
	ok := !q.isClosed() && len(q.funcs) < cap(q.funcs)
	if ok {
		q.funcs = append(q.funcs, f)
		q.cond.Signal()
	}
	q.mu.Unlock()
	return ok
}
786193
func (rkq *RekeyQueueStandard) Enqueue(id tlf.ID) {
	rkq.mu.Lock()
	defer rkq.mu.Unlock()
	rkq.pendings[id] = true

	select {
	case rkq.queue <- id:
	default:
		// The queue is full; drop this one for now until the next
		// request to the server for more rekeys.
		rkq.log.Debug("Rekey queue is full; dropping %s", id)
	}
}
58065
func (c *Controller) Requeue(obj interface{}) error {
	return c.config.Queue.AddIfNotPresent(cache.Deltas{
		cache.Delta{
			Type:   cache.Sync,
			Object: obj,
		},
	})
}
937070
func (e *Endpoint) RLockAlive() error {
	e.mutex.RLock()
	if e.IsDisconnecting() {
		e.mutex.RUnlock()
		return ErrNotAlive
	}
	return nil
}
21174
func Pattern(glob string) string {
	s := fmt.Sprintf("%%%s%%", strings.Replace(glob, "*", "%", -1))
	return squeezer.ReplaceAllString(s, "%")
}
401633
func (p *PriorityQueue) Pop() (*v1.Pod, error) {
	p.lock.Lock()
	defer p.lock.Unlock()
	for p.activeQ.Len() == 0 {
		// When the queue is empty, invocation of Pop() is blocked until new item is enqueued.
		// When Close() is called, the p.closed is set and the condition is broadcast,
		// which causes this loop to continue and return from the Pop().
		if p.closed {
			return nil, fmt.Errorf(queueClosed)
		}
		p.cond.Wait()
	}
	obj, err := p.activeQ.Pop()
	if err != nil {
		return nil, err
	}
	pInfo := obj.(*podInfo)
	p.schedulingCycle++
	return pInfo.pod, err
}
1164627
func (dbp *Process) SwitchThread(tid int) error {
	if dbp.exited {
		return &proc.ErrProcessExited{Pid: dbp.Pid()}
	}
	if th, ok := dbp.threads[tid]; ok {
		dbp.currentThread = th
		dbp.selectedGoroutine, _ = proc.GetG(dbp.currentThread)
		return nil
	}
	return fmt.Errorf("thread %d does not exist", tid)
}
1052951
func Pop(queue Queue) interface{} {
	var result interface{}
	queue.Pop(func(obj interface{}) error {
		result = obj
		return nil
	})
	return result
}
1216508
func DownlinkQueue(prefix string) storage.MigrateFunction {
	return func(client *redis.Client, key string, obj map[string]string) (string, map[string]string, error) {
		var err error
		nextDownlink, ok := obj["next_downlink"]
		if ok {
			delete(obj, "next_downlink")
			scheduleKey := prefix + ":downlink" + strings.TrimPrefix(key, prefix+":device")
			err = client.LPush(scheduleKey, nextDownlink).Err()
		}
		return "2.4.2", obj, err
	}
}