LU-2109 llog: introduce llog handle refcounter

llog handle is allocated by llog_open and freed by llog_close,
meawhile there is 'backdoor' to get already opened handle - it is
llog_cat_id2handle() function, it can find handle in the list. That
leads to possible use-after-free.

llog_refcount was introduced to solve this.
- llog_alloc_handle() set refcount to 1
- llog_handle_put() drops refcount and free handle if refcount == 0
- llog_cat_i2handle() takes extra reference
- llog_handle_put() must be called if handle was taken by
  llog_cat_id2handle()
- llog_free_handle() calls llog_handle_put() at the end.

Additional changes in patch:
- llog_cat_init_and_process() processes llogs without fork, making
  sure that server will not start until all old llogs are cancelled.
- llog_cat_cancel_records() doesn't break cycle if handle is not
  found but continue to check other handles
- wrong GOTO() parameter in llog_close. Should be 'rc = -EOPNOTSUPP'
  but not just '-EOPNOTSUPP'.
- indentation fixes.

Signed-off-by: Mikhail Pershin <tappro@whamcloud.com>
Change-Id: I0dd0a934eb79c63c6726902540132ad877183d10
Reviewed-on: http://review.whamcloud.com/4745
Tested-by: Hudson
Reviewed-by: Li Wei <wei.g.li@intel.com>
Tested-by: Maloo <whamcloud.maloo@gmail.com>
Reviewed-by: Alex Zhuravlev <bzzz@whamcloud.com>
Reviewed-by: Prakash Surya <surya1@llnl.gov>

diff --git a/lustre/include/lustre_log.h b/lustre/include/lustre_log.h
index ecb4c8f..45c248b 100644 (file)
--- a/lustre/include/lustre_log.h
+++ b/lustre/include/lustre_log.h
@@ -350,6 +350,7 @@ struct llog_handle {
        char                    *lgh_name;
        void                    *private_data;
        struct llog_operations  *lgh_logops;
        char                    *lgh_name;
        void                    *private_data;
        struct llog_operations  *lgh_logops;

+       cfs_atomic_t             lgh_refcount;

 };
 
 /* llog_lvfs.c */
 };
 
 /* llog_lvfs.c */

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index dbc1961..21bd4d6 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -70,6 +70,7 @@ struct llog_handle *llog_alloc_handle(void)
        init_rwsem(&loghandle->lgh_lock);
        spin_lock_init(&loghandle->lgh_hdr_lock);
        CFS_INIT_LIST_HEAD(&loghandle->u.phd.phd_entry);
        init_rwsem(&loghandle->lgh_lock);
        spin_lock_init(&loghandle->lgh_hdr_lock);
        CFS_INIT_LIST_HEAD(&loghandle->u.phd.phd_entry);

+       cfs_atomic_set(&loghandle->lgh_refcount, 1);

 
        return loghandle;
 }
 
        return loghandle;
 }


@@ -79,22 +80,34 @@ struct llog_handle *llog_alloc_handle(void)
  */
 void llog_free_handle(struct llog_handle *loghandle)
 {
  */
 void llog_free_handle(struct llog_handle *loghandle)
 {

-       if (!loghandle)
-               return;
+       LASSERT(loghandle != NULL);

 
 

+       /* failed llog_init_handle */

        if (!loghandle->lgh_hdr)
                goto out;
        if (!loghandle->lgh_hdr)
                goto out;

+

        if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)
        if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN)

-               cfs_list_del_init(&loghandle->u.phd.phd_entry);
-       if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT)
+               LASSERT(cfs_list_empty(&loghandle->u.phd.phd_entry));
+       else if (loghandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT)

                LASSERT(cfs_list_empty(&loghandle->u.chd.chd_head));
        LASSERT(sizeof(*(loghandle->lgh_hdr)) == LLOG_CHUNK_SIZE);
        OBD_FREE(loghandle->lgh_hdr, LLOG_CHUNK_SIZE);
                LASSERT(cfs_list_empty(&loghandle->u.chd.chd_head));
        LASSERT(sizeof(*(loghandle->lgh_hdr)) == LLOG_CHUNK_SIZE);
        OBD_FREE(loghandle->lgh_hdr, LLOG_CHUNK_SIZE);

