DPDK并行計算

參考文獻：

《深入淺出DPDK》

https://www.cnblogs.com/LubinLew/p/cpu_affinity.html

......................................................................

前言：

　　處理器提高性能主要是通過兩個途徑，一個是提高IPC（CPU每一時鐘周期內所執行的指令多少），另一個是提高處理器的主頻率。每一代微架構的調整都伴隨著對IPC的提高，從而提高處理器的性能，只是提升幅度有限。但是提高處理器主頻率對于性能的提升作用史明顯而且直接的。但是一味的提高主頻很快會觸及頻率墻，因為功耗正比與主頻的三次方

　　所以最終我們還是回到了提升IPC的方式上做突破，后來發現通過提高指令的并行度來提高IPC來提高IPC，而提高并行度有兩個方法，一種是提高微架構的指令并行度，另一種是采用多核并發，下面我們就了解DPDK是如何利用這兩種方式提高性能的

一. 多核性能和可擴展性

　　多核處理器是指一個處理器中集中兩個或者多個完整的內核（及計算引擎），如果把處理器性能伴隨著頻率的提升看作是垂直擴展，那么多核處理器的出現使得性能水平擴展成為可能。原本在單核上執行的任務按照邏輯劃分為若干個子任務，分別在不同的核上并行執行，在任務顆粒度上使得指令執行的并行度得到提升

　　那么隨著核數的增加，性能是否會持續提升呢？？？？Amdahl定律說：假如一個任務的工作量不變，多核并行計算理論時的延時加速上取決于那些不能并行處理部分的比例，也就是說不能完全依賴核數的數量讓性能一直線性提高

　　對于DPDK的主要領域--數據包處理，多核場景并不是完成一個固定的工作量任務，更關注單位時間內的吞吐量。Gustafson定律對于固定時間下的推導給我們更多的指導意義，多核并行計算的吞吐率隨著核數的增加而線性擴展，可并行處理器部分占整個任務比重越高，則增長的斜率越大。DPDK或許就是利用的這一點來提高性能的

二. 親和性

　　CPU親核性就是指在Linux系統中能夠將一個或多個進程綁定到一個或多個處理器上運行.

　　一個進程的CPU親合力掩碼決定了該進程將在哪個或哪幾個CPU上運行.在一個多處理器系統中,設置CPU親合力的掩碼可能會獲得更好的性能

　　在linux內核中，所有的線程都有一個相關的數據結構，稱為task_struct。linux內核API提供了一些方法讓用戶可以修改位掩碼或者查看當前的位掩碼

sched_set_affinity()：用來修改位掩碼
sched_get_affinity()：用來查看當前的位掩碼

　　注意：cpu_affinity會被傳遞給子線程，因此應該適當調用sched_set_affinity

　　為什要介紹親核性呢？為什么DPDK使用親核性呢？

　　將線程與cpu綁定，最直觀的好處是提高了CPU Cache 的命中率，從而減少內存訪問損耗，提高程序速度

　? 我們簡單用個例子來看一下affinity 如何使用的

這個例子來源于Linux的man page.

 1 #define _GNU_SOURCE
 2 #include <pthread.h> //不用再包含<sched.h>
 3 #include <stdio.h>
 4 #include <stdlib.h>
 5 #include <errno.h>
 6 
 7 #define handle_error_en(en, msg) \
 8         do { errno = en; perror(msg); exit(EXIT_FAILURE); } while (0)
 9 
