lotus/extern/sector-storage/sched.go

573 lines
13 KiB
Go
Raw Normal View History

package sectorstorage
2020-03-23 11:40:02 +00:00
import (
"context"
2020-07-09 10:58:52 +00:00
"math/rand"
"sort"
"sync"
2020-06-23 09:42:47 +00:00
"time"
"github.com/google/uuid"
2020-03-23 11:40:02 +00:00
"golang.org/x/xerrors"
2020-09-07 03:49:10 +00:00
"github.com/filecoin-project/go-state-types/abi"
"github.com/filecoin-project/specs-storage/storage"
"github.com/filecoin-project/lotus/extern/sector-storage/sealtasks"
"github.com/filecoin-project/lotus/extern/sector-storage/storiface"
2020-03-23 11:40:02 +00:00
)
2020-06-24 21:06:56 +00:00
type schedPrioCtxKey int
var SchedPriorityKey schedPrioCtxKey
var DefaultSchedPriority = 0
2020-07-09 10:58:52 +00:00
var SelectorTimeout = 5 * time.Second
var InitWait = 3 * time.Second
2020-07-09 10:58:52 +00:00
var (
SchedWindows = 2
)
2020-06-24 21:06:56 +00:00
func getPriority(ctx context.Context) int {
sp := ctx.Value(SchedPriorityKey)
if p, ok := sp.(int); ok {
return p
}
return DefaultSchedPriority
}
func WithPriority(ctx context.Context, priority int) context.Context {
return context.WithValue(ctx, SchedPriorityKey, priority)
}
2020-03-23 11:40:02 +00:00
const mib = 1 << 20
type WorkerAction func(ctx context.Context, w Worker) error
type WorkerSelector interface {
2020-06-15 12:32:17 +00:00
Ok(ctx context.Context, task sealtasks.TaskType, spt abi.RegisteredSealProof, a *workerHandle) (bool, error) // true if worker is acceptable for performing a task
Cmp(ctx context.Context, task sealtasks.TaskType, a, b *workerHandle) (bool, error) // true if a is preferred over b
}
type scheduler struct {
workersLk sync.RWMutex
workers map[WorkerID]*workerHandle
2020-05-01 18:00:17 +00:00
2020-07-09 10:58:52 +00:00
schedule chan *workerRequest
windowRequests chan *schedWindowRequest
workerChange chan struct{} // worker added / changed/freed resources
workerDisable chan workerDisableReq
2020-07-09 10:58:52 +00:00
// owned by the sh.runSched goroutine
schedQueue *requestQueue
openWindows []*schedWindowRequest
2020-10-28 13:23:38 +00:00
workTracker *workTracker
2020-09-23 12:56:37 +00:00
info chan func(interface{})
2020-07-16 23:32:49 +00:00
closing chan struct{}
2020-07-17 10:59:12 +00:00
closed chan struct{}
2020-07-16 23:26:55 +00:00
testSync chan struct{} // used for testing
2020-07-09 10:58:52 +00:00
}
type workerHandle struct {
2020-10-28 13:23:38 +00:00
workerRpc Worker
2020-07-09 10:58:52 +00:00
info storiface.WorkerInfo
preparing *activeResources // use with workerHandle.lk
active *activeResources // use with workerHandle.lk
2020-07-17 10:59:12 +00:00
lk sync.Mutex // can be taken inside sched.workersLk.RLock
2020-08-03 12:18:11 +00:00
wndLk sync.Mutex // can be taken inside sched.workersLk.RLock
activeWindows []*schedWindow
enabled bool
2020-07-21 18:01:25 +00:00
2020-07-17 10:59:12 +00:00
// for sync manager goroutine closing
cleanupStarted bool
closedMgr chan struct{}
closingMgr chan struct{}
2020-07-09 10:58:52 +00:00
}
type schedWindowRequest struct {
worker WorkerID
done chan *schedWindow
}
type schedWindow struct {
2020-07-09 12:40:53 +00:00
allocated activeResources
2020-07-09 10:58:52 +00:00
todo []*workerRequest
}
type workerDisableReq struct {
activeWindows []*schedWindow
wid WorkerID
done func()
}
2020-07-09 10:58:52 +00:00
type activeResources struct {
memUsedMin uint64
memUsedMax uint64
gpuUsed bool
cpuUse uint64
2021-09-15 14:37:27 +00:00
cond *sync.Cond
waiting int
2020-07-09 10:58:52 +00:00
}
type workerRequest struct {
sector storage.SectorRef
2020-07-09 10:58:52 +00:00
taskType sealtasks.TaskType
priority int // larger values more important
sel WorkerSelector
prepare WorkerAction
work WorkerAction
start time.Time
2020-07-09 10:58:52 +00:00
index int // The index of the item in the heap.
2020-07-27 06:21:29 +00:00
indexHeap int
2020-07-27 11:21:36 +00:00
ret chan<- workerResponse
ctx context.Context
2020-07-09 10:58:52 +00:00
}
2020-07-09 10:58:52 +00:00
type workerResponse struct {
err error
}
func newScheduler() *scheduler {
return &scheduler{
workers: map[WorkerID]*workerHandle{},
2020-05-01 18:00:17 +00:00
2020-07-09 12:40:53 +00:00
schedule: make(chan *workerRequest),
windowRequests: make(chan *schedWindowRequest, 20),
workerChange: make(chan struct{}, 20),
workerDisable: make(chan workerDisableReq),
2020-05-07 23:38:05 +00:00
schedQueue: &requestQueue{},
2020-07-09 12:40:53 +00:00
2020-10-28 13:23:38 +00:00
workTracker: &workTracker{
2020-09-23 12:56:37 +00:00
done: map[storiface.CallID]struct{}{},
running: map[storiface.CallID]trackedWork{},
},
info: make(chan func(interface{})),
2020-07-09 12:40:53 +00:00
closing: make(chan struct{}),
2020-07-17 10:59:12 +00:00
closed: make(chan struct{}),
}
}
func (sh *scheduler) Schedule(ctx context.Context, sector storage.SectorRef, taskType sealtasks.TaskType, sel WorkerSelector, prepare WorkerAction, work WorkerAction) error {
ret := make(chan workerResponse)
select {
case sh.schedule <- &workerRequest{
sector: sector,
taskType: taskType,
2020-06-24 21:06:56 +00:00
priority: getPriority(ctx),
sel: sel,
prepare: prepare,
work: work,
start: time.Now(),
ret: ret,
ctx: ctx,
}:
case <-sh.closing:
return xerrors.New("closing")
case <-ctx.Done():
return ctx.Err()
}
select {
case resp := <-ret:
return resp.err
case <-sh.closing:
return xerrors.New("closing")
case <-ctx.Done():
return ctx.Err()
}
}
func (r *workerRequest) respond(err error) {
2020-03-23 11:40:02 +00:00
select {
case r.ret <- workerResponse{err: err}:
case <-r.ctx.Done():
2020-03-23 11:40:02 +00:00
log.Warnf("request got cancelled before we could respond")
}
}
type SchedDiagRequestInfo struct {
Sector abi.SectorID
TaskType sealtasks.TaskType
Priority int
}
type SchedDiagInfo struct {
2020-07-27 11:21:36 +00:00
Requests []SchedDiagRequestInfo
OpenWindows []string
}
func (sh *scheduler) runSched() {
2020-07-17 10:59:12 +00:00
defer close(sh.closed)
iw := time.After(InitWait)
var initialised bool
2020-03-23 11:40:02 +00:00
for {
var doSched bool
var toDisable []workerDisableReq
2020-03-23 11:40:02 +00:00
select {
case <-sh.workerChange:
doSched = true
case dreq := <-sh.workerDisable:
toDisable = append(toDisable, dreq)
doSched = true
2020-07-09 10:58:52 +00:00
case req := <-sh.schedule:
sh.schedQueue.Push(req)
doSched = true
2020-07-09 10:58:52 +00:00
2020-07-16 23:26:55 +00:00
if sh.testSync != nil {
sh.testSync <- struct{}{}
}
2020-07-09 10:58:52 +00:00
case req := <-sh.windowRequests:
sh.openWindows = append(sh.openWindows, req)
doSched = true
case ireq := <-sh.info:
ireq(sh.diag())
case <-iw:
initialised = true
iw = nil
doSched = true
case <-sh.closing:
sh.schedClose()
2020-03-24 23:49:45 +00:00
return
2020-03-23 11:40:02 +00:00
}
if doSched && initialised {
// First gather any pending tasks, so we go through the scheduling loop
// once for every added task
loop:
for {
select {
case <-sh.workerChange:
case dreq := <-sh.workerDisable:
toDisable = append(toDisable, dreq)
case req := <-sh.schedule:
sh.schedQueue.Push(req)
if sh.testSync != nil {
sh.testSync <- struct{}{}
}
case req := <-sh.windowRequests:
sh.openWindows = append(sh.openWindows, req)
default:
break loop
}
}
for _, req := range toDisable {
for _, window := range req.activeWindows {
for _, request := range window.todo {
sh.schedQueue.Push(request)
}
}
openWindows := make([]*schedWindowRequest, 0, len(sh.openWindows))
for _, window := range sh.openWindows {
if window.worker != req.wid {
openWindows = append(openWindows, window)
}
}
sh.openWindows = openWindows
sh.workersLk.Lock()
sh.workers[req.wid].enabled = false
sh.workersLk.Unlock()
req.done()
}
sh.trySched()
}
2020-03-23 11:40:02 +00:00
}
}
func (sh *scheduler) diag() SchedDiagInfo {
var out SchedDiagInfo
for sqi := 0; sqi < sh.schedQueue.Len(); sqi++ {
task := (*sh.schedQueue)[sqi]
out.Requests = append(out.Requests, SchedDiagRequestInfo{
Sector: task.sector.ID,
TaskType: task.taskType,
Priority: task.priority,
})
}
sh.workersLk.RLock()
defer sh.workersLk.RUnlock()
for _, window := range sh.openWindows {
out.OpenWindows = append(out.OpenWindows, uuid.UUID(window.worker).String())
}
return out
}
2020-07-09 10:58:52 +00:00
func (sh *scheduler) trySched() {
/*
This assigns tasks to workers based on:
- Task priority (achieved by handling sh.schedQueue in order, since it's already sorted by priority)
- Worker resource availability
- Task-specified worker preference (acceptableWindows array below sorted by this preference)
- Window request age
2020-05-01 18:00:17 +00:00
2020-07-09 10:58:52 +00:00
1. For each task in the schedQueue find windows which can handle them
1.1. Create list of windows capable of handling a task
1.2. Sort windows according to task selector preferences
2. Going through schedQueue again, assign task to first acceptable window
with resources available
3. Submit windows with scheduled tasks to workers
2020-07-09 10:58:52 +00:00
*/
2020-08-03 12:18:11 +00:00
sh.workersLk.RLock()
defer sh.workersLk.RUnlock()
2020-09-25 14:41:29 +00:00
windowsLen := len(sh.openWindows)
queueLen := sh.schedQueue.Len()
2020-09-25 14:41:29 +00:00
log.Debugf("SCHED %d queued; %d open windows", queueLen, windowsLen)
2020-09-25 14:41:29 +00:00
if windowsLen == 0 || queueLen == 0 {
// nothing to schedule on
return
}
2020-08-03 12:18:11 +00:00
2020-09-25 14:41:29 +00:00
windows := make([]schedWindow, windowsLen)
acceptableWindows := make([][]int, queueLen)
2020-09-25 14:41:29 +00:00
2020-07-09 10:58:52 +00:00
// Step 1
2020-09-25 14:41:29 +00:00
throttle := make(chan struct{}, windowsLen)
var wg sync.WaitGroup
wg.Add(queueLen)
for i := 0; i < queueLen; i++ {
throttle <- struct{}{}
go func(sqi int) {
defer wg.Done()
defer func() {
<-throttle
}()
task := (*sh.schedQueue)[sqi]
needRes := ResourceTable[task.taskType][task.sector.ProofType]
task.indexHeap = sqi
for wnd, windowRequest := range sh.openWindows {
worker, ok := sh.workers[windowRequest.worker]
if !ok {
log.Errorf("worker referenced by windowRequest not found (worker: %s)", windowRequest.worker)
// TODO: How to move forward here?
continue
}
if !worker.enabled {
log.Debugw("skipping disabled worker", "worker", windowRequest.worker)
continue
}
// TODO: allow bigger windows
2021-06-21 19:49:16 +00:00
if !windows[wnd].allocated.canHandleRequest(needRes, windowRequest.worker, "schedAcceptable", worker.info) {
continue
}
rpcCtx, cancel := context.WithTimeout(task.ctx, SelectorTimeout)
ok, err := task.sel.Ok(rpcCtx, task.taskType, task.sector.ProofType, worker)
cancel()
if err != nil {
log.Errorf("trySched(1) req.sel.Ok error: %+v", err)
continue
}
if !ok {
continue
}
acceptableWindows[sqi] = append(acceptableWindows[sqi], wnd)
2020-07-09 10:58:52 +00:00
}
if len(acceptableWindows[sqi]) == 0 {
return
2020-07-09 10:58:52 +00:00
}
2020-06-23 09:42:47 +00:00
// Pick best worker (shuffle in case some workers are equally as good)
rand.Shuffle(len(acceptableWindows[sqi]), func(i, j int) {
acceptableWindows[sqi][i], acceptableWindows[sqi][j] = acceptableWindows[sqi][j], acceptableWindows[sqi][i] // nolint:scopelint
})
sort.SliceStable(acceptableWindows[sqi], func(i, j int) bool {
wii := sh.openWindows[acceptableWindows[sqi][i]].worker // nolint:scopelint
wji := sh.openWindows[acceptableWindows[sqi][j]].worker // nolint:scopelint
if wii == wji {
// for the same worker prefer older windows
return acceptableWindows[sqi][i] < acceptableWindows[sqi][j] // nolint:scopelint
}
wi := sh.workers[wii]
wj := sh.workers[wji]
rpcCtx, cancel := context.WithTimeout(task.ctx, SelectorTimeout)
defer cancel()
r, err := task.sel.Cmp(rpcCtx, task.taskType, wi, wj)
if err != nil {
2020-11-24 11:09:48 +00:00
log.Errorf("selecting best worker: %s", err)
}
return r
})
}(i)
}
wg.Wait()
log.Debugf("SCHED windows: %+v", windows)
log.Debugf("SCHED Acceptable win: %+v", acceptableWindows)
2020-07-09 10:58:52 +00:00
// Step 2
scheduled := 0
rmQueue := make([]int, 0, queueLen)
for sqi := 0; sqi < queueLen; sqi++ {
2020-07-09 10:58:52 +00:00
task := (*sh.schedQueue)[sqi]
needRes := ResourceTable[task.taskType][task.sector.ProofType]
2020-07-09 10:58:52 +00:00
selectedWindow := -1
2020-07-27 06:21:29 +00:00
for _, wnd := range acceptableWindows[task.indexHeap] {
2020-07-09 10:58:52 +00:00
wid := sh.openWindows[wnd].worker
2021-06-21 19:49:16 +00:00
info := sh.workers[wid].info
log.Debugf("SCHED try assign sqi:%d sector %d to window %d", sqi, task.sector.ID.Number, wnd)
2020-07-16 23:26:55 +00:00
2020-07-09 10:58:52 +00:00
// TODO: allow bigger windows
2021-06-21 19:49:16 +00:00
if !windows[wnd].allocated.canHandleRequest(needRes, wid, "schedAssign", info) {
2020-07-09 10:58:52 +00:00
continue
}
log.Debugf("SCHED ASSIGNED sqi:%d sector %d task %s to window %d", sqi, task.sector.ID.Number, task.taskType, wnd)
2020-07-16 23:26:55 +00:00
2021-06-21 19:49:16 +00:00
windows[wnd].allocated.add(info.Resources, needRes)
// TODO: We probably want to re-sort acceptableWindows here based on new
// workerHandle.utilization + windows[wnd].allocated.utilization (workerHandle.utilization is used in all
// task selectors, but not in the same way, so need to figure out how to do that in a non-O(n^2 way), and
// without additional network roundtrips (O(n^2) could be avoided by turning acceptableWindows.[] into heaps))
2020-03-23 11:40:02 +00:00
2020-07-09 10:58:52 +00:00
selectedWindow = wnd
break
}
2020-03-23 11:40:02 +00:00
2020-07-09 12:40:53 +00:00
if selectedWindow < 0 {
// all windows full
continue
}
2020-07-09 10:58:52 +00:00
windows[selectedWindow].todo = append(windows[selectedWindow].todo, task)
2020-03-23 11:40:02 +00:00
2020-09-25 14:13:27 +00:00
rmQueue = append(rmQueue, sqi)
2020-07-09 10:58:52 +00:00
scheduled++
}
2020-03-23 11:40:02 +00:00
2020-09-25 14:13:27 +00:00
if len(rmQueue) > 0 {
2020-09-25 14:59:21 +00:00
for i := len(rmQueue) - 1; i >= 0; i-- {
2020-09-25 14:13:27 +00:00
sh.schedQueue.Remove(rmQueue[i])
}
}
2020-07-09 10:58:52 +00:00
// Step 3
2020-03-23 11:40:02 +00:00
2020-07-09 10:58:52 +00:00
if scheduled == 0 {
return
}
2020-07-09 10:58:52 +00:00
scheduledWindows := map[int]struct{}{}
for wnd, window := range windows {
if len(window.todo) == 0 {
// Nothing scheduled here, keep the window open
continue
}
2020-07-09 10:58:52 +00:00
scheduledWindows[wnd] = struct{}{}
2020-07-09 12:40:53 +00:00
window := window // copy
2020-07-09 10:58:52 +00:00
select {
case sh.openWindows[wnd].done <- &window:
default:
log.Error("expected sh.openWindows[wnd].done to be buffered")
}
2020-07-09 10:58:52 +00:00
}
2020-07-09 10:58:52 +00:00
// Rewrite sh.openWindows array, removing scheduled windows
2020-09-25 14:41:29 +00:00
newOpenWindows := make([]*schedWindowRequest, 0, windowsLen-len(scheduledWindows))
2020-07-09 10:58:52 +00:00
for wnd, window := range sh.openWindows {
2020-07-09 17:17:15 +00:00
if _, scheduled := scheduledWindows[wnd]; scheduled {
2020-07-09 10:58:52 +00:00
// keep unscheduled windows open
continue
}
2020-07-09 10:58:52 +00:00
newOpenWindows = append(newOpenWindows, window)
}
2020-07-09 10:58:52 +00:00
sh.openWindows = newOpenWindows
}
func (sh *scheduler) schedClose() {
sh.workersLk.Lock()
defer sh.workersLk.Unlock()
2020-07-17 10:59:12 +00:00
log.Debugf("closing scheduler")
2020-03-24 23:49:45 +00:00
for i, w := range sh.workers {
2020-07-17 10:59:12 +00:00
sh.workerCleanup(i, w)
2020-03-24 23:49:45 +00:00
}
}
func (sh *scheduler) Info(ctx context.Context) (interface{}, error) {
ch := make(chan interface{}, 1)
sh.info <- func(res interface{}) {
ch <- res
}
select {
2020-07-27 11:21:36 +00:00
case res := <-ch:
return res, nil
case <-ctx.Done():
return nil, ctx.Err()
}
}
2020-07-17 10:59:12 +00:00
func (sh *scheduler) Close(ctx context.Context) error {
close(sh.closing)
2020-07-17 10:59:12 +00:00
select {
case <-sh.closed:
case <-ctx.Done():
return ctx.Err()
}
return nil
}