LU-2683 lov: release all locks in closure to release sublock

[fs/lustre-release.git] / lustre / obdclass / llog.c

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index dbc1961..f08fd6e 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);
+       cfs_atomic_set(&loghandle->lgh_refcount, 1);
 
        return loghandle;
 }
@@ -79,22 +80,34 @@ struct llog_handle *llog_alloc_handle(void)
  */
 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->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);
-
 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)
@@ -259,6 +272,32 @@ out:
 }
 EXPORT_SYMBOL(llog_init_handle);
 
+int llog_copy_handler(const struct lu_env *env,
+                     struct llog_handle *llh,
+                     struct llog_rec_hdr *rec,
+                     void *data)
+{
+       struct llog_rec_hdr local_rec = *rec;
+       struct llog_handle *local_llh = (struct llog_handle *)data;
+       char *cfg_buf = (char*) (rec + 1);
+       struct lustre_cfg *lcfg;
+       int rc = 0;
+       ENTRY;
+
+       /* Append all records */
+       local_rec.lrh_len -= sizeof(*rec) + sizeof(struct llog_rec_tail);
+       rc = llog_write(env, local_llh, &local_rec, NULL, 0,
+                       (void *)cfg_buf, -1);
+
+       lcfg = (struct lustre_cfg *)cfg_buf;
+       CDEBUG(D_INFO, "idx=%d, rc=%d, len=%d, cmd %x %s %s\n",
+              rec->lrh_index, rc, rec->lrh_len, lcfg->lcfg_command,
+              lustre_cfg_string(lcfg, 0), lustre_cfg_string(lcfg, 1));
+
+       RETURN(rc);
+}
+EXPORT_SYMBOL(llog_copy_handler);
+
 static int llog_process_thread(void *arg)
 {
        struct llog_process_info        *lpi = arg;
@@ -404,7 +443,7 @@ static int llog_process_thread_daemonize(void *arg)
        cfs_daemonize_ctxt("llog_process_thread");
 
        /* client env has no keys, tags is just 0 */
-       rc = lu_env_init(&env, LCT_LOCAL);
+       rc = lu_env_init(&env, LCT_LOCAL | LCT_MG_THREAD);
        if (rc)
                goto out;
        lpi->lpi_env = &env;
@@ -930,10 +969,10 @@ int llog_close(const struct lu_env *env, struct llog_handle *loghandle)
        if (rc)
                GOTO(out, rc);
        if (lop->lop_close == NULL)
-               GOTO(out, -EOPNOTSUPP);
+               GOTO(out, rc = -EOPNOTSUPP);
        rc = lop->lop_close(env, loghandle);
 out:
-       llog_free_handle(loghandle);
+       llog_handle_put(loghandle);
        RETURN(rc);
 }
 EXPORT_SYMBOL(llog_close);