fs/nfs/flexfilelayout/flexfilelayout.c

   1 /*
   2  * Module for pnfs flexfile layout driver.
   3  *
   4  * Copyright (c) 2014, Primary Data, Inc. All rights reserved.
   5  *
   6  * Tao Peng <bergwolf@primarydata.com>
   7  */
   8
   9 #include <linux/nfs_fs.h>
  10 #include <linux/nfs_page.h>
  11 #include <linux/module.h>
  12
  13 #include <linux/sunrpc/metrics.h>
  14
  15 #include "flexfilelayout.h"
  16 #include "../nfs4session.h"
  17 #include "../nfs4idmap.h"
  18 #include "../internal.h"
  19 #include "../delegation.h"
  20 #include "../nfs4trace.h"
  21 #include "../iostat.h"
  22 #include "../nfs.h"
  23 #include "../nfs42.h"
  24
  25 #define NFSDBG_FACILITY         NFSDBG_PNFS_LD
  26
  27 #define FF_LAYOUT_POLL_RETRY_MAX     (15*HZ)
  28 #define FF_LAYOUTRETURN_MAXERR 20
  29
  30
  31 static struct group_info        *ff_zero_group;
  32
  33 static void ff_layout_read_record_layoutstats_done(struct rpc_task *task,
  34                 struct nfs_pgio_header *hdr);
  35
  36 static struct pnfs_layout_hdr *
  37 ff_layout_alloc_layout_hdr(struct inode *inode, gfp_t gfp_flags)
  38 {
  39         struct nfs4_flexfile_layout *ffl;
  40
  41         ffl = kzalloc(sizeof(*ffl), gfp_flags);
  42         if (ffl) {
  43                 INIT_LIST_HEAD(&ffl->error_list);
  44                 INIT_LIST_HEAD(&ffl->mirrors);
  45                 ffl->last_report_time = ktime_get();
  46                 return &ffl->generic_hdr;
  47         } else
  48                 return NULL;
  49 }
  50
  51 static void
  52 ff_layout_free_layout_hdr(struct pnfs_layout_hdr *lo)
  53 {
  54         struct nfs4_ff_layout_ds_err *err, *n;
  55
  56         list_for_each_entry_safe(err, n, &FF_LAYOUT_FROM_HDR(lo)->error_list,
  57                                  list) {
  58                 list_del(&err->list);
  59                 kfree(err);
  60         }
  61         kfree(FF_LAYOUT_FROM_HDR(lo));
  62 }
  63
  64 static int decode_pnfs_stateid(struct xdr_stream *xdr, nfs4_stateid *stateid)
  65 {
  66         __be32 *p;
  67
  68         p = xdr_inline_decode(xdr, NFS4_STATEID_SIZE);
  69         if (unlikely(p == NULL))
  70                 return -ENOBUFS;
  71         stateid->type = NFS4_PNFS_DS_STATEID_TYPE;
  72         memcpy(stateid->data, p, NFS4_STATEID_SIZE);
  73         dprintk("%s: stateid id= [%x%x%x%x]\n", __func__,
  74                 p[0], p[1], p[2], p[3]);
  75         return 0;
  76 }
  77
  78 static int decode_deviceid(struct xdr_stream *xdr, struct nfs4_deviceid *devid)
  79 {
  80         __be32 *p;
  81
  82         p = xdr_inline_decode(xdr, NFS4_DEVICEID4_SIZE);
  83         if (unlikely(!p))
  84                 return -ENOBUFS;
  85         memcpy(devid, p, NFS4_DEVICEID4_SIZE);
  86         nfs4_print_deviceid(devid);
  87         return 0;
  88 }
  89
  90 static int decode_nfs_fh(struct xdr_stream *xdr, struct nfs_fh *fh)
  91 {
  92         __be32 *p;
  93
  94         p = xdr_inline_decode(xdr, 4);
  95         if (unlikely(!p))
  96                 return -ENOBUFS;
  97         fh->size = be32_to_cpup(p++);
  98         if (fh->size > sizeof(struct nfs_fh)) {
  99                 printk(KERN_ERR "NFS flexfiles: Too big fh received %d\n",
 100                        fh->size);
 101                 return -EOVERFLOW;
 102         }
 103         /* fh.data */
 104         p = xdr_inline_decode(xdr, fh->size);
 105         if (unlikely(!p))
 106                 return -ENOBUFS;
 107         memcpy(&fh->data, p, fh->size);
 108         dprintk("%s: fh len %d\n", __func__, fh->size);
 109
 110         return 0;
 111 }
 112
 113 /*
 114  * Currently only stringified uids and gids are accepted.
 115  * I.e., kerberos is not supported to the DSes, so no pricipals.
 116  *
 117  * That means that one common function will suffice, but when
 118  * principals are added, this should be split to accomodate
 119  * calls to both nfs_map_name_to_uid() and nfs_map_group_to_gid().
 120  */
 121 static int
 122 decode_name(struct xdr_stream *xdr, u32 *id)
 123 {
 124         __be32 *p;
 125         int len;
 126
 127         /* opaque_length(4)*/
 128         p = xdr_inline_decode(xdr, 4);
 129         if (unlikely(!p))
 130                 return -ENOBUFS;
 131         len = be32_to_cpup(p++);
 132         if (len < 0)
 133                 return -EINVAL;
 134
 135         dprintk("%s: len %u\n", __func__, len);
 136
 137         /* opaque body */
 138         p = xdr_inline_decode(xdr, len);
 139         if (unlikely(!p))
 140                 return -ENOBUFS;
 141
 142         if (!nfs_map_string_to_numeric((char *)p, len, id))
 143                 return -EINVAL;
 144
 145         return 0;
 146 }
 147
 148 static bool ff_mirror_match_fh(const struct nfs4_ff_layout_mirror *m1,
 149                 const struct nfs4_ff_layout_mirror *m2)
 150 {
 151         int i, j;
 152
 153         if (m1->fh_versions_cnt != m2->fh_versions_cnt)
 154                 return false;
 155         for (i = 0; i < m1->fh_versions_cnt; i++) {
 156                 bool found_fh = false;
 157                 for (j = 0; j < m2->fh_versions_cnt; j++) {
 158                         if (nfs_compare_fh(&m1->fh_versions[i],
 159                                         &m2->fh_versions[j]) == 0) {
 160                                 found_fh = true;
 161                                 break;
 162                         }
 163                 }
 164                 if (!found_fh)
 165                         return false;
 166         }
 167         return true;
 168 }
 169
 170 static struct nfs4_ff_layout_mirror *
 171 ff_layout_add_mirror(struct pnfs_layout_hdr *lo,
 172                 struct nfs4_ff_layout_mirror *mirror)
 173 {
 174         struct nfs4_flexfile_layout *ff_layout = FF_LAYOUT_FROM_HDR(lo);
 175         struct nfs4_ff_layout_mirror *pos;
 176         struct inode *inode = lo->plh_inode;
 177
 178         spin_lock(&inode->i_lock);
 179         list_for_each_entry(pos, &ff_layout->mirrors, mirrors) {
 180                 if (mirror->mirror_ds != pos->mirror_ds)
 181                         continue;
 182                 if (!ff_mirror_match_fh(mirror, pos))
 183                         continue;
 184                 if (atomic_inc_not_zero(&pos->ref)) {
 185                         spin_unlock(&inode->i_lock);
 186                         return pos;
 187                 }
 188         }
 189         list_add(&mirror->mirrors, &ff_layout->mirrors);
 190         mirror->layout = lo;
 191         spin_unlock(&inode->i_lock);
 192         return mirror;
 193 }
 194
 195 static void
 196 ff_layout_remove_mirror(struct nfs4_ff_layout_mirror *mirror)
 197 {
 198         struct inode *inode;
 199         if (mirror->layout == NULL)
 200                 return;
 201         inode = mirror->layout->plh_inode;
 202         spin_lock(&inode->i_lock);
 203         list_del(&mirror->mirrors);
 204         spin_unlock(&inode->i_lock);
 205         mirror->layout = NULL;
 206 }
 207
 208 static struct nfs4_ff_layout_mirror *ff_layout_alloc_mirror(gfp_t gfp_flags)
 209 {
 210         struct nfs4_ff_layout_mirror *mirror;
 211
 212         mirror = kzalloc(sizeof(*mirror), gfp_flags);
 213         if (mirror != NULL) {
 214                 spin_lock_init(&mirror->lock);
 215                 atomic_set(&mirror->ref, 1);
 216                 INIT_LIST_HEAD(&mirror->mirrors);
 217         }
 218         return mirror;
 219 }
 220
 221 static void ff_layout_free_mirror(struct nfs4_ff_layout_mirror *mirror)
 222 {
 223         struct rpc_cred *cred;
 224
 225         ff_layout_remove_mirror(mirror);
 226         kfree(mirror->fh_versions);
 227         cred = rcu_access_pointer(mirror->ro_cred);
 228         if (cred)
 229                 put_rpccred(cred);
 230         cred = rcu_access_pointer(mirror->rw_cred);
 231         if (cred)
 232                 put_rpccred(cred);
 233         nfs4_ff_layout_put_deviceid(mirror->mirror_ds);
 234         kfree(mirror);
 235 }
 236
 237 static void ff_layout_put_mirror(struct nfs4_ff_layout_mirror *mirror)
 238 {
 239         if (mirror != NULL && atomic_dec_and_test(&mirror->ref))
 240                 ff_layout_free_mirror(mirror);
 241 }
 242
 243 static void ff_layout_free_mirror_array(struct nfs4_ff_layout_segment *fls)
 244 {
 245         int i;
 246
 247         if (fls->mirror_array) {
 248                 for (i = 0; i < fls->mirror_array_cnt; i++) {
 249                         /* normally mirror_ds is freed in
 250                          * .free_deviceid_node but we still do it here
 251                          * for .alloc_lseg error path */
 252                         ff_layout_put_mirror(fls->mirror_array[i]);
 253                 }
 254                 kfree(fls->mirror_array);
 255                 fls->mirror_array = NULL;
 256         }
 257 }
 258
 259 static int ff_layout_check_layout(struct nfs4_layoutget_res *lgr)
 260 {
 261         int ret = 0;
 262
 263         dprintk("--> %s\n", __func__);
 264
 265         /* FIXME: remove this check when layout segment support is added */
 266         if (lgr->range.offset != 0 ||
 267             lgr->range.length != NFS4_MAX_UINT64) {
 268                 dprintk("%s Only whole file layouts supported. Use MDS i/o\n",
 269                         __func__);
 270                 ret = -EINVAL;
 271         }
 272
 273         dprintk("--> %s returns %d\n", __func__, ret);
 274         return ret;
 275 }
 276
 277 static void _ff_layout_free_lseg(struct nfs4_ff_layout_segment *fls)
 278 {
 279         if (fls) {
 280                 ff_layout_free_mirror_array(fls);
 281                 kfree(fls);
 282         }
 283 }
 284
 285 static bool
 286 ff_lseg_range_is_after(const struct pnfs_layout_range *l1,
 287                 const struct pnfs_layout_range *l2)
 288 {
 289         u64 end1, end2;
 290
 291         if (l1->iomode != l2->iomode)
 292                 return l1->iomode != IOMODE_READ;
 293         end1 = pnfs_calc_offset_end(l1->offset, l1->length);
 294         end2 = pnfs_calc_offset_end(l2->offset, l2->length);
 295         if (end1 < l2->offset)
 296                 return false;
 297         if (end2 < l1->offset)
 298                 return true;
 299         return l2->offset <= l1->offset;
 300 }
 301
 302 static bool
 303 ff_lseg_merge(struct pnfs_layout_segment *new,
 304                 struct pnfs_layout_segment *old)
 305 {
 306         u64 new_end, old_end;
 307
 308         if (test_bit(NFS_LSEG_LAYOUTRETURN, &old->pls_flags))
 309                 return false;
 310         if (new->pls_range.iomode != old->pls_range.iomode)
 311                 return false;
 312         old_end = pnfs_calc_offset_end(old->pls_range.offset,
 313                         old->pls_range.length);
 314         if (old_end < new->pls_range.offset)
 315                 return false;
 316         new_end = pnfs_calc_offset_end(new->pls_range.offset,
 317                         new->pls_range.length);
 318         if (new_end < old->pls_range.offset)
 319                 return false;
 320
 321         /* Mergeable: copy info from 'old' to 'new' */
 322         if (new_end < old_end)
 323                 new_end = old_end;
 324         if (new->pls_range.offset < old->pls_range.offset)
 325                 new->pls_range.offset = old->pls_range.offset;
 326         new->pls_range.length = pnfs_calc_offset_length(new->pls_range.offset,
 327                         new_end);
 328         if (test_bit(NFS_LSEG_ROC, &old->pls_flags))
 329                 set_bit(NFS_LSEG_ROC, &new->pls_flags);
 330         return true;
 331 }
 332
 333 static void
 334 ff_layout_add_lseg(struct pnfs_layout_hdr *lo,
 335                 struct pnfs_layout_segment *lseg,
 336                 struct list_head *free_me)
 337 {
 338         pnfs_generic_layout_insert_lseg(lo, lseg,
 339                         ff_lseg_range_is_after,
 340                         ff_lseg_merge,
 341                         free_me);
 342 }
 343
 344 static void ff_layout_sort_mirrors(struct nfs4_ff_layout_segment *fls)
 345 {
 346         int i, j;
 347
 348         for (i = 0; i < fls->mirror_array_cnt - 1; i++) {
 349                 for (j = i + 1; j < fls->mirror_array_cnt; j++)
 350                         if (fls->mirror_array[i]->efficiency <
 351                             fls->mirror_array[j]->efficiency)
 352                                 swap(fls->mirror_array[i],
 353                                      fls->mirror_array[j]);
 354         }
 355 }
 356
 357 static void ff_layout_mark_devices_valid(struct nfs4_ff_layout_segment *fls)
 358 {
 359         struct nfs4_deviceid_node *node;
 360         int i;
 361
 362         if (!(fls->flags & FF_FLAGS_NO_IO_THRU_MDS))
 363                 return;
 364         for (i = 0; i < fls->mirror_array_cnt; i++) {
 365                 node = &fls->mirror_array[i]->mirror_ds->id_node;
 366                 clear_bit(NFS_DEVICEID_UNAVAILABLE, &node->flags);
 367         }
 368 }
 369
 370 static struct pnfs_layout_segment *
 371 ff_layout_alloc_lseg(struct pnfs_layout_hdr *lh,
 372                      struct nfs4_layoutget_res *lgr,
 373                      gfp_t gfp_flags)
 374 {
 375         struct pnfs_layout_segment *ret;
 376         struct nfs4_ff_layout_segment *fls = NULL;
 377         struct xdr_stream stream;
 378         struct xdr_buf buf;
 379         struct page *scratch;
 380         u64 stripe_unit;
 381         u32 mirror_array_cnt;
 382         __be32 *p;
 383         int i, rc;
 384
 385         dprintk("--> %s\n", __func__);
 386         scratch = alloc_page(gfp_flags);
 387         if (!scratch)
 388                 return ERR_PTR(-ENOMEM);
 389
 390         xdr_init_decode_pages(&stream, &buf, lgr->layoutp->pages,
 391                               lgr->layoutp->len);
 392         xdr_set_scratch_buffer(&stream, page_address(scratch), PAGE_SIZE);
 393
 394         /* stripe unit and mirror_array_cnt */
 395         rc = -EIO;
 396         p = xdr_inline_decode(&stream, 8 + 4);
 397         if (!p)
 398                 goto out_err_free;
 399
 400         p = xdr_decode_hyper(p, &stripe_unit);
 401         mirror_array_cnt = be32_to_cpup(p++);
 402         dprintk("%s: stripe_unit=%llu mirror_array_cnt=%u\n", __func__,
 403                 stripe_unit, mirror_array_cnt);
 404
 405         if (mirror_array_cnt > NFS4_FLEXFILE_LAYOUT_MAX_MIRROR_CNT ||
 406             mirror_array_cnt == 0)
 407                 goto out_err_free;
 408
 409         rc = -ENOMEM;
 410         fls = kzalloc(sizeof(*fls), gfp_flags);
 411         if (!fls)
 412                 goto out_err_free;
 413
 414         fls->mirror_array_cnt = mirror_array_cnt;
 415         fls->stripe_unit = stripe_unit;
 416         fls->mirror_array = kcalloc(fls->mirror_array_cnt,
 417                                     sizeof(fls->mirror_array[0]), gfp_flags);
 418         if (fls->mirror_array == NULL)
 419                 goto out_err_free;
 420
 421         for (i = 0; i < fls->mirror_array_cnt; i++) {
 422                 struct nfs4_ff_layout_mirror *mirror;
 423                 struct nfs4_deviceid devid;
 424                 struct nfs4_deviceid_node *idnode;
 425                 struct auth_cred acred = { .group_info = ff_zero_group };
 426                 struct rpc_cred __rcu *cred;
 427                 u32 ds_count, fh_count, id;
 428                 int j;
 429
 430                 rc = -EIO;
 431                 p = xdr_inline_decode(&stream, 4);
 432                 if (!p)
 433                         goto out_err_free;
 434                 ds_count = be32_to_cpup(p);
 435
 436                 /* FIXME: allow for striping? */
 437                 if (ds_count != 1)
 438                         goto out_err_free;
 439
 440                 fls->mirror_array[i] = ff_layout_alloc_mirror(gfp_flags);
 441                 if (fls->mirror_array[i] == NULL) {
 442                         rc = -ENOMEM;
 443                         goto out_err_free;
 444                 }
 445
 446                 fls->mirror_array[i]->ds_count = ds_count;
 447
 448                 /* deviceid */
 449                 rc = decode_deviceid(&stream, &devid);
 450                 if (rc)
 451                         goto out_err_free;
 452
 453                 idnode = nfs4_find_get_deviceid(NFS_SERVER(lh->plh_inode),
 454                                                 &devid, lh->plh_lc_cred,
 455                                                 gfp_flags);
 456                 /*
 457                  * upon success, mirror_ds is allocated by previous
 458                  * getdeviceinfo, or newly by .alloc_deviceid_node
 459                  * nfs4_find_get_deviceid failure is indeed getdeviceinfo falure
 460                  */
 461                 if (idnode)
 462                         fls->mirror_array[i]->mirror_ds =
 463                                 FF_LAYOUT_MIRROR_DS(idnode);
 464                 else
 465                         goto out_err_free;
 466
 467                 /* efficiency */
 468                 rc = -EIO;
 469                 p = xdr_inline_decode(&stream, 4);
 470                 if (!p)
 471                         goto out_err_free;
 472                 fls->mirror_array[i]->efficiency = be32_to_cpup(p);
 473
 474                 /* stateid */
 475                 rc = decode_pnfs_stateid(&stream, &fls->mirror_array[i]->stateid);
 476                 if (rc)
 477                         goto out_err_free;
 478
 479                 /* fh */
 480                 p = xdr_inline_decode(&stream, 4);
 481                 if (!p)
 482                         goto out_err_free;
 483                 fh_count = be32_to_cpup(p);
 484
 485                 fls->mirror_array[i]->fh_versions =
 486                         kzalloc(fh_count * sizeof(struct nfs_fh),
 487                                 gfp_flags);
 488                 if (fls->mirror_array[i]->fh_versions == NULL) {
 489                         rc = -ENOMEM;
 490                         goto out_err_free;
 491                 }
 492
 493                 for (j = 0; j < fh_count; j++) {
 494                         rc = decode_nfs_fh(&stream,
 495                                            &fls->mirror_array[i]->fh_versions[j]);
 496                         if (rc)
 497                                 goto out_err_free;
 498                 }
 499
 500                 fls->mirror_array[i]->fh_versions_cnt = fh_count;
 501
 502                 /* user */
 503                 rc = decode_name(&stream, &id);
 504                 if (rc)
 505                         goto out_err_free;
 506
 507                 acred.uid = make_kuid(&init_user_ns, id);
 508
 509                 /* group */
 510                 rc = decode_name(&stream, &id);
 511                 if (rc)
 512                         goto out_err_free;
 513
 514                 acred.gid = make_kgid(&init_user_ns, id);
 515
 516                 /* find the cred for it */
 517                 rcu_assign_pointer(cred, rpc_lookup_generic_cred(&acred, 0, gfp_flags));
 518                 if (IS_ERR(cred)) {
 519                         rc = PTR_ERR(cred);
 520                         goto out_err_free;
 521                 }
 522
 523                 if (lgr->range.iomode == IOMODE_READ)
 524                         rcu_assign_pointer(fls->mirror_array[i]->ro_cred, cred);
 525                 else
 526                         rcu_assign_pointer(fls->mirror_array[i]->rw_cred, cred);
 527
 528                 mirror = ff_layout_add_mirror(lh, fls->mirror_array[i]);
 529                 if (mirror != fls->mirror_array[i]) {
 530                         /* swap cred ptrs so free_mirror will clean up old */
 531                         if (lgr->range.iomode == IOMODE_READ) {
 532                                 cred = xchg(&mirror->ro_cred, cred);
 533                                 rcu_assign_pointer(fls->mirror_array[i]->ro_cred, cred);
 534                         } else {
 535                                 cred = xchg(&mirror->rw_cred, cred);
 536                                 rcu_assign_pointer(fls->mirror_array[i]->rw_cred, cred);
 537                         }
 538                         ff_layout_free_mirror(fls->mirror_array[i]);
 539                         fls->mirror_array[i] = mirror;
 540                 }
 541
 542                 dprintk("%s: iomode %s uid %u gid %u\n", __func__,
 543                         lgr->range.iomode == IOMODE_READ ? "READ" : "RW",
 544                         from_kuid(&init_user_ns, acred.uid),
 545                         from_kgid(&init_user_ns, acred.gid));
 546         }
 547
 548         p = xdr_inline_decode(&stream, 4);
 549         if (!p)
 550                 goto out_sort_mirrors;
 551         fls->flags = be32_to_cpup(p);
 552
 553         p = xdr_inline_decode(&stream, 4);
 554         if (!p)
 555                 goto out_sort_mirrors;
 556         for (i=0; i < fls->mirror_array_cnt; i++)
 557                 fls->mirror_array[i]->report_interval = be32_to_cpup(p);
 558
 559 out_sort_mirrors:
 560         ff_layout_sort_mirrors(fls);
 561         rc = ff_layout_check_layout(lgr);
 562         if (rc)
 563                 goto out_err_free;
 564         ff_layout_mark_devices_valid(fls);
 565
 566         ret = &fls->generic_hdr;
 567         dprintk("<-- %s (success)\n", __func__);
 568 out_free_page:
 569         __free_page(scratch);
 570         return ret;
 571 out_err_free:
 572         _ff_layout_free_lseg(fls);
 573         ret = ERR_PTR(rc);
 574         dprintk("<-- %s (%d)\n", __func__, rc);
 575         goto out_free_page;
 576 }
 577
 578 static bool ff_layout_has_rw_segments(struct pnfs_layout_hdr *layout)
 579 {
 580         struct pnfs_layout_segment *lseg;
 581
 582         list_for_each_entry(lseg, &layout->plh_segs, pls_list)
 583                 if (lseg->pls_range.iomode == IOMODE_RW)
 584                         return true;
 585
 586         return false;
 587 }
 588
 589 static void
 590 ff_layout_free_lseg(struct pnfs_layout_segment *lseg)
 591 {
 592         struct nfs4_ff_layout_segment *fls = FF_LAYOUT_LSEG(lseg);
 593
 594         dprintk("--> %s\n", __func__);
 595
 596         if (lseg->pls_range.iomode == IOMODE_RW) {
 597                 struct nfs4_flexfile_layout *ffl;
 598                 struct inode *inode;
 599
 600                 ffl = FF_LAYOUT_FROM_HDR(lseg->pls_layout);
 601                 inode = ffl->generic_hdr.plh_inode;
 602                 spin_lock(&inode->i_lock);
 603                 if (!ff_layout_has_rw_segments(lseg->pls_layout)) {
 604                         ffl->commit_info.nbuckets = 0;
 605                         kfree(ffl->commit_info.buckets);
 606                         ffl->commit_info.buckets = NULL;
 607                 }
 608                 spin_unlock(&inode->i_lock);
 609         }
 610         _ff_layout_free_lseg(fls);
 611 }
 612
 613 /* Return 1 until we have multiple lsegs support */
 614 static int
 615 ff_layout_get_lseg_count(struct nfs4_ff_layout_segment *fls)
 616 {
 617         return 1;
 618 }
 619
 620 static void
 621 nfs4_ff_start_busy_timer(struct nfs4_ff_busy_timer *timer, ktime_t now)
 622 {
 623         /* first IO request? */
 624         if (atomic_inc_return(&timer->n_ops) == 1) {
 625                 timer->start_time = now;
 626         }
 627 }
 628
 629 static ktime_t
 630 nfs4_ff_end_busy_timer(struct nfs4_ff_busy_timer *timer, ktime_t now)
 631 {
 632         ktime_t start;
 633
 634         if (atomic_dec_return(&timer->n_ops) < 0)
 635                 WARN_ON_ONCE(1);
 636
 637         start = timer->start_time;
 638         timer->start_time = now;
 639         return ktime_sub(now, start);
 640 }
 641
 642 static bool
 643 nfs4_ff_layoutstat_start_io(struct nfs4_ff_layout_mirror *mirror,
 644                             struct nfs4_ff_layoutstat *layoutstat,
 645                             ktime_t now)
 646 {
 647         static const ktime_t notime = {0};
 648         s64 report_interval = FF_LAYOUTSTATS_REPORT_INTERVAL;
 649         struct nfs4_flexfile_layout *ffl = FF_LAYOUT_FROM_HDR(mirror->layout);
 650
 651         nfs4_ff_start_busy_timer(&layoutstat->busy_timer, now);
 652         if (ktime_equal(mirror->start_time, notime))
 653                 mirror->start_time = now;
 654         if (mirror->report_interval != 0)
 655                 report_interval = (s64)mirror->report_interval * 1000LL;
 656         else if (layoutstats_timer != 0)
 657                 report_interval = (s64)layoutstats_timer * 1000LL;
 658         if (ktime_to_ms(ktime_sub(now, ffl->last_report_time)) >=
 659                         report_interval) {
 660                 ffl->last_report_time = now;
 661                 return true;
 662         }
 663
 664         return false;
 665 }
 666
 667 static void
 668 nfs4_ff_layout_stat_io_update_requested(struct nfs4_ff_layoutstat *layoutstat,
 669                 __u64 requested)
 670 {
 671         struct nfs4_ff_io_stat *iostat = &layoutstat->io_stat;
 672
 673         iostat->ops_requested++;
 674         iostat->bytes_requested += requested;
 675 }
 676
 677 static void
 678 nfs4_ff_layout_stat_io_update_completed(struct nfs4_ff_layoutstat *layoutstat,
 679                 __u64 requested,
 680                 __u64 completed,
 681                 ktime_t time_completed,
 682                 ktime_t time_started)
 683 {
 684         struct nfs4_ff_io_stat *iostat = &layoutstat->io_stat;
 685         ktime_t completion_time = ktime_sub(time_completed, time_started);
 686         ktime_t timer;
 687
 688         iostat->ops_completed++;
 689         iostat->bytes_completed += completed;
 690         iostat->bytes_not_delivered += requested - completed;
 691
 692         timer = nfs4_ff_end_busy_timer(&layoutstat->busy_timer, time_completed);
 693         iostat->total_busy_time =
 694                         ktime_add(iostat->total_busy_time, timer);
 695         iostat->aggregate_completion_time =
 696                         ktime_add(iostat->aggregate_completion_time,
 697                                         completion_time);
 698 }
 699
 700 static void
 701 nfs4_ff_layout_stat_io_start_read(struct inode *inode,
 702                 struct nfs4_ff_layout_mirror *mirror,
 703                 __u64 requested, ktime_t now)
 704 {
 705         bool report;
 706
 707         spin_lock(&mirror->lock);
 708         report = nfs4_ff_layoutstat_start_io(mirror, &mirror->read_stat, now);
 709         nfs4_ff_layout_stat_io_update_requested(&mirror->read_stat, requested);
 710         set_bit(NFS4_FF_MIRROR_STAT_AVAIL, &mirror->flags);
 711         spin_unlock(&mirror->lock);
 712
 713         if (report)
 714                 pnfs_report_layoutstat(inode, GFP_KERNEL);
 715 }
 716
 717 static void
 718 nfs4_ff_layout_stat_io_end_read(struct rpc_task *task,
 719                 struct nfs4_ff_layout_mirror *mirror,
 720                 __u64 requested,
 721                 __u64 completed)
 722 {
 723         spin_lock(&mirror->lock);
 724         nfs4_ff_layout_stat_io_update_completed(&mirror->read_stat,
 725                         requested, completed,
 726                         ktime_get(), task->tk_start);
 727         set_bit(NFS4_FF_MIRROR_STAT_AVAIL, &mirror->flags);
 728         spin_unlock(&mirror->lock);
 729 }
 730
 731 static void
 732 nfs4_ff_layout_stat_io_start_write(struct inode *inode,
 733                 struct nfs4_ff_layout_mirror *mirror,
 734                 __u64 requested, ktime_t now)
 735 {
 736         bool report;
 737
 738         spin_lock(&mirror->lock);
 739         report = nfs4_ff_layoutstat_start_io(mirror , &mirror->write_stat, now);
 740         nfs4_ff_layout_stat_io_update_requested(&mirror->write_stat, requested);
 741         set_bit(NFS4_FF_MIRROR_STAT_AVAIL, &mirror->flags);
 742         spin_unlock(&mirror->lock);
 743
 744         if (report)
 745                 pnfs_report_layoutstat(inode, GFP_NOIO);
 746 }
 747
 748 static void
 749 nfs4_ff_layout_stat_io_end_write(struct rpc_task *task,
 750                 struct nfs4_ff_layout_mirror *mirror,
 751                 __u64 requested,
 752                 __u64 completed,
 753                 enum nfs3_stable_how committed)
 754 {
 755         if (committed == NFS_UNSTABLE)
 756                 requested = completed = 0;
 757
 758         spin_lock(&mirror->lock);
 759         nfs4_ff_layout_stat_io_update_completed(&mirror->write_stat,
 760                         requested, completed, ktime_get(), task->tk_start);
 761         set_bit(NFS4_FF_MIRROR_STAT_AVAIL, &mirror->flags);
 762         spin_unlock(&mirror->lock);
 763 }
 764
 765 static int
 766 ff_layout_alloc_commit_info(struct pnfs_layout_segment *lseg,
 767                             struct nfs_commit_info *cinfo,
 768                             gfp_t gfp_flags)
 769 {
 770         struct nfs4_ff_layout_segment *fls = FF_LAYOUT_LSEG(lseg);
 771         struct pnfs_commit_bucket *buckets;
 772         int size;
 773
 774         if (cinfo->ds->nbuckets != 0) {
 775                 /* This assumes there is only one RW lseg per file.
 776                  * To support multiple lseg per file, we need to
 777                  * change struct pnfs_commit_bucket to allow dynamic
 778                  * increasing nbuckets.
 779                  */
 780                 return 0;
 781         }
 782
 783         size = ff_layout_get_lseg_count(fls) * FF_LAYOUT_MIRROR_COUNT(lseg);
 784
 785         buckets = kcalloc(size, sizeof(struct pnfs_commit_bucket),
 786                           gfp_flags);
 787         if (!buckets)
 788                 return -ENOMEM;
 789         else {
 790                 int i;
 791
 792                 spin_lock(&cinfo->inode->i_lock);
 793                 if (cinfo->ds->nbuckets != 0)
 794                         kfree(buckets);
 795                 else {
 796                         cinfo->ds->buckets = buckets;
 797                         cinfo->ds->nbuckets = size;
 798                         for (i = 0; i < size; i++) {
 799                                 INIT_LIST_HEAD(&buckets[i].written);
 800                                 INIT_LIST_HEAD(&buckets[i].committing);
 801                                 /* mark direct verifier as unset */
 802                                 buckets[i].direct_verf.committed =
 803                                         NFS_INVALID_STABLE_HOW;
 804                         }
 805                 }
 806                 spin_unlock(&cinfo->inode->i_lock);
 807                 return 0;
 808         }
 809 }
 810
 811 static struct nfs4_pnfs_ds *
 812 ff_layout_choose_best_ds_for_read(struct pnfs_layout_segment *lseg,
 813                                   int start_idx,
 814                                   int *best_idx)
 815 {
 816         struct nfs4_ff_layout_segment *fls = FF_LAYOUT_LSEG(lseg);
 817         struct nfs4_pnfs_ds *ds;
 818         bool fail_return = false;
 819         int idx;
 820
 821         /* mirrors are sorted by efficiency */
 822         for (idx = start_idx; idx < fls->mirror_array_cnt; idx++) {
 823                 if (idx+1 == fls->mirror_array_cnt)
 824                         fail_return = true;
 825                 ds = nfs4_ff_layout_prepare_ds(lseg, idx, fail_return);
 826                 if (ds) {
 827                         *best_idx = idx;
 828                         return ds;
 829                 }
 830         }
 831
 832         return NULL;
 833 }
 834
 835 static void
 836 ff_layout_pg_get_read(struct nfs_pageio_descriptor *pgio,
 837                       struct nfs_page *req,
 838                       bool strict_iomode)
 839 {
 840 retry_strict:
 841         pnfs_put_lseg(pgio->pg_lseg);
 842         pgio->pg_lseg = pnfs_update_layout(pgio->pg_inode,
 843                                            req->wb_context,
 844                                            0,
 845                                            NFS4_MAX_UINT64,
 846                                            IOMODE_READ,
 847                                            strict_iomode,
 848                                            GFP_KERNEL);
 849         if (IS_ERR(pgio->pg_lseg)) {
 850                 pgio->pg_error = PTR_ERR(pgio->pg_lseg);
 851                 pgio->pg_lseg = NULL;
 852         }
 853
 854         /* If we don't have checking, do get a IOMODE_RW
 855          * segment, and the server wants to avoid READs
 856          * there, then retry!
 857          */
 858         if (pgio->pg_lseg && !strict_iomode &&
 859             ff_layout_avoid_read_on_rw(pgio->pg_lseg)) {
 860                 strict_iomode = true;
 861                 goto retry_strict;
 862         }
 863 }
 864
 865 static void
 866 ff_layout_pg_init_read(struct nfs_pageio_descriptor *pgio,
 867                         struct nfs_page *req)
 868 {
 869         struct nfs_pgio_mirror *pgm;
 870         struct nfs4_ff_layout_mirror *mirror;
 871         struct nfs4_pnfs_ds *ds;
 872         int ds_idx;
 873
 874 retry:
 875         /* Use full layout for now */
 876         if (!pgio->pg_lseg)
 877                 ff_layout_pg_get_read(pgio, req, false);
 878         else if (ff_layout_avoid_read_on_rw(pgio->pg_lseg))
 879                 ff_layout_pg_get_read(pgio, req, true);
 880
 881         /* If no lseg, fall back to read through mds */
 882         if (pgio->pg_lseg == NULL)
 883                 goto out_mds;
 884
 885         ds = ff_layout_choose_best_ds_for_read(pgio->pg_lseg, 0, &ds_idx);
 886         if (!ds) {
 887                 if (!ff_layout_no_fallback_to_mds(pgio->pg_lseg))
 888                         goto out_mds;
 889                 pnfs_put_lseg(pgio->pg_lseg);
 890                 pgio->pg_lseg = NULL;
 891                 /* Sleep for 1 second before retrying */
 892                 ssleep(1);
 893                 goto retry;
 894         }
 895
 896         mirror = FF_LAYOUT_COMP(pgio->pg_lseg, ds_idx);
 897
 898         pgio->pg_mirror_idx = ds_idx;
 899
 900         /* read always uses only one mirror - idx 0 for pgio layer */
 901         pgm = &pgio->pg_mirrors[0];
 902         pgm->pg_bsize = mirror->mirror_ds->ds_versions[0].rsize;
 903
 904         return;
 905 out_mds:
 906         pnfs_put_lseg(pgio->pg_lseg);
 907         pgio->pg_lseg = NULL;
 908         nfs_pageio_reset_read_mds(pgio);
 909 }
 910
 911 static void
 912 ff_layout_pg_init_write(struct nfs_pageio_descriptor *pgio,
 913                         struct nfs_page *req)
 914 {
 915         struct nfs4_ff_layout_mirror *mirror;
 916         struct nfs_pgio_mirror *pgm;
 917         struct nfs_commit_info cinfo;
 918         struct nfs4_pnfs_ds *ds;
 919         int i;
 920         int status;
 921
 922 retry:
 923         if (!pgio->pg_lseg) {
 924                 pgio->pg_lseg = pnfs_update_layout(pgio->pg_inode,
 925                                                    req->wb_context,
 926                                                    0,
 927                                                    NFS4_MAX_UINT64,
 928                                                    IOMODE_RW,
 929                                                    false,
 930                                                    GFP_NOFS);
 931                 if (IS_ERR(pgio->pg_lseg)) {
 932                         pgio->pg_error = PTR_ERR(pgio->pg_lseg);
 933                         pgio->pg_lseg = NULL;
 934                         return;
 935                 }
 936         }
 937         /* If no lseg, fall back to write through mds */
 938         if (pgio->pg_lseg == NULL)
 939                 goto out_mds;
 940
 941         nfs_init_cinfo(&cinfo, pgio->pg_inode, pgio->pg_dreq);
 942         status = ff_layout_alloc_commit_info(pgio->pg_lseg, &cinfo, GFP_NOFS);
 943         if (status < 0)
 944                 goto out_mds;
 945
 946         /* Use a direct mapping of ds_idx to pgio mirror_idx */
 947         if (WARN_ON_ONCE(pgio->pg_mirror_count !=
 948             FF_LAYOUT_MIRROR_COUNT(pgio->pg_lseg)))
 949                 goto out_mds;
 950
 951         for (i = 0; i < pgio->pg_mirror_count; i++) {
 952                 ds = nfs4_ff_layout_prepare_ds(pgio->pg_lseg, i, true);
 953                 if (!ds) {
 954                         if (!ff_layout_no_fallback_to_mds(pgio->pg_lseg))
 955                                 goto out_mds;
 956                         pnfs_put_lseg(pgio->pg_lseg);
 957                         pgio->pg_lseg = NULL;
 958                         /* Sleep for 1 second before retrying */
 959                         ssleep(1);
 960                         goto retry;
 961                 }
 962                 pgm = &pgio->pg_mirrors[i];
 963                 mirror = FF_LAYOUT_COMP(pgio->pg_lseg, i);
 964                 pgm->pg_bsize = mirror->mirror_ds->ds_versions[0].wsize;
 965         }
 966
 967         return;
 968
 969 out_mds:
 970         pnfs_put_lseg(pgio->pg_lseg);
 971         pgio->pg_lseg = NULL;
 972         nfs_pageio_reset_write_mds(pgio);
 973 }
 974
 975 static unsigned int
 976 ff_layout_pg_get_mirror_count_write(struct nfs_pageio_descriptor *pgio,
 977                                     struct nfs_page *req)
 978 {
 979         if (!pgio->pg_lseg) {
 980                 pgio->pg_lseg = pnfs_update_layout(pgio->pg_inode,
 981                                                    req->wb_context,
 982                                                    0,
 983                                                    NFS4_MAX_UINT64,
 984                                                    IOMODE_RW,
 985                                                    false,
 986                                                    GFP_NOFS);
 987                 if (IS_ERR(pgio->pg_lseg)) {
 988                         pgio->pg_error = PTR_ERR(pgio->pg_lseg);
 989                         pgio->pg_lseg = NULL;
 990                         goto out;
 991                 }
 992         }
 993         if (pgio->pg_lseg)
 994                 return FF_LAYOUT_MIRROR_COUNT(pgio->pg_lseg);
 995
 996         /* no lseg means that pnfs is not in use, so no mirroring here */
 997         nfs_pageio_reset_write_mds(pgio);
 998 out:
 999         return 1;
1000 }
1001
1002 static const struct nfs_pageio_ops ff_layout_pg_read_ops = {
1003         .pg_init = ff_layout_pg_init_read,
1004         .pg_test = pnfs_generic_pg_test,
1005         .pg_doio = pnfs_generic_pg_readpages,
1006         .pg_cleanup = pnfs_generic_pg_cleanup,
1007 };
1008
1009 static const struct nfs_pageio_ops ff_layout_pg_write_ops = {
1010         .pg_init = ff_layout_pg_init_write,
1011         .pg_test = pnfs_generic_pg_test,
1012         .pg_doio = pnfs_generic_pg_writepages,
1013         .pg_get_mirror_count = ff_layout_pg_get_mirror_count_write,
1014         .pg_cleanup = pnfs_generic_pg_cleanup,
1015 };
1016
1017 static void ff_layout_reset_write(struct nfs_pgio_header *hdr, bool retry_pnfs)
1018 {
1019         struct rpc_task *task = &hdr->task;
1020
1021         pnfs_layoutcommit_inode(hdr->inode, false);
1022
1023         if (retry_pnfs) {
1024                 dprintk("%s Reset task %5u for i/o through pNFS "
1025                         "(req %s/%llu, %u bytes @ offset %llu)\n", __func__,
1026                         hdr->task.tk_pid,
1027                         hdr->inode->i_sb->s_id,
1028                         (unsigned long long)NFS_FILEID(hdr->inode),
1029                         hdr->args.count,
1030                         (unsigned long long)hdr->args.offset);
1031
1032                 hdr->completion_ops->reschedule_io(hdr);
1033                 return;
1034         }
1035
1036         if (!test_and_set_bit(NFS_IOHDR_REDO, &hdr->flags)) {
1037                 dprintk("%s Reset task %5u for i/o through MDS "
1038                         "(req %s/%llu, %u bytes @ offset %llu)\n", __func__,
1039                         hdr->task.tk_pid,
1040                         hdr->inode->i_sb->s_id,
1041                         (unsigned long long)NFS_FILEID(hdr->inode),
1042                         hdr->args.count,
1043                         (unsigned long long)hdr->args.offset);
1044
1045                 task->tk_status = pnfs_write_done_resend_to_mds(hdr);
1046         }
1047 }
1048
1049 static void ff_layout_reset_read(struct nfs_pgio_header *hdr)
1050 {
1051         struct rpc_task *task = &hdr->task;
1052
1053         pnfs_layoutcommit_inode(hdr->inode, false);
1054
1055         if (!test_and_set_bit(NFS_IOHDR_REDO, &hdr->flags)) {
1056                 dprintk("%s Reset task %5u for i/o through MDS "
1057                         "(req %s/%llu, %u bytes @ offset %llu)\n", __func__,
1058                         hdr->task.tk_pid,
1059                         hdr->inode->i_sb->s_id,
1060                         (unsigned long long)NFS_FILEID(hdr->inode),
1061                         hdr->args.count,
1062                         (unsigned long long)hdr->args.offset);
1063
1064                 task->tk_status = pnfs_read_done_resend_to_mds(hdr);
1065         }
1066 }
1067
1068 static int ff_layout_async_handle_error_v4(struct rpc_task *task,
1069                                            struct nfs4_state *state,
1070                                            struct nfs_client *clp,
1071                                            struct pnfs_layout_segment *lseg,
1072                                            int idx)
1073 {
1074         struct pnfs_layout_hdr *lo = lseg->pls_layout;
1075         struct inode *inode = lo->plh_inode;
1076         struct nfs_server *mds_server = NFS_SERVER(inode);
1077
1078         struct nfs4_deviceid_node *devid = FF_LAYOUT_DEVID_NODE(lseg, idx);
1079         struct nfs_client *mds_client = mds_server->nfs_client;
1080         struct nfs4_slot_table *tbl = &clp->cl_session->fc_slot_table;
1081
1082         if (task->tk_status >= 0)
1083                 return 0;
1084
1085         switch (task->tk_status) {
1086         /* MDS state errors */
1087         case -NFS4ERR_DELEG_REVOKED:
1088         case -NFS4ERR_ADMIN_REVOKED:
1089         case -NFS4ERR_BAD_STATEID:
1090                 if (state == NULL)
1091                         break;
1092                 nfs_remove_bad_delegation(state->inode, NULL);
1093         case -NFS4ERR_OPENMODE:
1094                 if (state == NULL)
1095                         break;
1096                 if (nfs4_schedule_stateid_recovery(mds_server, state) < 0)
1097                         goto out_bad_stateid;
1098                 goto wait_on_recovery;
1099         case -NFS4ERR_EXPIRED:
1100                 if (state != NULL) {
1101                         if (nfs4_schedule_stateid_recovery(mds_server, state) < 0)
1102                                 goto out_bad_stateid;
1103                 }
1104                 nfs4_schedule_lease_recovery(mds_client);
1105                 goto wait_on_recovery;
1106         /* DS session errors */
1107         case -NFS4ERR_BADSESSION:
1108         case -NFS4ERR_BADSLOT:
1109         case -NFS4ERR_BAD_HIGH_SLOT:
1110         case -NFS4ERR_DEADSESSION:
1111         case -NFS4ERR_CONN_NOT_BOUND_TO_SESSION:
1112         case -NFS4ERR_SEQ_FALSE_RETRY:
1113         case -NFS4ERR_SEQ_MISORDERED:
1114                 dprintk("%s ERROR %d, Reset session. Exchangeid "
1115                         "flags 0x%x\n", __func__, task->tk_status,
1116                         clp->cl_exchange_flags);
1117                 nfs4_schedule_session_recovery(clp->cl_session, task->tk_status);
1118                 break;
1119         case -NFS4ERR_DELAY:
1120         case -NFS4ERR_GRACE:
1121                 rpc_delay(task, FF_LAYOUT_POLL_RETRY_MAX);
1122                 break;
1123         case -NFS4ERR_RETRY_UNCACHED_REP:
1124                 break;
1125         /* Invalidate Layout errors */
1126         case -NFS4ERR_PNFS_NO_LAYOUT:
1127         case -ESTALE:           /* mapped NFS4ERR_STALE */
1128         case -EBADHANDLE:       /* mapped NFS4ERR_BADHANDLE */
1129         case -EISDIR:           /* mapped NFS4ERR_ISDIR */
1130         case -NFS4ERR_FHEXPIRED:
1131         case -NFS4ERR_WRONG_TYPE:
1132                 dprintk("%s Invalid layout error %d\n", __func__,
1133                         task->tk_status);
1134                 /*
1135                  * Destroy layout so new i/o will get a new layout.
1136                  * Layout will not be destroyed until all current lseg
1137                  * references are put. Mark layout as invalid to resend failed
1138                  * i/o and all i/o waiting on the slot table to the MDS until
1139                  * layout is destroyed and a new valid layout is obtained.
1140                  */
1141                 pnfs_destroy_layout(NFS_I(inode));
1142                 rpc_wake_up(&tbl->slot_tbl_waitq);
1143                 goto reset;
1144         /* RPC connection errors */
1145         case -ECONNREFUSED:
1146         case -EHOSTDOWN:
1147         case -EHOSTUNREACH:
1148         case -ENETUNREACH:
1149         case -EIO:
1150         case -ETIMEDOUT:
1151         case -EPIPE:
1152                 dprintk("%s DS connection error %d\n", __func__,
1153                         task->tk_status);
1154                 nfs4_mark_deviceid_unavailable(devid);
1155                 rpc_wake_up(&tbl->slot_tbl_waitq);
1156                 /* fall through */
1157         default:
1158                 if (ff_layout_avoid_mds_available_ds(lseg))
1159                         return -NFS4ERR_RESET_TO_PNFS;
1160 reset:
1161                 dprintk("%s Retry through MDS. Error %d\n", __func__,
1162                         task->tk_status);
1163                 return -NFS4ERR_RESET_TO_MDS;
1164         }
1165 out:
1166         task->tk_status = 0;
1167         return -EAGAIN;
1168 out_bad_stateid:
1169         task->tk_status = -EIO;
1170         return 0;
1171 wait_on_recovery:
1172         rpc_sleep_on(&mds_client->cl_rpcwaitq, task, NULL);
1173         if (test_bit(NFS4CLNT_MANAGER_RUNNING, &mds_client->cl_state) == 0)
1174                 rpc_wake_up_queued_task(&mds_client->cl_rpcwaitq, task);
1175         goto out;
1176 }
1177
1178 /* Retry all errors through either pNFS or MDS except for -EJUKEBOX */
1179 static int ff_layout_async_handle_error_v3(struct rpc_task *task,
1180                                            struct pnfs_layout_segment *lseg,
1181                                            int idx)
1182 {
1183         struct nfs4_deviceid_node *devid = FF_LAYOUT_DEVID_NODE(lseg, idx);
1184
1185         if (task->tk_status >= 0)
1186                 return 0;
1187
1188         switch (task->tk_status) {
1189         /* File access problems. Don't mark the device as unavailable */
1190         case -EACCES:
1191         case -ESTALE:
1192         case -EISDIR:
1193         case -EBADHANDLE:
1194         case -ELOOP:
1195         case -ENOSPC:
1196                 break;
1197         case -EJUKEBOX:
1198                 nfs_inc_stats(lseg->pls_layout->plh_inode, NFSIOS_DELAY);
1199                 goto out_retry;
1200         default:
1201                 dprintk("%s DS connection error %d\n", __func__,
1202                         task->tk_status);
1203                 nfs4_mark_deviceid_unavailable(devid);
1204         }
1205         /* FIXME: Need to prevent infinite looping here. */
1206         return -NFS4ERR_RESET_TO_PNFS;
1207 out_retry:
1208         task->tk_status = 0;
1209         rpc_restart_call_prepare(task);
1210         rpc_delay(task, NFS_JUKEBOX_RETRY_TIME);
1211         return -EAGAIN;
1212 }
1213
1214 static int ff_layout_async_handle_error(struct rpc_task *task,
1215                                         struct nfs4_state *state,
1216                                         struct nfs_client *clp,
1217                                         struct pnfs_layout_segment *lseg,
1218                                         int idx)
1219 {
1220         int vers = clp->cl_nfs_mod->rpc_vers->number;
1221
1222         switch (vers) {
1223         case 3:
1224                 return ff_layout_async_handle_error_v3(task, lseg, idx);
1225         case 4:
1226                 return ff_layout_async_handle_error_v4(task, state, clp,
1227                                                        lseg, idx);
1228         default:
1229                 /* should never happen */
1230                 WARN_ON_ONCE(1);
1231                 return 0;
1232         }
1233 }
1234
1235 static void ff_layout_io_track_ds_error(struct pnfs_layout_segment *lseg,
1236                                         int idx, u64 offset, u64 length,
1237                                         u32 status, int opnum, int error)
1238 {
1239         struct nfs4_ff_layout_mirror *mirror;
1240         int err;
1241
1242         if (status == 0) {
1243                 switch (error) {
1244                 case -ETIMEDOUT:
1245                 case -EPFNOSUPPORT:
1246                 case -EPROTONOSUPPORT:
1247                 case -EOPNOTSUPP:
1248                 case -ECONNREFUSED:
1249                 case -ECONNRESET:
1250                 case -EHOSTDOWN:
1251                 case -EHOSTUNREACH:
1252                 case -ENETUNREACH:
1253                 case -EADDRINUSE:
1254                 case -ENOBUFS:
1255                 case -EPIPE:
1256                 case -EPERM:
1257                         status = NFS4ERR_NXIO;
1258                         break;
1259                 case -EACCES:
1260                         status = NFS4ERR_ACCESS;
1261                         break;
1262                 default:
1263                         return;
1264                 }
1265         }
1266
1267         switch (status) {
1268         case NFS4ERR_DELAY:
1269         case NFS4ERR_GRACE:
1270                 return;
1271         default:
1272                 break;
1273         }
1274
1275         mirror = FF_LAYOUT_COMP(lseg, idx);
1276         err = ff_layout_track_ds_error(FF_LAYOUT_FROM_HDR(lseg->pls_layout),
1277                                        mirror, offset, length, status, opnum,
1278                                        GFP_NOIO);
1279         pnfs_error_mark_layout_for_return(lseg->pls_layout->plh_inode, lseg);
1280         dprintk("%s: err %d op %d status %u\n", __func__, err, opnum, status);
1281 }
1282
1283 /* NFS_PROTO call done callback routines */
1284 static int ff_layout_read_done_cb(struct rpc_task *task,
1285                                 struct nfs_pgio_header *hdr)
1286 {
1287         int err;
1288
1289         trace_nfs4_pnfs_read(hdr, task->tk_status);
1290         if (task->tk_status < 0)
1291                 ff_layout_io_track_ds_error(hdr->lseg, hdr->pgio_mirror_idx,
1292                                             hdr->args.offset, hdr->args.count,
1293                                             hdr->res.op_status, OP_READ,
1294                                             task->tk_status);
1295         err = ff_layout_async_handle_error(task, hdr->args.context->state,
1296                                            hdr->ds_clp, hdr->lseg,
1297                                            hdr->pgio_mirror_idx);
1298
1299         switch (err) {
1300         case -NFS4ERR_RESET_TO_PNFS:
1301                 if (ff_layout_choose_best_ds_for_read(hdr->lseg,
1302                                         hdr->pgio_mirror_idx + 1,
1303                                         &hdr->pgio_mirror_idx))
1304                         goto out_eagain;
1305                 ff_layout_read_record_layoutstats_done(task, hdr);
1306                 pnfs_read_resend_pnfs(hdr);
1307                 return task->tk_status;
1308         case -NFS4ERR_RESET_TO_MDS:
1309                 ff_layout_reset_read(hdr);
1310                 return task->tk_status;
1311         case -EAGAIN:
1312                 goto out_eagain;
1313         }
1314
1315         return 0;
1316 out_eagain:
1317         rpc_restart_call_prepare(task);
1318         return -EAGAIN;
1319 }
1320
1321 static bool
1322 ff_layout_need_layoutcommit(struct pnfs_layout_segment *lseg)
1323 {
1324         return !(FF_LAYOUT_LSEG(lseg)->flags & FF_FLAGS_NO_LAYOUTCOMMIT);
1325 }
1326
1327 /*
1328  * We reference the rpc_cred of the first WRITE that triggers the need for
1329  * a LAYOUTCOMMIT, and use it to send the layoutcommit compound.
1330  * rfc5661 is not clear about which credential should be used.
1331  *
1332  * Flexlayout client should treat DS replied FILE_SYNC as DATA_SYNC, so
1333  * to follow http://www.rfc-editor.org/errata_search.php?rfc=5661&eid=2751
1334  * we always send layoutcommit after DS writes.
1335  */
1336 static void
1337 ff_layout_set_layoutcommit(struct inode *inode,
1338                 struct pnfs_layout_segment *lseg,
1339                 loff_t end_offset)
1340 {
1341         if (!ff_layout_need_layoutcommit(lseg))
1342                 return;
1343
1344         pnfs_set_layoutcommit(inode, lseg, end_offset);
1345         dprintk("%s inode %lu pls_end_pos %llu\n", __func__, inode->i_ino,
1346                 (unsigned long long) NFS_I(inode)->layout->plh_lwb);
1347 }
1348
1349 static bool
1350 ff_layout_device_unavailable(struct pnfs_layout_segment *lseg, int idx)
1351 {
1352         /* No mirroring for now */
1353         struct nfs4_deviceid_node *node = FF_LAYOUT_DEVID_NODE(lseg, idx);
1354
1355         return ff_layout_test_devid_unavailable(node);
1356 }
1357
1358 static void ff_layout_read_record_layoutstats_start(struct rpc_task *task,
1359                 struct nfs_pgio_header *hdr)
1360 {
1361         if (test_and_set_bit(NFS_IOHDR_STAT, &hdr->flags))
1362                 return;
1363         nfs4_ff_layout_stat_io_start_read(hdr->inode,
1364                         FF_LAYOUT_COMP(hdr->lseg, hdr->pgio_mirror_idx),
1365                         hdr->args.count,
1366                         task->tk_start);
1367 }
1368
1369 static void ff_layout_read_record_layoutstats_done(struct rpc_task *task,
1370                 struct nfs_pgio_header *hdr)
1371 {
1372         if (!test_and_clear_bit(NFS_IOHDR_STAT, &hdr->flags))
1373                 return;
1374         nfs4_ff_layout_stat_io_end_read(task,
1375                         FF_LAYOUT_COMP(hdr->lseg, hdr->pgio_mirror_idx),
1376                         hdr->args.count,
1377                         hdr->res.count);
1378 }
1379
1380 static int ff_layout_read_prepare_common(struct rpc_task *task,
1381                                          struct nfs_pgio_header *hdr)
1382 {
1383         if (unlikely(test_bit(NFS_CONTEXT_BAD, &hdr->args.context->flags))) {
1384                 rpc_exit(task, -EIO);
1385                 return -EIO;
1386         }
1387         if (ff_layout_device_unavailable(hdr->lseg, hdr->pgio_mirror_idx)) {
1388                 rpc_exit(task, -EHOSTDOWN);
1389                 return -EAGAIN;
1390         }
1391
1392         ff_layout_read_record_layoutstats_start(task, hdr);
1393         return 0;
1394 }
1395
1396 /*
1397  * Call ops for the async read/write cases
1398  * In the case of dense layouts, the offset needs to be reset to its
1399  * original value.
1400  */
1401 static void ff_layout_read_prepare_v3(struct rpc_task *task, void *data)
1402 {
1403         struct nfs_pgio_header *hdr = data;
1404
1405         if (ff_layout_read_prepare_common(task, hdr))
1406                 return;
1407
1408         rpc_call_start(task);
1409 }
1410
1411 static int ff_layout_setup_sequence(struct nfs_client *ds_clp,
1412                                     struct nfs4_sequence_args *args,
1413                                     struct nfs4_sequence_res *res,
1414                                     struct rpc_task *task)
1415 {
1416         if (ds_clp->cl_session)
1417                 return nfs41_setup_sequence(ds_clp->cl_session,
1418                                            args,
1419                                            res,
1420                                            task);
1421         return nfs40_setup_sequence(ds_clp->cl_slot_tbl,
1422                                    args,
1423                                    res,
1424                                    task);
1425 }
1426
1427 static void ff_layout_read_prepare_v4(struct rpc_task *task, void *data)
1428 {
1429         struct nfs_pgio_header *hdr = data;
1430
1431         if (ff_layout_setup_sequence(hdr->ds_clp,
1432                                      &hdr->args.seq_args,
1433                                      &hdr->res.seq_res,
1434                                      task))
1435                 return;
1436
1437         if (ff_layout_read_prepare_common(task, hdr))
1438                 return;
1439
1440         if (nfs4_set_rw_stateid(&hdr->args.stateid, hdr->args.context,
1441                         hdr->args.lock_context, FMODE_READ) == -EIO)
1442                 rpc_exit(task, -EIO); /* lost lock, terminate I/O */
1443 }
1444
1445 static void ff_layout_read_call_done(struct rpc_task *task, void *data)
1446 {
1447         struct nfs_pgio_header *hdr = data;
1448
1449         dprintk("--> %s task->tk_status %d\n", __func__, task->tk_status);
1450
1451         if (test_bit(NFS_IOHDR_REDO, &hdr->flags) &&
1452             task->tk_status == 0) {
1453                 nfs4_sequence_done(task, &hdr->res.seq_res);
1454                 return;
1455         }
1456
1457         /* Note this may cause RPC to be resent */
1458         hdr->mds_ops->rpc_call_done(task, hdr);
1459 }
1460
1461 static void ff_layout_read_count_stats(struct rpc_task *task, void *data)
1462 {
1463         struct nfs_pgio_header *hdr = data;
1464
1465         ff_layout_read_record_layoutstats_done(task, hdr);
1466         rpc_count_iostats_metrics(task,
1467             &NFS_CLIENT(hdr->inode)->cl_metrics[NFSPROC4_CLNT_READ]);
1468 }
1469
1470 static void ff_layout_read_release(void *data)
1471 {
1472         struct nfs_pgio_header *hdr = data;
1473
1474         ff_layout_read_record_layoutstats_done(&hdr->task, hdr);
1475         pnfs_generic_rw_release(data);
1476 }
1477
1478
1479 static int ff_layout_write_done_cb(struct rpc_task *task,
1480                                 struct nfs_pgio_header *hdr)
1481 {
1482         loff_t end_offs = 0;
1483         int err;
1484
1485         trace_nfs4_pnfs_write(hdr, task->tk_status);
1486         if (task->tk_status < 0)
1487                 ff_layout_io_track_ds_error(hdr->lseg, hdr->pgio_mirror_idx,
1488                                             hdr->args.offset, hdr->args.count,
1489                                             hdr->res.op_status, OP_WRITE,
1490                                             task->tk_status);
1491         err = ff_layout_async_handle_error(task, hdr->args.context->state,
1492                                            hdr->ds_clp, hdr->lseg,
1493                                            hdr->pgio_mirror_idx);
1494
1495         switch (err) {
1496         case -NFS4ERR_RESET_TO_PNFS:
1497                 ff_layout_reset_write(hdr, true);
1498                 return task->tk_status;
1499         case -NFS4ERR_RESET_TO_MDS:
1500                 ff_layout_reset_write(hdr, false);
1501                 return task->tk_status;
1502         case -EAGAIN:
1503                 return -EAGAIN;
1504         }
1505
1506         if (hdr->res.verf->committed == NFS_FILE_SYNC ||
1507             hdr->res.verf->committed == NFS_DATA_SYNC)
1508                 end_offs = hdr->mds_offset + (loff_t)hdr->res.count;
1509
1510         /* Note: if the write is unstable, don't set end_offs until commit */
1511         ff_layout_set_layoutcommit(hdr->inode, hdr->lseg, end_offs);
1512
1513         /* zero out fattr since we don't care DS attr at all */
1514         hdr->fattr.valid = 0;
1515         if (task->tk_status >= 0)
1516                 nfs_writeback_update_inode(hdr);
1517
1518         return 0;
1519 }
1520
1521 static int ff_layout_commit_done_cb(struct rpc_task *task,
1522                                      struct nfs_commit_data *data)
1523 {
1524         int err;
1525
1526         trace_nfs4_pnfs_commit_ds(data, task->tk_status);
1527         if (task->tk_status < 0)
1528                 ff_layout_io_track_ds_error(data->lseg, data->ds_commit_index,
1529                                             data->args.offset, data->args.count,
1530                                             data->res.op_status, OP_COMMIT,
1531                                             task->tk_status);
1532         err = ff_layout_async_handle_error(task, NULL, data->ds_clp,
1533                                            data->lseg, data->ds_commit_index);
1534
1535         switch (err) {
1536         case -NFS4ERR_RESET_TO_PNFS:
1537                 pnfs_generic_prepare_to_resend_writes(data);
1538                 return -EAGAIN;
1539         case -NFS4ERR_RESET_TO_MDS:
1540                 pnfs_generic_prepare_to_resend_writes(data);
1541                 return -EAGAIN;
1542         case -EAGAIN:
1543                 rpc_restart_call_prepare(task);
1544                 return -EAGAIN;
1545         }
1546
1547         ff_layout_set_layoutcommit(data->inode, data->lseg, data->lwb);
1548
1549         return 0;
1550 }
1551
1552 static void ff_layout_write_record_layoutstats_start(struct rpc_task *task,
1553                 struct nfs_pgio_header *hdr)
1554 {
1555         if (test_and_set_bit(NFS_IOHDR_STAT, &hdr->flags))
1556                 return;
1557         nfs4_ff_layout_stat_io_start_write(hdr->inode,
1558                         FF_LAYOUT_COMP(hdr->lseg, hdr->pgio_mirror_idx),
1559                         hdr->args.count,
1560                         task->tk_start);
1561 }
1562
1563 static void ff_layout_write_record_layoutstats_done(struct rpc_task *task,
1564                 struct nfs_pgio_header *hdr)
1565 {
1566         if (!test_and_clear_bit(NFS_IOHDR_STAT, &hdr->flags))
1567                 return;
1568         nfs4_ff_layout_stat_io_end_write(task,
1569                         FF_LAYOUT_COMP(hdr->lseg, hdr->pgio_mirror_idx),
1570                         hdr->args.count, hdr->res.count,
1571                         hdr->res.verf->committed);
1572 }
1573
1574 static int ff_layout_write_prepare_common(struct rpc_task *task,
1575                                           struct nfs_pgio_header *hdr)
1576 {
1577         if (unlikely(test_bit(NFS_CONTEXT_BAD, &hdr->args.context->flags))) {
1578                 rpc_exit(task, -EIO);
1579                 return -EIO;
1580         }
1581
1582         if (ff_layout_device_unavailable(hdr->lseg, hdr->pgio_mirror_idx)) {
1583                 rpc_exit(task, -EHOSTDOWN);
1584                 return -EAGAIN;
1585         }
1586
1587         ff_layout_write_record_layoutstats_start(task, hdr);
1588         return 0;
1589 }
1590
1591 static void ff_layout_write_prepare_v3(struct rpc_task *task, void *data)
1592 {
1593         struct nfs_pgio_header *hdr = data;
1594
1595         if (ff_layout_write_prepare_common(task, hdr))
1596                 return;
1597
1598         rpc_call_start(task);
1599 }
1600
1601 static void ff_layout_write_prepare_v4(struct rpc_task *task, void *data)
1602 {
1603         struct nfs_pgio_header *hdr = data;
1604
1605         if (ff_layout_setup_sequence(hdr->ds_clp,
1606                                      &hdr->args.seq_args,
1607                                      &hdr->res.seq_res,
1608                                      task))
1609                 return;
1610
1611         if (ff_layout_write_prepare_common(task, hdr))
1612                 return;
1613
1614         if (nfs4_set_rw_stateid(&hdr->args.stateid, hdr->args.context,
1615                         hdr->args.lock_context, FMODE_WRITE) == -EIO)
1616                 rpc_exit(task, -EIO); /* lost lock, terminate I/O */
1617 }
1618
1619 static void ff_layout_write_call_done(struct rpc_task *task, void *data)
1620 {
1621         struct nfs_pgio_header *hdr = data;
1622
1623         if (test_bit(NFS_IOHDR_REDO, &hdr->flags) &&
1624             task->tk_status == 0) {
1625                 nfs4_sequence_done(task, &hdr->res.seq_res);
1626                 return;
1627         }
1628
1629         /* Note this may cause RPC to be resent */
1630         hdr->mds_ops->rpc_call_done(task, hdr);
1631 }
1632
1633 static void ff_layout_write_count_stats(struct rpc_task *task, void *data)
1634 {
1635         struct nfs_pgio_header *hdr = data;
1636
1637         ff_layout_write_record_layoutstats_done(task, hdr);
1638         rpc_count_iostats_metrics(task,
1639             &NFS_CLIENT(hdr->inode)->cl_metrics[NFSPROC4_CLNT_WRITE]);
1640 }
1641
1642 static void ff_layout_write_release(void *data)
1643 {
1644         struct nfs_pgio_header *hdr = data;
1645
1646         ff_layout_write_record_layoutstats_done(&hdr->task, hdr);
1647         pnfs_generic_rw_release(data);
1648 }
1649
1650 static void ff_layout_commit_record_layoutstats_start(struct rpc_task *task,
1651                 struct nfs_commit_data *cdata)
1652 {
1653         if (test_and_set_bit(NFS_IOHDR_STAT, &cdata->flags))
1654                 return;
1655         nfs4_ff_layout_stat_io_start_write(cdata->inode,
1656                         FF_LAYOUT_COMP(cdata->lseg, cdata->ds_commit_index),
1657                         0, task->tk_start);
1658 }
1659
1660 static void ff_layout_commit_record_layoutstats_done(struct rpc_task *task,
1661                 struct nfs_commit_data *cdata)
1662 {
1663         struct nfs_page *req;
1664         __u64 count = 0;
1665
1666         if (!test_and_clear_bit(NFS_IOHDR_STAT, &cdata->flags))
1667                 return;
1668
1669         if (task->tk_status == 0) {
1670                 list_for_each_entry(req, &cdata->pages, wb_list)
1671                         count += req->wb_bytes;
1672         }
1673         nfs4_ff_layout_stat_io_end_write(task,
1674                         FF_LAYOUT_COMP(cdata->lseg, cdata->ds_commit_index),
1675                         count, count, NFS_FILE_SYNC);
1676 }
1677
1678 static void ff_layout_commit_prepare_common(struct rpc_task *task,
1679                 struct nfs_commit_data *cdata)
1680 {
1681         ff_layout_commit_record_layoutstats_start(task, cdata);
1682 }
1683
1684 static void ff_layout_commit_prepare_v3(struct rpc_task *task, void *data)
1685 {
1686         ff_layout_commit_prepare_common(task, data);
1687         rpc_call_start(task);
1688 }
1689
1690 static void ff_layout_commit_prepare_v4(struct rpc_task *task, void *data)
1691 {
1692         struct nfs_commit_data *wdata = data;
1693
1694         if (ff_layout_setup_sequence(wdata->ds_clp,
1695                                  &wdata->args.seq_args,
1696                                  &wdata->res.seq_res,
1697                                  task))
1698                 return;
1699         ff_layout_commit_prepare_common(task, data);
1700 }
1701
1702 static void ff_layout_commit_done(struct rpc_task *task, void *data)
1703 {
1704         pnfs_generic_write_commit_done(task, data);
1705 }
1706
1707 static void ff_layout_commit_count_stats(struct rpc_task *task, void *data)
1708 {
1709         struct nfs_commit_data *cdata = data;
1710
1711         ff_layout_commit_record_layoutstats_done(task, cdata);
1712         rpc_count_iostats_metrics(task,
1713             &NFS_CLIENT(cdata->inode)->cl_metrics[NFSPROC4_CLNT_COMMIT]);
1714 }
1715
1716 static void ff_layout_commit_release(void *data)
1717 {
1718         struct nfs_commit_data *cdata = data;
1719
1720         ff_layout_commit_record_layoutstats_done(&cdata->task, cdata);
1721         pnfs_generic_commit_release(data);
1722 }
1723
1724 static const struct rpc_call_ops ff_layout_read_call_ops_v3 = {
1725         .rpc_call_prepare = ff_layout_read_prepare_v3,
1726         .rpc_call_done = ff_layout_read_call_done,
1727         .rpc_count_stats = ff_layout_read_count_stats,
1728         .rpc_release = ff_layout_read_release,
1729 };
1730
1731 static const struct rpc_call_ops ff_layout_read_call_ops_v4 = {
1732         .rpc_call_prepare = ff_layout_read_prepare_v4,
1733         .rpc_call_done = ff_layout_read_call_done,
1734         .rpc_count_stats = ff_layout_read_count_stats,
1735         .rpc_release = ff_layout_read_release,
1736 };
1737
1738 static const struct rpc_call_ops ff_layout_write_call_ops_v3 = {
1739         .rpc_call_prepare = ff_layout_write_prepare_v3,
1740         .rpc_call_done = ff_layout_write_call_done,
1741         .rpc_count_stats = ff_layout_write_count_stats,
1742         .rpc_release = ff_layout_write_release,
1743 };
1744
1745 static const struct rpc_call_ops ff_layout_write_call_ops_v4 = {
1746         .rpc_call_prepare = ff_layout_write_prepare_v4,
1747         .rpc_call_done = ff_layout_write_call_done,
1748         .rpc_count_stats = ff_layout_write_count_stats,
1749         .rpc_release = ff_layout_write_release,
1750 };
1751
1752 static const struct rpc_call_ops ff_layout_commit_call_ops_v3 = {
1753         .rpc_call_prepare = ff_layout_commit_prepare_v3,
1754         .rpc_call_done = ff_layout_commit_done,
1755         .rpc_count_stats = ff_layout_commit_count_stats,
1756         .rpc_release = ff_layout_commit_release,
1757 };
1758
1759 static const struct rpc_call_ops ff_layout_commit_call_ops_v4 = {
1760         .rpc_call_prepare = ff_layout_commit_prepare_v4,
1761         .rpc_call_done = ff_layout_commit_done,
1762         .rpc_count_stats = ff_layout_commit_count_stats,
1763         .rpc_release = ff_layout_commit_release,
1764 };
1765
1766 static enum pnfs_try_status
1767 ff_layout_read_pagelist(struct nfs_pgio_header *hdr)
1768 {
1769         struct pnfs_layout_segment *lseg = hdr->lseg;
1770         struct nfs4_pnfs_ds *ds;
1771         struct rpc_clnt *ds_clnt;
1772         struct rpc_cred *ds_cred;
1773         loff_t offset = hdr->args.offset;
1774         u32 idx = hdr->pgio_mirror_idx;
1775         int vers;
1776         struct nfs_fh *fh;
1777
1778         dprintk("--> %s ino %lu pgbase %u req %Zu@%llu\n",
1779                 __func__, hdr->inode->i_ino,
1780                 hdr->args.pgbase, (size_t)hdr->args.count, offset);
1781
1782         ds = nfs4_ff_layout_prepare_ds(lseg, idx, false);
1783         if (!ds)
1784                 goto out_failed;
1785
1786         ds_clnt = nfs4_ff_find_or_create_ds_client(lseg, idx, ds->ds_clp,
1787                                                    hdr->inode);
1788         if (IS_ERR(ds_clnt))
1789                 goto out_failed;
1790
1791         ds_cred = ff_layout_get_ds_cred(lseg, idx, hdr->cred);
1792         if (!ds_cred)
1793                 goto out_failed;
1794
1795         vers = nfs4_ff_layout_ds_version(lseg, idx);
1796
1797         dprintk("%s USE DS: %s cl_count %d vers %d\n", __func__,
1798                 ds->ds_remotestr, atomic_read(&ds->ds_clp->cl_count), vers);
1799
1800         hdr->pgio_done_cb = ff_layout_read_done_cb;
1801         atomic_inc(&ds->ds_clp->cl_count);
1802         hdr->ds_clp = ds->ds_clp;
1803         fh = nfs4_ff_layout_select_ds_fh(lseg, idx);
1804         if (fh)
1805                 hdr->args.fh = fh;
1806         /*
1807          * Note that if we ever decide to split across DSes,
1808          * then we may need to handle dense-like offsets.
1809          */
1810         hdr->args.offset = offset;
1811         hdr->mds_offset = offset;
1812
1813         /* Perform an asynchronous read to ds */
1814         nfs_initiate_pgio(ds_clnt, hdr, ds_cred, ds->ds_clp->rpc_ops,
1815                           vers == 3 ? &ff_layout_read_call_ops_v3 :
1816                                       &ff_layout_read_call_ops_v4,
1817                           0, RPC_TASK_SOFTCONN);
1818         put_rpccred(ds_cred);
1819         return PNFS_ATTEMPTED;
1820
1821 out_failed:
1822         if (ff_layout_avoid_mds_available_ds(lseg))
1823                 return PNFS_TRY_AGAIN;
1824         return PNFS_NOT_ATTEMPTED;
1825 }
1826
1827 /* Perform async writes. */
1828 static enum pnfs_try_status
1829 ff_layout_write_pagelist(struct nfs_pgio_header *hdr, int sync)
1830 {
1831         struct pnfs_layout_segment *lseg = hdr->lseg;
1832         struct nfs4_pnfs_ds *ds;
1833         struct rpc_clnt *ds_clnt;
1834         struct rpc_cred *ds_cred;
1835         loff_t offset = hdr->args.offset;
1836         int vers;
1837         struct nfs_fh *fh;
1838         int idx = hdr->pgio_mirror_idx;
1839
1840         ds = nfs4_ff_layout_prepare_ds(lseg, idx, true);
1841         if (!ds)
1842                 return PNFS_NOT_ATTEMPTED;
1843
1844         ds_clnt = nfs4_ff_find_or_create_ds_client(lseg, idx, ds->ds_clp,
1845                                                    hdr->inode);
1846         if (IS_ERR(ds_clnt))
1847                 return PNFS_NOT_ATTEMPTED;
1848
1849         ds_cred = ff_layout_get_ds_cred(lseg, idx, hdr->cred);
1850         if (!ds_cred)
1851                 return PNFS_NOT_ATTEMPTED;
1852
1853         vers = nfs4_ff_layout_ds_version(lseg, idx);
1854
1855         dprintk("%s ino %lu sync %d req %Zu@%llu DS: %s cl_count %d vers %d\n",
1856                 __func__, hdr->inode->i_ino, sync, (size_t) hdr->args.count,
1857                 offset, ds->ds_remotestr, atomic_read(&ds->ds_clp->cl_count),
1858                 vers);
1859
1860         hdr->pgio_done_cb = ff_layout_write_done_cb;
1861         atomic_inc(&ds->ds_clp->cl_count);
1862         hdr->ds_clp = ds->ds_clp;
1863         hdr->ds_commit_idx = idx;
1864         fh = nfs4_ff_layout_select_ds_fh(lseg, idx);
1865         if (fh)
1866                 hdr->args.fh = fh;
1867
1868         /*
1869          * Note that if we ever decide to split across DSes,
1870          * then we may need to handle dense-like offsets.
1871          */
1872         hdr->args.offset = offset;
1873
1874         /* Perform an asynchronous write */
1875         nfs_initiate_pgio(ds_clnt, hdr, ds_cred, ds->ds_clp->rpc_ops,
1876                           vers == 3 ? &ff_layout_write_call_ops_v3 :
1877                                       &ff_layout_write_call_ops_v4,
1878                           sync, RPC_TASK_SOFTCONN);
1879         put_rpccred(ds_cred);
1880         return PNFS_ATTEMPTED;
1881 }
1882
1883 static u32 calc_ds_index_from_commit(struct pnfs_layout_segment *lseg, u32 i)
1884 {
1885         return i;
1886 }
1887
1888 static struct nfs_fh *
1889 select_ds_fh_from_commit(struct pnfs_layout_segment *lseg, u32 i)
1890 {
1891         struct nfs4_ff_layout_segment *flseg = FF_LAYOUT_LSEG(lseg);
1892
1893         /* FIXME: Assume that there is only one NFS version available
1894          * for the DS.
1895          */
1896         return &flseg->mirror_array[i]->fh_versions[0];
1897 }
1898
1899 static int ff_layout_initiate_commit(struct nfs_commit_data *data, int how)
1900 {
1901         struct pnfs_layout_segment *lseg = data->lseg;
1902         struct nfs4_pnfs_ds *ds;
1903         struct rpc_clnt *ds_clnt;
1904         struct rpc_cred *ds_cred;
1905         u32 idx;
1906         int vers, ret;
1907         struct nfs_fh *fh;
1908
1909         idx = calc_ds_index_from_commit(lseg, data->ds_commit_index);
1910         ds = nfs4_ff_layout_prepare_ds(lseg, idx, true);
1911         if (!ds)
1912                 goto out_err;
1913
1914         ds_clnt = nfs4_ff_find_or_create_ds_client(lseg, idx, ds->ds_clp,
1915                                                    data->inode);
1916         if (IS_ERR(ds_clnt))
1917                 goto out_err;
1918
1919         ds_cred = ff_layout_get_ds_cred(lseg, idx, data->cred);
1920         if (!ds_cred)
1921                 goto out_err;
1922
1923         vers = nfs4_ff_layout_ds_version(lseg, idx);
1924
1925         dprintk("%s ino %lu, how %d cl_count %d vers %d\n", __func__,
1926                 data->inode->i_ino, how, atomic_read(&ds->ds_clp->cl_count),
1927                 vers);
1928         data->commit_done_cb = ff_layout_commit_done_cb;
1929         data->cred = ds_cred;
1930         atomic_inc(&ds->ds_clp->cl_count);
1931         data->ds_clp = ds->ds_clp;
1932         fh = select_ds_fh_from_commit(lseg, data->ds_commit_index);
1933         if (fh)
1934                 data->args.fh = fh;
1935
1936         ret = nfs_initiate_commit(ds_clnt, data, ds->ds_clp->rpc_ops,
1937                                    vers == 3 ? &ff_layout_commit_call_ops_v3 :
1938                                                &ff_layout_commit_call_ops_v4,
1939                                    how, RPC_TASK_SOFTCONN);
1940         put_rpccred(ds_cred);
1941         return ret;
1942 out_err:
1943         pnfs_generic_prepare_to_resend_writes(data);
1944         pnfs_generic_commit_release(data);
1945         return -EAGAIN;
1946 }
1947
1948 static int
1949 ff_layout_commit_pagelist(struct inode *inode, struct list_head *mds_pages,
1950                            int how, struct nfs_commit_info *cinfo)
1951 {
1952         return pnfs_generic_commit_pagelist(inode, mds_pages, how, cinfo,
1953                                             ff_layout_initiate_commit);
1954 }
1955
1956 static struct pnfs_ds_commit_info *
1957 ff_layout_get_ds_info(struct inode *inode)
1958 {
1959         struct pnfs_layout_hdr *layout = NFS_I(inode)->layout;
1960
1961         if (layout == NULL)
1962                 return NULL;
1963
1964         return &FF_LAYOUT_FROM_HDR(layout)->commit_info;
1965 }
1966
1967 static void
1968 ff_layout_free_deviceid_node(struct nfs4_deviceid_node *d)
1969 {
1970         nfs4_ff_layout_free_deviceid(container_of(d, struct nfs4_ff_layout_ds,
1971                                                   id_node));
1972 }
1973
1974 static int ff_layout_encode_ioerr(struct nfs4_flexfile_layout *flo,
1975                                   struct xdr_stream *xdr,
1976                                   const struct nfs4_layoutreturn_args *args,
1977                                   const struct nfs4_flexfile_layoutreturn_args *ff_args)
1978 {
1979         __be32 *start;
1980
1981         start = xdr_reserve_space(xdr, 4);
1982         if (unlikely(!start))
1983                 return -E2BIG;
1984
1985         *start = cpu_to_be32(ff_args->num_errors);
1986         /* This assume we always return _ALL_ layouts */
1987         return ff_layout_encode_ds_ioerr(xdr, &ff_args->errors);
1988 }
1989
1990 /* report nothing for now */
1991 static void ff_layout_encode_iostats_array(struct nfs4_flexfile_layout *flo,
1992                                      struct xdr_stream *xdr,
1993                                      const struct nfs4_layoutreturn_args *args)
1994 {
1995         __be32 *p;
1996
1997         p = xdr_reserve_space(xdr, 4);
1998         if (likely(p))
1999                 *p = cpu_to_be32(0);
2000 }
2001
2002 static struct nfs4_deviceid_node *
2003 ff_layout_alloc_deviceid_node(struct nfs_server *server,
2004                               struct pnfs_device *pdev, gfp_t gfp_flags)
2005 {
2006         struct nfs4_ff_layout_ds *dsaddr;
2007
2008         dsaddr = nfs4_ff_alloc_deviceid_node(server, pdev, gfp_flags);
2009         if (!dsaddr)
2010                 return NULL;
2011         return &dsaddr->id_node;
2012 }
2013
2014 static void
2015 ff_layout_encode_layoutreturn(struct xdr_stream *xdr,
2016                 const void *voidargs,
2017                 const struct nfs4_xdr_opaque_data *ff_opaque)
2018 {
2019         const struct nfs4_layoutreturn_args *args = voidargs;
2020         struct pnfs_layout_hdr *lo = args->layout;
2021         struct nfs4_flexfile_layout *flo = FF_LAYOUT_FROM_HDR(lo);
2022         __be32 *start;
2023
2024         dprintk("%s: Begin\n", __func__);
2025         start = xdr_reserve_space(xdr, 4);
2026         BUG_ON(!start);
2027
2028         ff_layout_encode_ioerr(flo, xdr, args, ff_opaque->data);
2029         ff_layout_encode_iostats_array(flo, xdr, args);
2030
2031         *start = cpu_to_be32((xdr->p - start - 1) * 4);
2032         dprintk("%s: Return\n", __func__);
2033 }
2034
2035 static void
2036 ff_layout_free_layoutreturn(struct nfs4_xdr_opaque_data *args)
2037 {
2038         struct nfs4_flexfile_layoutreturn_args *ff_args;
2039
2040         if (!args->data)
2041                 return;
2042         ff_args = args->data;
2043         args->data = NULL;
2044
2045         ff_layout_free_ds_ioerr(&ff_args->errors);
2046
2047         kfree(ff_args);
2048 }
2049
2050 const struct nfs4_xdr_opaque_ops layoutreturn_ops = {
2051         .encode = ff_layout_encode_layoutreturn,
2052         .free = ff_layout_free_layoutreturn,
2053 };
2054
2055 static int
2056 ff_layout_prepare_layoutreturn(struct nfs4_layoutreturn_args *args)
2057 {
2058         struct nfs4_flexfile_layoutreturn_args *ff_args;
2059
2060         ff_args = kmalloc(sizeof(*ff_args), GFP_KERNEL);
2061         if (!ff_args)
2062                 return -ENOMEM;
2063
2064         INIT_LIST_HEAD(&ff_args->errors);
2065         ff_args->num_errors = ff_layout_fetch_ds_ioerr(args->layout,
2066                         &args->range, &ff_args->errors,
2067                         FF_LAYOUTRETURN_MAXERR);
2068
2069         args->ld_private->ops = &layoutreturn_ops;
2070         args->ld_private->data = ff_args;
2071         return 0;
2072 }
2073
2074 static int
2075 ff_layout_ntop4(const struct sockaddr *sap, char *buf, const size_t buflen)
2076 {
2077         const struct sockaddr_in *sin = (struct sockaddr_in *)sap;
2078
2079         return snprintf(buf, buflen, "%pI4", &sin->sin_addr);
2080 }
2081
2082 static size_t
2083 ff_layout_ntop6_noscopeid(const struct sockaddr *sap, char *buf,
2084                           const int buflen)
2085 {
2086         const struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)sap;
2087         const struct in6_addr *addr = &sin6->sin6_addr;
2088
2089         /*
2090          * RFC 4291, Section 2.2.2
2091          *
2092          * Shorthanded ANY address
2093          */
2094         if (ipv6_addr_any(addr))
2095                 return snprintf(buf, buflen, "::");
2096
2097         /*
2098          * RFC 4291, Section 2.2.2
2099          *
2100          * Shorthanded loopback address
2101          */
2102         if (ipv6_addr_loopback(addr))
2103                 return snprintf(buf, buflen, "::1");
2104
2105         /*
2106          * RFC 4291, Section 2.2.3
2107          *
2108          * Special presentation address format for mapped v4
2109          * addresses.
2110          */
2111         if (ipv6_addr_v4mapped(addr))
2112                 return snprintf(buf, buflen, "::ffff:%pI4",
2113                                         &addr->s6_addr32[3]);
2114
2115         /*
2116          * RFC 4291, Section 2.2.1
2117          */
2118         return snprintf(buf, buflen, "%pI6c", addr);
2119 }
2120
2121 /* Derived from rpc_sockaddr2uaddr */
2122 static void
2123 ff_layout_encode_netaddr(struct xdr_stream *xdr, struct nfs4_pnfs_ds_addr *da)
2124 {
2125         struct sockaddr *sap = (struct sockaddr *)&da->da_addr;
2126         char portbuf[RPCBIND_MAXUADDRPLEN];
2127         char addrbuf[RPCBIND_MAXUADDRLEN];
2128         char *netid;
2129         unsigned short port;
2130         int len, netid_len;
2131         __be32 *p;
2132
2133         switch (sap->sa_family) {
2134         case AF_INET:
2135                 if (ff_layout_ntop4(sap, addrbuf, sizeof(addrbuf)) == 0)
2136                         return;
2137                 port = ntohs(((struct sockaddr_in *)sap)->sin_port);
2138                 netid = "tcp";
2139                 netid_len = 3;
2140                 break;
2141         case AF_INET6:
2142                 if (ff_layout_ntop6_noscopeid(sap, addrbuf, sizeof(addrbuf)) == 0)
2143                         return;
2144                 port = ntohs(((struct sockaddr_in6 *)sap)->sin6_port);
2145                 netid = "tcp6";
2146                 netid_len = 4;
2147                 break;
2148         default:
2149                 /* we only support tcp and tcp6 */
2150                 WARN_ON_ONCE(1);
2151                 return;
2152         }
2153
2154         snprintf(portbuf, sizeof(portbuf), ".%u.%u", port >> 8, port & 0xff);
2155         len = strlcat(addrbuf, portbuf, sizeof(addrbuf));
2156
2157         p = xdr_reserve_space(xdr, 4 + netid_len);
2158         xdr_encode_opaque(p, netid, netid_len);
2159
2160         p = xdr_reserve_space(xdr, 4 + len);
2161         xdr_encode_opaque(p, addrbuf, len);
2162 }
2163
2164 static void
2165 ff_layout_encode_nfstime(struct xdr_stream *xdr,
2166                          ktime_t t)
2167 {
2168         struct timespec64 ts;
2169         __be32 *p;
2170
2171         p = xdr_reserve_space(xdr, 12);
2172         ts = ktime_to_timespec64(t);
2173         p = xdr_encode_hyper(p, ts.tv_sec);
2174         *p++ = cpu_to_be32(ts.tv_nsec);
2175 }
2176
2177 static void
2178 ff_layout_encode_io_latency(struct xdr_stream *xdr,
2179                             struct nfs4_ff_io_stat *stat)
2180 {
2181         __be32 *p;
2182
2183         p = xdr_reserve_space(xdr, 5 * 8);
2184         p = xdr_encode_hyper(p, stat->ops_requested);
2185         p = xdr_encode_hyper(p, stat->bytes_requested);
2186         p = xdr_encode_hyper(p, stat->ops_completed);
2187         p = xdr_encode_hyper(p, stat->bytes_completed);
2188         p = xdr_encode_hyper(p, stat->bytes_not_delivered);
2189         ff_layout_encode_nfstime(xdr, stat->total_busy_time);
2190         ff_layout_encode_nfstime(xdr, stat->aggregate_completion_time);
2191 }
2192
2193 static void
2194 ff_layout_encode_ff_layoutupdate(struct xdr_stream *xdr,
2195                               const struct nfs42_layoutstat_devinfo *devinfo,
2196                               struct nfs4_ff_layout_mirror *mirror)
2197 {
2198         struct nfs4_pnfs_ds_addr *da;
2199         struct nfs4_pnfs_ds *ds = mirror->mirror_ds->ds;
2200         struct nfs_fh *fh = &mirror->fh_versions[0];
2201         __be32 *p;
2202
2203         da = list_first_entry(&ds->ds_addrs, struct nfs4_pnfs_ds_addr, da_node);
2204         dprintk("%s: DS %s: encoding address %s\n",
2205                 __func__, ds->ds_remotestr, da->da_remotestr);
2206         /* netaddr4 */
2207         ff_layout_encode_netaddr(xdr, da);
2208         /* nfs_fh4 */
2209         p = xdr_reserve_space(xdr, 4 + fh->size);
2210         xdr_encode_opaque(p, fh->data, fh->size);
2211         /* ff_io_latency4 read */
2212         spin_lock(&mirror->lock);
2213         ff_layout_encode_io_latency(xdr, &mirror->read_stat.io_stat);
2214         /* ff_io_latency4 write */
2215         ff_layout_encode_io_latency(xdr, &mirror->write_stat.io_stat);
2216         spin_unlock(&mirror->lock);
2217         /* nfstime4 */
2218         ff_layout_encode_nfstime(xdr, ktime_sub(ktime_get(), mirror->start_time));
2219         /* bool */
2220         p = xdr_reserve_space(xdr, 4);
2221         *p = cpu_to_be32(false);
2222 }
2223
2224 static void
2225 ff_layout_encode_layoutstats(struct xdr_stream *xdr, const void *args,
2226                              const struct nfs4_xdr_opaque_data *opaque)
2227 {
2228         struct nfs42_layoutstat_devinfo *devinfo = container_of(opaque,
2229                         struct nfs42_layoutstat_devinfo, ld_private);
2230         __be32 *start;
2231
2232         /* layoutupdate length */
2233         start = xdr_reserve_space(xdr, 4);
2234         ff_layout_encode_ff_layoutupdate(xdr, devinfo, opaque->data);
2235
2236         *start = cpu_to_be32((xdr->p - start - 1) * 4);
2237 }
2238
2239 static void
2240 ff_layout_free_layoutstats(struct nfs4_xdr_opaque_data *opaque)
2241 {
2242         struct nfs4_ff_layout_mirror *mirror = opaque->data;
2243
2244         ff_layout_put_mirror(mirror);
2245 }
2246
2247 static const struct nfs4_xdr_opaque_ops layoutstat_ops = {
2248         .encode = ff_layout_encode_layoutstats,
2249         .free   = ff_layout_free_layoutstats,
2250 };
2251
2252 static int
2253 ff_layout_mirror_prepare_stats(struct pnfs_layout_hdr *lo,
2254                                struct nfs42_layoutstat_devinfo *devinfo,
2255                                int dev_limit)
2256 {
2257         struct nfs4_flexfile_layout *ff_layout = FF_LAYOUT_FROM_HDR(lo);
2258         struct nfs4_ff_layout_mirror *mirror;
2259         struct nfs4_deviceid_node *dev;
2260         int i = 0;
2261
2262         list_for_each_entry(mirror, &ff_layout->mirrors, mirrors) {
2263                 if (i >= dev_limit)
2264                         break;
2265                 if (!mirror->mirror_ds)
2266                         continue;
2267                 if (!test_and_clear_bit(NFS4_FF_MIRROR_STAT_AVAIL, &mirror->flags))
2268                         continue;
2269                 /* mirror refcount put in cleanup_layoutstats */
2270                 if (!atomic_inc_not_zero(&mirror->ref))
2271                         continue;
2272                 dev = &mirror->mirror_ds->id_node;
2273                 memcpy(&devinfo->dev_id, &dev->deviceid, NFS4_DEVICEID4_SIZE);
2274                 devinfo->offset = 0;
2275                 devinfo->length = NFS4_MAX_UINT64;
2276                 spin_lock(&mirror->lock);
2277                 devinfo->read_count = mirror->read_stat.io_stat.ops_completed;
2278                 devinfo->read_bytes = mirror->read_stat.io_stat.bytes_completed;
2279                 devinfo->write_count = mirror->write_stat.io_stat.ops_completed;
2280                 devinfo->write_bytes = mirror->write_stat.io_stat.bytes_completed;
2281                 spin_unlock(&mirror->lock);
2282                 devinfo->layout_type = LAYOUT_FLEX_FILES;
2283                 devinfo->ld_private.ops = &layoutstat_ops;
2284                 devinfo->ld_private.data = mirror;
2285
2286                 devinfo++;
2287                 i++;
2288         }
2289         return i;
2290 }
2291
2292 static int
2293 ff_layout_prepare_layoutstats(struct nfs42_layoutstat_args *args)
2294 {
2295         struct nfs4_flexfile_layout *ff_layout;
2296         const int dev_count = PNFS_LAYOUTSTATS_MAXDEV;
2297
2298         /* For now, send at most PNFS_LAYOUTSTATS_MAXDEV statistics */
2299         args->devinfo = kmalloc_array(dev_count, sizeof(*args->devinfo), GFP_NOIO);
2300         if (!args->devinfo)
2301                 return -ENOMEM;
2302
2303         spin_lock(&args->inode->i_lock);
2304         ff_layout = FF_LAYOUT_FROM_HDR(NFS_I(args->inode)->layout);
2305         args->num_dev = ff_layout_mirror_prepare_stats(&ff_layout->generic_hdr,
2306                         &args->devinfo[0], dev_count);
2307         spin_unlock(&args->inode->i_lock);
2308         if (!args->num_dev) {
2309                 kfree(args->devinfo);
2310                 args->devinfo = NULL;
2311                 return -ENOENT;
2312         }
2313
2314         return 0;
2315 }
2316
2317 static struct pnfs_layoutdriver_type flexfilelayout_type = {
2318         .id                     = LAYOUT_FLEX_FILES,
2319         .name                   = "LAYOUT_FLEX_FILES",
2320         .owner                  = THIS_MODULE,
2321         .alloc_layout_hdr       = ff_layout_alloc_layout_hdr,
2322         .free_layout_hdr        = ff_layout_free_layout_hdr,
2323         .alloc_lseg             = ff_layout_alloc_lseg,
2324         .free_lseg              = ff_layout_free_lseg,
2325         .add_lseg               = ff_layout_add_lseg,
2326         .pg_read_ops            = &ff_layout_pg_read_ops,
2327         .pg_write_ops           = &ff_layout_pg_write_ops,
2328         .get_ds_info            = ff_layout_get_ds_info,
2329         .free_deviceid_node     = ff_layout_free_deviceid_node,
2330         .mark_request_commit    = pnfs_layout_mark_request_commit,
2331         .clear_request_commit   = pnfs_generic_clear_request_commit,
2332         .scan_commit_lists      = pnfs_generic_scan_commit_lists,
2333         .recover_commit_reqs    = pnfs_generic_recover_commit_reqs,
2334         .commit_pagelist        = ff_layout_commit_pagelist,
2335         .read_pagelist          = ff_layout_read_pagelist,
2336         .write_pagelist         = ff_layout_write_pagelist,
2337         .alloc_deviceid_node    = ff_layout_alloc_deviceid_node,
2338         .prepare_layoutreturn   = ff_layout_prepare_layoutreturn,
2339         .sync                   = pnfs_nfs_generic_sync,
2340         .prepare_layoutstats    = ff_layout_prepare_layoutstats,
2341 };
2342
2343 static int __init nfs4flexfilelayout_init(void)
2344 {
2345         printk(KERN_INFO "%s: NFSv4 Flexfile Layout Driver Registering...\n",
2346                __func__);
2347         if (!ff_zero_group) {
2348                 ff_zero_group = groups_alloc(0);
2349                 if (!ff_zero_group)
2350                         return -ENOMEM;
2351         }
2352         return pnfs_register_layoutdriver(&flexfilelayout_type);
2353 }
2354
2355 static void __exit nfs4flexfilelayout_exit(void)
2356 {
2357         printk(KERN_INFO "%s: NFSv4 Flexfile Layout Driver Unregistering...\n",
2358                __func__);
2359         pnfs_unregister_layoutdriver(&flexfilelayout_type);
2360         if (ff_zero_group) {
2361                 put_group_info(ff_zero_group);
2362                 ff_zero_group = NULL;
2363         }
2364 }
2365
2366 MODULE_ALIAS("nfs-layouttype4-4");
2367
2368 MODULE_LICENSE("GPL");
2369 MODULE_DESCRIPTION("The NFSv4 flexfile layout driver");
2370
2371 module_init(nfs4flexfilelayout_init);
2372 module_exit(nfs4flexfilelayout_exit);