fs/nfs/pnfs_nfs.c

   1 /*
   2  * Common NFS I/O  operations for the pnfs file based
   3  * layout drivers.
   4  *
   5  * Copyright (c) 2014, Primary Data, Inc. All rights reserved.
   6  *
   7  * Tom Haynes <loghyr@primarydata.com>
   8  */
   9
  10 #include <linux/nfs_fs.h>
  11 #include <linux/nfs_page.h>
  12 #include <linux/sunrpc/addr.h>
  13 #include <linux/module.h>
  14
  15 #include "nfs4session.h"
  16 #include "internal.h"
  17 #include "pnfs.h"
  18
  19 #define NFSDBG_FACILITY         NFSDBG_PNFS
  20
  21 static void pnfs_generic_fenceme(struct inode *inode,
  22                                  struct pnfs_layout_hdr *lo)
  23 {
  24         if (!test_and_clear_bit(NFS_LAYOUT_RETURN, &lo->plh_flags))
  25                 return;
  26         pnfs_return_layout(inode);
  27 }
  28
  29 void pnfs_generic_rw_release(void *data)
  30 {
  31         struct nfs_pgio_header *hdr = data;
  32         struct pnfs_layout_hdr *lo = hdr->lseg->pls_layout;
  33
  34         pnfs_generic_fenceme(lo->plh_inode, lo);
  35         nfs_put_client(hdr->ds_clp);
  36         hdr->mds_ops->rpc_release(data);
  37 }
  38 EXPORT_SYMBOL_GPL(pnfs_generic_rw_release);
  39
  40 /* Fake up some data that will cause nfs_commit_release to retry the writes. */
  41 void pnfs_generic_prepare_to_resend_writes(struct nfs_commit_data *data)
  42 {
  43         struct nfs_page *first = nfs_list_entry(data->pages.next);
  44
  45         data->task.tk_status = 0;
  46         memcpy(&data->verf.verifier, &first->wb_verf,
  47                sizeof(data->verf.verifier));
  48         data->verf.verifier.data[0]++; /* ensure verifier mismatch */
  49 }
  50 EXPORT_SYMBOL_GPL(pnfs_generic_prepare_to_resend_writes);
  51
  52 void pnfs_generic_write_commit_done(struct rpc_task *task, void *data)
  53 {
  54         struct nfs_commit_data *wdata = data;
  55
  56         /* Note this may cause RPC to be resent */
  57         wdata->mds_ops->rpc_call_done(task, data);
  58 }
  59 EXPORT_SYMBOL_GPL(pnfs_generic_write_commit_done);
  60
  61 void pnfs_generic_commit_release(void *calldata)
  62 {
  63         struct nfs_commit_data *data = calldata;
  64
  65         data->completion_ops->completion(data);
  66         pnfs_put_lseg(data->lseg);
  67         nfs_put_client(data->ds_clp);
  68         nfs_commitdata_release(data);
  69 }
  70 EXPORT_SYMBOL_GPL(pnfs_generic_commit_release);
  71
  72 /* The generic layer is about to remove the req from the commit list.
  73  * If this will make the bucket empty, it will need to put the lseg reference.
  74  * Note this must be called holding the inode (/cinfo) lock
  75  */
  76 void
  77 pnfs_generic_clear_request_commit(struct nfs_page *req,
  78                                   struct nfs_commit_info *cinfo)
  79 {
  80         struct pnfs_layout_segment *freeme = NULL;
  81
  82         if (!test_and_clear_bit(PG_COMMIT_TO_DS, &req->wb_flags))
  83                 goto out;
  84         cinfo->ds->nwritten--;
  85         if (list_is_singular(&req->wb_list)) {
  86                 struct pnfs_commit_bucket *bucket;
  87
  88                 bucket = list_first_entry(&req->wb_list,
  89                                           struct pnfs_commit_bucket,
  90                                           written);
  91                 freeme = bucket->wlseg;
  92                 bucket->wlseg = NULL;
  93         }
  94 out:
  95         nfs_request_remove_commit_list(req, cinfo);
  96         pnfs_put_lseg_locked(freeme);
  97 }
  98 EXPORT_SYMBOL_GPL(pnfs_generic_clear_request_commit);
  99
 100 static int
 101 pnfs_generic_transfer_commit_list(struct list_head *src, struct list_head *dst,
 102                                   struct nfs_commit_info *cinfo, int max)
 103 {
 104         struct nfs_page *req, *tmp;
 105         int ret = 0;
 106
 107         list_for_each_entry_safe(req, tmp, src, wb_list) {
 108                 if (!nfs_lock_request(req))
 109                         continue;
 110                 kref_get(&req->wb_kref);
 111                 if (cond_resched_lock(cinfo->lock))
 112                         list_safe_reset_next(req, tmp, wb_list);
 113                 nfs_request_remove_commit_list(req, cinfo);
 114                 clear_bit(PG_COMMIT_TO_DS, &req->wb_flags);
 115                 nfs_list_add_request(req, dst);
 116                 ret++;
 117                 if ((ret == max) && !cinfo->dreq)
 118                         break;
 119         }
 120         return ret;
 121 }
 122
 123 static int
 124 pnfs_generic_scan_ds_commit_list(struct pnfs_commit_bucket *bucket,
 125                                  struct nfs_commit_info *cinfo,
 126                                  int max)
 127 {
 128         struct list_head *src = &bucket->written;
 129         struct list_head *dst = &bucket->committing;
 130         int ret;
 131
 132         lockdep_assert_held(cinfo->lock);
 133         ret = pnfs_generic_transfer_commit_list(src, dst, cinfo, max);
 134         if (ret) {
 135                 cinfo->ds->nwritten -= ret;
 136                 cinfo->ds->ncommitting += ret;
 137                 bucket->clseg = bucket->wlseg;
 138                 if (list_empty(src))
 139                         bucket->wlseg = NULL;
 140                 else
 141                         pnfs_get_lseg(bucket->clseg);
 142         }
 143         return ret;
 144 }
 145
 146 /* Move reqs from written to committing lists, returning count
 147  * of number moved.
 148  */
 149 int pnfs_generic_scan_commit_lists(struct nfs_commit_info *cinfo,
 150                                    int max)
 151 {
 152         int i, rv = 0, cnt;
 153
 154         lockdep_assert_held(cinfo->lock);
 155         for (i = 0; i < cinfo->ds->nbuckets && max != 0; i++) {
 156                 cnt = pnfs_generic_scan_ds_commit_list(&cinfo->ds->buckets[i],
 157                                                        cinfo, max);
 158                 max -= cnt;
 159                 rv += cnt;
 160         }
 161         return rv;
 162 }
 163 EXPORT_SYMBOL_GPL(pnfs_generic_scan_commit_lists);
 164
 165 /* Pull everything off the committing lists and dump into @dst.  */
 166 void pnfs_generic_recover_commit_reqs(struct list_head *dst,
 167                                       struct nfs_commit_info *cinfo)
 168 {
 169         struct pnfs_commit_bucket *b;
 170         struct pnfs_layout_segment *freeme;
 171         int i;
 172
 173         lockdep_assert_held(cinfo->lock);
 174 restart:
 175         for (i = 0, b = cinfo->ds->buckets; i < cinfo->ds->nbuckets; i++, b++) {
 176                 if (pnfs_generic_transfer_commit_list(&b->written, dst,
 177                                                       cinfo, 0)) {
 178                         freeme = b->wlseg;
 179                         b->wlseg = NULL;
 180                         spin_unlock(cinfo->lock);
 181                         pnfs_put_lseg(freeme);
 182                         spin_lock(cinfo->lock);
 183                         goto restart;
 184                 }
 185         }
 186         cinfo->ds->nwritten = 0;
 187 }
 188 EXPORT_SYMBOL_GPL(pnfs_generic_recover_commit_reqs);
 189
 190 static void pnfs_generic_retry_commit(struct nfs_commit_info *cinfo, int idx)
 191 {
 192         struct pnfs_ds_commit_info *fl_cinfo = cinfo->ds;
 193         struct pnfs_commit_bucket *bucket;
 194         struct pnfs_layout_segment *freeme;
 195         int i;
 196
 197         for (i = idx; i < fl_cinfo->nbuckets; i++) {
 198                 bucket = &fl_cinfo->buckets[i];
 199                 if (list_empty(&bucket->committing))
 200                         continue;
 201                 nfs_retry_commit(&bucket->committing, bucket->clseg, cinfo);
 202                 spin_lock(cinfo->lock);
 203                 freeme = bucket->clseg;
 204                 bucket->clseg = NULL;
 205                 spin_unlock(cinfo->lock);
 206                 pnfs_put_lseg(freeme);
 207         }
 208 }
 209
 210 static unsigned int
 211 pnfs_generic_alloc_ds_commits(struct nfs_commit_info *cinfo,
 212                               struct list_head *list)
 213 {
 214         struct pnfs_ds_commit_info *fl_cinfo;
 215         struct pnfs_commit_bucket *bucket;
 216         struct nfs_commit_data *data;
 217         int i;
 218         unsigned int nreq = 0;
 219
 220         fl_cinfo = cinfo->ds;
 221         bucket = fl_cinfo->buckets;
 222         for (i = 0; i < fl_cinfo->nbuckets; i++, bucket++) {
 223                 if (list_empty(&bucket->committing))
 224                         continue;
 225                 data = nfs_commitdata_alloc();
 226                 if (!data)
 227                         break;
 228                 data->ds_commit_index = i;
 229                 spin_lock(cinfo->lock);
 230                 data->lseg = bucket->clseg;
 231                 bucket->clseg = NULL;
 232                 spin_unlock(cinfo->lock);
 233                 list_add(&data->pages, list);
 234                 nreq++;
 235         }
 236
 237         /* Clean up on error */
 238         pnfs_generic_retry_commit(cinfo, i);
 239         return nreq;
 240 }
 241
 242 /* This follows nfs_commit_list pretty closely */
 243 int
 244 pnfs_generic_commit_pagelist(struct inode *inode, struct list_head *mds_pages,
 245                              int how, struct nfs_commit_info *cinfo,
 246                              int (*initiate_commit)(struct nfs_commit_data *data,
 247                                                     int how))
 248 {
 249         struct nfs_commit_data *data, *tmp;
 250         LIST_HEAD(list);
 251         unsigned int nreq = 0;
 252
 253         if (!list_empty(mds_pages)) {
 254                 data = nfs_commitdata_alloc();
 255                 if (data != NULL) {
 256                         data->lseg = NULL;
 257                         list_add(&data->pages, &list);
 258                         nreq++;
 259                 } else {
 260                         nfs_retry_commit(mds_pages, NULL, cinfo);
 261                         pnfs_generic_retry_commit(cinfo, 0);
 262                         cinfo->completion_ops->error_cleanup(NFS_I(inode));
 263                         return -ENOMEM;
 264                 }
 265         }
 266
 267         nreq += pnfs_generic_alloc_ds_commits(cinfo, &list);
 268
 269         if (nreq == 0) {
 270                 cinfo->completion_ops->error_cleanup(NFS_I(inode));
 271                 goto out;
 272         }
 273
 274         atomic_add(nreq, &cinfo->mds->rpcs_out);
 275
 276         list_for_each_entry_safe(data, tmp, &list, pages) {
 277                 list_del_init(&data->pages);
 278                 if (!data->lseg) {
 279                         nfs_init_commit(data, mds_pages, NULL, cinfo);
 280                         nfs_initiate_commit(NFS_CLIENT(inode), data,
 281                                             NFS_PROTO(data->inode),
 282                                             data->mds_ops, how, 0);
 283                 } else {
 284                         struct pnfs_commit_bucket *buckets;
 285
 286                         buckets = cinfo->ds->buckets;
 287                         nfs_init_commit(data,
 288                                         &buckets[data->ds_commit_index].committing,
 289                                         data->lseg,
 290                                         cinfo);
 291                         initiate_commit(data, how);
 292                 }
 293         }
 294 out:
 295         cinfo->ds->ncommitting = 0;
 296         return PNFS_ATTEMPTED;
 297 }
 298 EXPORT_SYMBOL_GPL(pnfs_generic_commit_pagelist);
 299
 300 /*
 301  * Data server cache
 302  *
 303  * Data servers can be mapped to different device ids.
 304  * nfs4_pnfs_ds reference counting
 305  *   - set to 1 on allocation
 306  *   - incremented when a device id maps a data server already in the cache.
 307  *   - decremented when deviceid is removed from the cache.
 308  */
 309 static DEFINE_SPINLOCK(nfs4_ds_cache_lock);
 310 static LIST_HEAD(nfs4_data_server_cache);
 311
 312 /* Debug routines */
 313 static void
 314 print_ds(struct nfs4_pnfs_ds *ds)
 315 {
 316         if (ds == NULL) {
 317                 printk(KERN_WARNING "%s NULL device\n", __func__);
 318                 return;
 319         }
 320         printk(KERN_WARNING "        ds %s\n"
 321                 "        ref count %d\n"
 322                 "        client %p\n"
 323                 "        cl_exchange_flags %x\n",
 324                 ds->ds_remotestr,
 325                 atomic_read(&ds->ds_count), ds->ds_clp,
 326                 ds->ds_clp ? ds->ds_clp->cl_exchange_flags : 0);
 327 }
 328
 329 static bool
 330 same_sockaddr(struct sockaddr *addr1, struct sockaddr *addr2)
 331 {
 332         struct sockaddr_in *a, *b;
 333         struct sockaddr_in6 *a6, *b6;
 334
 335         if (addr1->sa_family != addr2->sa_family)
 336                 return false;
 337
 338         switch (addr1->sa_family) {
 339         case AF_INET:
 340                 a = (struct sockaddr_in *)addr1;
 341                 b = (struct sockaddr_in *)addr2;
 342
 343                 if (a->sin_addr.s_addr == b->sin_addr.s_addr &&
 344                     a->sin_port == b->sin_port)
 345                         return true;
 346                 break;
 347
 348         case AF_INET6:
 349                 a6 = (struct sockaddr_in6 *)addr1;
 350                 b6 = (struct sockaddr_in6 *)addr2;
 351
 352                 /* LINKLOCAL addresses must have matching scope_id */
 353                 if (ipv6_addr_src_scope(&a6->sin6_addr) ==
 354                     IPV6_ADDR_SCOPE_LINKLOCAL &&
 355                     a6->sin6_scope_id != b6->sin6_scope_id)
 356                         return false;
 357
 358                 if (ipv6_addr_equal(&a6->sin6_addr, &b6->sin6_addr) &&
 359                     a6->sin6_port == b6->sin6_port)
 360                         return true;
 361                 break;
 362
 363         default:
 364                 dprintk("%s: unhandled address family: %u\n",
 365                         __func__, addr1->sa_family);
 366                 return false;
 367         }
 368
 369         return false;
 370 }
 371
 372 static bool
 373 _same_data_server_addrs_locked(const struct list_head *dsaddrs1,
 374                                const struct list_head *dsaddrs2)
 375 {
 376         struct nfs4_pnfs_ds_addr *da1, *da2;
 377
 378         /* step through both lists, comparing as we go */
 379         for (da1 = list_first_entry(dsaddrs1, typeof(*da1), da_node),
 380              da2 = list_first_entry(dsaddrs2, typeof(*da2), da_node);
 381              da1 != NULL && da2 != NULL;
 382              da1 = list_entry(da1->da_node.next, typeof(*da1), da_node),
 383              da2 = list_entry(da2->da_node.next, typeof(*da2), da_node)) {
 384                 if (!same_sockaddr((struct sockaddr *)&da1->da_addr,
 385                                    (struct sockaddr *)&da2->da_addr))
 386                         return false;
 387         }
 388         if (da1 == NULL && da2 == NULL)
 389                 return true;
 390
 391         return false;
 392 }
 393
 394 /*
 395  * Lookup DS by addresses.  nfs4_ds_cache_lock is held
 396  */
 397 static struct nfs4_pnfs_ds *
 398 _data_server_lookup_locked(const struct list_head *dsaddrs)
 399 {
 400         struct nfs4_pnfs_ds *ds;
 401
 402         list_for_each_entry(ds, &nfs4_data_server_cache, ds_node)
 403                 if (_same_data_server_addrs_locked(&ds->ds_addrs, dsaddrs))
 404                         return ds;
 405         return NULL;
 406 }
 407
 408 static void destroy_ds(struct nfs4_pnfs_ds *ds)
 409 {
 410         struct nfs4_pnfs_ds_addr *da;
 411
 412         dprintk("--> %s\n", __func__);
 413         ifdebug(FACILITY)
 414                 print_ds(ds);
 415
 416         nfs_put_client(ds->ds_clp);
 417
 418         while (!list_empty(&ds->ds_addrs)) {
 419                 da = list_first_entry(&ds->ds_addrs,
 420                                       struct nfs4_pnfs_ds_addr,
 421                                       da_node);
 422                 list_del_init(&da->da_node);
 423                 kfree(da->da_remotestr);
 424                 kfree(da);
 425         }
 426
 427         kfree(ds->ds_remotestr);
 428         kfree(ds);
 429 }
 430
 431 void nfs4_pnfs_ds_put(struct nfs4_pnfs_ds *ds)
 432 {
 433         if (atomic_dec_and_lock(&ds->ds_count,
 434                                 &nfs4_ds_cache_lock)) {
 435                 list_del_init(&ds->ds_node);
 436                 spin_unlock(&nfs4_ds_cache_lock);
 437                 destroy_ds(ds);
 438         }
 439 }
 440 EXPORT_SYMBOL_GPL(nfs4_pnfs_ds_put);
 441
 442 /*
 443  * Create a string with a human readable address and port to avoid
 444  * complicated setup around many dprinks.
 445  */
 446 static char *
 447 nfs4_pnfs_remotestr(struct list_head *dsaddrs, gfp_t gfp_flags)
 448 {
 449         struct nfs4_pnfs_ds_addr *da;
 450         char *remotestr;
 451         size_t len;
 452         char *p;
 453
 454         len = 3;        /* '{', '}' and eol */
 455         list_for_each_entry(da, dsaddrs, da_node) {
 456                 len += strlen(da->da_remotestr) + 1;    /* string plus comma */
 457         }
 458
 459         remotestr = kzalloc(len, gfp_flags);
 460         if (!remotestr)
 461                 return NULL;
 462
 463         p = remotestr;
 464         *(p++) = '{';
 465         len--;
 466         list_for_each_entry(da, dsaddrs, da_node) {
 467                 size_t ll = strlen(da->da_remotestr);
 468
 469                 if (ll > len)
 470                         goto out_err;
 471
 472                 memcpy(p, da->da_remotestr, ll);
 473                 p += ll;
 474                 len -= ll;
 475
 476                 if (len < 1)
 477                         goto out_err;
 478                 (*p++) = ',';
 479                 len--;
 480         }
 481         if (len < 2)
 482                 goto out_err;
 483         *(p++) = '}';
 484         *p = '\0';
 485         return remotestr;
 486 out_err:
 487         kfree(remotestr);
 488         return NULL;
 489 }
 490
 491 /*
 492  * Given a list of multipath struct nfs4_pnfs_ds_addr, add it to ds cache if
 493  * uncached and return cached struct nfs4_pnfs_ds.
 494  */
 495 struct nfs4_pnfs_ds *
 496 nfs4_pnfs_ds_add(struct list_head *dsaddrs, gfp_t gfp_flags)
 497 {
 498         struct nfs4_pnfs_ds *tmp_ds, *ds = NULL;
 499         char *remotestr;
 500
 501         if (list_empty(dsaddrs)) {
 502                 dprintk("%s: no addresses defined\n", __func__);
 503                 goto out;
 504         }
 505
 506         ds = kzalloc(sizeof(*ds), gfp_flags);
 507         if (!ds)
 508                 goto out;
 509
 510         /* this is only used for debugging, so it's ok if its NULL */
 511         remotestr = nfs4_pnfs_remotestr(dsaddrs, gfp_flags);
 512
 513         spin_lock(&nfs4_ds_cache_lock);
 514         tmp_ds = _data_server_lookup_locked(dsaddrs);
 515         if (tmp_ds == NULL) {
 516                 INIT_LIST_HEAD(&ds->ds_addrs);
 517                 list_splice_init(dsaddrs, &ds->ds_addrs);
 518                 ds->ds_remotestr = remotestr;
 519                 atomic_set(&ds->ds_count, 1);
 520                 INIT_LIST_HEAD(&ds->ds_node);
 521                 ds->ds_clp = NULL;
 522                 list_add(&ds->ds_node, &nfs4_data_server_cache);
 523                 dprintk("%s add new data server %s\n", __func__,
 524                         ds->ds_remotestr);
 525         } else {
 526                 kfree(remotestr);
 527                 kfree(ds);
 528                 atomic_inc(&tmp_ds->ds_count);
 529                 dprintk("%s data server %s found, inc'ed ds_count to %d\n",
 530                         __func__, tmp_ds->ds_remotestr,
 531                         atomic_read(&tmp_ds->ds_count));
 532                 ds = tmp_ds;
 533         }
 534         spin_unlock(&nfs4_ds_cache_lock);
 535 out:
 536         return ds;
 537 }
 538 EXPORT_SYMBOL_GPL(nfs4_pnfs_ds_add);
 539
 540 static void nfs4_wait_ds_connect(struct nfs4_pnfs_ds *ds)
 541 {
 542         might_sleep();
 543         wait_on_bit(&ds->ds_state, NFS4DS_CONNECTING,
 544                         TASK_KILLABLE);
 545 }
 546
 547 static void nfs4_clear_ds_conn_bit(struct nfs4_pnfs_ds *ds)
 548 {
 549         smp_mb__before_atomic();
 550         clear_bit(NFS4DS_CONNECTING, &ds->ds_state);
 551         smp_mb__after_atomic();
 552         wake_up_bit(&ds->ds_state, NFS4DS_CONNECTING);
 553 }
 554
 555 static struct nfs_client *(*get_v3_ds_connect)(
 556                         struct nfs_client *mds_clp,
 557                         const struct sockaddr *ds_addr,
 558                         int ds_addrlen,
 559                         int ds_proto,
 560                         unsigned int ds_timeo,
 561                         unsigned int ds_retrans,
 562                         rpc_authflavor_t au_flavor);
 563
 564 static bool load_v3_ds_connect(void)
 565 {
 566         if (!get_v3_ds_connect) {
 567                 get_v3_ds_connect = symbol_request(nfs3_set_ds_client);
 568                 WARN_ON_ONCE(!get_v3_ds_connect);
 569         }
 570
 571         return(get_v3_ds_connect != NULL);
 572 }
 573
 574 void __exit nfs4_pnfs_v3_ds_connect_unload(void)
 575 {
 576         if (get_v3_ds_connect) {
 577                 symbol_put(nfs3_set_ds_client);
 578                 get_v3_ds_connect = NULL;
 579         }
 580 }
 581 EXPORT_SYMBOL_GPL(nfs4_pnfs_v3_ds_connect_unload);
 582
 583 static int _nfs4_pnfs_v3_ds_connect(struct nfs_server *mds_srv,
 584                                  struct nfs4_pnfs_ds *ds,
 585                                  unsigned int timeo,
 586                                  unsigned int retrans,
 587                                  rpc_authflavor_t au_flavor)
 588 {
 589         struct nfs_client *clp = ERR_PTR(-EIO);
 590         struct nfs4_pnfs_ds_addr *da;
 591         int status = 0;
 592
 593         dprintk("--> %s DS %s au_flavor %d\n", __func__,
 594                 ds->ds_remotestr, au_flavor);
 595
 596         if (!load_v3_ds_connect())
 597                 goto out;
 598
 599         list_for_each_entry(da, &ds->ds_addrs, da_node) {
 600                 dprintk("%s: DS %s: trying address %s\n",
 601                         __func__, ds->ds_remotestr, da->da_remotestr);
 602
 603                 clp = get_v3_ds_connect(mds_srv->nfs_client,
 604                                         (struct sockaddr *)&da->da_addr,
 605                                         da->da_addrlen, IPPROTO_TCP,
 606                                         timeo, retrans, au_flavor);
 607                 if (!IS_ERR(clp))
 608                         break;
 609         }
 610
 611         if (IS_ERR(clp)) {
 612                 status = PTR_ERR(clp);
 613                 goto out;
 614         }
 615
 616         smp_wmb();
 617         ds->ds_clp = clp;
 618         dprintk("%s [new] addr: %s\n", __func__, ds->ds_remotestr);
 619 out:
 620         return status;
 621 }
 622
 623 static int _nfs4_pnfs_v4_ds_connect(struct nfs_server *mds_srv,
 624                                  struct nfs4_pnfs_ds *ds,
 625                                  unsigned int timeo,
 626                                  unsigned int retrans,
 627                                  u32 minor_version,
 628                                  rpc_authflavor_t au_flavor)
 629 {
 630         struct nfs_client *clp = ERR_PTR(-EIO);
 631         struct nfs4_pnfs_ds_addr *da;
 632         int status = 0;
 633
 634         dprintk("--> %s DS %s au_flavor %d\n", __func__, ds->ds_remotestr,
 635                 au_flavor);
 636
 637         list_for_each_entry(da, &ds->ds_addrs, da_node) {
 638                 dprintk("%s: DS %s: trying address %s\n",
 639                         __func__, ds->ds_remotestr, da->da_remotestr);
 640
 641                 clp = nfs4_set_ds_client(mds_srv->nfs_client,
 642                                         (struct sockaddr *)&da->da_addr,
 643                                         da->da_addrlen, IPPROTO_TCP,
 644                                         timeo, retrans, minor_version,
 645                                         au_flavor);
 646                 if (!IS_ERR(clp))
 647                         break;
 648         }
 649
 650         if (IS_ERR(clp)) {
 651                 status = PTR_ERR(clp);
 652                 goto out;
 653         }
 654
 655         status = nfs4_init_ds_session(clp, mds_srv->nfs_client->cl_lease_time);
 656         if (status)
 657                 goto out_put;
 658
 659         smp_wmb();
 660         ds->ds_clp = clp;
 661         dprintk("%s [new] addr: %s\n", __func__, ds->ds_remotestr);
 662 out:
 663         return status;
 664 out_put:
 665         nfs_put_client(clp);
 666         goto out;
 667 }
 668
 669 /*
 670  * Create an rpc connection to the nfs4_pnfs_ds data server.
 671  * Currently only supports IPv4 and IPv6 addresses.
 672  * If connection fails, make devid unavailable.
 673  */
 674 void nfs4_pnfs_ds_connect(struct nfs_server *mds_srv, struct nfs4_pnfs_ds *ds,
 675                           struct nfs4_deviceid_node *devid, unsigned int timeo,
 676                           unsigned int retrans, u32 version,
 677                           u32 minor_version, rpc_authflavor_t au_flavor)
 678 {
 679         if (test_and_set_bit(NFS4DS_CONNECTING, &ds->ds_state) == 0) {
 680                 int err = 0;
 681
 682                 if (version == 3) {
 683                         err = _nfs4_pnfs_v3_ds_connect(mds_srv, ds, timeo,
 684                                                        retrans, au_flavor);
 685                 } else if (version == 4) {
 686                         err = _nfs4_pnfs_v4_ds_connect(mds_srv, ds, timeo,
 687                                                        retrans, minor_version,
 688                                                        au_flavor);
 689                 } else {
 690                         dprintk("%s: unsupported DS version %d\n", __func__,
 691                                 version);
 692                         err = -EPROTONOSUPPORT;
 693                 }
 694
 695                 if (err)
 696                         nfs4_mark_deviceid_unavailable(devid);
 697                 nfs4_clear_ds_conn_bit(ds);
 698         } else {
 699                 nfs4_wait_ds_connect(ds);
 700         }
 701 }
 702 EXPORT_SYMBOL_GPL(nfs4_pnfs_ds_connect);
 703
 704 /*
 705  * Currently only supports ipv4, ipv6 and one multi-path address.
 706  */
 707 struct nfs4_pnfs_ds_addr *
 708 nfs4_decode_mp_ds_addr(struct net *net, struct xdr_stream *xdr, gfp_t gfp_flags)
 709 {
 710         struct nfs4_pnfs_ds_addr *da = NULL;
 711         char *buf, *portstr;
 712         __be16 port;
 713         int nlen, rlen;
 714         int tmp[2];
 715         __be32 *p;
 716         char *netid, *match_netid;
 717         size_t len, match_netid_len;
 718         char *startsep = "";
 719         char *endsep = "";
 720
 721
 722         /* r_netid */
 723         p = xdr_inline_decode(xdr, 4);
 724         if (unlikely(!p))
 725                 goto out_err;
 726         nlen = be32_to_cpup(p++);
 727
 728         p = xdr_inline_decode(xdr, nlen);
 729         if (unlikely(!p))
 730                 goto out_err;
 731
 732         netid = kmalloc(nlen+1, gfp_flags);
 733         if (unlikely(!netid))
 734                 goto out_err;
 735
 736         netid[nlen] = '\0';
 737         memcpy(netid, p, nlen);
 738
 739         /* r_addr: ip/ip6addr with port in dec octets - see RFC 5665 */
 740         p = xdr_inline_decode(xdr, 4);
 741         if (unlikely(!p))
 742                 goto out_free_netid;
 743         rlen = be32_to_cpup(p);
 744
 745         p = xdr_inline_decode(xdr, rlen);
 746         if (unlikely(!p))
 747                 goto out_free_netid;
 748
 749         /* port is ".ABC.DEF", 8 chars max */
 750         if (rlen > INET6_ADDRSTRLEN + IPV6_SCOPE_ID_LEN + 8) {
 751                 dprintk("%s: Invalid address, length %d\n", __func__,
 752                         rlen);
 753                 goto out_free_netid;
 754         }
 755         buf = kmalloc(rlen + 1, gfp_flags);
 756         if (!buf) {
 757                 dprintk("%s: Not enough memory\n", __func__);
 758                 goto out_free_netid;
 759         }
 760         buf[rlen] = '\0';
 761         memcpy(buf, p, rlen);
 762
 763         /* replace port '.' with '-' */
 764         portstr = strrchr(buf, '.');
 765         if (!portstr) {
 766                 dprintk("%s: Failed finding expected dot in port\n",
 767                         __func__);
 768                 goto out_free_buf;
 769         }
 770         *portstr = '-';
 771
 772         /* find '.' between address and port */
 773         portstr = strrchr(buf, '.');
 774         if (!portstr) {
 775                 dprintk("%s: Failed finding expected dot between address and "
 776                         "port\n", __func__);
 777                 goto out_free_buf;
 778         }
 779         *portstr = '\0';
 780
 781         da = kzalloc(sizeof(*da), gfp_flags);
 782         if (unlikely(!da))
 783                 goto out_free_buf;
 784
 785         INIT_LIST_HEAD(&da->da_node);
 786
 787         if (!rpc_pton(net, buf, portstr-buf, (struct sockaddr *)&da->da_addr,
 788                       sizeof(da->da_addr))) {
 789                 dprintk("%s: error parsing address %s\n", __func__, buf);
 790                 goto out_free_da;
 791         }
 792
 793         portstr++;
 794         sscanf(portstr, "%d-%d", &tmp[0], &tmp[1]);
 795         port = htons((tmp[0] << 8) | (tmp[1]));
 796
 797         switch (da->da_addr.ss_family) {
 798         case AF_INET:
 799                 ((struct sockaddr_in *)&da->da_addr)->sin_port = port;
 800                 da->da_addrlen = sizeof(struct sockaddr_in);
 801                 match_netid = "tcp";
 802                 match_netid_len = 3;
 803                 break;
 804
 805         case AF_INET6:
 806                 ((struct sockaddr_in6 *)&da->da_addr)->sin6_port = port;
 807                 da->da_addrlen = sizeof(struct sockaddr_in6);
 808                 match_netid = "tcp6";
 809                 match_netid_len = 4;
 810                 startsep = "[";
 811                 endsep = "]";
 812                 break;
 813
 814         default:
 815                 dprintk("%s: unsupported address family: %u\n",
 816                         __func__, da->da_addr.ss_family);
 817                 goto out_free_da;
 818         }
 819
 820         if (nlen != match_netid_len || strncmp(netid, match_netid, nlen)) {
 821                 dprintk("%s: ERROR: r_netid \"%s\" != \"%s\"\n",
 822                         __func__, netid, match_netid);
 823                 goto out_free_da;
 824         }
 825
 826         /* save human readable address */
 827         len = strlen(startsep) + strlen(buf) + strlen(endsep) + 7;
 828         da->da_remotestr = kzalloc(len, gfp_flags);
 829
 830         /* NULL is ok, only used for dprintk */
 831         if (da->da_remotestr)
 832                 snprintf(da->da_remotestr, len, "%s%s%s:%u", startsep,
 833                          buf, endsep, ntohs(port));
 834
 835         dprintk("%s: Parsed DS addr %s\n", __func__, da->da_remotestr);
 836         kfree(buf);
 837         kfree(netid);
 838         return da;
 839
 840 out_free_da:
 841         kfree(da);
 842 out_free_buf:
 843         dprintk("%s: Error parsing DS addr: %s\n", __func__, buf);
 844         kfree(buf);
 845 out_free_netid:
 846         kfree(netid);
 847 out_err:
 848         return NULL;
 849 }
 850 EXPORT_SYMBOL_GPL(nfs4_decode_mp_ds_addr);