3 * Copyright 2017 gRPC authors.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
31 // bufConn is a net.Conn implemented by a bytes.Buffer (which is a ReadWriter).
36 func (bufConn) Close() error { panic("unimplemented") }
37 func (bufConn) LocalAddr() net.Addr { panic("unimplemented") }
38 func (bufConn) RemoteAddr() net.Addr { panic("unimplemented") }
39 func (bufConn) SetDeadline(t time.Time) error { panic("unimplemneted") }
40 func (bufConn) SetReadDeadline(t time.Time) error { panic("unimplemneted") }
41 func (bufConn) SetWriteDeadline(t time.Time) error { panic("unimplemneted") }
43 func restoreHooks() func() {
52 func TestConn(t *testing.T) {
53 defer restoreHooks()()
56 now = func() time.Time { return time.Unix(123, 456) }
58 // Capture sleep times for checking later.
59 var sleepTimes []time.Duration
60 sleep = func(t time.Duration) { sleepTimes = append(sleepTimes, t) }
62 wantSleeps := func(want ...time.Duration) {
63 if !reflect.DeepEqual(want, sleepTimes) {
64 t.Fatalf("sleepTimes = %v; want %v", sleepTimes, want)
69 // Use a fairly high latency to cause a large BDP and avoid sleeps while
70 // writing due to simulation of full buffers.
71 latency := 1 * time.Second
72 c, err := (&Network{Kbps: 1, Latency: latency, MTU: 5}).Conn(bufConn{&bytes.Buffer{}})
74 t.Fatalf("Unexpected error creating connection: %v", err)
76 wantSleeps(latency) // Connection creation delay.
78 // 1 kbps = 128 Bps. Divides evenly by 1 second using nanos.
79 byteLatency := time.Duration(time.Second / 128)
81 write := func(b []byte) {
83 if n != len(b) || err != nil {
84 t.Fatalf("c.Write(%v) = %v, %v; want %v, nil", b, n, err, len(b))
88 write([]byte{1, 2, 3, 4, 5}) // One full packet
89 pkt1Time := latency + byteLatency*5
90 write([]byte{6}) // One partial packet
91 pkt2Time := pkt1Time + byteLatency
92 write([]byte{7, 8, 9, 10, 11, 12, 13}) // Two packets
93 pkt3Time := pkt2Time + byteLatency*5
94 pkt4Time := pkt3Time + byteLatency*2
96 // No reads, so no sleeps yet.
99 read := func(n int, want []byte) {
101 if rd, err := c.Read(b); err != nil || rd != len(want) {
102 t.Fatalf("c.Read(<%v bytes>) = %v, %v; want %v, nil", n, rd, err, len(want))
104 if !reflect.DeepEqual(b[:len(want)], want) {
105 t.Fatalf("read %v; want %v", b, want)
113 read(3, []byte{3, 4, 5})
117 read(2, []byte{7, 8})
119 read(10, []byte{9, 10, 11})
121 read(10, []byte{12, 13})
125 func TestSync(t *testing.T) {
126 defer restoreHooks()()
128 // Infinitely fast CPU: time doesn't pass unless sleep is called.
129 tn := time.Unix(123, 0)
130 now = func() time.Time { return tn }
131 sleep = func(d time.Duration) { tn = tn.Add(d) }
133 // Simulate a 20ms latency network, then run sync across that and expect to
134 // measure 20ms latency, or 10ms additional delay for a 30ms network.
135 slowConn, err := (&Network{Kbps: 0, Latency: 20 * time.Millisecond, MTU: 5}).Conn(bufConn{&bytes.Buffer{}})
137 t.Fatalf("Unexpected error creating connection: %v", err)
139 c, err := (&Network{Latency: 30 * time.Millisecond}).Conn(slowConn)
141 t.Fatalf("Unexpected error creating connection: %v", err)
143 if c.(*conn).delay != 10*time.Millisecond {
144 t.Fatalf("c.delay = %v; want 10ms", c.(*conn).delay)
148 func TestSyncTooSlow(t *testing.T) {
149 defer restoreHooks()()
151 // Infinitely fast CPU: time doesn't pass unless sleep is called.
152 tn := time.Unix(123, 0)
153 now = func() time.Time { return tn }
154 sleep = func(d time.Duration) { tn = tn.Add(d) }
156 // Simulate a 10ms latency network, then attempt to simulate a 5ms latency
157 // network and expect an error.
158 slowConn, err := (&Network{Kbps: 0, Latency: 10 * time.Millisecond, MTU: 5}).Conn(bufConn{&bytes.Buffer{}})
160 t.Fatalf("Unexpected error creating connection: %v", err)
163 errWant := "measured network latency (10ms) higher than desired latency (5ms)"
164 if _, err := (&Network{Latency: 5 * time.Millisecond}).Conn(slowConn); err == nil || err.Error() != errWant {
165 t.Fatalf("Conn() = _, %q; want _, %q", err, errWant)
169 func TestListenerAndDialer(t *testing.T) {
170 defer restoreHooks()()
172 tn := time.Unix(123, 0)
175 now = func() time.Time {
181 // Use a fairly high latency to cause a large BDP and avoid sleeps while
182 // writing due to simulation of full buffers.
183 n := &Network{Kbps: 2, Latency: 1 * time.Second, MTU: 10}
184 // 2 kbps = .25 kBps = 256 Bps
185 byteLatency := func(n int) time.Duration {
186 return time.Duration(n) * time.Second / 256
189 // Create a real listener and wrap it.
190 l, err := net.Listen("tcp", ":0")
192 t.Fatalf("Unexpected error creating listener: %v", err)
197 var serverConn net.Conn
199 scDone := make(chan struct{})
201 serverConn, scErr = l.Accept()
205 // Create a dialer and use it.
206 clientConn, err := n.TimeoutDialer(net.DialTimeout)("tcp", l.Addr().String(), 2*time.Second)
208 t.Fatalf("Unexpected error dialing: %v", err)
210 defer clientConn.Close()
212 // Block until server's Conn is available.
215 t.Fatalf("Unexpected error listening: %v", scErr)
217 defer serverConn.Close()
219 // sleep (only) advances tn. Done after connections established so sync detects zero delay.
220 sleep = func(d time.Duration) {
228 seq := func(a, b int) []byte {
229 buf := make([]byte, b-a)
230 for i := 0; i < b-a; i++ {
240 write := func(c net.Conn, b []byte) {
242 if n != len(b) || err != nil {
243 t.Fatalf("c.Write(%v) = %v, %v; want %v, nil", b, n, err, len(b))
247 write(serverConn, pkt1)
248 write(serverConn, pkt2)
249 write(serverConn, pkt3)
250 write(clientConn, pkt3)
251 write(clientConn, pkt1)
252 write(clientConn, pkt2)
255 t.Fatalf("unexpected sleep in write; tn = %v; want %v", tn, startTime)
258 read := func(c net.Conn, n int, want []byte, timeWant time.Time) {
260 if rd, err := c.Read(b); err != nil || rd != len(want) {
261 t.Fatalf("c.Read(<%v bytes>) = %v, %v; want %v, nil (read: %v)", n, rd, err, len(want), b[:rd])
263 if !reflect.DeepEqual(b[:len(want)], want) {
264 t.Fatalf("read %v; want %v", b, want)
266 if !tn.Equal(timeWant) {
267 t.Errorf("tn after read(%v) = %v; want %v", want, tn, timeWant)
271 read(clientConn, len(pkt1)+1, pkt1, startTime.Add(n.Latency+byteLatency(len(pkt1))))
272 read(serverConn, len(pkt3)+1, pkt3, tn) // tn was advanced by the above read; pkt3 is shorter than pkt1
274 read(clientConn, len(pkt2), pkt2[:10], startTime.Add(n.Latency+byteLatency(len(pkt1)+10)))
275 read(clientConn, len(pkt2), pkt2[10:], startTime.Add(n.Latency+byteLatency(len(pkt1)+len(pkt2))))
276 read(clientConn, len(pkt3), pkt3, startTime.Add(n.Latency+byteLatency(len(pkt1)+len(pkt2)+len(pkt3))))
278 read(serverConn, len(pkt1), pkt1, tn) // tn already past the arrival time due to prior reads
279 read(serverConn, len(pkt2), pkt2[:10], tn)
280 read(serverConn, len(pkt2), pkt2[10:], tn)
282 // Sleep awhile and make sure the read happens disregarding previous writes
283 // (lastSendEnd handling).
284 sleep(10 * time.Second)
285 write(clientConn, pkt1)
286 read(serverConn, len(pkt1), pkt1, tn.Add(n.Latency+byteLatency(len(pkt1))))
288 // Send, sleep longer than the network delay, then make sure the read happens
290 write(serverConn, pkt1)
291 sleep(10 * time.Second)
292 read(clientConn, len(pkt1), pkt1, tn)
295 func TestBufferBloat(t *testing.T) {
296 defer restoreHooks()()
298 // Infinitely fast CPU: time doesn't pass unless sleep is called.
299 tn := time.Unix(123, 0)
300 now = func() time.Time { return tn }
301 // Capture sleep times for checking later.
302 var sleepTimes []time.Duration
303 sleep = func(d time.Duration) {
304 sleepTimes = append(sleepTimes, d)
308 wantSleeps := func(want ...time.Duration) error {
309 if !reflect.DeepEqual(want, sleepTimes) {
310 return fmt.Errorf("sleepTimes = %v; want %v", sleepTimes, want)
316 n := &Network{Kbps: 8 /* 1KBps */, Latency: time.Second, MTU: 8}
317 bdpBytes := (n.Kbps * 1024 / 8) * int(n.Latency/time.Second) // 1024
318 c, err := n.Conn(bufConn{&bytes.Buffer{}})
320 t.Fatalf("Unexpected error creating connection: %v", err)
322 wantSleeps(n.Latency) // Connection creation delay.
324 write := func(n int, sleeps ...time.Duration) {
325 if wt, err := c.Write(make([]byte, n)); err != nil || wt != n {
326 t.Fatalf("c.Write(<%v bytes>) = %v, %v; want %v, nil", n, wt, err, n)
328 if err := wantSleeps(sleeps...); err != nil {
329 t.Fatalf("After writing %v bytes: %v", n, err)
333 read := func(n int, sleeps ...time.Duration) {
334 if rd, err := c.Read(make([]byte, n)); err != nil || rd != n {
335 t.Fatalf("c.Read(_) = %v, %v; want %v, nil", rd, err, n)
337 if err := wantSleeps(sleeps...); err != nil {
338 t.Fatalf("After reading %v bytes: %v", n, err)
342 write(8) // No reads and buffer not full, so no sleeps yet.
343 read(8, time.Second+n.pktTime(8))
345 write(bdpBytes) // Fill the buffer.
346 write(1) // We can send one extra packet even when the buffer is full.
347 write(n.MTU, n.pktTime(1)) // Make sure we sleep to clear the previous write.
348 write(1, n.pktTime(n.MTU))
349 write(n.MTU+1, n.pktTime(1), n.pktTime(n.MTU))
351 tn = tn.Add(10 * time.Second) // Wait long enough for the buffer to clear.
352 write(bdpBytes) // No sleeps required.