cloudflare/cloudflared
Publicmirrored from https://github.com/cloudflare/cloudflaredAvailable
h2mux/muxmetrics_test.go
180lines · modecode
| 1 | package h2mux |
| 2 | |
| 3 | import ( |
| 4 | "sync" |
| 5 | "testing" |
| 6 | "time" |
| 7 | |
| 8 | log "github.com/sirupsen/logrus" |
| 9 | "github.com/stretchr/testify/assert" |
| 10 | ) |
| 11 | |
| 12 | func ave(sum uint64, len int) float64 { |
| 13 | return float64(sum) / float64(len) |
| 14 | } |
| 15 | |
| 16 | func TestRTTUpdate(t *testing.T) { |
| 17 | r := newRTTData() |
| 18 | start := time.Now() |
| 19 | // send at 0 ms, receive at 2 ms, RTT = 2ms |
| 20 | m := &roundTripMeasurement{receiveTime: start.Add(2 * time.Millisecond), sendTime: start} |
| 21 | r.update(m) |
| 22 | assert.Equal(t, start, r.lastMeasurementTime) |
| 23 | assert.Equal(t, 2*time.Millisecond, r.rtt) |
| 24 | assert.Equal(t, 2*time.Millisecond, r.rttMin) |
| 25 | assert.Equal(t, 2*time.Millisecond, r.rttMax) |
| 26 | |
| 27 | // send at 3 ms, receive at 6 ms, RTT = 3ms |
| 28 | m = &roundTripMeasurement{receiveTime: start.Add(6 * time.Millisecond), sendTime: start.Add(3 * time.Millisecond)} |
| 29 | r.update(m) |
| 30 | assert.Equal(t, start.Add(3*time.Millisecond), r.lastMeasurementTime) |
| 31 | assert.Equal(t, 3*time.Millisecond, r.rtt) |
| 32 | assert.Equal(t, 2*time.Millisecond, r.rttMin) |
| 33 | assert.Equal(t, 3*time.Millisecond, r.rttMax) |
| 34 | |
| 35 | // send at 7 ms, receive at 8 ms, RTT = 1ms |
| 36 | m = &roundTripMeasurement{receiveTime: start.Add(8 * time.Millisecond), sendTime: start.Add(7 * time.Millisecond)} |
| 37 | r.update(m) |
| 38 | assert.Equal(t, start.Add(7*time.Millisecond), r.lastMeasurementTime) |
| 39 | assert.Equal(t, 1*time.Millisecond, r.rtt) |
| 40 | assert.Equal(t, 1*time.Millisecond, r.rttMin) |
| 41 | assert.Equal(t, 3*time.Millisecond, r.rttMax) |
| 42 | |
| 43 | // send at -4 ms, receive at 0 ms, RTT = 4ms, but this ping is before last measurement |
| 44 | // so it will be discarded |
| 45 | m = &roundTripMeasurement{receiveTime: start, sendTime: start.Add(-2 * time.Millisecond)} |
| 46 | r.update(m) |
| 47 | assert.Equal(t, start.Add(7*time.Millisecond), r.lastMeasurementTime) |
| 48 | assert.Equal(t, 1*time.Millisecond, r.rtt) |
| 49 | assert.Equal(t, 1*time.Millisecond, r.rttMin) |
| 50 | assert.Equal(t, 3*time.Millisecond, r.rttMax) |
| 51 | } |
| 52 | |
| 53 | func TestFlowControlDataUpdate(t *testing.T) { |
| 54 | f := newFlowControlData() |
| 55 | assert.Equal(t, 0, f.queue.Len()) |
| 56 | assert.Equal(t, float64(0), f.ave()) |
| 57 | |
| 58 | var sum uint64 |
| 59 | min := maxWindowSize - dataPoints |
| 60 | max := maxWindowSize |
| 61 | for i := 1; i <= dataPoints; i++ { |
| 62 | size := maxWindowSize - uint32(i) |
| 63 | f.update(size) |
| 64 | assert.Equal(t, max-uint32(1), f.max) |
| 65 | assert.Equal(t, size, f.min) |
| 66 | |
| 67 | assert.Equal(t, i, f.queue.Len()) |
| 68 | |
| 69 | sum += uint64(size) |
| 70 | assert.Equal(t, sum, f.sum) |
| 71 | assert.Equal(t, ave(sum, f.queue.Len()), f.ave()) |
| 72 | } |
| 73 | |
| 74 | // queue is full, should start to dequeue first element |
| 75 | for i := 1; i <= dataPoints; i++ { |
| 76 | f.update(max) |
| 77 | assert.Equal(t, max, f.max) |
| 78 | assert.Equal(t, min, f.min) |
| 79 | |
| 80 | assert.Equal(t, dataPoints, f.queue.Len()) |
| 81 | |
| 82 | sum += uint64(i) |
| 83 | assert.Equal(t, sum, f.sum) |
| 84 | assert.Equal(t, ave(sum, dataPoints), f.ave()) |
| 85 | } |
| 86 | } |
| 87 | |
| 88 | func TestMuxMetricsUpdater(t *testing.T) { |
| 89 | t.Skip("Race condition") |
| 90 | updateRTTChan := make(chan *roundTripMeasurement) |
| 91 | updateReceiveWindowChan := make(chan uint32) |
| 92 | updateSendWindowChan := make(chan uint32) |
| 93 | updateInBoundBytesChan := make(chan uint64) |
| 94 | updateOutBoundBytesChan := make(chan uint64) |
| 95 | abortChan := make(chan struct{}) |
| 96 | errChan := make(chan error) |
| 97 | compBefore, compAfter := NewAtomicCounter(0), NewAtomicCounter(0) |
| 98 | m := newMuxMetricsUpdater(updateRTTChan, |
| 99 | updateReceiveWindowChan, |
| 100 | updateSendWindowChan, |
| 101 | updateInBoundBytesChan, |
| 102 | updateOutBoundBytesChan, |
| 103 | abortChan, |
| 104 | compBefore, |
| 105 | compAfter, |
| 106 | ) |
| 107 | logger := log.NewEntry(log.New()) |
| 108 | |
| 109 | go func() { |
| 110 | errChan <- m.run(logger) |
| 111 | }() |
| 112 | |
| 113 | var wg sync.WaitGroup |
| 114 | wg.Add(2) |
| 115 | |
| 116 | // mock muxReader |
| 117 | readerStart := time.Now() |
| 118 | rm := &roundTripMeasurement{receiveTime: readerStart, sendTime: readerStart} |
| 119 | updateRTTChan <- rm |
| 120 | go func() { |
| 121 | defer wg.Done() |
| 122 | // Becareful if dataPoints is not divisibile by 4 |
| 123 | readerSend := readerStart.Add(time.Millisecond) |
| 124 | for i := 1; i <= dataPoints/4; i++ { |
| 125 | readerReceive := readerSend.Add(time.Duration(i) * time.Millisecond) |
| 126 | rm := &roundTripMeasurement{receiveTime: readerReceive, sendTime: readerSend} |
| 127 | updateRTTChan <- rm |
| 128 | readerSend = readerReceive.Add(time.Millisecond) |
| 129 | |
| 130 | updateReceiveWindowChan <- uint32(i) |
| 131 | updateSendWindowChan <- uint32(i) |
| 132 | |
| 133 | updateInBoundBytesChan <- uint64(i) |
| 134 | } |
| 135 | }() |
| 136 | |
| 137 | // mock muxWriter |
| 138 | go func() { |
| 139 | defer wg.Done() |
| 140 | for j := dataPoints/4 + 1; j <= dataPoints/2; j++ { |
| 141 | updateReceiveWindowChan <- uint32(j) |
| 142 | updateSendWindowChan <- uint32(j) |
| 143 | |
| 144 | // should always be disgard since the send time is before readerSend |
| 145 | rm := &roundTripMeasurement{receiveTime: readerStart, sendTime: readerStart.Add(-time.Duration(j*dataPoints) * time.Millisecond)} |
| 146 | updateRTTChan <- rm |
| 147 | |
| 148 | updateOutBoundBytesChan <- uint64(j) |
| 149 | } |
| 150 | |
| 151 | }() |
| 152 | wg.Wait() |
| 153 | |
| 154 | metrics := m.Metrics() |
| 155 | points := dataPoints / 2 |
| 156 | assert.Equal(t, time.Millisecond, metrics.RTTMin) |
| 157 | assert.Equal(t, time.Duration(dataPoints/4)*time.Millisecond, metrics.RTTMax) |
| 158 | |
| 159 | // sum(1..i) = i*(i+1)/2, ave(1..i) = i*(i+1)/2/i = (i+1)/2 |
| 160 | assert.Equal(t, float64(points+1)/float64(2), metrics.ReceiveWindowAve) |
| 161 | assert.Equal(t, uint32(1), metrics.ReceiveWindowMin) |
| 162 | assert.Equal(t, uint32(points), metrics.ReceiveWindowMax) |
| 163 | |
| 164 | assert.Equal(t, float64(points+1)/float64(2), metrics.SendWindowAve) |
| 165 | assert.Equal(t, uint32(1), metrics.SendWindowMin) |
| 166 | assert.Equal(t, uint32(points), metrics.SendWindowMax) |
| 167 | |
| 168 | assert.Equal(t, uint64(dataPoints/4), metrics.InBoundRateCurr) |
| 169 | assert.Equal(t, uint64(1), metrics.InBoundRateMin) |
| 170 | assert.Equal(t, uint64(dataPoints/4), metrics.InBoundRateMax) |
| 171 | |
| 172 | assert.Equal(t, uint64(dataPoints/2), metrics.OutBoundRateCurr) |
| 173 | assert.Equal(t, uint64(dataPoints/4+1), metrics.OutBoundRateMin) |
| 174 | assert.Equal(t, uint64(dataPoints/2), metrics.OutBoundRateMax) |
| 175 | |
| 176 | close(abortChan) |
| 177 | assert.Nil(t, <-errChan) |
| 178 | close(errChan) |
| 179 | |
| 180 | } |
| 181 | |