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 */ 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 = netfs_folioq_pool.alloc(gfp, netfs_folioq_pool.pool_data); 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 */ 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 */ 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 */ 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 */ 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 */ 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 */ 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 */ 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