// SPDX-License-Identifier: GPL-2.0-or-later/* Rolling buffer helpers * * Copyright (C) 2024 Red Hat, Inc. All Rights Reserved. * Written by David Howells (dhowells@redhat.com) */ #include <linux/bitops.h>#include <linux/mempool.h>#include <linux/pagemap.h>#include <linux/rolling_buffer.h>#include <linux/slab.h>#include "internal.h" static atomic_t debug_ids; /** * netfs_folioq_alloc - Allocate a folio_queue struct * @rreq_id: Associated debugging ID for tracing purposes * @gfp: Allocation constraints * @trace: Trace tag to indicate the purpose of the allocation * * Allocate, initialise and account the folio_queue struct and log a trace line * to mark the allocation. */struct folio_queue *netfs_folioq_alloc(unsigned int rreq_id, gfp_t gfp,				       unsigned int /*enum netfs_folioq_trace*/ trace){	struct folio_queue *fq; 	if (gfp == GFP_KERNEL)		fq = netfs_folioq_pool.alloc(gfp, netfs_folioq_pool.pool_data);	else		fq = mempool_alloc(&netfs_folioq_pool, gfp);	if (fq) {		netfs_stat(&netfs_n_folioq);		folioq_init(fq, rreq_id);		fq->debug_id = atomic_inc_return(&debug_ids);		trace_netfs_folioq(fq, trace);	}	return fq;}EXPORT_SYMBOL(netfs_folioq_alloc); /** * netfs_folioq_free - Free a folio_queue struct * @folioq: The object to free * @trace: Trace tag to indicate which free * * Free and unaccount the folio_queue struct. */void netfs_folioq_free(struct folio_queue *folioq,		       unsigned int /*enum netfs_trace_folioq*/ trace){	trace_netfs_folioq(folioq, trace);	netfs_stat_d(&netfs_n_folioq);	mempool_free(folioq, &netfs_folioq_pool);}EXPORT_SYMBOL(netfs_folioq_free); /* * Initialise a rolling buffer.  We allocate an empty folio queue struct to so * that the pointers can be independently driven by the producer and the * consumer. */int rolling_buffer_init(struct rolling_buffer *roll, unsigned int rreq_id,			unsigned int direction, gfp_t gfp){	struct folio_queue *fq; 	fq = netfs_folioq_alloc(rreq_id, gfp, netfs_trace_folioq_rollbuf_init);	if (!fq)		return -ENOMEM; 	roll->head = fq;	roll->tail = fq;	iov_iter_folio_queue(&roll->iter, direction, fq, 0, 0, 0);	return 0;} /* * Add another folio_queue to a rolling buffer if there's no space left. */int rolling_buffer_make_space(struct rolling_buffer *roll, gfp_t gfp){	struct folio_queue *fq, *head = roll->head; 	if (!folioq_full(head))		return 0; 	fq = netfs_folioq_alloc(head->rreq_id, gfp, netfs_trace_folioq_make_space);	if (!fq)		return -ENOMEM;	fq->prev = head; 	roll->head = fq;	if (folioq_full(head)) {		/* Make sure we don't leave the master iterator pointing to a		 * block that might get immediately consumed.		 */		if (roll->iter.folioq == head &&		    roll->iter.folioq_slot == folioq_nr_slots(head)) {			roll->iter.folioq = fq;			roll->iter.folioq_slot = 0;		}	} 	/* Make sure the initialisation is stored before the next pointer.	 *	 * [!] NOTE: After we set head->next, the consumer is at liberty to	 * immediately delete the old head.	 */	smp_store_release(&head->next, fq);	return 0;} /* * Decant the entire list of folios to read into a rolling buffer. */ssize_t rolling_buffer_bulk_load_from_ra(struct rolling_buffer *roll,					 struct readahead_control *ractl,					 unsigned int rreq_id, gfp_t gfp){	struct folio_queue *fq;	ssize_t loaded = 0; 	while (ractl->_nr_pages - ractl->_batch_count > 0) {		unsigned int nr; 		/* Allocate a folioq to put some folios into and attach it to		 * the rolling buffer.		 */		fq = netfs_folioq_alloc(rreq_id, gfp,					netfs_trace_folioq_make_space);		if (!fq)			goto nomem_unlock;		fq->prev = roll->head;		if (!roll->tail)			roll->tail = fq;		else			roll->head->next = fq;		roll->head = fq; 		/* Get a batch of folios and note their orders. */		nr = __readahead_batch(ractl, (struct page **)fq->vec.folios,				       folioq_nr_slots(fq));		if (WARN_ON_ONCE(!nr))			break;		fq->vec.nr = nr; 		for (int slot = 0; slot < nr; slot++) {			struct folio *folio = folioq_folio(fq, slot);			unsigned int order; 			order = folio_order(folio);			fq->orders[slot] = order;			loaded += PAGE_SIZE << order;			trace_netfs_folio(folio, netfs_folio_trace_read);		}	} 	WRITE_ONCE(roll->iter.count, loaded);	iov_iter_folio_queue(&roll->iter, ITER_DEST, roll->tail, 0, 0, loaded);	return loaded; nomem_unlock:	for (fq = roll->tail; fq; fq = fq->next) {		for (int slot = 0; slot < folioq_count(fq); slot++) {			folio_unlock(fq->vec.folios[slot]);			folioq_mark(fq, slot);		}	}	rolling_buffer_clear(roll);	roll->head = NULL;	roll->tail = NULL;	return -ENOMEM;} /* * Append a folio to the rolling buffer. */ssize_t rolling_buffer_append(struct rolling_buffer *roll, struct folio *folio,			      unsigned int flags, gfp_t gfp){	ssize_t size = folio_size(folio);	int slot; 	if (rolling_buffer_make_space(roll, gfp) < 0)		return -ENOMEM; 	slot = folioq_append(roll->head, folio);	if (flags & ROLLBUF_MARK_1)		folioq_mark(roll->head, slot);	if (flags & ROLLBUF_MARK_2)		folioq_mark2(roll->head, slot); 	WRITE_ONCE(roll->iter.count, roll->iter.count + size); 	/* Store the counter after setting the slot. */	smp_store_release(&roll->next_head_slot, slot);	return size;} /* * Delete a spent buffer from a rolling queue and return the next in line.  We * don't return the last buffer to keep the pointers independent, but return * NULL instead. */struct folio_queue *rolling_buffer_delete_spent(struct rolling_buffer *roll){	struct folio_queue *spent = roll->tail, *next = READ_ONCE(spent->next); 	if (!next)		return NULL;	next->prev = NULL;	netfs_folioq_free(spent, netfs_trace_folioq_delete);	roll->tail = next;	return next;} /* * Clear out a rolling queue.  Folios that have mark 1 set are put. */void rolling_buffer_clear(struct rolling_buffer *roll){	struct folio_batch fbatch;	struct folio_queue *p; 	folio_batch_init(&fbatch); 	while ((p = roll->tail)) {		roll->tail = p->next;		for (int slot = 0; slot < folioq_count(p); slot++) {			struct folio *folio = folioq_folio(p, slot); 			if (!folio)				continue;			if (folioq_is_marked(p, slot)) {				trace_netfs_folio(folio, netfs_folio_trace_put);				if (!folio_batch_add(&fbatch, folio))					folio_batch_release(&fbatch);			}		} 		netfs_folioq_free(p, netfs_trace_folioq_clear);	} 	folio_batch_release(&fbatch);}