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

add Future.Then

Browse Source
This commit is contained in:
Li Jie
2024-09-08 20:27:05 +08:00
parent cf53f3a347
commit 566d5ef96f
9 changed files with 56 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

393
x/async/README.md Normal file
View File

@@ -0,0 +1,393 @@
# Async I/O Design
## Async functions in different languages
### JavaScript
- [Async/Await](https://developer.mozilla.org/en-US/docs/Web/JavaScript/Reference/Statements/async_function)
Prototype:
```javascript
async function name(param0) {
statements;
}
async function name(param0, param1) {
statements;
}
async function name(param0, param1, /* …, */ paramN) {
statements;
}
```
Example:
```typescript
async function resolveAfter1Second(): Promise<string> {
return new Promise((resolve) => {
setTimeout(() => {
resolve("Resolved after 1 second");
}, 1000);
});
}
async function asyncCall(): Promise<string> {
const result = await resolveAfter1Second();
return `AsyncCall: ${result}`;
}
function asyncCall2(): Promise<string> {
return resolveAfter1Second();
}
function asyncCall3(): void {
resolveAfter1Second().then((result) => {
console.log(`AsyncCall3: ${result}`);
});
}
async function main() {
console.log("Starting AsyncCall");
const result1 = await asyncCall();
console.log(result1);
console.log("Starting AsyncCall2");
const result2 = await asyncCall2();
console.log(result2);
console.log("Starting AsyncCall3");
asyncCall3();
// Wait for AsyncCall3 to complete
await new Promise((resolve) => setTimeout(resolve, 1000));
console.log("Main function completed");
}
main().catch(console.error);
```
### Python
- [async def](https://docs.python.org/3/library/asyncio-task.html#coroutines)
Prototype:
```python
async def name(param0):
statements
```
Example:
```python
import asyncio
async def resolve_after_1_second() -> str:
await asyncio.sleep(1)
return "Resolved after 1 second"
async def async_call() -> str:
result = await resolve_after_1_second()
return f"AsyncCall: {result}"
def async_call2() -> asyncio.Task:
return resolve_after_1_second()
def async_call3() -> None:
asyncio.create_task(print_after_1_second())
async def print_after_1_second() -> None:
result = await resolve_after_1_second()
print(f"AsyncCall3: {result}")
async def main():
print("Starting AsyncCall")
result1 = await async_call()
print(result1)
print("Starting AsyncCall2")
result2 = await async_call2()
print(result2)
print("Starting AsyncCall3")
async_call3()
# Wait for AsyncCall3 to complete
await asyncio.sleep(1)
print("Main function completed")
# Run the main coroutine
asyncio.run(main())
```
### Rust
- [async fn](https://doc.rust-lang.org/std/keyword.async.html)
Prototype:
```rust
async fn name(param0: Type) -> ReturnType {
statements
}
```
Example:
```rust
use std::time::Duration;
use tokio::time::sleep;
use std::future::Future;
async fn resolve_after_1_second() -> String {
sleep(Duration::from_secs(1)).await;
"Resolved after 1 second".to_string()
}
async fn async_call() -> String {
let result = resolve_after_1_second().await;
format!("AsyncCall: {}", result)
}
fn async_call2() -> impl Future<Output = String> {
resolve_after_1_second()
}
fn async_call3() {
tokio::spawn(async {
let result = resolve_after_1_second().await;
println!("AsyncCall3: {}", result);
});
}
#[tokio::main]
async fn main() {
println!("Starting AsyncCall");
let result1 = async_call().await;
println!("{}", result1);
println!("Starting AsyncCall2");
let result2 = async_call2().await;
println!("{}", result2);
println!("Starting AsyncCall3");
async_call3();
// Wait for AsyncCall3 to complete
sleep(Duration::from_secs(2)).await;
println!("Main function completed");
}
```
### C#
- [async](https://docs.microsoft.com/en-us/dotnet/csharp/programming-guide/concepts/async/)
Prototype:
```csharp
async Task<ReturnType> NameAsync(Type param0)
{
statements;
}
```
Example:
```csharp
using System;
using System.Threading.Tasks;
class Program
{
static async Task<string> ResolveAfter1Second()
{
await Task.Delay(1000);
return "Resolved after 1 second";
}
static async Task<string> AsyncCall()
{
string result = await ResolveAfter1Second();
return $"AsyncCall: {result}";
}
static Task<string> AsyncCall2()
{
return ResolveAfter1Second();
}
static void AsyncCall3()
{
_ = Task.Run(async () =>
{
string result = await ResolveAfter1Second();
Console.WriteLine($"AsyncCall3: {result}");
});
}
static async Task Main()
{
Console.WriteLine("Starting AsyncCall");
string result1 = await AsyncCall();
Console.WriteLine(result1);
Console.WriteLine("Starting AsyncCall2");
string result2 = await AsyncCall2();
Console.WriteLine(result2);
Console.WriteLine("Starting AsyncCall3");
AsyncCall3();
// Wait for AsyncCall3 to complete
await Task.Delay(1000);
Console.WriteLine("Main method completed");
}
}
```
### C++ 20 Coroutines
- [co_await](https://en.cppreference.com/w/cpp/language/coroutines)
Prototype:
```cpp
TaskReturnType NameAsync(Type param0)
{
co_return co_await expression;
}
```
Example:
```cpp
#include <cppcoro/task.hpp>
#include <cppcoro/sync_wait.hpp>
#include <cppcoro/when_all.hpp>
#include <chrono>
#include <iostream>
#include <thread>
cppcoro::task<std::string> resolveAfter1Second() {
co_await std::chrono::seconds(1);
co_return "Resolved after 1 second";
}
cppcoro::task<std::string> asyncCall() {
auto result = co_await resolveAfter1Second();
co_return "AsyncCall: " + result;
}
cppcoro::task<std::string> asyncCall2() {
return resolveAfter1Second();
}
cppcoro::task<void> asyncCall3() {
auto result = co_await resolveAfter1Second();
std::cout << "AsyncCall3: " << result << std::endl;
}
cppcoro::task<void> main() {
std::cout << "Starting AsyncCall" << std::endl;
auto result1 = co_await asyncCall();
std::cout << result1 << std::endl;
std::cout << "Starting AsyncCall2" << std::endl;
auto result2 = co_await asyncCall2();
std::cout << result2 << std::endl;
std::cout << "Starting AsyncCall3" << std::endl;
auto asyncCall3Task = asyncCall3();
// Wait for AsyncCall3 to complete
co_await asyncCall3Task;
std::cout << "Main function completed" << std::endl;
}
int main() {
try {
cppcoro::sync_wait(::main());
} catch (const std::exception& e) {
std::cerr << "Error: " << e.what() << std::endl;
return 1;
}
return 0;
}
```
## Common concepts
### Promise, Future, Task, and Coroutine
- **Promise**: An object that represents the eventual completion (or failure) of an asynchronous operation and its resulting value. It is used to produce a value that will be consumed by a `Future`.
- **Future**: An object that represents the result of an asynchronous operation. It is used to obtain the value produced by a `Promise`.
- **Task**: A unit of work that can be scheduled and executed asynchronously. It is a higher-level abstraction that combines a `Promise` and a `Future`.
- **Coroutine**: A special type of function that can suspend its execution and return control to the caller without losing its state. It can be resumed later, allowing for asynchronous programming.
### `async`, `await` and similar keywords
- **`async`**: A keyword used to define a function that returns a `Promise` or `Task`. It allows the function to pause its execution and resume later.
- **`await`**: A keyword used to pause the execution of an `async` function until a `Promise` or `Task` is resolved. It unwraps the value of the `Promise` or `Task` and allows the function to continue.
- **`co_return`**: A keyword used in C++ coroutines to return a value from a coroutine. It is similar to `return` but is used in coroutines to indicate that the coroutine has completed. It's similar to `return` in `async` functions in other languages that boxes the value into a `Promise` or `Task`.
`async/await` and similar constructs provide a more readable and synchronous-like way of writing asynchronous code, it hides the type of `Promise`/`Future`/`Task` from the user and allows them to focus on the logic of the code.
### Executing Multiple Async Operations Concurrently
To run multiple promises concurrently, JavaScript provides `Promise.all`, `Promise.allSettled` and `Promise.any`, Python provides `asyncio.gather`, Rust provides `tokio::try_join`, C# provides `Task.WhenAll`, and C++ provides `cppcoro::when_all`.
In some situations, you may want to get the first result of multiple async operations. JavaScript provides `Promise.race` to get the first result of multiple promises. Python provides `asyncio.wait` to get the first result of multiple coroutines. Rust provides `tokio::select!` to get the first result of multiple futures. C# provides `Task.WhenAny` to get the first result of multiple tasks. C++ provides `cppcoro::when_any` to get the first result of multiple tasks. Those functions are very simular to `select` in Go.
### Error Handling
`await` commonly unwraps the value of a `Promise` or `Task`, but it also propagates errors. If the `Promise` or `Task` is rejected or throws an error, the error will be thrown in the `async` function by the `await` keyword. You can use `try/catch` blocks to handle errors in `async` functions.
## Common patterns
- `async` keyword hides the types of `Promise`/`Future`/`Task` in the function signature in Python and Rust, but not in JavaScript, C#, and C++.
- `await` keyword unwraps the value of a `Promise`/`Future`/`Task`.
- `return` keyword boxes the value into a `Promise`/`Future`/`Task` if it's not already.
## Design considerations in LLGo
- Don't introduce `async`/`await` keywords to compatible with Go compiler (just compiling)
- For performance reason don't implement async functions with goroutines
- Avoid implementing `Promise` by using `chan` to avoid blocking the thread, but it can be wrapped as a `chan` to make it compatible `select` statement
## Design
Introduce `async.IO[T]` type to represent an asynchronous operation, `async.Future[T]` type to represent the result of an asynchronous operation. `async.IO[T]` can be `bind` to a function that accepts `T` as an argument to chain multiple asynchronous operations. `async.IO[T]` can be `await` to get the value of the asynchronous operation.
```go
package async
type Future[T any] func(func(T))
func Await[T any](future Future[T]) T
func main() {
hello := func() Future[string] {
return func(resolve func(string)) {
resolve("Hello, World!")
}
}
future := hello()
future(func(value string) {
println(value)
})
println(Await(future))
}
```

16
x/async/TODO.md Normal file
View File

@@ -0,0 +1,16 @@
讨论：
1. Future 用 interface 还是闭包：性能应该差不多，如果没有其他方法要暴露，感觉也没有换成 interface 的必要。
2. 几个方法提供不同参数个数的版本还是用 tuple：如果编译器不支持可变泛型参数个数和特化，我倾向用 tuple 先简化实现，tuple 的开销应该也容易被编译器优化掉。多个方法让用户选择 Await2/Await3 这种也恶心。
3. 是否 Cancellable，暂时不加进去，多一个 context，也不一定能快速稳定下来，可以后面根据实践再改。
4. Executor 可能会变化，目前提供的 Run 是阻塞的，也可以把它做成异步。
5. 尽量再隐藏一些辅助类型，比如 TupleN，可能之提供 tuple 的构造和返回多值。内部的 libuv 如果隐藏可能要暴露同等接口，先不动了
6. 性能可能做个简单测试，但不是关键，只要别太差。未来可能会尽量减少 executor 的切换、尽量多并行
7. 异常兼容性：目前没考虑，这个要在回调里处理可能困难，要么就在 await 上处理，可以往后放一下，毕竟 golang 主要是以 error 为主
8. 可能先看一下如何在 go+里面集成，判断目前的设计实现是否合理
9. 多封装一些库看看通用性和易用性，\_demo 里几个简单例子基本符合预期，还需要更多检验
TODO：
10. select 兼容 （可能把 Future 改为 interface 更合理？）
11. Future 只会被执行一次

32
x/async/_demo/all/all.go
View File

@@ -32,7 +32,7 @@ func WriteFile(fileName string, content []byte) async.Future[error] {
func sleep(i int, d time.Duration) async.Future[int] {
return async.Async(func(resolve func(int)) {
timeout.Timeout(d)(func(async.Void) {
timeout.Timeout(d).Then(func(async.Void) {
resolve(i)
})
})
@@ -70,7 +70,7 @@ func RunIO() {
println("RunIO with BindIO")
async.Run(async.Async(func(resolve func(async.Void)) {
ReadFile("all.go")(func(v tuple.Tuple2[[]byte, error]) {
ReadFile("all.go").Then(func(v tuple.Tuple2[[]byte, error]) {
content, err := v.Get()
if err != nil {
fmt.Printf("read err: %v\n", err)
@@ -78,7 +78,7 @@ func RunIO() {
return
}
fmt.Printf("read content: %s\n", content)
WriteFile("2.out", content)(func(v error) {
WriteFile("2.out", content).Then(func(v error) {
err = v
if err != nil {
fmt.Printf("write err: %v\n", err)
@@ -100,7 +100,7 @@ func RunAllAndRace() {
println("Run All with Await")
async.Run(async.Async(func(resolve func(async.Void)) {
async.All(sleep(1, ms200), sleep(2, ms100), sleep(3, ms300))(func(v []int) {
async.All(sleep(1, ms200), sleep(2, ms100), sleep(3, ms300)).Then(func(v []int) {
fmt.Printf("All: %v\n", v)
resolve(async.Void{})
})
@@ -120,7 +120,7 @@ func RunAllAndRace() {
println("Run All with BindIO")
async.Run(async.Async(func(resolve func(async.Void)) {
async.All(sleep(1, ms200), sleep(2, ms100), sleep(3, ms300))(func(v []int) {
async.All(sleep(1, ms200), sleep(2, ms100), sleep(3, ms300)).Then(func(v []int) {
fmt.Printf("All: %v\n", v)
resolve(async.Void{})
})
@@ -129,7 +129,7 @@ func RunAllAndRace() {
println("Run Race with BindIO")
async.Run(async.Async(func(resolve func(async.Void)) {
async.Race(sleep(1, ms200), sleep(2, ms100), sleep(3, ms300))(func(v int) {
async.Race(sleep(1, ms200), sleep(2, ms100), sleep(3, ms300)).Then(func(v int) {
fmt.Printf("Race: %v\n", v)
resolve(async.Void{})
})
@@ -152,7 +152,7 @@ func RunTimeout() {
async.Run(async.Async(func(resolve func(async.Void)) {
fmt.Printf("Start 100 ms timeout\n")
timeout.Timeout(100 * time.Millisecond)(func(async.Void) {
timeout.Timeout(100 * time.Millisecond).Then(func(async.Void) {
fmt.Printf("timeout\n")
resolve(async.Void{})
})
@@ -165,15 +165,15 @@ func RunSocket() {
async.Run(async.Async(func(resolve func(async.Void)) {
println("RunServer")
RunServer()(func(async.Void) {
RunServer().Then(func(async.Void) {
println("RunServer done")
resolve(async.Void{})
})
println("RunClient")
timeout.Timeout(100 * time.Millisecond)(func(async.Void) {
RunClient()(func(async.Void) {
timeout.Timeout(100 * time.Millisecond).Then(func(async.Void) {
RunClient().Then(func(async.Void) {
println("RunClient done")
resolve(async.Void{})
})
@@ -184,7 +184,7 @@ func RunSocket() {
func RunClient() async.Future[async.Void] {
return async.Async(func(resolve func(async.Void)) {
addr := "127.0.0.1:3927"
socketio.Connect("tcp", addr)(func(v tuple.Tuple2[*socketio.Conn, error]) {
socketio.Connect("tcp", addr).Then(func(v tuple.Tuple2[*socketio.Conn, error]) {
client, err := v.Get()
println("Connected", client, err)
if err != nil {
@@ -195,17 +195,17 @@ func RunClient() async.Future[async.Void] {
loop = func(client *socketio.Conn) {
counter++
data := fmt.Sprintf("Hello %d", counter)
client.Write([]byte(data))(func(err error) {
client.Write([]byte(data)).Then(func(err error) {
if err != nil {
panic(err)
}
client.Read()(func(v tuple.Tuple2[[]byte, error]) {
client.Read().Then(func(v tuple.Tuple2[[]byte, error]) {
data, err := v.Get()
if err != nil {
panic(err)
}
println("Read from server:", string(data))
timeout.Timeout(1 * time.Second)(func(async.Void) {
timeout.Timeout(1 * time.Second).Then(func(async.Void) {
loop(client)
})
})
@@ -222,13 +222,13 @@ func RunServer() async.Future[async.Void] {
println("Client connected", client, err)
var loop func(client *socketio.Conn)
loop = func(client *socketio.Conn) {
client.Read()(func(v tuple.Tuple2[[]byte, error]) {
client.Read().Then(func(v tuple.Tuple2[[]byte, error]) {
data, err := v.Get()
if err != nil {
println("Read error", err)
} else {
println("Read from client:", string(data))
client.Write(data)(func(err error) {
client.Write(data).Then(func(err error) {
if err != nil {
println("Write error", err)
} else {

4
x/async/async.go
View File

@@ -24,6 +24,10 @@ type Void = [0]byte
type Future[T any] func(func(T))
func (f Future[T]) Then(cb func(T)) {
f(cb)
}
// Just for pure LLGo/Go, transpile to callback in Go+
func Await[T1 any](future Future[T1]) T1 {
return Run(future)

4
x/async/async_go.go
View File

@@ -34,7 +34,7 @@ func Race[T1 any](futures ...Future[T1]) Future[T1] {
ch := make(chan T1)
for _, future := range futures {
future := future
future(func(v T1) {
future.Then(func(v T1) {
defer func() {
// Avoid panic when the channel is closed.
_ = recover()
@@ -56,7 +56,7 @@ func All[T1 any](futures ...Future[T1]) Future[[]T1] {
wg.Add(n)
for i, future := range futures {
i := i
future(func(v T1) {
future.Then(func(v T1) {
results[i] = v
wg.Done()
})

4
x/async/async_llgo.go
View File

@@ -53,7 +53,7 @@ func Race[T1 any](futures ...Future[T1]) Future[T1] {
return Async(func(resolve func(T1)) {
done := atomic.Bool{}
for _, future := range futures {
future(func(v T1) {
future.Then(func(v T1) {
if !done.Swap(true) {
// Just resolve the first one.
resolve(v)
@@ -70,7 +70,7 @@ func All[T1 any](futures ...Future[T1]) Future[[]T1] {
var done uint32
for i, future := range futures {
i := i
future(func(v T1) {
future.Then(func(v T1) {
results[i] = v
if atomic.AddUint32(&done, 1) == uint32(n) {
// All done.

2
x/async/executor_go.go
View File

@@ -22,7 +22,7 @@ package async
func Run[T any](future Future[T]) T {
ch := make(chan T)
go func() {
future(func(v T) {
future.Then(func(v T) {
ch <- v
})
}()

2
x/async/executor_llgo.go
View File

@@ -59,7 +59,7 @@ func Run[T any](future Future[T]) T {
exec := &Executor{loop}
oldExec := setExec(exec)
var ret T
future(func(v T) {
future.Then(func(v T) {
ret = v
})
exec.Run()