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