blob: 0d86db17af179e18ce0259ce1335e281dfaa758f [file] [log] [blame]
# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
import logging
from autotest_lib.client.common_lib import error
from autotest_lib.client.common_lib import utils
from import iw_runner
from import tcpdump_analyzer
from import xmlrpc_datatypes
from import hostap_config
from import netperf_runner
from import wifi_cell_test_base
class network_WiFi_RateControl(wifi_cell_test_base.WiFiCellTestBase):
Test maximal achievable bandwidth on several channels per band.
Conducts a performance test for a set of specified router configurations
and reports results as keyval pairs.
version = 1
# In case of aggregated frames, we need to capture whole packet to be able
# to parse the A-MSDU subframe and extract the IP/UDP content.
# Per spec the max AMSDU size for 11n is 7935bytes, let use it here.
def parse_additional_arguments(self, commandline_args, additional_params):
Hook into super class to take control files parameters.
@param commandline_args: dict of parsed parameters from the autotest.
@param additional_params: list of HostapConfig objects.
self._ap_configs = additional_params
def get_highest_mcs_rate(self, frequency):
Get the highest MCS index supported by the DUT on |frequency|.
@param frequency: int frequency to look for supported MCS rates.
@return int highest rate supported.
# Figure out the highest MCS index supported by this hardware.
phys = iw_runner.IwRunner(
if len(phys) != 1:
raise error.TestFail('Test expects a single PHY, but we got %d' %
phy = phys[0]
bands = [band for band in phy.bands if frequency in band.frequencies]
if len(bands) != 1:
raise error.TestFail('Test expects a single possible band for a '
'given frequency, but this device has %d '
'such bands.' % len(bands))
# 32 is a special low throughput, high resilience mode. Ignore it.
possible_indices = filter(lambda x: x != 32, bands[0].mcs_indices)
if not possible_indices:
raise error.TestFail('No possible MCS indices on frequency %d' %
return max(possible_indices)
def check_bitrates_in_capture(self, pcap_result, max_mcs_index):
Check that frames in a packet capture have expected MCS indices.
@param pcap_result: RemoteCaptureResult tuple.
@param max_mcs_index: int MCS index representing the highest possible
bitrate on this device.
"""'Analyzing packet capture...')
display_filter = 'udp and ip.src==%s' % self.context.client.wifi_ip
frames = tcpdump_analyzer.get_frames(
bad_fcs='include')'Grouping frames by MCS index')
counts = {}
for frame in frames:
counts[frame.mcs_index] = counts.get(frame.mcs_index, 0) + 1'Saw WiFi frames with MCS indices: %r', counts)
# Now figure out the index which the device sent the most packets with.
dominant_index = None
num_packets_sent = -1
for index, num_packets in counts.iteritems():
if num_packets > num_packets_sent:
dominant_index = index
num_packets_sent = num_packets
# We should see that the device sent more frames with the maximal index
# than anything else. This checks that the rate controller is fairly
# aggressive and using all of the device's capabilities.
if dominant_index != max_mcs_index:
raise error.TestFail('Failed to use best possible MCS '
'index %d in a clean RF environment: %r' %
(max_mcs_index, counts))
def run_once(self):
"""Test body."""
if utils.host_could_be_in_afe(
# Just abort the test if we're in the lab and not on a
# machine known to be conducted. The performance
# requirements of this test are hard to meet, without
# strong multi-path effects. (Our conducted setups are
# designed to provide strong multi-path.)
if not self.context.client.conductive:
raise error.TestNAError(
'This test requires a great RF environment.')
logging.error('Unable to determine if DUT has conducted '
'connection to AP. Treat any TestFail with '
caps = [hostap_config.HostapConfig.N_CAPABILITY_GREENFIELD,
mode_11n = hostap_config.HostapConfig.MODE_11N_PURE
get_config = lambda channel: hostap_config.HostapConfig(
channel=channel, mode=mode_11n, n_capabilities=caps)
netperf_config = netperf_runner.NetperfConfig(
for i, ap_config in enumerate([get_config(1), get_config(157)]):
# Set up the router and associate the client with it.
assoc_params = xmlrpc_datatypes.AssociationParameters(
with netperf_runner.NetperfRunner(self.context.client,
netperf_config) as runner:
results = self.context.capture_host.stop_capture()
if len(results) != 1:
raise error.TestError('Expected to generate one packet '
'capture but got %d instead.' %
# The device should sense that it is in a clean RF environment and
# use the highest index to achieve maximal throughput.
max_mcs_index = self.get_highest_mcs_rate(ap_config.frequency)
self.check_bitrates_in_capture(results[0], max_mcs_index)
# Clean up router and client state for the next run.