[PATCH net-next v1 06/11] page_pool: Coordinate refill with memory providers

@@ -136,6 +136,16 @@ Driver RX Requirements
    circulation times, such as when userspace holds a reference in zerocopy
    scenarios.
 
+8. After each RX ring refill, call page_pool_refill_done() once per page
+   pool, with PP_REFILL_FULL only if all rings that use the pool are full.
+   Rings of a queue that take buffers from one such provider share one
+   page pool. Keep buffers in each of them, first in a ring that every
+   packet needs, like a header ring, so that a provider with few buffers
+   does not leave them all in one ring. If the call returns false, poll
+   again soon, for example by returning the full NAPI budget. Register the
+   NAPI of each RX queue with netif_queue_set_napi(); the core schedules
+   it when provider buffers become available.
+
 Driver TX Requirements
 ======================
 

@@ -33,6 +33,12 @@ struct netdev_rx_queue {
 	struct pp_memory_provider_params mp_params;
 	/* stamps the page pools of each queue allocation */
 	u32				mp_gen;
+	/* NAPI of the running queue, for memory provider wakeups. Set while
+	 * mp_armed, and if the provider has a refill_done op. Written under
+	 * the ops lock, read under RCU, see netif_mp_wake().
+	 */
+	struct napi_struct __rcu	*mp_napi;
+	bool				mp_armed;
 
 	/* If a queue is leased, then the lease pointer is always
 	 * valid. From the physical device it points to the virtual

@@ -523,6 +523,30 @@ static inline bool page_pool_is_unreadable(struct page_pool *pool)
 	return pool->mp_ops && !(pool->mp_ops->caps & MP_CAP_READABLE);
 }
 
+/**
+ * page_pool_refill_done() - report the end of an RX ring refill
+ * @pool: page pool of the rings
+ * @state: whether all rings that take buffers from @pool are full
+ *
+ * A memory provider with a fixed set of buffers must know when the rings
+ * are full, and when they wait for buffers. Call this once per page pool
+ * after each refill, also when the rings were full already.
+ *
+ * Context: the NAPI poll that allocates from @pool, or the driver while that
+ * NAPI is disabled, with IRQs enabled.
+ * Return: true if the refill is done, false if NAPI should retry it soon.
+ */
+static inline bool page_pool_refill_done(struct page_pool *pool,
+					 enum pp_refill state)
+{
+	const struct memory_provider_ops *ops = pool->mp_ops;
+
+	if (ops && ops->refill_done)
+		return ops->refill_done(pool, state);
+
+	return true;
+}
+
 /**
  * page_pool_supports_frag() - may page pool buffers be split
  * @pool: queried page pool

@@ -42,6 +42,7 @@ struct memory_provider_ops {
 	int (*nl_fill)(void *mp_priv, struct sk_buff *rsp,
 		       struct netdev_rx_queue *rxq);
 	void (*uninstall)(void *mp_priv, struct netdev_rx_queue *rxq);
+	bool (*refill_done)(struct page_pool *pool, enum pp_refill state);
 	u32 caps;
 };
 

@@ -55,6 +56,7 @@ int netif_mp_open_rxq(struct net_device *dev, unsigned int rxq_idx,
 int netif_mp_close_rxq(struct net_device *dev, unsigned int rxq_idx,
 		       const struct pp_memory_provider_params *old_p);
 bool netif_all_mps_have_cap(struct net_device *dev, u32 cap);
+bool netif_mp_wake(struct netdev_rx_queue *rxq);
 
 /**
   * net_mp_netmem_place_in_cache() - give a netmem to a page pool

@@ -40,6 +40,12 @@
  */
 #define PP_FLAG_MEM_PROVIDER	BIT(4)
 
+/* Ring state after a refill, see page_pool_refill_done() */
+enum pp_refill {
+	PP_REFILL_PARTIAL,	/* some ring of the pool has room left */
+	PP_REFILL_FULL,		/* all rings of the pool are full */
+};
+
 /* RX buffers a page pool supplies, see page_pool_params.rx_roles */
 enum pp_rx_role {
 	PP_RX_ROLE_HEAD		= BIT(0),	/* packet headers */

@@ -1771,6 +1771,7 @@ static int __dev_open(struct net_device *dev, struct netlink_ext_ack *extack)
 		dev_set_rx_mode(dev);
 		dev_activate(dev);
 		add_device_randomness(dev->dev_addr, dev->addr_len);
+		netif_mp_arm_dev(dev);
 	}
 
 	return ret;

@@ -1833,6 +1834,7 @@ static void __dev_close_many(struct list_head *head)
 
 		netdev_assert_locked_ops_compat(dev);
 
+		netif_mp_disarm_dev(dev);
 		if (ops->ndo_stop)
 			ops->ndo_stop(dev);
 

@@ -7559,6 +7561,7 @@ void netif_queue_set_napi(struct net_device *dev, unsigned int queue_index,
 	case NETDEV_QUEUE_TYPE_RX:
 		rxq = __netif_get_rx_queue(dev, queue_index);
 		rxq->napi = napi;
+		netif_mp_update_napi(rxq);
 		return;
 	case NETDEV_QUEUE_TYPE_TX:
 		txq = netdev_get_tx_queue(dev, queue_index);

@@ -7960,6 +7963,7 @@ void __netif_napi_del_locked(struct napi_struct *napi)
 	}
 
 	list_del_rcu(&napi->dev_list);
+	netif_mp_napi_del(napi);
 	napi_free_frags(napi);
 
 	gro_cleanup(&napi->gro);

@@ -125,6 +125,10 @@ bool netif_is_queue_leasee(const struct net_device *dev);
 
 void __netif_mp_uninstall_rxq(struct netdev_rx_queue *rxq,
 			      const struct pp_memory_provider_params *p);
+void netif_mp_update_napi(struct netdev_rx_queue *rxq);
+void netif_mp_arm_dev(struct net_device *dev);
+void netif_mp_disarm_dev(struct net_device *dev);
+void netif_mp_napi_del(struct napi_struct *napi);
 
 void netif_rxq_cleanup_unlease(struct netdev_rx_queue *phys_rxq,
 			       struct netdev_rx_queue *virt_rxq);

@@ -116,6 +116,94 @@ bool netif_all_mps_have_cap(struct net_device *dev, u32 cap)
 	return true;
 }
 
