net: Use k_fifo instead of k_work in RX and TX processing

The k_work handler cannot manipulate the used k_work. This means
that it is not easy to cleanup the net_pkt because it contains
k_work in it. Because of this, use k_fifo instead between
RX thread and network driver, and between application and TX
thread.

A echo-server/client run with IPv4 and UDP gave following
results:

Using k_work
------------
TX traffic class statistics:
TC  Priority	Sent pkts	bytes	time
[0] BK (1)	21922		5543071	103 us	[0->41->26->34=101 us]
[1] BE (0)	0		0	-
RX traffic class statistics:
TC  Priority	Recv pkts	bytes	time
[0] BK (0)	0		0	-
[1] BE (0)	21925		6039151	97 us	[0->21->16->37->20=94 us]

Using k_fifo
------------
TX traffic class statistics:
TC  Priority	Sent pkts	bytes	time
[0] BK (1)	15079		3811118	94 us	[0->36->23->32=91 us]
[1] BE (0)	0		0	-
RX traffic class statistics:
TC  Priority	Recv pkts	bytes	time
[0] BK (1)	0		0	-
[1] BE (0)	15073		4150947	79 us	[0->17->12->32->14=75 us]

So using k_fifo gives about 10% better performance with same workload.

Fixes #34690

Signed-off-by: Jukka Rissanen <[email protected]>

include/net/net_if.h

