examples/l2fwd-event: clean up worker state before exit

Event ports are configured to implicitly release the scheduler contexts
currently held in the next call to rte_event_dequeue_burst().
A worker core might still hold a scheduling context during exit, as the
next call to rte_event_dequeue_burst() is never made.
This might lead to deadlock based on the worker exit timing and when
there are very less number of flows.

Add clean up function to release any scheduling contexts held by the
worker by using RTE_EVENT_OP_RELEASE.

Signed-off-by: Pavan Nikhilesh <pbhagavatula@marvell.com>
Acked-by: Jerin Jacob <jerinj@marvell.com>

diff --git a/examples/l2fwd-event/l2fwd_common.c b/examples/l2fwd-event/l2fwd_common.c
index cf3d1b8aaf40bc29f2df8b39633aa14486b10425..15bfe790a0a331e9d225c71b86c900a1abeb7d43 100644 (file)
--- a/examples/l2fwd-event/l2fwd_common.c
+++ b/examples/l2fwd-event/l2fwd_common.c
@@ -114,3 +114,37 @@ l2fwd_event_init_ports(struct l2fwd_resources *rsrc)
 
        return nb_ports_available;
 }
+
+static void
+l2fwd_event_vector_array_free(struct rte_event events[], uint16_t num)
+{
+       uint16_t i;
+
+       for (i = 0; i < num; i++) {
+               rte_pktmbuf_free_bulk(events[i].vec->mbufs,
+                                     events[i].vec->nb_elem);
+               rte_mempool_put(rte_mempool_from_obj(events[i].vec),
+                               events[i].vec);
+       }
+}
+
+void
+l2fwd_event_worker_cleanup(uint8_t event_d_id, uint8_t port_id,
+                          struct rte_event events[], uint16_t nb_enq,
+                          uint16_t nb_deq, uint8_t is_vector)
+{
+       int i;
+
+       if (nb_deq) {
+               if (is_vector)
+                       l2fwd_event_vector_array_free(events + nb_enq,
+                                                     nb_deq - nb_enq);
+               else
+                       for (i = nb_enq; i < nb_deq; i++)
+                               rte_pktmbuf_free(events[i].mbuf);
+
+               for (i = 0; i < nb_deq; i++)
+                       events[i].op = RTE_EVENT_OP_RELEASE;
+               rte_event_enqueue_burst(event_d_id, port_id, events, nb_deq);
+       }
+}

diff --git a/examples/l2fwd-event/l2fwd_common.h b/examples/l2fwd-event/l2fwd_common.h
index 396e238c6a2b9334a524511046311ff9c4806947..bff3b65abf89fedb9b6c66334e07cf51d62b5990 100644 (file)
--- a/examples/l2fwd-event/l2fwd_common.h
+++ b/examples/l2fwd-event/l2fwd_common.h
@@ -140,5 +140,8 @@ l2fwd_get_rsrc(void)
 }
 
 int l2fwd_event_init_ports(struct l2fwd_resources *rsrc);
+void l2fwd_event_worker_cleanup(uint8_t event_d_id, uint8_t port_id,
+                               struct rte_event events[], uint16_t nb_enq,
+                               uint16_t nb_deq, uint8_t is_vector);
 
 #endif /* __L2FWD_COMMON_H__ */

diff --git a/examples/l2fwd-event/l2fwd_event.c b/examples/l2fwd-event/l2fwd_event.c
index 6df3cdfeab60450fef6311f8ac65551b3a5a760c..63450537fe30a50ff75987e73e29c83508fb3b20 100644 (file)
--- a/examples/l2fwd-event/l2fwd_event.c
+++ b/examples/l2fwd-event/l2fwd_event.c
@@ -193,6 +193,7 @@ l2fwd_event_loop_single(struct l2fwd_resources *rsrc,
                                        evt_rsrc->evq.nb_queues - 1];
        const uint64_t timer_period = rsrc->timer_period;
        const uint8_t event_d_id = evt_rsrc->event_d_id;
+       uint8_t enq = 0, deq = 0;
        struct rte_event ev;
 
        if (port_id < 0)
@@ -203,26 +204,28 @@ l2fwd_event_loop_single(struct l2fwd_resources *rsrc,
 
        while (!rsrc->force_quit) {
                /* Read packet from eventdev */
-               if (!rte_event_dequeue_burst(event_d_id, port_id, &ev, 1, 0))
+               deq = rte_event_dequeue_burst(event_d_id, port_id, &ev, 1, 0);
+               if (!deq)
                        continue;
 
                l2fwd_event_fwd(rsrc, &ev, tx_q_id, timer_period, flags);
 
                if (flags & L2FWD_EVENT_TX_ENQ) {
-                       while (rte_event_enqueue_burst(event_d_id, port_id,
-                                                      &ev, 1) &&
-                                       !rsrc->force_quit)
-                               ;
+                       do {
+                               enq = rte_event_enqueue_burst(event_d_id,
+                                                             port_id, &ev, 1);
+                       } while (!enq && !rsrc->force_quit);
                }
 
                if (flags & L2FWD_EVENT_TX_DIRECT) {
-                       while (!rte_event_eth_tx_adapter_enqueue(event_d_id,
-                                                               port_id,
-                                                               &ev, 1, 0) &&
-                                       !rsrc->force_quit)
-                               ;
+                       do {
+                               enq = rte_event_eth_tx_adapter_enqueue(
+                                       event_d_id, port_id, &ev, 1, 0);
+                       } while (!enq && !rsrc->force_quit);
                }
        }
+
+       l2fwd_event_worker_cleanup(event_d_id, port_id, &ev, enq, deq, 0);
 }
 
 static __rte_always_inline void
@@ -237,7 +240,7 @@ l2fwd_event_loop_burst(struct l2fwd_resources *rsrc,
        const uint8_t event_d_id = evt_rsrc->event_d_id;
        const uint8_t deq_len = evt_rsrc->deq_depth;
        struct rte_event ev[MAX_PKT_BURST];
-       uint16_t nb_rx, nb_tx;
+       uint16_t nb_rx = 0, nb_tx = 0;
        uint8_t i;
 
        if (port_id < 0)
@@ -280,6 +283,8 @@ l2fwd_event_loop_burst(struct l2fwd_resources *rsrc,
                                                ev + nb_tx, nb_rx - nb_tx, 0);
                }
        }
+
+       l2fwd_event_worker_cleanup(event_d_id, port_id, ev, nb_rx, nb_tx, 0);
 }
 
 static __rte_always_inline void
@@ -419,7 +424,7 @@ l2fwd_event_loop_vector(struct l2fwd_resources *rsrc, const uint32_t flags)
        const uint8_t event_d_id = evt_rsrc->event_d_id;
        const uint8_t deq_len = evt_rsrc->deq_depth;
        struct rte_event ev[MAX_PKT_BURST];
-       uint16_t nb_rx, nb_tx;
+       uint16_t nb_rx = 0, nb_tx = 0;
        uint8_t i;
 
        if (port_id < 0)
@@ -462,6 +467,8 @@ l2fwd_event_loop_vector(struct l2fwd_resources *rsrc, const uint32_t flags)
                                        nb_rx - nb_tx, 0);
                }
        }
+
+       l2fwd_event_worker_cleanup(event_d_id, port_id, ev, nb_rx, nb_tx, 1);
 }
 
 static void __rte_noinline