diff --git a/python-ecosys/iperf3/iperf3.py b/python-ecosys/iperf3/iperf3.py index 05d69f774..14be439b5 100644 --- a/python-ecosys/iperf3/iperf3.py +++ b/python-ecosys/iperf3/iperf3.py @@ -71,7 +71,8 @@ def fmt_size(val, div): class Stats: def __init__(self, param): - self.pacing_timer_us = param["pacing_timer"] * 1000 + # iperf3 clients before 3.2 do not send the pacing timer. + self.pacing_timer_us = param.get("pacing_timer", 1000) * 1000 self.udp = param.get("udp", False) self.reverse = param.get("reverse", False) self.running = False @@ -217,8 +218,10 @@ def _transfer(udp, reverse, addr, s_data, buf, udp_last_send, udp_packet_id, udp stats.add_bytes(n) else: if reverse: - recvninto(s_data, buf) - n = len(buf) + # The socket is non-blocking, so this reads only what is available. The + # sender is not required to end the stream on a block boundary, so waiting + # here for a full block could block forever and miss TEST_END. + n = recvinto(s_data, buf) or 0 else: n = s_data.send(buf) stats.add_bytes(n) @@ -233,7 +236,9 @@ def server_once(): s_listen = socket.socket(ai[0], socket.SOCK_STREAM) s_listen.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) s_listen.bind(ai[-1]) - s_listen.listen(1) + # The client opens its parallel streams all at once, so leave room to queue them. + # lwIP has 5 TCP pcbs by default, enough for the control connection and 4 streams. + s_listen.listen(4) s_ctrl, addr = s_listen.accept() # Read client's cookie @@ -251,15 +256,22 @@ def server_once(): if DEBUG: print(param) reverse = param.get("reverse", False) + parallel = param.get("parallel", 1) # Ask to create streams s_ctrl.sendall(bytes([CREATE_STREAMS])) if param.get("tcp", False): - # Accept stream - s_data, addr = s_listen.accept() - print("Accepted connection:", addr) - recvn(s_data, COOKIE_SIZE) + # Accept streams, the client opens one connection per parallel stream + s_data = [] + for _ in range(parallel): + s, addr = s_listen.accept() + print("Accepted connection:", addr) + recvn(s, COOKIE_SIZE) + if not reverse: + # Receive without blocking, see _transfer(). + s.setblocking(False) + s_data.append(s) udp = False udp_packet_id = 0 udp_interval = None @@ -267,10 +279,11 @@ def server_once(): elif param.get("udp", False): # Close TCP connection and open UDP "connection" s_listen.close() - s_data = socket.socket(ai[0], socket.SOCK_DGRAM) - s_data.bind(ai[-1]) - data, addr = s_data.recvfrom(4) - s_data.sendto(struct.pack("