Skip to content

Commit 4c78f32

Browse files
Xinyue-WangXinyue-Wang
authored
Merge pull request #32 from Xinyue-Wang/xinywa/succeed_fast
Add WaitAny
2 parents d9e07a1 + 8d6c81f commit 4c78f32

File tree

2 files changed

+257
-0
lines changed

2 files changed

+257
-0
lines changed

‎wait_any.go

Lines changed: 67 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,67 @@
1+
package asynctask
2+
3+
import (
4+
"context"
5+
"fmt"
6+
)
7+
8+
// WaitAnyOptions defines options for WaitAny function
9+
type WaitAnyOptions struct {
10+
// FailOnAnyError set to true will indicate WaitAny to return on first error it sees.
11+
FailOnAnyError bool
12+
}
13+
14+
// WaitAny block current thread til any of task finished.
15+
// first error from any tasks passed in will be returned if FailOnAnyError is set.
16+
// first task end without error will end wait and return nil
17+
func WaitAny(ctx context.Context, options *WaitAnyOptions, tasks ...Waitable) error {
18+
tasksCount := len(tasks)
19+
if tasksCount == 0 {
20+
return nil
21+
}
22+
23+
if options == nil {
24+
options = &WaitAnyOptions{}
25+
}
26+
27+
// tried to close channel before exit this func,
28+
// but it's complicated with routines, and we don't want to delay the return.
29+
// per https://stackoverflow.com/questions/8593645/is-it-ok-to-leave-a-channel-open, its ok to leave channel open, eventually it will be garbage collected.
30+
// this assumes the tasks eventually finish, otherwise we will have a routine leak.
31+
errorCh := make(chan error, tasksCount)
32+
33+
for _, tsk := range tasks {
34+
go waitOne(ctx, tsk, errorCh)
35+
}
36+
37+
runningTasks := tasksCount
38+
var errList []error
39+
for {
40+
select {
41+
case err := <-errorCh:
42+
runningTasks--
43+
if err != nil {
44+
// return immediately after receive first error if FailOnAnyError is set.
45+
if options.FailOnAnyError {
46+
return err
47+
}
48+
errList = append(errList, err)
49+
} else {
50+
// return immediately after first task completed.
51+
return nil
52+
}
53+
case <-ctx.Done():
54+
return fmt.Errorf("WaitAny %w", ctx.Err())
55+
}
56+
57+
// are we finished yet?
58+
if runningTasks == 0 {
59+
break
60+
}
61+
}
62+
63+
// when all tasks failed and FailOnAnyError is not set, return first one.
64+
// caller can get error for individual task by using Wait(),
65+
// it would return immediately after this WaitAny()
66+
return errList[0]
67+
}

‎wait_any_test.go

