pnfs_nfs.c 23.5 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11
/*
 * Common NFS I/O  operations for the pnfs file based
 * layout drivers.
 *
 * Copyright (c) 2014, Primary Data, Inc. All rights reserved.
 *
 * Tom Haynes <loghyr@primarydata.com>
 */

#include <linux/nfs_fs.h>
#include <linux/nfs_page.h>
12
#include <linux/sunrpc/addr.h>
13
#include <linux/module.h>
14

15
#include "nfs4session.h"
16 17 18
#include "internal.h"
#include "pnfs.h"

19 20
#define NFSDBG_FACILITY		NFSDBG_PNFS

21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63
void pnfs_generic_rw_release(void *data)
{
	struct nfs_pgio_header *hdr = data;

	nfs_put_client(hdr->ds_clp);
	hdr->mds_ops->rpc_release(data);
}
EXPORT_SYMBOL_GPL(pnfs_generic_rw_release);

/* Fake up some data that will cause nfs_commit_release to retry the writes. */
void pnfs_generic_prepare_to_resend_writes(struct nfs_commit_data *data)
{
	struct nfs_page *first = nfs_list_entry(data->pages.next);

	data->task.tk_status = 0;
	memcpy(&data->verf.verifier, &first->wb_verf,
	       sizeof(data->verf.verifier));
	data->verf.verifier.data[0]++; /* ensure verifier mismatch */
}
EXPORT_SYMBOL_GPL(pnfs_generic_prepare_to_resend_writes);

void pnfs_generic_write_commit_done(struct rpc_task *task, void *data)
{
	struct nfs_commit_data *wdata = data;

	/* Note this may cause RPC to be resent */
	wdata->mds_ops->rpc_call_done(task, data);
}
EXPORT_SYMBOL_GPL(pnfs_generic_write_commit_done);

void pnfs_generic_commit_release(void *calldata)
{
	struct nfs_commit_data *data = calldata;

	data->completion_ops->completion(data);
	pnfs_put_lseg(data->lseg);
	nfs_put_client(data->ds_clp);
	nfs_commitdata_release(data);
}
EXPORT_SYMBOL_GPL(pnfs_generic_commit_release);

/* The generic layer is about to remove the req from the commit list.
 * If this will make the bucket empty, it will need to put the lseg reference.
64
 * Note this must be called holding nfsi->commit_mutex
65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85
 */
void
pnfs_generic_clear_request_commit(struct nfs_page *req,
				  struct nfs_commit_info *cinfo)
{
	struct pnfs_layout_segment *freeme = NULL;

	if (!test_and_clear_bit(PG_COMMIT_TO_DS, &req->wb_flags))
		goto out;
	cinfo->ds->nwritten--;
	if (list_is_singular(&req->wb_list)) {
		struct pnfs_commit_bucket *bucket;

		bucket = list_first_entry(&req->wb_list,
					  struct pnfs_commit_bucket,
					  written);
		freeme = bucket->wlseg;
		bucket->wlseg = NULL;
	}
out:
	nfs_request_remove_commit_list(req, cinfo);
86
	pnfs_put_lseg(freeme);
87 88 89 90 91 92 93 94 95 96 97 98
}
EXPORT_SYMBOL_GPL(pnfs_generic_clear_request_commit);

static int
pnfs_generic_scan_ds_commit_list(struct pnfs_commit_bucket *bucket,
				 struct nfs_commit_info *cinfo,
				 int max)
{
	struct list_head *src = &bucket->written;
	struct list_head *dst = &bucket->committing;
	int ret;

99
	lockdep_assert_held(&NFS_I(cinfo->inode)->commit_mutex);
100
	ret = nfs_scan_commit_list(src, dst, cinfo, max);
101 102 103
	if (ret) {
		cinfo->ds->nwritten -= ret;
		cinfo->ds->ncommitting += ret;
104 105 106
		if (bucket->clseg == NULL)
			bucket->clseg = pnfs_get_lseg(bucket->wlseg);
		if (list_empty(src)) {
107
			pnfs_put_lseg(bucket->wlseg);
108
			bucket->wlseg = NULL;
109
		}
110 111 112 113
	}
	return ret;
}

114 115
/* Move reqs from written to committing lists, returning count
 * of number moved.
116 117 118 119 120 121
 */
