mirror of
https://github.com/tendermint/tendermint.git
synced 2026-09-03 14:47:16 +00:00
Move libs/async to internal/libs/async. (#7449)
This commit is contained in:
M. J. Fromberger
@@ -1,184 +0,0 @@
|
||||
package async
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"runtime"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
//----------------------------------------
|
||||
// Task
|
||||
|
||||
// val: the value returned after task execution.
|
||||
// err: the error returned during task completion.
|
||||
// abort: tells Parallel to return, whether or not all tasks have completed.
|
||||
type Task func(i int) (val interface{}, abort bool, err error)
|
||||
|
||||
type TaskResult struct {
|
||||
Value interface{}
|
||||
Error error
|
||||
}
|
||||
|
||||
type TaskResultCh <-chan TaskResult
|
||||
|
||||
type taskResultOK struct {
|
||||
TaskResult
|
||||
OK bool
|
||||
}
|
||||
|
||||
type TaskResultSet struct {
|
||||
chz []TaskResultCh
|
||||
results []taskResultOK
|
||||
}
|
||||
|
||||
func newTaskResultSet(chz []TaskResultCh) *TaskResultSet {
|
||||
return &TaskResultSet{
|
||||
chz: chz,
|
||||
results: make([]taskResultOK, len(chz)),
|
||||
}
|
||||
}
|
||||
|
||||
func (trs *TaskResultSet) Channels() []TaskResultCh {
|
||||
return trs.chz
|
||||
}
|
||||
|
||||
func (trs *TaskResultSet) LatestResult(index int) (TaskResult, bool) {
|
||||
if len(trs.results) <= index {
|
||||
return TaskResult{}, false
|
||||
}
|
||||
resultOK := trs.results[index]
|
||||
return resultOK.TaskResult, resultOK.OK
|
||||
}
|
||||
|
||||
// NOTE: Not concurrency safe.
|
||||
// Writes results to trs.results without waiting for all tasks to complete.
|
||||
func (trs *TaskResultSet) Reap() *TaskResultSet {
|
||||
for i := 0; i < len(trs.results); i++ {
|
||||
var trch = trs.chz[i]
|
||||
select {
|
||||
case result, ok := <-trch:
|
||||
if ok {
|
||||
// Write result.
|
||||
trs.results[i] = taskResultOK{
|
||||
TaskResult: result,
|
||||
OK: true,
|
||||
}
|
||||
}
|
||||
// else {
|
||||
// We already wrote it.
|
||||
// }
|
||||
default:
|
||||
// Do nothing.
|
||||
}
|
||||
}
|
||||
return trs
|
||||
}
|
||||
|
||||
// NOTE: Not concurrency safe.
|
||||
// Like Reap() but waits until all tasks have returned or panic'd.
|
||||
func (trs *TaskResultSet) Wait() *TaskResultSet {
|
||||
for i := 0; i < len(trs.results); i++ {
|
||||
var trch = trs.chz[i]
|
||||
result, ok := <-trch
|
||||
if ok {
|
||||
// Write result.
|
||||
trs.results[i] = taskResultOK{
|
||||
TaskResult: result,
|
||||
OK: true,
|
||||
}
|
||||
}
|
||||
// else {
|
||||
// We already wrote it.
|
||||
// }
|
||||
}
|
||||
return trs
|
||||
}
|
||||
|
||||
// Returns the firstmost (by task index) error as
|
||||
// discovered by all previous Reap() calls.
|
||||
func (trs *TaskResultSet) FirstValue() interface{} {
|
||||
for _, result := range trs.results {
|
||||
if result.Value != nil {
|
||||
return result.Value
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Returns the firstmost (by task index) error as
|
||||
// discovered by all previous Reap() calls.
|
||||
func (trs *TaskResultSet) FirstError() error {
|
||||
for _, result := range trs.results {
|
||||
if result.Error != nil {
|
||||
return result.Error
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
//----------------------------------------
|
||||
// Parallel
|
||||
|
||||
// Run tasks in parallel, with ability to abort early.
|
||||
// Returns ok=false iff any of the tasks returned abort=true.
|
||||
// NOTE: Do not implement quit features here. Instead, provide convenient
|
||||
// concurrent quit-like primitives, passed implicitly via Task closures. (e.g.
|
||||
// it's not Parallel's concern how you quit/abort your tasks).
|
||||
func Parallel(tasks ...Task) (trs *TaskResultSet, ok bool) {
|
||||
var taskResultChz = make([]TaskResultCh, len(tasks)) // To return.
|
||||
var taskDoneCh = make(chan bool, len(tasks)) // A "wait group" channel, early abort if any true received.
|
||||
var numPanics = new(int32) // Keep track of panics to set ok=false later.
|
||||
|
||||
// We will set it to false iff any tasks panic'd or returned abort.
|
||||
ok = true
|
||||
|
||||
// Start all tasks in parallel in separate goroutines.
|
||||
// When the task is complete, it will appear in the
|
||||
// respective taskResultCh (associated by task index).
|
||||
for i, task := range tasks {
|
||||
var taskResultCh = make(chan TaskResult, 1) // Capacity for 1 result.
|
||||
taskResultChz[i] = taskResultCh
|
||||
go func(i int, task Task, taskResultCh chan TaskResult) {
|
||||
// Recovery
|
||||
defer func() {
|
||||
if pnk := recover(); pnk != nil {
|
||||
atomic.AddInt32(numPanics, 1)
|
||||
// Send panic to taskResultCh.
|
||||
const size = 64 << 10
|
||||
buf := make([]byte, size)
|
||||
buf = buf[:runtime.Stack(buf, false)]
|
||||
taskResultCh <- TaskResult{nil, fmt.Errorf("panic in task %v : %s", pnk, buf)}
|
||||
// Closing taskResultCh lets trs.Wait() work.
|
||||
close(taskResultCh)
|
||||
// Decrement waitgroup.
|
||||
taskDoneCh <- false
|
||||
}
|
||||
}()
|
||||
// Run the task.
|
||||
var val, abort, err = task(i)
|
||||
// Send val/err to taskResultCh.
|
||||
// NOTE: Below this line, nothing must panic/
|
||||
taskResultCh <- TaskResult{val, err}
|
||||
// Closing taskResultCh lets trs.Wait() work.
|
||||
close(taskResultCh)
|
||||
// Decrement waitgroup.
|
||||
taskDoneCh <- abort
|
||||
}(i, task, taskResultCh)
|
||||
}
|
||||
|
||||
// Wait until all tasks are done, or until abort.
|
||||
// DONE_LOOP:
|
||||
for i := 0; i < len(tasks); i++ {
|
||||
abort := <-taskDoneCh
|
||||
if abort {
|
||||
ok = false
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
// Ok is also false if there were any panics.
|
||||
// We must do this check here (after DONE_LOOP).
|
||||
ok = ok && (atomic.LoadInt32(numPanics) == 0)
|
||||
|
||||
return newTaskResultSet(taskResultChz).Reap(), ok
|
||||
}
|
||||
@@ -1,160 +0,0 @@
|
||||
package async
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync/atomic"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
)
|
||||
|
||||
func TestParallel(t *testing.T) {
|
||||
|
||||
// Create tasks.
|
||||
var counter = new(int32)
|
||||
var tasks = make([]Task, 100*1000)
|
||||
for i := 0; i < len(tasks); i++ {
|
||||
tasks[i] = func(i int) (res interface{}, abort bool, err error) {
|
||||
atomic.AddInt32(counter, 1)
|
||||
return -1 * i, false, nil
|
||||
}
|
||||
}
|
||||
|
||||
// Run in parallel.
|
||||
var trs, ok = Parallel(tasks...)
|
||||
assert.True(t, ok)
|
||||
|
||||
// Verify.
|
||||
assert.Equal(t, int(*counter), len(tasks), "Each task should have incremented the counter already")
|
||||
var failedTasks int
|
||||
for i := 0; i < len(tasks); i++ {
|
||||
taskResult, ok := trs.LatestResult(i)
|
||||
switch {
|
||||
case !ok:
|
||||
assert.Fail(t, "Task #%v did not complete.", i)
|
||||
failedTasks++
|
||||
case taskResult.Error != nil:
|
||||
assert.Fail(t, "Task should not have errored but got %v", taskResult.Error)
|
||||
failedTasks++
|
||||
case !assert.Equal(t, -1*i, taskResult.Value.(int)):
|
||||
assert.Fail(t, "Task should have returned %v but got %v", -1*i, taskResult.Value.(int))
|
||||
failedTasks++
|
||||
}
|
||||
// else {
|
||||
// Good!
|
||||
// }
|
||||
}
|
||||
assert.Equal(t, failedTasks, 0, "No task should have failed")
|
||||
assert.Nil(t, trs.FirstError(), "There should be no errors")
|
||||
assert.Equal(t, 0, trs.FirstValue(), "First value should be 0")
|
||||
}
|
||||
|
||||
func TestParallelAbort(t *testing.T) {
|
||||
|
||||
var flow1 = make(chan struct{}, 1)
|
||||
var flow2 = make(chan struct{}, 1)
|
||||
var flow3 = make(chan struct{}, 1) // Cap must be > 0 to prevent blocking.
|
||||
var flow4 = make(chan struct{}, 1)
|
||||
|
||||
// Create tasks.
|
||||
var tasks = []Task{
|
||||
func(i int) (res interface{}, abort bool, err error) {
|
||||
assert.Equal(t, i, 0)
|
||||
flow1 <- struct{}{}
|
||||
return 0, false, nil
|
||||
},
|
||||
func(i int) (res interface{}, abort bool, err error) {
|
||||
assert.Equal(t, i, 1)
|
||||
flow2 <- <-flow1
|
||||
return 1, false, errors.New("some error")
|
||||
},
|
||||
func(i int) (res interface{}, abort bool, err error) {
|
||||
assert.Equal(t, i, 2)
|
||||
flow3 <- <-flow2
|
||||
return 2, true, nil
|
||||
},
|
||||
func(i int) (res interface{}, abort bool, err error) {
|
||||
assert.Equal(t, i, 3)
|
||||
<-flow4
|
||||
return 3, false, nil
|
||||
},
|
||||
}
|
||||
|
||||
// Run in parallel.
|
||||
var taskResultSet, ok = Parallel(tasks...)
|
||||
assert.False(t, ok, "ok should be false since we aborted task #2.")
|
||||
|
||||
// Verify task #3.
|
||||
// Initially taskResultSet.chz[3] sends nothing since flow4 didn't send.
|
||||
waitTimeout(t, taskResultSet.chz[3], "Task #3")
|
||||
|
||||
// Now let the last task (#3) complete after abort.
|
||||
flow4 <- <-flow3
|
||||
|
||||
// Wait until all tasks have returned or panic'd.
|
||||
taskResultSet.Wait()
|
||||
|
||||
// Verify task #0, #1, #2.
|
||||
checkResult(t, taskResultSet, 0, 0, nil, nil)
|
||||
checkResult(t, taskResultSet, 1, 1, errors.New("some error"), nil)
|
||||
checkResult(t, taskResultSet, 2, 2, nil, nil)
|
||||
checkResult(t, taskResultSet, 3, 3, nil, nil)
|
||||
}
|
||||
|
||||
func TestParallelRecover(t *testing.T) {
|
||||
|
||||
// Create tasks.
|
||||
var tasks = []Task{
|
||||
func(i int) (res interface{}, abort bool, err error) {
|
||||
return 0, false, nil
|
||||
},
|
||||
func(i int) (res interface{}, abort bool, err error) {
|
||||
return 1, false, errors.New("some error")
|
||||
},
|
||||
func(i int) (res interface{}, abort bool, err error) {
|
||||
panic(2)
|
||||
},
|
||||
}
|
||||
|
||||
// Run in parallel.
|
||||
var taskResultSet, ok = Parallel(tasks...)
|
||||
assert.False(t, ok, "ok should be false since we panic'd in task #2.")
|
||||
|
||||
// Verify task #0, #1, #2.
|
||||
checkResult(t, taskResultSet, 0, 0, nil, nil)
|
||||
checkResult(t, taskResultSet, 1, 1, errors.New("some error"), nil)
|
||||
checkResult(t, taskResultSet, 2, nil, nil, fmt.Errorf("panic in task %v", 2).Error())
|
||||
}
|
||||
|
||||
// Wait for result
|
||||
func checkResult(t *testing.T, taskResultSet *TaskResultSet, index int,
|
||||
val interface{}, err error, pnk interface{}) {
|
||||
taskResult, ok := taskResultSet.LatestResult(index)
|
||||
taskName := fmt.Sprintf("Task #%v", index)
|
||||
assert.True(t, ok, "TaskResultCh unexpectedly closed for %v", taskName)
|
||||
assert.Equal(t, val, taskResult.Value, taskName)
|
||||
switch {
|
||||
case err != nil:
|
||||
assert.Equal(t, err.Error(), taskResult.Error.Error(), taskName)
|
||||
case pnk != nil:
|
||||
assert.Contains(t, taskResult.Error.Error(), pnk, taskName)
|
||||
default:
|
||||
assert.Nil(t, taskResult.Error, taskName)
|
||||
}
|
||||
}
|
||||
|
||||
// Wait for timeout (no result)
|
||||
func waitTimeout(t *testing.T, taskResultCh TaskResultCh, taskName string) {
|
||||
select {
|
||||
case _, ok := <-taskResultCh:
|
||||
if !ok {
|
||||
assert.Fail(t, "TaskResultCh unexpectedly closed (%v)", taskName)
|
||||
} else {
|
||||
assert.Fail(t, "TaskResultCh unexpectedly returned for %v", taskName)
|
||||
}
|
||||
case <-time.After(1 * time.Second): // TODO use deterministic time?
|
||||
// Good!
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user