WIP

hephbuild · May 6, 2024 · 1154162 · 1154162
1 parent 6d0f970
commit 1154162
Show file tree

Hide file tree

Showing 18 changed files with 184 additions and 146 deletions.
diff --git a/bootstrap/bootstrap.go b/bootstrap/bootstrap.go
@@ -25,7 +25,7 @@ import (
 	"github.com/hephbuild/heph/upgrade"
 	"github.com/hephbuild/heph/utils/finalizers"
 	"github.com/hephbuild/heph/worker2"
-	"github.com/pbnjay/memory"
+	"maps"
 	"os"
 	"path/filepath"
 	"strings"
@@ -59,17 +59,20 @@ func findRoot(cwd string) (string, error) {
 	return "", errors.New("root not found, are you running this command in the repo directory?")
 }
 
+type BootBaseOpts struct {
+	Profiles     []string
+	Params       map[string]string
+	PostBootBase func(bs BaseBootstrap) error
+}
+
 type BootOpts struct {
-	Profiles              []string
+	BootBaseOpts
 	Workers               int
-	Params                map[string]string
 	Summary               bool
 	JaegerEndpoint        string
 	DisableCloudTelemetry bool
 	Pool                  *worker2.Engine
 
-	PostBootBase func(bs BaseBootstrap) error
-
 	FlowID string
 }
 
@@ -79,7 +82,7 @@ type BaseBootstrap struct {
 	Config *config.Config
 }
 
-func BootBase(ctx context.Context, opts BootOpts) (BaseBootstrap, error) {
+func BootBase(ctx context.Context, opts BootBaseOpts) (BaseBootstrap, error) {
 	lipgloss.SetDefaultRenderer(log.Renderer())
 
 	bs := BaseBootstrap{}
@@ -142,19 +145,19 @@ type Bootstrap struct {
 	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{
+func DefaultScheduler(requests map[string]float64, cpu int) *worker2.ResourceScheduler {
+	requests = maps.Clone(requests)
+	requests["cpu"] = float64(cpu)
+
+	return worker2.NewResourceScheduler(requests, map[string]float64{
 		"cpu": float64(1),
 	})
 }
 
 func Boot(ctx context.Context, opts BootOpts) (Bootstrap, error) {
 	bs := Bootstrap{}
 
-	bbs, err := BootBase(ctx, opts)
+	bbs, err := BootBase(ctx, opts.BootBaseOpts)
 	if err != nil {
 		return bs, err
 	}
@@ -194,7 +197,7 @@ func Boot(ctx context.Context, opts BootOpts) (Bootstrap, error) {
 	pool := opts.Pool
 	if pool == nil {
 		pool = worker2.NewEngine()
-		pool.SetDefaultScheduler(DefaultScheduler(opts.Workers))
+		pool.SetDefaultScheduler(DefaultScheduler(cfg.Engine.Resources, opts.Workers))
 		go pool.Run()
 	}
 	bs.Pool = pool
@@ -274,7 +277,7 @@ func Boot(ctx context.Context, opts BootOpts) (Bootstrap, error) {
 func BootScheduler(ctx context.Context, bs Bootstrap) (*scheduler.Scheduler, error) {
 	fins := &finalizers.Finalizers{}
 
-	localCache, err := lcache.NewState(bs.Root, bs.Pool, bs.Graph.Targets(), bs.Observability, fins, bs.Config.Engine.GC, bs.Config.Engine.ParallelCaching)
+	localCache, err := lcache.NewState(bs.Root, bs.Pool, bs.Graph.Targets(), bs.Observability, fins, bs.Config.Engine.GC)
 	if err != nil {
 		return nil, err
 	}

diff --git a/bootstrap/config.go b/bootstrap/config.go
@@ -6,7 +6,9 @@ import (
 	"github.com/hephbuild/heph/config"
 	"github.com/hephbuild/heph/hroot"
 	"github.com/hephbuild/heph/log/log"
+	"github.com/pbnjay/memory"
 	"os"
+	"runtime"
 	"time"
 )
 
@@ -21,6 +23,11 @@ func BuildConfig(root *hroot.State, profiles []string) (*config.Config, error) {
 	cfg.ProgressInterval = time.Second
 	cfg.Engine.ParallelCaching = true
 	cfg.Engine.SmartGen = true
+	cfg.Engine.Resources = map[string]float64{
+		"cpu":    float64(runtime.NumCPU()),
+		"memory": float64(memory.TotalMemory()),
+		"cdisk":  float64(runtime.NumCPU() * 5),
+	}
 	cfg.CacheOrder = config.CacheOrderLatency
 	cfg.BuildFiles.Patterns = []string{"**/{BUILD,*.BUILD}"}
 	cfg.Platforms = map[string]config.Platform{

diff --git a/bootstrap/rrs.go b/bootstrap/rrs.go
@@ -8,15 +8,20 @@ import (
 	"github.com/hephbuild/heph/log/log"
 	"github.com/hephbuild/heph/scheduler"
 	"github.com/hephbuild/heph/specs"
+	"github.com/hephbuild/heph/status"
 	"github.com/hephbuild/heph/targetrun"
+	"github.com/hephbuild/heph/tgt"
 	"github.com/hephbuild/heph/utils/ads"
 	"github.com/hephbuild/heph/utils/sets"
+	"github.com/hephbuild/heph/worker2"
 	"github.com/hephbuild/heph/worker2/poolwait"
 )
 
 var errHasExprDep = errors.New("has expr, bailing out")
 
-func generateRRs(ctx context.Context, g *graph.State, m specs.Matcher, args []string, opts targetrun.RequestOpts, bailOutOnExpr bool) (targetrun.Requests, error) {
+func generateRRs(ctx context.Context, e *scheduler.Scheduler, m specs.Matcher, args []string, opts targetrun.RequestOpts, bailOutOnExpr, plain bool) (targetrun.Requests, error) {
+	g := e.Graph
+
 	targets, err := g.Targets().Filter(m)
 	if err != nil {
 		return nil, err
@@ -33,19 +38,38 @@ func generateRRs(ctx context.Context, g *graph.State, m specs.Matcher, args []st
 		}
 	}
 
-	rrs := make(targetrun.Requests, 0, targets.Len())
+	linkDeps := worker2.NewGroup()
 	for _, target := range targets.Slice() {
-		if err := ctx.Err(); err != nil {
-			return nil, err
-		}
+		target := target
+		a := worker2.NewAction(worker2.ActionConfig{
+			Ctx:  ctx,
+			Name: "link " + target.Addr,
+			Do: func(ctx context.Context, ins worker2.InStore, outs worker2.OutStore) error {
+				status.Emit(ctx, tgt.TargetStatus(target, "Linking..."))
+				err := check(target)
+				if err != nil {
+					return err
+				}
 
-		err := check(target)
-		if err != nil {
-			return nil, err
-		}
+				err = g.LinkTarget(target, nil)
+				if err != nil {
+					return err
+				}
 
-		err = g.LinkTarget(target, nil)
-		if err != nil {
+				return nil
+			},
+		})
+		linkDeps.AddDep(a)
+	}
+
+	err = poolwait.Wait(ctx, "Link", e.Pool, linkDeps, plain, e.Config.ProgressInterval)
+	if err != nil {
+		return nil, err
+	}
+
+	rrs := make(targetrun.Requests, 0, targets.Len())
+	for _, target := range targets.Slice() {
+		if err := ctx.Err(); err != nil {
 			return nil, err
 		}
 
@@ -197,7 +221,7 @@ func Query(ctx context.Context, e *scheduler.Scheduler, m specs.Matcher, plain,
 func GenerateRRs(ctx context.Context, e *scheduler.Scheduler, m specs.Matcher, targs []string, opts targetrun.RequestOpts, plain, gen bool) (targetrun.Requests, error) {
 	if !e.Config.Engine.SmartGen {
 		if specs.IsMatcherExplicit(m) {
-			rrs, err := generateRRs(ctx, e.Graph, m, targs, opts, true)
+			rrs, err := generateRRs(ctx, e, m, targs, opts, true, plain)
 			if err != nil {
 				if !(errors.Is(err, errHasExprDep) || errors.Is(err, specs.TargetNotFoundErr{})) {
 					return nil, err
@@ -214,7 +238,7 @@ func GenerateRRs(ctx context.Context, e *scheduler.Scheduler, m specs.Matcher, t
 	}
 
 	if !gen {
-		return generateRRs(ctx, e.Graph, m, targs, opts, false)
+		return generateRRs(ctx, e, m, targs, opts, false, plain)
 	}
 
 	err := RunGen(ctx, e, plain, func() (func(gent *graph.Target) bool, error) {
@@ -260,5 +284,5 @@ func GenerateRRs(ctx context.Context, e *scheduler.Scheduler, m specs.Matcher, t
 		return nil, err
 	}
 
-	return generateRRs(ctx, e.Graph, m, targs, opts, false)
+	return generateRRs(ctx, e, m, targs, opts, false, plain)
 }
diff --git a/bootstrap/run.go b/bootstrap/run.go
@@ -9,7 +9,9 @@ import (
 	"github.com/hephbuild/heph/sandbox"
 	"github.com/hephbuild/heph/scheduler"
 	"github.com/hephbuild/heph/specs"
+	"github.com/hephbuild/heph/status"
 	"github.com/hephbuild/heph/targetrun"
+	"github.com/hephbuild/heph/worker2"
 	"github.com/hephbuild/heph/worker2/poolwait"
 	"os"
 	"os/exec"
@@ -70,15 +72,28 @@ func RunMode(ctx context.Context, e *scheduler.Scheduler, rrs targetrun.Requests
 	if inlineRR != nil {
 		skip = []specs.Specer{inlineRR.Target}
 	}
-	tdepsMap, tracker, err := e.ScheduleTargetRRsWithDeps(ctx, rrs, skip)
-	if err != nil {
-		return err
-	}
 
-	tdeps := tdepsMap.All()
-	tdeps.AddDep(tracker.Group())
+	waitDeps := worker2.NewGroup()
+
+	schedule := worker2.NewAction(worker2.ActionConfig{
+		Ctx: ctx,
+		Do: func(ctx context.Context, ins worker2.InStore, outs worker2.OutStore) error {
+			status.Emit(ctx, status.String("Scheduling..."))
+
+			tdepsMap, tracker, err := e.ScheduleTargetRRsWithDeps(ctx, rrs, skip)
+			if err != nil {
+				return err
+			}
+
+			waitDeps.AddDep(tdepsMap.All())
+			waitDeps.AddDep(tracker.Group())
+
+			return nil
+		},
+	})
+	waitDeps.AddDep(schedule)
 
-	err = poolwait.Wait(ctx, "Run", e.Pool, tdeps, runopts.Plain, e.Config.ProgressInterval)
+	err := poolwait.Wait(ctx, "Run", e.Pool, waitDeps, runopts.Plain, e.Config.ProgressInterval)
 	if err != nil {
 		return err
 	}

diff --git a/bootstrapwatch/bootstrap.go b/bootstrapwatch/bootstrap.go
@@ -88,16 +88,16 @@ func Boot(ctx context.Context, root *hroot.State, bootopts bootstrap.BootOpts, c
 		return nil, err
 	}
 
-	pool := worker2.NewEngine()
-	pool.SetDefaultScheduler(bootstrap.DefaultScheduler(bootopts.Workers))
-	go pool.Run()
-	bootopts.Pool = pool
-
-	bbs, err := bootstrap.BootBase(ctx, bootopts)
+	bbs, err := bootstrap.BootBase(ctx, bootopts.BootBaseOpts)
 	if err != nil {
 		return nil, err
 	}
 
+	pool := worker2.NewEngine()
+	pool.SetDefaultScheduler(bootstrap.DefaultScheduler(bbs.Config.Engine.Resources, bootopts.Workers))
+	go pool.Run()
+	bootopts.Pool = pool
+
 	sigCh := make(chan sigEvent)
 
 	s := &State{

diff --git a/cmd/heph/init.go b/cmd/heph/init.go
@@ -25,9 +25,11 @@ func bootstrapOptions() (bootstrap.BootOpts, error) {
 	}
 
 	return bootstrap.BootOpts{
-		Profiles:              *profiles,
+		BootBaseOpts: bootstrap.BootBaseOpts{
+			Profiles: *profiles,
+			Params:   paramsm,
+		},
 		Workers:               workers,
-		Params:                paramsm,
 		Summary:               *summary || *summaryGen,
 		JaegerEndpoint:        *jaegerEndpoint,
 		DisableCloudTelemetry: *noCloudTelemetry,
@@ -86,7 +88,7 @@ func bootstrapBase(ctx context.Context) (bootstrap.BaseBootstrap, error) {
 		return bootstrap.BaseBootstrap{}, err
 	}
 
-	return bootstrap.BootBase(ctx, opts)
+	return bootstrap.BootBase(ctx, opts.BootBaseOpts)
 }
 
 func bootstrapInit(ctx context.Context) (bootstrap.Bootstrap, error) {

diff --git a/cmd/heph/watch2.go b/cmd/heph/watch2.go
@@ -34,7 +34,7 @@ var watchCmd = &cobra.Command{
 			return err
 		}
 
-		bbs, err := bootstrap.BootBase(ctx, opts)
+		bbs, err := bootstrap.BootBase(ctx, opts.BootBaseOpts)
 		if err != nil {
 			return err
 		}

diff --git a/config/config.go b/config/config.go
@@ -32,13 +32,13 @@ type Config struct {
 		Project string `yaml:"project"`
 	} `yaml:"cloud"`
 	Engine struct {
-		GC              bool `yaml:"gc"`
-		CacheHints      bool `yaml:"cache_hints"`
-		GitCacheHints   bool `yaml:"git_cache_hints"`
-		InstallTools    bool `yaml:"install_tools"`
-		KeepSandbox     bool `yaml:"keep_sandbox"`
-		ParallelCaching bool `yaml:"parallel_caching"`
-		SmartGen        bool `yaml:"smart_gen"`
+		GC            bool               `yaml:"gc"`
+		CacheHints    bool               `yaml:"cache_hints"`
+		GitCacheHints bool               `yaml:"git_cache_hints"`
+		InstallTools  bool               `yaml:"install_tools"`
+		KeepSandbox   bool               `yaml:"keep_sandbox"`
+		SmartGen      bool               `yaml:"smart_gen"`
+		Resources     map[string]float64 `yaml:"resources"`
 	} `yaml:"engine"`
 	Platforms  map[string]Platform `yaml:"platforms"`
 	BuildFiles struct {

diff --git a/config/file_config.go b/config/file_config.go
@@ -15,13 +15,13 @@ type FileConfig struct {
 		Project string `yaml:"project"`
 	} `yaml:"cloud"`
 	Engine struct {
-		GC              *bool `yaml:"gc"`
-		CacheHints      *bool `yaml:"cache_hints"`
-		GitCacheHints   *bool `yaml:"git_cache_hints"`
-		InstallTools    *bool `yaml:"install_tools"`
-		KeepSandbox     *bool `yaml:"keep_sandbox"`
-		ParallelCaching *bool `yaml:"parallel_caching"`
-		SmartGen        *bool `yaml:"smart_gen"`
+		GC            *bool              `yaml:"gc"`
+		CacheHints    *bool              `yaml:"cache_hints"`
+		GitCacheHints *bool              `yaml:"git_cache_hints"`
+		InstallTools  *bool              `yaml:"install_tools"`
+		KeepSandbox   *bool              `yaml:"keep_sandbox"`
+		SmartGen      *bool              `yaml:"smart_gen"`
+		Resources     map[string]float64 `yaml:"resources,omitempty"`
 	} `yaml:"engine"`
 	Platforms  map[string]FilePlatform `yaml:"platforms"`
 	BuildFiles struct {
@@ -65,10 +65,6 @@ func (fc FileConfig) ApplyTo(c Config) Config {
 		c.Engine.KeepSandbox = *fc.Engine.KeepSandbox
 	}
 
-	if fc.Engine.ParallelCaching != nil {
-		c.Engine.ParallelCaching = *fc.Engine.ParallelCaching
-	}
-
 	if fc.Engine.SmartGen != nil {
 		c.Engine.SmartGen = *fc.Engine.SmartGen
 	}
@@ -89,6 +85,13 @@ func (fc FileConfig) ApplyTo(c Config) Config {
 		c.Engine.InstallTools = *fc.Engine.InstallTools
 	}
 
+	if c.Engine.Resources == nil {
+		c.Engine.Resources = map[string]float64{}
+	}
+	for k, v := range fc.Engine.Resources {
+		c.Engine.Resources[k] = v
+	}
+
 	if fc.ProgressInterval != nil {
 		c.ProgressInterval = time.Duration(*fc.ProgressInterval) * time.Second
 	}

diff --git a/graph/link_target.go b/graph/link_target.go
@@ -15,6 +15,7 @@ import (
 	"github.com/hephbuild/heph/utils/xcontext"
 	"github.com/hephbuild/heph/utils/xfs"
 	"github.com/hephbuild/heph/utils/xmath"
+	"github.com/hephbuild/heph/worker2"
 	"math"
 	"os"
 	"path/filepath"
@@ -194,7 +195,9 @@ func (e *State) LinkTarget(t *Target, breadcrumb *sets.StringSet) (rerr error) {
 
 	//logPrefix := strings.Repeat("|", breadcrumb.Len()-1)
 
-	t.m.Lock()
+	worker2.Wait(context.Background(), func() {
+		t.m.Lock()
+	})
 	if t.AllTargetDeps == nil {
 		t.AllTargetDeps = NewTargets(0)
 	}