飛凌嵌入式iMX8MP 開發(fā)板試用體驗(yàn)--編譯內(nèi)核源碼&編譯DPDK源碼實(shí)現(xiàn)rte_ring無鎖環(huán)隊(duì)列進(jìn)程間通信

原創(chuàng)

donatello1996 2021-11-23 14:19:00 iMX8MP

來源：電子發(fā)燒友作者：donatello1996

FETMX8MP-C核心板基于NXP i.MX 8M Plus處理器開發(fā)設(shè)計(jì)，該系列處理器專注于機(jī)器學(xué)習(xí)與視覺、高級多媒體以及具有高可靠性的工業(yè)自動(dòng)化。旨在滿足智慧城市、工業(yè)互聯(lián)網(wǎng)、智能醫(yī)療、智慧交通等應(yīng)用的需求。 ·強(qiáng)大的四核或雙核Arm? Cortex?-A53處理器，主頻高達(dá)1.6GHz，帶有神經(jīng)處理單元(NPU)，最高運(yùn)行速率可達(dá)2.3 TOPS。

iMX8MP核心板

本文主要介紹編譯內(nèi)核源碼&編譯DPDK源碼實(shí)現(xiàn)rte_ring無鎖環(huán)隊(duì)列進(jìn)程間通信。

正文開始：

下載并解壓飛凌廠商提供的iMX8MP內(nèi)核源碼壓縮包分卷：

在虛擬機(jī)中合并壓縮分卷并解壓得出內(nèi)核源碼包文件夾OK8MP-linux-kernel，將文件夾使用tar打包并復(fù)制到TF卡文件系統(tǒng)中解壓：

找到內(nèi)核源碼中的配置文件OK8MP-C_defconfig：

這個(gè)就是make選項(xiàng)，使用

make OK8MP-C_defconfig

指令即可配置編譯選項(xiàng)：

make -j4

開始編譯：

注意開始編譯前需要安裝常用軟件：

apt install bison bc flex

增量編譯完畢：

然后接下來就可以下載DPDK并運(yùn)行rte_ring無鎖環(huán)隊(duì)列Demo應(yīng)用，需要從

https://www.dpdk.org/

官網(wǎng)中下載DPDK 19.11.10 (LTS)長期支持版本：

在根目錄下的mk/文件夾下找到名為rte_vars.mk設(shè)置文件，找到環(huán)境變量RTE_KERNELDIR，修改為上述的內(nèi)核源碼路徑：

1.RTE_KERNELDIR ?= /home/OK8MP-linux-kernel/

進(jìn)入usertools文件夾，找到dpdk-setup.sh腳本并運(yùn)行

選擇8，ARM64-armv8a-linuxapp-gcc，

這個(gè)選項(xiàng)會(huì)使dpdk的gcc交叉編譯鏈生成適用于armv8a處理器的外部庫，外部庫中有kmod和lib等ko文件和so文件，

是用于第三方程序開發(fā)和運(yùn)行的：

使用指令

insmod /home/dpdk-stable-19.11.10/arm64-armv8a-linuxapp-gcc/kmod/igb_uio.ko

加載igb_uio.ko驅(qū)動(dòng)文件，這是進(jìn)行dpdk開發(fā)必備的步驟：

然后是使用dpdk-devbind.py腳本手動(dòng)進(jìn)行hugepage大頁內(nèi)存綁定，此處為numa方式：

此舉會(huì)將/mnt/huge文件mount成hugepage映射文件，并實(shí)實(shí)在在地占用內(nèi)存空間：

