When multiple goroutine calls consume, mb.newShard(md) will be called multiple times.

[luyuanx2]

Describe the bug

The problematic code is as shown below:

Steps to reproduce

Here is a simple code showing this bug:

package main

import (
	"fmt"
	"sync"
	"time"
)

type multiShardBatcher struct {
	batchers sync.Map

	lock sync.Mutex
	size int
}
type shard struct {
}

func (mb *multiShardBatcher) newShard() *shard {
	b := &shard{}
	fmt.Println("new shard")
	time.Sleep(1 * time.Second)
	return b
}

func (mb *multiShardBatcher) consume(key string) *shard {

	b, ok := mb.batchers.Load(key)
	if !ok {
		mb.lock.Lock()
		b, _ = mb.batchers.LoadOrStore(key, mb.newShard())
		mb.lock.Unlock()
	}
	return b.(*shard)
}

func main() {

	mb := &multiShardBatcher{}

	group := sync.WaitGroup{}
	for i := 0; i < 10; i++ {
		group.Add(1)
		go func() {
			mb.consume("key")
			group.Done()
		}()
	}
	group.Wait()

}

What did you expect to see?

What did you see instead?

What version did you use?

What config did you use?

Environment

Additional context

[mx-psi]

cc @dmitryax

[andrzej-stencel]

This report looks valid to me. The mb.newShard(md) method not only creates a new shard, but also starts a new goroutine on it, so we should not be doing that if we might not keep the shard around (which is what happens when the shard had already been created in another goroutine). Otherwise we end up with a goroutine leak.

We should move mb.lock.Lock() to before the initial mb.batchers.Load(aset).

This code was introduced in #7714. @bogdandrutu can you take a look?

cc @jmacd as the original contributor of the feature (#7578).

[andrzej-stencel]

Proposed a fix for this in #9814.