-

 out:
        OBD_FREE_PTR(loghandle);
 }
 
 out:
        OBD_FREE_PTR(loghandle);
 }
 

+void llog_handle_get(struct llog_handle *loghandle)
+{
+       cfs_atomic_inc(&loghandle->lgh_refcount);
+}
+
+void llog_handle_put(struct llog_handle *loghandle)
+{
+       LASSERT(cfs_atomic_read(&loghandle->lgh_refcount) > 0);
+       if (cfs_atomic_dec_and_test(&loghandle->lgh_refcount))
+               llog_free_handle(loghandle);
+}
+

 /* returns negative on error; 0 if success; 1 if success & log destroyed */
 int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
                    int index)
 /* returns negative on error; 0 if success; 1 if success & log destroyed */
 int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
                    int index)


@@ -930,10 +943,10 @@ int llog_close(const struct lu_env *env, struct llog_handle *loghandle)
        if (rc)
                GOTO(out, rc);
        if (lop->lop_close == NULL)
        if (rc)
                GOTO(out, rc);
        if (lop->lop_close == NULL)

-               GOTO(out, -EOPNOTSUPP);
+               GOTO(out, rc = -EOPNOTSUPP);

        rc = lop->lop_close(env, loghandle);
 out:
        rc = lop->lop_close(env, loghandle);
 out:

-       llog_free_handle(loghandle);
+       llog_handle_put(loghandle);

        RETURN(rc);
 }
 EXPORT_SYMBOL(llog_close);
        RETURN(rc);
 }
 EXPORT_SYMBOL(llog_close);

diff --git a/lustre/obdclass/llog_cat.c b/lustre/obdclass/llog_cat.c
index 1bd130e..3fe67f6 100644 (file)
--- a/lustre/obdclass/llog_cat.c
+++ b/lustre/obdclass/llog_cat.c
@@ -147,6 +147,9 @@ out_destroy:
  *
  * Assumes caller has already pushed us into the kernel context and is locking.
  * We return a lock on the handle to ensure nobody yanks it from us.
  *
  * Assumes caller has already pushed us into the kernel context and is locking.
  * We return a lock on the handle to ensure nobody yanks it from us.

+ *
+ * This takes extra reference on llog_handle via llog_handle_get() and require
+ * this reference to be put by caller using llog_handle_put()

  */
 int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle,
                       struct llog_handle **res, struct llog_logid *logid)
  */
 int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle,
                       struct llog_handle **res, struct llog_logid *logid)


@@ -185,13 +188,13 @@ int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle,
                CERROR("%s: error opening log id "LPX64":%x: rc = %d\n",
                       cathandle->lgh_ctxt->loc_obd->obd_name,
                       logid->lgl_oid, logid->lgl_ogen, rc);
                CERROR("%s: error opening log id "LPX64":%x: rc = %d\n",
                       cathandle->lgh_ctxt->loc_obd->obd_name,
                       logid->lgl_oid, logid->lgl_ogen, rc);

-               GOTO(out, rc);
+               RETURN(rc);

        }
 
        rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
        if (rc < 0) {
                llog_close(env, loghandle);
        }
 
        rc = llog_init_handle(env, loghandle, LLOG_F_IS_PLAIN, NULL);
        if (rc < 0) {
                llog_close(env, loghandle);

-               GOTO(out, rc);
+               RETURN(rc);

        }
 
        down_write(&cathandle->lgh_lock);
        }
 
        down_write(&cathandle->lgh_lock);


@@ -204,8 +207,9 @@ int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle,
                                loghandle->lgh_hdr->llh_cat_idx;
        EXIT;
 out:
                                loghandle->lgh_hdr->llh_cat_idx;
        EXIT;
 out:

+       llog_handle_get(loghandle);

        *res = loghandle;
        *res = loghandle;

-       return rc;
+       return 0;

 }
 
 int llog_cat_close(const struct lu_env *env, struct llog_handle *cathandle)
 }
 
 int llog_cat_close(const struct lu_env *env, struct llog_handle *cathandle)