int pnfs_generic_scan_commit_lists(struct nfs_commit_info *cinfo,
				   int max)
{
	int i, rv = 0, cnt;

122
	lockdep_assert_held(&NFS_I(cinfo->inode)->commit_mutex);
123 124 125 126 127 128 129 130 131 132
	for (i = 0; i < cinfo->ds->nbuckets && max != 0; i++) {
		cnt = pnfs_generic_scan_ds_commit_list(&cinfo->ds->buckets[i],
						       cinfo, max);
		max -= cnt;
		rv += cnt;
	}
	return rv;
}
EXPORT_SYMBOL_GPL(pnfs_generic_scan_commit_lists);

133
/* Pull everything off the committing lists and dump into @dst.  */
134 135 136 137 138
void pnfs_generic_recover_commit_reqs(struct list_head *dst,
				      struct nfs_commit_info *cinfo)
{
	struct pnfs_commit_bucket *b;
	struct pnfs_layout_segment *freeme;
139
	int nwritten;
140 141
	int i;

142
	lockdep_assert_held(&NFS_I(cinfo->inode)->commit_mutex);
143 144
restart:
	for (i = 0, b = cinfo->ds->buckets; i < cinfo->ds->nbuckets; i++, b++) {
145
		nwritten = nfs_scan_commit_list(&b->written, dst, cinfo, 0);
146 147 148 149
		if (!nwritten)
			continue;
		cinfo->ds->nwritten -= nwritten;
		if (list_empty(&b->written)) {
150 151 152 153 154 155 156 157 158 159 160 161 162 163
			freeme = b->wlseg;
			b->wlseg = NULL;
			pnfs_put_lseg(freeme);
			goto restart;
		}
	}
}
EXPORT_SYMBOL_GPL(pnfs_generic_recover_commit_reqs);

static void pnfs_generic_retry_commit(struct nfs_commit_info *cinfo, int idx)
{
	struct pnfs_ds_commit_info *fl_cinfo = cinfo->ds;
	struct pnfs_commit_bucket *bucket;
	struct pnfs_layout_segment *freeme;
164
	struct list_head *pos;
165
	LIST_HEAD(pages);
166 167
	int i;

168
	mutex_lock(&NFS_I(cinfo->inode)->commit_mutex);
169 170 171 172 173 174
	for (i = idx; i < fl_cinfo->nbuckets; i++) {
		bucket = &fl_cinfo->buckets[i];
		if (list_empty(&bucket->committing))
			continue;
		freeme = bucket->clseg;
		bucket->clseg = NULL;
175 176
		list_for_each(pos, &bucket->committing)
			cinfo->ds->ncommitting--;
177
		list_splice_init(&bucket->committing, &pages);
178
		mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
179
		nfs_retry_commit(&pages, freeme, cinfo, i);
180
		pnfs_put_lseg(freeme);
181
		mutex_lock(&NFS_I(cinfo->inode)->commit_mutex);
182
	}
183
	mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200
}

static unsigned int
pnfs_generic_alloc_ds_commits(struct nfs_commit_info *cinfo,
			      struct list_head *list)
{
	struct pnfs_ds_commit_info *fl_cinfo;
	struct pnfs_commit_bucket *bucket;
	struct nfs_commit_data *data;
	int i;
	unsigned int nreq = 0;

	fl_cinfo = cinfo->ds;
	bucket = fl_cinfo->buckets;
	for (i = 0; i < fl_cinfo->nbuckets; i++, bucket++) {
		if (list_empty(&bucket->committing))
			continue;
NeilBrown's avatar
NeilBrown committed
201
		data = nfs_commitdata_alloc(false);
202 203 204 205 206 207 208 209 210 211 212 213
		if (!data)
			break;
		data->ds_commit_index = i;
		list_add(&data->pages, list);
		nreq++;
	}

	/* Clean up on error */
	pnfs_generic_retry_commit(cinfo, i);
	return nreq;
}

214 215 216 217 218 219
static inline
void pnfs_fetch_commit_bucket_list(struct list_head *pages,
		struct nfs_commit_data *data,
		struct nfs_commit_info *cinfo)
{
	struct pnfs_commit_bucket *bucket;
220
	struct list_head *pos;
221 222

	bucket = &cinfo->ds->buckets[data->ds_commit_index];
223
	mutex_lock(&NFS_I(cinfo->inode)->commit_mutex);
224 225
	list_for_each(pos, &bucket->committing)
		cinfo->ds->ncommitting--;
Peng Tao's avatar
Peng Tao committed
226
	list_splice_init(&bucket->committing, pages);
227 228
	data->lseg = bucket->clseg;
	bucket->clseg = NULL;
229
	mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
230 231 232

}

