Function asyncControllable

interpreter/async_test.go:452–473

asyncControllable returns a channel-based AsyncOp whose calls block until release is closed, tracking the number of concurrently live calls and the high-water mark.

(release <-chan struct{}, live, maxLive *atomic.Int32)

// asyncControllable returns a channel-based AsyncOp whose calls block until release is closed,
// tracking the number of concurrently live calls and the high-water mark.
func asyncControllable(release <-chan struct{}, live, maxLive *atomic.Int32) functions.AsyncOp {
	return func(ctx context.Context, args ...ref.Val) <-chan ref.Val {
		ch := make(chan ref.Val, 1)
		go func() {
			cur := live.Add(1)
			for {
				old := maxLive.Load()
				if cur <= old || maxLive.CompareAndSwap(old, cur) {
					break
				}
			}
			select {
			case <-release:
			case <-ctx.Done():
			}
			live.Add(-1)
			ch <- args[0]
			close(ch)
		}()
		return ch
	}
}

func TestLaunchAdmissionAndBounding(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())