mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2024-01-19 02:48:24 +00:00
add future list
This commit is contained in:
parent
1484cb224f
commit
648b33beb0
|
@ -2,7 +2,11 @@ package util
|
||||||
|
|
||||||
// initial version comes from https://hackernoon.com/asyncawait-in-golang-an-introductory-guide-ol1e34sg
|
// initial version comes from https://hackernoon.com/asyncawait-in-golang-an-introductory-guide-ol1e34sg
|
||||||
|
|
||||||
import "context"
|
import (
|
||||||
|
"container/list"
|
||||||
|
"context"
|
||||||
|
"sync"
|
||||||
|
)
|
||||||
|
|
||||||
type Future interface {
|
type Future interface {
|
||||||
Await() interface{}
|
Await() interface{}
|
||||||
|
@ -17,23 +21,27 @@ func (f future) Await() interface{} {
|
||||||
}
|
}
|
||||||
|
|
||||||
type LimitedAsyncExecutor struct {
|
type LimitedAsyncExecutor struct {
|
||||||
executor *LimitedConcurrentExecutor
|
executor *LimitedConcurrentExecutor
|
||||||
|
futureList *list.List
|
||||||
|
futureListCond *sync.Cond
|
||||||
}
|
}
|
||||||
|
|
||||||
func NewLimitedAsyncExecutor(limit int) *LimitedAsyncExecutor {
|
func NewLimitedAsyncExecutor(limit int) *LimitedAsyncExecutor {
|
||||||
return &LimitedAsyncExecutor{
|
return &LimitedAsyncExecutor{
|
||||||
executor: NewLimitedConcurrentExecutor(limit),
|
executor: NewLimitedConcurrentExecutor(limit),
|
||||||
|
futureList: list.New(),
|
||||||
|
futureListCond: sync.NewCond(&sync.Mutex{}),
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ae *LimitedAsyncExecutor) Execute(job func() interface{}) Future {
|
func (ae *LimitedAsyncExecutor) Execute(job func() interface{}) {
|
||||||
var result interface{}
|
var result interface{}
|
||||||
c := make(chan struct{})
|
c := make(chan struct{})
|
||||||
ae.executor.Execute(func() {
|
ae.executor.Execute(func() {
|
||||||
defer close(c)
|
defer close(c)
|
||||||
result = job()
|
result = job()
|
||||||
})
|
})
|
||||||
return future{await: func(ctx context.Context) interface{} {
|
f := future{await: func(ctx context.Context) interface{} {
|
||||||
select {
|
select {
|
||||||
case <-ctx.Done():
|
case <-ctx.Done():
|
||||||
return ctx.Err()
|
return ctx.Err()
|
||||||
|
@ -41,4 +49,18 @@ func (ae *LimitedAsyncExecutor) Execute(job func() interface{}) Future {
|
||||||
return result
|
return result
|
||||||
}
|
}
|
||||||
}}
|
}}
|
||||||
|
ae.futureListCond.L.Lock()
|
||||||
|
ae.futureList.PushBack(f)
|
||||||
|
ae.futureListCond.Signal()
|
||||||
|
ae.futureListCond.L.Unlock()
|
||||||
|
}
|
||||||
|
|
||||||
|
func (ae *LimitedAsyncExecutor) NextFuture() Future {
|
||||||
|
ae.futureListCond.L.Lock()
|
||||||
|
for ae.futureList.Len() == 0 {
|
||||||
|
ae.futureListCond.Wait()
|
||||||
|
}
|
||||||
|
f := ae.futureList.Remove(ae.futureList.Front())
|
||||||
|
ae.futureListCond.L.Unlock()
|
||||||
|
return f.(Future)
|
||||||
}
|
}
|
||||||
|
|
|
@ -10,17 +10,17 @@ import (
|
||||||
|
|
||||||
func TestAsyncPool(t *testing.T) {
|
func TestAsyncPool(t *testing.T) {
|
||||||
p := NewLimitedAsyncExecutor(3)
|
p := NewLimitedAsyncExecutor(3)
|
||||||
var results []Future
|
|
||||||
|
|
||||||
results = append(results, p.Execute(FirstFunc))
|
p.Execute(FirstFunc)
|
||||||
results = append(results, p.Execute(SecondFunc))
|
p.Execute(SecondFunc)
|
||||||
results = append(results, p.Execute(ThirdFunc))
|
p.Execute(ThirdFunc)
|
||||||
results = append(results, p.Execute(FourthFunc))
|
p.Execute(FourthFunc)
|
||||||
results = append(results, p.Execute(FifthFunc))
|
p.Execute(FifthFunc)
|
||||||
|
|
||||||
var sorted_results []int
|
var sorted_results []int
|
||||||
for _, r := range results {
|
for i := 0; i < 5; i++ {
|
||||||
x := r.Await().(int)
|
f := p.NextFuture()
|
||||||
|
x := f.Await().(int)
|
||||||
println(x)
|
println(x)
|
||||||
sorted_results = append(sorted_results, x)
|
sorted_results = append(sorted_results, x)
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in a new issue