233
/* Helper function for pnfs_generic_commit_pagelist to catch an empty
234 235 236 237 238
 * page list. This can happen when two commits race.
 *
 * This must be called instead of nfs_init_commit - call one or the other, but
 * not both!
 */
239 240 241 242 243 244 245
static bool
pnfs_generic_commit_cancel_empty_pagelist(struct list_head *pages,
					  struct nfs_commit_data *data,
					  struct nfs_commit_info *cinfo)
{
	if (list_empty(pages)) {
		if (atomic_dec_and_test(&cinfo->mds->rpcs_out))
246
			wake_up_var(&cinfo->mds->rpcs_out);
247 248 249 250 251
		/* don't call nfs_commitdata_release - it tries to put
		 * the open_context which is not acquired until nfs_init_commit
		 * which has not been called on @data */
		WARN_ON_ONCE(data->context);
		nfs_commit_free(data);
252 253 254 255 256 257
		return true;
	}

	return false;
}

258 259 260 261 262 263 264 265 266 267 268 269
/* This follows nfs_commit_list pretty closely */
int
pnfs_generic_commit_pagelist(struct inode *inode, struct list_head *mds_pages,
			     int how, struct nfs_commit_info *cinfo,
			     int (*initiate_commit)(struct nfs_commit_data *data,
						    int how))
{
	struct nfs_commit_data *data, *tmp;
	LIST_HEAD(list);
	unsigned int nreq = 0;

	if (!list_empty(mds_pages)) {
NeilBrown's avatar
NeilBrown committed
270 271 272 273
		data = nfs_commitdata_alloc(true);
		data->ds_commit_index = -1;
		list_add(&data->pages, &list);
		nreq++;
274 275 276 277
	}

	nreq += pnfs_generic_alloc_ds_commits(cinfo, &list);

278
	if (nreq == 0)
279 280 281 282 283 284
		goto out;

	atomic_add(nreq, &cinfo->mds->rpcs_out);

	list_for_each_entry_safe(data, tmp, &list, pages) {
		list_del_init(&data->pages);
285
		if (data->ds_commit_index < 0) {
286 287 288 289 290
			/* another commit raced with us */
			if (pnfs_generic_commit_cancel_empty_pagelist(mds_pages,
				data, cinfo))
				continue;

291 292
			nfs_init_commit(data, mds_pages, NULL, cinfo);
			nfs_initiate_commit(NFS_CLIENT(inode), data,
293
					    NFS_PROTO(data->inode),
294 295
					    data->mds_ops, how, 0);
		} else {
296
			LIST_HEAD(pages);
297

298
			pnfs_fetch_commit_bucket_list(&pages, data, cinfo);
299 300 301 302 303 304

			/* another commit raced with us */
			if (pnfs_generic_commit_cancel_empty_pagelist(&pages,
				data, cinfo))
				continue;

305
			nfs_init_commit(data, &pages, data->lseg, cinfo);
306 307 308 309 310 311 312
			initiate_commit(data, how);
		}
	}
out:
	return PNFS_ATTEMPTED;
}
EXPORT_SYMBOL_GPL(pnfs_generic_commit_pagelist);
313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338

/*
 * Data server cache
 *
 * Data servers can be mapped to different device ids.
 * nfs4_pnfs_ds reference counting
 *   - set to 1 on allocation
 *   - incremented when a device id maps a data server already in the cache.
 *   - decremented when deviceid is removed from the cache.
 */
static DEFINE_SPINLOCK(nfs4_ds_cache_lock);
static LIST_HEAD(nfs4_data_server_cache);

/* Debug routines */
static void
print_ds(struct nfs4_pnfs_ds *ds)
{
	if (ds == NULL) {
		printk(KERN_WARNING "%s NULL device\n", __func__);
		return;
	}
	printk(KERN_WARNING "        ds %s\n"
		"        ref count %d\n"
		"        client %p\n"
		"        cl_exchange_flags %x\n",
		ds->ds_remotestr,
339
		refcount_read(&ds->ds_count), ds->ds_clp,
340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385
		ds->ds_clp ? ds->ds_clp->cl_exchange_flags : 0);
}

