feat: prep for the demo

- add SIM_MAX_TOPICS
- add maxPeers to config.go
- fix readme steps
- fix AddPeerEnter
- fix tracing in test-discovery-states
- add reset-env

Author: pancsta

README.md

@@ -126,14 +126,19 @@ Common setup for all the scenarios:
 
 - run `task start-env`
 - run `task start-am-dbg`
-- visit Grafana at http://localhost:3000/dashboard/import 
-  - import `assets/sim-grafana.dashboard.json`
-  - or `task gen-grafana-dashboard` and provide machine IDs via `GRAFANA_IDS`
-- switch TTY 
+- set up Grafana
+  - visit http://localhost:3000/connections/datasources/new
+    - login admin/admin, skip
+    - add a Prometheus data source http://prometheus:9090
+  - visit http://localhost:3000/dashboard/import
+    - login admin/admin, skip
+    - paste import [`assets/sim-grafana.dashboard.json`](https://raw.githubusercontent.com/pancsta/go-libp2p-pubsub-benchmark/main/assets/sim-grafana.dashboard.json)
+      - or `task gen-grafana-dashboard` and provide machine IDs via `GRAFANA_IDS`
+- switch TTY
 - run `task start-sim`
 - visit the first TTY for am-dbg history
   - requires `SIM_AM_DEBUG=1`
-- visit the Grafana dashboard at http://localhost:3000/d/f4ac0cf1-0f3d-4b41-9e04-d6b0e68b49bc/?orgId=1&refresh=5s&from=now-5m&to=now
+- visit the Grafana dashboard
   - requires `SIM_METRICS=1`
 
 ## Benchmarking a custom implementation
@@ -156,11 +161,16 @@ Common setup for all the scenarios:
 
 - replace `sim.env` and `internal/sim/config.go` with `sim.yml`
   - allow for env overrides
+- abstract discovery
+  - fix DHT lag
+  - enable mDNS
 - reimplement more things as machines
   - gossipsub, score, topic
 - fix benchmark failures
+- fix pubsub "Topic not found" err
 - replace `bench.env` and all the benchmark repo defs with `bench.yml`
   - allow for env overrides
 - bind [universal connectivity UI](https://github.com/libp2p/universal-connectivity/tree/main/go-peer) to see the traffic
   - needs a list of topics, doesn't need a msg form
 - subtract `time.Sleep()` from benchmark results
+- "delivered/missed msgs" metrics should also be averaged, not just counters

Taskfile.yml

@@ -21,13 +21,13 @@ tasks:
 
   test-discovery-states:
     cmds:
-      - pushd $REPOS_PATH/states; go test -test.run "^\QTestSimpleDiscovery" -v *.go
-    dotenv: ['nodebug.env']
+      - pushd $REPOS_PATH/states;
+        go test -test.run "^\QTestSimpleDiscovery" -v *.go
 
   test-discovery-origin:
     cmds:
-      - pushd $REPOS_PATH/origin; go test -test.run "^\QTestSimpleDiscovery" -v *.go
-    dotenv: ['nodebug.env']
+      - pushd $REPOS_PATH/origin;
+        go test -test.run "^\QTestSimpleDiscovery" -v *.go
 
   bench-pdf:
     cmds:
@@ -121,19 +121,24 @@ tasks:
   start-env:
     desc: Starts the docker environment (prometheus, grafana, pushgateway, jaeger)
     cmds:
-      - |
-        docker-compose --file config/docker-compose.yml \
-          --project-name go-libp2p-pubsub-benchmark \
+      - docker-compose --file config/docker-compose.yml
+          --project-name go-libp2p-pubsub-benchmark
           up -d
 
   stop-env:
     desc: Stops the docker environment (prometheus, grafana, pushgateway, jaeger)
     cmds:
-      - |
-        docker-compose --file config/docker-compose.yml \
-          --project-name go-libp2p-pubsub-benchmark \
+      - docker-compose --file config/docker-compose.yml
+          --project-name go-libp2p-pubsub-benchmark
           down
 
+  reset-env:
+    desc: Stops the docker environment (prometheus, grafana, pushgateway, jaeger)
+    cmds:
+      - docker-compose --file config/docker-compose.yml
+          --project-name go-libp2p-pubsub-benchmark
+          down -v
+
   gen-grafana-dashboard:
     desc: Generate a Grafana dashboard for the simulator and first 3 hosts
     dir: scripts

cmd/sim/cmd_sim.go

@@ -18,7 +18,7 @@ import (
 
 func main() {
 	// load .env
-	_ = godotenv.Load("sim.env", ".env")
+	_ = godotenv.Load(".env", "sim.env")
 
 	// init the simulator
 	ctx := context.Background()
@@ -29,14 +29,20 @@ func main() {
 	}
 
 	// parse max peers
-	maxPeers := 30
 	if os.Getenv("SIM_MAX_PEERS") != "" {
 		v, err := strconv.Atoi(os.Getenv("SIM_MAX_PEERS"))
 		if err == nil {
-			maxPeers = v
+			s.MaxPeers = v
+		}
+	}
+
+	// parse max topics
+	if os.Getenv("SIM_MAX_TOPICS") != "" {
+		v, err := strconv.Atoi(os.Getenv("SIM_MAX_TOPICS"))
+		if err == nil {
+			s.MaxTopics = v
 		}
 	}
-	s.MaxPeers = maxPeers
 
 	// parse duration
 	duration := 5 * time.Minute

go.mod

@@ -1,16 +1,14 @@
 module github.com/pancsta/go-libp2p-pubsub-benchmark
 
-replace github.com/pancsta/go-libp2p-pubsub => ../go-libp2p-pubsub
-
 go 1.22.3
 
 require (
 	github.com/brianvoe/gofakeit/v7 v7.0.2
 	github.com/joho/godotenv v1.5.1
 	github.com/libp2p/go-libp2p v0.35.0
 	github.com/libp2p/go-libp2p-kad-dht v0.25.2
-	github.com/pancsta/asyncmachine-go v0.6.0
-	github.com/pancsta/go-libp2p-pubsub v0.11.3
+	github.com/pancsta/asyncmachine-go v0.6.2
+	github.com/pancsta/go-libp2p-pubsub v0.11.4
 	github.com/prometheus/client_golang v1.19.1
 	github.com/samber/lo v1.39.0
 	github.com/sourcegraph/conc v0.3.0

go.sum

@@ -319,6 +319,9 @@ github.com/opentracing/opentracing-go v1.2.0/go.mod h1:GxEUsuufX4nBwe+T+Wl9TAgYr
 github.com/openzipkin/zipkin-go v0.1.1/go.mod h1:NtoC/o8u3JlF1lSlyPNswIbeQH9bJTmOf0Erfk+hxe8=
 github.com/pancsta/asyncmachine-go v0.6.0 h1:5gqvELRyjULDqOSoMFZ2K0KOoYzVCL28/UQUIByPl04=
 github.com/pancsta/asyncmachine-go v0.6.0/go.mod h1:hoG7VB315eqgnRvVfZQQcw7B7CrJMkwqNNHH8N+7CGs=
+github.com/pancsta/asyncmachine-go v0.6.1/go.mod h1:hoG7VB315eqgnRvVfZQQcw7B7CrJMkwqNNHH8N+7CGs=
+github.com/pancsta/asyncmachine-go v0.6.2/go.mod h1:hoG7VB315eqgnRvVfZQQcw7B7CrJMkwqNNHH8N+7CGs=
+github.com/pancsta/go-libp2p-pubsub v0.11.4/go.mod h1:owZ/Fi/draxxvWwljUejC9bTfHwdKqOWd16JirWs058=
 github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58 h1:onHthvaw9LFnH4t2DcNVpwGmV9E1BkGknEliJkfwQj0=
 github.com/pbnjay/memory v0.0.0-20210728143218-7b4eea64cf58/go.mod h1:DXv8WO4yhMYhSNPKjeNKa5WY9YCIEBRbNzFFPJbWO6Y=
 github.com/pion/datachannel v1.5.6 h1:1IxKJntfSlYkpUj8LlYRSWpYiTTC02nUrOE8T3DqGeg=

internal/sim/config.go

@@ -28,6 +28,7 @@ var (
 
 	maxHistoryEntries = 1000
 	maxTopics         = 20
+	maxPeers          = 30
 	maxTopicPerPeer   = 15
 	maxPeersPerTopic  = 10
 	maxFriendsPerPeer = 10

internal/sim/sim.go

@@ -56,6 +56,7 @@ type Sim struct {
 	promUrl       string
 	discServer    *mockDiscoveryServer
 	MaxPeers      int
+	MaxTopics     int
 }
 
 func NewSim(ctx context.Context, exportMetrics, debugAM bool) (*Sim, error) {
@@ -67,7 +68,9 @@ func NewSim(ctx context.Context, exportMetrics, debugAM bool) (*Sim, error) {
 		exportMetrics: exportMetrics,
 		promUrl:       promUrl,
 		debugAM:       debugAM,
-		MaxPeers:      30,
+		// config to override
+		MaxPeers:  maxPeers,
+		MaxTopics: maxTopics,
 	}
 
 	opts := &am.Opts{}
@@ -179,6 +182,10 @@ func (s *Sim) StartState(e *am.Event) {
 	//	return
 	// }
 
+	if printMetrics {
+		// TODO print config
+	}
+
 	s.discServer = newMockDiscoveryServer()
 
 	// init the initial simulation
@@ -256,15 +263,14 @@ func (s *Sim) HeartbeatState(e *am.Event) {
 }
 
 func (s *Sim) AddPeerEnter(e *am.Event) bool {
-	_, ok := e.Args["amount"].(int)
-	return ok && len(s.peers) < s.MaxPeers
+	return len(s.peers) < s.MaxPeers
 }
 
 func (s *Sim) AddPeerState(e *am.Event) {
 	s.Mach.Remove1(ss.AddPeer, nil)
 
-	amount := e.Args["amount"].(int)
-	if amount == 0 {
+	amount, ok := e.Args["amount"].(int)
+	if !ok {
 		amount = 1
 	}
 
@@ -329,7 +335,7 @@ func (s *Sim) AddTopicEnter(e *am.Event) bool {
 	p := s.pickRandPeerCond(func(p *Peer) bool {
 		return len(p.simTopics) < maxTopicPerPeer
 	})
-	return len(s.topics) < maxTopics && p != nil
+	return len(s.topics) < s.MaxTopics && p != nil
 }
 
 func (s *Sim) AddTopicState(e *am.Event) {
@@ -342,7 +348,7 @@ func (s *Sim) AddTopicState(e *am.Event) {
 
 	// add N topics
 	for i := 0; i < amount; i++ {
-		if len(s.topics) >= maxTopics {
+		if len(s.topics) >= s.MaxTopics {
 			break
 		}
 
@@ -440,7 +446,6 @@ func (s *Sim) AddRandomFriendState(e *am.Event) {
 		p2.simTopicJoined[randTopic] = time.Now()
 		p2.mach.Add1(ssp.JoiningTopic, am.A{"Topic.id": randTopic})
 	}
-	s.Mach.Log("Added friend: %s + %s", p1.id, p2.id)
 }
 
 // GCState does various periodic cleanups.
@@ -456,6 +461,7 @@ func (s *Sim) GCState(e *am.Event) {
 				s.Mach.Log("Friend not found: %s", fid)
 				continue
 			}
+
 			if len(lo.Intersect(p1.simTopics, p2.simTopics)) > 0 {
 				newFriends = append(newFriends, fid)
 			} else {
@@ -464,6 +470,7 @@ func (s *Sim) GCState(e *am.Event) {
 				s.Mach.Log("Removed friend: %s - %s", p1.id, p2.id)
 			}
 		}
+
 		p1.simFriends = newFriends
 	}
 
@@ -825,7 +832,7 @@ func (s *Sim) topicPeerChange(topic string, amount int) {
 	s.Mach.Eval("topicPeerChange", func() {
 		t, ok := s.topics[topic]
 		if !ok {
-			s.Mach.Log("Error: topic not found", topic)
+			s.Mach.Log("Error: topic not found %s", topic)
 			return
 		}
 		t.peersCount += amount
@@ -842,7 +849,7 @@ func (s *Sim) topicMsgs(topic string, amount int) {
 	s.Mach.Eval("topicPeerChange", func() {
 		t, ok := s.topics[topic]
 		if !ok {
-			s.Mach.Log("Error: topic not found", topic)
+			s.Mach.Log("Error: topic not found %s", topic)
 			return
 		}
 		t.msgsLog = append(t.msgsLog, TopicMsgsLog{time.Now(), amount})

pkg/psmon/psmon.go

@@ -58,7 +58,7 @@ func NewPSMon(endpoint string, newOtelProvider NewOtelProviderFn) *PSMon {
 	isSim := strings.HasSuffix(os.Args[0], "sim") ||
 		strings.HasSuffix(os.Args[0], "sim_go")
 	if !isSim {
-		_ = godotenv.Load("bench.env", ".env")
+		_ = godotenv.Load(".env", "bench.env")
 	}
 
 	// load .env