Use fastrand instead of prands (#4901)

In order to avoid top-level lock contention in the `math/rand` package,
we were previously initialising our own `prand` sources in various
structs. However these still needed to be protected by locks and they
also just weren't that fast.

This PR removes those `prand`s and replaces those calls with calls to
the Go runtime's `fastrand`, which is a couple hundred times faster and
is, more importantly, lock-free:
```
goos: darwin
goarch: arm64
pkg: github.com/nats-io/nats-server/v2/server/fastrand
BenchmarkFastRand32
BenchmarkFastRand32-10          1000000000               0.2564 ns/op
BenchmarkFastRand64
BenchmarkFastRand64-10          1000000000               0.5019 ns/op
BenchmarkDefaultRand32
BenchmarkDefaultRand32-10       11997840               104.4 ns/op
BenchmarkDefaultRand64
BenchmarkDefaultRand64-10       11745747               105.8 ns/op
PASS
ok      github.com/nats-io/nats-server/v2/server/fastrand       3.768s
```

Signed-off-by: Neil Twigg <neil@nats.io>

internal/fastrand/LICENSE

@@ -0,0 +1,27 @@
+Copyright (c) 2011 The LevelDB-Go Authors. All rights reserved.
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are
+met:
+
+   * Redistributions of source code must retain the above copyright
+notice, this list of conditions and the following disclaimer.
+   * Redistributions in binary form must reproduce the above
+copyright notice, this list of conditions and the following disclaimer
+in the documentation and/or other materials provided with the
+distribution.
+   * Neither the name of Google Inc. nor the names of its
+contributors may be used to endorse or promote products derived from
+this software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

internal/fastrand/fastrand.go

@@ -0,0 +1,23 @@
+// Copyright 2020-2023 The LevelDB-Go, Pebble and NATS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package fastrand
+
+import _ "unsafe" // required by go:linkname
+
+// Uint32 returns a lock free uint32 value.
+//
+//go:linkname Uint32 runtime.fastrand
+func Uint32() uint32
+
+// Uint32n returns a lock free uint32 value in the interval [0, n).
+//
+//go:linkname Uint32n runtime.fastrandn
+func Uint32n(n uint32) uint32
+
+// Uint32 returns a lock free uint64 value.
+func Uint64() uint64 {
+	v := uint64(Uint32())
+	return v<<32 | uint64(Uint32())
+}

internal/fastrand/fastrand_test.go

@@ -0,0 +1,72 @@
+// Copyright 2020-23 The LevelDB-Go, Pebble and NATS Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+
+package fastrand
+
+import (
+	"math/rand"
+	"sync"
+	"testing"
+	"time"
+)
+
+type defaultRand struct {
+	mu  sync.Mutex
+	src rand.Source64
+}
+
+func newDefaultRand() *defaultRand {
+	r := &defaultRand{
+		src: rand.New(rand.NewSource(time.Now().UnixNano())),
+	}
+	return r
+}
+
+func (r *defaultRand) Uint32() uint32 {
+	r.mu.Lock()
+	i := uint32(r.src.Uint64())
+	r.mu.Unlock()
+	return i
+}
+
+func (r *defaultRand) Uint64() uint64 {
+	r.mu.Lock()
+	i := uint64(r.src.Uint64())
+	r.mu.Unlock()
+	return i
+}
+
+func BenchmarkFastRand32(b *testing.B) {
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			Uint32()
+		}
+	})
+}
+
+func BenchmarkFastRand64(b *testing.B) {
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			Uint64()
+		}
+	})
+}
+
+func BenchmarkDefaultRand32(b *testing.B) {
+	r := newDefaultRand()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			r.Uint32()
+		}
+	})
+}
+
+func BenchmarkDefaultRand64(b *testing.B) {
+	r := newDefaultRand()
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			r.Uint64()
+		}
+	})
+}

server/accounts.go

@@ -33,6 +33,7 @@ import (
 	"time"
 
 	"github.com/nats-io/jwt/v2"
+	"github.com/nats-io/nats-server/v2/internal/fastrand"
 	"github.com/nats-io/nkeys"
 	"github.com/nats-io/nuid"
 )
@@ -85,10 +86,9 @@ type Account struct {
 	incomplete   bool
 	signingKeys  map[string]jwt.Scope
 	extAuth      *jwt.ExternalAuthorization
-	srv          *Server    // server this account is registered with (possibly nil)
-	lds          string     // loop detection subject for leaf nodes
-	siReply      []byte     // service reply prefix, will form wildcard subscription.
-	prand        *rand.Rand // NOT threadsafe, must have WRITE lock on Account
+	srv          *Server // server this account is registered with (possibly nil)
+	lds          string  // loop detection subject for leaf nodes
+	siReply      []byte  // service reply prefix, will form wildcard subscription.
 	eventIds     *nuid.NUID
 	eventIdsMu   sync.Mutex
 	defaultPerms *Permissions
@@ -238,7 +238,6 @@ func NewAccount(name string) *Account {
 		Name:     name,
 		limits:   limits{-1, -1, -1, -1, false},
 		eventIds: nuid.New(),
-		prand:    rand.New(rand.NewSource(time.Now().UnixNano())),
 	}
 	return a
 }
@@ -290,7 +289,6 @@ func (a *Account) shallowCopy(na *Account) {
 		}
 	}
 	na.mappings = a.mappings
