mds_join.c

lustre 1.6.5 source code

        struct ll_fid head_fid;
        int rc;
        ENTRY;

        if (lockh)
                ldlm_lock_decref(lockh, LCK_EX);

        head_inode = dchild->d_inode;
        mdc_pack_fid(&head_fid, head_inode->i_ino, head_inode->i_generation,
                      head_inode->i_mode & S_IFMT);

        rc = mds_get_parents_children_locked(obd, mds, &join_rec->jr_fid,
                                             &de_tailparent, &head_fid,
                                             &de_head, LCK_EX, rec->ur_name,
                                             rec->ur_namelen, &de_tail,
                                             NULL, 0, NULL, dlm_handles,
                                             LCK_EX);
        if (rc)
                GOTO(cleanup, rc);

        *lockh = dlm_handles[1];
        LASSERT(de_tailparent);
        tail_inode = de_tail->d_inode;
        if (tail_inode == NULL) {
                CERROR("tail inode doesn't exist(dir %lu,name %s)!\n",
                       de_tailparent? de_tailparent->d_inode->i_ino : 0,
                       rec->ur_name);
                GOTO(cleanup, rc = -ENOENT);
        }

        if (!S_ISREG(tail_inode->i_mode)) {
                CERROR("tail file is not a regular file (dir %lu, name %s)!\n",
                       de_tailparent? de_tailparent->d_inode->i_ino : 0,
                       rec->ur_name);
                GOTO(cleanup, rc = -EINVAL);
        }

        *handle = fsfilt_start(obd, head_inode, FSFILT_OP_JOIN, NULL);
        if (IS_ERR(*handle)) {
                rc = PTR_ERR(*handle);
                GOTO(cleanup, rc);
        }

        rc = mds_get_md(obd, tail_inode, tail_lmm, &lmm_size, 1, 0);
        if (rc < 0) /* get md fails */
                GOTO(cleanup, rc);

        LASSERT(le32_to_cpu(tail_lmm->lmm_magic) == LOV_MAGIC_JOIN ||
                le32_to_cpu(tail_lmm->lmm_magic) == LOV_MAGIC);

        LASSERT(de_tailparent);
        rc = vfs_unlink(de_tailparent->d_inode, de_tail);

        if (rc == 0) {
                CDEBUG(D_INODE, "delete the tail inode %lu/%u \n",
                       tail_inode->i_ino, tail_inode->i_generation);
        }
cleanup:
        if (dlm_handles[2].cookie != 0)
                ldlm_lock_decref(&dlm_handles[2], LCK_EX);

        if (dlm_handles[0].cookie != 0) {
                if (rc)
                        ldlm_lock_decref(&dlm_handles[0], LCK_EX);
                else
                        ptlrpc_save_lock(req, &dlm_handles[0], LCK_EX);
        }
        if (de_tail)
                l_dput(de_tail);

        if (de_tailparent)
                l_dput(de_tailparent);

        if (de_head)
                l_dput(de_head);

        RETURN(rc);
}

int mds_join_file(struct mds_update_record *rec, struct ptlrpc_request *req,
                  struct dentry *de_head, struct lustre_handle *lockh)
{
        struct mds_obd *mds = mds_req2mds(req);
        struct obd_device *obd = req->rq_export->exp_obd;
        struct inode *head_inode = NULL;
        struct lvfs_run_ctxt saved;
        void *handle = NULL;
        struct lov_mds_md *head_lmm, *tail_lmm;
        struct lov_mds_md_join *head_lmmj = NULL, *tail_lmmj = NULL;
        int lmm_size, rc = 0, cleanup_phase = 0, size;
        struct llog_handle *llh_head = NULL, *llh_tail = NULL;
        struct llog_ctxt *ctxt = NULL;
        struct mds_rec_join *join_rec;
        ENTRY;

        join_rec = lustre_swab_reqbuf(req, DLM_INTENT_REC_OFF + 3,
                                      sizeof(*join_rec),
                                      lustre_swab_mds_rec_join);
        if (join_rec == NULL)
                RETURN (-EFAULT);

        DEBUG_REQ(D_INODE, req,"head "LPU64"/%u, ptail ino "LPU64"/%u, tail %s",
                  rec->ur_fid1->id, rec->ur_fid1->generation,
                  join_rec->jr_fid.id, join_rec->jr_fid.generation,
                  rec->ur_name);

        size = mds->mds_max_mdsize;
        lmm_size = mds->mds_max_mdsize;
        OBD_ALLOC(head_lmm, lmm_size);
        OBD_ALLOC(tail_lmm, lmm_size);
        if (!head_lmm || !tail_lmm)
                GOTO(cleanup, rc = -ENOMEM);

        /* acquire head's dentry */
        LASSERT(de_head);
        head_inode = de_head->d_inode;
        if (head_inode == NULL) {
                CERROR("head inode doesn't exist!\n");
                GOTO(cleanup, rc = -ENOENT);
        }

        /*Unlink tail inode and get the lmm back*/
        rc = mds_join_unlink_tail_inode(rec, req, join_rec, tail_lmm, lmm_size,
                                        de_head, &handle, lockh);
        if (rc) {
                CERROR("unlink tail_inode error %d\n", rc);
                GOTO(cleanup, rc);
        }