static bool
same_sockaddr(struct sockaddr *addr1, struct sockaddr *addr2)
{
	struct sockaddr_in *a, *b;
	struct sockaddr_in6 *a6, *b6;

	if (addr1->sa_family != addr2->sa_family)
		return false;

	switch (addr1->sa_family) {
	case AF_INET:
		a = (struct sockaddr_in *)addr1;
		b = (struct sockaddr_in *)addr2;

		if (a->sin_addr.s_addr == b->sin_addr.s_addr &&
		    a->sin_port == b->sin_port)
			return true;
		break;

	case AF_INET6:
		a6 = (struct sockaddr_in6 *)addr1;
		b6 = (struct sockaddr_in6 *)addr2;

		/* LINKLOCAL addresses must have matching scope_id */
		if (ipv6_addr_src_scope(&a6->sin6_addr) ==
		    IPV6_ADDR_SCOPE_LINKLOCAL &&
		    a6->sin6_scope_id != b6->sin6_scope_id)
			return false;

		if (ipv6_addr_equal(&a6->sin6_addr, &b6->sin6_addr) &&
		    a6->sin6_port == b6->sin6_port)
			return true;
		break;

	default:
		dprintk("%s: unhandled address family: %u\n",
			__func__, addr1->sa_family);
		return false;
	}

	return false;
}

386 387 388 389
/*
 * Checks if 'dsaddrs1' contains a subset of 'dsaddrs2'. If it does,
 * declare a match.
 */
390 391 392 393 394
static bool
_same_data_server_addrs_locked(const struct list_head *dsaddrs1,
			       const struct list_head *dsaddrs2)
{
	struct nfs4_pnfs_ds_addr *da1, *da2;
395 396 397 398 399 400 401 402 403 404 405 406 407 408
	struct sockaddr *sa1, *sa2;
	bool match = false;

	list_for_each_entry(da1, dsaddrs1, da_node) {
		sa1 = (struct sockaddr *)&da1->da_addr;
		match = false;
		list_for_each_entry(da2, dsaddrs2, da_node) {
			sa2 = (struct sockaddr *)&da2->da_addr;
			match = same_sockaddr(sa1, sa2);
			if (match)
				break;
		}
		if (!match)
			break;
409
	}
410
	return match;
411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451
}

/*
 * Lookup DS by addresses.  nfs4_ds_cache_lock is held
 */
static struct nfs4_pnfs_ds *
_data_server_lookup_locked(const struct list_head *dsaddrs)
{
	struct nfs4_pnfs_ds *ds;

	list_for_each_entry(ds, &nfs4_data_server_cache, ds_node)
		if (_same_data_server_addrs_locked(&ds->ds_addrs, dsaddrs))
			return ds;
	return NULL;
}

static void destroy_ds(struct nfs4_pnfs_ds *ds)
{
	struct nfs4_pnfs_ds_addr *da;

	dprintk("--> %s\n", __func__);
	ifdebug(FACILITY)
		print_ds(ds);

	nfs_put_client(ds->ds_clp);

	while (!list_empty(&ds->ds_addrs)) {
		da = list_first_entry(&ds->ds_addrs,
				      struct nfs4_pnfs_ds_addr,
				      da_node);
		list_del_init(&da->da_node);
		kfree(da->da_remotestr);
		kfree(da);
	}

	kfree(ds->ds_remotestr);
	kfree(ds);
}

void nfs4_pnfs_ds_put(struct nfs4_pnfs_ds *ds)
{
452
	if (refcount_dec_and_lock(&ds->ds_count,
453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537
				&nfs4_ds_cache_lock)) {
		list_del_init(&ds->ds_node);
		spin_unlock(&nfs4_ds_cache_lock);
		destroy_ds(ds);
	}
}
EXPORT_SYMBOL_GPL(nfs4_pnfs_ds_put);

/*
 * Create a string with a human readable address and port to avoid
 * complicated setup around many dprinks.
 */
static char *
nfs4_pnfs_remotestr(struct list_head *dsaddrs, gfp_t gfp_flags)
{
	struct nfs4_pnfs_ds_addr *da;
	char *remotestr;
	size_t len;
	char *p;

	len = 3;        /* '{', '}' and eol */
	list_for_each_entry(da, dsaddrs, da_node) {
		len += strlen(da->da_remotestr) + 1;    /* string plus comma */
	}

	remotestr = kzalloc(len, gfp_flags);
	if (!remotestr)
		return NULL;

	p = remotestr;
	*(p++) = '{';
	len--;
	list_for_each_entry(da, dsaddrs, da_node) {
		size_t ll = strlen(da->da_remotestr);

		if (ll > len)
			goto out_err;

		memcpy(p, da->da_remotestr, ll);
		p += ll;
		len -= ll;

		if (len < 1)
			goto out_err;
		(*p++) = ',';
		len--;
	}
	if (len < 2)
		goto out_err;
	*(p++) = '}';
	*p = '\0';
	return remotestr;
out_err:
	kfree(remotestr);
	return NULL;
}

