// A Locker represents an object that can be locked and unlocked.
type Locker interface {
     Lock()
     Unlock()
}

//實(shí)現(xiàn)Locker接口
type spinLock uint32
//最大回退次數(shù)
const maxBackoff = 16
// 加鎖
func (sl *spinLock) Lock() {
    backoff := 1
    //基于CAS機(jī)制，嘗試獲取鎖
    for !atomic.CompareAndSwapUint32((*uint32)(sl), 0, 1) {
        //執(zhí)行backoff次 cpu讓出時(shí)間片次數(shù)
        for i := 0; i < backoff; i++ {
            //使當(dāng)前goroutine讓出CPU時(shí)間片
            runtime.Gosched()
        }
        if backoff < maxBackoff {
            //左移后賦值 等于 backoff = backoff << 1
            //左移一位就是乘以 2的1次方
            backoff <<= 1
        }
    }
}

//釋放鎖
func (sl *spinLock) Unlock() {
    atomic.StoreUint32((*uint32)(sl), 0)
}

type Pool struct {
    // 協(xié)程池容量 
    capacity int32
    // 當(dāng)前協(xié)程池中正在運(yùn)行的協(xié)程數(shù)
    running int32
    // ants 實(shí)現(xiàn)的自旋鎖，用于同步并發(fā)操作
    lock sync.Locker
    // 存放一組Worker
    workers workerArray
    // 協(xié)程池狀態(tài) （1-關(guān)閉、0-開(kāi)啟）
    state int32
    // 并發(fā)協(xié)調(diào)器，用于阻塞模式下，掛起和喚醒等待資源的協(xié)程
    cond *sync.Cond
    // worker 對(duì)象池
    workerCache sync.Pool
    // 等待的協(xié)程數(shù)量
    waiting int32
    // 回收協(xié)程是否關(guān)閉
    heartbeatDone int32
    // 閉回收協(xié)程的控制器函數(shù)
    stopHeartbeat context.CancelFunc
    // 協(xié)程池的配置
    options *Options
}

type goWorker struct {
    //goWorker 所屬的協(xié)程池
    pool *Pool
    //接收實(shí)際執(zhí)行任務(wù)的管道
    task chan func()
    //goWorker 回收到協(xié)程池的時(shí)間
    recycleTime time.Time
}

type workerArray interface {
 // worker 列表長(zhǎng)度
 len() int 
 // 是否為空
 isEmpty() bool
 // 插入一個(gè)goworker
 insert(worker *goWorker) error 
 // 從WorkerArray獲取可用的goworker
 detach() *goWorker 
 // 清理pool.workers中的過(guò)期goworker
 retrieveExpiry(duration time.Duration) []*goWorker  
 // 重置，清空WorkerArray中所有的goWorker
 reset() 
}

func NewPool(size int, options ...Option) (*Pool, error) {
    //讀取一些自定義的配置
    opts := loadOptions(options...)

    ...
    // 創(chuàng)建 Pool 對(duì)象
    p := &Pool{
        capacity: int32(size),
        lock:     internal.NewSpinLock(),
        options:  opts,
    }
     // 指定 sync.Pool 創(chuàng)建 worker 的方法
    p.workerCache.New = func() interface{} {
        return &goWorker{
            pool: p,
            task: make(chan func(), workerChanCap),
        }
    }
    // 初始化Pool時(shí)是否進(jìn)行內(nèi)存預(yù)分配
    // 區(qū)分workerArray 的實(shí)現(xiàn)方式
    if p.options.PreAlloc {
        if size == -1 {
            return nil, ErrInvalidPreAllocSize
        }
        // 預(yù)先分配固定 Size 的池子
        p.workers = newWorkerArray(loopQueueType, size)
    } else {
        // 初始化不創(chuàng)建，運(yùn)行時(shí)再創(chuàng)建
        p.workers = newWorkerArray(stackType, 0)
    }

    p.cond = sync.NewCond(p.lock)

    // 開(kāi)啟一個(gè)goroutine清理過(guò)期的 worker
    go p.purgePeriodically()

    return p, nil
}

