testsuites/vstf/vstf_scripts/vstf/agent/perf/qperf.py

   1 ##############################################################################
   2 # Copyright (c) 2015 Huawei Technologies Co.,Ltd and others.
   3 #
   4 # All rights reserved. This program and the accompanying materials
   5 # are made available under the terms of the Apache License, Version 2.0
   6 # which accompanies this distribution, and is available at
   7 # http://www.apache.org/licenses/LICENSE-2.0
   8 ##############################################################################
   9
  10 import subprocess
  11 import time
  12 import logging
  13 import vstf.common.decorator as deco
  14 from vstf.common import perfmark as mark
  15 from vstf.common.utils import kill_by_name, my_popen
  16
  17 LOG = logging.getLogger(__name__)
  18
  19
  20 class Qperf(object):
  21
  22     def __init__(self):
  23         self._send_processes = []
  24         self._receive_processes = []
  25
  26     @deco.check("protocol", choices=['tcp_lat', 'udp_lat'])
  27     @deco.check("namespace", defaults=None)
  28     @deco.check("dst")
  29     @deco.check("time", defaults=10)
  30     @deco.check("size", defaults=64)
  31     def send_start(self, **kwargs):
  32         cmd = self.format_send_start(**kwargs)
  33         LOG.info("cmd:%s", cmd)
  34         process = my_popen(
  35             cmd.split(),
  36             stdout=subprocess.PIPE,
  37             stderr=subprocess.PIPE)
  38         time.sleep(0.5)
  39         ret = process.poll()
  40         if ret is None:
  41             ret = 0
  42             error_str = "start qperf send success"
  43             self._send_processes.append(process)
  44         else:
  45             print ret
  46             error_str = "start qperf send failed, %s" % (str(kwargs))
  47             process.wait()
  48
  49         return ret, error_str
  50
  51     @deco.namespace()
  52     def format_send_start(self, **kwargs):
  53         cmd = "qperf %(dst_ip)s -t %(time)s -m %(pkt_size)s -vu %(type)s "
  54         context = {
  55             'dst_ip': kwargs['dst'][0]['ip'],
  56             'type': kwargs['protocol'],
  57             'time': kwargs['time'],
  58             'pkt_size': kwargs['size'],
  59         }
  60         cmd = cmd % context
  61         return cmd
  62
  63     def send_stop(self, **kwargs):
  64         results = []
  65         for process in self._send_processes:
  66             process.wait()
  67             read = process.stdout.read()
  68             read = self._parse_data(read)
  69             ret = 0
  70             results.append((ret, read))
  71         self._send_processes = []
  72         return results
  73
  74     @deco.namespace()
  75     def format_receive_start(self, **kwargs):
  76         cmd = 'qperf'
  77         return cmd
  78
  79     def receive_start(self, **kwargs):
  80         cmd = self.format_receive_start(**kwargs)
  81         LOG.info("cmd:%s", cmd)
  82
  83         process = my_popen(
  84             cmd.split(),
  85             stdout=subprocess.PIPE,
  86             stderr=subprocess.PIPE)
  87         time.sleep(0.5)
  88         ret = process.poll()
  89         if ret is None:
  90             ret = 0
  91             error_str = "start qperf receive success"
  92             self._receive_processes.append(process)
  93         else:
  94             print ret
  95             error_str = "start qperf receive failed, %s" % (str(kwargs))
  96             process.wait()
  97             raise Exception(error_str)
  98         return ret, error_str
  99
 100     def receive_stop(self, **kwargs):
 101         ret = 0
 102         for process in self._receive_processes:
 103             process.kill()
 104             process.wait()
 105         self._receive_processes = []
 106         error_str = "stop qperf receive success"
 107         return ret, error_str
 108
 109     def receive_kill(self):
 110         kill_by_name('qperf')
 111         self._receive_processes = []
 112         return True
 113
 114     def clean(self):
 115         for process in self._receive_processes:
 116             process.kill()
 117             process.wait()
 118             LOG.info("process.kill(qperf daemon:%s)", process.pid)
 119         for process in self._send_processes:
 120             LOG.info("process.wait(qperf client:%s)", process.pid)
 121             process.wait()
 122         self._receive_processes = []
 123         self._send_processes = []
 124         return True
 125
 126     def force_clean(self):
 127         LOG.info("%s %s start", self.__class__, self.force_clean.__name__)
 128         kill_by_name('qperf')
 129         self._send_processes = []
 130         self._receive_processes = []
 131         return True
 132
 133     def _parse_data(self, data):
 134         LOG.info(data)
 135         latency = 0
 136         if data:
 137             buf = data.splitlines()
 138             if "latency" in buf[1]:
 139                 data = buf[1].strip().split()
 140                 if data[3] == "us":
 141                     latency = float(data[2]) / 1000
 142                 else:
 143                     latency = float(data[2])
 144         result = {
 145             mark.minLatency: latency,
 146             mark.avgLatency: latency,
 147             mark.maxLatency: latency
 148         }
 149         return result
 150
 151
 152 def unit_test():
 153     perf = Qperf()
 154     perf.receive_start(namespace='receive')
 155
 156     send = {
 157         "namespace": "send",
 158         "protocol": "udp_lat",
 159         "dst": [
 160             {"ip": "192.168.1.102"}
 161         ],
 162         "size": 64,
 163     }
 164     print perf.send_start(**send)
 165     time.sleep(10)
 166     print perf.send_stop()
 167     print perf.receive_stop()
 168
 169
 170 if __name__ == "__main__":
 171     from vstf.common.log import setup_logging
 172
 173     setup_logging(
 174         level=logging.DEBUG,
 175         log_file="/var/log/vstf/vstf-qperf.log",
 176         clevel=logging.DEBUG)
 177     unit_test()