@@ -396,15 +396,18 @@ struct net_if_config {
  *
  * Traffic classes are used when sending or receiving data that is classified
  * with different priorities. So some traffic can be marked as high priority
- * and it will be sent or received first. There is always at least one work
- * queue in the system for Rx and Tx. Each network packet that is transmitted
- * or received goes through a work queue thread that will transmit it.
+ * and it will be sent or received first. Each network packet that is
+ * transmitted or received goes through a fifo to a thread that will transmit
+ * it.
  */
 struct net_traffic_class {
- /** Work queue for handling this Tx or Rx packet */
- struct k_work_q work_q;
+ /** Fifo for handling this Tx or Rx packet */
+ struct k_fifo fifo;
 
- /** Stack for this work queue */
+ /** Traffic class handler thread */
+ struct k_thread handler;
+
+ /** Stack for this handler */
  k_thread_stack_t *stack;
 };
 

include/net/net_pkt.h

@@ -60,18 +60,11 @@ struct net_pkt_cursor {
  * net_pkt_clone() function.
  */
 struct net_pkt {
- union {
- /** Internal variable that is used when packet is sent
- * or received.
- */
- struct k_work work;
- /** Socket layer will queue received net_pkt into a k_fifo.
- * Since this happens after consuming net_pkt's k_work on
- * RX path, it is then fine to have both attributes sharing
- * the same memory area.
- */
- intptr_t sock_recv_fifo;
- };
+ /**
+ * The fifo is used by RX/TX threads and by socket layer. The net_pkt
+ * is queued via fifo to the processing thread.
+ */
+ intptr_t fifo;
 
  /** Slab pointer from where it belongs to */
  struct k_mem_slab *slab;
@@ -264,11 +257,6 @@ struct net_pkt {
 
 /** @cond ignore */
 
-static inline struct k_work *net_pkt_work(struct net_pkt *pkt)
-{
- return &pkt->work;
-}
-
 /* The interface real ll address */
 static inline struct net_linkaddr *net_pkt_lladdr_if(struct net_pkt *pkt)
 {

subsys/net/ip/net_core.c

@@ -364,12 +364,8 @@ static void net_rx(struct net_if *iface, struct net_pkt *pkt)
  net_pkt_print();
 }
 
-static void process_rx_packet(struct k_work *work)
+void net_process_rx_packet(struct net_pkt *pkt)
 {
- struct net_pkt *pkt;
-
- pkt = CONTAINER_OF(work, struct net_pkt, work);
-
  net_pkt_set_rx_stats_tick(pkt, k_cycle_get_32());
 
  net_capture_pkt(net_pkt_iface(pkt), pkt);
@@ -382,8 +378,6 @@ static void net_queue_rx(struct net_if *iface, struct net_pkt *pkt)
  uint8_t prio = net_pkt_priority(pkt);
  uint8_t tc = net_rx_priority2tc(prio);
 
- k_work_init(net_pkt_work(pkt), process_rx_packet);
-
 #if defined(CONFIG_NET_STATISTICS)
  net_stats_update_tc_recv_pkt(iface, tc);
  net_stats_update_tc_recv_bytes(iface, tc, net_pkt_get_len(pkt));
@@ -395,7 +389,7 @@ static void net_queue_rx(struct net_if *iface, struct net_pkt *pkt)
 #endif
 
  if (NET_TC_RX_COUNT == 0) {
- process_rx_packet(net_pkt_work(pkt));
+ net_process_rx_packet(pkt);
  } else {
  net_tc_submit_to_rx_queue(tc, pkt);
  }

subsys/net/ip/net_if.c

@@ -316,12 +316,9 @@ static bool net_if_tx(struct net_if *iface, struct net_pkt *pkt)
  return true;
 }
 
-static void process_tx_packet(struct k_work *work)
+void net_process_tx_packet(struct net_pkt *pkt)
 {
  struct net_if *iface;
- struct net_pkt *pkt;
-
- pkt = CONTAINER_OF(work, struct net_pkt, work);
 
  net_pkt_set_tx_stats_tick(pkt, k_cycle_get_32());
 
@@ -355,8 +352,6 @@ void net_if_queue_tx(struct net_if *iface, struct net_pkt *pkt)
  return;
  }
 
- k_work_init(net_pkt_work(pkt), process_tx_packet);
-
 #if NET_TC_TX_COUNT > 1
  NET_DBG("TC %d with prio %d pkt %p", tc, prio, pkt);
 #endif

subsys/net/ip/net_private.h

@@ -47,6 +47,8 @@ extern void net_if_post_init(void);
 extern void net_if_carrier_down(struct net_if *iface);
 extern void net_if_stats_reset(struct net_if *iface);
 extern void net_if_stats_reset_all(void);
+extern void net_process_rx_packet(struct net_pkt *pkt);
+extern void net_process_tx_packet(struct net_pkt *pkt);
 
 #if defined(CONFIG_NET_NATIVE) || defined(CONFIG_NET_OFFLOAD)
 extern void net_context_init(void);

subsys/net/ip/net_tc.c

@@ -40,12 +40,19 @@ static struct net_traffic_class tx_classes[NET_TC_TX_COUNT];
 static struct net_traffic_class rx_classes[NET_TC_RX_COUNT];
 #endif
 
+#if NET_TC_RX_COUNT > 0 || NET_TC_TX_COUNT > 0
+static void submit_to_queue(struct k_fifo *queue, struct net_pkt *pkt)
+{
+ k_fifo_put(queue, pkt);
+}
+#endif
+
 bool net_tc_submit_to_tx_queue(uint8_t tc, struct net_pkt *pkt)
 {
 #if NET_TC_TX_COUNT > 0
  net_pkt_set_tx_stats_tick(pkt, k_cycle_get_32());
 
- k_work_submit_to_queue(&tx_classes[tc].work_q, net_pkt_work(pkt));
+ submit_to_queue(&tx_classes[tc].fifo, pkt);
 #else
  ARG_UNUSED(tc);
  ARG_UNUSED(pkt);
@@ -58,7 +65,7 @@ void net_tc_submit_to_rx_queue(uint8_t tc, struct net_pkt *pkt)
 #if NET_TC_RX_COUNT > 0
  net_pkt_set_rx_stats_tick(pkt, k_cycle_get_32());
 
- k_work_submit_to_queue(&rx_classes[tc].work_q, net_pkt_work(pkt));
+ submit_to_queue(&rx_classes[tc].fifo, pkt);
 #else
  ARG_UNUSED(tc);
  ARG_UNUSED(pkt);
@@ -229,9 +236,40 @@ static void net_tc_rx_stats_priority_setup(struct net_if *iface,
 #endif
 #endif
 
-/* Create workqueue for each traffic class we are using. All the network
- * traffic goes through these classes. There needs to be at least one traffic
- * class in the system.
+#if NET_TC_RX_COUNT > 0
+static void tc_rx_handler(struct k_fifo *fifo)
+{
+ struct net_pkt *pkt;
+
+ while (1) {
+ pkt = k_fifo_get(fifo, K_FOREVER);
+ if (pkt == NULL) {
+ continue;
+ }
+
+ net_process_rx_packet(pkt);
+ }
+}
+#endif
+
+#if NET_TC_TX_COUNT > 0
+static void tc_tx_handler(struct k_fifo *fifo)
+{
+ struct net_pkt *pkt;
+
+ while (1) {
+ pkt = k_fifo_get(fifo, K_FOREVER);
+ if (pkt == NULL) {
+ continue;
+ }
+
+ net_process_tx_packet(pkt);
+ }
+}
+#endif
+
+/* Create a fifo for each traffic class we are using. All the network
+ * traffic goes through these classes.
  */
 void net_tc_tx_init(void)
 {
@@ -250,32 +288,43 @@ void net_tc_tx_init(void)
  for (i = 0; i < NET_TC_TX_COUNT; i++) {
  uint8_t thread_priority;
  int priority;
+ k_tid_t tid;
 
  thread_priority = tx_tc2thread(i);
 
  priority = IS_ENABLED(CONFIG_NET_TC_THREAD_COOPERATIVE) ?
  K_PRIO_COOP(thread_priority) :
  K_PRIO_PREEMPT(thread_priority);
 
- NET_DBG("[%d] Starting TX queue %p stack size %zd "
+ NET_DBG("[%d] Starting TX handler %p stack size %zd "
  "prio %d %s(%d)", i,
- &tx_classes[i].work_q,
+ &tx_classes[i].handler,
  K_KERNEL_STACK_SIZEOF(tx_stack[i]),
  thread_priority,
  IS_ENABLED(CONFIG_NET_TC_THREAD_COOPERATIVE) ?
  "coop" : "preempt",
  priority);
 
- k_work_queue_start(&tx_classes[i].work_q, tx_stack[i],
- K_KERNEL_STACK_SIZEOF(tx_stack[i]), priority,
- NULL);
+ k_fifo_init(&tx_classes[i].fifo);
+
+ tid = k_thread_create(&tx_classes[i].handler, tx_stack[i],
+ K_KERNEL_STACK_SIZEOF(tx_stack[i]),
+ (k_thread_entry_t)tc_tx_handler,
+ &tx_classes[i].fifo, NULL, NULL,
+ priority, 0, K_FOREVER);
+ if (!tid) {
+ NET_ERR("Cannot create TC handler thread %d", i);
+ continue;
+ }
 
  if (IS_ENABLED(CONFIG_THREAD_NAME)) {
  char name[MAX_NAME_LEN];
 
  snprintk(name, sizeof(name), "tx_q[%d]", i);
- k_thread_name_set(&tx_classes[i].work_q.thread, name);
+ k_thread_name_set(tid, name);
  }
+
+ k_thread_start(tid);
  }
 #endif
 }
@@ -297,32 +346,43 @@ void net_tc_rx_init(void)
  for (i = 0; i < NET_TC_RX_COUNT; i++) {
  uint8_t thread_priority;
  int priority;
+ k_tid_t tid;
 
  thread_priority = rx_tc2thread(i);
 
  priority = IS_ENABLED(CONFIG_NET_TC_THREAD_COOPERATIVE) ?
  K_PRIO_COOP(thread_priority) :
  K_PRIO_PREEMPT(thread_priority);
 
- NET_DBG("[%d] Starting RX queue %p stack size %zd "
+ NET_DBG("[%d] Starting RX handler %p stack size %zd "
  "prio %d %s(%d)", i,
- &rx_classes[i].work_q,
+ &rx_classes[i].handler,
  K_KERNEL_STACK_SIZEOF(rx_stack[i]),
  thread_priority,
  IS_ENABLED(CONFIG_NET_TC_THREAD_COOPERATIVE) ?
  "coop" : "preempt",
  priority);
 
- k_work_queue_start(&rx_classes[i].work_q, rx_stack[i],
- K_KERNEL_STACK_SIZEOF(rx_stack[i]), priority,
- NULL);
+ k_fifo_init(&rx_classes[i].fifo);
+
+ tid = k_thread_create(&rx_classes[i].handler, rx_stack[i],
+ K_KERNEL_STACK_SIZEOF(rx_stack[i]),
+ (k_thread_entry_t)tc_rx_handler,
+ &rx_classes[i].fifo, NULL, NULL,
+ priority, 0, K_FOREVER);
+ if (!tid) {
+ NET_ERR("Cannot create TC handler thread %d", i);
+ continue;
+ }
 
  if (IS_ENABLED(CONFIG_THREAD_NAME)) {
  char name[MAX_NAME_LEN];
 
  snprintk(name, sizeof(name), "rx_q[%d]", i);
- k_thread_name_set(&rx_classes[i].work_q.thread, name);
+ k_thread_name_set(tid, name);
  }
+
+ k_thread_start(tid);
  }
 #endif
 }