/*
 * Given a list of multipath struct nfs4_pnfs_ds_addr, add it to ds cache if
 * uncached and return cached struct nfs4_pnfs_ds.
 */
struct nfs4_pnfs_ds *
nfs4_pnfs_ds_add(struct list_head *dsaddrs, gfp_t gfp_flags)
{
	struct nfs4_pnfs_ds *tmp_ds, *ds = NULL;
	char *remotestr;

	if (list_empty(dsaddrs)) {
		dprintk("%s: no addresses defined\n", __func__);
		goto out;
	}

	ds = kzalloc(sizeof(*ds), gfp_flags);
	if (!ds)
		goto out;

	/* this is only used for debugging, so it's ok if its NULL */
	remotestr = nfs4_pnfs_remotestr(dsaddrs, gfp_flags);

	spin_lock(&nfs4_ds_cache_lock);
	tmp_ds = _data_server_lookup_locked(dsaddrs);
	if (tmp_ds == NULL) {
		INIT_LIST_HEAD(&ds->ds_addrs);
		list_splice_init(dsaddrs, &ds->ds_addrs);
		ds->ds_remotestr = remotestr;
538
		refcount_set(&ds->ds_count, 1);
539 540 541 542 543 544 545 546
		INIT_LIST_HEAD(&ds->ds_node);
		ds->ds_clp = NULL;
		list_add(&ds->ds_node, &nfs4_data_server_cache);
		dprintk("%s add new data server %s\n", __func__,
			ds->ds_remotestr);
	} else {
		kfree(remotestr);
		kfree(ds);
547
		refcount_inc(&tmp_ds->ds_count);
548 549
		dprintk("%s data server %s found, inc'ed ds_count to %d\n",
			__func__, tmp_ds->ds_remotestr,
550
			refcount_read(&tmp_ds->ds_count));
551 552 553 554 555 556 557
		ds = tmp_ds;
	}
	spin_unlock(&nfs4_ds_cache_lock);
out:
	return ds;
}
EXPORT_SYMBOL_GPL(nfs4_pnfs_ds_add);
558

559 560 561 562 563 564 565 566 567 568 569 570 571 572 573
static void nfs4_wait_ds_connect(struct nfs4_pnfs_ds *ds)
{
	might_sleep();
	wait_on_bit(&ds->ds_state, NFS4DS_CONNECTING,
			TASK_KILLABLE);
}

static void nfs4_clear_ds_conn_bit(struct nfs4_pnfs_ds *ds)
{
	smp_mb__before_atomic();
	clear_bit(NFS4DS_CONNECTING, &ds->ds_state);
	smp_mb__after_atomic();
	wake_up_bit(&ds->ds_state, NFS4DS_CONNECTING);
}

574
static struct nfs_client *(*get_v3_ds_connect)(
575
			struct nfs_server *mds_srv,
576 577 578 579
			const struct sockaddr *ds_addr,
			int ds_addrlen,
			int ds_proto,
			unsigned int ds_timeo,
580
			unsigned int ds_retrans);
581 582 583 584 585 586 587 588 589 590 591

static bool load_v3_ds_connect(void)
{
	if (!get_v3_ds_connect) {
		get_v3_ds_connect = symbol_request(nfs3_set_ds_client);
		WARN_ON_ONCE(!get_v3_ds_connect);
	}

	return(get_v3_ds_connect != NULL);
}

592
void nfs4_pnfs_v3_ds_connect_unload(void)
593 594 595 596 597 598 599 600 601 602
{
	if (get_v3_ds_connect) {
		symbol_put(nfs3_set_ds_client);
		get_v3_ds_connect = NULL;
	}
}

static int _nfs4_pnfs_v3_ds_connect(struct nfs_server *mds_srv,
				 struct nfs4_pnfs_ds *ds,
				 unsigned int timeo,
603
				 unsigned int retrans)
