* Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
* Use is subject to license terms.
*
- * Copyright (c) 2012, 2015, Intel Corporation.
+ * Copyright (c) 2012, 2016, Intel Corporation.
*/
/*
* This file is part of Lustre, http://www.lustre.org/
#include <sys/sa_impl.h>
#include <sys/txg.h>
-static char *osd_zerocopy_tag = "zerocopy";
+static char *osd_0copy_tag = "zerocopy";
static void record_start_io(struct osd_device *osd, int rw, int discont_pages)
unsigned long start;
LASSERT(dt_object_exists(dt));
- LASSERT(obj->oo_db);
+ LASSERT(obj->oo_dn);
start = cfs_time_current();
record_start_io(osd, READ, 0);
- rc = -dmu_read(osd->od_os, obj->oo_db->db_object, *pos, size,
- buf->lb_buf, DMU_READ_PREFETCH);
+ rc = osd_dmu_read(osd, obj->oo_dn, *pos, size, buf->lb_buf,
+ DMU_READ_PREFETCH);
record_end_io(osd, READ, cfs_time_current() - start, size,
size >> PAGE_SHIFT);
/* size change (in dnode) will be declared by dmu_tx_hold_write() */
if (dt_object_exists(dt))
- oid = obj->oo_db->db_object;
+ oid = obj->oo_dn->dn_object;
else
oid = DMU_NEW_OBJECT;
if (pos == -1)
pos = max_t(loff_t, 256 * 8 * LLOG_MIN_CHUNK_SIZE,
obj->oo_attr.la_size + (2 << 20));
- dmu_tx_hold_write(oh->ot_tx, oid, pos, buf->lb_len);
+ osd_tx_hold_write(oh->ot_tx, oid, obj->oo_dn, pos, buf->lb_len);
/* dt_declare_write() is usually called for system objects, such
* as llog or last_rcvd files. We needn't enforce quota on those
ENTRY;
LASSERT(dt_object_exists(dt));
- LASSERT(obj->oo_db);
+ LASSERT(obj->oo_dn);
LASSERT(th != NULL);
oh = container_of0(th, struct osd_thandle, ot_super);
- dmu_write(osd->od_os, obj->oo_db->db_object, offset,
- (uint64_t)buf->lb_len, buf->lb_buf, oh->ot_tx);
+ osd_dmu_write(osd, obj->oo_dn, offset, (uint64_t)buf->lb_len,
+ buf->lb_buf, oh->ot_tx);
write_lock(&obj->oo_attr_lock);
if (obj->oo_attr.la_size < offset + buf->lb_len) {
obj->oo_attr.la_size = offset + buf->lb_len;
int i;
LASSERT(dt_object_exists(dt));
- LASSERT(obj->oo_db);
+ LASSERT(obj->oo_dn);
for (i = 0; i < npages; i++) {
if (lnb[i].lnb_page == NULL)
ptr = (unsigned long)lnb[i].lnb_data;
if (ptr & 1UL) {
ptr &= ~1UL;
- dmu_buf_rele((void *)ptr, osd_zerocopy_tag);
+ dmu_buf_rele((void *)ptr, osd_0copy_tag);
atomic_dec(&osd->od_zerocopy_pin);
} else if (lnb[i].lnb_data != NULL) {
+ int j, apages, abufsz;
+ abufsz = arc_buf_size(lnb[i].lnb_data);
+ apages = abufsz / PAGE_SIZE;
+ /* these references to pages must be invalidated
+ * to prevent access in osd_bufs_put() */
+ for (j = 0; j < apages; j++)
+ lnb[i + j].lnb_page = NULL;
dmu_return_arcbuf(lnb[i].lnb_data);
atomic_dec(&osd->od_zerocopy_loan);
}
static inline struct page *kmem_to_page(void *addr)
{
+ LASSERT(!((unsigned long)addr & ~PAGE_MASK));
if (is_vmalloc_addr(addr))
return vmalloc_to_page(addr);
else
* \retval negative error number of failure
*/
static int osd_bufs_get_read(const struct lu_env *env, struct osd_object *obj,
- loff_t off, ssize_t len, struct niobuf_local *lnb)
+ loff_t off, ssize_t len, struct niobuf_local *lnb)
{
struct osd_device *osd = osd_obj2dev(obj);
unsigned long start = cfs_time_current();
* can get own replacement for dmu_buf_hold_array_by_bonus().
*/
while (len > 0) {
- rc = -dmu_buf_hold_array_by_bonus(obj->oo_db, off, len, TRUE,
- osd_zerocopy_tag, &numbufs,
- &dbp);
+ rc = -dmu_buf_hold_array_by_bonus(&obj->oo_dn->dn_bonus->db,
+ off, len, TRUE, osd_0copy_tag,
+ &numbufs, &dbp);
if (unlikely(rc))
GOTO(err, rc);
dbp[i] = NULL;
}
- dmu_buf_rele_array(dbp, numbufs, osd_zerocopy_tag);
+ dmu_buf_rele_array(dbp, numbufs, osd_0copy_tag);
}
record_end_io(osd, READ, cfs_time_current() - start,
RETURN(rc);
}
+static inline arc_buf_t *osd_request_arcbuf(dnode_t *dn, size_t bs)
+{
+ arc_buf_t *abuf;
+
+ abuf = dmu_request_arcbuf(&dn->dn_bonus->db, bs);
+ if (unlikely(!abuf))
+ return ERR_PTR(-ENOMEM);
+
+#if ZFS_VERSION_CODE < OBD_OCD_VERSION(0, 7, 0, 0)
+ /**
+ * ZFS prior to 0.7.0 doesn't guarantee PAGE_SIZE alignment for zio
+ * blocks smaller than (PAGE_SIZE << 2). This poses a problem of
+ * setting up page array for RDMA transfer. See LU-9305.
+ */
+ if ((unsigned long)abuf->b_data & ~PAGE_MASK) {
+ dmu_return_arcbuf(abuf);
+ return NULL;
+ }
+#endif
+
+ return abuf;
+}
+
static int osd_bufs_get_write(const struct lu_env *env, struct osd_object *obj,
- loff_t off, ssize_t len, struct niobuf_local *lnb)
+ loff_t off, ssize_t len, struct niobuf_local *lnb)
{
struct osd_device *osd = osd_obj2dev(obj);
int plen, off_in_block, sz_in_block;
int rc, i = 0, npages = 0;
- arc_buf_t *abuf;
- uint32_t bs;
- uint64_t dummy;
+ dnode_t *dn = obj->oo_dn;
+ arc_buf_t *abuf;
+ uint32_t bs = dn->dn_datablksz;
ENTRY;
- dmu_object_size_from_db(obj->oo_db, &bs, &dummy);
-
/*
* currently only full blocks are subject to zerocopy approach:
* so that we're sure nobody is trying to update the same block
off_in_block = off & (bs - 1);
sz_in_block = min_t(int, bs - off_in_block, len);
+ abuf = NULL;
if (sz_in_block == bs) {
/* full block, try to use zerocopy */
+ abuf = osd_request_arcbuf(dn, bs);
+ if (unlikely(IS_ERR(abuf)))
+ GOTO(out_err, rc = PTR_ERR(abuf));
+ }
- abuf = dmu_request_arcbuf(obj->oo_db, bs);
- if (unlikely(abuf == NULL))
- GOTO(out_err, rc = -ENOMEM);
-
+ if (abuf != NULL) {
atomic_inc(&osd->od_zerocopy_loan);
/* go over pages arcbuf contains, put them as
static int osd_bufs_get(const struct lu_env *env, struct dt_object *dt,
loff_t offset, ssize_t len, struct niobuf_local *lnb,
- int rw)
+ enum dt_bufs_type rw)
{
struct osd_object *obj = osd_dt_obj(dt);
int rc;
LASSERT(dt_object_exists(dt));
- LASSERT(obj->oo_db);
+ LASSERT(obj->oo_dn);
- if (rw == 0)
- rc = osd_bufs_get_read(env, obj, offset, len, lnb);
- else
+ if (rw & DT_BUFS_TYPE_WRITE)
rc = osd_bufs_get_write(env, obj, offset, len, lnb);
+ else
+ rc = osd_bufs_get_read(env, obj, offset, len, lnb);
return rc;
}
struct osd_object *obj = osd_dt_obj(dt);
LASSERT(dt_object_exists(dt));
- LASSERT(obj->oo_db);
+ LASSERT(obj->oo_dn);
return 0;
}
-static inline uint32_t osd_get_blocksz(struct osd_object *obj)
-{
- uint32_t blksz;
- u_longlong_t unused;
-
- LASSERT(obj->oo_db);
-
- dmu_object_size_from_db(obj->oo_db, &blksz, &unused);
- return blksz;
-}
-
static inline uint64_t osd_roundup2blocksz(uint64_t size,
uint64_t offset,
uint32_t blksz)
size += offset % blksz;
- if (likely(IS_PO2(blksz)))
+ if (likely(is_power_of_2(blksz)))
return PO2_ROUNDUP_TYPED(size, blksz, uint64_t);
size += blksz - 1;
struct osd_thandle *oh;
uint64_t offset = 0;
uint32_t size = 0;
- uint32_t blksz = osd_get_blocksz(obj);
+ uint32_t blksz = obj->oo_dn->dn_datablksz;
int i, rc, flags = 0;
bool ignore_quota = false, synced = false;
long long space = 0;
ENTRY;
LASSERT(dt_object_exists(dt));
- LASSERT(obj->oo_db);
+ LASSERT(obj->oo_dn);
LASSERT(lnb);
LASSERT(npages > 0);
continue;
}
- dmu_tx_hold_write(oh->ot_tx, obj->oo_db->db_object,
- offset, size);
+ osd_tx_hold_write(oh->ot_tx, obj->oo_dn->dn_object,
+ obj->oo_dn, offset, size);
/* Estimating space to be consumed by a write is rather
* complicated with ZFS. As a consequence, we don't account for
* indirect blocks and just use as a rough estimate the worse
}
if (size) {
- dmu_tx_hold_write(oh->ot_tx, obj->oo_db->db_object,
+ osd_tx_hold_write(oh->ot_tx, obj->oo_dn->dn_object, obj->oo_dn,
offset, size);
space += osd_roundup2blocksz(size, offset, blksz);
}
- dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
-
oh->ot_write_commit = 1; /* used in osd_trans_start() for fail_loc */
/* backend zfs filesystem might be configured to store multiple data
uint64_t start, uint64_t end)
{
struct osd_device *osd = osd_obj2dev(obj);
- dmu_buf_impl_t *db = (dmu_buf_impl_t *)obj->oo_db;
- dnode_t *dn;
+ dnode_t *dn = obj->oo_dn;
uint32_t blksz;
int rc = 0;
ENTRY;
- DB_DNODE_ENTER(db);
- dn = DB_DNODE(db);
-
if (dn->dn_maxblkid > 0) /* can't change block size */
GOTO(out, rc);
out_unlock:
up_write(&obj->oo_guard);
out:
- DB_DNODE_EXIT(db);
return rc;
}
ENTRY;
LASSERT(dt_object_exists(dt));
- LASSERT(obj->oo_db);
+ LASSERT(obj->oo_dn);
LASSERT(th != NULL);
oh = container_of0(th, struct osd_thandle, ot_super);
continue;
}
+ if (new_size < lnb[i].lnb_file_offset + lnb[i].lnb_len)
+ new_size = lnb[i].lnb_file_offset + lnb[i].lnb_len;
+ if (lnb[i].lnb_page == NULL)
+ continue;
+
if (lnb[i].lnb_page->mapping == (void *)obj) {
- dmu_write(osd->od_os, obj->oo_db->db_object,
- lnb[i].lnb_file_offset, lnb[i].lnb_len,
- kmap(lnb[i].lnb_page), oh->ot_tx);
+ osd_dmu_write(osd, obj->oo_dn, lnb[i].lnb_file_offset,
+ lnb[i].lnb_len, kmap(lnb[i].lnb_page),
+ oh->ot_tx);
kunmap(lnb[i].lnb_page);
+ iosize += lnb[i].lnb_len;
} else if (lnb[i].lnb_data) {
+ int j, apages, abufsz;
LASSERT(((unsigned long)lnb[i].lnb_data & 1) == 0);
/* buffer loaned for zerocopy, try to use it.
* notice that dmu_assign_arcbuf() is smart
* enough to recognize changed blocksize
* in this case it fallbacks to dmu_write() */
- dmu_assign_arcbuf(obj->oo_db, lnb[i].lnb_file_offset,
+ abufsz = arc_buf_size(lnb[i].lnb_data);
+ LASSERT(abufsz & PAGE_MASK);
+ apages = abufsz / PAGE_SIZE;
+ LASSERT(i + apages <= npages);
+ /* these references to pages must be invalidated
+ * to prevent access in osd_bufs_put() */
+ for (j = 0; j < apages; j++)
+ lnb[i + j].lnb_page = NULL;
+ dmu_assign_arcbuf(&obj->oo_dn->dn_bonus->db,
+ lnb[i].lnb_file_offset,
lnb[i].lnb_data, oh->ot_tx);
/* drop the reference, otherwise osd_put_bufs()
* will be releasing it - bad! */
lnb[i].lnb_data = NULL;
atomic_dec(&osd->od_zerocopy_loan);
+ iosize += abufsz;
}
- if (new_size < lnb[i].lnb_file_offset + lnb[i].lnb_len)
- new_size = lnb[i].lnb_file_offset + lnb[i].lnb_len;
- iosize += lnb[i].lnb_len;
}
up_read(&obj->oo_guard);
loff_t eof;
LASSERT(dt_object_exists(dt));
- LASSERT(obj->oo_db);
+ LASSERT(obj->oo_dn);
read_lock(&obj->oo_attr_lock);
eof = obj->oo_attr.la_size;
* dmu_tx_hold_sa() and if off < size, dmu_tx_hold_free()
* called and then assigned to a transaction group.
*/
-static int __osd_object_punch(objset_t *os, dmu_buf_t *db, dmu_tx_t *tx,
+static int __osd_object_punch(objset_t *os, dnode_t *dn, dmu_tx_t *tx,
uint64_t size, uint64_t off, uint64_t len)
{
int rc = 0;
if (len == DMU_OBJECT_END && size == off)
return 0;
+ /* XXX: dnode_free_range() can be used to save on dnode lookup */
if (off < size)
- rc = -dmu_free_range(os, db->db_object, off, len, tx);
+ dmu_free_range(os, dn->dn_object, off, len, tx);
return rc;
}
len = end - start;
write_unlock(&obj->oo_attr_lock);
- rc = __osd_object_punch(osd->od_os, obj->oo_db, oh->ot_tx,
+ rc = __osd_object_punch(osd->od_os, obj->oo_dn, oh->ot_tx,
obj->oo_attr.la_size, start, len);
/* set new size */
if (len == DMU_OBJECT_END) {
/* declare we'll free some blocks ... */
if (start < obj->oo_attr.la_size) {
read_unlock(&obj->oo_attr_lock);
- dmu_tx_hold_free(oh->ot_tx, obj->oo_db->db_object, start, len);
+ dmu_tx_hold_free(oh->ot_tx, obj->oo_dn->dn_object, start, len);
} else {
read_unlock(&obj->oo_attr_lock);
}
- /* ... and we'll modify size attribute */
- dmu_tx_hold_sa(oh->ot_tx, obj->oo_sa_hdl, 0);
-
RETURN(osd_declare_quota(env, osd, obj->oo_attr.la_uid,
obj->oo_attr.la_gid, 0, oh, true, NULL,
false));
git://git.whamcloud.com / fs / lustre-release.git / blobdiff