client: Add massively concurrent test, reduce contention

Add a massively concurrent unit test.

Reduce lock contention by only waking up one waiter when a matching
response has been received and unanswered requests remain.

Signed-off-by: Jarno Rajahalme <jarno@isovalent.com>

shared_client.go

@@ -49,7 +49,6 @@ func (s *SharedClients) GetSharedClient(key string, conf *Client, serverAddrStr
 	return client, func() {
 		s.Lock()
 		defer s.Unlock()
-
 		client.refcount--
 		if client.refcount == 0 {
 			// Make client unreachable and close it's connection.
@@ -185,13 +184,17 @@ func (c *SharedClient) ExchangeSharedContext(ctx context.Context, m *Msg) (r *Ms
 			// Releasing the reader lock before sending errors on waiter's channels
 			// so that when they get them, one of them can take the reader lock.
 			c.readerLock.Unlock()
-			for id, ch := range c.reqs {
-				// Another reader will pick up if any errNoReader errors are sent.
-				// Only delete the pending request in other error cases.
-				if !errors.Is(err, errNoReader) {
+			if errors.Is(err, errNoReader) {
+				// only wake one waiting requester to do the reading
+				for _, ch := range c.reqs {
+					ch <- Response{err: err}
+					break
+				}
+			} else {
+				for id, ch := range c.reqs {
 					delete(c.reqs, id)
+					ch <- Response{err: err}
 				}
-				ch <- Response{err: err}
 			}
 		}
 		// Get the response of error from the current reader.

shared_client_test.go

@@ -6,9 +6,11 @@ package dns
 import (
 	"context"
 	"crypto/tls"
+	"fmt"
 	"net"
 	"strconv"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 )
@@ -58,6 +60,7 @@ func TestSharedClientSync(t *testing.T) {
 	if c2 != c {
 		t.Fatal("client not really shared")
 	}
+	m.Id = uint16(42)
 	r, _, err = c2.ExchangeShared(m)
 	if err != nil {
 		t.Errorf("failed to exchange: %v", err)
@@ -67,6 +70,60 @@ func TestSharedClientSync(t *testing.T) {
 	}
 }
 
+func TestSharedClientConcurrentSync(t *testing.T) {
+	HandleFunc("miek.nl.", HelloServer)
+	defer HandleRemove("miek.nl.")
+
+	s, addrstr, _, err := RunLocalUDPServer(":0")
+	if err != nil {
+		t.Fatalf("unable to run test server: %v", err)
+	}
+	defer s.Shutdown()
+
+	conf := &Client{
+		Timeout: time.Second,
+	}
+
+	errors := sync.Map{}
+	var wg sync.WaitGroup
+	f1 := func(id uint16) {
+		defer wg.Done()
+
+		m := new(Msg)
+		m.SetQuestion("miek.nl.", TypeSOA)
+		m.Id = id
+
+		c, closer := clients.GetSharedClient("concurrent-client", conf, addrstr)
+		defer closer()
+		r, _, err := c.ExchangeShared(m)
+		if err != nil {
+			errors.Store(id, fmt.Errorf("failed to exchange: %v", err))
+		} else {
+			if r == nil {
+				errors.Store(id, fmt.Errorf("response is nil"))
+			} else {
+				if r.Id != id {
+					errors.Store(id, fmt.Errorf("incorrect id (%d != %d)", r.Id, id))
+				}
+				if r.Rcode != RcodeSuccess {
+					errors.Store(id, fmt.Errorf("failed to get an valid answer\n%v", r))
+				}
+			}
+		}
+	}
+
+	for id := uint16(1); id <= 100; id++ {
+		wg.Add(1)
+		go f1(id)
+	}
+	wg.Wait()
+
+	errors.Range(func(key, value any) bool {
+		t.Fatalf("Id: %v, error: %v", key, value)
+		return false
+	})
+}
+
 func TestSharedClientLocalAddress(t *testing.T) {
 	HandleFunc("miek.nl.", HelloServerEchoAddrPort)
 	defer HandleRemove("miek.nl.")