101 lines
2.1 KiB
Go
101 lines
2.1 KiB
Go
package jobs
|
|
|
|
import (
|
|
"context"
|
|
"slices"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
const slotWaitInterval = 500 * time.Millisecond
|
|
|
|
const (
|
|
queueFirst = "first"
|
|
queueUp = "up"
|
|
queueDown = "down"
|
|
)
|
|
|
|
type slotLimiter struct {
|
|
mutex sync.Mutex
|
|
active int
|
|
waiting []string
|
|
}
|
|
|
|
func (limiter *slotLimiter) enqueue(id string) {
|
|
limiter.mutex.Lock()
|
|
defer limiter.mutex.Unlock()
|
|
if !slices.Contains(limiter.waiting, id) {
|
|
limiter.waiting = append(limiter.waiting, id)
|
|
}
|
|
}
|
|
|
|
func (limiter *slotLimiter) acquire(ctx context.Context, id string, limit func() int) bool {
|
|
limiter.enqueue(id)
|
|
ticker := time.NewTicker(slotWaitInterval)
|
|
defer ticker.Stop()
|
|
for {
|
|
if limiter.tryAcquire(id, limit()) {
|
|
return true
|
|
}
|
|
select {
|
|
case <-ctx.Done():
|
|
limiter.remove(id)
|
|
return false
|
|
case <-ticker.C:
|
|
}
|
|
}
|
|
}
|
|
|
|
func (limiter *slotLimiter) tryAcquire(id string, limit int) bool {
|
|
limiter.mutex.Lock()
|
|
defer limiter.mutex.Unlock()
|
|
if limiter.active >= limit || len(limiter.waiting) == 0 || limiter.waiting[0] != id {
|
|
return false
|
|
}
|
|
limiter.waiting = limiter.waiting[1:]
|
|
limiter.active++
|
|
return true
|
|
}
|
|
|
|
func (limiter *slotLimiter) release() {
|
|
limiter.mutex.Lock()
|
|
defer limiter.mutex.Unlock()
|
|
limiter.active--
|
|
}
|
|
|
|
func (limiter *slotLimiter) remove(id string) {
|
|
limiter.mutex.Lock()
|
|
defer limiter.mutex.Unlock()
|
|
limiter.waiting = slices.DeleteFunc(limiter.waiting, func(waitingID string) bool {
|
|
return waitingID == id
|
|
})
|
|
}
|
|
|
|
func (limiter *slotLimiter) position(id string) int {
|
|
limiter.mutex.Lock()
|
|
defer limiter.mutex.Unlock()
|
|
return slices.Index(limiter.waiting, id) + 1
|
|
}
|
|
|
|
func (limiter *slotLimiter) reorder(id, direction string) bool {
|
|
limiter.mutex.Lock()
|
|
defer limiter.mutex.Unlock()
|
|
index := slices.Index(limiter.waiting, id)
|
|
if index < 0 {
|
|
return false
|
|
}
|
|
|
|
target := index
|
|
switch direction {
|
|
case queueFirst:
|
|
target = 0
|
|
case queueUp:
|
|
target = max(index-1, 0)
|
|
case queueDown:
|
|
target = min(index+1, len(limiter.waiting)-1)
|
|
}
|
|
limiter.waiting = slices.Delete(limiter.waiting, index, index+1)
|
|
limiter.waiting = slices.Insert(limiter.waiting, target, id)
|
|
return true
|
|
}
|