ORNL-10: Basic IR implementation

[fs/lustre-release.git] / lustre / mgs / mgs_handler.c

diff --git a/lustre/mgs/mgs_handler.c b/lustre/mgs/mgs_handler.c
index 4829c1f..45c74b2 100644 (file)
--- a/lustre/mgs/mgs_handler.c
+++ b/lustre/mgs/mgs_handler.c
@@ -196,6 +196,9 @@ static int mgs_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
                 GOTO(err_ops, rc = -EROFS);
         }
 
+        obd->u.obt.obt_magic = OBT_MAGIC;
+        obd->u.obt.obt_instance = 0;
+
         /* namespace for mgs llog */
         obd->obd_namespace = ldlm_namespace_new(obd ,"MGS",
                                                 LDLM_NAMESPACE_SERVER,
@@ -225,6 +228,7 @@ static int mgs_setup(struct obd_device *obd, struct lustre_cfg *lcfg)
         /* Internal mgs setup */
         mgs_init_fsdb_list(obd);
         cfs_sema_init(&mgs->mgs_sem, 1);
+        mgs->mgs_start_time = cfs_time_current_sec();
 
         /* Setup proc */
         lprocfs_mgs_init_vars(&lvars);
@@ -325,50 +329,55 @@ static int mgs_cleanup(struct obd_device *obd)
 }
 
 /* similar to filter_prepare_destroy */
-static int mgs_get_cfg_lock(struct obd_device *obd, char *fsname,
-                            struct lustre_handle *lockh)
+int mgs_get_lock(struct obd_device *obd, struct ldlm_res_id *res,
+                 struct lustre_handle *lockh)
 {
-        struct ldlm_res_id res_id;
         int rc, flags = 0;
         ENTRY;
 
-        rc = mgc_fsname2resid(fsname, &res_id, CONFIG_T_CONFIG);
-        if (!rc)
-                rc = ldlm_cli_enqueue_local(obd->obd_namespace, &res_id,
-                                            LDLM_PLAIN, NULL, LCK_EX,
-                                            &flags, ldlm_blocking_ast,
-                                            ldlm_completion_ast, NULL,
-                                            fsname, 0, NULL, lockh);
+        rc = ldlm_cli_enqueue_local(obd->obd_namespace, res,
+                                    LDLM_PLAIN, NULL, LCK_EX,
+                                    &flags, ldlm_blocking_ast,
+                                    ldlm_completion_ast, NULL,
+                                    NULL, 0, NULL, lockh);
         if (rc)
-                CERROR("can't take cfg lock for %s (%d)\n", fsname, rc);
+                CERROR("can't take cfg lock for "LPX64"/"LPX64"(%d)\n",
+                       le64_to_cpu(res->name[0]), le64_to_cpu(res->name[1]),
+                       rc);
 
         RETURN(rc);
 }
 