-	na.prand = rand.New(rand.NewSource(time.Now().UnixNano()))
 	na.hasMapped.Store(len(na.mappings) > 0)
 
 	// JetStream
@@ -802,7 +800,7 @@ func (a *Account) selectMappedSubject(dest string) (string, bool) {
 	if len(dests) == 1 && dests[0].weight == 100 {
 		d = dests[0]
 	} else {
-		w := uint8(a.prand.Int31n(100))
+		w := uint8(fastrand.Uint32n(100))
 		for _, rm := range dests {
 			if w < rm.weight {
 				d = rm
@@ -2186,7 +2184,7 @@ func (a *Account) processServiceImportResponse(sub *subscription, c *client, _ *
 // Lock should be held.
 func (a *Account) createRespWildcard() {
 	var b = [baseServerLen]byte{'_', 'R', '_', '.'}
-	rn := a.prand.Uint64()
+	rn := fastrand.Uint64()
 	for i, l := replyPrefixLen, rn; i < len(b); i++ {
 		b[i] = digits[l%base]
 		l /= base
@@ -2205,7 +2203,7 @@ func isTrackedReply(reply []byte) bool {
 func (a *Account) newServiceReply(tracking bool) []byte {
 	a.mu.Lock()
 	s := a.srv
-	rn := a.prand.Uint64()
+	rn := fastrand.Uint64()
 
 	// Check if we need to create the reply here.
 	var createdSiReply bool

server/client.go

@@ -35,6 +35,7 @@ import (
 
 	"github.com/klauspost/compress/s2"
 	"github.com/nats-io/jwt/v2"
+	"github.com/nats-io/nats-server/v2/internal/fastrand"
 )
 
 // Type of client connection.
@@ -442,8 +443,6 @@ type readCache struct {
 	// to make sure to only send one message and properly scope to queues as needed.
 	rts []routeTarget
 
-	prand *rand.Rand
-
 	// These are all temporary totals for an invocation of a read in readloop.
 	msgs  int32
 	bytes int32
@@ -4505,12 +4504,6 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 		goto sendToRoutesOrLeafs
 	}
 
-	// Check to see if we have our own rand yet. Global rand
-	// has contention with lots of clients, etc.
-	if c.in.prand == nil {
-		c.in.prand = rand.New(rand.NewSource(time.Now().UnixNano()))
-	}
-
 	// Process queue subs
 	for i := 0; i < len(r.qsubs); i++ {
 		qsubs := r.qsubs[i]
@@ -4558,7 +4551,7 @@ func (c *client) processMsgResults(acc *Account, r *SublistResult, msg, deliver,
 		sindex := 0
 		lqs := len(qsubs)
 		if lqs > 1 {
-			sindex = c.in.prand.Int() % lqs
+			sindex = int(fastrand.Uint32()) % lqs
 		}
 
 		// Find a subscription that is able to deliver this message starting at a random index.

server/raft.go

@@ -31,6 +31,8 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/nats-io/nats-server/v2/internal/fastrand"
+
 	"github.com/minio/highwayhash"
 )
 
@@ -202,8 +204,6 @@ type raft struct {
 	stepdown *ipQueue[string]               // Stepdown requests
 	leadc    chan bool                      // Leader changes
 	quit     chan struct{}                  // Raft group shutdown
-
-	prand *rand.Rand // Random generator, used to generate inboxes for instance
 }
 
 // cacthupState structure that holds our subscription, and catchup term and index
@@ -349,7 +349,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 	sq := s.sys.sq
 	sacc := s.sys.account
 	hash := s.sys.shash
-	pub := s.info.ID
 	s.mu.RUnlock()
 
 	// Do this here to process error quicker.
@@ -362,12 +361,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 	}
 
 	qpfx := fmt.Sprintf("[ACC:%s] RAFT '%s' ", accName, cfg.Name)
-	rsrc := time.Now().UnixNano()
-	if len(pub) >= 32 {
-		if h, _ := highwayhash.New64([]byte(pub[:32])); h != nil {
-			rsrc += int64(h.Sum64())
-		}
-	}
 	n := &raft{
 		created:  time.Now(),
 		id:       hash[:idLen],
@@ -399,7 +392,6 @@ func (s *Server) startRaftNode(accName string, cfg *RaftConfig, labels pprofLabe
 		leadc:    make(chan bool, 1),
 		observer: cfg.Observer,
 		extSt:    ps.domainExt,
-		prand:    rand.New(rand.NewSource(rsrc)),
 	}
 	n.c.registerWithAccount(sacc)
 
@@ -1685,7 +1677,7 @@ const (
 // Lock should be held (due to use of random generator)
 func (n *raft) newCatchupInbox() string {
 	var b [replySuffixLen]byte
-	rn := n.prand.Int63()
+	rn := fastrand.Uint64()
 	for i, l := 0, rn; i < len(b); i++ {
 		b[i] = digits[l%base]
 		l /= base
@@ -1695,7 +1687,7 @@ func (n *raft) newCatchupInbox() string {
 
 func (n *raft) newInbox() string {
 	var b [replySuffixLen]byte
-	rn := n.prand.Int63()
+	rn := fastrand.Uint64()
 	for i, l := 0, rn; i < len(b); i++ {
 		b[i] = digits[l%base]
 		l /= base