func (p *Pool) purgePeriodically(ctx context.Context) {
    // ExpiryDuration 默認(rèn)是1s
    heartbeat := time.NewTicker(p.options.ExpiryDuration)
    ...
    for {
        select {
        case <-heartbeat.C:
        case <-ctx.Done():
            return
        }
        // pool關(guān)閉
        if p.IsClosed() {
            break
        }
        // 從 workers 中獲取過(guò)期的 worker
        p.lock.Lock()
        expiredWorkers := p.workers.retrieveExpiry(p.options.ExpiryDuration)
        p.lock.Unlock()
        // 清理過(guò)期的worker
        for i := range expiredWorkers {
            expiredWorkers[i].task <- nil
            expiredWorkers[i] = nil
        }
        // 喚醒所有等待的線程
        if p.Running() == 0 || (p.Waiting() > 0 && p.Free() > 0) {
            p.cond.Broadcast()
        }
    }
}

func (p *Pool) Submit(task func()) error {
    // pool是否關(guān)閉
    if p.IsClosed() {
        return ErrPoolClosed
    }
    var w *goWorker
    // 嘗試獲取空閑的goWorker
    if w = p.retrieveWorker(); w == nil {
        return ErrPoolOverload
    }
    // 發(fā)送到 goWorker的channel中
    w.task <- task
    return nil
}

func (p *Pool) retrieveWorker() (w *goWorker) {
    //創(chuàng)建一個(gè)新的goWorker，并執(zhí)行
    spawnWorker := func() {
        //實(shí)例化 worker
        w = p.workerCache.Get().(*goWorker)
        // 運(yùn)行
        w.run()
    }

    // 加鎖
    p.lock.Lock()
    // 從workers 中取出一個(gè) goWorker
    // workerStack 實(shí)現(xiàn)了p.workers的方法
    w = p.workers.detach()
    if w != nil { 
        p.lock.Unlock()
    // Pool容量大于正在工作的 goWorker 數(shù)量）
    //則調(diào)用 spawnWorker() 新建一個(gè) goWorker
    } else if capacity := p.Cap(); capacity == -1 || capacity > p.Running() {
        
        p.lock.Unlock()
        spawnWorker()
    } else { 
          // options設(shè)置了非阻塞選項(xiàng)，直接返回 nil
          if p.options.Nonblocking {
            p.lock.Unlock()
            return
        }
    retry:
        //option設(shè)置了最大阻塞隊(duì)列，當(dāng)前阻塞等待的任務(wù)數(shù)量已經(jīng)達(dá)設(shè)置上限，直接返回 nil
        if p.options.MaxBlockingTasks != 0 && p.Waiting() >= p.options.MaxBlockingTasks {
            p.lock.Unlock()
            return
        }
        ...
        var nw int
        //如果正在執(zhí)行的worker數(shù)量為0時(shí)，則重新創(chuàng)建woker
        if nw = p.Running(); nw == 0 { 
            p.lock.Unlock()
            spawnWorker()
            return
        }
        //p.workers中獲取可用的worker
        //執(zhí)行開(kāi)頭創(chuàng)建的spawnWorker
        if w = p.workers.detach(); w == nil {
            if nw < p.Cap() {
                p.lock.Unlock()
                spawnWorker()
                return
            }
            goto retry
        }
        p.lock.Unlock()
    }
    return
}

func (w *goWorker) run() {
    // pool的running 加 一
    w.pool.addRunning(1)
    go func() {
        defer func() {
            ...
            if p := recover(); p != nil {
                //處理捕獲的panic
            }
            w.pool.cond.Signal()
        }()
        //任務(wù)執(zhí)行
        for f := range w.task {
            if f == nil {
                return
            }
            f()
            //執(zhí)行完后回收worker
            if ok := w.pool.revertWorker(w); !ok {
                return
            }
        }
    }()
}

func (p *Pool) revertWorker(worker *goWorker) bool {
    if capacity := p.Cap(); (capacity > 0 && p.Running() > capacity) || p.IsClosed() {
        p.cond.Broadcast()
        return false
    }
    // 重置空閑計(jì)時(shí)器，用于判定過(guò)期
    worker.recycleTime = p.nowTime()
    p.lock.Lock()
    ...
    // 調(diào)用works的insert方法放回Pool
    err := p.workers.insert(worker)
    if err != nil {
        p.lock.Unlock()
        return false
    }
    // p.cond.Signal() 喚醒一個(gè)可能等待的線程
    p.cond.Signal()
    p.lock.Unlock()
    return true
}