admin/llgo
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

asyncio: redesign

Browse Source
This commit is contained in:
Li Jie
2024-07-23 17:02:40 +08:00
parent eb7a94bb55
commit 7230e19166
3 changed files with 102 additions and 88 deletions
Download Patch File Download Diff File Expand all files Collapse all files

68
x/io/README.md
View File

@@ -366,23 +366,77 @@ In some situations, you may want to get the first result of multiple async opera
## Design ## Design
Introduce `Promise` type to represent the eventual completion of an asynchronous operation and its resulting value. `Promise` can be resolved with a value or rejected with an error. `Promise` can be awaited to get the value or error.
`Promise` just a type indicating the result of an asynchronous operation, it injected by the LLGo compiler, and the user can't create a `Promise` directly.
```go ```go
func resolveAfter1Second() Promise[string] { // Some native async functions
return Async(func (context) { func timeoutAsync(d time.Duration, cb func()) {
context.ScheduleAfter(1 * time.Second, func() { go func() {
context.Resolve("Resolved after 1 second") time.Sleep(d)
}) cb()
}()
}
// Wrap callback-based async function into Promise
func resolveAfter1Second() (resolve Promise[string]) {
timeoutAsync(1 * time.Second, func() {
resolve("Resolved after 1 second", nil)
}) })
} }
func asyncCall() Promise[string] { // Compiled to:
return Async(resolveAfter1Second().Await()) func resolveAfter1Second() (resolve PromiseImpl[string]) {
promise := io.NewPromiseImpl[string](resolve func(value string, err error) {
resolve: func(value string, err error) {
for true {
switch (promise.prev = promise.next) {
case 0:
timeoutAsync(1 * time.Second, func() {
resolve("Resolved after 1 second", nil)
})
}
}
},
}
return promise
} }
func asyncCall() (resolve Promise[string]) {
str, err := resolveAfter1Second().Await()
resolve("AsyncCall: " + str, err)
}
// Compiled to:
func asyncCall() (resolve PromiseImpl[string]) {
promise := io.NewPromiseImpl[string](resolve func(value string, err error) {
for true {
switch (promise.prev = promise.next) {
case 0:
resolveAfter1Second()
return
case 1:
str, err := promise.value, promise.err
resolve("AsyncCall: " + str, err)
return
}
}
})
return promise
}
// Directly return Promise
func asyncCall2() Promise[string] { func asyncCall2() Promise[string] {
return resolveAfter1Second() return resolveAfter1Second()
} }
// Compiled to:
func asyncCall2() PromiseImpl[string] {
return resolveAfter1Second()
}
// Don't wait for Promise to complete
func asyncCall3() { func asyncCall3() {
resolveAfter1Second().Then(func(result string) { resolveAfter1Second().Then(func(result string) {
fmt.Println("AsyncCall3: " + result) fmt.Println("AsyncCall3: " + result)

94
x/io/_demo/asyncdemo/async.go
View File

@@ -18,84 +18,53 @@ func (r *Response) mock(body string) {
r.mockBody = body r.mockBody = body
} }
func (r *Response) Text() *io.Promise[string] { func (r *Response) Text() (resolve io.Promise[string]) {
return io.NewPromise[string](func(resolve func(string, error)) { resolve(r.mockBody, nil)
resolve(r.mockBody, nil) return
})
} }
func HttpGet(url string, callback func(resp *Response, err error)) { func HttpGet(url string, callback func(resp *Response, err error)) {
panic("todo: Get") panic("todo: Get")
} }
func AsyncHttpGet(url string) io.AsyncCall[*Response] { func AsyncHttpGet(url string) (resolve io.Promise[*Response]) {
return io.NewPromise[*Response](func(resolve func(*Response, error)) { HttpGet(url, resolve)
HttpGet(url, resolve) return
})
} }
type User struct { type User struct {
Name string Name string
} }
func GetUser(uid string) io.AsyncCall[User] { func GetUser(uid string) (resolve io.Promise[User]) {
return io.NewPromise[User](func(resolve func(User, error)) { resp, err := io.Await[*Response](AsyncHttpGet("http://example.com/user/" + uid))
resp, err := io.Await(AsyncHttpGet("http://example.com/user/" + uid)) if err != nil {
if err != nil { resolve(User{}, err)
resolve(User{}, err) return
return }
}
if resp.StatusCode != 200 { if resp.StatusCode != 200 {
resolve(User{}, fmt.Errorf("http status code: %d", resp.StatusCode)) resolve(User{}, fmt.Errorf("http status code: %d", resp.StatusCode))
return return
} }
resp.mock(`{"name":"Alice"}`) resp.mock(`{"name":"Alice"}`)
body, err := io.Await[string](resp.Text()) body, err := io.Await[string](resp.Text())
if err != nil { if err != nil {
resolve(User{}, err) resolve(User{}, err)
return return
} }
user := User{} user := User{}
if err := json.Unmarshal([]byte(body), &user); err != nil { if err := json.Unmarshal([]byte(body), &user); err != nil {
resolve(User{}, err) resolve(User{}, err)
return return
} }
resolve(user, nil) resolve(user, nil)
}) return
} }
// func GetUser1(uid string) (resolve io.AsyncCall[User]) {
// resp, err := io.Await(AsyncHttpGet("http://example.com/user/" + uid))
// if err != nil {
// resolve(User{}, err)
// return
// }
// if resp.StatusCode != 200 {
// resolve(User{}, fmt.Errorf("http status code: %d", resp.StatusCode))
// return
// }
// resp.mock(`{"name":"Alice"}`)
// body, err := io.Await[string](resp.Text())
// if err != nil {
// resolve(User{}, err)
// return
// }
// user := User{}
// if err := json.Unmarshal([]byte(body), &user); err != nil {
// resolve(User{}, err)
// return
// }
// resolve(user, nil)
// }
func GetScore() *io.Promise[float64] { func GetScore() *io.Promise[float64] {
panic("todo: GetScore") panic("todo: GetScore")
} }
@@ -108,9 +77,12 @@ func main() {
user, err := GetUser("123").Await() user, err := GetUser("123").Await()
fmt.Println(user, err) fmt.Println(user, err)
user, err = io.Race(GetUser("123"), GetUser("456"), GetUser("789")).Await() user, err = io.Race[User](GetUser("123"), GetUser("456"), GetUser("789")).Await()
fmt.Println(user, err) fmt.Println(user, err)
users, err := io.All[User]([]io.AsyncCall[User]{GetUser("123"), GetUser("456"), GetUser("789")}).Await()
fmt.Println(users, err)
user, score, _, err := io.Await3[User, float64, io.Void](GetUser("123"), GetScore(), DoUpdate("update sth.")) user, score, _, err := io.Await3[User, float64, io.Void](GetUser("123"), GetScore(), DoUpdate("update sth."))
fmt.Println(user, score, err) fmt.Println(user, score, err)

28
x/io/io.go
View File

@@ -48,6 +48,10 @@ func Race[OutT any](acs ...AsyncCall[OutT]) (ret AsyncCall[OutT]) {
return return
} }
func All[OutT any](acs []AsyncCall[OutT]) (ret AsyncCall[[]OutT]) {
return nil
}
// llgo:link Await2 llgo.await // llgo:link Await2 llgo.await
func Await2[OutT1, OutT2 any]( func Await2[OutT1, OutT2 any](
ac1 AsyncCall[OutT1], ac2 AsyncCall[OutT2], ac1 AsyncCall[OutT1], ac2 AsyncCall[OutT2],
@@ -64,31 +68,15 @@ func Await3[OutT1, OutT2, OutT3 any](
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------
type Promise[OutT any] struct { type Promise[OutT any] func(OutT, error)
}
func NewPromise[OutT any](fn func(resolve func(OutT, error))) (ret *Promise[OutT]) { // llgo:link Promise.Await llgo.await
ret = &Promise[OutT]{} func (p Promise[OutT]) Await(timeout ...time.Duration) (ret OutT, err error) {
return return
} }
func NewPromiseFromValue[OutT any](value OutT) (ret *Promise[OutT]) { func (p Promise[OutT]) Chan() <-chan OutT {
return NewPromise[OutT](func(resolve func(OutT, error)) {
resolve(value, nil)
})
}
func (p *Promise[OutT]) Await(timeout ...time.Duration) (ret OutT, err error) {
return
}
func (p *Promise[OutT]) Chan() <-chan OutT {
return nil return nil
} }
// llgo:link Async llgo.async
func Async[OutT any](fn any) (ret Promise[OutT]) {
return
}
// ----------------------------------------------------------------------------- // -----------------------------------------------------------------------------