11const { hrtime } = require ( 'process' ) ;
22const { randomBytes } = require ( 'crypto' ) ;
3+ const PERCENTILES = [ 50 , 75 , 90 , 95 , 99 , 99.9 , 99.99 , 100 ] ;
34
45const TERMINATE_TIMEOUT_MS = process . env . TERMINATE_TIMEOUT_MS ? + process . env . TERMINATE_TIMEOUT_MS : 600000 ;
56const AUTO_COMMIT = process . env . AUTO_COMMIT || 'false' ;
@@ -58,8 +59,86 @@ function genericProduceToTopic(producer, topic, messages) {
5859 } ) ;
5960}
6061
62+
63+ // We use a simple count-sketch for latency percentiles to avoid storing all latencies in memory.
64+ // because we're also measuring the memory usage of the consumer as part of the performance tests.
65+ class LatencyCountSketch {
66+ #numBuckets;
67+ #minValue;
68+ #maxValue;
69+ #buckets;
70+ #counts;
71+ #changeBaseLogarithm;
72+ #totalCount = 0 ;
73+ #base;
74+
75+ constructor ( {
76+ error = 0.01 , // 1% error
77+ minValue = 0.01 , // min 10μs latency
78+ maxValue = 60000 , // max 60s latency
79+ } ) {
80+ // Each bucket represents [x, x * (1 + error))
81+ this . #base = 1 + error ;
82+ // Change base from natural log to log base this.#base
83+ this . #changeBaseLogarithm = Math . log ( this . #base) ;
84+ this . #numBuckets = Math . ceil ( Math . log ( maxValue / minValue ) / Math . log ( this . #base) ) ;
85+ this . #maxValue = maxValue ;
86+
87+ this . #buckets = new Array ( this . #numBuckets + 2 ) . fill ( 0 ) ;
88+ this . #buckets[ this . #numBuckets + 1 ] = Number . POSITIVE_INFINITY ;
89+ this . #buckets[ this . #numBuckets] = this . #maxValue;
90+ this . #buckets[ 0 ] = 0 ;
91+ let i = this . #numBuckets - 1 ;
92+ let currentValue = maxValue ;
93+ while ( i >= 1 ) {
94+ let nextMinimum = currentValue / this . #base;
95+ this . #buckets[ i ] = nextMinimum ;
96+ currentValue = nextMinimum ;
97+ i -- ;
98+ }
99+ this . #minValue = this . #buckets[ 1 ] ;
100+ this . #counts = new Array ( this . #numBuckets + 2 ) . fill ( 0 ) ;
101+ }
102+
103+ add ( latency ) {
104+ let idx = 0 ;
105+ if ( latency > 0 )
106+ idx = Math . ceil ( Math . log ( latency / this . #minValue) / this . #changeBaseLogarithm) ;
107+ idx = ( idx < 0 ) ? 0 :
108+ ( idx > this . #buckets. length - 2 ) ? ( this . #buckets. length - 2 ) :
109+ idx ;
110+
111+ this . #counts[ idx ] ++ ;
112+ this . #totalCount++ ;
113+ }
114+
115+ percentiles ( percentilesArray ) {
116+ const percentileCounts = percentilesArray . map ( p => Math . ceil ( this . #totalCount * p / 100 ) ) ;
117+ const percentileResults = new Array ( percentilesArray . length ) ;
118+ var totalCountSoFar = 0 ;
119+ let j = 0 ;
120+ let sum = 0 ;
121+ for ( let i = 0 ; i < this . #counts. length ; i ++ ) {
122+ sum += this . #counts[ i ] ;
123+ }
124+ for ( let i = 0 ; i < percentileCounts . length ; i ++ ) {
125+ while ( ( totalCountSoFar < percentileCounts [ i ] ) && ( j < this . #counts. length - 1 ) ) {
126+ totalCountSoFar += this . #counts[ j ] ;
127+ j ++ ;
128+ }
129+ const bucketIndex = ( j < this . #counts. length - 1 ) ? j : this . #counts. length - 2 ;
130+ percentileResults [ i ] = [ this . #buckets[ bucketIndex ] , totalCountSoFar , this . #totalCount] ;
131+ }
132+ return percentileResults ;
133+ }
134+ }
135+
61136async function runConsumer ( consumer , topic , warmupMessages , totalMessageCnt , eachBatch , partitionsConsumedConcurrently , stats , actionOnMessages ) {
62137 const handlers = installHandlers ( totalMessageCnt === - 1 ) ;
138+ if ( stats ) {
139+ stats . percentilesTOT1 = new LatencyCountSketch ( { } ) ;
140+ stats . percentilesTOT2 = new LatencyCountSketch ( { } ) ;
141+ }
63142 while ( true ) {
64143 try {
65144 await consumer . connect ( ) ;
@@ -96,7 +175,17 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
96175 return ;
97176
98177 const sentAt = Number ( decoder . decode ( message . value . slice ( 0 , 13 ) ) ) ;
99- const latency = receivedAt - sentAt ;
178+ let latency = receivedAt - sentAt ;
179+
180+ if ( isNaN ( latency ) ) {
181+ console . log ( `WARN: NaN latency received message timestamp: ${ message . value . slice ( 0 , 13 ) } ) ;
182+ return ;
183+ } else if ( latency < 0 ) {
184+ console . log ( `WARN: negative latency ${ latency } ${ sentAt } ${ receivedAt } ) ;
185+ latency = 0 ;
186+ } else if ( latency > 60000 ) {
187+ console . log ( `WARN: received large latency ${ latency } ${ sentAt } ${ receivedAt } ) ;
188+ }
100189
101190 if ( ! isT0T2 ) {
102191 if ( ! stats . maxLatencyT0T1 ) {
@@ -106,6 +195,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
106195 stats . maxLatencyT0T1 = Math . max ( stats . maxLatencyT0T1 , latency ) ;
107196 stats . avgLatencyT0T1 = ( ( stats . avgLatencyT0T1 * ( numMessages - 1 ) ) + latency ) / numMessages ;
108197 }
198+ stats . percentilesTOT1 . add ( latency ) ;
109199 } else {
110200 if ( ! stats . maxLatencyT0T2 ) {
111201 stats . maxLatencyT0T2 = latency ;
@@ -114,6 +204,7 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
114204 stats . maxLatencyT0T2 = Math . max ( stats . maxLatencyT0T2 , latency ) ;
115205 stats . avgLatencyT0T2 = ( ( stats . avgLatencyT0T2 * ( numMessages - 1 ) ) + latency ) / numMessages ;
116206 }
207+ stats . percentilesTOT2 . add ( latency ) ;
117208 }
118209 } ;
119210
@@ -257,6 +348,18 @@ async function runConsumer(consumer, topic, warmupMessages, totalMessageCnt, eac
257348 stats . messageRate = durationSeconds > 0 ?
258349 ( messagesMeasured / durationSeconds ) : Infinity ;
259350 stats . durationSeconds = durationSeconds ;
351+ stats . percentilesTOT1 = stats . percentilesTOT1 . percentiles ( PERCENTILES ) . map ( ( value , index ) => ( {
352+ percentile : PERCENTILES [ index ] ,
353+ value : value [ 0 ] ,
354+ count : value [ 1 ] ,
355+ total : value [ 2 ] ,
356+ } ) ) ;
357+ stats . percentilesTOT2 = stats . percentilesTOT2 . percentiles ( PERCENTILES ) . map ( ( value , index ) => ( {
358+ percentile : PERCENTILES [ index ] ,
359+ value : value [ 0 ] ,
360+ count : value [ 1 ] ,
361+ total : value [ 2 ] ,
362+ } ) ) ;
260363 }
261364 removeHandlers ( handlers ) ;
262365 return rate ;
0 commit comments