Remove concept of worker

worker2/engine.go

@@ -15,7 +15,7 @@ type Engine struct {
 	execUid          uint64
 	wg               sync.WaitGroup
 	defaultScheduler Scheduler
-	workers          []*Worker
+	liveExecs        []*Execution
 	m                sync.RWMutex
 	eventsCh         chan Event
 	hooks            []Hook
@@ -34,11 +34,11 @@ func (e *Engine) SetDefaultScheduler(s Scheduler) {
 	e.defaultScheduler = s
 }
 
-func (e *Engine) GetWorkers() []*Worker {
+func (e *Engine) GetLiveExecutions() []*Execution {
 	e.m.Lock()
 	defer e.m.Unlock()
 
-	return e.workers[:]
+	return e.liveExecs[:]
 }
 
 func (e *Engine) loop() {
@@ -62,6 +62,11 @@ func (e *Engine) handle(event Event) {
 			e.finalize(event.Execution, ExecStateSucceeded, nil)
 		}
 		e.runHooks(event, event.Execution)
+	case EventSuspended:
+		go func() {
+			<-event.Bag.WaitResume()
+			e.queue(event.Execution)
+		}()
 	default:
 		if event, ok := event.(WithExecution); ok {
 			defer e.runHooks(event, event.getExecution())
@@ -257,24 +262,15 @@ func (e *Engine) start(exec *Execution) {
 	e.m.Lock()
 	defer e.m.Unlock()
 
-	w := &Worker{
-		ctx:  exec.Dep.GetCtx(),
-		exec: exec,
-		queue: func() {
-			e.queue(exec)
-		},
-	}
-	e.workers = append(e.workers, w)
+	e.liveExecs = append(e.liveExecs, exec)
 
 	go func() {
-		w.Run()
+		exec.Run()
 
 		e.m.Lock()
 		defer e.m.Unlock()
 
-		e.workers = ads.Filter(e.workers, func(worker *Worker) bool {
-			return worker != w
-		})
+		e.liveExecs = ads.Remove(e.liveExecs, exec)
 	}()
 
 	e.runHooks(EventStarted{Execution: exec}, exec)

worker2/engine_test.go

@@ -130,7 +130,7 @@ func TestStatus(t *testing.T) {
 	<-emittedCh
 
 	var emittedStatus status.Statuser
-	for _, worker := range e.GetWorkers() {
+	for _, worker := range e.GetLiveExecutions() {
 		emittedStatus = worker.status
 		if emittedStatus != nil {
 			break
@@ -664,7 +664,7 @@ func TestSuspend(t *testing.T) {
 	assert.Equal(t, "end_wait", <-logCh)
 	assert.Equal(t, "leave", <-logCh)
 
-	<-e.Wait()
+	<-a.Wait()
 	close(eventCh)
 
 	events := make([]string, 0)
@@ -702,15 +702,23 @@ func TestSuspendStress(t *testing.T) {
 	go e.Run()
 	defer e.Stop()
 
+	doneCh := make(chan struct{})
+	defer close(doneCh)
+
 	go func() {
 		for {
-			time.Sleep(time.Second)
-			t.Log(done)
+			select {
+			case <-doneCh:
+				return
+			case <-time.After(time.Second):
+			}
+
+			t.Log(atomic.LoadInt64(&done))
 		}
 	}()
 
 	wg.Wait()
-	time.Sleep(time.Second) // todo figure out why things are trying to send events after the pool is stopped
+	<-e.Wait()
 }
 
 func TestSuspendLimit(t *testing.T) {

worker2/events.go

@@ -68,6 +68,7 @@ func (e EventReady) getExecution() *Execution {
 type EventSuspended struct {
 	At        time.Time
 	Execution *Execution
+	Bag       *SuspendBag
 }
 
 func (e EventSuspended) getExecution() *Execution {

worker2/execution.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"fmt"
 	"github.com/dlsniper/debugger"
+	"github.com/hephbuild/heph/status"
 	"runtime/debug"
 	"strconv"
 	"sync"
@@ -74,6 +75,8 @@ type Execution struct {
 	StartedAt   time.Time
 	QueuedAt    time.Time
 
+	status status.Statuser
+
 	debugString string
 }
 
@@ -93,15 +96,7 @@ func (e *Execution) GetOutput() Value {
 	return e.outStore.Get()
 }
 
-type ErrSuspended struct {
-	Bag *SuspendBag
-}
-
-func (e ErrSuspended) Error() string {
-	return "suspended"
-}
-
-func (e *Execution) Run(ctx context.Context) error {
+func (e *Execution) Run() {
 	e.m.Lock()
 	if e.errCh == nil {
 		e.errCh = make(chan error)
@@ -114,6 +109,10 @@ func (e *Execution) Run(ctx context.Context) error {
 		e.StartedAt = time.Now()
 
 		go func() {
+			ctx := e.Dep.GetCtx()
+			ctx = contextWithExecution(ctx, e)
+			ctx = status.ContextWithHandler(ctx, e)
+
 			err := e.run(ctx)
 			e.errCh <- err
 		}()
@@ -125,10 +124,19 @@ func (e *Execution) Run(ctx context.Context) error {
 
 	select {
 	case sb := <-e.WaitSuspend():
+		e.scheduler.Done(e.Dep)
+
 		e.State = ExecStateSuspended
-		return ErrSuspended{Bag: sb}
+
+		e.eventsCh <- EventSuspended{Execution: e, Bag: sb}
 	case err := <-e.errCh:
-		return err
+		e.scheduler.Done(e.Dep)
+
+		e.eventsCh <- EventCompleted{
+			Execution: e,
+			Output:    e.outStore.Get(),
+			Error:     err,
+		}
 	}
 }
 
@@ -221,3 +229,19 @@ func (e *Execution) ResumeAck() {
 func (e *Execution) WaitSuspend() chan *SuspendBag {
 	return e.suspendCh
 }
+
+func (e *Execution) Status(status status.Statuser) {
+	e.status = status
+}
+
+func (e *Execution) GetStatus() status.Statuser {
+	s := e.status
+	if s == nil {
+		s = status.String("")
+	}
+	return s
+}
+
+func (e *Execution) Interactive() bool {
+	return true
+}

worker2/poolui/tui.go

@@ -78,12 +78,7 @@ func (m *Model) updateMsg(final bool) UpdateMessage {
 	}
 
 	var workers []workerEntry
-	for _, w := range m.pool.GetWorkers() {
-		exec := w.Execution()
-		if exec == nil {
-			continue
-		}
-
+	for _, exec := range m.pool.GetLiveExecutions() {
 		if _, ok := exec.Dep.(*worker2.Group); ok {
 			continue
 		}
@@ -98,7 +93,7 @@ func (m *Model) updateMsg(final bool) UpdateMessage {
 		}
 
 		workers = append(workers, workerEntry{
-			status:   w.GetStatus(),
+			status:   exec.GetStatus(),
 			duration: duration,
 			exec:     exec,
 		})

worker2/poolwait/log.go

@@ -47,14 +47,14 @@ func logUI(name string, deps worker2.Dep, pool *worker2.Engine, interval time.Du
 
 	printWorkersStatus := func() {
 		statusm := map[string]struct{}{}
-		for _, job := range pool.GetWorkers() {
-			if job.Execution().State != worker2.ExecStateSuspended {
+		for _, exec := range pool.GetLiveExecutions() {
+			if exec.State != worker2.ExecStateSuspended {
 				continue
 			}
 
-			duration := time.Since(job.Execution().StartedAt)
+			duration := time.Since(exec.StartedAt)
 
-			status := job.GetStatus().String(log.Renderer())
+			status := exec.GetStatus().String(log.Renderer())
 			if status == "" {
 				status = "Suspended..."
 			}
@@ -71,20 +71,15 @@ func logUI(name string, deps worker2.Dep, pool *worker2.Engine, interval time.Du
 		if len(statusm) > 0 {
 			fmt.Fprintf(os.Stderr, "===\n")
 		}
-		for _, w := range pool.GetWorkers() {
-			j := w.Execution()
-			if j == nil {
-				continue
-			}
-
-			duration := time.Since(j.StartedAt)
+		for _, exec := range pool.GetLiveExecutions() {
+			duration := time.Since(exec.StartedAt)
 
 			if duration < 5*time.Second {
 				// Skip printing short jobs
 				continue
 			}
 
-			status := w.GetStatus().String(log.Renderer())
+			status := exec.GetStatus().String(log.Renderer())
 			if status == "" {
 				status = "Waiting..."
 			}