Alexander Lobakin <aleksander.lobakin@xxxxxxxxx> writes:
> The main reason for this change was to allow mixing pages from different
> &page_pools within one &xdp_buff/&xdp_frame. Why not?
> Adjust xdp_return_frame_bulk() and page_pool_put_page_bulk(), so that
> they won't be tied to a particular pool. Let the latter create a
> separate bulk of pages which's PP is different and flush it recursively.
> This greatly optimizes xdp_return_frame_bulk(): no more hashtable
> lookups. Also make xdp_flush_frame_bulk() inline, as it's just one if +
> function call + one u32 read, not worth extending the call ladder.
>
> Signed-off-by: Alexander Lobakin <aleksander.lobakin@xxxxxxxxx>
Neat idea, but one comment, see below:
[...]
> /**
> * page_pool_put_page_bulk() - release references on multiple pages
> - * @pool: pool from which pages were allocated
> * @data: array holding page pointers
> * @count: number of pages in @data
> + * @rec: whether it's called recursively by itself
> *
> * Tries to refill a number of pages into the ptr_ring cache holding ptr_ring
> * producer lock. If the ptr_ring is full, page_pool_put_page_bulk()
> @@ -854,21 +865,43 @@ EXPORT_SYMBOL(page_pool_put_unrefed_page);
> * Please note the caller must not use data area after running
> * page_pool_put_page_bulk(), as this function overwrites it.
> */
> -void page_pool_put_page_bulk(struct page_pool *pool, struct page **data,
> - u32 count)
> +void page_pool_put_page_bulk(struct page **data, u32 count, bool rec)
> {
> + struct page_pool *pool = NULL;
> + struct xdp_frame_bulk sub;
> int i, bulk_len = 0;
> bool allow_direct;
> bool in_softirq;
>
> - allow_direct = page_pool_napi_local(pool);
> + xdp_frame_bulk_init(&sub);
>
> for (i = 0; i < count; i++) {
> - netmem_ref netmem = page_to_netmem(compound_head(data[i]));
> + struct page *page;
> + netmem_ref netmem;
> +
> + if (!rec) {
> + page = compound_head(data[i]);
> + netmem = page_to_netmem(page);
>
> - /* It is not the last user for the page frag case */
> - if (!page_pool_is_last_ref(netmem))
> + /* It is not the last user for the page frag case */
> + if (!page_pool_is_last_ref(netmem))
> + continue;
> + } else {
> + page = data[i];
> + netmem = page_to_netmem(page);
> + }
> +
> + if (unlikely(!pool)) {
> + pool = page->pp;
> + allow_direct = page_pool_napi_local(pool);
> + } else if (page->pp != pool) {
> + /*
> + * If the page belongs to a different page_pool, save
> + * it to handle recursively after the main loop.
> + */
> + page_pool_bulk_rec_add(&sub, page);
> continue;
> + }
>
> netmem = __page_pool_put_page(pool, netmem, -1, allow_direct);
> /* Approved for bulk recycling in ptr_ring cache */
> @@ -876,6 +909,9 @@ void page_pool_put_page_bulk(struct page_pool *pool, struct page **data,
> data[bulk_len++] = (__force void *)netmem;
> }
>
> + if (sub.count)
> + page_pool_put_page_bulk(sub.q, sub.count, true);
> +
In the worst case here, this function can recursively call itself
XDP_BULK_QUEUE_SIZE (=16) times. Which will blow ~2.5k of stack size,
and lots of function call overhead. I'm not saying this level of
recursion is likely to happen today, but who knows about future uses? So
why not make it iterative instead of recursive (same basic idea, but
some kind of 'goto begin', or loop, instead of the recursive call)?
Something like:
void page_pool_put_page_bulk(void **data, int count)
{
struct page *bulk_prod[XDP_BULK_QUEUE_SIZE];
int page_count = 0, pages_left, bulk_len, i;
bool allow_direct;
bool in_softirq;
for (i = 0; i < count; i++) {
struct page *p = compound_head(data[i]));
if (page_pool_is_last_ref(page_to_netmem(p)))
data[page_count++] = p;
}
begin:
pool = data[0]->pp;
allow_direct = page_pool_napi_local(pool);
pages_left = 0;
bulk_len = 0;
for (i = 0; i < page_count; i++) {
struct page *p = data[i];
netmem_ref netmem;
if (unlikely(p->pp != pool)) {
data[pages_left++] = p;
continue;
}
netmem = __page_pool_put_page(pool, page_to_netmem(p), -1, allow_direct);
/* Approved for bulk recycling in ptr_ring cache */
if (netmem)
bulk_prod[bulk_len++] = (__force void *)netmem;
}
if (!bulk_len)
goto out;
/* Bulk producer into ptr_ring page_pool cache */
in_softirq = page_pool_producer_lock(pool);
for (i = 0; i < bulk_len; i++) {
if (__ptr_ring_produce(&pool->ring, bulk_prod[i])) {
/* ring full */
recycle_stat_inc(pool, ring_full);
break;
}
}
recycle_stat_add(pool, ring, i);
page_pool_producer_unlock(pool, in_softirq);
/* Hopefully all pages was return into ptr_ring */
if (likely(i == bulk_len))
goto out;
/* ptr_ring cache full, free remaining pages outside producer lock
* since put_page() with refcnt == 1 can be an expensive operation
*/
for (; i < bulk_len; i++)
page_pool_return_page(pool, (__force netmem_ref)bulk_prod[i]);
out:
if (pages_left) {
page_count = pages_left;
goto begin;
}
}
Personally I also think this is easier to read, and it gets rid of the
'rec' function parameter wart... :)
-Toke
