Cherry-pick PRs into v2.10.8 release branch (#4912)

Includes:

#4901
#4903
#4905
#4909
#4910

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/filestore.go

@@ -795,7 +795,7 @@ var blkPoolSmall sync.Pool  // 2MB
 
 // Get a new msg block based on sz estimate.
 func getMsgBlockBuf(sz int) (buf []byte) {
-	var pb interface{}
+	var pb any
 	if sz <= defaultSmallBlockSize {
 		pb = blkPoolSmall.Get()
 	} else if sz <= defaultMediumBlockSize {
@@ -6132,21 +6132,23 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 	for i := 0; i < len(fs.blks); i++ {
 		mb := fs.blks[i]
 		mb.mu.Lock()
-		if err := mb.ensurePerSubjectInfoLoaded(); err != nil {
-			mb.mu.Unlock()
-			continue
+		var shouldExpire bool
+		if mb.cacheNotLoaded() {
+			mb.loadMsgsWithLock()
+			shouldExpire = true
 		}
+
 		t, f, l := mb.filteredPendingLocked(subject, wc, atomic.LoadUint64(&mb.first.seq))
 		if t == 0 {
+			// Expire if we were responsible for loading.
+			if shouldExpire {
+				// Expire this cache before moving on.
+				mb.tryForceExpireCacheLocked()
+			}
 			mb.mu.Unlock()
 			continue
 		}
 
-		var shouldExpire bool
-		if mb.cacheNotLoaded() {
-			mb.loadMsgsWithLock()
-			shouldExpire = true
-		}
 		if sequence > 1 && sequence <= l {
 			l = sequence - 1
 		}

server/filestore_test.go

@@ -6286,6 +6286,38 @@ func TestFileStoreCorruptPSIMOnDisk(t *testing.T) {
 	require_True(t, bytes.Equal(sm.msg, []byte("XYZ")))
 }
 
+func TestFileStorePurgeExBufPool(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, BlockSize: 1024},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := bytes.Repeat([]byte("ABC"), 33) // ~100bytes
+	for i := 0; i < 1000; i++ {
+		fs.StoreMsg("foo.foo", nil, msg)
+		fs.StoreMsg("foo.bar", nil, msg)
+	}
+
+	p, err := fs.PurgeEx("foo.bar", 1, 0)
+	require_NoError(t, err)
+	require_Equal(t, p, 1000)
+
+	// Now make sure we do not have all of the msg blocks cache's loaded.
+	var loaded int
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.mu.RLock()
+		if mb.cacheAlreadyLoaded() {
+			loaded++
+		}
+		mb.mu.RUnlock()
+	}
+	fs.mu.RUnlock()
+	require_Equal(t, loaded, 1)
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////