10 int
11 main(int argc, char *argv[])
12 {
13     int s, j;
14     cpu_set_t cpuset;
15     pthread_t thread;
16     
17     thread = pthread_self();
18     
19     /* Set affinity mask to include CPUs 0 to 7 */
20     CPU_ZERO(&cpuset);
21     for (j = 0; j < 8; j++)
22         CPU_SET(j, &cpuset);
23     
24     s = pthread_setaffinity_np(thread, sizeof(cpu_set_t), &cpuset);
25     if (s != 0)
26     {
27         handle_error_en(s, "pthread_setaffinity_np");
28     }
29     
30     /* Check the actual affinity mask assigned to the thread */
31     s = pthread_getaffinity_np(thread, sizeof(cpu_set_t), &cpuset);
32     if (s != 0)
33     {
34         handle_error_en(s, "pthread_getaffinity_np");
35     }
36     
37     printf("Set returned by pthread_getaffinity_np() contained:\n");
38     for (j = 0; j < CPU_SETSIZE; j++) //CPU_SETSIZE 是定義在<sched.h>中的宏，通常是1024
39     {
40         if (CPU_ISSET(j, &cpuset))
41         {
42             printf("    CPU %d\n", j);
43         }
44     }
45     exit(EXIT_SUCCESS);
46 }

　　除了affinity,?? linux? 還提供了一個命令可以綁定：taskset

　　man taskset出現
　　CPU affinity is a scheduler property that "bonds" a process to a given set of CPUs on the system. The Linux scheduler will honor the given CPU affinity and the process will not run on any other CPUs. Note that the Linux scheduler also supports natural CPU affinity:
　　翻譯：
　　　　taskset設定cpu親和力，cpu親和力是指
　　　　CPU調度程序屬性關聯性是“鎖定”一個進程，使他只能在一個或幾個cpu線程上運行。　　對于一個給定的系統上設置的cpu。給定CPU親和力和進程不會運行在任何其他CPU。注意,Linux調度器還支持自然CPU關聯:（不能讓這個cpu只為這一個進程服務）
這里要注意的是我們可以把某個程序限定在某一些CPU上運行，但這并不意味著該程序可以獨占這些CPU，其實其他程序還是可以利用這些CPU運行。如果要精確控制CPU，taskset就略嫌不足，cpuset才是可以

選項以及使用：
-a, --all-tasks 操作所有的任務線程-p, --pid 操作已存在的pid-c, --cpu-list 通過列表顯示方式設置CPU
（1）指定1和2號cpu運行25718線程的程序
taskset -cp 1,2 25718
（2），讓某程序運行在指定的cpu上 taskset -c 1,2,4-7 tar jcf test.tar.gz test
（3）指定在1號CPU上后臺執行指定的perl程序
taskset –c 1 nohup perl pi.pl & ?

?三.? DPDK 的多線程

　　DPDK的多線程是基于pthread接口創建的，屬于搶占式線程模型，受內核支配。DPDK通過在多核設備上創建多個線程，每個線程綁定到單獨的核上，減少線程調度的開銷，來提高性能

　　DPDK可以作為控制線程也可以作為數據線程，控制線程一般綁定到主核上，受用戶配置，傳遞配置參數給數據線程，數據線程分布在不同核上處理數據包

　　1）EAL中的lcore

　　DPDK的lcore指的是EAL線程，本質是基于pthread 封裝實現。Lcore由remote_launch函數指定任務創建并管理，每個EAL pthread 中，有一個TLS稱為_lcore_id。當DPDK的EAL 'c' 參數指定coremask的時候，EAL pthread 生成相應個數的lcore并默認是1：1 親和到coremask 對應的cpu邏輯核，_lcore_id 和 CPU ID是一致的

　　在這里我們簡單介紹一下lcore的初始化：

　　1） rte_eal_cpu_init() 函數中，讀取 /sys/devices/system/cpu/? 下的信息，確定當前每個核屬于那個CPU Socket

　　2）eal_parse_args()函數，解析-c 參數，確定那些CPU核是可以使用的

　　3）給每個SLAVE核創建線程，調用eal_thread_set_affinity() 綁定CPU。

　　注冊：

　　不同模塊需要調用rte_dal_mp_remote_launch()，將自己的回調函數注冊到lcore_config[].f中，以了l2fwd為例，注冊回調處理函數是：

　　　l2fwd_launch_on_lcore()

四. lcore親和性

　　默認情況下，lcore和邏輯核是一一綁定的，帶來性能提升的同時也犧牲了一定的靈活性

