alice.pds.demo.boltless.dev
This repository has no description
1package engine
2
3import (
4 "context"
5 "errors"
6
7 "tangled.org/core/spindle/models"
8)
9
10var ErrNoWorkflowSlots = errors.New("no workflow slots available")
11
12type WorkflowSlot interface {
13 Release()
14}
15
16// governs blocking behaviour when acquiring a slot
17type AcquireMode int
18
19const (
20 // blocks until a slot is free
21 Wait AcquireMode = iota
22 // fails immediately if full
23 // executors use this because the mill owns the backlog
24 NoWait
25)
26
27type WorkflowSlotter interface {
28 AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode AcquireMode) (WorkflowSlot, error)
29}
30
31type releaseFunc func()
32
33func (f releaseFunc) Release() {
34 if f != nil {
35 f()
36 }
37}
38
39type NoopSlot struct{}
40
41func (NoopSlot) Release() {}
42
43// limit by concurrent workflow count
44type SemaphoreSlotter struct {
45 slots chan struct{}
46}
47
48func NewSemaphoreSlotter(maxConcurrent int) *SemaphoreSlotter {
49 if maxConcurrent <= 0 {
50 return &SemaphoreSlotter{}
51 }
52 return &SemaphoreSlotter{slots: make(chan struct{}, maxConcurrent)}
53}
54
55func (a *SemaphoreSlotter) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode AcquireMode) (WorkflowSlot, error) {
56 if a == nil || a.slots == nil {
57 return NoopSlot{}, nil
58 }
59 if mode == NoWait {
60 select {
61 case a.slots <- struct{}{}:
62 return releaseFunc(func() { <-a.slots }), nil
63 default:
64 return nil, ErrNoWorkflowSlots
65 }
66 }
67 select {
68 case a.slots <- struct{}{}:
69 return releaseFunc(func() { <-a.slots }), nil
70 case <-ctx.Done():
71 return nil, ctx.Err()
72 }
73}