@@ -234,15 +238,7 @@ int llog_cat_close(const struct lu_env *env, struct llog_handle *cathandle)
                                       rc);
 
                        index = loghandle->u.phd.phd_cookie.lgc_index;
                                       rc);
 
                        index = loghandle->u.phd.phd_cookie.lgc_index;

-
-                       LASSERT(index);
-                       llog_cat_set_first_idx(cathandle, index);
-                       rc = llog_cancel_rec(env, cathandle, index);
-                       if (rc == 0)
-                               CDEBUG(D_RPCTRACE,
-                                      "cancel plain log at index %u of "
-                                      "catalog "LPX64"\n",
-                                      index, cathandle->lgh_id.lgl_oid);
+                       llog_cat_cleanup(env, cathandle, NULL, index);

                }
                llog_close(env, loghandle);
        }
                }
                llog_close(env, loghandle);
        }


@@ -513,25 +509,15 @@ int llog_cat_cancel_records(const struct lu_env *env,
                        CERROR("%s: cannot find handle for llog "LPX64": %d\n",
                               cathandle->lgh_ctxt->loc_obd->obd_name,
                               lgl->lgl_oid, rc);
                        CERROR("%s: cannot find handle for llog "LPX64": %d\n",
                               cathandle->lgh_ctxt->loc_obd->obd_name,
                               lgl->lgl_oid, rc);

-                       break;
+                       failed++;
+                       continue;

                }
 
                lrc = llog_cancel_rec(env, loghandle, cookies->lgc_index);
                if (lrc == 1) {          /* log has been destroyed */
                        index = loghandle->u.phd.phd_cookie.lgc_index;
                }
 
                lrc = llog_cancel_rec(env, loghandle, cookies->lgc_index);
                if (lrc == 1) {          /* log has been destroyed */
                        index = loghandle->u.phd.phd_cookie.lgc_index;

-                       down_write(&cathandle->lgh_lock);
-                       if (cathandle->u.chd.chd_current_log == loghandle)
-                               cathandle->u.chd.chd_current_log = NULL;
-                       up_write(&cathandle->lgh_lock);
-                       llog_close(env, loghandle);
-
-                       LASSERT(index);
-                       llog_cat_set_first_idx(cathandle, index);
-                       lrc = llog_cancel_rec(env, cathandle, index);
-                       if (lrc == 0)
-                               CDEBUG(D_RPCTRACE, "cancel plain log at index"
-                                      " %u of catalog "LPX64"\n",
-                                      index, cathandle->lgh_id.lgl_oid);
+                       rc = llog_cat_cleanup(env, cathandle, loghandle,
+                                             index);

                } else if (lrc == -ENOENT) {
                        if (rc == 0) /* ENOENT shouldn't rewrite any error */
                                rc = lrc;
                } else if (lrc == -ENOENT) {
                        if (rc == 0) /* ENOENT shouldn't rewrite any error */
                                rc = lrc;


@@ -539,6 +525,7 @@ int llog_cat_cancel_records(const struct lu_env *env,
                        failed++;
                        rc = lrc;
                }
                        failed++;
                        rc = lrc;
                }

+               llog_handle_put(loghandle);

        }
        if (rc)
                CERROR("%s: fail to cancel %d of %d llog-records: rc = %d\n",
        }
        if (rc)
                CERROR("%s: fail to cancel %d of %d llog-records: rc = %d\n",


@@ -585,14 +572,15 @@ int llog_cat_process_cb(const struct lu_env *env, struct llog_handle *cat_llh,
                 cd.lpcd_last_idx = 0;
                rc = llog_process_or_fork(env, llh, d->lpd_cb, d->lpd_data,
                                          &cd, false);
                 cd.lpcd_last_idx = 0;
                rc = llog_process_or_fork(env, llh, d->lpd_cb, d->lpd_data,
                                          &cd, false);

-                /* Continue processing the next log from idx 0 */
-                d->lpd_startidx = 0;
-        } else {
+               /* Continue processing the next log from idx 0 */
+               d->lpd_startidx = 0;
+       } else {

                rc = llog_process_or_fork(env, llh, d->lpd_cb, d->lpd_data,
                                          NULL, false);
                rc = llog_process_or_fork(env, llh, d->lpd_cb, d->lpd_data,
                                          NULL, false);

-        }
+       }
+       llog_handle_put(llh);

 
 

