async

README

async

萬物皆有靈，有靈以為生。

A Go library for asynchronous programming.

This is quite an unusual implementation that coroutines do not exchange data directly. They just yield on awaiting events and resume on event notifications. Communications between coroutines are done by sending event notifications to each other. A coroutine can watch multiple events before yielding and an event notification can resume multiple coroutines.

Continue reading, or visit Go Reference if you are not there.

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 Goroutines

Wanted to execute pieces of code from goroutines in a single-threaded way?

An Executor is designed to be able to run tasks spawned in 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 or a file, 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 spawned with a Coroutine to take care of it. In this user-provided function, one can return a specific Result to tell a coroutine to watch and await some events (e.g. Signal, State and Memo, etc.), and the coroutine can just re-run the task 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 coroutine from within the task function. Simple.

Use Case 3: Easy State Machines

A Coroutine can also make a transition from one Task to another, just like a state machine can make a transition from one state to another. This is done by returning another specific Result from within a task function. A coroutine can transition from one task to another until a task ends it.

With the ability to transition, async is able to provide more advanced control structures, like Block, Loop and Func, to ease the process of writing async code. The experience now feels similar to that of writing sync code.

Spawning Async Tasks vs. Passing Data over Go Channels

It's not recommended to have channel operations in an async Task for a Coroutine to do, since they tend to block. For an Executor, if one coroutine blocks, no other coroutines 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 tasks 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 task.

Example

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	// Create an executor.
	var myExecutor async.Executor

	// Set up an autorun function to run an executor automatically whenever a coroutine is spawned or resumed.
	// The best practice is to pass a function that does not block. See Example (NonBlocking).
	myExecutor.Autorun(myExecutor.Run)

	// Create some states.
	s1, s2 := async.NewState(1), async.NewState(2)
	op := async.NewState('+')

	// 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 coroutines to read or update states.

	// Create a coroutine to print the sum or the product of s1 and s2, depending on what op is.
	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		co.Watch(op) // Let co depend on op, so co can re-run whenever op changes.

		fmt.Println("op =", "'"+string(op.Get())+"'")

		switch op.Get() {
		case '+':
			// Using an inner coroutine to narrow down what has to react whenever a state changes might be a good idea.
			// The following creates an inner coroutine, it runs immediately and re-runs whenever s1 or s2 changes.
			co.Spawn(func(co *async.Coroutine) async.Result {
				fmt.Println("s1 + s2 =", s1.Get()+s2.Get())
				return co.Await(s1, s2) // Watches s1 and s2, and awaits.
			})
		case '*':
			co.Spawn(func(co *async.Coroutine) async.Result {
				fmt.Println("s1 * s2 =", s1.Get()*s2.Get())
				return co.Await(s1, s2)
			})
		}

		return co.Await() // Awaits anything that has been watched (in this case, op).
	})

	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 (CleanupFunc)

This example demonstrates how to add a function call before a task re-runs, or after a task ends.

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		co.Watch(&myState)

		v := myState.Get()
		co.CleanupFunc(func() { fmt.Println(v, myState.Get()) })

		if v < 3 {
			return co.Await()
		}

		return co.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 (Conditional)

This example demonstrates how a task can conditionally depend on a state.

package main

import (
	"fmt"

	"github.com/b97tsk/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(func(co *async.Coroutine) async.Result {
		co.Watch(s1, s2) // Always depends on s1 and s2.

		v := s1.Get() + s2.Get()
		if v%2 == 0 {
			co.Watch(s3) // Conditionally depends on s3.
			v *= s3.Get()
		}

		fmt.Println(v)
		return co.Await()
	})

	inc := func(i int) int { return i + 1 }

	myExecutor.Spawn(async.Do(func() { s3.Notify() })) // Nothing happens.
	myExecutor.Spawn(async.Do(func() { s1.Update(inc) }))
	myExecutor.Spawn(async.Do(func() { s3.Notify() }))
	myExecutor.Spawn(async.Do(func() { s2.Update(inc) }))
	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"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	s1, s2, s3 := async.NewState(1), async.NewState(2), async.NewState(7)

	const memoWeight = 999 // Memos usually run before other tasks.

	m := async.NewMemo(&myExecutor, memoWeight, func(co *async.Coroutine, s *async.State[int]) {
		co.Watch(s1, s2) // Always depends on s1 and s2.

		v := s1.Get() + s2.Get()
		if v%2 == 0 {
			co.Watch(s3) // Conditionally depends on s3.
			v *= s3.Get()
		}

		s.Set(v)
	})

	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		co.Watch(m)
		fmt.Println(m.Get())
		return co.Await()
	})

	inc := func(i int) int { return i + 1 }

	myExecutor.Spawn(async.Do(func() { s3.Notify() })) // Nothing happens.
	myExecutor.Spawn(async.Do(func() { s1.Update(inc) }))
	myExecutor.Spawn(async.Do(func() { s3.Notify() }))
	myExecutor.Spawn(async.Do(func() { s2.Update(inc) }))
	myExecutor.Spawn(async.Do(func() { s3.Notify() })) // Nothing happens.

}

Output:

