Repository: gocql/gocql (go)

Issue:
roundRobinHostPolicy uses lots of resources for every query
Please answer these questions before submitting your issue. Thanks!

### What version of Cassandra are you using?
3.11.5

### What version of Gocql are you using?
1982a06ad6b987c24beec15336019914d71b2e31

### What did you do?
Due to #1400 we were accidentally using the round robin host selection policy for every query.

### What did you expect to see?
I expected to see normal client side performance.

### What did you see instead?
I did CPU profile and noticed 10% CPU used by `rand.NewSource`. In addition, it allocates a lot, taking up ~30% CPU via GC.


Pull request description:
Allocation-Free Round Robin Policy (fixes #1401)
Copy-free and allocation-free version of round robin.

Uses offset increased by atomic package to select next query destination.

Fixes memory allocation in #1401 downing it to zero.

Patch:
diff --git a/README.md b/README.md
index a35d9f3..ccb4803 100644
--- a/README.md
+++ b/README.md
@@ -166,6 +166,22 @@ func main() {
 }
 ```
 
+
+Authentication 
+-------
+
+```go
+cluster := gocql.NewCluster("192.168.1.1", "192.168.1.2", "192.168.1.3")
+cluster.Authenticator = gocql.PasswordAuthenticator{
+	Username: "user",
+	Password: "password"
+}
+cluster.Keyspace = "example"
+cluster.Consistency = gocql.Quorum
+session, _ := cluster.CreateSession()
+defer session.Close()
+```
+
 Data Binding
 ------------
 
diff --git a/policies.go b/policies.go
index a93bd66..62d809c 100644
--- a/policies.go
+++ b/policies.go
@@ -6,8 +6,6 @@ package gocql
 
 import (
 	"context"
-	crand "crypto/rand"
-	"encoding/binary"
 	"errors"
 	"fmt"
 	"math"
@@ -335,7 +333,8 @@ func RoundRobinHostPolicy() HostSelectionPolicy {
 }
 
 type roundRobinHostPolicy struct {
-	hosts cowHostList
+	hosts           cowHostList
+	lastUsedHostIdx uint64
 }
 
 func (r *roundRobinHostPolicy) IsLocal(*HostInfo) bool              { return true }
@@ -343,26 +342,9 @@ func (r *roundRobinHostPolicy) KeyspaceChanged(KeyspaceUpdateEvent) {}
 func (r *roundRobinHostPolicy) SetPartitioner(partitioner string)   {}
 func (r *roundRobinHostPolicy) Init(*Session)                       {}
 
-var (
-	randPool = sync.Pool{
-		New: func() interface{} {
-			return rand.New(randSource())
-		},
-	}
-)
-
 func (r *roundRobinHostPolicy) Pick(qry ExecutableQuery) NextHost {
-	src := r.hosts.get()
-	hosts := make([]*HostInfo, len(src))
-	copy(hosts, src)
-
-	rand := randPool.Get().(*rand.Rand)
-	defer randPool.Put(rand)
-	rand.Shuffle(len(hosts), func(i, j int) {
-		hosts[i], hosts[j] = hosts[j], hosts[i]
-	})
-
-	return roundRobbin(hosts)
+	nextStartOffset := atomic.AddUint64(&r.lastUsedHostIdx, 1)
+	return roundRobbin(int(nextStartOffset), r.hosts.get())
 }
 
 func (r *roundRobinHostPolicy) AddHost(host *HostInfo) {
@@ -811,9 +793,10 @@ func (host selectedHostPoolHost) Mark(err error) {
 }
 
 type dcAwareRR struct {
-	local       string
-	localHosts  cowHostList
-	remoteHosts cowHostList
+	local           string
+	localHosts      cowHostList
+	remoteHosts     cowHostList
+	lastUsedHostIdx uint64
 }
 
 // DCAwareRoundRobinPolicy is a host selection policies which will prioritize and
@@ -850,55 +833,51 @@ func (d *dcAwareRR) RemoveHost(host *HostInfo) {
 func (d *dcAwareRR) HostUp(host *HostInfo)   { d.AddHost(host) }
 func (d *dcAwareRR) HostDown(host *HostInfo) { d.RemoveHost(host) }
 
-var randSeed int64
-
-func init() {
-	p := make([]byte, 8)
-	if _, err := crand.Read(p); err != nil {
-		panic(err)
-	}
-	randSeed = int64(binary.BigEndian.Uint64(p))
-}
-
-func randSource() rand.Source {
-	return rand.NewSource(atomic.AddInt64(&randSeed, 1))
-}
+// This function is supposed to be called in a fashion
+// roundRobbin(offset, hostsPriority1, hostsPriority2, hostsPriority3 ... )
+//
+// E.g. for DC-naive strategy:
+// roundRobbin(offset, allHosts)
+//
+// For tiered and DC-aware strategy:
+// roundRobbin(offset, localHosts, remoteHosts)
+func roundRobbin(shift int, hosts ...[]*HostInfo) NextHost {
+	currentLayer := 0
+	currentlyObserved := 0
 
-func roundRobbin(hosts []*HostInfo) NextHost {
-	var i int
 	return func() SelectedHost {
-		for i < len(hosts) {
-			h := hosts[i]
-			i++
 
-			if h.IsUp() {
-				return (*selectedHost)(h)
+		// iterate over layers
+		for {
+			if currentLayer == len(hosts) {
+				return nil
 			}
-		}
 
-		return nil
-	}
-}
+			currentLayerSize := len(hosts[currentLayer])
 
-func (d *dcAwareRR) Pick(q ExecutableQuery) NextHost {
-	local := d.localHosts.get()
-	remote := d.remoteHosts.get()
+			// iterate over hosts within a layer
+			for {
+				currentlyObserved++
+				if currentlyObserved > currentLayerSize {
+					currentLayer++
+					currentlyObserved = 0
+					break
+				}
 
-	hosts := make([]*HostInfo, len(local)+len(remote))
-	n := copy(hosts, local)
-	copy(hosts[n:], remote)
+				h := hosts[currentLayer][(shift+currentlyObserved)%currentLayerSize]
 
-	// TODO: use random chose-2 but that will require plumbing information
-	// about connection/host load to here
-	r := randPool.Get().(*rand.Rand)
-	defer randPool.Put(r)
-	for _, l := range [][]*HostInfo{hosts[:len(local)], hosts[len(local):]} {
-		r.Shuffle(len(l), func(i, j int) {
-			l[i], l[j] = l[j], l[i]
-		})
+				if h.IsUp() {
+					return (*selectedHost)(h)
+				}
+
+			}
+		}
 	}
+}
 
-	return roundRobbin(hosts)
+func (d *dcAwareRR) Pick(q ExecutableQuery) NextHost {
+	nextStartOffset := atomic.AddUint64(&d.lastUsedHostIdx, 1)
+	return roundRobbin(int(nextStartOffset), d.localHosts.get(), d.remoteHosts.get())
 }
 
 // ConvictionPolicy interface is used by gocql to determine if a host should be
@@ -953,10 +932,15 @@ func (c *ConstantReconnectionPolicy) GetMaxRetries() int {
 type ExponentialReconnectionPolicy struct {
 	MaxRetries      int
 	InitialInterval time.Duration
+	MaxInterval     time.Duration
 }
 
 func (e *ExponentialReconnectionPolicy) GetInterval(currentRetry int) time.Duration {
-	return getExponentialTime(e.InitialInterval, math.MaxInt16*time.Second, currentRetry)
+	max := e.MaxInterval
+	if max < e.InitialInterval {
+		max = math.MaxInt16 * time.Second
+	}
+	return getExponentialTime(e.InitialInterval, max, currentRetry)
 }
 
 func (e *ExponentialReconnectionPolicy) GetMaxRetries() int {
diff --git a/token.go b/token.go
index 8ab7b85..51d2b71 100644
--- a/token.go
+++ b/token.go
@@ -130,8 +130,15 @@ func (ht hostToken) String() string {
 // a data structure for organizing the relationship between tokens and hosts
 type tokenRing struct {
 	partitioner partitioner
-	tokens      []hostToken
-	hosts       []*HostInfo
+
+	// tokens map token range to primary replica.
+	// The elements in tokens are sorted by token ascending.
+	// The range for a given item in tokens starts after preceding range and ends with the token specified in
+	// token. The end token is part of the range.
+	// The lowest (i.e. index 0) range wraps around the ring (its preceding range is the one with largest index).
+	tokens []hostToken
+
+	hosts []*HostInfo
 }
 
 func newTokenRing(partitioner string, hosts []*HostInfo) (*tokenRing, error) {
diff --git a/topology.go b/topology.go
index 008f4a7..5ff7fd2 100644
--- a/topology.go
+++ b/topology.go
@@ -8,10 +8,16 @@ import (
 )
 
 type hostTokens struct {
+	// token is end (inclusive) of token range these hosts belong to
 	token token
 	hosts []*HostInfo
 }
 
+// tokenRingReplicas maps token ranges to list of replicas.
+// The elements in tokenRingReplicas are sorted by token ascending.
+// The range for a given item in tokenRingReplicas starts after preceding range and ends with the token specified in
+// token. The end token is part of the range.
+// The lowest (i.e. index 0) range wraps around the ring (its preceding range is the one with largest index).
 type tokenRingReplicas []hostTokens
 
 func (h tokenRingReplicas) Less(i, j int) bool { return h[i].token.Less(h[j].token) }
@@ -27,19 +33,9 @@ func (h tokenRingReplicas) replicasFor(t token) *hostTokens {
 		return !h[i].token.Less(t)
 	})
 
-	// TODO: simplify this
-	if p < len(h) && h[p].token == t {
-		return &h[p]
-	}
-
-	p--
-
 	if p >= len(h) {
 		// rollover
 		p = 0
-	} else if p < 0 {
-		// rollunder
-		p = len(h) - 1
 	}
 
 	return &h[p]


Test patch:
diff --git a/policies_test.go b/policies_test.go
index e65a11f..826b2cd 100644
--- a/policies_test.go
+++ b/policies_test.go
@@ -106,8 +106,8 @@ func TestHostPolicy_TokenAware_SimpleStrategy(t *testing.T) {
 	// now the token ring is configured
 	query.RoutingKey([]byte("20"))
 	iter = policy.Pick(query)
-	iterCheck(t, iter, "0")
 	iterCheck(t, iter, "1")
+	iterCheck(t, iter, "2")
 }
 
 // Tests of the host pool host selection policy implementation
@@ -608,7 +608,7 @@ func TestHostPolicy_TokenAware_NetworkStrategy(t *testing.T) {
 	}, policyInternal.getMetadataReadOnly().replicas)
 
 	// now the token ring is configured
-	query.RoutingKey([]byte("23"))
+	query.RoutingKey([]byte("18"))
 	iter = policy.Pick(query)
 	// first should be hosts with matching token from the local DC
 	iterCheck(t, iter, "4")