Merge branch 'main' into dirigent_workflow_rps_support

Author: tostocker

.github/configs/wordlist.txt

@@ -827,4 +827,6 @@ InData
 RpsFile
 RpsRequestedGpu
 gpus
-SoCC
+SoCC
+SweepOptions
+SweepType

.github/workflows/tools-tests.yaml

@@ -22,7 +22,7 @@ jobs:
     strategy:
       fail-fast: false
       matrix:
-        module: [ tools/plotter, tools/multi_loader/runner]
+        module: [ tools/plotter, tools/multi_loader/runner, tools/multi_loader/common]
     steps:
     - name: Set up Golang
       uses: actions/setup-go@v5

docs/multi_loader.md

@@ -28,6 +28,8 @@ As a wrapper around loader, multi-loader requires the initial cluster setup to b
 | Verbosity             | string                 | "info", "debug", "trace"      | "info"        | (Optional) Verbosity level for logging the experiment             |
 | PreScript             | string                 | any bash Command              | ""           | (Optional) Local script that runs this specific experiment |
 | PostScript            | string                 | any bash Command              | ""           | (Optional) Local script that runs this specific experiment |
+| Sweep                 | [][SweepOptions](#sweepoptions)        | N/A                           | N/A           | (Optional) List of sweep options for the experiment |
+| SweepType             | string                 | "linear", "grid"              | "grid"        | (Optional) Determines how the sweep options are applied: `grid` means all permutations, while `linear` pairs corresponding elements from sweep options, similar to Python's zip() function |
 
 > **_Important_**: 
 >
@@ -51,6 +53,37 @@ As a wrapper around loader, multi-loader requires the initial cluster setup to b
 > Any field defined in `Config` will override the corresponding value from the configuration in `BaseConfigPath`, but only for that specific experiment. 
 > For example, if `BaseConfigPath` has `ExperimentDuration` set to 5 minutes, and you define `ExperimentDuration` as 10 minutes in `Config`, that particular experiment will run for 10 minutes instead.
 
+### SweepOptions
+
+| Parameter name | Data type     | Possible values               | Default value | Description                                                                |
+|----------------|---------------|-------------------------------|---------------|----------------------------------------------------------------------------|
+| Field          | string        | Any valid field name          | N/A           | Any field in [LoaderConfiguration](https://github.com/vhive-serverless/invitro/blob/main/docs/configuration.md#loader-configuration-file-format), `PreScript` and `PostScript`. Cannot be `TracePath`, `OutputDir` or `Platform` as they are reserved fields |
+| Values         | []interface{} | Any valid value type          | N/A           | List of values for the sweep options. These will be used as the sweep values for the experiment |
+| Format         | string        | Any string containing `{}`    | N/A           | Optional format string to customize the sweep. The `{}` placeholder should be included to replace with sweep values |
+
+#### Example:
+  - `Field` should not be empty.
+  - Reserved fields like `TracePath` and `OutputDir` cannot be used.
+  - `Values` must contain at least one item.
+  - If `Format` is provided, it must include the placeholder `{}`.
+  
+  ```json
+  {
+    "Field": "ExperimentDuration",
+    "Values": [10, 20],
+  },
+  {
+    "Field": "Prescript",
+    "Values": ["example1", "example2", "example3"],
+    "Format": "command_{}_to_run"
+  }
+  ```
+
+> **_Note_**: 
+> You can use PreScript and PostScript as sweep options; however, these scripts will only execute at the local level. This means that the global PreScript and PostScript will not be overridden by the sweep-specific scripts.
+
+
+
 ## Command Flags
 
 The multi-loader accepts the following command-line flags. 

pkg/common/utilities.go

@@ -162,14 +162,14 @@ func DeepCopy[T any](a T) (T, error) {
 	return b, err
 }
 
-func RunScript(command string) {
+func RunCommand(command string) {
 	if command == "" {
 		return
 	}
 	logger.Debug("Running command ", command)
-	cmd, err := exec.Command("/bin/sh", command).Output()
+	cmd, err := exec.Command("sh", "-c", command).Output()
 	if err != nil {
-		log.Fatal(err)
+		logger.Fatal(err)
 	}
 	logger.Debug(string(cmd))
 }

tools/multi_loader/common/constants.go

@@ -3,3 +3,10 @@ package common
 const (
 	TraceFormatString = "{}"
 )
+
+const (
+	GridSweepType   = "grid"
+	LinearSweepType = "linear"
+)
+
+var SweepTypes = []string{GridSweepType, LinearSweepType}

tools/multi_loader/common/utils.go

@@ -2,7 +2,10 @@ package common
 
 import (
 	"encoding/json"
+	"fmt"
 	"os"
+	"path"
+	"path/filepath"
 
 	log "github.com/sirupsen/logrus"
 
@@ -50,3 +53,75 @@ func DeterminePlatformFromConfig(multiLoaderConfig types.MultiLoaderConfiguratio
 	}
 	return loaderConfig.Platform
 }
+
+/**
+ * NextCProduct generates the next Cartesian product of the given limits
+ **/
+func NextCProduct(limits []int) func() []int {
+	permutations := make([]int, len(limits))
+	indices := make([]int, len(limits))
+	done := false
+
+	return func() []int {
+		// Check if there are more permutations
+		if done {
+			return nil
+		}
+
+		// Generate the current permutation
+		copy(permutations, indices)
+
+		// Generate the next permutation
+		for i := len(indices) - 1; i >= 0; i-- {
+			indices[i]++
+			if indices[i] <= limits[i] {
+				break
+			}
+			indices[i] = 0
+			if i == 0 {
+				// All permutations have been generated
+				done = true
+			}
+		}
+
+		return permutations
+	}
+}
+
+func SplitPath(path string) []string {
+	dir, last := filepath.Split(path)
+	if dir == "" {
+		return []string{last}
+	}
+	return append(SplitPath(filepath.Clean(dir)), last)
+}
+
+func SweepOptionsToPostfix(sweepOptions []types.SweepOptions, selectedSweepValues []int) string {
+	var postfix string
+	for i, sweepOption := range sweepOptions {
+		postfix += fmt.Sprintf("_%s_%v", sweepOption.Field, sweepOption.Values[selectedSweepValues[i]])
+	}
+	return postfix
+}
+
+func UpdateExperimentWithSweepIndices(experiment *types.LoaderExperiment, sweepOptions []types.SweepOptions, selectedSweepValues []int) {
+	experimentPostFix := SweepOptionsToPostfix(sweepOptions, selectedSweepValues)
+
+	experiment.Name = experiment.Name + experimentPostFix
+	paths := SplitPath(experiment.Config["OutputPathPrefix"].(string))
+	// update the last two paths with the sweep indices
+	paths[len(paths)-2] = paths[len(paths)-2] + experimentPostFix
+
+	experiment.Config["OutputPathPrefix"] = path.Join(paths...)
+
+	for sweepOptionI, sweepValueI := range selectedSweepValues {
+		sweepValue := sweepOptions[sweepOptionI].GetValue(sweepValueI)
+		if sweepOptions[sweepOptionI].Field == "PreScript" {
+			experiment.PreScript = sweepValue.(string)
+		} else if sweepOptions[sweepOptionI].Field == "PostScript" {
+			experiment.PostScript = sweepValue.(string)
+		} else {
+			experiment.Config[sweepOptions[sweepOptionI].Field] = sweepValue
+		}
+	}
+}

tools/multi_loader/common/utils_test.go

@@ -0,0 +1,88 @@
+package common
+
+import (
+	"path"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/vhive-serverless/loader/tools/multi_loader/types"
+)
+
+func TestNextProduct(t *testing.T) {
+	ints := []int{2, 1}
+	nextProduct := NextCProduct(ints)
+	expectedArrs := [][]int{{0, 0}, {0, 1}, {1, 0}, {1, 1}, {2, 0}, {2, 1}}
+	curI := 0
+
+	for {
+		product := nextProduct()
+		if len(product) == 0 {
+			if curI != len(expectedArrs) {
+				t.Fatalf("Expected %d products, got %d", len(expectedArrs), curI)
+			}
+			break
+		}
+		if len(product) != len(ints) {
+			t.Fatalf("Expected product length %d, got %d", len(ints), len(product))
+		}
+		for i, v := range product {
+			if v != expectedArrs[curI][i] {
+				t.Fatalf("Expected %v, got %v", expectedArrs[curI], product)
+			}
+		}
+		curI++
+	}
+}
+
+func TestSplitPath(t *testing.T) {
+	assert.Equal(t, []string{"file.txt"}, SplitPath("file.txt"), "Expected ['file.txt'] for single file")
+	assert.Equal(t, []string{"home", "user", "docs", "file.txt"}, SplitPath(filepath.Join("home", "user", "docs", "file.txt")), "Expected full path split")
+}
+
+func TestSweepOptionsToPostfix(t *testing.T) {
+	t.Run("Test Post Fix Naming Util", func(t *testing.T) {
+		result := SweepOptionsToPostfix(
+			[]types.SweepOptions{
+				{Field: "PreScript", Values: []interface{}{"PreValue_1", "PreValue_2"}},
+				{Field: "CPULimit", Values: []interface{}{"1vCPU", "2vCPU", "4vCPU"}},
+				{Field: "ExperimentDuration", Values: []interface{}{"10", "20", "30"}},
+				{Field: "PostScript", Values: []interface{}{"PostValue_1", "PostValue_2", "PostValue_3"}},
+			},
+			[]int{1, 2, 0, 2},
+		)
+		assert.Equal(t, "_PreScript_PreValue_2_CPULimit_4vCPU_ExperimentDuration_10_PostScript_PostValue_3", result, "Unexpected postfix result")
+	})
+}
+
+func TestUpdateExperimentWithSweepIndices(t *testing.T) {
+	experiment := &types.LoaderExperiment{
+		Name:   "test_experiment",
+		Config: map[string]interface{}{"OutputPathPrefix": "first/second/third"},
+	}
+
+	// Test sweep options
+	sweepOptions := []types.SweepOptions{
+		{Field: "PreScript", Values: []interface{}{"pre1", "pre2"}, Format: "test_{}"},
+		{Field: "PostScript", Values: []interface{}{"post1", "post2"}, Format: ""},
+		{Field: "ExperimentDuration", Values: []interface{}{"1", "2"}, Format: ""},
+	}
+
+	selectedSweepValues := []int{1, 0, 1}
+
+	UpdateExperimentWithSweepIndices(experiment, sweepOptions, selectedSweepValues)
+
+	expectedPostfix := "_PreScript_pre2_PostScript_post1_ExperimentDuration_2"
+
+	// Check experiment name
+	assert.Equal(t, "test_experiment"+expectedPostfix, experiment.Name)
+
+	// Check OutputPathPrefix
+	expectedOutputPathPrefix := path.Join("first", "second"+expectedPostfix, "third")
+	assert.Equal(t, expectedOutputPathPrefix, experiment.Config["OutputPathPrefix"])
+
+	// Verify that the sweep options have been updated
+	assert.Equal(t, "test_pre2", experiment.PreScript)
+	assert.Equal(t, "post1", experiment.PostScript)
+	assert.Equal(t, "2", experiment.Config["ExperimentDuration"])
+}

tools/multi_loader/common/validators.go

@@ -2,6 +2,7 @@ package common
 
 import (
 	"path"
+	"slices"
 	"strings"
 
 	log "github.com/sirupsen/logrus"
@@ -51,6 +52,21 @@ func CheckMultiLoaderConfig(multiLoaderConfig types.MultiLoaderConfiguration) {
 				log.Warn("Setting default output directory to ", study.OutputDir)
 			}
 		}
+
+		// Check sweep options
+		for _, sweep := range study.Sweep {
+			if err := sweep.Validate(); err != nil {
+				log.Fatal(err)
+			}
+		}
+		// check sweep type
+		CheckSweepType(study.SweepType)
 	}
 	log.Debug("All experiments configs are valid")
 }
+
+func CheckSweepType(sweepType string) {
+	if sweepType != "" && !slices.Contains(SweepTypes, sweepType) {
+		log.Fatal("Invalid Sweep Type ", sweepType)
+	}
+}

tools/multi_loader/common/validators_test.go

@@ -0,0 +1,40 @@
+package common
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	log "github.com/sirupsen/logrus"
+)
+
+func TestSweepTypeValidator(t *testing.T) {
+	t.Run("Check Grid Sweep Type", func(t *testing.T) {
+		CheckSweepType("grid")
+	})
+	t.Run("Check Random Sweep Type", func(t *testing.T) {
+		expectFatal(t, func() {
+			CheckSweepType("random")
+		})
+	})
+	t.Run(("Check Linear Sweep Type"), func(t *testing.T) {
+		CheckSweepType("linear")
+	})
+
+}
+
+func expectFatal(t *testing.T, funcToTest func()) {
+	fatal := false
+	originalExitFunc := log.StandardLogger().ExitFunc
+	log.Info("Expecting a fatal message during the test, overriding the exit function")
+	// Replace logrus exit function
+	log.StandardLogger().ExitFunc = func(int) {
+		fatal = true
+		t.SkipNow()
+	}
+	defer func() {
+		log.StandardLogger().ExitFunc = originalExitFunc
+		assert.True(t, fatal, "Expected log.Fatal to be called")
+	}()
+	funcToTest()
+}