数组越界BUG

[buguang01]

func (m *Master) AdjustSize(newSize int) {
    if int64(newSize) > m.maxNum {
        newSize = int(m.maxNum)
    }

    m.Lock()
    defer m.Unlock()

    if diff := newSize - int(m.ingNum); diff > 0 {
        for i := 0; i < diff; i++ {
            m.workers[m.ingNum] = newWorker()
            atomic.AddInt64(&m.ingNum, 1)
        }
    } else if diff < 0 {
        atomic.StoreInt64(&m.ingNum, int64(newSize))
        if cursor := atomic.LoadInt64(&m.cursor); cursor > int64(newSize) {
            atomic.StoreInt64(&m.cursor, int64(newSize))
        }
        for _, w := range m.workers[newSize:] {
            if w == nil {
                break
            }
            w.shutdown()
        }
        m.workers = m.workers[0:newSize]
    }
}

func (m *Master) Running() int64 {
    return atomic.LoadInt64(&m.ingNum)
}

func (m *Master) Shutdown() {
    m.AdjustSize(0) // 关闭所有worker
}

func (m *Master) getWorker() *worker {
    atomic.CompareAndSwapInt64(&m.cursor, m.ingNum, 0)
    idx := atomic.AddInt64(&m.cursor, 1)
    w := m.workers[idx-1]
    return w
}

当cursor与原ingNum相等时，你要进行缩容 这个时候，如果是先把ingNum改成了新的值，再 atomic.CompareAndSwapInt64(&m.cursor, m.ingNum, 0) 运行这行时，就不会修改成功，然后你对cursor,进行了累加，导致越界，或有可能拿到一个nil对象。 还有就是你的代码里，对workers初始化的时候，用的是maxNum 但后面修改的时候，又用了ingNum做为他的长度。 我原以为你是要一开始声明一个确定长度的数组。但后面又换了，如果按你后面的逻辑 你一开始应该是 workers: make([]*worker, initNum,maxNum) 才对。

[letsfire]

大佬，帮忙Review

[buguang01]

你写了一个比之前复杂的保证安全的逻辑，但仔细看，还是会有问题

if atomic.CompareAndSwapInt32(&m.tryFlag, 0, 1) {
        atomic.StoreInt64(&m.cursor, 0)
        atomic.StoreInt32(&m.tryFlag, 0)
        m.synCond.Broadcast()
    } else {
        m.synCond.L.Lock()
        m.synCond.Wait()
        m.synCond.L.Unlock()
    }

你不能保证你在m.synCond.Broadcast()时，所有其他协程都在m.synCond.Wait() 虽然你的逻辑也不能保证m.synCond.Broadcast()之前，只有一个协程进入if的then里，但如果只有一个进入了then而有2个进入了else其中一个到了Wait，还有一个在Lock()等待进入中 这个时候你的m.synCond.Broadcast()只能让之前在wait的出来，然后Lock()就进入了wait。 这个时候，如果没有新的协程去m.synCond.Broadcast()，那这个Wait就会一直wiat下去。

虽然Wait方法里面会在添加监听后，释放锁，让别的协程可以进来，但是依然存在着我上面说的那种情况发生的概率；

[buguang01]

你把workers从数组改成了双map进行切换，但你写的并没有达到目标，只是map在获取的时候，可以检查而已，并不需要双map 如果使用双map那每次就应该只更新其中一个map,但那样的话就会导致资源的不释放问题，和你写这个东西的初衷就违背了。

[buguang01]

还有一个问题，那就是你的双map是同一个map，导致一修改数量，就会出现 fatal error: concurrent map read and map write

[buguang01]

func TestMaster(t *testing.T) {
    wg := sync.WaitGroup{}
    wg2 := sync.WaitGroup{}
    w := NewMaster(10, 2)
    ctx, cel := context.WithCancel(context.Background())
    li := w.AddLine("testli", func(e interface{}) {
        defer wg.Done()
    })
    for i := 0; i < 100; i++ {
        go func() {
            wg2.Add(1)
            defer wg2.Done()
            for {
                select {
                case <-ctx.Done():
                    return
                default:
                    wg.Add(1)
                    li.Submit(0)
                }
            }
        }()
    }
    time.Sleep(time.Second * 5)
    fmt.Println("adjustsize")
    w.AdjustSize(6)
    time.Sleep(time.Second * 2)
    w.AdjustSize(1)
    time.Sleep(time.Second * 2)
    cel() //关闭发协程
    w.AdjustSize(2)
    time.Sleep(time.Second * 2)
    wg2.Wait() //确认发协程是否能关闭，多半会卡在这里
    fmt.Println("down")
    w.Shutdown()
    fmt.Println("Wait")
    wg.Wait()
    fmt.Println("End")
}

给你一段测试代码。 改到你能看到输出end就行

[buguang01]

记的多跑几次，几个问题都会暴露出来的。

[letsfire]

感谢，这段工作较忙，问题已经修复，使用了sync.Map，您的测试用例已通过。再次感谢

[buguang01]

你的代码需要优化。用了好多互斥锁。单线程，原子锁。

[letsfire]

还请指点，resetGuard里面是多，但是这段代码执行少，复杂度可以忽略吧

[buguang01]

Master.resetGuard 这个是用来干什么用的？

[letsfire]

确保只有一个协程能重置m.cursor

[buguang01]

确保只有一个协程能重置m.cursor

原因呢？ 你里面本来就是一个原子操作了

[buguang01]

还有就是getWorker为什么是递归？不应该是循环吗？

[letsfire]

m.workers.Load(idx) 同一时间会有多个协程拿不到worker进而去下面重置cursor，这时候下面的guard就是仅让其中一个协程进行重置，其他等待重置完毕继续获取从头获取