3
28
28
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"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		co.Watch(&myState)

		v := myState.Get()
		fmt.Println(v)

		if v < 3 {
			return co.Await()
		}

		return co.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"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	s1, s2 := async.NewState(1), async.NewState(2)

	const memoWeight = 999 // Memos usually run before other tasks.

	sum := async.NewMemo(&myExecutor, memoWeight, func(co *async.Coroutine, s *async.State[int]) {
		co.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, memoWeight, func(co *async.Coroutine, s *async.State[int]) {
		co.Watch(s1, s2)
		if v := s1.Get() * s2.Get(); v != s.Get() {
			s.Set(v)
		}
	})

	op := async.NewState('+')

	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		co.Watch(op)

		fmt.Println("op =", "'"+string(op.Get())+"'")

		switch op.Get() {
		case '+':
			co.Spawn(func(co *async.Coroutine) async.Result {
				fmt.Println("s1 + s2 =", sum.Get())
				return co.Await(sum)
			})
		case '*':
			co.Spawn(func(co *async.Coroutine) async.Result {
				fmt.Println("s1 * s2 =", product.Get())
				return co.Await(product)
			})
		}

		return co.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 coroutine is spawned or resumed.

package main

import (
	"fmt"
	"sync"

	"github.com/b97tsk/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)

	const memoWeight = 999 // Memos usually run before other tasks.

	sum := async.NewMemo(&myExecutor, memoWeight, func(co *async.Coroutine, s *async.State[int]) {
		co.Watch(s1, s2)
		if v := s1.Get() + s2.Get(); v != s.Get() {
			s.Set(v)
		}
	})

	product := async.NewMemo(&myExecutor, memoWeight, func(co *async.Coroutine, s *async.State[int]) {
		co.Watch(s1, s2)
		if v := s1.Get() * s2.Get(); v != s.Get() {
			s.Set(v)
		}
	})

	op := async.NewState('+')

	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		co.Watch(op)

		fmt.Println("op =", "'"+string(op.Get())+"'")

		switch op.Get() {
		case '+':
			co.Spawn(func(co *async.Coroutine) async.Result {
				fmt.Println("s1 + s2 =", sum.Get())
				return co.Await(sum)
			})
		case '*':
			co.Spawn(func(co *async.Coroutine) async.Result {
				fmt.Println("s1 * s2 =", product.Get())
				return co.Await(product)
			})
		}

		return co.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 (Transition)

This example demonstrates how a coroutine can transition from one task to another.

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		co.Watch(&myState)

		v := myState.Get()
		fmt.Println(v)

		if v < 3 {
			return co.Await()
		}

		return co.Transition(func(co *async.Coroutine) async.Result {
			co.Watch(&myState)

			v := myState.Get()
			fmt.Println(v, "(transitioned)")

			if v < 5 {
				return co.Await()
			}

			return co.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 (transitioned)
4 (transitioned)
5 (transitioned)
7

Index

• type Cleanup
• type CleanupFunc
  • func (f CleanupFunc) Cleanup()
• type Coroutine
  • func (co *Coroutine) Await(ev ...Event) Result
  • func (co *Coroutine) AwaitSeq(seq iter.Seq[Event]) Result
  • func (co *Coroutine) Break() Result
  • func (co *Coroutine) BreakLabel(l Label) Result
  • func (co *Coroutine) Cleanup(c Cleanup)
  • func (co *Coroutine) CleanupFunc(f func())
  • func (co *Coroutine) Continue() Result
  • func (co *Coroutine) ContinueLabel(l Label) Result
  • func (co *Coroutine) Defer(t Task)
  • func (co *Coroutine) End() Result
  • func (co *Coroutine) Executor() *Executor
  • func (co *Coroutine) Exit() Result
  • func (co *Coroutine) Exiting() bool
  • func (co *Coroutine) Return() Result
  • func (co *Coroutine) Spawn(t Task)
  • func (co *Coroutine) Transition(t Task) Result
  • func (co *Coroutine) Watch(ev ...Event)
  • func (co *Coroutine) WatchSeq(seq iter.Seq[Event])
  • func (co *Coroutine) Weight() Weight
  • func (co *Coroutine) Yield(t Task) Result
• type Event
• type Executor
  • func (e *Executor) Autorun(f func())
  • func (e *Executor) Run()
  • func (e *Executor) Spawn(t Task)
  • func (e *Executor) SpawnWeighted(w Weight, t Task)
  • func (e *Executor) UsePool(p *sync.Pool)
• type Label
• type Memo
  • func NewMemo[T any](e *Executor, w Weight, f func(co *Coroutine, s *State[T])) *Memo[T]
  • func NewStrictMemo[T any](e *Executor, w Weight, f func(co *Coroutine, s *State[T])) *Memo[T]
  • func (m *Memo[T]) Get() T
• type Result
• type Semaphore
  • func NewSemaphore(n int64) *Semaphore
  • func (s *Semaphore) Acquire(n int64) Task
  • func (s *Semaphore) Release(n int64)
• type Signal
  • func (s *Signal) Notify()
• type State
  • func NewState[T any](v T) *State[T]
  • func (s *State[T]) Get() T
  • func (s *State[T]) Set(v T)
  • func (s *State[T]) Update(f func(v T) T)
• type Task
  • func Await(ev ...Event) Task
  • func AwaitSeq(seq iter.Seq[Event]) Task
  • func Block(s ...Task) Task
  • func Break() Task
  • func BreakLabel(l Label) Task
  • func Continue() Task
  • func ContinueLabel(l Label) Task
  • func Defer(t Task) Task
  • func Do(f func()) Task
  • func End() Task
  • func Exit() Task
  • func FromSeq(seq iter.Seq[Task]) Task
  • func Func(t Task) Task
  • func Join(s ...Task) Task
  • func Loop(t Task) Task
  • func LoopLabel(l Label, t Task) Task
  • func LoopLabelN[Int intType](l Label, n Int, t Task) Task
  • func LoopN[Int intType](n Int, t Task) Task
  • func Return() Task
  • func Select(s ...Task) Task
  • func (t Task) Then(next Task) Task
• type WaitGroup
  • func (wg *WaitGroup) Add(delta int)
  • func (wg *WaitGroup) Await() Task
  • func (wg *WaitGroup) Done()
• type Weight

Examples

• Package
• Package (CleanupFunc)
• Package (Conditional)
• Package (ConditionalMemo)
• Package (End)
• Package (Memo)
• Package (NonBlocking)
• Package (Transition)
• Await
• Block
• FromSeq
• Func
• Func (Exit)
• Func (Panic)
• Func (Tailcall)
• Join
• Loop
• LoopN
• Select
• Select (WithCancel)
• Semaphore
• Semaphore (Cancel)
• Task.Then
• WaitGroup

Types

type Cleanup

type Cleanup interface {
	Cleanup()
}

Cleanup represents any type that carries a Cleanup method. A Cleanup can be added to a coroutine in a Task function for making an effect some time later when the coroutine resumes or ends, or when the coroutine is making a transition to work on another Task.

type CleanupFunc

type CleanupFunc func()

A CleanupFunc is a func() that implements the Cleanup interface.

func (CleanupFunc) Cleanup

func (f CleanupFunc) Cleanup()

Cleanup implements the Cleanup interface.

type Coroutine

type Coroutine struct {
	// contains filtered or unexported fields
}

A Coroutine is an execution of code, similar to a goroutine but cooperative and stackless.

A coroutine is created with a function called Task. A coroutine's job is to end the task. When an Executor spawns a coroutine with a task, it runs the coroutine by calling the task function with the coroutine as the argument. The return value determines whether to end the coroutine or to yield it so that it could resume later.

In order for a coroutine to resume, the coroutine must watch at least one Event (e.g. Signal, State and Memo, etc.), when calling the task function. A notification of such an event resumes the coroutine. When a coroutine is resumed, the executor runs the coroutine again.

A coroutine can also make a transition to work on another task according to the return value of the task function. A coroutine can transition from one task to another until a task ends it.

func (*Coroutine) Await

func (co *Coroutine) Await(ev ...Event) Result

Await returns a Result that will cause co to yield. Await also accepts additional events to watch.

Yielding is not allowed when a coroutine is exiting. If co is exiting, Await panics.

func (*Coroutine) AwaitSeq

func (co *Coroutine) AwaitSeq(seq iter.Seq[Event]) Result

AwaitSeq returns a Result that will cause co to yield. AwaitSeq also accepts a sequence of events from seq to watch.

Yielding is not allowed when a coroutine is exiting. If co is exiting, AwaitSeq panics.

func (*Coroutine) Break

func (co *Coroutine) Break() Result

Break returns a Result that will cause co to break a loop.

func (*Coroutine) BreakLabel

func (co *Coroutine) BreakLabel(l Label) Result

BreakLabel returns a Result that will cause co to break a loop with label l.

func (*Coroutine) Cleanup

func (co *Coroutine) Cleanup(c Cleanup)

Cleanup adds something to clean up when co resumes or ends, or when co is making a transition to work on another Task.

func (*Coroutine) CleanupFunc

func (co *Coroutine) CleanupFunc(f func())

CleanupFunc adds a function call when co resumes or ends, or when co is making a transition to work on another Task.

func (*Coroutine) Continue

func (co *Coroutine) Continue() Result

Continue returns a Result that will cause co to continue a loop.

func (*Coroutine) ContinueLabel

func (co *Coroutine) ContinueLabel(l Label) Result

ContinueLabel returns a Result that will cause co to continue a loop with label l.

func (*Coroutine) Defer

func (co *Coroutine) Defer(t Task)

Defer adds a Task for execution when returning from a Func. Deferred tasks are executed in last-in-first-out (LIFO) order.

func (*Coroutine) End

func (co *Coroutine) End() Result

End returns a Result that will cause co to end, or make a transition to work on another Task if there are advanced control structures taking control.

func (*Coroutine) Executor

func (co *Coroutine) Executor() *Executor

Executor returns the executor that spawned co.

Since co can be recycled by an executor, it is recommended to save the return value in a variable first.

func (*Coroutine) Exit

func (co *Coroutine) Exit() Result

Exit returns a Result that will cause co to exit. All deferred tasks will be run before co exits.

func (*Coroutine) Exiting

func (co *Coroutine) Exiting() bool

Exiting reports whether co is exiting.

func (*Coroutine) Return

func (co *Coroutine) Return() Result

Return returns a Result that will cause co to return from a Func.

func (*Coroutine) Spawn

func (co *Coroutine) Spawn(t Task)

Spawn creates an inner coroutine with the same weight as co to work on t.

Inner coroutines, if not yet ended, are forcely exited when the outer one resumes or ends, or when the outer one is making a transition to work on another task.

func (*Coroutine) Transition

func (co *Coroutine) Transition(t Task) Result

Transition returns a Result that will cause co to make a transition to work on t.

func (*Coroutine) Watch

func (co *Coroutine) Watch(ev ...Event)

Watch watches some events so that, when any of them notifies, co resumes.

func (*Coroutine) WatchSeq

func (co *Coroutine) WatchSeq(seq iter.Seq[Event])

WatchSeq watches a sequence of events from seq so that, when any of them notifies, co resumes.

func (*Coroutine) Weight

func (co *Coroutine) Weight() Weight

Weight returns the weight of co.

func (*Coroutine) Yield

func (co *Coroutine) Yield(t Task) Result

Yield returns a Result that will cause co to yield and, when co is resumed, make a transition to work on t instead.

Yielding is not allowed when a coroutine is exiting. If co is exiting, Yield panics.

type Event

type Event interface {
	// contains filtered or unexported methods
}

Event is the interface of any type that can be watched by a coroutine.

The following types implement Event: Signal, State and Memo. Any type that embeds Signal also implements Event, e.g. State.

type Executor

type Executor struct {
	// contains filtered or unexported fields
}

An Executor is a coroutine spawner, and a coroutine runner.

When a coroutine 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 coroutine blocks, no other coroutines can run. The best practice is not to block.

The internal queue is a priority queue. Coroutines added in the queue are sorted by their weights. Coroutines with the same weight are sorted by their levels (inner coroutines have one level higher than their outer ones). Coroutines with the same weight and level are sorted by their arrival order (FIFO). Popping the queue removes the first coroutine with the highest weight or the least level.

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 coroutine is spawned or resumed. An 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 coroutine 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 coroutine in the queue until the queue is emptied.

Run must not be called twice at the same time.

func (*Executor) Spawn

func (e *Executor) Spawn(t Task)

Spawn creates a coroutine with default weight to work on t.

The coroutine is added in a queue. To run it, either call the Run method, or call the Autorun method to set up an autorun function beforehand.

Spawn is safe for concurrent use.

func (*Executor) SpawnWeighted

func (e *Executor) SpawnWeighted(w Weight, t Task)

SpawnWeighted creates a coroutine with weight w to work on t.

The coroutine is added in a queue. To run it, either call the Run method, or call the Autorun method to set up an autorun function beforehand.

SpawnWeighted is safe for concurrent use.

func (*Executor) UsePool

func (e *Executor) UsePool(p *sync.Pool)

UsePool tells e to use p, rather than a new one, to cache unused coroutine objects for later reuse.

type Label

type Label string

const NoLabel Label = ""

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 a given Task-like function driven by an internal coroutine.

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

func NewMemo[T any](e *Executor, w Weight, f func(co *Coroutine, s *State[T])) *Memo[T]

NewMemo returns a new non-strict Memo. The arguments are used to initialize an internal coroutine.

One must pass a function that watches some states, computes a value from these states, and then updates the given state if the value differs.

Like any Event, a memo can be watched by multiple coroutines. The watch list increases and decreases over time. For a non-strict memo, when the last coroutine in the list unwatches it, it does not immediately end its internal coroutine. Ending the internal coroutine would only put the memo into a stale state because the memo no longer detects dependency changes. By not immediately ending the internal coroutine, a non-strict memo prevents an extra computation when a new coroutine watches it, provided that there are no dependency changes.

On the other hand, a strict memo immediately ends its internal coroutine whenever the last coroutine in the watch list unwatches it. The memo becomes stale. The next time a new coroutine watches it, it has to make a fresh computation.

func NewStrictMemo

func NewStrictMemo[T any](e *Executor, w Weight, f func(co *Coroutine, s *State[T])) *Memo[T]

NewStrictMemo returns a new strict Memo.

See NewMemo for more information.

func (*Memo[T]) Get

func (m *Memo[T]) Get() T

Get retrieves the value of m.

One should only call this method in a Task function.

type Result

type Result struct {
	// contains filtered or unexported fields
}

Result is the type of the return value of a Task function. A Result determines what next for a coroutine to do after running a task.

A Result can be created by calling one of the following methods:

• Coroutine.End: for ending a coroutine or making a transition to work on another task if there are advanced control structures taking control;
• Coroutine.Await: for yielding a coroutine with additional events to watch;
• Coroutine.Yield: for yielding a coroutine with another task to which will be transitioned later when resuming;
• Coroutine.Transition: for making a transition to work on another task;
• Coroutine.Break: for breaking a loop;
• Coroutine.BreakLabel: for breaking a loop with a specific label;
• Coroutine.Continue: for continuing a loop;
• Coroutine.ContinueLabel: for continuing a loop with a specific label;
• Coroutine.Return: for returning from a Func;
• Coroutine.Exit: for exiting a coroutine.

These methods may have side effects. One should never store a Result in a variable and overwrite it with another, before returning it. Instead, one should just return a Result right after it is created.

type Semaphore

type Semaphore struct {
	// contains filtered or unexported fields
}

Semaphore provides a way to bound asynchronous access to a resource. The callers can request access with a given weight.

Note that this Semaphore type does not provide backpressure for spawning a lot of tasks. One should instead look for a sync implementation.

A Semaphore must not be shared by more than one Executor.

Example

package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/b97tsk/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()
		}()
	})

	mySemaphore := async.NewSemaphore(12)

	for n := int64(1); n <= 8; n++ {
		myExecutor.Spawn(mySemaphore.Acquire(n).Then(async.Do(func() {
			fmt.Println(n)
			wg.Add(1)
			go func() {
				defer wg.Done()
				time.Sleep(100 * time.Millisecond)
				myExecutor.Spawn(async.Do(func() { mySemaphore.Release(n) }))
			}()
		})))
	}

	wg.Wait()

}

