xref: /openbmc/linux/fs/nfs/pnfs_nfs.c (revision 11c416e3)
1 // SPDX-License-Identifier: GPL-2.0-only
2 /*
3  * Common NFS I/O  operations for the pnfs file based
4  * layout drivers.
5  *
6  * Copyright (c) 2014, Primary Data, Inc. All rights reserved.
7  *
8  * Tom Haynes <loghyr@primarydata.com>
9  */
10 
11 #include <linux/nfs_fs.h>
12 #include <linux/nfs_page.h>
13 #include <linux/sunrpc/addr.h>
14 #include <linux/module.h>
15 
16 #include "nfs4session.h"
17 #include "internal.h"
18 #include "pnfs.h"
19 
20 #define NFSDBG_FACILITY		NFSDBG_PNFS
21 
22 void pnfs_generic_rw_release(void *data)
23 {
24 	struct nfs_pgio_header *hdr = data;
25 
26 	nfs_put_client(hdr->ds_clp);
27 	hdr->mds_ops->rpc_release(data);
28 }
29 EXPORT_SYMBOL_GPL(pnfs_generic_rw_release);
30 
31 /* Fake up some data that will cause nfs_commit_release to retry the writes. */
32 void pnfs_generic_prepare_to_resend_writes(struct nfs_commit_data *data)
33 {
34 	struct nfs_writeverf *verf = data->res.verf;
35 
36 	data->task.tk_status = 0;
37 	memset(&verf->verifier, 0, sizeof(verf->verifier));
38 	verf->committed = NFS_UNSTABLE;
39 }
40 EXPORT_SYMBOL_GPL(pnfs_generic_prepare_to_resend_writes);
41 
42 void pnfs_generic_write_commit_done(struct rpc_task *task, void *data)
43 {
44 	struct nfs_commit_data *wdata = data;
45 
46 	/* Note this may cause RPC to be resent */
47 	wdata->mds_ops->rpc_call_done(task, data);
48 }
49 EXPORT_SYMBOL_GPL(pnfs_generic_write_commit_done);
50 
51 void pnfs_generic_commit_release(void *calldata)
52 {
53 	struct nfs_commit_data *data = calldata;
54 
55 	data->completion_ops->completion(data);
56 	pnfs_put_lseg(data->lseg);
57 	nfs_put_client(data->ds_clp);
58 	nfs_commitdata_release(data);
59 }
60 EXPORT_SYMBOL_GPL(pnfs_generic_commit_release);
61 
62 static struct pnfs_layout_segment *
63 pnfs_free_bucket_lseg(struct pnfs_commit_bucket *bucket)
64 {
65 	if (list_empty(&bucket->committing) && list_empty(&bucket->written)) {
66 		struct pnfs_layout_segment *freeme = bucket->lseg;
67 		bucket->lseg = NULL;
68 		return freeme;
69 	}
70 	return NULL;
71 }
72 
73 /* The generic layer is about to remove the req from the commit list.
74  * If this will make the bucket empty, it will need to put the lseg reference.
75  * Note this must be called holding nfsi->commit_mutex
76  */
77 void
78 pnfs_generic_clear_request_commit(struct nfs_page *req,
79 				  struct nfs_commit_info *cinfo)
80 {
81 	struct pnfs_layout_segment *freeme = NULL;
82 
83 	if (!test_and_clear_bit(PG_COMMIT_TO_DS, &req->wb_flags))
84 		goto out;
85 	cinfo->ds->nwritten--;
86 	if (list_is_singular(&req->wb_list)) {
87 		struct pnfs_commit_bucket *bucket;
88 
89 		bucket = list_first_entry(&req->wb_list,
90 					  struct pnfs_commit_bucket,
91 					  written);
92 		freeme = pnfs_free_bucket_lseg(bucket);
93 	}
94 out:
95 	nfs_request_remove_commit_list(req, cinfo);
96 	pnfs_put_lseg(freeme);
97 }
98 EXPORT_SYMBOL_GPL(pnfs_generic_clear_request_commit);
99 
100 struct pnfs_commit_array *
101 pnfs_alloc_commit_array(size_t n, gfp_t gfp_flags)
102 {
103 	struct pnfs_commit_array *p;
104 	struct pnfs_commit_bucket *b;
105 
106 	p = kmalloc(struct_size(p, buckets, n), gfp_flags);
107 	if (!p)
108 		return NULL;
109 	p->nbuckets = n;
110 	INIT_LIST_HEAD(&p->cinfo_list);
111 	INIT_LIST_HEAD(&p->lseg_list);
112 	p->lseg = NULL;
113 	for (b = &p->buckets[0]; n != 0; b++, n--) {
114 		INIT_LIST_HEAD(&b->written);
115 		INIT_LIST_HEAD(&b->committing);
116 		b->lseg = NULL;
117 		b->direct_verf.committed = NFS_INVALID_STABLE_HOW;
118 	}
119 	return p;
120 }
121 EXPORT_SYMBOL_GPL(pnfs_alloc_commit_array);
122 
123 void
124 pnfs_free_commit_array(struct pnfs_commit_array *p)
125 {
126 	kfree_rcu(p, rcu);
127 }
128 EXPORT_SYMBOL_GPL(pnfs_free_commit_array);
129 
130 static struct pnfs_commit_array *
131 pnfs_find_commit_array_by_lseg(struct pnfs_ds_commit_info *fl_cinfo,
132 		struct pnfs_layout_segment *lseg)
133 {
134 	struct pnfs_commit_array *array;
135 
136 	list_for_each_entry_rcu(array, &fl_cinfo->commits, cinfo_list) {
137 		if (array->lseg == lseg)
138 			return array;
139 	}
140 	return NULL;
141 }
142 
143 struct pnfs_commit_array *
144 pnfs_add_commit_array(struct pnfs_ds_commit_info *fl_cinfo,
145 		struct pnfs_commit_array *new,
146 		struct pnfs_layout_segment *lseg)
147 {
148 	struct pnfs_commit_array *array;
149 
150 	array = pnfs_find_commit_array_by_lseg(fl_cinfo, lseg);
151 	if (array)
152 		return array;
153 	new->lseg = lseg;
154 	refcount_set(&new->refcount, 1);
155 	list_add_rcu(&new->cinfo_list, &fl_cinfo->commits);
156 	list_add(&new->lseg_list, &lseg->pls_commits);
157 	return new;
158 }
159 EXPORT_SYMBOL_GPL(pnfs_add_commit_array);
160 
161 static struct pnfs_commit_array *
162 pnfs_lookup_commit_array(struct pnfs_ds_commit_info *fl_cinfo,
163 		struct pnfs_layout_segment *lseg)
164 {
165 	struct pnfs_commit_array *array;
166 
167 	rcu_read_lock();
168 	array = pnfs_find_commit_array_by_lseg(fl_cinfo, lseg);
169 	if (!array) {
170 		rcu_read_unlock();
171 		fl_cinfo->ops->setup_ds_info(fl_cinfo, lseg);
172 		rcu_read_lock();
173 		array = pnfs_find_commit_array_by_lseg(fl_cinfo, lseg);
174 	}
175 	rcu_read_unlock();
176 	return array;
177 }
178 
179 static void
180 pnfs_release_commit_array_locked(struct pnfs_commit_array *array)
181 {
182 	list_del_rcu(&array->cinfo_list);
183 	list_del(&array->lseg_list);
184 	pnfs_free_commit_array(array);
185 }
186 
187 static void
188 pnfs_put_commit_array_locked(struct pnfs_commit_array *array)
189 {
190 	if (refcount_dec_and_test(&array->refcount))
191 		pnfs_release_commit_array_locked(array);
192 }
193 
194 static void
195 pnfs_put_commit_array(struct pnfs_commit_array *array, struct inode *inode)
196 {
197 	if (refcount_dec_and_lock(&array->refcount, &inode->i_lock)) {
198 		pnfs_release_commit_array_locked(array);
199 		spin_unlock(&inode->i_lock);
200 	}
201 }
202 
203 static struct pnfs_commit_array *
204 pnfs_get_commit_array(struct pnfs_commit_array *array)
205 {
206 	if (refcount_inc_not_zero(&array->refcount))
207 		return array;
208 	return NULL;
209 }
210 
211 static void
212 pnfs_remove_and_free_commit_array(struct pnfs_commit_array *array)
213 {
214 	array->lseg = NULL;
215 	list_del_init(&array->lseg_list);
216 	pnfs_put_commit_array_locked(array);
217 }
218 
219 void
220 pnfs_generic_ds_cinfo_release_lseg(struct pnfs_ds_commit_info *fl_cinfo,
221 		struct pnfs_layout_segment *lseg)
222 {
223 	struct pnfs_commit_array *array, *tmp;
224 
225 	list_for_each_entry_safe(array, tmp, &lseg->pls_commits, lseg_list)
226 		pnfs_remove_and_free_commit_array(array);
227 }
228 EXPORT_SYMBOL_GPL(pnfs_generic_ds_cinfo_release_lseg);
229 
230 void
231 pnfs_generic_ds_cinfo_destroy(struct pnfs_ds_commit_info *fl_cinfo)
232 {
233 	struct pnfs_commit_array *array, *tmp;
234 
235 	list_for_each_entry_safe(array, tmp, &fl_cinfo->commits, cinfo_list)
236 		pnfs_remove_and_free_commit_array(array);
237 }
238 EXPORT_SYMBOL_GPL(pnfs_generic_ds_cinfo_destroy);
239 
240 /*
241  * Locks the nfs_page requests for commit and moves them to
242  * @bucket->committing.
243  */
244 static int
245 pnfs_bucket_scan_ds_commit_list(struct pnfs_commit_bucket *bucket,
246 				struct nfs_commit_info *cinfo,
247 				int max)
248 {
249 	struct list_head *src = &bucket->written;
250 	struct list_head *dst = &bucket->committing;
251 	int ret;
252 
253 	lockdep_assert_held(&NFS_I(cinfo->inode)->commit_mutex);
254 	ret = nfs_scan_commit_list(src, dst, cinfo, max);
255 	if (ret) {
256 		cinfo->ds->nwritten -= ret;
257 		cinfo->ds->ncommitting += ret;
258 	}
259 	return ret;
260 }
261 
262 static int pnfs_bucket_scan_array(struct nfs_commit_info *cinfo,
263 				  struct pnfs_commit_bucket *buckets,
264 				  unsigned int nbuckets,
265 				  int max)
266 {
267 	unsigned int i;
268 	int rv = 0, cnt;
269 
270 	for (i = 0; i < nbuckets && max != 0; i++) {
271 		cnt = pnfs_bucket_scan_ds_commit_list(&buckets[i], cinfo, max);
272 		rv += cnt;
273 		max -= cnt;
274 	}
275 	return rv;
276 }
277 
278 /* Move reqs from written to committing lists, returning count
279  * of number moved.
280  */
281 int pnfs_generic_scan_commit_lists(struct nfs_commit_info *cinfo, int max)
282 {
283 	struct pnfs_ds_commit_info *fl_cinfo = cinfo->ds;
284 	struct pnfs_commit_array *array;
285 	int rv = 0, cnt;
286 
287 	rcu_read_lock();
288 	list_for_each_entry_rcu(array, &fl_cinfo->commits, cinfo_list) {
289 		if (!array->lseg || !pnfs_get_commit_array(array))
290 			continue;
291 		rcu_read_unlock();
292 		cnt = pnfs_bucket_scan_array(cinfo, array->buckets,
293 				array->nbuckets, max);
294 		rcu_read_lock();
295 		pnfs_put_commit_array(array, cinfo->inode);
296 		rv += cnt;
297 		max -= cnt;
298 		if (!max)
299 			break;
300 	}
301 	rcu_read_unlock();
302 	return rv;
303 }
304 EXPORT_SYMBOL_GPL(pnfs_generic_scan_commit_lists);
305 
306 static unsigned int
307 pnfs_bucket_recover_commit_reqs(struct list_head *dst,
308 			        struct pnfs_commit_bucket *buckets,
309 				unsigned int nbuckets,
310 				struct nfs_commit_info *cinfo)
311 {
312 	struct pnfs_commit_bucket *b;
313 	struct pnfs_layout_segment *freeme;
314 	unsigned int nwritten, ret = 0;
315 	unsigned int i;
316 
317 restart:
318 	for (i = 0, b = buckets; i < nbuckets; i++, b++) {
319 		nwritten = nfs_scan_commit_list(&b->written, dst, cinfo, 0);
320 		if (!nwritten)
321 			continue;
322 		ret += nwritten;
323 		freeme = pnfs_free_bucket_lseg(b);
324 		if (freeme) {
325 			pnfs_put_lseg(freeme);
326 			goto restart;
327 		}
328 	}
329 	return ret;
330 }
331 
332 /* Pull everything off the committing lists and dump into @dst.  */
333 void pnfs_generic_recover_commit_reqs(struct list_head *dst,
334 				      struct nfs_commit_info *cinfo)
335 {
336 	struct pnfs_ds_commit_info *fl_cinfo = cinfo->ds;
337 	struct pnfs_commit_array *array;
338 	unsigned int nwritten;
339 
340 	lockdep_assert_held(&NFS_I(cinfo->inode)->commit_mutex);
341 	rcu_read_lock();
342 	list_for_each_entry_rcu(array, &fl_cinfo->commits, cinfo_list) {
343 		if (!array->lseg || !pnfs_get_commit_array(array))
344 			continue;
345 		rcu_read_unlock();
346 		nwritten = pnfs_bucket_recover_commit_reqs(dst,
347 							   array->buckets,
348 							   array->nbuckets,
349 							   cinfo);
350 		rcu_read_lock();
351 		pnfs_put_commit_array(array, cinfo->inode);
352 		fl_cinfo->nwritten -= nwritten;
353 	}
354 	rcu_read_unlock();
355 }
356 EXPORT_SYMBOL_GPL(pnfs_generic_recover_commit_reqs);
357 
358 static struct nfs_page *
359 pnfs_bucket_search_commit_reqs(struct pnfs_commit_bucket *buckets,
360 		unsigned int nbuckets, struct page *page)
361 {
362 	struct nfs_page *req;
363 	struct pnfs_commit_bucket *b;
364 	unsigned int i;
365 
366 	/* Linearly search the commit lists for each bucket until a matching
367 	 * request is found */
368 	for (i = 0, b = buckets; i < nbuckets; i++, b++) {
369 		list_for_each_entry(req, &b->written, wb_list) {
370 			if (req->wb_page == page)
371 				return req->wb_head;
372 		}
373 		list_for_each_entry(req, &b->committing, wb_list) {
374 			if (req->wb_page == page)
375 				return req->wb_head;
376 		}
377 	}
378 	return NULL;
379 }
380 
381 /* pnfs_generic_search_commit_reqs - Search lists in @cinfo for the head reqest
382  *				   for @page
383  * @cinfo - commit info for current inode
384  * @page - page to search for matching head request
385  *
386  * Returns a the head request if one is found, otherwise returns NULL.
387  */
388 struct nfs_page *
389 pnfs_generic_search_commit_reqs(struct nfs_commit_info *cinfo, struct page *page)
390 {
391 	struct pnfs_ds_commit_info *fl_cinfo = cinfo->ds;
392 	struct pnfs_commit_array *array;
393 	struct nfs_page *req;
394 
395 	list_for_each_entry(array, &fl_cinfo->commits, cinfo_list) {
396 		req = pnfs_bucket_search_commit_reqs(array->buckets,
397 				array->nbuckets, page);
398 		if (req)
399 			return req;
400 	}
401 	return NULL;
402 }
403 EXPORT_SYMBOL_GPL(pnfs_generic_search_commit_reqs);
404 
405 static struct pnfs_layout_segment *
406 pnfs_bucket_get_committing(struct list_head *head,
407 			   struct pnfs_commit_bucket *bucket,
408 			   struct nfs_commit_info *cinfo)
409 {
410 	struct list_head *pos;
411 
412 	list_for_each(pos, &bucket->committing)
413 		cinfo->ds->ncommitting--;
414 	list_splice_init(&bucket->committing, head);
415 	return pnfs_free_bucket_lseg(bucket);
416 }
417 
418 static struct nfs_commit_data *
419 pnfs_bucket_fetch_commitdata(struct pnfs_commit_bucket *bucket,
420 			     struct nfs_commit_info *cinfo)
421 {
422 	struct nfs_commit_data *data = nfs_commitdata_alloc(false);
423 
424 	if (!data)
425 		return NULL;
426 	data->lseg = pnfs_bucket_get_committing(&data->pages, bucket, cinfo);
427 	if (!data->lseg)
428 		data->lseg = pnfs_get_lseg(bucket->lseg);
429 	return data;
430 }
431 
432 static void pnfs_generic_retry_commit(struct pnfs_commit_bucket *buckets,
433 				      unsigned int nbuckets,
434 				      struct nfs_commit_info *cinfo,
435 				      unsigned int idx)
436 {
437 	struct pnfs_commit_bucket *bucket;
438 	struct pnfs_layout_segment *freeme;
439 	LIST_HEAD(pages);
440 
441 	for (bucket = buckets; idx < nbuckets; bucket++, idx++) {
442 		if (list_empty(&bucket->committing))
443 			continue;
444 		mutex_lock(&NFS_I(cinfo->inode)->commit_mutex);
445 		freeme = pnfs_bucket_get_committing(&pages, bucket, cinfo);
446 		mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
447 		nfs_retry_commit(&pages, freeme, cinfo, idx);
448 		pnfs_put_lseg(freeme);
449 	}
450 }
451 
452 static unsigned int
453 pnfs_bucket_alloc_ds_commits(struct list_head *list,
454 			     struct pnfs_commit_bucket *buckets,
455 			     unsigned int nbuckets,
456 			     struct nfs_commit_info *cinfo)
457 {
458 	struct pnfs_commit_bucket *bucket;
459 	struct nfs_commit_data *data;
460 	unsigned int i;
461 	unsigned int nreq = 0;
462 
463 	for (i = 0, bucket = buckets; i < nbuckets; i++, bucket++) {
464 		if (list_empty(&bucket->committing))
465 			continue;
466 		mutex_lock(&NFS_I(cinfo->inode)->commit_mutex);
467 		if (!list_empty(&bucket->committing)) {
468 			data = pnfs_bucket_fetch_commitdata(bucket, cinfo);
469 			if (!data)
470 				goto out_error;
471 			data->ds_commit_index = i;
472 			list_add_tail(&data->list, list);
473 			atomic_inc(&cinfo->mds->rpcs_out);
474 			nreq++;
475 		}
476 		mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
477 	}
478 	return nreq;
479 out_error:
480 	mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
481 	/* Clean up on error */
482 	pnfs_generic_retry_commit(buckets, nbuckets, cinfo, i);
483 	return nreq;
484 }
485 
486 static unsigned int
487 pnfs_alloc_ds_commits_list(struct list_head *list,
488 			   struct pnfs_ds_commit_info *fl_cinfo,
489 			   struct nfs_commit_info *cinfo)
490 {
491 	struct pnfs_commit_array *array;
492 	unsigned int ret = 0;
493 
494 	rcu_read_lock();
495 	list_for_each_entry_rcu(array, &fl_cinfo->commits, cinfo_list) {
496 		if (!array->lseg || !pnfs_get_commit_array(array))
497 			continue;
498 		rcu_read_unlock();
499 		ret += pnfs_bucket_alloc_ds_commits(list, array->buckets,
500 				array->nbuckets, cinfo);
501 		rcu_read_lock();
502 		pnfs_put_commit_array(array, cinfo->inode);
503 	}
504 	rcu_read_unlock();
505 	return ret;
506 }
507 
508 /* This follows nfs_commit_list pretty closely */
509 int
510 pnfs_generic_commit_pagelist(struct inode *inode, struct list_head *mds_pages,
511 			     int how, struct nfs_commit_info *cinfo,
512 			     int (*initiate_commit)(struct nfs_commit_data *data,
513 						    int how))
514 {
515 	struct pnfs_ds_commit_info *fl_cinfo = cinfo->ds;
516 	struct nfs_commit_data *data, *tmp;
517 	LIST_HEAD(list);
518 	unsigned int nreq = 0;
519 
520 	if (!list_empty(mds_pages)) {
521 		data = nfs_commitdata_alloc(true);
522 		data->ds_commit_index = -1;
523 		list_splice_init(mds_pages, &data->pages);
524 		list_add_tail(&data->list, &list);
525 		atomic_inc(&cinfo->mds->rpcs_out);
526 		nreq++;
527 	}
528 
529 	nreq += pnfs_alloc_ds_commits_list(&list, fl_cinfo, cinfo);
530 	if (nreq == 0)
531 		goto out;
532 
533 	list_for_each_entry_safe(data, tmp, &list, list) {
534 		list_del(&data->list);
535 		if (data->ds_commit_index < 0) {
536 			nfs_init_commit(data, NULL, NULL, cinfo);
537 			nfs_initiate_commit(NFS_CLIENT(inode), data,
538 					    NFS_PROTO(data->inode),
539 					    data->mds_ops, how,
540 					    RPC_TASK_CRED_NOREF);
541 		} else {
542 			nfs_init_commit(data, NULL, data->lseg, cinfo);
543 			initiate_commit(data, how);
544 		}
545 	}
546 out:
547 	return PNFS_ATTEMPTED;
548 }
549 EXPORT_SYMBOL_GPL(pnfs_generic_commit_pagelist);
550 
551 /*
552  * Data server cache
553  *
554  * Data servers can be mapped to different device ids.
555  * nfs4_pnfs_ds reference counting
556  *   - set to 1 on allocation
557  *   - incremented when a device id maps a data server already in the cache.
558  *   - decremented when deviceid is removed from the cache.
559  */
560 static DEFINE_SPINLOCK(nfs4_ds_cache_lock);
561 static LIST_HEAD(nfs4_data_server_cache);
562 
563 /* Debug routines */
564 static void
565 print_ds(struct nfs4_pnfs_ds *ds)
566 {
567 	if (ds == NULL) {
568 		printk(KERN_WARNING "%s NULL device\n", __func__);
569 		return;
570 	}
571 	printk(KERN_WARNING "        ds %s\n"
572 		"        ref count %d\n"
573 		"        client %p\n"
574 		"        cl_exchange_flags %x\n",
575 		ds->ds_remotestr,
576 		refcount_read(&ds->ds_count), ds->ds_clp,
577 		ds->ds_clp ? ds->ds_clp->cl_exchange_flags : 0);
578 }
579 
580 static bool
581 same_sockaddr(struct sockaddr *addr1, struct sockaddr *addr2)
582 {
583 	struct sockaddr_in *a, *b;
584 	struct sockaddr_in6 *a6, *b6;
585 
586 	if (addr1->sa_family != addr2->sa_family)
587 		return false;
588 
589 	switch (addr1->sa_family) {
590 	case AF_INET:
591 		a = (struct sockaddr_in *)addr1;
592 		b = (struct sockaddr_in *)addr2;
593 
594 		if (a->sin_addr.s_addr == b->sin_addr.s_addr &&
595 		    a->sin_port == b->sin_port)
596 			return true;
597 		break;
598 
599 	case AF_INET6:
600 		a6 = (struct sockaddr_in6 *)addr1;
601 		b6 = (struct sockaddr_in6 *)addr2;
602 
603 		/* LINKLOCAL addresses must have matching scope_id */
604 		if (ipv6_addr_src_scope(&a6->sin6_addr) ==
605 		    IPV6_ADDR_SCOPE_LINKLOCAL &&
606 		    a6->sin6_scope_id != b6->sin6_scope_id)
607 			return false;
608 
609 		if (ipv6_addr_equal(&a6->sin6_addr, &b6->sin6_addr) &&
610 		    a6->sin6_port == b6->sin6_port)
611 			return true;
612 		break;
613 
614 	default:
615 		dprintk("%s: unhandled address family: %u\n",
616 			__func__, addr1->sa_family);
617 		return false;
618 	}
619 
620 	return false;
621 }
622 
623 /*
624  * Checks if 'dsaddrs1' contains a subset of 'dsaddrs2'. If it does,
625  * declare a match.
626  */
627 static bool
628 _same_data_server_addrs_locked(const struct list_head *dsaddrs1,
629 			       const struct list_head *dsaddrs2)
630 {
631 	struct nfs4_pnfs_ds_addr *da1, *da2;
632 	struct sockaddr *sa1, *sa2;
633 	bool match = false;
634 
635 	list_for_each_entry(da1, dsaddrs1, da_node) {
636 		sa1 = (struct sockaddr *)&da1->da_addr;
637 		match = false;
638 		list_for_each_entry(da2, dsaddrs2, da_node) {
639 			sa2 = (struct sockaddr *)&da2->da_addr;
640 			match = same_sockaddr(sa1, sa2);
641 			if (match)
642 				break;
643 		}
644 		if (!match)
645 			break;
646 	}
647 	return match;
648 }
649 
650 /*
651  * Lookup DS by addresses.  nfs4_ds_cache_lock is held
652  */
653 static struct nfs4_pnfs_ds *
654 _data_server_lookup_locked(const struct list_head *dsaddrs)
655 {
656 	struct nfs4_pnfs_ds *ds;
657 
658 	list_for_each_entry(ds, &nfs4_data_server_cache, ds_node)
659 		if (_same_data_server_addrs_locked(&ds->ds_addrs, dsaddrs))
660 			return ds;
661 	return NULL;
662 }
663 
664 static void destroy_ds(struct nfs4_pnfs_ds *ds)
665 {
666 	struct nfs4_pnfs_ds_addr *da;
667 
668 	dprintk("--> %s\n", __func__);
669 	ifdebug(FACILITY)
670 		print_ds(ds);
671 
672 	nfs_put_client(ds->ds_clp);
673 
674 	while (!list_empty(&ds->ds_addrs)) {
675 		da = list_first_entry(&ds->ds_addrs,
676 				      struct nfs4_pnfs_ds_addr,
677 				      da_node);
678 		list_del_init(&da->da_node);
679 		kfree(da->da_remotestr);
680 		kfree(da);
681 	}
682 
683 	kfree(ds->ds_remotestr);
684 	kfree(ds);
685 }
686 
687 void nfs4_pnfs_ds_put(struct nfs4_pnfs_ds *ds)
688 {
689 	if (refcount_dec_and_lock(&ds->ds_count,
690 				&nfs4_ds_cache_lock)) {
691 		list_del_init(&ds->ds_node);
692 		spin_unlock(&nfs4_ds_cache_lock);
693 		destroy_ds(ds);
694 	}
695 }
696 EXPORT_SYMBOL_GPL(nfs4_pnfs_ds_put);
697 
698 /*
699  * Create a string with a human readable address and port to avoid
700  * complicated setup around many dprinks.
701  */
702 static char *
703 nfs4_pnfs_remotestr(struct list_head *dsaddrs, gfp_t gfp_flags)
704 {
705 	struct nfs4_pnfs_ds_addr *da;
706 	char *remotestr;
707 	size_t len;
708 	char *p;
709 
710 	len = 3;        /* '{', '}' and eol */
711 	list_for_each_entry(da, dsaddrs, da_node) {
712 		len += strlen(da->da_remotestr) + 1;    /* string plus comma */
713 	}
714 
715 	remotestr = kzalloc(len, gfp_flags);
716 	if (!remotestr)
717 		return NULL;
718 
719 	p = remotestr;
720 	*(p++) = '{';
721 	len--;
722 	list_for_each_entry(da, dsaddrs, da_node) {
723 		size_t ll = strlen(da->da_remotestr);
724 
725 		if (ll > len)
726 			goto out_err;
727 
728 		memcpy(p, da->da_remotestr, ll);
729 		p += ll;
730 		len -= ll;
731 
732 		if (len < 1)
733 			goto out_err;
734 		(*p++) = ',';
735 		len--;
736 	}
737 	if (len < 2)
738 		goto out_err;
739 	*(p++) = '}';
740 	*p = '\0';
741 	return remotestr;
742 out_err:
743 	kfree(remotestr);
744 	return NULL;
745 }
746 
747 /*
748  * Given a list of multipath struct nfs4_pnfs_ds_addr, add it to ds cache if
749  * uncached and return cached struct nfs4_pnfs_ds.
750  */
751 struct nfs4_pnfs_ds *
752 nfs4_pnfs_ds_add(struct list_head *dsaddrs, gfp_t gfp_flags)
753 {
754 	struct nfs4_pnfs_ds *tmp_ds, *ds = NULL;
755 	char *remotestr;
756 
757 	if (list_empty(dsaddrs)) {
758 		dprintk("%s: no addresses defined\n", __func__);
759 		goto out;
760 	}
761 
762 	ds = kzalloc(sizeof(*ds), gfp_flags);
763 	if (!ds)
764 		goto out;
765 
766 	/* this is only used for debugging, so it's ok if its NULL */
767 	remotestr = nfs4_pnfs_remotestr(dsaddrs, gfp_flags);
768 
769 	spin_lock(&nfs4_ds_cache_lock);
770 	tmp_ds = _data_server_lookup_locked(dsaddrs);
771 	if (tmp_ds == NULL) {
772 		INIT_LIST_HEAD(&ds->ds_addrs);
773 		list_splice_init(dsaddrs, &ds->ds_addrs);
774 		ds->ds_remotestr = remotestr;
775 		refcount_set(&ds->ds_count, 1);
776 		INIT_LIST_HEAD(&ds->ds_node);
777 		ds->ds_clp = NULL;
778 		list_add(&ds->ds_node, &nfs4_data_server_cache);
779 		dprintk("%s add new data server %s\n", __func__,
780 			ds->ds_remotestr);
781 	} else {
782 		kfree(remotestr);
783 		kfree(ds);
784 		refcount_inc(&tmp_ds->ds_count);
785 		dprintk("%s data server %s found, inc'ed ds_count to %d\n",
786 			__func__, tmp_ds->ds_remotestr,
787 			refcount_read(&tmp_ds->ds_count));
788 		ds = tmp_ds;
789 	}
790 	spin_unlock(&nfs4_ds_cache_lock);
791 out:
792 	return ds;
793 }
794 EXPORT_SYMBOL_GPL(nfs4_pnfs_ds_add);
795 
796 static void nfs4_wait_ds_connect(struct nfs4_pnfs_ds *ds)
797 {
798 	might_sleep();
799 	wait_on_bit(&ds->ds_state, NFS4DS_CONNECTING,
800 			TASK_KILLABLE);
801 }
802 
803 static void nfs4_clear_ds_conn_bit(struct nfs4_pnfs_ds *ds)
804 {
805 	smp_mb__before_atomic();
806 	clear_bit(NFS4DS_CONNECTING, &ds->ds_state);
807 	smp_mb__after_atomic();
808 	wake_up_bit(&ds->ds_state, NFS4DS_CONNECTING);
809 }
810 
811 static struct nfs_client *(*get_v3_ds_connect)(
812 			struct nfs_server *mds_srv,
813 			const struct sockaddr *ds_addr,
814 			int ds_addrlen,
815 			int ds_proto,
816 			unsigned int ds_timeo,
817 			unsigned int ds_retrans);
818 
819 static bool load_v3_ds_connect(void)
820 {
821 	if (!get_v3_ds_connect) {
822 		get_v3_ds_connect = symbol_request(nfs3_set_ds_client);
823 		WARN_ON_ONCE(!get_v3_ds_connect);
824 	}
825 
826 	return(get_v3_ds_connect != NULL);
827 }
828 
829 void nfs4_pnfs_v3_ds_connect_unload(void)
830 {
831 	if (get_v3_ds_connect) {
832 		symbol_put(nfs3_set_ds_client);
833 		get_v3_ds_connect = NULL;
834 	}
835 }
836 
837 static int _nfs4_pnfs_v3_ds_connect(struct nfs_server *mds_srv,
838 				 struct nfs4_pnfs_ds *ds,
839 				 unsigned int timeo,
840 				 unsigned int retrans)
841 {
842 	struct nfs_client *clp = ERR_PTR(-EIO);
843 	struct nfs4_pnfs_ds_addr *da;
844 	int status = 0;
845 
846 	dprintk("--> %s DS %s\n", __func__, ds->ds_remotestr);
847 
848 	if (!load_v3_ds_connect())
849 		goto out;
850 
851 	list_for_each_entry(da, &ds->ds_addrs, da_node) {
852 		dprintk("%s: DS %s: trying address %s\n",
853 			__func__, ds->ds_remotestr, da->da_remotestr);
854 
855 		if (!IS_ERR(clp)) {
856 			struct xprt_create xprt_args = {
857 				.ident = XPRT_TRANSPORT_TCP,
858 				.net = clp->cl_net,
859 				.dstaddr = (struct sockaddr *)&da->da_addr,
860 				.addrlen = da->da_addrlen,
861 				.servername = clp->cl_hostname,
862 			};
863 			/* Add this address as an alias */
864 			rpc_clnt_add_xprt(clp->cl_rpcclient, &xprt_args,
865 					rpc_clnt_test_and_add_xprt, NULL);
866 			continue;
867 		}
868 		clp = get_v3_ds_connect(mds_srv,
869 				(struct sockaddr *)&da->da_addr,
870 				da->da_addrlen, IPPROTO_TCP,
871 				timeo, retrans);
872 		if (IS_ERR(clp))
873 			continue;
874 		clp->cl_rpcclient->cl_softerr = 0;
875 		clp->cl_rpcclient->cl_softrtry = 0;
876 	}
877 
878 	if (IS_ERR(clp)) {
879 		status = PTR_ERR(clp);
880 		goto out;
881 	}
882 
883 	smp_wmb();
884 	ds->ds_clp = clp;
885 	dprintk("%s [new] addr: %s\n", __func__, ds->ds_remotestr);
886 out:
887 	return status;
888 }
889 
890 static int _nfs4_pnfs_v4_ds_connect(struct nfs_server *mds_srv,
891 				 struct nfs4_pnfs_ds *ds,
892 				 unsigned int timeo,
893 				 unsigned int retrans,
894 				 u32 minor_version)
895 {
896 	struct nfs_client *clp = ERR_PTR(-EIO);
897 	struct nfs4_pnfs_ds_addr *da;
898 	int status = 0;
899 
900 	dprintk("--> %s DS %s\n", __func__, ds->ds_remotestr);
901 
902 	list_for_each_entry(da, &ds->ds_addrs, da_node) {
903 		dprintk("%s: DS %s: trying address %s\n",
904 			__func__, ds->ds_remotestr, da->da_remotestr);
905 
906 		if (!IS_ERR(clp) && clp->cl_mvops->session_trunk) {
907 			struct xprt_create xprt_args = {
908 				.ident = XPRT_TRANSPORT_TCP,
909 				.net = clp->cl_net,
910 				.dstaddr = (struct sockaddr *)&da->da_addr,
911 				.addrlen = da->da_addrlen,
912 				.servername = clp->cl_hostname,
913 			};
914 			struct nfs4_add_xprt_data xprtdata = {
915 				.clp = clp,
916 				.cred = nfs4_get_clid_cred(clp),
917 			};
918 			struct rpc_add_xprt_test rpcdata = {
919 				.add_xprt_test = clp->cl_mvops->session_trunk,
920 				.data = &xprtdata,
921 			};
922 
923 			/**
924 			* Test this address for session trunking and
925 			* add as an alias
926 			*/
927 			rpc_clnt_add_xprt(clp->cl_rpcclient, &xprt_args,
928 					  rpc_clnt_setup_test_and_add_xprt,
929 					  &rpcdata);
930 			if (xprtdata.cred)
931 				put_cred(xprtdata.cred);
932 		} else {
933 			clp = nfs4_set_ds_client(mds_srv,
934 						(struct sockaddr *)&da->da_addr,
935 						da->da_addrlen, IPPROTO_TCP,
936 						timeo, retrans, minor_version);
937 			if (IS_ERR(clp))
938 				continue;
939 
940 			status = nfs4_init_ds_session(clp,
941 					mds_srv->nfs_client->cl_lease_time);
942 			if (status) {
943 				nfs_put_client(clp);
944 				clp = ERR_PTR(-EIO);
945 				continue;
946 			}
947 
948 		}
949 	}
950 
951 	if (IS_ERR(clp)) {
952 		status = PTR_ERR(clp);
953 		goto out;
954 	}
955 
956 	smp_wmb();
957 	ds->ds_clp = clp;
958 	dprintk("%s [new] addr: %s\n", __func__, ds->ds_remotestr);
959 out:
960 	return status;
961 }
962 
963 /*
964  * Create an rpc connection to the nfs4_pnfs_ds data server.
965  * Currently only supports IPv4 and IPv6 addresses.
966  * If connection fails, make devid unavailable and return a -errno.
967  */
968 int nfs4_pnfs_ds_connect(struct nfs_server *mds_srv, struct nfs4_pnfs_ds *ds,
969 			  struct nfs4_deviceid_node *devid, unsigned int timeo,
970 			  unsigned int retrans, u32 version, u32 minor_version)
971 {
972 	int err;
973 
974 again:
975 	err = 0;
976 	if (test_and_set_bit(NFS4DS_CONNECTING, &ds->ds_state) == 0) {
977 		if (version == 3) {
978 			err = _nfs4_pnfs_v3_ds_connect(mds_srv, ds, timeo,
979 						       retrans);
980 		} else if (version == 4) {
981 			err = _nfs4_pnfs_v4_ds_connect(mds_srv, ds, timeo,
982 						       retrans, minor_version);
983 		} else {
984 			dprintk("%s: unsupported DS version %d\n", __func__,
985 				version);
986 			err = -EPROTONOSUPPORT;
987 		}
988 
989 		nfs4_clear_ds_conn_bit(ds);
990 	} else {
991 		nfs4_wait_ds_connect(ds);
992 
993 		/* what was waited on didn't connect AND didn't mark unavail */
994 		if (!ds->ds_clp && !nfs4_test_deviceid_unavailable(devid))
995 			goto again;
996 	}
997 
998 	/*
999 	 * At this point the ds->ds_clp should be ready, but it might have
1000 	 * hit an error.
1001 	 */
1002 	if (!err) {
1003 		if (!ds->ds_clp || !nfs_client_init_is_complete(ds->ds_clp)) {
1004 			WARN_ON_ONCE(ds->ds_clp ||
1005 				!nfs4_test_deviceid_unavailable(devid));
1006 			return -EINVAL;
1007 		}
1008 		err = nfs_client_init_status(ds->ds_clp);
1009 	}
1010 
1011 	return err;
1012 }
1013 EXPORT_SYMBOL_GPL(nfs4_pnfs_ds_connect);
1014 
1015 /*
1016  * Currently only supports ipv4, ipv6 and one multi-path address.
1017  */
1018 struct nfs4_pnfs_ds_addr *
1019 nfs4_decode_mp_ds_addr(struct net *net, struct xdr_stream *xdr, gfp_t gfp_flags)
1020 {
1021 	struct nfs4_pnfs_ds_addr *da = NULL;
1022 	char *buf, *portstr;
1023 	__be16 port;
1024 	int nlen, rlen;
1025 	int tmp[2];
1026 	__be32 *p;
1027 	char *netid, *match_netid;
1028 	size_t len, match_netid_len;
1029 	char *startsep = "";
1030 	char *endsep = "";
1031 
1032 
1033 	/* r_netid */
1034 	p = xdr_inline_decode(xdr, 4);
1035 	if (unlikely(!p))
1036 		goto out_err;
1037 	nlen = be32_to_cpup(p++);
1038 
1039 	p = xdr_inline_decode(xdr, nlen);
1040 	if (unlikely(!p))
1041 		goto out_err;
1042 
1043 	netid = kmalloc(nlen+1, gfp_flags);
1044 	if (unlikely(!netid))
1045 		goto out_err;
1046 
1047 	netid[nlen] = '\0';
1048 	memcpy(netid, p, nlen);
1049 
1050 	/* r_addr: ip/ip6addr with port in dec octets - see RFC 5665 */
1051 	p = xdr_inline_decode(xdr, 4);
1052 	if (unlikely(!p))
1053 		goto out_free_netid;
1054 	rlen = be32_to_cpup(p);
1055 
1056 	p = xdr_inline_decode(xdr, rlen);
1057 	if (unlikely(!p))
1058 		goto out_free_netid;
1059 
1060 	/* port is ".ABC.DEF", 8 chars max */
1061 	if (rlen > INET6_ADDRSTRLEN + IPV6_SCOPE_ID_LEN + 8) {
1062 		dprintk("%s: Invalid address, length %d\n", __func__,
1063 			rlen);
1064 		goto out_free_netid;
1065 	}
1066 	buf = kmalloc(rlen + 1, gfp_flags);
1067 	if (!buf) {
1068 		dprintk("%s: Not enough memory\n", __func__);
1069 		goto out_free_netid;
1070 	}
1071 	buf[rlen] = '\0';
1072 	memcpy(buf, p, rlen);
1073 
1074 	/* replace port '.' with '-' */
1075 	portstr = strrchr(buf, '.');
1076 	if (!portstr) {
1077 		dprintk("%s: Failed finding expected dot in port\n",
1078 			__func__);
1079 		goto out_free_buf;
1080 	}
1081 	*portstr = '-';
1082 
1083 	/* find '.' between address and port */
1084 	portstr = strrchr(buf, '.');
1085 	if (!portstr) {
1086 		dprintk("%s: Failed finding expected dot between address and "
1087 			"port\n", __func__);
1088 		goto out_free_buf;
1089 	}
1090 	*portstr = '\0';
1091 
1092 	da = kzalloc(sizeof(*da), gfp_flags);
1093 	if (unlikely(!da))
1094 		goto out_free_buf;
1095 
1096 	INIT_LIST_HEAD(&da->da_node);
1097 
1098 	if (!rpc_pton(net, buf, portstr-buf, (struct sockaddr *)&da->da_addr,
1099 		      sizeof(da->da_addr))) {
1100 		dprintk("%s: error parsing address %s\n", __func__, buf);
1101 		goto out_free_da;
1102 	}
1103 
1104 	portstr++;
1105 	sscanf(portstr, "%d-%d", &tmp[0], &tmp[1]);
1106 	port = htons((tmp[0] << 8) | (tmp[1]));
1107 
1108 	switch (da->da_addr.ss_family) {
1109 	case AF_INET:
1110 		((struct sockaddr_in *)&da->da_addr)->sin_port = port;
1111 		da->da_addrlen = sizeof(struct sockaddr_in);
1112 		match_netid = "tcp";
1113 		match_netid_len = 3;
1114 		break;
1115 
1116 	case AF_INET6:
1117 		((struct sockaddr_in6 *)&da->da_addr)->sin6_port = port;
1118 		da->da_addrlen = sizeof(struct sockaddr_in6);
1119 		match_netid = "tcp6";
1120 		match_netid_len = 4;
1121 		startsep = "[";
1122 		endsep = "]";
1123 		break;
1124 
1125 	default:
1126 		dprintk("%s: unsupported address family: %u\n",
1127 			__func__, da->da_addr.ss_family);
1128 		goto out_free_da;
1129 	}
1130 
1131 	if (nlen != match_netid_len || strncmp(netid, match_netid, nlen)) {
1132 		dprintk("%s: ERROR: r_netid \"%s\" != \"%s\"\n",
1133 			__func__, netid, match_netid);
1134 		goto out_free_da;
1135 	}
1136 
1137 	/* save human readable address */
1138 	len = strlen(startsep) + strlen(buf) + strlen(endsep) + 7;
1139 	da->da_remotestr = kzalloc(len, gfp_flags);
1140 
1141 	/* NULL is ok, only used for dprintk */
1142 	if (da->da_remotestr)
1143 		snprintf(da->da_remotestr, len, "%s%s%s:%u", startsep,
1144 			 buf, endsep, ntohs(port));
1145 
1146 	dprintk("%s: Parsed DS addr %s\n", __func__, da->da_remotestr);
1147 	kfree(buf);
1148 	kfree(netid);
1149 	return da;
1150 
1151 out_free_da:
1152 	kfree(da);
1153 out_free_buf:
1154 	dprintk("%s: Error parsing DS addr: %s\n", __func__, buf);
1155 	kfree(buf);
1156 out_free_netid:
1157 	kfree(netid);
1158 out_err:
1159 	return NULL;
1160 }
1161 EXPORT_SYMBOL_GPL(nfs4_decode_mp_ds_addr);
1162 
1163 void
1164 pnfs_layout_mark_request_commit(struct nfs_page *req,
1165 				struct pnfs_layout_segment *lseg,
1166 				struct nfs_commit_info *cinfo,
1167 				u32 ds_commit_idx)
1168 {
1169 	struct list_head *list;
1170 	struct pnfs_commit_array *array;
1171 	struct pnfs_commit_bucket *bucket;
1172 
1173 	mutex_lock(&NFS_I(cinfo->inode)->commit_mutex);
1174 	array = pnfs_lookup_commit_array(cinfo->ds, lseg);
1175 	if (!array || !pnfs_is_valid_lseg(lseg))
1176 		goto out_resched;
1177 	bucket = &array->buckets[ds_commit_idx];
1178 	list = &bucket->written;
1179 	/* Non-empty buckets hold a reference on the lseg.  That ref
1180 	 * is normally transferred to the COMMIT call and released
1181 	 * there.  It could also be released if the last req is pulled
1182 	 * off due to a rewrite, in which case it will be done in
1183 	 * pnfs_common_clear_request_commit
1184 	 */
1185 	if (!bucket->lseg)
1186 		bucket->lseg = pnfs_get_lseg(lseg);
1187 	set_bit(PG_COMMIT_TO_DS, &req->wb_flags);
1188 	cinfo->ds->nwritten++;
1189 
1190 	nfs_request_add_commit_list_locked(req, list, cinfo);
1191 	mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
1192 	nfs_mark_page_unstable(req->wb_page, cinfo);
1193 	return;
1194 out_resched:
1195 	mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
1196 	cinfo->completion_ops->resched_write(cinfo, req);
1197 }
1198 EXPORT_SYMBOL_GPL(pnfs_layout_mark_request_commit);
1199 
1200 int
1201 pnfs_nfs_generic_sync(struct inode *inode, bool datasync)
1202 {
1203 	int ret;
1204 
1205 	if (!pnfs_layoutcommit_outstanding(inode))
1206 		return 0;
1207 	ret = nfs_commit_inode(inode, FLUSH_SYNC);
1208 	if (ret < 0)
1209 		return ret;
1210 	if (datasync)
1211 		return 0;
1212 	return pnfs_layoutcommit_inode(inode, true);
1213 }
1214 EXPORT_SYMBOL_GPL(pnfs_nfs_generic_sync);
1215 
1216