準(zhǔn)備工作完成，我們接下來可以進(jìn)行rte_ring無鎖環(huán)隊(duì)列Demo代碼的編寫，但是在編寫之前，需要對無鎖環(huán)隊(duì)列有一個(gè)基本的認(rèn)識(shí)：https://blog.csdn.net/chen98765432101/article/details/69367633
無論是dpdk第三方開發(fā)的rte_ring還是Linux內(nèi)核中本就存在的無鎖環(huán)隊(duì)列，其基本原理類似，在一條分配好的隊(duì)列型內(nèi)存空間中，讀寫方式為FIFO（先進(jìn)先出），讀和寫的動(dòng)作分別有兩個(gè)進(jìn)程或兩個(gè)線程進(jìn)行，寫進(jìn)程不斷往地址自增的內(nèi)存位置寫入數(shù)據(jù)，讀進(jìn)程不斷讀取地址自增的內(nèi)存位置的數(shù)據(jù)，當(dāng)寫位置的內(nèi)存地址已為隊(duì)列中內(nèi)存的最高值時(shí)，需要釋放隊(duì)列中內(nèi)存地址最低值的空間供寫進(jìn)程繼續(xù)寫，方式仍與上一周期相同（不斷往地址自增的內(nèi)存位置寫入數(shù)據(jù)），釋放過程需要保證對末尾內(nèi)存地址空間的鎖定與解鎖，避免讀寫過程出錯(cuò)。而不同的是，Linux內(nèi)核中的無鎖環(huán)隊(duì)列，地址管理和讀寫控制均由內(nèi)核進(jìn)行，而dpdk的rte_ring則由dpdk內(nèi)部的控制器進(jìn)行，因?yàn)閐pdk這一模塊完整地接管了所分配內(nèi)存空間的管理權(quán)，是直接繞過Linux內(nèi)核進(jìn)行管理的，內(nèi)核也無權(quán)訪問dpdk控制器的具體管理細(xì)節(jié)。

編寫無鎖環(huán)隊(duì)列兩個(gè)進(jìn)程的Demo，先寫Primary進(jìn)程：

#include <stdio.h>

#include <string.h>

#include <stdint.h>

#include <stdlib.h>

#include <inttypes.h>

#include <stdarg.h>

#include <errno.h>

#include <unistd.h>

#include <termios.h>

#include <sys/queue.h>

#include <rte_mempool.h>

#include <rte_lcore.h>

#define RTE_LOGTYPE_APP RTE_LOGTYPE_USER1

static const char *_MSG_POOL = "MSG_POOL";

static const char *_SEC_2_PRI = "SEC_2_PRI";

static const char *_PRI_2_SEC = "PRI_2_SEC";

static const char *_PRI_2_THI = "PRI_2_THI";

struct rte_ring *send_ring, *recv_ring , *send_ring_third;

struct rte_mempool *message_pool;

volatile int quit = 0;

static void * lcore_recv(void *arg)

{

unsigned lcore_id = rte_lcore_id();

printf("Starting core %u\n", lcore_id);

while (!quit){

void *msg;

if (rte_ring_dequeue(recv_ring, &msg) < 0)

{

usleep(5);

continue;

}

printf("lcore_id = %d Received '%s'\n" , lcore_id , (char *)msg);

rte_mempool_put(message_pool , msg);

}

return 0;

}

int string_size = 100;

int elt_size = 128;

pthread_t id1;

int main(int argc, char **argv)

{

const unsigned flags = 0;

const unsigned ring_size = 64;

const unsigned pool_size = 1024;

const unsigned pool_cache = 32;

const unsigned priv_data_sz = 0;

int ret;

unsigned lcore_id;

ret = rte_eal_init(argc, argv);

if (ret < 0)

rte_exit(EXIT_FAILURE, "Cannot init EAL\n");

send_ring = rte_ring_create(_PRI_2_SEC, ring_size, rte_socket_id(), flags);

recv_ring = rte_ring_create(_SEC_2_PRI, ring_size, rte_socket_id(), flags);

send_ring_third = rte_ring_create(_PRI_2_THI, ring_size, rte_socket_id(), flags);

message_pool = rte_mempool_create(_MSG_POOL, pool_size,

elt_size, pool_cache, priv_data_sz,

NULL, NULL, NULL, NULL,

rte_socket_id(), flags);

if (send_ring == NULL)

rte_exit(EXIT_FAILURE, "Problem getting sending ring\n");

if (recv_ring == NULL)

rte_exit(EXIT_FAILURE, "Problem getting receiving ring\n");

if (send_ring_third == NULL)

rte_exit(EXIT_FAILURE, "Problem getting send_ring_third\n");

if (message_pool == NULL)

rte_exit(EXIT_FAILURE, "Problem getting message pool\n");

pthread_create(&id1 , NULL , lcore_recv , NULL);

while(1)

{

void *msg = NULL;

if (rte_mempool_get(message_pool, &msg) < 0)

continue;

snprintf((char *)msg, string_size, "%s", "primary to secondary");

if (rte_ring_enqueue(send_ring , msg) < 0)

{

rte_mempool_put(message_pool, msg);

}

if (rte_mempool_get(message_pool, &msg) < 0)

continue;

snprintf((char *)msg, string_size, "%s", "primary to third");

if (rte_ring_enqueue(send_ring_third , msg) < 0)

{

rte_mempool_put(message_pool, msg);

}

sleep(1);

}

return 0;

}

