chann: a few more generic chann functions

Author: changkun

README.md

@@ -1,14 +1,14 @@
 # chann ![example workflow](https://github.com/golang-design/chann/actions/workflows/chann.yml/badge.svg) ![](https://changkun.de/urlstat?mode=github&repo=golang-design/chann)
 
-a unified representation of buffered, unbuffered, and unbounded channels in Go
+a unified channel package in Go
 
 ```go
 import "golang.design/x/chann"
 ```
 
 This package requires Go 1.18.
 
-## Usage
+## Basic Usage
 
 Different types of channels:
 
@@ -34,8 +34,8 @@ ch.Close()
 Channel properties:
 
 ```go
-ch.ApproxLen() // an (approx. of) length of the channel
-ch.Cap()       // the capacity of the channel
+ch.Len() // the length of the channel
+ch.Cap() // the capacity of the channel
 ```
 
 See https://golang.design/research/ultimate-channel for more details of

chann.go

@@ -4,8 +4,7 @@
 //
 // Written by Changkun Ou <changkun.de>
 
-// Package chann provides a unified representation of buffered,
-// unbuffered, and unbounded channels in Go.
+// Package chann providesa a unified channel package.
 //
 // The package is compatible with existing buffered and unbuffered
 // channels. For example, in Go, to create a buffered or unbuffered
@@ -204,11 +203,11 @@ func (ch *Chann[T]) Close() {
 	}
 }
 
-// ApproxLen returns an approximation of the length of the channel.
+// Len returns an approximation of the length of the channel.
 //
 // Note that in a concurrent scenario, the returned length of a channel
 // may never be accurate. Hence the function is named with an Approx prefix.
