Workers v2 (#17)

.github/workflows/release.yml

@@ -26,7 +26,7 @@ jobs:
       - name: Set up Go
         uses: magnetikonline/action-golang-cache@v4
         with:
-          go-version: 1.21.4
+          go-version: ~1.22
           cache-key-suffix: -build
 
       - name: Cache
@@ -59,6 +59,12 @@ jobs:
           echo $VERSION
           echo "version=$VERSION" >> $GITHUB_OUTPUT
 
+#      - name: Setup tmate session
+#        if: ${{ failure() }}
+#        run: |
+#          curl -L https://github.com/tmate-io/tmate/releases/download/2.4.0/tmate-2.4.0-static-linux-amd64.tar.xz | tar -xJ --strip-components 1
+#          ./tmate -F
+
   upload_gcs:
     name: Upload binaries
     needs: [ build ]
@@ -125,7 +131,7 @@ jobs:
       - name: Set up Go
         uses: magnetikonline/action-golang-cache@v4
         with:
-          go-version: ~1.21
+          go-version: ~1.22
           cache-key-suffix: -test
 
       - uses: actions/download-artifact@v3
@@ -170,9 +176,3 @@ jobs:
       - name: Cleanup .heph
         run: |
           rm -rf .heph/cache/test
-
-#      - name: Setup tmate session
-#        if: ${{ failure() }}
-#        run: |
-#          curl -L https://github.com/tmate-io/tmate/releases/download/2.4.0/tmate-2.4.0-static-linux-amd64.tar.xz | tar -xJ --strip-components 1
-#          ./tmate -F

BUILD

@@ -4,7 +4,7 @@ load("//backend/node", "yarn_toolchain")
 
 go_toolchain(
     name = "go",
-    version = "1.21.4",
+    version = "1.22.2",
     architectures = [
         "darwin_amd64",
         "darwin_arm64",

bootstrap/bootstrap.go

@@ -8,6 +8,7 @@ import (
 	"github.com/hephbuild/heph/buildfiles"
 	"github.com/hephbuild/heph/config"
 	"github.com/hephbuild/heph/exprs"
+	"github.com/hephbuild/heph/gitstatus"
 	"github.com/hephbuild/heph/graph"
 	"github.com/hephbuild/heph/hbuiltin"
 	"github.com/hephbuild/heph/hroot"
@@ -23,7 +24,8 @@ import (
 	"github.com/hephbuild/heph/targetrun"
 	"github.com/hephbuild/heph/upgrade"
 	"github.com/hephbuild/heph/utils/finalizers"
-	"github.com/hephbuild/heph/worker"
+	"github.com/hephbuild/heph/worker2"
+	"github.com/pbnjay/memory"
 	"os"
 	"path/filepath"
 	"strings"
@@ -64,7 +66,7 @@ type BootOpts struct {
 	Summary               bool
 	JaegerEndpoint        string
 	DisableCloudTelemetry bool
-	Pool                  *worker.Pool
+	Pool                  *worker2.Engine
 
 	PostBootBase func(bs BaseBootstrap) error
 
@@ -133,13 +135,22 @@ type Bootstrap struct {
 	Observability     *observability.Observability
 	Cloud             Cloud
 	Summary           *obsummary.Summary
-	Pool              *worker.Pool
+	Pool              *worker2.Engine
 	Packages          *packages.Registry
 	BuildFiles        *buildfiles.State
 	Graph             *graph.State
 	PlatformProviders []platform.PlatformProvider
 }
 
+func DefaultScheduler(cpu int) *worker2.ResourceScheduler {
+	return worker2.NewResourceScheduler(map[string]float64{
+		"cpu":    float64(cpu),
+		"memory": float64(memory.TotalMemory()),
+	}, map[string]float64{
+		"cpu": float64(1),
+	})
+}
+
 func Boot(ctx context.Context, opts BootOpts) (Bootstrap, error) {
 	bs := Bootstrap{}
 
@@ -182,7 +193,9 @@ func Boot(ctx context.Context, opts BootOpts) (Bootstrap, error) {
 
 	pool := opts.Pool
 	if pool == nil {
-		pool = worker.NewPool(opts.Workers)
+		pool = worker2.NewEngine()
+		pool.SetDefaultScheduler(DefaultScheduler(opts.Workers))
+		go pool.Run()
 	}
 	bs.Pool = pool
 
@@ -294,21 +307,26 @@ func BootScheduler(ctx context.Context, bs Bootstrap) (*scheduler.Scheduler, err
 	}
 
 	e := scheduler.New(scheduler.Scheduler{
-		Cwd:             bs.Cwd,
-		Root:            bs.Root,
-		Config:          bs.Config,
-		Observability:   bs.Observability,
-		GetFlowID:       getFlowId,
-		LocalCache:      localCache,
-		RemoteCache:     remoteCache,
-		Packages:        bs.Packages,
-		BuildFilesState: bs.BuildFiles,
-		Graph:           bs.Graph,
-		Pool:            bs.Pool,
-		Finalizers:      fins,
-		Runner:          runner,
+		Cwd:               bs.Cwd,
+		Root:              bs.Root,
+		Config:            bs.Config,
+		Observability:     bs.Observability,
+		GetFlowID:         getFlowId,
+		LocalCache:        localCache,
+		RemoteCache:       remoteCache,
+		Packages:          bs.Packages,
+		BuildFilesState:   bs.BuildFiles,
+		Graph:             bs.Graph,
+		Pool:              bs.Pool,
+		BackgroundTracker: worker2.NewRunningTracker(),
+		Finalizers:        fins,
+		Runner:            runner,
 	})
 
+	if bs.Config.Engine.GitCacheHints {
+		e.GitStatus = gitstatus.New(bs.Root.Root.Abs())
+	}
+
 	bs.Finalizers.RegisterWithErr(func(err error) {
 		fins.Run(err)
 	})

bootstrap/config.go

@@ -7,6 +7,7 @@ import (
 	"github.com/hephbuild/heph/hroot"
 	"github.com/hephbuild/heph/log/log"
 	"os"
+	"time"
 )
 
 func BuildConfig(root *hroot.State, profiles []string) (*config.Config, error) {
@@ -16,6 +17,8 @@ func BuildConfig(root *hroot.State, profiles []string) (*config.Config, error) {
 	cfg.CacheHistory = 3
 	cfg.Engine.GC = true
 	cfg.Engine.CacheHints = true
+	cfg.Engine.GitCacheHints = false
+	cfg.ProgressInterval = time.Second
 	cfg.Engine.ParallelCaching = true
 	cfg.Engine.SmartGen = true
 	cfg.CacheOrder = config.CacheOrderLatency

bootstrap/error.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"github.com/hephbuild/heph/log/log"
 	"github.com/hephbuild/heph/targetrun"
-	"github.com/hephbuild/heph/worker"
+	"github.com/hephbuild/heph/worker2"
 	"go.uber.org/multierr"
 	"io"
 	"os"
@@ -42,7 +42,7 @@ func printErrTargetFailed(err error) bool {
 }
 
 func PrintHumanError(err error) {
-	errs := multierr.Errors(worker.CollectRootErrors(err))
+	errs := worker2.CollectRootErrors(err)
 	skippedCount := 0
 	skipSpacing := true
 
@@ -57,18 +57,19 @@ func PrintHumanError(err error) {
 	for _, err := range errs {
 		if printErrTargetFailed(err) {
 			// Printed !
+			continue
+		}
+
+		var jerr worker2.Error
+		if errors.As(err, &jerr) && jerr.Skipped() {
+			skippedCount++
+			skipSpacing = true
+			log.Debugf("skipped: %v", jerr)
 		} else {
-			var jerr worker.JobError
-			if errors.As(err, &jerr) && jerr.Skipped() {
-				skippedCount++
+			for _, err := range multierr.Errors(err) {
 				skipSpacing = true
-				log.Debugf("skipped: %v", jerr)
-			} else {
-				for _, err := range multierr.Errors(err) {
-					skipSpacing = true
-					separate()
-					log.Error(err)
-				}
+				separate()
+				log.Error(err)
 			}
 		}
 	}

bootstrap/error_test.go

@@ -0,0 +1,33 @@
+package bootstrap
+
+import (
+	"errors"
+	"github.com/hephbuild/heph/worker2"
+	"github.com/stretchr/testify/assert"
+	"testing"
+)
+
+func TestCollectRootErrors(t *testing.T) {
+	err := worker2.Error{
+		ID:    2,
+		Name:  "a2",
+		State: worker2.ExecStateSkipped,
+		Err: worker2.Error{
+			ID:    1,
+			Name:  "a1",
+			State: worker2.ExecStateFailed,
+			Err:   errors.New("sad beep bop"),
+		},
+	}
+
+	errs := worker2.CollectRootErrors(err)
+
+	assert.EqualValues(t, []error{
+		worker2.Error{
+			ID:    1,
+			Name:  "a1",
+			State: worker2.ExecStateFailed,
+			Err:   errors.New("sad beep bop"),
+		},
+	}, errs)
+}

bootstrap/rrs.go

@@ -11,7 +11,7 @@ import (
 	"github.com/hephbuild/heph/targetrun"
 	"github.com/hephbuild/heph/utils/ads"
 	"github.com/hephbuild/heph/utils/sets"
-	"github.com/hephbuild/heph/worker/poolwait"
+	"github.com/hephbuild/heph/worker2/poolwait"
 )
 
 var errHasExprDep = errors.New("has expr, bailing out")
@@ -125,7 +125,7 @@ func RunGen(ctx context.Context, e *scheduler.Scheduler, plain bool, filterFacto
 			return err
 		}
 
-		err = poolwait.Wait(ctx, fmt.Sprintf("Gen run %v", i), e.Pool, deps, plain)
+		err = poolwait.Wait(ctx, fmt.Sprintf("Gen run %v", i), e.Pool, deps, plain, e.Config.ProgressInterval)
 		if err != nil {
 			return err
 		}

bootstrap/run.go

@@ -4,13 +4,13 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"github.com/dlsniper/debugger"
 	"github.com/hephbuild/heph/log/log"
 	"github.com/hephbuild/heph/sandbox"
 	"github.com/hephbuild/heph/scheduler"
 	"github.com/hephbuild/heph/specs"
 	"github.com/hephbuild/heph/targetrun"
-	"github.com/hephbuild/heph/worker"
-	"github.com/hephbuild/heph/worker/poolwait"
+	"github.com/hephbuild/heph/worker2/poolwait"
 	"os"
 	"os/exec"
 )
@@ -37,6 +37,12 @@ func Run(ctx context.Context, e *scheduler.Scheduler, rrs targetrun.Requests, ru
 }
 
 func RunMode(ctx context.Context, e *scheduler.Scheduler, rrs targetrun.Requests, runopts RunOpts, inlineSingle bool, mode string, iocfg sandbox.IOConfig) error {
+	debugger.SetLabels(func() []string {
+		return []string{
+			"where", "RunMode",
+		}
+	})
+
 	for i := range rrs {
 		rrs[i].Mode = mode
 	}
@@ -60,32 +66,19 @@ func RunMode(ctx context.Context, e *scheduler.Scheduler, rrs targetrun.Requests
 		inlineRR = &rrs[0]
 	}
 
-	// fgDeps will include deps created inside the scheduled jobs to be waited for in the foreground
-	// The DoneSem() must be called after all the tdeps have finished
-	ctx, fgDeps := poolwait.ContextWithForegroundWaitGroup(ctx)
-	fgDeps.AddSem()
-
 	var skip []specs.Specer
 	if inlineRR != nil {
 		skip = []specs.Specer{inlineRR.Target}
 	}
-	tdepsMap, err := e.ScheduleTargetRRsWithDeps(ctx, rrs, skip)
+	tdepsMap, tracker, err := e.ScheduleTargetRRsWithDeps(ctx, rrs, skip)
 	if err != nil {
-		fgDeps.DoneSem()
 		return err
 	}
 
 	tdeps := tdepsMap.All()
-	go func() {
-		<-tdeps.Done()
-		fgDeps.DoneSem()
-	}()
-
-	runDeps := &worker.WaitGroup{}
-	runDeps.AddChild(tdeps)
-	runDeps.AddChild(fgDeps)
+	tdeps.AddDep(tracker.Group())
 
-	err = poolwait.Wait(ctx, "Run", e.Pool, runDeps, runopts.Plain)
+	err = poolwait.Wait(ctx, "Run", e.Pool, tdeps, runopts.Plain, e.Config.ProgressInterval)
 	if err != nil {
 		return err
 	}
@@ -119,7 +112,7 @@ func RunMode(ctx context.Context, e *scheduler.Scheduler, rrs targetrun.Requests
 		iocfg.Stdout = os.Stderr
 	}
 
-	err = e.RunWithSpan(ctx, *inlineRR, iocfg)
+	err = e.RunWithSpan(ctx, *inlineRR, iocfg, nil)
 	if err != nil {
 		var eerr *exec.ExitError
 		if errors.As(err, &eerr) {