lazy: Fix concurrent initialization order

Fixes #5901
This commit is contained in:
Bjørn Erik Pedersen 2019-04-29 19:05:28 +02:00
parent 1cbb501be8
commit 009076e5ee
2 changed files with 131 additions and 56 deletions

View file

@ -77,30 +77,19 @@ func (ini *Init) Do() (interface{}, error) {
} }
ini.init.Do(func() { ini.init.Do(func() {
var (
dependencies []*Init
children []*Init
)
prev := ini.prev prev := ini.prev
for prev != nil { if prev != nil {
// A branch. Initialize the ancestors.
if prev.shouldInitialize() { if prev.shouldInitialize() {
dependencies = append(dependencies, prev) _, err := prev.Do()
} if err != nil {
prev = prev.prev ini.err = err
} return
}
for _, child := range ini.children { } else if prev.inProgress() {
if child.shouldInitialize() { // Concurrent initialization. The following init func
children = append(children, child) // may depend on earlier state, so wait.
} prev.wait()
}
for _, dep := range dependencies {
_, err := dep.Do()
if err != nil {
ini.err = err
return
} }
} }
@ -108,16 +97,25 @@ func (ini *Init) Do() (interface{}, error) {
ini.out, ini.err = ini.f() ini.out, ini.err = ini.f()
} }
for _, dep := range children { for _, child := range ini.children {
_, err := dep.Do() if child.shouldInitialize() {
if err != nil { _, err := child.Do()
ini.err = err if err != nil {
return ini.err = err
return
}
} }
} }
}) })
ini.wait()
return ini.out, ini.err
}
// TODO(bep) investigate if we can use sync.Cond for this.
func (ini *Init) wait() {
var counter time.Duration var counter time.Duration
for !ini.init.Done() { for !ini.init.Done() {
counter += 10 counter += 10
@ -126,8 +124,10 @@ func (ini *Init) Do() (interface{}, error) {
} }
time.Sleep(counter * time.Microsecond) time.Sleep(counter * time.Microsecond)
} }
}
return ini.out, ini.err func (ini *Init) inProgress() bool {
return ini != nil && ini.init.InProgress()
} }
func (ini *Init) shouldInitialize() bool { func (ini *Init) shouldInitialize() bool {
@ -147,20 +147,19 @@ func (ini *Init) add(branch bool, initFn func() (interface{}, error)) *Init {
ini.mu.Lock() ini.mu.Lock()
defer ini.mu.Unlock() defer ini.mu.Unlock()
if !branch { if branch {
ini.checkDone() return &Init{
f: initFn,
prev: ini,
}
} }
init := &Init{ ini.checkDone()
f: initFn, ini.children = append(ini.children, &Init{
prev: ini, f: initFn,
} })
if !branch { return ini
ini.children = append(ini.children, init)
}
return init
} }
func (ini *Init) checkDone() { func (ini *Init) checkDone() {

View file

@ -25,32 +25,41 @@ import (
"github.com/stretchr/testify/require" "github.com/stretchr/testify/require"
) )
var (
rnd = rand.New(rand.NewSource(time.Now().UnixNano()))
bigOrSmall = func() int {
if rnd.Intn(10) < 5 {
return 10000 + rnd.Intn(100000)
}
return 1 + rnd.Intn(50)
}
)
func doWork() {
doWorkOfSize(bigOrSmall())
}
func doWorkOfSize(size int) {
_ = strings.Repeat("Hugo Rocks! ", size)
}
func TestInit(t *testing.T) { func TestInit(t *testing.T) {
assert := require.New(t) assert := require.New(t)
var result string var result string
bigOrSmall := func() int {
if rand.Intn(10) < 3 {
return 10000 + rand.Intn(100000)
}
return 1 + rand.Intn(50)
}
f1 := func(name string) func() (interface{}, error) { f1 := func(name string) func() (interface{}, error) {
return func() (interface{}, error) { return func() (interface{}, error) {
result += name + "|" result += name + "|"
size := bigOrSmall() doWork()
_ = strings.Repeat("Hugo Rocks! ", size)
return name, nil return name, nil
} }
} }
f2 := func() func() (interface{}, error) { f2 := func() func() (interface{}, error) {
return func() (interface{}, error) { return func() (interface{}, error) {
size := bigOrSmall() doWork()
_ = strings.Repeat("Hugo Rocks! ", size) return nil, nil
return size, nil
} }
} }
@ -73,16 +82,15 @@ func TestInit(t *testing.T) {
go func(i int) { go func(i int) {
defer wg.Done() defer wg.Done()
var err error var err error
if rand.Intn(10) < 5 { if rnd.Intn(10) < 5 {
_, err = root.Do() _, err = root.Do()
assert.NoError(err) assert.NoError(err)
} }
// Add a new branch on the fly. // Add a new branch on the fly.
if rand.Intn(10) > 5 { if rnd.Intn(10) > 5 {
branch := branch1_2.Branch(f2()) branch := branch1_2.Branch(f2())
init := branch.Add(f2()) _, err = branch.Do()
_, err = init.Do()
assert.NoError(err) assert.NoError(err)
} else { } else {
_, err = branch1_2_1.Do() _, err = branch1_2_1.Do()
@ -148,3 +156,71 @@ func TestInitAddWithTimeoutError(t *testing.T) {
assert.Error(err) assert.Error(err)
} }
type T struct {
sync.Mutex
V1 string
V2 string
}
func (t *T) Add1(v string) {
t.Lock()
t.V1 += v
t.Unlock()
}
func (t *T) Add2(v string) {
t.Lock()
t.V2 += v
t.Unlock()
}
// https://github.com/gohugoio/hugo/issues/5901
func TestInitBranchOrder(t *testing.T) {
assert := require.New(t)
base := New()
work := func(size int, f func()) func() (interface{}, error) {
return func() (interface{}, error) {
doWorkOfSize(size)
if f != nil {
f()
}
return nil, nil
}
}
state := &T{}
base = base.Add(work(10000, func() {
state.Add1("A")
}))
inits := make([]*Init, 2)
for i := range inits {
inits[i] = base.Branch(work(i+1*100, func() {
// V1 is A
ab := state.V1 + "B"
state.Add2(ab)
}))
}
var wg sync.WaitGroup
for _, v := range inits {
v := v
wg.Add(1)
go func() {
defer wg.Done()
_, err := v.Do()
assert.NoError(err)
}()
}
wg.Wait()
assert.Equal("ABAB", state.V2)
}