Lines changed: 190 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,190 @@
1+
package asynctask_test
2+
3+
import (
4+
"fmt"
5+
"testing"
6+
"time"
7+
8+
"github.com/Azure/go-asynctask"
9+
"github.com/stretchr/testify/assert"
10+
)
11+
12+
func TestWaitAnyNoTask(t *testing.T) {
13+
t.Parallel()
14+
ctx, _ := newTestContextWithTimeout(t, 2*time.Second)
15+
16+
err := asynctask.WaitAny(ctx, nil)
17+
assert.NoError(t, err)
18+
}
19+
20+
func TestWaitAny(t *testing.T) {
21+
t.Parallel()
22+
ctx, cancelTaskExecution := newTestContextWithTimeout(t, 2*time.Second)
23+
24+
start := time.Now()
25+
countingTsk3 := asynctask.Start(ctx, getCountingTask(10, "countingPer2ms", 2*time.Millisecond))
26+
result := "something"
27+
completedTsk := asynctask.NewCompletedTask(&result)
28+
29+
err := asynctask.WaitAny(ctx, nil, countingTsk3, completedTsk)
30+
elapsed := time.Since(start)
31+
assert.NoError(t, err)
32+
// should finish after right away
33+
assert.True(t, elapsed < 2*time.Millisecond, fmt.Sprintf("actually elapsed: %v", elapsed))
34+
35+
start = time.Now()
36+
countingTsk1 := asynctask.Start(ctx, getCountingTask(10, "countingPer40ms", 40*time.Millisecond))
37+
countingTsk2 := asynctask.Start(ctx, getCountingTask(10, "countingPer20ms", 20*time.Millisecond))
38+
countingTsk3 = asynctask.Start(ctx, getCountingTask(10, "countingPer2ms", 2*time.Millisecond))
39+
err = asynctask.WaitAny(ctx, &asynctask.WaitAnyOptions{FailOnAnyError: true}, countingTsk1, countingTsk2, countingTsk3)
40+
elapsed = time.Since(start)
41+
assert.NoError(t, err)
42+
cancelTaskExecution()
43+
44+
// should finish right after countingTsk3
45+
assert.True(t, elapsed >= 20*time.Millisecond && elapsed < 200*time.Millisecond, fmt.Sprintf("actually elapsed: %v", elapsed))
46+
47+
// counting task do testing.Logf in another go routine
48+
// while testing.Logf would cause DataRace error when test is already finished: https://github.com/golang/go/issues/40343
49+
// wait minor time for the go routine to finish.
50+
time.Sleep(1 * time.Millisecond)
51+
}
52+
53+
func TestWaitAnyContextCancel(t *testing.T) {
54+
t.Parallel()
55+
ctx, cancelTaskExecution := newTestContextWithTimeout(t, 2*time.Second)
56+
57+
start := time.Now()
58+
59+
countingTsk1 := asynctask.Start(ctx, getCountingTask(10, "countingPer40ms", 40*time.Millisecond))
60+
countingTsk2 := asynctask.Start(ctx, getCountingTask(10, "countingPer20ms", 20*time.Millisecond))
61+
go func() {
62+
time.Sleep(5 * time.Millisecond)
63+
cancelTaskExecution()
64+
}()
65+
err := asynctask.WaitAny(ctx, nil, countingTsk1, countingTsk2)
66+
elapsed := time.Since(start)
67+
assert.Error(t, err)
68+
assert.Equal(t, "WaitAny context canceled", err.Error(), "expecting context canceled error")
69+
// should finish right after countingTsk3
70+
assert.True(t, elapsed >= 5*time.Millisecond && elapsed < 200*time.Millisecond, fmt.Sprintf("actually elapsed: %v", elapsed))
71+
72+
// counting task do testing.Logf in another go routine
73+
// while testing.Logf would cause DataRace error when test is already finished: https://github.com/golang/go/issues/40343
74+
// wait minor time for the go routine to finish.
75+
time.Sleep(1 * time.Millisecond)
76+
}
77+
78+
func TestWaitAnyErrorCase(t *testing.T) {
79+
t.Parallel()
80+
ctx, cancelTaskExecution := newTestContextWithTimeout(t, 3*time.Second)
81+
82+
start := time.Now()
83+
errorTsk := asynctask.Start(ctx, getErrorTask("expected error", 10*time.Millisecond))
84+
result := "something"
85+
completedTsk := asynctask.NewCompletedTask(&result)
86+
err := asynctask.WaitAny(ctx, nil, errorTsk, completedTsk)
87+
assert.NoError(t, err)
88+
elapsed := time.Since(start)
89+
// should finish after right away
90+
assert.True(t, elapsed < 20*time.Millisecond, fmt.Sprintf("actually elapsed: %v", elapsed))
91+
completedTskState := completedTsk.State()
92+
assert.Equal(t, asynctask.StateCompleted, completedTskState, "completed task should finished")
93+
94+
start = time.Now()
95+
countingTsk := asynctask.Start(ctx, getCountingTask(10, "countingPer40ms", 40*time.Millisecond))
96+
errorTsk = asynctask.Start(ctx, getErrorTask("expected error", 10*time.Millisecond))
97+
panicTsk := asynctask.Start(ctx, getPanicTask(20*time.Millisecond))
98+
err = asynctask.WaitAny(ctx, nil, countingTsk, errorTsk, panicTsk)
99+
// there is a succeed task
100+
assert.NoError(t, err)
101+
elapsed = time.Since(start)
102+
103+
countingTskState := countingTsk.State()
104+
panicTskState := panicTsk.State()
105+
errTskState := errorTsk.State()
106+
cancelTaskExecution() // all assertion variable captured, cancel counting task
107+
108+
// should only finish after longest task.
109+
assert.True(t, elapsed >= 40*10*time.Millisecond, fmt.Sprintf("actually elapsed: %v", elapsed))
110+
111+
assert.Equal(t, asynctask.StateCompleted, countingTskState, "countingTask should NOT finished")
112+
assert.Equal(t, asynctask.StateFailed, errTskState, "error task should failed")
113+
assert.Equal(t, asynctask.StateFailed, panicTskState, "panic task should Not failed")
114+
115+
// counting task do testing.Logf in another go routine
116+
// while testing.Logf would cause DataRace error when test is already finished: https://github.com/golang/go/issues/40343
117+
// wait minor time for the go routine to finish.
118+
time.Sleep(1 * time.Millisecond)
119+
}
120+
121+
func TestWaitAnyAllFailCase(t *testing.T) {
122+
t.Parallel()
123+
ctx, cancelTaskExecution := newTestContextWithTimeout(t, 3*time.Second)
124+
125+
start := time.Now()
126+
errorTsk := asynctask.Start(ctx, getErrorTask("expected error", 10*time.Millisecond))
127+
panicTsk := asynctask.Start(ctx, getPanicTask(20*time.Millisecond))
128+
err := asynctask.WaitAny(ctx, nil, errorTsk, panicTsk)
129+
assert.Error(t, err)
130+
131+
panicTskState := panicTsk.State()
132+
errTskState := errorTsk.State()
133+
elapsed := time.Since(start)
134+
cancelTaskExecution() // all assertion variable captured, cancel counting task
135+
136+
assert.Equal(t, "expected error", err.Error(), "expecting first error")
137+
// should finsh after both error
138+
assert.True(t, elapsed >= 20*time.Millisecond, fmt.Sprintf("actually elapsed: %v", elapsed))
139+
140+
assert.Equal(t, asynctask.StateFailed, errTskState, "error task should failed")
141+
assert.Equal(t, asynctask.StateFailed, panicTskState, "panic task should Not failed")
142+
143+
// counting task do testing.Logf in another go routine
144+
// while testing.Logf would cause DataRace error when test is already finished: https://github.com/golang/go/issues/40343
145+
// wait minor time for the go routine to finish.
146+
time.Sleep(1 * time.Millisecond)
147+
}
148+
149+
func TestWaitAnyErrorWithFailOnAnyErrorCase(t *testing.T) {
150+
t.Parallel()
151+
ctx, cancelTaskExecution := newTestContextWithTimeout(t, 3*time.Second)
152+
153+
start := time.Now()
154+
errorTsk := asynctask.Start(ctx, getErrorTask("expected error", 10*time.Millisecond))
155+
result := "something"
156+
completedTsk := asynctask.NewCompletedTask(&result)
157+
err := asynctask.WaitAny(ctx, &asynctask.WaitAnyOptions{FailOnAnyError: true}, errorTsk, completedTsk)
158+
assert.NoError(t, err)
159+
elapsed := time.Since(start)
160+
// should finish after right away
161+
assert.True(t, elapsed < 20*time.Millisecond, fmt.Sprintf("actually elapsed: %v", elapsed))
162+
163+
start = time.Now()
164+
countingTsk := asynctask.Start(ctx, getCountingTask(10, "countingPer40ms", 40*time.Millisecond))
165+
errorTsk = asynctask.Start(ctx, getErrorTask("expected error", 10*time.Millisecond))
166+
panicTsk := asynctask.Start(ctx, getPanicTask(20*time.Millisecond))
167+
err = asynctask.WaitAny(ctx, &asynctask.WaitAnyOptions{FailOnAnyError: true}, countingTsk, errorTsk, panicTsk)
168+
assert.Error(t, err)
169+
completedTskState := completedTsk.State()
170+
assert.Equal(t, asynctask.StateCompleted, completedTskState, "completed task should finished")
171+
172+
countingTskState := countingTsk.State()
173+
panicTskState := panicTsk.State()
174+
errTskState := errorTsk.State()
175+
elapsed = time.Since(start)
176+
cancelTaskExecution() // all assertion variable captured, cancel counting task
177+
178+
assert.Equal(t, "expected error", err.Error(), "expecting first error")
179+
// should finsh after first error
180+
assert.True(t, elapsed >= 10*time.Millisecond && elapsed < 20*time.Millisecond, fmt.Sprintf("actually elapsed: %v", elapsed))
181+
182+
assert.Equal(t, asynctask.StateRunning, countingTskState, "countingTask should NOT finished")
183+
assert.Equal(t, asynctask.StateFailed, errTskState, "error task should failed")
184+
assert.Equal(t, asynctask.StateRunning, panicTskState, "panic task should Not failed")
185+
186+
// counting task do testing.Logf in another go routine
187+
// while testing.Logf would cause DataRace error when test is already finished: https://github.com/golang/go/issues/40343
188+
// wait minor time for the go routine to finish.
189+
time.Sleep(1 * time.Millisecond)
190+
}

0 commit comments

Comments
 (0)