Output:

1
2
3
4
5
6
7
8

Example (Cancel)

package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/b97tsk/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()
		}()
	})

	mySemaphore := async.NewSemaphore(3)

	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		// Four Acquire calls, only two of them can succeed;
		// the other two get canceled later when co ends.
		for n := int64(1); n <= 4; n++ {
			co.Spawn(mySemaphore.Acquire(n).Then(async.Do(func() {
				fmt.Println(n)
			})))
		}

		var sig async.Signal

		wg.Add(1)
		go func() {
			defer wg.Done()
			time.Sleep(100 * time.Millisecond)
			myExecutor.Spawn(async.Do(func() { sig.Notify() }))
		}()

		co.Watch(&sig)
		return co.Yield(async.End())
	})

	wg.Wait()

}

Output:

1
2

func NewSemaphore

func NewSemaphore(n int64) *Semaphore

NewSemaphore creates a new weighted semaphore with the given maximum combined weight.

func (*Semaphore) Acquire

func (s *Semaphore) Acquire(n int64) Task

Acquire returns a Task that awaits until a weight of n is acquired from the semaphore, and then ends.

func (*Semaphore) Release

func (s *Semaphore) Release(n int64)

Release releases the semaphore with a weight of n.

One should only call this method in a Task function.

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 a Task function, resumes any coroutine that is watching the signal.

