Merge pull request #292 from sylwiaszunejko/debounce_fill

Start debouncing pool.fill() calls

connectionpool.go

@@ -10,6 +10,8 @@ import (
 	"net"
 	"sync"
 	"time"
+
+	"github.com/gocql/gocql/debounce"
 )
 
 // interface to implement to receive the host information
@@ -203,7 +205,7 @@ func (p *policyConnPool) addHost(host *HostInfo) {
 	}
 	p.mu.Unlock()
 
-	pool.fill()
+	pool.fill_debounce()
 }
 
 func (p *policyConnPool) removeHost(hostID string) {
@@ -232,6 +234,7 @@ type hostConnPool struct {
 	connPicker ConnPicker
 	closed     bool
 	filling    bool
+	debouncer  *debounce.SimpleDebouncer
 
 	logger StdLogger
 }
@@ -256,6 +259,7 @@ func newHostConnPool(session *Session, host *HostInfo, port, size int,
 		filling:    false,
 		closed:     false,
 		logger:     session.logger,
+		debouncer:  debounce.NewSimpleDebouncer(),
 	}
 
 	// the pool is not filled or connected
@@ -274,7 +278,7 @@ func (pool *hostConnPool) Pick(token Token, qry ExecutableQuery) *Conn {
 	size, missing := pool.connPicker.Size()
 	if missing > 0 {
 		// try to fill the pool
-		go pool.fill()
+		go pool.fill_debounce()
 
 		if size == 0 {
 			return nil
@@ -385,6 +389,10 @@ func (pool *hostConnPool) fill() {
 	}()
 }
 
+func (pool *hostConnPool) fill_debounce() {
+	pool.debouncer.Debounce(pool.fill)
+}
+
 func (pool *hostConnPool) logConnectErr(err error) {
 	if opErr, ok := err.(*net.OpError); ok && (opErr.Op == "dial" || opErr.Op == "read") {
 		// connection refused
@@ -551,5 +559,5 @@ func (pool *hostConnPool) HandleError(conn *Conn, err error, closed bool) {
 	}
 
 	pool.connPicker.Remove(conn)
-	go pool.fill()
+	go pool.fill_debounce()
 }

debounce/simple_debouncer.go

@@ -0,0 +1,25 @@
+package debounce
+
+// SimpleDebouncer debounce function call with simple logc:
+// 1. If call is currently pending, function call should go through
+// 2. If call is scheduled, but not pending, function call should be voided
+type SimpleDebouncer struct {
+	channel chan struct{}
+}
+
+// NewDebouncer creates a new Debouncer with a buffered channel of size 1
+func NewSimpleDebouncer() *SimpleDebouncer {
+	return &SimpleDebouncer{
+		channel: make(chan struct{}, 1),
+	}
+}
+
+// Debounce attempts to execute the function if the channel allows it
+func (d *SimpleDebouncer) Debounce(fn func()) {
+	select {
+	case d.channel <- struct{}{}:
+		fn()
+		<-d.channel
+	default:
+	}
+}

debounce/simple_debouncer_test.go

@@ -0,0 +1,55 @@
+package debounce
+
+import (
+	"runtime"
+	"sync/atomic"
+	"testing"
+)
+
+// TestDebouncer tests that the debouncer allows only one function to execute at a time
+func TestSimpleDebouncer(t *testing.T) {
+	d := NewSimpleDebouncer()
+	var executions int32
+	startedCh := make(chan struct{}, 1)
+	doneCh := make(chan struct{}, 1)
+	// Function to increment executions
+	fn := func() {
+		<-startedCh // Simulate work
+		atomic.AddInt32(&executions, 1)
+		<-doneCh // Simulate work
+	}
+	t.Run("Case 1", func(t *testing.T) {
+		// Case 1: Normal single execution
+		startedCh <- struct{}{}
+		doneCh <- struct{}{}
+		d.Debounce(fn)
+		// We expect that the function has only executed once due to debouncing
+		if atomic.LoadInt32(&executions) != 1 {
+			t.Errorf("Expected function to be executed only once, but got %d executions", executions)
+		}
+	})
+	atomic.StoreInt32(&executions, 0)
+	t.Run("Case 2", func(t *testing.T) {
+		// Case 2: Debounce the function multiple times at row when body is started
+		go d.Debounce(fn)
+		startedCh <- struct{}{}
+		go d.Debounce(fn)
+		go d.Debounce(fn)
+		doneCh <- struct{}{}
+		startedCh <- struct{}{}
+		doneCh <- struct{}{}
+		waitTillChannelIsEmpty(doneCh)
+		// We expect that the function has only executed once due to debouncing
+		if atomic.LoadInt32(&executions) != 2 {
+			t.Errorf("Expected function to be executed twice, but got %d executions", executions)
+		}
+	})
+}
+func waitTillChannelIsEmpty(ch chan struct{}) {
+	for {
+		if len(ch) == 0 {
+			return
+		}
+		runtime.Gosched()
+	}
+}

scylla_shard_aware_port_common_test.go

@@ -240,7 +240,7 @@ func triggerPoolsRefill(sess *Session) {
 	hosts := sess.ring.allHosts()
 	for _, host := range hosts {
 		hostPool, _ := sess.pool.getPool(host)
-		go hostPool.fill()
+		go hostPool.fill_debounce()
 	}
 }