From e328fa0c8649f1be17cc9f83c798bc8022ccd6ab Mon Sep 17 00:00:00 2001 From: aya Date: Wed, 14 May 2025 18:53:27 +0300 Subject: [PATCH] save results to csv file --- utils/utils.go | 1 + waku/nwaku_test_utils.go | 14 ++++ waku/stress_test.go | 141 +++++++-------------------------------- 3 files changed, 39 insertions(+), 117 deletions(-) diff --git a/utils/utils.go b/utils/utils.go index c1f03bd..0df2b6b 100644 --- a/utils/utils.go +++ b/utils/utils.go @@ -75,3 +75,4 @@ func GetRSSKB() (uint64, error) { pageSize := os.Getpagesize() return (rssPages * uint64(pageSize)) / 1024, nil } + diff --git a/waku/nwaku_test_utils.go b/waku/nwaku_test_utils.go index a2d64a0..4da8272 100644 --- a/waku/nwaku_test_utils.go +++ b/waku/nwaku_test_utils.go @@ -9,6 +9,7 @@ import ( "io" "net/http" "os" + "runtime" "strconv" "sync" "time" @@ -16,6 +17,7 @@ import ( "github.com/cenkalti/backoff/v3" "github.com/libp2p/go-libp2p/core/peer" "github.com/waku-org/go-waku/waku/v2/protocol/pb" + "github.com/waku-org/waku-go-bindings/utils" "github.com/waku-org/waku-go-bindings/waku/common" "google.golang.org/protobuf/proto" ) @@ -288,3 +290,15 @@ func recordMemoryMetricsPX(testName, phase string, heapAllocKB, rssKB uint64) er } return writer.Write(row) } + +func captureMemory(testName, phase string) { + var ms runtime.MemStats + runtime.ReadMemStats(&ms) + + heapKB := ms.HeapAlloc / 1024 + rssKB, _ := utils.GetRSSKB() + + Debug("[%s] Memory usage (%s): %d KB (RSS %d KB)", testName, phase, heapKB, rssKB) + + _ = recordMemoryMetricsPX(testName, phase, heapKB, rssKB) +} \ No newline at end of file diff --git a/waku/stress_test.go b/waku/stress_test.go index ba7c3cc..dc65b65 100644 --- a/waku/stress_test.go +++ b/waku/stress_test.go @@ -11,7 +11,6 @@ import ( "time" "github.com/stretchr/testify/require" - "github.com/waku-org/waku-go-bindings/utils" "github.com/waku-org/waku-go-bindings/waku/common" // "go.uber.org/zap/zapcore" @@ -21,9 +20,7 @@ import ( func TestStressMemoryUsageForThreeNodes(t *testing.T) { testName := t.Name() var err error - var memStats runtime.MemStats - runtime.ReadMemStats(&memStats) - Debug("[%s] Memory usage BEFORE creating nodes: %d KB", testName, memStats.HeapAlloc/1024) + captureMemory(t.Name(), "start") node1Cfg := DefaultWakuConfig node1Cfg.TcpPort, node1Cfg.Discv5UdpPort, err = GetFreePortIfNeeded(0, 0) require.NoError(t, err) @@ -41,8 +38,7 @@ func TestStressMemoryUsageForThreeNodes(t *testing.T) { node3, err := NewWakuNode(&node3Cfg, "node3") require.NoError(t, err) - runtime.ReadMemStats(&memStats) - Debug("[%s] Memory usage AFTER creating nodes: %d KB", testName, memStats.HeapAlloc/1024) + captureMemory(t.Name(), "before nodes start") err = node1.Start() require.NoError(t, err) @@ -51,8 +47,7 @@ func TestStressMemoryUsageForThreeNodes(t *testing.T) { err = node3.Start() require.NoError(t, err) - runtime.ReadMemStats(&memStats) - Debug("[%s] Memory usage AFTER starting nodes: %d KB", testName, memStats.HeapAlloc/1024) + captureMemory(t.Name(), "after nodes run") time.Sleep(2 * time.Second) @@ -64,8 +59,7 @@ func TestStressMemoryUsageForThreeNodes(t *testing.T) { time.Sleep(1 * time.Second) runtime.GC() - runtime.ReadMemStats(&memStats) - Debug("[%s] Memory usage AFTER destroying nodes: %d KB", testName, memStats.HeapAlloc/1024) + captureMemory(t.Name(), "at end") Debug("[%s] Test completed successfully", testName) } @@ -91,12 +85,10 @@ func TestStressStoreQuery5kMessagesWithPagination(t *testing.T) { node2.StopAndDestroy() }() - var memStats runtime.MemStats - iterations := 2000 + + iterations := 5000 - runtime.ReadMemStats(&memStats) - initialHeapAlloc := memStats.HeapAlloc - Debug("Initial memory usage check before publishing %d MB", initialHeapAlloc/1024/1024) + captureMemory(t.Name(), "at start") queryTimestamp := proto.Int64(time.Now().UnixNano()) @@ -107,9 +99,6 @@ func TestStressStoreQuery5kMessagesWithPagination(t *testing.T) { require.NoError(t, err, "Failed to publish message") if i%10 == 0 { - runtime.ReadMemStats(&memStats) - Debug("Memory usage at iteration %d: HeapAlloc=%v MB, NumGC=%v", - i, memStats.HeapAlloc/1024/1024, memStats.NumGC) storeQueryRequest := &common.StoreQueryRequest{ TimeStart: queryTimestamp, @@ -123,11 +112,7 @@ func TestStressStoreQuery5kMessagesWithPagination(t *testing.T) { } } - runtime.ReadMemStats(&memStats) - finalHeapAlloc := memStats.HeapAlloc - Debug("Memory before test: %v KB, Memory after test: %v KB", initialHeapAlloc/1024, finalHeapAlloc/1024) - - //require.LessOrEqual(t, finalHeapAlloc, initialHeapAlloc*2, "Memory usage has grown too much") + captureMemory(t.Name(), "at end") Debug("[%s] Test completed successfully", t.Name()) } @@ -149,15 +134,9 @@ func TestStressHighThroughput10kPublish(t *testing.T) { err = node1.ConnectPeer(node2) require.NoError(t, err, "Failed to connect node1 to node2") - var memStats runtime.MemStats - runtime.ReadMemStats(&memStats) - startHeapKB := memStats.HeapAlloc / 1024 - startRSSKB, err := utils.GetRSSKB() - require.NoError(t, err, "Failed to read initial RSS") + captureMemory(t.Name(), "at start") - Debug("Memory usage BEFORE sending => HeapAlloc: %d KB, RSS: %d KB", startHeapKB, startRSSKB) - - totalMessages := 6000 + totalMessages := 2000 pubsubTopic := DefaultPubsubTopic for i := 0; i < totalMessages; i++ { @@ -170,21 +149,11 @@ func TestStressHighThroughput10kPublish(t *testing.T) { Debug("###Iteration number#%d", i) } - runtime.ReadMemStats(&memStats) - endHeapKB := memStats.HeapAlloc / 1024 - endRSSKB, err := utils.GetRSSKB() - require.NoError(t, err, "Failed to read final RSS") - - Debug("Memory usage AFTER sending => HeapAlloc: %d KB, RSS: %d KB", endHeapKB, endRSSKB) + captureMemory(t.Name(), "at end") } -func TestStressConnectDisconnect500Iteration(t *testing.T) { - var memStats runtime.MemStats - runtime.ReadMemStats(&memStats) - startHeapKB := memStats.HeapAlloc / 1024 - startRSSKB, err := utils.GetRSSKB() - require.NoError(t, err) - Debug("Before test: HeapAlloc = %d KB, RSS = %d KB", startHeapKB, startRSSKB) +func TestStressConnectDisconnect1kIteration(t *testing.T) { + captureMemory(t.Name(), "at start") node0Cfg := DefaultWakuConfig node0Cfg.Relay = true @@ -219,11 +188,7 @@ func TestStressConnectDisconnect500Iteration(t *testing.T) { Debug("Iteration %d: node0 disconnected from node1", i) time.Sleep(2 * time.Second) } - runtime.ReadMemStats(&memStats) - endHeapKB := memStats.HeapAlloc / 1024 - endRSSKB, err := utils.GetRSSKB() - require.NoError(t, err) - Debug("After test: HeapAlloc = %d KB, RSS = %d KB", endHeapKB, endRSSKB) + captureMemory(t.Name(), "at end") } func TestStressRandomNodesInMesh(t *testing.T) { @@ -245,12 +210,7 @@ func TestStressRandomNodesInMesh(t *testing.T) { time.Sleep(1 * time.Second) require.NoError(t, err, "Failed to connect initial nodes with ConnectAllPeers") - var memStats runtime.MemStats - runtime.ReadMemStats(&memStats) - startHeapKB := memStats.HeapAlloc / 1024 - startRSSKB, err2 := utils.GetRSSKB() - require.NoError(t, err2, "Failed to read initial RSS") - Debug("Memory at start of test: HeapAlloc=%d KB, RSS=%d KB", startHeapKB, startRSSKB) + captureMemory(t.Name(), "at start") testDuration := 30 * time.Minute endTime := time.Now().Add(testDuration) @@ -308,11 +268,7 @@ func TestStressRandomNodesInMesh(t *testing.T) { n.StopAndDestroy() } - runtime.ReadMemStats(&memStats) - endHeapKB := memStats.HeapAlloc / 1024 - endRSSKB, err3 := utils.GetRSSKB() - require.NoError(t, err3, "Failed to read final RSS") - Debug("Memory at end of test: HeapAlloc=%d KB, RSS=%d KB", endHeapKB, endRSSKB) + captureMemory(t.Name(), "at end") } func TestStressLargePayloadEphemeralMessagesEndurance(t *testing.T) { @@ -340,12 +296,7 @@ func TestStressLargePayloadEphemeralMessagesEndurance(t *testing.T) { time.Sleep(2 * time.Second) - var memStats runtime.MemStats - runtime.ReadMemStats(&memStats) - startHeapKB := memStats.HeapAlloc / 1024 - startRSSKB, err := utils.GetRSSKB() - require.NoError(t, err) - Debug("Before endurance test: HeapAlloc = %d KB, RSS = %d KB", startHeapKB, startRSSKB) + captureMemory(t.Name(), "at start") maxIterations := 5000 payloadSize := 100 * 1024 @@ -372,25 +323,14 @@ func TestStressLargePayloadEphemeralMessagesEndurance(t *testing.T) { Debug("###Iteration number %d", i+1) } - runtime.ReadMemStats(&memStats) - endHeapKB := memStats.HeapAlloc / 1024 - endRSSKB, err := utils.GetRSSKB() - require.NoError(t, err) - Debug("After endurance test: HeapAlloc = %d KB, RSS = %d KB", endHeapKB, endRSSKB) + captureMemory(t.Name(), "at end") } -func TestStress2Nodes500IterationTearDown(t *testing.T) { - - var memStats runtime.MemStats - runtime.ReadMemStats(&memStats) - initialMem := memStats.HeapAlloc - Debug("[%s] Memory usage at test START: %d KB", t.Name(), initialMem/1024) - - initialRSS, err := utils.GetRSSKB() - require.NoError(t, err) - Debug("[%s] OS-level RSS at test START: %d KB", t.Name(), initialRSS) +func TestStress2Nodes2kIterationTearDown(t *testing.T) { + captureMemory(t.Name(), "at start") + var err error totalIterations := 2000 for i := 1; i <= totalIterations; i++ { var nodes []*WakuNode @@ -421,32 +361,16 @@ func TestStress2Nodes500IterationTearDown(t *testing.T) { runtime.GC() time.Sleep(250 * time.Millisecond) runtime.GC() - if i == 250 || i == 500 { - runtime.ReadMemStats(&memStats) - Debug("Iteration %d, usage after teardown: %d KB", i, memStats.HeapAlloc/1024) - require.LessOrEqual(t, memStats.HeapAlloc, initialMem*3, "Memory usage soared above threshold after iteration %d", i) - rssNow, err := utils.GetRSSKB() - require.NoError(t, err) - Debug("Iteration %d, OS-level RSS after teardown: %d KB", i, rssNow) - //require.LessOrEqual(t, rssNow, initialRSS*10, "OS-level RSS soared above threshold after iteration %d", i) - } Debug("Iteration numberrrrrr %d", i) } runtime.GC() time.Sleep(500 * time.Millisecond) runtime.GC() - runtime.ReadMemStats(&memStats) - finalMem := memStats.HeapAlloc - Debug("[%s] Memory usage at test END: %d KB", t.Name(), finalMem/1024) - // require.LessOrEqual(t, finalMem, initialMem*3, "Memory usage soared above threshold after %d cycles", totalIterations) - finalRSS, err := utils.GetRSSKB() - require.NoError(t, err) - Debug("[%s] OS-level RSS at test END: %d KB", t.Name(), finalRSS) + captureMemory(t.Name(), "at end") //require.LessOrEqual(t, finalRSS, initialRSS*3, "OS-level RSS soared above threshold after %d cycles", totalIterations) } func TestPeerExchangePXLoad(t *testing.T) { - testName := "PeerExchangePXLoad" pxServerCfg := DefaultWakuConfig pxServerCfg.PeerExchange = true pxServerCfg.Relay = true @@ -469,16 +393,7 @@ func TestPeerExchangePXLoad(t *testing.T) { time.Sleep(2 * time.Second) - var memStats runtime.MemStats - runtime.ReadMemStats(&memStats) - startHeapKB := memStats.HeapAlloc / 1024 - startRSSKB, err := utils.GetRSSKB() - require.NoError(t, err, "Failed to get initial RSS") - Debug("%s: Before test: HeapAlloc=%d KB, RSS=%d KB", testName, startHeapKB, startRSSKB) - - // Save the initial memory reading to CSV - err = recordMemoryMetricsPX(testName, "start", startHeapKB, startRSSKB) - require.NoError(t, err, "Failed to record start metrics") + captureMemory(t.Name(), "at start") testDuration := 30 * time.Minute endTime := time.Now().Add(testDuration) @@ -514,13 +429,5 @@ func TestPeerExchangePXLoad(t *testing.T) { time.Sleep(1 * time.Second) } - runtime.ReadMemStats(&memStats) - endHeapKB := memStats.HeapAlloc / 1024 - endRSSKB, err := utils.GetRSSKB() - require.NoError(t, err, "Failed to get final RSS") - Debug("Memory %s: After test: HeapAlloc=%d KB, RSS=%d KB", testName, endHeapKB, endRSSKB) - - // Save the final memory reading to CSV - err = recordMemoryMetricsPX(testName, "end", endHeapKB, endRSSKB) - require.NoError(t, err, "Failed to record end metrics") + captureMemory(t.Name(), "at end") }