注意在Makefile文件里面要關(guān)閉WERROR相關(guān)編譯選項(xiàng)：

# BSD LICENSE

# Redistribution and use in source and binary forms, with or without

# modification, are permitted provided that the following conditions

# are met:

# * Redistributions of source code must retain the above copyright

# notice, this list of conditions and the following disclaimer.

# * Redistributions in binary form must reproduce the above copyright

# notice, this list of conditions and the following disclaimer in

# the documentation and/or other materials provided with the

# distribution.

# * Neither the name of Intel Corporation nor the names of its

# contributors may be used to endorse or promote products derived

# from this software without specific prior written permission.

# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS

# "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT

# LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR

# A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT

# OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,

# SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT

# LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,

# DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY

# THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT

# (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

# OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

ifeq ($(RTE_SDK),)

$(error "Please define RTE_SDK environment variable")

endif

# Default target, can be overridden by command line or environment

RTE_TARGET ?= arm64-armv8a-linuxapp-gcc

include $(RTE_SDK)/mk/rte.vars.mk

# binary name

APP = rte_ring_primary

# all source are stored in SRCS-y

SRCS-y := main.c

CFLAGS += -O0

CFLAGS +=

include $(RTE_SDK)/mk/rte.extapp.mk

Secondary進(jìn)程：

#include <stdio.h>

#include <string.h>

#include <stdint.h>

#include <stdlib.h>

#include <inttypes.h>

#include <stdarg.h>

#include <errno.h>

#include <unistd.h>

#include <termios.h>

#include <sys/queue.h>

#include <rte_mempool.h>

#include <rte_lcore.h>

#define RTE_LOGTYPE_APP RTE_LOGTYPE_USER1

static const char *_MSG_POOL = "MSG_POOL";

static const char *_SEC_2_PRI = "SEC_2_PRI";

static const char *_PRI_2_SEC = "PRI_2_SEC";

struct rte_ring *send_ring, *recv_ring;

struct rte_mempool *message_pool;

volatile int quit = 0;

int string_size = 100;

static int lcore_send(__attribute__((unused)) void *arg)

{

unsigned lcore_id = rte_lcore_id();

while(1)

{

void *msg = NULL;

if (rte_mempool_get(message_pool, &msg) < 0)

continue;

snprintf((char *)msg , string_size , "%s", "secondary to primary");

if (rte_ring_enqueue(send_ring , msg) < 0)

{

rte_mempool_put(message_pool, msg);

}

sleep(1);

}

return 0;

}

pthread_t id1;

int main(int argc, char **argv)

{

const unsigned flags = 0;

const unsigned ring_size = 64;

const unsigned pool_size = 1024;

const unsigned pool_cache = 32;

const unsigned priv_data_sz = 0;

int ret;

unsigned lcore_id;

ret = rte_eal_init(argc, argv);

if (ret < 0)

rte_exit(EXIT_FAILURE, "Cannot init EAL\n");

recv_ring = rte_ring_lookup(_PRI_2_SEC);

send_ring = rte_ring_lookup(_SEC_2_PRI);

message_pool = rte_mempool_lookup(_MSG_POOL);

if (send_ring == NULL)

rte_exit(EXIT_FAILURE, "Problem getting sending ring\n");

if (recv_ring == NULL)

rte_exit(EXIT_FAILURE, "Problem getting receiving ring\n");

if (message_pool == NULL)

rte_exit(EXIT_FAILURE, "Problem getting message pool\n");

pthread_create(&id1 , NULL , lcore_send , NULL);

while (1)

{

lcore_id = rte_lcore_id();

void * msg = NULL;

if (rte_ring_dequeue(recv_ring, &msg) < 0)

{

usleep(5);

continue;

}

printf("lcore_id = %d Received: %s\n" , lcore_id , (char *)msg);

rte_mempool_put(message_pool, msg);

}

return 0;

}

同樣在Makefile文件里面要關(guān)閉WERROR相關(guān)編譯選項(xiàng)：