contextとchannelを用いた実装サンプル Receiver:Senderが 1:Nの場合

# 概要

goで非同期を扱う際には、channelでgo routine間のコミュニケーションを行い、
contextでgo routineの寿命を管理することが主流であるようだ。

基本的なパターンとして、Receiver:Senderが
- 1:N
- N:1
- N:N

の場合があると思う。今回は最初のパターンについて実装サンプルを載せたい。


# サンプル

## 設計

ReceiverからSenderを作成し、非同期に走らせる関数の引数として渡し、
関数が終了したらエラーか結果をSendし、ReceiverはそれらをListenする。

Receiverが持っているchannelをCloseする際には、
SenderがすべてCloseされたかを確認する必要があるため、
カウンタを用意し、これが0になったらcloseするようにしている。

## 詳細
まずはReceiverに対して何個のSenderが参照しているか（CloseされてないSenderが何個あるか)
を計算する奴を作る。

```go: 

type ReferenceCounter struct {
	mu  sync.Mutex
	ctr int
}

func NewReferenceCounter() *ReferenceCounter {
	return &ReferenceCounter{}
}

func (r ReferenceCounter) IsMinus() bool {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.ctr <= 0
}

func (r *ReferenceCounter) Increment() {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.ctr++
}

func (r *ReferenceCounter) Decrement() {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.ctr--
}

```

Receiverの実装

```go:
type Result int

type Receiver struct {
	ctx    context.Context
	cancel func()
	errCh  chan error
	resCh  chan Result
	refCtr *ReferenceCounter // 作成されたSender数
}

func NewReceiver(ctx context.Context) Receiver {
	ctx, cancel := context.WithTimeout(ctx, timeout)
	return Receiver{
		ctx:    ctx,
		cancel: cancel,
		errCh:  make(chan error, 1),
		resCh:  make(chan Result, 10),
		refCtr: NewReferenceCounter(),
	}
}

func (r Receiver) Close() {
	r.cancel()
    // すべてのSenderがCloseされてrefCtrが0になったらchannelをCloseしている。
	go func() {
		if r.refCtr.IsMinus() {
			close(r.resCh)
			close(r.errCh)
		}
	}()
}

func (r Receiver) Receive() (Result, error) {
	for {
		select {
		case <-r.ctx.Done():
			return 0, r.ctx.Err()
		case rslt := <-r.resCh:
			return rslt, nil
		case err := <-r.errCh:
			return 0, err
		default:
			continue
		}
	}

}

```

Senderの実装

```go:
type Sender struct {
	errCh  chan<- error
	resCh  chan<- Result
	refCtr *ReferenceCounter
}

func (r Receiver) NewSender() Sender {
   // 作成されたら参照数を増やす
	r.refCtr.Increment()
	return Sender{
		errCh:  r.errCh,
		resCh:  r.resCh,
		refCtr: r.refCtr,
	}
}

// 終了したら参照数を減らす
func (s Sender) Close() {
	s.refCtr.Decrement()
}

func (s Sender) SendError(err error) {
	s.errCh <- err
}

func (s Sender) SendResult(res Result) {
	s.resCh <- res
}

```

上記の使用部

```go:

func RunParallele(ctx context.Context, timeout time.Duration, f func(ctx context.Context, param int, sender Sender)) ([]Result, error) {
	r := NewReceiver(ctx)
	defer r.Close()

	for i := 0; i < 10; i++ {
		go f(ctx, i, r.NewSender())
	}

	var rs []Result
	for {
		rslt, err := r.Receive()
		if err != nil {
			return nil, err
		}

		rs = append(rs, rslt)
		if len(rs) == 10 {
			return rs, nil
		}
	}
}

func FailedFunc(ctx context.Context, param int, sender Sender) {
	defer sender.Close()
	sender.SendError(errors.New("failed"))
}

func SucceededFunc(ctx context.Context, param int, sender Sender) {
	defer sender.Close()
	sender.SendResult(Result(param))
}

func TimedOutFunc(ctx context.Context, param int, sender Sender) {
	defer sender.Close()
	for {
		time.Sleep(time.Millisecond)
	}
}

func main() {
	for _, f := range []func(context.Context, int, Sender){
		SucceededFunc, // => [9 0 1 2 3 4 5 6 7 8]
		FailedFunc,    // => failed
		TimedOutFunc,  // => context deadline exceeded
	} {
		ctx := context.Background()
		rs, err := RunParallele(ctx, timeout, f)
		if err != nil {
			fmt.Println(err)
		} else {
			fmt.Println(rs)
		}
	}
}
```

# 私見
このパターンでchannelをちゃんとcloseしたいと思うことはそもそもアンチパターンなのかもしれない。
mutexを使いはじめたときには設計が狂っていることが多いので、おそらくあるべき姿は

- 「channelを発行し、Senderを用いるgo routineを複数発行し、終了したらclose」をあまり柔軟に（例えば関数が呼ばれるたびに）行うべきではない
- 可能ならばできるだけchannelをcloseせず、再利用する

になるかと思った。