fs/nfs/pnfs_nfs.c

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