mirror
/
ants
mirror of https://github.com/panjf2000/ants.git
1
0
Fork 1
Code Issues Packages Projects Releases Wiki Activity

opt: speed up ReleaseTimeout() for multi-pool (#332)

This commit is contained in:
Andy Pan 2024-06-18 02:42:55 +08:00 committed by GitHub
parent 95dad45c7d
commit da22980e2c
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 62 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

35
multipool.go
View File

@ -28,6 +28,8 @@ import (
"strings"
"sync/atomic"
"time"
"golang.org/x/sync/errgroup"
)
// LoadBalancingStrategy represents the type of load-balancing algorithm.
@ -182,14 +184,35 @@ func (mp *MultiPool) ReleaseTimeout(timeout time.Duration) error {
return ErrPoolClosed
}
var errStr strings.Builder
errCh := make(chan error, len(mp.pools))
var wg errgroup.Group
for i, pool := range mp.pools {
if err := pool.ReleaseTimeout(timeout); err != nil {
errStr.WriteString(fmt.Sprintf("pool %d: %v\n", i, err))
if i < len(mp.pools)-1 {
errStr.WriteString(" | ")
func(p *Pool, idx int) {
wg.Go(func() error {
err := p.ReleaseTimeout(timeout)
if err != nil {
err = fmt.Errorf("pool %d: %v", idx, err)
}
errCh <- err
return err
})
}(pool, i)
}
_ = wg.Wait()
var (
i int
errStr strings.Builder
)
for err := range errCh {
i++
if i == len(mp.pools) {
break
}
if err != nil {
errStr.WriteString(err.Error())
errStr.WriteString(" | ")
}
}
@ -197,7 +220,7 @@ func (mp *MultiPool) ReleaseTimeout(timeout time.Duration) error {
return nil
}
return errors.New(errStr.String())
return errors.New(strings.TrimSuffix(errStr.String(), " | "))
}
// Reboot reboots a released multi-pool.

35
multipool_func.go
View File

@ -28,6 +28,8 @@ import (
"strings"
"sync/atomic"
"time"
"golang.org/x/sync/errgroup"
)
// MultiPoolWithFunc consists of multiple pools, from which you will benefit the
@ -172,14 +174,35 @@ func (mp *MultiPoolWithFunc) ReleaseTimeout(timeout time.Duration) error {
return ErrPoolClosed
}
var errStr strings.Builder
errCh := make(chan error, len(mp.pools))
var wg errgroup.Group
for i, pool := range mp.pools {
if err := pool.ReleaseTimeout(timeout); err != nil {
errStr.WriteString(fmt.Sprintf("pool %d: %v\n", i, err))
if i < len(mp.pools)-1 {
errStr.WriteString(" | ")
func(p *PoolWithFunc, idx int) {
wg.Go(func() error {
err := p.ReleaseTimeout(timeout)
if err != nil {
err = fmt.Errorf("pool %d: %v", idx, err)
}
errCh <- err
return err
})
}(pool, i)
}
_ = wg.Wait()
var (
i int
errStr strings.Builder
)
for err := range errCh {
i++
if i == len(mp.pools) {
break
}
if err != nil {
errStr.WriteString(err.Error())
errStr.WriteString(" | ")
}
}
@ -187,7 +210,7 @@ func (mp *MultiPoolWithFunc) ReleaseTimeout(timeout time.Duration) error {
return nil
}
return errors.New(errStr.String())
return errors.New(strings.TrimSuffix(errStr.String(), " | "))
}
// Reboot reboots a released multi-pool.