　　下圖是多線程的場景圖：

下面解析一下代碼如何處理運作的：

?rte_eal_cpu_init函數主要設置每個線程lcore_config相關信息

 1 /*
 2  * Parse /sys/devices/system/cpu to get the number of physical and logical
 3  * processors on the machine. The function will fill the cpu_info
 4  * structure.
 5  */
 6 int
 7 rte_eal_cpu_init(void)
 8 {
 9     /* pointer to global configuration */
10     struct rte_config *config = rte_eal_get_configuration(); //獲取全局變量rte_config結構體的指針；
11     unsigned lcore_id;  //id號
12     unsigned count = 0; //使用的lcore的數量
13     unsigned int socket_id, prev_socket_id;
14     int lcore_to_socket_id[RTE_MAX_LCORE];
15 
16     /*
17      * Parse the maximum set of logical cores, detect the subset of running
18      * ones and enable them by default.
19      */
20     for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
21         lcore_config[lcore_id].core_index = count;
22 
23         /* init cpuset for per lcore config */
24         CPU_ZERO(&lcore_config[lcore_id].cpuset);
25 
26         /* find socket first */
27         socket_id = eal_cpu_socket_id(lcore_id);
28         if (socket_id >= RTE_MAX_NUMA_NODES) {
29 #ifdef RTE_EAL_ALLOW_INV_SOCKET_ID
30             socket_id = 0;
31 #else
32             RTE_LOG(ERR, EAL, "Socket ID (%u) is greater than RTE_MAX_NUMA_NODES (%d)\n",
33                     socket_id, RTE_MAX_NUMA_NODES);
34             return -1;
35 #endif
36         }
37         lcore_to_socket_id[lcore_id] = socket_id;
38 
39         /* in 1:1 mapping, record related cpu detected state */
40         lcore_config[lcore_id].detected = eal_cpu_detected(lcore_id);
41         if (lcore_config[lcore_id].detected == 0) {
42             config->lcore_role[lcore_id] = ROLE_OFF;
43             lcore_config[lcore_id].core_index = -1;
44             continue;
45         }
46 
47         /* By default, lcore 1:1 map to cpu id */
48         CPU_SET(lcore_id, &lcore_config[lcore_id].cpuset);
49 
50         /* By default, each detected core is enabled */
51         config->lcore_role[lcore_id] = ROLE_RTE;
52         lcore_config[lcore_id].core_role = ROLE_RTE;
53         lcore_config[lcore_id].core_id = eal_cpu_core_id(lcore_id);
54         lcore_config[lcore_id].socket_id = socket_id;
55         RTE_LOG(DEBUG, EAL, "Detected lcore %u as "
56                 "core %u on socket %u\n",
57                 lcore_id, lcore_config[lcore_id].core_id,
58                 lcore_config[lcore_id].socket_id);
59         count++;
60     }
61     /* Set the count of enabled logical cores of the EAL configuration */
62     config->lcore_count = count; //有效的lcore數
63     RTE_LOG(DEBUG, EAL,
64         "Support maximum %u logical core(s) by configuration.\n",
65         RTE_MAX_LCORE);
66     RTE_LOG(INFO, EAL, "Detected %u lcore(s)\n", config->lcore_count);
67 
68     /* sort all socket id's in ascending order */
69     qsort(lcore_to_socket_id, RTE_DIM(lcore_to_socket_id),
70             sizeof(lcore_to_socket_id[0]), socket_id_cmp);
71 
72     prev_socket_id = -1;
73     config->numa_node_count = 0;
74     for (lcore_id = 0; lcore_id < RTE_MAX_LCORE; lcore_id++) {
75         socket_id = lcore_to_socket_id[lcore_id];
76         if (socket_id != prev_socket_id)
77             config->numa_nodes[config->numa_node_count++] =
78                     socket_id;
79         prev_socket_id = socket_id;
80     }
81     RTE_LOG(INFO, EAL, "Detected %u NUMA nodes\n", config->numa_node_count);
82 
83     return 0;
84 }

