fix the bug of thread pool hanging

This commit is contained in:
M09Ic 2024-08-12 15:12:43 +08:00
parent 28aacea18c
commit 411f24d94d
5 changed files with 28 additions and 14 deletions

View File

@ -349,7 +349,7 @@ func (opt *Option) NewRunner() (*Runner, error) {
} }
opt.PrintPlugin() opt.PrintPlugin()
if r.IsCheck == false { if r.IsCheck == true {
logs.Log.Important("enabling brute mod, because of enabled brute plugin") logs.Log.Important("enabling brute mod, because of enabled brute plugin")
} }

View File

@ -51,7 +51,7 @@ func NewBrutePool(ctx context.Context, config *Config) (*BrutePool, error) {
additionCh: make(chan *Unit, config.Thread), additionCh: make(chan *Unit, config.Thread),
closeCh: make(chan struct{}), closeCh: make(chan struct{}),
processCh: make(chan *pkg.Baseline, config.Thread), processCh: make(chan *pkg.Baseline, config.Thread),
wg: sync.WaitGroup{}, wg: &sync.WaitGroup{},
}, },
base: u.Scheme + "://" + u.Host, base: u.Scheme + "://" + u.Host,
isDir: strings.HasSuffix(u.Path, "/"), isDir: strings.HasSuffix(u.Path, "/"),
@ -196,7 +196,7 @@ func (pool *BrutePool) Upgrade(bl *pkg.Baseline) error {
return nil return nil
} }
func (pool *BrutePool) Run(offset, limit int) { func (pool *BrutePool) Run(ctx context.Context, offset, limit int) {
pool.Worder.Run() pool.Worder.Run()
if pool.Active { if pool.Active {
pool.wg.Add(1) pool.wg.Add(1)
@ -279,7 +279,7 @@ Loop:
} }
case <-pool.closeCh: case <-pool.closeCh:
break Loop break Loop
case <-pool.ctx.Done(): case <-ctx.Done():
break Loop break Loop
case <-pool.ctx.Done(): case <-pool.ctx.Done():
break Loop break Loop

View File

@ -29,7 +29,7 @@ func NewCheckPool(ctx context.Context, config *Config) (*CheckPool, error) {
Timeout: time.Duration(config.Timeout) * time.Second, Timeout: time.Duration(config.Timeout) * time.Second,
ProxyAddr: config.ProxyAddr, ProxyAddr: config.ProxyAddr,
}), }),
wg: sync.WaitGroup{}, wg: &sync.WaitGroup{},
additionCh: make(chan *Unit, 1024), additionCh: make(chan *Unit, 1024),
closeCh: make(chan struct{}), closeCh: make(chan struct{}),
processCh: make(chan *pkg.Baseline, config.Thread), processCh: make(chan *pkg.Baseline, config.Thread),
@ -50,21 +50,35 @@ type CheckPool struct {
func (pool *CheckPool) Run(ctx context.Context, offset, limit int) { func (pool *CheckPool) Run(ctx context.Context, offset, limit int) {
pool.Worder.Run() pool.Worder.Run()
var done bool
// 挂起一个监控goroutine, 每100ms判断一次done, 如果已经done, 则关闭closeCh, 然后通过Loop中的select case closeCh去break, 实现退出
go func() {
for {
if done {
pool.wg.Wait()
close(pool.closeCh)
return
}
time.Sleep(100 * time.Millisecond)
}
}()
Loop: Loop:
for { for {
select { select {
case u, ok := <-pool.Worder.C: case u, ok := <-pool.Worder.C:
if !ok { if !ok {
break Loop done = true
continue
} }
if pool.reqCount < offset { if pool.reqCount < offset {
pool.reqCount++ pool.reqCount++
break Loop continue
} }
if pool.reqCount > limit { if pool.reqCount > limit {
break Loop continue
} }
pool.wg.Add(1) pool.wg.Add(1)
@ -82,7 +96,7 @@ Loop:
break Loop break Loop
} }
} }
pool.wg.Wait()
pool.Close() pool.Close()
} }
@ -128,6 +142,9 @@ func (pool *CheckPool) Invoke(v interface{}) {
} else { } else {
bl = pkg.NewBaseline(req.URI(), req.Host(), resp) bl = pkg.NewBaseline(req.URI(), req.Host(), resp)
bl.Collect() bl.Collect()
if bl.Status == 400 {
pool.doUpgrade(bl)
}
} }
bl.ReqDepth = unit.depth bl.ReqDepth = unit.depth
bl.Source = unit.source bl.Source = unit.source
@ -141,9 +158,6 @@ func (pool *CheckPool) Handler() {
if bl.RedirectURL != "" { if bl.RedirectURL != "" {
pool.doRedirect(bl, bl.ReqDepth) pool.doRedirect(bl, bl.ReqDepth)
pool.putToOutput(bl) pool.putToOutput(bl)
} else if bl.Status == 400 {
pool.doUpgrade(bl)
pool.putToOutput(bl)
} else { } else {
params := map[string]interface{}{ params := map[string]interface{}{
"current": bl, "current": bl,

View File

@ -29,7 +29,7 @@ type BasePool struct {
failedCount int failedCount int
additionCh chan *Unit additionCh chan *Unit
closeCh chan struct{} closeCh chan struct{}
wg sync.WaitGroup wg *sync.WaitGroup
} }
func (pool *BasePool) doRedirect(bl *pkg.Baseline, depth int) { func (pool *BasePool) doRedirect(bl *pkg.Baseline, depth int) {

View File

@ -207,7 +207,7 @@ func (r *Runner) Prepare(ctx context.Context) error {
} }
} }
brutePool.Run(brutePool.Statistor.Offset, limit) brutePool.Run(ctx, brutePool.Statistor.Offset, limit)
if brutePool.IsFailed && len(brutePool.FailedBaselines) > 0 { if brutePool.IsFailed && len(brutePool.FailedBaselines) > 0 {
// 如果因为错误积累退出, end将指向第一个错误发生时, 防止resume时跳过大量目标 // 如果因为错误积累退出, end将指向第一个错误发生时, 防止resume时跳过大量目标