func (q *rocksQueue) notify() {
	q.waitmu.Lock()
	defer q.waitmu.Unlock()
	for e := q.waiters.Front(); e != nil; e = e.Next() {
		qw := e.Value.(*QueueWaiter)
		q.waiters.Remove(e)
		deadline, _ := qw.ctx.Deadline()
		if time.Now().Before(deadline) {
			qw.notifier <- NEW_JOB
			break
		}
	}
}
Iterates through our current list of waiters,
finds one whose deadline has not passed and signals
the waiter via its channel.

This is best effort: it's possible for the waiter's
deadline to pass just as it is selected to wake up.
package storage
import (
	"container/list"
	"context"
	"encoding/binary"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/contribsys/faktory/storage/brodal"
	"github.com/contribsys/faktory/util"
	"github.com/contribsys/gorocksdb"
)
var (
	DefaultMaxSize = uint64(1000000)
)
const (
	NEW_JOB = iota
	CLOSE
)
type Backpressure struct {
	QueueName   string
	CurrentSize uint64
	MaxSize     uint64
}
func (bp Backpressure) Error() string {
	return fmt.Sprintf("%s is too large, currently %d, max size is %d", bp.QueueName, bp.CurrentSize, bp.MaxSize)
}
type queuePointer struct {
	high     uint64
	low      uint64
	priority uint8
}
func (p *queuePointer) Value() int {
	return int(-p.priority)
}
type rocksQueue struct {
	name    string
	size    uint64
	store   *rocksStore
	cf      *gorocksdb.ColumnFamilyHandle
	mu      sync.Mutex
	maxsz   uint64
	waiters *list.List
	waitmu  sync.RWMutex
	done    bool
	orderedPointers *brodal.Heap
	pointers        map[uint8]*queuePointer
}
func (q *rocksQueue) Close() {
	q.mu.Lock()
	defer q.mu.Unlock()
	q.done = true
	q.clearWaiters()
}
func (q *rocksQueue) Name() string {
	return q.name
}
func (q *rocksQueue) Page(start int64, count int64, fn func(index int, k, v []byte) error) error {
	index := 0
	upper := upperBound(q.name)
	ro := queueReadOptions(false)
	ro.SetIterateUpperBound(upper)
	ro.SetFillCache(false)
	defer ro.Destroy()
	it := q.store.db.NewIteratorCF(ro, q.cf)
	defer it.Close()
	prefix := append([]byte(q.name), 0xFF)
	it.Seek(prefix)
	if it.Err() != nil {
		return it.Err()
	}
	for i := start; i > 0; i-- {
		if !it.Valid() {
			return nil
		}
		it.Next()
	}
	for ; it.Valid(); it.Next() {
		if count == 0 {
			break
		}
		if err := it.Err(); err != nil {
			return err
		}
		k := it.Key()
		v := it.Value()
		value := v.Data()
		key := k.Data()
		err := fn(index, key, value)
		index++
		k.Free()
		v.Free()
		if err != nil {
			return err
		}
		count--
	}
	if it.Err() != nil {
		return it.Err()
	}
	return nil
}
func (q *rocksQueue) Each(fn func(index int, k, v []byte) error) error {
	return q.Page(0, -1, fn)
}
func (q *rocksQueue) Clear() (uint64, error) {
	count := uint64(0)
	q.mu.Lock()
	defer q.mu.Unlock()
	upper := upperBound(q.name)
	ro := queueReadOptions(true)
	ro.SetIterateUpperBound(upper)
	ro.SetFillCache(false)
	defer ro.Destroy()
	it := q.store.db.NewIteratorCF(ro, q.cf)
	defer it.Close()
	prefix := append([]byte(q.name), 0xFF)
	it.Seek(prefix)
	if it.Err() != nil {
		return 0, it.Err()
	}
	if !it.Valid() {
		return 0, nil
	}
	wo := queueWriteOptions()
	defer wo.Destroy()
	wb := gorocksdb.NewWriteBatch()
	defer wb.Destroy()
	for ; it.Valid(); it.Next() {
		k := it.Key()
		key := k.Data()
		wb.DeleteCF(q.cf, key)
		k.Free()
		count++
	}
	err := q.store.db.Write(wo, wb)
	if err != nil {
		return count, err
	}
	q.orderedPointers = brodal.NewHeap()
	q.pointers = make(map[uint8]*queuePointer)
	atomic.StoreUint64(&q.size, 0)
	return count, nil
}
func (q *rocksQueue) Init() error {
	q.mu = sync.Mutex{}
	upper := upperBound(q.name)
	ro := queueReadOptions(false)
	ro.SetIterateUpperBound(upper)
	ro.SetFillCache(false)
	defer ro.Destroy()
	var count uint64
	it := q.store.db.NewIteratorCF(ro, q.cf)
	defer it.Close()
	prefix := append([]byte(q.name), 0xFF)
	it.Seek(prefix)
	if err := it.Err(); err != nil {
		return err
	}
	for ; it.Valid(); it.Next() {
		k := it.Key()
		key := k.Data()
		_, priority, seq := decodeKey(q.name, key)
		if p, ok := q.pointers[priority]; ok {
			if seq > p.high {
				p.high = seq
			} else if seq < p.low {
				p.low = seq
			}
		} else {
			p = &queuePointer{
				priority: priority,
				high:     seq,
				low:      seq,
			}
			q.orderedPointers.Insert(p)
			q.pointers[priority] = p
		}
		k.Free()
		count++
	}
	it.SeekToLast()
	if err := it.Err(); err != nil {
		return err
	}
	atomic.StoreUint64(&q.size, count)
	q.waiters = list.New()
	util.Debugf("Queue init: %s %d elements", q.name, atomic.LoadUint64(&q.size))
	return nil
}
func (q *rocksQueue) Size() uint64 {
	return atomic.LoadUint64(&q.size)
}
func (q *rocksQueue) Push(priority uint8, payload []byte) error {
	q.mu.Lock()
	defer q.mu.Unlock()
	if size := atomic.LoadUint64(&q.size); size > q.maxsz {
		return Backpressure{q.name, size, q.maxsz}
	}
	k := q.nextkey(priority)
	v := payload
	wo := queueWriteOptions()
	defer wo.Destroy()
	err := q.store.db.PutCF(wo, q.cf, k, v)
	if err != nil {
		return err
	}
	atomic.AddUint64(&q.size, 1)
	q.notify()
	return nil
}
func (q *rocksQueue) Pop() ([]byte, error) {
	q.mu.Lock()
	defer q.mu.Unlock()
	if atomic.LoadUint64(&q.size) == 0 || q.done {
		return nil, nil
	}
	return q._pop()
}
func (q *rocksQueue) _pop() ([]byte, error) {
	var value []byte
	var err error
	if len(q.pointers) == 0 {
		return nil, nil
	}
	p := q.orderedPointers.Peek().(*queuePointer)
	key := makeKey(q.name, p.priority, p.high)
	ro := queueReadOptions(true)
	ro.SetIterateUpperBound(key)
	defer ro.Destroy()
	for {
		key = makeKey(q.name, p.priority, p.low)
		value, err = q.store.db.GetBytesCF(ro, q.cf, key)
		if err != nil {
			return nil, err
		}
		if value != nil {
			break
		}
		if p.low < p.high {
			atomic.AddUint64(&p.low, 1)
			continue
		}
		q.orderedPointers.Pop()
		delete(q.pointers, p.priority)
		return nil, nil
	}
	wo := queueWriteOptions()
	defer wo.Destroy()
	err = q.store.db.DeleteCF(wo, q.cf, key)
	if err != nil {
		return nil, err
	}
	atomic.AddUint64(&p.low, 1)
	if p.low > p.high {
		q.orderedPointers.Pop()
		delete(q.pointers, p.priority)
	}
	atomic.AddUint64(&q.size, ^uint64(0))
	return value, nil
}
type QueueWaiter struct {
	ctx      context.Context
	notifier chan int
}