下面是設置親核性：

 1 /* set affinity for current thread */
 2 static int
 3 eal_thread_set_affinity(void)
 4 {
 5     unsigned lcore_id = rte_lcore_id();
 6 
 7     /* acquire system unique id  */
 8     rte_gettid();
 9 
10     /* update EAL thread core affinity */
11     return rte_thread_set_affinity(&lcore_config[lcore_id].cpuset);
12 }

綁定主線程親和性：

1 void eal_thread_init_master(unsigned lcore_id)
2 {
3     /* set the lcore ID in per-lcore memory area */
4     RTE_PER_LCORE(_lcore_id) = lcore_id;
5 
6     /* set CPU affinity */
7     if (eal_thread_set_affinity() < 0)
8         rte_panic("cannot set affinity\n");
9 }

slave lcore的主循環函數

 1 /* main loop of threads */
 2 __attribute__((noreturn)) void *
 3 eal_thread_loop(__attribute__((unused)) void *arg)
 4 {
 5     char c;
 6     int n, ret;
 7     unsigned lcore_id;
 8     pthread_t thread_id;
 9     int m2s, s2m;
10     char cpuset[RTE_CPU_AFFINITY_STR_LEN];
11 
12     thread_id = pthread_self();
13     // 根據tid找到對應的lcore_id
14     /* retrieve our lcore_id from the configuration structure */
15     RTE_LCORE_FOREACH_SLAVE(lcore_id) {
16         if (thread_id == lcore_config[lcore_id].thread_id)
17             break;
18     }
19     if (lcore_id == RTE_MAX_LCORE)
20         rte_panic("cannot retrieve lcore id\n");
21 
22     m2s = lcore_config[lcore_id].pipe_master2slave[0];
23     s2m = lcore_config[lcore_id].pipe_slave2master[1];
24 
25     /* set the lcore ID in per-lcore memory area */
26     RTE_PER_LCORE(_lcore_id) = lcore_id;
27     //綁定SLAVE lcore到logical CPU
28     /* set CPU affinity */
29     if (eal_thread_set_affinity() < 0)
30         rte_panic("cannot set affinity\n");
31 
32     ret = eal_thread_dump_affinity(cpuset, sizeof(cpuset));
33 
34     RTE_LOG(DEBUG, EAL, "lcore %u is ready (tid=%p;cpuset=[%s%s])\n",
35         lcore_id, thread_id, cpuset, ret == 0 ? "" : "...");
36 
37     /* read on our pipe to get commands */
38     while (1) {
39         void *fct_arg;
40         //等待MASTER lcore的消息
41         /* wait command */
42         do {
43             n = read(m2s, &c, 1);
44         } while (n < 0 && errno == EINTR);
45 
46         if (n <= 0)
47             rte_panic("cannot read on configuration pipe\n");
48 
49         lcore_config[lcore_id].state = RUNNING;
50         //發送確認給MASTER lcore
51         /* send ack */
52         n = 0;
53         while (n == 0 || (n < 0 && errno == EINTR))
54             n = write(s2m, &c, 1);
55         if (n < 0)
56             rte_panic("cannot write on configuration pipe\n");
57 
58         if (lcore_config[lcore_id].f == NULL)
59             rte_panic("NULL function pointer\n");
60         //執行MASTER lcore通過rte_eal_remote_launch()注冊的回調函數 大部分DPDK應用的回調函數都是一個死循環，SLAVE lcore會阻塞在這里
61         /* call the function and store the return value */
62         fct_arg = lcore_config[lcore_id].arg;
63         ret = lcore_config[lcore_id].f(fct_arg);
64         lcore_config[lcore_id].ret = ret;
65         rte_wmb();
66         lcore_config[lcore_id].state = FINISHED;  //設置SLAVE lcore的狀態為FINISHED
67     }
68 
69     /* never reached */
70     /* pthread_exit(NULL); */
71     /* return NULL; */
72 }