存在任务还没有运行就提前结束的异常情况

[yp-app]

按照预期，pool.Wait()应该会等待所有任务运行结束
实际情况在少量任务时会出现任务还没有开始执行，pool.Wait()就结束了

代码部分

以下是从日志上检查到的异常情况的示例
异常情况1，任务已添加，但是任务没有运行，pool.Wait()直接就结束了

异常情况2，任务已添加，pool.Wait()提前结束，任务才开始调度运行

[wyy-class]

感觉是len()方法不是原子操作，在计算期间存在并发安全问题。
解决方案：
把判断条件放到临界区内（加锁区域中）试试。

[daniel-hutao]

@wyy-class 看起来不像是 len 的问题；在 Pool 开始工作后，taskQueue 会逐渐变小，到0时就是没有任务在排队了；这时候 workers 长度和 workerStack 相等，应该就是所有“worker”都已经忙完，任务结束了才对；

@yp-app 能不能提供一段能够复现问题的代码片段，我在本地调试下

[wyy-class]

@daniel-hutao 试了一下，还是有问题。没搞懂为什么i会重复？

package main

import (
	"fmt"

	"github.com/devchat-ai/gopool"
)

func main() {
	pool := gopool.NewGoPool(10, gopool.WithResultCallback(func(i interface{}) {
		fmt.Println(i.(string))
	}), gopool.WithTaskQueueSize(5))

	for i := 0; i < 20; i++ {
		pool.AddTask(func() (interface{}, error) {
			str := fmt.Sprintf("task-%d", i)
			return str, nil
		})
	}
	pool.Wait()
	fmt.Println("all task success!")
}

[wyy-class]

中间加个局部变量就没问题了

for i := 0; i < 20; i++ {
		// 创建一个局部变量以捕获当前值 i
		taskNum := i
		pool.AddTask(func() (interface{}, error) {
			str := fmt.Sprintf("task-%d", taskNum)
			return str, nil
		})
	}

[yp-app]

@daniel-hutao
我大致的代码逻辑是这样的

func job(ctx context.Context) {
	var errorCount atomic.Int64
	// 打开文件
	file, err := os.Open("/Users/yp/a.txt")
	if err != nil {
		log.Println(err)
		return
	}
	defer file.Close()
	// 创建一个Scanner来读取文件
	scanner := bufio.NewScanner(file)
	// 创建任务池
	pool := gopool.NewGoPool(50, gopool.WithErrorCallback(func(err error) {
		log.Println(err, "error")
		errorCount.Add(1)
	}))
	defer pool.Release()

	for scanner.Scan() {
		select {
		case <-ctx.Done():
			log.Println("job cancel")
			return
		default:
			line := scanner.Text()
			log.Println("add task")
			pool.AddTask(func() (interface{}, error) {
				log.Println("task running", line)
				return task(line)
			})
		}
	}
	pool.Wait()
	log.Println("job finished", errorCount.Load())
}

@wyy-class 针对你提的加锁的问题，我引入源码修改测试了一下，还是会存在同样的问题

[yp-app]

通过加的这部分日志，任务确实是添加成功了的，但是pool确实提前退出了（或者说退出的时候认为没有任务）
异常情况1：pool结束，任务没有运行

异常情况2：pool结束后任务才运行

[yp-app]

@daniel-hutao @wyy-class
问题好像是这里，异常情况的时候，任务添加完成之后，任务队列的长度居然是0，导致Wait判断没有任务，就结束了

这是正常情况的时候

[wyy-class]

@yp-app

感觉只能是dispatch有问题了。
在任务出队和获取worker中间，是满足wait跳出的判断条件的。你可以打印试一下。
解决方案：任务不能先出队，需要先获取worker，然后任务出队。

简单修改一下。你可以试试

[yp-app]

@wyy-class
这样改不行，如果先popWorker了，没有任务了，就会阻塞在t:=<-p.taskQueue这，Wait就不会退出了

[yp-app]

@wyy-class @daniel-hutao
目前我是这样修改的，经过测试没有发现异常情况了，但是不确定对性能会产生多少影响

构造一个taskAndWorker结构，添加task的时候，预分配一个worker，taskQueue传递的是task和分配好的worker，这样就能保证Wait不会提前退出，dispatch也只需要做单独的任务分发就可以了，完整代码如下

