dt: add ->del() method to interator, and implemente it in osd

diff --git a/lustre/include/dt_object.h b/lustre/include/dt_object.h
index 5bd55dd..ca34c76 100644 (file)
--- a/lustre/include/dt_object.h
+++ b/lustre/include/dt_object.h
@@ -301,7 +301,7 @@ struct dt_index_operations {
                  * precondition: dt_object_exists(dt);
                  */
                 struct dt_it *(*init)(const struct lu_context *ctxt,
-                                      struct dt_object *dt);
+                                      struct dt_object *dt, int writable);
                 void          (*fini)(const struct lu_context *ctxt,
                                       struct dt_it *di);
                 int            (*get)(const struct lu_context *ctxt,
@@ -309,6 +309,8 @@ struct dt_index_operations {
                                       const struct dt_key *key);
                 void           (*put)(const struct lu_context *ctxt,
                                       struct dt_it *di);
+                int            (*del)(const struct lu_context *ctxt,
+                                      struct dt_it *di, struct thandle *th);
                 int           (*next)(const struct lu_context *ctxt,
                                       struct dt_it *di);
                 struct dt_key *(*key)(const struct lu_context *ctxt,
@@ -414,7 +416,7 @@ struct dt_txn_callback {
                              struct txn_param *param, void *cookie);
         int (*dtc_txn_stop)(const struct lu_context *ctx,
                             struct thandle *txn, void *cookie);
-        int (*dtc_txn_commit)(const struct lu_context *ctx, 
+        int (*dtc_txn_commit)(const struct lu_context *ctx,
                               struct thandle *txn, void *cookie);
         void            *dtc_cookie;
         struct list_head dtc_linkage;

diff --git a/lustre/mdd/mdd_handler.c b/lustre/mdd/mdd_handler.c
index a307136..3280f6c 100644 (file)
--- a/lustre/mdd/mdd_handler.c
+++ b/lustre/mdd/mdd_handler.c
@@ -277,7 +277,7 @@ static int __mdd_lmm_get(const struct lu_context *ctxt,
         int rc;
 
         LASSERT(ma->ma_lmm != NULL && ma->ma_lmm_size > 0);
-        rc = mdd_get_md(ctxt, mdd_obj, ma->ma_lmm, &ma->ma_lmm_size, 0, 
+        rc = mdd_get_md(ctxt, mdd_obj, ma->ma_lmm, &ma->ma_lmm_size, 0,
                         MDS_LOV_MD_NAME);
         if (rc > 0) {
                 ma->ma_valid |= MA_LOV;
@@ -314,16 +314,16 @@ static int mdd_attr_get_internal(const struct lu_context *ctxt,
                 rc = __mdd_iattr_get(ctxt, mdd_obj, ma);
 
         if (rc == 0 && ma->ma_need & MA_LOV) {
-                if (S_ISREG(mdd_object_type(mdd_obj)) || 
+                if (S_ISREG(mdd_object_type(mdd_obj)) ||
                     S_ISDIR(mdd_object_type(mdd_obj)))
                         rc = __mdd_lmm_get(ctxt, mdd_obj, ma);
         }
-#ifdef HAVE_SPLIT_SUPPORT 
+#ifdef HAVE_SPLIT_SUPPORT
         if (rc == 0 && ma->ma_need & MA_LMV) {
                 if (S_ISDIR(mdd_object_type(mdd_obj)))
                         rc = __mdd_lmv_get(ctxt, mdd_obj, ma);
         }
-#endif        
+#endif
         CDEBUG(D_INODE, "after getattr rc = %d, ma_valid = "LPX64"\n",
                         rc, ma->ma_valid);
         RETURN(rc);
@@ -841,7 +841,7 @@ static int mdd_attr_set(const struct lu_context *ctxt,
                 if (lmm == NULL)
                         GOTO(cleanup, rc = -ENOMEM);
 
-                rc = mdd_get_md(ctxt, mdd_obj, lmm, &lmm_size, 1, 
+                rc = mdd_get_md(ctxt, mdd_obj, lmm, &lmm_size, 1,
                                 MDS_LOV_MD_NAME);
 
                 if (rc < 0)
@@ -1115,7 +1115,7 @@ static int mdd_dir_is_empty(const struct lu_context *ctx,
 
         obj = mdd_object_child(dir);
         iops = &obj->do_index_ops->dio_it;
-        it = iops->init(ctx, obj);
+        it = iops->init(ctx, obj, 0);
         if (it != NULL) {
                 result = iops->get(ctx, it, (const void *)"");
                 if (result > 0) {
@@ -2074,7 +2074,7 @@ static int mdd_close(const struct lu_context *ctxt, struct md_object *obj,
                         handle = mdd_trans_start(ctxt, mdo2mdd(obj));
                         if (IS_ERR(handle))
                                 GOTO(out, rc = -ENOMEM);
-                        
+
                         rc = __mdd_orphan_del(ctxt, mdd_obj, handle);
 
                         mdd_trans_stop(ctxt, mdo2mdd(obj), rc, handle);

diff --git a/lustre/osd/osd_handler.c b/lustre/osd/osd_handler.c
index ff4ddd3..86e36f6 100644 (file)
--- a/lustre/osd/osd_handler.c
+++ b/lustre/osd/osd_handler.c
@@ -105,8 +105,8 @@ struct osd_device {
 
         /* Thread context for transaction commit callback.
          * Currently, OSD is based on ext3/JBD. Transaction commit in ext3/JBD
-         * is serialized, that is there is no more than one transaction commit 
-         * at a time (JBD journal_commit_transaction() is serialized). 
+         * is serialized, that is there is no more than one transaction commit
+         * at a time (JBD journal_commit_transaction() is serialized).
          * This means that it's enough to have _one_ lu_context.
          */
         struct lu_context         od_ctx_for_commit;
@@ -173,6 +173,8 @@ static int   osd_it_get        (const struct lu_context *ctx,
                                 struct dt_it *di, const struct dt_key *key);
 static void  osd_it_put        (const struct lu_context *ctx, struct dt_it *di);
 static int   osd_it_next       (const struct lu_context *ctx, struct dt_it *di);
+static int   osd_it_del        (const struct lu_context *ctx, struct dt_it *di,
+                                struct thandle *th);
 static int   osd_it_key_size   (const struct lu_context *ctx,
                                 const struct dt_it *di);
 static void  osd_conf_get      (const struct lu_context *ctx,
@@ -202,7 +204,7 @@ static struct inode       *osd_iget         (struct osd_thread_info *info,
                                              const struct osd_inode_id *id);
 static struct super_block *osd_sb           (const struct osd_device *dev);
 static struct dt_it       *osd_it_init      (const struct lu_context *ctx,
-                                             struct dt_object *dt);
+                                             struct dt_object *dt, int wable);
 static struct dt_key      *osd_it_key       (const struct lu_context *ctx,
                                              const struct dt_it *di);
 static struct dt_rec      *osd_it_rec       (const struct lu_context *ctx,
@@ -465,8 +467,8 @@ static void osd_trans_commit_cb(struct journal_callback *jcb, int error)
                 CERROR("transaction @0x%p commit error: %d\n", th, error);
         } else {
                 /* This dd_ctx_for_commit is only for commit usage.
-                 * see "struct dt_device" 
-                 */ 
+                 * see "struct dt_device"
+                 */
                 dt_txn_hook_commit(&osd_dt_dev(dev)->od_ctx_for_commit, th);
         }
 
@@ -589,12 +591,12 @@ static void osd_ro(const struct lu_context *ctx,
         th = osd_trans_start(ctx, d, &param);
         if (!IS_ERR(th))
                 osd_trans_stop(ctx, th);
-        
+
         if (sync)
                 osd_sync(ctx, d);
-        
+
         lvfs_set_rdonly(lvfs_sbdev(osd_sb(osd_dt_dev(d))));
-        EXIT;        
+        EXIT;
 }
 
 
@@ -1111,7 +1113,7 @@ static int osd_dir_page_build(const struct lu_context *ctx, int first,
         if (first) {
                 area += sizeof (struct lu_dirpage);
                 nob  -= sizeof (struct lu_dirpage);
-                
+
         }
 
         LASSERT(nob > sizeof *ent);
@@ -1197,7 +1199,7 @@ static int osd_readpage(const struct lu_context *ctxt,
          * iterating through directory and fill pages from @rdpg
          */
         iops = &dt->do_index_ops->dio_it;
-        it = iops->init(ctxt, dt);
+        it = iops->init(ctxt, dt, 0);
         if (it == NULL)
                 return -ENOMEM;
         /*
@@ -1217,7 +1219,7 @@ static int osd_readpage(const struct lu_context *ctxt,
                         rc = osd_dir_page_build(ctxt, !i, kmap(pg),
                                                 min_t(int, nob, CFS_PAGE_SIZE),
                                                 iops, it,
-                                                &hash_start, &hash_end, 
+                                                &hash_start, &hash_end,
                                                 rdpg->rp_hash_end, &last);
                         if (rc != 0 || i == rdpg->rp_npages - 1)
                                 last->lde_reclen = 0;
@@ -1465,21 +1467,22 @@ struct osd_it {
 };
 
 static struct dt_it *osd_it_init(const struct lu_context *ctx,
-                                 struct dt_object *dt)
+                                 struct dt_object *dt, int writable)
 {
         struct osd_it     *it;
         struct osd_object *obj = osd_dt_obj(dt);
         struct lu_object  *lo  = &dt->do_lu;
+        __u32              flags;
 
         LASSERT(lu_object_exists(lo));
         LASSERT(obj->oo_ipd != NULL);
 
+        flags = writable ? IAM_IT_MOVE|IAM_IT_WRITE : IAM_IT_MOVE;
         OBD_ALLOC_PTR(it);
         if (it != NULL) {
                 it->oi_obj = obj;
                 lu_object_get(lo);
-                iam_it_init(&it->oi_it,
-                            &obj->oo_container, IAM_IT_MOVE, obj->oo_ipd);
+                iam_it_init(&it->oi_it, &obj->oo_container, flags, obj->oo_ipd);
         }
         return (struct dt_it *)it;
 }
@@ -1513,6 +1516,18 @@ static int osd_it_next(const struct lu_context *ctx, struct dt_it *di)
         return iam_it_next(&it->oi_it);
 }
 
+static int osd_it_del(const struct lu_context *ctx, struct dt_it *di,
+                      struct thandle *th)
+{
+        struct osd_it      *it = (struct osd_it *)di;
+        struct osd_thandle *oh;
+
+        oh = container_of0(th, struct osd_thandle, ot_super);
+        LASSERT(oh->ot_handle != NULL);
+
+        return iam_it_rec_delete(oh->ot_handle, &it->oi_it);
+}
+
 static struct dt_key *osd_it_key(const struct lu_context *ctx,
                                  const struct dt_it *di)
 {
@@ -1555,6 +1570,7 @@ static struct dt_index_operations osd_index_ops = {
                 .fini     = osd_it_fini,
                 .get      = osd_it_get,
                 .put      = osd_it_put,
+                .del      = osd_it_del,
                 .next     = osd_it_next,
                 .key      = osd_it_key,
                 .key_size = osd_it_key_size,
@@ -1640,12 +1656,13 @@ static int osd_index_compat_lookup(const struct lu_context *ctxt,
                                                        (struct lu_fid *)rec);
                         else
                                 result = -ENOENT;
+                        d = dentry;
                 } else {
                         /* What? Disconnected alias? Ppheeeww... */
                         CERROR("Aliasing where not expected\n");
                         result = -EIO;
-                        dput(d);
                 }
+                dput(d);
                 dput(dentry);
         } else
                 result = -ENOMEM;
@@ -2030,7 +2047,7 @@ static struct inode *osd_iget(struct osd_thread_info *info,
                 iput(inode);
                 inode = ERR_PTR(-ESTALE);
         }
-        
+
         return inode;
 
 }