filter_io_24.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 545 行 · 第 1/2 页

        rc = fsfilt_send_bio(rw, obd, inode, iobuf);

        CDEBUG(D_INFO, "tried to %s %d pages, rc = %d\n",
               rw & OBD_BRW_WRITE ? "write" : "read", iobuf->nr_pages, rc);

        if (rc > 0)
                rc = 0;

        EXIT;
cleanup:
        if (!committed && (rw == OBD_BRW_WRITE)) {
                int err = fsfilt_commit_async(obd, inode,
                                              oti->oti_handle, wait_handle);
                if (err)
                        CERROR("can't close transaction: %d\n", err);
                /*
                 * this is error path, so we prefer to return
                 * original error, not this one
                 */
        }

        switch(cleanup_phase) {
        case 3:
        case 2:
                unlock_kiovec(1, &iobuf);
        case 1:
        case 0:
                if (cleanup_phase != 3 && rw == OBD_BRW_WRITE)
                        up(&inode->i_sem);
                break;
        default:
                CERROR("corrupt cleanup_phase (%d)?\n", cleanup_phase);
                LBUG();
                break;
        }
        return rc;
}

/* See if there are unallocated parts in given file region */
int filter_range_is_mapped(struct inode *inode, obd_size offset, int len)
{
        int (*fs_bmap)(struct address_space *, long) =
                inode->i_mapping->a_ops->bmap;
        int j;

        /* We can't know if the range is mapped already or not */
        if (fs_bmap == NULL)
                return 0;

        offset >>= inode->i_blkbits;
        len >>= inode->i_blkbits;

        for (j = 0; j < len; j++)
                if (fs_bmap(inode->i_mapping, offset + j) == 0)
                        return 0;

        return 1;
}

/* some kernels require alloc_kiovec callers to zero members through the use of
 * map_user_kiobuf and unmap_.. we don't use those, so we have a little helper
 * that makes sure we don't break the rules. */
static void clear_kiobuf(struct kiobuf *iobuf)
{
        int i;

        for (i = 0; i < iobuf->array_len; i++)
                iobuf->maplist[i] = NULL;

        iobuf->nr_pages = 0;
        iobuf->offset = 0;
        iobuf->length = 0;
}

struct filter_iobuf *filter_alloc_iobuf(struct filter_obd *filter,
                                        int rw, int num_pages)
{
        struct kiobuf *iobuf;
        int rc;
        ENTRY;

        LASSERTF(rw == OBD_BRW_WRITE || rw == OBD_BRW_READ, "%x\n", rw);

        rc = alloc_kiovec(1, &iobuf);
        if (rc)
                RETURN(ERR_PTR(rc));

        rc = expand_kiobuf(iobuf, num_pages);
        if (rc) {
                free_kiovec(1, &iobuf);
                RETURN(ERR_PTR(rc));
        }

#ifdef HAVE_KIOBUF_DOVARY
        iobuf->dovary = 0; /* this prevents corruption, not present in 2.4.20 */
#endif
        clear_kiobuf(iobuf);
        RETURN((void *)iobuf);
}

void filter_free_iobuf(struct filter_iobuf *buf)
{
        struct kiobuf *iobuf = (void *)buf;

        clear_kiobuf(iobuf);
        free_kiovec(1, &iobuf);
}

void filter_iobuf_put(struct filter_obd *filter, struct filter_iobuf *iobuf,
                      struct obd_trans_info *oti)
{
        int thread_id = oti ? oti->oti_thread_id : -1;

        if (unlikely(thread_id < 0)) {
                filter_free_iobuf(iobuf);
                return;
        }

        LASSERTF(filter->fo_iobuf_pool[thread_id] == iobuf,
                 "iobuf mismatch for thread %d: pool %p iobuf %p\n",
                 thread_id, filter->fo_iobuf_pool[thread_id], iobuf);
        clear_kiobuf((void *)iobuf);
}

int filter_iobuf_add_page(struct obd_device *obd, struct filter_iobuf *buf,
                           struct inode *inode, struct page *page)
{
        struct kiobuf *iobuf = (void *)buf;

        iobuf->maplist[iobuf->nr_pages++] = page;
        iobuf->length += CFS_PAGE_SIZE;

        return 0;
}