package gopool

import (
	"context"
	"sort"
	"sync"
	"time"
)

// GoPool represents a pool of workers.
type GoPool interface {
	// AddTask adds a task to the pool.
	AddTask(t task)
	// Wait waits for all tasks to be dispatched and completed.
	Wait()
	// Release releases the pool and all its workers.
	Release()
	// GetRunning returns the number of running workers.
	Running() int
	// GetWorkerCount returns the number of workers.
	GetWorkerCount() int
	// GetTaskQueueSize returns the size of the task queue.
	GetTaskQueueSize() int
}

// task represents a function that will be executed by a worker.
// It returns a result and an error.
type task func() (interface{}, error)

type taskAndWorker struct {
	task
	workerIndex int
}

// goPool represents a pool of workers.
type goPool struct {
	workers     []*worker
	workerStack []int
	maxWorkers  int
	// Set by WithMinWorkers(), used to adjust the number of workers. Default equals to maxWorkers.
	minWorkers int
	// tasks are added to this channel first, then dispatched to workers. Default buffer size is 1 million.
	taskQueue chan taskAndWorker
	// Set by WithTaskQueueSize(), used to set the size of the task queue. Default is 1e6.
	taskQueueSize int
	// Set by WithRetryCount(), used to retry a task when it fails. Default is 0.
	retryCount int
	lock       sync.Locker
	cond       *sync.Cond
	// Set by WithTimeout(), used to set a timeout for a task. Default is 0, which means no timeout.
	timeout time.Duration
	// Set by WithResultCallback(), used to handle the result of a task. Default is nil.
	resultCallback func(interface{})
	// Set by WithErrorCallback(), used to handle the error of a task. Default is nil.
	errorCallback func(error)
	// adjustInterval is the interval to adjust the number of workers. Default is 1 second.
	adjustInterval time.Duration
	ctx            context.Context
	// cancel is used to cancel the context. It is called when Release() is called.
	cancel context.CancelFunc
}

// NewGoPool creates a new pool of workers.
func NewGoPool(maxWorkers int, opts ...Option) GoPool {
	ctx, cancel := context.WithCancel(context.Background())
	pool := &goPool{
		maxWorkers: maxWorkers,
		// Set minWorkers to maxWorkers by default
		minWorkers: maxWorkers,
		// workers and workerStack should be initialized after WithMinWorkers() is called
		workers:        nil,
		workerStack:    nil,
		taskQueue:      nil,
		taskQueueSize:  1e6,
		retryCount:     0,
		lock:           new(sync.Mutex),
		timeout:        0,
		adjustInterval: 1 * time.Second,
		ctx:            ctx,
		cancel:         cancel,
	}
	// Apply options
	for _, opt := range opts {
		opt(pool)
	}

	pool.taskQueue = make(chan taskAndWorker, pool.taskQueueSize)
	pool.workers = make([]*worker, pool.minWorkers)
	pool.workerStack = make([]int, pool.minWorkers)

	if pool.cond == nil {
		pool.cond = sync.NewCond(pool.lock)
	}
	// Create workers with the minimum number. Don't use pushWorker() here.
	for i := 0; i < pool.minWorkers; i++ {
		worker := newWorker()
		pool.workers[i] = worker
		pool.workerStack[i] = i
		worker.start(pool, i)
	}
	go pool.adjustWorkers()
	go pool.dispatch()
	return pool
}

// AddTask adds a task to the pool.
func (p *goPool) AddTask(t task) {
	index := p.popWorker()
	p.taskQueue <- taskAndWorker{
		task:        t,
		workerIndex: index,
	}
}

// Wait waits for all tasks to be dispatched and completed.
func (p *goPool) Wait() {
	for {
		p.lock.Lock()
		workerStackLen := len(p.workerStack)
		p.lock.Unlock()

		if len(p.taskQueue) == 0 && workerStackLen == len(p.workers) {
			break
		}

		time.Sleep(50 * time.Millisecond)
	}
}

// Release stops all workers and releases resources.
func (p *goPool) Release() {
	close(p.taskQueue)
	p.cancel()
	p.cond.L.Lock()
	for len(p.workerStack) != p.minWorkers {
		p.cond.Wait()
	}
	p.cond.L.Unlock()
	for _, worker := range p.workers {
		close(worker.taskQueue)
	}
	p.workers = nil
	p.workerStack = nil
}

