GOCBC-1827: Range scan - Use atomic types for variables intended to only be used with atomic operations

Motivation
==========
We use a number of variables in the range scan code which are only ever supposed to be used with atomic operations. We should use the more ergonomic atomic types, to avoid any accidental non-atomic usage of these variables. This also allows us to remove the direct dependency of gocb on the "unsafe" package.

These were added with Go 1.19 that we still supported when range scan was added, but since we no longer maintain compatibility with Go 1.19 we can now make use of these types.

Changes
=======
* Use appropriate atomic types for the various counters that are used with atomic operations
* Replace use of unsafe.Pointer for the peeked scan item with atomic.Pointer[ScanResultItem]

Change-Id: If0ef05dfc322c3ef3d2b1f42ba366918cedffda6
Reviewed-on: https://review.couchbase.org/c/gocb/+/245686
Tested-by: Build Bot <build@couchbase.com>
Reviewed-by: Charles Dixon <chvckd@gmail.com>

Author: DemetrisChr

rangescanopmanager.go

@@ -10,7 +10,6 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
-	"unsafe"
 
 	"github.com/couchbase/gocbcore/v10"
 )
@@ -60,7 +59,7 @@ type rangeScanOpManager struct {
 
 	result *ScanResult
 
-	cancelled uint32
+	cancelled atomic.Bool
 }
 
 type rangeScanVbucket struct {
@@ -344,7 +343,7 @@ func (m *rangeScanOpManager) IsRangeScan() bool {
 }
 
 func (m *rangeScanOpManager) cancelScan(err error) {
-	if atomic.CompareAndSwapUint32(&m.cancelled, 0, 1) {
+	if m.cancelled.CompareAndSwap(false, true) {
 		if err != nil {
 			m.result.setErr(err)
 		}
@@ -373,27 +372,26 @@ func (m *rangeScanOpManager) Scan(ctx context.Context) (*ScanResult, error) {
 
 	balancer := m.createLoadBalancer()
 
-	var complete uint32
-	var seenData uint32
-
 	// We keep separate counts of running and completed to simplify shutdown of the scan.
-	scansRunning := int32(m.maxConcurrency)
+	var complete atomic.Uint32
+	var scansRunning atomic.Int32
+	scansRunning.Store(int32(m.maxConcurrency))
 
 	isRangeScan := m.IsRangeScan()
 
 	var i uint16
 	for i = 0; i < m.maxConcurrency; i++ {
 		go func() {
 			defer func() {
-				if atomic.AddUint32(&complete, 1) == uint32(m.maxConcurrency) {
+				if complete.Add(1) == uint32(m.maxConcurrency) {
 					m.Finish()
 					balancer.close()
 					close(resultCh)
 				}
 			}()
 
 			for vbucket, ok := balancer.selectVbucket(); ok; vbucket, ok = balancer.selectVbucket() {
-				if atomic.LoadUint32(&m.cancelled) == 1 {
+				if m.cancelled.Load() {
 					return
 				}
 
@@ -415,7 +413,7 @@ func (m *rangeScanOpManager) Scan(ctx context.Context) (*ScanResult, error) {
 							if errors.Is(err, gocbcore.ErrBusy) {
 								// Busy indicates that the server is reporting too many active scans.
 								// Shut ourselves down if we're not the only runner remaining.
-								running := atomic.AddInt32(&scansRunning, -1)
+								running := scansRunning.Add(-1)
 								if running >= 1 {
 									// Shutdown this worker.
 									logDebugf("Shutting down scan runner, remaining %d", running)
@@ -479,8 +477,7 @@ func (m *rangeScanOpManager) Scan(ctx context.Context) (*ScanResult, error) {
 	case item, more := <-resultCh:
 		// more could be false if no sampling scans returned any data, but that isn't an error case.
 		if more {
-			atomic.StoreUint32(&seenData, 1)
-			r.peeked = unsafe.Pointer(item) //nolint:gosec
+			r.peeked.Store(item)
 		}
 	}
 
@@ -567,7 +564,7 @@ func (m *rangeScanOpManager) scanPartition(ctx context.Context, deadline time.Ti
 				return 0, nil
 			}
 		}
-		if atomic.LoadUint32(&m.cancelled) == 1 {
+		if m.cancelled.Load() {
 			m.cancelStream(ctx, span, deadline, createRes)
 			return 0, nil
 		}
@@ -775,15 +772,15 @@ func (b *rangeScanLoadBalancer) selectVbucket() (rangeScanVbucket, bool) {
 
 	var selectedServer int
 	selected := false
-	min := uint32(math.MaxUint32)
+	minimum := uint32(math.MaxUint32)
 
 	for s := range b.servers {
 		if len(b.vbucketChannels[s]) == 0 {
 			continue
 		}
 		activeScans := b.numActiveScans(s).Load()
-		if activeScans < min {
-			min = activeScans
+		if activeScans < minimum {
+			minimum = activeScans
 			selectedServer = s
 			selected = true
 		}

rangescanopmanager_test.go

@@ -229,11 +229,11 @@ func (r *rangeScanCreateResult) RangeScanCancel(opts gocbcore.RangeScanCancelOpt
 func (suite *UnitTestSuite) TestScanFirstCreateFailsUnknownError() {
 	expectedErr := errors.New("unknown error")
 	test := func(scan ScanType) (*ScanResult, error) {
-		var calls uint32
+		var calls atomic.Uint32
 		provider := makeRangeScanProvider(func(args mock.Arguments) {
 			cb := args.Get(2).(gocbcore.RangeScanCreateCallback)
 
-			if atomic.AddUint32(&calls, 1) == 1 {
+			if calls.Add(1) == 1 {
 				cb(nil, expectedErr)
 				return
 			}
@@ -285,11 +285,11 @@ func (suite *UnitTestSuite) TestScanFirstCreateFailsUnknownError() {
 func (suite *UnitTestSuite) TestScanSecondCreateFailsUnknownError() {
 	expectedErr := errors.New("unknown error")
 	test := func(scan ScanType) (*ScanResult, error) {
-		var calls uint32
+		var calls atomic.Uint32
 		provider := makeRangeScanProvider(func(args mock.Arguments) {
 			cb := args.Get(2).(gocbcore.RangeScanCreateCallback)
 
-			if atomic.AddUint32(&calls, 1) == 2 {
+			if calls.Add(1) == 2 {
 				cb(nil, expectedErr)
 				return
 			}
@@ -349,15 +349,15 @@ func (suite *UnitTestSuite) TestScanNMV() {
 	keyAfter := "keyafter"
 	test := func(scan ScanType) (map[string]struct{}, string, error) {
 		var firstTermAfter string
-		var calls uint32
+		var calls atomic.Uint32
 		var vbID int16 = -1
 		var createCalls uint32
 		provider := makeRangeScanProvider(func(args mock.Arguments) {
 			thisVBID := int16(args.Get(0).(uint16))
 			opts := args.Get(1).(gocbcore.RangeScanCreateOptions)
 			cb := args.Get(2).(gocbcore.RangeScanCreateCallback)
 
-			if atomic.AddUint32(&calls, 1) == 2 || thisVBID == vbID {
+			if calls.Add(1) == 2 || thisVBID == vbID {
 				vbID = thisVBID
 				if createCalls == 0 {
 					createCalls++

results.go

@@ -6,7 +6,6 @@ import (
 	"sync"
 	"sync/atomic"
 	"time"
-	"unsafe"
 
 	"github.com/couchbase/gocbcore/v10/memd"
 )
@@ -461,9 +460,9 @@ type ScanResult struct {
 	errLock    sync.Mutex
 
 	limit    uint64
-	numItems uint64
+	numItems atomic.Uint64
 
-	peeked unsafe.Pointer
+	peeked atomic.Pointer[ScanResultItem]
 }
 
 func (sr *ScanResult) setErr(err error) {
@@ -474,10 +473,10 @@ func (sr *ScanResult) setErr(err error) {
 
 // Next returns the next item on the stream, if there are no items remaining then nil is returned.
 func (sr *ScanResult) Next() *ScanResultItem {
-	peeked := atomic.SwapPointer(&sr.peeked, nil)
+	peeked := sr.peeked.Swap(nil)
 	if peeked != nil {
-		atomic.AddUint64(&sr.numItems, 1)
-		return (*ScanResultItem)(peeked)
+		sr.numItems.Add(1)
+		return peeked
 	}
 
 	item, more := <-sr.resultChan
@@ -488,7 +487,7 @@ func (sr *ScanResult) Next() *ScanResultItem {
 	// If we're doing a sampling scan then we need to only write data into the channel
 	// if we haven't seen the number of items that the user requested. Otherwise
 	// we need to cancel the streams
-	numItems := atomic.AddUint64(&sr.numItems, 1)
+	numItems := sr.numItems.Add(1)
 	if sr.limit == 0 || numItems <= sr.limit {
 		return item
 	}