Branch: HEAD

[fs/lustre-release.git] / lustre / obdfilter / filter.c

diff --git a/lustre/obdfilter/filter.c b/lustre/obdfilter/filter.c
index 7cec53e..fdccecc 100644 (file)
--- a/lustre/obdfilter/filter.c
+++ b/lustre/obdfilter/filter.c
@@ -202,12 +202,9 @@ static int filter_export_stats_init(struct obd_device *obd,
                                     struct obd_export *exp,
                                     void *client_nid)
 {
-        struct filter_export_data *fed = &exp->exp_filter_data;
         int rc, newnid = 0;
         ENTRY;
 
-        init_brw_stats(&fed->fed_brw_stats);
-
         if (obd_uuid_equals(&exp->exp_client_uuid, &obd->obd_uuid))
                 /* Self-export gets no proc entry */
                 RETURN(0);
@@ -545,7 +542,7 @@ struct filter_mod_data *filter_fmd_get(struct obd_export *exp,
         struct filter_export_data *fed = &exp->exp_filter_data;
         struct filter_mod_data *found = NULL, *fmd_new = NULL;
 
-        OBD_SLAB_ALLOC(fmd_new, ll_fmd_cachep, CFS_ALLOC_IO, sizeof(*fmd_new));
+        OBD_SLAB_ALLOC_PTR_GFP(fmd_new, ll_fmd_cachep, CFS_ALLOC_IO);
 
         spin_lock(&fed->fed_lock);
         found = filter_fmd_find_nolock(&exp->exp_obd->u.filter,fed,objid,group);
@@ -2673,7 +2670,13 @@ static int filter_connect_internal(struct obd_export *exp,
         exp->exp_connect_flags = data->ocd_connect_flags;
         data->ocd_version = LUSTRE_VERSION_CODE;
 
+        /* Kindly make sure the SKIP_ORPHAN flag is from MDS. */
+        if (!ergo(data->ocd_connect_flags & OBD_CONNECT_SKIP_ORPHAN,
+                  data->ocd_connect_flags & OBD_CONNECT_MDS))
+                RETURN(-EPROTO);
+
         if (exp->exp_connect_flags & OBD_CONNECT_GRANT) {
+                struct filter_obd *filter = &exp->exp_obd->u.filter;
                 obd_size left, want;
 
                 spin_lock(&exp->exp_obd->obd_osfs_lock);
@@ -2687,6 +2690,8 @@ static int filter_connect_internal(struct obd_export *exp,
                        LPU64" left: "LPU64"\n", exp->exp_obd->obd_name,
                        exp->exp_client_uuid.uuid, exp,
                        data->ocd_grant, want, left);
+                
+                filter->fo_tot_granted_clients ++;
         }
 
         if (data->ocd_connect_flags & OBD_CONNECT_INDEX) {
@@ -2984,6 +2989,12 @@ static int filter_destroy_export(struct obd_export *exp)
         filter_grant_discard(exp);
         filter_fmd_cleanup(exp);
 
+        if (exp->exp_connect_flags & OBD_CONNECT_GRANT_SHRINK) {
+                struct filter_obd *filter = &exp->exp_obd->u.filter;
+                if (filter->fo_tot_granted_clients > 0)
+                        filter->fo_tot_granted_clients --;
+        }
+
         if (!(exp->exp_flags & OBD_OPT_FORCE))
                 filter_grant_sanity_check(exp->exp_obd, __FUNCTION__);
 
@@ -3494,11 +3505,12 @@ static int filter_unpackmd(struct obd_export *exp, struct lov_stripe_md **lsmp,
 
 /* caller must hold fo_create_locks[oa->o_gr] */
 static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa,
-                                      struct filter_obd *filter)
+                                     struct filter_obd *filter)
 {
         struct obdo doa; /* XXX obdo on stack */
         obd_id last, id;
-        int rc;
+        int rc = 0;
+        int skip_orphan;
         ENTRY;
 
         LASSERT(oa);
@@ -3520,8 +3532,11 @@ static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa,
 
         last = filter_last_id(filter, doa.o_gr);
 
-        CWARN("%s: deleting orphan objects from "LPU64" to "LPU64"\n",
-               exp->exp_obd->obd_name, oa->o_id + 1, last);
+        skip_orphan = !!(exp->exp_connect_flags & OBD_CONNECT_SKIP_ORPHAN);
+
+        CWARN("%s: deleting orphan objects from "LPU64" to "LPU64"%s\n",
+               exp->exp_obd->obd_name, oa->o_id + 1, last,
+               skip_orphan ? ", orphan objids won't be reused any more." : ".");
 
         for (id = last; id > oa->o_id; id--) {
                 doa.o_id = id;
@@ -3529,17 +3544,26 @@ static int filter_destroy_precreated(struct obd_export *exp, struct obdo *oa,
                 if (rc && rc != -ENOENT) /* this is pretty fatal... */
                         CEMERG("error destroying precreate objid "LPU64": %d\n",
                                id, rc);
-                filter_set_last_id(filter, id - 1, doa.o_gr);
+
                 /* update last_id on disk periodically so that if we restart
                  * we don't need to re-scan all of the just-deleted objects. */
-                if ((id & 511) == 0)
+                if ((id & 511) == 0 && !skip_orphan) {
+                        filter_set_last_id(filter, id - 1, doa.o_gr);
                         filter_update_last_objid(exp->exp_obd, doa.o_gr, 0);
+                }
         }
 
         CDEBUG(D_HA, "%s: after destroy: set last_objids["LPU64"] = "LPU64"\n",
                exp->exp_obd->obd_name, doa.o_gr, oa->o_id);
 
-        rc = filter_update_last_objid(exp->exp_obd, doa.o_gr, 1);
+        if (!skip_orphan) {
+                filter_set_last_id(filter, id, doa.o_gr);
+                rc = filter_update_last_objid(exp->exp_obd, doa.o_gr, 1);
+        } else {
+                /* don't reuse orphan object, return last used objid */
+                oa->o_id = last;
+                rc = 0;
+        }
         clear_bit(doa.o_gr, &filter->fo_destroys_in_progress);
 
         RETURN(rc);
@@ -3558,6 +3582,8 @@ static int filter_handle_precreate(struct obd_export *exp, struct obdo *oa,
 
         /* delete orphans request */
         if ((oa->o_valid & OBD_MD_FLFLAGS) && (oa->o_flags & OBD_FL_DELORPHAN)){
+                obd_id last = filter_last_id(filter, group);
+
                 if (oti->oti_conn_cnt < exp->exp_conn_cnt) {
                         CERROR("%s: dropping old orphan cleanup request\n",
                                obd->obd_name);
@@ -3572,14 +3598,14 @@ static int filter_handle_precreate(struct obd_export *exp, struct obdo *oa,
                         up(&filter->fo_create_locks[group]);
                         RETURN(0);
                 }
-                diff = oa->o_id - filter_last_id(filter, group);
+                diff = oa->o_id - last;
                 CDEBUG(D_HA, "filter_last_id() = "LPU64" -> diff = %d\n",
-                       filter_last_id(filter, group), diff);
+                       last, diff);
 
                 if (-diff > OST_MAX_PRECREATE) {
                         CERROR("%s: ignoring bogus orphan destroy request: "
                                "obdid "LPU64" last_id "LPU64"\n", obd->obd_name,
-                               oa->o_id, filter_last_id(filter, group));
+                               oa->o_id, last);
                         /* FIXME: should reset precreate_next_id on MDS */
                         GOTO(out, rc = -EINVAL);
                 }
@@ -4264,14 +4290,10 @@ static int filter_get_info(struct obd_export *exp, __u32 keylen,
                 push_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
                 rc = fsfilt_iocontrol(obd, dentry->d_inode, NULL,
                                       EXT3_IOC_FIEMAP, (long)fiemap);
-                if (rc) {
-                        f_dput(dentry);
-                        RETURN(rc);
-                }
                 pop_ctxt(&saved, &obd->obd_lvfs_ctxt, NULL);
 
                 f_dput(dentry);
-                RETURN(0);
+                RETURN(rc);
         }
 
         CDEBUG(D_IOCTL, "invalid key\n");
@@ -4331,6 +4353,15 @@ static int filter_set_info_async(struct obd_export *exp, __u32 keylen,
                 RETURN(0);
         }
 
+        if (KEY_IS(KEY_GRANT_SHRINK)) {
+                struct ost_body *body = (struct ost_body *)val;
+                /* handle shrink grant */
+                spin_lock(&exp->exp_obd->obd_osfs_lock);
+                filter_grant_incoming(exp, &body->oa);
+                spin_unlock(&exp->exp_obd->obd_osfs_lock);
+                RETURN(rc);
+        }
+
         if (!KEY_IS(KEY_MDS_CONN))
                 RETURN(-EINVAL);