feat(throughput): add NotReceivedP2PEventCount to ThroughoutMeasure

Author: tosettil-polimi

pkg/coordinator/tasks/tx_pool_throughput_analysis/task.go

@@ -27,8 +27,9 @@ var (
 )
 
 type ThroughoutMeasure struct {
-	LoadTPS      int `json:"load_tps"`
-	ProcessedTPS int `json:"processed_tps"`
+	LoadTPS                  int `json:"load_tps"`
+	ProcessedTPS             int `json:"processed_tps"`
+	NotReceivedP2PEventCount int `json:"not_received_p2p_event_count"`
 }
 
 type Task struct {
@@ -136,9 +137,11 @@ func (t *Task) Execute(ctx context.Context) error {
 	t.logger.Infof("Iterating over the TPS range, starting TPS: %d, ending TPS: %d, increment TPS: %d",
 		t.config.StartingTPS, t.config.EndingTPS, t.config.IncrementTPS)
 
+	missedP2PEventCount := 0
+
 	for sendingTps := t.config.StartingTPS; sendingTps <= t.config.EndingTPS; sendingTps += t.config.IncrementTPS {
 		// measure the throughput with the current sendingTps
-		processedTps, err := t.measureTpsWithLoad(loadTarget, sendingTps, t.config.DurationS, singleMeasureDeadline, percentile)
+		processedTps, notReceivedP2PEventCount, err := t.measureTpsWithLoad(loadTarget, sendingTps, t.config.DurationS, singleMeasureDeadline, percentile)
 		if err != nil {
 			t.logger.Errorf("Error during throughput measurement with sendingTps=%d, duration=%d: %v", sendingTps, t.config.DurationS, err)
 			t.ctx.SetResult(types.TaskResultFailure)
@@ -148,9 +151,12 @@ func (t *Task) Execute(ctx context.Context) error {
 
 		// add to throughoutMeasures
 		throughoutMeasures = append(throughoutMeasures, ThroughoutMeasure{
-			LoadTPS:      sendingTps,
-			ProcessedTPS: processedTps,
+			LoadTPS:                  sendingTps,
+			ProcessedTPS:             processedTps,
+			NotReceivedP2PEventCount: notReceivedP2PEventCount,
 		})
+
+		missedP2PEventCount += notReceivedP2PEventCount
 	}
 
 	t.logger.Infof("Finished measuring throughput, collected %d measures", len(throughoutMeasures))
@@ -160,7 +166,8 @@ func (t *Task) Execute(ctx context.Context) error {
 	t.ctx.Outputs.SetVar("throughput_measures", throughoutMeasures) // log coordinated_omission_event_count and missed_p2p_event_count?
 
 	outputs := map[string]interface{}{
-		"throughput_measures": throughoutMeasures,
+		"throughput_measures":    throughoutMeasures,
+		"missed_p2p_event_count": missedP2PEventCount,
 	}
 
 	outputsJSON, _ := json.Marshal(outputs)
@@ -173,7 +180,7 @@ func (t *Task) Execute(ctx context.Context) error {
 }
 
 func (t *Task) measureTpsWithLoad(loadTarget *txloadtool.LoadTarget, sendingTps, durationS int,
-	testDeadline time.Time, percentile float64) (int, error) {
+	testDeadline time.Time, percentile float64) (processedTps, notReceivedP2PEventCount int, err error) {
 	t.logger.Infof("Single measure of throughput, sending TPS: %d, duration: %d secs", sendingTps, durationS)
 
 	// Prepare to collect transaction latencies
@@ -185,7 +192,7 @@ func (t *Task) measureTpsWithLoad(loadTarget *txloadtool.LoadTarget, sendingTps,
 		t.logger.Errorf("Error during transaction load execution: %v", execErr)
 		t.ctx.SetResult(types.TaskResultFailure)
 
-		return 0, execErr
+		return 0, 0, execErr
 	}
 
 	// Collect the transactions and their latencies
@@ -194,12 +201,12 @@ func (t *Task) measureTpsWithLoad(loadTarget *txloadtool.LoadTarget, sendingTps,
 		t.logger.Errorf("Error measuring transaction propagation latencies: %v", measureErr)
 		t.ctx.SetResult(types.TaskResultFailure)
 
-		return 0, measureErr
+		return 0, 0, measureErr
 	}
 
 	// Check if the context was cancelled or other errors occurred
 	if result.Failed {
-		return 0, fmt.Errorf("error measuring transaction propagation latencies: load failed")
+		return 0, 0, fmt.Errorf("error measuring transaction propagation latencies: load failed")
 	}
 
 	// Send txes to other clients, for speeding up tx mining
@@ -226,19 +233,20 @@ func (t *Task) measureTpsWithLoad(loadTarget *txloadtool.LoadTarget, sendingTps,
 		// Calculate the percentile of latencies using result.LatenciesMus
 		// Not implemented yet
 		notImpl := errors.New("percentile selection not implemented, use 0.99")
-		return 0, notImpl
+		return 0, 0, notImpl
 	}
 
 	t.logger.Infof("Using 0.99 percentile for latency calculation")
 
 	t.logger.Infof("Last measure delay since start time: %s", result.LastMeasureDelay)
 
 	processedTpsF := float64(result.TotalTxs) / result.LastMeasureDelay.Seconds()
-	processedTps := int(processedTpsF) // round
+	processedTps = int(processedTpsF) // round
+	notReceivedP2PEventCount = result.NotReceivedP2PEventCount
 
 	t.logger.Infof("Processed %d transactions in %.2fs, mean throughput: %.2f tx/s",
 		result.TotalTxs, result.LastMeasureDelay.Seconds(), processedTpsF)
 	t.logger.Infof("Sent %d transactions in %.2fs", result.TotalTxs, result.LastMeasureDelay.Seconds())
 
-	return processedTps, nil
+	return processedTps, notReceivedP2PEventCount, nil
 }