Merge pull request #26827 from vespa-engine/mpolden/feed-client-9

Less locking

7 files changed, 39 insertions, 28 deletions

diff --git a/client/go/go.mod b/client/go/go.mod
index 18e3853868d..94f69c8286a 100644
--- a/client/go/go.mod
+++ b/client/go/go.mod
@@ -13,8 +13,8 @@ require (
 	github.com/spf13/pflag v1.0.5
 	github.com/stretchr/testify v1.7.0
 	github.com/zalando/go-keyring v0.1.1
-	golang.org/x/net v0.8.0
-	golang.org/x/sys v0.6.0
+	golang.org/x/net v0.9.0
+	golang.org/x/sys v0.7.0
 	gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b
 )
 
@@ -28,7 +28,7 @@ require (
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/russross/blackfriday/v2 v2.1.0 // indirect
 	github.com/stretchr/objx v0.1.1 // indirect
-	golang.org/x/text v0.8.0 // indirect
+	golang.org/x/text v0.9.0 // indirect
 	gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
 	gopkg.in/yaml.v2 v2.4.0 // indirect
 )
diff --git a/client/go/go.sum b/client/go/go.sum
index 2af8bb1e4c0..ac662c9fd43 100644
--- a/client/go/go.sum
+++ b/client/go/go.sum
@@ -47,16 +47,16 @@ github.com/stretchr/testify v1.7.0 h1:nwc3DEeHmmLAfoZucVR881uASk0Mfjw8xYJ99tb5Cc
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/zalando/go-keyring v0.1.1 h1:w2V9lcx/Uj4l+dzAf1m9s+DJ1O8ROkEHnynonHjTcYE=
 github.com/zalando/go-keyring v0.1.1/go.mod h1:OIC+OZ28XbmwFxU/Rp9V7eKzZjamBJwRzC8UFJH9+L8=
-golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
-golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
+golang.org/x/net v0.9.0 h1:aWJ/m6xSmxWBx+V0XRHTlrYrPG56jKsLdTFmsSsCzOM=
+golang.org/x/net v0.9.0/go.mod h1:d48xBJpPfHeWQsugry2m+kC02ZBRGRgulfHnEXEuWns=
 golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210616045830-e2b7044e8c71/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
-golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/text v0.8.0 h1:57P1ETyNKtuIjB4SRd15iJxuhj8Gc416Y78H3qgMh68=
-golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
+golang.org/x/sys v0.7.0 h1:3jlCCIQZPdOYu1h8BkNvLz8Kgwtae2cagcG/VamtZRU=
+golang.org/x/sys v0.7.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/text v0.9.0 h1:2sjJmO8cDvYveuX97RDLsxlyUxLl+GHoLxBiRdHllBE=
+golang.org/x/text v0.9.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo=
 gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
diff --git a/client/go/internal/cli/cmd/feed.go b/client/go/internal/cli/cmd/feed.go
index 06568dd35c3..8635f4aa41b 100644
--- a/client/go/internal/cli/cmd/feed.go
+++ b/client/go/internal/cli/cmd/feed.go
@@ -139,6 +139,7 @@ func feed(r io.Reader, cli *CLI, options feedOptions) error {
 		Route:       options.route,
 		TraceLevel:  options.traceLevel,
 		BaseURL:     service.BaseURL,
+		NowFunc:     cli.now,
 	}, clients)
 	throttler := document.NewThrottler(options.connections)
 	// TODO(mpolden): Make doom duration configurable
