diff options
Diffstat (limited to 'scripts/automation/trex_control_plane')
20 files changed, 611 insertions, 212 deletions
diff --git a/scripts/automation/trex_control_plane/server/outer_packages.py b/scripts/automation/trex_control_plane/server/outer_packages.py index 313a93a6..f49a9925 100755 --- a/scripts/automation/trex_control_plane/server/outer_packages.py +++ b/scripts/automation/trex_control_plane/server/outer_packages.py @@ -2,11 +2,12 @@ import sys import os +python_ver = 'python%s' % sys.version_info.major CURRENT_PATH = os.path.dirname(os.path.realpath(__file__)) ROOT_PATH = os.path.abspath(os.path.join(CURRENT_PATH, os.pardir)) # path to trex_control_plane directory PATH_TO_PYTHON_LIB = os.path.abspath(os.path.join(ROOT_PATH, os.pardir, os.pardir, 'external_libs')) -PATH_TO_PLATFORM_LIB = os.path.abspath(os.path.join(PATH_TO_PYTHON_LIB, 'pyzmq-14.5.0', 'python2', 'fedora18', '64bit')) +PATH_TO_PLATFORM_LIB = os.path.abspath(os.path.join(PATH_TO_PYTHON_LIB, 'pyzmq-14.5.0', python_ver , 'fedora18', '64bit')) SERVER_MODULES = ['enum34-1.0.4', 'zmq', diff --git a/scripts/automation/trex_control_plane/server/singleton_daemon.py b/scripts/automation/trex_control_plane/server/singleton_daemon.py index 1784cc42..cd16d173 100755 --- a/scripts/automation/trex_control_plane/server/singleton_daemon.py +++ b/scripts/automation/trex_control_plane/server/singleton_daemon.py @@ -140,6 +140,7 @@ class SingletonDaemon(object): def restart(self, timeout = 15): if self.is_running(): self.kill(timeout) + sleep(0.5) return self.start(timeout) diff --git a/scripts/automation/trex_control_plane/server/trex_launch_thread.py b/scripts/automation/trex_control_plane/server/trex_launch_thread.py index 22606753..f4ee0d6b 100755 --- a/scripts/automation/trex_control_plane/server/trex_launch_thread.py +++ b/scripts/automation/trex_control_plane/server/trex_launch_thread.py @@ -33,7 +33,8 @@ class AsynchronousTRexSession(threading.Thread): try:
with open(self.export_path, 'w') as output_file:
self.time_stamps['start'] = self.time_stamps['run_time'] = time.time()
- self.session = subprocess.Popen(shlex.split(self.cmd), cwd = self.launch_path, stdout = output_file, preexec_fn=os.setsid, close_fds = True)
+ self.session = subprocess.Popen(shlex.split(self.cmd), cwd = self.launch_path, stdout = output_file,
+ stderr = subprocess.STDOUT, preexec_fn=os.setsid, close_fds = True)
logger.info("TRex session initialized successfully, Parent process pid is {pid}.".format( pid = self.session.pid ))
while self.session.poll() is None: # subprocess is NOT finished
time.sleep(0.5)
diff --git a/scripts/automation/trex_control_plane/server/trex_server.py b/scripts/automation/trex_control_plane/server/trex_server.py index 9fe7d70b..2b718a69 100755 --- a/scripts/automation/trex_control_plane/server/trex_server.py +++ b/scripts/automation/trex_control_plane/server/trex_server.py @@ -109,7 +109,7 @@ class CTRexServer(object): # initialize the server instance with given resources register_socket('trex_daemon_server') try: - print "Firing up TRex REST daemon @ port {trex_port} ...\n".format( trex_port = self.trex_daemon_port ) + print("Firing up TRex REST daemon @ port {trex_port} ...\n".format( trex_port = self.trex_daemon_port )) logger.info("Firing up TRex REST daemon @ port {trex_port} ...".format( trex_port = self.trex_daemon_port )) logger.info("current working dir is: {0}".format(self.TREX_PATH) ) logger.info("current files dir is : {0}".format(self.trex_files_path) ) @@ -119,7 +119,7 @@ class CTRexServer(object): except socket.error as e: if e.errno == errno.EADDRINUSE: logger.error("TRex server requested address already in use. Aborting server launching.") - print "TRex server requested address already in use. Aborting server launching." + print("TRex server requested address already in use. Aborting server launching.") raise socket.error(errno.EADDRINUSE, "TRex daemon requested address already in use. " "Server launch aborted. Please make sure no other process is " "using the desired server properties.") @@ -142,6 +142,7 @@ class CTRexServer(object): self.server.register_function(self.get_running_info) self.server.register_function(self.get_running_status) self.server.register_function(self.get_trex_cmds) + self.server.register_function(self.get_trex_config) self.server.register_function(self.get_trex_daemon_log) self.server.register_function(self.get_trex_log) self.server.register_function(self.get_trex_version) @@ -172,7 +173,7 @@ class CTRexServer(object): try: with open(filepath, 'rb') as f: file_content = f.read() - return binascii.b2a_base64(file_content) + return binascii.b2a_base64(file_content).decode(errors='replace') except Exception as e: err_str = "Can't get requested file %s: %s" % (filepath, e) logger.error(err_str) @@ -215,6 +216,11 @@ class CTRexServer(object): logger.info("Processing get_trex_log() command.") return self._pull_file('/tmp/trex.txt') + # get /etc/trex_cfg.yaml + def get_trex_config(self): + logger.info("Processing get_trex_config() command.") + return self._pull_file('/etc/trex_cfg.yaml') + # get daemon log /var/log/trex/trex_daemon_server.log def get_trex_daemon_log (self): logger.info("Processing get_trex_daemon_log() command.") @@ -229,11 +235,11 @@ class CTRexServer(object): search_result = re.search('\n\s*(Version\s*:.+)', stdout, re.DOTALL) if not search_result: raise Exception('Could not determine version from ./t-rex-64 --help') - self.trex_version = binascii.b2a_base64(search_result.group(1)) + self.trex_version = binascii.b2a_base64(search_result.group(1).encode(errors='replace')) if base64: - return self.trex_version + return self.trex_version.decode(errors='replace') else: - return binascii.a2b_base64(self.trex_version) + return binascii.a2b_base64(self.trex_version).decode(errors='replace') except Exception as e: err_str = "Can't get trex version, error: %s" % e logger.error(err_str) @@ -246,6 +252,14 @@ class CTRexServer(object): self.stop_trex(self.trex.get_seq()) sys.exit(0) + def assert_zmq_ok(self): + if self.trex.zmq_error: + raise Exception('ZMQ thread got error: %s' % self.trex.zmq_error) + if not self.zmq_monitor.is_alive(): + if self.trex.get_status() != TRexStatus.Idle: + self.force_trex_kill() + raise Exception('ZMQ thread is dead.') + def is_running (self): run_status = self.trex.get_status() logger.info("Processing is_running() command. Running status is: {stat}".format(stat = run_status) ) @@ -309,8 +323,8 @@ class CTRexServer(object): assert(self.__reservation is None) return False - def start_trex(self, trex_cmd_options, user, block_to_success = True, timeout = 40, stateless = False, debug_image = False, trex_args = ''): + self.assert_zmq_ok() with self.start_lock: logger.info("Processing start_trex() command.") if self.is_reserved(): @@ -337,6 +351,7 @@ class CTRexServer(object): break else: time.sleep(0.5) + self.assert_zmq_ok() # check for TRex run started normally if trex_state == TRexStatus.Starting: # reached timeout @@ -403,12 +418,15 @@ class CTRexServer(object): trex_state = None start_time = time.time() while (time.time() - start_time) < timeout : + self.assert_zmq_ok() trex_state = self.trex.get_status() if trex_state != TRexStatus.Starting: return + sleep(0.1) return Fault(-12, 'TimeoutError: TRex initiation outcome could not be obtained, since TRex stays at Starting state beyond defined timeout.') # raise at client TRexWarning def get_running_info (self): + self.assert_zmq_ok() logger.info("Processing get_running_info() command.") return self.trex.get_running_info() @@ -441,7 +459,7 @@ class CTRexServer(object): # adding additional options to the command trex_cmd_options = '' - for key, value in kwargs.iteritems(): + for key, value in kwargs.items(): tmp_key = key.replace('_','-').lstrip('-') dash = ' -' if (len(key)==1) else ' --' if value is True: @@ -474,13 +492,13 @@ class CTRexServer(object): def __check_trex_path_validity(self): # check for executable existance if not os.path.exists(self.TREX_PATH+'/t-rex-64'): - print "The provided TRex path do not contain an executable TRex file.\nPlease check the path and retry." + print("The provided TRex path do not contain an executable TRex file.\nPlease check the path and retry.") logger.error("The provided TRex path do not contain an executable TRex file") exit(-1) # check for executable permissions st = os.stat(self.TREX_PATH+'/t-rex-64') if not bool(st.st_mode & (stat.S_IXUSR ) ): - print "The provided TRex path do not contain an TRex file with execution privileges.\nPlease check the files permissions and retry." + print("The provided TRex path do not contain an TRex file with execution privileges.\nPlease check the files permissions and retry.") logger.error("The provided TRex path do not contain an TRex file with execution privileges") exit(-1) else: @@ -490,16 +508,16 @@ class CTRexServer(object): # first, check for path existance. otherwise, try creating it with appropriate credentials if not os.path.exists(self.trex_files_path): try: - os.makedirs(self.trex_files_path, 0660) + os.makedirs(self.trex_files_path, 0o660) return except os.error as inst: - print "The provided files path does not exist and cannot be created with needed access credentials using root user.\nPlease check the path's permissions and retry." + print("The provided files path does not exist and cannot be created with needed access credentials using root user.\nPlease check the path's permissions and retry.") logger.error("The provided files path does not exist and cannot be created with needed access credentials using root user.") exit(-1) elif os.access(self.trex_files_path, os.W_OK): return else: - print "The provided files path has insufficient access credentials for root user.\nPlease check the path's permissions and retry." + print("The provided files path has insufficient access credentials for root user.\nPlease check the path's permissions and retry.") logger.error("The provided files path has insufficient access credentials for root user") exit(-1) @@ -511,6 +529,7 @@ class CTRex(object): self.session = None self.zmq_monitor = None self.zmq_dump = None + self.zmq_error = None self.seq = None self.expect_trex = threading.Event() self.encoder = JSONEncoder() diff --git a/scripts/automation/trex_control_plane/server/zipmsg.py b/scripts/automation/trex_control_plane/server/zipmsg.py new file mode 100644 index 00000000..397ada16 --- /dev/null +++ b/scripts/automation/trex_control_plane/server/zipmsg.py @@ -0,0 +1,32 @@ +import zlib +import struct + +class ZippedMsg: + + MSG_COMPRESS_THRESHOLD = 256 + MSG_COMPRESS_HEADER_MAGIC = 0xABE85CEA + + def check_threshold (self, msg): + return len(msg) >= self.MSG_COMPRESS_THRESHOLD + + def compress (self, msg): + # compress + compressed = zlib.compress(msg) + new_msg = struct.pack(">II", self.MSG_COMPRESS_HEADER_MAGIC, len(msg)) + compressed + return new_msg + + + def decompress (self, msg): + if len(msg) < 8: + return None + + t = struct.unpack(">II", msg[:8]) + if (t[0] != self.MSG_COMPRESS_HEADER_MAGIC): + return None + + x = zlib.decompress(msg[8:]) + if len(x) != t[1]: + return None + + return x + diff --git a/scripts/automation/trex_control_plane/server/zmq_monitor_thread.py b/scripts/automation/trex_control_plane/server/zmq_monitor_thread.py index db9bf7da..f559ebc1 100755 --- a/scripts/automation/trex_control_plane/server/zmq_monitor_thread.py +++ b/scripts/automation/trex_control_plane/server/zmq_monitor_thread.py @@ -6,6 +6,7 @@ import zmq import threading
import logging
import CCustomLogger
+import zipmsg
from json import JSONDecoder
from common.trex_status_e import TRexStatus
@@ -24,28 +25,33 @@ class ZmqMonitorSession(threading.Thread): self.trexObj = trexObj
self.expect_trex = self.trexObj.expect_trex # used to signal if TRex is expected to run and if data should be considered
self.decoder = JSONDecoder()
+ self.zipped = zipmsg.ZippedMsg()
logger.info("ZMQ monitor initialization finished")
def run(self):
- self.context = zmq.Context()
- self.socket = self.context.socket(zmq.SUB)
- logger.info("ZMQ monitor started listening @ {pub}".format(pub=self.zmq_publisher))
- self.socket.connect(self.zmq_publisher)
- self.socket.setsockopt(zmq.SUBSCRIBE, '')
-
- while not self.stoprequest.is_set():
- try:
- zmq_dump = self.socket.recv() # This call is BLOCKING until data received!
- if self.expect_trex.is_set():
- self.parse_and_update_zmq_dump(zmq_dump)
- logger.debug("ZMQ dump received on socket, and saved to trexObject.")
- except Exception as e:
- if self.stoprequest.is_set():
- # allow this exception since it comes from ZMQ monitor termination
- pass
- else:
- logger.error("ZMQ monitor thrown an exception. Received exception: {ex}".format(ex=e))
- raise
+ try:
+ self.context = zmq.Context()
+ self.socket = self.context.socket(zmq.SUB)
+ logger.info("ZMQ monitor started listening @ {pub}".format(pub=self.zmq_publisher))
+ self.socket.connect(self.zmq_publisher)
+ self.socket.setsockopt(zmq.SUBSCRIBE, b'')
+
+ while not self.stoprequest.is_set():
+ try:
+ zmq_dump = self.socket.recv() # This call is BLOCKING until data received!
+ if self.expect_trex.is_set():
+ self.parse_and_update_zmq_dump(zmq_dump)
+ logger.debug("ZMQ dump received on socket, and saved to trexObject.")
+ except Exception as e:
+ if self.stoprequest.is_set():
+ # allow this exception since it comes from ZMQ monitor termination
+ pass
+ else:
+ logger.error("ZMQ monitor thrown an exception. Received exception: {ex}".format(ex=e))
+ raise
+ except Exception as e:
+ logger.error('ZMQ monitor error: %s' % e)
+ self.trexObj.zmq_error = e
def join(self, timeout=None):
self.stoprequest.set()
@@ -56,14 +62,16 @@ class ZmqMonitorSession(threading.Thread): super(ZmqMonitorSession, self).join(timeout)
def parse_and_update_zmq_dump(self, zmq_dump):
- try:
- dict_obj = self.decoder.decode(zmq_dump)
- except ValueError:
- logger.error("ZMQ dump failed JSON-RPC decode. Ignoring. Bad dump was: {dump}".format(dump=zmq_dump))
- dict_obj = None
+ unzipped = self.zipped.decompress(zmq_dump)
+ if unzipped:
+ zmq_dump = unzipped
+ dict_obj = self.decoder.decode(zmq_dump.decode(errors = 'replace'))
+
+ if type(dict_obj) is not dict:
+ raise Exception('Expected ZMQ dump of type dict, got: %s' % type(dict_obj))
# add to trex_obj zmq latest dump, based on its 'name' header
- if dict_obj is not None and dict_obj != {}:
+ if dict_obj != {}:
self.trexObj.zmq_dump[dict_obj['name']] = dict_obj
if self.first_dump:
# change TRexStatus from starting to Running once the first ZMQ dump is obtained and parsed successfully
diff --git a/scripts/automation/trex_control_plane/stl/console/trex_console.py b/scripts/automation/trex_control_plane/stl/console/trex_console.py index 110457d6..5d23d8da 100755 --- a/scripts/automation/trex_control_plane/stl/console/trex_console.py +++ b/scripts/automation/trex_control_plane/stl/console/trex_console.py @@ -440,6 +440,7 @@ class TRexConsole(TRexGeneralCmd): if (l > 2) and (s[l - 2] in file_flags): return TRexConsole.tree_autocomplete(s[l - 1]) + complete_push = complete_start @verify_connected def do_start(self, line): diff --git a/scripts/automation/trex_control_plane/stl/console/trex_tui.py b/scripts/automation/trex_control_plane/stl/console/trex_tui.py index a69c4165..e769b9b2 100644 --- a/scripts/automation/trex_control_plane/stl/console/trex_tui.py +++ b/scripts/automation/trex_control_plane/stl/console/trex_tui.py @@ -1,11 +1,18 @@ +from __future__ import print_function + import termios import sys import os import time +import threading + from collections import OrderedDict, deque +from texttable import ansi_len + + import datetime import readline -from texttable import ansi_len + if sys.version_info > (3,0): from io import StringIO @@ -41,11 +48,11 @@ class SimpleBar(object): self.pattern_len = len(pattern) self.index = 0 - def show (self): + def show (self, buffer): if self.desc: - print(format_text("{0} {1}".format(self.desc, self.pattern[self.index]), 'bold')) + print(format_text("{0} {1}".format(self.desc, self.pattern[self.index]), 'bold'), file = buffer) else: - print(format_text("{0}".format(self.pattern[self.index]), 'bold')) + print(format_text("{0}".format(self.pattern[self.index]), 'bold'), file = buffer) self.index = (self.index + 1) % self.pattern_len @@ -59,7 +66,7 @@ class TrexTUIPanel(object): self.stateless_client = mng.stateless_client self.is_graph = False - def show (self): + def show (self, buffer): raise NotImplementedError("must implement this") def get_key_actions (self): @@ -108,11 +115,11 @@ class TrexTUIDashBoard(TrexTUIPanel): return self.toggle_filter.filter_items() - def show (self): + def show (self, buffer): stats = self.stateless_client._get_formatted_stats(self.get_showed_ports()) # print stats to screen for stat_type, stat_data in stats.items(): - text_tables.print_table_with_header(stat_data.text_table, stat_type) + text_tables.print_table_with_header(stat_data.text_table, stat_type, buffer = buffer) def get_key_actions (self): @@ -203,11 +210,11 @@ class TrexTUIStreamsStats(TrexTUIPanel): self.key_actions['c'] = {'action': self.action_clear, 'legend': 'clear', 'show': True} - def show (self): + def show (self, buffer): stats = self.stateless_client._get_formatted_stats(port_id_list = None, stats_mask = trex_stl_stats.SS_COMPAT) # print stats to screen for stat_type, stat_data in stats.items(): - text_tables.print_table_with_header(stat_data.text_table, stat_type) + text_tables.print_table_with_header(stat_data.text_table, stat_type, buffer = buffer) pass @@ -230,7 +237,7 @@ class TrexTUILatencyStats(TrexTUIPanel): self.is_histogram = False - def show (self): + def show (self, buffer): if self.is_histogram: stats = self.stateless_client._get_formatted_stats(port_id_list = None, stats_mask = trex_stl_stats.LH_COMPAT) else: @@ -241,7 +248,7 @@ class TrexTUILatencyStats(TrexTUIPanel): untouched_header = ' (usec)' else: untouched_header = '' - text_tables.print_table_with_header(stat_data.text_table, stat_type, untouched_header = untouched_header) + text_tables.print_table_with_header(stat_data.text_table, stat_type, untouched_header = untouched_header, buffer = buffer) def get_key_actions (self): return self.key_actions @@ -261,11 +268,11 @@ class TrexTUIUtilizationStats(TrexTUIPanel): super(TrexTUIUtilizationStats, self).__init__(mng, "ustats") self.key_actions = {} - def show (self): + def show (self, buffer): stats = self.stateless_client._get_formatted_stats(port_id_list = None, stats_mask = trex_stl_stats.UT_COMPAT) # print stats to screen for stat_type, stat_data in stats.items(): - text_tables.print_table_with_header(stat_data.text_table, stat_type) + text_tables.print_table_with_header(stat_data.text_table, stat_type, buffer = buffer) def get_key_actions (self): return self.key_actions @@ -279,16 +286,16 @@ class TrexTUILog(): def add_event (self, msg): self.log.append("[{0}] {1}".format(str(datetime.datetime.now().time()), msg)) - def show (self, max_lines = 4): + def show (self, buffer, max_lines = 4): cut = len(self.log) - max_lines if cut < 0: cut = 0 - print(format_text("\nLog:", 'bold', 'underline')) + print(format_text("\nLog:", 'bold', 'underline'), file = buffer) for msg in self.log[cut:]: - print(msg) + print(msg, file = buffer) # a predicate to wrap function as a bool @@ -366,14 +373,14 @@ class TrexTUIPanelManager(): self.legend += "{:}".format(x) - def print_connection_status (self): + def print_connection_status (self, buffer): if self.tui.get_state() == self.tui.STATE_ACTIVE: - self.conn_bar.show() + self.conn_bar.show(buffer = buffer) else: - self.dis_bar.show() + self.dis_bar.show(buffer = buffer) - def print_legend (self): - print(format_text(self.legend, 'bold')) + def print_legend (self, buffer): + print(format_text(self.legend, 'bold'), file = buffer) # on window switch or turn on / off of the TUI we call this @@ -382,16 +389,16 @@ class TrexTUIPanelManager(): self.locked = locked self.generate_legend() - def show (self, show_legend): - self.main_panel.show() - self.print_connection_status() + def show (self, show_legend, buffer): + self.main_panel.show(buffer) + self.print_connection_status(buffer) if show_legend: self.generate_legend() - self.print_legend() + self.print_legend(buffer) if self.show_log: - self.log.show() + self.log.show(buffer) def handle_key (self, ch): @@ -452,6 +459,89 @@ class TrexTUIPanelManager(): self.init(self.show_log) return "" + + +# ScreenBuffer is a class designed to +# avoid inline delays when reprinting the screen +class ScreenBuffer(): + def __init__ (self, redraw_cb): + self.snapshot = '' + self.lock = threading.Lock() + + self.redraw_cb = redraw_cb + self.update_flag = False + + + def start (self): + self.active = True + self.t = threading.Thread(target = self.__handler) + self.t.setDaemon(True) + self.t.start() + + def stop (self): + self.active = False + self.t.join() + + + # request an update + def update (self): + self.update_flag = True + + # fetch the screen, return None if no new screen exists yet + def get (self): + + if not self.snapshot: + return None + + # we have a snapshot - fetch it + with self.lock: + x = self.snapshot + self.snapshot = None + return x + + + def __handler (self): + + while self.active: + if self.update_flag: + self.__redraw() + + time.sleep(0.01) + + # redraw the next screen + def __redraw (self): + buffer = StringIO() + + self.redraw_cb(buffer) + + with self.lock: + self.snapshot = buffer + self.update_flag = False + +# a policer class to make sure no too-fast redraws +# occurs - it filters fast bursts of redraws +class RedrawPolicer(): + def __init__ (self, rate): + self.ts = 0 + self.marked = False + self.rate = rate + self.force = False + + def mark_for_redraw (self, force = False): + self.marked = True + if force: + self.force = True + + def should_redraw (self): + dt = time.time() - self.ts + return self.force or (self.marked and (dt > self.rate)) + + def reset (self, restart = False): + self.ts = time.time() + self.marked = restart + self.force = False + + # shows a textual top style window class TrexTUI(): @@ -463,6 +553,7 @@ class TrexTUI(): MIN_ROWS = 50 MIN_COLS = 111 + class ScreenSizeException(Exception): def __init__ (self, cols, rows): msg = "TUI requires console screen size of at least {0}x{1}, current is {2}x{3}".format(TrexTUI.MIN_COLS, @@ -471,11 +562,18 @@ class TrexTUI(): rows) super(TrexTUI.ScreenSizeException, self).__init__(msg) + def __init__ (self, stateless_client): self.stateless_client = stateless_client + self.tui_global_lock = threading.Lock() self.pm = TrexTUIPanelManager(self) - + self.sb = ScreenBuffer(self.redraw_handler) + + def redraw_handler (self, buffer): + # this is executed by the screen buffer - should be protected against TUI commands + with self.tui_global_lock: + self.pm.show(show_legend = self.async_keys.is_legend_mode(), buffer = buffer) def clear_screen (self, lines = 50): # reposition the cursor @@ -490,7 +588,6 @@ class TrexTUI(): # reposition the cursor sys.stdout.write("\x1b[0;0H") - #sys.stdout.write("\x1b[2J\x1b[H") def show (self, client, save_console_history, show_log = False, locked = False): @@ -499,7 +596,7 @@ class TrexTUI(): if (int(rows) < TrexTUI.MIN_ROWS) or (int(cols) < TrexTUI.MIN_COLS): raise self.ScreenSizeException(rows = rows, cols = cols) - with AsyncKeys(client, save_console_history, locked) as async_keys: + with AsyncKeys(client, save_console_history, self.tui_global_lock, locked) as async_keys: sys.stdout.write("\x1bc") self.async_keys = async_keys self.show_internal(show_log, locked) @@ -511,84 +608,107 @@ class TrexTUI(): self.pm.init(show_log, locked) self.state = self.STATE_ACTIVE - self.last_redraw_ts = 0 + + # create print policers + self.full_redraw = RedrawPolicer(0.5) + self.keys_redraw = RedrawPolicer(0.05) + self.full_redraw.mark_for_redraw() + try: + self.sb.start() + while True: # draw and handle user input status = self.async_keys.tick(self.pm) - self.draw_screen(status) + # prepare the next frame + self.prepare(status) + time.sleep(0.01) + self.draw_screen() - # speedup for keys, slower for no keys - if status == AsyncKeys.STATUS_NONE: - time.sleep(0.01) - else: - time.sleep(0.001) + with self.tui_global_lock: + self.handle_state_machine() - # regular state - if self.state == self.STATE_ACTIVE: - # if no connectivity - move to lost connecitivty - if not self.stateless_client.async_client.is_alive(): - self.stateless_client._invalidate_stats(self.pm.ports) - self.state = self.STATE_LOST_CONT + except TUIQuit: + print("\nExiting TUI...") + finally: + self.sb.stop() - # lost connectivity - elif self.state == self.STATE_LOST_CONT: - # got it back - if self.stateless_client.async_client.is_alive(): - # move to state reconnect - self.state = self.STATE_RECONNECT + print("") + - # restored connectivity - try to reconnect - elif self.state == self.STATE_RECONNECT: + # handle state machine + def handle_state_machine (self): + # regular state + if self.state == self.STATE_ACTIVE: + # if no connectivity - move to lost connecitivty + if not self.stateless_client.async_client.is_alive(): + self.stateless_client._invalidate_stats(self.pm.ports) + self.state = self.STATE_LOST_CONT - try: - self.stateless_client.connect() - self.state = self.STATE_ACTIVE - except STLError: - self.state = self.STATE_LOST_CONT + # lost connectivity + elif self.state == self.STATE_LOST_CONT: + # got it back + if self.stateless_client.async_client.is_alive(): + # move to state reconnect + self.state = self.STATE_RECONNECT - except TUIQuit: - print("\nExiting TUI...") - print("") + # restored connectivity - try to reconnect + elif self.state == self.STATE_RECONNECT: + try: + self.stateless_client.connect() + self.stateless_client.acquire() + self.state = self.STATE_ACTIVE + except STLError: + self.state = self.STATE_LOST_CONT - # draw once - def draw_screen (self, status): - t = time.time() - self.last_redraw_ts - redraw = (t >= 0.5) or (status == AsyncKeys.STATUS_REDRAW_ALL) - if redraw: - # capture stdout to a string - old_stdout = sys.stdout - sys.stdout = mystdout = StringIO() - self.pm.show(show_legend = self.async_keys.is_legend_mode()) - self.last_snap = mystdout.getvalue() + # logic before printing + def prepare (self, status): + if status == AsyncKeys.STATUS_REDRAW_ALL: + self.full_redraw.mark_for_redraw(force = True) - self.async_keys.draw() - sys.stdout = old_stdout + elif status == AsyncKeys.STATUS_REDRAW_KEYS: + self.keys_redraw.mark_for_redraw() - self.clear_screen() + if self.full_redraw.should_redraw(): + self.sb.update() + self.full_redraw.reset(restart = True) - sys.stdout.write(mystdout.getvalue()) - + return + + + # draw once + def draw_screen (self): + + # check for screen buffer's new screen + x = self.sb.get() + + # we have a new screen to draw + if x: + self.clear_screen() + + self.async_keys.draw(x) + sys.stdout.write(x.getvalue()) sys.stdout.flush() - self.last_redraw_ts = time.time() - elif status == AsyncKeys.STATUS_REDRAW_KEYS: + # maybe we need to redraw the keys + elif self.keys_redraw.should_redraw(): sys.stdout.write("\x1b[4A") - - self.async_keys.draw() + self.async_keys.draw(sys.stdout) sys.stdout.flush() - return + # reset the policer for next time + self.keys_redraw.reset() + def get_state (self): return self.state @@ -607,7 +727,9 @@ class AsyncKeys: STATUS_REDRAW_KEYS = 1 STATUS_REDRAW_ALL = 2 - def __init__ (self, client, save_console_history, locked = False): + def __init__ (self, client, save_console_history, tui_global_lock, locked = False): + self.tui_global_lock = tui_global_lock + self.engine_console = AsyncKeysEngineConsole(self, client, save_console_history) self.engine_legend = AsyncKeysEngineLegend(self) self.locked = locked @@ -679,8 +801,8 @@ class AsyncKeys: return self.engine.tick(seq, pm) - def draw (self): - self.engine.draw() + def draw (self, buffer): + self.engine.draw(buffer) @@ -704,7 +826,7 @@ class AsyncKeysEngineLegend: rc = pm.handle_key(seq) return AsyncKeys.STATUS_REDRAW_ALL if rc else AsyncKeys.STATUS_NONE - def draw (self): + def draw (self, buffer): pass @@ -720,8 +842,10 @@ class AsyncKeysEngineConsole: self.ac = {'start' : client.start_line, 'stop' : client.stop_line, 'pause' : client.pause_line, + 'clear' : client.clear_stats_line, 'push' : client.push_line, 'resume' : client.resume_line, + 'reset' : client.reset_line, 'update' : client.update_line, 'connect' : client.connect_line, 'disconnect' : client.disconnect_line, @@ -804,7 +928,6 @@ class AsyncKeysEngineConsole: def handle_single_key (self, ch): - # newline if ch == '\n': self.handle_cmd() @@ -921,6 +1044,7 @@ class AsyncKeysEngineConsole: def handle_cmd (self): + cmd = self.lines[self.line_index].get().strip() if not cmd: return @@ -929,7 +1053,8 @@ class AsyncKeysEngineConsole: func = self.ac.get(op) if func: - func_rc = func(param) + with self.async.tui_global_lock: + func_rc = func(param) # take out the empty line empty_line = self.lines.popleft() @@ -962,7 +1087,6 @@ class AsyncKeysEngineConsole: # success if func_rc: self.last_status = format_text("[OK]", 'green') - # errors else: err_msgs = ascii_split(str(func_rc)) @@ -971,16 +1095,18 @@ class AsyncKeysEngineConsole: self.last_status += " [{0} more errors messages]".format(len(err_msgs) - 1) color = 'red' + + # trim too long lines if ansi_len(self.last_status) > TrexTUI.MIN_COLS: self.last_status = format_text(self.last_status[:TrexTUI.MIN_COLS] + "...", color, 'bold') - def draw (self): - sys.stdout.write("\nPress 'ESC' for navigation panel...\n") - sys.stdout.write("status: \x1b[0K{0}\n".format(self.last_status)) - sys.stdout.write("\n{0}\x1b[0K".format(self.generate_prompt(prefix = 'tui'))) - self.lines[self.line_index].draw() + def draw (self, buffer): + buffer.write("\nPress 'ESC' for navigation panel...\n") + buffer.write("status: \x1b[0K{0}\n".format(self.last_status)) + buffer.write("\n{0}\x1b[0K".format(self.generate_prompt(prefix = 'tui'))) + self.lines[self.line_index].draw(buffer) # a readline alike command line - can be modified during edit @@ -1055,7 +1181,7 @@ class CmdLine(object): def go_right (self): self.cursor_index = min(len(self.get()), self.cursor_index + 1) - def draw (self): - sys.stdout.write(self.get()) - sys.stdout.write('\b' * (len(self.get()) - self.cursor_index)) + def draw (self, buffer): + buffer.write(self.get()) + buffer.write('\b' * (len(self.get()) - self.cursor_index)) diff --git a/scripts/automation/trex_control_plane/stl/examples/using_rpc_proxy.py b/scripts/automation/trex_control_plane/stl/examples/using_rpc_proxy.py index 065f4284..d2fcdff3 100755 --- a/scripts/automation/trex_control_plane/stl/examples/using_rpc_proxy.py +++ b/scripts/automation/trex_control_plane/stl/examples/using_rpc_proxy.py @@ -4,6 +4,7 @@ import argparse import sys import os from time import sleep +from pprint import pprint # ext libs ext_libs = os.path.join(os.pardir, os.pardir, os.pardir, os.pardir, 'external_libs') @@ -74,14 +75,40 @@ if __name__ == '__main__': print('Sending pcap to ports %s' % ports) verify(server.push_remote(pcap_filename = 'stl/sample.pcap')) + sleep(3) print('Getting stats') res = verify(server.get_stats()) - print('Stats: %s' % res[1]) + pprint(res[1]) print('Resetting all ports') verify(server.reset()) + imix_path_1 = '../../../../stl/imix.py' + imix_path_2 = '../../stl/imix.py' + if os.path.exists(imix_path_1): + imix_path = imix_path_1 + elif os.path.exists(imix_path_2): + imix_path = imix_path_2 + else: + print('Could not find path of imix profile, skipping') + imix_path = None + + if imix_path: + print('Adding profile %s' % imix_path) + verify(server.native_method(func_name = 'add_profile', filename = imix_path)) + + print('Start traffic for 5 sec') + verify(server.native_method('start')) + sleep(5) + + print('Getting stats') + res = verify(server.get_stats()) + pprint(res[1]) + + print('Resetting all ports') + verify(server.reset()) + print('Deleting Native Client instance') verify(server.native_proxy_del()) @@ -116,7 +143,7 @@ if __name__ == '__main__': print('Getting stats') res = verify_hlt(server.traffic_stats(mode = 'aggregate', port_handle = ports[:2])) - print(res) + pprint(res) print('Deleting HLTAPI Client instance') verify_hlt(server.hltapi_proxy_del()) diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_async_client.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_async_client.py index 0f73792a..2c95844b 100644 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_async_client.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_async_client.py @@ -13,6 +13,7 @@ from .trex_stl_jsonrpc_client import JsonRpcClient, BatchMessage from .utils.text_opts import * from .trex_stl_stats import * from .trex_stl_types import * +from .utils.zipmsg import ZippedMsg # basic async stats class class CTRexAsyncStats(object): @@ -156,7 +157,9 @@ class CTRexAsyncClient(): self.monitor = AsyncUtil() self.connected = False - + + self.zipped = ZippedMsg() + # connects the async channel def connect (self): @@ -214,7 +217,7 @@ class CTRexAsyncClient(): # done self.connected = False - + # thread function def _run (self): @@ -232,10 +235,17 @@ class CTRexAsyncClient(): try: with self.monitor: - line = self.socket.recv_string() + line = self.socket.recv() self.monitor.on_recv_msg(line) + # try to decomrpess + unzipped = self.zipped.decompress(line) + if unzipped: + line = unzipped + + line = line.decode() + self.last_data_recv_ts = time.time() # signal once diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py index 4e3d3092..7101b8a2 100755 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_client.py @@ -453,6 +453,10 @@ class CCommLink(object): class STLClient(object): """TRex Stateless client object - gives operations per TRex/user""" + # different modes for attaching traffic to ports + CORE_MASK_SPLIT = 1 + CORE_MASK_PIN = 2 + def __init__(self, username = common.get_current_user(), server = "localhost", @@ -511,7 +515,7 @@ class STLClient(object): self.connected = False # API classes - self.api_vers = [ {'type': 'core', 'major': 1, 'minor': 3 } ] + self.api_vers = [ {'type': 'core', 'major': 2, 'minor': 3 } ] self.api_h = {'core': None} # logger @@ -675,14 +679,23 @@ class STLClient(object): return self.ports[port_id].get_stream_id_list() - def __start (self, multiplier, duration, port_id_list = None, force = False): + def __start (self, + multiplier, + duration, + port_id_list, + force, + core_mask): port_id_list = self.__ports(port_id_list) rc = RC() + for port_id in port_id_list: - rc.add(self.ports[port_id].start(multiplier, duration, force)) + rc.add(self.ports[port_id].start(multiplier, + duration, + force, + core_mask[port_id])) return rc @@ -800,13 +813,14 @@ class STLClient(object): self.server_version = rc.data() self.global_stats.server_version = rc.data() - + # cache system info rc = self._transmit("get_system_info") if not rc: return rc self.system_info = rc.data() + self.global_stats.system_info = rc.data() # cache supported commands rc = self._transmit("get_supported_cmds") @@ -907,6 +921,37 @@ class STLClient(object): return stats + def __decode_core_mask (self, ports, core_mask): + + # predefined modes + if isinstance(core_mask, int): + if core_mask not in [self.CORE_MASK_PIN, self.CORE_MASK_SPLIT]: + raise STLError("'core_mask' can be either CORE_MASK_PIN, CORE_MASK_SPLIT or a list of masks") + + decoded_mask = {} + for port in ports: + # a pin mode was requested and we have + # the second port from the group in the start list + if (core_mask == self.CORE_MASK_PIN) and ( (port ^ 0x1) in ports ): + decoded_mask[port] = 0x55555555 if( port % 2) == 0 else 0xAAAAAAAA + else: + decoded_mask[port] = None + + return decoded_mask + + # list of masks + elif isinstance(core_mask, list): + if len(ports) != len(core_mask): + raise STLError("'core_mask' list must be the same length as 'ports' list") + + decoded_mask = {} + for i, port in enumerate(ports): + decoded_mask[port] = core_mask[i] + + return decoded_mask + + + ############ functions used by other classes but not users ############## def _validate_port_list (self, port_id_list): @@ -1018,7 +1063,7 @@ class STLClient(object): try: ret = f(*args, **kwargs) except KeyboardInterrupt as e: - raise STLError("Test was interrupted by a keyboard signal (probably ctrl + c)") + raise STLError("Interrupted by a keyboard signal (probably ctrl + c)") return ret return wrap2 @@ -1536,7 +1581,6 @@ class STLClient(object): @__api_check(False) def connect (self): """ - def connect(self): Connects to the TRex server @@ -1829,6 +1873,33 @@ class STLClient(object): # return the stream IDs return rc.data() + @__api_check(True) + def add_profile(self, filename, ports = None, **kwargs): + """ | Add streams from profile by its type. Supported types are: + | .py + | .yaml + | .pcap file that converted to profile automatically + + :parameters: + filename : string + filename (with path) of the profile + ports : list + list of ports to add the profile (default: all acquired) + kwargs : dict + forward those key-value pairs to the profile (tunables) + + :returns: + List of stream IDs in order of the stream list + + :raises: + + :exc:`STLError` + + """ + + validate_type('filename', filename, basestring) + profile = STLProfile.load(filename, **kwargs) + return self.add_streams(profile.get_streams(), ports) + @__api_check(True) def remove_streams (self, stream_id_list, ports = None): @@ -1875,7 +1946,8 @@ class STLClient(object): mult = "1", force = False, duration = -1, - total = False): + total = False, + core_mask = CORE_MASK_SPLIT): """ Start traffic on port(s) @@ -1901,21 +1973,31 @@ class STLClient(object): True: Divide bandwidth among the ports False: Duplicate + core_mask: CORE_MASK_SPLIT, CORE_MASK_PIN or a list of masks (one per port) + Determine the allocation of cores per port + In CORE_MASK_SPLIT all the traffic will be divided equally between all the cores + associated with each port + In CORE_MASK_PIN, for each dual ports (a group that shares the same cores) + the cores will be divided half pinned for each port :raises: + :exc:`STLError` """ - ports = ports if ports is not None else self.get_acquired_ports() ports = self._validate_port_list(ports) - validate_type('mult', mult, basestring) validate_type('force', force, bool) validate_type('duration', duration, (int, float)) validate_type('total', total, bool) + validate_type('core_mask', core_mask, (int, list)) + + ######################### + # decode core mask argument + decoded_mask = self.__decode_core_mask(ports, core_mask) + ####################### # verify multiplier mult_obj = parsing_opts.decode_multiplier(mult, @@ -1938,7 +2020,7 @@ class STLClient(object): # start traffic self.logger.pre_cmd("Starting traffic on port(s) {0}:".format(ports)) - rc = self.__start(mult_obj, duration, ports, force) + rc = self.__start(mult_obj, duration, ports, force, decoded_mask) self.logger.post_cmd(rc) if not rc: @@ -2621,12 +2703,22 @@ class STLClient(object): parsing_opts.DURATION, parsing_opts.TUNABLES, parsing_opts.MULTIPLIER_STRICT, - parsing_opts.DRY_RUN) + parsing_opts.DRY_RUN, + parsing_opts.CORE_MASK_GROUP) opts = parser.parse_args(line.split(), default_ports = self.get_acquired_ports(), verify_acquired = True) if not opts: return opts + # core mask + if opts.core_mask is not None: + core_mask = opts.core_mask + else: + core_mask = self.CORE_MASK_PIN if opts.pin_cores else self.CORE_MASK_SPLIT + + # just for sanity - will be checked on the API as well + self.__decode_core_mask(opts.ports, core_mask) + active_ports = list_intersect(self.get_active_ports(), opts.ports) if active_ports: if not opts.force: @@ -2682,11 +2774,13 @@ class STLClient(object): if opts.dry: self.validate(opts.ports, opts.mult, opts.duration, opts.total) else: + self.start(opts.ports, opts.mult, opts.force, opts.duration, - opts.total) + opts.total, + core_mask) return RC_OK() @@ -2834,7 +2928,7 @@ class STLClient(object): self.clear_stats(opts.ports) - + return RC_OK() @__console diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_ext.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_ext.py index ed0c393d..306302dc 100644 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_ext.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_ext.py @@ -54,7 +54,7 @@ def import_module_list(modules_list): print("Unable to find required module library: '{0}'".format(p['name'])) print("Please provide the correct path using TREX_STL_EXT_PATH variable") print("current path used: '{0}'".format(full_path)) - exit(0) + exit(1) sys.path.insert(1, full_path) diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_jsonrpc_client.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_jsonrpc_client.py index 065a1442..609ea076 100644 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_jsonrpc_client.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_jsonrpc_client.py @@ -9,6 +9,7 @@ import struct from .trex_stl_types import * from .utils.common import random_id_gen +from .utils.zipmsg import ZippedMsg class bcolors: BLUE = '\033[94m' @@ -43,9 +44,6 @@ class BatchMessage(object): # JSON RPC v2.0 client class JsonRpcClient(object): - MSG_COMPRESS_THRESHOLD = 4096 - MSG_COMPRESS_HEADER_MAGIC = 0xABE85CEA - def __init__ (self, default_server, default_port, client): self.client_api = client.api_h self.logger = client.logger @@ -56,7 +54,7 @@ class JsonRpcClient(object): self.server = default_server self.id_gen = random_id_gen() - + self.zipper = ZippedMsg() def get_connection_details (self): rc = {} @@ -121,28 +119,7 @@ class JsonRpcClient(object): return self.send_msg(msg) - - def compress_msg (self, msg): - # compress - compressed = zlib.compress(msg) - new_msg = struct.pack(">II", self.MSG_COMPRESS_HEADER_MAGIC, len(msg)) + compressed - return new_msg - - - def decompress_msg (self, msg): - if len(msg) < 8: - return None - - t = struct.unpack(">II", msg[:8]) - if (t[0] != self.MSG_COMPRESS_HEADER_MAGIC): - return None - - x = zlib.decompress(msg[8:]) - if len(x) != t[1]: - return None - - return x - + def send_msg (self, msg): # print before if self.logger.check_verbose(self.logger.VERBOSE_HIGH): @@ -151,10 +128,10 @@ class JsonRpcClient(object): # encode string to buffer buffer = msg.encode() - if len(buffer) > self.MSG_COMPRESS_THRESHOLD: - response = self.send_raw_msg(self.compress_msg(buffer)) + if self.zipper.check_threshold(buffer): + response = self.send_raw_msg(self.zipper.compress(buffer)) if response: - response = self.decompress_msg(response) + response = self.zipper.decompress(response) else: response = self.send_raw_msg(buffer) diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py index d239fc57..890ce7de 100644 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_port.py @@ -36,6 +36,8 @@ class Port(object): STATE_PAUSE = 4 STATE_PCAP_TX = 5 + MASK_ALL = ((1 << 64) - 1) + PortState = namedtuple('PortState', ['state_id', 'state_name']) STATES_MAP = {STATE_DOWN: "DOWN", STATE_IDLE: "IDLE", @@ -100,7 +102,7 @@ class Port(object): # decorator to check server is readable (port not down and etc.) def writeable(func): - def func_wrapper(*args): + def func_wrapper(*args, **kwargs): port = args[0] if not port.is_up(): @@ -112,7 +114,7 @@ class Port(object): if not port.is_writeable(): return port.err("{0} - port is not in a writeable state".format(func.__name__)) - return func(*args) + return func(*args, **kwargs) return func_wrapper @@ -396,16 +398,17 @@ class Port(object): @writeable - def start (self, mul, duration, force): + def start (self, mul, duration, force, mask): if self.state == self.STATE_IDLE: return self.err("unable to start traffic - no streams attached to port") - params = {"handler": self.handler, - "port_id": self.port_id, - "mul": mul, - "duration": duration, - "force": force} + params = {"handler": self.handler, + "port_id": self.port_id, + "mul": mul, + "duration": duration, + "force": force, + "core_mask": mask if mask is not None else self.MASK_ALL} # must set this before to avoid race with the async response last_state = self.state @@ -707,7 +710,7 @@ class Port(object): ('L2 len', len(obj['pkt']) + 4), ('mode', obj['mode']), ('rate', obj['rate']), - ('next_stream', obj['next_id']) + ('next_stream', obj['next_id'] if not '-1' else 'None') ]) return {"streams" : OrderedDict(sorted(data.items())) } diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_sim.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_sim.py index 62724e64..3e63c4e2 100644 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_sim.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_sim.py @@ -40,6 +40,8 @@ class BpSimException(Exception): # stateless simulation class STLSim(object): + MASK_ALL = ((1 << 64) - 1) + def __init__ (self, bp_sim_path, handler = 0, port_id = 0, api_h = "dummy"): self.bp_sim_path = os.path.abspath(bp_sim_path) @@ -61,7 +63,8 @@ class STLSim(object): "force": force, "port_id": self.port_id, "mul": parsing_opts.decode_multiplier(mult), - "duration": duration} + "duration": duration, + "core_mask": self.MASK_ALL} } diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_stats.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_stats.py index 1bf0a9a4..afb01791 100644 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_stats.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_stats.py @@ -210,8 +210,11 @@ class CTRexInfoGenerator(object): ("version", "{ver}, UUID: {uuid}".format(ver=global_stats.server_version.get("version", "N/A"), uuid="N/A")), - ("cpu_util.", "{0}% {1}".format( format_threshold(round_float(global_stats.get("m_cpu_util")), [85, 100], [0, 85]), - global_stats.get_trend_gui("m_cpu_util", use_raw = True))), + ("cpu_util.", "{0}% @ {2} cores ({3} per port) {1}".format( format_threshold(round_float(global_stats.get("m_cpu_util")), [85, 100], [0, 85]), + global_stats.get_trend_gui("m_cpu_util", use_raw = True), + global_stats.system_info.get('dp_core_count'), + global_stats.system_info.get('dp_core_count_per_port'), + )), ("rx_cpu_util.", "{0}% {1}".format( format_threshold(round_float(global_stats.get("m_rx_cpu_util")), [85, 100], [0, 85]), global_stats.get_trend_gui("m_rx_cpu_util", use_raw = True))), @@ -234,7 +237,7 @@ class CTRexInfoGenerator(object): ("total_pps", "{0} {1}".format( global_stats.get("m_tx_pps", format=True, suffix="pkt/sec"), global_stats.get_trend_gui("m_tx_pps"))), - (" ", ""), + #(" ", ""), ("drop_rate", "{0}".format( format_num(global_stats.get("m_rx_drop_bps"), suffix = 'b/sec', @@ -422,21 +425,35 @@ class CTRexInfoGenerator(object): def _generate_cpu_util_stats(self): util_stats = self._util_stats_ref.get_stats(use_1sec_cache = True) + stats_table = text_tables.TRexTextTable() if util_stats: if 'cpu' not in util_stats: raise Exception("Excepting 'cpu' section in stats %s" % util_stats) cpu_stats = util_stats['cpu'] - hist_len = len(cpu_stats[0]) + hist_len = len(cpu_stats[0]["history"]) avg_len = min(5, hist_len) show_len = min(15, hist_len) stats_table.header(['Thread', 'Avg', 'Latest'] + list(range(-1, 0 - show_len, -1))) stats_table.set_cols_align(['l'] + ['r'] * (show_len + 1)) - stats_table.set_cols_width([8, 3, 6] + [3] * (show_len - 1)) + stats_table.set_cols_width([10, 3, 6] + [3] * (show_len - 1)) stats_table.set_cols_dtype(['t'] * (show_len + 2)) + for i in range(min(14, len(cpu_stats))): - avg = int(round(sum(cpu_stats[i][:avg_len]) / avg_len)) - stats_table.add_row([i, avg] + cpu_stats[i][:show_len]) + history = cpu_stats[i]["history"] + ports = cpu_stats[i]["ports"] + avg = int(round(sum(history[:avg_len]) / avg_len)) + + # decode active ports for core + if ports == [-1, -1]: + interfaces = "(IDLE)" + elif not -1 in ports: + interfaces = "({:},{:})".format(ports[0], ports[1]) + else: + interfaces = "({:})".format(ports[0] if ports[0] != -1 else ports[1]) + + thread = "{:2} {:^7}".format(i, interfaces) + stats_table.add_row([thread, avg] + history[:show_len]) else: stats_table.add_row(['No Data.']) return {'cpu_util(%)': ExportableStats(None, stats_table)} @@ -542,6 +559,7 @@ class CTRexInfoGenerator(object): per_field_stats = OrderedDict([("owner", []), ("state", []), ("speed", []), + ("CPU util.", []), ("--", []), ("Tx bps L2", []), ("Tx bps L1", []), @@ -1037,7 +1055,8 @@ class CPortStats(CTRexStats): return {"owner": owner, "state": "{0}".format(state), "speed": self._port_obj.get_formatted_speed() if self._port_obj else '', - + "CPU util.": "{0} {1}%".format(self.get_trend_gui("m_cpu_util", use_raw = True), + format_threshold(round_float(self.get("m_cpu_util")), [85, 100], [0, 85])) if self._port_obj else '' , "--": " ", "---": " ", "----": " ", @@ -1401,6 +1420,7 @@ class CUtilStats(CTRexStats): self.history.append(rc.data()) else: self.history.append({}) + return self.history[-1] if __name__ == "__main__": diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py index e9451940..fc0bc78c 100755 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/trex_stl_streams.py @@ -918,9 +918,9 @@ class STLProfile(object): imp.reload(module) # reload the update t = STLProfile.get_module_tunables(module) - for arg in kwargs: - if not arg in t: - raise STLError("Profile {0} does not support tunable '{1}' - supported tunables are: '{2}'".format(python_file, arg, t)) + #for arg in kwargs: + # if not arg in t: + # raise STLError("Profile {0} does not support tunable '{1}' - supported tunables are: '{2}'".format(python_file, arg, t)) streams = module.register().get_streams(direction = direction, port_id = port_id, diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/parsing_opts.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/parsing_opts.py index af7e90c1..9ed6c0f8 100755 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/parsing_opts.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/parsing_opts.py @@ -37,6 +37,8 @@ PROMISCUOUS_SWITCH = 21 TUNABLES = 22 REMOTE_FILE = 23 LOCKED = 24 +PIN_CORES = 25 +CORE_MASK = 26 GLOBAL_STATS = 50 PORT_STATS = 51 @@ -47,6 +49,8 @@ CPU_STATS = 55 MBUF_STATS = 56 STREAMS_MASK = 60 +CORE_MASK_GROUP = 61 + # ALL_STREAMS = 61 # STREAM_LIST_WITH_ALL = 62 @@ -79,14 +83,23 @@ def match_time_unit(val): "-d 10m : in min \n" "-d 1h : in hours") + match_multiplier_help = """Multiplier should be passed in the following format: - [number][<empty> | bps | kbps | mbps | gbps | pps | kpps | mpps | %% ]. + [number][<empty> | bps | kbps | mbps | gbps | pps | kpps | mpps | %% ]. + no suffix will provide an absoulute factor and percentage will provide a percentage of the line rate. examples - '-m 10', '-m 10kbps', '-m 10mpps', '-m 23%%' - '-m 23%%' : is 23%% L1 bandwidth - '-m 23mbps' : is 23mbps in L2 bandwidth (including FCS+4) + '-m 10', + '-m 10kbps', + '-m 10kbpsl1', + '-m 10mpps', + '-m 23%% ' + + '-m 23%%' : is 23%% L1 bandwidth + '-m 23mbps': is 23mbps in L2 bandwidth (including FCS+4) + '-m 23mbpsl1': is 23mbps in L1 bandwidth + """ @@ -183,6 +196,14 @@ def match_multiplier_strict(val): return val +def hex_int (val): + pattern = r"0x[1-9a-fA-F][0-9a-fA-F]*" + + if not re.match(pattern, val): + raise argparse.ArgumentTypeError("{0} is not a valid positive HEX formatted number".format(val)) + + return int(val, 16) + def is_valid_file(filename): if not os.path.isfile(filename): @@ -264,7 +285,7 @@ OPTIONS_DB = {MULTIPLIER: ArgumentPack(['-m', '--multiplier'], {"nargs": '+', 'dest':'ports', 'metavar': 'PORTS', - 'type': int, + 'type': int, 'help': "A list of ports on which to apply the command", 'default': []}), @@ -314,7 +335,6 @@ OPTIONS_DB = {MULTIPLIER: ArgumentPack(['-m', '--multiplier'], 'default': False, 'help': "Dry run - no traffic will be injected"}), - XTERM: ArgumentPack(['-x', '--xterm'], {'action': 'store_true', 'dest': 'xterm', @@ -364,6 +384,21 @@ OPTIONS_DB = {MULTIPLIER: ArgumentPack(['-m', '--multiplier'], 'default': []}), + PIN_CORES: ArgumentPack(['--pin'], + {'action': 'store_true', + 'dest': 'pin_cores', + 'default': False, + 'help': "Pin cores to interfaces - cores will be divided between interfaces (performance boot for symetric profiles)"}), + + CORE_MASK: ArgumentPack(['--core_mask'], + {'action': 'store', + 'nargs': '+', + 'type': hex_int, + 'dest': 'core_mask', + 'default': None, + 'help': "Core mask - only cores responding to the bit mask will be active"}), + + # promiscuous PROMISCUOUS_SWITCH: ArgumentGroup(MUTEX, [PROMISCUOUS, NO_PROMISCUOUS], @@ -383,7 +418,13 @@ OPTIONS_DB = {MULTIPLIER: ArgumentPack(['-m', '--multiplier'], STREAMS_STATS, CPU_STATS, MBUF_STATS], - {}) + {}), + + + CORE_MASK_GROUP: ArgumentGroup(MUTEX, [PIN_CORES, + CORE_MASK], + {'required': False}), + } diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/text_tables.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/text_tables.py index 4b7e9b3e..393ba111 100644 --- a/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/text_tables.py +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/text_tables.py @@ -1,7 +1,10 @@ +from __future__ import print_function + import sys from texttable import Texttable from .text_opts import format_text + class TRexTextTable(Texttable): def __init__(self): @@ -21,11 +24,11 @@ class TRexTextInfo(Texttable): def generate_trex_stats_table(): pass -def print_table_with_header(texttable_obj, header="", untouched_header=""): +def print_table_with_header(texttable_obj, header="", untouched_header="", buffer=sys.stdout): header = header.replace("_", " ").title() + untouched_header - print(format_text(header, 'cyan', 'underline') + "\n") + print(format_text(header, 'cyan', 'underline') + "\n", file=buffer) - print((texttable_obj.draw() + "\n")) + print((texttable_obj.draw() + "\n"), file=buffer) if __name__ == "__main__": pass diff --git a/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/zipmsg.py b/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/zipmsg.py new file mode 100644 index 00000000..397ada16 --- /dev/null +++ b/scripts/automation/trex_control_plane/stl/trex_stl_lib/utils/zipmsg.py @@ -0,0 +1,32 @@ +import zlib +import struct + +class ZippedMsg: + + MSG_COMPRESS_THRESHOLD = 256 + MSG_COMPRESS_HEADER_MAGIC = 0xABE85CEA + + def check_threshold (self, msg): + return len(msg) >= self.MSG_COMPRESS_THRESHOLD + + def compress (self, msg): + # compress + compressed = zlib.compress(msg) + new_msg = struct.pack(">II", self.MSG_COMPRESS_HEADER_MAGIC, len(msg)) + compressed + return new_msg + + + def decompress (self, msg): + if len(msg) < 8: + return None + + t = struct.unpack(">II", msg[:8]) + if (t[0] != self.MSG_COMPRESS_HEADER_MAGIC): + return None + + x = zlib.decompress(msg[8:]) + if len(x) != t[1]: + return None + + return x + |