-func (ch *Chann[T]) ApproxLen() int {
+func (ch *Chann[T]) Len() int {
 	switch ch.cfg.typ {
 	case buffered, unbuffered:
 		return len(ch.in)

chann_test.go

@@ -186,14 +186,14 @@ func TestChan(t *testing.T) {
 		{
 			// Test len/cap.
 			c := chann.New[int](chann.Cap(chanCap))
-			if c.ApproxLen() != 0 || c.Cap() != chanCap {
-				t.Fatalf("chan[%d]: bad len/cap, expect %v/%v, got %v/%v", chanCap, 0, chanCap, c.ApproxLen(), c.Cap())
+			if c.Len() != 0 || c.Cap() != chanCap {
+				t.Fatalf("chan[%d]: bad len/cap, expect %v/%v, got %v/%v", chanCap, 0, chanCap, c.Len(), c.Cap())
 			}
 			for i := 0; i < chanCap; i++ {
 				c.In() <- i
 			}
-			if c.ApproxLen() != chanCap || c.Cap() != chanCap {
-				t.Fatalf("chan[%d]: bad len/cap, expect %v/%v, got %v/%v", chanCap, chanCap, chanCap, c.ApproxLen(), c.Cap())
+			if c.Len() != chanCap || c.Cap() != chanCap {
+				t.Fatalf("chan[%d]: bad len/cap, expect %v/%v, got %v/%v", chanCap, chanCap, chanCap, c.Len(), c.Cap())
 			}
 		}
 	}

lb.go

@@ -0,0 +1,55 @@
+// Copyright 2022 The golang.design Initiative Authors.
+// All rights reserved. Use of this source code is governed
+// by a MIT license that can be found in the LICENSE file.
+//
+// Written by Changkun Ou <changkun.de>
+
+package chann
+
+import (
+	"math/rand"
+	"sync"
+)
+
+// Fanin provides a generic fan-in functionality for variadic channels.
+func Fanin[T any](chans ...*Chann[T]) *Chann[T] {
+	buf := 0
+	for _, ch := range chans {
+		if ch.Len() > buf {
+			buf = ch.Len()
+		}
+	}
+	out := New[T](Cap(buf))
+	wg := sync.WaitGroup{}
+	wg.Add(len(chans))
+	for _, ch := range chans {
+		go func(ch *Chann[T]) {
+			for v := range ch.Out() {
+				out.In() <- v
+			}
+			wg.Done()
+		}(ch)
+	}
+	go func() {
+		wg.Wait()
+		out.Close()
+	}()
+	return out
+}
+
+// Fanout provides a generic fan-out functionality for variadic channels.
+func Fanout[T any](randomizer func(max int) int, in *Chann[T], outs ...*Chann[T]) {
+	l := len(outs)
+	for v := range in.Out() {
+		i := randomizer(l)
+		if i < 0 || i > l {
+			i = rand.Intn(l)
+		}
+		go func(v T) { outs[i].In() <- v }(v)
+	}
+}
+
+// LB load balances the given input channels to the output channels.
+func LB[T any](randomizer func(max int) int, ins []*Chann[T], outs []*Chann[T]) {
+	Fanout(randomizer, Fanin(ins...), outs...)
+}

lb_test.go

@@ -0,0 +1,54 @@
+// Copyright 2022 The golang.design Initiative Authors.
+// All rights reserved. Use of this source code is governed
+// by a MIT license that can be found in the LICENSE file.
+//
+// Written by Changkun Ou <changkun.de>
+
+package chann_test
+
+import (
+	"math/rand"
+	"testing"
+
+	"golang.design/x/chann"
+)
+
+func getInputChan() *chann.Chann[int] {
+	input := chann.New[int](chann.Cap(20))
+	numbers := []int{0, 1, 2, 3, 4, 5, 6, 7, 8, 9}
+	go func() {
+		for num := range numbers {
+			input.In() <- num
+		}
+		input.Close()
+	}()
+	return input
+}
+
+func TestFanin(t *testing.T) {
+	chs := make([]*chann.Chann[int], 10)
+	for i := 0; i < 10; i++ {
+		chs[i] = getInputChan()
+	}
+
+	out := chann.Fanin(chs...)
+	count := 0
+	for range out.Out() {
+		count++
+	}
+	if count != 100 {
+		t.Fatalf("Fanin failed")
+	}
+}
+
+func TestLB(t *testing.T) {
+	ins := make([]*chann.Chann[int], 10)
+	for i := 0; i < 10; i++ {
+		ins[i] = getInputChan()
+	}
+	outs := make([]*chann.Chann[int], 10)
+	for i := 0; i < 10; i++ {
+		outs[i] = chann.New[int](chann.Cap(10))
+	}
+	chann.LB(func(m int) int { return rand.Intn(m) }, ins, outs)
+}

utils.go

@@ -0,0 +1,64 @@
+// Copyright 2022 The golang.design Initiative Authors.
+// All rights reserved. Use of this source code is governed
+// by a MIT license that can be found in the LICENSE file.
+//
+// Written by Changkun Ou <changkun.de>
+
+package chann
+
+import "runtime"
+
+// Ranger returns a Sender and a Receiver. The Receiver provides a
+// Next method to retrieve values. The Sender provides a Send method
+// to send values and a Close method to stop sending values. The Next
+// method indicates when the Sender has been closed, and the Send
+// method indicates when the Receiver has been freed.
+//
+// This is a convenient way to exit a goroutine sending values when
+// the receiver stops reading them.
+func Ranger[T any]() (*Sender[T], *Receiver[T]) {
+	c := New[T]()
+	d := New[bool]()
+	s := &Sender[T]{values: c, done: d}
+	r := &Receiver[T]{values: c, done: d}
+	runtime.SetFinalizer(r, func(r *Receiver[T]) { r.finalize() })
+	return s, r
+}
+
+// A sender is used to send values to a Receiver.
+type Sender[T any] struct {
+	values *Chann[T]
+	done   *Chann[bool]
+}
+
+// Send sends a value to the receiver. It returns whether any more
+// values may be sent; if it returns false the value was not sent.
+func (s *Sender[T]) Send(v T) bool {
+	select {
+	case s.values.In() <- v:
+		return true
+	case <-s.done.Out():
+		return false
+	}
+}
+
+// Close tells the receiver that no more values will arrive.
+// After Close is called, the Sender may no longer be used.
+func (s *Sender[T]) Close() { s.values.Close() }
+
+// A Receiver receives values from a Sender.
+type Receiver[T any] struct {
+	values *Chann[T]
+	done   *Chann[bool]
+}
+
+// Next returns the next value from the channel. The bool result
+// indicates whether the value is valid, or whether the Sender has
+// been closed and no more values will be received.
+func (r *Receiver[T]) Next() (T, bool) {
+	v, ok := <-r.values.Out()
+	return v, ok
+}
+
+// finalize is a finalizer for the receiver.
+func (r *Receiver[T]) finalize() { r.done.Close() }

utils_test.go

@@ -0,0 +1,27 @@
+// Copyright 2021 The golang.design Initiative Authors.
+// All rights reserved. Use of this source code is governed
+// by a MIT license that can be found in the LICENSE file.
+//
+// Written by Changkun Ou <changkun.de>
+
+package chann_test
+
+import (
+	"testing"
+
+	"golang.design/x/chann"
+)
+
+func TestRanger(t *testing.T) {
+	s, r := chann.Ranger[int]()
+
+	go func() {
+		s.Send(42)
+	}()
+
+	n, ok := r.Next()
+	if !ok {
+		t.Fatalf("cannot receive from senter")
+	}
+	t.Log(n)
+}