A Signal must not be shared by more than one Executor.

func (*Signal) Notify

func (s *Signal) Notify()

Notify resumes any coroutine that is watching s.

One should only call this method in a Task function.

type State

type State[T any] struct {
	Signal
	// contains filtered or unexported fields
}

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 a Task function, updates the value and resumes any coroutine that is watching the state.

A State must not be shared by more than one Executor.

func NewState

func NewState[T any](v T) *State[T]

NewState creates a new State with its initial value set to v.

func (*State[T]) Get

func (s *State[T]) Get() T

Get retrieves the value of s.

Without proper synchronization, one should only call this method in a Task function.

func (*State[T]) Set

func (s *State[T]) Set(v T)

Set updates the value of s and resumes any coroutine that is watching s.

One should only call this method in a Task function.

func (*State[T]) Update

func (s *State[T]) Update(f func(v T) T)

Update sets the value of s to f(s.Get()) and resumes any coroutine that is watching s.

One should only call this method in a Task function.

type Task

type Task func(co *Coroutine) Result

A Task is a piece of work that a coroutine is given to do when it is spawned. The return value of a task, a Result, determines what next for a coroutine to do.

The argument co must not escape, because co can be recycled by an Executor when co ends.

func Await

func Await(ev ...Event) Task

Await returns a Task that awaits some events until any of them notifies, and then ends. If ev is empty, Await returns a Task that never ends.

