Cooperative Event Scheduling

Author: teddyknox

op-node/node/node.go

@@ -173,7 +173,7 @@ func (n *OpNode) init(ctx context.Context, cfg *config.Config) error {
 
 func (n *OpNode) initEventSystem() {
 	// This executor will be configurable in the future, for parallel event processing
-	executor := event.NewGlobalSynchronous(n.resourcesCtx).WithMetrics(n.metrics)
+	executor := event.NewCooperative(n.resourcesCtx).WithMetrics(n.metrics)
 	sys := event.NewSystem(n.log, executor)
 	sys.AddTracer(event.NewMetricsTracer(n.metrics))
 	sys.Register("node", event.DeriverFunc(n.onEvent))

op-service/event/async_fn.go

@@ -0,0 +1,197 @@
+package event
+
+import (
+	"context"
+	"fmt"
+	"strconv"
+	"sync"
+	"sync/atomic"
+)
+
+// Async arities with explicit error type parameter E.
+type Async0[E interface{ error }] func(ctx context.Context) E
+type Async1[T any, E interface{ error }] func(ctx context.Context) (T, E)
+type Async2[T1 any, T2 any, E interface{ error }] func(ctx context.Context) (T1, T2, E)
+
+// asyncInvokeEvent schedules a function by ID via the cooperative loop.
+type asyncInvokeEvent struct{ id string }
+
+func (asyncInvokeEvent) String() string { return "async-invoke" }
+
+// taskRunner receives asyncInvokeEvent and runs the associated function.
+type taskRunner struct {
+	exec  *CooperativeExec
+	em    Emitter
+	sys   System
+	mu    sync.Mutex
+	tasks map[string]func(ctx context.Context)
+	ctr   atomic.Uint64
+}
+
+func (tr *taskRunner) nextID() string { return strconv.FormatUint(tr.ctr.Add(1), 10) }
+
+func (tr *taskRunner) OnEvent(ctx context.Context, ev Event) bool {
+	a, ok := ev.(asyncInvokeEvent)
+	if !ok {
+		return false
+	}
+	tr.mu.Lock()
+	fn := tr.tasks[a.id]
+	if fn == nil {
+		tr.mu.Unlock()
+		return false
+	}
+	delete(tr.tasks, a.id)
+	tr.mu.Unlock()
+	// ensure the context is system-aware for ctx-only APIs
+	ctx = WithSystem(ctx, tr.sys)
+	fn(ctx)
+	return true
+}
+
+var (
+	runnersMu sync.Mutex
+	// runners is keyed by System, then by Priority. This allows us to respect spawn priority
+	// while keeping the ability to collapse to a single Normal-priority runner.
+	runners = make(map[*Sys]map[Priority]*taskRunner)
+	// Toggle to enable/disable respecting SpawnN priority. If false, all spawns use Normal.
+	respectSpawnPriority = true
+)
+
+// SetSpawnPriorityEnabled toggles whether SpawnN respects the provided priority (default: true).
+func SetSpawnPriorityEnabled(enabled bool) { respectSpawnPriority = enabled }
+
+func ensureTaskRunner(s System, pr Priority) (*taskRunner, error) {
+	sys, ok := s.(*Sys)
+	if !ok {
+		return nil, fmt.Errorf("unexpected system type")
+	}
+	exec, ok := sys.executor.(*CooperativeExec)
+	if !ok {
+		return nil, fmt.Errorf("cooperative executor required")
+	}
+	// Optionally collapse to Normal priority runner.
+	effectivePr := pr
+	if !respectSpawnPriority {
+		effectivePr = Normal
+	}
+
+	runnersMu.Lock()
+	defer runnersMu.Unlock()
+	byPr, ok := runners[sys]
+	if !ok {
+		byPr = make(map[Priority]*taskRunner)
+		runners[sys] = byPr
+	}
+	if tr := byPr[effectivePr]; tr != nil {
+		return tr, nil
+	}
+	// Register a new runner with both executor and emitter priorities set.
+	tr := &taskRunner{exec: exec, sys: sys, tasks: make(map[string]func(context.Context))}
+	name := "async-runner-" + strconv.Itoa(int(effectivePr))
+	em := sys.Register(name, tr, WithExecPriority(effectivePr), WithEmitPriority(effectivePr))
+	tr.em = em
+	byPr[effectivePr] = tr
+	return tr, nil
+}
+
+// Spawn options to configure priority and name (optional).
+type SpawnOption func(*SpawnConfig)
+
+type SpawnConfig struct {
+	Priority Priority
+	Name     string
+}
+
+func defaultSpawnConfig() *SpawnConfig {
+	return &SpawnConfig{Priority: Normal, Name: ""}
+}
+
+func WithSpawnPriority(p Priority) SpawnOption { return func(c *SpawnConfig) { c.Priority = p } }
+func WithSpawnName(name string) SpawnOption    { return func(c *SpawnConfig) { c.Name = name } }
+
+// Context-based spawn helpers that derive the System from the context.
+// Spawn0 schedules f and returns a Promise0.
+func Spawn0[E interface{ error }](ctx context.Context, f Async0[E], opts ...SpawnOption) (Promise0[E], error) {
+	sys, ok := SystemFromContext(ctx)
+	if !ok {
+		var p Promise0[E]
+		return p, fmt.Errorf("no system in context")
+	}
+	// implement using taskRunner, resolving the promise on completion
+	cfg := defaultSpawnConfig()
+	for _, o := range opts {
+		o(cfg)
+	}
+	tr, _ := ensureTaskRunner(sys, cfg.Priority)
+	id := tr.nextID()
+	p, r := NewPromise0[E]()
+	tr.mu.Lock()
+	tr.tasks[id] = func(ctx context.Context) {
+		err := f(ctx)
+		if any(err) != nil {
+			r.Reject(err)
+			return
+		}
+		r.Resolve()
+	}
+	tr.mu.Unlock()
+	tr.em.Emit(ctx, asyncInvokeEvent{id: id})
+	return p, nil
+}
+
+// Spawn1 schedules f and returns a Promise1.
+func Spawn1[T any, E interface{ error }](ctx context.Context, f Async1[T, E], opts ...SpawnOption) (Promise1[T, E], error) {
+	sys, ok := SystemFromContext(ctx)
+	if !ok {
+		var p Promise1[T, E]
+		return p, fmt.Errorf("no system in context")
+	}
+	cfg := defaultSpawnConfig()
+	for _, o := range opts {
+		o(cfg)
+	}
+	tr, _ := ensureTaskRunner(sys, cfg.Priority)
+	id := tr.nextID()
+	p, r := NewPromise1[T, E]()
+	tr.mu.Lock()
+	tr.tasks[id] = func(ctx context.Context) {
+		t, e := f(ctx)
+		if any(e) != nil {
+			r.Reject(e)
+			return
+		}
+		r.Resolve(t)
+	}
+	tr.mu.Unlock()
+	tr.em.Emit(ctx, asyncInvokeEvent{id: id})
+	return p, nil
+}
+
+// Spawn2 schedules f and returns a Promise2.
+func Spawn2[T1 any, T2 any, E interface{ error }](ctx context.Context, f Async2[T1, T2, E], opts ...SpawnOption) (Promise2[T1, T2, E], error) {
+	sys, ok := SystemFromContext(ctx)
+	if !ok {
+		var p Promise2[T1, T2, E]
+		return p, fmt.Errorf("no system in context")
+	}
+	cfg := defaultSpawnConfig()
+	for _, o := range opts {
+		o(cfg)
+	}
+	tr, _ := ensureTaskRunner(sys, cfg.Priority)
+	id := tr.nextID()
+	p, r := NewPromise2[T1, T2, E]()
+	tr.mu.Lock()
+	tr.tasks[id] = func(ctx context.Context) {
+		a, b, e := f(ctx)
+		if any(e) != nil {
+			r.Reject(e)
+			return
+		}
+		r.Resolve(a, b)
+	}
+	tr.mu.Unlock()
+	tr.em.Emit(ctx, asyncInvokeEvent{id: id})
+	return p, nil
+}