[#300] pool: Move 'healthCheck' to separate file

Signed-off-by: Alexander Chuprov <a.chuprov@yadro.com>
This commit is contained in:
Alexander Chuprov 2025-02-12 18:07:23 +03:00
parent 08cd49b070
commit 7e92df62a7
Signed by: achuprov
GPG key ID: 2D916FFD803B0EDD
2 changed files with 47 additions and 41 deletions

47
pool/healthcheck.go Normal file
View file

@ -0,0 +1,47 @@
package pool
import (
"context"
"time"
)
type healthCheck struct {
cancel context.CancelFunc
closedCh chan struct{}
clientRebalanceInterval time.Duration
}
func newHealthCheck(clientRebalanceInterval time.Duration) *healthCheck {
var h healthCheck
h.clientRebalanceInterval = clientRebalanceInterval
h.closedCh = make(chan struct{})
return &h
}
// startRebalance runs loop to monitor connection healthy status.
func (h *healthCheck) startRebalance(ctx context.Context, callback func(ctx context.Context)) {
ctx, cancel := context.WithCancel(ctx)
h.cancel = cancel
go func() {
ticker := time.NewTicker(h.clientRebalanceInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
close(h.closedCh)
return
case <-ticker.C:
callback(ctx)
ticker.Reset(h.clientRebalanceInterval)
}
}
}()
}
func (h *healthCheck) stopRebalance() {
h.cancel()
<-h.closedCh
}

View file

@ -1951,20 +1951,6 @@ type connectionManager struct {
healthChecker *healthCheck
}
type healthCheck struct {
cancel context.CancelFunc
closedCh chan struct{}
clientRebalanceInterval time.Duration
}
func newHealthCheck(clientRebalanceInterval time.Duration) *healthCheck {
var h healthCheck
h.clientRebalanceInterval = clientRebalanceInterval
h.closedCh = make(chan struct{})
return &h
}
type innerPool struct {
lock sync.RWMutex
sampler *sampler
@ -2200,33 +2186,6 @@ func adjustNodeParams(nodeParams []NodeParam) ([]*nodesParam, error) {
return nodesParams, nil
}
// startRebalance runs loop to monitor connection healthy status.
func (h *healthCheck) startRebalance(ctx context.Context, callback func(ctx context.Context)) {
ctx, cancel := context.WithCancel(ctx)
h.cancel = cancel
go func() {
ticker := time.NewTicker(h.clientRebalanceInterval)
defer ticker.Stop()
for {
select {
case <-ctx.Done():
close(h.closedCh)
return
case <-ticker.C:
callback(ctx)
ticker.Reset(h.clientRebalanceInterval)
}
}
}()
}
func (h *healthCheck) stopRebalance() {
h.cancel()
<-h.closedCh
}
func (cm *connectionManager) updateNodesHealth(ctx context.Context, buffers [][]float64) {
wg := sync.WaitGroup{}
for i, inner := range cm.innerPools {