604 605 606 607 608
{
	struct nfs_client *clp = ERR_PTR(-EIO);
	struct nfs4_pnfs_ds_addr *da;
	int status = 0;

609
	dprintk("--> %s DS %s\n", __func__, ds->ds_remotestr);
610 611 612 613 614 615 616 617

	if (!load_v3_ds_connect())
		goto out;

	list_for_each_entry(da, &ds->ds_addrs, da_node) {
		dprintk("%s: DS %s: trying address %s\n",
			__func__, ds->ds_remotestr, da->da_remotestr);

618 619 620 621 622 623 624 625 626 627 628 629
		if (!IS_ERR(clp)) {
			struct xprt_create xprt_args = {
				.ident = XPRT_TRANSPORT_TCP,
				.net = clp->cl_net,
				.dstaddr = (struct sockaddr *)&da->da_addr,
				.addrlen = da->da_addrlen,
				.servername = clp->cl_hostname,
			};
			/* Add this address as an alias */
			rpc_clnt_add_xprt(clp->cl_rpcclient, &xprt_args,
					rpc_clnt_test_and_add_xprt, NULL);
		} else
630
			clp = get_v3_ds_connect(mds_srv,
631 632
					(struct sockaddr *)&da->da_addr,
					da->da_addrlen, IPPROTO_TCP,
633
					timeo, retrans);
634 635 636 637 638 639 640 641 642 643 644 645 646 647 648
	}

	if (IS_ERR(clp)) {
		status = PTR_ERR(clp);
		goto out;
	}

	smp_wmb();
	ds->ds_clp = clp;
	dprintk("%s [new] addr: %s\n", __func__, ds->ds_remotestr);
out:
	return status;
}

static int _nfs4_pnfs_v4_ds_connect(struct nfs_server *mds_srv,
649 650
				 struct nfs4_pnfs_ds *ds,
				 unsigned int timeo,
651
				 unsigned int retrans,
652
				 u32 minor_version)
653 654 655 656 657
{
	struct nfs_client *clp = ERR_PTR(-EIO);
	struct nfs4_pnfs_ds_addr *da;
	int status = 0;

658
	dprintk("--> %s DS %s\n", __func__, ds->ds_remotestr);
659 660 661 662 663

	list_for_each_entry(da, &ds->ds_addrs, da_node) {
		dprintk("%s: DS %s: trying address %s\n",
			__func__, ds->ds_remotestr, da->da_remotestr);

664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688
		if (!IS_ERR(clp) && clp->cl_mvops->session_trunk) {
			struct xprt_create xprt_args = {
				.ident = XPRT_TRANSPORT_TCP,
				.net = clp->cl_net,
				.dstaddr = (struct sockaddr *)&da->da_addr,
				.addrlen = da->da_addrlen,
				.servername = clp->cl_hostname,
			};
			struct nfs4_add_xprt_data xprtdata = {
				.clp = clp,
				.cred = nfs4_get_clid_cred(clp),
			};
			struct rpc_add_xprt_test rpcdata = {
				.add_xprt_test = clp->cl_mvops->session_trunk,
				.data = &xprtdata,
			};

			/**
			* Test this address for session trunking and
			* add as an alias
			*/
			rpc_clnt_add_xprt(clp->cl_rpcclient, &xprt_args,
					  rpc_clnt_setup_test_and_add_xprt,
					  &rpcdata);
			if (xprtdata.cred)
689
				put_cred(xprtdata.cred);
690 691 692 693
		} else {
			clp = nfs4_set_ds_client(mds_srv,
						(struct sockaddr *)&da->da_addr,
						da->da_addrlen, IPPROTO_TCP,
694
						timeo, retrans, minor_version);
695 696 697 698 699 700 701 702 703 704 705 706
			if (IS_ERR(clp))
				continue;

			status = nfs4_init_ds_session(clp,
					mds_srv->nfs_client->cl_lease_time);
			if (status) {
				nfs_put_client(clp);
				clp = ERR_PTR(-EIO);
				continue;
			}

		}
707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723
	}

	if (IS_ERR(clp)) {
		status = PTR_ERR(clp);
		goto out;
	}

	smp_wmb();
	ds->ds_clp = clp;
	dprintk("%s [new] addr: %s\n", __func__, ds->ds_remotestr);
out:
	return status;
}

/*
 * Create an rpc connection to the nfs4_pnfs_ds data server.
 * Currently only supports IPv4 and IPv6 addresses.
724
 * If connection fails, make devid unavailable and return a -errno.
725
 */
