mds_reint.c

lustre 1.6.5 source code

                        if (rc)
                                GOTO(cleanup, rc);
                } else {
                        rc = obd_iocontrol(OBD_IOC_LOV_SETSTRIPE,
                                           mds->mds_osc_exp, 0,
                                           &lsm, rec->ur_eadata);
                        if (rc)
                                GOTO(cleanup, rc);

                        obd_free_memmd(mds->mds_osc_exp, &lsm);

                        rc = fsfilt_set_md(obd, inode, handle, rec->ur_eadata,
                                           rec->ur_eadatalen, "lov");
                        if (rc)
                                GOTO(cleanup, rc);
                }
        }

        body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
        mds_pack_inode2fid(&body->fid1, inode);
        mds_pack_inode2body(body, inode);

        /* don't return OST-specific attributes if we didn't just set them. */
        if (ia_valid & ATTR_SIZE)
                body->valid |= OBD_MD_FLSIZE | OBD_MD_FLBLOCKS;
        if (ia_valid & (ATTR_MTIME | ATTR_MTIME_SET))
                body->valid |= OBD_MD_FLMTIME;
        if (ia_valid & (ATTR_ATIME | ATTR_ATIME_SET))
                body->valid |= OBD_MD_FLATIME;

        if (rc == 0 && rec->ur_cookielen && !IS_ERR(mds->mds_osc_obd)) {
                OBD_ALLOC(mlcd, sizeof(*mlcd) + rec->ur_cookielen +
                          rec->ur_eadatalen);
                if (mlcd) {
                        mlcd->mlcd_size = sizeof(*mlcd) + rec->ur_cookielen +
                                rec->ur_eadatalen;
                        mlcd->mlcd_eadatalen = rec->ur_eadatalen;
                        mlcd->mlcd_cookielen = rec->ur_cookielen;
                        mlcd->mlcd_lmm = (void *)&mlcd->mlcd_cookies +
                                mlcd->mlcd_cookielen;
                        memcpy(&mlcd->mlcd_cookies, rec->ur_logcookies,
                               mlcd->mlcd_cookielen);
                        memcpy(mlcd->mlcd_lmm, rec->ur_eadata,
                               mlcd->mlcd_eadatalen);
                } else {
                        CERROR("unable to allocate log cancel data\n");
                }
        }
        EXIT;
 cleanup:
        if (mlcd != NULL)
                sync = fsfilt_add_journal_cb(req->rq_export->exp_obd, 0, handle,
                                             mds_cancel_cookies_cb, mlcd);
        err = mds_finish_transno(mds, inode, handle, req, rc, 0, sync);
        /* do mds to ost setattr if needed */
        if (!rc && !err && lmm_size)
                mds_osc_setattr_async(obd, inode, lmm, lmm_size,
                                      logcookies, rec->ur_fid1);

        switch (cleanup_phase) {
        case 2:
                OBD_FREE(lmm, mds->mds_max_mdsize);
                if (logcookies)
                        OBD_FREE(logcookies, cookie_size);
        case 1:
                if ((S_ISREG(inode->i_mode) || S_ISDIR(inode->i_mode)) &&
                    rec->ur_eadata != NULL)
                        UNLOCK_INODE_MUTEX(inode);
                l_dput(de);
                if (locked) {
                        if (rc) {
                                ldlm_lock_decref(&lockh, LCK_EX);
                        } else {
                                ptlrpc_save_lock (req, &lockh, LCK_EX);
                        }
                }
        case 0:
                break;
        default:
                LBUG();
        }
        if (err && !rc)
                rc = err;

        req->rq_status = rc;

        /* trigger dqrel/dqacq for original owner and new owner */
        if (ia_valid & (ATTR_UID | ATTR_GID))
                lquota_adjust(mds_quota_interface_ref, obd, qcids, qpids, rc,
                              FSFILT_OP_SETATTR);

        return 0;
}

static void reconstruct_reint_create(struct mds_update_record *rec, int offset,
                                     struct ptlrpc_request *req)
{
        struct mds_export_data *med = &req->rq_export->exp_mds_data;
        struct mds_obd *obd = &req->rq_export->exp_obd->u.mds;
        struct dentry *parent, *child;
        struct mds_body *body;

        mds_req_from_mcd(req, med->med_mcd);

        if (req->rq_status)
                return;

        parent = mds_fid2dentry(obd, rec->ur_fid1, NULL);
        LASSERT(!IS_ERR(parent));
        child = ll_lookup_one_len(rec->ur_name, parent, rec->ur_namelen - 1);
        LASSERT(!IS_ERR(child));

        body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
        mds_pack_inode2fid(&body->fid1, child->d_inode);
        mds_pack_inode2body(body, child->d_inode);

        l_dput(parent);
        l_dput(child);
}

static int mds_reint_create(struct mds_update_record *rec, int offset,
                            struct ptlrpc_request *req,
                            struct lustre_handle *lh)
{
        struct dentry *dparent = NULL;
        struct mds_obd *mds = mds_req2mds(req);
        struct obd_device *obd = req->rq_export->exp_obd;
        struct dentry *dchild = NULL;
        struct inode *dir = NULL;
        void *handle = NULL;
        struct lustre_handle lockh;
        int rc = 0, err, type = rec->ur_mode & S_IFMT, cleanup_phase = 0;
        int created = 0;
        unsigned int qcids[MAXQUOTAS] = { current->fsuid, current->fsgid };
        unsigned int qpids[MAXQUOTAS] = { 0, 0 };
        struct lvfs_dentry_params dp = LVFS_DENTRY_PARAMS_INIT;
        int rec_pending = 0;
        unsigned int gid = current->fsgid;
        ENTRY;

        LASSERT(offset == REQ_REC_OFF);
        offset = REPLY_REC_OFF;

        LASSERT(!strcmp(req->rq_export->exp_obd->obd_type->typ_name,
                        LUSTRE_MDS_NAME));

        DEBUG_REQ(D_INODE, req, "parent "LPU64"/%u name %s mode %o",
                  rec->ur_fid1->id, rec->ur_fid1->generation,
                  rec->ur_name, rec->ur_mode);

        MDS_CHECK_RESENT(req, reconstruct_reint_create(rec, offset, req));

        if (OBD_FAIL_CHECK(OBD_FAIL_MDS_REINT_CREATE))
                GOTO(cleanup, rc = -ESTALE);

        if (rec->ur_dlm)
                ldlm_request_cancel(req, rec->ur_dlm, 0);

        dparent = mds_fid2locked_dentry(obd, rec->ur_fid1, NULL, LCK_EX, &lockh,
                                        rec->ur_name, rec->ur_namelen - 1,
                                        MDS_INODELOCK_UPDATE);
        if (IS_ERR(dparent)) {
                rc = PTR_ERR(dparent);
                CDEBUG(D_DENTRY, "parent "LPU64"/%u lookup error %d\n",
                               rec->ur_fid1->id, rec->ur_fid1->generation, rc);
                GOTO(cleanup, rc);
        }
        cleanup_phase = 1; /* locked parent dentry */
        dir = dparent->d_inode;
        LASSERT(dir);

        ldlm_lock_dump_handle(D_OTHER, &lockh);

        dchild = ll_lookup_one_len(rec->ur_name, dparent, rec->ur_namelen - 1);
        if (IS_ERR(dchild)) {
                rc = PTR_ERR(dchild);
                CDEBUG(D_DENTRY, "child lookup error %d\n", rc);
                GOTO(cleanup, rc);
        }

        cleanup_phase = 2; /* child dentry */

        OBD_FAIL_WRITE(obd, OBD_FAIL_MDS_REINT_CREATE_WRITE, dir->i_sb);

        if (req->rq_export->exp_connect_flags & OBD_CONNECT_RDONLY) {
                if (dchild->d_inode)
                        GOTO(cleanup, rc = -EEXIST);
                GOTO(cleanup, rc = -EROFS);
        }

        if (dir->i_mode & S_ISGID && S_ISDIR(rec->ur_mode))
                rec->ur_mode |= S_ISGID;

        dchild->d_fsdata = (void *)&dp;
        dp.ldp_inum = (unsigned long)rec->ur_fid2->id;
        dp.ldp_ptr = req;

        if (dir->i_mode & S_ISGID)
                gid = dir->i_gid;
        else
                gid = current->fsgid;

        /* we try to get enough quota to write here, and let ldiskfs
         * decide if it is out of quota or not b=14783 */
        lquota_chkquota(mds_quota_interface_ref, obd,
                        current->fsuid, gid, 1, &rec_pending);

        switch (type) {
        case S_IFREG:{
                handle = fsfilt_start(obd, dir, FSFILT_OP_CREATE, NULL);
                if (IS_ERR(handle))
                        GOTO(cleanup, rc = PTR_ERR(handle));
                rc = ll_vfs_create(dir, dchild, rec->ur_mode, NULL);
                mds_counter_incr(req->rq_export, LPROC_MDS_MKNOD);
                EXIT;
                break;
        }
        case S_IFDIR:{
                handle = fsfilt_start(obd, dir, FSFILT_OP_MKDIR, NULL);
                if (IS_ERR(handle))
                        GOTO(cleanup, rc = PTR_ERR(handle));
                rc = vfs_mkdir(dir, dchild, rec->ur_mode);
                mds_counter_incr(req->rq_export, LPROC_MDS_MKDIR);
                EXIT;
                break;
        }
        case S_IFLNK:{
                handle = fsfilt_start(obd, dir, FSFILT_OP_SYMLINK, NULL);
                if (IS_ERR(handle))
                        GOTO(cleanup, rc = PTR_ERR(handle));
                if (rec->ur_tgt == NULL)        /* no target supplied */
                        rc = -EINVAL;           /* -EPROTO? */
                else
                        rc = ll_vfs_symlink(dir, dchild, rec->ur_tgt, S_IALLUGO);
                mds_counter_incr(req->rq_export, LPROC_MDS_MKNOD);
                EXIT;
                break;
        }
        case S_IFCHR:
        case S_IFBLK:
        case S_IFIFO:
        case S_IFSOCK:{
                int rdev = rec->ur_rdev;
                handle = fsfilt_start(obd, dir, FSFILT_OP_MKNOD, NULL);
                if (IS_ERR(handle))
                        GOTO(cleanup, rc = PTR_ERR(handle));
                rc = vfs_mknod(dir, dchild, rec->ur_mode, rdev);
                mds_counter_incr(req->rq_export, LPROC_MDS_MKNOD);
                EXIT;
                break;
        }
        default:
                CERROR("bad file type %o creating %s\n", type, rec->ur_name);
                dchild->d_fsdata = NULL;
                GOTO(cleanup, rc = -EINVAL);
        }

        /* In case we stored the desired inum in here, we want to clean up. */
        if (dchild->d_fsdata == (void *)(unsigned long)rec->ur_fid2->id)
                dchild->d_fsdata = NULL;

        if (rc) {
                CDEBUG(D_INODE, "error during create: %d\n", rc);
                GOTO(cleanup, rc);
        } else {
                struct iattr iattr;
                struct inode *inode = dchild->d_inode;
                struct mds_body *body;

                created = 1;
                LTIME_S(iattr.ia_atime) = rec->ur_time;
                LTIME_S(iattr.ia_ctime) = rec->ur_time;
                LTIME_S(iattr.ia_mtime) = rec->ur_time;
                iattr.ia_uid = current->fsuid;  /* set by push_ctxt already */
                iattr.ia_gid = gid;
                iattr.ia_valid = ATTR_UID | ATTR_GID | ATTR_ATIME |
                        ATTR_MTIME | ATTR_CTIME;

                if (rec->ur_fid2->id) {
                        LASSERT(rec->ur_fid2->id == inode->i_ino);
                        inode->i_generation = rec->ur_fid2->generation;
                        /* Dirtied and committed by the upcoming setattr. */
                        CDEBUG(D_INODE, "recreated ino %lu with gen %u\n",
                               inode->i_ino, inode->i_generation);
                } else {
                        CDEBUG(D_INODE, "created ino %lu with gen %x\n",
                               inode->i_ino, inode->i_generation);
                }

                rc = fsfilt_setattr(obd, dchild, handle, &iattr, 0);
                if (rc)
                        CERROR("error on child setattr: rc = %d\n", rc);

                iattr.ia_valid = ATTR_MTIME | ATTR_CTIME;
                rc = fsfilt_setattr(obd, dparent, handle, &iattr, 0);
                if (rc)
                        CERROR("error on parent setattr: rc = %d\n", rc);

                if (S_ISDIR(inode->i_mode)) {
                        struct lov_mds_md lmm;
                        int lmm_size = sizeof(lmm);
                        rc = mds_get_md(obd, dir, &lmm, &lmm_size, 1, 0);
                        if (rc > 0) {
                                LOCK_INODE_MUTEX(inode);
                                rc = fsfilt_set_md(obd, inode, handle,
                                                   &lmm, lmm_size, "lov");
                                UNLOCK_INODE_MUTEX(inode);
                        }
                        if (rc)
                                CERROR("error on copy stripe info: rc = %d\n",
                                        rc);
                }

                body = lustre_msg_buf(req->rq_repmsg, offset, sizeof(*body));
                mds_pack_inode2fid(&body->fid1, inode);
                mds_pack_inode2body(body, inode);
        }
        EXIT;

cleanup:
        err = mds_finish_transno(mds, dir, handle, req, rc, 0, 0);