int filter_commitrw_write(struct obd_export *exp, struct obdo *oa, int objcount,
                          struct obd_ioobj *obj, int niocount,
                          struct niobuf_local *res, struct obd_trans_info *oti,
                          int rc)
{
        struct obd_device *obd = exp->exp_obd;
        struct lvfs_run_ctxt saved;
        struct niobuf_local *lnb;
        struct fsfilt_objinfo fso;
        struct iattr iattr = { 0 };
        void *iobuf = NULL;
        struct inode *inode = NULL;
        int i, n, cleanup_phase = 0, err;
        unsigned long now = jiffies; /* DEBUGGING OST TIMEOUTS */
        void *wait_handle;
        ENTRY;
        LASSERT(oti != NULL);
        LASSERT(objcount == 1);
        LASSERT(current->journal_info == NULL);

        if (rc != 0)
                GOTO(cleanup, rc);

        iobuf = filter_iobuf_get(&obd->u.filter, oti);
        if (IS_ERR(iobuf))
                GOTO(cleanup, rc = PTR_ERR(iobuf));
        cleanup_phase = 1;

        fso.fso_dentry = res->dentry;
        fso.fso_bufcnt = obj->ioo_bufcnt;
        inode = res->dentry->d_inode;

        for (i = 0, lnb = res, n = 0; i < obj->ioo_bufcnt; i++, lnb++) {
                loff_t this_size;

                /* If overwriting an existing block, we don't need a grant */
                if (!(lnb->flags & OBD_BRW_GRANTED) && lnb->rc == -ENOSPC &&
                    filter_range_is_mapped(inode, lnb->offset, lnb->len))
                        lnb->rc = 0;

                if (lnb->rc) /* ENOSPC, network RPC error */
                        continue;

                filter_iobuf_add_page(obd, iobuf, inode, lnb->page);

                /* We expect these pages to be in offset order, but we'll
                 * be forgiving */
                this_size = lnb->offset + lnb->len;
                if (this_size > iattr.ia_size)
                        iattr.ia_size = this_size;
        }

        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        cleanup_phase = 2;

        down(&inode->i_sem);
        oti->oti_handle = fsfilt_brw_start(obd, objcount, &fso, niocount, res,
                                           oti);
        if (IS_ERR(oti->oti_handle)) {
                up(&inode->i_sem);
                rc = PTR_ERR(oti->oti_handle);
                CDEBUG(rc == -ENOSPC ? D_INODE : D_ERROR,
                       "error starting transaction: rc = %d\n", rc);
                oti->oti_handle = NULL;
                GOTO(cleanup, rc);
        }

        fsfilt_check_slow(obd, now, "brw_start");

        i = OBD_MD_FLATIME | OBD_MD_FLMTIME | OBD_MD_FLCTIME;

        /* If the inode still has SUID+SGID bits set (see filter_precreate())
         * then we will accept the UID+GID if sent by the client for
         * initializing the ownership of this inode.  We only allow this to
         * happen once (so clear these bits) and later only allow setattr. */
        if (inode->i_mode & S_ISUID)
                i |= OBD_MD_FLUID;
        if (inode->i_mode & S_ISGID)
                i |= OBD_MD_FLGID;

        iattr_from_obdo(&iattr, oa, i);
        if (iattr.ia_valid & (ATTR_UID | ATTR_GID)) {
                CDEBUG(D_INODE, "update UID/GID to %lu/%lu\n",
                       (unsigned long)oa->o_uid, (unsigned long)oa->o_gid);

                cap_raise(current->cap_effective, CAP_SYS_RESOURCE);

                iattr.ia_valid |= ATTR_MODE;
                iattr.ia_mode = inode->i_mode;
                if (iattr.ia_valid & ATTR_UID)
                        iattr.ia_mode &= ~S_ISUID;
                if (iattr.ia_valid & ATTR_GID)
                        iattr.ia_mode &= ~S_ISGID;

                rc = filter_update_fidea(exp, inode, oti->oti_handle, oa);
        }

        /* filter_direct_io drops i_sem */
        rc = filter_direct_io(OBD_BRW_WRITE, res->dentry, iobuf, exp, &iattr,
                              oti, &wait_handle);
        if (rc == 0)
                obdo_from_inode(oa, inode, FILTER_VALID_FLAGS);

        fsfilt_check_slow(obd, now, "direct_io");

        err = fsfilt_commit_wait(obd, inode, wait_handle);
        if (err) {
                CERROR("Failure to commit OST transaction (%d)?\n", err);
                rc = err;
        }
        if (obd->obd_replayable && !rc)
                LASSERTF(oti->oti_transno <= obd->obd_last_committed,
                         "oti_transno "LPU64" last_committed "LPU64"\n",
                         oti->oti_transno, obd->obd_last_committed);
        fsfilt_check_slow(obd, now, "commitrw commit");

cleanup:
        filter_grant_commit(exp, niocount, res);

        switch (cleanup_phase) {
        case 2:
                pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
                LASSERT(current->journal_info == NULL);
        case 1:
                filter_iobuf_put(&obd->u.filter, iobuf, oti);
        case 0:
                /*
                 * lnb->page automatically returns back into per-thread page
                 * pool (bug 5137)
                 */
                f_dput(res->dentry);
        }

        RETURN(rc);
}