726
int nfs4_pnfs_ds_connect(struct nfs_server *mds_srv, struct nfs4_pnfs_ds *ds,
727
			  struct nfs4_deviceid_node *devid, unsigned int timeo,
728
			  unsigned int retrans, u32 version, u32 minor_version)
729
{
730
	int err;
731

732 733 734
again:
	err = 0;
	if (test_and_set_bit(NFS4DS_CONNECTING, &ds->ds_state) == 0) {
735 736
		if (version == 3) {
			err = _nfs4_pnfs_v3_ds_connect(mds_srv, ds, timeo,
737
						       retrans);
738 739
		} else if (version == 4) {
			err = _nfs4_pnfs_v4_ds_connect(mds_srv, ds, timeo,
740
						       retrans, minor_version);
741 742 743 744 745 746
		} else {
			dprintk("%s: unsupported DS version %d\n", __func__,
				version);
			err = -EPROTONOSUPPORT;
		}

747 748 749
		nfs4_clear_ds_conn_bit(ds);
	} else {
		nfs4_wait_ds_connect(ds);
750 751 752 753

		/* what was waited on didn't connect AND didn't mark unavail */
		if (!ds->ds_clp && !nfs4_test_deviceid_unavailable(devid))
			goto again;
754
	}
755 756 757 758 759

	/*
	 * At this point the ds->ds_clp should be ready, but it might have
	 * hit an error.
	 */
760 761 762 763 764 765 766
	if (!err) {
		if (!ds->ds_clp || !nfs_client_init_is_complete(ds->ds_clp)) {
			WARN_ON_ONCE(ds->ds_clp ||
				!nfs4_test_deviceid_unavailable(devid));
			return -EINVAL;
		}
		err = nfs_client_init_status(ds->ds_clp);
767 768
	}

769
	return err;
770 771 772
}
EXPORT_SYMBOL_GPL(nfs4_pnfs_ds_connect);

773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919
/*
 * Currently only supports ipv4, ipv6 and one multi-path address.
 */
struct nfs4_pnfs_ds_addr *
nfs4_decode_mp_ds_addr(struct net *net, struct xdr_stream *xdr, gfp_t gfp_flags)
{
	struct nfs4_pnfs_ds_addr *da = NULL;
	char *buf, *portstr;
	__be16 port;
	int nlen, rlen;
	int tmp[2];
	__be32 *p;
	char *netid, *match_netid;
	size_t len, match_netid_len;
	char *startsep = "";
	char *endsep = "";


	/* r_netid */
	p = xdr_inline_decode(xdr, 4);
	if (unlikely(!p))
		goto out_err;
	nlen = be32_to_cpup(p++);

	p = xdr_inline_decode(xdr, nlen);
	if (unlikely(!p))
		goto out_err;

	netid = kmalloc(nlen+1, gfp_flags);
	if (unlikely(!netid))
		goto out_err;

	netid[nlen] = '\0';
	memcpy(netid, p, nlen);

	/* r_addr: ip/ip6addr with port in dec octets - see RFC 5665 */
	p = xdr_inline_decode(xdr, 4);
	if (unlikely(!p))
		goto out_free_netid;
	rlen = be32_to_cpup(p);

	p = xdr_inline_decode(xdr, rlen);
	if (unlikely(!p))
		goto out_free_netid;

	/* port is ".ABC.DEF", 8 chars max */
	if (rlen > INET6_ADDRSTRLEN + IPV6_SCOPE_ID_LEN + 8) {
		dprintk("%s: Invalid address, length %d\n", __func__,
			rlen);
		goto out_free_netid;
	}
	buf = kmalloc(rlen + 1, gfp_flags);
	if (!buf) {
		dprintk("%s: Not enough memory\n", __func__);
		goto out_free_netid;
	}
	buf[rlen] = '\0';
	memcpy(buf, p, rlen);

	/* replace port '.' with '-' */
	portstr = strrchr(buf, '.');
	if (!portstr) {
		dprintk("%s: Failed finding expected dot in port\n",
			__func__);
		goto out_free_buf;
	}
	*portstr = '-';

	/* find '.' between address and port */
	portstr = strrchr(buf, '.');
	if (!portstr) {
		dprintk("%s: Failed finding expected dot between address and "
			"port\n", __func__);
		goto out_free_buf;
	}
	*portstr = '\0';

	da = kzalloc(sizeof(*da), gfp_flags);
	if (unlikely(!da))
		goto out_free_buf;

	INIT_LIST_HEAD(&da->da_node);

	if (!rpc_pton(net, buf, portstr-buf, (struct sockaddr *)&da->da_addr,
		      sizeof(da->da_addr))) {
		dprintk("%s: error parsing address %s\n", __func__, buf);
		goto out_free_da;
	}

	portstr++;
	sscanf(portstr, "%d-%d", &tmp[0], &tmp[1]);
	port = htons((tmp[0] << 8) | (tmp[1]));

	switch (da->da_addr.ss_family) {
	case AF_INET:
		((struct sockaddr_in *)&da->da_addr)->sin_port = port;
		da->da_addrlen = sizeof(struct sockaddr_in);
		match_netid = "tcp";
		match_netid_len = 3;
		break;

	case AF_INET6:
		((struct sockaddr_in6 *)&da->da_addr)->sin6_port = port;
		da->da_addrlen = sizeof(struct sockaddr_in6);
		match_netid = "tcp6";
		match_netid_len = 4;
		startsep = "[";
		endsep = "]";
		break;

	default:
		dprintk("%s: unsupported address family: %u\n",
			__func__, da->da_addr.ss_family);
		goto out_free_da;
	}

	if (nlen != match_netid_len || strncmp(netid, match_netid, nlen)) {
		dprintk("%s: ERROR: r_netid \"%s\" != \"%s\"\n",
			__func__, netid, match_netid);
		goto out_free_da;
	}

	/* save human readable address */
	len = strlen(startsep) + strlen(buf) + strlen(endsep) + 7;
	da->da_remotestr = kzalloc(len, gfp_flags);

	/* NULL is ok, only used for dprintk */
	if (da->da_remotestr)
		snprintf(da->da_remotestr, len, "%s%s%s:%u", startsep,
			 buf, endsep, ntohs(port));

	dprintk("%s: Parsed DS addr %s\n", __func__, da->da_remotestr);
	kfree(buf);
	kfree(netid);
	return da;

out_free_da:
	kfree(da);
out_free_buf:
	dprintk("%s: Error parsing DS addr: %s\n", __func__, buf);
	kfree(buf);
out_free_netid:
	kfree(netid);
out_err:
	return NULL;
}
EXPORT_SYMBOL_GPL(nfs4_decode_mp_ds_addr);
920 921 922 923 924 925 926 927 928 929

