Goの複数のgoroutineに対して、一斉ブロードキャストを行いたい

Question

前提・実現したいこと

現在、Golangを使ってGIS系の沢山の人間や自動車が動き回るシミュレータを作成しております。

具体的にはgoroutineの中でstructで定義したオブジェクトを大量に発生させて、それらをバラバラに動かすことを考えております。

時々、そのすべてのgoroutineの中のオブジェクトに対して、main()より指示を出したいと思っています(例えば、『現在の位置情報を私(main())に教えろ』というようなもの)。

そこで、単一のchannelを使ったブロードキャストのようなことを試みたいと思っています。
複数のgoroutineに同時に指示が出せるのであれば、手段は問いません。

何卒ご教示頂けますよう、よろしくお願い申し上げます。

発生している問題・エラーメッセージ

これについては、(下記のコードのように)goroutineの数だけchannelを作れば実現可能であることは分かったのですが、問題点が2つほどあります。
(1)作成するgoroutineの数が大きいこと(数千から数万)
(2)goroutineは不定期に発生して、(役目を終えたら)自動的に消滅する(させる)
となっております。

該当のソースコード

package main

import (
	"fmt"
	"sync"
	"time"
)

type BUS struct {
	number int
}

func bus_func(i int, wg *sync.WaitGroup, ch1 chan int, ch2 chan int) {
	defer wg.Done()

	t := time.NewTicker(1 * time.Second) // 1秒おきに通知

	bus := BUS{}
	bus.number = i

	pointer_bus := &bus

	fmt.Println(pointer_bus)

	// 待受
	for {
		select {
		case v := <-ch1: // 受信
			if v == -1 { //終了コードの受信
				return // スレッドを自滅させる
			} else {
				fmt.Println(v)
			}
		case <-t.C: // 送信 (1秒まったら、むりやりこっちにくる)
			ch2 <- i + 100
		}
	}

}

//var c = make([]chan int, 3) // 間違い
// var c []chan int  これは実行時にエラーとなる

var c1 [3]chan int // チャネルをグローバル変数で置く方法の、現時点での正解
var c2 [3]chan int // チャネルをグローバル変数で置く方法の、現時点での正解

func main() {
	var wg sync.WaitGroup
	defer wg.Wait()

	//c := make(chan int)

	// バスエージェントの生成 3台
	for i := 0; i < 3; i++ {
		wg.Add(1)

		c1[i] = make(chan int)
		c2[i] = make(chan int)
		go bus_func(i, &wg, c1[i], c2[i])
	}

	// バスエージェントにメッセージを送る

	c1[0] <- 50
	c1[1] <- 30
	c1[2] <- 10

	c1[0] <- 50
	c1[1] <- 30
	c1[2] <- 10

	c1[0] <- 50
	c1[1] <- 30
	c1[2] <- 10

	fmt.Println(<-c2[0])
	fmt.Println(<-c2[1])
	fmt.Println(<-c2[2])

	c1[0] <- -1 // スレッド終了コードの送信
	c1[1] <- -1 // スレッド終了コードの送信
	c1[2] <- -1 // スレッド終了コードの送信

}

試したこと

上記に記載するサンプルプログラムでは、発生させたgoroutineとの間で個別にchannelを作る必要があり、goroutineの管理(発生から消滅まで)をしなければなりません。

(当初、『これくらいのことは簡単に見つかるだろう』と思っていたのですが、昨日1日かけて発見することができませんでした)

補足情報（FW/ツールのバージョンなど）

go version go1.17 windows/amd64

Answer 1

いっせいブロードキャストをするのに一般的に使われているのはPubSubと呼ばれる方式です。

サブスクライブを複数あらかじめおこなっておき、パブリッシュでメッセージを送ると複数のサブスクライブ先に同じメッセージを配信するというものです。

おそらくこの方式は発見済みで、想像するに複数のサブスクライブ先をループで巡って複数回送信することでブロードキャスト相当を実現するのではなく、もっと真にブロードキャストしたいということが質問者さんの意図なのかなと。

そういうものを実現するのに「sync.Cond」という標準ライブラリ機能があります。
これの活用方法は実はちゃんとした実例が見つけにくいです。たいてい前者のやり方で済ましてしまっているのと、sync.Condの挙動は若干わかりづらいです。

すこし解説は端折りますが、以下のように記述することで実現できると思います。

ポイントは

• タイミングだけをsync.CondのBroadcastで伝える
• 複数のタスクには共有メモリを通して渡したいメッセージを伝えます
• 送る方も受ける方も排他ロックを併用するのがCondの使い方でロック期間であれば共有メモリをコンフリクトなくアクセスできます

解答例

package main

import (
	"fmt"
	"log"
	"sync"
	"time"
)

type BroadCaster struct {
	cond *sync.Cond
	id   int64
	msg  string
}

func (bc *BroadCaster) Send(msg string) {
	bc.cond.L.Lock()
	defer bc.cond.L.Unlock()
	bc.id++
	bc.msg = msg
	bc.cond.Broadcast()
}

func (bc *BroadCaster) Recv(last int64) (int64, string) {
	bc.cond.L.Lock()
	defer bc.cond.L.Unlock()
	for bc.id == last {
		bc.cond.Wait()
	}
	return bc.id, bc.msg
}

var (
	broadcaster = &BroadCaster{
		cond: sync.NewCond(&sync.Mutex{}),
	}
)

func task(i int) {
	log.Println("task:", i, " start")
	defer log.Println("task:", i, " stop")
	last := int64(0)
	for {
		id, msg := broadcaster.Recv(last)
		last = id
		log.Println("task:", i, msg)
	}
}

func main() {
	for i := 0; i < 3; i++ {
		go task(i)
	}
	for i := 0; i < 3; i++ {
		time.Sleep(1 * time.Second)
		broadcaster.Send(fmt.Sprintf("hello, world: %d", i))
	}
	time.Sleep(1 * time.Second)
}

補足：

• この方法はPubSubにくらべ、共有メモリをすべてのgoroutineタスクに伝播したかどうかを保証する仕組みがないです
• つまり、この方法は「低頻度のイベントを大量のタスクに配信」するか、もしくは「最新の値さえ受け取れればOK」という用途向けです。

Comments

[TomMurphy]

お返事が遅れまして申し訳ございません(メールが変なところに飛ばされていました)。
コードまで展開して頂き、本当に嬉しいです。早速試させて頂きます。
引き続き、ご指導頂けましたら幸いでございます。