LU-8900 snapshot: simulate readonly device

[fs/lustre-release.git] / lustre / mdt / mdt_coordinator.c

diff --git a/lustre/mdt/mdt_coordinator.c b/lustre/mdt/mdt_coordinator.c
index 6fcbb4d..a981809 100644 (file)
--- a/lustre/mdt/mdt_coordinator.c
+++ b/lustre/mdt/mdt_coordinator.c
@@ -417,8 +417,8 @@ static int mdt_coordinator(void *data)
        int                      request_sz;
        ENTRY;
 
-       cdt->cdt_thread.t_flags = SVC_RUNNING;
-       wake_up(&cdt->cdt_thread.t_ctl_waitq);
+       cdt->cdt_flags = SVC_RUNNING;
+       wake_up(&cdt->cdt_waitq);
 
        CDEBUG(D_HSM, "%s: coordinator thread starting, pid=%d\n",
               mdt_obd_name(mdt), current_pid());
@@ -442,23 +442,22 @@ static int mdt_coordinator(void *data)
 
                lwi = LWI_TIMEOUT(cfs_time_seconds(cdt->cdt_loop_period),
                                  NULL, NULL);
-               l_wait_event(cdt->cdt_thread.t_ctl_waitq,
-                            (cdt->cdt_thread.t_flags &
-                             (SVC_STOPPING|SVC_EVENT)),
+               l_wait_event(cdt->cdt_waitq,
+                            cdt->cdt_flags & (SVC_STOPPING|SVC_EVENT),
                             &lwi);
 
                CDEBUG(D_HSM, "coordinator resumes\n");
 
-               if (cdt->cdt_thread.t_flags & SVC_STOPPING ||
+               if (cdt->cdt_flags & SVC_STOPPING ||
                    cdt->cdt_state == CDT_STOPPING) {
-                       cdt->cdt_thread.t_flags &= ~SVC_STOPPING;
+                       cdt->cdt_flags &= ~SVC_STOPPING;
                        rc = 0;
                        break;
                }
 
                /* wake up before timeout, new work arrives */
-               if (cdt->cdt_thread.t_flags & SVC_EVENT)
-                       cdt->cdt_thread.t_flags &= ~SVC_EVENT;
+               if (cdt->cdt_flags & SVC_EVENT)
+                       cdt->cdt_flags &= ~SVC_EVENT;
 
                /* if coordinator is suspended continue to wait */
                if (cdt->cdt_state == CDT_DISABLE) {
@@ -567,8 +566,8 @@ out:
                 * by mdt_stop_coordinator(), we have to ack
                 * and cdt cleaning will be done by event sender
                 */
-               cdt->cdt_thread.t_flags = SVC_STOPPED;
-               wake_up(&cdt->cdt_thread.t_ctl_waitq);
+               cdt->cdt_flags = SVC_STOPPED;
+               wake_up(&cdt->cdt_waitq);
        }
 
        if (rc != 0)
@@ -750,8 +749,8 @@ int mdt_hsm_cdt_wakeup(struct mdt_device *mdt)
                RETURN(-ESRCH);
 
        /* wake up coordinator */
-       cdt->cdt_thread.t_flags = SVC_EVENT;
-       wake_up(&cdt->cdt_thread.t_ctl_waitq);
+       cdt->cdt_flags = SVC_EVENT;
+       wake_up(&cdt->cdt_waitq);
 
        RETURN(0);
 }
@@ -771,7 +770,7 @@ int mdt_hsm_cdt_init(struct mdt_device *mdt)
 
        cdt->cdt_state = CDT_STOPPED;
 
-       init_waitqueue_head(&cdt->cdt_thread.t_ctl_waitq);
+       init_waitqueue_head(&cdt->cdt_waitq);
        mutex_init(&cdt->cdt_llog_lock);
        init_rwsem(&cdt->cdt_agent_lock);
        init_rwsem(&cdt->cdt_request_lock);
@@ -883,6 +882,9 @@ static int mdt_hsm_cdt_start(struct mdt_device *mdt)
                       " for registered restore: %d\n",
                       mdt_obd_name(mdt), rc);
 