func (p *goPool) popWorker() int {
	p.cond.L.Lock()
	for len(p.workerStack) == 0 {
		p.cond.Wait()
	}
	p.cond.L.Unlock()
	p.lock.Lock()
	workerIndex := p.workerStack[len(p.workerStack)-1]
	p.workerStack = p.workerStack[:len(p.workerStack)-1]
	p.lock.Unlock()
	return workerIndex
}

func (p *goPool) pushWorker(workerIndex int) {
	p.lock.Lock()
	p.workerStack = append(p.workerStack, workerIndex)
	p.lock.Unlock()
	p.cond.Signal()
}

// adjustWorkers adjusts the number of workers according to the number of tasks in the queue.
func (p *goPool) adjustWorkers() {
	ticker := time.NewTicker(p.adjustInterval)
	defer ticker.Stop()

	var adjustFlag bool

	for {
		adjustFlag = false
		select {
		case <-ticker.C:
			p.cond.L.Lock()
			if len(p.taskQueue) > len(p.workers)*3/4 && len(p.workers) < p.maxWorkers {
				adjustFlag = true
				// Double the number of workers until it reaches the maximum
				newWorkers := min(len(p.workers)*2, p.maxWorkers) - len(p.workers)
				for i := 0; i < newWorkers; i++ {
					worker := newWorker()
					p.workers = append(p.workers, worker)
					// Don't use len(p.workerStack)-1 here, because it will be less than len(p.workers)-1 when the pool is busy
					p.workerStack = append(p.workerStack, len(p.workers)-1)
					worker.start(p, len(p.workers)-1)
				}
			} else if len(p.taskQueue) == 0 && len(p.workerStack) == len(p.workers) && len(p.workers) > p.minWorkers {
				adjustFlag = true
				// Halve the number of workers until it reaches the minimum
				removeWorkers := (len(p.workers) - p.minWorkers + 1) / 2
				// Sort the workerStack before removing workers.
				// [1,2,3,4,5] -working-> [1,2,3] -expansive-> [1,2,3,6,7] -idle-> [1,2,3,6,7,4,5]
				sort.Ints(p.workerStack)
				p.workers = p.workers[:len(p.workers)-removeWorkers]
				p.workerStack = p.workerStack[:len(p.workerStack)-removeWorkers]
			}
			p.cond.L.Unlock()
			if adjustFlag {
				p.cond.Broadcast()
			}
		case <-p.ctx.Done():
			return
		}
	}
}

// dispatch dispatches tasks to workers.
func (p *goPool) dispatch() {
	for t := range p.taskQueue {
		p.workers[t.workerIndex].taskQueue <- t.task
	}
}

func min(a, b int) int {
	if a < b {
		return a
	}
	return b
}

// Running returns the number of workers that are currently working.
func (p *goPool) Running() int {
	p.lock.Lock()
	defer p.lock.Unlock()
	return len(p.workers) - len(p.workerStack)
}

// GetWorkerCount returns the number of workers in the pool.
func (p *goPool) GetWorkerCount() int {
	p.lock.Lock()
	defer p.lock.Unlock()
	return len(p.workers)
}

// GetTaskQueueSize returns the size of the task queue.
func (p *goPool) GetTaskQueueSize() int {
	p.lock.Lock()
	defer p.lock.Unlock()
	return p.taskQueueSize
}

[wyy-class]

测试代码

func Test(t *testing.T) {
	pool := gopool.NewGoPool(10, gopool.WithResultCallback(func(i interface{}) {
		fmt.Println(i.(string))
	}))
	go func() {
		for i := 0; i < 20; i++ {
			j := i
			pool.AddTask(func() (interface{}, error) {
				s := fmt.Sprintf("task-%d", j)
				time.Sleep(2 * time.Second)
				return s, nil
			})
		}
	}()
	pool.Wait()
	fmt.Println("all task success!")
	time.Sleep(4 * time.Second)
}

输出

[wyy-class]

@yp-app 测试了一下，还是存在这样的问题。

[yp-app]

@wyy-class
你这样写是有问题的，goroutine还没开始运行呢，主线程Wait判断没有任务是正常的，后面goroutine才把任务加进去的

[wyy-class]

@yp-app 学习到了，输出解决了