We have a use-case that has one RX thread and two TX threads operating on same port. The RX thread uses the rte_pktmbuf_free to free the mbuf after processing received packets. The TX threads allocate mbuf using rte_pktmbuf_alloc call and transmit on the port.
When using same pool for RX descriptor init, and for mbuf allocation in TX threads, we see that sometimes there are unexpected mbuf leaks and allocation failures. If we use separate pools for RX and each of the TX threads, then we do not see these issues
We have not used any flags in the mempool_create call. The DPDK documentation seems to indicate mempool operations are thread safe?
We are using DPDK version: 19.11
Are we supposed to create unique mempool for each of the RX and the two TX threads?
EDIT1: Code snippet added as requested by Vipin
DPDK Primary Process
mbuf_pool = rte_pktmbuf_pool_create("fast_pkt", 8192,
256, 0, RTE_MBUF_DEFAULT_BUF_SIZE, rte_socket_id());
/* Configure the Ethernet device. */
retval = rte_eth_dev_configure(port, rx_rings, tx_rings, &port_conf);
/* Allocate and set up 1 RX queue per Ethernet port. */
retval = rte_eth_rx_queue_setup(port, q, 128,
rte_eth_dev_socket_id(port), NULL, mbuf_pool);
/* Allocate and set up 2 TX queue */
for (q = 0; q < tx_rings; q++) {
retval = rte_eth_tx_queue_setup(port, q, 512,
rte_eth_dev_socket_id(port), NULL);
}
/* Start the Ethernet port. */
retval = rte_eth_dev_start(port);
Secondary process
Main thread
mbuf_pool = rte_mempool_lookup("fast_pkt");
/* Create rx thread attach it to vCPU #4 */
rte_eal_remote_launch( main_rx, NULL, 4 );
/* create two tx threads and assign affinity to different vCPUs*/
pthread_create(&thread_id, NULL, tx_1, NULL );
CPU_ZERO(&cpuset);
CPU_SET(5, &cpuset);
pthread_setaffinity_np(thread_id, sizeof(cpuset), &cpuset);
pthread_create(&thread_id, NULL, tx_2, NULL );
CPU_ZERO(&cpuset);
CPU_SET(6, &cpuset);
pthread_setaffinity_np(thread_id, sizeof(cpuset), &cpuset);
Receive thread (main_rx)
for (;; )
{
nb_rx = rte_eth_rx_burst(port , 0, pkts_burst, 32);
for ( i = 0; i < nb_rx ; i++ )
{
process_packet( pkts_burst[ i] );
rte_pktmbuf_free(pkts_burst[ i]);
}
}
Transmit threads (tx_1, tx_2)
tx_q = <thread_sepcific queue ID>
for ( ;; )
{
/* Receive Internal message for TX*/
get_packet ( work_buffer [Burst]);
for ( iterate through each index in work_buffer )
{
tx_mbuf[i] = rte_pktmbuf_alloc(mbuf_pool);
/* combination of following calls to construct packet */
rte_pktmbuf_mtod( .. );
/* follow above call with copy from work buffer */
rte_pktmbuf_prepend( .. );
rte_pktmbuf_append( .. );
}
rte_eth_tx_burst(start_port_id, tx_q, tx_mbuf, num_burst)
/* loop through packets not processed by rte_eth_tx_burst */
for(.. )
{
rte_pktmbuf_free (... );
}
}