Trex threads - pin DPDK master thread to the master core

also, some names to the threads to make things clear
and a script to show them

Signed-off-by: imarom <imarom@cisco.com>

8 files changed, 141 insertions, 29 deletions

diff --git a/scripts/trex_show_threads.py b/scripts/trex_show_threads.py
new file mode 100755
index 00000000..fabe6d68
--- /dev/null
+++ b/scripts/trex_show_threads.py
@@ -0,0 +1,80 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import os
+
+def read_task_stats (task_path):
+
+    # files
+    status = task_path + "/status"
+    stat  = task_path + "/stat"
+
+    stats_dict = {}
+    for line in open(status, 'r').readlines():
+        name, value = line.split(':', 1)
+        stats_dict[name.strip().lower()] = value.strip()
+
+    stat_data = open(stat, 'r').readline().split()
+
+    stats_dict['last_sched_cpu'] = stat_data[-14]
+
+    return stats_dict
+
+
+def show_threads (pid):
+    process_dir = "/proc/{0}/task".format(pid)
+    task_paths = ["{0}/{1}".format(process_dir, task) for task in os.listdir(process_dir)]
+
+    
+    header = [ 'Task Name', 'PID', 'Allowed CPU', 'Last Sched CPU', 'Asked Ctx Switch', 'Forced Ctx Switch']
+    for x in header:
+        print('{:^20}'.format(x)),
+    print("")
+
+    tasks = []
+    for task_path in task_paths:
+        task = read_task_stats(task_path)
+        tasks.append(task)
+
+    tasks = sorted(tasks, key = lambda x: int(x['cpus_allowed_list']))
+    for task in tasks:
+        # name
+        print("{:<20}".format(task['name'])),
+        print("{:^20}".format(task['pid'])),
+        print("{:^20}".format(task['cpus_allowed_list'])),
+        print("{:^20}".format(task['last_sched_cpu'])),
+        print("{:^20}".format(task['voluntary_ctxt_switches'])),
+        print("{:^20}".format(task['nonvoluntary_ctxt_switches'])),
+        print("")
+
+def isnum (x):
+    try:
+        int(x)
+        return True
+    except ValueError:
+        return False
+
+
+def find_trex_pid ():
+    procs = [x for x in os.listdir('/proc/') if isnum(x)]
+    for proc in procs:
+        cmd = open('/proc/{0}/{1}'.format(proc, 'cmdline')).readline()
+        if '_t-rex' in cmd:
+            return proc
+
+    return None
+
+def main ():
+    trex_pid = find_trex_pid()
+    if trex_pid is None:
+        print("Unable to find Trex PID")
+        exit(1)
+
+    print("\nTrex PID on {0}\n".format(trex_pid))
+    show_threads(trex_pid)
+
+
+
+if __name__ == '__main__':
+     main()
+
diff --git a/src/bp_sim.cpp b/src/bp_sim.cpp
index 24959b7b..62e8d822 100755
--- a/src/bp_sim.cpp
+++ b/src/bp_sim.cpp
@@ -177,8 +177,8 @@ physical_thread_id_t CPlatformSocketInfoNoConfig::thread_virt_to_phy(virtual_thr
     return (virt_id);
 }
 