+/**
+ * netif_mp_wake - schedule the NAPI that refills a memory provider queue
+ * @rxq: physical RX queue of the provider
+ *
+ * Context: IRQs enabled, not in hardirq. The caller keeps the device of
+ * @rxq alive.
+ *
+ * Return: false if no NAPI refills the queue.
+ */
+bool netif_mp_wake(struct netdev_rx_queue *rxq)
+{
+	struct napi_struct *napi;
+	bool running = false;
+
+	rcu_read_lock();
+	napi = rcu_dereference(rxq->mp_napi);
+	if (napi) {
+		local_bh_disable();
+		napi_schedule(napi);
+		local_bh_enable();
+		running = true;
+	}
+	rcu_read_unlock();
+
+	return running;
+}
+
+/* Point wakeups at the NAPI of the running queue, and schedule it once: a
+ * refill during the start may have found no buffers.
+ */
+void netif_mp_update_napi(struct netdev_rx_queue *rxq)
+{
+	const struct memory_provider_ops *ops = rxq->mp_params.mp_ops;
+	struct napi_struct *napi = NULL;
+
+	if (rxq->mp_armed && ops && ops->refill_done)
+		napi = rxq->napi;
+	if (rcu_access_pointer(rxq->mp_napi) == napi)
+		return;
+
+	rcu_assign_pointer(rxq->mp_napi, napi);
+	if (napi)
+		netif_mp_wake(rxq);
+}
+
+static void netif_mp_arm_napi(struct netdev_rx_queue *rxq)
+{
+	rxq->mp_armed = true;
+	netif_mp_update_napi(rxq);
+}
+
+static void netif_mp_disarm_napi(struct netdev_rx_queue *rxq)
+{
+	rxq->mp_armed = false;
+	RCU_INIT_POINTER(rxq->mp_napi, NULL);
+}
+
+void netif_mp_arm_dev(struct net_device *dev)
+{
+	unsigned int i;
+
+	for (i = 0; i < dev->real_num_rx_queues; i++)
+		netif_mp_arm_napi(__netif_get_rx_queue(dev, i));
+}
+
+void netif_mp_disarm_dev(struct net_device *dev)
+{
+	unsigned int i;
+
+	for (i = 0; i < dev->num_rx_queues; i++)
+		netif_mp_disarm_napi(&dev->_rx[i]);
+}
+
+/* The grace period of netif_napi_del() covers wakeups that found @napi. */
+void netif_mp_napi_del(struct napi_struct *napi)
+{
+	struct net_device *dev = napi->dev;
+	unsigned int i;
+
+	/* free_netdev() frees the queues before it deletes the last NAPIs */
+	if (dev->reg_state != NETREG_REGISTERED)
+		return;
+
+	for (i = 0; i < dev->num_rx_queues; i++)
+		if (rcu_access_pointer(dev->_rx[i].mp_napi) == napi)
+			RCU_INIT_POINTER(dev->_rx[i].mp_napi, NULL);
+}
+
 static int netdev_rx_queue_reconfig(struct net_device *dev,
 				    unsigned int rxq_idx,
 				    struct netdev_queue_config *qcfg_old,

@@ -152,13 +240,15 @@ static int netdev_rx_queue_reconfig(struct net_device *dev,
 		goto err_free_new_queue_mem;
 
 	if (netif_running(dev)) {
+		netif_mp_disarm_napi(rxq);
 		err = qops->ndo_queue_stop(dev, old_mem, rxq_idx);
 		if (err)
-			goto err_free_new_queue_mem;
+			goto err_arm_old_queue;
 
 		err = qops->ndo_queue_start(dev, qcfg_new, new_mem, rxq_idx);
 		if (err)
 			goto err_start_queue;
+		netif_mp_arm_napi(rxq);
 	} else {
 		swap(new_mem, old_mem);
 	}

@@ -183,8 +273,13 @@ static int netdev_rx_queue_reconfig(struct net_device *dev,
 		     "Failed to restart old queue in error path. RX queue %d may be unhealthy.",
 		     rxq_idx);
 		qops->ndo_queue_mem_free(dev, old_mem);
+		goto err_free_new_queue_mem;
 	}
 
+err_arm_old_queue:
+	/* The old queue may run */
+	netif_mp_arm_napi(rxq);
+
 err_free_new_queue_mem:
 	qops->ndo_queue_mem_free(dev, new_mem);
 

@@ -794,6 +794,10 @@ static int xsk_wakeup(struct xdp_sock *xs, u8 flags)
 {
 	struct net_device *dev = xs->dev;
 
+	/* The core wakes the queue for both RX and TX. */
+	if (xs->pool->pp)
+		return xsk_pp_wakeup(xs->pool->pp);
+
 	return dev->netdev_ops->ndo_xsk_wakeup(dev, xs->queue_id, flags);
 }
 

@@ -42,6 +42,7 @@ bool xsk_pp_supported(struct net_device *dev);
 int xsk_pp_bind(struct xsk_buff_pool *pool, struct net_device *dev,
 		u16 queue_id);
 void xsk_pp_unbind(struct xsk_buff_pool *pool);
+int xsk_pp_wakeup(struct xsk_pp *pp);
 int xp_alloc_heads(struct xsk_buff_pool *pool);
 void xsk_clear_pool_at_qid(struct net_device *dev, u16 queue_id);
 int xsk_reg_pool_at_qid(struct net_device *dev, struct xsk_buff_pool *pool,

@@ -36,7 +36,8 @@ struct xsk_pp {
 	u32 nr_pools;
 	bool dma_need_sync;
 	/* RX queue of the provider, NULL once it is removed */
-	struct netdev_rx_queue *rxq;
+	struct netdev_rx_queue __rcu *rxq;
+	bool starved;
 	u8 *state;
 	u32 reuse_cnt;
 	u32 reuse[];

@@ -219,11 +220,31 @@ static netmem_ref xsk_pp_alloc_netmems(struct page_pool *pool, gfp_t gfp)
 	return netmems[n - 1];
 }
 
+static bool xsk_pp_wake(struct xsk_pp *pp)
+{
+	struct netdev_rx_queue *rxq;
+	bool running = false;
+
+	rcu_read_lock();
+	rxq = rcu_dereference(pp->rxq);
+	if (rxq)
+		running = netif_mp_wake(rxq);
+	rcu_read_unlock();
+
+	return running;
+}
+
+int xsk_pp_wakeup(struct xsk_pp *pp)
+{
+	return xsk_pp_wake(pp) ? 0 : -ENETDOWN;
+}
+
 static bool xsk_pp_release_netmem(struct page_pool *pool, netmem_ref netmem)
 {
 	struct net_iov *niov = netmem_to_net_iov(netmem);
 	struct xsk_pp *pp = pool->mp_priv;
 	u32 idx = net_iov_idx(niov);
+	bool starved;
 
 	spin_lock_bh(&pp->lock);
 	net_mp_niov_clear_page_pool(niov);

@@ -232,11 +253,47 @@ static bool xsk_pp_release_netmem(struct page_pool *pool, netmem_ref netmem)
 		pp->state[idx] = XSK_PP_REUSE;
 		pp->reuse[pp->reuse_cnt++] = idx;
 	}
+	starved = pp->starved;
+	WRITE_ONCE(pp->starved, false);
 	spin_unlock_bh(&pp->lock);
 
+	/* A starved queue may wait for this release */
+	if (starved)
+		xsk_pp_wake(pp);
+
 	return false;
 }
 
+static bool xsk_pp_refill_done(struct page_pool *pool, enum pp_refill state)
+{
+	struct xsk_pp *pp = pool->mp_priv;
+	struct xsk_buff_pool *xp = pp->pool;
+	bool wait = xp->uses_need_wakeup;
+	bool more;
+
+	if (state == PP_REFILL_FULL && !READ_ONCE(pp->starved))
+		return true;
+
+	spin_lock_bh(&pp->lock);
+	if (!pp->fq) {
+		spin_unlock_bh(&pp->lock);
+		return true;
+	}
+
+	if (state == PP_REFILL_FULL) {
+		WRITE_ONCE(pp->starved, false);
+		spin_unlock_bh(&pp->lock);
+		return true;
+	}
+
+	/* With need_wakeup, userspace wakes the queue after it posts FILL */
+	more = pp->reuse_cnt || xskq_cons_nb_entries(pp->fq, 1);
+	WRITE_ONCE(pp->starved, !more);
+	spin_unlock_bh(&pp->lock);
+
+	return !more && wait;
+}
+
 static int xsk_pp_nl_fill(void *mp_priv, struct sk_buff *rsp,
 			  struct netdev_rx_queue *rxq)
 {

@@ -246,10 +303,12 @@ static int xsk_pp_nl_fill(void *mp_priv, struct sk_buff *rsp,
 /* The queue no longer uses the provider, and its DMA stopped */
 static void xsk_pp_detach(struct xsk_pp *pp)
 {
-	pp->rxq = NULL;
+	RCU_INIT_POINTER(pp->rxq, NULL);
 	spin_lock_bh(&pp->lock);
 	pp->fq = NULL;
 	spin_unlock_bh(&pp->lock);
+	/* Wait for wakeups that found the queue */
+	synchronize_net();
 	xp_dma_unmap(pp->pool, XSK_PP_DMA_ATTR);
 }
 

@@ -266,6 +325,7 @@ const struct memory_provider_ops xsk_pp_ops = {
 	.release_netmem	= xsk_pp_release_netmem,
 	.nl_fill	= xsk_pp_nl_fill,
 	.uninstall	= xsk_pp_uninstall,
+	.refill_done	= xsk_pp_refill_done,
 	.caps		= MP_CAP_READABLE,
 };
 

@@ -428,17 +488,22 @@ int xsk_pp_bind(struct xsk_buff_pool *pool, struct net_device *dev,
 		return PTR_ERR(pp);
 	}
 
+	rcu_assign_pointer(pp->rxq, __netif_get_rx_queue(dev, queue_id));
 	mp.mp_priv = pp;
 	mp.rx_page_size = pool->chunk_size;
 	mp.rx_headroom = xsk_pool_get_headroom(pool);
 	err = netif_mp_open_rxq(dev, queue_id, &mp, NULL);
 	if (!err) {
-		pp->rxq = __netif_get_rx_queue(dev, queue_id);
 		pool->pp = pp;
 		pool->xdp_zc_max_segs = dev->xdp_zc_max_segs;
+		/* See xsk_pp_refill_done() */
+		if (pool->uses_need_wakeup)
+			xsk_set_rx_need_wakeup(pool);
 		return 0;
 	}
 
+	RCU_INIT_POINTER(pp->rxq, NULL);
+	synchronize_net();
 	xp_dma_unmap(pool, XSK_PP_DMA_ATTR);
 	/* Copy mode takes the chunks that the provider holds */
 	if (!xp_alloc_heads(pool) && xsk_pp_abandon(pp))

@@ -459,7 +524,7 @@ void xsk_pp_unbind(struct xsk_buff_pool *pool)
 	};
 
 	/* Unregister removed the provider already */
-	if (!pp->rxq)
+	if (!rcu_access_pointer(pp->rxq))
 		return;
 
 	netif_mp_close_rxq(pool->netdev, pool->queue_id, &mp);