xfs_buf.c revision 04fcad80
1// SPDX-License-Identifier: GPL-2.0
2/*
3 * Copyright (c) 2000-2006 Silicon Graphics, Inc.
4 * All Rights Reserved.
5 */
6#include "xfs.h"
7#include <linux/backing-dev.h>
8
9#include "xfs_shared.h"
10#include "xfs_format.h"
11#include "xfs_log_format.h"
12#include "xfs_trans_resv.h"
13#include "xfs_mount.h"
14#include "xfs_trace.h"
15#include "xfs_log.h"
16#include "xfs_log_recover.h"
17#include "xfs_trans.h"
18#include "xfs_buf_item.h"
19#include "xfs_errortag.h"
20#include "xfs_error.h"
21#include "xfs_ag.h"
22
23static kmem_zone_t *xfs_buf_zone;
24
25/*
26 * Locking orders
27 *
28 * xfs_buf_ioacct_inc:
29 * xfs_buf_ioacct_dec:
30 *	b_sema (caller holds)
31 *	  b_lock
32 *
33 * xfs_buf_stale:
34 *	b_sema (caller holds)
35 *	  b_lock
36 *	    lru_lock
37 *
38 * xfs_buf_rele:
39 *	b_lock
40 *	  pag_buf_lock
41 *	    lru_lock
42 *
43 * xfs_buftarg_drain_rele
44 *	lru_lock
45 *	  b_lock (trylock due to inversion)
46 *
47 * xfs_buftarg_isolate
48 *	lru_lock
49 *	  b_lock (trylock due to inversion)
50 */
51
52static int __xfs_buf_submit(struct xfs_buf *bp, bool wait);
53
54static inline int
55xfs_buf_submit(
56	struct xfs_buf		*bp)
57{
58	return __xfs_buf_submit(bp, !(bp->b_flags & XBF_ASYNC));
59}
60
61static inline int
62xfs_buf_is_vmapped(
63	struct xfs_buf	*bp)
64{
65	/*
66	 * Return true if the buffer is vmapped.
67	 *
68	 * b_addr is null if the buffer is not mapped, but the code is clever
69	 * enough to know it doesn't have to map a single page, so the check has
70	 * to be both for b_addr and bp->b_page_count > 1.
71	 */
72	return bp->b_addr && bp->b_page_count > 1;
73}
74
75static inline int
76xfs_buf_vmap_len(
77	struct xfs_buf	*bp)
78{
79	return (bp->b_page_count * PAGE_SIZE);
80}
81
82/*
83 * Bump the I/O in flight count on the buftarg if we haven't yet done so for
84 * this buffer. The count is incremented once per buffer (per hold cycle)
85 * because the corresponding decrement is deferred to buffer release. Buffers
86 * can undergo I/O multiple times in a hold-release cycle and per buffer I/O
87 * tracking adds unnecessary overhead. This is used for sychronization purposes
88 * with unmount (see xfs_buftarg_drain()), so all we really need is a count of
89 * in-flight buffers.
90 *
91 * Buffers that are never released (e.g., superblock, iclog buffers) must set
92 * the XBF_NO_IOACCT flag before I/O submission. Otherwise, the buftarg count
93 * never reaches zero and unmount hangs indefinitely.
94 */
95static inline void
96xfs_buf_ioacct_inc(
97	struct xfs_buf	*bp)
98{
99	if (bp->b_flags & XBF_NO_IOACCT)
100		return;
101
102	ASSERT(bp->b_flags & XBF_ASYNC);
103	spin_lock(&bp->b_lock);
104	if (!(bp->b_state & XFS_BSTATE_IN_FLIGHT)) {
105		bp->b_state |= XFS_BSTATE_IN_FLIGHT;
106		percpu_counter_inc(&bp->b_target->bt_io_count);
107	}
108	spin_unlock(&bp->b_lock);
109}
110
111/*
112 * Clear the in-flight state on a buffer about to be released to the LRU or
113 * freed and unaccount from the buftarg.
114 */
115static inline void
116__xfs_buf_ioacct_dec(
117	struct xfs_buf	*bp)
118{
119	lockdep_assert_held(&bp->b_lock);
120
121	if (bp->b_state & XFS_BSTATE_IN_FLIGHT) {
122		bp->b_state &= ~XFS_BSTATE_IN_FLIGHT;
123		percpu_counter_dec(&bp->b_target->bt_io_count);
124	}
125}
126
127static inline void
128xfs_buf_ioacct_dec(
129	struct xfs_buf	*bp)
130{
131	spin_lock(&bp->b_lock);
132	__xfs_buf_ioacct_dec(bp);
133	spin_unlock(&bp->b_lock);
134}
135
136/*
137 * When we mark a buffer stale, we remove the buffer from the LRU and clear the
138 * b_lru_ref count so that the buffer is freed immediately when the buffer
139 * reference count falls to zero. If the buffer is already on the LRU, we need
140 * to remove the reference that LRU holds on the buffer.
141 *
142 * This prevents build-up of stale buffers on the LRU.
143 */
144void
145xfs_buf_stale(
146	struct xfs_buf	*bp)
147{
148	ASSERT(xfs_buf_islocked(bp));
149
150	bp->b_flags |= XBF_STALE;
151
152	/*
153	 * Clear the delwri status so that a delwri queue walker will not
154	 * flush this buffer to disk now that it is stale. The delwri queue has
155	 * a reference to the buffer, so this is safe to do.
156	 */
157	bp->b_flags &= ~_XBF_DELWRI_Q;
158
159	/*
160	 * Once the buffer is marked stale and unlocked, a subsequent lookup
161	 * could reset b_flags. There is no guarantee that the buffer is
162	 * unaccounted (released to LRU) before that occurs. Drop in-flight
163	 * status now to preserve accounting consistency.
164	 */
165	spin_lock(&bp->b_lock);
166	__xfs_buf_ioacct_dec(bp);
167
168	atomic_set(&bp->b_lru_ref, 0);
169	if (!(bp->b_state & XFS_BSTATE_DISPOSE) &&
170	    (list_lru_del(&bp->b_target->bt_lru, &bp->b_lru)))
171		atomic_dec(&bp->b_hold);
172
173	ASSERT(atomic_read(&bp->b_hold) >= 1);
174	spin_unlock(&bp->b_lock);
175}
176
177static int
178xfs_buf_get_maps(
179	struct xfs_buf		*bp,
180	int			map_count)
181{
182	ASSERT(bp->b_maps == NULL);
183	bp->b_map_count = map_count;
184
185	if (map_count == 1) {
186		bp->b_maps = &bp->__b_map;
187		return 0;
188	}
189
190	bp->b_maps = kmem_zalloc(map_count * sizeof(struct xfs_buf_map),
191				KM_NOFS);
192	if (!bp->b_maps)
193		return -ENOMEM;
194	return 0;
195}
196
197/*
198 *	Frees b_pages if it was allocated.
199 */
200static void
201xfs_buf_free_maps(
202	struct xfs_buf	*bp)
203{
204	if (bp->b_maps != &bp->__b_map) {
205		kmem_free(bp->b_maps);
206		bp->b_maps = NULL;
207	}
208}
209
210static int
211_xfs_buf_alloc(
212	struct xfs_buftarg	*target,
213	struct xfs_buf_map	*map,
214	int			nmaps,
215	xfs_buf_flags_t		flags,
216	struct xfs_buf		**bpp)
217{
218	struct xfs_buf		*bp;
219	int			error;
220	int			i;
221
222	*bpp = NULL;
223	bp = kmem_cache_zalloc(xfs_buf_zone, GFP_NOFS | __GFP_NOFAIL);
224
225	/*
226	 * We don't want certain flags to appear in b_flags unless they are
227	 * specifically set by later operations on the buffer.
228	 */
229	flags &= ~(XBF_UNMAPPED | XBF_TRYLOCK | XBF_ASYNC | XBF_READ_AHEAD);
230
231	atomic_set(&bp->b_hold, 1);
232	atomic_set(&bp->b_lru_ref, 1);
233	init_completion(&bp->b_iowait);
234	INIT_LIST_HEAD(&bp->b_lru);
235	INIT_LIST_HEAD(&bp->b_list);
236	INIT_LIST_HEAD(&bp->b_li_list);
237	sema_init(&bp->b_sema, 0); /* held, no waiters */
238	spin_lock_init(&bp->b_lock);
239	bp->b_target = target;
240	bp->b_mount = target->bt_mount;
241	bp->b_flags = flags;
242
243	/*
244	 * Set length and io_length to the same value initially.
245	 * I/O routines should use io_length, which will be the same in
246	 * most cases but may be reset (e.g. XFS recovery).
247	 */
248	error = xfs_buf_get_maps(bp, nmaps);
249	if (error)  {
250		kmem_cache_free(xfs_buf_zone, bp);
251		return error;
252	}
253
254	bp->b_bn = map[0].bm_bn;
255	bp->b_length = 0;
256	for (i = 0; i < nmaps; i++) {
257		bp->b_maps[i].bm_bn = map[i].bm_bn;
258		bp->b_maps[i].bm_len = map[i].bm_len;
259		bp->b_length += map[i].bm_len;
260	}
261
262	atomic_set(&bp->b_pin_count, 0);
263	init_waitqueue_head(&bp->b_waiters);
264
265	XFS_STATS_INC(bp->b_mount, xb_create);
266	trace_xfs_buf_init(bp, _RET_IP_);
267
268	*bpp = bp;
269	return 0;
270}
271
272static void
273xfs_buf_free_pages(
274	struct xfs_buf	*bp)
275{
276	uint		i;
277
278	ASSERT(bp->b_flags & _XBF_PAGES);
279
280	if (xfs_buf_is_vmapped(bp))
281		vm_unmap_ram(bp->b_addr, bp->b_page_count);
282
283	for (i = 0; i < bp->b_page_count; i++) {
284		if (bp->b_pages[i])
285			__free_page(bp->b_pages[i]);
286	}
287	if (current->reclaim_state)
288		current->reclaim_state->reclaimed_slab += bp->b_page_count;
289
290	if (bp->b_pages != bp->b_page_array)
291		kmem_free(bp->b_pages);
292	bp->b_pages = NULL;
293	bp->b_flags &= ~_XBF_PAGES;
294}
295
296static void
297xfs_buf_free(
298	struct xfs_buf		*bp)
299{
300	trace_xfs_buf_free(bp, _RET_IP_);
301
302	ASSERT(list_empty(&bp->b_lru));
303
304	if (bp->b_flags & _XBF_PAGES)
305		xfs_buf_free_pages(bp);
306	else if (bp->b_flags & _XBF_KMEM)
307		kmem_free(bp->b_addr);
308
309	xfs_buf_free_maps(bp);
310	kmem_cache_free(xfs_buf_zone, bp);
311}
312
313static int
314xfs_buf_alloc_kmem(
315	struct xfs_buf	*bp,
316	xfs_buf_flags_t	flags)
317{
318	xfs_km_flags_t	kmflag_mask = KM_NOFS;
319	size_t		size = BBTOB(bp->b_length);
320
321	/* Assure zeroed buffer for non-read cases. */
322	if (!(flags & XBF_READ))
323		kmflag_mask |= KM_ZERO;
324
325	bp->b_addr = kmem_alloc(size, kmflag_mask);
326	if (!bp->b_addr)
327		return -ENOMEM;
328
329	if (((unsigned long)(bp->b_addr + size - 1) & PAGE_MASK) !=
330	    ((unsigned long)bp->b_addr & PAGE_MASK)) {
331		/* b_addr spans two pages - use alloc_page instead */
332		kmem_free(bp->b_addr);
333		bp->b_addr = NULL;
334		return -ENOMEM;
335	}
336	bp->b_offset = offset_in_page(bp->b_addr);
337	bp->b_pages = bp->b_page_array;
338	bp->b_pages[0] = kmem_to_page(bp->b_addr);
339	bp->b_page_count = 1;
340	bp->b_flags |= _XBF_KMEM;
341	return 0;
342}
343
344static int
345xfs_buf_alloc_pages(
346	struct xfs_buf	*bp,
347	xfs_buf_flags_t	flags)
348{
349	gfp_t		gfp_mask = __GFP_NOWARN;
350	long		filled = 0;
351
352	if (flags & XBF_READ_AHEAD)
353		gfp_mask |= __GFP_NORETRY;
354	else
355		gfp_mask |= GFP_NOFS;
356
357	/* Make sure that we have a page list */
358	bp->b_page_count = DIV_ROUND_UP(BBTOB(bp->b_length), PAGE_SIZE);
359	if (bp->b_page_count <= XB_PAGES) {
360		bp->b_pages = bp->b_page_array;
361	} else {
362		bp->b_pages = kzalloc(sizeof(struct page *) * bp->b_page_count,
363					gfp_mask);
364		if (!bp->b_pages)
365			return -ENOMEM;
366	}
367	bp->b_flags |= _XBF_PAGES;
368
369	/* Assure zeroed buffer for non-read cases. */
370	if (!(flags & XBF_READ))
371		gfp_mask |= __GFP_ZERO;
372
373	/*
374	 * Bulk filling of pages can take multiple calls. Not filling the entire
375	 * array is not an allocation failure, so don't back off if we get at
376	 * least one extra page.
377	 */
378	for (;;) {
379		long	last = filled;
380
381		filled = alloc_pages_bulk_array(gfp_mask, bp->b_page_count,
382						bp->b_pages);
383		if (filled == bp->b_page_count) {
384			XFS_STATS_INC(bp->b_mount, xb_page_found);
385			break;
386		}
387
388		if (filled != last)
389			continue;
390
391		if (flags & XBF_READ_AHEAD) {
392			xfs_buf_free_pages(bp);
393			return -ENOMEM;
394		}
395
396		XFS_STATS_INC(bp->b_mount, xb_page_retries);
397		congestion_wait(BLK_RW_ASYNC, HZ / 50);
398	}
399	return 0;
400}
401
402/*
403 *	Map buffer into kernel address-space if necessary.
404 */
405STATIC int
406_xfs_buf_map_pages(
407	struct xfs_buf		*bp,
408	uint			flags)
409{
410	ASSERT(bp->b_flags & _XBF_PAGES);
411	if (bp->b_page_count == 1) {
412		/* A single page buffer is always mappable */
413		bp->b_addr = page_address(bp->b_pages[0]);
414	} else if (flags & XBF_UNMAPPED) {
415		bp->b_addr = NULL;
416	} else {
417		int retried = 0;
418		unsigned nofs_flag;
419
420		/*
421		 * vm_map_ram() will allocate auxiliary structures (e.g.
422		 * pagetables) with GFP_KERNEL, yet we are likely to be under
423		 * GFP_NOFS context here. Hence we need to tell memory reclaim
424		 * that we are in such a context via PF_MEMALLOC_NOFS to prevent
425		 * memory reclaim re-entering the filesystem here and
426		 * potentially deadlocking.
427		 */
428		nofs_flag = memalloc_nofs_save();
429		do {
430			bp->b_addr = vm_map_ram(bp->b_pages, bp->b_page_count,
431						-1);
432			if (bp->b_addr)
433				break;
434			vm_unmap_aliases();
435		} while (retried++ <= 1);
436		memalloc_nofs_restore(nofs_flag);
437
438		if (!bp->b_addr)
439			return -ENOMEM;
440	}
441
442	return 0;
443}
444
445/*
446 *	Finding and Reading Buffers
447 */
448static int
449_xfs_buf_obj_cmp(
450	struct rhashtable_compare_arg	*arg,
451	const void			*obj)
452{
453	const struct xfs_buf_map	*map = arg->key;
454	const struct xfs_buf		*bp = obj;
455
456	/*
457	 * The key hashing in the lookup path depends on the key being the
458	 * first element of the compare_arg, make sure to assert this.
459	 */
460	BUILD_BUG_ON(offsetof(struct xfs_buf_map, bm_bn) != 0);
461
462	if (bp->b_bn != map->bm_bn)
463		return 1;
464
465	if (unlikely(bp->b_length != map->bm_len)) {
466		/*
467		 * found a block number match. If the range doesn't
468		 * match, the only way this is allowed is if the buffer
469		 * in the cache is stale and the transaction that made
470		 * it stale has not yet committed. i.e. we are
471		 * reallocating a busy extent. Skip this buffer and
472		 * continue searching for an exact match.
473		 */
474		ASSERT(bp->b_flags & XBF_STALE);
475		return 1;
476	}
477	return 0;
478}
479
480static const struct rhashtable_params xfs_buf_hash_params = {
481	.min_size		= 32,	/* empty AGs have minimal footprint */
482	.nelem_hint		= 16,
483	.key_len		= sizeof(xfs_daddr_t),
484	.key_offset		= offsetof(struct xfs_buf, b_bn),
485	.head_offset		= offsetof(struct xfs_buf, b_rhash_head),
486	.automatic_shrinking	= true,
487	.obj_cmpfn		= _xfs_buf_obj_cmp,
488};
489
490int
491xfs_buf_hash_init(
492	struct xfs_perag	*pag)
493{
494	spin_lock_init(&pag->pag_buf_lock);
495	return rhashtable_init(&pag->pag_buf_hash, &xfs_buf_hash_params);
496}
497
498void
499xfs_buf_hash_destroy(
500	struct xfs_perag	*pag)
501{
502	rhashtable_destroy(&pag->pag_buf_hash);
503}
504
505/*
506 * Look up a buffer in the buffer cache and return it referenced and locked
507 * in @found_bp.
508 *
509 * If @new_bp is supplied and we have a lookup miss, insert @new_bp into the
510 * cache.
511 *
512 * If XBF_TRYLOCK is set in @flags, only try to lock the buffer and return
513 * -EAGAIN if we fail to lock it.
514 *
515 * Return values are:
516 *	-EFSCORRUPTED if have been supplied with an invalid address
517 *	-EAGAIN on trylock failure
518 *	-ENOENT if we fail to find a match and @new_bp was NULL
519 *	0, with @found_bp:
520 *		- @new_bp if we inserted it into the cache
521 *		- the buffer we found and locked.
522 */
523static int
524xfs_buf_find(
525	struct xfs_buftarg	*btp,
526	struct xfs_buf_map	*map,
527	int			nmaps,
528	xfs_buf_flags_t		flags,
529	struct xfs_buf		*new_bp,
530	struct xfs_buf		**found_bp)
531{
532	struct xfs_perag	*pag;
533	struct xfs_buf		*bp;
534	struct xfs_buf_map	cmap = { .bm_bn = map[0].bm_bn };
535	xfs_daddr_t		eofs;
536	int			i;
537
538	*found_bp = NULL;
539
540	for (i = 0; i < nmaps; i++)
541		cmap.bm_len += map[i].bm_len;
542
543	/* Check for IOs smaller than the sector size / not sector aligned */
544	ASSERT(!(BBTOB(cmap.bm_len) < btp->bt_meta_sectorsize));
545	ASSERT(!(BBTOB(cmap.bm_bn) & (xfs_off_t)btp->bt_meta_sectormask));
546
547	/*
548	 * Corrupted block numbers can get through to here, unfortunately, so we
549	 * have to check that the buffer falls within the filesystem bounds.
550	 */
551	eofs = XFS_FSB_TO_BB(btp->bt_mount, btp->bt_mount->m_sb.sb_dblocks);
552	if (cmap.bm_bn < 0 || cmap.bm_bn >= eofs) {
553		xfs_alert(btp->bt_mount,
554			  "%s: daddr 0x%llx out of range, EOFS 0x%llx",
555			  __func__, cmap.bm_bn, eofs);
556		WARN_ON(1);
557		return -EFSCORRUPTED;
558	}
559
560	pag = xfs_perag_get(btp->bt_mount,
561			    xfs_daddr_to_agno(btp->bt_mount, cmap.bm_bn));
562
563	spin_lock(&pag->pag_buf_lock);
564	bp = rhashtable_lookup_fast(&pag->pag_buf_hash, &cmap,
565				    xfs_buf_hash_params);
566	if (bp) {
567		atomic_inc(&bp->b_hold);
568		goto found;
569	}
570
571	/* No match found */
572	if (!new_bp) {
573		XFS_STATS_INC(btp->bt_mount, xb_miss_locked);
574		spin_unlock(&pag->pag_buf_lock);
575		xfs_perag_put(pag);
576		return -ENOENT;
577	}
578
579	/* the buffer keeps the perag reference until it is freed */
580	new_bp->b_pag = pag;
581	rhashtable_insert_fast(&pag->pag_buf_hash, &new_bp->b_rhash_head,
582			       xfs_buf_hash_params);
583	spin_unlock(&pag->pag_buf_lock);
584	*found_bp = new_bp;
585	return 0;
586
587found:
588	spin_unlock(&pag->pag_buf_lock);
589	xfs_perag_put(pag);
590
591	if (!xfs_buf_trylock(bp)) {
592		if (flags & XBF_TRYLOCK) {
593			xfs_buf_rele(bp);
594			XFS_STATS_INC(btp->bt_mount, xb_busy_locked);
595			return -EAGAIN;
596		}
597		xfs_buf_lock(bp);
598		XFS_STATS_INC(btp->bt_mount, xb_get_locked_waited);
599	}
600
601	/*
602	 * if the buffer is stale, clear all the external state associated with
603	 * it. We need to keep flags such as how we allocated the buffer memory
604	 * intact here.
605	 */
606	if (bp->b_flags & XBF_STALE) {
607		ASSERT((bp->b_flags & _XBF_DELWRI_Q) == 0);
608		bp->b_flags &= _XBF_KMEM | _XBF_PAGES;
609		bp->b_ops = NULL;
610	}
611
612	trace_xfs_buf_find(bp, flags, _RET_IP_);
613	XFS_STATS_INC(btp->bt_mount, xb_get_locked);
614	*found_bp = bp;
615	return 0;
616}
617
618struct xfs_buf *
619xfs_buf_incore(
620	struct xfs_buftarg	*target,
621	xfs_daddr_t		blkno,
622	size_t			numblks,
623	xfs_buf_flags_t		flags)
624{
625	struct xfs_buf		*bp;
626	int			error;
627	DEFINE_SINGLE_BUF_MAP(map, blkno, numblks);
628
629	error = xfs_buf_find(target, &map, 1, flags, NULL, &bp);
630	if (error)
631		return NULL;
632	return bp;
633}
634
635/*
636 * Assembles a buffer covering the specified range. The code is optimised for
637 * cache hits, as metadata intensive workloads will see 3 orders of magnitude
638 * more hits than misses.
639 */
640int
641xfs_buf_get_map(
642	struct xfs_buftarg	*target,
643	struct xfs_buf_map	*map,
644	int			nmaps,
645	xfs_buf_flags_t		flags,
646	struct xfs_buf		**bpp)
647{
648	struct xfs_buf		*bp;
649	struct xfs_buf		*new_bp;
650	int			error;
651
652	*bpp = NULL;
653	error = xfs_buf_find(target, map, nmaps, flags, NULL, &bp);
654	if (!error)
655		goto found;
656	if (error != -ENOENT)
657		return error;
658
659	error = _xfs_buf_alloc(target, map, nmaps, flags, &new_bp);
660	if (error)
661		return error;
662
663	/*
664	 * For buffers that fit entirely within a single page, first attempt to
665	 * allocate the memory from the heap to minimise memory usage. If we
666	 * can't get heap memory for these small buffers, we fall back to using
667	 * the page allocator.
668	 */
669	if (BBTOB(new_bp->b_length) >= PAGE_SIZE ||
670	    xfs_buf_alloc_kmem(new_bp, flags) < 0) {
671		error = xfs_buf_alloc_pages(new_bp, flags);
672		if (error)
673			goto out_free_buf;
674	}
675
676	error = xfs_buf_find(target, map, nmaps, flags, new_bp, &bp);
677	if (error)
678		goto out_free_buf;
679
680	if (bp != new_bp)
681		xfs_buf_free(new_bp);
682
683found:
684	if (!bp->b_addr) {
685		error = _xfs_buf_map_pages(bp, flags);
686		if (unlikely(error)) {
687			xfs_warn_ratelimited(target->bt_mount,
688				"%s: failed to map %u pages", __func__,
689				bp->b_page_count);
690			xfs_buf_relse(bp);
691			return error;
692		}
693	}
694
695	/*
696	 * Clear b_error if this is a lookup from a caller that doesn't expect
697	 * valid data to be found in the buffer.
698	 */
699	if (!(flags & XBF_READ))
700		xfs_buf_ioerror(bp, 0);
701
702	XFS_STATS_INC(target->bt_mount, xb_get);
703	trace_xfs_buf_get(bp, flags, _RET_IP_);
704	*bpp = bp;
705	return 0;
706out_free_buf:
707	xfs_buf_free(new_bp);
708	return error;
709}
710
711int
712_xfs_buf_read(
713	struct xfs_buf		*bp,
714	xfs_buf_flags_t		flags)
715{
716	ASSERT(!(flags & XBF_WRITE));
717	ASSERT(bp->b_maps[0].bm_bn != XFS_BUF_DADDR_NULL);
718
719	bp->b_flags &= ~(XBF_WRITE | XBF_ASYNC | XBF_READ_AHEAD | XBF_DONE);
720	bp->b_flags |= flags & (XBF_READ | XBF_ASYNC | XBF_READ_AHEAD);
721
722	return xfs_buf_submit(bp);
723}
724
725/*
726 * Reverify a buffer found in cache without an attached ->b_ops.
727 *
728 * If the caller passed an ops structure and the buffer doesn't have ops
729 * assigned, set the ops and use it to verify the contents. If verification
730 * fails, clear XBF_DONE. We assume the buffer has no recorded errors and is
731 * already in XBF_DONE state on entry.
732 *
733 * Under normal operations, every in-core buffer is verified on read I/O
734 * completion. There are two scenarios that can lead to in-core buffers without
735 * an assigned ->b_ops. The first is during log recovery of buffers on a V4
736 * filesystem, though these buffers are purged at the end of recovery. The
737 * other is online repair, which intentionally reads with a NULL buffer ops to
738 * run several verifiers across an in-core buffer in order to establish buffer
739 * type.  If repair can't establish that, the buffer will be left in memory
740 * with NULL buffer ops.
741 */
742int
743xfs_buf_reverify(
744	struct xfs_buf		*bp,
745	const struct xfs_buf_ops *ops)
746{
747	ASSERT(bp->b_flags & XBF_DONE);
748	ASSERT(bp->b_error == 0);
749
750	if (!ops || bp->b_ops)
751		return 0;
752
753	bp->b_ops = ops;
754	bp->b_ops->verify_read(bp);
755	if (bp->b_error)
756		bp->b_flags &= ~XBF_DONE;
757	return bp->b_error;
758}
759
760int
761xfs_buf_read_map(
762	struct xfs_buftarg	*target,
763	struct xfs_buf_map	*map,
764	int			nmaps,
765	xfs_buf_flags_t		flags,
766	struct xfs_buf		**bpp,
767	const struct xfs_buf_ops *ops,
768	xfs_failaddr_t		fa)
769{
770	struct xfs_buf		*bp;
771	int			error;
772
773	flags |= XBF_READ;
774	*bpp = NULL;
775
776	error = xfs_buf_get_map(target, map, nmaps, flags, &bp);
777	if (error)
778		return error;
779
780	trace_xfs_buf_read(bp, flags, _RET_IP_);
781
782	if (!(bp->b_flags & XBF_DONE)) {
783		/* Initiate the buffer read and wait. */
784		XFS_STATS_INC(target->bt_mount, xb_get_read);
785		bp->b_ops = ops;
786		error = _xfs_buf_read(bp, flags);
787
788		/* Readahead iodone already dropped the buffer, so exit. */
789		if (flags & XBF_ASYNC)
790			return 0;
791	} else {
792		/* Buffer already read; all we need to do is check it. */
793		error = xfs_buf_reverify(bp, ops);
794
795		/* Readahead already finished; drop the buffer and exit. */
796		if (flags & XBF_ASYNC) {
797			xfs_buf_relse(bp);
798			return 0;
799		}
800
801		/* We do not want read in the flags */
802		bp->b_flags &= ~XBF_READ;
803		ASSERT(bp->b_ops != NULL || ops == NULL);
804	}
805
806	/*
807	 * If we've had a read error, then the contents of the buffer are
808	 * invalid and should not be used. To ensure that a followup read tries
809	 * to pull the buffer from disk again, we clear the XBF_DONE flag and
810	 * mark the buffer stale. This ensures that anyone who has a current
811	 * reference to the buffer will interpret it's contents correctly and
812	 * future cache lookups will also treat it as an empty, uninitialised
813	 * buffer.
814	 */
815	if (error) {
816		if (!xfs_is_shutdown(target->bt_mount))
817			xfs_buf_ioerror_alert(bp, fa);
818
819		bp->b_flags &= ~XBF_DONE;
820		xfs_buf_stale(bp);
821		xfs_buf_relse(bp);
822
823		/* bad CRC means corrupted metadata */
824		if (error == -EFSBADCRC)
825			error = -EFSCORRUPTED;
826		return error;
827	}
828
829	*bpp = bp;
830	return 0;
831}
832
833/*
834 *	If we are not low on memory then do the readahead in a deadlock
835 *	safe manner.
836 */
837void
838xfs_buf_readahead_map(
839	struct xfs_buftarg	*target,
840	struct xfs_buf_map	*map,
841	int			nmaps,
842	const struct xfs_buf_ops *ops)
843{
844	struct xfs_buf		*bp;
845
846	if (bdi_read_congested(target->bt_bdev->bd_bdi))
847		return;
848
849	xfs_buf_read_map(target, map, nmaps,
850		     XBF_TRYLOCK | XBF_ASYNC | XBF_READ_AHEAD, &bp, ops,
851		     __this_address);
852}
853
854/*
855 * Read an uncached buffer from disk. Allocates and returns a locked
856 * buffer containing the disk contents or nothing.
857 */
858int
859xfs_buf_read_uncached(
860	struct xfs_buftarg	*target,
861	xfs_daddr_t		daddr,
862	size_t			numblks,
863	int			flags,
864	struct xfs_buf		**bpp,
865	const struct xfs_buf_ops *ops)
866{
867	struct xfs_buf		*bp;
868	int			error;
869
870	*bpp = NULL;
871
872	error = xfs_buf_get_uncached(target, numblks, flags, &bp);
873	if (error)
874		return error;
875
876	/* set up the buffer for a read IO */
877	ASSERT(bp->b_map_count == 1);
878	bp->b_bn = XFS_BUF_DADDR_NULL;  /* always null for uncached buffers */
879	bp->b_maps[0].bm_bn = daddr;
880	bp->b_flags |= XBF_READ;
881	bp->b_ops = ops;
882
883	xfs_buf_submit(bp);
884	if (bp->b_error) {
885		error = bp->b_error;
886		xfs_buf_relse(bp);
887		return error;
888	}
889
890	*bpp = bp;
891	return 0;
892}
893
894int
895xfs_buf_get_uncached(
896	struct xfs_buftarg	*target,
897	size_t			numblks,
898	int			flags,
899	struct xfs_buf		**bpp)
900{
901	int			error;
902	struct xfs_buf		*bp;
903	DEFINE_SINGLE_BUF_MAP(map, XFS_BUF_DADDR_NULL, numblks);
904
905	*bpp = NULL;
906
907	/* flags might contain irrelevant bits, pass only what we care about */
908	error = _xfs_buf_alloc(target, &map, 1, flags & XBF_NO_IOACCT, &bp);
909	if (error)
910		return error;
911
912	error = xfs_buf_alloc_pages(bp, flags);
913	if (error)
914		goto fail_free_buf;
915
916	error = _xfs_buf_map_pages(bp, 0);
917	if (unlikely(error)) {
918		xfs_warn(target->bt_mount,
919			"%s: failed to map pages", __func__);
920		goto fail_free_buf;
921	}
922
923	trace_xfs_buf_get_uncached(bp, _RET_IP_);
924	*bpp = bp;
925	return 0;
926
927fail_free_buf:
928	xfs_buf_free(bp);
929	return error;
930}
931
932/*
933 *	Increment reference count on buffer, to hold the buffer concurrently
934 *	with another thread which may release (free) the buffer asynchronously.
935 *	Must hold the buffer already to call this function.
936 */
937void
938xfs_buf_hold(
939	struct xfs_buf		*bp)
940{
941	trace_xfs_buf_hold(bp, _RET_IP_);
942	atomic_inc(&bp->b_hold);
943}
944
945/*
946 * Release a hold on the specified buffer. If the hold count is 1, the buffer is
947 * placed on LRU or freed (depending on b_lru_ref).
948 */
949void
950xfs_buf_rele(
951	struct xfs_buf		*bp)
952{
953	struct xfs_perag	*pag = bp->b_pag;
954	bool			release;
955	bool			freebuf = false;
956
957	trace_xfs_buf_rele(bp, _RET_IP_);
958
959	if (!pag) {
960		ASSERT(list_empty(&bp->b_lru));
961		if (atomic_dec_and_test(&bp->b_hold)) {
962			xfs_buf_ioacct_dec(bp);
963			xfs_buf_free(bp);
964		}
965		return;
966	}
967
968	ASSERT(atomic_read(&bp->b_hold) > 0);
969
970	/*
971	 * We grab the b_lock here first to serialise racing xfs_buf_rele()
972	 * calls. The pag_buf_lock being taken on the last reference only
973	 * serialises against racing lookups in xfs_buf_find(). IOWs, the second
974	 * to last reference we drop here is not serialised against the last
975	 * reference until we take bp->b_lock. Hence if we don't grab b_lock
976	 * first, the last "release" reference can win the race to the lock and
977	 * free the buffer before the second-to-last reference is processed,
978	 * leading to a use-after-free scenario.
979	 */
980	spin_lock(&bp->b_lock);
981	release = atomic_dec_and_lock(&bp->b_hold, &pag->pag_buf_lock);
982	if (!release) {
983		/*
984		 * Drop the in-flight state if the buffer is already on the LRU
985		 * and it holds the only reference. This is racy because we
986		 * haven't acquired the pag lock, but the use of _XBF_IN_FLIGHT
987		 * ensures the decrement occurs only once per-buf.
988		 */
989		if ((atomic_read(&bp->b_hold) == 1) && !list_empty(&bp->b_lru))
990			__xfs_buf_ioacct_dec(bp);
991		goto out_unlock;
992	}
993
994	/* the last reference has been dropped ... */
995	__xfs_buf_ioacct_dec(bp);
996	if (!(bp->b_flags & XBF_STALE) && atomic_read(&bp->b_lru_ref)) {
997		/*
998		 * If the buffer is added to the LRU take a new reference to the
999		 * buffer for the LRU and clear the (now stale) dispose list
1000		 * state flag
1001		 */
1002		if (list_lru_add(&bp->b_target->bt_lru, &bp->b_lru)) {
1003			bp->b_state &= ~XFS_BSTATE_DISPOSE;
1004			atomic_inc(&bp->b_hold);
1005		}
1006		spin_unlock(&pag->pag_buf_lock);
1007	} else {
1008		/*
1009		 * most of the time buffers will already be removed from the
1010		 * LRU, so optimise that case by checking for the
1011		 * XFS_BSTATE_DISPOSE flag indicating the last list the buffer
1012		 * was on was the disposal list
1013		 */
1014		if (!(bp->b_state & XFS_BSTATE_DISPOSE)) {
1015			list_lru_del(&bp->b_target->bt_lru, &bp->b_lru);
1016		} else {
1017			ASSERT(list_empty(&bp->b_lru));
1018		}
1019
1020		ASSERT(!(bp->b_flags & _XBF_DELWRI_Q));
1021		rhashtable_remove_fast(&pag->pag_buf_hash, &bp->b_rhash_head,
1022				       xfs_buf_hash_params);
1023		spin_unlock(&pag->pag_buf_lock);
1024		xfs_perag_put(pag);
1025		freebuf = true;
1026	}
1027
1028out_unlock:
1029	spin_unlock(&bp->b_lock);
1030
1031	if (freebuf)
1032		xfs_buf_free(bp);
1033}
1034
1035
1036/*
1037 *	Lock a buffer object, if it is not already locked.
1038 *
1039 *	If we come across a stale, pinned, locked buffer, we know that we are
1040 *	being asked to lock a buffer that has been reallocated. Because it is
1041 *	pinned, we know that the log has not been pushed to disk and hence it
1042 *	will still be locked.  Rather than continuing to have trylock attempts
1043 *	fail until someone else pushes the log, push it ourselves before
1044 *	returning.  This means that the xfsaild will not get stuck trying
1045 *	to push on stale inode buffers.
1046 */
1047int
1048xfs_buf_trylock(
1049	struct xfs_buf		*bp)
1050{
1051	int			locked;
1052
1053	locked = down_trylock(&bp->b_sema) == 0;
1054	if (locked)
1055		trace_xfs_buf_trylock(bp, _RET_IP_);
1056	else
1057		trace_xfs_buf_trylock_fail(bp, _RET_IP_);
1058	return locked;
1059}
1060
1061/*
1062 *	Lock a buffer object.
1063 *
1064 *	If we come across a stale, pinned, locked buffer, we know that we
1065 *	are being asked to lock a buffer that has been reallocated. Because
1066 *	it is pinned, we know that the log has not been pushed to disk and
1067 *	hence it will still be locked. Rather than sleeping until someone
1068 *	else pushes the log, push it ourselves before trying to get the lock.
1069 */
1070void
1071xfs_buf_lock(
1072	struct xfs_buf		*bp)
1073{
1074	trace_xfs_buf_lock(bp, _RET_IP_);
1075
1076	if (atomic_read(&bp->b_pin_count) && (bp->b_flags & XBF_STALE))
1077		xfs_log_force(bp->b_mount, 0);
1078	down(&bp->b_sema);
1079
1080	trace_xfs_buf_lock_done(bp, _RET_IP_);
1081}
1082
1083void
1084xfs_buf_unlock(
1085	struct xfs_buf		*bp)
1086{
1087	ASSERT(xfs_buf_islocked(bp));
1088
1089	up(&bp->b_sema);
1090	trace_xfs_buf_unlock(bp, _RET_IP_);
1091}
1092
1093STATIC void
1094xfs_buf_wait_unpin(
1095	struct xfs_buf		*bp)
1096{
1097	DECLARE_WAITQUEUE	(wait, current);
1098
1099	if (atomic_read(&bp->b_pin_count) == 0)
1100		return;
1101
1102	add_wait_queue(&bp->b_waiters, &wait);
1103	for (;;) {
1104		set_current_state(TASK_UNINTERRUPTIBLE);
1105		if (atomic_read(&bp->b_pin_count) == 0)
1106			break;
1107		io_schedule();
1108	}
1109	remove_wait_queue(&bp->b_waiters, &wait);
1110	set_current_state(TASK_RUNNING);
1111}
1112
1113static void
1114xfs_buf_ioerror_alert_ratelimited(
1115	struct xfs_buf		*bp)
1116{
1117	static unsigned long	lasttime;
1118	static struct xfs_buftarg *lasttarg;
1119
1120	if (bp->b_target != lasttarg ||
1121	    time_after(jiffies, (lasttime + 5*HZ))) {
1122		lasttime = jiffies;
1123		xfs_buf_ioerror_alert(bp, __this_address);
1124	}
1125	lasttarg = bp->b_target;
1126}
1127
1128/*
1129 * Account for this latest trip around the retry handler, and decide if
1130 * we've failed enough times to constitute a permanent failure.
1131 */
1132static bool
1133xfs_buf_ioerror_permanent(
1134	struct xfs_buf		*bp,
1135	struct xfs_error_cfg	*cfg)
1136{
1137	struct xfs_mount	*mp = bp->b_mount;
1138
1139	if (cfg->max_retries != XFS_ERR_RETRY_FOREVER &&
1140	    ++bp->b_retries > cfg->max_retries)
1141		return true;
1142	if (cfg->retry_timeout != XFS_ERR_RETRY_FOREVER &&
1143	    time_after(jiffies, cfg->retry_timeout + bp->b_first_retry_time))
1144		return true;
1145
1146	/* At unmount we may treat errors differently */
1147	if (xfs_is_unmounting(mp) && mp->m_fail_unmount)
1148		return true;
1149
1150	return false;
1151}
1152
1153/*
1154 * On a sync write or shutdown we just want to stale the buffer and let the
1155 * caller handle the error in bp->b_error appropriately.
1156 *
1157 * If the write was asynchronous then no one will be looking for the error.  If
1158 * this is the first failure of this type, clear the error state and write the
1159 * buffer out again. This means we always retry an async write failure at least
1160 * once, but we also need to set the buffer up to behave correctly now for
1161 * repeated failures.
1162 *
1163 * If we get repeated async write failures, then we take action according to the
1164 * error configuration we have been set up to use.
1165 *
1166 * Returns true if this function took care of error handling and the caller must
1167 * not touch the buffer again.  Return false if the caller should proceed with
1168 * normal I/O completion handling.
1169 */
1170static bool
1171xfs_buf_ioend_handle_error(
1172	struct xfs_buf		*bp)
1173{
1174	struct xfs_mount	*mp = bp->b_mount;
1175	struct xfs_error_cfg	*cfg;
1176
1177	/*
1178	 * If we've already decided to shutdown the filesystem because of I/O
1179	 * errors, there's no point in giving this a retry.
1180	 */
1181	if (xfs_is_shutdown(mp))
1182		goto out_stale;
1183
1184	xfs_buf_ioerror_alert_ratelimited(bp);
1185
1186	/*
1187	 * We're not going to bother about retrying this during recovery.
1188	 * One strike!
1189	 */
1190	if (bp->b_flags & _XBF_LOGRECOVERY) {
1191		xfs_force_shutdown(mp, SHUTDOWN_META_IO_ERROR);
1192		return false;
1193	}
1194
1195	/*
1196	 * Synchronous writes will have callers process the error.
1197	 */
1198	if (!(bp->b_flags & XBF_ASYNC))
1199		goto out_stale;
1200
1201	trace_xfs_buf_iodone_async(bp, _RET_IP_);
1202
1203	cfg = xfs_error_get_cfg(mp, XFS_ERR_METADATA, bp->b_error);
1204	if (bp->b_last_error != bp->b_error ||
1205	    !(bp->b_flags & (XBF_STALE | XBF_WRITE_FAIL))) {
1206		bp->b_last_error = bp->b_error;
1207		if (cfg->retry_timeout != XFS_ERR_RETRY_FOREVER &&
1208		    !bp->b_first_retry_time)
1209			bp->b_first_retry_time = jiffies;
1210		goto resubmit;
1211	}
1212
1213	/*
1214	 * Permanent error - we need to trigger a shutdown if we haven't already
1215	 * to indicate that inconsistency will result from this action.
1216	 */
1217	if (xfs_buf_ioerror_permanent(bp, cfg)) {
1218		xfs_force_shutdown(mp, SHUTDOWN_META_IO_ERROR);
1219		goto out_stale;
1220	}
1221
1222	/* Still considered a transient error. Caller will schedule retries. */
1223	if (bp->b_flags & _XBF_INODES)
1224		xfs_buf_inode_io_fail(bp);
1225	else if (bp->b_flags & _XBF_DQUOTS)
1226		xfs_buf_dquot_io_fail(bp);
1227	else
1228		ASSERT(list_empty(&bp->b_li_list));
1229	xfs_buf_ioerror(bp, 0);
1230	xfs_buf_relse(bp);
1231	return true;
1232
1233resubmit:
1234	xfs_buf_ioerror(bp, 0);
1235	bp->b_flags |= (XBF_DONE | XBF_WRITE_FAIL);
1236	xfs_buf_submit(bp);
1237	return true;
1238out_stale:
1239	xfs_buf_stale(bp);
1240	bp->b_flags |= XBF_DONE;
1241	bp->b_flags &= ~XBF_WRITE;
1242	trace_xfs_buf_error_relse(bp, _RET_IP_);
1243	return false;
1244}
1245
1246static void
1247xfs_buf_ioend(
1248	struct xfs_buf	*bp)
1249{
1250	trace_xfs_buf_iodone(bp, _RET_IP_);
1251
1252	/*
1253	 * Pull in IO completion errors now. We are guaranteed to be running
1254	 * single threaded, so we don't need the lock to read b_io_error.
1255	 */
1256	if (!bp->b_error && bp->b_io_error)
1257		xfs_buf_ioerror(bp, bp->b_io_error);
1258
1259	if (bp->b_flags & XBF_READ) {
1260		if (!bp->b_error && bp->b_ops)
1261			bp->b_ops->verify_read(bp);
1262		if (!bp->b_error)
1263			bp->b_flags |= XBF_DONE;
1264	} else {
1265		if (!bp->b_error) {
1266			bp->b_flags &= ~XBF_WRITE_FAIL;
1267			bp->b_flags |= XBF_DONE;
1268		}
1269
1270		if (unlikely(bp->b_error) && xfs_buf_ioend_handle_error(bp))
1271			return;
1272
1273		/* clear the retry state */
1274		bp->b_last_error = 0;
1275		bp->b_retries = 0;
1276		bp->b_first_retry_time = 0;
1277
1278		/*
1279		 * Note that for things like remote attribute buffers, there may
1280		 * not be a buffer log item here, so processing the buffer log
1281		 * item must remain optional.
1282		 */
1283		if (bp->b_log_item)
1284			xfs_buf_item_done(bp);
1285
1286		if (bp->b_flags & _XBF_INODES)
1287			xfs_buf_inode_iodone(bp);
1288		else if (bp->b_flags & _XBF_DQUOTS)
1289			xfs_buf_dquot_iodone(bp);
1290
1291	}
1292
1293	bp->b_flags &= ~(XBF_READ | XBF_WRITE | XBF_READ_AHEAD |
1294			 _XBF_LOGRECOVERY);
1295
1296	if (bp->b_flags & XBF_ASYNC)
1297		xfs_buf_relse(bp);
1298	else
1299		complete(&bp->b_iowait);
1300}
1301
1302static void
1303xfs_buf_ioend_work(
1304	struct work_struct	*work)
1305{
1306	struct xfs_buf		*bp =
1307		container_of(work, struct xfs_buf, b_ioend_work);
1308
1309	xfs_buf_ioend(bp);
1310}
1311
1312static void
1313xfs_buf_ioend_async(
1314	struct xfs_buf	*bp)
1315{
1316	INIT_WORK(&bp->b_ioend_work, xfs_buf_ioend_work);
1317	queue_work(bp->b_mount->m_buf_workqueue, &bp->b_ioend_work);
1318}
1319
1320void
1321__xfs_buf_ioerror(
1322	struct xfs_buf		*bp,
1323	int			error,
1324	xfs_failaddr_t		failaddr)
1325{
1326	ASSERT(error <= 0 && error >= -1000);
1327	bp->b_error = error;
1328	trace_xfs_buf_ioerror(bp, error, failaddr);
1329}
1330
1331void
1332xfs_buf_ioerror_alert(
1333	struct xfs_buf		*bp,
1334	xfs_failaddr_t		func)
1335{
1336	xfs_buf_alert_ratelimited(bp, "XFS: metadata IO error",
1337		"metadata I/O error in \"%pS\" at daddr 0x%llx len %d error %d",
1338				  func, (uint64_t)xfs_buf_daddr(bp),
1339				  bp->b_length, -bp->b_error);
1340}
1341
1342/*
1343 * To simulate an I/O failure, the buffer must be locked and held with at least
1344 * three references. The LRU reference is dropped by the stale call. The buf
1345 * item reference is dropped via ioend processing. The third reference is owned
1346 * by the caller and is dropped on I/O completion if the buffer is XBF_ASYNC.
1347 */
1348void
1349xfs_buf_ioend_fail(
1350	struct xfs_buf	*bp)
1351{
1352	bp->b_flags &= ~XBF_DONE;
1353	xfs_buf_stale(bp);
1354	xfs_buf_ioerror(bp, -EIO);
1355	xfs_buf_ioend(bp);
1356}
1357
1358int
1359xfs_bwrite(
1360	struct xfs_buf		*bp)
1361{
1362	int			error;
1363
1364	ASSERT(xfs_buf_islocked(bp));
1365
1366	bp->b_flags |= XBF_WRITE;
1367	bp->b_flags &= ~(XBF_ASYNC | XBF_READ | _XBF_DELWRI_Q |
1368			 XBF_DONE);
1369
1370	error = xfs_buf_submit(bp);
1371	if (error)
1372		xfs_force_shutdown(bp->b_mount, SHUTDOWN_META_IO_ERROR);
1373	return error;
1374}
1375
1376static void
1377xfs_buf_bio_end_io(
1378	struct bio		*bio)
1379{
1380	struct xfs_buf		*bp = (struct xfs_buf *)bio->bi_private;
1381
1382	if (!bio->bi_status &&
1383	    (bp->b_flags & XBF_WRITE) && (bp->b_flags & XBF_ASYNC) &&
1384	    XFS_TEST_ERROR(false, bp->b_mount, XFS_ERRTAG_BUF_IOERROR))
1385		bio->bi_status = BLK_STS_IOERR;
1386
1387	/*
1388	 * don't overwrite existing errors - otherwise we can lose errors on
1389	 * buffers that require multiple bios to complete.
1390	 */
1391	if (bio->bi_status) {
1392		int error = blk_status_to_errno(bio->bi_status);
1393
1394		cmpxchg(&bp->b_io_error, 0, error);
1395	}
1396
1397	if (!bp->b_error && xfs_buf_is_vmapped(bp) && (bp->b_flags & XBF_READ))
1398		invalidate_kernel_vmap_range(bp->b_addr, xfs_buf_vmap_len(bp));
1399
1400	if (atomic_dec_and_test(&bp->b_io_remaining) == 1)
1401		xfs_buf_ioend_async(bp);
1402	bio_put(bio);
1403}
1404
1405static void
1406xfs_buf_ioapply_map(
1407	struct xfs_buf	*bp,
1408	int		map,
1409	int		*buf_offset,
1410	int		*count,
1411	int		op)
1412{
1413	int		page_index;
1414	unsigned int	total_nr_pages = bp->b_page_count;
1415	int		nr_pages;
1416	struct bio	*bio;
1417	sector_t	sector =  bp->b_maps[map].bm_bn;
1418	int		size;
1419	int		offset;
1420
1421	/* skip the pages in the buffer before the start offset */
1422	page_index = 0;
1423	offset = *buf_offset;
1424	while (offset >= PAGE_SIZE) {
1425		page_index++;
1426		offset -= PAGE_SIZE;
1427	}
1428
1429	/*
1430	 * Limit the IO size to the length of the current vector, and update the
1431	 * remaining IO count for the next time around.
1432	 */
1433	size = min_t(int, BBTOB(bp->b_maps[map].bm_len), *count);
1434	*count -= size;
1435	*buf_offset += size;
1436
1437next_chunk:
1438	atomic_inc(&bp->b_io_remaining);
1439	nr_pages = bio_max_segs(total_nr_pages);
1440
1441	bio = bio_alloc(GFP_NOIO, nr_pages);
1442	bio_set_dev(bio, bp->b_target->bt_bdev);
1443	bio->bi_iter.bi_sector = sector;
1444	bio->bi_end_io = xfs_buf_bio_end_io;
1445	bio->bi_private = bp;
1446	bio->bi_opf = op;
1447
1448	for (; size && nr_pages; nr_pages--, page_index++) {
1449		int	rbytes, nbytes = PAGE_SIZE - offset;
1450
1451		if (nbytes > size)
1452			nbytes = size;
1453
1454		rbytes = bio_add_page(bio, bp->b_pages[page_index], nbytes,
1455				      offset);
1456		if (rbytes < nbytes)
1457			break;
1458
1459		offset = 0;
1460		sector += BTOBB(nbytes);
1461		size -= nbytes;
1462		total_nr_pages--;
1463	}
1464
1465	if (likely(bio->bi_iter.bi_size)) {
1466		if (xfs_buf_is_vmapped(bp)) {
1467			flush_kernel_vmap_range(bp->b_addr,
1468						xfs_buf_vmap_len(bp));
1469		}
1470		submit_bio(bio);
1471		if (size)
1472			goto next_chunk;
1473	} else {
1474		/*
1475		 * This is guaranteed not to be the last io reference count
1476		 * because the caller (xfs_buf_submit) holds a count itself.
1477		 */
1478		atomic_dec(&bp->b_io_remaining);
1479		xfs_buf_ioerror(bp, -EIO);
1480		bio_put(bio);
1481	}
1482
1483}
1484
1485STATIC void
1486_xfs_buf_ioapply(
1487	struct xfs_buf	*bp)
1488{
1489	struct blk_plug	plug;
1490	int		op;
1491	int		offset;
1492	int		size;
1493	int		i;
1494
1495	/*
1496	 * Make sure we capture only current IO errors rather than stale errors
1497	 * left over from previous use of the buffer (e.g. failed readahead).
1498	 */
1499	bp->b_error = 0;
1500
1501	if (bp->b_flags & XBF_WRITE) {
1502		op = REQ_OP_WRITE;
1503
1504		/*
1505		 * Run the write verifier callback function if it exists. If
1506		 * this function fails it will mark the buffer with an error and
1507		 * the IO should not be dispatched.
1508		 */
1509		if (bp->b_ops) {
1510			bp->b_ops->verify_write(bp);
1511			if (bp->b_error) {
1512				xfs_force_shutdown(bp->b_mount,
1513						   SHUTDOWN_CORRUPT_INCORE);
1514				return;
1515			}
1516		} else if (bp->b_bn != XFS_BUF_DADDR_NULL) {
1517			struct xfs_mount *mp = bp->b_mount;
1518
1519			/*
1520			 * non-crc filesystems don't attach verifiers during
1521			 * log recovery, so don't warn for such filesystems.
1522			 */
1523			if (xfs_has_crc(mp)) {
1524				xfs_warn(mp,
1525					"%s: no buf ops on daddr 0x%llx len %d",
1526					__func__, bp->b_bn, bp->b_length);
1527				xfs_hex_dump(bp->b_addr,
1528						XFS_CORRUPTION_DUMP_LEN);
1529				dump_stack();
1530			}
1531		}
1532	} else {
1533		op = REQ_OP_READ;
1534		if (bp->b_flags & XBF_READ_AHEAD)
1535			op |= REQ_RAHEAD;
1536	}
1537
1538	/* we only use the buffer cache for meta-data */
1539	op |= REQ_META;
1540
1541	/*
1542	 * Walk all the vectors issuing IO on them. Set up the initial offset
1543	 * into the buffer and the desired IO size before we start -
1544	 * _xfs_buf_ioapply_vec() will modify them appropriately for each
1545	 * subsequent call.
1546	 */
1547	offset = bp->b_offset;
1548	size = BBTOB(bp->b_length);
1549	blk_start_plug(&plug);
1550	for (i = 0; i < bp->b_map_count; i++) {
1551		xfs_buf_ioapply_map(bp, i, &offset, &size, op);
1552		if (bp->b_error)
1553			break;
1554		if (size <= 0)
1555			break;	/* all done */
1556	}
1557	blk_finish_plug(&plug);
1558}
1559
1560/*
1561 * Wait for I/O completion of a sync buffer and return the I/O error code.
1562 */
1563static int
1564xfs_buf_iowait(
1565	struct xfs_buf	*bp)
1566{
1567	ASSERT(!(bp->b_flags & XBF_ASYNC));
1568
1569	trace_xfs_buf_iowait(bp, _RET_IP_);
1570	wait_for_completion(&bp->b_iowait);
1571	trace_xfs_buf_iowait_done(bp, _RET_IP_);
1572
1573	return bp->b_error;
1574}
1575
1576/*
1577 * Buffer I/O submission path, read or write. Asynchronous submission transfers
1578 * the buffer lock ownership and the current reference to the IO. It is not
1579 * safe to reference the buffer after a call to this function unless the caller
1580 * holds an additional reference itself.
1581 */
1582static int
1583__xfs_buf_submit(
1584	struct xfs_buf	*bp,
1585	bool		wait)
1586{
1587	int		error = 0;
1588
1589	trace_xfs_buf_submit(bp, _RET_IP_);
1590
1591	ASSERT(!(bp->b_flags & _XBF_DELWRI_Q));
1592
1593	/* on shutdown we stale and complete the buffer immediately */
1594	if (xfs_is_shutdown(bp->b_mount)) {
1595		xfs_buf_ioend_fail(bp);
1596		return -EIO;
1597	}
1598
1599	/*
1600	 * Grab a reference so the buffer does not go away underneath us. For
1601	 * async buffers, I/O completion drops the callers reference, which
1602	 * could occur before submission returns.
1603	 */
1604	xfs_buf_hold(bp);
1605
1606	if (bp->b_flags & XBF_WRITE)
1607		xfs_buf_wait_unpin(bp);
1608
1609	/* clear the internal error state to avoid spurious errors */
1610	bp->b_io_error = 0;
1611
1612	/*
1613	 * Set the count to 1 initially, this will stop an I/O completion
1614	 * callout which happens before we have started all the I/O from calling
1615	 * xfs_buf_ioend too early.
1616	 */
1617	atomic_set(&bp->b_io_remaining, 1);
1618	if (bp->b_flags & XBF_ASYNC)
1619		xfs_buf_ioacct_inc(bp);
1620	_xfs_buf_ioapply(bp);
1621
1622	/*
1623	 * If _xfs_buf_ioapply failed, we can get back here with only the IO
1624	 * reference we took above. If we drop it to zero, run completion so
1625	 * that we don't return to the caller with completion still pending.
1626	 */
1627	if (atomic_dec_and_test(&bp->b_io_remaining) == 1) {
1628		if (bp->b_error || !(bp->b_flags & XBF_ASYNC))
1629			xfs_buf_ioend(bp);
1630		else
1631			xfs_buf_ioend_async(bp);
1632	}
1633
1634	if (wait)
1635		error = xfs_buf_iowait(bp);
1636
1637	/*
1638	 * Release the hold that keeps the buffer referenced for the entire
1639	 * I/O. Note that if the buffer is async, it is not safe to reference
1640	 * after this release.
1641	 */
1642	xfs_buf_rele(bp);
1643	return error;
1644}
1645
1646void *
1647xfs_buf_offset(
1648	struct xfs_buf		*bp,
1649	size_t			offset)
1650{
1651	struct page		*page;
1652
1653	if (bp->b_addr)
1654		return bp->b_addr + offset;
1655
1656	page = bp->b_pages[offset >> PAGE_SHIFT];
1657	return page_address(page) + (offset & (PAGE_SIZE-1));
1658}
1659
1660void
1661xfs_buf_zero(
1662	struct xfs_buf		*bp,
1663	size_t			boff,
1664	size_t			bsize)
1665{
1666	size_t			bend;
1667
1668	bend = boff + bsize;
1669	while (boff < bend) {
1670		struct page	*page;
1671		int		page_index, page_offset, csize;
1672
1673		page_index = (boff + bp->b_offset) >> PAGE_SHIFT;
1674		page_offset = (boff + bp->b_offset) & ~PAGE_MASK;
1675		page = bp->b_pages[page_index];
1676		csize = min_t(size_t, PAGE_SIZE - page_offset,
1677				      BBTOB(bp->b_length) - boff);
1678
1679		ASSERT((csize + page_offset) <= PAGE_SIZE);
1680
1681		memset(page_address(page) + page_offset, 0, csize);
1682
1683		boff += csize;
1684	}
1685}
1686
1687/*
1688 * Log a message about and stale a buffer that a caller has decided is corrupt.
1689 *
1690 * This function should be called for the kinds of metadata corruption that
1691 * cannot be detect from a verifier, such as incorrect inter-block relationship
1692 * data.  Do /not/ call this function from a verifier function.
1693 *
1694 * The buffer must be XBF_DONE prior to the call.  Afterwards, the buffer will
1695 * be marked stale, but b_error will not be set.  The caller is responsible for
1696 * releasing the buffer or fixing it.
1697 */
1698void
1699__xfs_buf_mark_corrupt(
1700	struct xfs_buf		*bp,
1701	xfs_failaddr_t		fa)
1702{
1703	ASSERT(bp->b_flags & XBF_DONE);
1704
1705	xfs_buf_corruption_error(bp, fa);
1706	xfs_buf_stale(bp);
1707}
1708
1709/*
1710 *	Handling of buffer targets (buftargs).
1711 */
1712
1713/*
1714 * Wait for any bufs with callbacks that have been submitted but have not yet
1715 * returned. These buffers will have an elevated hold count, so wait on those
1716 * while freeing all the buffers only held by the LRU.
1717 */
1718static enum lru_status
1719xfs_buftarg_drain_rele(
1720	struct list_head	*item,
1721	struct list_lru_one	*lru,
1722	spinlock_t		*lru_lock,
1723	void			*arg)
1724
1725{
1726	struct xfs_buf		*bp = container_of(item, struct xfs_buf, b_lru);
1727	struct list_head	*dispose = arg;
1728
1729	if (atomic_read(&bp->b_hold) > 1) {
1730		/* need to wait, so skip it this pass */
1731		trace_xfs_buf_drain_buftarg(bp, _RET_IP_);
1732		return LRU_SKIP;
1733	}
1734	if (!spin_trylock(&bp->b_lock))
1735		return LRU_SKIP;
1736
1737	/*
1738	 * clear the LRU reference count so the buffer doesn't get
1739	 * ignored in xfs_buf_rele().
1740	 */
1741	atomic_set(&bp->b_lru_ref, 0);
1742	bp->b_state |= XFS_BSTATE_DISPOSE;
1743	list_lru_isolate_move(lru, item, dispose);
1744	spin_unlock(&bp->b_lock);
1745	return LRU_REMOVED;
1746}
1747
1748/*
1749 * Wait for outstanding I/O on the buftarg to complete.
1750 */
1751void
1752xfs_buftarg_wait(
1753	struct xfs_buftarg	*btp)
1754{
1755	/*
1756	 * First wait on the buftarg I/O count for all in-flight buffers to be
1757	 * released. This is critical as new buffers do not make the LRU until
1758	 * they are released.
1759	 *
1760	 * Next, flush the buffer workqueue to ensure all completion processing
1761	 * has finished. Just waiting on buffer locks is not sufficient for
1762	 * async IO as the reference count held over IO is not released until
1763	 * after the buffer lock is dropped. Hence we need to ensure here that
1764	 * all reference counts have been dropped before we start walking the
1765	 * LRU list.
1766	 */
1767	while (percpu_counter_sum(&btp->bt_io_count))
1768		delay(100);
1769	flush_workqueue(btp->bt_mount->m_buf_workqueue);
1770}
1771
1772void
1773xfs_buftarg_drain(
1774	struct xfs_buftarg	*btp)
1775{
1776	LIST_HEAD(dispose);
1777	int			loop = 0;
1778	bool			write_fail = false;
1779
1780	xfs_buftarg_wait(btp);
1781
1782	/* loop until there is nothing left on the lru list. */
1783	while (list_lru_count(&btp->bt_lru)) {
1784		list_lru_walk(&btp->bt_lru, xfs_buftarg_drain_rele,
1785			      &dispose, LONG_MAX);
1786
1787		while (!list_empty(&dispose)) {
1788			struct xfs_buf *bp;
1789			bp = list_first_entry(&dispose, struct xfs_buf, b_lru);
1790			list_del_init(&bp->b_lru);
1791			if (bp->b_flags & XBF_WRITE_FAIL) {
1792				write_fail = true;
1793				xfs_buf_alert_ratelimited(bp,
1794					"XFS: Corruption Alert",
1795"Corruption Alert: Buffer at daddr 0x%llx had permanent write failures!",
1796					(long long)bp->b_bn);
1797			}
1798			xfs_buf_rele(bp);
1799		}
1800		if (loop++ != 0)
1801			delay(100);
1802	}
1803
1804	/*
1805	 * If one or more failed buffers were freed, that means dirty metadata
1806	 * was thrown away. This should only ever happen after I/O completion
1807	 * handling has elevated I/O error(s) to permanent failures and shuts
1808	 * down the fs.
1809	 */
1810	if (write_fail) {
1811		ASSERT(xfs_is_shutdown(btp->bt_mount));
1812		xfs_alert(btp->bt_mount,
1813	      "Please run xfs_repair to determine the extent of the problem.");
1814	}
1815}
1816
1817static enum lru_status
1818xfs_buftarg_isolate(
1819	struct list_head	*item,
1820	struct list_lru_one	*lru,
1821	spinlock_t		*lru_lock,
1822	void			*arg)
1823{
1824	struct xfs_buf		*bp = container_of(item, struct xfs_buf, b_lru);
1825	struct list_head	*dispose = arg;
1826
1827	/*
1828	 * we are inverting the lru lock/bp->b_lock here, so use a trylock.
1829	 * If we fail to get the lock, just skip it.
1830	 */
1831	if (!spin_trylock(&bp->b_lock))
1832		return LRU_SKIP;
1833	/*
1834	 * Decrement the b_lru_ref count unless the value is already
1835	 * zero. If the value is already zero, we need to reclaim the
1836	 * buffer, otherwise it gets another trip through the LRU.
1837	 */
1838	if (atomic_add_unless(&bp->b_lru_ref, -1, 0)) {
1839		spin_unlock(&bp->b_lock);
1840		return LRU_ROTATE;
1841	}
1842
1843	bp->b_state |= XFS_BSTATE_DISPOSE;
1844	list_lru_isolate_move(lru, item, dispose);
1845	spin_unlock(&bp->b_lock);
1846	return LRU_REMOVED;
1847}
1848
1849static unsigned long
1850xfs_buftarg_shrink_scan(
1851	struct shrinker		*shrink,
1852	struct shrink_control	*sc)
1853{
1854	struct xfs_buftarg	*btp = container_of(shrink,
1855					struct xfs_buftarg, bt_shrinker);
1856	LIST_HEAD(dispose);
1857	unsigned long		freed;
1858
1859	freed = list_lru_shrink_walk(&btp->bt_lru, sc,
1860				     xfs_buftarg_isolate, &dispose);
1861
1862	while (!list_empty(&dispose)) {
1863		struct xfs_buf *bp;
1864		bp = list_first_entry(&dispose, struct xfs_buf, b_lru);
1865		list_del_init(&bp->b_lru);
1866		xfs_buf_rele(bp);
1867	}
1868
1869	return freed;
1870}
1871
1872static unsigned long
1873xfs_buftarg_shrink_count(
1874	struct shrinker		*shrink,
1875	struct shrink_control	*sc)
1876{
1877	struct xfs_buftarg	*btp = container_of(shrink,
1878					struct xfs_buftarg, bt_shrinker);
1879	return list_lru_shrink_count(&btp->bt_lru, sc);
1880}
1881
1882void
1883xfs_free_buftarg(
1884	struct xfs_buftarg	*btp)
1885{
1886	unregister_shrinker(&btp->bt_shrinker);
1887	ASSERT(percpu_counter_sum(&btp->bt_io_count) == 0);
1888	percpu_counter_destroy(&btp->bt_io_count);
1889	list_lru_destroy(&btp->bt_lru);
1890
1891	blkdev_issue_flush(btp->bt_bdev);
1892
1893	kmem_free(btp);
1894}
1895
1896int
1897xfs_setsize_buftarg(
1898	xfs_buftarg_t		*btp,
1899	unsigned int		sectorsize)
1900{
1901	/* Set up metadata sector size info */
1902	btp->bt_meta_sectorsize = sectorsize;
1903	btp->bt_meta_sectormask = sectorsize - 1;
1904
1905	if (set_blocksize(btp->bt_bdev, sectorsize)) {
1906		xfs_warn(btp->bt_mount,
1907			"Cannot set_blocksize to %u on device %pg",
1908			sectorsize, btp->bt_bdev);
1909		return -EINVAL;
1910	}
1911
1912	/* Set up device logical sector size mask */
1913	btp->bt_logical_sectorsize = bdev_logical_block_size(btp->bt_bdev);
1914	btp->bt_logical_sectormask = bdev_logical_block_size(btp->bt_bdev) - 1;
1915
1916	return 0;
1917}
1918
1919/*
1920 * When allocating the initial buffer target we have not yet
1921 * read in the superblock, so don't know what sized sectors
1922 * are being used at this early stage.  Play safe.
1923 */
1924STATIC int
1925xfs_setsize_buftarg_early(
1926	xfs_buftarg_t		*btp,
1927	struct block_device	*bdev)
1928{
1929	return xfs_setsize_buftarg(btp, bdev_logical_block_size(bdev));
1930}
1931
1932xfs_buftarg_t *
1933xfs_alloc_buftarg(
1934	struct xfs_mount	*mp,
1935	struct block_device	*bdev,
1936	struct dax_device	*dax_dev)
1937{
1938	xfs_buftarg_t		*btp;
1939
1940	btp = kmem_zalloc(sizeof(*btp), KM_NOFS);
1941
1942	btp->bt_mount = mp;
1943	btp->bt_dev =  bdev->bd_dev;
1944	btp->bt_bdev = bdev;
1945	btp->bt_daxdev = dax_dev;
1946
1947	/*
1948	 * Buffer IO error rate limiting. Limit it to no more than 10 messages
1949	 * per 30 seconds so as to not spam logs too much on repeated errors.
1950	 */
1951	ratelimit_state_init(&btp->bt_ioerror_rl, 30 * HZ,
1952			     DEFAULT_RATELIMIT_BURST);
1953
1954	if (xfs_setsize_buftarg_early(btp, bdev))
1955		goto error_free;
1956
1957	if (list_lru_init(&btp->bt_lru))
1958		goto error_free;
1959
1960	if (percpu_counter_init(&btp->bt_io_count, 0, GFP_KERNEL))
1961		goto error_lru;
1962
1963	btp->bt_shrinker.count_objects = xfs_buftarg_shrink_count;
1964	btp->bt_shrinker.scan_objects = xfs_buftarg_shrink_scan;
1965	btp->bt_shrinker.seeks = DEFAULT_SEEKS;
1966	btp->bt_shrinker.flags = SHRINKER_NUMA_AWARE;
1967	if (register_shrinker(&btp->bt_shrinker))
1968		goto error_pcpu;
1969	return btp;
1970
1971error_pcpu:
1972	percpu_counter_destroy(&btp->bt_io_count);
1973error_lru:
1974	list_lru_destroy(&btp->bt_lru);
1975error_free:
1976	kmem_free(btp);
1977	return NULL;
1978}
1979
1980/*
1981 * Cancel a delayed write list.
1982 *
1983 * Remove each buffer from the list, clear the delwri queue flag and drop the
1984 * associated buffer reference.
1985 */
1986void
1987xfs_buf_delwri_cancel(
1988	struct list_head	*list)
1989{
1990	struct xfs_buf		*bp;
1991
1992	while (!list_empty(list)) {
1993		bp = list_first_entry(list, struct xfs_buf, b_list);
1994
1995		xfs_buf_lock(bp);
1996		bp->b_flags &= ~_XBF_DELWRI_Q;
1997		list_del_init(&bp->b_list);
1998		xfs_buf_relse(bp);
1999	}
2000}
2001
2002/*
2003 * Add a buffer to the delayed write list.
2004 *
2005 * This queues a buffer for writeout if it hasn't already been.  Note that
2006 * neither this routine nor the buffer list submission functions perform
2007 * any internal synchronization.  It is expected that the lists are thread-local
2008 * to the callers.
2009 *
2010 * Returns true if we queued up the buffer, or false if it already had
2011 * been on the buffer list.
2012 */
2013bool
2014xfs_buf_delwri_queue(
2015	struct xfs_buf		*bp,
2016	struct list_head	*list)
2017{
2018	ASSERT(xfs_buf_islocked(bp));
2019	ASSERT(!(bp->b_flags & XBF_READ));
2020
2021	/*
2022	 * If the buffer is already marked delwri it already is queued up
2023	 * by someone else for imediate writeout.  Just ignore it in that
2024	 * case.
2025	 */
2026	if (bp->b_flags & _XBF_DELWRI_Q) {
2027		trace_xfs_buf_delwri_queued(bp, _RET_IP_);
2028		return false;
2029	}
2030
2031	trace_xfs_buf_delwri_queue(bp, _RET_IP_);
2032
2033	/*
2034	 * If a buffer gets written out synchronously or marked stale while it
2035	 * is on a delwri list we lazily remove it. To do this, the other party
2036	 * clears the  _XBF_DELWRI_Q flag but otherwise leaves the buffer alone.
2037	 * It remains referenced and on the list.  In a rare corner case it
2038	 * might get readded to a delwri list after the synchronous writeout, in
2039	 * which case we need just need to re-add the flag here.
2040	 */
2041	bp->b_flags |= _XBF_DELWRI_Q;
2042	if (list_empty(&bp->b_list)) {
2043		atomic_inc(&bp->b_hold);
2044		list_add_tail(&bp->b_list, list);
2045	}
2046
2047	return true;
2048}
2049
2050/*
2051 * Compare function is more complex than it needs to be because
2052 * the return value is only 32 bits and we are doing comparisons
2053 * on 64 bit values
2054 */
2055static int
2056xfs_buf_cmp(
2057	void			*priv,
2058	const struct list_head	*a,
2059	const struct list_head	*b)
2060{
2061	struct xfs_buf	*ap = container_of(a, struct xfs_buf, b_list);
2062	struct xfs_buf	*bp = container_of(b, struct xfs_buf, b_list);
2063	xfs_daddr_t		diff;
2064
2065	diff = ap->b_maps[0].bm_bn - bp->b_maps[0].bm_bn;
2066	if (diff < 0)
2067		return -1;
2068	if (diff > 0)
2069		return 1;
2070	return 0;
2071}
2072
2073/*
2074 * Submit buffers for write. If wait_list is specified, the buffers are
2075 * submitted using sync I/O and placed on the wait list such that the caller can
2076 * iowait each buffer. Otherwise async I/O is used and the buffers are released
2077 * at I/O completion time. In either case, buffers remain locked until I/O
2078 * completes and the buffer is released from the queue.
2079 */
2080static int
2081xfs_buf_delwri_submit_buffers(
2082	struct list_head	*buffer_list,
2083	struct list_head	*wait_list)
2084{
2085	struct xfs_buf		*bp, *n;
2086	int			pinned = 0;
2087	struct blk_plug		plug;
2088
2089	list_sort(NULL, buffer_list, xfs_buf_cmp);
2090
2091	blk_start_plug(&plug);
2092	list_for_each_entry_safe(bp, n, buffer_list, b_list) {
2093		if (!wait_list) {
2094			if (xfs_buf_ispinned(bp)) {
2095				pinned++;
2096				continue;
2097			}
2098			if (!xfs_buf_trylock(bp))
2099				continue;
2100		} else {
2101			xfs_buf_lock(bp);
2102		}
2103
2104		/*
2105		 * Someone else might have written the buffer synchronously or
2106		 * marked it stale in the meantime.  In that case only the
2107		 * _XBF_DELWRI_Q flag got cleared, and we have to drop the
2108		 * reference and remove it from the list here.
2109		 */
2110		if (!(bp->b_flags & _XBF_DELWRI_Q)) {
2111			list_del_init(&bp->b_list);
2112			xfs_buf_relse(bp);
2113			continue;
2114		}
2115
2116		trace_xfs_buf_delwri_split(bp, _RET_IP_);
2117
2118		/*
2119		 * If we have a wait list, each buffer (and associated delwri
2120		 * queue reference) transfers to it and is submitted
2121		 * synchronously. Otherwise, drop the buffer from the delwri
2122		 * queue and submit async.
2123		 */
2124		bp->b_flags &= ~_XBF_DELWRI_Q;
2125		bp->b_flags |= XBF_WRITE;
2126		if (wait_list) {
2127			bp->b_flags &= ~XBF_ASYNC;
2128			list_move_tail(&bp->b_list, wait_list);
2129		} else {
2130			bp->b_flags |= XBF_ASYNC;
2131			list_del_init(&bp->b_list);
2132		}
2133		__xfs_buf_submit(bp, false);
2134	}
2135	blk_finish_plug(&plug);
2136
2137	return pinned;
2138}
2139
2140/*
2141 * Write out a buffer list asynchronously.
2142 *
2143 * This will take the @buffer_list, write all non-locked and non-pinned buffers
2144 * out and not wait for I/O completion on any of the buffers.  This interface
2145 * is only safely useable for callers that can track I/O completion by higher
2146 * level means, e.g. AIL pushing as the @buffer_list is consumed in this
2147 * function.
2148 *
2149 * Note: this function will skip buffers it would block on, and in doing so
2150 * leaves them on @buffer_list so they can be retried on a later pass. As such,
2151 * it is up to the caller to ensure that the buffer list is fully submitted or
2152 * cancelled appropriately when they are finished with the list. Failure to
2153 * cancel or resubmit the list until it is empty will result in leaked buffers
2154 * at unmount time.
2155 */
2156int
2157xfs_buf_delwri_submit_nowait(
2158	struct list_head	*buffer_list)
2159{
2160	return xfs_buf_delwri_submit_buffers(buffer_list, NULL);
2161}
2162
2163/*
2164 * Write out a buffer list synchronously.
2165 *
2166 * This will take the @buffer_list, write all buffers out and wait for I/O
2167 * completion on all of the buffers. @buffer_list is consumed by the function,
2168 * so callers must have some other way of tracking buffers if they require such
2169 * functionality.
2170 */
2171int
2172xfs_buf_delwri_submit(
2173	struct list_head	*buffer_list)
2174{
2175	LIST_HEAD		(wait_list);
2176	int			error = 0, error2;
2177	struct xfs_buf		*bp;
2178
2179	xfs_buf_delwri_submit_buffers(buffer_list, &wait_list);
2180
2181	/* Wait for IO to complete. */
2182	while (!list_empty(&wait_list)) {
2183		bp = list_first_entry(&wait_list, struct xfs_buf, b_list);
2184
2185		list_del_init(&bp->b_list);
2186
2187		/*
2188		 * Wait on the locked buffer, check for errors and unlock and
2189		 * release the delwri queue reference.
2190		 */
2191		error2 = xfs_buf_iowait(bp);
2192		xfs_buf_relse(bp);
2193		if (!error)
2194			error = error2;
2195	}
2196
2197	return error;
2198}
2199
2200/*
2201 * Push a single buffer on a delwri queue.
2202 *
2203 * The purpose of this function is to submit a single buffer of a delwri queue
2204 * and return with the buffer still on the original queue. The waiting delwri
2205 * buffer submission infrastructure guarantees transfer of the delwri queue
2206 * buffer reference to a temporary wait list. We reuse this infrastructure to
2207 * transfer the buffer back to the original queue.
2208 *
2209 * Note the buffer transitions from the queued state, to the submitted and wait
2210 * listed state and back to the queued state during this call. The buffer
2211 * locking and queue management logic between _delwri_pushbuf() and
2212 * _delwri_queue() guarantee that the buffer cannot be queued to another list
2213 * before returning.
2214 */
2215int
2216xfs_buf_delwri_pushbuf(
2217	struct xfs_buf		*bp,
2218	struct list_head	*buffer_list)
2219{
2220	LIST_HEAD		(submit_list);
2221	int			error;
2222
2223	ASSERT(bp->b_flags & _XBF_DELWRI_Q);
2224
2225	trace_xfs_buf_delwri_pushbuf(bp, _RET_IP_);
2226
2227	/*
2228	 * Isolate the buffer to a new local list so we can submit it for I/O
2229	 * independently from the rest of the original list.
2230	 */
2231	xfs_buf_lock(bp);
2232	list_move(&bp->b_list, &submit_list);
2233	xfs_buf_unlock(bp);
2234
2235	/*
2236	 * Delwri submission clears the DELWRI_Q buffer flag and returns with
2237	 * the buffer on the wait list with the original reference. Rather than
2238	 * bounce the buffer from a local wait list back to the original list
2239	 * after I/O completion, reuse the original list as the wait list.
2240	 */
2241	xfs_buf_delwri_submit_buffers(&submit_list, buffer_list);
2242
2243	/*
2244	 * The buffer is now locked, under I/O and wait listed on the original
2245	 * delwri queue. Wait for I/O completion, restore the DELWRI_Q flag and
2246	 * return with the buffer unlocked and on the original queue.
2247	 */
2248	error = xfs_buf_iowait(bp);
2249	bp->b_flags |= _XBF_DELWRI_Q;
2250	xfs_buf_unlock(bp);
2251
2252	return error;
2253}
2254
2255int __init
2256xfs_buf_init(void)
2257{
2258	xfs_buf_zone = kmem_cache_create("xfs_buf", sizeof(struct xfs_buf), 0,
2259					 SLAB_HWCACHE_ALIGN |
2260					 SLAB_RECLAIM_ACCOUNT |
2261					 SLAB_MEM_SPREAD,
2262					 NULL);
2263	if (!xfs_buf_zone)
2264		goto out;
2265
2266	return 0;
2267
2268 out:
2269	return -ENOMEM;
2270}
2271
2272void
2273xfs_buf_terminate(void)
2274{
2275	kmem_cache_destroy(xfs_buf_zone);
2276}
2277
2278void xfs_buf_set_ref(struct xfs_buf *bp, int lru_ref)
2279{
2280	/*
2281	 * Set the lru reference count to 0 based on the error injection tag.
2282	 * This allows userspace to disrupt buffer caching for debug/testing
2283	 * purposes.
2284	 */
2285	if (XFS_TEST_ERROR(false, bp->b_mount, XFS_ERRTAG_BUF_LRU_REF))
2286		lru_ref = 0;
2287
2288	atomic_set(&bp->b_lru_ref, lru_ref);
2289}
2290
2291/*
2292 * Verify an on-disk magic value against the magic value specified in the
2293 * verifier structure. The verifier magic is in disk byte order so the caller is
2294 * expected to pass the value directly from disk.
2295 */
2296bool
2297xfs_verify_magic(
2298	struct xfs_buf		*bp,
2299	__be32			dmagic)
2300{
2301	struct xfs_mount	*mp = bp->b_mount;
2302	int			idx;
2303
2304	idx = xfs_has_crc(mp);
2305	if (WARN_ON(!bp->b_ops || !bp->b_ops->magic[idx]))
2306		return false;
2307	return dmagic == bp->b_ops->magic[idx];
2308}
2309/*
2310 * Verify an on-disk magic value against the magic value specified in the
2311 * verifier structure. The verifier magic is in disk byte order so the caller is
2312 * expected to pass the value directly from disk.
2313 */
2314bool
2315xfs_verify_magic16(
2316	struct xfs_buf		*bp,
2317	__be16			dmagic)
2318{
2319	struct xfs_mount	*mp = bp->b_mount;
2320	int			idx;
2321
2322	idx = xfs_has_crc(mp);
2323	if (WARN_ON(!bp->b_ops || !bp->b_ops->magic16[idx]))
2324		return false;
2325	return dmagic == bp->b_ops->magic16[idx];
2326}
2327