1 /*
2  * Device operations for the pnfs nfs4 file layout driver.
3  *
4  * Copyright (c) 2014, Primary Data, Inc. All rights reserved.
5  *
6  * Tao Peng <bergwolf@primarydata.com>
7  */
8 
9 #include <linux/nfs_fs.h>
10 #include <linux/vmalloc.h>
11 #include <linux/module.h>
12 #include <linux/sunrpc/addr.h>
13 
14 #include "../internal.h"
15 #include "../nfs4session.h"
16 #include "flexfilelayout.h"
17 
18 #define NFSDBG_FACILITY		NFSDBG_PNFS_LD
19 
20 static unsigned int dataserver_timeo = NFS_DEF_TCP_RETRANS;
21 static unsigned int dataserver_retrans;
22 
23 static bool ff_layout_has_available_ds(struct pnfs_layout_segment *lseg);
24 
25 void nfs4_ff_layout_put_deviceid(struct nfs4_ff_layout_ds *mirror_ds)
26 {
27 	if (!IS_ERR_OR_NULL(mirror_ds))
28 		nfs4_put_deviceid_node(&mirror_ds->id_node);
29 }
30 
31 void nfs4_ff_layout_free_deviceid(struct nfs4_ff_layout_ds *mirror_ds)
32 {
33 	nfs4_print_deviceid(&mirror_ds->id_node.deviceid);
34 	nfs4_pnfs_ds_put(mirror_ds->ds);
35 	kfree(mirror_ds->ds_versions);
36 	kfree_rcu(mirror_ds, id_node.rcu);
37 }
38 
39 /* Decode opaque device data and construct new_ds using it */
40 struct nfs4_ff_layout_ds *
41 nfs4_ff_alloc_deviceid_node(struct nfs_server *server, struct pnfs_device *pdev,
42 			    gfp_t gfp_flags)
43 {
44 	struct xdr_stream stream;
45 	struct xdr_buf buf;
46 	struct page *scratch;
47 	struct list_head dsaddrs;
48 	struct nfs4_pnfs_ds_addr *da;
49 	struct nfs4_ff_layout_ds *new_ds = NULL;
50 	struct nfs4_ff_ds_version *ds_versions = NULL;
51 	u32 mp_count;
52 	u32 version_count;
53 	__be32 *p;
54 	int i, ret = -ENOMEM;
55 
56 	/* set up xdr stream */
57 	scratch = alloc_page(gfp_flags);
58 	if (!scratch)
59 		goto out_err;
60 
61 	new_ds = kzalloc(sizeof(struct nfs4_ff_layout_ds), gfp_flags);
62 	if (!new_ds)
63 		goto out_scratch;
64 
65 	nfs4_init_deviceid_node(&new_ds->id_node,
66 				server,
67 				&pdev->dev_id);
68 	INIT_LIST_HEAD(&dsaddrs);
69 
70 	xdr_init_decode_pages(&stream, &buf, pdev->pages, pdev->pglen);
71 	xdr_set_scratch_buffer(&stream, page_address(scratch), PAGE_SIZE);
72 
73 	/* multipath count */
74 	p = xdr_inline_decode(&stream, 4);
75 	if (unlikely(!p))
76 		goto out_err_drain_dsaddrs;
77 	mp_count = be32_to_cpup(p);
78 	dprintk("%s: multipath ds count %d\n", __func__, mp_count);
79 
80 	for (i = 0; i < mp_count; i++) {
81 		/* multipath ds */
82 		da = nfs4_decode_mp_ds_addr(server->nfs_client->cl_net,
83 					    &stream, gfp_flags);
84 		if (da)
85 			list_add_tail(&da->da_node, &dsaddrs);
86 	}
87 	if (list_empty(&dsaddrs)) {
88 		dprintk("%s: no suitable DS addresses found\n",
89 			__func__);
90 		ret = -ENOMEDIUM;
91 		goto out_err_drain_dsaddrs;
92 	}
93 
94 	/* version count */
95 	p = xdr_inline_decode(&stream, 4);
96 	if (unlikely(!p))
97 		goto out_err_drain_dsaddrs;
98 	version_count = be32_to_cpup(p);
99 	dprintk("%s: version count %d\n", __func__, version_count);
100 
101 	ds_versions = kzalloc(version_count * sizeof(struct nfs4_ff_ds_version),
102 			      gfp_flags);
103 	if (!ds_versions)
104 		goto out_scratch;
105 
106 	for (i = 0; i < version_count; i++) {
107 		/* 20 = version(4) + minor_version(4) + rsize(4) + wsize(4) +
108 		 * tightly_coupled(4) */
109 		p = xdr_inline_decode(&stream, 20);
110 		if (unlikely(!p))
111 			goto out_err_drain_dsaddrs;
112 		ds_versions[i].version = be32_to_cpup(p++);
113 		ds_versions[i].minor_version = be32_to_cpup(p++);
114 		ds_versions[i].rsize = nfs_block_size(be32_to_cpup(p++), NULL);
115 		ds_versions[i].wsize = nfs_block_size(be32_to_cpup(p++), NULL);
116 		ds_versions[i].tightly_coupled = be32_to_cpup(p);
117 
118 		if (ds_versions[i].rsize > NFS_MAX_FILE_IO_SIZE)
119 			ds_versions[i].rsize = NFS_MAX_FILE_IO_SIZE;
120 		if (ds_versions[i].wsize > NFS_MAX_FILE_IO_SIZE)
121 			ds_versions[i].wsize = NFS_MAX_FILE_IO_SIZE;
122 
123 		/*
124 		 * check for valid major/minor combination.
125 		 * currently we support dataserver which talk:
126 		 *   v3, v4.0, v4.1, v4.2
127 		 */
128 		if (!((ds_versions[i].version == 3 && ds_versions[i].minor_version == 0) ||
129 			(ds_versions[i].version == 4 && ds_versions[i].minor_version < 3))) {
130 			dprintk("%s: [%d] unsupported ds version %d-%d\n", __func__,
131 				i, ds_versions[i].version,
132 				ds_versions[i].minor_version);
133 			ret = -EPROTONOSUPPORT;
134 			goto out_err_drain_dsaddrs;
135 		}
136 
137 		dprintk("%s: [%d] vers %u minor_ver %u rsize %u wsize %u coupled %d\n",
138 			__func__, i, ds_versions[i].version,
139 			ds_versions[i].minor_version,
140 			ds_versions[i].rsize,
141 			ds_versions[i].wsize,
142 			ds_versions[i].tightly_coupled);
143 	}
144 
145 	new_ds->ds_versions = ds_versions;
146 	new_ds->ds_versions_cnt = version_count;
147 
148 	new_ds->ds = nfs4_pnfs_ds_add(&dsaddrs, gfp_flags);
149 	if (!new_ds->ds)
150 		goto out_err_drain_dsaddrs;
151 
152 	/* If DS was already in cache, free ds addrs */
153 	while (!list_empty(&dsaddrs)) {
154 		da = list_first_entry(&dsaddrs,
155 				      struct nfs4_pnfs_ds_addr,
156 				      da_node);
157 		list_del_init(&da->da_node);
158 		kfree(da->da_remotestr);
159 		kfree(da);
160 	}
161 
162 	__free_page(scratch);
163 	return new_ds;
164 
165 out_err_drain_dsaddrs:
166 	while (!list_empty(&dsaddrs)) {
167 		da = list_first_entry(&dsaddrs, struct nfs4_pnfs_ds_addr,
168 				      da_node);
169 		list_del_init(&da->da_node);
170 		kfree(da->da_remotestr);
171 		kfree(da);
172 	}
173 
174 	kfree(ds_versions);
175 out_scratch:
176 	__free_page(scratch);
177 out_err:
178 	kfree(new_ds);
179 
180 	dprintk("%s ERROR: returning %d\n", __func__, ret);
181 	return NULL;
182 }
183 
184 static void ff_layout_mark_devid_invalid(struct pnfs_layout_segment *lseg,
185 		struct nfs4_deviceid_node *devid)
186 {
187 	nfs4_delete_deviceid(devid->ld, devid->nfs_client, &devid->deviceid);
188 	if (!ff_layout_has_available_ds(lseg))
189 		pnfs_error_mark_layout_for_return(lseg->pls_layout->plh_inode,
190 				lseg);
191 }
192 
193 static bool ff_layout_mirror_valid(struct pnfs_layout_segment *lseg,
194 				   struct nfs4_ff_layout_mirror *mirror,
195 				   bool create)
196 {
197 	if (mirror == NULL || IS_ERR(mirror->mirror_ds))
198 		goto outerr;
199 	if (mirror->mirror_ds == NULL) {
200 		if (create) {
201 			struct nfs4_deviceid_node *node;
202 			struct pnfs_layout_hdr *lh = lseg->pls_layout;
203 			struct nfs4_ff_layout_ds *mirror_ds = ERR_PTR(-ENODEV);
204 
205 			node = nfs4_find_get_deviceid(NFS_SERVER(lh->plh_inode),
206 					&mirror->devid, lh->plh_lc_cred,
207 					GFP_KERNEL);
208 			if (node)
209 				mirror_ds = FF_LAYOUT_MIRROR_DS(node);
210 
211 			/* check for race with another call to this function */
212 			if (cmpxchg(&mirror->mirror_ds, NULL, mirror_ds) &&
213 			    mirror_ds != ERR_PTR(-ENODEV))
214 				nfs4_put_deviceid_node(node);
215 		} else
216 			goto outerr;
217 	}
218 
219 	if (IS_ERR(mirror->mirror_ds))
220 		goto outerr;
221 
222 	if (mirror->mirror_ds->ds == NULL) {
223 		struct nfs4_deviceid_node *devid;
224 		devid = &mirror->mirror_ds->id_node;
225 		ff_layout_mark_devid_invalid(lseg, devid);
226 		return false;
227 	}
228 	return true;
229 outerr:
230 	pnfs_error_mark_layout_for_return(lseg->pls_layout->plh_inode, lseg);
231 	return false;
232 }
233 
234 static void extend_ds_error(struct nfs4_ff_layout_ds_err *err,
235 			    u64 offset, u64 length)
236 {
237 	u64 end;
238 
239 	end = max_t(u64, pnfs_end_offset(err->offset, err->length),
240 		    pnfs_end_offset(offset, length));
241 	err->offset = min_t(u64, err->offset, offset);
242 	err->length = end - err->offset;
243 }
244 
245 static int
246 ff_ds_error_match(const struct nfs4_ff_layout_ds_err *e1,
247 		const struct nfs4_ff_layout_ds_err *e2)
248 {
249 	int ret;
250 
251 	if (e1->opnum != e2->opnum)
252 		return e1->opnum < e2->opnum ? -1 : 1;
253 	if (e1->status != e2->status)
254 		return e1->status < e2->status ? -1 : 1;
255 	ret = memcmp(e1->stateid.data, e2->stateid.data,
256 			sizeof(e1->stateid.data));
257 	if (ret != 0)
258 		return ret;
259 	ret = memcmp(&e1->deviceid, &e2->deviceid, sizeof(e1->deviceid));
260 	if (ret != 0)
261 		return ret;
262 	if (pnfs_end_offset(e1->offset, e1->length) < e2->offset)
263 		return -1;
264 	if (e1->offset > pnfs_end_offset(e2->offset, e2->length))
265 		return 1;
266 	/* If ranges overlap or are contiguous, they are the same */
267 	return 0;
268 }
269 
270 static void
271 ff_layout_add_ds_error_locked(struct nfs4_flexfile_layout *flo,
272 			      struct nfs4_ff_layout_ds_err *dserr)
273 {
274 	struct nfs4_ff_layout_ds_err *err, *tmp;
275 	struct list_head *head = &flo->error_list;
276 	int match;
277 
278 	/* Do insertion sort w/ merges */
279 	list_for_each_entry_safe(err, tmp, &flo->error_list, list) {
280 		match = ff_ds_error_match(err, dserr);
281 		if (match < 0)
282 			continue;
283 		if (match > 0) {
284 			/* Add entry "dserr" _before_ entry "err" */
285 			head = &err->list;
286 			break;
287 		}
288 		/* Entries match, so merge "err" into "dserr" */
289 		extend_ds_error(dserr, err->offset, err->length);
290 		list_replace(&err->list, &dserr->list);
291 		kfree(err);
292 		return;
293 	}
294 
295 	list_add_tail(&dserr->list, head);
296 }
297 
298 int ff_layout_track_ds_error(struct nfs4_flexfile_layout *flo,
299 			     struct nfs4_ff_layout_mirror *mirror, u64 offset,
300 			     u64 length, int status, enum nfs_opnum4 opnum,
301 			     gfp_t gfp_flags)
302 {
303 	struct nfs4_ff_layout_ds_err *dserr;
304 
305 	if (status == 0)
306 		return 0;
307 
308 	if (mirror->mirror_ds == NULL)
309 		return -EINVAL;
310 
311 	dserr = kmalloc(sizeof(*dserr), gfp_flags);
312 	if (!dserr)
313 		return -ENOMEM;
314 
315 	INIT_LIST_HEAD(&dserr->list);
316 	dserr->offset = offset;
317 	dserr->length = length;
318 	dserr->status = status;
319 	dserr->opnum = opnum;
320 	nfs4_stateid_copy(&dserr->stateid, &mirror->stateid);
321 	memcpy(&dserr->deviceid, &mirror->mirror_ds->id_node.deviceid,
322 	       NFS4_DEVICEID4_SIZE);
323 
324 	spin_lock(&flo->generic_hdr.plh_inode->i_lock);
325 	ff_layout_add_ds_error_locked(flo, dserr);
326 	spin_unlock(&flo->generic_hdr.plh_inode->i_lock);
327 
328 	return 0;
329 }
330 
331 static struct rpc_cred *
332 ff_layout_get_mirror_cred(struct nfs4_ff_layout_mirror *mirror, u32 iomode)
333 {
334 	struct rpc_cred *cred, __rcu **pcred;
335 
336 	if (iomode == IOMODE_READ)
337 		pcred = &mirror->ro_cred;
338 	else
339 		pcred = &mirror->rw_cred;
340 
341 	rcu_read_lock();
342 	do {
343 		cred = rcu_dereference(*pcred);
344 		if (!cred)
345 			break;
346 
347 		cred = get_rpccred_rcu(cred);
348 	} while(!cred);
349 	rcu_read_unlock();
350 	return cred;
351 }
352 
353 struct nfs_fh *
354 nfs4_ff_layout_select_ds_fh(struct pnfs_layout_segment *lseg, u32 mirror_idx)
355 {
356 	struct nfs4_ff_layout_mirror *mirror = FF_LAYOUT_COMP(lseg, mirror_idx);
357 	struct nfs_fh *fh = NULL;
358 
359 	if (!ff_layout_mirror_valid(lseg, mirror, false)) {
360 		pr_err_ratelimited("NFS: %s: No data server for mirror offset index %d\n",
361 			__func__, mirror_idx);
362 		goto out;
363 	}
364 
365 	/* FIXME: For now assume there is only 1 version available for the DS */
366 	fh = &mirror->fh_versions[0];
367 out:
368 	return fh;
369 }
370 
371 /**
372  * nfs4_ff_layout_prepare_ds - prepare a DS connection for an RPC call
373  * @lseg: the layout segment we're operating on
374  * @ds_idx: index of the DS to use
375  * @fail_return: return layout on connect failure?
376  *
377  * Try to prepare a DS connection to accept an RPC call. This involves
378  * selecting a mirror to use and connecting the client to it if it's not
379  * already connected.
380  *
381  * Since we only need a single functioning mirror to satisfy a read, we don't
382  * want to return the layout if there is one. For writes though, any down
383  * mirror should result in a LAYOUTRETURN. @fail_return is how we distinguish
384  * between the two cases.
385  *
386  * Returns a pointer to a connected DS object on success or NULL on failure.
387  */
388 struct nfs4_pnfs_ds *
389 nfs4_ff_layout_prepare_ds(struct pnfs_layout_segment *lseg, u32 ds_idx,
390 			  bool fail_return)
391 {
392 	struct nfs4_ff_layout_mirror *mirror = FF_LAYOUT_COMP(lseg, ds_idx);
393 	struct nfs4_pnfs_ds *ds = NULL;
394 	struct nfs4_deviceid_node *devid;
395 	struct inode *ino = lseg->pls_layout->plh_inode;
396 	struct nfs_server *s = NFS_SERVER(ino);
397 	unsigned int max_payload;
398 	int status;
399 
400 	if (!ff_layout_mirror_valid(lseg, mirror, true)) {
401 		pr_err_ratelimited("NFS: %s: No data server for offset index %d\n",
402 			__func__, ds_idx);
403 		goto out;
404 	}
405 
406 	devid = &mirror->mirror_ds->id_node;
407 	if (ff_layout_test_devid_unavailable(devid))
408 		goto out_fail;
409 
410 	ds = mirror->mirror_ds->ds;
411 	/* matching smp_wmb() in _nfs4_pnfs_v3/4_ds_connect */
412 	smp_rmb();
413 	if (ds->ds_clp)
414 		goto out;
415 
416 	/* FIXME: For now we assume the server sent only one version of NFS
417 	 * to use for the DS.
418 	 */
419 	status = nfs4_pnfs_ds_connect(s, ds, devid, dataserver_timeo,
420 			     dataserver_retrans,
421 			     mirror->mirror_ds->ds_versions[0].version,
422 			     mirror->mirror_ds->ds_versions[0].minor_version);
423 
424 	/* connect success, check rsize/wsize limit */
425 	if (!status) {
426 		max_payload =
427 			nfs_block_size(rpc_max_payload(ds->ds_clp->cl_rpcclient),
428 				       NULL);
429 		if (mirror->mirror_ds->ds_versions[0].rsize > max_payload)
430 			mirror->mirror_ds->ds_versions[0].rsize = max_payload;
431 		if (mirror->mirror_ds->ds_versions[0].wsize > max_payload)
432 			mirror->mirror_ds->ds_versions[0].wsize = max_payload;
433 		goto out;
434 	}
435 out_fail:
436 	ff_layout_track_ds_error(FF_LAYOUT_FROM_HDR(lseg->pls_layout),
437 				 mirror, lseg->pls_range.offset,
438 				 lseg->pls_range.length, NFS4ERR_NXIO,
439 				 OP_ILLEGAL, GFP_NOIO);
440 	if (fail_return || !ff_layout_has_available_ds(lseg))
441 		pnfs_error_mark_layout_for_return(ino, lseg);
442 	ds = NULL;
443 out:
444 	return ds;
445 }
446 
447 struct rpc_cred *
448 ff_layout_get_ds_cred(struct pnfs_layout_segment *lseg, u32 ds_idx,
449 		      struct rpc_cred *mdscred)
450 {
451 	struct nfs4_ff_layout_mirror *mirror = FF_LAYOUT_COMP(lseg, ds_idx);
452 	struct rpc_cred *cred;
453 
454 	if (mirror) {
455 		cred = ff_layout_get_mirror_cred(mirror, lseg->pls_range.iomode);
456 		if (!cred)
457 			cred = get_rpccred(mdscred);
458 	} else {
459 		cred = get_rpccred(mdscred);
460 	}
461 	return cred;
462 }
463 
464 /**
465 * Find or create a DS rpc client with th MDS server rpc client auth flavor
466 * in the nfs_client cl_ds_clients list.
467 */
468 struct rpc_clnt *
469 nfs4_ff_find_or_create_ds_client(struct pnfs_layout_segment *lseg, u32 ds_idx,
470 				 struct nfs_client *ds_clp, struct inode *inode)
471 {
472 	struct nfs4_ff_layout_mirror *mirror = FF_LAYOUT_COMP(lseg, ds_idx);
473 
474 	switch (mirror->mirror_ds->ds_versions[0].version) {
475 	case 3:
476 		/* For NFSv3 DS, flavor is set when creating DS connections */
477 		return ds_clp->cl_rpcclient;
478 	case 4:
479 		return nfs4_find_or_create_ds_client(ds_clp, inode);
480 	default:
481 		BUG();
482 	}
483 }
484 
485 void ff_layout_free_ds_ioerr(struct list_head *head)
486 {
487 	struct nfs4_ff_layout_ds_err *err;
488 
489 	while (!list_empty(head)) {
490 		err = list_first_entry(head,
491 				struct nfs4_ff_layout_ds_err,
492 				list);
493 		list_del(&err->list);
494 		kfree(err);
495 	}
496 }
497 
498 /* called with inode i_lock held */
499 int ff_layout_encode_ds_ioerr(struct xdr_stream *xdr, const struct list_head *head)
500 {
501 	struct nfs4_ff_layout_ds_err *err;
502 	__be32 *p;
503 
504 	list_for_each_entry(err, head, list) {
505 		/* offset(8) + length(8) + stateid(NFS4_STATEID_SIZE)
506 		 * + array length + deviceid(NFS4_DEVICEID4_SIZE)
507 		 * + status(4) + opnum(4)
508 		 */
509 		p = xdr_reserve_space(xdr,
510 				28 + NFS4_STATEID_SIZE + NFS4_DEVICEID4_SIZE);
511 		if (unlikely(!p))
512 			return -ENOBUFS;
513 		p = xdr_encode_hyper(p, err->offset);
514 		p = xdr_encode_hyper(p, err->length);
515 		p = xdr_encode_opaque_fixed(p, &err->stateid,
516 					    NFS4_STATEID_SIZE);
517 		/* Encode 1 error */
518 		*p++ = cpu_to_be32(1);
519 		p = xdr_encode_opaque_fixed(p, &err->deviceid,
520 					    NFS4_DEVICEID4_SIZE);
521 		*p++ = cpu_to_be32(err->status);
522 		*p++ = cpu_to_be32(err->opnum);
523 		dprintk("%s: offset %llu length %llu status %d op %d\n",
524 			__func__, err->offset, err->length, err->status,
525 			err->opnum);
526 	}
527 
528 	return 0;
529 }
530 
531 static
532 unsigned int do_layout_fetch_ds_ioerr(struct pnfs_layout_hdr *lo,
533 				      const struct pnfs_layout_range *range,
534 				      struct list_head *head,
535 				      unsigned int maxnum)
536 {
537 	struct nfs4_flexfile_layout *flo = FF_LAYOUT_FROM_HDR(lo);
538 	struct inode *inode = lo->plh_inode;
539 	struct nfs4_ff_layout_ds_err *err, *n;
540 	unsigned int ret = 0;
541 
542 	spin_lock(&inode->i_lock);
543 	list_for_each_entry_safe(err, n, &flo->error_list, list) {
544 		if (!pnfs_is_range_intersecting(err->offset,
545 				pnfs_end_offset(err->offset, err->length),
546 				range->offset,
547 				pnfs_end_offset(range->offset, range->length)))
548 			continue;
549 		if (!maxnum)
550 			break;
551 		list_move(&err->list, head);
552 		maxnum--;
553 		ret++;
554 	}
555 	spin_unlock(&inode->i_lock);
556 	return ret;
557 }
558 
559 unsigned int ff_layout_fetch_ds_ioerr(struct pnfs_layout_hdr *lo,
560 				      const struct pnfs_layout_range *range,
561 				      struct list_head *head,
562 				      unsigned int maxnum)
563 {
564 	unsigned int ret;
565 
566 	ret = do_layout_fetch_ds_ioerr(lo, range, head, maxnum);
567 	/* If we're over the max, discard all remaining entries */
568 	if (ret == maxnum) {
569 		LIST_HEAD(discard);
570 		do_layout_fetch_ds_ioerr(lo, range, &discard, -1);
571 		ff_layout_free_ds_ioerr(&discard);
572 	}
573 	return ret;
574 }
575 
576 static bool ff_read_layout_has_available_ds(struct pnfs_layout_segment *lseg)
577 {
578 	struct nfs4_ff_layout_mirror *mirror;
579 	struct nfs4_deviceid_node *devid;
580 	u32 idx;
581 
582 	for (idx = 0; idx < FF_LAYOUT_MIRROR_COUNT(lseg); idx++) {
583 		mirror = FF_LAYOUT_COMP(lseg, idx);
584 		if (mirror) {
585 			if (!mirror->mirror_ds)
586 				return true;
587 			if (IS_ERR(mirror->mirror_ds))
588 				continue;
589 			devid = &mirror->mirror_ds->id_node;
590 			if (!ff_layout_test_devid_unavailable(devid))
591 				return true;
592 		}
593 	}
594 
595 	return false;
596 }
597 
598 static bool ff_rw_layout_has_available_ds(struct pnfs_layout_segment *lseg)
599 {
600 	struct nfs4_ff_layout_mirror *mirror;
601 	struct nfs4_deviceid_node *devid;
602 	u32 idx;
603 
604 	for (idx = 0; idx < FF_LAYOUT_MIRROR_COUNT(lseg); idx++) {
605 		mirror = FF_LAYOUT_COMP(lseg, idx);
606 		if (!mirror || IS_ERR(mirror->mirror_ds))
607 			return false;
608 		if (!mirror->mirror_ds)
609 			continue;
610 		devid = &mirror->mirror_ds->id_node;
611 		if (ff_layout_test_devid_unavailable(devid))
612 			return false;
613 	}
614 
615 	return FF_LAYOUT_MIRROR_COUNT(lseg) != 0;
616 }
617 
618 static bool ff_layout_has_available_ds(struct pnfs_layout_segment *lseg)
619 {
620 	if (lseg->pls_range.iomode == IOMODE_READ)
621 		return  ff_read_layout_has_available_ds(lseg);
622 	/* Note: RW layout needs all mirrors available */
623 	return ff_rw_layout_has_available_ds(lseg);
624 }
625 
626 bool ff_layout_avoid_mds_available_ds(struct pnfs_layout_segment *lseg)
627 {
628 	return ff_layout_no_fallback_to_mds(lseg) ||
629 	       ff_layout_has_available_ds(lseg);
630 }
631 
632 bool ff_layout_avoid_read_on_rw(struct pnfs_layout_segment *lseg)
633 {
634 	return lseg->pls_range.iomode == IOMODE_RW &&
635 	       ff_layout_no_read_on_rw(lseg);
636 }
637 
638 module_param(dataserver_retrans, uint, 0644);
639 MODULE_PARM_DESC(dataserver_retrans, "The  number of times the NFSv4.1 client "
640 			"retries a request before it attempts further "
641 			" recovery  action.");
642 module_param(dataserver_timeo, uint, 0644);
643 MODULE_PARM_DESC(dataserver_timeo, "The time (in tenths of a second) the "
644 			"NFSv4.1  client  waits for a response from a "
645 			" data server before it retries an NFS request.");
646