| #!/usr/bin/env python3 |
| # |
| # Copyright 2020 - The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import logging |
| import multiprocessing |
| import time |
| |
| from datetime import datetime |
| from uuid import uuid4 |
| |
| from acts import signals |
| from acts import tracelogger |
| from acts import utils |
| from acts.controllers import iperf_client |
| from acts.controllers import iperf_server |
| |
| AC_VO = 'AC_VO' |
| AC_VI = 'AC_VI' |
| AC_BE = 'AC_BE' |
| AC_BK = 'AC_BK' |
| |
| # TODO(fxb/61421): Add tests to check all DSCP classes are mapped to the correct |
| # AC (there are many that aren't included here). Requires implementation of |
| # sniffer. |
| DEFAULT_AC_TO_TOS_TAG_MAP = { |
| AC_VO: '0xC0', |
| AC_VI: '0x80', |
| AC_BE: '0x0', |
| AC_BK: '0x20' |
| } |
| UDP = 'udp' |
| TCP = 'tcp' |
| DEFAULT_IPERF_PORT = 5201 |
| DEFAULT_STREAM_TIME = 10 |
| DEFAULT_IP_ADDR_TIMEOUT = 15 |
| PROCESS_JOIN_TIMEOUT = 60 |
| AVAILABLE = True |
| UNAVAILABLE = False |
| |
| |
| class WmmTransceiverError(signals.ControllerError): |
| pass |
| |
| |
| def create(config, identifier=None, wlan_devices=None, access_points=None): |
| """Creates a WmmTransceiver from a config. |
| |
| Args: |
| config: dict, config parameters for the transceiver. Contains: |
| - iperf_config: dict, the config to use for creating IPerfClients |
| and IPerfServers (excluding port). |
| - port_range_start: int, the lower bound of the port range to use |
| for creating IPerfServers. Defaults to 5201. |
| - wlan_device: string, the identifier of the wlan_device used for |
| this WmmTransceiver (optional) |
| |
| identifier: string, identifier for the WmmTransceiver. Must be provided |
| either as arg or in the config. |
| wlan_devices: list of WlanDevice objects from which to get the |
| wlan_device, if any, used as this transceiver |
| access_points: list of AccessPoint objects from which to get the |
| access_point, if any, used as this transceiver |
| """ |
| try: |
| # If identifier is not provided as func arg, it must be provided via |
| # config file. |
| if not identifier: |
| identifier = config['identifier'] |
| iperf_config = config['iperf_config'] |
| |
| except KeyError as err: |
| raise WmmTransceiverError( |
| 'Parameter not provided as func arg, nor found in config: %s' % |
| err) |
| |
| if wlan_devices is None: |
| wlan_devices = [] |
| |
| if access_points is None: |
| access_points = [] |
| |
| port_range_start = config.get('port_range_start', DEFAULT_IPERF_PORT) |
| |
| wd = None |
| ap = None |
| if 'wlan_device' in config: |
| wd = _find_wlan_device(config['wlan_device'], wlan_devices) |
| elif 'access_point' in config: |
| ap = _find_access_point(config['access_point'], access_points) |
| |
| return WmmTransceiver(iperf_config, |
| identifier, |
| wlan_device=wd, |
| access_point=ap, |
| port_range_start=port_range_start) |
| |
| |
| def _find_wlan_device(wlan_device_identifier, wlan_devices): |
| """Returns WlanDevice based on string identifier (e.g. ip, serial, etc.) |
| |
| Args: |
| wlan_device_identifier: string, identifier for the desired WlanDevice |
| wlan_devices: list, WlanDevices to search through |
| |
| Returns: |
| WlanDevice, with identifier matching wlan_device_identifier |
| |
| Raises: |
| WmmTransceiverError, if no WlanDevice matches identifier |
| """ |
| for wd in wlan_devices: |
| if wlan_device_identifier == wd.identifier: |
| return wd |
| raise WmmTransceiverError('No WlanDevice with identifier: %s' % |
| wlan_device_identifier) |
| |
| |
| def _find_access_point(access_point_ip, access_points): |
| """Returns AccessPoint based on string ip address |
| |
| Args: |
| access_point_ip: string, control plane ip addr of the desired AP, |
| access_points: list, AccessPoints to search through |
| |
| Returns: |
| AccessPoint, with hostname matching access_point_ip |
| |
| Raises: |
| WmmTransceiverError, if no AccessPoint matches ip""" |
| for ap in access_points: |
| if ap.ssh_settings.hostname == access_point_ip: |
| return ap |
| raise WmmTransceiverError('No AccessPoint with ip: %s' % access_point_ip) |
| |
| |
| class WmmTransceiver(object): |
| """Object for handling WMM tagged streams between devices""" |
| def __init__(self, |
| iperf_config, |
| identifier, |
| wlan_device=None, |
| access_point=None, |
| port_range_start=5201): |
| |
| self.identifier = identifier |
| self.log = tracelogger.TraceLogger( |
| WmmTransceiverLoggerAdapter(logging.getLogger(), |
| {'identifier': self.identifier})) |
| # WlanDevice or AccessPoint, that is used as the transceiver. Only one |
| # will be set. This helps consolodate association, setup, teardown, etc. |
| self.wlan_device = wlan_device |
| self.access_point = access_point |
| |
| # Parameters used to create IPerfClient and IPerfServer objects on |
| # device |
| self._iperf_config = iperf_config |
| self._test_interface = self._iperf_config.get('test_interface') |
| self._port_range_start = port_range_start |
| self._next_server_port = port_range_start |
| |
| # Maps IPerfClients, used for streams from this device, to True if |
| # available, False if reserved |
| self._iperf_clients = {} |
| |
| # Maps IPerfServers, used to receive streams from other devices, to True |
| # if available, False if reserved |
| self._iperf_servers = {} |
| |
| # Maps ports of servers, which are provided to other transceivers, to |
| # the actual IPerfServer objects |
| self._iperf_server_ports = {} |
| |
| # Maps stream UUIDs to IPerfClients reserved for that streams use |
| self._reserved_clients = {} |
| |
| # Maps stream UUIDs to (WmmTransceiver, IPerfServer) tuples, where the |
| # server is reserved on the transceiver for that streams use |
| self._reserved_servers = {} |
| |
| # Maps with shared memory functionality to be used across the parallel |
| # streams. active_streams holds UUIDs of streams that are currently |
| # running on this device (mapped to True, since there is no |
| # multiprocessing set). stream_results maps UUIDs of streams completed |
| # on this device to IPerfResult results for that stream. |
| self._manager = multiprocessing.Manager() |
| self._active_streams = self._manager.dict() |
| self._stream_results = self._manager.dict() |
| |
| # Holds parameters for streams that are prepared to run asynchronously |
| # (i.e. resources have been allocated). Maps UUIDs of the future streams |
| # to a dict, containing the stream parameters. |
| self._pending_async_streams = {} |
| |
| # Set of UUIDs of asynchronous streams that have at least started, but |
| # have not had their resources reclaimed yet |
| self._ran_async_streams = set() |
| |
| # Set of stream parallel process, which can be joined if completed |
| # successfully, or terminated and joined in the event of an error |
| self._running_processes = set() |
| |
| def run_synchronous_traffic_stream(self, stream_parameters, subnet): |
| """Runs a traffic stream with IPerf3 between two WmmTransceivers and |
| saves the results. |
| |
| Args: |
| stream_parameters: dict, containing parameters to used for the |
| stream. See _parse_stream_parameters for details. |
| subnet: string, the subnet of the network to use for the stream |
| |
| Returns: |
| uuid: UUID object, identifier of the stream |
| """ |
| (receiver, access_category, bandwidth, |
| stream_time) = self._parse_stream_parameters(stream_parameters) |
| uuid = uuid4() |
| |
| (client, server_ip, |
| server_port) = self._get_stream_resources(uuid, receiver, subnet) |
| |
| self._validate_server_address(server_ip, uuid) |
| |
| self.log.info('Running synchronous stream to %s WmmTransceiver' % |
| receiver.identifier) |
| self._run_traffic(uuid, |
| client, |
| server_ip, |
| server_port, |
| self._active_streams, |
| self._stream_results, |
| access_category=access_category, |
| bandwidth=bandwidth, |
| stream_time=stream_time) |
| |
| self._return_stream_resources(uuid) |
| return uuid |
| |
| def prepare_asynchronous_stream(self, stream_parameters, subnet): |
| """Reserves resources and saves configs for upcoming asynchronous |
| traffic streams, so they can be started more simultaneously. |
| |
| Args: |
| stream_parameters: dict, containing parameters to used for the |
| stream. See _parse_stream_parameters for details. |
| subnet: string, the subnet of the network to use for the stream |
| |
| Returns: |
| uuid: UUID object, identifier of the stream |
| """ |
| (receiver, access_category, bandwidth, |
| time) = self._parse_stream_parameters(stream_parameters) |
| uuid = uuid4() |
| |
| (client, server_ip, |
| server_port) = self._get_stream_resources(uuid, receiver, subnet) |
| |
| self._validate_server_address(server_ip, uuid) |
| |
| pending_stream_config = { |
| 'client': client, |
| 'server_ip': server_ip, |
| 'server_port': server_port, |
| 'access_category': access_category, |
| 'bandwidth': bandwidth, |
| 'time': time |
| } |
| |
| self._pending_async_streams[uuid] = pending_stream_config |
| self.log.info('Stream to %s WmmTransceiver prepared.' % |
| receiver.identifier) |
| return uuid |
| |
| def start_asynchronous_streams(self, start_time=None): |
| """Starts pending asynchronous streams between two WmmTransceivers as |
| parallel processes. |
| |
| Args: |
| start_time: float, time, seconds since epoch, at which to start the |
| stream (for better synchronicity). If None, start immediately. |
| """ |
| for uuid in self._pending_async_streams: |
| pending_stream_config = self._pending_async_streams[uuid] |
| client = pending_stream_config['client'] |
| server_ip = pending_stream_config['server_ip'] |
| server_port = pending_stream_config['server_port'] |
| access_category = pending_stream_config['access_category'] |
| bandwidth = pending_stream_config['bandwidth'] |
| time = pending_stream_config['time'] |
| |
| process = multiprocessing.Process(target=self._run_traffic, |
| args=[ |
| uuid, client, server_ip, |
| server_port, |
| self._active_streams, |
| self._stream_results |
| ], |
| kwargs={ |
| 'access_category': |
| access_category, |
| 'bandwidth': bandwidth, |
| 'stream_time': time, |
| 'start_time': start_time |
| }) |
| |
| # This needs to be set here to ensure its marked active before |
| # it even starts. |
| self._active_streams[uuid] = True |
| process.start() |
| self._ran_async_streams.add(uuid) |
| self._running_processes.add(process) |
| |
| self._pending_async_streams.clear() |
| |
| def cleanup_asynchronous_streams(self, timeout=PROCESS_JOIN_TIMEOUT): |
| """Releases reservations on resources (IPerfClients and IPerfServers) |
| that were held for asynchronous streams, both pending and finished. |
| Attempts to join any running processes, logging an error if timeout is |
| exceeded. |
| |
| Args: |
| timeout: time, in seconds, to wait for each running process, if any, |
| to join |
| """ |
| self.log.info('Cleaning up any asynchronous streams.') |
| |
| # Releases resources for any streams that were prepared, but no run |
| for uuid in self._pending_async_streams: |
| self.log.error( |
| 'Pending asynchronous stream %s never ran. Cleaning.' % uuid) |
| self._return_stream_resources(uuid) |
| self._pending_async_streams.clear() |
| |
| # Attempts to join any running streams, terminating them after timeout |
| # if necessary. |
| while self._running_processes: |
| process = self._running_processes.pop() |
| process.join(timeout) |
| if process.is_alive(): |
| self.log.error( |
| 'Stream process failed to join in %s seconds. Terminating.' |
| % timeout) |
| process.terminate() |
| process.join() |
| self._active_streams.clear() |
| |
| # Release resources for any finished streams |
| while self._ran_async_streams: |
| uuid = self._ran_async_streams.pop() |
| self._return_stream_resources(uuid) |
| |
| def get_results(self, uuid): |
| """Retrieves a streams IPerfResults from stream_results |
| |
| Args: |
| uuid: UUID object, identifier of the stream |
| """ |
| return self._stream_results.get(uuid, None) |
| |
| def destroy_resources(self): |
| for server in self._iperf_servers: |
| server.stop() |
| self._iperf_servers.clear() |
| self._iperf_server_ports.clear() |
| self._iperf_clients.clear() |
| self._next_server_port = self._port_range_start |
| self._stream_results.clear() |
| |
| @property |
| def has_active_streams(self): |
| return bool(self._active_streams) |
| |
| # Helper Functions |
| |
| def _run_traffic(self, |
| uuid, |
| client, |
| server_ip, |
| server_port, |
| active_streams, |
| stream_results, |
| access_category=None, |
| bandwidth=None, |
| stream_time=DEFAULT_STREAM_TIME, |
| start_time=None): |
| """Runs an iperf3 stream. |
| |
| 1. Adds stream UUID to active_streams |
| 2. Runs stream |
| 3. Saves results to stream_results |
| 4. Removes stream UUID from active_streams |
| |
| Args: |
| uuid: UUID object, identifier for stream |
| client: IPerfClient object on device |
| server_ip: string, ip address of IPerfServer for stream |
| server_port: int, port of the IPerfServer for stream |
| active_streams: multiprocessing.Manager.dict, which holds stream |
| UUIDs of active streams on the device |
| stream_results: multiprocessing.Manager.dict, which maps stream |
| UUIDs of streams to IPerfResult objects |
| access_category: string, WMM access category to use with iperf |
| (AC_BK, AC_BE, AC_VI, AC_VO). Unset if None. |
| bandwidth: int, bandwidth in mbps to use with iperf. Implies UDP. |
| Unlimited if None. |
| stream_time: int, time in seconds, to run iperf stream |
| start_time: float, time, seconds since epoch, at which to start the |
| stream (for better synchronicity). If None, start immediately. |
| """ |
| active_streams[uuid] = True |
| # SSH sessions must be started within the process that is going to |
| # use it. |
| if type(client) == iperf_client.IPerfClientOverSsh: |
| with utils.SuppressLogOutput(): |
| client.start_ssh() |
| |
| ac_flag = '' |
| bandwidth_flag = '' |
| time_flag = '-t %s' % stream_time |
| |
| if access_category: |
| ac_flag = ' -S %s' % DEFAULT_AC_TO_TOS_TAG_MAP[access_category] |
| |
| if bandwidth: |
| bandwidth_flag = ' -u -b %sM' % bandwidth |
| |
| iperf_flags = '-p %s -i 1 %s%s%s -J' % (server_port, time_flag, |
| ac_flag, bandwidth_flag) |
| if not start_time: |
| start_time = time.time() |
| time_str = datetime.fromtimestamp(start_time).strftime('%H:%M:%S.%f') |
| self.log.info( |
| 'At %s, starting %s second stream to %s:%s with (AC: %s, Bandwidth: %s)' |
| % (time_str, stream_time, server_ip, server_port, access_category, |
| bandwidth if bandwidth else 'Unlimited')) |
| |
| # If present, wait for stream start time |
| if start_time: |
| current_time = time.time() |
| while current_time < start_time: |
| current_time = time.time() |
| path = client.start(server_ip, iperf_flags, '%s' % uuid) |
| stream_results[uuid] = iperf_server.IPerfResult( |
| path, reporting_speed_units='mbps') |
| |
| if type(client) == iperf_client.IPerfClientOverSsh: |
| client.close_ssh() |
| active_streams.pop(uuid) |
| |
| def _get_stream_resources(self, uuid, receiver, subnet): |
| """Reserves an IPerfClient and IPerfServer for a stream. |
| |
| Args: |
| uuid: UUID object, identifier of the stream |
| receiver: WmmTransceiver object, which will be the streams receiver |
| subnet: string, subnet of test network, to retrieve the appropriate |
| server address |
| |
| Returns: |
| (IPerfClient, string, int) representing the client, server address, |
| and server port to use for the stream |
| """ |
| client = self._get_client(uuid) |
| server_ip, server_port = self._get_server(receiver, uuid, subnet) |
| return (client, server_ip, server_port) |
| |
| def _return_stream_resources(self, uuid): |
| """Releases reservations on a streams IPerfClient and IPerfServer, so |
| they can be used by a future stream. |
| |
| Args: |
| uuid: UUID object, identifier of the stream |
| """ |
| if uuid in self._active_streams: |
| raise EnvironmentError('Resource still being used by stream %s' % |
| uuid) |
| (receiver, server_port) = self._reserved_servers.pop(uuid) |
| receiver._release_server(server_port) |
| client = self._reserved_clients.pop(uuid) |
| self._iperf_clients[client] = AVAILABLE |
| |
| def _get_client(self, uuid): |
| """Retrieves and reserves IPerfClient for use in a stream. If none are |
| available, a new one is created. |
| |
| Args: |
| uuid: UUID object, identifier for stream, used to link client to |
| stream for teardown |
| |
| Returns: |
| IPerfClient on device |
| """ |
| reserved_client = None |
| for client in self._iperf_clients: |
| if self._iperf_clients[client] == AVAILABLE: |
| reserved_client = client |
| break |
| else: |
| reserved_client = iperf_client.create([self._iperf_config])[0] |
| # Due to the nature of multiprocessing, ssh connections must |
| # be started inside the parallel processes, so it must be closed |
| # here. |
| if type(reserved_client) == iperf_client.IPerfClientOverSsh: |
| reserved_client.close_ssh() |
| |
| self._iperf_clients[reserved_client] = UNAVAILABLE |
| self._reserved_clients[uuid] = reserved_client |
| return reserved_client |
| |
| def _get_server(self, receiver, uuid, subnet): |
| """Retrieves the address and port of a reserved IPerfServer object from |
| the receiver object for use in a stream. |
| |
| Args: |
| receiver: WmmTransceiver, to get an IPerfServer from |
| uuid: UUID, identifier for stream, used to link server to stream |
| for teardown |
| subnet: string, subnet of test network, to retrieve the appropriate |
| server address |
| |
| Returns: |
| (string, int) representing the IPerfServer address and port |
| """ |
| (server_ip, server_port) = receiver._reserve_server(subnet) |
| self._reserved_servers[uuid] = (receiver, server_port) |
| return (server_ip, server_port) |
| |
| def _reserve_server(self, subnet): |
| """Reserves an available IPerfServer for use in a stream from another |
| WmmTransceiver. If none are available, a new one is created. |
| |
| Args: |
| subnet: string, subnet of test network, to retrieve the appropriate |
| server address |
| |
| Returns: |
| (string, int) representing the IPerfServer address and port |
| """ |
| reserved_server = None |
| for server in self._iperf_servers: |
| if self._iperf_servers[server] == AVAILABLE: |
| reserved_server = server |
| break |
| else: |
| iperf_server_config = self._iperf_config |
| iperf_server_config.update({'port': self._next_server_port}) |
| self._next_server_port += 1 |
| reserved_server = iperf_server.create([iperf_server_config])[0] |
| self._iperf_server_ports[reserved_server.port] = reserved_server |
| |
| self._iperf_servers[reserved_server] = UNAVAILABLE |
| reserved_server.start() |
| end_time = time.time() + DEFAULT_IP_ADDR_TIMEOUT |
| while time.time() < end_time: |
| if self.wlan_device: |
| addresses = utils.get_interface_ip_addresses( |
| self.wlan_device.device, self._test_interface) |
| else: |
| addresses = reserved_server.get_interface_ip_addresses( |
| self._test_interface) |
| for addr in addresses['ipv4_private']: |
| if utils.ip_in_subnet(addr, subnet): |
| return (addr, reserved_server.port) |
| raise AttributeError( |
| 'Reserved server has no ipv4 address in the %s subnet' % subnet) |
| |
| def _release_server(self, server_port): |
| """Releases reservation on IPerfServer, which was held for a stream |
| from another WmmTransceiver. |
| |
| Args: |
| server_port: int, the port of the IPerfServer being returned (since) |
| it is the identifying characteristic |
| """ |
| server = self._iperf_server_ports[server_port] |
| server.stop() |
| self._iperf_servers[server] = AVAILABLE |
| |
| def _validate_server_address(self, server_ip, uuid, timeout=60): |
| """ Verifies server address can be pinged before attempting to run |
| traffic, since iperf is unforgiving when the server is unreachable. |
| |
| Args: |
| server_ip: string, ip address of the iperf server |
| uuid: string, uuid of the stream to use this server |
| timeout: int, time in seconds to wait for server to respond to pings |
| |
| Raises: |
| WmmTransceiverError, if, after timeout, server ip is unreachable. |
| """ |
| self.log.info('Verifying server address (%s) is reachable.' % |
| server_ip) |
| end_time = time.time() + timeout |
| while time.time() < end_time: |
| if self.can_ping(server_ip): |
| break |
| else: |
| self.log.debug( |
| 'Could not ping server address (%s). Retrying in 1 second.' |
| % (server_ip)) |
| time.sleep(1) |
| else: |
| self._return_stream_resources(uuid) |
| raise WmmTransceiverError('IPerfServer address (%s) unreachable.' % |
| server_ip) |
| |
| def can_ping(self, dest_ip): |
| """ Utilizes can_ping function in wlan_device or access_point device to |
| ping dest_ip |
| |
| Args: |
| dest_ip: string, ip address to ping |
| |
| Returns: |
| True, if dest address is reachable |
| False, otherwise |
| """ |
| if self.wlan_device: |
| return self.wlan_device.can_ping(dest_ip) |
| else: |
| return self.access_point.can_ping(dest_ip) |
| |
| def _parse_stream_parameters(self, stream_parameters): |
| """Parses stream_parameters from dictionary. |
| |
| Args: |
| stream_parameters: dict of stream parameters |
| 'receiver': WmmTransceiver, the receiver for the stream |
| 'access_category': String, the access category to use for the |
| stream. Unset if None. |
| 'bandwidth': int, bandwidth in mbps for the stream. If set, |
| implies UDP. If unset, implies TCP and unlimited bandwidth. |
| 'time': int, time in seconds to run stream. |
| |
| Returns: |
| (receiver, access_category, bandwidth, time) as |
| (WmmTransceiver, String, int, int) |
| """ |
| receiver = stream_parameters['receiver'] |
| access_category = stream_parameters.get('access_category', None) |
| bandwidth = stream_parameters.get('bandwidth', None) |
| time = stream_parameters.get('time', DEFAULT_STREAM_TIME) |
| return (receiver, access_category, bandwidth, time) |
| |
| |
| class WmmTransceiverLoggerAdapter(logging.LoggerAdapter): |
| def process(self, msg, kwargs): |
| if self.extra['identifier']: |
| log_identifier = ' | %s' % self.extra['identifier'] |
| else: |
| log_identifier = '' |
| msg = "[WmmTransceiver%s] %s" % (log_identifier, msg) |
| return (msg, kwargs) |