-        RETURN(rc);
+       RETURN(rc);

 }
 
 int llog_cat_process_or_fork(const struct lu_env *env,
 }
 
 int llog_cat_process_or_fork(const struct lu_env *env,


@@ -737,7 +725,8 @@ static int llog_cat_reverse_process_cb(const struct lu_env *env,
        }
 
        rc = llog_reverse_process(env, llh, d->lpd_cb, d->lpd_data, NULL);
        }
 
        rc = llog_reverse_process(env, llh, d->lpd_cb, d->lpd_data, NULL);

-        RETURN(rc);
+       llog_handle_put(llh);
+       RETURN(rc);

 }
 
 int llog_cat_reverse_process(const struct lu_env *env,
 }
 
 int llog_cat_reverse_process(const struct lu_env *env,


@@ -813,13 +802,42 @@ out:
         RETURN(0);
 }
 
         RETURN(0);
 }
 

+/* Cleanup deleted plain llog traces from catalog */
+int llog_cat_cleanup(const struct lu_env *env, struct llog_handle *cathandle,
+                    struct llog_handle *loghandle, int index)
+{
+       int rc;
+
+       LASSERT(index);
+       if (loghandle != NULL) {
+               /* remove destroyed llog from catalog list and
+                * chd_current_log variable */
+               down_write(&cathandle->lgh_lock);
+               if (cathandle->u.chd.chd_current_log == loghandle)
+                       cathandle->u.chd.chd_current_log = NULL;
+               cfs_list_del_init(&loghandle->u.phd.phd_entry);
+               up_write(&cathandle->lgh_lock);
+               LASSERT(index == loghandle->u.phd.phd_cookie.lgc_index);
+               /* llog was opened and keep in a list, close it now */
+               llog_close(env, loghandle);
+       }
+       /* remove plain llog entry from catalog by index */
+       llog_cat_set_first_idx(cathandle, index);
+       rc = llog_cancel_rec(env, cathandle, index);
+       if (rc == 0)
+               CDEBUG(D_HA, "cancel plain log at index"
+                      " %u of catalog "LPX64"\n",
+                      index, cathandle->lgh_id.lgl_oid);
+       return rc;
+}
+

 int cat_cancel_cb(const struct lu_env *env, struct llog_handle *cathandle,
                  struct llog_rec_hdr *rec, void *data)
 {
        struct llog_logid_rec   *lir = (struct llog_logid_rec *)rec;
        struct llog_handle      *loghandle;
        struct llog_log_hdr     *llh;
 int cat_cancel_cb(const struct lu_env *env, struct llog_handle *cathandle,
                  struct llog_rec_hdr *rec, void *data)
 {
        struct llog_logid_rec   *lir = (struct llog_logid_rec *)rec;
        struct llog_handle      *loghandle;
        struct llog_log_hdr     *llh;

-       int                      rc, index;
+       int                      rc;

 
        ENTRY;
 
 
        ENTRY;
 


@@ -838,8 +856,8 @@ int cat_cancel_cb(const struct lu_env *env, struct llog_handle *cathandle,
                       cathandle->lgh_ctxt->loc_obd->obd_name,
                       lir->lid_id.lgl_oid, rc);
                if (rc == -ENOENT || rc == -ESTALE) {
                       cathandle->lgh_ctxt->loc_obd->obd_name,
                       lir->lid_id.lgl_oid, rc);
                if (rc == -ENOENT || rc == -ESTALE) {

-                       index = rec->lrh_index;
-                       goto cat_cleanup;
+                       /* remove index from catalog */
+                       llog_cat_cleanup(env, cathandle, NULL, rec->lrh_index);

                }
                RETURN(rc);
        }
                }
                RETURN(rc);
        }


