changed numeric aggregators to use float64

Author: htemelski-redis

go.mod

@@ -9,6 +9,8 @@ require (
 	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f
 )
 
+require go.uber.org/atomic v1.11.0
+
 retract (
 	v9.15.1 // This version is used to retract v9.15.0
 	v9.15.0 // This version was accidentally released. It is identical to 9.15.0-beta.2

go.sum

@@ -4,5 +4,10 @@ github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=
 github.com/bsm/gomega v1.27.10/go.mod h1:JyEr/xRbxbtgWNi8tIEVPUYZ5Dzef52k01W3YH0H+O0=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
 github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f h1:lO4WD4F/rVNCu3HqELle0jiPLLBs70cWOduZpkS1E78=
 github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f/go.mod h1:cuUVRXasLTGF7a8hSLbxyZXjz+1KgoB3wDUb6vlszIc=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
+go.uber.org/atomic v1.11.0 h1:ZvwS0R+56ePWxUNi+Atn9dWONBPp/AUETXlHW0DxSjE=
+go.uber.org/atomic v1.11.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0=

internal/routing/aggregator.go

@@ -5,9 +5,11 @@ import (
 	"fmt"
 	"math"
 	"sync"
+
 	"sync/atomic"
 
 	"github.com/redis/go-redis/v9/internal/util"
+	uberAtomic "go.uber.org/atomic"
 )
 
 var (
@@ -214,7 +216,7 @@ func (a *OneSucceededAggregator) Result() (interface{}, error) {
 // AggSumAggregator sums numeric replies from all shards.
 type AggSumAggregator struct {
 	err atomic.Value
-	res int64
+	res uberAtomic.Float64
 }
 
 func (a *AggSumAggregator) Add(result interface{}, err error) error {
@@ -223,12 +225,12 @@ func (a *AggSumAggregator) Add(result interface{}, err error) error {
 	}
 
 	if result != nil {
-		val, err := toInt64(result)
+		val, err := toFloat64(result)
 		if err != nil {
 			a.err.CompareAndSwap(nil, err)
 			return err
 		}
-		atomic.AddInt64(&a.res, val)
+		a.res.Add(val)
 	}
 
 	return nil
@@ -277,7 +279,7 @@ func (a *AggSumAggregator) BatchSlice(results []AggregatorResErr) error {
 }
 
 func (a *AggSumAggregator) Result() (interface{}, error) {
-	res, err := atomic.LoadInt64(&a.res), a.err.Load()
+	res, err := a.res.Load(), a.err.Load()
 	if err != nil {
 		return nil, err.(error)
 	}
@@ -297,13 +299,13 @@ func (a *AggMinAggregator) Add(result interface{}, err error) error {
 		return nil
 	}
 
-	intVal, e := toInt64(result)
+	floatVal, e := toFloat64(result)
 	if e != nil {
 		a.err.CompareAndSwap(nil, err)
 		return nil
 	}
 
-	a.res.Value(intVal)
+	a.res.Value(floatVal)
 
 	return nil
 }
@@ -337,22 +339,22 @@ func (a *AggMinAggregator) AddWithKey(key string, result interface{}, err error)
 }
 
 func (a *AggMinAggregator) BatchSlice(results []AggregatorResErr) error {
-	min := int64(math.MaxInt64)
+	min := float64(math.MaxFloat64)
 
 	for _, res := range results {
 		if res.Err != nil {
 			_ = a.Add(nil, res.Err)
 			return nil
 		}
 
-		resInt, err := toInt64(res.Result)
+		floatVal, err := toFloat64(res.Result)
 		if err != nil {
 			_ = a.Add(nil, res.Err)
 			return nil
 		}
 
-		if resInt < min {
-			min = resInt
+		if floatVal < min {
+			min = floatVal
 		}
 
 	}
@@ -385,13 +387,13 @@ func (a *AggMaxAggregator) Add(result interface{}, err error) error {
 		return nil
 	}
 
-	intVal, e := toInt64(result)
+	floatVal, e := toFloat64(result)
 	if e != nil {
 		a.err.CompareAndSwap(nil, err)
 		return nil
 	}
 
-	a.res.Value(intVal)
+	a.res.Value(floatVal)
 
 	return nil
 }
@@ -650,6 +652,27 @@ func toInt64(val interface{}) (int64, error) {
 	}
 }
 
+func toFloat64(val interface{}) (float64, error) {
+	if val == nil {
+		return 0, nil
+	}
+
+	switch v := val.(type) {
+	case float64:
+		return v, nil
+	case int:
+		return float64(v), nil
+	case int32:
+		return float64(v), nil
+	case int64:
+		return float64(v), nil
+	case float32:
+		return float64(v), nil
+	default:
+		return 0, fmt.Errorf("cannot convert %T to float64", val)
+	}
+}
+
 func toBool(val interface{}) (bool, error) {
 	if val == nil {
 		return false, nil

internal/util/atomic_max.go

@@ -3,14 +3,15 @@
 ISC License
 
 Modified by htemelski-redis
-Removed the treshold, adapted it to work with int64
+Removed the treshold, adapted it to work with float64
 */
 
 package util
 
 import (
 	"math"
-	"sync/atomic"
+
+	"go.uber.org/atomic"
 )
 
 // AtomicMax is a thread-safe max container
@@ -22,7 +23,7 @@ import (
 type AtomicMax struct {
 
 	// value is current max
-	value atomic.Int64
+	value atomic.Float64
 	// whether [AtomicMax.Value] has been invoked
 	// with value equal or greater to threshold
 	hasValue atomic.Bool
@@ -32,7 +33,7 @@ type AtomicMax struct {
 //   - if threshold is not used, AtomicMax is initialization-free
 func NewAtomicMax() (atomicMax *AtomicMax) {
 	m := AtomicMax{}
-	m.value.Store(math.MinInt64)
+	m.value.Store((-math.MaxFloat64))
 	return &m
 }
 
@@ -44,14 +45,14 @@ func NewAtomicMax() (atomicMax *AtomicMax) {
 //   - upon return, Max and Max1 are guaranteed to reflect the invocation
 //   - the return order of concurrent Value invocations is not guaranteed
 //   - Thread-safe
-func (m *AtomicMax) Value(value int64) (isNewMax bool) {
-	//  math.MinInt64 as max case
+func (m *AtomicMax) Value(value float64) (isNewMax bool) {
+	//  -math.MaxFloat64 as max case
 	var hasValue0 = m.hasValue.Load()
-	if value == math.MinInt64 {
+	if value == (-math.MaxFloat64) {
 		if !hasValue0 {
 			isNewMax = m.hasValue.CompareAndSwap(false, true)
 		}
-		return // math.MinInt64 as max: isNewMax true for first 0 writer
+		return // -math.MaxFloat64 as max: isNewMax true for first 0 writer
 	}
 
 	// check against present value
@@ -82,7 +83,7 @@ func (m *AtomicMax) Value(value int64) (isNewMax bool) {
 //   - hasValue true indicates that value reflects a Value invocation
 //   - hasValue false: value is zero-value
 //   - Thread-safe
-func (m *AtomicMax) Max() (value int64, hasValue bool) {
+func (m *AtomicMax) Max() (value float64, hasValue bool) {
 	if hasValue = m.hasValue.Load(); !hasValue {
 		return
 	}
@@ -93,4 +94,4 @@ func (m *AtomicMax) Max() (value int64, hasValue bool) {
 // Max1 returns current maximum whether zero-value or set by Value
 //   - threshold is ignored
 //   - Thread-safe
-func (m *AtomicMax) Max1() (value int64) { return m.value.Load() }
+func (m *AtomicMax) Max1() (value float64) { return m.value.Load() }

internal/util/atomic_min.go

@@ -10,7 +10,8 @@ Adapted from the modified atomic_max, but with inverted logic
 
 import (
 	"math"
-	"sync/atomic"
+
+	"go.uber.org/atomic"
 )
 
 // AtomicMin is a thread-safe Min container
@@ -21,7 +22,7 @@ import (
 type AtomicMin struct {
 
 	// value is current Min
-	value atomic.Int64
+	value atomic.Float64
 	// whether [AtomicMin.Value] has been invoked
 	// with value equal or greater to threshold
 	hasValue atomic.Bool
@@ -31,7 +32,7 @@ type AtomicMin struct {
 //   - if threshold is not used, AtomicMin is initialization-free
 func NewAtomicMin() (atomicMin *AtomicMin) {
 	m := AtomicMin{}
-	m.value.Store(math.MaxInt64)
+	m.value.Store(math.MaxFloat64)
 	return &m
 }
 
@@ -43,14 +44,14 @@ func NewAtomicMin() (atomicMin *AtomicMin) {
 //   - upon return, Min and Min1 are guaranteed to reflect the invocation
 //   - the return order of concurrent Value invocations is not guaranteed
 //   - Thread-safe
-func (m *AtomicMin) Value(value int64) (isNewMin bool) {
-	//  math.MaxInt64 as Min case
+func (m *AtomicMin) Value(value float64) (isNewMin bool) {
+	//  math.MaxFloat64 as Min case
 	var hasValue0 = m.hasValue.Load()
-	if value == math.MaxInt64 {
+	if value == math.MaxFloat64 {
 		if !hasValue0 {
 			isNewMin = m.hasValue.CompareAndSwap(false, true)
 		}
-		return // math.MaxInt64 as Min: isNewMin true for first 0 writer
+		return // math.MaxFloat64 as Min: isNewMin true for first 0 writer
 	}
 
 	// check against present value
@@ -81,7 +82,7 @@ func (m *AtomicMin) Value(value int64) (isNewMin bool) {
 //   - hasValue true indicates that value reflects a Value invocation
 //   - hasValue false: value is zero-value
 //   - Thread-safe
-func (m *AtomicMin) Min() (value int64, hasValue bool) {
+func (m *AtomicMin) Min() (value float64, hasValue bool) {
 	if hasValue = m.hasValue.Load(); !hasValue {
 		return
 	}
@@ -92,4 +93,4 @@ func (m *AtomicMin) Min() (value int64, hasValue bool) {
 // Min1 returns current Minimum whether zero-value or set by Value
 //   - threshold is ignored
 //   - Thread-safe
-func (m *AtomicMin) Min1() (value int64) { return m.value.Load() }
+func (m *AtomicMin) Min1() (value float64) { return m.value.Load() }