Clean up and optimize http.batch(), fix batchPerHost

cmd/options.go

@@ -52,7 +52,7 @@ func optionFlagSet() *pflag.FlagSet {
 	flags.BoolP("paused", "p", false, "start the test in a paused state")
 	flags.Int64("max-redirects", 10, "follow at most n redirects")
 	flags.Int64("batch", 20, "max parallel batch reqs")
-	flags.Int64("batch-per-host", 20, "max parallel batch reqs per host")
+	flags.Int64("batch-per-host", 6, "max parallel batch reqs per host")
 	flags.Int64("rps", 0, "limit requests per second")
 	flags.String("user-agent", fmt.Sprintf("k6/%s (https://k6.io/)", consts.Version), "user agent for http requests")
 	flags.String("http-debug", "", "log all HTTP requests and responses. Excludes body by default. To include body use '--http-debug=full'")
@@ -94,6 +94,7 @@ func getOptions(flags *pflag.FlagSet) (lib.Options, error) {
 		Paused:                getNullBool(flags, "paused"),
 		MaxRedirects:          getNullInt64(flags, "max-redirects"),
 		Batch:                 getNullInt64(flags, "batch"),
+		BatchPerHost:          getNullInt64(flags, "batch-per-host"),
 		RPS:                   getNullInt64(flags, "rps"),
 		UserAgent:             getNullString(flags, "user-agent"),
 		HTTPDebug:             getNullString(flags, "http-debug"),

core/engine_test.go

@@ -574,6 +574,7 @@ func TestSentReceivedMetrics(t *testing.T) {
 			Hosts:                 tb.Dialer.Hosts,
 			InsecureSkipTLSVerify: null.BoolFrom(true),
 			NoVUConnectionReuse:   null.BoolFrom(noConnReuse),
+			Batch:                 null.IntFrom(20),
 		}
 
 		r.SetOptions(options)

js/modules/k6/http/request.go

@@ -30,7 +30,6 @@ import (
 	"net/url"
 	"reflect"
 	"strings"
-	"sync"
 	"time"
 
 	"github.com/dop251/goja"
@@ -357,70 +356,98 @@ func (h *HTTP) parseRequest(
 	return result, nil
 }
 
+func (h *HTTP) prepareBatchArray(
+	ctx context.Context, requests []interface{},
+) ([]httpext.BatchParsedHTTPRequest, []*Response, error) {
+	reqCount := len(requests)
+	batchReqs := make([]httpext.BatchParsedHTTPRequest, reqCount)
+	results := make([]*Response, reqCount)
+
+	for i, req := range requests {
+		parsedReq, err := h.parseBatchRequest(ctx, i, req)
+		if err != nil {
+			return nil, nil, err
+		}
+		response := new(httpext.Response)
+		batchReqs[i] = httpext.BatchParsedHTTPRequest{
+			ParsedHTTPRequest: parsedReq,
+			Response:          response,
+		}
+		results[i] = &Response{response}
+	}
+
+	return batchReqs, results, nil
+}
+
+func (h *HTTP) prepareBatchObject(
+	ctx context.Context, requests map[string]interface{},
+) ([]httpext.BatchParsedHTTPRequest, map[string]*Response, error) {
+	reqCount := len(requests)
+	batchReqs := make([]httpext.BatchParsedHTTPRequest, reqCount)
+	results := make(map[string]*Response, reqCount)
+
+	i := 0
+	for key, req := range requests {
+		parsedReq, err := h.parseBatchRequest(ctx, key, req)
+		if err != nil {
+			return nil, nil, err
+		}
+		response := new(httpext.Response)
+		batchReqs[i] = httpext.BatchParsedHTTPRequest{
+			ParsedHTTPRequest: parsedReq,
+			Response:          response,
+		}
+		results[key] = &Response{response}
+		i++
+	}
+
+	return batchReqs, results, nil
+}
+
 // Batch makes multiple simultaneous HTTP requests. The provideds reqsV should be an array of request
 // objects. Batch returns an array of responses and/or error
 func (h *HTTP) Batch(ctx context.Context, reqsV goja.Value) (goja.Value, error) {
 	state := lib.GetState(ctx)
 	if state == nil {
 		return nil, ErrBatchForbiddenInInitContext
 	}
-	rt := common.GetRuntime(ctx)
-
-	reqs := reqsV.ToObject(rt)
-	keys := reqs.Keys()
-	parsedReqs := map[string]*httpext.ParsedHTTPRequest{}
-	for _, key := range keys {
-		parsedReq, err := h.parseBatchRequest(ctx, key, reqs.Get(key))
-		if err != nil {
-			return nil, err
-		}
-		parsedReqs[key] = parsedReq
-	}
 
 	var (
-		// Return values; retval must be guarded by the mutex.
-		mutex  sync.Mutex
-		retval = rt.NewObject()
-		errs   = make(chan error)
-
-		// Concurrency limits.
-		globalLimiter  = NewSlotLimiter(int(state.Options.Batch.Int64))
-		perHostLimiter = NewMultiSlotLimiter(int(state.Options.BatchPerHost.Int64))
+		err       error
+		batchReqs []httpext.BatchParsedHTTPRequest
+		results   interface{} // either []*Response or map[string]*Response
 	)
-	for k, pr := range parsedReqs {
-		go func(key string, parsedReq *httpext.ParsedHTTPRequest) {
-			globalLimiter.Begin()
-			defer globalLimiter.End()
-
-			if hl := perHostLimiter.Slot(parsedReq.URL.GetURL().Host); hl != nil {
-				hl.Begin()
-				defer hl.End()
-			}
-
-			res, err := httpext.MakeRequest(ctx, parsedReq)
-			if err != nil {
-				errs <- err
-				return
-			}
 
-			mutex.Lock()
-			_ = retval.Set(key, responseFromHttpext(res))
-			mutex.Unlock()
+	switch v := reqsV.Export().(type) {
+	case []interface{}:
+		batchReqs, results, err = h.prepareBatchArray(ctx, v)
+	case map[string]interface{}:
+		batchReqs, results, err = h.prepareBatchObject(ctx, v)
+	default:
+		return nil, fmt.Errorf("invalid http.batch() argument type %T", v)
+	}
 
-			errs <- nil
-		}(k, pr)
+	if err != nil {
+		return nil, err
 	}
 
-	var err error
-	for range keys {
-		if e := <-errs; e != nil {
+	reqCount := len(batchReqs)
+	errs := httpext.MakeBatchRequests(
+		ctx, batchReqs, reqCount,
+		int(state.Options.Batch.Int64), int(state.Options.BatchPerHost.Int64),
+	)
+
+	for i := 0; i < reqCount; i++ {
+		if e := <-errs; e != nil && err == nil { // Save only the first error
 			err = e
 		}
 	}
-	return retval, err
+	return common.GetRuntime(ctx).ToValue(results), err
 }
 
-func (h *HTTP) parseBatchRequest(ctx context.Context, key string, val goja.Value) (*httpext.ParsedHTTPRequest, error) {
+func (h *HTTP) parseBatchRequest(
+	ctx context.Context, key interface{}, val interface{},
+) (*httpext.ParsedHTTPRequest, error) {
 	var (
 		method = HTTP_METHOD_GET
 		ok     bool
@@ -431,7 +458,7 @@ func (h *HTTP) parseBatchRequest(ctx context.Context, key string, val goja.Value
 		rt     = common.GetRuntime(ctx)
 	)
 
-	switch data := val.Export().(type) {
+	switch data := val.(type) {
 	case []interface{}:
 		// Handling of ["GET", "http://example.com/"]
 		dataLen := len(data)
@@ -456,7 +483,7 @@ func (h *HTTP) parseBatchRequest(ctx context.Context, key string, val goja.Value
 	case map[string]interface{}:
 		// Handling of {method: "GET", url: "http://test.loadimpact.com"}
 		if murl, ok := data["url"]; !ok {
-			return nil, fmt.Errorf("batch request %s doesn't have an url key", key)
+			return nil, fmt.Errorf("batch request %q doesn't have an url key", key)
 		} else if reqURL, err = ToURL(murl); err != nil {
 			return nil, err
 		}

js/modules/k6/http/request_test.go

@@ -123,6 +123,8 @@ func newRuntime(
 		UserAgent:    null.StringFrom("TestUserAgent"),
 		Throw:        null.BoolFrom(true),
 		SystemTags:   &stats.DefaultSystemTagSet,
+		Batch:        null.IntFrom(20),
+		BatchPerHost: null.IntFrom(20),
 		//HTTPDebug:    null.StringFrom("full"),
 	}
 	samples := make(chan stats.SampleContainer, 1000)
@@ -1037,6 +1039,10 @@ func TestRequestAndBatch(t *testing.T) {
 	}
 
 	t.Run("Batch", func(t *testing.T) {
+		t.Run("error", func(t *testing.T) {
+			_, err := common.RunString(rt, `let res = http.batch("https://somevalidurl.com");`)
+			require.Error(t, err)
+		})
 		t.Run("GET", func(t *testing.T) {
 			_, err := common.RunString(rt, sr(`
 			let reqs = [
@@ -1048,7 +1054,7 @@ func TestRequestAndBatch(t *testing.T) {
 				if (res[key].status != 200) { throw new Error("wrong status: " + res[key].status); }
 				if (res[key].url != reqs[key][1]) { throw new Error("wrong url: " + res[key].url); }
 			}`))
-			assert.NoError(t, err)
+			require.NoError(t, err)
 			bufSamples := stats.GetBufferedSamples(samples)
 			assertRequestMetricsEmitted(t, bufSamples, "GET", sr("HTTPBIN_URL/"), "", 200, "")
 			assertRequestMetricsEmitted(t, bufSamples, "GET", sr("HTTPBIN_IP_URL/"), "", 200, "")

js/modules/k6/http/response.go

@@ -21,7 +21,6 @@
 package http
 
 import (
-	"context"
 	"errors"
 	"fmt"
 	"net/url"
@@ -34,22 +33,18 @@ import (
 )
 
 // Response is a representation of an HTTP response to be returned to the goja VM
-// TODO: refactor after https://github.com/dop251/goja/issues/84
-type Response httpext.Response
-
-// GetCtx returns the Context of the httpext.Response
-func (res *Response) GetCtx() context.Context {
-	return ((*httpext.Response)(res)).GetCtx()
+type Response struct {
+	*httpext.Response `js:"-"`
 }
 
 func responseFromHttpext(resp *httpext.Response) *Response {
-	res := Response(*resp)
+	res := Response{resp}
 	return &res
 }
 
 // JSON parses the body of a response as json and returns it to the goja VM
 func (res *Response) JSON(selector ...string) goja.Value {
-	v, err := ((*httpext.Response)(res)).JSON(selector...)
+	v, err := res.Response.JSON(selector...)
 	if err != nil {
 		common.Throw(common.GetRuntime(res.GetCtx()), err)
 	}

js/modules/k6/http/limiter.go → lib/limiter.go

@@ -18,7 +18,7 @@
  *
  */
 
-package http
+package lib
 
 import (
 	"sync"

js/modules/k6/http/limiter_test.go → lib/limiter_test.go

@@ -17,7 +17,7 @@
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  *
  */
-package http
+package lib
 
 import (
 	"fmt"
@@ -69,11 +69,12 @@ func TestSlotLimiters(t *testing.T) {
 			l := NewSlotLimiter(tc.limit)
 			wg := sync.WaitGroup{}
 
-			if tc.limit == 0 {
+			switch {
+			case tc.limit == 0:
 				wg.Add(tc.launches)
-			} else if tc.launches < tc.limit {
+			case tc.launches < tc.limit:
 				wg.Add(tc.launches)
-			} else {
+			default:
 				wg.Add(tc.limit)
 			}
 

lib/netext/httpext/batch.go

@@ -0,0 +1,84 @@
+/*
+ *
+ * k6 - a next-generation load testing tool
+ * Copyright (C) 2019 Load Impact
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as
+ * published by the Free Software Foundation, either version 3 of the
+ * License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ *
+ */
+
+package httpext
+
+import (
+	"context"
+	"sync/atomic"
+
+	"github.com/loadimpact/k6/lib"
+)
+
+// BatchParsedHTTPRequest extends the normal parsed HTTP request with a pointer
+// to a Response object, so that the batch goroutines can concurrently store the
+// responses they receive, without any locking.
+type BatchParsedHTTPRequest struct {
+	*ParsedHTTPRequest
+	Response *Response // this is modified by MakeBatchRequests()
+}
+
+// MakeBatchRequests concurrently makes multiple requests. It spawns
+// min(reqCount, globalLimit) goroutines that asynchronously process all
+// requests coming from the requests channel. Responses are recorded in the
+// pointers contained in each BatchParsedHTTPRequest object, so they need to be
+// pre-initialized. In addition, each processed request would emit either a nil
+// value, or an error, via the returned errors channel. The goroutines exit when
+// the requests channel is closed.
+func MakeBatchRequests(
+	ctx context.Context,
+	requests []BatchParsedHTTPRequest,
+	reqCount, globalLimit, perHostLimit int,
+) <-chan error {
+	workers := globalLimit
+	if reqCount < workers {
+		workers = reqCount
+	}
+	result := make(chan error, reqCount)
+	perHostLimiter := lib.NewMultiSlotLimiter(perHostLimit)
+
+	makeRequest := func(req BatchParsedHTTPRequest) {
+		if hl := perHostLimiter.Slot(req.URL.GetURL().Host); hl != nil {
+			hl.Begin()
+			defer hl.End()
+		}
+
+		resp, err := MakeRequest(ctx, req.ParsedHTTPRequest)
+		if resp != nil {
+			*req.Response = *resp
+		}
+		result <- err
+	}
+
+	counter, i32reqCount := int32(-1), int32(reqCount)
+	for i := 0; i < workers; i++ {
+		go func() {
+			for {
+				reqNum := atomic.AddInt32(&counter, 1)
+				if reqNum >= i32reqCount {
+					return
+				}
+				makeRequest(requests[reqNum])
+			}
+		}()
+	}
+
+	return result
+}