How to change channel in for-select loop

I want to do a dynamic channel pool ,which can listen Hundreds of thousands channels and all of these are under control ,and as i excepted i want it to be auto upgradable if there are too much channels listening (goroutine => reflect => selectn)
But during selectN channel watcher coding , i was blocked by channel replacing

I want to replace chan during runtime which is in for select loop, and I have tried it for some times to make it available, but things not went well.

func Test_Change(t *testing.T) {
type A struct {
	ch chan interface{}
}
a := &A{
	ch: make(chan interface{}),
}
go func() {
	for {
		select {
		case v := <-a.ch:
			fmt.Println(v)
		}
	}
}()

newCh := make(chan interface{})
go func() {
	for i := 0; i < 200; i++ {
		a.ch <- i
	}
	a.ch = newCh
}()
go func() {
	for i := 1000; i < 1010; i++ {
		newCh <- i
	}
}()
for {
	select {}
}}

It blocked

func Test_Change(t *testing.T) {
type A struct {
	ch chan interface{}
	bh chan interface{}
}
a := &A{
	ch: make(chan interface{}),
	bh: make(chan interface{}),
}
notify := make(chan struct{})
go func() {
	for {
		select {
		case v := <-a.ch:
			fmt.Println(v)
		case <-notify:
			fmt.Println("notify")
		}
	}
}()
newCh := make(chan interface{})
go func() {
	for i := 0; i < 200; i++ {
		a.ch <- i
	}
	a.ch = newCh
	notify <- struct{}{}
}()
go func() {
	for i := 1000; i < 1010; i++ {
		newCh <- i
	}
}()
for {
	select {}
}}

And it worked

You have correctly determined that a select block on an arbitrarily large dynamic number of chans is not possible with the usual syntax (fixed number of cases), and is possible using the reflect package.

However, I'm not sure this is the best way to achieve your goal. If you do have thousands of channels to watch (e.g. thousands of remote clients connected at the same time), then you may use a "fan-in" pattern to write everything to a very small fixed number of channels, and select on that.

Instead of

    for {
        select {
        case <-sigterm:
            cleanup()
            os.Exit(1)
        case msg := <-client1:
            // process msg...
        case msg := <-client2:
            // process msg...
        // HOW CAN I DYNAMICALLY ADD AND REMOVE A CLIENT HERE?
        }
    }

Think of something like:

    for {
        select {
        case <-sigterm:
            cleanup()
            os.Exit(1)
        case msg := <-clients:
            // process msg...
        }
    }

func addClient(client chan Message) {
    // Fan-in: read all future messages from client, and write them
    // to clients.
    go func(){
        for msg := range client {
            clients <- msg
        }
    }()
}

Replacing the value of a channel variable is not thread-safe (can be a data race), however it is perfectly fine to have several goroutines write to and read from the same channel clients, concurrently.

The data race on a.ch is a serious bug and a design flaw. You can detect data races by running the tests with go test -race or the program with go run -race program.go.

It is possible to replace the value of the chans used inside a for/select loop, as long as it's properly done inside the body of a case, not in the code of another concurrent goroutine.

	replace := time.After(3 * time.Second)

	for {
		select {
		case v, ok := <-ch1:
			// use v...
		case v, ok := <-ch2:
			// use v...
		case <-replace:
			ch1 = anotherChannel
		}
	}

This sample runnable code is racy (don't do it). You can save it on your workstation and try with the data race detector.

This fixed sample code is not racy.

Maybe this might work;

test it very very extensively, it is doable, but hard to get it right.

I expect it to be not perfect. And it lacks extensive docs about do's and dont's.

It is also not the version that merges input channel, it always ever consume only one input channel at a time. Which might a problem for performance.

The only guarantee i give is that it is race free.

Though I leave as an exercise to the reader the task to write a fan-in-ed version.

package main

import (
	"fmt"
)

func main() {
	m := New()
	go m.Run()
	input := m.Resize(0)
	input <- 5
	input <- 4
	close(input)

	input = m.Resize(10)
	input <- 6
	input <- 7
	close(input)

	input = m.Resize(2)
	input <- 8
	input <- 9
	close(input)

	m.Close()

	fmt.Println()

}

type masterOfThings struct {
	notify    chan notification
	wantClose chan chan bool
}

func New() masterOfThings {
	return masterOfThings{
		notify:    make(chan notification, 1),
		wantClose: make(chan chan bool),
	}
}

type notification struct {
	N   int
	out chan chan interface{}
}

func (m masterOfThings) Resize(n int) chan<- interface{} {
	N := notification{
		N:   n,
		out: make(chan chan interface{}, 1),
	}
	m.notify <- N
	return <-N.out
}

func (m masterOfThings) Close() {
	closed := make(chan bool)
	m.wantClose <- closed
	<-closed
}

func (m masterOfThings) Run() {
	var input chan interface{}
	inputs := []chan interface{}{}
	closers := []chan bool{}
	defer func() {
		for _, c := range closers {
			close(c)
		}
	}()
	var wantClose bool
	for {
		select {
		case m := <-m.wantClose:
			closers = append(closers, m)
			wantClose = true
			if len(inputs) < 1 && input == nil {
				return
			}
		case n, ok := <-input:
			if !ok {
				input = nil
				if len(inputs) > 0 {
					input = inputs[0]
					copy(inputs, inputs[1:])
					inputs = inputs[:len(inputs)-1]
				} else if wantClose {
					return
				}
				continue
			}
			fmt.Println(n)
		case n := <-m.notify:
			nInput := make(chan interface{}, n.N)
			if input == nil {
				input = nInput
			} else {
				inputs = append(inputs, nInput)
			}
			n.out <- nInput
		}
	}
}