LU-3365 lmv: support DNE with HSM.

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

diff --git a/lustre/obdclass/llog.c b/lustre/obdclass/llog.c
index cd457f5..2cd9598 100644 (file)
--- a/lustre/obdclass/llog.c
+++ b/lustre/obdclass/llog.c
@@ -26,6 +26,8 @@
 /*
  * Copyright (c) 2003, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
+ *
+ * Copyright (c) 2012, 2013, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
@@ -68,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;
 }
@@ -77,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)
@@ -101,8 +116,8 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
         int rc = 0;
         ENTRY;
 
-        CDEBUG(D_RPCTRACE, "Canceling %d in log "LPX64"\n",
-               index, loghandle->lgh_id.lgl_oid);
+        CDEBUG(D_RPCTRACE, "Canceling %d in log "DOSTID"\n",
+               index, POSTID(&loghandle->lgh_id.lgl_oi));
 
         if (index == 0) {
                 CERROR("Can't cancel index 0 which is header\n");
@@ -124,11 +139,10 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
                spin_unlock(&loghandle->lgh_hdr_lock);
                rc = llog_destroy(env, loghandle);
                if (rc < 0) {
-                       CERROR("%s: can't destroy empty llog #"LPX64"#"LPX64
+                       CERROR("%s: can't destroy empty llog #"DOSTID
                               "#%08x: rc = %d\n",
                               loghandle->lgh_ctxt->loc_obd->obd_name,
-                              loghandle->lgh_id.lgl_oid,
-                              loghandle->lgh_id.lgl_oseq,
+                              POSTID(&loghandle->lgh_id.lgl_oi),
                               loghandle->lgh_id.lgl_ogen, rc);
                        GOTO(out_err, rc);
                }
@@ -138,11 +152,10 @@ int llog_cancel_rec(const struct lu_env *env, struct llog_handle *loghandle,
 
        rc = llog_write(env, loghandle, &llh->llh_hdr, NULL, 0, NULL, 0);
        if (rc < 0) {
-               CERROR("%s: fail to write header for llog #"LPX64"#"LPX64
+               CERROR("%s: fail to write header for llog #"DOSTID
                       "#%08x: rc = %d\n",
                       loghandle->lgh_ctxt->loc_obd->obd_name,
-                      loghandle->lgh_id.lgl_oid,
-                      loghandle->lgh_id.lgl_oseq,
+                      POSTID(&loghandle->lgh_id.lgl_oi),
                       loghandle->lgh_id.lgl_ogen, rc);
                GOTO(out_err, rc);
        }
@@ -257,6 +270,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;
@@ -399,10 +438,10 @@ static int llog_process_thread_daemonize(void *arg)
        struct lu_env                    env;
        int                              rc;
 
-       cfs_daemonize_ctxt("llog_process_thread");
+       unshare_fs_struct();
 
        /* 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;
@@ -441,9 +480,9 @@ int llog_process_or_fork(const struct lu_env *env,
                 * init the new one in llog_process_thread_daemonize. */
                lpi->lpi_env = NULL;
                init_completion(&lpi->lpi_completion);
-               rc = cfs_create_thread(llog_process_thread_daemonize, lpi,
-                                      CFS_DAEMON_FLAGS);
-               if (rc < 0) {
+               rc = PTR_ERR(kthread_run(llog_process_thread_daemonize, lpi,
+                                            "llog_process_thread"));
+               if (IS_ERR_VALUE(rc)) {
                        CERROR("%s: cannot start thread: rc = %d\n",
                               loghandle->lgh_ctxt->loc_obd->obd_name, rc);
                        OBD_FREE_PTR(lpi);
@@ -928,10 +967,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);