Branch: b_new_cmd

1)set max md/cookie size for mdc of cmm
2)fix unlink group log problem
3)some other fixes

diff --git a/lustre/cmm/cmm_device.c b/lustre/cmm/cmm_device.c
index a649d6e..fe6d349 100644 (file)
--- a/lustre/cmm/cmm_device.c
+++ b/lustre/cmm/cmm_device.c
@@ -121,6 +121,27 @@ static struct md_device_operations cmm_md_ops = {
 
 extern struct lu_device_type mdc_device_type;
 
+static int cmm_post_init_mdc(const struct lu_env *env, 
+                             struct cmm_device *cmm)
+{
+        int max_mdsize, max_cookiesize, rc;
+        struct mdc_device *mc, *tmp;
+
+        /* get the max mdsize and cookiesize from lower layer */
+        rc = cmm_maxsize_get(env, &cmm->cmm_md_dev, &max_mdsize, 
+                                                &max_cookiesize);
+        if (rc)
+                RETURN(rc);
+
+        spin_lock(&cmm->cmm_tgt_guard);
+        list_for_each_entry_safe(mc, tmp, &cmm->cmm_targets,
+                                 mc_linkage) {
+                mdc_init_ea_size(env, mc, max_mdsize, max_cookiesize);
+        }
+        spin_unlock(&cmm->cmm_tgt_guard); 
+        RETURN(rc);
+}
+
 /* --- cmm_lu_operations --- */
 /* add new MDC to the CMM, create MDC lu_device and connect it to mdc_obd */
 static int cmm_add_mdc(const struct lu_env *env,
@@ -163,32 +184,37 @@ static int cmm_add_mdc(const struct lu_env *env,
         }
         /* pass config to the just created MDC */
         rc = ld->ld_ops->ldo_process_config(env, ld, cfg);
-        if (rc == 0) {
-                spin_lock(&cm->cmm_tgt_guard);
-                list_for_each_entry_safe(mc, tmp, &cm->cmm_targets,
-                                         mc_linkage) {
-                        if (mc->mc_num == mdc_num) {
-                                spin_unlock(&cm->cmm_tgt_guard);
-                                ldt->ldt_ops->ldto_device_fini(env, ld);
-                                ldt->ldt_ops->ldto_device_free(env, ld);
-                                RETURN(-EEXIST);
-                        }
+        if (rc)
+                RETURN(rc);
+
+        spin_lock(&cm->cmm_tgt_guard);
+        list_for_each_entry_safe(mc, tmp, &cm->cmm_targets,
+                                 mc_linkage) {
+                if (mc->mc_num == mdc_num) {
+                        spin_unlock(&cm->cmm_tgt_guard);
+                        ldt->ldt_ops->ldto_device_fini(env, ld);
+                        ldt->ldt_ops->ldto_device_free(env, ld);
+                        RETURN(-EEXIST);
                 }
-                mc = lu2mdc_dev(ld);
-                list_add_tail(&mc->mc_linkage, &cm->cmm_targets);
-                cm->cmm_tgt_count++;
-                spin_unlock(&cm->cmm_tgt_guard);
+        }
+        mc = lu2mdc_dev(ld);
+        list_add_tail(&mc->mc_linkage, &cm->cmm_targets);
+        cm->cmm_tgt_count++;
+        spin_unlock(&cm->cmm_tgt_guard);
 
-                lu_device_get(cmm2lu_dev(cm));
+        lu_device_get(cmm2lu_dev(cm));
 
-                ls = cm->cmm_md_dev.md_lu_dev.ld_site;
+        ls = cm->cmm_md_dev.md_lu_dev.ld_site;
 
-                target.ft_srv = NULL;
-                target.ft_idx = mc->mc_num;
-                target.ft_exp = mc->mc_desc.cl_exp;
+        target.ft_srv = NULL;
+        target.ft_idx = mc->mc_num;
+        target.ft_exp = mc->mc_desc.cl_exp;
 
-                fld_client_add_target(ls->ls_client_fld, &target);
-        }
+        fld_client_add_target(ls->ls_client_fld, &target);
+        
+        /* set max md size for the mdc */
+        rc = cmm_post_init_mdc(env, cm);
+        
         RETURN(rc);
 }
 
@@ -287,8 +313,15 @@ int cmm_upcall(const struct lu_env *env, struct md_device *md,
         upcall_dev = md->md_upcall.mu_upcall_dev;
 
         LASSERT(upcall_dev);
-        rc = upcall_dev->md_upcall.mu_upcall(env, md->md_upcall.mu_upcall_dev, ev);
-
+        switch (ev) {
+                case MD_LOV_SYNC:
+                        rc = cmm_post_init_mdc(env, md2cmm_dev(md)); 
+                        if (rc) 
+                                CERROR("can not init md size %d\n", rc);
+                default:
+                        rc = upcall_dev->md_upcall.mu_upcall(env, 
+                                        md->md_upcall.mu_upcall_dev, ev);
+        }
         RETURN(rc);
 }
 

diff --git a/lustre/cmm/mdc_device.c b/lustre/cmm/mdc_device.c
index ae41ca3..ed8628e 100644 (file)
--- a/lustre/cmm/mdc_device.c
+++ b/lustre/cmm/mdc_device.c
@@ -192,6 +192,15 @@ static struct lu_device_operations mdc_lu_ops = {
         .ldo_process_config = mdc_process_config
 };
 
+void mdc_init_ea_size(const struct lu_env *env, struct mdc_device *mc, 
+                      int max_mdsize, int max_cookiesize)
+{
+        struct obd_device *obd = class_exp2obd(mc->mc_desc.cl_exp);
+       
+        obd->u.cli.cl_max_mds_easize = max_mdsize;
+        obd->u.cli.cl_max_mds_cookiesize = max_cookiesize;
+}
+
 static int mdc_device_init(const struct lu_env *env,
                            struct lu_device *ld, struct lu_device *next)
 {

diff --git a/lustre/cmm/mdc_internal.h b/lustre/cmm/mdc_internal.h
index 1be19ff..afe9ca4 100644 (file)
--- a/lustre/cmm/mdc_internal.h
+++ b/lustre/cmm/mdc_internal.h
@@ -96,6 +96,9 @@ static inline struct mdc_device *lu2mdc_dev(struct lu_device *ld)
 struct lu_object *mdc_object_alloc(const struct lu_env *,
                                    const struct lu_object_header *,
                                    struct lu_device *);
+
+void mdc_init_ea_size(const struct lu_env *env, struct mdc_device *mc, 
+                      int max_mdsize, int max_cookiesize);
 #ifdef HAVE_SPLIT_SUPPORT
 int mdc_send_page(struct cmm_device *cmm, const struct lu_env *env,
                   struct md_object *mo, struct page *page, __u32 end);

diff --git a/lustre/include/obd.h b/lustre/include/obd.h
index 3493fec..f52abc2 100644 (file)
--- a/lustre/include/obd.h
+++ b/lustre/include/obd.h
@@ -573,9 +573,6 @@ struct mds_obd {
                                          mds_fl_acl:1;
 
 
-        /* For CMD add mds_num */
-        int                              mds_num;
-
         struct upcall_cache             *mds_identity_cache;
         struct upcall_cache             *mds_rmtacl_cache;
 

diff --git a/lustre/mdd/mdd_lov.c b/lustre/mdd/mdd_lov.c
index ac48d69..6d0bcc8 100644 (file)
--- a/lustre/mdd/mdd_lov.c
+++ b/lustre/mdd/mdd_lov.c
@@ -127,6 +127,7 @@ int mdd_init_obd(const struct lu_env *env, struct mdd_device *mdd,
                 CERROR("can not find obd %s \n", MDD_OBD_NAME);
                 LBUG();
         }
+
         obd->u.mds.mds_id = index;
         obd->obd_recovering = 1;
         rc = class_setup(obd, lcfg);
@@ -139,7 +140,6 @@ int mdd_init_obd(const struct lu_env *env, struct mdd_device *mdd,
         obd->obd_upcall.onu_owner = mdd;
         obd->obd_upcall.onu_upcall = mdd_lov_update;
         mdd->mdd_obd_dev = obd;
-        obd->u.mds.mds_num = mdd2lu_dev(mdd)->ld_site->ls_node_id;
 class_detach:
         if (rc)
                 class_detach(obd, lcfg);

diff --git a/lustre/mds/mds_fs.c b/lustre/mds/mds_fs.c
index 198e0cb..1442802 100644 (file)
--- a/lustre/mds/mds_fs.c
+++ b/lustre/mds/mds_fs.c
@@ -716,7 +716,7 @@ int mds_obd_create(struct obd_export *exp, struct obdo *oa,
         err = fsfilt_commit(exp->exp_obd, mds->mds_objects_dir->d_inode,
                             handle, 0);
         if (!err) {
-                oa->o_gr = FILTER_GROUP_MDS0 + mds->mds_num;
+                oa->o_gr = FILTER_GROUP_MDS0 + mds->mds_id;
                 oa->o_valid |= OBD_MD_FLID | OBD_MD_FLGENER | OBD_MD_FLGROUP;
         } else if (!rc)
                 rc = err;

diff --git a/lustre/mdt/mdt_handler.c b/lustre/mdt/mdt_handler.c
index 2b1f9c4..292c4c3 100644 (file)
--- a/lustre/mdt/mdt_handler.c
+++ b/lustre/mdt/mdt_handler.c
@@ -756,12 +756,13 @@ out:
  * it here and do not define more moo api anymore for
  * this.
  */
-static int mdt_write_dir_page(struct mdt_thread_info *info, struct page *page)
+static int mdt_write_dir_page(struct mdt_thread_info *info, struct page *page,
+                              int size)
 {
         struct mdt_object *object = info->mti_object;
         struct lu_dirpage *dp;
         struct lu_dirent *ent;
-        int rc = 0;
+        int rc = 0, offset = 0, is_dir;
 
         ENTRY;
 
@@ -770,26 +771,27 @@ static int mdt_write_dir_page(struct mdt_thread_info *info, struct page *page)
         info->mti_no_need_trans = 1;
         kmap(page);
         dp = page_address(page);
-        for (ent = lu_dirent_start(dp); ent != NULL;
+        for (ent = lu_dirent_start(dp); ent != NULL && offset < size;
                           ent = lu_dirent_next(ent)) {
                 struct lu_fid *lf = &ent->lde_fid;
+                char *name;
+                offset = (int)((__u32)ent - (__u32)dp);
+                
+                if (!strncmp(ent->lde_name, ".", ent->lde_namelen) || 
+                    !strncmp(ent->lde_name, "..", ent->lde_namelen))
+                        continue;
 
-                /* FIXME: multi-trans for this name insert */
-                if (strncmp(ent->lde_name, ".", ent->lde_namelen) &&
-                    strncmp(ent->lde_name, "..", ent->lde_namelen)) {
-                        char *name;
-                        int is_dir = le32_to_cpu(ent->lde_hash) & 
-                                        MAX_HASH_HIGHEST_BIT;
-                        
-                        OBD_ALLOC(name, ent->lde_namelen + 1);
-                        memcpy(name, ent->lde_name, ent->lde_namelen);
-                        rc = mdo_name_insert(info->mti_env,
-                                             md_object_next(&object->mot_obj),
-                                             name, lf, is_dir);
-                        OBD_FREE(name, ent->lde_namelen + 1);
-                        if (rc)
-                                GOTO(out, rc);
-                }
+                is_dir = le32_to_cpu(ent->lde_hash) & MAX_HASH_HIGHEST_BIT;
+                
+                OBD_ALLOC(name, ent->lde_namelen + 1);
+                memcpy(name, ent->lde_name, ent->lde_namelen);
+                CDEBUG(D_INFO, "insert name %s offset %d \n", name, offset);
+                rc = mdo_name_insert(info->mti_env,
+                                     md_object_next(&object->mot_obj),
+                                     name, lf, is_dir);
+                OBD_FREE(name, ent->lde_namelen + 1);
+                if (rc)
+                        GOTO(out, rc);
         }
 out:
         kunmap(page);
@@ -874,7 +876,7 @@ static int mdt_writepage(struct mdt_thread_info *info)
         }
         if (rc)
                 GOTO(cleanup_lwi, rc);
-        rc = mdt_write_dir_page(info, page);
+        rc = mdt_write_dir_page(info, page, reqbody->nlink);
 
 cleanup_lwi:
         OBD_FREE_PTR(lwi);