+       if (mdt->mdt_bottom->dd_rdonly)
+               RETURN(0);
+
        task = kthread_run(mdt_coordinator, cdt_mti, "hsm_cdtr");
        if (IS_ERR(task)) {
                rc = PTR_ERR(task);
@@ -896,8 +898,8 @@ static int mdt_hsm_cdt_start(struct mdt_device *mdt)
                rc = 0;
        }
 
-       wait_event(cdt->cdt_thread.t_ctl_waitq,
-                      (cdt->cdt_thread.t_flags & SVC_RUNNING));
+       wait_event(cdt->cdt_waitq,
+                      (cdt->cdt_flags & SVC_RUNNING));
 
        cdt->cdt_state = CDT_RUNNING;
        mdt->mdt_opts.mo_coordinator = 1;
@@ -917,6 +919,9 @@ int mdt_hsm_cdt_stop(struct mdt_device *mdt)
        struct mdt_thread_info          *cdt_mti;
        ENTRY;
 
+       if (mdt->mdt_opts.mo_coordinator == 0)
+               RETURN(0);
+
        if (cdt->cdt_state == CDT_STOPPED) {
                CERROR("%s: Coordinator already stopped\n",
                       mdt_obd_name(mdt));
@@ -925,10 +930,10 @@ int mdt_hsm_cdt_stop(struct mdt_device *mdt)
 
        if (cdt->cdt_state != CDT_STOPPING) {
                /* stop coordinator thread before cleaning */
-               cdt->cdt_thread.t_flags = SVC_STOPPING;
-               wake_up(&cdt->cdt_thread.t_ctl_waitq);
-               wait_event(cdt->cdt_thread.t_ctl_waitq,
-                          cdt->cdt_thread.t_flags & SVC_STOPPED);
+               cdt->cdt_flags = SVC_STOPPING;
+               wake_up(&cdt->cdt_waitq);
+               wait_event(cdt->cdt_waitq,
+                          cdt->cdt_flags & SVC_STOPPED);
        }
        cdt->cdt_state = CDT_STOPPED;
 
@@ -1502,6 +1507,8 @@ static int mdt_cancel_all_cb(const struct lu_env *env,
  */
 static int hsm_cancel_all_actions(struct mdt_device *mdt)
 {
+       struct lu_env                    env;
+       struct lu_context                session;
        struct mdt_thread_info          *mti;
        struct coordinator              *cdt = &mdt->mdt_coordinator;
        struct cdt_agent_req            *car;
@@ -1512,8 +1519,25 @@ static int hsm_cancel_all_actions(struct mdt_device *mdt)
        enum cdt_states                  save_state;
        ENTRY;
 
-       /* retrieve coordinator context */
-       mti = lu_context_key_get(&cdt->cdt_env.le_ctx, &mdt_thread_key);
+       rc = lu_env_init(&env, LCT_MD_THREAD);
+       if (rc < 0)
+               RETURN(rc);
+
+       /* for mdt_ucred(), lu_ucred stored in lu_ucred_key */
+       rc = lu_context_init(&session, LCT_SERVER_SESSION);
+       if (rc < 0)
+               GOTO(out_env, rc);
+
+       lu_context_enter(&session);
+       env.le_ses = &session;
+
+       mti = lu_context_key_get(&env.le_ctx, &mdt_thread_key);
+       LASSERT(mti != NULL);
+
+       mti->mti_env = &env;
+       mti->mti_mdt = mdt;
+
+       hsm_init_ucred(mdt_ucred(mti));
 
        /* disable coordinator */
        save_state = cdt->cdt_state;
@@ -1549,7 +1573,7 @@ static int hsm_cancel_all_actions(struct mdt_device *mdt)
                        if (hal == NULL) {
                                mdt_cdt_put_request(car);
                                up_read(&cdt->cdt_request_lock);
-                               GOTO(out, rc = -ENOMEM);
+                               GOTO(out_cdt_state, rc = -ENOMEM);
                        }
                }
 
@@ -1589,9 +1613,13 @@ static int hsm_cancel_all_actions(struct mdt_device *mdt)
 
        rc = cdt_llog_process(mti->mti_env, mti->mti_mdt,
                              mdt_cancel_all_cb, &hcad);
-out:
+out_cdt_state:
        /* enable coordinator */
        cdt->cdt_state = save_state;
+       lu_context_exit(&session);
+       lu_context_fini(&session);
+out_env:
+       lu_env_fini(&env);
 
        RETURN(rc);
 }