Core changes for k6/x/execution JS module

core/local/local.go

@@ -157,15 +157,15 @@ func (e *ExecutionScheduler) GetExecutionPlan() []lib.ExecutionStep {
 func (e *ExecutionScheduler) initVU(
 	samplesOut chan<- stats.SampleContainer, logger *logrus.Entry,
 ) (lib.InitializedVU, error) {
-	// Get the VU ID here, so that the VUs are (mostly) ordered by their
+	// Get the VU IDs here, so that the VUs are (mostly) ordered by their
 	// number in the channel buffer
-	vuID := e.state.GetUniqueVUIdentifier()
-	vu, err := e.runner.NewVU(int64(vuID), samplesOut)
+	vuIDLocal, vuIDGlobal := e.state.GetUniqueVUIdentifiers()
+	vu, err := e.runner.NewVU(vuIDLocal, vuIDGlobal, samplesOut)
 	if err != nil {
-		return nil, errext.WithHint(err, fmt.Sprintf("error while initializing VU #%d", vuID))
+		return nil, errext.WithHint(err, fmt.Sprintf("error while initializing VU #%d", vuIDGlobal))
 	}
 
-	logger.Debugf("Initialized VU #%d", vuID)
+	logger.Debugf("Initialized VU #%d", vuIDGlobal)
 	return vu, nil
 }
 
@@ -362,6 +362,7 @@ func (e *ExecutionScheduler) Run(globalCtx, runCtx context.Context, engineOut ch
 
 	runResults := make(chan error, executorsCount) // nil values are successful runs
 
+	runCtx = lib.WithExecutionState(runCtx, e.state)
 	runSubCtx, cancel := context.WithCancel(runCtx)
 	defer cancel() // just in case, and to shut up go vet...
 

js/bundle.go

@@ -236,7 +236,7 @@ func (b *Bundle) getExports(logger logrus.FieldLogger, rt *goja.Runtime, options
 }
 
 // Instantiate creates a new runtime from this bundle.
-func (b *Bundle) Instantiate(logger logrus.FieldLogger, vuID int64) (bi *BundleInstance, instErr error) {
+func (b *Bundle) Instantiate(logger logrus.FieldLogger, vuID uint64) (bi *BundleInstance, instErr error) {
 	// TODO: actually use a real context here, so that the instantiation can be killed
 	// Placeholder for a real context.
 	ctxPtr := new(context.Context)
@@ -283,7 +283,7 @@ func (b *Bundle) Instantiate(logger logrus.FieldLogger, vuID int64) (bi *BundleI
 
 // Instantiates the bundle into an existing runtime. Not public because it also messes with a bunch
 // of other things, will potentially thrash data and makes a mess in it if the operation fails.
-func (b *Bundle) instantiate(logger logrus.FieldLogger, rt *goja.Runtime, init *InitContext, vuID int64) error {
+func (b *Bundle) instantiate(logger logrus.FieldLogger, rt *goja.Runtime, init *InitContext, vuID uint64) error {
 	rt.SetParserOptions(parser.WithDisableSourceMaps)
 	rt.SetFieldNameMapper(common.FieldNameMapper{})
 	rt.SetRandSource(common.NewRandSource())

js/bundle_test.go

@@ -910,7 +910,7 @@ func TestBundleNotSharable(t *testing.T) {
 		t.Run(name, func(t *testing.T) {
 			t.Parallel()
 			for i := 0; i < vus; i++ {
-				bi, err := b.Instantiate(logger, int64(i))
+				bi, err := b.Instantiate(logger, uint64(i))
 				require.NoError(t, err)
 				for j := 0; j < iters; j++ {
 					bi.Runtime.Set("__ITER", j)

js/console_test.go

@@ -143,7 +143,7 @@ func TestConsole(t *testing.T) {
 					assert.NoError(t, err)
 
 					samples := make(chan stats.SampleContainer, 100)
-					initVU, err := r.newVU(1, samples)
+					initVU, err := r.newVU(1, 1, samples)
 					assert.NoError(t, err)
 
 					ctx, cancel := context.WithCancel(context.Background())
@@ -245,7 +245,7 @@ func TestFileConsole(t *testing.T) {
 							assert.NoError(t, err)
 
 							samples := make(chan stats.SampleContainer, 100)
-							initVU, err := r.newVU(1, samples)
+							initVU, err := r.newVU(1, 1, samples)
 							assert.NoError(t, err)
 
 							ctx, cancel := context.WithCancel(context.Background())

js/empty_iteartions_bench_test.go

@@ -26,7 +26,7 @@ func BenchmarkEmptyIteration(b *testing.B) {
 		for range ch {
 		}
 	}()
-	initVU, err := r.NewVU(1, ch)
+	initVU, err := r.NewVU(1, 1, ch)
 	if !assert.NoError(b, err) {
 		return
 	}

js/http_bench_test.go

@@ -62,7 +62,7 @@ func BenchmarkHTTPRequests(b *testing.B) {
 		for range ch {
 		}
 	}()
-	initVU, err := r.NewVU(1, ch)
+	initVU, err := r.NewVU(1, 1, ch)
 	if !assert.NoError(b, err) {
 		return
 	}
@@ -105,7 +105,7 @@ func BenchmarkHTTPRequestsBase(b *testing.B) {
 		for range ch {
 		}
 	}()
-	initVU, err := r.NewVU(1, ch)
+	initVU, err := r.NewVU(1, 1, ch)
 	if !assert.NoError(b, err) {
 		return
 	}

js/init_and_modules_test.go

@@ -28,17 +28,18 @@ import (
 	"testing"
 	"time"
 
+	"github.com/spf13/afero"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/guregu/null.v3"
+
 	"go.k6.io/k6/js"
 	"go.k6.io/k6/js/common"
 	"go.k6.io/k6/js/modules"
 	"go.k6.io/k6/lib"
 	"go.k6.io/k6/lib/testutils"
 	"go.k6.io/k6/loader"
 	"go.k6.io/k6/stats"
-	"github.com/spf13/afero"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-	"gopkg.in/guregu/null.v3"
 )
 
 type CheckModule struct {
@@ -94,7 +95,7 @@ func TestNewJSRunnerWithCustomModule(t *testing.T) {
 	assert.Equal(t, checkModule.initCtxCalled, 1)
 	assert.Equal(t, checkModule.vuCtxCalled, 0)
 
-	vu, err := runner.NewVU(1, make(chan stats.SampleContainer, 100))
+	vu, err := runner.NewVU(1, 1, make(chan stats.SampleContainer, 100))
 	require.NoError(t, err)
 	assert.Equal(t, checkModule.initCtxCalled, 2)
 	assert.Equal(t, checkModule.vuCtxCalled, 0)
@@ -118,7 +119,7 @@ func TestNewJSRunnerWithCustomModule(t *testing.T) {
 	require.NoError(t, err)
 	assert.Equal(t, checkModule.initCtxCalled, 3) // changes because we need to get the exported functions
 	assert.Equal(t, checkModule.vuCtxCalled, 2)
-	vuFromArc, err := runnerFromArc.NewVU(2, make(chan stats.SampleContainer, 100))
+	vuFromArc, err := runnerFromArc.NewVU(2, 2, make(chan stats.SampleContainer, 100))
 	require.NoError(t, err)
 	assert.Equal(t, checkModule.initCtxCalled, 4)
 	assert.Equal(t, checkModule.vuCtxCalled, 2)

js/module_loading_test.go

@@ -116,7 +116,7 @@ func TestLoadOnceGlobalVars(t *testing.T) {
 					t.Parallel()
 					ch := newDevNullSampleChannel()
 					defer close(ch)
-					initVU, err := r.NewVU(1, ch)
+					initVU, err := r.NewVU(1, 1, ch)
 
 					ctx, cancel := context.WithCancel(context.Background())
 					defer cancel()
@@ -167,7 +167,7 @@ func TestLoadExportsIsUsableInModule(t *testing.T) {
 			t.Parallel()
 			ch := newDevNullSampleChannel()
 			defer close(ch)
-			initVU, err := r.NewVU(1, ch)
+			initVU, err := r.NewVU(1, 1, ch)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
 			vu := initVU.Activate(&lib.VUActivationParams{RunContext: ctx})
@@ -215,7 +215,7 @@ func TestLoadDoesntBreakHTTPGet(t *testing.T) {
 			t.Parallel()
 			ch := newDevNullSampleChannel()
 			defer close(ch)
-			initVU, err := r.NewVU(1, ch)
+			initVU, err := r.NewVU(1, 1, ch)
 			require.NoError(t, err)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
@@ -261,7 +261,7 @@ func TestLoadGlobalVarsAreNotSharedBetweenVUs(t *testing.T) {
 			t.Parallel()
 			ch := newDevNullSampleChannel()
 			defer close(ch)
-			initVU, err := r.NewVU(1, ch)
+			initVU, err := r.NewVU(1, 1, ch)
 			require.NoError(t, err)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
@@ -270,7 +270,7 @@ func TestLoadGlobalVarsAreNotSharedBetweenVUs(t *testing.T) {
 			require.NoError(t, err)
 
 			// run a second VU
-			initVU, err = r.NewVU(2, ch)
+			initVU, err = r.NewVU(2, 2, ch)
 			require.NoError(t, err)
 			ctx, cancel = context.WithCancel(context.Background())
 			defer cancel()
@@ -326,7 +326,7 @@ func TestLoadCycle(t *testing.T) {
 			t.Parallel()
 			ch := newDevNullSampleChannel()
 			defer close(ch)
-			initVU, err := r.NewVU(1, ch)
+			initVU, err := r.NewVU(1, 1, ch)
 			require.NoError(t, err)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
@@ -389,7 +389,7 @@ func TestLoadCycleBinding(t *testing.T) {
 			t.Parallel()
 			ch := newDevNullSampleChannel()
 			defer close(ch)
-			initVU, err := r.NewVU(1, ch)
+			initVU, err := r.NewVU(1, 1, ch)
 			require.NoError(t, err)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
@@ -455,7 +455,7 @@ func TestBrowserified(t *testing.T) {
 			t.Parallel()
 			ch := make(chan stats.SampleContainer, 100)
 			defer close(ch)
-			initVU, err := r.NewVU(1, ch)
+			initVU, err := r.NewVU(1, 1, ch)
 			require.NoError(t, err)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
@@ -499,7 +499,7 @@ func TestLoadingUnexistingModuleDoesntPanic(t *testing.T) {
 			t.Parallel()
 			ch := newDevNullSampleChannel()
 			defer close(ch)
-			initVU, err := r.NewVU(1, ch)
+			initVU, err := r.NewVU(1, 1, ch)
 			require.NoError(t, err)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()
@@ -534,7 +534,7 @@ func TestLoadingSourceMapsDoesntErrorOut(t *testing.T) {
 			t.Parallel()
 			ch := newDevNullSampleChannel()
 			defer close(ch)
-			initVU, err := r.NewVU(1, ch)
+			initVU, err := r.NewVU(1, 1, ch)
 			require.NoError(t, err)
 			ctx, cancel := context.WithCancel(context.Background())
 			defer cancel()

js/modules/k6/marshalling_test.go

@@ -136,7 +136,7 @@ func TestSetupDataMarshalling(t *testing.T) {
 	if !assert.NoError(t, runner.Setup(context.Background(), samples)) {
 		return
 	}
-	initVU, err := runner.NewVU(1, samples)
+	initVU, err := runner.NewVU(1, 1, samples)
 	if assert.NoError(t, err) {
 		ctx, cancel := context.WithCancel(context.Background())
 		defer cancel()