-bool CPlatformSocketInfoNoConfig::thread_phy_is_master(physical_thread_id_t  phy_id){
-    return (phy_id==0);
+physical_thread_id_t CPlatformSocketInfoNoConfig::get_master_phy_id() {
+    return (0);
 }
 
 bool CPlatformSocketInfoNoConfig::thread_phy_is_rx(physical_thread_id_t  phy_id){
@@ -402,8 +402,8 @@ physical_thread_id_t CPlatformSocketInfoConfig::thread_virt_to_phy(virtual_threa
     return ( m_thread_virt_to_phy[virt_id]);
 }
 
-bool CPlatformSocketInfoConfig::thread_phy_is_master(physical_thread_id_t  phy_id){
-    return (m_platform->m_master_thread==phy_id?true:false);
+physical_thread_id_t CPlatformSocketInfoConfig::get_master_phy_id() {
+    return m_platform->m_master_thread;
 }
 
 bool CPlatformSocketInfoConfig::thread_phy_is_rx(physical_thread_id_t  phy_id){
@@ -481,6 +481,10 @@ bool CPlatformSocketInfo::thread_phy_is_master(physical_thread_id_t  phy_id){
     return ( m_obj->thread_phy_is_master(phy_id));
 }
 
+physical_thread_id_t CPlatformSocketInfo::get_master_phy_id() {
+    return ( m_obj->get_master_phy_id());
+}
+
 bool CPlatformSocketInfo::thread_phy_is_rx(physical_thread_id_t  phy_id) {
     return ( m_obj->thread_phy_is_rx(phy_id));
 }
diff --git a/src/bp_sim.h b/src/bp_sim.h
index f7a1e73c..cd0f6a13 100755
--- a/src/bp_sim.h
+++ b/src/bp_sim.h
@@ -957,10 +957,16 @@ public:
     /* return  the map betwean virtual to phy id */
     virtual physical_thread_id_t thread_virt_to_phy(virtual_thread_id_t virt_id)=0;
 
-    virtual bool thread_phy_is_master(physical_thread_id_t  phy_id)=0;
+
+    virtual physical_thread_id_t get_master_phy_id() = 0;
     virtual bool thread_phy_is_rx(physical_thread_id_t  phy_id)=0;
 
     virtual void dump(FILE *fd)=0;
+
+    bool thread_phy_is_master(physical_thread_id_t  phy_id) {
+        return (get_master_phy_id() == phy_id);
+    }
+
 };
 
 class CPlatformSocketInfoNoConfig : public CPlatformSocketInfoBase {
@@ -999,7 +1005,7 @@ public:
     /* return  the map betwean virtual to phy id */
     physical_thread_id_t thread_virt_to_phy(virtual_thread_id_t virt_id);
 
-    bool thread_phy_is_master(physical_thread_id_t  phy_id);
+    physical_thread_id_t get_master_phy_id();
     bool thread_phy_is_rx(physical_thread_id_t  phy_id);
 
     virtual void dump(FILE *fd);
@@ -1045,7 +1051,7 @@ public:
     /* return  the map betwean virtual to phy id */
     physical_thread_id_t thread_virt_to_phy(virtual_thread_id_t virt_id);
 
-    bool thread_phy_is_master(physical_thread_id_t  phy_id);
+    physical_thread_id_t get_master_phy_id();
     bool thread_phy_is_rx(physical_thread_id_t  phy_id);
 
 public:
@@ -1110,6 +1116,7 @@ public:
     physical_thread_id_t thread_virt_to_phy(virtual_thread_id_t virt_id);
 
     bool thread_phy_is_master(physical_thread_id_t  phy_id);
+    physical_thread_id_t get_master_phy_id();
     bool thread_phy_is_rx(physical_thread_id_t  phy_id);
 
     void dump(FILE *fd);
diff --git a/src/main_dpdk.cpp b/src/main_dpdk.cpp
index de0f2327..9f2b62b2 100644
--- a/src/main_dpdk.cpp
+++ b/src/main_dpdk.cpp
@@ -128,7 +128,7 @@ static char * global_dpdk_args[MAX_DPDK_ARGS];
 static char global_cores_str[100];
 static char global_prefix_str[100];
 static char global_loglevel_str[20];
-
+static char global_master_id_str[10];
 
 class CTRexExtendedDriverBase {
 public:
@@ -4391,6 +4391,8 @@ void CGlobalTRex::shutdown() {
 
 int CGlobalTRex::run_in_master() {
 
+    //rte_thread_setname(pthread_self(), "TRex Control");
+
     if ( get_is_stateless() ) {
         m_trex_stateless->launch_control_plane();
     }
@@ -4441,6 +4443,8 @@ int CGlobalTRex::run_in_master() {
 
 int CGlobalTRex::run_in_rx_core(void){
 
+    rte_thread_setname(pthread_self(), "TRex RX");
+
     if (get_is_stateless()) {
         m_sl_rx_running = true;
         m_rx_sl.start();
@@ -4457,7 +4461,9 @@ int CGlobalTRex::run_in_rx_core(void){
 
 int CGlobalTRex::run_in_core(virtual_thread_id_t virt_core_id){
     std::stringstream ss;
-    ss << "DP core " << int(virt_core_id);
+
+    ss << "Trex DP core " << int(virt_core_id);
+    rte_thread_setname(pthread_self(), ss.str().c_str());
 
     CPreviewMode *lp=&CGlobalInfo::m_options.preview;
     if ( lp->getSingleCore() &&
@@ -5069,25 +5075,28 @@ int  update_dpdk_args(void){
     }
 
     /* set the DPDK options */
-    global_dpdk_args_num =7;
+    global_dpdk_args_num = 0;
 
-    global_dpdk_args[0]=(char *)"xx";
-    global_dpdk_args[1]=(char *)"-c";
-    global_dpdk_args[2]=(char *)global_cores_str;
-    global_dpdk_args[3]=(char *)"-n";
-    global_dpdk_args[4]=(char *)"4";
+    global_dpdk_args[global_dpdk_args_num++]=(char *)"xx";
+    global_dpdk_args[global_dpdk_args_num++]=(char *)"-c";
+    global_dpdk_args[global_dpdk_args_num++]=(char *)global_cores_str;
+    global_dpdk_args[global_dpdk_args_num++]=(char *)"-n";
+    global_dpdk_args[global_dpdk_args_num++]=(char *)"4";
 
     if ( CGlobalInfo::m_options.preview.getVMode() == 0  ) {
-        global_dpdk_args[5]=(char *)"--log-level";
+        global_dpdk_args[global_dpdk_args_num++]=(char *)"--log-level";
         snprintf(global_loglevel_str, sizeof(global_loglevel_str), "%d", 4);
-        global_dpdk_args[6]=(char *)global_loglevel_str;
+        global_dpdk_args[global_dpdk_args_num++]=(char *)global_loglevel_str;
     }else{
-        global_dpdk_args[5]=(char *)"--log-level";
+        global_dpdk_args[global_dpdk_args_num++]=(char *)"--log-level";
         snprintf(global_loglevel_str, sizeof(global_loglevel_str), "%d", CGlobalInfo::m_options.preview.getVMode()+1);
-        global_dpdk_args[6]=(char *)global_loglevel_str;
+        global_dpdk_args[global_dpdk_args_num++]=(char *)global_loglevel_str;
     }
 
-    global_dpdk_args_num = 7;
+    global_dpdk_args[global_dpdk_args_num++] = (char *)"--master-lcore";
+
+    snprintf(global_master_id_str, sizeof(global_master_id_str), "%u", lpsock->get_master_phy_id());
+    global_dpdk_args[global_dpdk_args_num++] = global_master_id_str;
 
     /* add white list */
     if (lpop->m_run_mode == CParserOption::RUN_MODE_DUMP_INFO and lpop->dump_interfaces.size()) {
@@ -5178,6 +5187,7 @@ void dump_interfaces_info() {
 
 int main_test(int argc , char * argv[]){
 
+
     utl_termio_init();
 
     int ret;
@@ -5233,6 +5243,12 @@ int main_test(int argc , char * argv[]){
         return (-1);
     }
 
+    /* set affinity to the master core as default */
+    cpu_set_t mask;
+    CPU_ZERO(&mask);
+    CPU_SET(CGlobalInfo::m_socket.get_master_phy_id(), &mask);
+    pthread_setaffinity_np(pthread_self(), sizeof(mask), &mask);
+
     ret = rte_eal_init(global_dpdk_args_num, (char **)global_dpdk_args);
     if (ret < 0){
         printf(" You might need to run ./trex-cfg  once  \n");
diff --git a/src/publisher/trex_publisher.cpp b/src/publisher/trex_publisher.cpp
index 67670593..224604e9 100644
--- a/src/publisher/trex_publisher.cpp
+++ b/src/publisher/trex_publisher.cpp
@@ -33,6 +33,8 @@ limitations under the License.
 bool 
 TrexPublisher::Create(uint16_t port, bool disable){
 
+    char thread_name[256];
+
     if (disable) {
         return (true);
     }
@@ -42,7 +44,15 @@ TrexPublisher::Create(uint16_t port, bool disable){
         show_zmq_last_error("can't connect to ZMQ library");
     }
 
+    /* change the pthread name temporarly for the socket creation */
+    pthread_getname_np(pthread_self(), thread_name, sizeof(thread_name));
+    pthread_setname_np(pthread_self(), "Trex Publisher");
+
     m_publisher = zmq_socket (m_context, ZMQ_PUB);
+
+    /* restore it */
+    pthread_setname_np(pthread_self(), thread_name);
+
     if ( m_context == 0 ) {
         show_zmq_last_error("can't create ZMQ socket");
     }
diff --git a/src/rpc-server/trex_rpc_req_resp_server.cpp b/src/rpc-server/trex_rpc_req_resp_server.cpp
index e0e7635c..28bf1d80 100644
--- a/src/rpc-server/trex_rpc_req_resp_server.cpp
+++ b/src/rpc-server/trex_rpc_req_resp_server.cpp
@@ -56,6 +56,8 @@ void TrexRpcServerReqRes::_rpc_thread_cb() {
     std::stringstream ss;
     int zmq_rc;
 
+    pthread_setname_np(pthread_self(), "Trex ZMQ sync");
+
     m_monitor.create(m_name, 1);
     TrexWatchDog::getInstance().register_monitor(&m_monitor);
 
diff --git a/src/stateless/cp/trex_stateless.cpp b/src/stateless/cp/trex_stateless.cpp
index 6029cbd5..0a7f8533 100644
--- a/src/stateless/cp/trex_stateless.cpp
+++ b/src/stateless/cp/trex_stateless.cpp
@@ -112,15 +112,6 @@ void TrexStateless::shutdown() {
 void
 TrexStateless::launch_control_plane() {
 
-    /* pin this process to the current running CPU
-       any new thread will be called on the same CPU
-       (control plane restriction)
-     */
-    cpu_set_t mask;
-    CPU_ZERO(&mask);
-    CPU_SET(sched_getcpu(), &mask);
-    sched_setaffinity(0, sizeof(mask), &mask);
-
     /* start RPC server */
     m_rpc_server->start();
 }
diff --git a/src/trex_watchdog.cpp b/src/trex_watchdog.cpp
index 88711b7a..9b6f5865 100644
--- a/src/trex_watchdog.cpp
+++ b/src/trex_watchdog.cpp
@@ -218,6 +218,8 @@ void TrexWatchDog::stop() {
  */
 void TrexWatchDog::_main() {
 
+    pthread_setname_np(pthread_self(), "Trex Watchdog");
+
     assert(m_enable == true);
 
     /* start main loop */