@@ -852,20 +870,10 @@ int cat_cancel_cb(const struct lu_env *env, struct llog_handle *cathandle,
                        CERROR("%s: fail to destroy empty log: rc = %d\n",
                               loghandle->lgh_ctxt->loc_obd->obd_name, rc);
 
                        CERROR("%s: fail to destroy empty log: rc = %d\n",
                               loghandle->lgh_ctxt->loc_obd->obd_name, rc);
 

-               index = loghandle->u.phd.phd_cookie.lgc_index;
-               llog_close(env, loghandle);
-
-cat_cleanup:
-               LASSERT(index);
-               llog_cat_set_first_idx(cathandle, index);
-               rc = llog_cancel_rec(env, cathandle, index);
-               if (rc == 0)
-                       CDEBUG(D_HA,
-                              "cancel log "LPX64":%x at index %u of catalog "
-                              LPX64"\n", lir->lid_id.lgl_oid,
-                              lir->lid_id.lgl_ogen, rec->lrh_index,
-                              cathandle->lgh_id.lgl_oid);
+               llog_cat_cleanup(env, cathandle, loghandle,
+                                loghandle->u.phd.phd_cookie.lgc_index);

        }
        }

+       llog_handle_put(loghandle);

 
        RETURN(rc);
 }
 
        RETURN(rc);
 }


