fs/nfs/objlayout/objio_osd.c

   1 /*
   2  *  pNFS Objects layout implementation over open-osd initiator library
   3  *
   4  *  Copyright (C) 2009 Panasas Inc. [year of first publication]
   5  *  All rights reserved.
   6  *
   7  *  Benny Halevy <bhalevy@panasas.com>
   8  *  Boaz Harrosh <bharrosh@panasas.com>
   9  *
  10  *  This program is free software; you can redistribute it and/or modify
  11  *  it under the terms of the GNU General Public License version 2
  12  *  See the file COPYING included with this distribution for more details.
  13  *
  14  *  Redistribution and use in source and binary forms, with or without
  15  *  modification, are permitted provided that the following conditions
  16  *  are met:
  17  *
  18  *  1. Redistributions of source code must retain the above copyright
  19  *     notice, this list of conditions and the following disclaimer.
  20  *  2. Redistributions in binary form must reproduce the above copyright
  21  *     notice, this list of conditions and the following disclaimer in the
  22  *     documentation and/or other materials provided with the distribution.
  23  *  3. Neither the name of the Panasas company nor the names of its
  24  *     contributors may be used to endorse or promote products derived
  25  *     from this software without specific prior written permission.
  26  *
  27  *  THIS SOFTWARE IS PROVIDED ``AS IS'' AND ANY EXPRESS OR IMPLIED
  28  *  WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
  29  *  MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
  30  *  DISCLAIMED. IN NO EVENT SHALL THE REGENTS OR CONTRIBUTORS BE LIABLE
  31  *  FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
  32  *  CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
  33  *  SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR
  34  *  BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
  35  *  LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
  36  *  NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
  37  *  SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  38  */
  39
  40 #include <linux/module.h>
  41 #include <scsi/osd_initiator.h>
  42
  43 #include "objlayout.h"
  44
  45 #define NFSDBG_FACILITY         NFSDBG_PNFS_LD
  46
  47 #define _LLU(x) ((unsigned long long)x)
  48
  49 enum { BIO_MAX_PAGES_KMALLOC =
  50                 (PAGE_SIZE - sizeof(struct bio)) / sizeof(struct bio_vec),
  51 };
  52
  53 struct objio_dev_ent {
  54         struct nfs4_deviceid_node id_node;
  55         struct osd_dev *od;
  56 };
  57
  58 static void
  59 objio_free_deviceid_node(struct nfs4_deviceid_node *d)
  60 {
  61         struct objio_dev_ent *de = container_of(d, struct objio_dev_ent, id_node);
  62
  63         dprintk("%s: free od=%p\n", __func__, de->od);
  64         osduld_put_device(de->od);
  65         kfree(de);
  66 }
  67
  68 static struct objio_dev_ent *_dev_list_find(const struct nfs_server *nfss,
  69         const struct nfs4_deviceid *d_id)
  70 {
  71         struct nfs4_deviceid_node *d;
  72         struct objio_dev_ent *de;
  73
  74         d = nfs4_find_get_deviceid(nfss->pnfs_curr_ld, nfss->nfs_client, d_id);
  75         if (!d)
  76                 return NULL;
  77
  78         de = container_of(d, struct objio_dev_ent, id_node);
  79         return de;
  80 }
  81
  82 static struct objio_dev_ent *
  83 _dev_list_add(const struct nfs_server *nfss,
  84         const struct nfs4_deviceid *d_id, struct osd_dev *od,
  85         gfp_t gfp_flags)
  86 {
  87         struct nfs4_deviceid_node *d;
  88         struct objio_dev_ent *de = kzalloc(sizeof(*de), gfp_flags);
  89         struct objio_dev_ent *n;
  90
  91         if (!de) {
  92                 dprintk("%s: -ENOMEM od=%p\n", __func__, od);
  93                 return NULL;
  94         }
  95
  96         dprintk("%s: Adding od=%p\n", __func__, od);
  97         nfs4_init_deviceid_node(&de->id_node,
  98                                 nfss->pnfs_curr_ld,
  99                                 nfss->nfs_client,
 100                                 d_id);
 101         de->od = od;
 102
 103         d = nfs4_insert_deviceid_node(&de->id_node);
 104         n = container_of(d, struct objio_dev_ent, id_node);
 105         if (n != de) {
 106                 dprintk("%s: Race with other n->od=%p\n", __func__, n->od);
 107                 objio_free_deviceid_node(&de->id_node);
 108                 de = n;
 109         }
 110
 111         return de;
 112 }
 113
 114 struct caps_buffers {
 115         u8 caps_key[OSD_CRYPTO_KEYID_SIZE];
 116         u8 creds[OSD_CAP_LEN];
 117 };
 118
 119 struct objio_segment {
 120         struct pnfs_layout_segment lseg;
 121
 122         struct pnfs_osd_object_cred *comps;
 123
 124         unsigned mirrors_p1;
 125         unsigned stripe_unit;
 126         unsigned group_width;   /* Data stripe_units without integrity comps */
 127         u64 group_depth;
 128         unsigned group_count;
 129
 130         unsigned max_io_size;
 131
 132         unsigned comps_index;
 133         unsigned num_comps;
 134         /* variable length */
 135         struct objio_dev_ent *ods[];
 136 };
 137
 138 static inline struct objio_segment *
 139 OBJIO_LSEG(struct pnfs_layout_segment *lseg)
 140 {
 141         return container_of(lseg, struct objio_segment, lseg);
 142 }
 143
 144 struct objio_state;
 145 typedef ssize_t (*objio_done_fn)(struct objio_state *ios);
 146
 147 struct objio_state {
 148         /* Generic layer */
 149         struct objlayout_io_state ol_state;
 150
 151         struct objio_segment *layout;
 152
 153         struct kref kref;
 154         objio_done_fn done;
 155         void *private;
 156
 157         unsigned long length;
 158         unsigned numdevs; /* Actually used devs in this IO */
 159         /* A per-device variable array of size numdevs */
 160         struct _objio_per_comp {
 161                 struct bio *bio;
 162                 struct osd_request *or;
 163                 unsigned long length;
 164                 u64 offset;
 165                 unsigned dev;
 166         } per_dev[];
 167 };
 168
 169 /* Send and wait for a get_device_info of devices in the layout,
 170    then look them up with the osd_initiator library */
 171 static struct objio_dev_ent *_device_lookup(struct pnfs_layout_hdr *pnfslay,
 172                                 struct objio_segment *objio_seg, unsigned comp,
 173                                 gfp_t gfp_flags)
 174 {
 175         struct pnfs_osd_deviceaddr *deviceaddr;
 176         struct nfs4_deviceid *d_id;
 177         struct objio_dev_ent *ode;
 178         struct osd_dev *od;
 179         struct osd_dev_info odi;
 180         int err;
 181
 182         d_id = &objio_seg->comps[comp].oc_object_id.oid_device_id;
 183
 184         ode = _dev_list_find(NFS_SERVER(pnfslay->plh_inode), d_id);
 185         if (ode)
 186                 return ode;
 187
 188         err = objlayout_get_deviceinfo(pnfslay, d_id, &deviceaddr, gfp_flags);
 189         if (unlikely(err)) {
 190                 dprintk("%s: objlayout_get_deviceinfo dev(%llx:%llx) =>%d\n",
 191                         __func__, _DEVID_LO(d_id), _DEVID_HI(d_id), err);
 192                 return ERR_PTR(err);
 193         }
 194
 195         odi.systemid_len = deviceaddr->oda_systemid.len;
 196         if (odi.systemid_len > sizeof(odi.systemid)) {
 197                 err = -EINVAL;
 198                 goto out;
 199         } else if (odi.systemid_len)
 200                 memcpy(odi.systemid, deviceaddr->oda_systemid.data,
 201                        odi.systemid_len);
 202         odi.osdname_len  = deviceaddr->oda_osdname.len;
 203         odi.osdname      = (u8 *)deviceaddr->oda_osdname.data;
 204
 205         if (!odi.osdname_len && !odi.systemid_len) {
 206                 dprintk("%s: !odi.osdname_len && !odi.systemid_len\n",
 207                         __func__);
 208                 err = -ENODEV;
 209                 goto out;
 210         }
 211
 212         od = osduld_info_lookup(&odi);
 213         if (unlikely(IS_ERR(od))) {
 214                 err = PTR_ERR(od);
 215                 dprintk("%s: osduld_info_lookup => %d\n", __func__, err);
 216                 goto out;
 217         }
 218
 219         ode = _dev_list_add(NFS_SERVER(pnfslay->plh_inode), d_id, od,
 220                             gfp_flags);
 221
 222 out:
 223         dprintk("%s: return=%d\n", __func__, err);
 224         objlayout_put_deviceinfo(deviceaddr);
 225         return err ? ERR_PTR(err) : ode;
 226 }
 227
 228 static int objio_devices_lookup(struct pnfs_layout_hdr *pnfslay,
 229         struct objio_segment *objio_seg,
 230         gfp_t gfp_flags)
 231 {
 232         unsigned i;
 233         int err;
 234
 235         /* lookup all devices */
 236         for (i = 0; i < objio_seg->num_comps; i++) {
 237                 struct objio_dev_ent *ode;
 238
 239                 ode = _device_lookup(pnfslay, objio_seg, i, gfp_flags);
 240                 if (unlikely(IS_ERR(ode))) {
 241                         err = PTR_ERR(ode);
 242                         goto out;
 243                 }
 244                 objio_seg->ods[i] = ode;
 245         }
 246         err = 0;
 247
 248 out:
 249         dprintk("%s: return=%d\n", __func__, err);
 250         return err;
 251 }
 252
 253 static int _verify_data_map(struct pnfs_osd_layout *layout)
 254 {
 255         struct pnfs_osd_data_map *data_map = &layout->olo_map;
 256         u64 stripe_length;
 257         u32 group_width;
 258
 259 /* FIXME: Only raid0 for now. if not go through MDS */
 260         if (data_map->odm_raid_algorithm != PNFS_OSD_RAID_0) {
 261                 printk(KERN_ERR "Only RAID_0 for now\n");
 262                 return -ENOTSUPP;
 263         }
 264         if (0 != (data_map->odm_num_comps % (data_map->odm_mirror_cnt + 1))) {
 265                 printk(KERN_ERR "Data Map wrong, num_comps=%u mirrors=%u\n",
 266                           data_map->odm_num_comps, data_map->odm_mirror_cnt);
 267                 return -EINVAL;
 268         }
 269
 270         if (data_map->odm_group_width)
 271                 group_width = data_map->odm_group_width;
 272         else
 273                 group_width = data_map->odm_num_comps /
 274                                                 (data_map->odm_mirror_cnt + 1);
 275
 276         stripe_length = (u64)data_map->odm_stripe_unit * group_width;
 277         if (stripe_length >= (1ULL << 32)) {
 278                 printk(KERN_ERR "Total Stripe length(0x%llx)"
 279                           " >= 32bit is not supported\n", _LLU(stripe_length));
 280                 return -ENOTSUPP;
 281         }
 282
 283         if (0 != (data_map->odm_stripe_unit & ~PAGE_MASK)) {
 284                 printk(KERN_ERR "Stripe Unit(0x%llx)"
 285                           " must be Multples of PAGE_SIZE(0x%lx)\n",
 286                           _LLU(data_map->odm_stripe_unit), PAGE_SIZE);
 287                 return -ENOTSUPP;
 288         }
 289
 290         return 0;
 291 }
 292
 293 static void copy_single_comp(struct pnfs_osd_object_cred *cur_comp,
 294                              struct pnfs_osd_object_cred *src_comp,
 295                              struct caps_buffers *caps_p)
 296 {
 297         WARN_ON(src_comp->oc_cap_key.cred_len > sizeof(caps_p->caps_key));
 298         WARN_ON(src_comp->oc_cap.cred_len > sizeof(caps_p->creds));
 299
 300         *cur_comp = *src_comp;
 301
 302         memcpy(caps_p->caps_key, src_comp->oc_cap_key.cred,
 303                sizeof(caps_p->caps_key));
 304         cur_comp->oc_cap_key.cred = caps_p->caps_key;
 305
 306         memcpy(caps_p->creds, src_comp->oc_cap.cred,
 307                sizeof(caps_p->creds));
 308         cur_comp->oc_cap.cred = caps_p->creds;
 309 }
 310
 311 int objio_alloc_lseg(struct pnfs_layout_segment **outp,
 312         struct pnfs_layout_hdr *pnfslay,
 313         struct pnfs_layout_range *range,
 314         struct xdr_stream *xdr,
 315         gfp_t gfp_flags)
 316 {
 317         struct objio_segment *objio_seg;
 318         struct pnfs_osd_xdr_decode_layout_iter iter;
 319         struct pnfs_osd_layout layout;
 320         struct pnfs_osd_object_cred *cur_comp, src_comp;
 321         struct caps_buffers *caps_p;
 322         int err;
 323
 324         err = pnfs_osd_xdr_decode_layout_map(&layout, &iter, xdr);
 325         if (unlikely(err))
 326                 return err;
 327
 328         err = _verify_data_map(&layout);
 329         if (unlikely(err))
 330                 return err;
 331
 332         objio_seg = kzalloc(sizeof(*objio_seg) +
 333                             sizeof(objio_seg->ods[0]) * layout.olo_num_comps +
 334                             sizeof(*objio_seg->comps) * layout.olo_num_comps +
 335                             sizeof(struct caps_buffers) * layout.olo_num_comps,
 336                             gfp_flags);
 337         if (!objio_seg)
 338                 return -ENOMEM;
 339
 340         objio_seg->comps = (void *)(objio_seg->ods + layout.olo_num_comps);
 341         cur_comp = objio_seg->comps;
 342         caps_p = (void *)(cur_comp + layout.olo_num_comps);
 343         while (pnfs_osd_xdr_decode_layout_comp(&src_comp, &iter, xdr, &err))
 344                 copy_single_comp(cur_comp++, &src_comp, caps_p++);
 345         if (unlikely(err))
 346                 goto err;
 347
 348         objio_seg->num_comps = layout.olo_num_comps;
 349         objio_seg->comps_index = layout.olo_comps_index;
 350         err = objio_devices_lookup(pnfslay, objio_seg, gfp_flags);
 351         if (err)
 352                 goto err;
 353
 354         objio_seg->mirrors_p1 = layout.olo_map.odm_mirror_cnt + 1;
 355         objio_seg->stripe_unit = layout.olo_map.odm_stripe_unit;
 356         if (layout.olo_map.odm_group_width) {
 357                 objio_seg->group_width = layout.olo_map.odm_group_width;
 358                 objio_seg->group_depth = layout.olo_map.odm_group_depth;
 359                 objio_seg->group_count = layout.olo_map.odm_num_comps /
 360                                                 objio_seg->mirrors_p1 /
 361                                                 objio_seg->group_width;
 362         } else {
 363                 objio_seg->group_width = layout.olo_map.odm_num_comps /
 364                                                 objio_seg->mirrors_p1;
 365                 objio_seg->group_depth = -1;
 366                 objio_seg->group_count = 1;
 367         }
 368
 369         /* Cache this calculation it will hit for every page */
 370         objio_seg->max_io_size = (BIO_MAX_PAGES_KMALLOC * PAGE_SIZE -
 371                                   objio_seg->stripe_unit) *
 372                                  objio_seg->group_width;
 373
 374         *outp = &objio_seg->lseg;
 375         return 0;
 376
 377 err:
 378         kfree(objio_seg);
 379         dprintk("%s: Error: return %d\n", __func__, err);
 380         *outp = NULL;
 381         return err;
 382 }
 383
 384 void objio_free_lseg(struct pnfs_layout_segment *lseg)
 385 {
 386         int i;
 387         struct objio_segment *objio_seg = OBJIO_LSEG(lseg);
 388
 389         for (i = 0; i < objio_seg->num_comps; i++) {
 390                 if (!objio_seg->ods[i])
 391                         break;
 392                 nfs4_put_deviceid_node(&objio_seg->ods[i]->id_node);
 393         }
 394         kfree(objio_seg);
 395 }
 396
 397 int objio_alloc_io_state(struct pnfs_layout_segment *lseg,
 398                          struct objlayout_io_state **outp,
 399                          gfp_t gfp_flags)
 400 {
 401         struct objio_segment *objio_seg = OBJIO_LSEG(lseg);
 402         struct objio_state *ios;
 403         const unsigned first_size = sizeof(*ios) +
 404                                 objio_seg->num_comps * sizeof(ios->per_dev[0]);
 405         const unsigned sec_size = objio_seg->num_comps *
 406                                                 sizeof(ios->ol_state.ioerrs[0]);
 407
 408         ios = kzalloc(first_size + sec_size, gfp_flags);
 409         if (unlikely(!ios))
 410                 return -ENOMEM;
 411
 412         ios->layout = objio_seg;
 413         ios->ol_state.ioerrs = ((void *)ios) + first_size;
 414         ios->ol_state.num_comps = objio_seg->num_comps;
 415
 416         *outp = &ios->ol_state;
 417         return 0;
 418 }
 419
 420 void objio_free_io_state(struct objlayout_io_state *ol_state)
 421 {
 422         struct objio_state *ios = container_of(ol_state, struct objio_state,
 423                                                ol_state);
 424
 425         kfree(ios);
 426 }
 427
 428 enum pnfs_osd_errno osd_pri_2_pnfs_err(enum osd_err_priority oep)
 429 {
 430         switch (oep) {
 431         case OSD_ERR_PRI_NO_ERROR:
 432                 return (enum pnfs_osd_errno)0;
 433
 434         case OSD_ERR_PRI_CLEAR_PAGES:
 435                 BUG_ON(1);
 436                 return 0;
 437
 438         case OSD_ERR_PRI_RESOURCE:
 439                 return PNFS_OSD_ERR_RESOURCE;
 440         case OSD_ERR_PRI_BAD_CRED:
 441                 return PNFS_OSD_ERR_BAD_CRED;
 442         case OSD_ERR_PRI_NO_ACCESS:
 443                 return PNFS_OSD_ERR_NO_ACCESS;
 444         case OSD_ERR_PRI_UNREACHABLE:
 445                 return PNFS_OSD_ERR_UNREACHABLE;
 446         case OSD_ERR_PRI_NOT_FOUND:
 447                 return PNFS_OSD_ERR_NOT_FOUND;
 448         case OSD_ERR_PRI_NO_SPACE:
 449                 return PNFS_OSD_ERR_NO_SPACE;
 450         default:
 451                 WARN_ON(1);
 452                 /* fallthrough */
 453         case OSD_ERR_PRI_EIO:
 454                 return PNFS_OSD_ERR_EIO;
 455         }
 456 }
 457
 458 static void _clear_bio(struct bio *bio)
 459 {
 460         struct bio_vec *bv;
 461         unsigned i;
 462
 463         __bio_for_each_segment(bv, bio, i, 0) {
 464                 unsigned this_count = bv->bv_len;
 465
 466                 if (likely(PAGE_SIZE == this_count))
 467                         clear_highpage(bv->bv_page);
 468                 else
 469                         zero_user(bv->bv_page, bv->bv_offset, this_count);
 470         }
 471 }
 472
 473 static int _io_check(struct objio_state *ios, bool is_write)
 474 {
 475         enum osd_err_priority oep = OSD_ERR_PRI_NO_ERROR;
 476         int lin_ret = 0;
 477         int i;
 478
 479         for (i = 0; i <  ios->numdevs; i++) {
 480                 struct osd_sense_info osi;
 481                 struct osd_request *or = ios->per_dev[i].or;
 482                 int ret;
 483
 484                 if (!or)
 485                         continue;
 486
 487                 ret = osd_req_decode_sense(or, &osi);
 488                 if (likely(!ret))
 489                         continue;
 490
 491                 if (OSD_ERR_PRI_CLEAR_PAGES == osi.osd_err_pri) {
 492                         /* start read offset passed endof file */
 493                         BUG_ON(is_write);
 494                         _clear_bio(ios->per_dev[i].bio);
 495                         dprintk("%s: start read offset passed end of file "
 496                                 "offset=0x%llx, length=0x%lx\n", __func__,
 497                                 _LLU(ios->per_dev[i].offset),
 498                                 ios->per_dev[i].length);
 499
 500                         continue; /* we recovered */
 501                 }
 502                 objlayout_io_set_result(&ios->ol_state, i,
 503                                         &ios->layout->comps[i].oc_object_id,
 504                                         osd_pri_2_pnfs_err(osi.osd_err_pri),
 505                                         ios->per_dev[i].offset,
 506                                         ios->per_dev[i].length,
 507                                         is_write);
 508
 509                 if (osi.osd_err_pri >= oep) {
 510                         oep = osi.osd_err_pri;
 511                         lin_ret = ret;
 512                 }
 513         }
 514
 515         return lin_ret;
 516 }
 517
 518 /*
 519  * Common IO state helpers.
 520  */
 521 static void _io_free(struct objio_state *ios)
 522 {
 523         unsigned i;
 524
 525         for (i = 0; i < ios->numdevs; i++) {
 526                 struct _objio_per_comp *per_dev = &ios->per_dev[i];
 527
 528                 if (per_dev->or) {
 529                         osd_end_request(per_dev->or);
 530                         per_dev->or = NULL;
 531                 }
 532
 533                 if (per_dev->bio) {
 534                         bio_put(per_dev->bio);
 535                         per_dev->bio = NULL;
 536                 }
 537         }
 538 }
 539
 540 struct osd_dev *_io_od(struct objio_state *ios, unsigned dev)
 541 {
 542         unsigned min_dev = ios->layout->comps_index;
 543         unsigned max_dev = min_dev + ios->layout->num_comps;
 544
 545         BUG_ON(dev < min_dev || max_dev <= dev);
 546         return ios->layout->ods[dev - min_dev]->od;
 547 }
 548
 549 struct _striping_info {
 550         u64 obj_offset;
 551         u64 group_length;
 552         unsigned dev;
 553         unsigned unit_off;
 554 };
 555
 556 static void _calc_stripe_info(struct objio_state *ios, u64 file_offset,
 557                               struct _striping_info *si)
 558 {
 559         u32     stripe_unit = ios->layout->stripe_unit;
 560         u32     group_width = ios->layout->group_width;
 561         u64     group_depth = ios->layout->group_depth;
 562         u32     U = stripe_unit * group_width;
 563
 564         u64     T = U * group_depth;
 565         u64     S = T * ios->layout->group_count;
 566         u64     M = div64_u64(file_offset, S);
 567
 568         /*
 569         G = (L - (M * S)) / T
 570         H = (L - (M * S)) % T
 571         */
 572         u64     LmodU = file_offset - M * S;
 573         u32     G = div64_u64(LmodU, T);
 574         u64     H = LmodU - G * T;
 575
 576         u32     N = div_u64(H, U);
 577
 578         div_u64_rem(file_offset, stripe_unit, &si->unit_off);
 579         si->obj_offset = si->unit_off + (N * stripe_unit) +
 580                                   (M * group_depth * stripe_unit);
 581
 582         /* "H - (N * U)" is just "H % U" so it's bound to u32 */
 583         si->dev = (u32)(H - (N * U)) / stripe_unit + G * group_width;
 584         si->dev *= ios->layout->mirrors_p1;
 585
 586         si->group_length = T - H;
 587 }
 588
 589 static int _add_stripe_unit(struct objio_state *ios,  unsigned *cur_pg,
 590                 unsigned pgbase, struct _objio_per_comp *per_dev, int len,
 591                 gfp_t gfp_flags)
 592 {
 593         unsigned pg = *cur_pg;
 594         int cur_len = len;
 595         struct request_queue *q =
 596                         osd_request_queue(_io_od(ios, per_dev->dev));
 597
 598         if (per_dev->bio == NULL) {
 599                 unsigned pages_in_stripe = ios->layout->group_width *
 600                                       (ios->layout->stripe_unit / PAGE_SIZE);
 601                 unsigned bio_size = (ios->ol_state.nr_pages + pages_in_stripe) /
 602                                     ios->layout->group_width;
 603
 604                 if (BIO_MAX_PAGES_KMALLOC < bio_size)
 605                         bio_size = BIO_MAX_PAGES_KMALLOC;
 606
 607                 per_dev->bio = bio_kmalloc(gfp_flags, bio_size);
 608                 if (unlikely(!per_dev->bio)) {
 609                         dprintk("Faild to allocate BIO size=%u\n", bio_size);
 610                         return -ENOMEM;
 611                 }
 612         }
 613
 614         while (cur_len > 0) {
 615                 unsigned pglen = min_t(unsigned, PAGE_SIZE - pgbase, cur_len);
 616                 unsigned added_len;
 617
 618                 BUG_ON(ios->ol_state.nr_pages <= pg);
 619                 cur_len -= pglen;
 620
 621                 added_len = bio_add_pc_page(q, per_dev->bio,
 622                                         ios->ol_state.pages[pg], pglen, pgbase);
 623                 if (unlikely(pglen != added_len))
 624                         return -ENOMEM;
 625                 pgbase = 0;
 626                 ++pg;
 627         }
 628         BUG_ON(cur_len);
 629
 630         per_dev->length += len;
 631         *cur_pg = pg;
 632         return 0;
 633 }
 634
 635 static int _prepare_one_group(struct objio_state *ios, u64 length,
 636                               struct _striping_info *si, unsigned *last_pg,
 637                               gfp_t gfp_flags)
 638 {
 639         unsigned stripe_unit = ios->layout->stripe_unit;
 640         unsigned mirrors_p1 = ios->layout->mirrors_p1;
 641         unsigned devs_in_group = ios->layout->group_width * mirrors_p1;
 642         unsigned dev = si->dev;
 643         unsigned first_dev = dev - (dev % devs_in_group);
 644         unsigned max_comp = ios->numdevs ? ios->numdevs - mirrors_p1 : 0;
 645         unsigned cur_pg = *last_pg;
 646         int ret = 0;
 647
 648         while (length) {
 649                 struct _objio_per_comp *per_dev = &ios->per_dev[dev - first_dev];
 650                 unsigned cur_len, page_off = 0;
 651
 652                 if (!per_dev->length) {
 653                         per_dev->dev = dev;
 654                         if (dev < si->dev) {
 655                                 per_dev->offset = si->obj_offset + stripe_unit -
 656                                                                    si->unit_off;
 657                                 cur_len = stripe_unit;
 658                         } else if (dev == si->dev) {
 659                                 per_dev->offset = si->obj_offset;
 660                                 cur_len = stripe_unit - si->unit_off;
 661                                 page_off = si->unit_off & ~PAGE_MASK;
 662                                 BUG_ON(page_off &&
 663                                       (page_off != ios->ol_state.pgbase));
 664                         } else { /* dev > si->dev */
 665                                 per_dev->offset = si->obj_offset - si->unit_off;
 666                                 cur_len = stripe_unit;
 667                         }
 668
 669                         if (max_comp < dev - first_dev)
 670                                 max_comp = dev - first_dev;
 671                 } else {
 672                         cur_len = stripe_unit;
 673                 }
 674                 if (cur_len >= length)
 675                         cur_len = length;
 676
 677                 ret = _add_stripe_unit(ios, &cur_pg, page_off , per_dev,
 678                                        cur_len, gfp_flags);
 679                 if (unlikely(ret))
 680                         goto out;
 681
 682                 dev += mirrors_p1;
 683                 dev = (dev % devs_in_group) + first_dev;
 684
 685                 length -= cur_len;
 686                 ios->length += cur_len;
 687         }
 688 out:
 689         ios->numdevs = max_comp + mirrors_p1;
 690         *last_pg = cur_pg;
 691         return ret;
 692 }
 693
 694 static int _io_rw_pagelist(struct objio_state *ios, gfp_t gfp_flags)
 695 {
 696         u64 length = ios->ol_state.count;
 697         u64 offset = ios->ol_state.offset;
 698         struct _striping_info si;
 699         unsigned last_pg = 0;
 700         int ret = 0;
 701
 702         while (length) {
 703                 _calc_stripe_info(ios, offset, &si);
 704
 705                 if (length < si.group_length)
 706                         si.group_length = length;
 707
 708                 ret = _prepare_one_group(ios, si.group_length, &si, &last_pg, gfp_flags);
 709                 if (unlikely(ret))
 710                         goto out;
 711
 712                 offset += si.group_length;
 713                 length -= si.group_length;
 714         }
 715
 716 out:
 717         if (!ios->length)
 718                 return ret;
 719
 720         return 0;
 721 }
 722
 723 static ssize_t _sync_done(struct objio_state *ios)
 724 {
 725         struct completion *waiting = ios->private;
 726
 727         complete(waiting);
 728         return 0;
 729 }
 730
 731 static void _last_io(struct kref *kref)
 732 {
 733         struct objio_state *ios = container_of(kref, struct objio_state, kref);
 734
 735         ios->done(ios);
 736 }
 737
 738 static void _done_io(struct osd_request *or, void *p)
 739 {
 740         struct objio_state *ios = p;
 741
 742         kref_put(&ios->kref, _last_io);
 743 }
 744
 745 static ssize_t _io_exec(struct objio_state *ios)
 746 {
 747         DECLARE_COMPLETION_ONSTACK(wait);
 748         ssize_t status = 0; /* sync status */
 749         unsigned i;
 750         objio_done_fn saved_done_fn = ios->done;
 751         bool sync = ios->ol_state.sync;
 752
 753         if (sync) {
 754                 ios->done = _sync_done;
 755                 ios->private = &wait;
 756         }
 757
 758         kref_init(&ios->kref);
 759
 760         for (i = 0; i < ios->numdevs; i++) {
 761                 struct osd_request *or = ios->per_dev[i].or;
 762
 763                 if (!or)
 764                         continue;
 765
 766                 kref_get(&ios->kref);
 767                 osd_execute_request_async(or, _done_io, ios);
 768         }
 769
 770         kref_put(&ios->kref, _last_io);
 771
 772         if (sync) {
 773                 wait_for_completion(&wait);
 774                 status = saved_done_fn(ios);
 775         }
 776
 777         return status;
 778 }
 779
 780 /*
 781  * read
 782  */
 783 static ssize_t _read_done(struct objio_state *ios)
 784 {
 785         ssize_t status;
 786         int ret = _io_check(ios, false);
 787
 788         _io_free(ios);
 789
 790         if (likely(!ret))
 791                 status = ios->length;
 792         else
 793                 status = ret;
 794
 795         objlayout_read_done(&ios->ol_state, status, ios->ol_state.sync);
 796         return status;
 797 }
 798
 799 static int _read_mirrors(struct objio_state *ios, unsigned cur_comp)
 800 {
 801         struct osd_request *or = NULL;
 802         struct _objio_per_comp *per_dev = &ios->per_dev[cur_comp];
 803         unsigned dev = per_dev->dev;
 804         struct pnfs_osd_object_cred *cred =
 805                         &ios->layout->comps[cur_comp];
 806         struct osd_obj_id obj = {
 807                 .partition = cred->oc_object_id.oid_partition_id,
 808                 .id = cred->oc_object_id.oid_object_id,
 809         };
 810         int ret;
 811
 812         or = osd_start_request(_io_od(ios, dev), GFP_KERNEL);
 813         if (unlikely(!or)) {
 814                 ret = -ENOMEM;
 815                 goto err;
 816         }
 817         per_dev->or = or;
 818
 819         osd_req_read(or, &obj, per_dev->offset, per_dev->bio, per_dev->length);
 820
 821         ret = osd_finalize_request(or, 0, cred->oc_cap.cred, NULL);
 822         if (ret) {
 823                 dprintk("%s: Faild to osd_finalize_request() => %d\n",
 824                         __func__, ret);
 825                 goto err;
 826         }
 827
 828         dprintk("%s:[%d] dev=%d obj=0x%llx start=0x%llx length=0x%lx\n",
 829                 __func__, cur_comp, dev, obj.id, _LLU(per_dev->offset),
 830                 per_dev->length);
 831
 832 err:
 833         return ret;
 834 }
 835
 836 static ssize_t _read_exec(struct objio_state *ios)
 837 {
 838         unsigned i;
 839         int ret;
 840
 841         for (i = 0; i < ios->numdevs; i += ios->layout->mirrors_p1) {
 842                 if (!ios->per_dev[i].length)
 843                         continue;
 844                 ret = _read_mirrors(ios, i);
 845                 if (unlikely(ret))
 846                         goto err;
 847         }
 848
 849         ios->done = _read_done;
 850         return _io_exec(ios); /* In sync mode exec returns the io status */
 851
 852 err:
 853         _io_free(ios);
 854         return ret;
 855 }
 856
 857 ssize_t objio_read_pagelist(struct objlayout_io_state *ol_state)
 858 {
 859         struct objio_state *ios = container_of(ol_state, struct objio_state,
 860                                                ol_state);
 861         int ret;
 862
 863         ret = _io_rw_pagelist(ios, GFP_KERNEL);
 864         if (unlikely(ret))
 865                 return ret;
 866
 867         return _read_exec(ios);
 868 }
 869
 870 /*
 871  * write
 872  */
 873 static ssize_t _write_done(struct objio_state *ios)
 874 {
 875         ssize_t status;
 876         int ret = _io_check(ios, true);
 877
 878         _io_free(ios);
 879
 880         if (likely(!ret)) {
 881                 /* FIXME: should be based on the OSD's persistence model
 882                  * See OSD2r05 Section 4.13 Data persistence model */
 883                 ios->ol_state.committed = NFS_FILE_SYNC;
 884                 status = ios->length;
 885         } else {
 886                 status = ret;
 887         }
 888
 889         objlayout_write_done(&ios->ol_state, status, ios->ol_state.sync);
 890         return status;
 891 }
 892
 893 static int _write_mirrors(struct objio_state *ios, unsigned cur_comp)
 894 {
 895         struct _objio_per_comp *master_dev = &ios->per_dev[cur_comp];
 896         unsigned dev = ios->per_dev[cur_comp].dev;
 897         unsigned last_comp = cur_comp + ios->layout->mirrors_p1;
 898         int ret;
 899
 900         for (; cur_comp < last_comp; ++cur_comp, ++dev) {
 901                 struct osd_request *or = NULL;
 902                 struct pnfs_osd_object_cred *cred =
 903                                         &ios->layout->comps[cur_comp];
 904                 struct osd_obj_id obj = {
 905                         .partition = cred->oc_object_id.oid_partition_id,
 906                         .id = cred->oc_object_id.oid_object_id,
 907                 };
 908                 struct _objio_per_comp *per_dev = &ios->per_dev[cur_comp];
 909                 struct bio *bio;
 910
 911                 or = osd_start_request(_io_od(ios, dev), GFP_NOFS);
 912                 if (unlikely(!or)) {
 913                         ret = -ENOMEM;
 914                         goto err;
 915                 }
 916                 per_dev->or = or;
 917
 918                 if (per_dev != master_dev) {
 919                         bio = bio_kmalloc(GFP_NOFS,
 920                                           master_dev->bio->bi_max_vecs);
 921                         if (unlikely(!bio)) {
 922                                 dprintk("Faild to allocate BIO size=%u\n",
 923                                         master_dev->bio->bi_max_vecs);
 924                                 ret = -ENOMEM;
 925                                 goto err;
 926                         }
 927
 928                         __bio_clone(bio, master_dev->bio);
 929                         bio->bi_bdev = NULL;
 930                         bio->bi_next = NULL;
 931                         per_dev->bio = bio;
 932                         per_dev->dev = dev;
 933                         per_dev->length = master_dev->length;
 934                         per_dev->offset =  master_dev->offset;
 935                 } else {
 936                         bio = master_dev->bio;
 937                         bio->bi_rw |= REQ_WRITE;
 938                 }
 939
 940                 osd_req_write(or, &obj, per_dev->offset, bio, per_dev->length);
 941
 942                 ret = osd_finalize_request(or, 0, cred->oc_cap.cred, NULL);
 943                 if (ret) {
 944                         dprintk("%s: Faild to osd_finalize_request() => %d\n",
 945                                 __func__, ret);
 946                         goto err;
 947                 }
 948
 949                 dprintk("%s:[%d] dev=%d obj=0x%llx start=0x%llx length=0x%lx\n",
 950                         __func__, cur_comp, dev, obj.id, _LLU(per_dev->offset),
 951                         per_dev->length);
 952         }
 953
 954 err:
 955         return ret;
 956 }
 957
 958 static ssize_t _write_exec(struct objio_state *ios)
 959 {
 960         unsigned i;
 961         int ret;
 962
 963         for (i = 0; i < ios->numdevs; i += ios->layout->mirrors_p1) {
 964                 if (!ios->per_dev[i].length)
 965                         continue;
 966                 ret = _write_mirrors(ios, i);
 967                 if (unlikely(ret))
 968                         goto err;
 969         }
 970
 971         ios->done = _write_done;
 972         return _io_exec(ios); /* In sync mode exec returns the io->status */
 973
 974 err:
 975         _io_free(ios);
 976         return ret;
 977 }
 978
 979 ssize_t objio_write_pagelist(struct objlayout_io_state *ol_state, bool stable)
 980 {
 981         struct objio_state *ios = container_of(ol_state, struct objio_state,
 982                                                ol_state);
 983         int ret;
 984
 985         /* TODO: ios->stable = stable; */
 986         ret = _io_rw_pagelist(ios, GFP_NOFS);
 987         if (unlikely(ret))
 988                 return ret;
 989
 990         return _write_exec(ios);
 991 }
 992
 993 static bool objio_pg_test(struct nfs_pageio_descriptor *pgio,
 994                           struct nfs_page *prev, struct nfs_page *req)
 995 {
 996         if (!pnfs_generic_pg_test(pgio, prev, req))
 997                 return false;
 998
 999         return pgio->pg_count + req->wb_bytes <=
1000                         OBJIO_LSEG(pgio->pg_lseg)->max_io_size;
1001 }
1002
1003 static const struct nfs_pageio_ops objio_pg_read_ops = {
1004         .pg_init = pnfs_generic_pg_init_read,
1005         .pg_test = objio_pg_test,
1006         .pg_doio = pnfs_generic_pg_readpages,
1007 };
1008
1009 static const struct nfs_pageio_ops objio_pg_write_ops = {
1010         .pg_init = pnfs_generic_pg_init_write,
1011         .pg_test = objio_pg_test,
1012         .pg_doio = pnfs_generic_pg_writepages,
1013 };
1014
1015 static struct pnfs_layoutdriver_type objlayout_type = {
1016         .id = LAYOUT_OSD2_OBJECTS,
1017         .name = "LAYOUT_OSD2_OBJECTS",
1018         .flags                   = PNFS_LAYOUTRET_ON_SETATTR,
1019
1020         .alloc_layout_hdr        = objlayout_alloc_layout_hdr,
1021         .free_layout_hdr         = objlayout_free_layout_hdr,
1022
1023         .alloc_lseg              = objlayout_alloc_lseg,
1024         .free_lseg               = objlayout_free_lseg,
1025
1026         .read_pagelist           = objlayout_read_pagelist,
1027         .write_pagelist          = objlayout_write_pagelist,
1028         .pg_read_ops             = &objio_pg_read_ops,
1029         .pg_write_ops            = &objio_pg_write_ops,
1030
1031         .free_deviceid_node      = objio_free_deviceid_node,
1032
1033         .encode_layoutcommit     = objlayout_encode_layoutcommit,
1034         .encode_layoutreturn     = objlayout_encode_layoutreturn,
1035 };
1036
1037 MODULE_DESCRIPTION("pNFS Layout Driver for OSD2 objects");
1038 MODULE_AUTHOR("Benny Halevy <bhalevy@panasas.com>");
1039 MODULE_LICENSE("GPL");
1040
1041 static int __init
1042 objlayout_init(void)
1043 {
1044         int ret = pnfs_register_layoutdriver(&objlayout_type);
1045
1046         if (ret)
1047                 printk(KERN_INFO
1048                         "%s: Registering OSD pNFS Layout Driver failed: error=%d\n",
1049                         __func__, ret);
1050         else
1051                 printk(KERN_INFO "%s: Registered OSD pNFS Layout Driver\n",
1052                         __func__);
1053         return ret;
1054 }
1055
1056 static void __exit
1057 objlayout_exit(void)
1058 {
1059         pnfs_unregister_layoutdriver(&objlayout_type);
1060         printk(KERN_INFO "%s: Unregistered OSD pNFS Layout Driver\n",
1061                __func__);
1062 }
1063
1064 MODULE_ALIAS("nfs-layouttype4-2");
1065
1066 module_init(objlayout_init);
1067 module_exit(objlayout_exit);