~james/core

core/spindle/engine/slot.go -rw-r--r-- 1.7 KiB
452f17b7 — james chore(import): snapshot james.dad.town/core at f70c71c7696a 17 hours ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
package engine

import (
	"context"
	"errors"
	"time"

	"tangled.org/core/spindle/models"
)

var ErrNoWorkflowSlots = errors.New("no workflow slots available")

type WorkflowSlot interface {
	Release()
}

// governs blocking behaviour when acquiring a slot
type AcquireMode int

const (
	// blocks until a slot is free
	Wait AcquireMode = iota
	// fails immediately if full
	// executors use this because the mill owns the backlog
	NoWait
)

type WorkflowSlotter interface {
	AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode AcquireMode) (WorkflowSlot, error)
}

// a slot whose run gets less time than the engine's WorkflowTimeout, zero when it doesn't know
type RunBudgeter interface {
	RunBudget() time.Duration
}

type releaseFunc func()

func (f releaseFunc) Release() {
	if f != nil {
		f()
	}
}

type NoopSlot struct{}

func (NoopSlot) Release() {}

// limit by concurrent workflow count
type SemaphoreSlotter struct {
	slots chan struct{}
}

func NewSemaphoreSlotter(maxConcurrent int) *SemaphoreSlotter {
	if maxConcurrent <= 0 {
		return &SemaphoreSlotter{}
	}
	return &SemaphoreSlotter{slots: make(chan struct{}, maxConcurrent)}
}

func (a *SemaphoreSlotter) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode AcquireMode) (WorkflowSlot, error) {
	if a == nil || a.slots == nil {
		return NoopSlot{}, nil
	}
	if mode == NoWait {
		select {
		case a.slots <- struct{}{}:
			return releaseFunc(func() { <-a.slots }), nil
		default:
			return nil, ErrNoWorkflowSlots
		}
	}
	select {
	case a.slots <- struct{}{}:
		return releaseFunc(func() { <-a.slots }), nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}