        LOCK_INODE_MUTEX(head_inode);
        cleanup_phase = 1;
        rc = mds_get_md(obd, head_inode, head_lmm, &size, 0, 0);
        if (rc < 0)
                GOTO(cleanup, rc);

        LASSERT(le32_to_cpu(head_lmm->lmm_magic) == LOV_MAGIC_JOIN ||
                le32_to_cpu(head_lmm->lmm_magic) == LOV_MAGIC);

        push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        ctxt = llog_get_context(obd, LLOG_LOVEA_ORIG_CTXT);
        LASSERT(ctxt != NULL);
        cleanup_phase = 2;
        if (le32_to_cpu(head_lmm->lmm_magic) == LOV_MAGIC) { /*simple file */
                struct llog_logid *llog_array;

                rc = llog_create(ctxt, &llh_head, NULL, NULL);
                if (rc) {
                        CERROR("cannot create new log, error = %d\n", rc);
                        GOTO(cleanup, rc);
                }
                cleanup_phase = 3;
                llog_array = &llh_head->lgh_id;
                CDEBUG(D_INFO,"create arrary for %lu with id "LPU64":"LPU64"\n",
                       head_inode->i_ino, llog_array->lgl_oid,
                       llog_array->lgl_ogr);
                rc = llog_init_handle(llh_head, LLOG_F_IS_PLAIN, NULL);
                if (rc)
                        GOTO(cleanup, rc);
                OBD_ALLOC_PTR(head_lmmj);
                if (head_lmmj == NULL)
                        GOTO(cleanup, rc = -ENOMEM);
                mds_init_stripe_join(head_lmmj, head_lmm, llog_array);
                mds_insert_join_lmm(llh_head, head_lmm, 0,join_rec->jr_headsize,
                                    head_lmmj);
        } else { /*head lmm is join file */
                head_lmmj = (struct lov_mds_md_join *)head_lmm;
                /* construct and fill extent llog object */
                rc = llog_create(ctxt, &llh_head,
                                 &head_lmmj->lmmj_array_id, NULL);
                if (rc) {
                        CERROR("cannot open existing log, error = %d\n", rc);
                        GOTO(cleanup, rc);
                }
                cleanup_phase = 3;
                rc = llog_init_handle(llh_head, LLOG_F_IS_PLAIN, NULL);
                if (rc)
                        GOTO(cleanup, rc);
                rc = mds_adjust_last_extent(llh_head, join_rec->jr_headsize);
                if (rc) {
                        CERROR("can't adjust last extent of obj rc=%d\n", rc);
                        GOTO(cleanup, rc);
                }
        }

        if (le32_to_cpu(tail_lmm->lmm_magic) != LOV_MAGIC_JOIN) {
                mds_insert_join_lmm(llh_head, tail_lmm, join_rec->jr_headsize,
                                    -1, head_lmmj);
        } else {
                struct mdsea_cb_data cbdata;
                tail_lmmj = (struct lov_mds_md_join *)tail_lmm;

                rc = llog_create(ctxt,&llh_tail,&tail_lmmj->lmmj_array_id,NULL);
                if (rc) {
                        CERROR("cannot open existing log, error = %d\n", rc);
                        GOTO(cleanup, rc);
                }
                rc = llog_init_handle(llh_tail, LLOG_F_IS_PLAIN, NULL);
                if (rc) {
                        llog_close(llh_tail);
                        GOTO(cleanup, rc);
                }
                cbdata.mc_llh = llh_head;
                cbdata.mc_headfile_sz = join_rec->jr_headsize;
                cbdata.mc_lmm_join = head_lmmj;
                rc = mdsea_iterate(llh_tail, (llog_cb_t)mdsea_append_extent,
                                   &cbdata);
                if (rc) {
                        llog_close(llh_tail);
                        CERROR("can not append extent log error %d \n", rc);
                        GOTO(cleanup, rc);
                }
                rc = llog_destroy(llh_tail);
                if (rc) {
                        llog_close(llh_tail);
                        CERROR("can not destroy log error %d \n", rc);
                        GOTO(cleanup, rc);
                }
                llog_free_handle(llh_tail);
        }
        LASSERT(head_inode);
        CDEBUG(D_INODE, "join finish, set lmm V2 to inode %lu \n",
               head_inode->i_ino);
        fsfilt_set_md(obd, head_inode, handle, head_lmmj,
                      sizeof(struct lov_mds_md_join), "lov");
        mds_finish_join(mds, req, head_inode, head_lmmj);
cleanup:
        rc = mds_finish_transno(mds, head_inode, handle, req, rc, 0, 0);
        switch(cleanup_phase){
        case 3:
                llog_close(llh_head);
        case 2:
                llog_ctxt_put(ctxt);
                if (head_lmmj && ((void*)head_lmmj != (void*)head_lmm))
                        OBD_FREE_PTR(head_lmmj);

                pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
        case 1:
                UNLOCK_INODE_MUTEX(head_inode);
        case 0:
                if (tail_lmm != NULL)
                        OBD_FREE(tail_lmm, lmm_size);
                if (head_lmm != NULL)
                        OBD_FREE(head_lmm, lmm_size);
                break;
        default:
                CERROR("invalid cleanup_phase %d\n", cleanup_phase);
                LBUG();
        }
        req->rq_status = rc;
        RETURN(rc);
}