golangのgoroutineで作ったエージェントの動きを同期させるプログラム

2023年10月2日

今迄、エージェントプログラムでは、エージェントの中に内部時計(正確にはSleep())を組み込んで、エージェント間の時刻同期については、無視してきたのですが、交通シミュレーションに時刻表を入れる必要になって、無視できなくなってきました。

今や、私の十八番となったredisを使ったgolangのgoroutineで作った複数エージェントへのメッセージのブロードキャストを使って、時刻同期をするサンプルプログラムを作ってみました。

BaseClock()から、所定時間間隔で、シミュレーション共通の時刻をブロードキャストして、person()のエージェントで、その時刻情報を割り込み受信させることにしました。

ブロードキャストによる遅延や割り込み取りこぼしが心配ではありますが、これまでの経験上redisのブロードキャストは十分に高速なようなので、多分大丈夫だと思います。

// C:\Users\ebata\tomioka3B\src\others\main26.go

package main

import (
	"encoding/json"
	"fmt"
	"sync"
	"time"

	"github.com/gomodule/redigo/redis"
)

var Clock1 chan interface{}

type Clock_Info struct {
	VirtualTime time.Time
	RealTime    time.Time
}

func BaseClock(wg *sync.WaitGroup) {
	defer wg.Done()

	// 接続
	conn, err := redis.Dial("tcp", "localhost:6379")
	if err != nil {
		panic(err)
	}
	defer conn.Close()

	// スタート時刻を指定
	startTime := time.Date(2023, 10, 1, 7, 0, 0, 0, time.UTC)

	// 1秒値を保持する変数
	seconds := 0

	var ci Clock_Info

	// ループを開始
	for {
		// 現在の時刻を計算
		ci.VirtualTime = startTime.Add(time.Duration(seconds) * time.Second)
		ci.RealTime = time.Now()

		// 現在の時刻を表示
		fmt.Println("シミュレータの時刻:", ci.VirtualTime.Format("2006/01/02 15:04:05"))
		fmt.Println("現在の時刻:", ci.RealTime.Format("2006/01/02 15:04:05")) // "2006/01/02 15:04:05"はフォーマットの形を真似るもので、内容に意味なし

		// パブリッシュ
		json_ci, _ := json.Marshal(ci)
		r, err := redis.Int(conn.Do("PUBLISH", "ClockInfo_1", json_ci))
		if err != nil {
			panic(err)
		}
		fmt.Println(r)

		// 5秒待つ (実際は、0.05秒くらいだが、確認用に長くしている)
		time.Sleep(5000 * time.Millisecond)

		// 1秒値を増加させる
		seconds++
	}
}

func person(person_num int, wg *sync.WaitGroup) {
	defer wg.Done()
	// 接続
	conn, err := redis.Dial("tcp", "localhost:6379")
	if err != nil {
		panic(err)
	}
	defer conn.Close()

	psc := redis.PubSubConn{Conn: conn}
	psc.Subscribe("ClockInfo_1")

	for {
		ci := new(Clock_Info)
		switch v := psc.Receive().(type) {
		case redis.Message:
			_ = json.Unmarshal(v.Data, &ci)
			fmt.Println("Person:", person_num, "VirtualTime:", ci.VirtualTime)
			fmt.Println("Person:", person_num, "RealTime:", ci.RealTime)

		case redis.Subscription:
			fmt.Printf("%s: %s %d\n", v.Channel, v.Kind, v.Count)

		case error:
			return
		}
	}
}

func main() {

	wg := sync.WaitGroup{}

	wg.Add(1)
	go BaseClock(&wg)

	for i := 0; i < 10; i++ { // 10人
		wg.Add(1)
		go person(i, &wg)
	}

	wg.Wait()
	fmt.Println("end of ... main()")
}