cl: compile async functions
This commit is contained in:
Li Jie
@@ -1,623 +0,0 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/goplus/llgo/x/async"
|
||||
"github.com/goplus/llgo/x/async/naive"
|
||||
"github.com/goplus/llgo/x/tuple"
|
||||
)
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
func http(method string, url string, callback func(resp *Response, err error)) {
|
||||
go func() {
|
||||
body := ""
|
||||
if strings.HasPrefix(url, "http://example.com/user/") {
|
||||
name := url[len("http://example.com/user/"):]
|
||||
body = `{"name":"` + name + `"}`
|
||||
} else if strings.HasPrefix(url, "http://example.com/score/") {
|
||||
body = "99.5"
|
||||
}
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
resp := &Response{StatusCode: 200, Body: body}
|
||||
callback(resp, nil)
|
||||
}()
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
type Response struct {
|
||||
StatusCode int
|
||||
|
||||
Body string
|
||||
}
|
||||
|
||||
func (r *Response) Text() (co *async.Promise[tuple.Tuple2[string, error]]) {
|
||||
co.Return(tuple.Tuple2[string, error]{V1: r.Body, V2: nil})
|
||||
return
|
||||
}
|
||||
|
||||
func (r *Response) TextCompiled() *naive.PromiseImpl[tuple.Tuple2[string, error]] {
|
||||
co := &naive.PromiseImpl[tuple.Tuple2[string, error]]{}
|
||||
co.Debug = "Text"
|
||||
co.Func = func() {
|
||||
switch co.Next {
|
||||
case 0:
|
||||
co.Next = -1
|
||||
co.Return(tuple.Tuple2[string, error]{V1: r.Body, V2: nil})
|
||||
return
|
||||
default:
|
||||
panic("Promise already done")
|
||||
}
|
||||
}
|
||||
return co
|
||||
}
|
||||
|
||||
// async AsyncHttpGet(url string) (resp *Response, err error) {
|
||||
// http("GET", url, func(resp *Response, err error) {
|
||||
// return resp, err
|
||||
// })
|
||||
// }
|
||||
func AsyncHttpGet(url string) *async.Promise[tuple.Tuple2[*Response, error]] {
|
||||
co := &async.Promise[tuple.Tuple2[*Response, error]]{}
|
||||
http("GET", url, func(resp *Response, err error) {
|
||||
co.Return(tuple.Tuple2[*Response, error]{V1: resp, V2: nil})
|
||||
})
|
||||
co.Suspend()
|
||||
return co
|
||||
}
|
||||
|
||||
func AsyncHttpGetCompiled(url string) *naive.PromiseImpl[tuple.Tuple2[*Response, error]] {
|
||||
co := &naive.PromiseImpl[tuple.Tuple2[*Response, error]]{}
|
||||
co.Debug = "HttpGet"
|
||||
co.Func = func() {
|
||||
switch co.Next {
|
||||
case 0:
|
||||
co.Next = -1
|
||||
http("GET", url, func(resp *Response, err error) {
|
||||
co.Return(tuple.Tuple2[*Response, error]{V1: resp, V2: nil})
|
||||
})
|
||||
co.Suspend()
|
||||
return
|
||||
default:
|
||||
panic("Promise already done")
|
||||
}
|
||||
}
|
||||
return co
|
||||
}
|
||||
|
||||
func AsyncHttpPost(url string) *async.Promise[tuple.Tuple2[*Response, error]] {
|
||||
co := &async.Promise[tuple.Tuple2[*Response, error]]{}
|
||||
http("POST", url, func(resp *Response, err error) {
|
||||
co.Return(tuple.Tuple2[*Response, error]{V1: resp, V2: nil})
|
||||
})
|
||||
co.Suspend()
|
||||
return co
|
||||
}
|
||||
|
||||
func AsyncHttpPostCompiled(url string) *naive.PromiseImpl[tuple.Tuple2[*Response, error]] {
|
||||
P := &naive.PromiseImpl[tuple.Tuple2[*Response, error]]{}
|
||||
P.Debug = "HttpPost"
|
||||
P.Func = func() {
|
||||
switch P.Next {
|
||||
case 0:
|
||||
P.Next = -1
|
||||
http("POST", url, func(resp *Response, err error) {
|
||||
P.Return(tuple.Tuple2[*Response, error]{V1: resp, V2: nil})
|
||||
})
|
||||
return
|
||||
default:
|
||||
panic("Promise already done")
|
||||
}
|
||||
}
|
||||
return P
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
type User struct {
|
||||
Name string
|
||||
}
|
||||
|
||||
func GetUser(name string) (co *naive.PromiseImpl[tuple.Tuple2[User, error]]) {
|
||||
resp, err := AsyncHttpGet("http://example.com/user/" + name).Await().Values()
|
||||
if err != nil {
|
||||
// return User{}, err
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
// return User{}, fmt.Errorf("http status code: %d", resp.StatusCode)
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: fmt.Errorf("http status code: %d", resp.StatusCode)})
|
||||
return
|
||||
}
|
||||
|
||||
body, err := resp.Text().Await().Values()
|
||||
if err != nil {
|
||||
// return User{}, err
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: err})
|
||||
return
|
||||
}
|
||||
user := User{}
|
||||
if err := json.Unmarshal([]byte(body), &user); err != nil {
|
||||
// return User{}, err
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
// return user, nil
|
||||
co.Return(tuple.Tuple2[User, error]{V1: user, V2: nil})
|
||||
return
|
||||
}
|
||||
|
||||
func GetUserCompiled(name string) (co *naive.PromiseImpl[tuple.Tuple2[User, error]]) {
|
||||
var state1 *naive.PromiseImpl[tuple.Tuple2[*Response, error]]
|
||||
var state2 *naive.PromiseImpl[tuple.Tuple2[string, error]]
|
||||
|
||||
co = &naive.PromiseImpl[tuple.Tuple2[User, error]]{}
|
||||
co.Debug = "GetUser"
|
||||
co.Func = func() {
|
||||
switch co.Next {
|
||||
case 0:
|
||||
co.Next = 1
|
||||
state1 = AsyncHttpGetCompiled("http://example.com/user/" + name)
|
||||
state1.Exec = co.Exec
|
||||
state1.Parent = co
|
||||
state1.Call()
|
||||
return
|
||||
case 1:
|
||||
co.Next = 2
|
||||
resp, err := state1.Value().Values()
|
||||
log.Printf("resp: %v, err: %v\n", resp, err)
|
||||
if err != nil {
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: fmt.Errorf("http status code: %d", resp.StatusCode)})
|
||||
return
|
||||
}
|
||||
|
||||
state2 = resp.TextCompiled()
|
||||
state2.Exec = co.Exec
|
||||
state2.Parent = co
|
||||
state2.Call()
|
||||
return
|
||||
case 2:
|
||||
co.Next = -1
|
||||
body, err := state2.Value().Values()
|
||||
if err != nil {
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: err})
|
||||
return
|
||||
}
|
||||
user := User{}
|
||||
log.Printf("body: %v\n", body)
|
||||
if err := json.Unmarshal([]byte(body), &user); err != nil {
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
log.Printf("resolve user: %+v\n", user)
|
||||
co.Return(tuple.Tuple2[User, error]{V1: user, V2: nil})
|
||||
return
|
||||
default:
|
||||
panic(fmt.Errorf("Promise already done, %+v", co))
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
func GetScore() (co *naive.PromiseImpl[tuple.Tuple2[float64, error]]) {
|
||||
resp, err := AsyncHttpGet("http://example.com/score/").Await().Values()
|
||||
if err != nil {
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
// return 0, fmt.Errorf("http status code: %d", resp.StatusCode)
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: fmt.Errorf("http status code: %d", resp.StatusCode)})
|
||||
return
|
||||
}
|
||||
|
||||
body, err := resp.Text().Await().Values()
|
||||
if err != nil {
|
||||
// return 0, err
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
score := 0.0
|
||||
if _, err := fmt.Sscanf(body, "%f", &score); err != nil {
|
||||
// return 0, err
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
// return score, nil
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: score, V2: nil})
|
||||
return
|
||||
}
|
||||
|
||||
func GetScoreCompiled() *naive.PromiseImpl[tuple.Tuple2[float64, error]] {
|
||||
var state1 *naive.PromiseImpl[tuple.Tuple2[*Response, error]]
|
||||
var state2 *naive.PromiseImpl[tuple.Tuple2[string, error]]
|
||||
|
||||
co := &naive.PromiseImpl[tuple.Tuple2[float64, error]]{}
|
||||
co.Debug = "GetScore"
|
||||
co.Func = func() {
|
||||
switch co.Next {
|
||||
case 0:
|
||||
co.Next = 1
|
||||
state1 = AsyncHttpGetCompiled("http://example.com/score/")
|
||||
state1.Exec = co.Exec
|
||||
state1.Parent = co
|
||||
state1.Call()
|
||||
return
|
||||
case 1:
|
||||
co.Next = 2
|
||||
|
||||
resp, err := state1.Value().Values()
|
||||
if err != nil {
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: fmt.Errorf("http status code: %d", resp.StatusCode)})
|
||||
return
|
||||
}
|
||||
|
||||
state2 = resp.TextCompiled()
|
||||
state2.Exec = co.Exec
|
||||
state2.Parent = co
|
||||
state2.Call()
|
||||
|
||||
return
|
||||
case 2:
|
||||
co.Next = -1
|
||||
body, err := state2.Value().Values()
|
||||
if err != nil {
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
score := 0.0
|
||||
if _, err := fmt.Sscanf(body, "%f", &score); err != nil {
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: err})
|
||||
return
|
||||
}
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: score, V2: nil})
|
||||
return
|
||||
default:
|
||||
panic("Promise already done")
|
||||
}
|
||||
}
|
||||
return co
|
||||
}
|
||||
|
||||
func DoUpdate(op string) (co *naive.PromiseImpl[error]) {
|
||||
resp, err := AsyncHttpPost("http://example.com/update/" + op).Await().Values()
|
||||
if err != nil {
|
||||
co.Return(err)
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
co.Return(fmt.Errorf("http status code: %d", resp.StatusCode))
|
||||
}
|
||||
|
||||
co.Return(nil)
|
||||
return
|
||||
}
|
||||
|
||||
func DoUpdateCompiled(op string) *naive.PromiseImpl[error] {
|
||||
var state1 *naive.PromiseImpl[tuple.Tuple2[*Response, error]]
|
||||
|
||||
co := &naive.PromiseImpl[error]{}
|
||||
co.Debug = "DoUpdate"
|
||||
co.Func = func() {
|
||||
switch co.Next {
|
||||
case 0:
|
||||
co.Next = 1
|
||||
state1 = AsyncHttpPostCompiled("http://example.com/update/" + op)
|
||||
state1.Exec = co.Exec
|
||||
state1.Parent = co
|
||||
state1.Call()
|
||||
return
|
||||
case 1:
|
||||
co.Next = -1
|
||||
resp, err := state1.Value().Values()
|
||||
if err != nil {
|
||||
co.Return(err)
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
co.Return(fmt.Errorf("http status code: %d", resp.StatusCode))
|
||||
return
|
||||
}
|
||||
|
||||
co.Return(nil)
|
||||
return
|
||||
default:
|
||||
panic("Promise already done")
|
||||
}
|
||||
}
|
||||
return co
|
||||
}
|
||||
|
||||
func GenInts() (co *naive.PromiseImpl[int]) {
|
||||
co.Yield(3)
|
||||
co.Yield(2)
|
||||
co.Yield(5)
|
||||
return
|
||||
}
|
||||
|
||||
func GenIntsCompiled() *naive.PromiseImpl[int] {
|
||||
co := &naive.PromiseImpl[int]{}
|
||||
co.Debug = "GenInts"
|
||||
co.Func = func() {
|
||||
switch co.Next {
|
||||
case 0:
|
||||
co.Next = 1
|
||||
co.Yield(3)
|
||||
return
|
||||
case 1:
|
||||
co.Next = 2
|
||||
co.Yield(2)
|
||||
return
|
||||
case 2:
|
||||
co.Next = 3
|
||||
co.Yield(5)
|
||||
return
|
||||
case 3:
|
||||
co.Next = -1
|
||||
default:
|
||||
panic("Generator already done")
|
||||
}
|
||||
}
|
||||
return co
|
||||
}
|
||||
|
||||
// Generator with async calls and panic
|
||||
func GenUsers() (co *naive.PromiseImpl[User]) {
|
||||
u, err := GetUser("Alice").Await().Values()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
co.Yield(u)
|
||||
u, err = GetUser("Bob").Await().Values()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
co.Yield(u)
|
||||
u, err = GetUser("Cindy").Await().Values()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
co.Yield(u)
|
||||
log.Printf("genUsers done\n")
|
||||
return
|
||||
}
|
||||
|
||||
func GenUsersCompiled() (resolve *naive.PromiseImpl[User]) {
|
||||
var state1, state2, state3 *naive.PromiseImpl[tuple.Tuple2[User, error]]
|
||||
|
||||
co := &naive.PromiseImpl[User]{}
|
||||
co.Debug = "GenUsers"
|
||||
co.Func = func() {
|
||||
switch co.Next {
|
||||
case 0:
|
||||
co.Next = 1
|
||||
state1 = GetUserCompiled("Alice")
|
||||
state1.Exec = co.Exec
|
||||
state1.Parent = co
|
||||
state1.Call()
|
||||
return
|
||||
case 1:
|
||||
co.Next = 2
|
||||
u, err := state1.Value().Values()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
} else {
|
||||
co.Yield(u)
|
||||
}
|
||||
return
|
||||
case 2:
|
||||
co.Next = 3
|
||||
state2 = GetUserCompiled("Bob")
|
||||
state2.Exec = co.Exec
|
||||
state2.Parent = co
|
||||
state2.Call()
|
||||
return
|
||||
case 3:
|
||||
co.Next = 4
|
||||
u, err := state2.Value().Values()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
} else {
|
||||
co.Yield(u)
|
||||
}
|
||||
return
|
||||
case 4:
|
||||
co.Next = 5
|
||||
state3 = GetUserCompiled("Cindy")
|
||||
state3.Exec = co.Exec
|
||||
state3.Parent = co
|
||||
state3.Call()
|
||||
return
|
||||
case 5:
|
||||
co.Next = 6
|
||||
u, err := state3.Value().Values()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
} else {
|
||||
co.Yield(u)
|
||||
}
|
||||
return
|
||||
case 6:
|
||||
co.Next = -1
|
||||
default:
|
||||
panic("Generator already done")
|
||||
}
|
||||
}
|
||||
return co
|
||||
}
|
||||
|
||||
func Demo() (co *async.Promise[async.Void]) {
|
||||
user, err := GetUser("1").Await().Values()
|
||||
log.Println(user, err)
|
||||
|
||||
user, err = naive.Race[tuple.Tuple2[User, error]](GetUser("2"), GetUser("3"), GetUser("4")).Value().Values()
|
||||
log.Println(user, err)
|
||||
|
||||
users := naive.All[tuple.Tuple2[User, error]]([]naive.AsyncCall[tuple.Tuple2[User, error]]{GetUser("5"), GetUser("6"), GetUser("7")}).Value()
|
||||
log.Println(users, err)
|
||||
|
||||
user, score, _ := naive.Await3Compiled[User, float64, async.Void](GetUser("8"), GetScore(), DoUpdate("update sth.")).Value().Values()
|
||||
log.Println(user, score, err)
|
||||
|
||||
// for loop with generator
|
||||
g := GenInts()
|
||||
for {
|
||||
g.Call()
|
||||
if g.Done() {
|
||||
break
|
||||
}
|
||||
log.Println("genInt:", g.Value(), g.Done())
|
||||
}
|
||||
|
||||
// for loop with async generator
|
||||
// for u, err := range GenUsers() {...}
|
||||
g1 := GenUsers()
|
||||
for {
|
||||
g1.Call()
|
||||
u := g1.Await()
|
||||
if g1.Done() {
|
||||
break
|
||||
}
|
||||
log.Println("genUser:", u, err)
|
||||
}
|
||||
|
||||
// TODO(lijie): select from multiple promises without channel
|
||||
// select {
|
||||
// case user := <-GetUser("123").Chan():
|
||||
// log.Println("user:", user)
|
||||
// case score := <-GetScore().Chan():
|
||||
// log.Println("score:", score)
|
||||
// case <-async.Timeout(5 * time.Second).Chan():
|
||||
// log.Println("timeout")
|
||||
// }
|
||||
|
||||
log.Println("Demo done")
|
||||
co.Return(async.Void{})
|
||||
return
|
||||
}
|
||||
|
||||
func DemoCompiled() *naive.PromiseImpl[async.Void] {
|
||||
var state1 *naive.PromiseImpl[tuple.Tuple2[User, error]]
|
||||
var state2 *naive.PromiseImpl[tuple.Tuple2[User, error]]
|
||||
var state3 *naive.PromiseImpl[[]tuple.Tuple2[User, error]]
|
||||
var state4 *naive.PromiseImpl[tuple.Tuple3[tuple.Tuple2[User, error], tuple.Tuple2[float64, error], error]]
|
||||
var g1 *naive.PromiseImpl[int]
|
||||
var g2 *naive.PromiseImpl[User]
|
||||
|
||||
P := &naive.PromiseImpl[async.Void]{}
|
||||
P.Debug = "Demo"
|
||||
P.Func = func() {
|
||||
switch P.Next {
|
||||
case 0:
|
||||
P.Next = 1
|
||||
state1 = GetUserCompiled("1")
|
||||
state1.Exec = P.Exec
|
||||
state1.Parent = P
|
||||
state1.Call()
|
||||
return
|
||||
case 1:
|
||||
P.Next = 2
|
||||
user, err := state1.Value().Values()
|
||||
log.Printf("user: %+v, err: %v\n", user, err)
|
||||
|
||||
state2 = naive.Race[tuple.Tuple2[User, error]](GetUserCompiled("2"), GetUserCompiled("3"), GetUserCompiled("4"))
|
||||
state2.Exec = P.Exec
|
||||
state2.Parent = P
|
||||
state2.Call()
|
||||
return
|
||||
case 2:
|
||||
P.Next = 3
|
||||
user, err := state2.Value().Values()
|
||||
log.Printf("race user: %+v, err: %v\n", user, err)
|
||||
|
||||
state3 = naive.All[tuple.Tuple2[User, error]]([]naive.AsyncCall[tuple.Tuple2[User, error]]{GetUserCompiled("5"), GetUserCompiled("6"), GetUserCompiled("7")})
|
||||
state3.Exec = P.Exec
|
||||
state3.Parent = P
|
||||
state3.Call()
|
||||
return
|
||||
case 3:
|
||||
|
||||
P.Next = 4
|
||||
users := state3.Value()
|
||||
log.Println(users)
|
||||
|
||||
state4 = naive.Await3Compiled[tuple.Tuple2[User, error], tuple.Tuple2[float64, error], error](GetUserCompiled("8"), GetScoreCompiled(), DoUpdateCompiled("update sth."))
|
||||
state4.Exec = P.Exec
|
||||
state4.Parent = P
|
||||
state4.Call()
|
||||
return
|
||||
case 4:
|
||||
P.Next = 5
|
||||
user, score, _ := state4.Value().Values()
|
||||
log.Println(user, score)
|
||||
|
||||
g1 = GenIntsCompiled()
|
||||
for {
|
||||
g1.Call()
|
||||
if g1.Done() {
|
||||
break
|
||||
}
|
||||
|
||||
log.Printf("genInt: %+v, done: %v\n", g1.Value(), g1.Done())
|
||||
}
|
||||
|
||||
g2 = GenUsersCompiled()
|
||||
g2.Exec = P.Exec
|
||||
g2.Parent = P
|
||||
g2.Call()
|
||||
return
|
||||
case 5:
|
||||
g2.Call()
|
||||
if g2.Done() {
|
||||
P.Next = -1
|
||||
log.Printf("Demo done\n")
|
||||
P.Return(async.Void{})
|
||||
return
|
||||
}
|
||||
log.Printf("genUser: %+v, done: %v\n", g2.Value(), g2.Done())
|
||||
return
|
||||
default:
|
||||
panic("Promise already done")
|
||||
}
|
||||
}
|
||||
return P
|
||||
}
|
||||
|
||||
func main() {
|
||||
log.SetFlags(log.Lshortfile | log.LstdFlags)
|
||||
log.Printf("=========== Run Naive Demo ===========\n")
|
||||
v := naive.RunImpl[async.Void](DemoCompiled())
|
||||
log.Println(v)
|
||||
log.Printf("=========== Run Naive Demo finished ===========\n")
|
||||
|
||||
log.Printf("=========== Run Demo ===========\n")
|
||||
v1 := Demo()
|
||||
log.Println(v1)
|
||||
log.Printf("=========== Run Demo finished ===========\n")
|
||||
}
|
||||
229
x/async/_demo/asyncdemo/asyncdemo.go
Normal file
229
x/async/_demo/asyncdemo/asyncdemo.go
Normal file
@@ -0,0 +1,229 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"log"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/goplus/llgo/x/async"
|
||||
"github.com/goplus/llgo/x/tuple"
|
||||
)
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
func http(method string, url string, callback func(resp *Response, err error)) {
|
||||
go func() {
|
||||
body := ""
|
||||
if strings.HasPrefix(url, "http://example.com/user/") {
|
||||
name := url[len("http://example.com/user/"):]
|
||||
body = `{"name":"` + name + `"}`
|
||||
} else if strings.HasPrefix(url, "http://example.com/score/") {
|
||||
body = "99.5"
|
||||
}
|
||||
time.Sleep(200 * time.Millisecond)
|
||||
resp := &Response{StatusCode: 200, Body: body}
|
||||
callback(resp, nil)
|
||||
}()
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
type Response struct {
|
||||
StatusCode int
|
||||
|
||||
Body string
|
||||
}
|
||||
|
||||
func (r *Response) Text() (co async.Promise[tuple.Tuple2[string, error]]) {
|
||||
co.Return(tuple.Tuple2[string, error]{V1: r.Body, V2: nil})
|
||||
return
|
||||
}
|
||||
|
||||
// async AsyncHttpGet(url string) (resp *Response, err error) {
|
||||
// http("GET", url, func(resp *Response, err error) {
|
||||
// return resp, err
|
||||
// })
|
||||
// }
|
||||
func AsyncHttpGet(url string) (co async.Promise[tuple.Tuple2[*Response, error]]) {
|
||||
return co.Async(func(resolve func(tuple.Tuple2[*Response, error])) {
|
||||
http("GET", url, func(resp *Response, err error) {
|
||||
resolve(tuple.Tuple2[*Response, error]{V1: resp, V2: nil})
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
func AsyncHttpPost(url string) (co async.Promise[tuple.Tuple2[*Response, error]]) {
|
||||
http("POST", url, func(resp *Response, err error) {
|
||||
// co.Return(tuple.Tuple2[*Response, error]{V1: resp, V2: nil})
|
||||
})
|
||||
co.Suspend()
|
||||
return
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
type User struct {
|
||||
Name string
|
||||
}
|
||||
|
||||
func GetUser(name string) (co async.Promise[tuple.Tuple2[User, error]]) {
|
||||
resp, err := AsyncHttpGet("http://example.com/user/" + name).Await().Values()
|
||||
if err != nil {
|
||||
// return User{}, err
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
// return User{}, fmt.Errorf("http status code: %d", resp.StatusCode)
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: fmt.Errorf("http status code: %d", resp.StatusCode)})
|
||||
return
|
||||
}
|
||||
|
||||
body, err := resp.Text().Await().Values()
|
||||
if err != nil {
|
||||
// return User{}, err
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: err})
|
||||
return
|
||||
}
|
||||
user := User{}
|
||||
if err := json.Unmarshal([]byte(body), &user); err != nil {
|
||||
// return User{}, err
|
||||
co.Return(tuple.Tuple2[User, error]{V1: User{}, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
// return user, nil
|
||||
co.Return(tuple.Tuple2[User, error]{V1: user, V2: nil})
|
||||
return
|
||||
}
|
||||
|
||||
func GetScore() (co *async.Promise[tuple.Tuple2[float64, error]]) {
|
||||
resp, err := AsyncHttpGet("http://example.com/score/").Await().Values()
|
||||
if err != nil {
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
// return 0, fmt.Errorf("http status code: %d", resp.StatusCode)
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: fmt.Errorf("http status code: %d", resp.StatusCode)})
|
||||
return
|
||||
}
|
||||
|
||||
body, err := resp.Text().Await().Values()
|
||||
if err != nil {
|
||||
// return 0, err
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
score := 0.0
|
||||
if _, err := fmt.Sscanf(body, "%f", &score); err != nil {
|
||||
// return 0, err
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: 0, V2: err})
|
||||
return
|
||||
}
|
||||
|
||||
// return score, nil
|
||||
co.Return(tuple.Tuple2[float64, error]{V1: score, V2: nil})
|
||||
return
|
||||
}
|
||||
|
||||
func DoUpdate(op string) (co *async.Promise[error]) {
|
||||
resp, err := AsyncHttpPost("http://example.com/update/" + op).Await().Values()
|
||||
if err != nil {
|
||||
co.Return(err)
|
||||
return
|
||||
}
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
co.Return(fmt.Errorf("http status code: %d", resp.StatusCode))
|
||||
}
|
||||
|
||||
co.Return(nil)
|
||||
return
|
||||
}
|
||||
|
||||
func GenInts() (co *async.Promise[int]) {
|
||||
co.Yield(3)
|
||||
co.Yield(2)
|
||||
co.Yield(5)
|
||||
return
|
||||
}
|
||||
|
||||
// Generator with async calls and panic
|
||||
func GenUsers() (co *async.Promise[User]) {
|
||||
u, err := GetUser("Alice").Await().Values()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
co.Yield(u)
|
||||
u, err = GetUser("Bob").Await().Values()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
co.Yield(u)
|
||||
u, err = GetUser("Cindy").Await().Values()
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
co.Yield(u)
|
||||
log.Printf("genUsers done\n")
|
||||
return
|
||||
}
|
||||
|
||||
func Demo() (co *async.Promise[async.Void]) {
|
||||
user, err := GetUser("1").Await().Values()
|
||||
log.Println(user, err)
|
||||
|
||||
// user, err = naive.Race[tuple.Tuple2[User, error]](GetUser("2"), GetUser("3"), GetUser("4")).Value().Values()
|
||||
// log.Println(user, err)
|
||||
|
||||
// users := naive.All[tuple.Tuple2[User, error]]([]naive.AsyncCall[tuple.Tuple2[User, error]]{GetUser("5"), GetUser("6"), GetUser("7")}).Value()
|
||||
// log.Println(users, err)
|
||||
|
||||
// user, score, _ := naive.Await3Compiled[User, float64, async.Void](GetUser("8"), GetScore(), DoUpdate("update sth.")).Value().Values()
|
||||
// log.Println(user, score, err)
|
||||
|
||||
// for loop with generator
|
||||
g := GenInts()
|
||||
for !g.Done() {
|
||||
log.Println("genInt:", g.Value(), g.Done())
|
||||
g.Resume()
|
||||
}
|
||||
|
||||
// for loop with async generator
|
||||
// for u, err := range GenUsers() {...}
|
||||
g1 := GenUsers()
|
||||
for !g1.Done() {
|
||||
u := g1.Value()
|
||||
log.Println("genUser:", u)
|
||||
g1.Resume()
|
||||
}
|
||||
|
||||
// TODO(lijie): select from multiple promises without channel
|
||||
// select {
|
||||
// case user := <-GetUser("123").Chan():
|
||||
// log.Println("user:", user)
|
||||
// case score := <-GetScore().Chan():
|
||||
// log.Println("score:", score)
|
||||
// case <-async.Timeout(5 * time.Second).Chan():
|
||||
// log.Println("timeout")
|
||||
// }
|
||||
|
||||
log.Println("Demo done")
|
||||
co.Return(async.Void{})
|
||||
return
|
||||
}
|
||||
|
||||
func main() {
|
||||
log.SetFlags(log.Lshortfile | log.LstdFlags)
|
||||
|
||||
log.Printf("=========== Run Demo ===========\n")
|
||||
v1 := Demo()
|
||||
log.Println(v1)
|
||||
log.Printf("=========== Run Demo finished ===========\n")
|
||||
}
|
||||
21
x/async/_demo/gendemo/gendemo.go
Normal file
21
x/async/_demo/gendemo/gendemo.go
Normal file
@@ -0,0 +1,21 @@
|
||||
package main
|
||||
|
||||
import "github.com/goplus/llgo/x/async"
|
||||
|
||||
func GenInts() (co *async.Promise[int]) {
|
||||
print("1")
|
||||
co.Yield(1)
|
||||
print("2")
|
||||
co.Yield(2)
|
||||
print("3")
|
||||
co.Yield(3)
|
||||
print("4")
|
||||
return
|
||||
}
|
||||
|
||||
func main() {
|
||||
co := GenInts()
|
||||
for !co.Done() {
|
||||
print(co.Next())
|
||||
}
|
||||
}
|
||||
@@ -25,12 +25,8 @@ const (
|
||||
LLGoPackage = "decl"
|
||||
)
|
||||
|
||||
const debugAsync = false
|
||||
|
||||
type Void = [0]byte
|
||||
|
||||
type AsyncCall[TOut any] interface{}
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
type Promise[TOut any] struct {
|
||||
@@ -38,36 +34,45 @@ type Promise[TOut any] struct {
|
||||
value TOut
|
||||
}
|
||||
|
||||
// llgo:link PromiseImpl llgo.coAwait
|
||||
// // llgo:link (*Promise).Await llgo.coAwait
|
||||
func (p *Promise[TOut]) Await() TOut {
|
||||
panic("should not executed")
|
||||
}
|
||||
|
||||
// llgo:link Return llgo.coReturn
|
||||
func (p *Promise[TOut]) Return(v TOut) {
|
||||
panic("should not executed")
|
||||
p.value = v
|
||||
coReturn(p.hdl)
|
||||
}
|
||||
|
||||
// llgo:link Yield llgo.coYield
|
||||
func (p *Promise[TOut]) Yield(v TOut) {
|
||||
panic("should not executed")
|
||||
}
|
||||
// llgo:link (*Promise).Yield llgo.coYield
|
||||
func (p *Promise[TOut]) Yield(v TOut) {}
|
||||
|
||||
// llgo:link Suspend llgo.coSuspend
|
||||
func (p *Promise[TOut]) Suspend() {
|
||||
panic("should not executed")
|
||||
}
|
||||
// llgo:link (*Promise).Suspend llgo.coSuspend
|
||||
func (p *Promise[TOut]) Suspend() {}
|
||||
|
||||
// llgo:link Resume llgo.coResume
|
||||
func (p *Promise[TOut]) Resume() {
|
||||
panic("should not executed")
|
||||
coResume(p.hdl)
|
||||
}
|
||||
|
||||
func (p *Promise[TOut]) Next() TOut {
|
||||
coResume(p.hdl)
|
||||
return p.value
|
||||
}
|
||||
|
||||
// TODO(lijie): should merge to Yield()
|
||||
// call by llgo.coYield
|
||||
func (p *Promise[TOut]) setValue(v TOut) {
|
||||
p.value = v
|
||||
}
|
||||
|
||||
func (p *Promise[TOut]) Value() TOut {
|
||||
return p.value
|
||||
}
|
||||
|
||||
// llgo:link Run llgo.coRun
|
||||
func (p *Promise[TOut]) Done() bool {
|
||||
return coDone(p.hdl) != 0
|
||||
}
|
||||
|
||||
func Run[TOut any](f func() TOut) TOut {
|
||||
panic("should not executed")
|
||||
}
|
||||
|
||||
39
x/async/coro.go
Normal file
39
x/async/coro.go
Normal file
@@ -0,0 +1,39 @@
|
||||
/*
|
||||
* Copyright (c) 2024 The GoPlus Authors (goplus.org). All rights reserved.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package async
|
||||
|
||||
import (
|
||||
"unsafe"
|
||||
_ "unsafe"
|
||||
|
||||
"github.com/goplus/llgo/c"
|
||||
)
|
||||
|
||||
// llgo:link coDone llgo.coDone
|
||||
func coDone(hdl unsafe.Pointer) c.Char {
|
||||
panic("should not executed")
|
||||
}
|
||||
|
||||
// llgo:link coResume llgo.coResume
|
||||
func coResume(hdl unsafe.Pointer) {
|
||||
panic("should not executed")
|
||||
}
|
||||
|
||||
// llgo:link coReturn llgo.coReturn
|
||||
func coReturn(hdl unsafe.Pointer) {
|
||||
panic("should not executed")
|
||||
}
|
||||
@@ -1,285 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2024 The GoPlus Authors (goplus.org). All rights reserved.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package naive
|
||||
|
||||
import (
|
||||
"log"
|
||||
"sync"
|
||||
"time"
|
||||
_ "unsafe"
|
||||
|
||||
"github.com/goplus/llgo/x/async"
|
||||
"github.com/goplus/llgo/x/tuple"
|
||||
)
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
func TimeoutCompiled(d time.Duration) *PromiseImpl[async.Void] {
|
||||
P := &PromiseImpl[async.Void]{}
|
||||
P.Debug = "Timeout"
|
||||
P.Func = func() {
|
||||
go func() {
|
||||
time.Sleep(d)
|
||||
P.Return(async.Void{})
|
||||
}()
|
||||
}
|
||||
return P
|
||||
}
|
||||
|
||||
type Result[T any] struct {
|
||||
V T
|
||||
Err error
|
||||
}
|
||||
|
||||
func Race[OutT any](acs ...AsyncCall[OutT]) *PromiseImpl[OutT] {
|
||||
if len(acs) == 0 {
|
||||
panic("race: no promise")
|
||||
}
|
||||
ps := make([]*PromiseImpl[OutT], len(acs))
|
||||
for idx, ac := range acs {
|
||||
ps[idx] = ac.(*PromiseImpl[OutT])
|
||||
}
|
||||
remaining := len(acs)
|
||||
returned := false
|
||||
P := &PromiseImpl[OutT]{}
|
||||
P.Debug = "Race"
|
||||
P.Func = func() {
|
||||
switch P.Next {
|
||||
case 0:
|
||||
P.Next = 1
|
||||
for _, p := range ps {
|
||||
p.Exec = P.Exec
|
||||
p.Parent = P
|
||||
p.Call()
|
||||
}
|
||||
return
|
||||
case 1:
|
||||
remaining--
|
||||
if remaining < 0 {
|
||||
log.Fatalf("race: remaining < 0: %+v\n", remaining)
|
||||
}
|
||||
if returned {
|
||||
return
|
||||
}
|
||||
|
||||
for _, p := range ps {
|
||||
if p.Done() {
|
||||
if debugAsync {
|
||||
log.Printf("async.Race done: %+v won the race\n", p)
|
||||
}
|
||||
returned = true
|
||||
P.Return(p.value)
|
||||
return
|
||||
}
|
||||
}
|
||||
log.Fatalf("no promise done: %+v\n", ps)
|
||||
return
|
||||
default:
|
||||
panic("unreachable")
|
||||
}
|
||||
}
|
||||
return P
|
||||
}
|
||||
|
||||
func All[OutT any](acs []AsyncCall[OutT]) *PromiseImpl[[]OutT] {
|
||||
ps := make([]*PromiseImpl[OutT], len(acs))
|
||||
for idx, ac := range acs {
|
||||
ps[idx] = ac.(*PromiseImpl[OutT])
|
||||
}
|
||||
done := 0
|
||||
P := &PromiseImpl[[]OutT]{}
|
||||
P.Debug = "All"
|
||||
P.Func = func() {
|
||||
switch P.Next {
|
||||
case 0:
|
||||
P.Next = 1
|
||||
for _, p := range ps {
|
||||
p.Exec = P.Exec
|
||||
p.Parent = P
|
||||
p.Call()
|
||||
}
|
||||
return
|
||||
case 1:
|
||||
done++
|
||||
if done < len(acs) {
|
||||
return
|
||||
}
|
||||
P.Next = -1
|
||||
|
||||
for _, p := range ps {
|
||||
if !p.Done() {
|
||||
log.Fatalf("async.All: not done: %+v\n", p)
|
||||
}
|
||||
}
|
||||
|
||||
ret := make([]OutT, len(acs))
|
||||
for idx, p := range ps {
|
||||
ret[idx] = p.value
|
||||
}
|
||||
if debugAsync {
|
||||
log.Printf("async.All done: %+v\n", ret)
|
||||
}
|
||||
P.Return(ret)
|
||||
return
|
||||
default:
|
||||
panic("unreachable")
|
||||
}
|
||||
}
|
||||
return P
|
||||
}
|
||||
|
||||
// llgo:link Await2 llgo.await
|
||||
func Await2Compiled[OutT1, OutT2 any](
|
||||
ac1 AsyncCall[OutT1], ac2 AsyncCall[OutT2],
|
||||
timeout ...time.Duration) (ret *PromiseImpl[tuple.Tuple3[OutT1, OutT2, error]]) {
|
||||
p1 := ac1.(*PromiseImpl[OutT1])
|
||||
p2 := ac2.(*PromiseImpl[OutT2])
|
||||
remaining := 2
|
||||
P := &PromiseImpl[tuple.Tuple3[OutT1, OutT2, error]]{}
|
||||
P.Debug = "Await2"
|
||||
P.Func = func() {
|
||||
switch P.Next {
|
||||
case 0:
|
||||
P.Next = 1
|
||||
p1.Exec = P.Exec
|
||||
p1.Parent = P
|
||||
p1.Call()
|
||||
|
||||
p2.Exec = P.Exec
|
||||
p2.Parent = P
|
||||
p2.Call()
|
||||
return
|
||||
case 1:
|
||||
remaining--
|
||||
if remaining > 0 {
|
||||
return
|
||||
}
|
||||
P.Next = -1
|
||||
if !p1.Done() || !p2.Done() {
|
||||
log.Fatalf("async.Await2: not done: %+v, %+v\n", p1, p2)
|
||||
}
|
||||
|
||||
P.Return(tuple.Tuple3[OutT1, OutT2, error]{
|
||||
V1: p1.value,
|
||||
V2: p2.value,
|
||||
V3: nil,
|
||||
})
|
||||
return
|
||||
default:
|
||||
panic("unreachable")
|
||||
}
|
||||
}
|
||||
return P
|
||||
}
|
||||
|
||||
// llgo:link Await2 llgo.await
|
||||
func Await3Compiled[OutT1, OutT2, OutT3 any](
|
||||
ac1 AsyncCall[OutT1], ac2 AsyncCall[OutT2], ac3 AsyncCall[OutT3],
|
||||
timeout ...time.Duration) *PromiseImpl[tuple.Tuple3[OutT1, OutT2, OutT3]] {
|
||||
p1 := ac1.(*PromiseImpl[OutT1])
|
||||
p2 := ac2.(*PromiseImpl[OutT2])
|
||||
p3 := ac3.(*PromiseImpl[OutT3])
|
||||
remaining := 3
|
||||
P := &PromiseImpl[tuple.Tuple3[OutT1, OutT2, OutT3]]{}
|
||||
P.Debug = "Await3"
|
||||
P.Func = func() {
|
||||
switch P.Next {
|
||||
case 0:
|
||||
P.Next = 1
|
||||
p1.Exec = P.Exec
|
||||
p1.Parent = P
|
||||
p1.Call()
|
||||
|
||||
p2.Exec = P.Exec
|
||||
p2.Parent = P
|
||||
p2.Call()
|
||||
|
||||
p3.Exec = P.Exec
|
||||
p3.Parent = P
|
||||
p3.Call()
|
||||
return
|
||||
case 1:
|
||||
remaining--
|
||||
if remaining > 0 {
|
||||
return
|
||||
}
|
||||
P.Next = -1
|
||||
// TODO(lijie): return every error?
|
||||
if !p1.Done() || !p2.Done() || !p3.Done() {
|
||||
log.Fatalf("async.Await3: not done: %+v, %+v, %+v\n", p1, p2, p3)
|
||||
}
|
||||
|
||||
P.Return(tuple.Tuple3[OutT1, OutT2, OutT3]{
|
||||
V1: p1.value,
|
||||
V2: p2.value,
|
||||
V3: p3.value,
|
||||
})
|
||||
return
|
||||
default:
|
||||
panic("unreachable")
|
||||
}
|
||||
}
|
||||
return P
|
||||
}
|
||||
|
||||
func PAllCompiled[OutT any](acs ...AsyncCall[OutT]) *PromiseImpl[[]OutT] {
|
||||
P := &PromiseImpl[[]OutT]{}
|
||||
P.Debug = "Parallel"
|
||||
P.Func = func() {
|
||||
ret := make([]OutT, len(acs))
|
||||
wg := sync.WaitGroup{}
|
||||
for idx, ac := range acs {
|
||||
idx := idx
|
||||
ac := ac
|
||||
wg.Add(1)
|
||||
go func(ac AsyncCall[OutT]) {
|
||||
v := RunImpl[OutT](ac)
|
||||
ret[idx] = v
|
||||
wg.Done()
|
||||
}(ac)
|
||||
}
|
||||
wg.Wait()
|
||||
P.Return(ret)
|
||||
}
|
||||
return P
|
||||
}
|
||||
|
||||
func PAwait3Compiled[OutT1, OutT2, OutT3 any](
|
||||
ac1 AsyncCall[OutT1], ac2 AsyncCall[OutT2], ac3 AsyncCall[OutT3]) *PromiseImpl[tuple.Tuple4[OutT1, OutT2, OutT3, error]] {
|
||||
P := &PromiseImpl[tuple.Tuple4[OutT1, OutT2, OutT3, error]]{}
|
||||
P.Debug = "PAwait3"
|
||||
P.Func = func() {
|
||||
ret := tuple.Tuple4[OutT1, OutT2, OutT3, error]{}
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(3)
|
||||
go func() {
|
||||
ret.V1 = RunImpl[OutT1](ac1)
|
||||
wg.Done()
|
||||
}()
|
||||
go func() {
|
||||
ret.V2 = RunImpl[OutT2](ac2)
|
||||
wg.Done()
|
||||
}()
|
||||
go func() {
|
||||
ret.V3 = RunImpl[OutT3](ac3)
|
||||
wg.Done()
|
||||
}()
|
||||
wg.Wait()
|
||||
P.Return(ret)
|
||||
}
|
||||
return P
|
||||
}
|
||||
@@ -1,154 +0,0 @@
|
||||
/*
|
||||
* Copyright (c) 2024 The GoPlus Authors (goplus.org). All rights reserved.
|
||||
*
|
||||
* Licensed under the Apache License, Version 2.0 (the "License");
|
||||
* you may not use this file except in compliance with the License.
|
||||
* You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing, software
|
||||
* distributed under the License is distributed on an "AS IS" BASIS,
|
||||
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||
* See the License for the specific language governing permissions and
|
||||
* limitations under the License.
|
||||
*/
|
||||
|
||||
package naive
|
||||
|
||||
import (
|
||||
"log"
|
||||
"sync"
|
||||
)
|
||||
|
||||
const debugAsync = false
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
type asyncCall interface {
|
||||
parent() asyncCall
|
||||
Resume()
|
||||
Call()
|
||||
Done() bool
|
||||
}
|
||||
|
||||
type AsyncCall[OutT any] interface {
|
||||
Resume()
|
||||
}
|
||||
|
||||
type executor struct {
|
||||
acs []asyncCall
|
||||
mu sync.Mutex
|
||||
cond *sync.Cond
|
||||
}
|
||||
|
||||
func newExecutor() *executor {
|
||||
e := &executor{}
|
||||
e.cond = sync.NewCond(&e.mu)
|
||||
return e
|
||||
}
|
||||
|
||||
func (e *executor) schedule(ac asyncCall) {
|
||||
e.mu.Lock()
|
||||
e.acs = append(e.acs, ac)
|
||||
e.mu.Unlock()
|
||||
e.cond.Signal()
|
||||
}
|
||||
|
||||
func RunImpl[OutT any](ac AsyncCall[OutT]) OutT {
|
||||
e := newExecutor()
|
||||
p := ac.(*PromiseImpl[OutT])
|
||||
p.Exec = e
|
||||
var rootAc asyncCall = p
|
||||
e.schedule(rootAc)
|
||||
|
||||
for {
|
||||
e.mu.Lock()
|
||||
for len(e.acs) == 0 {
|
||||
e.cond.Wait()
|
||||
}
|
||||
e.mu.Unlock()
|
||||
ac := e.acs[0]
|
||||
e.acs = e.acs[1:]
|
||||
ac.Call()
|
||||
if ac.Done() && ac == rootAc {
|
||||
return p.value
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------------
|
||||
|
||||
type PromiseImpl[TOut any] struct {
|
||||
Debug string
|
||||
Next int
|
||||
Exec *executor
|
||||
Parent asyncCall
|
||||
|
||||
Func func()
|
||||
value TOut
|
||||
c chan TOut
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) parent() asyncCall {
|
||||
return p.Parent
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) Resume() {
|
||||
if debugAsync {
|
||||
log.Printf("Resume task: %+v\n", p)
|
||||
}
|
||||
p.Exec.schedule(p)
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) Done() bool {
|
||||
return p.Next == -1
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) Call() {
|
||||
p.Func()
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) Suspend() {
|
||||
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) Return(v TOut) {
|
||||
// TODO(lijie): panic if already resolved
|
||||
p.value = v
|
||||
if p.c != nil {
|
||||
p.c <- v
|
||||
}
|
||||
if debugAsync {
|
||||
log.Printf("Return task: %+v\n", p)
|
||||
}
|
||||
if p.Parent != nil {
|
||||
p.Parent.Resume()
|
||||
}
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) Yield(v TOut) {
|
||||
p.value = v
|
||||
if debugAsync {
|
||||
log.Printf("Yield task: %+v\n", p)
|
||||
}
|
||||
if p.Parent != nil {
|
||||
p.Parent.Resume()
|
||||
}
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) Value() TOut {
|
||||
return p.value
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) Chan() <-chan TOut {
|
||||
if p.c == nil {
|
||||
p.c = make(chan TOut, 1)
|
||||
p.Func()
|
||||
}
|
||||
return p.c
|
||||
}
|
||||
|
||||
func (p *PromiseImpl[TOut]) Await() (ret TOut) {
|
||||
panic("should not called")
|
||||
}
|
||||
Reference in New Issue
Block a user