134 lines
2.0 KiB
Go
134 lines
2.0 KiB
Go
// backend\tasks_exclusive.go
|
|
|
|
package main
|
|
|
|
import (
|
|
"context"
|
|
"sync/atomic"
|
|
)
|
|
|
|
type exclusiveTaskRequest struct {
|
|
id uint64
|
|
granted chan struct{}
|
|
}
|
|
|
|
type exclusiveTaskCancelRequest struct {
|
|
id uint64
|
|
ack chan bool
|
|
}
|
|
|
|
var (
|
|
exclusiveTaskReqCh = make(chan *exclusiveTaskRequest)
|
|
exclusiveTaskCancelCh = make(chan exclusiveTaskCancelRequest)
|
|
exclusiveTaskReleaseCh = make(chan struct{}, 1)
|
|
exclusiveTaskNextID atomic.Uint64
|
|
)
|
|
|
|
func init() {
|
|
go runExclusiveTaskQueue()
|
|
}
|
|
|
|
func runExclusiveTaskQueue() {
|
|
queue := make([]*exclusiveTaskRequest, 0, 8)
|
|
|
|
locked := false
|
|
var activeID uint64
|
|
|
|
grantNext := func() {
|
|
if locked {
|
|
return
|
|
}
|
|
|
|
for len(queue) > 0 {
|
|
req := queue[0]
|
|
queue = queue[1:]
|
|
|
|
if req == nil {
|
|
continue
|
|
}
|
|
|
|
locked = true
|
|
activeID = req.id
|
|
close(req.granted)
|
|
return
|
|
}
|
|
|
|
activeID = 0
|
|
}
|
|
|
|
for {
|
|
grantNext()
|
|
|
|
select {
|
|
case req := <-exclusiveTaskReqCh:
|
|
if req != nil {
|
|
queue = append(queue, req)
|
|
}
|
|
|
|
case cancelReq := <-exclusiveTaskCancelCh:
|
|
grantedAlready := locked && activeID == cancelReq.id
|
|
|
|
if !grantedAlready {
|
|
for i, req := range queue {
|
|
if req != nil && req.id == cancelReq.id {
|
|
queue = append(queue[:i], queue[i+1:]...)
|
|
break
|
|
}
|
|
}
|
|
}
|
|
|
|
cancelReq.ack <- grantedAlready
|
|
|
|
case <-exclusiveTaskReleaseCh:
|
|
if locked {
|
|
locked = false
|
|
activeID = 0
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func acquireExclusiveTask(ctx context.Context) error {
|
|
if ctx == nil {
|
|
ctx = context.Background()
|
|
}
|
|
|
|
req := &exclusiveTaskRequest{
|
|
id: exclusiveTaskNextID.Add(1),
|
|
granted: make(chan struct{}),
|
|
}
|
|
|
|
select {
|
|
case exclusiveTaskReqCh <- req:
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
|
|
select {
|
|
case <-req.granted:
|
|
return nil
|
|
|
|
case <-ctx.Done():
|
|
ack := make(chan bool, 1)
|
|
|
|
exclusiveTaskCancelCh <- exclusiveTaskCancelRequest{
|
|
id: req.id,
|
|
ack: ack,
|
|
}
|
|
|
|
grantedAlready := <-ack
|
|
if grantedAlready {
|
|
return nil
|
|
}
|
|
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
|
|
func releaseExclusiveTask() {
|
|
select {
|
|
case exclusiveTaskReleaseCh <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|