Retracted
README
¶
Documentation
¶
Overview ¶
Package async is a library for asynchronous programming.
Since Go has already done a great job in bringing green/virtual threads into life, this library only implements a single-threaded Executor type, which some refer to as an async runtime. One can create as many Executors as they like.
While Go excels at forking, async, on the other hand, excels at joining.
Use Case #1: Fan-in Executing Code From Various Goroutines ¶
Wanted to execute pieces of code from various goroutines in a single-threaded way?
An Executor is designed to be able to run Tasks spawned in various goroutines sequentially. This comes in handy when one wants to do a series of operations on a single thread, for example, to read or update states that are not safe for concurrent access, to write data to the console, to update one's user interfaces, etc.
No backpressure alert. Task spawning is designed not to block. If spawning outruns execution, an Executor could easily consume a lot of memory over time. To mitigate, one could introduce a semaphore per hot spot.
Use Case #2: Event-driven Reactiveness ¶
A Task can be reactive.
A Task is created with an Operation function. In this user-provided function, one can return a specific Result to tell a Task to watch and await some Events (any of Signal, State and Memo), and the Task can just re-run whenever any of these Events notifies.
This is useful when one wants to do something repeatedly. It works like a loop. To exit this loop, just return a Result that ends the Task from within the Operation function. Simple.
Use Case #3: Easy State Machines ¶
A Task can also switch from one Operation to another, just like a state machine can transit from one state to another. This is done by returning another specific Result from within an Operation function. A Task can switch from one Operation to another until an Operation ends it.
Spawning Async Tasks vs. Passing Data over Go Channels ¶
It's not recommended to have channel operations in an async Operation for a Task to do, since they tend to block. For an Executor, if one Task blocks, no other Tasks can run. So instead of passing data around, one would just handle data in place.
One of the advantages of passing data over channels is to be able to reduce allocation. Unfortunately, async Operations always escape to heap. Any variable they captured also escapes to heap. One should always stay alert and take measures in hot spot, like repeatedly using a same Operation.
Example ¶
This example demonstrates how to create async Tasks with different paths. The lower path, the higher priority. This example creates a Task with path "aa" for additional computations and another Task with path "zz" for printing results. The former runs before the latter because "aa" < "zz".
package main
import (
"fmt"
"async"
)
func main() {
// Create an Executor.
var myExecutor async.Executor
// Set up an autorun function to run an Executor automatically whenever a Task is spawned or resumed.
// The best practice is to pass a function that does not block. See Example (NonBlocking).
myExecutor.Autorun(myExecutor.Run)
// Create two States.
s1, s2 := async.NewState(1), async.NewState(2)
// Although States can be created without the help of Executors,
// they might only be safe for use by one and only one Executor because of data races.
// Without proper synchronization, it's better only to spawn Tasks to read or update States.
var sum, product async.State[int]
myExecutor.Spawn("aa", func(t *async.Task) async.Result { // The path of t is "aa".
t.Watch(s1, s2) // Let t depend on s1 and s2, so t can re-run whenever s1 or s2 changes.
sum.Set(s1.Get() + s2.Get())
product.Set(s1.Get() * s2.Get())
return t.Await() // Awaits signals or state changes.
})
// The above Task re-runs whenever s1 or s2 changes. As an example, this is fine.
// In practice, one should probably use Memos to avoid unnecessary recomputations. See Example (Memo).
op := async.NewState('+')
myExecutor.Spawn("zz", func(t *async.Task) async.Result { // The path of t is "zz".
t.Watch(op)
fmt.Println("op =", "'"+string(op.Get())+"'")
switch op.Get() {
case '+':
// The path of an inner Task is relative to its outer one.
t.Spawn("sum", func(t *async.Task) async.Result { // The path of inner t is "zz/sum".
fmt.Println("s1 + s2 =", sum.Get())
return t.Await(&sum)
})
case '*':
t.Spawn("product", func(t *async.Task) async.Result { // The path of inner t is "zz/product".
fmt.Println("s1 * s2 =", product.Get())
return t.Await(&product)
})
}
return t.Await()
})
fmt.Println("--- SEPARATOR ---")
// The followings create several Tasks to mutate States.
// They share the same path, "/", which is lower than "aa" and "zz".
// Remember that, the lower path, the higher priority.
// Updating States should have higher priority, so that when there are multiple update Tasks,
// they can run together before any read Task.
// This reduces the number of reads that have to react on update.
myExecutor.Spawn("/", async.Do(func() {
s1.Set(3)
s2.Set(4)
}))
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
op.Set('*')
}))
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
s1.Set(5)
s2.Set(6)
}))
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
s1.Set(7)
s2.Set(8)
op.Set('+')
}))
}
Output: op = '+' s1 + s2 = 3 --- SEPARATOR --- s1 + s2 = 7 --- SEPARATOR --- op = '*' s1 * s2 = 12 --- SEPARATOR --- s1 * s2 = 30 --- SEPARATOR --- op = '+' s1 + s2 = 15
Example (Chain) ¶
This example demonstrates how to chain multiple Operations together to be worked on in sequence by a Task.
package main
import (
"fmt"
"async"
)
func main() {
var myExecutor async.Executor
myExecutor.Autorun(myExecutor.Run)
var myState async.State[int]
myExecutor.Spawn("zz", async.Chain(
func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
fmt.Println(v, "(first)")
if v < 3 {
return t.Await()
}
return t.Switch(func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
fmt.Println(v, "(switched)")
if v < 5 {
return t.Await()
}
return t.End()
})
},
func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
fmt.Println(v, "(second)")
if v < 7 {
return t.Await()
}
return t.End()
},
))
for i := 1; i <= 9; i++ {
myExecutor.Spawn("/", async.Do(func() { myState.Set(i) }))
}
fmt.Println(myState.Get()) // Prints 9.
}
Output: 0 (first) 1 (first) 2 (first) 3 (first) 3 (switched) 4 (switched) 5 (switched) 5 (second) 6 (second) 7 (second) 9
Example (Conditional) ¶
This example demonstrates how a Task can conditionally depend on a State.
package main
import (
"fmt"
"async"
)
func main() {
var myExecutor async.Executor
myExecutor.Autorun(myExecutor.Run)
s1, s2, s3 := async.NewState(1), async.NewState(2), async.NewState(7)
myExecutor.Spawn("aa", func(t *async.Task) async.Result {
t.Watch(s1, s2) // Always depends on s1 and s2.
v := s1.Get() + s2.Get()
if v%2 == 0 {
t.Watch(s3) // Conditionally depends on s3.
v *= s3.Get()
}
fmt.Println(v)
return t.Await()
})
myExecutor.Spawn("/", async.Do(func() { s3.Notify() })) // Nothing happens.
myExecutor.Spawn("/", async.Do(func() { s1.Set(s1.Get() + 1) }))
myExecutor.Spawn("/", async.Do(func() { s3.Notify() }))
myExecutor.Spawn("/", async.Do(func() { s2.Set(s2.Get() + 1) }))
myExecutor.Spawn("/", async.Do(func() { s3.Notify() })) // Nothing happens.
}
Output: 3 28 28 5
Example (ConditionalMemo) ¶
This example demonstrates how a Memo can conditionally depend on a State.
package main
import (
"fmt"
"async"
)
func main() {
var myExecutor async.Executor
myExecutor.Autorun(myExecutor.Run)
s1, s2, s3 := async.NewState(1), async.NewState(2), async.NewState(7)
m := async.NewMemo(&myExecutor, "aa", func(t *async.Task, s *async.State[int]) {
t.Watch(s1, s2) // Always depends on s1 and s2.
v := s1.Get() + s2.Get()
if v%2 == 0 {
t.Watch(s3) // Conditionally depends on s3.
v *= s3.Get()
}
s.Set(v)
})
myExecutor.Spawn("zz", func(t *async.Task) async.Result {
t.Watch(m)
fmt.Println(m.Get())
return t.Await()
})
myExecutor.Spawn("/", async.Do(func() { s3.Notify() })) // Nothing happens.
myExecutor.Spawn("/", async.Do(func() { s1.Set(s1.Get() + 1) }))
myExecutor.Spawn("/", async.Do(func() { s3.Notify() }))
myExecutor.Spawn("/", async.Do(func() { s2.Set(s2.Get() + 1) }))
myExecutor.Spawn("/", async.Do(func() { s3.Notify() })) // Nothing happens.
}
Output: 3 28 28 5
Example (Defer) ¶
This example demonstrates how to add a function call before a Task re-runs, or after a Task ends.
package main
import (
"fmt"
"async"
)
func main() {
var myExecutor async.Executor
myExecutor.Autorun(myExecutor.Run)
var myState async.State[int]
myExecutor.Spawn("zz", func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
t.Defer(func() { fmt.Println(v, myState.Get()) })
if v < 3 {
return t.Await()
}
return t.End()
})
for i := 1; i <= 5; i++ {
myExecutor.Spawn("/", async.Do(func() { myState.Set(i) }))
}
fmt.Println(myState.Get()) // Prints 5.
}
Output: 0 1 1 2 2 3 3 3 5
Example (End) ¶
This example demonstrates how to end a Task. It creates a Task that prints the value of a State whenever it changes. The Task only prints 0, 1, 2 and 3 because it is ended after 3.
package main
import (
"fmt"
"async"
)
func main() {
var myExecutor async.Executor
myExecutor.Autorun(myExecutor.Run)
var myState async.State[int]
myExecutor.Spawn("zz", func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
fmt.Println(v)
if v < 3 {
return t.Await()
}
return t.End()
})
for i := 1; i <= 5; i++ {
myExecutor.Spawn("/", async.Do(func() { myState.Set(i) }))
}
fmt.Println(myState.Get()) // Prints 5.
}
Output: 0 1 2 3 5
Example (Memo) ¶
This example demonstrates how to use Memos to memoize cheap computations. Memos are evaluated lazily. They take effect only when they are acquired.
package main
import (
"fmt"
"async"
)
func main() {
var myExecutor async.Executor
myExecutor.Autorun(myExecutor.Run)
s1, s2 := async.NewState(1), async.NewState(2)
sum := async.NewMemo(&myExecutor, "aa", func(t *async.Task, s *async.State[int]) {
t.Watch(s1, s2)
if v := s1.Get() + s2.Get(); v != s.Get() {
s.Set(v) // Update s only when its value changes to stop unnecessary propagation.
}
})
product := async.NewMemo(&myExecutor, "aa", func(t *async.Task, s *async.State[int]) {
t.Watch(s1, s2)
if v := s1.Get() * s2.Get(); v != s.Get() {
s.Set(v)
}
})
op := async.NewState('+')
myExecutor.Spawn("zz", func(t *async.Task) async.Result {
t.Watch(op)
fmt.Println("op =", "'"+string(op.Get())+"'")
switch op.Get() {
case '+':
t.Spawn("sum", func(t *async.Task) async.Result {
fmt.Println("s1 + s2 =", sum.Get())
return t.Await(sum)
})
case '*':
t.Spawn("product", func(t *async.Task) async.Result {
fmt.Println("s1 * s2 =", product.Get())
return t.Await(product)
})
}
return t.Await()
})
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
s1.Set(3)
s2.Set(4)
}))
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
op.Set('*')
}))
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
s1.Set(5)
s2.Set(6)
}))
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
s1.Set(7)
s2.Set(8)
op.Set('+')
}))
}
Output: op = '+' s1 + s2 = 3 --- SEPARATOR --- s1 + s2 = 7 --- SEPARATOR --- op = '*' s1 * s2 = 12 --- SEPARATOR --- s1 * s2 = 30 --- SEPARATOR --- op = '+' s1 + s2 = 15
Example (NonBlocking) ¶
This example demonstrates how to set up an autorun function to run an Executor in a goroutine automatically whenever a Task is spawned or resumed.
package main
import (
"fmt"
"sync"
"async"
)
func main() {
var wg sync.WaitGroup // For keeping track of goroutines.
var myExecutor async.Executor
myExecutor.Autorun(func() {
wg.Add(1)
go func() {
defer wg.Done()
myExecutor.Run()
}()
})
s1, s2 := async.NewState(1), async.NewState(2)
sum := async.NewMemo(&myExecutor, "aa", func(t *async.Task, s *async.State[int]) {
t.Watch(s1, s2)
if v := s1.Get() + s2.Get(); v != s.Get() {
s.Set(v)
}
})
product := async.NewMemo(&myExecutor, "aa", func(t *async.Task, s *async.State[int]) {
t.Watch(s1, s2)
if v := s1.Get() * s2.Get(); v != s.Get() {
s.Set(v)
}
})
op := async.NewState('+')
myExecutor.Spawn("zz", func(t *async.Task) async.Result {
t.Watch(op)
fmt.Println("op =", "'"+string(op.Get())+"'")
switch op.Get() {
case '+':
t.Spawn("sum", func(t *async.Task) async.Result {
fmt.Println("s1 + s2 =", sum.Get())
return t.Await(sum)
})
case '*':
t.Spawn("product", func(t *async.Task) async.Result {
fmt.Println("s1 * s2 =", product.Get())
return t.Await(product)
})
}
return t.Await()
})
wg.Wait() // Wait for autorun to complete.
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
s1.Set(3)
s2.Set(4)
}))
wg.Wait()
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
op.Set('*')
}))
wg.Wait()
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
s1.Set(5)
s2.Set(6)
}))
wg.Wait()
fmt.Println("--- SEPARATOR ---")
myExecutor.Spawn("/", async.Do(func() {
s1.Set(7)
s2.Set(8)
op.Set('+')
}))
wg.Wait()
}
Output: op = '+' s1 + s2 = 3 --- SEPARATOR --- s1 + s2 = 7 --- SEPARATOR --- op = '*' s1 * s2 = 12 --- SEPARATOR --- s1 * s2 = 30 --- SEPARATOR --- op = '+' s1 + s2 = 15
Example (Switch) ¶
This example demonstrates how a Task can switch from one Operation to another.
package main
import (
"fmt"
"async"
)
func main() {
var myExecutor async.Executor
myExecutor.Autorun(myExecutor.Run)
var myState async.State[int]
myExecutor.Spawn("zz", func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
fmt.Println(v)
if v < 3 {
return t.Await()
}
return t.Switch(func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
fmt.Println(v, "(switched)")
if v < 5 {
return t.Await()
}
return t.End()
})
})
for i := 1; i <= 7; i++ {
myExecutor.Spawn("/", async.Do(func() { myState.Set(i) }))
}
fmt.Println(myState.Get()) // Prints 7.
}
Output: 0 1 2 3 3 (switched) 4 (switched) 5 (switched) 7
Example (Yield) ¶
This example demonstrates how to yield a Task only for it to resume later with another Operation. It computes two values in separate goroutines sequentially, then prints their sum. It showcases what yielding can do, not that it's a useful pattern.
package main
import (
"fmt"
"sync"
"time"
"async"
)
func main() {
var wg sync.WaitGroup // For keeping track of goroutines.
var myExecutor async.Executor
myExecutor.Autorun(func() {
wg.Add(1)
go func() {
defer wg.Done()
myExecutor.Run()
}()
})
var myState struct {
async.Signal
v1, v2 int
}
myExecutor.Spawn("/", func(t *async.Task) async.Result {
wg.Add(1)
go func() {
defer wg.Done()
time.Sleep(500 * time.Millisecond) // Heavy work #1 here.
ans := 15
myExecutor.Spawn("/", async.Do(func() {
myState.v1 = ans
myState.Notify()
}))
}()
t.Watch(&myState)
// Yield preserves Events that are being watched.
return t.Yield(func(t *async.Task) async.Result {
wg.Add(1)
go func() {
defer wg.Done()
time.Sleep(500 * time.Millisecond) // Heavy work #2 here.
ans := 27
myExecutor.Spawn("/", async.Do(func() {
myState.v2 = ans
myState.Notify()
}))
}()
t.Watch(&myState)
return t.Yield(async.Do(func() {
fmt.Println("v1 + v2 =", myState.v1+myState.v2)
}))
})
})
wg.Wait()
}
Output: v1 + v2 = 42
Index ¶
- type Event
- type Executor
- type Memo
- type Operation
- type Result
- type Signal
- type State
- type Task
- func (t *Task) Await(s ...Event) Result
- func (t *Task) Defer(f func())
- func (t *Task) End() Result
- func (t *Task) Executor() *Executor
- func (t *Task) Path() string
- func (t *Task) Spawn(p string, op Operation)
- func (t *Task) Switch(op Operation) Result
- func (t *Task) Watch(s ...Event)
- func (t *Task) Yield(op Operation) Result
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Event ¶
type Event interface {
// contains filtered or unexported methods
}
Event is the interface of any type that can be watched by a Task.
The following types implement Event: Signal, State and Memo.
type Executor ¶
type Executor struct {
// contains filtered or unexported fields
}
An Executor is a Task spawner, and a Task runner.
When a Task is spawned or resumed, it is added into an internal queue. The Run method then pops and runs each of them from the queue until the queue is emptied. It is done in a single-threaded manner. If one Task blocks, no other Tasks can run. The best practice is not to block.
The internal queue is a priority queue. Tasks added in the queue are sorted by their paths. Tasks with the same path are sorted by their arrival order (FIFO). Popping the queue removes the first Task with the least path.
Manually calling the Run method is usually not desired. One would instead use the Autorun method to set up an autorun function to calling the Run method automatically whenever a Task is spawned or resumed. The Executor never calls the autorun function twice at the same time.
func (*Executor) Autorun ¶
func (e *Executor) Autorun(f func())
Autorun sets up an autorun function to calling the Run method automatically whenever a Task is spawned or resumed.
One must pass a function that calls the Run method.
If f blocks, the Spawn method may block too. The best practice is not to block.
func (*Executor) Run ¶
func (e *Executor) Run()
Run pops and runs every Task in the queue until the queue is emptied.
Run must not be called twice at the same time.
type Memo ¶
type Memo[T any] struct { // contains filtered or unexported fields }
A Memo is a State-like structure that carries a value that can only be set in an Operation-like function.
A Memo is designed to have a value that is computed from other States. What make a Memo useful are that:
- A Memo can prevent unnecessary computations when it isn't used;
- A Memo can prevent unnecessary propagations when its value isn't changed.
To create a Memo, use NewMemo or NewStrictMemo.
A Memo must not be shared by more than one Executor.
func NewMemo ¶
NewMemo returns a new non-strict Memo. The arguments are used to initialize an internal Task.
One must pass a function that watches some States, computes a value from these States, and then updates the provided State if the value differs.
Like any Event, a Memo can be watched by multiple Tasks. The watch list increases and decreases over time. For a non-strict Memo, when the last Task in the list unwatches it, it does not immediately end its internal Task. Ending the internal Task would only put the Memo into a stale state because the Memo no longer detects dependency changes. By not immediately ending the internal Task, a non-strict Memo prevents an extra computation when a new Task watches it, provided that there are no dependency changes.
On the other hand, a strict Memo immediately ends its internal Task whenever the last Task in the watch list unwatches it. The Memo becomes stale. The next time a new Task watches it, it has to make a fresh computation.
type Operation ¶
An Operation is a piece of work that a Task is given to do when it is spawned. The return value of an Operation, a Result, determines what next for a Task to do.
The argument t must not escape, because t can be recycled by an Executor when t ends.
func Chain ¶
Chain returns an Operation that will work on each of the provided Operations in sequence. When one Operation completes, Chain works on another.
func Never ¶
func Never() Operation
Never returns an Operation that never completes. Operations in a Chain after Never are never getting worked on.
func (Operation) Then ¶
Then returns an Operation that first works on op, then switches to work on next after op completes.
To chain multiple Operations, use Chain function.
Example ¶
package main
import (
"fmt"
"async"
)
func main() {
var myExecutor async.Executor
myExecutor.Autorun(myExecutor.Run)
var myState async.State[int]
a := func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
fmt.Println(v, "(a)")
if v < 3 {
return t.Await()
}
return t.Switch(func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
fmt.Println(v, "(switched)")
if v < 5 {
return t.Await()
}
return t.End()
})
}
b := func(t *async.Task) async.Result {
t.Watch(&myState)
v := myState.Get()
fmt.Println(v, "(b)")
if v < 7 {
return t.Await()
}
return t.End()
}
myExecutor.Spawn("zz", async.Operation(a).Then(b))
for i := 1; i <= 9; i++ {
myExecutor.Spawn("/", async.Do(func() { myState.Set(i) }))
}
fmt.Println(myState.Get()) // Prints 9.
}
Output: 0 (a) 1 (a) 2 (a) 3 (a) 3 (switched) 4 (switched) 5 (switched) 5 (b) 6 (b) 7 (b) 9
type Result ¶
type Result struct {
// contains filtered or unexported fields
}
Result is the type of the return value of an Operation function. A Result determines what next for a Task to do after calling an Operation function.
A Result can be created by calling one of the following method of Task:
- Task.End: for ending a Task;
- Task.Await: for yielding a Task with additional Events to watch;
- Task.Yield: for yielding a Task with another Operation to which will be switched later when resuming;
- Task.Switch: for switching to another Operation.
type Signal ¶
type Signal struct {
// contains filtered or unexported fields
}
Signal is a type that implements Event.
Calling the Notify method of a Signal, in an Operation function, resumes any Task that is watching the Signal.
A Signal must not be shared by more than one Executor.
type State ¶
A State is a Signal that carries a value. To retrieve the value, call the Get method.
Calling the Set method of a State, in an Operation function, updates the value and resumes any Task that is watching the State.
A State must not be shared by more than one Executor.
type Task ¶
type Task struct {
// contains filtered or unexported fields
}
A Task is an execution of code, similar to a goroutine but cooperative and stackless.
A Task is created with a function called Operation. A Task's job is to complete it. When an Executor spawns a Task, it runs the Task by calling the Operation function with the Task as the argument. The return value determines whether to end the Task or to yield it so that it could resume later.
In order for a Task to resume, the Task must watch at least one Event, which must be a Signal, a State or a Memo, when calling the Operation function. A notification of such an Event resumes the Task. When a Task is resumed, the Executor runs the Task again.
A Task can also switch to work on another Operation function according to the return value of the Operation function. A Task can switch from one Operation to another until an Operation ends it.
func (*Task) Await ¶
Await returns a Result that will cause t to yield. Await also accepts additional Events to be awaited for.
func (*Task) Defer ¶
func (t *Task) Defer(f func())
Defer adds a function call when t resumes or ends, or when t is switching to work on another Operation.
func (*Task) End ¶
End returns a Result that will cause t to end or switch to work on another Operation in a Chain.
func (*Task) Executor ¶
Executor returns the Executor that spawned t.
Since t can be recycled by an Executor, it is recommended to save the return value in a variable first.
func (*Task) Path ¶
Path returns the path of t.
Since t can be recycled by an Executor, it is recommended to save the return value in a variable first.
func (*Task) Spawn ¶
Spawn creates an inner Task to work on op, using the result of path.Join(t.Path(), p) as its path.
Inner Tasks are ended automatically when the outer one resumes or ends, or when the outer one is switching to work on another Operation.
func (*Task) Switch ¶
Switch returns a Result that will cause t to switch to work on op. t will be reset and op will be called immediately as the current Operation of t.
Source Files