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_ore.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 ore_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.od);
  64         osduld_put_device(de->od.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 = 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.od);
 107                 objio_free_deviceid_node(&de->id_node);
 108                 de = n;
 109         }
 110
 111         return de;
 112 }
 113
 114 struct objio_segment {
 115         struct pnfs_layout_segment lseg;
 116
 117         struct ore_layout layout;
 118         struct ore_components oc;
 119 };
 120
 121 static inline struct objio_segment *
 122 OBJIO_LSEG(struct pnfs_layout_segment *lseg)
 123 {
 124         return container_of(lseg, struct objio_segment, lseg);
 125 }
 126
 127 struct objio_state;
 128 typedef int (*objio_done_fn)(struct objio_state *ios);
 129
 130 struct objio_state {
 131         /* Generic layer */
 132         struct objlayout_io_res oir;
 133
 134         struct page **pages;
 135         unsigned pgbase;
 136         unsigned nr_pages;
 137         unsigned long count;
 138         loff_t offset;
 139         bool sync;
 140
 141         struct ore_layout *layout;
 142         struct ore_components *oc;
 143
 144         struct kref kref;
 145         objio_done_fn done;
 146         void *private;
 147
 148         unsigned long length;
 149         unsigned numdevs; /* Actually used devs in this IO */
 150         /* A per-device variable array of size numdevs */
 151         struct _objio_per_comp {
 152                 struct bio *bio;
 153                 struct osd_request *or;
 154                 unsigned long length;
 155                 u64 offset;
 156                 unsigned dev;
 157         } per_dev[];
 158 };
 159
 160 /* Send and wait for a get_device_info of devices in the layout,
 161    then look them up with the osd_initiator library */
 162 static int objio_devices_lookup(struct pnfs_layout_hdr *pnfslay,
 163         struct objio_segment *objio_seg, unsigned c, struct nfs4_deviceid *d_id,
 164         gfp_t gfp_flags)
 165 {
 166         struct pnfs_osd_deviceaddr *deviceaddr;
 167         struct objio_dev_ent *ode;
 168         struct osd_dev *od;
 169         struct osd_dev_info odi;
 170         int err;
 171
 172         ode = _dev_list_find(NFS_SERVER(pnfslay->plh_inode), d_id);
 173         if (ode) {
 174                 objio_seg->oc.ods[c] = &ode->od; /* must use container_of */
 175                 return 0;
 176         }
 177
 178         err = objlayout_get_deviceinfo(pnfslay, d_id, &deviceaddr, gfp_flags);
 179         if (unlikely(err)) {
 180                 dprintk("%s: objlayout_get_deviceinfo dev(%llx:%llx) =>%d\n",
 181                         __func__, _DEVID_LO(d_id), _DEVID_HI(d_id), err);
 182                 return err;
 183         }
 184
 185         odi.systemid_len = deviceaddr->oda_systemid.len;
 186         if (odi.systemid_len > sizeof(odi.systemid)) {
 187                 dprintk("%s: odi.systemid_len > sizeof(systemid=%zd)\n",
 188                         __func__, sizeof(odi.systemid));
 189                 err = -EINVAL;
 190                 goto out;
 191         } else if (odi.systemid_len)
 192                 memcpy(odi.systemid, deviceaddr->oda_systemid.data,
 193                        odi.systemid_len);
 194         odi.osdname_len  = deviceaddr->oda_osdname.len;
 195         odi.osdname      = (u8 *)deviceaddr->oda_osdname.data;
 196
 197         if (!odi.osdname_len && !odi.systemid_len) {
 198                 dprintk("%s: !odi.osdname_len && !odi.systemid_len\n",
 199                         __func__);
 200                 err = -ENODEV;
 201                 goto out;
 202         }
 203
 204         od = osduld_info_lookup(&odi);
 205         if (unlikely(IS_ERR(od))) {
 206                 err = PTR_ERR(od);
 207                 dprintk("%s: osduld_info_lookup => %d\n", __func__, err);
 208                 goto out;
 209         }
 210
 211         ode = _dev_list_add(NFS_SERVER(pnfslay->plh_inode), d_id, od,
 212                             gfp_flags);
 213         objio_seg->oc.ods[c] = &ode->od; /* must use container_of */
 214         dprintk("Adding new dev_id(%llx:%llx)\n",
 215                 _DEVID_LO(d_id), _DEVID_HI(d_id));
 216 out:
 217         objlayout_put_deviceinfo(deviceaddr);
 218         return err;
 219 }
 220
 221 #if 0
 222 static int _verify_data_map(struct pnfs_osd_layout *layout)
 223 {
 224         struct pnfs_osd_data_map *data_map = &layout->olo_map;
 225         u64 stripe_length;
 226         u32 group_width;
 227
 228 /* FIXME: Only raid0 for now. if not go through MDS */
 229         if (data_map->odm_raid_algorithm != PNFS_OSD_RAID_0) {
 230                 printk(KERN_ERR "Only RAID_0 for now\n");
 231                 return -ENOTSUPP;
 232         }
 233         if (0 != (data_map->odm_num_comps % (data_map->odm_mirror_cnt + 1))) {
 234                 printk(KERN_ERR "Data Map wrong, num_comps=%u mirrors=%u\n",
 235                           data_map->odm_num_comps, data_map->odm_mirror_cnt);
 236                 return -EINVAL;
 237         }
 238
 239         if (data_map->odm_group_width)
 240                 group_width = data_map->odm_group_width;
 241         else
 242                 group_width = data_map->odm_num_comps /
 243                                                 (data_map->odm_mirror_cnt + 1);
 244
 245         stripe_length = (u64)data_map->odm_stripe_unit * group_width;
 246         if (stripe_length >= (1ULL << 32)) {
 247                 printk(KERN_ERR "Total Stripe length(0x%llx)"
 248                           " >= 32bit is not supported\n", _LLU(stripe_length));
 249                 return -ENOTSUPP;
 250         }
 251
 252         if (0 != (data_map->odm_stripe_unit & ~PAGE_MASK)) {
 253                 printk(KERN_ERR "Stripe Unit(0x%llx)"
 254                           " must be Multples of PAGE_SIZE(0x%lx)\n",
 255                           _LLU(data_map->odm_stripe_unit), PAGE_SIZE);
 256                 return -ENOTSUPP;
 257         }
 258
 259         return 0;
 260 }
 261 #endif
 262
 263 static void copy_single_comp(struct ore_components *oc, unsigned c,
 264                              struct pnfs_osd_object_cred *src_comp)
 265 {
 266         struct ore_comp *ocomp = &oc->comps[c];
 267
 268         WARN_ON(src_comp->oc_cap_key.cred_len > 0); /* libosd is NO_SEC only */
 269         WARN_ON(src_comp->oc_cap.cred_len > sizeof(ocomp->cred));
 270
 271         ocomp->obj.partition = src_comp->oc_object_id.oid_partition_id;
 272         ocomp->obj.id = src_comp->oc_object_id.oid_object_id;
 273
 274         memcpy(ocomp->cred, src_comp->oc_cap.cred, sizeof(ocomp->cred));
 275 }
 276
 277 int __alloc_objio_seg(unsigned numdevs, gfp_t gfp_flags,
 278                        struct objio_segment **pseg)
 279 {
 280         struct __alloc_objio_segment {
 281                 struct objio_segment olseg;
 282                 struct ore_dev *ods[numdevs];
 283                 struct ore_comp comps[numdevs];
 284         } *aolseg;
 285
 286         aolseg = kzalloc(sizeof(*aolseg), gfp_flags);
 287         if (unlikely(!aolseg)) {
 288                 dprintk("%s: Faild allocation numdevs=%d size=%zd\n", __func__,
 289                         numdevs, sizeof(*aolseg));
 290                 return -ENOMEM;
 291         }
 292
 293         aolseg->olseg.oc.numdevs = numdevs;
 294         aolseg->olseg.oc.single_comp = EC_MULTPLE_COMPS;
 295         aolseg->olseg.oc.comps = aolseg->comps;
 296         aolseg->olseg.oc.ods = aolseg->ods;
 297
 298         *pseg = &aolseg->olseg;
 299         return 0;
 300 }
 301
 302 int objio_alloc_lseg(struct pnfs_layout_segment **outp,
 303         struct pnfs_layout_hdr *pnfslay,
 304         struct pnfs_layout_range *range,
 305         struct xdr_stream *xdr,
 306         gfp_t gfp_flags)
 307 {
 308         struct objio_segment *objio_seg;
 309         struct pnfs_osd_xdr_decode_layout_iter iter;
 310         struct pnfs_osd_layout layout;
 311         struct pnfs_osd_object_cred src_comp;
 312         unsigned cur_comp;
 313         int err;
 314
 315         err = pnfs_osd_xdr_decode_layout_map(&layout, &iter, xdr);
 316         if (unlikely(err))
 317                 return err;
 318
 319         err = __alloc_objio_seg(layout.olo_num_comps, gfp_flags, &objio_seg);
 320         if (unlikely(err))
 321                 return err;
 322
 323         objio_seg->layout.stripe_unit = layout.olo_map.odm_stripe_unit;
 324         objio_seg->layout.group_width = layout.olo_map.odm_group_width;
 325         objio_seg->layout.group_depth = layout.olo_map.odm_group_depth;
 326         objio_seg->layout.mirrors_p1 = layout.olo_map.odm_mirror_cnt + 1;
 327         objio_seg->layout.raid_algorithm = layout.olo_map.odm_raid_algorithm;
 328
 329         err = ore_verify_layout(layout.olo_map.odm_num_comps,
 330                                           &objio_seg->layout);
 331         if (unlikely(err))
 332                 goto err;
 333
 334         objio_seg->oc.first_dev = layout.olo_comps_index;
 335         cur_comp = 0;
 336         while (pnfs_osd_xdr_decode_layout_comp(&src_comp, &iter, xdr, &err)) {
 337                 copy_single_comp(&objio_seg->oc, cur_comp, &src_comp);
 338                 err = objio_devices_lookup(pnfslay, objio_seg, cur_comp,
 339                                            &src_comp.oc_object_id.oid_device_id,
 340                                            gfp_flags);
 341                 if (err)
 342                         goto err;
 343                 ++cur_comp;
 344         }
 345         /* pnfs_osd_xdr_decode_layout_comp returns false on error */
 346         if (unlikely(err))
 347                 goto err;
 348
 349         *outp = &objio_seg->lseg;
 350         return 0;
 351
 352 err:
 353         kfree(objio_seg);
 354         dprintk("%s: Error: return %d\n", __func__, err);
 355         *outp = NULL;
 356         return err;
 357 }
 358
 359 void objio_free_lseg(struct pnfs_layout_segment *lseg)
 360 {
 361         int i;
 362         struct objio_segment *objio_seg = OBJIO_LSEG(lseg);
 363
 364         for (i = 0; i < objio_seg->oc.numdevs; i++) {
 365                 struct ore_dev *od = objio_seg->oc.ods[i];
 366                 struct objio_dev_ent *ode;
 367
 368                 if (!od)
 369                         break;
 370                 ode = container_of(od, typeof(*ode), od);
 371                 nfs4_put_deviceid_node(&ode->id_node);
 372         }
 373         kfree(objio_seg);
 374 }
 375
 376 static int
 377 objio_alloc_io_state(struct pnfs_layout_hdr *pnfs_layout_type,
 378         struct pnfs_layout_segment *lseg, struct page **pages, unsigned pgbase,
 379         loff_t offset, size_t count, void *rpcdata, gfp_t gfp_flags,
 380         struct objio_state **outp)
 381 {
 382         struct objio_segment *objio_seg = OBJIO_LSEG(lseg);
 383         struct objio_state *ios;
 384         struct __alloc_objio_state {
 385                 struct objio_state objios;
 386                 struct _objio_per_comp per_dev[objio_seg->oc.numdevs];
 387                 struct pnfs_osd_ioerr ioerrs[objio_seg->oc.numdevs];
 388         } *aos;
 389
 390         aos = kzalloc(sizeof(*aos), gfp_flags);
 391         if (unlikely(!aos))
 392                 return -ENOMEM;
 393
 394         ios = &aos->objios;
 395
 396         ios->layout = &objio_seg->layout;
 397         ios->oc = &objio_seg->oc;
 398         objlayout_init_ioerrs(&aos->objios.oir, objio_seg->oc.numdevs,
 399                         aos->ioerrs, rpcdata, pnfs_layout_type);
 400
 401         ios->pages = pages;
 402         ios->pgbase = pgbase;
 403         ios->nr_pages = (pgbase + count + PAGE_SIZE - 1) >> PAGE_SHIFT;
 404         ios->offset = offset;
 405         ios->count = count;
 406         ios->sync = 0;
 407         BUG_ON(ios->nr_pages > (pgbase + count + PAGE_SIZE - 1) >> PAGE_SHIFT);
 408
 409         *outp = ios;
 410         return 0;
 411 }
 412
 413 void objio_free_result(struct objlayout_io_res *oir)
 414 {
 415         struct objio_state *ios = container_of(oir, struct objio_state, oir);
 416
 417         kfree(ios);
 418 }
 419
 420 enum pnfs_osd_errno osd_pri_2_pnfs_err(enum osd_err_priority oep)
 421 {
 422         switch (oep) {
 423         case OSD_ERR_PRI_NO_ERROR:
 424                 return (enum pnfs_osd_errno)0;
 425
 426         case OSD_ERR_PRI_CLEAR_PAGES:
 427                 BUG_ON(1);
 428                 return 0;
 429
 430         case OSD_ERR_PRI_RESOURCE:
 431                 return PNFS_OSD_ERR_RESOURCE;
 432         case OSD_ERR_PRI_BAD_CRED:
 433                 return PNFS_OSD_ERR_BAD_CRED;
 434         case OSD_ERR_PRI_NO_ACCESS:
 435                 return PNFS_OSD_ERR_NO_ACCESS;
 436         case OSD_ERR_PRI_UNREACHABLE:
 437                 return PNFS_OSD_ERR_UNREACHABLE;
 438         case OSD_ERR_PRI_NOT_FOUND:
 439                 return PNFS_OSD_ERR_NOT_FOUND;
 440         case OSD_ERR_PRI_NO_SPACE:
 441                 return PNFS_OSD_ERR_NO_SPACE;
 442         default:
 443                 WARN_ON(1);
 444                 /* fallthrough */
 445         case OSD_ERR_PRI_EIO:
 446                 return PNFS_OSD_ERR_EIO;
 447         }
 448 }
 449
 450 static void __on_dev_error(struct objio_state *ios, bool is_write,
 451         struct ore_dev *od, unsigned dev_index, enum osd_err_priority oep,
 452         u64 dev_offset, u64  dev_len)
 453 {
 454         struct objio_state *objios = ios->private;
 455         struct pnfs_osd_objid pooid;
 456         struct objio_dev_ent *ode = container_of(od, typeof(*ode), od);
 457         /* FIXME: what to do with more-then-one-group layouts. We need to
 458          * translate from ore_io_state index to oc->comps index
 459          */
 460         unsigned comp = dev_index;
 461
 462         pooid.oid_device_id = ode->id_node.deviceid;
 463         pooid.oid_partition_id = ios->oc->comps[comp].obj.partition;
 464         pooid.oid_object_id = ios->oc->comps[comp].obj.id;
 465
 466         objlayout_io_set_result(&objios->oir, comp,
 467                                 &pooid, osd_pri_2_pnfs_err(oep),
 468                                 dev_offset, dev_len, is_write);
 469 }
 470
 471 static void _clear_bio(struct bio *bio)
 472 {
 473         struct bio_vec *bv;
 474         unsigned i;
 475
 476         __bio_for_each_segment(bv, bio, i, 0) {
 477                 unsigned this_count = bv->bv_len;
 478
 479                 if (likely(PAGE_SIZE == this_count))
 480                         clear_highpage(bv->bv_page);
 481                 else
 482                         zero_user(bv->bv_page, bv->bv_offset, this_count);
 483         }
 484 }
 485
 486 static int _io_check(struct objio_state *ios, bool is_write)
 487 {
 488         enum osd_err_priority oep = OSD_ERR_PRI_NO_ERROR;
 489         int lin_ret = 0;
 490         int i;
 491
 492         for (i = 0; i <  ios->numdevs; i++) {
 493                 struct osd_sense_info osi;
 494                 struct osd_request *or = ios->per_dev[i].or;
 495                 int ret;
 496
 497                 if (!or)
 498                         continue;
 499
 500                 ret = osd_req_decode_sense(or, &osi);
 501                 if (likely(!ret))
 502                         continue;
 503
 504                 if (OSD_ERR_PRI_CLEAR_PAGES == osi.osd_err_pri) {
 505                         /* start read offset passed endof file */
 506                         BUG_ON(is_write);
 507                         _clear_bio(ios->per_dev[i].bio);
 508                         dprintk("%s: start read offset passed end of file "
 509                                 "offset=0x%llx, length=0x%lx\n", __func__,
 510                                 _LLU(ios->per_dev[i].offset),
 511                                 ios->per_dev[i].length);
 512
 513                         continue; /* we recovered */
 514                 }
 515                 __on_dev_error(ios, is_write, ios->oc->ods[i],
 516                                 ios->per_dev[i].dev, osi.osd_err_pri,
 517                                 ios->per_dev[i].offset, ios->per_dev[i].length);
 518
 519                 if (osi.osd_err_pri >= oep) {
 520                         oep = osi.osd_err_pri;
 521                         lin_ret = ret;
 522                 }
 523         }
 524
 525         return lin_ret;
 526 }
 527
 528 /*
 529  * Common IO state helpers.
 530  */
 531 static void _io_free(struct objio_state *ios)
 532 {
 533         unsigned i;
 534
 535         for (i = 0; i < ios->numdevs; i++) {
 536                 struct _objio_per_comp *per_dev = &ios->per_dev[i];
 537
 538                 if (per_dev->or) {
 539                         osd_end_request(per_dev->or);
 540                         per_dev->or = NULL;
 541                 }
 542
 543                 if (per_dev->bio) {
 544                         bio_put(per_dev->bio);
 545                         per_dev->bio = NULL;
 546                 }
 547         }
 548 }
 549
 550 struct osd_dev *_io_od(struct objio_state *ios, unsigned dev)
 551 {
 552         unsigned min_dev = ios->oc->first_dev;
 553         unsigned max_dev = min_dev + ios->oc->numdevs;
 554
 555         BUG_ON(dev < min_dev || max_dev <= dev);
 556         return ios->oc->ods[dev - min_dev]->od;
 557 }
 558
 559 struct _striping_info {
 560         u64 obj_offset;
 561         u64 group_length;
 562         unsigned dev;
 563         unsigned unit_off;
 564 };
 565
 566 static void _calc_stripe_info(struct objio_state *ios, u64 file_offset,
 567                               struct _striping_info *si)
 568 {
 569         u32     stripe_unit = ios->layout->stripe_unit;
 570         u32     group_width = ios->layout->group_width;
 571         u64     group_depth = ios->layout->group_depth;
 572         u32     U = stripe_unit * group_width;
 573
 574         u64     T = U * group_depth;
 575         u64     S = T * ios->layout->group_count;
 576         u64     M = div64_u64(file_offset, S);
 577
 578         /*
 579         G = (L - (M * S)) / T
 580         H = (L - (M * S)) % T
 581         */
 582         u64     LmodU = file_offset - M * S;
 583         u32     G = div64_u64(LmodU, T);
 584         u64     H = LmodU - G * T;
 585
 586         u32     N = div_u64(H, U);
 587
 588         div_u64_rem(file_offset, stripe_unit, &si->unit_off);
 589         si->obj_offset = si->unit_off + (N * stripe_unit) +
 590                                   (M * group_depth * stripe_unit);
 591
 592         /* "H - (N * U)" is just "H % U" so it's bound to u32 */
 593         si->dev = (u32)(H - (N * U)) / stripe_unit + G * group_width;
 594         si->dev *= ios->layout->mirrors_p1;
 595
 596         si->group_length = T - H;
 597 }
 598
 599 static int _add_stripe_unit(struct objio_state *ios,  unsigned *cur_pg,
 600                 unsigned pgbase, struct _objio_per_comp *per_dev, int len,
 601                 gfp_t gfp_flags)
 602 {
 603         unsigned pg = *cur_pg;
 604         int cur_len = len;
 605         struct request_queue *q =
 606                         osd_request_queue(_io_od(ios, per_dev->dev));
 607
 608         if (per_dev->bio == NULL) {
 609                 unsigned pages_in_stripe = ios->layout->group_width *
 610                                       (ios->layout->stripe_unit / PAGE_SIZE);
 611                 unsigned bio_size = (ios->nr_pages + pages_in_stripe) /
 612                                     ios->layout->group_width;
 613
 614                 if (BIO_MAX_PAGES_KMALLOC < bio_size)
 615                         bio_size = BIO_MAX_PAGES_KMALLOC;
 616
 617                 per_dev->bio = bio_kmalloc(gfp_flags, bio_size);
 618                 if (unlikely(!per_dev->bio)) {
 619                         dprintk("Faild to allocate BIO size=%u\n", bio_size);
 620                         return -ENOMEM;
 621                 }
 622         }
 623
 624         while (cur_len > 0) {
 625                 unsigned pglen = min_t(unsigned, PAGE_SIZE - pgbase, cur_len);
 626                 unsigned added_len;
 627
 628                 BUG_ON(ios->nr_pages <= pg);
 629                 cur_len -= pglen;
 630
 631                 added_len = bio_add_pc_page(q, per_dev->bio,
 632                                         ios->pages[pg], pglen, pgbase);
 633                 if (unlikely(pglen != added_len))
 634                         return -ENOMEM;
 635                 pgbase = 0;
 636                 ++pg;
 637         }
 638         BUG_ON(cur_len);
 639
 640         per_dev->length += len;
 641         *cur_pg = pg;
 642         return 0;
 643 }
 644
 645 static int _prepare_one_group(struct objio_state *ios, u64 length,
 646                               struct _striping_info *si, unsigned *last_pg,
 647                               gfp_t gfp_flags)
 648 {
 649         unsigned stripe_unit = ios->layout->stripe_unit;
 650         unsigned mirrors_p1 = ios->layout->mirrors_p1;
 651         unsigned devs_in_group = ios->layout->group_width * mirrors_p1;
 652         unsigned dev = si->dev;
 653         unsigned first_dev = dev - (dev % devs_in_group);
 654         unsigned max_comp = ios->numdevs ? ios->numdevs - mirrors_p1 : 0;
 655         unsigned cur_pg = *last_pg;
 656         int ret = 0;
 657
 658         while (length) {
 659                 struct _objio_per_comp *per_dev = &ios->per_dev[dev - first_dev];
 660                 unsigned cur_len, page_off = 0;
 661
 662                 if (!per_dev->length) {
 663                         per_dev->dev = dev;
 664                         if (dev < si->dev) {
 665                                 per_dev->offset = si->obj_offset + stripe_unit -
 666                                                                    si->unit_off;
 667                                 cur_len = stripe_unit;
 668                         } else if (dev == si->dev) {
 669                                 per_dev->offset = si->obj_offset;
 670                                 cur_len = stripe_unit - si->unit_off;
 671                                 page_off = si->unit_off & ~PAGE_MASK;
 672                                 BUG_ON(page_off &&
 673                                       (page_off != ios->pgbase));
 674                         } else { /* dev > si->dev */
 675                                 per_dev->offset = si->obj_offset - si->unit_off;
 676                                 cur_len = stripe_unit;
 677                         }
 678
 679                         if (max_comp < dev - first_dev)
 680                                 max_comp = dev - first_dev;
 681                 } else {
 682                         cur_len = stripe_unit;
 683                 }
 684                 if (cur_len >= length)
 685                         cur_len = length;
 686
 687                 ret = _add_stripe_unit(ios, &cur_pg, page_off , per_dev,
 688                                        cur_len, gfp_flags);
 689                 if (unlikely(ret))
 690                         goto out;
 691
 692                 dev += mirrors_p1;
 693                 dev = (dev % devs_in_group) + first_dev;
 694
 695                 length -= cur_len;
 696                 ios->length += cur_len;
 697         }
 698 out:
 699         ios->numdevs = max_comp + mirrors_p1;
 700         *last_pg = cur_pg;
 701         return ret;
 702 }
 703
 704 static int _io_rw_pagelist(struct objio_state *ios, gfp_t gfp_flags)
 705 {
 706         u64 length = ios->count;
 707         u64 offset = ios->offset;
 708         struct _striping_info si;
 709         unsigned last_pg = 0;
 710         int ret = 0;
 711
 712         while (length) {
 713                 _calc_stripe_info(ios, offset, &si);
 714
 715                 if (length < si.group_length)
 716                         si.group_length = length;
 717
 718                 ret = _prepare_one_group(ios, si.group_length, &si, &last_pg, gfp_flags);
 719                 if (unlikely(ret))
 720                         goto out;
 721
 722                 offset += si.group_length;
 723                 length -= si.group_length;
 724         }
 725
 726 out:
 727         if (!ios->length)
 728                 return ret;
 729
 730         return 0;
 731 }
 732
 733 static int _sync_done(struct objio_state *ios)
 734 {
 735         struct completion *waiting = ios->private;
 736
 737         complete(waiting);
 738         return 0;
 739 }
 740
 741 static void _last_io(struct kref *kref)
 742 {
 743         struct objio_state *ios = container_of(kref, struct objio_state, kref);
 744
 745         ios->done(ios);
 746 }
 747
 748 static void _done_io(struct osd_request *or, void *p)
 749 {
 750         struct objio_state *ios = p;
 751
 752         kref_put(&ios->kref, _last_io);
 753 }
 754
 755 static int _io_exec(struct objio_state *ios)
 756 {
 757         DECLARE_COMPLETION_ONSTACK(wait);
 758         int ret = 0;
 759         unsigned i;
 760         objio_done_fn saved_done_fn = ios->done;
 761         bool sync = ios->sync;
 762
 763         if (sync) {
 764                 ios->done = _sync_done;
 765                 ios->private = &wait;
 766         }
 767
 768         kref_init(&ios->kref);
 769
 770         for (i = 0; i < ios->numdevs; i++) {
 771                 struct osd_request *or = ios->per_dev[i].or;
 772
 773                 if (!or)
 774                         continue;
 775
 776                 kref_get(&ios->kref);
 777                 osd_execute_request_async(or, _done_io, ios);
 778         }
 779
 780         kref_put(&ios->kref, _last_io);
 781
 782         if (sync) {
 783                 wait_for_completion(&wait);
 784                 ret = saved_done_fn(ios);
 785         }
 786
 787         return ret;
 788 }
 789
 790 /*
 791  * read
 792  */
 793 static int _read_done(struct objio_state *ios)
 794 {
 795         ssize_t status;
 796         int ret = _io_check(ios, false);
 797
 798         _io_free(ios);
 799
 800         if (likely(!ret))
 801                 status = ios->length;
 802         else
 803                 status = ret;
 804
 805         objlayout_read_done(&ios->oir, status, ios->sync);
 806         return ret;
 807 }
 808
 809 static int _read_mirrors(struct objio_state *ios, unsigned cur_comp)
 810 {
 811         struct osd_request *or = NULL;
 812         struct _objio_per_comp *per_dev = &ios->per_dev[cur_comp];
 813         unsigned dev = per_dev->dev;
 814         struct ore_comp *cred =
 815                         &ios->oc->comps[cur_comp];
 816         struct osd_obj_id obj = cred->obj;
 817         int ret;
 818
 819         or = osd_start_request(_io_od(ios, dev), GFP_KERNEL);
 820         if (unlikely(!or)) {
 821                 ret = -ENOMEM;
 822                 goto err;
 823         }
 824         per_dev->or = or;
 825
 826         osd_req_read(or, &obj, per_dev->offset, per_dev->bio, per_dev->length);
 827
 828         ret = osd_finalize_request(or, 0, cred->cred, NULL);
 829         if (ret) {
 830                 dprintk("%s: Faild to osd_finalize_request() => %d\n",
 831                         __func__, ret);
 832                 goto err;
 833         }
 834
 835         dprintk("%s:[%d] dev=%d obj=0x%llx start=0x%llx length=0x%lx\n",
 836                 __func__, cur_comp, dev, obj.id, _LLU(per_dev->offset),
 837                 per_dev->length);
 838
 839 err:
 840         return ret;
 841 }
 842
 843 static int _read_exec(struct objio_state *ios)
 844 {
 845         unsigned i;
 846         int ret;
 847
 848         for (i = 0; i < ios->numdevs; i += ios->layout->mirrors_p1) {
 849                 if (!ios->per_dev[i].length)
 850                         continue;
 851                 ret = _read_mirrors(ios, i);
 852                 if (unlikely(ret))
 853                         goto err;
 854         }
 855
 856         ios->done = _read_done;
 857         return _io_exec(ios);
 858
 859 err:
 860         _io_free(ios);
 861         return ret;
 862 }
 863
 864 int objio_read_pagelist(struct nfs_read_data *rdata)
 865 {
 866         struct objio_state *ios;
 867         int ret;
 868
 869         ret = objio_alloc_io_state(NFS_I(rdata->inode)->layout,
 870                         rdata->lseg, rdata->args.pages, rdata->args.pgbase,
 871                         rdata->args.offset, rdata->args.count, rdata,
 872                         GFP_KERNEL, &ios);
 873         if (unlikely(ret))
 874                 return ret;
 875
 876         ret = _io_rw_pagelist(ios, GFP_KERNEL);
 877         if (unlikely(ret))
 878                 return ret;
 879
 880         return _read_exec(ios);
 881 }
 882
 883 /*
 884  * write
 885  */
 886 static int _write_done(struct objio_state *ios)
 887 {
 888         ssize_t status;
 889         int ret = _io_check(ios, true);
 890
 891         _io_free(ios);
 892
 893         if (likely(!ret)) {
 894                 /* FIXME: should be based on the OSD's persistence model
 895                  * See OSD2r05 Section 4.13 Data persistence model */
 896                 ios->oir.committed = NFS_FILE_SYNC;
 897                 status = ios->length;
 898         } else {
 899                 status = ret;
 900         }
 901
 902         objlayout_write_done(&ios->oir, status, ios->sync);
 903         return ret;
 904 }
 905
 906 static int _write_mirrors(struct objio_state *ios, unsigned cur_comp)
 907 {
 908         struct _objio_per_comp *master_dev = &ios->per_dev[cur_comp];
 909         unsigned dev = ios->per_dev[cur_comp].dev;
 910         unsigned last_comp = cur_comp + ios->layout->mirrors_p1;
 911         int ret;
 912
 913         for (; cur_comp < last_comp; ++cur_comp, ++dev) {
 914                 struct osd_request *or = NULL;
 915                 struct ore_comp *cred = &ios->oc->comps[cur_comp];
 916                 struct osd_obj_id obj = cred->obj;
 917                 struct _objio_per_comp *per_dev = &ios->per_dev[cur_comp];
 918                 struct bio *bio;
 919
 920                 or = osd_start_request(_io_od(ios, dev), GFP_NOFS);
 921                 if (unlikely(!or)) {
 922                         ret = -ENOMEM;
 923                         goto err;
 924                 }
 925                 per_dev->or = or;
 926
 927                 if (per_dev != master_dev) {
 928                         bio = bio_kmalloc(GFP_NOFS,
 929                                           master_dev->bio->bi_max_vecs);
 930                         if (unlikely(!bio)) {
 931                                 dprintk("Faild to allocate BIO size=%u\n",
 932                                         master_dev->bio->bi_max_vecs);
 933                                 ret = -ENOMEM;
 934                                 goto err;
 935                         }
 936
 937                         __bio_clone(bio, master_dev->bio);
 938                         bio->bi_bdev = NULL;
 939                         bio->bi_next = NULL;
 940                         per_dev->bio = bio;
 941                         per_dev->dev = dev;
 942                         per_dev->length = master_dev->length;
 943                         per_dev->offset =  master_dev->offset;
 944                 } else {
 945                         bio = master_dev->bio;
 946                         bio->bi_rw |= REQ_WRITE;
 947                 }
 948
 949                 osd_req_write(or, &obj, per_dev->offset, bio, per_dev->length);
 950
 951                 ret = osd_finalize_request(or, 0, cred->cred, NULL);
 952                 if (ret) {
 953                         dprintk("%s: Faild to osd_finalize_request() => %d\n",
 954                                 __func__, ret);
 955                         goto err;
 956                 }
 957
 958                 dprintk("%s:[%d] dev=%d obj=0x%llx start=0x%llx length=0x%lx\n",
 959                         __func__, cur_comp, dev, obj.id, _LLU(per_dev->offset),
 960                         per_dev->length);
 961         }
 962
 963 err:
 964         return ret;
 965 }
 966
 967 static int _write_exec(struct objio_state *ios)
 968 {
 969         unsigned i;
 970         int ret;
 971
 972         for (i = 0; i < ios->numdevs; i += ios->layout->mirrors_p1) {
 973                 if (!ios->per_dev[i].length)
 974                         continue;
 975                 ret = _write_mirrors(ios, i);
 976                 if (unlikely(ret))
 977                         goto err;
 978         }
 979
 980         ios->done = _write_done;
 981         return _io_exec(ios);
 982
 983 err:
 984         _io_free(ios);
 985         return ret;
 986 }
 987
 988 int objio_write_pagelist(struct nfs_write_data *wdata, int how)
 989 {
 990         struct objio_state *ios;
 991         int ret;
 992
 993         ret = objio_alloc_io_state(NFS_I(wdata->inode)->layout,
 994                         wdata->lseg, wdata->args.pages, wdata->args.pgbase,
 995                         wdata->args.offset, wdata->args.count, wdata, GFP_NOFS,
 996                         &ios);
 997         if (unlikely(ret))
 998                 return ret;
 999
1000         ios->sync = 0 != (how & FLUSH_SYNC);
1001
1002         /* TODO: ios->stable = stable; */
1003         ret = _io_rw_pagelist(ios, GFP_NOFS);
1004         if (unlikely(ret))
1005                 return ret;
1006
1007         return _write_exec(ios);
1008 }
1009
1010 static bool objio_pg_test(struct nfs_pageio_descriptor *pgio,
1011                           struct nfs_page *prev, struct nfs_page *req)
1012 {
1013         if (!pnfs_generic_pg_test(pgio, prev, req))
1014                 return false;
1015
1016         return pgio->pg_count + req->wb_bytes <=
1017                         OBJIO_LSEG(pgio->pg_lseg)->layout.max_io_length;
1018 }
1019
1020 static const struct nfs_pageio_ops objio_pg_read_ops = {
1021         .pg_init = pnfs_generic_pg_init_read,
1022         .pg_test = objio_pg_test,
1023         .pg_doio = pnfs_generic_pg_readpages,
1024 };
1025
1026 static const struct nfs_pageio_ops objio_pg_write_ops = {
1027         .pg_init = pnfs_generic_pg_init_write,
1028         .pg_test = objio_pg_test,
1029         .pg_doio = pnfs_generic_pg_writepages,
1030 };
1031
1032 static struct pnfs_layoutdriver_type objlayout_type = {
1033         .id = LAYOUT_OSD2_OBJECTS,
1034         .name = "LAYOUT_OSD2_OBJECTS",
1035         .flags                   = PNFS_LAYOUTRET_ON_SETATTR,
1036
1037         .alloc_layout_hdr        = objlayout_alloc_layout_hdr,
1038         .free_layout_hdr         = objlayout_free_layout_hdr,
1039
1040         .alloc_lseg              = objlayout_alloc_lseg,
1041         .free_lseg               = objlayout_free_lseg,
1042
1043         .read_pagelist           = objlayout_read_pagelist,
1044         .write_pagelist          = objlayout_write_pagelist,
1045         .pg_read_ops             = &objio_pg_read_ops,
1046         .pg_write_ops            = &objio_pg_write_ops,
1047
1048         .free_deviceid_node      = objio_free_deviceid_node,
1049
1050         .encode_layoutcommit     = objlayout_encode_layoutcommit,
1051         .encode_layoutreturn     = objlayout_encode_layoutreturn,
1052 };
1053
1054 MODULE_DESCRIPTION("pNFS Layout Driver for OSD2 objects");
1055 MODULE_AUTHOR("Benny Halevy <bhalevy@panasas.com>");
1056 MODULE_LICENSE("GPL");
1057
1058 static int __init
1059 objlayout_init(void)
1060 {
1061         int ret = pnfs_register_layoutdriver(&objlayout_type);
1062
1063         if (ret)
1064                 printk(KERN_INFO
1065                         "%s: Registering OSD pNFS Layout Driver failed: error=%d\n",
1066                         __func__, ret);
1067         else
1068                 printk(KERN_INFO "%s: Registered OSD pNFS Layout Driver\n",
1069                         __func__);
1070         return ret;
1071 }
1072
1073 static void __exit
1074 objlayout_exit(void)
1075 {
1076         pnfs_unregister_layoutdriver(&objlayout_type);
1077         printk(KERN_INFO "%s: Unregistered OSD pNFS Layout Driver\n",
1078                __func__);
1079 }
1080
1081 MODULE_ALIAS("nfs-layouttype4-2");
1082
1083 module_init(objlayout_init);
1084 module_exit(objlayout_exit);