1 // SPDX-License-Identifier: GPL-2.0-only
2 /* Network filesystem write subrequest result collection, assessment
3 * and retrying.
4 *
5 * Copyright (C) 2024 Red Hat, Inc. All Rights Reserved.
6 * Written by David Howells (dhowells@redhat.com)
7 */
8
9 #include <linux/export.h>
10 #include <linux/fs.h>
11 #include <linux/mm.h>
12 #include <linux/pagemap.h>
13 #include <linux/slab.h>
14 #include "internal.h"
15
16 /* Notes made in the collector */
17 #define HIT_PENDING 0x01 /* A front op was still pending */
18 #define NEED_REASSESS 0x02 /* Need to loop round and reassess */
19 #define MADE_PROGRESS 0x04 /* Made progress cleaning up a stream or the folio set */
20 #define NEED_UNLOCK 0x08 /* The pagecache needs unlocking */
21 #define NEED_RETRY 0x10 /* A front op requests retrying */
22 #define SAW_FAILURE 0x20 /* One stream or hit a permanent failure */
23
netfs_dump_request(const struct netfs_io_request * rreq)24 static void netfs_dump_request(const struct netfs_io_request *rreq)
25 {
26 pr_err("Request R=%08x r=%d fl=%lx or=%x e=%ld\n",
27 rreq->debug_id, refcount_read(&rreq->ref), rreq->flags,
28 rreq->origin, rreq->error);
29 pr_err(" st=%llx tsl=%zx/%llx/%llx\n",
30 rreq->start, rreq->transferred, rreq->submitted, rreq->len);
31 pr_err(" cci=%llx/%llx/%llx\n",
32 rreq->cleaned_to, rreq->collected_to, atomic64_read(&rreq->issued_to));
33 pr_err(" iw=%pSR\n", rreq->netfs_ops->issue_write);
34 for (int i = 0; i < NR_IO_STREAMS; i++) {
35 const struct netfs_io_subrequest *sreq;
36 const struct netfs_io_stream *s = &rreq->io_streams[i];
37
38 pr_err(" str[%x] s=%x e=%d acnf=%u,%u,%u,%u\n",
39 s->stream_nr, s->source, s->error,
40 s->avail, s->active, s->need_retry, s->failed);
41 pr_err(" str[%x] ct=%llx t=%zx\n",
42 s->stream_nr, s->collected_to, s->transferred);
43 list_for_each_entry(sreq, &s->subrequests, rreq_link) {
44 pr_err(" sreq[%x:%x] sc=%u s=%llx t=%zx/%zx r=%d f=%lx\n",
45 sreq->stream_nr, sreq->debug_index, sreq->source,
46 sreq->start, sreq->transferred, sreq->len,
47 refcount_read(&sreq->ref), sreq->flags);
48 }
49 }
50 }
51
52 /*
53 * Successful completion of write of a folio to the server and/or cache. Note
54 * that we are not allowed to lock the folio here on pain of deadlocking with
55 * truncate.
56 */
netfs_folio_written_back(struct folio * folio)57 int netfs_folio_written_back(struct folio *folio)
58 {
59 enum netfs_folio_trace why = netfs_folio_trace_endwb;
60 struct inode *inode = folio_inode(folio);
61 struct netfs_inode *ictx = netfs_inode(inode);
62 struct netfs_folio *finfo;
63 struct netfs_group *group = NULL;
64 int gcount = 0;
65
66 if ((finfo = netfs_folio_info(folio))) {
67 /* Streaming writes cannot be redirtied whilst under writeback,
68 * so discard the streaming record.
69 */
70 uoff_t fend;
71
72 fend = folio_pos(folio) + finfo->dirty_offset + finfo->dirty_len;
73 spin_lock(&ictx->inode.i_lock);
74 if (fend > ictx->_zero_point)
75 netfs_write_zero_point(inode, fend);
76 spin_unlock(&ictx->inode.i_lock);
77
78 folio_detach_private(folio);
79 group = finfo->netfs_group;
80 gcount++;
81 kfree(finfo);
82 why = netfs_folio_trace_endwb_s;
83 goto end_wb;
84 }
85
86 if ((group = netfs_folio_group(folio))) {
87 if (group == NETFS_FOLIO_COPY_TO_CACHE) {
88 why = netfs_folio_trace_endwb_cc;
89 folio_detach_private(folio);
90 goto end_wb;
91 }
92
93 /* Need to detach the group pointer if the page didn't get
94 * redirtied. If it has been redirtied, then it must be within
95 * the same group.
96 */
97 why = netfs_folio_trace_redirtied;
98 if (!folio_test_dirty(folio)) {
99 folio_detach_private(folio);
100 gcount++;
101 why = netfs_folio_trace_endwb_g;
102 }
103 }
104
105 end_wb:
106 trace_netfs_folio(folio, why);
107 folio_end_writeback(folio);
108 return gcount;
109 }
110
111 /*
112 * Unlock any folios we've finished with.
113 */
netfs_writeback_unlock_folios(struct netfs_io_request * wreq,unsigned int * notes)114 static void netfs_writeback_unlock_folios(struct netfs_io_request *wreq,
115 unsigned int *notes)
116 {
117 struct folio_queue *folioq = wreq->buffer.tail;
118 unsigned int slot = wreq->buffer.first_tail_slot;
119 uoff_t collected_to = wreq->collected_to;
120
121 if (WARN_ON_ONCE(!folioq)) {
122 pr_err("[!] Writeback unlock found empty rolling buffer!\n");
123 netfs_dump_request(wreq);
124 return;
125 }
126
127 if (wreq->origin == NETFS_PGPRIV2_COPY_TO_CACHE) {
128 if (netfs_pgpriv2_unlock_copied_folios(wreq))
129 *notes |= MADE_PROGRESS;
130 return;
131 }
132
133 if (slot >= folioq_nr_slots(folioq)) {
134 folioq = rolling_buffer_delete_spent(&wreq->buffer);
135 if (!folioq)
136 return;
137 slot = 0;
138 }
139
140 for (;;) {
141 struct folio *folio;
142 struct netfs_folio *finfo;
143 uoff_t fpos, fend;
144 size_t fsize, flen;
145
146 folio = folioq_folio(folioq, slot);
147 if (WARN_ONCE(!folio_test_writeback(folio),
148 "R=%08x: folio %lx is not under writeback\n",
149 wreq->debug_id, folio->index))
150 trace_netfs_folio(folio, netfs_folio_trace_not_under_wback);
151
152 fpos = folio_pos(folio);
153 fsize = folio_size(folio);
154 finfo = netfs_folio_info(folio);
155 flen = finfo ? finfo->dirty_offset + finfo->dirty_len : fsize;
156
157 fend = min_t(uoff_t, fpos + flen, wreq->i_size);
158
159 trace_netfs_collect_folio(wreq, folio);
160
161 /* Unlock any folio we've transferred all of. */
162 if (collected_to < fend)
163 break;
164
165 wreq->nr_group_rel += netfs_folio_written_back(folio);
166 wreq->cleaned_to = fpos + fsize;
167 *notes |= MADE_PROGRESS;
168
169 /* Clean up the head folioq. If we clear an entire folioq, then
170 * we can get rid of it provided it's not also the tail folioq
171 * being filled by the issuer.
172 */
173 folioq_clear(folioq, slot);
174 slot++;
175 if (slot >= folioq_nr_slots(folioq)) {
176 folioq = rolling_buffer_delete_spent(&wreq->buffer);
177 if (!folioq)
178 goto done;
179 slot = 0;
180 }
181
182 if (fpos + fsize >= collected_to)
183 break;
184 }
185
186 wreq->buffer.tail = folioq;
187 done:
188 wreq->buffer.first_tail_slot = slot;
189 }
190
191 /*
192 * Collect cache results.
193 */
netfs_cache_collect(struct netfs_io_request * wreq,struct netfs_io_stream * stream,enum netfs_cache_collect block_type)194 static void netfs_cache_collect(struct netfs_io_request *wreq,
195 struct netfs_io_stream *stream,
196 enum netfs_cache_collect block_type)
197 {
198 struct netfs_cache_resources *cres = &wreq->cache_resources;
199
200 if (stream->source != NETFS_WRITE_TO_CACHE ||
201 wreq->cache_coll_to >= stream->collected_to)
202 return;
203
204 if (cres->ops && cres->ops->collect_write)
205 cres->ops->collect_write(wreq, wreq->cache_coll_to,
206 stream->collected_to - wreq->cache_coll_to,
207 block_type);
208 wreq->cache_coll_to = stream->collected_to;
209 }
210
211 /*
212 * Collect and assess the results of various write subrequests. We may need to
213 * retry some of the results - or even do an RMW cycle for content crypto.
214 *
215 * Note that we have a number of parallel, overlapping lists of subrequests,
216 * one to the server and one to the local cache for example, which may not be
217 * the same size or starting position and may not even correspond in boundary
218 * alignment.
219 */
netfs_collect_write_results(struct netfs_io_request * wreq)220 static void netfs_collect_write_results(struct netfs_io_request *wreq)
221 {
222 struct netfs_io_subrequest *front, *remove;
223 struct netfs_io_stream *stream;
224 unsigned int notes;
225 uoff_t collected_to, issued_to;
226 int s;
227
228 _enter("%llx-%llx", wreq->start, wreq->start + wreq->len);
229 trace_netfs_collect(wreq);
230 trace_netfs_rreq(wreq, netfs_rreq_trace_collect);
231
232 reassess_streams:
233 issued_to = atomic64_read(&wreq->issued_to);
234 smp_rmb();
235 collected_to = ULLONG_MAX;
236 if (wreq->origin == NETFS_WRITEBACK ||
237 wreq->origin == NETFS_PGPRIV2_COPY_TO_CACHE)
238 notes = NEED_UNLOCK;
239 else
240 notes = 0;
241
242 /* Remove completed subrequests from the front of the streams and
243 * advance the completion point on each stream. We stop when we hit
244 * something that's in progress. The issuer thread may be adding stuff
245 * to the tail whilst we're doing this.
246 */
247 for (s = 0; s < NR_IO_STREAMS; s++) {
248 stream = &wreq->io_streams[s];
249 /* Read active flag before list pointers */
250 if (!smp_load_acquire(&stream->active))
251 continue;
252
253 front = list_first_entry_or_null_acquire(&stream->subrequests,
254 struct netfs_io_subrequest, rreq_link);
255 /* Read first subreq pointer before IN_PROGRESS flag. */
256
257 while (front) {
258 enum netfs_cache_collect cache_collect;
259
260 trace_netfs_collect_sreq(wreq, front);
261 //_debug("sreq [%x] %llx %zx/%zx",
262 // front->debug_index, front->start, front->transferred, front->len);
263
264 if (stream->collected_to < front->start) {
265 trace_netfs_collect_gap(wreq, stream, issued_to, 'F');
266 if (stream->cache_collect != NETFS_CACHE_COLLECT_WRITE_GAP)
267 netfs_cache_collect(wreq, stream, stream->cache_collect);
268 stream->collected_to = front->start;
269 netfs_cache_collect(wreq, stream, NETFS_CACHE_COLLECT_WRITE_GAP);
270 stream->cache_collect = NETFS_CACHE_COLLECT_WRITE_GAP;
271 }
272
273 /* Stall if the front is still undergoing I/O. */
274 if (netfs_check_subreq_in_progress(front)) {
275 notes |= HIT_PENDING;
276 break;
277 }
278
279 if (stream->failed) {
280 stream->collected_to = front->start + front->len;
281 notes |= MADE_PROGRESS | SAW_FAILURE;
282 goto cancel;
283 }
284 if (front->start + front->transferred > stream->collected_to) {
285 stream->collected_to = front->start + front->transferred;
286 stream->transferred = stream->collected_to - wreq->start;
287 stream->transferred_valid = true;
288 notes |= MADE_PROGRESS;
289 }
290
291 /* Handle failed or cancelled subreqs. Failure of
292 * cache writes are handled differently to upload
293 * failures. Cache writes aren't fatal, provided we're
294 * not doing disconnected operation, and so we can kind
295 * of treat them as if they had succeeded - except that
296 * we need to log any holes they cause.
297 */
298 switch (stream->source) {
299 case NETFS_UPLOAD_TO_SERVER:
300 if (test_bit(NETFS_SREQ_FAILED, &front->flags)) {
301 if (!stream->failed) {
302 stream->failed = true;
303 stream->error = front->error;
304 mapping_set_error(wreq->mapping, front->error);
305 break;
306 }
307 notes |= NEED_REASSESS | SAW_FAILURE;
308 }
309 break;
310
311 case NETFS_WRITE_TO_CACHE:
312 cache_collect = test_bit(NETFS_SREQ_CANCELLED, &front->flags) ?
313 NETFS_CACHE_COLLECT_WRITE_CANCEL :
314 NETFS_CACHE_COLLECT_WRITE_DATA;
315 if (cache_collect != stream->cache_collect &&
316 stream->cache_collect != NETFS_CACHE_COLLECT_WRITE_GAP) {
317 trace_netfs_rreq(wreq, netfs_rreq_trace_cache_fail_collect);
318 netfs_cache_collect(wreq, stream, stream->cache_collect);
319 }
320 stream->cache_collect = cache_collect;
321 break;
322
323 default:
324 WARN_ON(1);
325 break;
326 }
327 if (test_bit(NETFS_SREQ_NEED_RETRY, &front->flags)) {
328 stream->need_retry = true;
329 notes |= NEED_RETRY | MADE_PROGRESS;
330 break;
331 }
332
333 cancel:
334 /* Remove if completely consumed. */
335 spin_lock(&wreq->lock);
336
337 remove = front;
338 list_del_init(&front->rreq_link);
339 front = list_first_entry_or_null(&stream->subrequests,
340 struct netfs_io_subrequest, rreq_link);
341 spin_unlock(&wreq->lock);
342 netfs_put_subrequest(remove,
343 notes & SAW_FAILURE ?
344 netfs_sreq_trace_put_cancel :
345 netfs_sreq_trace_put_done);
346 }
347
348 /* If we have an empty stream, we need to jump it forward
349 * otherwise the collection point will never advance.
350 */
351 if (!front && issued_to > stream->collected_to) {
352 trace_netfs_collect_gap(wreq, stream, issued_to, 'E');
353 stream->collected_to = issued_to;
354 }
355
356 if (stream->collected_to < collected_to)
357 collected_to = stream->collected_to;
358 }
359
360 if (collected_to != ULLONG_MAX && collected_to > wreq->collected_to)
361 wreq->collected_to = collected_to;
362
363 for (s = 0; s < NR_IO_STREAMS; s++) {
364 stream = &wreq->io_streams[s];
365 if (stream->active)
366 trace_netfs_collect_stream(wreq, stream);
367 }
368
369 trace_netfs_collect_state(wreq, wreq->collected_to, notes);
370
371 /* Unlock any folios that we have now finished with. */
372 if (notes & NEED_UNLOCK) {
373 if (wreq->cleaned_to < wreq->collected_to)
374 netfs_writeback_unlock_folios(wreq, ¬es);
375 } else {
376 wreq->cleaned_to = wreq->collected_to;
377 }
378
379 // TODO: Discard encryption buffers
380
381 if (notes & NEED_RETRY)
382 goto need_retry;
383
384 if (notes & MADE_PROGRESS) {
385 netfs_wake_rreq_flag(wreq, NETFS_RREQ_PAUSE, netfs_rreq_trace_unpause);
386 //cond_resched();
387 goto reassess_streams;
388 }
389
390 if (notes & NEED_REASSESS) {
391 //cond_resched();
392 goto reassess_streams;
393 }
394
395 out:
396 netfs_put_group_many(wreq->group, wreq->nr_group_rel);
397 wreq->nr_group_rel = 0;
398 _leave(" = %x", notes);
399 return;
400
401 need_retry:
402 /* Okay... We're going to have to retry one or both streams. Note
403 * that any partially completed op will have had any wholly transferred
404 * folios removed from it.
405 */
406 _debug("retry");
407 netfs_retry_writes(wreq);
408 goto out;
409 }
410
411 /*
412 * Perform the collection of subrequests, folios and encryption buffers.
413 */
netfs_write_collection(struct netfs_io_request * wreq)414 bool netfs_write_collection(struct netfs_io_request *wreq)
415 {
416 struct netfs_io_stream *cstream = &wreq->io_streams[1];
417 struct netfs_inode *ictx = netfs_inode(wreq->inode);
418 size_t transferred;
419 bool transferred_valid = false;
420 int s;
421
422 _enter("R=%x", wreq->debug_id);
423
424 netfs_collect_write_results(wreq);
425
426 /* We're done when the app thread has finished posting subreqs and all
427 * the queues in all the streams are empty.
428 */
429 if (!netfs_are_all_subreqs_queued(wreq))
430 return false;
431
432 transferred = LONG_MAX;
433 for (s = 0; s < NR_IO_STREAMS; s++) {
434 struct netfs_io_stream *stream = &wreq->io_streams[s];
435 if (!stream->active)
436 continue;
437 if (!list_empty(&stream->subrequests))
438 return false;
439 if (stream->transferred_valid &&
440 stream->transferred < transferred) {
441 transferred = stream->transferred;
442 transferred_valid = true;
443 }
444 }
445
446 /* Okay, declare that all I/O is complete. */
447 if (transferred_valid)
448 wreq->transferred = transferred;
449 trace_netfs_rreq(wreq, netfs_rreq_trace_write_done);
450
451 if (cstream->active) {
452 if (test_bit(NETFS_RREQ_CACHE_ERROR, &wreq->flags)) {
453 if (ictx->ops->invalidate_cache) {
454 /* Cache write failure doesn't prevent
455 * writeback completion unless we're in
456 * disconnected mode.
457 */
458 trace_netfs_rreq(wreq, netfs_rreq_trace_inval_cache);
459 ictx->ops->invalidate_cache(wreq);
460 }
461 } else if (!cstream->failed) {
462 netfs_cache_collect(wreq, cstream, cstream->cache_collect);
463 }
464 }
465
466 _debug("finished");
467 netfs_wake_rreq_flag(wreq, NETFS_RREQ_IN_PROGRESS, netfs_rreq_trace_wake_ip);
468 /* As we cleared NETFS_RREQ_IN_PROGRESS, we acquired its ref. */
469
470 switch (wreq->origin) {
471 case NETFS_WRITEBACK:
472 case NETFS_WRITEBACK_SINGLE:
473 netfs_wb_end(ictx);
474 break;
475 default:
476 break;
477 }
478
479 if (wreq->iocb) {
480 size_t written = min(wreq->transferred, wreq->len);
481 wreq->iocb->ki_pos += written;
482 if (wreq->iocb->ki_complete) {
483 trace_netfs_rreq(wreq, netfs_rreq_trace_ki_complete);
484 wreq->iocb->ki_complete(
485 wreq->iocb, wreq->error ? wreq->error : written);
486 }
487 wreq->iocb = VFS_PTR_POISON;
488 }
489
490 netfs_clear_subrequests(wreq);
491 return true;
492 }
493
netfs_write_collection_worker(struct work_struct * work)494 void netfs_write_collection_worker(struct work_struct *work)
495 {
496 struct netfs_io_request *rreq = container_of(work, struct netfs_io_request, work);
497
498 netfs_see_request(rreq, netfs_rreq_trace_see_work);
499 if (netfs_check_rreq_in_progress(rreq)) {
500 if (netfs_write_collection(rreq))
501 /* Drop the ref from the IN_PROGRESS flag. */
502 netfs_put_request(rreq, netfs_rreq_trace_put_work_ip);
503 else
504 netfs_see_request(rreq, netfs_rreq_trace_see_work_complete);
505 }
506 }
507
508 /**
509 * netfs_write_subrequest_terminated - Note the termination of a write operation.
510 * @_op: The I/O request that has terminated.
511 * @transferred_or_error: The amount of data transferred or an error code.
512 *
513 * This tells the library that a contributory write I/O operation has
514 * terminated, one way or another, and that it should collect the results.
515 *
516 * The caller indicates in @transferred_or_error the outcome of the operation,
517 * supplying a positive value to indicate the number of bytes transferred or a
518 * negative error code. The library will look after reissuing I/O operations
519 * as appropriate and writing downloaded data to the cache.
520 *
521 * When this is called, ownership of the subrequest is transferred back to the
522 * library, along with a ref.
523 *
524 * Note that %_op is a void* so that the function can be passed to
525 * kiocb::term_func without the need for a casting wrapper.
526 */
netfs_write_subrequest_terminated(void * _op,ssize_t transferred_or_error)527 void netfs_write_subrequest_terminated(void *_op, ssize_t transferred_or_error)
528 {
529 struct netfs_io_subrequest *subreq = _op;
530 struct netfs_io_request *wreq = subreq->rreq;
531
532 _enter("%x[%x] %zd", wreq->debug_id, subreq->debug_index, transferred_or_error);
533
534 switch (subreq->source) {
535 case NETFS_UPLOAD_TO_SERVER:
536 netfs_stat(&netfs_n_wh_upload_done);
537 break;
538 case NETFS_WRITE_TO_CACHE:
539 netfs_stat(&netfs_n_wh_write_done);
540 break;
541 default:
542 BUG();
543 }
544
545 if (IS_ERR_VALUE(transferred_or_error)) {
546 subreq->error = transferred_or_error;
547
548 switch (subreq->source) {
549 case NETFS_WRITE_TO_CACHE:
550 /* We don't mark a cache-write subreq as failed.
551 * Instead we tell the issuer to produce dummy subreqs
552 * instead and make a note if we need to invalidate the
553 * cache at the end. We also don't pause the loop that
554 * grabs pages and launches upload subreqs.
555 *
556 * Note that we need to distinguish between -ENOBUFS
557 * (no space available in the cache) and other errors.
558 * In the former case, we can keep the data we have,
559 * though we might have to change the way the on-disk
560 * data is tracked.
561 */
562 netfs_stat(&netfs_n_wh_write_failed);
563 if (test_bit(NETFS_SREQ_NEED_RETRY, &subreq->flags))
564 break;
565
566 trace_netfs_failure(wreq, subreq, transferred_or_error, netfs_fail_write);
567 __set_bit(NETFS_SREQ_CANCELLED, &subreq->flags);
568 set_bit(NETFS_RREQ_CACHE_STOP, &wreq->flags);
569 if (transferred_or_error == -ENOBUFS)
570 trace_netfs_rreq(wreq, netfs_rreq_trace_cache_no_space);
571 else if (!test_and_set_bit(NETFS_RREQ_CACHE_ERROR, &wreq->flags))
572 trace_netfs_rreq(wreq, netfs_rreq_trace_cache_failed);
573 subreq->transferred = subreq->len;
574 break;
575
576 case NETFS_UPLOAD_TO_SERVER:
577 /* If need_retry is set, error should not matter */
578 if (!test_bit(NETFS_SREQ_NEED_RETRY, &subreq->flags)) {
579 set_bit(NETFS_SREQ_FAILED, &subreq->flags);
580 trace_netfs_failure(wreq, subreq, transferred_or_error,
581 netfs_fail_upload);
582 }
583
584 set_bit(NETFS_RREQ_PAUSE, &wreq->flags);
585 trace_netfs_rreq(wreq, netfs_rreq_trace_set_pause);
586 netfs_stat(&netfs_n_wh_upload_failed);
587 break;
588
589 default:
590 break;
591 }
592 } else {
593 if (WARN(transferred_or_error > subreq->len - subreq->transferred,
594 "Subreq excess write: R=%x[%x] %zd > %zu - %zu",
595 wreq->debug_id, subreq->debug_index,
596 transferred_or_error, subreq->len, subreq->transferred))
597 transferred_or_error = subreq->len - subreq->transferred;
598
599 subreq->error = 0;
600 subreq->transferred += transferred_or_error;
601
602 if (subreq->transferred < subreq->len)
603 set_bit(NETFS_SREQ_NEED_RETRY, &subreq->flags);
604 }
605
606 trace_netfs_sreq(subreq, netfs_sreq_trace_terminated);
607 netfs_subreq_clear_in_progress(subreq);
608 netfs_put_subrequest(subreq, netfs_sreq_trace_put_terminated);
609 }
610 EXPORT_SYMBOL(netfs_write_subrequest_terminated);
611