Example

This example computes two values in separate goroutines sequentially, then prints their sum.

package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/b97tsk/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(co *async.Coroutine) 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()
			}))
		}()

		return co.Transition(async.Await(&myState).Then(
			func(co *async.Coroutine) 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()
					}))
				}()

				return co.Transition(async.Await(&myState).Then(
					async.Do(func() {
						fmt.Println("v1 + v2 =", myState.v1+myState.v2)
					}),
				))
			},
		))
	})

	wg.Wait()

}

Output:

v1 + v2 = 42

func AwaitSeq

func AwaitSeq(seq iter.Seq[Event]) Task

AwaitSeq returns a Task that awaits a sequence of events from seq until any of them notifies, and then ends. If seq is empty, AwaitSeq returns a Task that never ends.

func Block

func Block(s ...Task) Task

Block returns a Task that runs each of the given tasks in sequence. When one task ends, Block runs another.

Example

This example demonstrates how to run a block of tasks. A block can have zero or more tasks. A block runs tasks in sequence.

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		var t async.Task

		t = async.Block(
			async.Await(&myState),
			async.Do(func() {
				if v := myState.Get(); v%2 != 0 {
					fmt.Println(v)
				}
			}),
			func(co *async.Coroutine) async.Result {
				if v := myState.Get(); v >= 7 {
					return co.End()
				}
				return co.Transition(t) // Transition to t again to form a loop.
			},
		)

		return co.Transition(t)
	})

	for i := 1; i <= 9; i++ {
		myExecutor.Spawn(async.Do(func() { myState.Set(i) }))
	}

	fmt.Println(myState.Get()) // Prints 9.

}

Output:

1
3
5
7
9

func Break

func Break() Task

Break returns a Task that breaks a loop.

func BreakLabel

func BreakLabel(l Label) Task

BreakLabel returns a Task that breaks a loop with label l.

func Continue

func Continue() Task

Continue returns a Task that continues a loop.

func ContinueLabel

func ContinueLabel(l Label) Task

ContinueLabel returns a Task that continues a loop with label l.

func Defer

func Defer(t Task) Task

Defer returns a Task that adds t for execution when returning from a Func. Deferred tasks are executed in last-in-first-out (LIFO) order.

func Do

func Do(f func()) Task