void
pnfs_layout_mark_request_commit(struct nfs_page *req,
				struct pnfs_layout_segment *lseg,
				struct nfs_commit_info *cinfo,
				u32 ds_commit_idx)
{
	struct list_head *list;
	struct pnfs_commit_bucket *buckets;

930
	mutex_lock(&NFS_I(cinfo->inode)->commit_mutex);
931 932 933
	buckets = cinfo->ds->buckets;
	list = &buckets[ds_commit_idx].written;
	if (list_empty(list)) {
934
		if (!pnfs_is_valid_lseg(lseg)) {
935
			mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
936 937 938
			cinfo->completion_ops->resched_write(cinfo, req);
			return;
		}
939 940 941 942 943 944 945 946 947 948 949 950
		/* Non-empty buckets hold a reference on the lseg.  That ref
		 * is normally transferred to the COMMIT call and released
		 * there.  It could also be released if the last req is pulled
		 * off due to a rewrite, in which case it will be done in
		 * pnfs_common_clear_request_commit
		 */
		WARN_ON_ONCE(buckets[ds_commit_idx].wlseg != NULL);
		buckets[ds_commit_idx].wlseg = pnfs_get_lseg(lseg);
	}
	set_bit(PG_COMMIT_TO_DS, &req->wb_flags);
	cinfo->ds->nwritten++;

951
	nfs_request_add_commit_list_locked(req, list, cinfo);
952
	mutex_unlock(&NFS_I(cinfo->inode)->commit_mutex);
953
	nfs_mark_page_unstable(req->wb_page, cinfo);
954 955
}
EXPORT_SYMBOL_GPL(pnfs_layout_mark_request_commit);
956 957 958 959

int
pnfs_nfs_generic_sync(struct inode *inode, bool datasync)
{
960 961 962 963 964 965 966
	int ret;

	if (!pnfs_layoutcommit_outstanding(inode))
		return 0;
	ret = nfs_commit_inode(inode, FLUSH_SYNC);
	if (ret < 0)
		return ret;
967 968 969 970 971 972
	if (datasync)
		return 0;
	return pnfs_layoutcommit_inode(inode, true);
}
EXPORT_SYMBOL_GPL(pnfs_nfs_generic_sync);