@@ -881,7 +889,7 @@ int llog_cat_init_and_process(const struct lu_env *env,
        if (rc)
                RETURN(rc);
 
        if (rc)
                RETURN(rc);
 

-       rc = llog_process(env, llh, cat_cancel_cb, NULL, NULL);
+       rc = llog_process_or_fork(env, llh, cat_cancel_cb, NULL, NULL, false);

        if (rc)
                CERROR("%s: llog_process() with cat_cancel_cb failed: rc = "
                       "%d\n", llh->lgh_ctxt->loc_obd->obd_name, rc);
        if (rc)
                CERROR("%s: llog_process() with cat_cancel_cb failed: rc = "
                       "%d\n", llh->lgh_ctxt->loc_obd->obd_name, rc);

diff --git a/lustre/obdclass/llog_internal.h b/lustre/obdclass/llog_internal.h
index 7ff15c6..9d51c6a 100644 (file)
--- a/lustre/obdclass/llog_internal.h
+++ b/lustre/obdclass/llog_internal.h
@@ -81,6 +81,8 @@ static inline struct llog_thread_info *llog_info(const struct lu_env *env)
 int llog_info_init(void);
 void llog_info_fini(void);
 
 int llog_info_init(void);
 void llog_info_fini(void);
 

+void llog_handle_get(struct llog_handle *loghandle);
+void llog_handle_put(struct llog_handle *loghandle);

 int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle,
                       struct llog_handle **res, struct llog_logid *logid);
 int class_config_dump_handler(const struct lu_env *env,
 int llog_cat_id2handle(const struct lu_env *env, struct llog_handle *cathandle,
                       struct llog_handle **res, struct llog_logid *logid);
 int class_config_dump_handler(const struct lu_env *env,


@@ -90,5 +92,6 @@ int class_config_parse_rec(struct llog_rec_hdr *rec, char *buf, int size);
 int llog_process_or_fork(const struct lu_env *env,
                         struct llog_handle *loghandle,
                         llog_cb_t cb, void *data, void *catdata, bool fork);
 int llog_process_or_fork(const struct lu_env *env,
                         struct llog_handle *loghandle,
                         llog_cb_t cb, void *data, void *catdata, bool fork);

-int llog_cat_set_first_idx(struct llog_handle *cathandle, int index);
+int llog_cat_cleanup(const struct lu_env *env, struct llog_handle *cathandle,
+                    struct llog_handle *loghandle, int index);

 #endif
 #endif

diff --git a/lustre/obdclass/llog_ioctl.c b/lustre/obdclass/llog_ioctl.c
index 92d3d8b..f377397 100644 (file)
--- a/lustre/obdclass/llog_ioctl.c
+++ b/lustre/obdclass/llog_ioctl.c
@@ -116,9 +116,9 @@ static int llog_check_cb(const struct lu_env *env, struct llog_handle *handle,
        if (to > 0 && cur_index > to)
                RETURN(-LLOG_EEMPTY);
 
        if (to > 0 && cur_index > to)
                RETURN(-LLOG_EEMPTY);
 

-        if (handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT) {
-                struct llog_logid_rec *lir = (struct llog_logid_rec *)rec;
-                struct llog_handle *log_handle;
+       if (handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT) {
+               struct llog_logid_rec   *lir = (struct llog_logid_rec *)rec;
+               struct llog_handle      *loghandle;

 
                 if (rec->lrh_type != LLOG_LOGID_MAGIC) {
                         l = snprintf(out, remains, "[index]: %05d  [type]: "
 
                 if (rec->lrh_type != LLOG_LOGID_MAGIC) {
                         l = snprintf(out, remains, "[index]: %05d  [type]: "


@@ -128,17 +128,17 @@ static int llog_check_cb(const struct lu_env *env, struct llog_handle *handle,
                 }
                 if (handle->lgh_ctxt == NULL)
                         RETURN(-EOPNOTSUPP);
                 }
                 if (handle->lgh_ctxt == NULL)
                         RETURN(-EOPNOTSUPP);

-               rc = llog_cat_id2handle(env, handle, &log_handle, &lir->lid_id);
-                if (rc) {
-                        CDEBUG(D_IOCTL,
-                               "cannot find log #"LPX64"#"LPX64"#%08x\n",
-                               lir->lid_id.lgl_oid, lir->lid_id.lgl_oseq,
-                               lir->lid_id.lgl_ogen);
-                        RETURN(rc);
-                }
-               rc = llog_process(env, log_handle, llog_check_cb, NULL, NULL);
-               llog_close(env, log_handle);
-        } else {
+               rc = llog_cat_id2handle(env, handle, &loghandle, &lir->lid_id);
+               if (rc) {
+                       CDEBUG(D_IOCTL,
+                              "cannot find log #"LPX64"#"LPX64"#%08x\n",
+                              lir->lid_id.lgl_oid, lir->lid_id.lgl_oseq,
+                              lir->lid_id.lgl_ogen);
+                       RETURN(rc);
+               }
+               rc = llog_process(env, loghandle, llog_check_cb, NULL, NULL);
+               llog_handle_put(loghandle);
+       } else {

                bool ok;
 
                 switch (rec->lrh_type) {
                bool ok;
 
                 switch (rec->lrh_type) {


@@ -238,33 +238,26 @@ static int llog_print_cb(const struct lu_env *env, struct llog_handle *handle,
 static int llog_remove_log(const struct lu_env *env, struct llog_handle *cat,
                           struct llog_logid *logid)
 {
 static int llog_remove_log(const struct lu_env *env, struct llog_handle *cat,
                           struct llog_logid *logid)
 {

-        struct llog_handle *log;
-        int rc, index = 0;
+       struct llog_handle      *log;
+       int                      rc;

 
 

-        ENTRY;
+       ENTRY;

 
        rc = llog_cat_id2handle(env, cat, &log, logid);
 
        rc = llog_cat_id2handle(env, cat, &log, logid);

-        if (rc) {
-                CDEBUG(D_IOCTL, "cannot find log #"LPX64"#"LPX64"#%08x\n",
-                       logid->lgl_oid, logid->lgl_oseq, logid->lgl_ogen);
-                GOTO(out, rc = -ENOENT);
-        }
+       if (rc) {
+               CDEBUG(D_IOCTL, "cannot find log #"LPX64"#"LPX64"#%08x\n",
+                      logid->lgl_oid, logid->lgl_oseq, logid->lgl_ogen);
+               RETURN(-ENOENT);
+       }

 
 

-        index = log->u.phd.phd_cookie.lgc_index;
-        LASSERT(index);

        rc = llog_destroy(env, log);
        if (rc) {
                CDEBUG(D_IOCTL, "cannot destroy log\n");
                GOTO(out, rc);
        }
        rc = llog_destroy(env, log);
        if (rc) {
                CDEBUG(D_IOCTL, "cannot destroy log\n");
                GOTO(out, rc);
        }

-       down_write(&cat->lgh_lock);
-       if (cat->u.chd.chd_current_log == log)
-               cat->u.chd.chd_current_log = NULL;
-       up_write(&cat->lgh_lock);
-       llog_cat_set_first_idx(cat, index);
-       rc = llog_cancel_rec(env, cat, index);
+       llog_cat_cleanup(env, cat, log, log->u.phd.phd_cookie.lgc_index);

 out:
 out:

-       llog_close(env, log);
+       llog_handle_put(log);

        RETURN(rc);
 
 }
        RETURN(rc);
 
 }