1 // SPDX-License-Identifier: GPL-2.0-or-later
2 /* Rolling buffer helpers
3 *
4 * Copyright (C) 2024 Red Hat, Inc. All Rights Reserved.
5 * Written by David Howells (dhowells@redhat.com)
6 */
7
8 #include <linux/bitops.h>
9 #include <linux/mempool.h>
10 #include <linux/pagemap.h>
11 #include <linux/rolling_buffer.h>
12 #include <linux/slab.h>
13 #include "internal.h"
14
15 static atomic_t debug_ids;
16
17 /**
18 * netfs_folioq_alloc - Allocate a folio_queue struct
19 * @rreq_id: Associated debugging ID for tracing purposes
20 * @gfp: Allocation constraints
21 * @trace: Trace tag to indicate the purpose of the allocation
22 *
23 * Allocate, initialise and account the folio_queue struct and log a trace line
24 * to mark the allocation.
25 */
netfs_folioq_alloc(unsigned int rreq_id,gfp_t gfp,unsigned int trace)26 struct folio_queue *netfs_folioq_alloc(unsigned int rreq_id, gfp_t gfp,
27 unsigned int /*enum netfs_folioq_trace*/ trace)
28 {
29 struct folio_queue *fq;
30
31 if (gfp == GFP_KERNEL)
32 fq = mempool_alloc_noreserve(&netfs_folioq_pool, gfp);
33 else
34 fq = mempool_alloc(&netfs_folioq_pool, gfp);
35 if (fq) {
36 netfs_stat(&netfs_n_folioq);
37 folioq_init(fq, rreq_id);
38 fq->debug_id = atomic_inc_return(&debug_ids);
39 trace_netfs_folioq(fq, trace);
40 }
41 return fq;
42 }
43 EXPORT_SYMBOL(netfs_folioq_alloc);
44
45 /**
46 * netfs_folioq_free - Free a folio_queue struct
47 * @folioq: The object to free
48 * @trace: Trace tag to indicate which free
49 *
50 * Free and unaccount the folio_queue struct.
51 */
netfs_folioq_free(struct folio_queue * folioq,unsigned int trace)52 void netfs_folioq_free(struct folio_queue *folioq,
53 unsigned int /*enum netfs_trace_folioq*/ trace)
54 {
55 trace_netfs_folioq(folioq, trace);
56 netfs_stat_d(&netfs_n_folioq);
57 mempool_free(folioq, &netfs_folioq_pool);
58 }
59 EXPORT_SYMBOL(netfs_folioq_free);
60
61 /*
62 * Initialise a rolling buffer. We allocate an empty folio queue struct to so
63 * that the pointers can be independently driven by the producer and the
64 * consumer.
65 */
rolling_buffer_init(struct rolling_buffer * roll,unsigned int rreq_id,unsigned int direction,gfp_t gfp)66 int rolling_buffer_init(struct rolling_buffer *roll, unsigned int rreq_id,
67 unsigned int direction, gfp_t gfp)
68 {
69 struct folio_queue *fq;
70
71 fq = netfs_folioq_alloc(rreq_id, gfp, netfs_trace_folioq_rollbuf_init);
72 if (!fq)
73 return -ENOMEM;
74
75 roll->head = fq;
76 roll->tail = fq;
77 iov_iter_folio_queue(&roll->iter, direction, fq, 0, 0, 0);
78 return 0;
79 }
80
81 /*
82 * Add another folio_queue to a rolling buffer if there's no space left.
83 */
rolling_buffer_make_space(struct rolling_buffer * roll,gfp_t gfp)84 int rolling_buffer_make_space(struct rolling_buffer *roll, gfp_t gfp)
85 {
86 struct folio_queue *fq, *head = roll->head;
87
88 if (!folioq_full(head))
89 return 0;
90
91 fq = netfs_folioq_alloc(head->rreq_id, gfp, netfs_trace_folioq_make_space);
92 if (!fq)
93 return -ENOMEM;
94 fq->prev = head;
95
96 roll->head = fq;
97 if (folioq_full(head)) {
98 /* Make sure we don't leave the master iterator pointing to a
99 * block that might get immediately consumed.
100 */
101 if (roll->iter.folioq == head &&
102 roll->iter.folioq_slot == folioq_nr_slots(head)) {
103 roll->iter.folioq = fq;
104 roll->iter.folioq_slot = 0;
105 }
106 }
107
108 /* Make sure the initialisation is stored before the next pointer.
109 *
110 * [!] NOTE: After we set head->next, the consumer is at liberty to
111 * immediately delete the old head.
112 */
113 smp_store_release(&head->next, fq);
114 return 0;
115 }
116
117 /*
118 * Decant the entire list of folios to read into a rolling buffer.
119 */
rolling_buffer_bulk_load_from_ra(struct rolling_buffer * roll,struct readahead_control * ractl,unsigned int rreq_id,gfp_t gfp)120 ssize_t rolling_buffer_bulk_load_from_ra(struct rolling_buffer *roll,
121 struct readahead_control *ractl,
122 unsigned int rreq_id, gfp_t gfp)
123 {
124 struct folio_queue *fq;
125 ssize_t loaded = 0;
126
127 while (ractl->_nr_pages - ractl->_batch_count > 0) {
128 unsigned int nr;
129
130 /* Allocate a folioq to put some folios into and attach it to
131 * the rolling buffer.
132 */
133 fq = netfs_folioq_alloc(rreq_id, gfp,
134 netfs_trace_folioq_make_space);
135 if (!fq)
136 goto nomem_unlock;
137 fq->prev = roll->head;
138 if (!roll->tail)
139 roll->tail = fq;
140 else
141 roll->head->next = fq;
142 roll->head = fq;
143
144 /* Get a batch of folios and note their orders. */
145 nr = __readahead_batch(ractl, (struct page **)fq->vec.folios,
146 folioq_nr_slots(fq));
147 if (WARN_ON_ONCE(!nr))
148 break;
149 fq->vec.nr = nr;
150
151 for (int slot = 0; slot < nr; slot++) {
152 struct folio *folio = folioq_folio(fq, slot);
153 unsigned int order;
154
155 order = folio_order(folio);
156 fq->orders[slot] = order;
157 loaded += PAGE_SIZE << order;
158 trace_netfs_folio(folio, netfs_folio_trace_read);
159 }
160 }
161
162 WRITE_ONCE(roll->iter.count, loaded);
163 iov_iter_folio_queue(&roll->iter, ITER_DEST, roll->tail, 0, 0, loaded);
164 return loaded;
165
166 nomem_unlock:
167 for (fq = roll->tail; fq; fq = fq->next) {
168 for (int slot = 0; slot < folioq_count(fq); slot++) {
169 folio_unlock(fq->vec.folios[slot]);
170 folioq_mark(fq, slot);
171 }
172 }
173 rolling_buffer_clear(roll);
174 roll->head = NULL;
175 roll->tail = NULL;
176 return -ENOMEM;
177 }
178
179 /*
180 * Append a folio to the rolling buffer.
181 */
rolling_buffer_append(struct rolling_buffer * roll,struct folio * folio,unsigned int flags,gfp_t gfp)182 ssize_t rolling_buffer_append(struct rolling_buffer *roll, struct folio *folio,
183 unsigned int flags, gfp_t gfp)
184 {
185 ssize_t size = folio_size(folio);
186 int slot;
187
188 if (rolling_buffer_make_space(roll, gfp) < 0)
189 return -ENOMEM;
190
191 slot = folioq_append(roll->head, folio);
192 if (flags & ROLLBUF_MARK_1)
193 folioq_mark(roll->head, slot);
194 if (flags & ROLLBUF_MARK_2)
195 folioq_mark2(roll->head, slot);
196
197 WRITE_ONCE(roll->iter.count, roll->iter.count + size);
198
199 /* Store the counter after setting the slot. */
200 smp_store_release(&roll->next_head_slot, slot);
201 return size;
202 }
203
204 /*
205 * Delete a spent buffer from a rolling queue and return the next in line. We
206 * don't return the last buffer to keep the pointers independent, but return
207 * NULL instead.
208 */
rolling_buffer_delete_spent(struct rolling_buffer * roll)209 struct folio_queue *rolling_buffer_delete_spent(struct rolling_buffer *roll)
210 {
211 struct folio_queue *spent = roll->tail, *next = READ_ONCE(spent->next);
212
213 if (!next)
214 return NULL;
215 next->prev = NULL;
216 netfs_folioq_free(spent, netfs_trace_folioq_delete);
217 roll->tail = next;
218 return next;
219 }
220
221 /*
222 * Clear out a rolling queue. Folios that have mark 1 set are put.
223 */
rolling_buffer_clear(struct rolling_buffer * roll)224 void rolling_buffer_clear(struct rolling_buffer *roll)
225 {
226 struct folio_batch fbatch;
227 struct folio_queue *p;
228
229 folio_batch_init(&fbatch);
230
231 while ((p = roll->tail)) {
232 roll->tail = p->next;
233 for (int slot = 0; slot < folioq_count(p); slot++) {
234 struct folio *folio = folioq_folio(p, slot);
235
236 if (!folio)
237 continue;
238 if (folioq_is_marked(p, slot)) {
239 trace_netfs_folio(folio, netfs_folio_trace_put);
240 if (!folio_batch_add(&fbatch, folio))
241 folio_batch_release(&fbatch);
242 }
243 }
244
245 netfs_folioq_free(p, netfs_trace_folioq_clear);
246 }
247
248 folio_batch_release(&fbatch);
249 }
250