diff --git a/.gitignore b/.gitignore index 5e2babe..aa7fabf 100644 --- a/.gitignore +++ b/.gitignore @@ -1,6 +1,6 @@ # Binaries *-sim -gossipsub +# gossipsub gossipsub/gossipsub topology-gen topology/gen/topology-gen @@ -11,7 +11,6 @@ shadow.data/ topology.json gossipsub/topology.json shadow-gossipsub.yaml -message_propagation.png # Go build artifacts *.out diff --git a/Makefile b/Makefile index 6194eed..00ec343 100644 --- a/Makefile +++ b/Makefile @@ -4,9 +4,12 @@ # Default parameters NODE_COUNT ?= 10 -MSG_SIZE ?= 256 +MESH_NODE_COUNT ?= 6 +MESH_ATTESTER_COUNT ?= 2 +NON_MESH_ATTESTER_COUNT ?= 2 TOPOLOGY_TYPE ?= random-regular PEER_COUNT ?= 4 +NON_MESH_NODE_PEER_COUNT ?= 4 BRANCHING ?= 2 PROGRESS ?= false LOG_LEVEL ?= info @@ -15,6 +18,7 @@ LOG_LEVEL ?= info TOPOLOGY_GEN_DIR = topology/gen GOSSIPSUB_DIR = gossipsub TOPOLOGY_FILE = topology.json +SIMCONFIG_FILE = simconfig.yaml # Check dependencies check-deps: @@ -33,7 +37,7 @@ build-topology-gen: # Generate topology file generate-topology: build-topology-gen @echo "Generating $(TOPOLOGY_TYPE) topology with $(NODE_COUNT) nodes..." - $(TOPOLOGY_GEN_DIR)/topology-gen -type $(TOPOLOGY_TYPE) -nodes $(NODE_COUNT) -degree $(PEER_COUNT) -branching $(BRANCHING) -output $(TOPOLOGY_FILE) + $(TOPOLOGY_GEN_DIR)/topology-gen -type $(TOPOLOGY_TYPE) -nodes $(NODE_COUNT) -degree $(PEER_COUNT) -non-mesh-node-degree $(NON_MESH_NODE_PEER_COUNT) -branching $(BRANCHING) -mesh-nodes $(MESH_NODE_COUNT) -mesh-attester-count $(MESH_ATTESTER_COUNT) -non-mesh-attester-count $(NON_MESH_ATTESTER_COUNT) -output $(TOPOLOGY_FILE) --simconfig-file $(SIMCONFIG_FILE) @test -f $(TOPOLOGY_FILE) || (echo "Topology generation failed" && exit 1) @echo "Topology generated: $(TOPOLOGY_FILE)" @@ -46,14 +50,20 @@ build: check-deps # Generate network graph and Shadow configuration generate-config: generate-topology - @echo "Generating Shadow network configuration for $(NODE_COUNT) nodes..." - uv run network_graph.py $(NODE_COUNT) $(MSG_SIZE) $(TOPOLOGY_FILE) + @echo "Generating Shadow network configuration for $(NODE_COUNT) nodes with $(MESH_NODE_COUNT) mesh nodes, $(MESH_ATTESTER_COUNT) mesh attesters nodes, $(NON_MESH_ATTESTER_COUNT) non mesh attesters nodes..." + uv run network_graph.py $(NODE_COUNT) $(TOPOLOGY_FILE) $(SIMCONFIG_FILE) @test -f shadow-gossipsub.yaml && test -f graph.gml || (echo "Config generation failed" && exit 1) @echo "Configuration generated" +generate-config-only: + @echo "Generating Shadow network configuration for $(NODE_COUNT) nodes with $(MESH_NODE_COUNT) mesh nodes, $(MESH_ATTESTER_COUNT) mesh attesters nodes, $(NON_MESH_ATTESTER_COUNT) non mesh attesters nodes..." + uv run network_graph.py $(NODE_COUNT) $(TOPOLOGY_FILE) $(SIMCONFIG_FILE) + @test -f shadow-gossipsub.yaml && test -f graph.gml || (echo "Config generation failed" && exit 1) + @echo "Configuration generated" + # Run the complete Shadow simulation run-sim: build generate-config - @echo "Starting GossipSub Shadow simulation ($(NODE_COUNT) nodes, $(MSG_SIZE) byte message)..." + @echo "Starting GossipSub Shadow simulation ($(NODE_COUNT) nodes, $(MESH_NODE_COUNT) mesh nodes, $(MESH_ATTESTER_COUNT) mesh attesters nodes, $(NON_MESH_ATTESTER_COUNT) non mesh attesters nodes, $(MSG_SIZE) byte message)..." @rm -rf shadow.data/ shadow --progress $(PROGRESS) shadow-gossipsub.yaml @echo "GossipSub simulation completed" @@ -61,13 +71,13 @@ run-sim: build generate-config # Test simulation results test: @echo "Testing GossipSub simulation results..." - uv run test_results.py $(NODE_COUNT) + uv run test_results.py $(NODE_COUNT) $(TOPOLOGY_FILE) # Plot message propagation plot: @echo "Plotting message propagation..." - uv run plot_propagation.py $(NODE_COUNT) - @test -f message_propagation.png && echo "Plot generated: message_propagation.png" || echo "Plot generation failed" + uv run plot_propagation.py $(NODE_COUNT) --topology-file $(TOPOLOGY_FILE) --peer-count $(PEER_COUNT) --non-mesh-node-peer-count $(NON_MESH_NODE_PEER_COUNT) --simconfig-file $(SIMCONFIG_FILE) -o new-latencies-plots/30slots_batch_50ms_node$(NODE_COUNT)_mesh_nodes$(MESH_NODE_COUNT)_mesh_attesters$(MESH_ATTESTER_COUNT)_non_mesh_attesters$(NON_MESH_ATTESTER_COUNT).png + @test -f new-latencies-plots/30slots_batch_50ms_node$(NODE_COUNT)_mesh_nodes$(MESH_NODE_COUNT)_mesh_attesters$(MESH_ATTESTER_COUNT)_non_mesh_attesters$(NON_MESH_ATTESTER_COUNT).png && echo "Plot generated: new-latencies-plots/30slots_batch_50ms_node$(NODE_COUNT)_mesh_nodes$(MESH_NODE_COUNT)_mesh_attesters$(MESH_ATTESTER_COUNT)_non_mesh_attesters$(NON_MESH_ATTESTER_COUNT).png" || echo "Plot generation failed" # Clean build artifacts and simulation results clean: @@ -103,20 +113,28 @@ help: @echo " help - Show this help message" @echo "" @echo "Configuration variables:" - @echo " NODE_COUNT - Number of nodes (default: $(NODE_COUNT))" - @echo " MSG_SIZE - Message size in bytes (default: $(MSG_SIZE))" - @echo " TOPOLOGY_TYPE - Topology type: mesh, tree, random-regular (default: $(TOPOLOGY_TYPE))" - @echo " PEER_COUNT - Peer count for random-regular (default: $(PEER_COUNT))" - @echo " BRANCHING - Branching factor for tree (default: $(BRANCHING))" - @echo " PROGRESS - Show Shadow progress bar (default: $(PROGRESS))" - @echo " LOG_LEVEL - Log level (default: $(LOG_LEVEL))" + @echo " NODE_COUNT - Number of nodes (default: $(NODE_COUNT))" + @echo " MESH_NODE_COUNT - Number of mesh nodes (default: $(MESH_NODE_COUNT))" + @echo " MESH_ATTESTER_COUNT - Number of mesh attesters nodes (default: $(MESH_ATTESTER_COUNT))" + @echo " NON_MESH_ATTESTER_COUNT - Number of non mesh attesters nodes (default: $(NON_MESH_ATTESTER_COUNT))" + @echo " MSG_SIZE - Message size in bytes (default: $(MSG_SIZE))" + @echo " TOPOLOGY_TYPE - Topology type: mesh, tree, random-regular (default: $(TOPOLOGY_TYPE))" + @echo " PEER_COUNT - Peer count for random-regular (default: $(PEER_COUNT))" + @echo " BRANCHING - Branching factor for tree (default: $(BRANCHING))" + @echo " PROGRESS - Show Shadow progress bar (default: $(PROGRESS))" + @echo " LOG_LEVEL - Log level (default: $(LOG_LEVEL))" + @echo " BATCH_INTERVAL - BLS batch interval in milliseconds (default: $(BATCH_INTERVAL))" + @echo " BATCH_VERIFIER_TIME - BLS batch verifier time in microseconds (default: $(BATCH_VERIFIER_TIME))" @echo "" @echo "Examples:" @echo " make all # Run simulation, test, and plot (random-regular)" @echo " make run-sim NODE_COUNT=20 MSG_SIZE=512 # Custom parameters" + @echo " make run-sim NODE_COUNT=20 NODES_TO_PUBLISH=5 # 20 nodes, only 5 publish" + @echo " make run-sim NODE_COUNT=20 NODES_TO_PUBLISH=5 NODES_TO_SUBSCRIBE=15 # 5 publishers, 15 subscribers" @echo " make run-sim TOPOLOGY_TYPE=mesh NODE_COUNT=10 # Mesh topology" @echo " make run-sim TOPOLOGY_TYPE=tree NODE_COUNT=31 BRANCHING=2 # Tree topology" @echo " make run-sim TOPOLOGY_TYPE=random-regular PEER_COUNT=6 # Random-regular with 6 peers" @echo " make run-sim PROGRESS=true # Run with progress bar" + @echo " make run-sim BATCH_INTERVAL=2 BATCH_VERIFIER_TIME=2000 # Custom batch parameters" @echo " make test # Test existing simulation results" @echo " make plot NODE_COUNT=10 # Plot message propagation" diff --git a/go.mod b/go.mod index d360001..147bcb5 100644 --- a/go.mod +++ b/go.mod @@ -9,14 +9,18 @@ require ( github.com/libp2p/go-libp2p v0.41.1 github.com/libp2p/go-libp2p-pubsub v0.15.0 github.com/multiformats/go-multiaddr v0.15.0 + github.com/stretchr/testify v1.10.0 ) +require github.com/libp2p/go-mplex v0.7.0 // indirect + require ( github.com/benbjohnson/clock v1.3.5 // indirect github.com/beorn7/perks v1.0.1 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/containerd/cgroups v1.1.0 // indirect github.com/coreos/go-systemd/v22 v22.5.0 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect github.com/davidlazar/go-crypto v0.0.0-20200604182044-b73af7476f6c // indirect github.com/decred/dcrd/dcrec/secp256k1/v4 v4.4.0 // indirect github.com/docker/go-units v0.5.0 // indirect @@ -41,6 +45,7 @@ require ( github.com/libp2p/go-buffer-pool v0.1.0 // indirect github.com/libp2p/go-flow-metrics v0.2.0 // indirect github.com/libp2p/go-libp2p-asn-util v0.4.1 // indirect + github.com/libp2p/go-libp2p-mplex v0.11.0 github.com/libp2p/go-msgio v0.3.0 // indirect github.com/libp2p/go-netroute v0.2.2 // indirect github.com/libp2p/go-reuseport v0.4.0 // indirect @@ -85,6 +90,7 @@ require ( github.com/pion/turn/v4 v4.0.0 // indirect github.com/pion/webrtc/v4 v4.0.10 // indirect github.com/pkg/errors v0.9.1 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect github.com/prometheus/client_golang v1.21.1 // indirect github.com/prometheus/client_model v0.6.1 // indirect github.com/prometheus/common v0.62.0 // indirect @@ -109,5 +115,6 @@ require ( golang.org/x/text v0.22.0 // indirect golang.org/x/tools v0.30.0 // indirect google.golang.org/protobuf v1.36.5 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect lukechampine.com/blake3 v1.4.0 // indirect ) diff --git a/go.sum b/go.sum index 4dc26cf..819a372 100644 --- a/go.sum +++ b/go.sum @@ -136,10 +136,14 @@ github.com/libp2p/go-libp2p v0.41.1 h1:8ecNQVT5ev/jqALTvisSJeVNvXYJyK4NhQx1nNRXQ github.com/libp2p/go-libp2p v0.41.1/go.mod h1:DcGTovJzQl/I7HMrby5ZRjeD0kQkGiy+9w6aEkSZpRI= github.com/libp2p/go-libp2p-asn-util v0.4.1 h1:xqL7++IKD9TBFMgnLPZR6/6iYhawHKHl950SO9L6n94= github.com/libp2p/go-libp2p-asn-util v0.4.1/go.mod h1:d/NI6XZ9qxw67b4e+NgpQexCIiFYJjErASrYW4PFDN8= +github.com/libp2p/go-libp2p-mplex v0.11.0 h1:0vwpLXRSfkTzshEjETIEgJaVxXvg+orbxYoIb3Ty5qM= +github.com/libp2p/go-libp2p-mplex v0.11.0/go.mod h1:QrsdNY3lzjpdo9V1goJfPb0O65Nms0sUR8CDAO18f6k= github.com/libp2p/go-libp2p-pubsub v0.15.0 h1:cG7Cng2BT82WttmPFMi50gDNV+58K626m/wR00vGL1o= github.com/libp2p/go-libp2p-pubsub v0.15.0/go.mod h1:lr4oE8bFgQaifRcoc2uWhWWiK6tPdOEKpUuR408GFN4= github.com/libp2p/go-libp2p-testing v0.12.0 h1:EPvBb4kKMWO29qP4mZGyhVzUyR25dvfUIK5WDu6iPUA= github.com/libp2p/go-libp2p-testing v0.12.0/go.mod h1:KcGDRXyN7sQCllucn1cOOS+Dmm7ujhfEyXQL5lvkcPg= +github.com/libp2p/go-mplex v0.7.0 h1:BDhFZdlk5tbr0oyFq/xv/NPGfjbnrsDam1EvutpBDbY= +github.com/libp2p/go-mplex v0.7.0/go.mod h1:rW8ThnRcYWft/Jb2jeORBmPd6xuG3dGxWN/W168L9EU= github.com/libp2p/go-msgio v0.3.0 h1:mf3Z8B1xcFN314sWX+2vOTShIE0Mmn2TXn3YCUQGNj0= github.com/libp2p/go-msgio v0.3.0/go.mod h1:nyRM819GmVaF9LX3l03RMh10QdOroF++NBbxAb0mmDM= github.com/libp2p/go-netroute v0.2.2 h1:Dejd8cQ47Qx2kRABg6lPwknU7+nBnFRpko45/fFPuZ8= diff --git a/gossipsub/main.go b/gossipsub/main.go index 43c49d4..709e9a5 100644 --- a/gossipsub/main.go +++ b/gossipsub/main.go @@ -2,31 +2,270 @@ package main import ( "context" + "crypto/sha256" + "encoding/hex" "flag" "fmt" "io" "log" + "math/rand" "net" + "slices" + "sync/atomic" "time" "github.com/libp2p/go-libp2p" + mplex "github.com/libp2p/go-libp2p-mplex" pubsub "github.com/libp2p/go-libp2p-pubsub" "github.com/libp2p/go-libp2p/core/crypto" "github.com/libp2p/go-libp2p/core/peer" + "github.com/libp2p/go-libp2p/core/protocol" + "github.com/libp2p/go-libp2p/p2p/security/noise" + libp2ptcp "github.com/libp2p/go-libp2p/p2p/transport/tcp" "github.com/multiformats/go-multiaddr" + "github.com/ethp2p/attsim/simconfig" "github.com/ethp2p/attsim/topology" + pubsubpb "github.com/libp2p/go-libp2p-pubsub/pb" + logging "github.com/ipfs/go-log/v2" ) +// CustomTracer implements pubsub.RawTracer to log RPC drops and undeliverable messages +type CustomTracer struct { + nodeID int +} + +// NewCustomTracer creates a new custom tracer +func NewCustomTracer(nodeID int) *CustomTracer { + return &CustomTracer{nodeID: nodeID} +} + +// AddPeer is invoked when a new peer is added +func (t *CustomTracer) AddPeer(p peer.ID, proto protocol.ID) {} + +// RemovePeer is invoked when a peer is removed +func (t *CustomTracer) RemovePeer(p peer.ID) {} + +// Join is invoked when a new topic is joined +func (t *CustomTracer) Join(topic string) {} + +// Leave is invoked when a topic is abandoned +func (t *CustomTracer) Leave(topic string) {} + +// Graft is invoked when a new peer is grafted on the mesh (gossipsub) +func (t *CustomTracer) Graft(p peer.ID, topic string) {} + +// Prune is invoked when a peer is pruned from the mesh (gossipsub) +func (t *CustomTracer) Prune(p peer.ID, topic string) {} + +// ValidateMessage is invoked when a message first enters the validation pipeline +func (t *CustomTracer) ValidateMessage(msg *pubsub.Message) {} + +// DeliverMessage is invoked when a message is delivered +func (t *CustomTracer) DeliverMessage(msg *pubsub.Message) {} + +// RejectMessage is invoked when a message is Rejected or Ignored +func (t *CustomTracer) RejectMessage(msg *pubsub.Message, reason string) {} + +// DuplicateMessage is invoked when a duplicate message is dropped +func (t *CustomTracer) DuplicateMessage(msg *pubsub.Message) { +} + +// ThrottlePeer is invoked when a peer is throttled by the peer gater +func (t *CustomTracer) ThrottlePeer(p peer.ID) { + log.Printf("[Node %d] PEER THROTTLED: %s", t.nodeID, p) +} + +// RecvRPC is invoked when an incoming RPC is received +func (t *CustomTracer) RecvRPC(rpc *pubsub.RPC) { +} + +// SendRPC is invoked when an RPC is sent +func (t *CustomTracer) SendRPC(rpc *pubsub.RPC, p peer.ID) { + // for _, msg := range rpc.GetPublish() { + // log.Printf("[Node %d] SENDING MESSAGE: %s with id: %d", t.nodeID, msg.Data, msg.Seqno) + // } +} + +// DropRPC is invoked when an RPC is dropped +func (t *CustomTracer) DropRPC(rpc *pubsub.RPC, p peer.ID) { + log.Printf("[Node %d] RPC DROPPED: to peer %s", t.nodeID, p) +} + +// UndeliverableMessage is invoked when a message is undeliverable +func (t *CustomTracer) UndeliverableMessage(msg *pubsub.Message) { + log.Printf("[Node %d] UNDELIVERABLE MESSAGE", t.nodeID) +} + +func GetMessageId(msg *pubsubpb.Message) string { + hash := sha256.Sum256([]byte(fmt.Sprintf("%s%d%s", *msg.Topic, len(*msg.Topic), string(msg.Data)))) + return hex.EncodeToString(hash[:])[:20] +} + +const ( + MeshNode = "mesh" + MeshAttesterNode = "mesh-attester" + NonMeshAttesterNode = "non-mesh-attester" + + PrysmClient = "prysm" + LighthouseClient = "lighthouse" +) + +// BLSBatchVerifier simulates a BLS signature batch verifier queue +type BLSBatchVerifier struct { + queue chan *ValidationRequest + ctx context.Context + cancel context.CancelFunc + batchVerifierTime time.Duration + batchIntervalTime time.Duration +} + +// ValidationRequest represents a message waiting for batch verification +type ValidationRequest struct { + Message *pubsub.Message + Callback chan pubsub.ValidationResult +} + +// NewBLSBatchVerifier creates a new batch verifier +func NewBLSBatchVerifier(batchVerifierTime time.Duration, batchIntervalTime time.Duration) *BLSBatchVerifier { + ctx, cancel := context.WithCancel(context.Background()) + + bv := &BLSBatchVerifier{ + queue: make(chan *ValidationRequest, 1000), // Buffer for high throughput + ctx: ctx, + cancel: cancel, + batchVerifierTime: batchVerifierTime, + batchIntervalTime: batchIntervalTime, + } + + // Start the batch processing goroutine + go bv.processBatches() + + return bv +} + +// processBatches processes messages in batches every 5ms +func (bv *BLSBatchVerifier) processBatches() { + ticker := time.NewTicker(bv.batchIntervalTime) + defer ticker.Stop() + + var batch []*ValidationRequest + + for { + select { + case <-bv.ctx.Done(): + return + case req := <-bv.queue: + batch = append(batch, req) + case <-ticker.C: + if len(batch) > 0 { + bv.processBatch(batch) + batch = batch[:0] // Reset slice but keep capacity + } + } + } +} + +// processBatch simulates batch verification taking 4.5ms +func (bv *BLSBatchVerifier) processBatch(batch []*ValidationRequest) { + if len(batch) == 0 { + return + } + + time.Sleep(bv.batchVerifierTime) + + // Send results back to all requests in the batch + for _, req := range batch { + select { + case req.Callback <- pubsub.ValidationAccept: + case <-time.After(1 * time.Millisecond): + // Timeout if callback channel is full + log.Printf("Warning: Callback timeout for message") + } + } +} + +// ValidateMessage queues a message for batch verification +func (bv *BLSBatchVerifier) ValidateMessage(ctx context.Context, msg *pubsub.Message) pubsub.ValidationResult { + // Create callback channel + callback := make(chan pubsub.ValidationResult, 1) + + // Create validation request + req := &ValidationRequest{ + Message: msg, + Callback: callback, + } + + // Queue the request + select { + case bv.queue <- req: + // Successfully queued + case <-ctx.Done(): + return pubsub.ValidationReject + default: + // Queue is full, reject message + log.Printf("Warning: Validation queue full, rejecting message") + return pubsub.ValidationReject + } + + // Wait for batch verification result + select { + case result := <-callback: + return result + case <-ctx.Done(): + return pubsub.ValidationReject + case <-time.After(100 * time.Millisecond): + // Timeout after 50ms (should be well within batch processing time) + log.Printf("Warning: Validation timeout for message") + return pubsub.ValidationReject + } +} + +// Close shuts down the batch verifier +func (bv *BLSBatchVerifier) Close() { + bv.cancel() + close(bv.queue) +} + +func GetClientType(nodeId int, topology *topology.Topology) (string, error) { + if slices.Contains(topology.PrysmNodeids, nodeId) { + return PrysmClient, nil + } + + if slices.Contains(topology.LighthouseNodeids, nodeId) { + return LighthouseClient, nil + } + + return "", fmt.Errorf("node %d is not a prysm or lighthouse client", nodeId) +} + +func GetNodeType(nodeId int, topology *topology.Topology) (string, error) { + if slices.Contains(topology.MeshNodeIds, nodeId) { + return MeshNode, nil + } + if slices.Contains(topology.MeshAttesterNodeIds, nodeId) { + return MeshAttesterNode, nil + } + if slices.Contains(topology.NonMeshAttesterNodeIds, nodeId) { + return NonMeshAttesterNode, nil + } + + return "", fmt.Errorf("node %d is not a mesh node, mesh attester node, or non mesh attester node", nodeId) +} + func main() { + // sleep for a random time between 0 and 100ms. This is to avoid all nodes starting at the + // same time to avoid clock synchronization + time.Sleep(time.Duration(rand.Intn(1000)) * time.Millisecond) + var ( - nodeID = flag.Int("node-id", 0, "Node ID for this simulation instance") - nodeCount = flag.Int("node-count", 10, "Total number of nodes in simulation") - msgSize = flag.Int("msg-size", 32, "Size of the message in bytes") - topologyFile = flag.String("topology-file", "", "Path to topology JSON file (required)") - logLevel = flag.String("log-level", "info", "Log level (debug, info, warn, error)") + nodeID = flag.Int("node-id", 0, "Node ID for this simulation instance") + nodeCount = flag.Int("node-count", 10, "Total number of nodes in simulation") + simConfigFile = flag.String("simconfig-file", "", "Path to simulation config file") + topologyFile = flag.String("topology-file", "", "Path to topology JSON file (required)") + logLevel = flag.String("log-level", "info", "Log level (debug, info, warn, error)") ) flag.Parse() @@ -47,11 +286,36 @@ func main() { log.Fatal("Topology file is required. Use -topology-file flag to specify a topology JSON file.") } + // Load simulation config + if *simConfigFile == "" { + log.Fatal("Simulation config file is required. Use -sim-config-file flag to specify a simulation config file.") + } + + simConfig, err := simconfig.ParseSimConfig(*simConfigFile) + if err != nil { + log.Fatalf("Failed to load simulation config from file: %v", err) + } + + log.Printf("Loaded simulation config from file: %s", *simConfigFile) + log.Printf("Genesis time: %d", simConfig.GenesisTime) + log.Printf("Slots to run: %d", simConfig.SlotsToRun) + topo, err := topology.LoadFromFile(*topologyFile) if err != nil { log.Fatalf("Failed to load topology from file: %v", err) } + nodeType, err := GetNodeType(*nodeID, topo) + if err != nil { + log.Fatalf("Failed to get node type: %v", err) + } + + clientType, err := GetClientType(*nodeID, topo) + if err != nil { + log.Fatalf("Failed to get client type: %v", err) + } + log.Printf("Node %d is a %s client", *nodeID, clientType) + // Validate node count if topo.NodeCount != *nodeCount { log.Fatalf("Topology file specifies %d nodes but simulation has %d nodes", @@ -61,8 +325,8 @@ func main() { log.Printf("Loaded topology from file: %s", *topologyFile) log.Printf("Starting libp2p GossipSub simulation") - log.Printf("Node ID: %d, Total nodes: %d, Message size: %d bytes", - *nodeID, *nodeCount, *msgSize) + log.Printf("Node ID: %d, Node Type: %s,Total nodes: %d, Message size: %d bytes", + *nodeID, nodeType, *nodeCount, simConfig.MsgSize) log.Printf("Topology: %s", topo.GetDescription()) ctx := context.Background() @@ -83,6 +347,11 @@ func main() { h, err := libp2p.New( libp2p.Identity(priv), libp2p.ListenAddrStrings(listenAddr), + libp2p.DefaultMuxers, + libp2p.Transport(libp2ptcp.NewTCPTransport), + libp2p.Muxer("/mplex/6.7.0", mplex.DefaultTransport), + libp2p.Security(noise.ID, noise.New), + libp2p.Ping(false), // Disable Ping Service. ) if err != nil { log.Fatalf("Failed to create libp2p host: %v", err) @@ -104,27 +373,73 @@ func main() { gossipsubParams.HistoryLength = 6 // mcache_len: number of windows to retain full messages gossipsubParams.HistoryGossip = 3 // mcache_gossip: number of windows to gossip about + // Create custom tracer for this node + tracer := NewCustomTracer(*nodeID) + ps, err := pubsub.NewGossipSub(ctx, h, pubsub.WithGossipSubParams(gossipsubParams), + pubsub.WithRawTracer(tracer), + pubsub.WithPeerOutboundQueueSize(2000), + pubsub.WithValidateQueueSize(2000), + pubsub.WithMessageIdFn(GetMessageId), ) if err != nil { log.Fatalf("Failed to create gossipsub: %v", err) } - // Join topic - topicName := "gossipsub-sim" - topic, err := ps.Join(topicName) - if err != nil { - log.Fatalf("Failed to join topic: %v", err) + batchIntervalTimeDuration := time.Duration(simConfig.PrysmValidator.BatchInterval) * time.Millisecond + batchVerificationTimeDuration := time.Duration(simConfig.PrysmValidator.BatchVerifierTime) * time.Microsecond + + // Create BLS batch verifier for Prysm clients + var batchVerifier *BLSBatchVerifier + if clientType == PrysmClient { + log.Printf("Creating BLS batch verifier for Prysm client with batch verification time %s and batch interval time %s", batchVerificationTimeDuration.String(), batchIntervalTimeDuration.String()) + batchVerifier = NewBLSBatchVerifier(batchVerificationTimeDuration, batchIntervalTimeDuration) + defer batchVerifier.Close() + } else { + log.Printf("Creating Lighthouse validator for Lighthouse client with validator time %s", time.Duration(simConfig.LighthouseValidator.ValidatorTime).String()) } + // Register topic validator with BLS batch verification + topicName := "gossipsub-simabcdefghijklmnopqrstuvwxyzabcdefgh" + err = ps.RegisterTopicValidator(topicName, func(ctx context.Context, pid peer.ID, msg *pubsub.Message) pubsub.ValidationResult { + if clientType == PrysmClient { + return batchVerifier.ValidateMessage(ctx, msg) + } else if clientType == LighthouseClient { + time.Sleep(time.Duration(simConfig.LighthouseValidator.ValidatorTime) * time.Microsecond) + return pubsub.ValidationAccept + } - // Subscribe to topic - sub, err := topic.Subscribe() + return pubsub.ValidationReject + }) if err != nil { - log.Fatalf("Failed to subscribe to topic: %v", err) + log.Fatalf("Failed to register topic validator: %v", err) } + log.Printf("Registered BLS batch verifier for %s (%s batching + %s verification)", topicName, batchIntervalTimeDuration.String(), batchVerificationTimeDuration.String()) + + // Join topic and conditionally subscribe + var topic *pubsub.Topic + var sub *pubsub.Subscription + + if nodeType == MeshAttesterNode || nodeType == MeshNode { + topic, err = ps.Join(topicName) + if err != nil { + log.Fatalf("Failed to join topic: %v", err) + } + + sub, err = topic.Subscribe(pubsub.WithBufferSize(4096)) + if err != nil { + log.Fatalf("Failed to subscribe to topic: %v", err) + } + + log.Printf("Successfully joined and subscribed to topic: %s (publisher)", topicName) + } else { + topic, err = ps.Join(topicName) + if err != nil { + log.Fatalf("Failed to join topic: %v", err) + } - log.Printf("Successfully joined topic: %s", topicName) + log.Printf("Successfully joined topic: %s (fanout peer)", topicName) + } // Wait for all nodes to start listening (important in Shadow) log.Printf("Waiting for all nodes to initialize...") @@ -193,42 +508,90 @@ func main() { time.Sleep(waitDuration) } - // Start message receiver goroutine - receivedCount := 0 - go func() { - for { - msg, err := sub.Next(ctx) - if err != nil { - log.Printf("Error receiving message: %v", err) - return - } - receivedCount++ - log.Printf("Received message %d: %s", receivedCount, string(msg.Data)) - } - }() + slotTicker := NewSlotTicker() + slotTicker.RunSlotTicker(int64(simConfig.GenesisTime), time.Duration(simConfig.SlotTime)*time.Second) - // All nodes publish a message - log.Printf("Publishing message") + publisherSubscriber := slotTicker.Subscribe() + receiverSubscriber := slotTicker.Subscribe() - // Create message with specified size, fully filled - msgContent := fmt.Sprintf("Message-from-node-%d", *nodeID) - msg := make([]byte, *msgSize) + // TODO - we could do better here. Having an atomic.Uint64 looks a bit clunky + slotsTicked := atomic.Uint64{} - // Fill the entire message buffer - for j := 0; j < *msgSize; j++ { - if j < len(msgContent) { - msg[j] = msgContent[j] - } else { - // Fill remaining bytes with a pattern to use full message size - msg[j] = byte('A' + (j % 26)) + log.Printf("Initializing slot ticker...") + go func() { + for { + slotInfo := <-receiverSubscriber + slotsTicked.Store(slotInfo.relativeSlot) + log.Printf("Slots ticked: %d", slotsTicked.Load()) } + }() + log.Printf("Slot ticker initialized") + + // Start message receiver goroutine (only for publishers) + var receivedCount int + if nodeType == MeshAttesterNode || nodeType == MeshNode { + receivedCount = 0 + go func() { + for { + msg, err := sub.Next(ctx) + if err != nil { + log.Printf("Error receiving message: %v", err) + return + } + // only log after the last run + if slotsTicked.Load() == uint64(simConfig.SlotsToRun) { + log.Printf("Received message %d: %s", receivedCount, string(msg.Data)) + receivedCount++ + } else { + log.Printf("Ignoring received message for slot %d (not the last run)", slotsTicked.Load()) + } + } + }() + } else { + log.Printf("Node %d is a fanout peer - will not receive messages directly", *nodeID) } - err = topic.Publish(ctx, msg) - if err != nil { - log.Printf("Failed to publish message: %v", err) + totalPublishers := len(topo.NonMeshAttesterNodeIds) + len(topo.MeshAttesterNodeIds) + + // Only some nodes publish messages + if nodeType == MeshAttesterNode || nodeType == NonMeshAttesterNode { + go func() { + log.Printf("Publisher goroutine started") + for { + // publish every time a new slot is ticked + newSlotInfo := <-publisherSubscriber + if newSlotInfo.relativeSlot > uint64(simConfig.SlotsToRun) { + log.Printf("Skipping publication for slot %d (already reached slots to run)", newSlotInfo.currentSlot) + continue + } + + log.Printf("Publishing message (publisher %d of %d) for slot %d", *nodeID+1, totalPublishers, newSlotInfo.currentSlot) + + // Create message with specified size, fully filled + msgContent := fmt.Sprintf("Message-from-node-%d-slot-%d", *nodeID, newSlotInfo.currentSlot) + msg := make([]byte, simConfig.MsgSize) + + // Fill the entire message buffer + for j := 0; j < int(simConfig.MsgSize); j++ { + if j < len(msgContent) { + msg[j] = msgContent[j] + } else { + // Fill remaining bytes with a pattern to use full message size + msg[j] = byte('A' + (j % 26)) + } + } + + err = topic.Publish(ctx, msg) + if err != nil { + log.Printf("Failed to publish message: %v", err) + } else { + log.Printf("Published message: %s", msgContent) + } + } + }() } else { - log.Printf("Published message: %s", msgContent) + log.Printf("Not publishing (node %d is not a publisher)", + *nodeID) } // Keep the process running for Shadow diff --git a/gossipsub/slot_ticker.go b/gossipsub/slot_ticker.go new file mode 100644 index 0000000..5105c36 --- /dev/null +++ b/gossipsub/slot_ticker.go @@ -0,0 +1,82 @@ +package main + +import ( + "log" + "sync" + "time" +) + +type SlotTicker struct { + mu sync.Mutex + subscribers map[chan SlotInfo]bool +} + +func NewSlotTicker() SlotTicker { + return SlotTicker { + subscribers: make(map[chan SlotInfo]bool), + } +} + +type SlotInfo struct { + currentSlot uint64 + relativeSlot uint64 +} + +func (s *SlotTicker) RunSlotTicker(genesisTime int64, slotTime time.Duration) { + go func() { + genesisTime := ConvertGenesisTime(genesisTime) + sinceGenesisTime := time.Since(genesisTime) + nextTick := sinceGenesisTime.Truncate(slotTime) + slotTime + nextTickTime := genesisTime.Add(nextTick) + currentSlot := nextTick / slotTime + relativeSlot := 1 + + for { + waitTime := time.Until(nextTickTime) + + <-time.After(waitTime) + s.mu.Lock() + for subscriber := range s.subscribers { + select { + case subscriber <- SlotInfo{ currentSlot: uint64(currentSlot), relativeSlot: uint64(relativeSlot)}: + continue + default: + // subscriber is not ready to receive the slot + log.Printf("Subscriber is not ready to receive the slot") + } + } + s.mu.Unlock() + + currentSlot += 1 + relativeSlot += 1 + nextTickTime = nextTickTime.Add(slotTime) + } + }() +} + +func (s *SlotTicker) Subscribe() chan SlotInfo { + subscriber := make(chan SlotInfo, 100) // buffered channel + s.addSubscriber(subscriber) + return subscriber +} + +func (s *SlotTicker) addSubscriber(subscriber chan SlotInfo) { + s.mu.Lock() + defer s.mu.Unlock() + s.subscribers[subscriber] = true +} + +func ConvertGenesisTime(genesisTime int64) time.Time { + return time.Unix(genesisTime, 0) +} + +func CurrentSlot(genesisTime time.Time, slotTime time.Duration) uint64 { + return At(genesisTime, time.Now(), slotTime) +} + +func At(genesisTime, tm time.Time, slotTime time.Duration) uint64 { + if tm.Before(genesisTime) { + return 0 + } + return uint64(tm.Sub(genesisTime) / slotTime) +} \ No newline at end of file diff --git a/gossipsub/slot_ticker_test.go b/gossipsub/slot_ticker_test.go new file mode 100644 index 0000000..bda2eed --- /dev/null +++ b/gossipsub/slot_ticker_test.go @@ -0,0 +1,138 @@ +package main + +import ( + "testing" + "time" + + "github.com/stretchr/testify/require" +) + +func TestSlotCalculation(t *testing.T) { + genesisTime := int64(1719859200) + slotTime := 12 * time.Second + + testCases := []struct { + description string + genesisTime int64 + timeToCheck int64 + expectedSlot uint64 + }{ + { + description: "Slot at genesis time", + genesisTime: genesisTime, + timeToCheck: genesisTime, + expectedSlot: 0, + }, + { + description: "Slot after genesis time", + genesisTime: genesisTime, + timeToCheck: genesisTime + 12, + expectedSlot: 1, + }, + { + description: "Slot after genesis time + 12 seconds", + genesisTime: genesisTime, + timeToCheck: genesisTime + 24, + expectedSlot: 2, + }, + } + + for _, testCase := range testCases { + t.Run(testCase.description, func(t *testing.T) { + slot := At(ConvertGenesisTime(testCase.genesisTime), time.Unix(int64(testCase.timeToCheck), 0), slotTime) + if slot != testCase.expectedSlot { + t.Fatalf("expected slot %d, got %d", testCase.expectedSlot, slot) + } + }) + } +} + +func TestSlotTicker(t *testing.T) { + genesisTime := time.Now().Unix() + slotTicker := NewSlotTicker() + slotTicker.RunSlotTicker(genesisTime, 1*time.Second) + + subscriber := slotTicker.Subscribe() + + slot_1 := <-subscriber + slot_2 := <-subscriber + slot_3 := <-subscriber + slot_4 := <-subscriber + slot_5 := <-subscriber + + require.Equal(t, slot_1.currentSlot, uint64(1)) + require.Equal(t, slot_2.currentSlot, uint64(2)) + require.Equal(t, slot_3.currentSlot, uint64(3)) + require.Equal(t, slot_4.currentSlot, uint64(4)) + require.Equal(t, slot_5.currentSlot, uint64(5)) + + require.Equal(t, slot_1.relativeSlot, uint64(1)) + require.Equal(t, slot_2.relativeSlot, uint64(2)) + require.Equal(t, slot_3.relativeSlot, uint64(3)) + require.Equal(t, slot_4.relativeSlot, uint64(4)) + require.Equal(t, slot_5.relativeSlot, uint64(5)) +} + +func TestSlotTickerGenesisTimeInPast(t *testing.T) { + genesisTime := time.Now().Add(-30 * time.Second) + + slotTicker := NewSlotTicker() + slotTicker.RunSlotTicker(genesisTime.Unix(), 1*time.Second) + + currentSlot := CurrentSlot(genesisTime, 1*time.Second) + require.Equal(t, currentSlot, uint64(30)) + + subsciber := slotTicker.Subscribe() + + slot_1 := <-subsciber + slot_2 := <-subsciber + slot_3 := <-subsciber + slot_4 := <-subsciber + slot_5 := <-subsciber + + require.Equal(t, slot_1.currentSlot, uint64(31)) + require.Equal(t, slot_2.currentSlot, uint64(32)) + require.Equal(t, slot_3.currentSlot, uint64(33)) + require.Equal(t, slot_4.currentSlot, uint64(34)) + require.Equal(t, slot_5.currentSlot, uint64(35)) + + require.Equal(t, slot_1.relativeSlot, uint64(1)) + require.Equal(t, slot_2.relativeSlot, uint64(2)) + require.Equal(t, slot_3.relativeSlot, uint64(3)) + require.Equal(t, slot_4.relativeSlot, uint64(4)) + require.Equal(t, slot_5.relativeSlot, uint64(5)) +} + +func TestSlotTickerMultipleSubscribers(t *testing.T) { + genesisTime := time.Now().Unix() + slotTicker := NewSlotTicker() + slotTicker.RunSlotTicker(genesisTime, 1*time.Second) + + subscriber_1 := slotTicker.Subscribe() + subscriber_2 := slotTicker.Subscribe() + subscriber_3 := slotTicker.Subscribe() + + slot_1_1 := <-subscriber_1 + slot_1_2 := <-subscriber_2 + slot_1_3 := <-subscriber_3 + + slot_2_1 := <-subscriber_1 + slot_2_2 := <-subscriber_2 + slot_2_3 := <-subscriber_3 + + require.Equal(t, slot_1_1.currentSlot, uint64(1)) + require.Equal(t, slot_1_2.currentSlot, uint64(1)) + require.Equal(t, slot_1_3.currentSlot, uint64(1)) + + require.Equal(t, slot_1_1.relativeSlot, uint64(1)) + require.Equal(t, slot_1_2.relativeSlot, uint64(1)) + require.Equal(t, slot_1_3.relativeSlot, uint64(1)) + + require.Equal(t, slot_2_1.currentSlot, uint64(2)) + require.Equal(t, slot_2_2.currentSlot, uint64(2)) + require.Equal(t, slot_2_3.currentSlot, uint64(2)) + + require.Equal(t, slot_2_1.relativeSlot, uint64(2)) + require.Equal(t, slot_2_2.relativeSlot, uint64(2)) + require.Equal(t, slot_2_3.relativeSlot, uint64(2)) +} \ No newline at end of file diff --git a/network_graph.py b/network_graph.py index 5e564b5..5e33837 100755 --- a/network_graph.py +++ b/network_graph.py @@ -40,108 +40,133 @@ class NodeType: south_africa = Location("south_africa", 47) south_america = Location("south_america", 36) -supernode = NodeType("supernode", 1024, 1024, 20) -fullnode = NodeType("fullnode", 50, 50, 80) -node_types = [supernode, fullnode] +supernode = NodeType("supernode", 50, 50, 0) +fullnode = NodeType("fullnode", 15, 50, 100) +node_types = [fullnode] locations = [australia, europe, east_asia, west_asia, na_east, na_west, south_africa, south_america] +""" +Edge latencies are picked up from https://www.cloudping.co/. We map regions based on their AWS Regions and set the latencies +based on the ping latencies between these regions. + +australia - ap-southeast-2 X +east_asia - ap-northeast-1 X +europe - eu-central-1 X +na_west - us-west-2 X +na_east - us-east-1 X +south_america - sa-east-1 X +south_africa - af-south-1 X +west_asia - me-south-1 X +""" + edges = [ Edge(australia, australia, 2), - Edge(australia, east_asia, 110), - Edge(australia, europe, 165), - Edge(australia, na_west, 110), - Edge(australia, na_east, 150), - Edge(australia, south_america, 190), - Edge(australia, south_africa, 220), - Edge(australia, west_asia, 180), - - Edge(east_asia, australia, 110), - Edge(east_asia, east_asia, 4), - Edge(east_asia, europe, 125), - Edge(east_asia, na_west, 100), - Edge(east_asia, na_east, 140), - Edge(east_asia, south_america, 175), - Edge(east_asia, south_africa, 175), - Edge(east_asia, west_asia, 110), - - Edge(europe, australia, 165), - Edge(europe, east_asia, 125), - Edge(europe, europe, 2), - Edge(europe, na_west, 110), - Edge(europe, na_east, 70), - Edge(europe, south_america, 140), - Edge(europe, south_africa, 95), - Edge(europe, west_asia, 60), - - Edge(na_west, australia, 110), - Edge(na_west, east_asia, 100), - Edge(na_west, europe, 110), - Edge(na_west, na_west, 2), - Edge(na_west, na_east, 60), - Edge(na_west, south_america, 100), - Edge(na_west, south_africa, 160), - Edge(na_west, west_asia, 150), - - Edge(na_east, australia, 150), - Edge(na_east, east_asia, 140), - Edge(na_east, europe, 70), - Edge(na_east, na_west, 60), - Edge(na_east, na_east, 2), - Edge(na_east, south_america, 100), - Edge(na_east, south_africa, 130), - Edge(na_east, west_asia, 110), - - Edge(south_america, australia, 190), - Edge(south_america, east_asia, 175), - Edge(south_america, europe, 140), - Edge(south_america, na_west, 100), - Edge(south_america, na_east, 100), - Edge(south_america, south_america, 7), - Edge(south_america, south_africa, 195), - Edge(south_america, west_asia, 145), - - Edge(south_africa, australia, 220), - Edge(south_africa, east_asia, 175), - Edge(south_africa, europe, 95), - Edge(south_africa, na_west, 160), - Edge(south_africa, na_east, 130), - Edge(south_africa, south_america, 190), - Edge(south_africa, south_africa, 7), - Edge(south_africa, west_asia, 110), - - Edge(west_asia, australia, 180), - Edge(west_asia, east_asia, 110), - Edge(west_asia, europe, 60), - Edge(west_asia, na_west, 150), - Edge(west_asia, na_east, 110), - Edge(west_asia, south_america, 145), - Edge(west_asia, south_africa, 110), - Edge(west_asia, west_asia, 5), + Edge(australia, east_asia, 107), + Edge(australia, europe, 251), + Edge(australia, na_west, 141), + Edge(australia, na_east, 202), + Edge(australia, south_america, 316), + Edge(australia, south_africa, 409), + Edge(australia, west_asia, 185), + + Edge(east_asia, australia, 120), + Edge(east_asia, east_asia, 8), + Edge(east_asia, europe, 227), + Edge(east_asia, na_west, 101), + Edge(east_asia, na_east, 155), + Edge(east_asia, south_america, 264), + Edge(east_asia, south_africa, 394), + Edge(east_asia, west_asia, 175), + + Edge(europe, australia, 251), + Edge(europe, east_asia, 230), + Edge(europe, europe, 6), + Edge(europe, na_west, 145), + Edge(europe, na_east, 96), + Edge(europe, south_america, 206), + Edge(europe, south_africa, 154), + Edge(europe, west_asia, 86), + + Edge(na_west, australia, 143), + Edge(na_west, east_asia, 102), + Edge(na_west, europe, 145), + Edge(na_west, na_west, 5), + Edge(na_west, na_east, 65), + Edge(na_west, south_america, 180), + Edge(na_west, south_africa, 274), + Edge(na_west, west_asia, 267), + + Edge(na_east, australia, 206), + Edge(na_east, east_asia, 154), + Edge(na_east, europe, 97), + Edge(na_east, na_west, 67), + Edge(na_east, na_east, 9), + Edge(na_east, south_america, 115), + Edge(na_east, south_africa, 227), + Edge(na_east, west_asia, 170), + + Edge(south_america, australia, 311), + Edge(south_america, east_asia, 265), + Edge(south_america, europe, 206), + Edge(south_america, na_west, 176), + Edge(south_america, na_east, 115), + Edge(south_america, south_america, 6), + Edge(south_america, south_africa, 340), + Edge(south_america, west_asia, 277), + + Edge(south_africa, australia, 415), + Edge(south_africa, east_asia, 391), + Edge(south_africa, europe, 159), + Edge(south_africa, na_west, 276), + Edge(south_africa, na_east, 229), + Edge(south_africa, south_america, 343), + Edge(south_africa, south_africa, 3), + Edge(south_africa, west_asia, 248), + + Edge(west_asia, australia, 185), + Edge(west_asia, east_asia, 175), + Edge(west_asia, europe, 87), + Edge(west_asia, na_west, 262), + Edge(west_asia, na_east, 168), + Edge(west_asia, south_america, 276), + Edge(west_asia, south_africa, 249), + Edge(west_asia, west_asia, 2), ] if len(sys.argv) < 3: - print("Usage: python network_graph.py [node-count] [msg-size] [topology-file]") + print("Usage: python network_graph.py [node-count] [msg-size] [topology-file] [simconfig-file] [batch-interval] [batch-verifier-time] [slots-to-run]") print(" node-count: Number of nodes in the simulation") - print(" msg-size: Size of the message in bytes") print(" topology-file: Path to topology JSON file (required)") + print(" simconfig-file: Path to simulation config file (required)") sys.exit(1) node_count = int(sys.argv[1]) -msg_size = int(sys.argv[2]) -topology_file = sys.argv[3] if len(sys.argv) > 3 else "" +topology_file = sys.argv[2] if len(sys.argv) > 2 else "" +simconfig_file = sys.argv[3] if len(sys.argv) > 3 else "" if not topology_file: print("Error: Topology file is required") sys.exit(1) +if not simconfig_file: + print("Error: Simulation config file is required") + sys.exit(1) + if not os.path.exists(topology_file): print(f"Error: Topology file not found: {topology_file}") sys.exit(1) +if not os.path.exists(simconfig_file): + print(f"Error: Simulation config file not found: {simconfig_file}") + sys.exit(1) + topology_file = os.path.abspath(topology_file) print(f"Using topology file: {topology_file}") +simconfig_file = os.path.abspath(simconfig_file) +print(f"Using simulation config file: {simconfig_file}") + ids = {} for node_type in node_types: for location in locations: @@ -175,13 +200,13 @@ class NodeType: config["hosts"] = {} for i in range(node_count): location = random.choices(locations, map(lambda lc: lc.weight, locations))[0] - if i == 0: - node_type = supernode - else: - node_type = random.choices(node_types, map(lambda nt: nt.weight, node_types))[0] + # if i == 0: + # node_type = supernode + # else: + node_type = random.choices(node_types, map(lambda nt: nt.weight, node_types))[0] - # Build args with topology file - args = f"-node-id {i} -node-count {node_count} -msg-size {msg_size} -topology-file {topology_file}" + # Build args with topology file, nodes to publish, and batch parameters + args = f"-node-id {i} -node-count {node_count} -topology-file {topology_file} --simconfig-file {simconfig_file}" config["hosts"][f"node{i}"] = { "network_node_id": ids[f"{location.name}-{node_type.name}"], @@ -197,5 +222,22 @@ class NodeType: print(f"Generated Shadow configuration: shadow-gossipsub.yaml") print(f" Nodes: {node_count}") +# Read simulation config to get message size and validation parameters +simconfig_file = "simconfig.yaml" +if not os.path.exists(simconfig_file): + print(f"Error: Simulation config file not found: {simconfig_file}") + sys.exit(1) + +with open(simconfig_file, "r") as file: + simconfig = yaml.safe_load(file) + +msg_size = simconfig.get("msg_size", 192) +batch_interval = simconfig.get("prysm_validator", {}).get("batch_interval", 5) +batch_verifier_time = simconfig.get("prysm_validator", {}).get("batch_verifier_time", 4500) +lighthouse_validator_time = simconfig.get("lighthouse_validator", {}).get("validator_time", 2500) + print(f" Message size: {msg_size} bytes") -print(f" Topology: {topology_file}") +print(f" Prysm Batch interval: {batch_interval}ms") +print(f" Prysm Batch verifier time: {batch_verifier_time}μs") +print(f" Lighthouse validator time: {lighthouse_validator_time}μs") +print(f" Topology: {topology_file}") \ No newline at end of file diff --git a/new-latencies-plots/30slots_batch_50ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png b/new-latencies-plots/30slots_batch_50ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png new file mode 100644 index 0000000..81b059b Binary files /dev/null and b/new-latencies-plots/30slots_batch_50ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png differ diff --git a/new-latencies-plots/30slots_batch_5ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png b/new-latencies-plots/30slots_batch_5ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png new file mode 100644 index 0000000..7e9dd6b Binary files /dev/null and b/new-latencies-plots/30slots_batch_5ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png differ diff --git a/plot_propagation.py b/plot_propagation.py index ae54466..c5953d5 100644 --- a/plot_propagation.py +++ b/plot_propagation.py @@ -13,6 +13,8 @@ from datetime import datetime from typing import List, Tuple import matplotlib.pyplot as plt +import json +import yaml def parse_timestamp(timestamp_str: str) -> float: @@ -25,16 +27,17 @@ def parse_timestamp(timestamp_str: str) -> float: return delta.total_seconds() -def parse_shadow_logs(node_count: int) -> List[Tuple[float, int]]: +def parse_shadow_logs(node_count: int, topology: dict) -> List[Tuple[float, int]]: """ - Parse Shadow logs to extract message reception times. + Parse Shadow logs to extract message reception times among subscribers. Returns: List of (timestamp, cumulative_total) tuples """ events = [] # List of (timestamp, node_id) - for node_id in range(node_count): + # Only parse logs for subscriber nodes (0 to nodes_to_subscribe-1) + for node_id in topology["mesh_node_ids"] + topology["mesh_attester_node_ids"]: log_file = f"shadow.data/hosts/node{node_id}/gossipsub.1000.stderr" if not os.path.exists(log_file): @@ -64,34 +67,173 @@ def parse_shadow_logs(node_count: int) -> List[Tuple[float, int]]: return cumulative_data -def plot_propagation(node_count: int, output_file: str = 'message_propagation.png'): +def plot_propagation(node_count: int, output_file: str = 'message_propagation.png', topology_file: str = "", peer_count: int = None, non_mesh_node_peer_count: int = None, simconfig_file: str = None): """Plot message propagation over time.""" - print(f"Parsing Shadow logs for {node_count} nodes...") + if peer_count is None: + peer_count = 0 # Default for non-random-regular topologies + if non_mesh_node_peer_count is None: + non_mesh_node_peer_count = 0 # Default for non-random-regular topologies + if simconfig_file is None: + simconfig_file = "simconfig.yaml" # Default simconfig file + + # Read the simconfig file to get slot_time + simconfig = None + with open(simconfig_file, 'r') as f: + simconfig = yaml.safe_load(f) + + batch_interval = simconfig.get("prysm_validator", {}).get("batch_interval", 0) + batch_verifier_time = simconfig.get("prysm_validator", {}).get("batch_verifier_time", 0) + lighthouse_validator_time = simconfig.get("lighthouse_validator", {}).get("validator_time", 0) + slots_to_run = simconfig.get("slots_to_run", 1) + + print(f"Prysm Batch interval: {batch_interval}ms") + print(f"Prysm Batch verifier time: {batch_verifier_time}μs") + print(f"Lighthouse validator time: {lighthouse_validator_time}μs") + print(f"Slots to run: {slots_to_run}") + + # Read the topology json file + topology = None + with open(topology_file, 'r') as f: + topology = json.load(f) + + mesh_nodes = len(topology["mesh_node_ids"]) + mesh_attester_nodes = len(topology["mesh_attester_node_ids"]) + non_mesh_attester_nodes = len(topology["non_mesh_attester_node_ids"]) + + # Calculate the start time for the last slot (Nth slot where N = slots_to_run) + # Genesis time is a Unix timestamp, but Shadow timestamps are relative to 2000/01/01 00:00:00 + # We need to convert genesis_time to Shadow's time system + shadow_epoch = datetime(2000, 1, 1, 0, 2, 0) + genesis_time_unix = simconfig["genesis_time"] + genesis_time_dt = datetime.fromtimestamp(genesis_time_unix) + + slot_time = simconfig["slot_time"] + + nodes_to_publish = non_mesh_attester_nodes + mesh_attester_nodes + nodes_to_subscribe = mesh_nodes + mesh_attester_nodes + + print(f"Parsing Shadow logs for {node_count} nodes with {non_mesh_attester_nodes} Non-Mesh Attesters and {mesh_attester_nodes} Mesh Attesters...") + print(f"Tracking propagation to {nodes_to_subscribe} Subscribers from {nodes_to_publish} Publishers...") + if peer_count > 0: + print(f"Network topology: random-regular with {peer_count} peers per node") - cumulative_data = parse_shadow_logs(node_count) + cumulative_data = parse_shadow_logs(node_count, topology) if not cumulative_data: print("Error: No message reception events found in logs") return - # Calculate metric: total_received / node_count - timestamps = [t for t, _ in cumulative_data] - avg_per_node = [count / node_count for _, count in cumulative_data] - - # Make timestamps relative to publish time (2000/01/01 00:02:00 = 120 seconds) - publish_time = 120.0 - timestamps = [(t - publish_time) * 1000 for t in timestamps] # Convert to milliseconds + # Debug: show timestamp range of all events + if cumulative_data: + all_timestamps = [t for t, _ in cumulative_data] + min_time = min(all_timestamps) + max_time = max(all_timestamps) + print(f"Debug: All event timestamps range from {min_time:.1f}s to {max_time:.1f}s") + + # Since messages are only logged for the last slot (see gossipsub/main.go line 520), + # all events in the logs are from the last slot only + # Use the earliest timestamp as the baseline (when the last slot started / messages were published) + earliest_timestamp = min(all_timestamps) + last_slot_start_time = earliest_timestamp + + print(f"All events are from the last slot (slot {slots_to_run})") + print(f"Using earliest event at {earliest_timestamp:.1f}s as slot start (0ms baseline)") + + # All events are already from the last slot, so no filtering needed + # Just recalculate cumulative count to restart from 1 (in case there were any other logs earlier) + filtered_data = [] + cumulative_count = 0 + for t, count in cumulative_data: + cumulative_count += 1 + filtered_data.append((t, cumulative_count)) + + # Calculate metric: percentage of total expected messages received + timestamps = [t for t, _ in filtered_data] + total_expected = nodes_to_publish * nodes_to_subscribe + percentage_received = [(count / total_expected) * 100 for _, count in filtered_data] + + # Normalize timestamps relative to the start of the last slot + # Subtract the earliest timestamp to get relative propagation time (0ms = when last slot started) + timestamps = [(t - last_slot_start_time) * 1000 for t in timestamps] # Convert to milliseconds # Create plot plt.figure(figsize=(10, 6)) - plt.plot(timestamps, avg_per_node, linewidth=2) + plt.plot(timestamps, percentage_received, linewidth=2) plt.xlabel('Time (ms)', fontsize=12) - plt.ylabel('Average messages received per node', fontsize=12) - plt.title(f'Message Propagation Over Time ({node_count} nodes)', fontsize=14) + plt.ylabel('Cumulative Percentage of Messages Received (%)', fontsize=12) + # Set x-axis to scale up to 700ms + plt.xlim(left=0, right=1400) + + # Create title and subtitle + title = 'Message Arrival Time Distribution' + + # Create a concise subtitle with key network info + subtitle = f'{node_count} nodes | {mesh_attester_nodes} Mesh Attesters | {mesh_nodes} Mesh Nodes | {non_mesh_attester_nodes} Non-Mesh Attesters' + + # Get client split info if available + client_split = simconfig.get("client_split", {}) + prysm_pct = client_split.get("prysm", 0) + lighthouse_pct = client_split.get("lighthouse", 0) + + # Create detailed info text box + info_lines = [ + f'Network: {node_count} nodes ({mesh_attester_nodes} MA, {mesh_nodes} MN, {non_mesh_attester_nodes} NMA)', + f'Clients: {prysm_pct}% Prysm, {lighthouse_pct}% Lighthouse', + f'Prysm: {batch_interval}ms interval, {batch_verifier_time}μs verifier', + f'Lighthouse: {lighthouse_validator_time}μs validator', + f'Topology: {peer_count} peers/node, {non_mesh_node_peer_count} peers/non-mesh', + f'Slot: {slots_to_run} slots × {slot_time}s' + ] + + info_text = '\n'.join(info_lines) + + # Set title and subtitle + plt.title(title, fontsize=14, fontweight='bold') + plt.suptitle(subtitle, fontsize=11, y=0.96, color='gray') + + # Add info box in mid-right corner of figure + plt.figtext(0.98, 0.5, info_text, + fontsize=8, + verticalalignment='center', + horizontalalignment='right', + bbox=dict(boxstyle='round', facecolor='wheat', alpha=0.8), + family='monospace') + plt.grid(True, alpha=0.3) - # Add horizontal line at expected final value (N messages per node for N nodes) - plt.axhline(y=node_count, color='r', linestyle='--', alpha=0.5, label=f'Expected final: {node_count}') + # Add horizontal line at 100% completion + plt.axhline(y=100, color='r', linestyle='--', alpha=0.5, label='100% Complete') + + # Calculate percentile times + percentile_times = {} + percentiles = [50, 75, 90, 95, 99] + + for p in percentiles: + target_percentage = p # Direct percentage value + percentile_time = None + + for i, percentage in enumerate(percentage_received): + if percentage >= target_percentage: + percentile_time = timestamps[i] + break + + percentile_times[p] = percentile_time + + # Calculate 100th percentile (time when last message is received) + if timestamps and percentage_received: + percentile_times[100] = timestamps[-1] # Last timestamp is when 100% is reached + + # Add percentile markers to plot + colors = ['orange', 'purple', 'brown', 'pink', 'gray'] + for i, p in enumerate(percentiles): + if percentile_times[p] is not None: + plt.axvline(x=percentile_times[p], color=colors[i], linestyle=':', alpha=0.7, + label=f'{p}th percentile: {percentile_times[p]:.1f}ms') + + # Mark 100th percentile (last message) with a distinct style + if percentile_times.get(100) is not None: + plt.axvline(x=percentile_times[100], color='red', linestyle='--', linewidth=2, alpha=0.8, + label=f'100th percentile (last msg): {percentile_times[100]:.1f}ms') plt.legend() plt.tight_layout() @@ -101,19 +243,36 @@ def plot_propagation(node_count: int, output_file: str = 'message_propagation.pn print(f"\nPlot saved to: {output_file}") # Print statistics - if cumulative_data: + if filtered_data: start_time = timestamps[0] end_time = timestamps[-1] duration = end_time - start_time - final_avg = avg_per_node[-1] - - print(f"\nPropagation Statistics:") - print(f" Start time: {start_time:.3f}ms (relative to publish)") - print(f" End time: {end_time:.3f}ms (relative to publish)") + final_percentage = percentage_received[-1] + + print(f"\nPropagation Statistics (Subscribers) - Last Slot Only:") + print(f" Network: {node_count} nodes, {nodes_to_publish} publishers, {nodes_to_subscribe} subscribers", end="") + if peer_count > 0: + print(f", {peer_count} peers/node") + else: + print() + print(f" Measuring slot {slots_to_run} (last slot)") + print(f" Start time: {start_time:.3f}ms (relative to slot {slots_to_run} start)") + print(f" End time: {end_time:.3f}ms (relative to slot {slots_to_run} start)") print(f" Duration: {duration:.3f}ms") - print(f" Final average: {final_avg:.2f} messages/node") - print(f" Expected: {node_count} messages/node") - print(f" Total events: {len(cumulative_data)}") + print(f" Final percentage: {final_percentage:.2f}% of messages received") + print(f" Expected: 100% of messages") + print(f" Total events in last slot: {len(filtered_data)}") + + print(f"\nPercentile Times (Time to reach X% of messages):") + for p in percentiles: + if percentile_times[p] is not None: + print(f" {p:2d}th percentile: {percentile_times[p]:.3f}ms") + else: + print(f" {p:2d}th percentile: Not reached") + + # Print 100th percentile separately + if percentile_times.get(100) is not None: + print(f" 100th percentile (last message): {percentile_times[100]:.3f}ms") def main(): @@ -125,6 +284,7 @@ def main(): python3 plot_propagation.py 10 python3 plot_propagation.py 20 -o propagation_20nodes.png python3 plot_propagation.py 10 --output results/test1.png + python3 plot_propagation.py 20 -p 5 -s 15 -o propagation_5p_15s.png """ ) @@ -133,17 +293,28 @@ def main(): parser.add_argument('-o', '--output', type=str, default='message_propagation.png', help='Output file path for the plot (default: message_propagation.png)') + parser.add_argument('--peer-count', type=int, + help='Number of peers per node for random-regular topology (default: 0 for other topologies)') + parser.add_argument('--non-mesh-node-peer-count', type=int, + help='Number of peers per non-mesh node for random-regular topology (default: 0 for other topologies)') + parser.add_argument('--topology-file', type=str, + help='Topology file path (default: topology.json)') + parser.add_argument('--simconfig-file', type=str, + help='Simulation config file path (default: simconfig.yaml)') args = parser.parse_args() if args.node_count <= 0: parser.error("node-count must be positive") + if args.topology_file is None: + parser.error("topology-file must be provided") + if not os.path.exists("shadow.data"): print("Error: shadow.data directory not found. Run simulation first with 'make run-sim'") sys.exit(1) - plot_propagation(args.node_count, args.output) + plot_propagation(args.node_count, args.output, args.topology_file, args.peer_count, args.non_mesh_node_peer_count, args.simconfig_file) if __name__ == "__main__": diff --git a/plots/batch_50ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png b/plots/batch_50ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png new file mode 100644 index 0000000..9d4d84e Binary files /dev/null and b/plots/batch_50ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png differ diff --git a/plots/batch_5ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png b/plots/batch_5ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png new file mode 100644 index 0000000..09db30e Binary files /dev/null and b/plots/batch_5ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png differ diff --git a/simconfig.yaml b/simconfig.yaml new file mode 100644 index 0000000..1d4eb61 --- /dev/null +++ b/simconfig.yaml @@ -0,0 +1,12 @@ +genesis_time: 946665000 +slot_time: 12 +client_split: + prysm: 44 + lighthouse: 56 +prysm_validator: + batch_interval: 50 + batch_verifier_time: 4500 +lighthouse_validator: + validator_time: 2500 +slots_to_run: 30 +msg_size: 192 diff --git a/simconfig/simconfig.go b/simconfig/simconfig.go new file mode 100644 index 0000000..ccde2fe --- /dev/null +++ b/simconfig/simconfig.go @@ -0,0 +1,71 @@ +package simconfig + +import ( + "fmt" + "log" + "os" + + "gopkg.in/yaml.v3" +) + +type PrysmValidatorInfo struct { + BatchInterval uint64 `yaml:"batch_interval"` + BatchVerifierTime uint64 `yaml:"batch_verifier_time"` +} + +type LighthouseValidatorInfo struct { + ValidatorTime uint64 `yaml:"validator_time"` +} + +type ClientSplit struct { + Prysm uint64 `yaml:"prysm"` + Lighthouse uint64 `yaml:"lighthouse"` +} + +func (c *ClientSplit) PrysmPercentage() float64 { + return float64(c.Prysm) / float64(c.Prysm+c.Lighthouse) +} + +func (c *ClientSplit) LighthousePercentage() float64 { + return float64(c.Lighthouse) / float64(c.Prysm+c.Lighthouse) +} + +func (c *ClientSplit) Validate() bool { + return c.Prysm+c.Lighthouse == 100 +} + +type SimConfig struct { + GenesisTime uint64 `yaml:"genesis_time"` + // TODO - add more fields here for other simulation params like + // mesh attester count, mesh node count etc. + SlotTime uint64 `yaml:"slot_time"` + + ClientSplit ClientSplit `yaml:"client_split"` + + PrysmValidator PrysmValidatorInfo `yaml:"prysm_validator"` + LighthouseValidator LighthouseValidatorInfo `yaml:"lighthouse_validator"` + + SlotsToRun uint64 `yaml:"slots_to_run"` + MsgSize uint64 `yaml:"msg_size"` +} + +func ParseSimConfig(configPath string) (*SimConfig, error) { + configFileData, err := os.ReadFile(configPath) + if err != nil { + return nil, fmt.Errorf("error reading simulation config file: %w", err) + } + + var config SimConfig + err = yaml.Unmarshal([]byte(configFileData), &config) + if err != nil { + return nil, fmt.Errorf("error unmarshalling config file: %w", err) + } + + log.Printf("Loaded simulation config: %+v", config) + + if !config.ClientSplit.Validate() { + return nil, fmt.Errorf("client split is not valid") + } + + return &config, nil +} diff --git a/test_results.py b/test_results.py index 53a53bb..4e5bf6a 100755 --- a/test_results.py +++ b/test_results.py @@ -1,26 +1,28 @@ #!/usr/bin/env python3 """ Test script to verify Shadow simulation results for GossipSub by reading log files. -Since each simulation publishes exactly one message, this verifies that all nodes received it. +Verifies that all subscribers received messages from all publishers. -Usage: python3 test_results.py [node_count] +Usage: python3 test_results.py [node_count] [nodes_to_publish] [nodes_to_subscribe] """ import sys import os import glob -from typing import Dict +from typing import Dict, List +import json -def parse_shadow_logs(node_count: int) -> Dict[int, int]: +def parse_shadow_logs(node_count: int, subscribed_node_ids: List[int]) -> Dict[int, int]: """ - Parse Shadow log files to extract message reception counts. + Parse Shadow log files to extract message reception counts for subscribers. Returns: - received_counts: Dict mapping node_id to number of messages received """ received_counts = {} - for node_id in range(node_count): + # Check subscribers (nodes 0 to nodes_to_subscribe-1) since they are the ones subscribed to the topic + for node_id in subscribed_node_ids: node_dir = f"shadow.data/hosts/node{node_id}" received_count = 0 @@ -50,21 +52,37 @@ def parse_shadow_logs(node_count: int) -> Dict[int, int]: return received_counts -def test_message_delivery(node_count: int) -> bool: - """Test that all nodes received messages from all publishers.""" +def test_message_delivery(node_count: int, topology_file: str) -> bool: + """Test that all subscribers received messages from all publishers.""" + topology = None + with open(topology_file, 'r') as f: + topology = json.load(f) + + mesh_nodes = len(topology["mesh_node_ids"]) + mesh_attester_nodes = len(topology["mesh_attester_node_ids"]) + non_mesh_attester_nodes = len(topology["non_mesh_attester_node_ids"]) + + nodes_to_subscribe = mesh_nodes + mesh_attester_nodes + nodes_to_publish = non_mesh_attester_nodes + mesh_attester_nodes + + subscribed_node_ids = topology["mesh_node_ids"] + topology["mesh_attester_node_ids"] + publisher_node_ids = topology["non_mesh_attester_node_ids"] + topology["mesh_attester_node_ids"] + print(f"Shadow GossipSub Simulation Test Results") print("=" * 60) + print(f"Total nodes: {node_count}, Mesh Nodes: {mesh_nodes}, Mesh Attesters: {mesh_attester_nodes}, Non Mesh Attesters: {non_mesh_attester_nodes}") + print(f"Testing message delivery to {nodes_to_subscribe} subscribers from {nodes_to_publish} publishers...") - received_counts = parse_shadow_logs(node_count) + received_counts = parse_shadow_logs(node_count, subscribed_node_ids) all_passed = True total_received = 0 - expected_messages = node_count # Each node publishes, so expect N messages per node + expected_messages = nodes_to_publish # Each subscriber should receive messages from all publishers - print(f"\nMessage Delivery (Expected: {expected_messages} messages per node)") + print(f"\nMessage Delivery to Subscribers (Expected: {expected_messages} messages per subscriber)") print("-" * 60) - for node_id in range(node_count): + for node_id in subscribed_node_ids: received = received_counts.get(node_id, 0) total_received += received @@ -74,19 +92,19 @@ def test_message_delivery(node_count: int) -> bool: status = "✗ FAIL" all_passed = False - print(f"Node {node_id:2d}: {received}/{expected_messages} messages {status}") + print(f"Subscriber {node_id:2d}: {received}/{expected_messages} messages {status}") print("-" * 60) - print(f"Total messages received: {total_received}/{node_count * expected_messages}") + print(f"Total messages received: {total_received}/{nodes_to_subscribe * expected_messages}") print() if all_passed: - print("✓ ALL TESTS PASSED: All nodes received all messages") + print("✓ ALL TESTS PASSED: All subscribers received all published messages") return True else: - failed_nodes = [i for i in range(node_count) if received_counts.get(i, 0) != expected_messages] - print(f"✗ TEST FAILED: {len(failed_nodes)} node(s) failed to receive all messages") - print(f"Failed nodes: {failed_nodes}") + failed_subscribers = [i for i in subscribed_node_ids if received_counts.get(i, 0) != expected_messages] + print(f"✗ TEST FAILED: {len(failed_subscribers)} subscriber(s) failed to receive all published messages") + print(f"Failed subscribers: {failed_subscribers}") return False def check_shadow_data_exists() -> bool: @@ -105,24 +123,21 @@ def check_shadow_data_exists() -> bool: def main(): """Main function.""" if len(sys.argv) < 2: - print("Usage: python3 test_results.py [node_count]") - print("Example: python3 test_results.py 10") + print("Usage: python3 test_results.py [node_count] [topology_file]") + print("Example: python3 test_results.py 10 topology.json") sys.exit(1) try: node_count = int(sys.argv[1]) + topology_file = sys.argv[2] except ValueError: - print("Error: node_count must be an integer") - sys.exit(1) - - if node_count < 1: - print("Error: node_count must be at least 1") + print("Error: node_count and topology_file must be valid") sys.exit(1) if not check_shadow_data_exists(): sys.exit(1) - success = test_message_delivery(node_count) + success = test_message_delivery(node_count, topology_file) sys.exit(0 if success else 1) if __name__ == "__main__": diff --git a/topic-name-increase-plots/30slots_batch_50ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png b/topic-name-increase-plots/30slots_batch_50ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png new file mode 100644 index 0000000..8b40211 Binary files /dev/null and b/topic-name-increase-plots/30slots_batch_50ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499.png differ diff --git a/topic-name-increase-plots/30slots_batch_5ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499_2.png b/topic-name-increase-plots/30slots_batch_5ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499_2.png new file mode 100644 index 0000000..1945683 Binary files /dev/null and b/topic-name-increase-plots/30slots_batch_5ms_node905_mesh_nodes390_mesh_attesters16_non_mesh_attesters499_2.png differ diff --git a/topology/gen/main.go b/topology/gen/main.go index 7d191a5..698ac1e 100644 --- a/topology/gen/main.go +++ b/topology/gen/main.go @@ -6,6 +6,7 @@ import ( "log" "os" + "github.com/ethp2p/attsim/simconfig" "github.com/ethp2p/attsim/topology" ) @@ -18,6 +19,16 @@ func main() { // Type-specific parameters branchingFactor = flag.Int("branching", 2, "Branching factor for tree topology") randomRegularDegree = flag.Int("degree", 3, "Node degree for random-regular topology") + nonMeshNodeDegree = flag.Int("non-mesh-node-degree", 3, "Node degree for non-mesh nodes") + + // node specific parameters + // number of nodes subscribed + // meshNodeCount + meshAttesterCount + nonMeshAttesterCount = nodeCount + meshNodeCount = flag.Int("mesh-nodes", 6, "Number of nodes for mesh topology") + meshAttesterCount = flag.Int("mesh-attester-count", 2, "Number of attesters for mesh topology") + nonMeshAttesterCount = flag.Int("non-mesh-attester-count", 2, "Number of attesters for non-mesh topology") + + simConfigFile = flag.String("simconfig-file", "", "Path to simulation config file") // Utility flags visualize = flag.Bool("visualize", false, "Print ASCII visualization of the topology") @@ -45,16 +56,53 @@ func main() { log.Fatal("Node count must be at least 2") } + // Check meshNodeIds + meshAttesterNodeIds + nonMeshAttesterNodeIds = NodeCount + if *meshNodeCount + *meshAttesterCount + *nonMeshAttesterCount != *nodeCount { + log.Fatalf("Mesh node count + mesh attester count + non mesh attester count must be equal to node count. meshNodeCount: %d, meshAttesterCount: %d, nonMeshAttesterCount: %d, nodeCount: %d", *meshNodeCount, *meshAttesterCount, *nonMeshAttesterCount, *nodeCount) + } + + // Load simulation config + if *simConfigFile == "" { + log.Fatal("Simulation config file is required. Use -simconfig-file flag to specify a simulation config file.") + } + + simConfig, err := simconfig.ParseSimConfig(*simConfigFile) + if err != nil { + log.Fatalf("Failed to load simulation config from file: %v", err) + } + + // reserve nodeIds for meshNodes, meshAttesterNodes and nonMeshAttesterNodes + // TODO - abstract these into methods in the topology package. + meshNodeIds := make(map[int]bool) + meshAttesterNodeIds := make(map[int]bool) + nonMeshAttesterNodeIds := make(map[int]bool) + + for i := 0; i < *meshNodeCount; i++ { + meshNodeIds[i] = true + } + + for i := 0; i < *meshAttesterCount; i++ { + meshAttesterNodeIds[*meshNodeCount + i] = true + } + + for i := 0; i < *nonMeshAttesterCount; i++ { + nonMeshAttesterNodeIds[*meshNodeCount + *meshAttesterCount + i] = true + } + + prysmNodeIds, lighthouseNodeIds := topology.SplitNodeIdsByClientSplit(*nodeCount, simConfig.ClientSplit.PrysmPercentage(), simConfig.ClientSplit.LighthousePercentage()) + // Create topology based on type var topo *topology.Topology + // TODO - let the topology generator methods take care of nodeId reservations. We should avoid doing it in the main function and passing + // in the nodeIds to the topology generator methods. switch *topologyType { case "mesh": - topo = topology.GenerateMesh(*nodeCount) + topo = topology.GenerateMesh(*nodeCount, meshNodeIds, meshAttesterNodeIds, nonMeshAttesterNodeIds, prysmNodeIds, lighthouseNodeIds) case "tree": - topo = topology.GenerateTree(*nodeCount, *branchingFactor) + topo = topology.GenerateTree(*nodeCount, *branchingFactor, meshNodeIds, meshAttesterNodeIds, nonMeshAttesterNodeIds, prysmNodeIds, lighthouseNodeIds) case "random-regular": - topo = topology.GenerateRandomRegular(*nodeCount, *randomRegularDegree) + topo = topology.GenerateRandomRegular(*nodeCount, *randomRegularDegree, *nonMeshNodeDegree, meshNodeIds, meshAttesterNodeIds, nonMeshAttesterNodeIds, prysmNodeIds, lighthouseNodeIds) default: log.Fatalf("Unknown topology type: %s. Supported types: mesh, tree, random-regular", *topologyType) } @@ -67,12 +115,12 @@ func main() { // Print statistics if requested if *stats { - printTopologyStats(topo) + topo.PrintTopologyStats() } // Visualize if requested if *visualize { - visualizeTopology(topo) + topo.VisualizeTopology() } // Save to file @@ -81,100 +129,4 @@ func main() { } fmt.Printf("Topology saved to: %s\n", *output) -} - -func printTopologyStats(topo *topology.Topology) { - fmt.Println("\n=== Topology Statistics ===") - - connections := topo.GetAllConnections() - nodeCount := topo.NodeCount - - // Count degree of each node - degree := make(map[int]int) - for _, conn := range connections { - degree[conn.From]++ - degree[conn.To]++ - } - - // Calculate statistics - minDegree := nodeCount - maxDegree := 0 - totalDegree := 0 - - for i := 0; i < nodeCount; i++ { - d := degree[i] - if d < minDegree { - minDegree = d - } - if d > maxDegree { - maxDegree = d - } - totalDegree += d - } - - avgDegree := float64(totalDegree) / float64(nodeCount) - - fmt.Printf("Nodes: %d\n", nodeCount) - fmt.Printf("Edges: %d\n", len(connections)) - fmt.Printf("Average degree: %.2f\n", avgDegree) - fmt.Printf("Min degree: %d\n", minDegree) - fmt.Printf("Max degree: %d\n", maxDegree) - - // Show degree distribution for small networks - if nodeCount <= 20 { - fmt.Println("\nDegree distribution:") - for i := 0; i < nodeCount; i++ { - fmt.Printf(" Node %2d: degree %d\n", i, degree[i]) - } - } -} - -func visualizeTopology(topo *topology.Topology) { - connections := topo.GetAllConnections() - nodeCount := topo.NodeCount - - if nodeCount > 20 { - fmt.Println("\n(Visualization skipped for networks with more than 20 nodes)") - return - } - - fmt.Println("\n=== Topology Visualization ===") - fmt.Println("Adjacency Matrix (1 = connected, 0 = not connected):") - - // Create adjacency matrix - matrix := make([][]bool, nodeCount) - for i := range matrix { - matrix[i] = make([]bool, nodeCount) - } - - for _, conn := range connections { - matrix[conn.From][conn.To] = true - matrix[conn.To][conn.From] = true - } - - // Print header - fmt.Print(" ") - for i := 0; i < nodeCount; i++ { - fmt.Printf("%2d ", i) - } - fmt.Println() - - // Print matrix - for i := 0; i < nodeCount; i++ { - fmt.Printf("%2d: ", i) - for j := 0; j < nodeCount; j++ { - if matrix[i][j] { - fmt.Print(" 1 ") - } else { - fmt.Print(" . ") - } - } - fmt.Println() - } - - // Print edge list for clarity - fmt.Println("\nEdge List:") - for _, conn := range connections { - fmt.Printf(" %d <-> %d\n", conn.From, conn.To) - } -} +} \ No newline at end of file diff --git a/topology/generators.go b/topology/generators.go index d4b007f..e1b6f7b 100644 --- a/topology/generators.go +++ b/topology/generators.go @@ -1,10 +1,47 @@ package topology -import () +import ( + "log" +) + + +func SplitNodeIdsByClientSplit(nodeCount int, prysmPercentage float64, lighthousePercentage float64) (map[int]bool, map[int]bool) { + prysmNodeCount := int(float64(nodeCount) * prysmPercentage) + lighthouseNodeCount := int(float64(nodeCount) * lighthousePercentage) + + log.Printf("Prysm node percentage: %f, Lighthouse node percentage: %f", prysmPercentage, lighthousePercentage) + log.Printf("Prysm node count: %d, Lighthouse node count: %d", prysmNodeCount, lighthouseNodeCount) + + prysmNodeIds := make(map[int]bool) + lighthouseNodeIds := make(map[int]bool) + + allNodeIds := make([]int, nodeCount) + for i := 0; i < nodeCount; i++ { + allNodeIds[i] = i + } + + // Shuffle the node IDs for random allocation + for i := len(allNodeIds) - 1; i > 0; i-- { + j := (i * 17 + 42) % (i + 1) // Simple deterministic shuffle + allNodeIds[i], allNodeIds[j] = allNodeIds[j], allNodeIds[i] + } + + // Assign first prysmNodeCount nodes to Prysm + for i := 0; i < prysmNodeCount; i++ { + prysmNodeIds[allNodeIds[i]] = true + } + + // Assign remaining nodes to Lighthouse + for i := prysmNodeCount; i < nodeCount; i++ { + lighthouseNodeIds[allNodeIds[i]] = true + } + + return prysmNodeIds, lighthouseNodeIds +} // GenerateMesh creates a fully connected mesh topology -func GenerateMesh(nodeCount int) *Topology { - topo := NewTopology(nodeCount) +func GenerateMesh(nodeCount int, meshNodeIds map[int]bool, meshAttesterNodeIds map[int]bool, nonMeshAttesterNodeIds map[int]bool, prysmNodeIds map[int]bool, lighthouseNodeIds map[int]bool) *Topology { + topo := NewTopology(nodeCount, meshNodeIds, meshAttesterNodeIds, nonMeshAttesterNodeIds, prysmNodeIds, lighthouseNodeIds) for i := 0; i < nodeCount; i++ { for j := i + 1; j < nodeCount; j++ { @@ -16,8 +53,8 @@ func GenerateMesh(nodeCount int) *Topology { } // GenerateTree creates a tree topology with specified branching factor -func GenerateTree(nodeCount int, branchingFactor int) *Topology { - topo := NewTopology(nodeCount) +func GenerateTree(nodeCount int, branchingFactor int, meshNodeIds map[int]bool, meshAttesterNodeIds map[int]bool, nonMeshAttesterNodeIds map[int]bool, prysmNodeIds map[int]bool, lighthouseNodeIds map[int]bool) *Topology { + topo := NewTopology(nodeCount, meshNodeIds, meshAttesterNodeIds, nonMeshAttesterNodeIds, prysmNodeIds, lighthouseNodeIds) if branchingFactor < 2 { branchingFactor = 2 @@ -36,13 +73,7 @@ func GenerateTree(nodeCount int, branchingFactor int) *Topology { return topo } -// GenerateRandomRegular creates a random regular graph where each node has exactly 'degree' connections -// to randomly selected other nodes. Guarantees connectivity by checking and regenerating until connected. -func GenerateRandomRegular(nodeCount int, degree int) *Topology { - if nodeCount < 2 { - return NewTopology(nodeCount) - } - +func NormalizeDegree(degree, nodeCount int) int { if degree < 1 { degree = 2 // Minimum degree for connectivity } @@ -60,10 +91,23 @@ func GenerateRandomRegular(nodeCount int, degree int) *Topology { degree-- // Decrease if we're at maximum } } + + return degree +} + +// GenerateRandomRegular creates a random regular graph where each node has exactly 'degree' connections +// to randomly selected other nodes. Guarantees connectivity by checking and regenerating until connected. +func GenerateRandomRegular(nodeCount int, degree int, nonMeshNodeDegree int,meshNodeIds map[int]bool, meshAttesterNodeIds map[int]bool, nonMeshAttesterNodeIds map[int]bool, prysmNodeIds map[int]bool, lighthouseNodeIds map[int]bool) *Topology { + if nodeCount < 2 { + return NewTopology(nodeCount, meshNodeIds, meshAttesterNodeIds, nonMeshAttesterNodeIds, prysmNodeIds, lighthouseNodeIds) + } + + degree = NormalizeDegree(degree, nodeCount) + nonMeshNodeDegree = NormalizeDegree(nonMeshNodeDegree, len(meshAttesterNodeIds) + len(meshNodeIds)); // Keep trying until we generate a connected graph for attempt := 0; ; attempt++ { - topo := NewTopology(nodeCount) + topo := NewTopology(nodeCount, meshNodeIds, meshAttesterNodeIds, nonMeshAttesterNodeIds, prysmNodeIds, lighthouseNodeIds) // Use attempt number as additional seed for variety seed := attempt * 137 @@ -71,30 +115,39 @@ func GenerateRandomRegular(nodeCount int, degree int) *Topology { // Use deterministic "random" selection based on node hash for reproducibility // This ensures consistent results across runs while appearing random for nodeID := 0; nodeID < nodeCount; nodeID++ { + + degreeToUse := degree + _, ok := nonMeshAttesterNodeIds[nodeID] + if ok { + degreeToUse = nonMeshNodeDegree + } + + // fmt.Printf("Connecting nodeID, %d\n", nodeID) connections := make(map[int]bool) connections[nodeID] = true // Can't connect to self - for len(connections)-1 < degree { // -1 because we exclude self + for len(connections)-1 < degreeToUse { // -1 because we exclude self + // fmt.Println("len(connections)", len(connections)) // Generate pseudo-random target using hash function with attempt seed - hash := (nodeID*17 + len(connections)*23 + seed + 42) % nodeCount + // we want to pick a nodeId of a node only in the mesh. + hash := (nodeID*17 + len(connections)*23 + seed + 42) % (len(meshNodeIds) + len(meshAttesterNodeIds)) target := hash // If target is already connected or is self, find next available for connections[target] { - target = (target + 1) % nodeCount + // fmt.Println("target", target, "is already connected, finding next available") + target = (target + 1) % (len(meshNodeIds) + len(meshAttesterNodeIds)) } connections[target] = true // Add bidirectional connection (avoid duplicates) - if nodeID < target { - topo.AddConnection(nodeID, target) - } + topo.AddConnection(nodeID, target) } } // Check if the graph is connected - if topo.IsConnected() { + if topo.IsConnected() && topo.AreNonMeshNodesConnectedToMeshNodes() { return topo } } diff --git a/topology/generators_test.go b/topology/generators_test.go new file mode 100644 index 0000000..62cf235 --- /dev/null +++ b/topology/generators_test.go @@ -0,0 +1,85 @@ +package topology + +import ( + "testing" + + "github.com/stretchr/testify/require" +) + +func TestRandomRegularTopology(t *testing.T) { + testCases := []struct { + description string + nodeCount int + degree int + nonMeshNodeDegree int + meshNodeCount int + meshAttesterCount int + nonMeshAttesterCount int + }{ + {"10 node network", 10, 4, 2, 5, 3, 2}, + {"20 node network", 20, 6, 3, 10, 5, 5}, + {"200 node network", 200, 25, 10, 32, 64, 100}, + {"905 node network", 905, 25, 10, 16, 380, 499}, + } + + for _, testCase := range testCases { + t.Run(testCase.description, func(t *testing.T) { + meshNodeIds := make(map[int]bool) + meshAttesterNodeIds := make(map[int]bool) + nonMeshAttesterNodeIds := make(map[int]bool) + + for i := 0; i < testCase.meshNodeCount; i++ { + meshNodeIds[i] = true + } + + for i := 0; i < testCase.meshAttesterCount; i++ { + meshAttesterNodeIds[testCase.meshNodeCount+i] = true + } + + for i := 0; i < testCase.nonMeshAttesterCount; i++ { + nonMeshAttesterNodeIds[testCase.meshNodeCount+testCase.meshAttesterCount+i] = true + } + + prysmNodeIds, lighthouseNodeIds := SplitNodeIdsByClientSplit(testCase.nodeCount, 0.5, 0.5) + + topo := GenerateRandomRegular(testCase.nodeCount, testCase.degree, testCase.nonMeshNodeDegree,meshNodeIds, meshAttesterNodeIds, nonMeshAttesterNodeIds, prysmNodeIds, lighthouseNodeIds) + require.NotNil(t, topo, "Topology should not be nil") + + require.True(t, topo.IsConnected(), "Topology should be connected") + + require.True(t, topo.AreNonMeshNodesConnectedToMeshNodes(), "Non-mesh attester nodes should be connected to mesh nodes or mesh attesters") + + normalizedNonMeshNodeDegree := NormalizeDegree(testCase.nonMeshNodeDegree, len(meshNodeIds) + len(meshAttesterNodeIds)) + + // check degree of non mesh nodes + for _, nodeId := range topo.NonMeshAttesterNodeIds { + connections := topo.GetConnections(nodeId) + require.Equal(t, normalizedNonMeshNodeDegree, len(connections), "Degree of non mesh attester node should be equal to non mesh node degree") + } + }) + } + +} + +func TestClientSplit(t *testing.T) { + testCases := []struct { + description string + nodeCount int + prysmPercentage float64 + lighthousePercentage float64 + }{ + {"50% Prysm, 50% Lighthouse", 10, 0.5, 0.5}, + {"70% Prysm, 30% Lighthouse", 20, 0.7, 0.3}, + {"30% Prysm, 70% Lighthouse", 30, 0.3, 0.7}, + } + + for _, testCase := range testCases { + t.Run(testCase.description, func(t *testing.T) { + prysmNodeIds, lighthouseNodeIds := SplitNodeIdsByClientSplit(testCase.nodeCount, testCase.prysmPercentage, testCase.lighthousePercentage) + require.NotNil(t, prysmNodeIds, "Prysm node IDs should not be nil") + require.NotNil(t, lighthouseNodeIds, "Lighthouse node IDs should not be nil") + require.Equal(t, testCase.nodeCount, len(prysmNodeIds) + len(lighthouseNodeIds), "Total node count should be equal to the sum of Prysm and Lighthouse node counts") + }) + } + +} \ No newline at end of file diff --git a/topology/topology.go b/topology/topology.go index 45ea87a..3e6e171 100644 --- a/topology/topology.go +++ b/topology/topology.go @@ -4,6 +4,7 @@ import ( "encoding/json" "fmt" "os" + "slices" ) // Connection represents a network connection between two nodes @@ -14,17 +15,47 @@ type Connection struct { // Topology represents a network topology as an adjacency list type Topology struct { - NodeCount int `json:"node_count"` - Connections []Connection `json:"connections"` - adjacencyList map[int][]int + NodeCount int `json:"node_count"` + Connections []Connection `json:"connections"` + adjacencyList map[int][]int + MeshNodeIds []int `json:"mesh_node_ids"` + MeshAttesterNodeIds []int `json:"mesh_attester_node_ids"` + NonMeshAttesterNodeIds []int `json:"non_mesh_attester_node_ids"` + PrysmNodeids []int `json:"prysm_node_ids"` + LighthouseNodeids []int `json:"lighthouse_node_ids"` } // NewTopology creates a new empty topology -func NewTopology(nodeCount int) *Topology { +func NewTopology(nodeCount int, meshNodeIds map[int]bool, meshAttesterNodeIds map[int]bool, nonMeshAttesterNodeIds map[int]bool, prysmNodeIds map[int]bool, lighthouseNodeIds map[int]bool) *Topology { + meshNodeIdsList := make([]int, 0, len(meshNodeIds)) + for nodeId := range meshNodeIds { + meshNodeIdsList = append(meshNodeIdsList, nodeId) + } + meshAttesterNodeIdsList := make([]int, 0, len(meshAttesterNodeIds)) + for nodeId := range meshAttesterNodeIds { + meshAttesterNodeIdsList = append(meshAttesterNodeIdsList, nodeId) + } + nonMeshAttesterNodeIdsList := make([]int, 0, len(nonMeshAttesterNodeIds)) + for nodeId := range nonMeshAttesterNodeIds { + nonMeshAttesterNodeIdsList = append(nonMeshAttesterNodeIdsList, nodeId) + } + prysmNodeIdsList := make([]int, 0, len(prysmNodeIds)) + for nodeId := range prysmNodeIds { + prysmNodeIdsList = append(prysmNodeIdsList, nodeId) + } + lighthouseNodeIdsList := make([]int, 0, len(lighthouseNodeIds)) + for nodeId := range lighthouseNodeIds { + lighthouseNodeIdsList = append(lighthouseNodeIdsList, nodeId) + } return &Topology{ - NodeCount: nodeCount, - Connections: []Connection{}, - adjacencyList: make(map[int][]int), + NodeCount: nodeCount, + Connections: []Connection{}, + adjacencyList: make(map[int][]int), + MeshNodeIds: meshNodeIdsList, + MeshAttesterNodeIds: meshAttesterNodeIdsList, + NonMeshAttesterNodeIds: nonMeshAttesterNodeIdsList, + PrysmNodeids: prysmNodeIdsList, + LighthouseNodeids: lighthouseNodeIdsList, } } @@ -138,3 +169,111 @@ func (t *Topology) IsConnected() bool { // Check if all nodes were visited return len(visited) == t.NodeCount } + +func (t *Topology) AreNonMeshNodesConnectedToMeshNodes() bool { + for _, nodeId := range t.NonMeshAttesterNodeIds { + connections := t.GetConnections(nodeId) + for _, connection := range connections { + if !slices.Contains(t.MeshNodeIds, connection) && !slices.Contains(t.MeshAttesterNodeIds, connection) { + return false + } + } + } + return true +} + +func (topo *Topology) PrintTopologyStats() { + fmt.Println("\n=== Topology Statistics ===") + + connections := topo.GetAllConnections() + nodeCount := topo.NodeCount + + // Count degree of each node + degree := make(map[int]int) + for _, conn := range connections { + degree[conn.From]++ + degree[conn.To]++ + } + + // Calculate statistics + minDegree := nodeCount + maxDegree := 0 + totalDegree := 0 + + for i := 0; i < nodeCount; i++ { + d := degree[i] + if d < minDegree { + minDegree = d + } + if d > maxDegree { + maxDegree = d + } + totalDegree += d + } + + avgDegree := float64(totalDegree) / float64(nodeCount) + + fmt.Printf("Nodes: %d\n", nodeCount) + fmt.Printf("Edges: %d\n", len(connections)) + fmt.Printf("Average degree: %.2f\n", avgDegree) + fmt.Printf("Min degree: %d\n", minDegree) + fmt.Printf("Max degree: %d\n", maxDegree) + + // Show degree distribution for small networks + if nodeCount <= 20 { + fmt.Println("\nDegree distribution:") + for i := 0; i < nodeCount; i++ { + fmt.Printf(" Node %2d: degree %d\n", i, degree[i]) + } + } +} + +func (topo *Topology) VisualizeTopology() { + connections := topo.GetAllConnections() + nodeCount := topo.NodeCount + + if nodeCount > 20 { + fmt.Println("\n(Visualization skipped for networks with more than 20 nodes)") + return + } + + fmt.Println("\n=== Topology Visualization ===") + fmt.Println("Adjacency Matrix (1 = connected, 0 = not connected):") + + // Create adjacency matrix + matrix := make([][]bool, nodeCount) + for i := range matrix { + matrix[i] = make([]bool, nodeCount) + } + + for _, conn := range connections { + matrix[conn.From][conn.To] = true + matrix[conn.To][conn.From] = true + } + + // Print header + fmt.Print(" ") + for i := 0; i < nodeCount; i++ { + fmt.Printf("%2d ", i) + } + fmt.Println() + + // Print matrix + for i := 0; i < nodeCount; i++ { + fmt.Printf("%2d: ", i) + for j := 0; j < nodeCount; j++ { + if matrix[i][j] { + fmt.Print(" 1 ") + } else { + fmt.Print(" . ") + } + } + fmt.Println() + } + + // Print edge list for clarity + fmt.Println("\nEdge List:") + for _, conn := range connections { + fmt.Printf(" %d <-> %d\n", conn.From, conn.To) + } +}