Do returns a Task that calls f, and then ends.

func End

func End() Task

End returns a Task that ends without doing anything.

func Exit

func Exit() Task

Exit returns a Task that causes the coroutine that runs it to exit. All deferred tasks are run before the coroutine exits.

func FromSeq

func FromSeq(seq iter.Seq[Task]) Task

FromSeq returns a Task that runs each of the tasks from seq in sequence.

Example

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	myExecutor.Spawn(async.FromSeq(
		func(yield func(async.Task) bool) {
			await := async.Await(&myState)
			for yield(await) {
				v := myState.Get()
				if v%2 != 0 {
					fmt.Println(v)
				}
				if v >= 7 {
					return
				}
			}
		},
	))

	for i := 1; i <= 9; i++ {
		myExecutor.Spawn(async.Do(func() { myState.Set(i) }))
	}

	fmt.Println(myState.Get()) // Prints 9.

}

Output:

1
3
5
7
9

func Func

func Func(t Task) Task

Func returns a Task that runs t in a function scope. Spawned tasks are considered surrounded by an invisible Func.

Example

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	myExecutor.Spawn(async.Block(
		async.Defer( // Note that spawned tasks are considered surrounded by an invisible async.Func.
			async.Do(func() { fmt.Println("defer 1") }),
		),
		async.Func(async.Block( // A block in a function scope.
			async.Defer(
				async.Do(func() { fmt.Println("defer 2") }),
			),
			async.Loop(async.Block(
				async.Await(&myState),
				func(co *async.Coroutine) async.Result {
					if v := myState.Get(); v%2 == 0 {
						return co.Continue()
					}
					return co.End()
				},
				async.Do(func() {
					fmt.Println(myState.Get())
				}),
				func(co *async.Coroutine) async.Result {
					if v := myState.Get(); v >= 7 {
						return co.Return() // Return here.
					}
					return co.End()
				},
			)),
			async.Do(func() { fmt.Println("after Loop") }), // Didn't run due to early return.
		)),
		async.Do(func() { fmt.Println("after Func") }),
	))

	for i := 1; i <= 9; i++ {
		myExecutor.Spawn(async.Do(func() { myState.Set(i) }))
	}

	fmt.Println(myState.Get()) // Prints 9.

}

Output:

1
3
5
7
defer 2
after Func
defer 1
9

Example (Exit)

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	myExecutor.Spawn(async.Block(
		async.Defer( // Note that spawned tasks are considered surrounded by an invisible async.Func.
			async.Do(func() { fmt.Println("defer 1") }),
		),
		async.Func(async.Block( // A block in a function scope.
			async.Defer(
				async.Do(func() { fmt.Println("defer 2") }),
			),
			async.Loop(async.Block(
				async.Await(&myState),
				func(co *async.Coroutine) async.Result {
					if v := myState.Get(); v%2 == 0 {
						return co.Continue()
					}
					return co.End()
				},
				async.Do(func() {
					fmt.Println(myState.Get())
				}),
				func(co *async.Coroutine) async.Result {
					if v := myState.Get(); v >= 7 {
						return co.Exit() // Exit here.
					}
					return co.End()
				},
			)),
			async.Do(func() { fmt.Println("after Loop") }), // Didn't run due to early exit.
		)),
		async.Do(func() { fmt.Println("after Func") }), // Didn't run due to early exit.
	))

	for i := 1; i <= 9; i++ {
		myExecutor.Spawn(async.Do(func() { myState.Set(i) }))
	}

	fmt.Println(myState.Get()) // Prints 9.

}

Output:

1
3
5
7
defer 2
defer 1
9

Example (Panic)

This example demonstrates how async can handle panickings.

package main

import (
	"errors"
	"fmt"
	"strings"

	"github.com/b97tsk/async"
)

func main() {
	dummyError := errors.New("dummy")

	var myExecutor async.Executor

	myExecutor.Autorun(func() {
		defer func() {
			v := recover()
			if v == nil {
				return
			}
			err, ok := v.(error)
			if ok && errors.Is(err, dummyError) && strings.Contains(err.Error(), "dummy") {
				// Use of strings.Contains(...) here is for code coverage, not that it is necessary.
				fmt.Println("recovered dummy error")
				return
			}
			panic(v) // Repanic unexpected recovered value.
		}()
		myExecutor.Run()
	})

	var myState async.State[int]

	myExecutor.Spawn(async.Block(
		async.Defer( // Note that spawned tasks are considered surrounded by an invisible async.Func.
			async.Do(func() { fmt.Println("defer 1") }),
		),
		async.Func(async.Block( // A block in a function scope.
			async.Defer(
				async.Do(func() { fmt.Println("defer 2") }),
			),
			async.Loop(async.Block(
				async.Await(&myState),
				func(co *async.Coroutine) async.Result {
					if v := myState.Get(); v%2 == 0 {
						return co.Continue()
					}
					return co.End()
				},
				async.Do(func() {
					fmt.Println(myState.Get())
				}),
				func(co *async.Coroutine) async.Result {
					if v := myState.Get(); v >= 7 {
						panic(dummyError) // Panic here.
					}
					return co.End()
				},
			)),
			async.Do(func() { fmt.Println("after Loop") }), // Didn't run due to early panicking.
		)),
		async.Do(func() { fmt.Println("after Func") }), // Didn't run due to early panicking.
	))

	for i := 1; i <= 9; i++ {
		myExecutor.Spawn(async.Do(func() { myState.Set(i) }))
	}

	fmt.Println(myState.Get()) // Prints 9.

}

