All of lore.kernel.org
 help / color / mirror / Atom feed
From: Paolo Abeni <pabeni@redhat.com>
To: "illusion.wang" <illusion.wang@nebula-matrix.com>,
	dimon.zhao@nebula-matrix.com, alvin.wang@nebula-matrix.com,
	sam.chen@nebula-matrix.com, netdev@vger.kernel.org
Cc: andrew+netdev@lunn.ch, corbet@lwn.net, kuba@kernel.org,
	horms@kernel.org, linux-doc@vger.kernel.org,
	vadim.fedorenko@linux.dev, lukas.bulwahn@redhat.com,
	edumazet@google.com, enelsonmoore@gmail.com,
	skhan@linuxfoundation.org, hkallweit1@gmail.com,
	open list <linux-kernel@vger.kernel.org>
Subject: Re: [PATCH v21 net-next 04/12] net/nebula-matrix: add channel layer
Date: Thu, 9 Jul 2026 10:09:37 +0200	[thread overview]
Message-ID: <698d9e2d-70e0-40e7-9398-e39ea681245a@redhat.com> (raw)
In-Reply-To: <20260708064742.35391-5-illusion.wang@nebula-matrix.com>

On 7/8/26 8:47 AM, illusion.wang wrote:
> From: illusion wang <illusion.wang@nebula-matrix.com>
> 
> A channel management layer provides a structured approach to handle
> communication between different components and drivers. Here's a summary
> of its key functionalities:
> 
> 1. Message Handling Framework
>    Message Registration: Functions (nbl_chan_register_msg) allow dynamic
>    registration of message handlers for specific message types, enabling
>    extensible communication protocols.
> 
>    Message Sending/Acknowledgment: Core functions (nbl_chan_send_msg,
>    nbl_chan_send_ack) handle message transmission, including asynchronous
>    operations with acknowledgment (ACK) support. Received ACKs are
>    processed via nbl_chan_recv_ack_msg.
> 
>    Hash-Based Handler Lookup: A hash table (`handle_hash_tbl`) stores
>    message handlers for efficient O(1) lookup by message type. The
>    entire table is removed via `nbl_chan_remove_msg_handler` during
>    driver teardown (per-message-type removal is not implemented
>    in this version).
> 
> 2. Channel Types and Queue Management
>    Mailbox Channel: For direct communication between PF0 and Other PF.
> 
>    Queue Initialization: Functions (nbl_chan_init_queue,
>    nbl_chan_init_tx_queue) allocate resources:
>    - TX descriptors: dmam_alloc_coherent()
>    - RX descriptors: dmam_alloc_coherent()
>    - TX/RX buffer metadata arrays (txq->buf, rxq->buf): devm_kcalloc()
> 
>    Queue Teardown: nbl_chan_teardown_queue() stops queues, cancels
>    pending work items (clean_task), and destroys mutexes. It does NOT
>    free DMA memory, which is released automatically via devm on driver
>    remove.
> 
>    IMPORTANT - Resource Lifecycle Design:
>    DMA memory allocated with dmam_alloc_coherent() is intentionally NOT
>    freed in nbl_chan_teardown_queue(). The queues are allocated once
>    during driver probe and freed only during driver remove (when all
>    devm_ resources are released). This assumes queues are NOT dynamically
>    torn down and recreated per-PF during normal operation.
> 
>    Queue Configuration: Hardware-specific queue parameters (e.g., buffer
>    sizes, entry counts) are set via nbl_chan_config_queue, with hardware
>    interactions delegated to hw_ops.
> 
> 3. Hardware Abstraction Layer (HW Ops)
>    Hardware-Specific Operations: The nbl_hw_ops structure abstracts
>    hardware interactions: queue configuration (config_mailbox_txq/rxq),
>    tail pointer updates (update_mailbox_queue_tail_ptr).
> 
> Signed-off-by: illusion wang <illusion.wang@nebula-matrix.com>
> ---
>  .../net/ethernet/nebula-matrix/nbl/Makefile   |    4 +-
>  .../nbl/nbl_channel/nbl_channel.c             | 1094 +++++++++++++++++
>  .../nbl/nbl_channel/nbl_channel.h             |  168 +++
>  .../nebula-matrix/nbl/nbl_common/nbl_common.c |  172 +++
>  .../nebula-matrix/nbl/nbl_common/nbl_common.h |   33 +
>  .../net/ethernet/nebula-matrix/nbl/nbl_core.h |   11 +
>  .../nbl/nbl_hw/nbl_hw_leonis/nbl_hw_leonis.c  |  173 +++
>  .../nbl/nbl_hw/nbl_hw_leonis/nbl_hw_leonis.h  |   56 +
>  .../nebula-matrix/nbl/nbl_hw/nbl_hw_reg.h     |   45 +
>  .../nbl/nbl_include/nbl_def_channel.h         |  113 ++
>  .../nbl/nbl_include/nbl_def_common.h          |   17 +
>  .../nbl/nbl_include/nbl_def_hw.h              |   15 +
>  .../nbl/nbl_include/nbl_include.h             |    6 +
>  .../net/ethernet/nebula-matrix/nbl/nbl_main.c |    7 +
>  14 files changed, 1913 insertions(+), 1 deletion(-)
>  create mode 100644 drivers/net/ethernet/nebula-matrix/nbl/nbl_channel/nbl_channel.c
>  create mode 100644 drivers/net/ethernet/nebula-matrix/nbl/nbl_channel/nbl_channel.h
>  create mode 100644 drivers/net/ethernet/nebula-matrix/nbl/nbl_common/nbl_common.c
>  create mode 100644 drivers/net/ethernet/nebula-matrix/nbl/nbl_common/nbl_common.h
> 
> diff --git a/drivers/net/ethernet/nebula-matrix/nbl/Makefile b/drivers/net/ethernet/nebula-matrix/nbl/Makefile
> index caa863d3a582..6dc1539cee1f 100644
> --- a/drivers/net/ethernet/nebula-matrix/nbl/Makefile
> +++ b/drivers/net/ethernet/nebula-matrix/nbl/Makefile
> @@ -3,5 +3,7 @@
>  
>  obj-$(CONFIG_NBL) := nbl.o
>  
> -nbl-objs +=       nbl_hw/nbl_hw_leonis/nbl_hw_leonis.o \
> +nbl-objs +=       nbl_common/nbl_common.o \
> +				nbl_channel/nbl_channel.o \
> +				nbl_hw/nbl_hw_leonis/nbl_hw_leonis.o \
>  				nbl_main.o
> diff --git a/drivers/net/ethernet/nebula-matrix/nbl/nbl_channel/nbl_channel.c b/drivers/net/ethernet/nebula-matrix/nbl/nbl_channel/nbl_channel.c
> new file mode 100644
> index 000000000000..220c740f68b9
> --- /dev/null
> +++ b/drivers/net/ethernet/nebula-matrix/nbl/nbl_channel/nbl_channel.c
> @@ -0,0 +1,1094 @@
> +// SPDX-License-Identifier: GPL-2.0
> +/*
> + * Copyright (c) 2025 Nebula Matrix Limited.
> + */
> +#include <linux/delay.h>
> +#include <linux/device.h>
> +#include <linux/bitfield.h>
> +#include <linux/pci.h>
> +#include <linux/bits.h>
> +#include <linux/dma-mapping.h>
> +#include <linux/atomic.h>
> +#include <linux/wait.h>
> +#include "nbl_channel.h"
> +
> +static int nbl_chan_add_msg_handler(struct nbl_channel_mgt *chan_mgt,
> +				    u16 msg_type, nbl_chan_resp func,
> +				    void *priv)
> +{
> +	struct nbl_chan_msg_node_data handler = { 0 };
> +	int ret;
> +
> +	handler.func = func;
> +	handler.priv = priv;
> +	ret = nbl_common_alloc_hash_node(chan_mgt->handle_hash_tbl, &msg_type,
> +					 &handler, NULL);
> +
> +	return ret;
> +}
> +
> +static int nbl_chan_init_msg_handler(struct nbl_channel_mgt *chan_mgt)
> +{
> +	struct nbl_common_info *common = chan_mgt->common;
> +	struct nbl_hash_tbl_key tbl_key = { 0 };
> +
> +	tbl_key.dev = common->dev;
> +	tbl_key.key_size = sizeof(u16);
> +	tbl_key.data_size = sizeof(struct nbl_chan_msg_node_data);
> +	tbl_key.bucket_size = NBL_CHAN_HANDLER_TBL_BUCKET_SIZE;
> +
> +	chan_mgt->handle_hash_tbl = nbl_common_init_hash_table(&tbl_key);
> +	if (!chan_mgt->handle_hash_tbl)
> +		return -ENOMEM;
> +
> +	return 0;
> +}
> +
> +static void nbl_chan_remove_msg_handler(struct nbl_channel_mgt *chan_mgt)
> +{
> +	if (!chan_mgt->handle_hash_tbl)
> +		return;
> +	nbl_common_remove_hash_table(chan_mgt->handle_hash_tbl);
> +	chan_mgt->handle_hash_tbl = NULL;
> +}
> +
> +static void nbl_chan_init_queue_param(struct nbl_chan_info *chan_info,
> +				      u16 num_txq_entries, u16 num_rxq_entries,
> +				      u16 txq_buf_size, u16 rxq_buf_size)
> +{
> +	mutex_init(&chan_info->txq_lock);
> +	chan_info->num_txq_entries = num_txq_entries;
> +	chan_info->num_rxq_entries = num_rxq_entries;
> +	chan_info->txq_buf_size = txq_buf_size;
> +	chan_info->rxq_buf_size = rxq_buf_size;
> +	atomic_set(&chan_info->inflight_tx_cnt, 0);
> +	chan_info->shutdown = false;
> +}
> +
> +static int nbl_chan_init_tx_queue(struct nbl_common_info *common,
> +				  struct nbl_chan_info *chan_info)
> +{
> +	struct nbl_chan_ring *txq = &chan_info->txq;
> +	struct device *dev = common->dev;
> +	size_t size =
> +		chan_info->num_txq_entries * sizeof(struct nbl_chan_tx_desc);
> +	int i;
> +
> +	txq->desc.tx_desc =
> +		dmam_alloc_coherent(dev, size, &txq->dma, GFP_KERNEL);
> +	if (!txq->desc.tx_desc)
> +		return -ENOMEM;
> +
> +	chan_info->wait = devm_kcalloc(dev, chan_info->num_txq_entries,
> +				       sizeof(*chan_info->wait), GFP_KERNEL);
> +	if (!chan_info->wait)
> +		return -ENOMEM;
> +	for (i = 0; i < chan_info->num_txq_entries; i++) {
> +		init_waitqueue_head(&chan_info->wait[i].wait_queue);
> +		chan_info->wait[i].status = NBL_MBX_STATUS_IDLE;
> +		spin_lock_init(&chan_info->wait[i].status_lock);
> +	}
> +
> +	txq->buf = devm_kcalloc(dev, chan_info->num_txq_entries,
> +				sizeof(*txq->buf), GFP_KERNEL);
> +	if (!txq->buf)
> +		return -ENOMEM;
> +
> +	return 0;
> +}
> +
> +static int nbl_chan_init_rx_queue(struct nbl_common_info *common,
> +				  struct nbl_chan_info *chan_info)
> +{
> +	struct nbl_chan_ring *rxq = &chan_info->rxq;
> +	struct device *dev = common->dev;
> +	size_t size =
> +		chan_info->num_rxq_entries * sizeof(struct nbl_chan_rx_desc);
> +
> +	rxq->desc.rx_desc =
> +		dmam_alloc_coherent(dev, size, &rxq->dma, GFP_KERNEL);
> +	if (!rxq->desc.rx_desc) {
> +		dev_err(dev,
> +			"Allocate DMA for chan rx descriptor ring failed\n");
> +		return -ENOMEM;
> +	}
> +
> +	rxq->buf = devm_kcalloc(dev, chan_info->num_rxq_entries,
> +				sizeof(*rxq->buf), GFP_KERNEL);
> +	if (!rxq->buf)
> +		return -ENOMEM;
> +
> +	return 0;
> +}
> +
> +static int nbl_chan_init_queue(struct nbl_common_info *common,
> +			       struct nbl_chan_info *chan_info)
> +{
> +	int err;
> +
> +	err = nbl_chan_init_tx_queue(common, chan_info);
> +	if (err)
> +		return err;
> +
> +	err = nbl_chan_init_rx_queue(common, chan_info);
> +
> +	return err;
> +}
> +
> +static void nbl_chan_config_queue(struct nbl_channel_mgt *chan_mgt,
> +				  struct nbl_chan_info *chan_info, bool tx)
> +{
> +	struct nbl_hw_ops *hw_ops = chan_mgt->hw_ops_tbl->ops;
> +	struct nbl_hw_mgt *p = chan_mgt->hw_ops_tbl->priv;
> +	int size_bwid;
> +	struct nbl_chan_ring *ring;
> +	dma_addr_t dma_addr;
> +
> +	if (tx)
> +		ring = &chan_info->txq;
> +	else
> +		ring = &chan_info->rxq;
> +	dma_addr = ring->dma;
> +	if (tx) {
> +		size_bwid = ilog2(chan_info->num_txq_entries);
> +		hw_ops->config_mailbox_txq(p, dma_addr, size_bwid);
> +	} else {
> +		size_bwid = ilog2(chan_info->num_rxq_entries);
> +		hw_ops->config_mailbox_rxq(p, dma_addr, size_bwid);
> +	}
> +}
> +
> +static int nbl_chan_alloc_all_tx_bufs(struct nbl_channel_mgt *chan_mgt,
> +				      struct nbl_chan_info *chan_info)
> +{
> +	struct nbl_chan_ring *txq = &chan_info->txq;
> +	struct device *dev = chan_mgt->common->dev;
> +	struct nbl_chan_buf *buf;
> +	u16 i;
> +
> +	for (i = 0; i < chan_info->num_txq_entries; i++) {
> +		buf = &txq->buf[i];
> +		buf->va = dmam_alloc_coherent(dev, chan_info->txq_buf_size,
> +					      &buf->pa, GFP_KERNEL);
> +		if (!buf->va) {
> +			dev_err(dev,
> +				"Allocate buffer for chan tx queue failed\n");
> +			return -ENOMEM;
> +		}
> +	}
> +
> +	txq->next_to_clean = 0;
> +	txq->next_to_use = 0;
> +	txq->tail_ptr = 0;
> +
> +	return 0;
> +}
> +
> +static void nbl_chan_cfg_qinfo_map_table(struct nbl_channel_mgt *chan_mgt)
> +{
> +	struct nbl_hw_ops *hw_ops = chan_mgt->hw_ops_tbl->ops;
> +	struct nbl_common_info *common = chan_mgt->common;
> +	struct nbl_hw_mgt *p = chan_mgt->hw_ops_tbl->priv;
> +	u8 func_id;
> +	u32 pf_mask;
> +
> +	pf_mask = hw_ops->get_host_pf_mask(p);
> +	for (func_id = 0; func_id < NBL_MAX_PF; func_id++) {
> +		if (!(pf_mask & (1 << func_id)))
> +			hw_ops->cfg_mailbox_qinfo(p, func_id, common->hw_bus,
> +						  common->devid,
> +						  common->function + func_id);
> +	}
> +}
> +
> +static inline void nbl_chan_update_tail_ptr(struct nbl_hw_ops *hw_ops,
> +					    void *hw_priv, u32 tail_ptr, u8 qid)
> +{
> +	hw_ops->update_mailbox_queue_tail_ptr(hw_priv, tail_ptr, qid);
> +}
> +
> +static int nbl_chan_alloc_all_rx_bufs(struct nbl_channel_mgt *chan_mgt,
> +				      struct nbl_chan_info *chan_info)
> +{
> +	struct nbl_chan_ring *rxq = &chan_info->rxq;
> +	struct device *dev = chan_mgt->common->dev;
> +	struct nbl_chan_rx_desc *desc;
> +	struct nbl_chan_buf *buf;
> +	u16 i;
> +
> +	for (i = 0; i < chan_info->num_rxq_entries; i++) {
> +		buf = &rxq->buf[i];
> +		buf->va = dmam_alloc_coherent(dev, chan_info->rxq_buf_size,
> +					      &buf->pa, GFP_KERNEL);
> +		if (!buf->va) {
> +			dev_err(dev,
> +				"Allocate buffer for chan rx queue failed\n");
> +			goto err;
> +		}
> +	}
> +
> +	desc = rxq->desc.rx_desc;
> +	for (i = 0; i < chan_info->num_rxq_entries - 1; i++) {
> +		buf = &rxq->buf[i];
> +		desc[i].buf_addr = cpu_to_le64(buf->pa);
> +		desc[i].buf_len = cpu_to_le32(chan_info->rxq_buf_size);
> +		desc[i].flags = cpu_to_le16(BIT(NBL_CHAN_RX_DESC_AVAIL));
> +	}
> +
> +	rxq->next_to_clean = 0;
> +	rxq->next_to_use = chan_info->num_rxq_entries - 1;
> +	rxq->tail_ptr = chan_info->num_rxq_entries - 1;
> +
> +	return 0;
> +err:
> +	return -ENOMEM;
> +}
> +
> +static int nbl_chan_alloc_all_bufs(struct nbl_channel_mgt *chan_mgt,
> +				   struct nbl_chan_info *chan_info)
> +{
> +	int err;
> +
> +	err = nbl_chan_alloc_all_tx_bufs(chan_mgt, chan_info);
> +	if (err)
> +		return err;
> +	err = nbl_chan_alloc_all_rx_bufs(chan_mgt, chan_info);
> +
> +	return err;
> +}
> +
> +static void nbl_chan_stop_queue(struct nbl_channel_mgt *chan_mgt)
> +{
> +	struct nbl_hw_ops *hw_ops = chan_mgt->hw_ops_tbl->ops;
> +
> +	hw_ops->stop_mailbox_rxq(chan_mgt->hw_ops_tbl->priv);
> +	hw_ops->stop_mailbox_txq(chan_mgt->hw_ops_tbl->priv);
> +}
> +
> +static int nbl_chan_teardown_queue(struct nbl_channel_mgt *chan_mgt,
> +				   u8 chan_type)
> +{
> +	struct nbl_chan_info *chan_info = chan_mgt->chan_info[chan_type];
> +	struct nbl_chan_waitqueue_head *wait_head;
> +	u16 i;
> +
> +	/* Step 1: Mark shutdown flag, reject all new send requests */
> +	WRITE_ONCE(chan_info->shutdown, true);
> +
> +	/* Stop hardware queues */
> +	nbl_chan_stop_queue(chan_mgt);
> +
> +	/* Cancel any pending cleanup work */
> +	if (chan_info->clean_task)
> +		cancel_work_sync(chan_info->clean_task);
> +	for (i = 0; i < chan_info->num_txq_entries; i++) {
> +		wait_head = &chan_info->wait[i];
> +		spin_lock_irq(&wait_head->status_lock);
> +		/* Only wake threads that are actually waiting */
> +		if (READ_ONCE(wait_head->status) == NBL_MBX_STATUS_WAITING) {
> +			/* Mark as timeout so waking threads know to abort */
> +			wait_head->status = NBL_MBX_STATUS_TIMEOUT;
> +			wait_head->acked = 1;
> +			wait_head->ack_err = -EIO;
> +			/* Ensure status is written */
> +			smp_wmb();
> +		}
> +		spin_unlock_irq(&wait_head->status_lock);
> +		if (READ_ONCE(wait_head->status) == NBL_MBX_STATUS_TIMEOUT)
> +			wake_up(&wait_head->wait_queue);
> +	}
> +
> +	/* Step 2: Wait all in-flight send_msg threads exit via counter */
> +	while (atomic_read(&chan_info->inflight_tx_cnt) != 0) {
> +		/* synchronize atomic counter load with other CPUs */
> +		smp_rmb();
> +		usleep_range(100, 1000);
> +	}
> +
> +	/* No concurrent thread holds txq_lock now, safe destroy mutex */
> +	mutex_destroy(&chan_info->txq_lock);
> +
> +	return 0;
> +}
> +
> +static int nbl_chan_setup_queue(struct nbl_channel_mgt *chan_mgt, u8 chan_type)
> +{
> +	struct nbl_chan_info *chan_info = chan_mgt->chan_info[chan_type];
> +	struct nbl_hw_ops *hw_ops = chan_mgt->hw_ops_tbl->ops;
> +	struct nbl_common_info *common = chan_mgt->common;
> +	struct nbl_chan_ring *rxq = &chan_info->rxq;
> +	int err;
> +
> +	if (chan_info->init_done)
> +		return 0;
> +	nbl_chan_init_queue_param(chan_info, NBL_CHAN_QUEUE_LEN,
> +				  NBL_CHAN_QUEUE_LEN, NBL_CHAN_BUF_LEN,
> +				  NBL_CHAN_BUF_LEN);
> +	err = nbl_chan_init_queue(common, chan_info);
> +	if (err)
> +		goto chan_setup_fail;
> +	err = nbl_chan_alloc_all_bufs(chan_mgt, chan_info);
> +	if (err)
> +		goto chan_setup_fail;
> +	nbl_chan_config_queue(chan_mgt, chan_info, true); /* tx */
> +	nbl_chan_config_queue(chan_mgt, chan_info, false); /* rx */
> +	nbl_chan_update_tail_ptr(hw_ops, chan_mgt->hw_ops_tbl->priv,
> +				 rxq->tail_ptr, NBL_MB_RX_QID);
> +	chan_info->init_done = true;
> +	return 0;
> +chan_setup_fail:
> +	mutex_destroy(&chan_info->txq_lock);
> +	return err;
> +}
> +
> +static int nbl_chan_update_txqueue(struct nbl_channel_mgt *chan_mgt,
> +				   struct nbl_chan_info *chan_info,
> +				   struct nbl_chan_tx_param *param)
> +{
> +	struct nbl_chan_ring *txq = &chan_info->txq;
> +	struct nbl_chan_tx_desc *tx_desc =
> +		NBL_CHAN_TX_RING_TO_DESC(txq, txq->next_to_use);
> +	struct nbl_chan_buf *tx_buf =
> +		NBL_CHAN_TX_RING_TO_BUF(txq, txq->next_to_use);
> +
> +	if (param->arg_len > NBL_CHAN_BUF_LEN - sizeof(*tx_desc))
> +		return -EINVAL;
> +
> +	tx_desc->dstid = cpu_to_le16(param->dstid);
> +	tx_desc->msg_type = cpu_to_le16(param->msg_type);
> +	tx_desc->msgid = cpu_to_le16(param->msgid);
> +
> +	if (param->arg_len > NBL_CHAN_TX_DESC_EMBEDDED_DATA_LEN) {
> +		memcpy(tx_buf->va, param->arg, param->arg_len);
> +		tx_desc->buf_addr = cpu_to_le64(tx_buf->pa);
> +		tx_desc->buf_len = cpu_to_le16(param->arg_len);
> +		tx_desc->data_len = 0;
> +		memset(tx_desc->data, 0, sizeof(tx_desc->data));
> +	} else {
> +		memset(tx_desc->data, 0, sizeof(tx_desc->data));
> +		memset(&tx_desc->buf_addr, 0, sizeof(tx_desc->buf_addr));
> +		memcpy(tx_desc->data, param->arg, param->arg_len);
> +		tx_desc->buf_len = 0;
> +		tx_desc->data_len = cpu_to_le16(param->arg_len);
> +	}
> +	dma_wmb();
> +	tx_desc->flags = cpu_to_le16(BIT(NBL_CHAN_TX_DESC_AVAIL));
> +
> +	txq->next_to_use =
> +		NBL_NEXT_ID(txq->next_to_use, chan_info->num_txq_entries - 1);
> +	txq->tail_ptr++;
> +
> +	return 0;
> +}
> +
> +static int nbl_chan_kick_tx_ring(struct nbl_channel_mgt *chan_mgt,
> +				 struct nbl_chan_info *chan_info)
> +{
> +	struct nbl_hw_ops *hw_ops = chan_mgt->hw_ops_tbl->ops;
> +	struct nbl_chan_ring *txq = &chan_info->txq;
> +	struct device *dev = chan_mgt->common->dev;
> +	int max_retries = NBL_CHAN_TX_WAIT_TIMES;
> +	struct nbl_chan_tx_desc *tx_desc;
> +	int retry_count = 0;
> +
> +	nbl_chan_update_tail_ptr(hw_ops, chan_mgt->hw_ops_tbl->priv,
> +				 txq->tail_ptr, NBL_MB_TX_QID);
> +
> +	tx_desc = NBL_CHAN_TX_RING_TO_DESC(txq, txq->next_to_clean);
> +	while (retry_count < max_retries) {
> +		if (le16_to_cpu(READ_ONCE(tx_desc->flags)) &
> +		    BIT(NBL_CHAN_TX_DESC_USED)) {
> +			dma_rmb();
> +			break;
> +		}
> +
> +		retry_count++;
> +		if (retry_count == max_retries) {
> +			dev_err(dev, "chan send message type: %d timeout\n",
> +				le16_to_cpu(READ_ONCE(tx_desc->msg_type)));
> +			txq->next_to_clean = txq->next_to_use;
> +			return -ETIMEDOUT;
> +		}
> +		usleep_range(NBL_CHAN_TX_WAIT_US, NBL_CHAN_TX_WAIT_US_MAX);
> +	}
> +
> +	txq->next_to_clean = txq->next_to_use;
> +
> +	return 0;
> +}
> +
> +static void nbl_chan_recv_ack_msg(void *priv, u16 srcid, u16 msgid, void *data,
> +				  u32 data_len)
> +{
> +	struct nbl_channel_mgt *chan_mgt = (struct nbl_channel_mgt *)priv;
> +	struct nbl_chan_waitqueue_head *wait_head = NULL;
> +	struct device *dev = chan_mgt->common->dev;
> +	struct nbl_chan_info *chan_info =
> +		chan_mgt->chan_info[NBL_CHAN_TYPE_MAILBOX];
> +	u32 ack_datalen, ack_msgtype = 0;
> +	u32 *payload = data;
> +	u16 ack_msgid = 0;
> +	u32 copy_len;
> +
> +	if (data_len > NBL_CHAN_BUF_LEN ||
> +	    data_len < NBL_CHAN_ACK_HEAD_LEN * sizeof(u32)) {
> +		dev_err(dev, "Invalid ACK data_len: %u\n", data_len);
> +		return;
> +	}
> +	ack_datalen = data_len - NBL_CHAN_ACK_HEAD_LEN * sizeof(u32);
> +	ack_msgtype = le16_to_cpu(*(__le16 *)(payload + NBL_CHAN_MSG_TYPE_POS));
> +	ack_msgid = le16_to_cpu(*(__le16 *)(payload + NBL_CHAN_MSG_ID_POS));
> +	if (FIELD_GET(NBL_CHAN_MSGID_LOC_MASK, ack_msgid) >=
> +	    chan_info->num_txq_entries) {
> +		dev_err(dev, "chan recv msg id: %d err\n", ack_msgid);
> +		return;
> +	}
> +	wait_head =
> +		&chan_info->wait[FIELD_GET(NBL_CHAN_MSGID_LOC_MASK, ack_msgid)];
> +	spin_lock_irq(&wait_head->status_lock);
> +	if (srcid != wait_head->dstid) {
> +		/* Do not modify the status; the slot remains WAITING,
> +		 * and the sender will time out normally
> +		 */
> +		spin_unlock_irq(&wait_head->status_lock);
> +		dev_err(dev, "ACK srcid=%u != dstid=%u, rejecting\n", srcid,
> +			wait_head->dstid);
> +		return;
> +	}
> +	if (READ_ONCE(wait_head->status) != NBL_MBX_STATUS_WAITING) {
> +		spin_unlock_irq(&wait_head->status_lock);
> +		dev_err(dev,
> +			"Skip ack with invalid status, wait_head msgtype:%u msg_index:%u status:%d ack_data_len:%d, ack msgtype:%u msgid:%u datalen:%d\n",
> +			READ_ONCE(wait_head->msg_type),
> +			READ_ONCE(wait_head->msg_index),
> +			READ_ONCE(wait_head->status), wait_head->ack_data_len,
> +			ack_msgtype, ack_msgid, ack_datalen);
> +		return;
> +	}
> +
> +	if (READ_ONCE(wait_head->msg_type) != ack_msgtype) {
> +		/*
> +		 * Mismatched ACK. Restore state to WAITING so the original
> +		 * sender will time out and not reuse the slot.
> +		 */
> +		wait_head->status = NBL_MBX_STATUS_WAITING;
> +
> +		dev_err(dev,
> +			"Skip ack msg type donot match, wait_head msgtype:%u msg_index:%u status:%d ack_data_len:%d, ack msgtype:%u msgid:%u datalen:%d\n",
> +			READ_ONCE(wait_head->msg_type),
> +			READ_ONCE(wait_head->msg_index),
> +			READ_ONCE(wait_head->status), wait_head->ack_data_len,
> +			ack_msgtype, ack_msgid, ack_datalen);
> +		spin_unlock_irq(&wait_head->status_lock);
> +		/* Wake up the sender to let it know the ACK was invalid */
> +		wake_up(&wait_head->wait_queue);
> +		return;
> +	}
> +	if (FIELD_GET(NBL_CHAN_MSGID_INDEX_MASK, ack_msgid) !=
> +	    READ_ONCE(wait_head->msg_index)) {
> +		/*
> +		 * Stale ACK. Restore state to WAITING so the original
> +		 * sender will time out and not reuse the slot.
> +		 */
> +		wait_head->status = NBL_MBX_STATUS_WAITING;
> +
> +		dev_err(dev,
> +			"Stale ACK: expected index=%u, got msgid %u\n",
> +			READ_ONCE(wait_head->msg_index), ack_msgid);
> +		spin_unlock_irq(&wait_head->status_lock);
> +		/* Wake up the sender to let it know the ACK was stale */
> +		wake_up(&wait_head->wait_queue);
> +		return;
> +	}
> +
> +	wait_head->ack_err =
> +		le32_to_cpu(*(__le32 *)(payload + NBL_CHAN_ACK_RET_POS));
> +
> +	copy_len = min_t(u32, wait_head->ack_data_len, ack_datalen);
> +	if (wait_head->ack_err >= 0 && copy_len > 0) {
> +		if (!wait_head->ack_data) {
> +			dev_err(dev, "ACK payload dropped: ack_data is NULL\n");
> +			wait_head->ack_data_len = 0;
> +			goto ack_done;
> +		}
> +		memcpy((char *)wait_head->ack_data,
> +		       payload + NBL_CHAN_ACK_HEAD_LEN, copy_len);
> +		wait_head->ack_data_len = (u16)copy_len;
> +	} else {
> +		wait_head->ack_data_len = 0;
> +	}
> +ack_done:
> +	/*
> +	 * Ensure all writes to ack_data and ack_data_len are completed
> +	 * before setting the 'acked' flag. This prevents other threads
> +	 * from observing stale or partially updated data.
> +	 */
> +	smp_wmb();
> +	wait_head->acked = 1;
> +	spin_unlock_irq(&wait_head->status_lock);
> +	if (READ_ONCE(wait_head->acked))
> +		wake_up(&wait_head->wait_queue);
> +}
> +
> +static void nbl_chan_recv_msg(struct nbl_channel_mgt *chan_mgt, void *data)
> +{
> +	struct device *dev = chan_mgt->common->dev;
> +	struct nbl_chan_msg_node_data *msg_handler;
> +	u16 msg_type, payload_len, srcid, msgid;
> +	struct nbl_chan_tx_desc *tx_desc;
> +	void *payload;
> +
> +	tx_desc = data;
> +	msg_type = le16_to_cpu(tx_desc->msg_type);
> +	dev_dbg(dev, "recv msg_type: %d\n", msg_type);
> +
> +	srcid = le16_to_cpu(tx_desc->srcid);
> +	msgid = le16_to_cpu(tx_desc->msgid);
> +	/* Only check if the value exceeds the maximum, relying on the hash
> +	 * table to filter invalid message IDs.
> +	 * The gap values are reserved for future protocol extensions.
> +	 */
> +	if (msg_type >= NBL_CHAN_MSG_MAILBOX_MAX)
> +		return;
> +
> +	if (tx_desc->data_len) {
> +		payload_len = le16_to_cpu(tx_desc->data_len);
> +		if (payload_len > NBL_CHAN_TX_DESC_EMBEDDED_DATA_LEN) {
> +			dev_err(dev,
> +				"data_len=%u exceeds embedded buffer size=%u\n",
> +				payload_len,
> +				NBL_CHAN_TX_DESC_EMBEDDED_DATA_LEN);
> +			return;
> +		}
> +		payload = tx_desc->data;
> +	} else {
> +		payload_len = le16_to_cpu(tx_desc->buf_len);
> +		if (payload_len > NBL_CHAN_BUF_LEN - sizeof(*tx_desc)) {
> +			dev_err(dev,
> +				"buf_len=%u exceeds external buffer size=%zu\n",
> +				payload_len,
> +				NBL_CHAN_BUF_LEN - sizeof(*tx_desc));
> +			return;
> +		}
> +		payload = tx_desc + 1;
> +	}
> +
> +	msg_handler =
> +		nbl_common_get_hash_node(chan_mgt->handle_hash_tbl, &msg_type);
> +	if (!msg_handler || !msg_handler->func) {
> +		dev_err(dev,
> +			"No handler for msg_type: %u (srcid=%u, msgid=%u)\n",
> +			msg_type, srcid, msgid);
> +		return;
> +	}
> +	msg_handler->func(msg_handler->priv, srcid, msgid, payload,
> +			  payload_len);
> +}
> +
> +static void nbl_chan_advance_rx_ring(struct nbl_channel_mgt *chan_mgt,
> +				     struct nbl_chan_info *chan_info,
> +				     struct nbl_chan_ring *rxq)
> +{
> +	struct nbl_hw_ops *hw_ops = chan_mgt->hw_ops_tbl->ops;
> +	struct nbl_chan_rx_desc *rx_desc;
> +	struct nbl_chan_buf *rx_buf;
> +	u16 next_to_use;
> +
> +	next_to_use = rxq->next_to_use;
> +	rx_desc = NBL_CHAN_RX_RING_TO_DESC(rxq, next_to_use);
> +	rx_buf = NBL_CHAN_RX_RING_TO_BUF(rxq, next_to_use);
> +
> +	rx_desc->buf_addr = cpu_to_le64(rx_buf->pa);
> +	rx_desc->buf_len = cpu_to_le32(chan_info->rxq_buf_size);
> +
> +	/*
> +	 * DMA Write Memory Barrier:
> +	 * Ensures all previous DMA-mapped writes (buffer address/length)
> +	 * are completed before the descriptor flags are updated.
> +	 * This prevents hardware from seeing a partially updated descriptor
> +	 * where flags are set but buffer info isn't ready yet.
> +	 */
> +	dma_wmb();
> +
> +	rx_desc->flags = cpu_to_le16(BIT(NBL_CHAN_RX_DESC_AVAIL));
> +
> +	/*
> +	 * CPU Write Memory Barrier:
> +	 * Ensures the descriptor flags update is visible to other CPUs
> +	 * before we update the tail pointer. This is important for:
> +	 * 1. Software cleaning threads that might be checking the tail pointer
> +	 * 2. Maintaining proper memory ordering in multi-core systems
> +	 */
> +	wmb();
> +	rxq->next_to_use++;
> +	if (rxq->next_to_use == chan_info->num_rxq_entries)
> +		rxq->next_to_use = 0;
> +	rxq->tail_ptr++;
> +
> +	nbl_chan_update_tail_ptr(hw_ops, chan_mgt->hw_ops_tbl->priv,
> +				 rxq->tail_ptr, NBL_MB_RX_QID);
> +}
> +
> +/*
> + * Since the channel operates in either polling mode or interrupt mode
> + * (mutually exclusive, configured via set_queue_state), nbl_chan_clean_queue
> + * is always called in a serialized manner:
> + * 1. In polling mode: nbl_chan_clean_queue is called directly within
> + * nbl_chan_send_msg, in the same thread after txq_lock has been released.
> + * No other thread can call it concurrently.
> + * 2. In interrupt mode: nbl_chan_clean_queue is called from a workqueue
> + * (nbl_dev_clean_mailbox_task). Linux workqueue guarantees that the same
> + * work item never runs concurrently on multiple CPUs.
> + * Therefore, at any given time, only one execution context can be inside
> + * nbl_chan_clean_queue. There is no concurrency, and thus no need for
> + * locking
> + */
> +static void nbl_chan_clean_queue(struct nbl_channel_mgt *chan_mgt,
> +				 struct nbl_chan_info *chan_info)
> +{
> +	struct nbl_chan_ring *rxq = &chan_info->rxq;
> +	struct device *dev = chan_mgt->common->dev;
> +	struct nbl_chan_rx_desc *rx_desc;
> +	struct nbl_chan_buf *rx_buf;
> +	bool more_work = false;
> +	u16 next_to_clean;
> +	u32 budget = 64;
> +
> +	next_to_clean = rxq->next_to_clean;
> +	rx_desc = NBL_CHAN_RX_RING_TO_DESC(rxq, next_to_clean);
> +	rx_buf = NBL_CHAN_RX_RING_TO_BUF(rxq, next_to_clean);
> +	while (le16_to_cpu(rx_desc->flags) & BIT(NBL_CHAN_RX_DESC_USED)) {
> +		if (!(le16_to_cpu(rx_desc->flags) &
> +		      BIT(NBL_CHAN_RX_DESC_WRITE)))
> +			dev_dbg(dev,
> +				"mailbox rx flag 0x%x has no NBL_CHAN_RX_DESC_WRITE\n",
> +				le16_to_cpu(rx_desc->flags));
> +
> +		dma_rmb();
> +		nbl_chan_recv_msg(chan_mgt, rx_buf->va);
> +		nbl_chan_advance_rx_ring(chan_mgt, chan_info, rxq);
> +		next_to_clean++;
> +		if (next_to_clean == chan_info->num_rxq_entries)
> +			next_to_clean = 0;
> +		rx_desc = NBL_CHAN_RX_RING_TO_DESC(rxq, next_to_clean);
> +		rx_buf = NBL_CHAN_RX_RING_TO_BUF(rxq, next_to_clean);
> +		if (--budget == 0) {
> +			more_work = true;
> +			break;
> +		}
> +	}
> +	rxq->next_to_clean = next_to_clean;
> +	/* If descriptors remain, reschedule work to avoid stalled RX ring */
> +	if (more_work)
> +		schedule_work(chan_info->clean_task);
> +}
> +
> +static void nbl_chan_clean_queue_subtask(struct nbl_channel_mgt *chan_mgt,
> +					 u8 chan_type)
> +{
> +	struct nbl_chan_info *chan_info = chan_mgt->chan_info[chan_type];
> +
> +	if (!test_bit(NBL_CHAN_INTERRUPT_READY, chan_info->state))
> +		return;
> +
> +	nbl_chan_clean_queue(chan_mgt, chan_info);
> +}
> +
> +static int nbl_chan_get_msg_id(struct nbl_chan_info *chan_info,
> +			       u16 *msgid)
> +{
> +	int valid_loc = chan_info->wait_head_index, i;
> +	struct nbl_chan_waitqueue_head *wait = NULL;
> +	int status;
> +
> +	for (i = 0; i < NBL_CHAN_QUEUE_LEN; i++) {
> +		wait = &chan_info->wait[valid_loc];
> +		status = READ_ONCE(wait->status);
> +		if (status == NBL_MBX_STATUS_IDLE ||
> +		    status == NBL_MBX_STATUS_TIMEOUT) {
> +			wait->msg_index = NBL_NEXT_ID(wait->msg_index,
> +						      NBL_CHAN_MSG_INDEX_MAX);
> +			*msgid =
> +				FIELD_PREP(NBL_CHAN_MSGID_INDEX_MASK,
> +					   wait->msg_index) |
> +				FIELD_PREP(NBL_CHAN_MSGID_LOC_MASK, valid_loc);
> +			valid_loc = NBL_NEXT_ID(valid_loc,
> +						chan_info->num_txq_entries - 1);
> +			chan_info->wait_head_index = valid_loc;
> +			return 0;
> +		}
> +
> +		valid_loc =
> +			NBL_NEXT_ID(valid_loc, chan_info->num_txq_entries - 1);
> +	}
> +
> +	/*
> +	 * the current NBL_CHAN_QUEUE_LEN configuration meets the design
> +	 * requirements and theoretically should not return errors, the
> +	 * following scenarios may still cause the waiting queue to
> +	 * become full:
> +	 * High-concurrency scenarios:
> +	 * If the sender (calling nbl_chan_send_msg()) generates messages
> +	 * at a rate far exceeding the receiver's ability to process
> +	 * acknowledgments (ACKs),the waiting queue may become fully occupied.
> +	 * Delayed or failed ACK handling by the receiver:
> +	 * The receiver may fail to send ACKs in a timely manner due to
> +	 * processing delays, blocking, or faults, causing the sender's
> +	 * waiting queue slots to remain occupied for an extended period.
> +	 */
> +	return -EAGAIN;
> +}
> +
> +static int nbl_chan_send_msg(struct nbl_channel_mgt *chan_mgt,
> +			     struct nbl_chan_send_info *chan_send)
> +{
> +	struct nbl_common_info *common = chan_mgt->common;
> +	struct nbl_chan_waitqueue_head *wait_head;
> +	struct nbl_chan_tx_param tx_param = { 0 };
> +	u16 msgid = 0;
> +	int i = NBL_CHAN_TX_WAIT_ACK_TIMES, ret;
> +	struct nbl_chan_info *chan_info =
> +		chan_mgt->chan_info[NBL_CHAN_TYPE_MAILBOX];
> +	struct device *dev = common->dev;
> +
> +	if (chan_send->resp_len > NBL_CHAN_BUF_LEN) {
> +		dev_err(dev, "resp_len %zu exceeds max %d\n",
> +			chan_send->resp_len, NBL_CHAN_BUF_LEN);
> +		return -EINVAL;
> +	}
> +
> +	/* Reject new send if shutdown already triggered */
> +	if (READ_ONCE(chan_info->shutdown))
> +		return -ESHUTDOWN;
> +	mutex_lock(&chan_info->txq_lock);
> +
> +	if (test_bit(NBL_CHAN_ABNORMAL, chan_info->state)) {
> +		mutex_unlock(&chan_info->txq_lock);
> +		return -EIO;
> +	}
> +	ret = nbl_chan_get_msg_id(chan_info, &msgid);
> +	if (ret) {
> +		mutex_unlock(&chan_info->txq_lock);
> +		dev_err(dev,
> +			"Channel tx wait head full, send msgtype:%u to dstid:%u failed\n",
> +			chan_send->msg_type, chan_send->dstid);
> +		return ret;
> +	}
> +
> +	tx_param.msg_type = chan_send->msg_type;
> +	tx_param.arg = chan_send->arg;
> +	tx_param.arg_len = chan_send->arg_len;
> +	tx_param.dstid = chan_send->dstid;
> +	tx_param.msgid = msgid;
> +
> +	ret = nbl_chan_update_txqueue(chan_mgt, chan_info, &tx_param);
> +	if (ret) {
> +		mutex_unlock(&chan_info->txq_lock);
> +		dev_err(dev,
> +			"Channel tx queue full, send msgtype:%u to dstid:%u failed\n",
> +			chan_send->msg_type, chan_send->dstid);
> +		return ret;
> +	}
> +
> +	wait_head =
> +		&chan_info->wait[FIELD_GET(NBL_CHAN_MSGID_LOC_MASK, msgid)];
> +	spin_lock_irq(&wait_head->status_lock);
> +	wait_head->acked = 0;
> +	wait_head->ack_data = chan_send->resp;
> +	wait_head->ack_data_len = chan_send->resp_len;
> +	wait_head->msg_type = chan_send->msg_type;
> +	wait_head->msg_index = FIELD_GET(NBL_CHAN_MSGID_INDEX_MASK, msgid);
> +	wait_head->dstid = chan_send->dstid;
> +	/* Ensure all fields above are visible before status update, so receiver
> +	 * won't see WAITING with stale data
> +	 */
> +	smp_wmb();
> +	wait_head->status = chan_send->ack ? NBL_MBX_STATUS_WAITING :
> +					     NBL_MBX_STATUS_IDLE;
> +	spin_unlock_irq(&wait_head->status_lock);
> +
> +	atomic_inc(&chan_info->inflight_tx_cnt);
> +	ret = nbl_chan_kick_tx_ring(chan_mgt, chan_info);
> +	if (ret) {
> +		mutex_lock(&chan_info->txq_lock);

coccinelle says:

@@ -94,0 +95 @@
+/srv/nipa-builds-contest/testing/wt-cocci/drivers/net/ethernet/nebula-matrix/nbl/nbl_channel/nbl_channel.c:763:1-11: second lock on line 813

this will deadlock.

Note that sashiko has more comments:

https://sashiko.dev/#/patchset/20260708064742.35391-1-illusion.wang%40nebula-matrix.com

/P


  reply	other threads:[~2026-07-09  8:09 UTC|newest]

Thread overview: 17+ messages / expand[flat|nested]  mbox.gz  Atom feed  top
2026-07-08  6:47 [PATCH v21 net-next 00/12] nbl driver for Nebulamatrix NICs illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 01/12] net/nebula-matrix: add minimum nbl build framework illusion.wang
2026-07-08 15:28   ` Breno Leitao
2026-07-10  2:45     ` 回复:[PATCH " Illusion Wang
2026-07-10  8:20       ` Breno Leitao
2026-07-08  6:47 ` [PATCH v21 net-next 02/12] net/nebula-matrix: add core driver architecture and HW layer initialization illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 03/12] net/nebula-matrix: add channel wire opcode enum definitions illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 04/12] net/nebula-matrix: add channel layer illusion.wang
2026-07-09  8:09   ` Paolo Abeni [this message]
2026-07-08  6:47 ` [PATCH v21 net-next 05/12] net/nebula-matrix: add common resource implementation illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 06/12] net/nebula-matrix: add intr " illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 07/12] net/nebula-matrix: add vsi " illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 08/12] net/nebula-matrix: dispatch: add control-level routing core infrastructure illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 09/12] net/nebula-matrix: dispatch: add cross-version channel message framework illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 10/12] net/nebula-matrix: dispatch: add resource ops lock protection illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 11/12] net/nebula-matrix: add common/ctrl dev init/remove operation illusion.wang
2026-07-08  6:47 ` [PATCH v21 net-next 12/12] net/nebula-matrix: add common dev start/stop operation illusion.wang

Reply instructions:

You may reply publicly to this message via plain-text email
using any one of the following methods:

* Save the following mbox file, import it into your mail client,
  and reply-to-all from there: mbox

  Avoid top-posting and favor interleaved quoting:
  https://en.wikipedia.org/wiki/Posting_style#Interleaved_style

* Reply using the --to, --cc, and --in-reply-to
  switches of git-send-email(1):

  git send-email \
    --in-reply-to=698d9e2d-70e0-40e7-9398-e39ea681245a@redhat.com \
    --to=pabeni@redhat.com \
    --cc=alvin.wang@nebula-matrix.com \
    --cc=andrew+netdev@lunn.ch \
    --cc=corbet@lwn.net \
    --cc=dimon.zhao@nebula-matrix.com \
    --cc=edumazet@google.com \
    --cc=enelsonmoore@gmail.com \
    --cc=hkallweit1@gmail.com \
    --cc=horms@kernel.org \
    --cc=illusion.wang@nebula-matrix.com \
    --cc=kuba@kernel.org \
    --cc=linux-doc@vger.kernel.org \
    --cc=linux-kernel@vger.kernel.org \
    --cc=lukas.bulwahn@redhat.com \
    --cc=netdev@vger.kernel.org \
    --cc=sam.chen@nebula-matrix.com \
    --cc=skhan@linuxfoundation.org \
    --cc=vadim.fedorenko@linux.dev \
    /path/to/YOUR_REPLY

  https://kernel.org/pub/software/scm/git/docs/git-send-email.html

* If your mail client supports setting the In-Reply-To header
  via mailto: links, try the mailto: link
Be sure your reply has a Subject: header at the top and a blank line before the message body.
This is an external index of several public inboxes,
see mirroring instructions on how to clone and mirror
all data and code used by this external index.