-static int mgs_put_cfg_lock(struct lustre_handle *lockh)
+int mgs_put_lock(struct lustre_handle *lockh)
 {
         ENTRY;
-        ldlm_lock_decref(lockh, LCK_EX);
+        ldlm_lock_decref_and_cancel(lockh, LCK_EX);
         RETURN(0);
 }
 
 void mgs_revoke_lock(struct obd_device *obd, struct fs_db *fsdb)
 {
         struct lustre_handle lockh;
+        struct ldlm_res_id   res_id;
         int                  lockrc;
+        int                  bit;
+        int                  rc;
 
         LASSERT(fsdb->fsdb_name[0] != '\0');
+        rc = mgc_fsname2resid(fsdb->fsdb_name, &res_id, CONFIG_T_CONFIG);
+        LASSERT(rc == 0);
 
-        if (cfs_test_and_set_bit(FSDB_REVOKING_LOCK, &fsdb->fsdb_flags) == 0) {
-                lockrc = mgs_get_cfg_lock(obd, fsdb->fsdb_name, &lockh);
+        bit = FSDB_REVOKING_LOCK;
+        if (!rc && cfs_test_and_set_bit(bit, &fsdb->fsdb_flags) == 0) {
+                lockrc = mgs_get_lock(obd, &res_id, &lockh);
                 /* clear the bit before lock put */
-                cfs_clear_bit(FSDB_REVOKING_LOCK, &fsdb->fsdb_flags);
+                cfs_clear_bit(bit, &fsdb->fsdb_flags);
 
                 if (lockrc != ELDLM_OK)
                         CERROR("lock error %d for fs %s\n",
                                lockrc, fsdb->fsdb_name);
                 else
-                        mgs_put_cfg_lock(&lockh);
+                        mgs_put_lock(&lockh);
         }
 }
 
@@ -433,6 +442,7 @@ static int mgs_handle_target_reg(struct ptlrpc_request *req)
         struct obd_device *obd = req->rq_export->exp_obd;
         struct mgs_target_info *mti, *rep_mti;
         struct fs_db *fsdb;
+        int opc;
         int rc = 0;
         ENTRY;
 
@@ -440,6 +450,26 @@ static int mgs_handle_target_reg(struct ptlrpc_request *req)
 
         mti = req_capsule_client_get(&req->rq_pill, &RMF_MGS_TARGET_INFO);
 
+        opc = mti->mti_flags & LDD_F_OPC_MASK;
+        if (opc == LDD_F_OPC_READY) {
+                CDEBUG(D_MGS, "fs: %s index: %d is ready to reconnect.\n",
+                       mti->mti_fsname, mti->mti_stripe_index);
+                rc = mgs_ir_update(obd, mti);
+                if (rc) {
+                        LASSERT(!(mti->mti_flags & LDD_F_IR_CAPABLE));
+                        CERROR("Update IR return with %d(ignore and IR "
+                               "disabled)\n", rc);
+                }
+                GOTO(out_nolock, rc);
+        }
+
+        /* Do not support unregistering right now. */
+        if (opc != LDD_F_OPC_REG)
+                GOTO(out_nolock, rc = -EINVAL);
+
+        CDEBUG(D_MGS, "fs: %s index: %d is registered to MGS.\n",
+               mti->mti_fsname, mti->mti_stripe_index);
+
         if (mti->mti_flags & LDD_F_NEED_INDEX)
                 mti->mti_flags |= LDD_F_WRITECONF;
 
@@ -534,6 +564,11 @@ out_nolock:
         CDEBUG(D_MGS, "replying with %s, index=%d, rc=%d\n", mti->mti_svname,
                mti->mti_stripe_index, rc);
         req->rq_status = rc;
+        if (rc)
+                /* we need an error flag to tell the target what's going on,
+                 * instead of just doing it by error code only. */
+                mti->mti_flags |= LDD_F_ERROR;
+
         rc = req_capsule_server_pack(&req->rq_pill);
         if (rc)
                 RETURN(rc);
@@ -581,6 +616,33 @@ static int mgs_set_info_rpc(struct ptlrpc_request *req)
         RETURN(rc);
 }
 
+static int mgs_config_read(struct ptlrpc_request *req)
+{
+        struct mgs_config_body *body;
+        int rc;
+        ENTRY;
+
+        body = req_capsule_client_get(&req->rq_pill, &RMF_MGS_CONFIG_BODY);
+        if (body == NULL)
+                RETURN(-EINVAL);
+
+        switch (body->mcb_type) {
+        case CONFIG_T_RECOVER:
+                rc = mgs_get_ir_logs(req);
+                break;
+
+        case CONFIG_T_CONFIG:
+                rc = -ENOTSUPP;
+                break;
+
+        default:
+                rc = -EINVAL;
+                break;
+        }
+
+        RETURN(rc);
+}
+
 /*
  * similar as in ost_connect_check_sptlrpc()
  */
@@ -721,7 +783,11 @@ int mgs_handle(struct ptlrpc_request *req)
                 req_capsule_set(&req->rq_pill, &RQF_MGS_SET_INFO);
                 rc = mgs_set_info_rpc(req);
                 break;
-
+        case MGS_CONFIG_READ:
+                DEBUG_REQ(D_MGS, req, "read config");
+                req_capsule_set(&req->rq_pill, &RQF_MGS_CONFIG_READ);
+                rc = mgs_config_read(req);
+                break;
         case LDLM_ENQUEUE:
                 DEBUG_REQ(D_MGS, req, "enqueue");
                 req_capsule_set(&req->rq_pill, &RQF_LDLM_ENQUEUE);