Output:

1
3
5
7
defer 2
defer 1
recovered dummy error
9

Example (Tailcall)

This example demonstrates how to make tail-calls in an async.Func. Tail-calls are not recommended and should be avoided when possible. Without tail-call optimization, this example shall panic.

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	// Case 1: Making tail-call in the last task of a block.
	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		var n int

		var t async.Task

		t = async.Func(async.Block(
			async.End(),
			async.End(),
			async.End(),
			func(co *async.Coroutine) async.Result { // Last task in the block.
				if n < 2000000 {
					n++
					return co.Transition(t) // Tail-call here.
				}
				return co.End()
			},
		))

		return co.Transition(t.Then(async.Do(func() { fmt.Println(n) })))
	})

	// Case 2: Making tail-call anywhere.
	myExecutor.Spawn(func(co *async.Coroutine) async.Result {
		var n int

		var t async.Task

		t = async.Func(async.Block(
			func(co *async.Coroutine) async.Result {
				if n < 2000000 {
					n++
					co.Defer(t)        // Tail-call here (using the only defer call as a workaround).
					return co.Return() // Early return.
				}
				return co.End()
			},
			async.End(),
			async.End(),
			async.End(),
		))

		return co.Transition(t.Then(async.Do(func() { fmt.Println(n) })))
	})

}

Output:

2000000
2000000

func Join

func Join(s ...Task) Task

Join returns a Task that runs each of the given tasks in an inner coroutine and awaits until all of them complete, and then ends.

Example

package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/b97tsk/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 s1, s2 async.State[int]

	myExecutor.Spawn(async.Block(
		async.Join(
			func(co *async.Coroutine) 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() { s1.Set(ans) }))
				}()
				return co.Transition(async.Await(&s1))
			},
			func(co *async.Coroutine) async.Result {
				wg.Add(1)
				go func() {
					defer wg.Done()
					time.Sleep(1500 * time.Millisecond) // Heavy work #2 here.
					ans := 27
					myExecutor.Spawn(async.Do(func() { s2.Set(ans) }))
				}()
				return co.Transition(async.Await(&s2))
			},
		),
		async.Do(func() { fmt.Println("s1 + s2 =", s1.Get()+s2.Get()) }),
	))

	wg.Wait()

}

Output:

s1 + s2 = 42

func Loop

func Loop(t Task) Task

Loop returns a Task that forms a loop, which would run t repeatedly. Both Coroutine.Break and Break can break this loop early. Both Coroutine.Continue and Continue can continue this loop early.

Example

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	myExecutor.Spawn(async.Loop(async.Block(
		async.Await(&myState),
		func(co *async.Coroutine) async.Result {
			if v := myState.Get(); v%2 == 0 {
				return co.Continue()
			}
			return co.End()
		},
		async.Do(func() {
			fmt.Println(myState.Get())
		}),
		func(co *async.Coroutine) async.Result {
			if v := myState.Get(); v >= 7 {
				return co.Break()
			}
			return co.End()
		},
	)))

	for i := 1; i <= 9; i++ {
		myExecutor.Spawn(async.Do(func() { myState.Set(i) }))
	}

	fmt.Println(myState.Get()) // Prints 9.

}

Output:

1
3
5
7
9

func LoopLabel

func LoopLabel(l Label, t Task) Task

LoopLabel returns a Task that forms a loop with label l, which would run t repeatedly. Both Coroutine.Break and Break can break this loop early. Both Coroutine.Continue and Continue can continue this loop early. Both Coroutine.BreakLabel and BreakLabel, with label l, can break this loop early. Both Coroutine.ContinueLabel and ContinueLabel, with label l, can continue this loop early.

func LoopLabelN

func LoopLabelN[Int intType](l Label, n Int, t Task) Task

LoopLabelN returns a Task that forms a loop with label l, which would run t repeatedly for n times. Both Coroutine.Break and Break can break this loop early. Both Coroutine.Continue and Continue can continue this loop early. Both Coroutine.BreakLabel and BreakLabel, with label l, can break this loop early. Both Coroutine.ContinueLabel and ContinueLabel, with label l, can continue this loop early.

func LoopN

func LoopN[Int intType](n Int, t Task) Task

LoopN returns a Task that forms a loop, which would run t repeatedly for n times. Both Coroutine.Break and Break can break this loop early. Both Coroutine.Continue and Continue can continue this loop early.

Example

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	myExecutor.Spawn(async.LoopN(7, async.Block(
		async.Await(&myState),
		func(co *async.Coroutine) async.Result {
			if v := myState.Get(); v%2 == 0 {
				return co.Continue()
			}
			return co.End()
		},
		async.Do(func() {
			fmt.Println(myState.Get())
		}),
	)))

	for i := 1; i <= 9; i++ {
		myExecutor.Spawn(async.Do(func() { myState.Set(i) }))
	}

	fmt.Println(myState.Get()) // Prints 9.

}

Output:

1
3
5
7
9

func Return

func Return() Task

Return returns a Task that returns from a surrounding Func.

func Select

func Select(s ...Task) Task

Select returns a Task that runs each of the given tasks in an inner coroutine and awaits until any of them completes, and then ends. When Select ends, tasks other than the one that completes are forcely exited.

