lustre/osd-zfs/osd_io.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  */
  30 /*
  31  * Copyright (c) 2012, 2013, Intel Corporation.
  32  * Use is subject to license terms.
  33  */
  34 /*
  35  * This file is part of Lustre, http://www.lustre.org/
  36  * Lustre is a trademark of Sun Microsystems, Inc.
  37  *
  38  * lustre/osd-zfs/osd_io.c
  39  *
  40  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  41  * Author: Mike Pershin <tappro@whamcloud.com>
  42  */
  43
  44 #define DEBUG_SUBSYSTEM S_OSD
  45
  46 #include <lustre_ver.h>
  47 #include <libcfs/libcfs.h>
  48 #include <obd_support.h>
  49 #include <lustre_net.h>
  50 #include <obd.h>
  51 #include <obd_class.h>
  52 #include <lustre_disk.h>
  53 #include <lustre_fid.h>
  54
  55 #include "osd_internal.h"
  56
  57 #include <sys/dnode.h>
  58 #include <sys/dbuf.h>
  59 #include <sys/spa.h>
  60 #include <sys/stat.h>
  61 #include <sys/zap.h>
  62 #include <sys/spa_impl.h>
  63 #include <sys/zfs_znode.h>
  64 #include <sys/dmu_tx.h>
  65 #include <sys/dmu_objset.h>
  66 #include <sys/dsl_prop.h>
  67 #include <sys/sa_impl.h>
  68 #include <sys/txg.h>
  69
  70 static char *osd_zerocopy_tag = "zerocopy";
  71
  72 static ssize_t osd_read(const struct lu_env *env, struct dt_object *dt,
  73                         struct lu_buf *buf, loff_t *pos,
  74                         struct lustre_capa *capa)
  75 {
  76         struct osd_object *obj  = osd_dt_obj(dt);
  77         struct osd_device *osd = osd_obj2dev(obj);
  78         uint64_t           old_size;
  79         int                size = buf->lb_len;
  80         int                rc;
  81
  82         LASSERT(dt_object_exists(dt));
  83         LASSERT(obj->oo_db);
  84
  85         read_lock(&obj->oo_attr_lock);
  86         old_size = obj->oo_attr.la_size;
  87         read_unlock(&obj->oo_attr_lock);
  88
  89         if (*pos + size > old_size) {
  90                 if (old_size < *pos)
  91                         return 0;
  92                 else
  93                         size = old_size - *pos;
  94         }
  95
  96         rc = -dmu_read(osd->od_objset.os, obj->oo_db->db_object, *pos, size,
  97                         buf->lb_buf, DMU_READ_PREFETCH);
  98         if (rc == 0) {
  99                 rc = size;
 100                 *pos += size;
 101
 102                 /* XXX: workaround for bug in HEAD: fsfilt_ldiskfs_read() returns
 103                  * requested number of bytes, not actually read ones */
 104                 if (S_ISLNK(obj->oo_dt.do_lu.lo_header->loh_attr))
 105                         rc = buf->lb_len;
 106         }
 107         return rc;
 108 }
 109
 110 static ssize_t osd_declare_write(const struct lu_env *env, struct dt_object *dt,
 111                                 const struct lu_buf *buf, loff_t pos,
 112                                 struct thandle *th)
 113 {
 114         struct osd_object  *obj  = osd_dt_obj(dt);
 115         struct osd_device  *osd = osd_obj2dev(obj);
 116         struct osd_thandle *oh;
 117         uint64_t            oid;
 118         ENTRY;
 119
 120         oh = container_of0(th, struct osd_thandle, ot_super);
 121
 122         /* in some cases declare can race with creation (e.g. llog)
 123          * and we need to wait till object is initialized. notice
 124          * LOHA_EXISTs is supposed to be the last step in the
 125          * initialization */
 126
 127         /* declare possible size change. notice we can't check
 128          * current size here as another thread can change it */
 129
 130         if (dt_object_exists(dt)) {
 131                 LASSERT(obj->oo_db);
 132                 oid = obj->oo_db->db_object;
 133
 134                 dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
 135         } else {
 136                 oid = DMU_NEW_OBJECT;
 137                 dmu_tx_hold_sa_create(oh->ot_tx, ZFS_SA_BASE_ATTR_SIZE);
 138         }
 139
 140         dmu_tx_hold_write(oh->ot_tx, oid, pos, buf->lb_len);
 141
 142         /* dt_declare_write() is usually called for system objects, such
 143          * as llog or last_rcvd files. We needn't enforce quota on those
 144          * objects, so always set the lqi_space as 0. */
 145         RETURN(osd_declare_quota(env, osd, obj->oo_attr.la_uid,
 146                                  obj->oo_attr.la_gid, 0, oh, true, NULL,
 147                                  false));
 148 }
 149
 150 static ssize_t osd_write(const struct lu_env *env, struct dt_object *dt,
 151                         const struct lu_buf *buf, loff_t *pos,
 152                         struct thandle *th, struct lustre_capa *capa,
 153                         int ignore_quota)
 154 {
 155         struct osd_object  *obj  = osd_dt_obj(dt);
 156         struct osd_device  *osd = osd_obj2dev(obj);
 157         udmu_objset_t      *uos = &osd->od_objset;
 158         struct osd_thandle *oh;
 159         uint64_t            offset = *pos;
 160         int                 rc;
 161         ENTRY;
 162
 163         LASSERT(dt_object_exists(dt));
 164         LASSERT(obj->oo_db);
 165
 166         LASSERT(th != NULL);
 167         oh = container_of0(th, struct osd_thandle, ot_super);
 168
 169         dmu_write(osd->od_objset.os, obj->oo_db->db_object, offset,
 170                 (uint64_t)buf->lb_len, buf->lb_buf, oh->ot_tx);
 171         write_lock(&obj->oo_attr_lock);
 172         if (obj->oo_attr.la_size < offset + buf->lb_len) {
 173                 obj->oo_attr.la_size = offset + buf->lb_len;
 174                 write_unlock(&obj->oo_attr_lock);
 175                 /* osd_object_sa_update() will be copying directly from oo_attr
 176                  * into dbuf.  any update within a single txg will copy the
 177                  * most actual */
 178                 rc = osd_object_sa_update(obj, SA_ZPL_SIZE(uos),
 179                                         &obj->oo_attr.la_size, 8, oh);
 180                 if (unlikely(rc))
 181                         GOTO(out, rc);
 182         } else {
 183                 write_unlock(&obj->oo_attr_lock);
 184         }
 185
 186         *pos += buf->lb_len;
 187         rc = buf->lb_len;
 188
 189 out:
 190         RETURN(rc);
 191 }
 192
 193 /*
 194  * XXX: for the moment I don't want to use lnb_flags for osd-internal
 195  *      purposes as it's not very well defined ...
 196  *      instead I use the lowest bit of the address so that:
 197  *        arc buffer:  .lnb_obj = abuf          (arc we loan for write)
 198  *        dbuf buffer: .lnb_obj = dbuf | 1      (dbuf we get for read)
 199  *        copy buffer: .lnb_page->mapping = obj (page we allocate for write)
 200  *
 201  *      bzzz, to blame
 202  */
 203 static int osd_bufs_put(const struct lu_env *env, struct dt_object *dt,
 204                         struct niobuf_local *lnb, int npages)
 205 {
 206         struct osd_object *obj  = osd_dt_obj(dt);
 207         struct osd_device *osd = osd_obj2dev(obj);
 208         unsigned long      ptr;
 209         int                i;
 210
 211         LASSERT(dt_object_exists(dt));
 212         LASSERT(obj->oo_db);
 213
 214         for (i = 0; i < npages; i++) {
 215                 if (lnb[i].page == NULL)
 216                         continue;
 217                 if (lnb[i].page->mapping == (void *)obj) {
 218                         /* this is anonymous page allocated for copy-write */
 219                         lnb[i].page->mapping = NULL;
 220                         __free_page(lnb[i].page);
 221                         atomic_dec(&osd->od_zerocopy_alloc);
 222                 } else {
 223                         /* see comment in osd_bufs_get_read() */
 224                         ptr = (unsigned long)lnb[i].dentry;
 225                         if (ptr & 1UL) {
 226                                 ptr &= ~1UL;
 227                                 dmu_buf_rele((void *)ptr, osd_zerocopy_tag);
 228                                 atomic_dec(&osd->od_zerocopy_pin);
 229                         } else if (lnb[i].dentry != NULL) {
 230                                 dmu_return_arcbuf((void *)lnb[i].dentry);
 231                                 atomic_dec(&osd->od_zerocopy_loan);
 232                         }
 233                 }
 234                 lnb[i].page = NULL;
 235                 lnb[i].dentry = NULL;
 236         }
 237
 238         return 0;
 239 }
 240
 241 static inline struct page *kmem_to_page(void *addr)
 242 {
 243         if (is_vmalloc_addr(addr))
 244                 return vmalloc_to_page(addr);
 245         else
 246                 return virt_to_page(addr);
 247 }
 248
 249 static int osd_bufs_get_read(const struct lu_env *env, struct osd_object *obj,
 250                                 loff_t off, ssize_t len, struct niobuf_local *lnb)
 251 {
 252         struct osd_device *osd = osd_obj2dev(obj);
 253         dmu_buf_t        **dbp;
 254         int                rc, i, numbufs, npages = 0;
 255         ENTRY;
 256
 257         /* grab buffers for read:
 258          * OSD API let us to grab buffers first, then initiate IO(s)
 259          * so that all required IOs will be done in parallel, but at the
 260          * moment DMU doesn't provide us with a method to grab buffers.
 261          * If we discover this is a vital for good performance we
 262          * can get own replacement for dmu_buf_hold_array_by_bonus().
 263          */
 264         while (len > 0) {
 265                 rc = -dmu_buf_hold_array_by_bonus(obj->oo_db, off, len, TRUE,
 266                                                   osd_zerocopy_tag, &numbufs,
 267                                                   &dbp);
 268                 if (unlikely(rc))
 269                         GOTO(err, rc);
 270
 271                 for (i = 0; i < numbufs; i++) {
 272                         int bufoff, tocpy, thispage;
 273                         void *dbf = dbp[i];
 274
 275                         LASSERT(len > 0);
 276
 277                         atomic_inc(&osd->od_zerocopy_pin);
 278
 279                         bufoff = off - dbp[i]->db_offset;
 280                         tocpy = min_t(int, dbp[i]->db_size - bufoff, len);
 281
 282                         /* kind of trick to differentiate dbuf vs. arcbuf */
 283                         LASSERT(((unsigned long)dbp[i] & 1) == 0);
 284                         dbf = (void *) ((unsigned long)dbp[i] | 1);
 285
 286                         while (tocpy > 0) {
 287                                 thispage = PAGE_CACHE_SIZE;
 288                                 thispage -= bufoff & (PAGE_CACHE_SIZE - 1);
 289                                 thispage = min(tocpy, thispage);
 290
 291                                 lnb->rc = 0;
 292                                 lnb->lnb_file_offset = off;
 293                                 lnb->lnb_page_offset = bufoff & ~CFS_PAGE_MASK;
 294                                 lnb->len = thispage;
 295                                 lnb->page = kmem_to_page(dbp[i]->db_data +
 296                                                                 bufoff);
 297                                 /* mark just a single slot: we need this
 298                                  * reference to dbuf to be release once */
 299                                 lnb->dentry = dbf;
 300                                 dbf = NULL;
 301
 302                                 tocpy -= thispage;
 303                                 len -= thispage;
 304                                 bufoff += thispage;
 305                                 off += thispage;
 306
 307                                 npages++;
 308                                 lnb++;
 309                         }
 310
 311                         /* steal dbuf so dmu_buf_rele_array() cant release it */
 312                         dbp[i] = NULL;
 313                 }
 314
 315                 dmu_buf_rele_array(dbp, numbufs, osd_zerocopy_tag);
 316         }
 317
 318         RETURN(npages);
 319
 320 err:
 321         LASSERT(rc < 0);
 322         osd_bufs_put(env, &obj->oo_dt, lnb - npages, npages);
 323         RETURN(rc);
 324 }
 325
 326 static int osd_bufs_get_write(const struct lu_env *env, struct osd_object *obj,
 327                                 loff_t off, ssize_t len, struct niobuf_local *lnb)
 328 {
 329         struct osd_device *osd = osd_obj2dev(obj);
 330         int                plen, off_in_block, sz_in_block;
 331         int                rc, i = 0, npages = 0;
 332         arc_buf_t         *abuf;
 333         uint32_t           bs;
 334         uint64_t           dummy;
 335         ENTRY;
 336
 337         dmu_object_size_from_db(obj->oo_db, &bs, &dummy);
 338
 339         /*
 340          * currently only full blocks are subject to zerocopy approach:
 341          * so that we're sure nobody is trying to update the same block
 342          */
 343         while (len > 0) {
 344                 LASSERT(npages < PTLRPC_MAX_BRW_PAGES);
 345
 346                 off_in_block = off & (bs - 1);
 347                 sz_in_block = min_t(int, bs - off_in_block, len);
 348
 349                 if (sz_in_block == bs) {
 350                         /* full block, try to use zerocopy */
 351
 352                         abuf = dmu_request_arcbuf(obj->oo_db, bs);
 353                         if (unlikely(abuf == NULL))
 354                                 GOTO(out_err, rc = -ENOMEM);
 355
 356                         atomic_inc(&osd->od_zerocopy_loan);
 357
 358                         /* go over pages arcbuf contains, put them as
 359                          * local niobufs for ptlrpc's bulks */
 360                         while (sz_in_block > 0) {
 361                                 plen = min_t(int, sz_in_block, PAGE_CACHE_SIZE);
 362
 363                                 lnb[i].lnb_file_offset = off;
 364                                 lnb[i].lnb_page_offset = 0;
 365                                 lnb[i].len = plen;
 366                                 lnb[i].rc = 0;
 367                                 if (sz_in_block == bs)
 368                                         lnb[i].dentry = (void *)abuf;
 369                                 else
 370                                         lnb[i].dentry = NULL;
 371
 372                                 /* this one is not supposed to fail */
 373                                 lnb[i].page = kmem_to_page(abuf->b_data +
 374                                                         off_in_block);
 375                                 LASSERT(lnb[i].page);
 376
 377                                 lprocfs_counter_add(osd->od_stats,
 378                                                 LPROC_OSD_ZEROCOPY_IO, 1);
 379
 380                                 sz_in_block -= plen;
 381                                 len -= plen;
 382                                 off += plen;
 383                                 off_in_block += plen;
 384                                 i++;
 385                                 npages++;
 386                         }
 387                 } else {
 388                         if (off_in_block == 0 && len < bs &&
 389                                         off + len >= obj->oo_attr.la_size)
 390                                 lprocfs_counter_add(osd->od_stats,
 391                                                 LPROC_OSD_TAIL_IO, 1);
 392
 393                         /* can't use zerocopy, allocate temp. buffers */
 394                         while (sz_in_block > 0) {
 395                                 plen = min_t(int, sz_in_block, PAGE_CACHE_SIZE);
 396
 397                                 lnb[i].lnb_file_offset = off;
 398                                 lnb[i].lnb_page_offset = 0;
 399                                 lnb[i].len = plen;
 400                                 lnb[i].rc = 0;
 401                                 lnb[i].dentry = NULL;
 402
 403                                 lnb[i].page = alloc_page(OSD_GFP_IO);
 404                                 if (unlikely(lnb[i].page == NULL))
 405                                         GOTO(out_err, rc = -ENOMEM);
 406
 407                                 LASSERT(lnb[i].page->mapping == NULL);
 408                                 lnb[i].page->mapping = (void *)obj;
 409
 410                                 atomic_inc(&osd->od_zerocopy_alloc);
 411                                 lprocfs_counter_add(osd->od_stats,
 412                                                 LPROC_OSD_COPY_IO, 1);
 413
 414                                 sz_in_block -= plen;
 415                                 len -= plen;
 416                                 off += plen;
 417                                 i++;
 418                                 npages++;
 419                         }
 420                 }
 421         }
 422
 423         RETURN(npages);
 424
 425 out_err:
 426         osd_bufs_put(env, &obj->oo_dt, lnb, npages);
 427         RETURN(rc);
 428 }
 429
 430 static int osd_bufs_get(const struct lu_env *env, struct dt_object *dt,
 431                         loff_t offset, ssize_t len, struct niobuf_local *lnb,
 432                         int rw, struct lustre_capa *capa)
 433 {
 434         struct osd_object *obj  = osd_dt_obj(dt);
 435         int                rc;
 436
 437         LASSERT(dt_object_exists(dt));
 438         LASSERT(obj->oo_db);
 439
 440         if (rw == 0)
 441                 rc = osd_bufs_get_read(env, obj, offset, len, lnb);
 442         else
 443                 rc = osd_bufs_get_write(env, obj, offset, len, lnb);
 444
 445         return rc;
 446 }
 447
 448 static int osd_write_prep(const struct lu_env *env, struct dt_object *dt,
 449                         struct niobuf_local *lnb, int npages)
 450 {
 451         struct osd_object *obj = osd_dt_obj(dt);
 452
 453         LASSERT(dt_object_exists(dt));
 454         LASSERT(obj->oo_db);
 455
 456         return 0;
 457 }
 458
 459 /* Return number of blocks that aren't mapped in the [start, start + size]
 460  * region */
 461 static int osd_count_not_mapped(struct osd_object *obj, uint64_t start,
 462                                 uint32_t size)
 463 {
 464         dmu_buf_impl_t  *dbi = (dmu_buf_impl_t *)obj->oo_db;
 465         dmu_buf_impl_t  *db;
 466         dnode_t         *dn;
 467         uint32_t         blkshift;
 468         uint64_t         end, blkid;
 469         int              rc;
 470         ENTRY;
 471
 472         DB_DNODE_ENTER(dbi);
 473         dn = DB_DNODE(dbi);
 474
 475         if (dn->dn_maxblkid == 0) {
 476                 if (start + size <= dn->dn_datablksz)
 477                         GOTO(out, size = 0);
 478                 if (start < dn->dn_datablksz)
 479                         start = dn->dn_datablksz;
 480                 /* assume largest block size */
 481                 blkshift = SPA_MAXBLOCKSHIFT;
 482         } else {
 483                 /* blocksize can't change */
 484                 blkshift = dn->dn_datablkshift;
 485         }
 486
 487         /* compute address of last block */
 488         end = (start + size - 1) >> blkshift;
 489         /* align start on block boundaries */
 490         start >>= blkshift;
 491
 492         /* size is null, can't be mapped */
 493         if (obj->oo_attr.la_size == 0 || dn->dn_maxblkid == 0)
 494                 GOTO(out, size = (end - start + 1) << blkshift);
 495
 496         /* beyond EOF, can't be mapped */
 497         if (start > dn->dn_maxblkid)
 498                 GOTO(out, size = (end - start + 1) << blkshift);
 499
 500         size = 0;
 501         for (blkid = start; blkid <= end; blkid++) {
 502                 if (blkid == dn->dn_maxblkid)
 503                         /* this one is mapped for sure */
 504                         continue;
 505                 if (blkid > dn->dn_maxblkid) {
 506                         size += (end - blkid + 1) << blkshift;
 507                         GOTO(out, size);
 508                 }
 509
 510                 rc = dbuf_hold_impl(dn, 0, blkid, TRUE, FTAG, &db);
 511                 if (rc) {
 512                         /* for ENOENT (block not mapped) and any other errors,
 513                          * assume the block isn't mapped */
 514                         size += 1 << blkshift;
 515                         continue;
 516                 }
 517                 dbuf_rele(db, FTAG);
 518         }
 519
 520         GOTO(out, size);
 521 out:
 522         DB_DNODE_EXIT(dbi);
 523         return size;
 524 }
 525
 526 static int osd_declare_write_commit(const struct lu_env *env,
 527                                 struct dt_object *dt,
 528                                 struct niobuf_local *lnb, int npages,
 529                                 struct thandle *th)
 530 {
 531         struct osd_object  *obj = osd_dt_obj(dt);
 532         struct osd_device  *osd = osd_obj2dev(obj);
 533         struct osd_thandle *oh;
 534         uint64_t            offset = 0;
 535         uint32_t            size = 0;
 536         int                 i, rc, flags = 0;
 537         bool                ignore_quota = false, synced = false;
 538         long long           space = 0;
 539         ENTRY;
 540
 541         LASSERT(dt_object_exists(dt));
 542         LASSERT(obj->oo_db);
 543
 544         LASSERT(lnb);
 545         LASSERT(npages > 0);
 546
 547         oh = container_of0(th, struct osd_thandle, ot_super);
 548
 549         for (i = 0; i < npages; i++) {
 550                 if (lnb[i].rc)
 551                         /* ENOSPC, network RPC error, etc.
 552                          * We don't want to book space for pages which will be
 553                          * skipped in osd_write_commit(). Hence we skip pages
 554                          * with lnb_rc != 0 here too */
 555                         continue;
 556                 /* ignore quota for the whole request if any page is from
 557                  * client cache or written by root.
 558                  *
 559                  * XXX once we drop the 1.8 client support, the checking
 560                  * for whether page is from cache can be simplified as:
 561                  * !(lnb[i].flags & OBD_BRW_SYNC)
 562                  *
 563                  * XXX we could handle this on per-lnb basis as done by
 564                  * grant. */
 565                 if ((lnb[i].flags & OBD_BRW_NOQUOTA) ||
 566                     (lnb[i].flags & (OBD_BRW_FROM_GRANT | OBD_BRW_SYNC)) ==
 567                     OBD_BRW_FROM_GRANT)
 568                         ignore_quota = true;
 569                 if (size == 0) {
 570                         /* first valid lnb */
 571                         offset = lnb[i].lnb_file_offset;
 572                         size = lnb[i].len;
 573                         continue;
 574                 }
 575                 if (offset + size == lnb[i].lnb_file_offset) {
 576                         /* this lnb is contiguous to the previous one */
 577                         size += lnb[i].len;
 578                         continue;
 579                 }
 580
 581                 dmu_tx_hold_write(oh->ot_tx, obj->oo_db->db_object, offset,size);
 582
 583                 /* estimating space that will be consumed by a write is rather
 584                  * complicated with ZFS. As a consequence, we don't account for
 585                  * indirect blocks and quota overrun will be adjusted once the
 586                  * operation is committed, if required. */
 587                 space += osd_count_not_mapped(obj, offset, size);
 588
 589                 offset = lnb->lnb_file_offset;
 590                 size = lnb->len;
 591         }
 592
 593         if (size) {
 594                 dmu_tx_hold_write(oh->ot_tx, obj->oo_db->db_object, offset,size);
 595                 space += osd_count_not_mapped(obj, offset, size);
 596         }
 597
 598         dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
 599
 600         oh->ot_write_commit = 1; /* used in osd_trans_start() for fail_loc */
 601
 602         /* backend zfs filesystem might be configured to store multiple data
 603          * copies */
 604         space  *= osd->od_objset.os->os_copies;
 605         space   = toqb(space);
 606         CDEBUG(D_QUOTA, "writting %d pages, reserving "LPD64"K of quota "
 607                "space\n", npages, space);
 608
 609 retry:
 610         /* acquire quota space if needed */
 611         rc = osd_declare_quota(env, osd, obj->oo_attr.la_uid,
 612                                obj->oo_attr.la_gid, space, oh, true, &flags,
 613                                ignore_quota);
 614
 615         if (!synced && rc == -EDQUOT && (flags & QUOTA_FL_SYNC) != 0) {
 616                 dt_sync(env, th->th_dev);
 617                 synced = true;
 618                 CDEBUG(D_QUOTA, "retry after sync\n");
 619                 flags = 0;
 620                 goto retry;
 621         }
 622
 623         /* we need only to store the overquota flags in the first lnb for
 624          * now, once we support multiple objects BRW, this code needs be
 625          * revised. */
 626         if (flags & QUOTA_FL_OVER_USRQUOTA)
 627                 lnb[0].flags |= OBD_BRW_OVER_USRQUOTA;
 628         if (flags & QUOTA_FL_OVER_GRPQUOTA)
 629                 lnb[0].flags |= OBD_BRW_OVER_GRPQUOTA;
 630
 631         RETURN(rc);
 632 }
 633
 634 static int osd_write_commit(const struct lu_env *env, struct dt_object *dt,
 635                         struct niobuf_local *lnb, int npages,
 636                         struct thandle *th)
 637 {
 638         struct osd_object  *obj  = osd_dt_obj(dt);
 639         struct osd_device  *osd = osd_obj2dev(obj);
 640         udmu_objset_t      *uos = &osd->od_objset;
 641         struct osd_thandle *oh;
 642         uint64_t            new_size = 0;
 643         int                 i, rc = 0;
 644         ENTRY;
 645
 646         LASSERT(dt_object_exists(dt));
 647         LASSERT(obj->oo_db);
 648
 649         LASSERT(th != NULL);
 650         oh = container_of0(th, struct osd_thandle, ot_super);
 651
 652         for (i = 0; i < npages; i++) {
 653                 CDEBUG(D_INODE, "write %u bytes at %u\n",
 654                         (unsigned) lnb[i].len,
 655                         (unsigned) lnb[i].lnb_file_offset);
 656
 657                 if (lnb[i].rc) {
 658                         /* ENOSPC, network RPC error, etc.
 659                          * Unlike ldiskfs, zfs allocates new blocks on rewrite,
 660                          * so we skip this page if lnb_rc is set to -ENOSPC */
 661                         CDEBUG(D_INODE, "obj "DFID": skipping lnb[%u]: rc=%d\n",
 662                                 PFID(lu_object_fid(&dt->do_lu)), i,
 663                                 lnb[i].rc);
 664                         continue;
 665                 }
 666
 667                 if (lnb[i].page->mapping == (void *)obj) {
 668                         dmu_write(osd->od_objset.os, obj->oo_db->db_object,
 669                                 lnb[i].lnb_file_offset, lnb[i].len,
 670                                 kmap(lnb[i].page), oh->ot_tx);
 671                         kunmap(lnb[i].page);
 672                 } else if (lnb[i].dentry) {
 673                         LASSERT(((unsigned long)lnb[i].dentry & 1) == 0);
 674                         /* buffer loaned for zerocopy, try to use it.
 675                          * notice that dmu_assign_arcbuf() is smart
 676                          * enough to recognize changed blocksize
 677                          * in this case it fallbacks to dmu_write() */
 678                         dmu_assign_arcbuf(obj->oo_db, lnb[i].lnb_file_offset,
 679                                         (void *)lnb[i].dentry, oh->ot_tx);
 680                         /* drop the reference, otherwise osd_put_bufs()
 681                          * will be releasing it - bad! */
 682                         lnb[i].dentry = NULL;
 683                         atomic_dec(&osd->od_zerocopy_loan);
 684                 }
 685
 686                 if (new_size < lnb[i].lnb_file_offset + lnb[i].len)
 687                         new_size = lnb[i].lnb_file_offset + lnb[i].len;
 688         }
 689
 690         if (unlikely(new_size == 0)) {
 691                 /* no pages to write, no transno is needed */
 692                 th->th_local = 1;
 693                 /* it is important to return 0 even when all lnb_rc == -ENOSPC
 694                  * since ofd_commitrw_write() retries several times on ENOSPC */
 695                 RETURN(0);
 696         }
 697
 698         write_lock(&obj->oo_attr_lock);
 699         if (obj->oo_attr.la_size < new_size) {
 700                 obj->oo_attr.la_size = new_size;
 701                 write_unlock(&obj->oo_attr_lock);
 702                 /* osd_object_sa_update() will be copying directly from
 703                  * oo_attr into dbuf. any update within a single txg will copy
 704                  * the most actual */
 705                 rc = osd_object_sa_update(obj, SA_ZPL_SIZE(uos),
 706                                         &obj->oo_attr.la_size, 8, oh);
 707         } else {
 708                 write_unlock(&obj->oo_attr_lock);
 709         }
 710
 711         RETURN(rc);
 712 }
 713
 714 static int osd_read_prep(const struct lu_env *env, struct dt_object *dt,
 715                         struct niobuf_local *lnb, int npages)
 716 {
 717         struct osd_object *obj  = osd_dt_obj(dt);
 718         struct lu_buf      buf;
 719         loff_t             offset;
 720         int                i;
 721
 722         LASSERT(dt_object_exists(dt));
 723         LASSERT(obj->oo_db);
 724
 725         for (i = 0; i < npages; i++) {
 726                 buf.lb_buf = kmap(lnb[i].page);
 727                 buf.lb_len = lnb[i].len;
 728                 offset = lnb[i].lnb_file_offset;
 729
 730                 CDEBUG(D_OTHER, "read %u bytes at %u\n",
 731                         (unsigned) lnb[i].len,
 732                         (unsigned) lnb[i].lnb_file_offset);
 733                 lnb[i].rc = osd_read(env, dt, &buf, &offset, NULL);
 734                 kunmap(lnb[i].page);
 735
 736                 if (lnb[i].rc < buf.lb_len) {
 737                         /* all subsequent rc should be 0 */
 738                         while (++i < npages)
 739                                 lnb[i].rc = 0;
 740                         break;
 741                 }
 742         }
 743
 744         return 0;
 745 }
 746
 747 /*
 748  * Punch/truncate an object
 749  *
 750  *      IN:     db  - dmu_buf of the object to free data in.
 751  *              off - start of section to free.
 752  *              len - length of section to free (DMU_OBJECT_END => to EOF).
 753  *
 754  *      RETURN: 0 if success
 755  *              error code if failure
 756  *
 757  * The transaction passed to this routine must have
 758  * dmu_tx_hold_sa() and if off < size, dmu_tx_hold_free()
 759  * called and then assigned to a transaction group.
 760  */
 761 static int __osd_object_punch(objset_t *os, dmu_buf_t *db, dmu_tx_t *tx,
 762                                 uint64_t size, uint64_t off, uint64_t len)
 763 {
 764         int rc = 0;
 765
 766         /* Assert that the transaction has been assigned to a
 767            transaction group. */
 768         LASSERT(tx->tx_txg != 0);
 769         /*
 770          * Nothing to do if file already at desired length.
 771          */
 772         if (len == DMU_OBJECT_END && size == off)
 773                 return 0;
 774
 775         if (off < size)
 776                 rc = -dmu_free_range(os, db->db_object, off, len, tx);
 777
 778         return rc;
 779 }
 780
 781 static int osd_punch(const struct lu_env *env, struct dt_object *dt,
 782                         __u64 start, __u64 end, struct thandle *th,
 783                         struct lustre_capa *capa)
 784 {
 785         struct osd_object  *obj = osd_dt_obj(dt);
 786         struct osd_device  *osd = osd_obj2dev(obj);
 787         udmu_objset_t      *uos = &osd->od_objset;
 788         struct osd_thandle *oh;
 789         __u64               len;
 790         int                 rc = 0;
 791         ENTRY;
 792
 793         LASSERT(dt_object_exists(dt));
 794         LASSERT(osd_invariant(obj));
 795
 796         LASSERT(th != NULL);
 797         oh = container_of0(th, struct osd_thandle, ot_super);
 798
 799         write_lock(&obj->oo_attr_lock);
 800         /* truncate */
 801         if (end == OBD_OBJECT_EOF || end >= obj->oo_attr.la_size)
 802                 len = DMU_OBJECT_END;
 803         else
 804                 len = end - start;
 805         write_unlock(&obj->oo_attr_lock);
 806
 807         rc = __osd_object_punch(osd->od_objset.os, obj->oo_db, oh->ot_tx,
 808                                 obj->oo_attr.la_size, start, len);
 809         /* set new size */
 810         if (len == DMU_OBJECT_END) {
 811                 write_lock(&obj->oo_attr_lock);
 812                 obj->oo_attr.la_size = start;
 813                 write_unlock(&obj->oo_attr_lock);
 814                 rc = osd_object_sa_update(obj, SA_ZPL_SIZE(uos),
 815                                         &obj->oo_attr.la_size, 8, oh);
 816         }
 817         RETURN(rc);
 818 }
 819
 820 static int osd_declare_punch(const struct lu_env *env, struct dt_object *dt,
 821                         __u64 start, __u64 end, struct thandle *handle)
 822 {
 823         struct osd_object  *obj = osd_dt_obj(dt);
 824         struct osd_device  *osd = osd_obj2dev(obj);
 825         struct osd_thandle *oh;
 826         __u64               len;
 827         ENTRY;
 828
 829         oh = container_of0(handle, struct osd_thandle, ot_super);
 830
 831         read_lock(&obj->oo_attr_lock);
 832         if (end == OBD_OBJECT_EOF || end >= obj->oo_attr.la_size)
 833                 len = DMU_OBJECT_END;
 834         else
 835                 len = end - start;
 836
 837         /* declare we'll free some blocks ... */
 838         if (start < obj->oo_attr.la_size) {
 839                 read_unlock(&obj->oo_attr_lock);
 840                 dmu_tx_hold_free(oh->ot_tx, obj->oo_db->db_object, start, len);
 841         } else {
 842                 read_unlock(&obj->oo_attr_lock);
 843         }
 844
 845         /* ... and we'll modify size attribute */
 846         dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
 847
 848         RETURN(osd_declare_quota(env, osd, obj->oo_attr.la_uid,
 849                                  obj->oo_attr.la_gid, 0, oh, true, NULL,
 850                                  false));
 851 }
 852
 853
 854 struct dt_body_operations osd_body_ops = {
 855         .dbo_read                       = osd_read,
 856         .dbo_declare_write              = osd_declare_write,
 857         .dbo_write                      = osd_write,
 858         .dbo_bufs_get                   = osd_bufs_get,
 859         .dbo_bufs_put                   = osd_bufs_put,
 860         .dbo_write_prep                 = osd_write_prep,
 861         .dbo_declare_write_commit       = osd_declare_write_commit,
 862         .dbo_write_commit               = osd_write_commit,
 863         .dbo_read_prep                  = osd_read_prep,
 864         .dbo_declare_punch              = osd_declare_punch,
 865         .dbo_punch                      = osd_punch,
 866 };
 867