rte_ring is a lock-free FIFO ring queue implemented with CAS, which supports multiple consumers/producers entering and exiting the queue at the same time, and is often used for communication between multiple threads/processes. For the specific principle, you can check the official DPDK documentation or read the source code. The hardware board used in this article is the Forlinx embedded OKMX8MP-C single-board computer, the system version is Linux5.4.70+Qt5.15.0. It mainly introduces the realization of rte_ring ring-free queue by compiling the DPDK source code. Interprocess communication.

1. Kernel compilation

Download and decompress the iMX8MP kernel source compressed package volume provided by the Forlinx manufacturer:

Combine the compressed volumes in the virtual machine and unzip to get the kernel source package folder OK8MP-linux-kernel, package the folder with tar and copy it to the TF card file system to unzip it:

Find the configuration file OK8MP-C_defconfig in the kernel source code:

This is the make option, use

make OK8MP-C_defconfig

command to configure compile options:

make -j4

Start compiling:

Note that common software needs to be installed before starting to compile:

apt install bison bc flex

Incremental compilation is complete:

2. DPDK compilation

Next, you can download DPDK and run the rte_ring lock-free ring queue Demo application, which needs to be downloaded from https://www.dpdk.org/

Download the DPDK 19.11.10 (LTS) long-term support version from the official website:

Find the setting file named rte_vars.mk in the mk/ folder in the root directory, find the environment variable RTE_KERNELDIR, and modify it to the above kernel source path:

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

Go to the usertools folder, find the dpdk-setup.sh script and run

select 8, ARM64-armv8a-linuxapp-gcc,

This option will cause the gcc cross-compilation chain of dpdk to generate an external library suitable for the armv8a processor. There are ko files and so files such as kmod and lib in the external library, which are used for third-party program development and operation:

use instructions

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

Load the igb_uio.ko driver file, which is a necessary step for dpdk development:

Then use the dpdk-devbind.py script to manually bind hugepage memory, here is the numa method:

This will mount the /mnt/huge file into a hugepage mapping file, and actually occupy the memory space:

3. rte_ring lock-free ring queue

The preparation work is completed, we can write the rte_ring lock-free ring queue Demo code next, but before writing, we need to have a basic understanding of the lock-free ring queue: https://blog.csdn.net/chen98765432101/article/ details/69367633

Whether it is the rte_ring developed by the third party of dpdk or the lock-free ring queue that already exists in the Linux kernel, the basic principle is similar. The writing action is carried out by two processes or two threads respectively. The writing process continuously writes data to the memory location with the self-incrementing address, and the reading process continuously reads the data of the memory location with the self-incrementing address. When the memory address of the writing location is When the memory in the queue reaches the highest value, the space with the lowest memory address in the queue needs to be released for the writing process to continue writing. Lock and unlock the memory address space at the end to avoid errors in the read and write process. The difference is that the lock-free ring queue, address management and read-write control in the Linux kernel are all performed by the kernel, while the rte_ring of dpdk is performed by the controller inside dpdk, because the dpdk module completely takes over the allocated memory The management right of the space is directly managed by bypassing the Linux kernel, and the kernel has no right to access the specific management details of the dpdk controller.

Write a Demo of two processes in a lock-free ring queue, first write the Primary process:

#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;
}

Note that the WERROR-related compilation options should be turned off in the Makefile:

#   BSD LICENSE
#
#   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
#   All rights reserved.
#
#   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 process:

#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;
}

Also in the Makefile, turn off the WERROR-related compilation options:

#   BSD LICENSE
#
#   Copyright(c) 2010-2014 Intel Corporation. All rights reserved.
#   All rights reserved.
#
#   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_secondary
# all source are stored in SRCS-y
SRCS-y := main.c
CFLAGS += -O3
CFLAGS += $()
include $(RTE_SDK)/mk/rte.extapp.mk

Running, let's talk about it here, based on rte_ring inter-process communication, the Secondary process is best to use the auto type:

./rte_ring_primary --proc-type primary
./rte_ring_secondary --proc-type auto

running result: