handler.c

lustre 1.6.5 source code

        if (rc)
                RETURN(rc);
        exp = class_conn2export(conn);
        LASSERT(exp);
        med = &exp->exp_mds_data;

        rc = mds_connect_internal(exp, data);
        if (rc)
                GOTO(out, rc);

        OBD_ALLOC(mcd, sizeof(*mcd));
        if (!mcd)
                GOTO(out, rc = -ENOMEM);

        memcpy(mcd->mcd_uuid, cluuid, sizeof(mcd->mcd_uuid));
        med->med_mcd = mcd;

        rc = mds_client_add(obd, exp, -1, *client_nid);
        GOTO(out, rc);

out:
        if (rc) {
                if (mcd) {
                        OBD_FREE(mcd, sizeof(*mcd));
                        med->med_mcd = NULL;
                }
                class_disconnect(exp);
        } else {
                class_export_put(exp);
        }

        RETURN(rc);
}

int mds_init_export(struct obd_export *exp)
{
        struct mds_export_data *med = &exp->exp_mds_data;

        INIT_LIST_HEAD(&med->med_open_head);
        spin_lock_init(&med->med_open_lock);
        
        spin_lock(&exp->exp_lock);
        exp->exp_connecting = 1;
        spin_unlock(&exp->exp_lock);

        RETURN(0);
}

static int mds_destroy_export(struct obd_export *export)
{
        struct mds_export_data *med;
        struct obd_device *obd = export->exp_obd;
        struct mds_obd *mds = &obd->u.mds;
        struct lvfs_run_ctxt saved;
        struct lov_mds_md *lmm;
        struct llog_cookie *logcookies;
        int rc = 0;
        ENTRY;

        med = &export->exp_mds_data;
        target_destroy_export(export);

        if (obd_uuid_equals(&export->exp_client_uuid, &obd->obd_uuid))
                RETURN(0);

        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        /* Close any open files (which may also cause orphan unlinking). */

        OBD_ALLOC(lmm, mds->mds_max_mdsize);
        if (lmm == NULL) {
                CWARN("%s: allocation failure during cleanup; can not force "
                      "close file handles on this service.\n", obd->obd_name);
                GOTO(out, rc = -ENOMEM);
        }

        OBD_ALLOC(logcookies, mds->mds_max_cookiesize);
        if (logcookies == NULL) {
                CWARN("%s: allocation failure during cleanup; can not force "
                      "close file handles on this service.\n", obd->obd_name);
                OBD_FREE(lmm, mds->mds_max_mdsize);
                GOTO(out, rc = -ENOMEM);
        }

        spin_lock(&med->med_open_lock);
        while (!list_empty(&med->med_open_head)) {
                struct list_head *tmp = med->med_open_head.next;
                struct mds_file_data *mfd =
                        list_entry(tmp, struct mds_file_data, mfd_list);
                int lmm_size = mds->mds_max_mdsize;
                umode_t mode = mfd->mfd_dentry->d_inode->i_mode;
                __u64 valid = 0;

                /* Remove mfd handle so it can't be found again.
                 * We are consuming the mfd_list reference here. */
                mds_mfd_unlink(mfd, 0);
                spin_unlock(&med->med_open_lock);

                /* If you change this message, be sure to update
                 * replay_single:test_46 */
                CDEBUG(D_INODE|D_IOCTL, "%s: force closing file handle for "
                       "%.*s (ino %lu)\n", obd->obd_name,
                       mfd->mfd_dentry->d_name.len,mfd->mfd_dentry->d_name.name,
                       mfd->mfd_dentry->d_inode->i_ino);

                rc = mds_get_md(obd, mfd->mfd_dentry->d_inode, lmm,
                                &lmm_size, 1, 0);
                if (rc < 0)
                        CWARN("mds_get_md failure, rc=%d\n", rc);
                else
                        valid |= OBD_MD_FLEASIZE;

                /* child orphan sem protects orphan_dec_test and
                 * is_orphan race, mds_mfd_close drops it */
                MDS_DOWN_WRITE_ORPHAN_SEM(mfd->mfd_dentry->d_inode);

                rc = mds_mfd_close(NULL, REQ_REC_OFF, obd, mfd,
                                   !(export->exp_flags & OBD_OPT_FAILOVER),
                                   lmm, lmm_size, logcookies,
                                   mds->mds_max_cookiesize,
                                   &valid);

                if (rc)
                        CDEBUG(D_INODE|D_IOCTL, "Error closing file: %d\n", rc);

                if (valid & OBD_MD_FLCOOKIE) {
                        rc = mds_osc_destroy_orphan(obd, mode, lmm,
                                                    lmm_size, logcookies, 1);
                        if (rc < 0) {
                                CDEBUG(D_INODE, "%s: destroy of orphan failed,"
                                       " rc = %d\n", obd->obd_name, rc);
                                rc = 0;
                        }
                        valid &= ~OBD_MD_FLCOOKIE;
                }

                spin_lock(&med->med_open_lock);
        }

        OBD_FREE(logcookies, mds->mds_max_cookiesize);
        OBD_FREE(lmm, mds->mds_max_mdsize);

        spin_unlock(&med->med_open_lock);

        pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        mds_client_free(export);

 out:
        RETURN(rc);
}

static int mds_disconnect(struct obd_export *exp)
{
        int rc = 0;
        ENTRY;

        LASSERT(exp);
        class_export_get(exp);

        /* Disconnect early so that clients can't keep using export */
        rc = class_disconnect(exp);
        if (exp->exp_obd->obd_namespace != NULL)
                ldlm_cancel_locks_for_export(exp);

        /* complete all outstanding replies */
        spin_lock(&exp->exp_lock);
        while (!list_empty(&exp->exp_outstanding_replies)) {
                struct ptlrpc_reply_state *rs =
                        list_entry(exp->exp_outstanding_replies.next,
                                   struct ptlrpc_reply_state, rs_exp_list);
                struct ptlrpc_service *svc = rs->rs_service;

                spin_lock(&svc->srv_lock);
                list_del_init(&rs->rs_exp_list);
                ptlrpc_schedule_difficult_reply(rs);
                spin_unlock(&svc->srv_lock);
        }
        spin_unlock(&exp->exp_lock);

        class_export_put(exp);
        RETURN(rc);
}

static int mds_getstatus(struct ptlrpc_request *req)
{
        struct mds_obd *mds = mds_req2mds(req);
        struct mds_body *body;
        int rc, size[2] = { sizeof(struct ptlrpc_body), sizeof(*body) };
        ENTRY;

        OBD_FAIL_RETURN(OBD_FAIL_MDS_GETSTATUS_PACK, req->rq_status = -ENOMEM);
        rc = lustre_pack_reply(req, 2, size, NULL);
        if (rc)
                RETURN(req->rq_status = rc);

        body = lustre_msg_buf(req->rq_repmsg, REPLY_REC_OFF, sizeof(*body));
        memcpy(&body->fid1, &mds->mds_rootfid, sizeof(body->fid1));

        /* the last_committed and last_xid fields are filled in for all
         * replies already - no need to do so here also.
         */
        RETURN(0);
}

/* get the LOV EA from @inode and store it into @md.  It can be at most
 * @size bytes, and @size is updated with the actual EA size.
 * The EA size is also returned on success, and -ve errno on failure. 
 * If there is no EA then 0 is returned. */
int mds_get_md(struct obd_device *obd, struct inode *inode, void *md,
               int *size, int lock, int flags)
{
        int rc = 0;
        int lmm_size = 0;

        if (lock)
                LOCK_INODE_MUTEX(inode);
        rc = fsfilt_get_md(obd, inode, md, *size, "lov");

        if (rc == 0 && flags == MDS_GETATTR)
                rc = mds_get_default_md(obd, md, &lmm_size);

        if (rc < 0) {
                CERROR("Error %d reading eadata for ino %lu\n",
                       rc, inode->i_ino);
        } else if (rc > 0) {
                lmm_size = rc;
                rc = mds_convert_lov_ea(obd, inode, md, lmm_size);

                if (rc == 0) {
                        *size = lmm_size;
                        rc = lmm_size;
                } else if (rc > 0) {
                        *size = rc;
                }
        } else {
                *size = 0;
        }
        if (lock)
                UNLOCK_INODE_MUTEX(inode);

        RETURN (rc);
}


/* Call with lock=1 if you want mds_pack_md to take the i_mutex.
 * Call with lock=0 if the caller has already taken the i_mutex. */
int mds_pack_md(struct obd_device *obd, struct lustre_msg *msg, int offset,
                struct mds_body *body, struct inode *inode, int lock, int flags)
{
        struct mds_obd *mds = &obd->u.mds;
        void *lmm;
        int lmm_size;
        int rc;
        ENTRY;

        lmm = lustre_msg_buf(msg, offset, 0);
        if (lmm == NULL) {
                /* Some problem with getting eadata when I sized the reply
                 * buffer... */
                CDEBUG(D_INFO, "no space reserved for inode %lu MD\n",
                       inode->i_ino);
                RETURN(0);
        }
        lmm_size = lustre_msg_buflen(msg, offset);

        /* I don't really like this, but it is a sanity check on the client
         * MD request.  However, if the client doesn't know how much space
         * to reserve for the MD, it shouldn't be bad to have too much space.
         */
        if (lmm_size > mds->mds_max_mdsize) {
                CWARN("Reading MD for inode %lu of %d bytes > max %d\n",
                       inode->i_ino, lmm_size, mds->mds_max_mdsize);
                // RETURN(-EINVAL);
        }

        rc = mds_get_md(obd, inode, lmm, &lmm_size, lock, flags);
        if (rc > 0) {
                if (S_ISDIR(inode->i_mode))
                        body->valid |= OBD_MD_FLDIREA;
                else
                        body->valid |= OBD_MD_FLEASIZE;
                body->eadatasize = lmm_size;
                rc = 0;
        }

        RETURN(rc);
}

#ifdef CONFIG_FS_POSIX_ACL
static
int mds_pack_posix_acl(struct inode *inode, struct lustre_msg *repmsg,
                       struct mds_body *repbody, int repoff)
{
        struct dentry de = { .d_inode = inode };
        int buflen, rc;
        ENTRY;

        LASSERT(repbody->aclsize == 0);
        LASSERT(lustre_msg_bufcount(repmsg) > repoff);

        buflen = lustre_msg_buflen(repmsg, repoff);
        if (!buflen)
                GOTO(out, 0);

        if (!inode->i_op || !inode->i_op->getxattr)
                GOTO(out, 0);

        lock_24kernel();
        rc = inode->i_op->getxattr(&de, MDS_XATTR_NAME_ACL_ACCESS,
                                   lustre_msg_buf(repmsg, repoff, buflen),
                                   buflen);
        unlock_24kernel();

        if (rc >= 0)
                repbody->aclsize = rc;
        else if (rc != -ENODATA) {
                CERROR("buflen %d, get acl: %d\n", buflen, rc);
                RETURN(rc);
        }
        EXIT;
out:
        repbody->valid |= OBD_MD_FLACL;
        return 0;
}
#else
#define mds_pack_posix_acl(inode, repmsg, repbody, repoff) 0
#endif

int mds_pack_acl(struct mds_export_data *med, struct inode *inode,
                 struct lustre_msg *repmsg, struct mds_body *repbody,
                 int repoff)
{
        return mds_pack_posix_acl(inode, repmsg, repbody, repoff);
}

static int mds_getattr_internal(struct obd_device *obd, struct dentry *dentry,
                                struct ptlrpc_request *req,
                                struct mds_body *reqbody, int reply_off)
{
        struct mds_body *body;
        struct inode *inode = dentry->d_inode;
        int rc = 0;
        int flags = 0;
        ENTRY;

        if (inode == NULL)
                RETURN(-ENOENT);

        body = lustre_msg_buf(req->rq_repmsg, reply_off, sizeof(*body));
        LASSERT(body != NULL);                 /* caller prepped reply */

        mds_pack_inode2fid(&body->fid1, inode);
        body->flags = reqbody->flags; /* copy MDS_BFLAG_EXT_FLAGS if present */
        mds_pack_inode2body(body, inode);
        reply_off++;