diff --git a/client/go/internal/cli/cmd/feed_test.go b/client/go/internal/cli/cmd/feed_test.go
index 521d2b2abd0..eb641005ab7 100644
--- a/client/go/internal/cli/cmd/feed_test.go
+++ b/client/go/internal/cli/cmd/feed_test.go
@@ -44,9 +44,9 @@ func TestFeed(t *testing.T) {
 
 	assert.Equal(t, "", stderr.String())
 	want := `{
-  "feeder.seconds": 1.000,
+  "feeder.seconds": 3.000,
   "feeder.ok.count": 1,
-  "feeder.ok.rate": 1.000,
+  "feeder.ok.rate": 0.333,
   "feeder.error.count": 0,
   "feeder.inflight.count": 0,
   "http.request.count": 1,
@@ -57,9 +57,9 @@ func TestFeed(t *testing.T) {
   "http.response.bytes": 16,
   "http.response.MBps": 0.000,
   "http.response.error.count": 0,
-  "http.response.latency.millis.min": 0,
-  "http.response.latency.millis.avg": 0,
-  "http.response.latency.millis.max": 0,
+  "http.response.latency.millis.min": 1000,
+  "http.response.latency.millis.avg": 1000,
+  "http.response.latency.millis.max": 1000,
   "http.response.code.counts": {
     "200": 1
   }
diff --git a/client/go/internal/vespa/document/dispatcher.go b/client/go/internal/vespa/document/dispatcher.go
index 5c99f3bf056..0f3d39d5a78 100644
--- a/client/go/internal/vespa/document/dispatcher.go
+++ b/client/go/internal/vespa/document/dispatcher.go
@@ -128,7 +128,6 @@ func (d *Dispatcher) start() {
 		return
 	}
 	d.listPool.New = func() any { return list.New() }
-	d.ready = make(chan Id, 4096)
 	d.results = make(chan Result, 4096)
 	d.msgs = make(chan string, 4096)
 	d.started = true
@@ -164,27 +163,18 @@ func (d *Dispatcher) enqueue(op documentOp) error {
 	}
 	d.mu.Unlock()
 	group.add(op, op.attempts > 0)
-	d.enqueueWithSlot(op.document.Id)
+	d.enqueueWithSlot(group)
 	return nil
 }
 
-func (d *Dispatcher) enqueueWithSlot(id Id) {
+func (d *Dispatcher) enqueueWithSlot(group *documentGroup) {
 	d.acquireSlot()
-	d.ready <- id
-	d.throttler.Sent()
-	d.dispatch()
-}
-
-func (d *Dispatcher) dispatch() {
 	d.workerWg.Add(1)
 	go func() {
 		defer d.workerWg.Done()
-		id := <-d.ready
-		d.mu.RLock()
-		group := d.inflight[id.String()]
-		d.mu.RUnlock()
 		d.sendDocumentIn(group)
 	}()
+	d.throttler.Sent()
 }
 
 func (d *Dispatcher) acquireSlot() {
diff --git a/client/go/internal/vespa/document/dispatcher_test.go b/client/go/internal/vespa/document/dispatcher_test.go
index d066f5bc9ae..c8f8e550ba4 100644
--- a/client/go/internal/vespa/document/dispatcher_test.go
+++ b/client/go/internal/vespa/document/dispatcher_test.go
@@ -130,3 +130,18 @@ func TestDispatcherOrderingWithFailures(t *testing.T) {
 	assert.Equal(t, int64(2), dispatcher.Stats().Errors)
 	assert.Equal(t, 6, len(feeder.documents))
 }
+
+func BenchmarkDocumentDispatching(b *testing.B) {
+	feeder := &mockFeeder{}
+	clock := &manualClock{tick: time.Second}
+	throttler := newThrottler(8, clock.now)
+	breaker := NewCircuitBreaker(time.Second, 0)
+	dispatcher := NewDispatcher(feeder, throttler, breaker, io.Discard, false)
+	doc := Document{Id: mustParseId("id:ns:type::doc1"), Operation: OperationPut, Body: []byte(`{"fields":{"foo": "123"}}`)}
+	b.ResetTimer() // ignore setup time
+
+	for n := 0; n < b.N; n++ {
+		dispatcher.enqueue(documentOp{document: doc})
+		dispatcher.workerWg.Wait()
+	}
+}
diff --git a/client/go/internal/vespa/document/http.go b/client/go/internal/vespa/document/http.go
index 51b6fa4de39..0530144747a 100644
--- a/client/go/internal/vespa/document/http.go
+++ b/client/go/internal/vespa/document/http.go
@@ -40,6 +40,7 @@ type ClientOptions struct {
 	Route       string
 	TraceLevel  int
 	Compression Compression
+	NowFunc     func() time.Time
 }
 
 type countingHTTPClient struct {
@@ -73,10 +74,14 @@ func NewClient(options ClientOptions, httpClients []util.HTTPClient) *Client {
 	for _, client := range httpClients {
 		countingClients = append(countingClients, countingHTTPClient{client: client})
 	}
+	nowFunc := options.NowFunc
+	if nowFunc == nil {
+		nowFunc = time.Now
+	}
 	return &Client{
 		options:     options,
 		httpClients: countingClients,
-		now:         time.Now,
+		now:         nowFunc,
 	}
 }