Example

package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/b97tsk/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 s1, s2 async.State[int]

	myExecutor.Spawn(async.Block(
		async.Select(
			func(co *async.Coroutine) 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() { s1.Set(ans) }))
				}()
				return co.Transition(async.Await(&s1))
			},
			func(co *async.Coroutine) async.Result {
				wg.Add(1)
				go func() {
					defer wg.Done()
					time.Sleep(1500 * time.Millisecond) // Heavy work #2 here.
					ans := 27
					myExecutor.Spawn(async.Do(func() { s2.Set(ans) }))
				}()
				return co.Transition(async.Await(&s2))
			},
		),
		async.Do(func() { fmt.Println("s1 + s2 =", s1.Get()+s2.Get()) }),
	))

	wg.Wait()

}

Output:

s1 + s2 = 15

Example (WithCancel)

Without cancellation, ExampleSelect takes the same amount of time as ExampleJoin, which is unacceptable. The following example fixes that.

package main

import (
	"context"
	"fmt"
	"sync"
	"time"

	"github.com/b97tsk/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 s1, s2 async.State[int]

	myExecutor.Spawn(async.Block(
		async.Func(
			func(co *async.Coroutine) async.Result {
				ctx, cancel := context.WithCancel(context.Background())
				co.Defer(async.Do(cancel))
				return co.Transition(async.Select(
					func(co *async.Coroutine) async.Result {
						wg.Add(1)
						go func() {
							defer wg.Done()
							select { // Heavy work #1 here.
							case <-time.After(500 * time.Millisecond):
							case <-ctx.Done():
								return // Cancel work when ctx gets canceled.
							}
							ans := 15
							myExecutor.Spawn(async.Do(func() { s1.Set(ans) }))
						}()
						return co.Transition(async.Await(&s1))
					},
					func(co *async.Coroutine) async.Result {
						wg.Add(1)
						go func() {
							defer wg.Done()
							select { // Heavy work #2 here.
							case <-time.After(1500 * time.Millisecond):
							case <-ctx.Done():
								return // Cancel work when ctx gets canceled.
							}
							ans := 27
							myExecutor.Spawn(async.Do(func() { s2.Set(ans) }))
						}()
						return co.Transition(async.Await(&s2))
					},
				))
			},
		),
		async.Do(func() { fmt.Println("s1 + s2 =", s1.Get()+s2.Get()) }),
	))

	wg.Wait()

}

Output:

s1 + s2 = 15

func (Task) Then

func (t Task) Then(next Task) Task

Then returns a Task that first works on t, then next after t ends.

To chain multiple tasks, use Block function.

Example

This example demonstrates how to run a task after another. To run multiple tasks in sequence, use async.Block instead.

package main

import (
	"fmt"

	"github.com/b97tsk/async"
)

func main() {
	var myExecutor async.Executor

	myExecutor.Autorun(myExecutor.Run)

	var myState async.State[int]

	a := func(co *async.Coroutine) async.Result {
		co.Watch(&myState)

		v := myState.Get()
		fmt.Println(v, "(a)")

		if v < 3 {
			return co.Await()
		}

		return co.Transition(func(co *async.Coroutine) async.Result {
			co.Watch(&myState)

			v := myState.Get()
			fmt.Println(v, "(transitioned)")

			if v < 5 {
				return co.Await()
			}

			return co.End()
		})
	}

	b := func(co *async.Coroutine) async.Result {
		co.Watch(&myState)

		v := myState.Get()
		fmt.Println(v, "(b)")

		if v < 7 {
			return co.Await()
		}

		return co.End()
	}

	myExecutor.Spawn(async.Task(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 (transitioned)
4 (transitioned)
5 (transitioned)
5 (b)
6 (b)
7 (b)
9

type WaitGroup

type WaitGroup struct {
	Signal
	// contains filtered or unexported fields
}

A WaitGroup is a Signal with a counter.

Calling the Add or Done method of a WaitGroup, in a Task function, updates the counter and, when the counter becomes zero, resumes any coroutine that is watching the WaitGroup.

A WaitGroup must not be shared by more than one Executor.

Example

package main

import (
	"fmt"
	"sync"
	"time"

	"github.com/b97tsk/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 {
		wg     async.WaitGroup
		v1, v2 int
	}

	myState.wg.Add(2) // Note that async.WaitGroup is not safe for concurrent use.

	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.wg.Done()
		}))
	}()

	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.wg.Done()
		}))
	}()

	myExecutor.Spawn(myState.wg.Await().Then(async.Do(func() {
		fmt.Println("v1 + v2 =", myState.v1+myState.v2)
	})))

	wg.Wait()

}

Output:

v1 + v2 = 42

func (*WaitGroup) Add

func (wg *WaitGroup) Add(delta int)

Add adds delta, which may be negative, to the WaitGroup counter. If the WaitGroup counter becomes zero, Add resumes any coroutine that is watching wg. If the WaitGroup counter is negative, Add panics.

func (*WaitGroup) Await

func (wg *WaitGroup) Await() Task

Await returns a Task that awaits until the WaitGroup counter becomes zero, and then ends.

func (*WaitGroup) Done

func (wg *WaitGroup) Done()

Done decrements the WaitGroup counter by one.

type Weight

type Weight int

Weight is the type of weight for use when spawning a weighted coroutine.