lustre/mgs/mgs_handler.c

   1 /*
   2  * GPL HEADER START
   3  *
   4  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   5  *
   6  * This program is free software; you can redistribute it and/or modify
   7  * it under the terms of the GNU General Public License version 2 only,
   8  * as published by the Free Software Foundation.
   9  *
  10  * This program is distributed in the hope that it will be useful, but
  11  * WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13  * General Public License version 2 for more details (a copy is included
  14  * in the LICENSE file that accompanied this code).
  15  *
  16  * You should have received a copy of the GNU General Public License
  17  * version 2 along with this program; If not, see
  18  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  19  *
  20  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  21  * CA 95054 USA or visit www.sun.com if you need additional information or
  22  * have any questions.
  23  *
  24  * GPL HEADER END
  25  */
  26 /*
  27  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  28  * Use is subject to license terms.
  29  *
  30  * Copyright (c) 2011, 2012, Whamcloud, Inc.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lustre/mgs/mgs_handler.c
  37  *
  38  * Author: Nathan Rutman <nathan@clusterfs.com>
  39  * Author: Alex Zhuravlev <bzzz@whamcloud.com>
  40  * Author: Mikhail Pershin <tappro@whamcloud.com>
  41  */
  42
  43 #define DEBUG_SUBSYSTEM S_MGS
  44 #define D_MGS D_CONFIG
  45
  46 #ifdef __KERNEL__
  47 # include <linux/module.h>
  48 # include <linux/pagemap.h>
  49 # include <linux/miscdevice.h>
  50 # include <linux/init.h>
  51 #else
  52 # include <liblustre.h>
  53 #endif
  54
  55 #include <obd_class.h>
  56 #include <lustre_dlm.h>
  57 #include <lprocfs_status.h>
  58 #include <lustre_fsfilt.h>
  59 #include <lustre_disk.h>
  60 #include <lustre_param.h>
  61 #include <lustre_log.h>
  62
  63 #include "mgs_internal.h"
  64
  65 /* Establish a connection to the MGS.*/
  66 static int mgs_connect(const struct lu_env *env,
  67                        struct obd_export **exp, struct obd_device *obd,
  68                        struct obd_uuid *cluuid, struct obd_connect_data *data,
  69                        void *localdata)
  70 {
  71         struct obd_export *lexp;
  72         struct lustre_handle conn = { 0 };
  73         int rc;
  74         ENTRY;
  75
  76         if (!exp || !obd || !cluuid)
  77                 RETURN(-EINVAL);
  78
  79         rc = class_connect(&conn, obd, cluuid);
  80         if (rc)
  81                 RETURN(rc);
  82
  83         lexp = class_conn2export(&conn);
  84         if (lexp == NULL)
  85                 GOTO(out, rc = -EFAULT);
  86
  87         mgs_counter_incr(lexp, LPROC_MGS_CONNECT);
  88
  89         if (data != NULL) {
  90                 data->ocd_connect_flags &= MGS_CONNECT_SUPPORTED;
  91                 lexp->exp_connect_flags = data->ocd_connect_flags;
  92                 data->ocd_version = LUSTRE_VERSION_CODE;
  93         }
  94
  95         rc = mgs_export_stats_init(obd, lexp, localdata);
  96
  97 out:
  98         if (rc) {
  99                 class_disconnect(lexp);
 100         } else {
 101                 *exp = lexp;
 102         }
 103
 104         RETURN(rc);
 105 }
 106
 107 static int mgs_reconnect(const struct lu_env *env,
 108                          struct obd_export *exp, struct obd_device *obd,
 109                          struct obd_uuid *cluuid, struct obd_connect_data *data,
 110                          void *localdata)
 111 {
 112         ENTRY;
 113
 114         if (exp == NULL || obd == NULL || cluuid == NULL)
 115                 RETURN(-EINVAL);
 116
 117         mgs_counter_incr(exp, LPROC_MGS_CONNECT);
 118
 119         if (data != NULL) {
 120                 data->ocd_connect_flags &= MGS_CONNECT_SUPPORTED;
 121                 exp->exp_connect_flags = data->ocd_connect_flags;
 122                 data->ocd_version = LUSTRE_VERSION_CODE;
 123         }
 124
 125         RETURN(mgs_export_stats_init(obd, exp, localdata));
 126 }
 127
 128 static int mgs_disconnect(struct obd_export *exp)
 129 {
 130         int rc;
 131         ENTRY;
 132
 133         LASSERT(exp);
 134
 135         mgs_fsc_cleanup(exp);
 136
 137         class_export_get(exp);
 138         mgs_counter_incr(exp, LPROC_MGS_DISCONNECT);
 139
 140         rc = server_disconnect_export(exp);
 141         class_export_put(exp);
 142         RETURN(rc);
 143 }
 144
 145 static int mgs_handle(struct ptlrpc_request *req);
 146
 147 static int mgs_llog_init(struct obd_device *obd, struct obd_llog_group *olg,
 148                          struct obd_device *tgt, int *index)
 149 {
 150         int rc;
 151         ENTRY;
 152
 153         LASSERT(olg == &obd->obd_olg);
 154         rc = llog_setup(obd, olg, LLOG_CONFIG_ORIG_CTXT, obd, 0, NULL,
 155                         &llog_lvfs_ops);
 156         RETURN(rc);
 157 }
 158
 159 static int mgs_llog_finish(struct obd_device *obd, int count)
 160 {
 161         struct llog_ctxt *ctxt;
 162         int rc = 0;
 163         ENTRY;
 164
 165         ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 166         if (ctxt)
 167                 rc = llog_cleanup(ctxt);
 168
 169         RETURN(rc);
 170 }
 171
 172 static int mgs_completion_ast_config(struct ldlm_lock *lock, int flags,
 173                                      void *cbdata)
 174 {
 175         ENTRY;
 176
 177         if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
 178                        LDLM_FL_BLOCK_CONV))) {
 179                 struct fs_db *fsdb = (struct fs_db *)lock->l_ast_data;
 180                 struct lustre_handle lockh;
 181
 182                 /* clear the bit before lock put */
 183                 cfs_clear_bit(FSDB_REVOKING_LOCK, &fsdb->fsdb_flags);
 184
 185                 ldlm_lock2handle(lock, &lockh);
 186                 ldlm_lock_decref_and_cancel(&lockh, LCK_EX);
 187         }
 188
 189         RETURN(ldlm_completion_ast(lock, flags, cbdata));
 190 }
 191
 192 static int mgs_completion_ast_ir(struct ldlm_lock *lock, int flags,
 193                                  void *cbdata)
 194 {
 195         ENTRY;
 196
 197         if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
 198                        LDLM_FL_BLOCK_CONV))) {
 199                 struct fs_db *fsdb;
 200
 201                 /* l_ast_data is used as a marker to avoid cancel ldlm lock
 202                  * twice. See LU-1259. */
 203                 lock_res_and_lock(lock);
 204                 fsdb = (struct fs_db *)lock->l_ast_data;
 205                 lock->l_ast_data = NULL;
 206                 unlock_res_and_lock(lock);
 207
 208                 if (fsdb != NULL) {
 209                         struct lustre_handle lockh;
 210
 211                         mgs_ir_notify_complete(fsdb);
 212
 213                         ldlm_lock2handle(lock, &lockh);
 214                         ldlm_lock_decref_and_cancel(&lockh, LCK_EX);
 215                 }
 216         }
 217
 218         RETURN(ldlm_completion_ast(lock, flags, cbdata));
 219 }
 220
 221 void mgs_revoke_lock(struct mgs_device *mgs, struct fs_db *fsdb, int type)
 222 {
 223         ldlm_completion_callback cp = NULL;
 224         struct lustre_handle     lockh = { 0 };
 225         struct ldlm_res_id       res_id;
 226         int flags = LDLM_FL_ATOMIC_CB;
 227         int rc;
 228         ENTRY;
 229
 230         LASSERT(fsdb->fsdb_name[0] != '\0');
 231         rc = mgc_fsname2resid(fsdb->fsdb_name, &res_id, type);
 232         LASSERT(rc == 0);
 233
 234         switch (type) {
 235         case CONFIG_T_CONFIG:
 236                 cp = mgs_completion_ast_config;
 237                 if (cfs_test_and_set_bit(FSDB_REVOKING_LOCK, &fsdb->fsdb_flags))
 238                         rc = -EALREADY;
 239                 break;
 240         case CONFIG_T_RECOVER:
 241                 cp = mgs_completion_ast_ir;
 242         default:
 243                 break;
 244         }
 245
 246         if (!rc) {
 247                 LASSERT(cp != NULL);
 248                 rc = ldlm_cli_enqueue_local(mgs->mgs_obd->obd_namespace,
 249                                             &res_id, LDLM_PLAIN, NULL, LCK_EX,
 250                                             &flags, ldlm_blocking_ast, cp,
 251                                             NULL, fsdb, 0, NULL, &lockh);
 252                 if (rc != ELDLM_OK) {
 253                         CERROR("can't take cfg lock for "LPX64"/"LPX64"(%d)\n",
 254                                le64_to_cpu(res_id.name[0]),
 255                                le64_to_cpu(res_id.name[1]), rc);
 256
 257                         if (type == CONFIG_T_CONFIG)
 258                                 cfs_clear_bit(FSDB_REVOKING_LOCK,
 259                                               &fsdb->fsdb_flags);
 260                 }
 261                 /* lock has been cancelled in completion_ast. */
 262         }
 263
 264         RETURN_EXIT;
 265 }
 266
 267 /* rc=0 means ok
 268       1 means update
 269      <0 means error */
 270 static int mgs_check_target(const struct lu_env *env,
 271                             struct mgs_device *mgs,
 272                             struct mgs_target_info *mti)
 273 {
 274         int rc;
 275         ENTRY;
 276
 277         rc = mgs_check_index(env, mgs, mti);
 278         if (rc == 0) {
 279                 LCONSOLE_ERROR_MSG(0x13b, "%s claims to have registered, but "
 280                                    "this MGS does not know about it, preventing "
 281                                    "registration.\n", mti->mti_svname);
 282                 rc = -ENOENT;
 283         } else if (rc == -1) {
 284                 LCONSOLE_ERROR_MSG(0x13c, "Client log %s-client has "
 285                                    "disappeared! Regenerating all logs.\n",
 286                                    mti->mti_fsname);
 287                 mti->mti_flags |= LDD_F_WRITECONF;
 288                 rc = 1;
 289         } else {
 290                 /* Index is correctly marked as used */
 291
 292                 /* If the logs don't contain the mti_nids then add
 293                    them as failover nids */
 294                 rc = mgs_check_failnid(env, mgs, mti);
 295         }
 296
 297         RETURN(rc);
 298 }
 299
 300 /* Ensure this is not a failover node that is connecting first*/
 301 static int mgs_check_failover_reg(struct mgs_target_info *mti)
 302 {
 303         lnet_nid_t nid;
 304         char *ptr;
 305         int i;
 306
 307         ptr = mti->mti_params;
 308         while (class_find_param(ptr, PARAM_FAILNODE, &ptr) == 0) {
 309                 while (class_parse_nid_quiet(ptr, &nid, &ptr) == 0) {
 310                         for (i = 0; i < mti->mti_nid_count; i++) {
 311                                 if (nid == mti->mti_nids[i]) {
 312                                         LCONSOLE_WARN("Denying initial registra"
 313                                                       "tion attempt from nid %s"
 314                                                       ", specified as failover"
 315                                                       "\n",libcfs_nid2str(nid));
 316                                         return -EADDRNOTAVAIL;
 317                                 }
 318                         }
 319                 }
 320         }
 321         return 0;
 322 }
 323
 324 /* Called whenever a target starts up.  Flags indicate first connect, etc. */
 325 static int mgs_handle_target_reg(struct ptlrpc_request *req)
 326 {
 327         struct obd_device *obd = req->rq_export->exp_obd;
 328         struct mgs_device *mgs = exp2mgs_dev(req->rq_export);
 329         struct lu_env     *env = req->rq_svc_thread->t_env;
 330         struct mgs_target_info *mti, *rep_mti;
 331         struct fs_db *fsdb;
 332         int opc;
 333         int rc = 0;
 334         ENTRY;
 335
 336         mgs_counter_incr(req->rq_export, LPROC_MGS_TARGET_REG);
 337
 338         mti = req_capsule_client_get(&req->rq_pill, &RMF_MGS_TARGET_INFO);
 339
 340         opc = mti->mti_flags & LDD_F_OPC_MASK;
 341         if (opc == LDD_F_OPC_READY) {
 342                 CDEBUG(D_MGS, "fs: %s index: %d is ready to reconnect.\n",
 343                        mti->mti_fsname, mti->mti_stripe_index);
 344                 rc = mgs_ir_update(env, mgs, mti);
 345                 if (rc) {
 346                         LASSERT(!(mti->mti_flags & LDD_F_IR_CAPABLE));
 347                         CERROR("Update IR return with %d(ignore and IR "
 348                                "disabled)\n", rc);
 349                 }
 350                 GOTO(out_nolock, rc);
 351         }
 352
 353         /* Do not support unregistering right now. */
 354         if (opc != LDD_F_OPC_REG)
 355                 GOTO(out_nolock, rc = -EINVAL);
 356
 357         CDEBUG(D_MGS, "fs: %s index: %d is registered to MGS.\n",
 358                mti->mti_fsname, mti->mti_stripe_index);
 359
 360         if (mti->mti_flags & LDD_F_NEED_INDEX)
 361                 mti->mti_flags |= LDD_F_WRITECONF;
 362
 363         if (!(mti->mti_flags & (LDD_F_WRITECONF | LDD_F_UPGRADE14 |
 364                                 LDD_F_UPDATE))) {
 365                 /* We're just here as a startup ping. */
 366                 CDEBUG(D_MGS, "Server %s is running on %s\n",
 367                        mti->mti_svname, obd_export_nid2str(req->rq_export));
 368                 rc = mgs_check_target(env, mgs, mti);
 369                 /* above will set appropriate mti flags */
 370                 if (rc <= 0)
 371                         /* Nothing wrong, or fatal error */
 372                         GOTO(out_nolock, rc);
 373         } else {
 374                 if (!(mti->mti_flags & LDD_F_NO_PRIMNODE)
 375                     && (rc = mgs_check_failover_reg(mti)))
 376                         GOTO(out_nolock, rc);
 377         }
 378
 379         OBD_FAIL_TIMEOUT(OBD_FAIL_MGS_PAUSE_TARGET_REG, 10);
 380
 381         if (mti->mti_flags & LDD_F_WRITECONF) {
 382                 if (mti->mti_flags & LDD_F_SV_TYPE_MDT &&
 383                     mti->mti_stripe_index == 0) {
 384                         rc = mgs_erase_logs(env, mgs, mti->mti_fsname);
 385                         LCONSOLE_WARN("%s: Logs for fs %s were removed by user "
 386                                       "request.  All servers must be restarted "
 387                                       "in order to regenerate the logs."
 388                                       "\n", obd->obd_name, mti->mti_fsname);
 389                 } else if (mti->mti_flags &
 390                            (LDD_F_SV_TYPE_OST | LDD_F_SV_TYPE_MDT)) {
 391                         rc = mgs_erase_log(env, mgs, mti->mti_svname);
 392                         LCONSOLE_WARN("%s: Regenerating %s log by user "
 393                                       "request.\n",
 394                                       obd->obd_name, mti->mti_svname);
 395                 }
 396                 mti->mti_flags |= LDD_F_UPDATE;
 397                 /* Erased logs means start from scratch. */
 398                 mti->mti_flags &= ~LDD_F_UPGRADE14;
 399                 if (rc)
 400                         GOTO(out_nolock, rc);
 401         }
 402
 403         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
 404         if (rc) {
 405                 CERROR("Can't get db for %s: %d\n", mti->mti_fsname, rc);
 406                 GOTO(out_nolock, rc);
 407         }
 408
 409         /*
 410          * Log writing contention is handled by the fsdb_mutex.
 411          *
 412          * It should be alright if someone was reading while we were
 413          * updating the logs - if we revoke at the end they will just update
 414          * from where they left off.
 415          */
 416
 417         if (mti->mti_flags & LDD_F_UPGRADE14) {
 418                 CERROR("Can't upgrade from 1.4 (%d)\n", rc);
 419                 GOTO(out, rc);
 420         }
 421
 422         if (mti->mti_flags & LDD_F_UPDATE) {
 423                 CDEBUG(D_MGS, "updating %s, index=%d\n", mti->mti_svname,
 424                        mti->mti_stripe_index);
 425
 426                 /* create or update the target log
 427                    and update the client/mdt logs */
 428                 rc = mgs_write_log_target(env, mgs, mti, fsdb);
 429                 if (rc) {
 430                         CERROR("Failed to write %s log (%d)\n",
 431                                mti->mti_svname, rc);
 432                         GOTO(out, rc);
 433                 }
 434
 435                 mti->mti_flags &= ~(LDD_F_VIRGIN | LDD_F_UPDATE |
 436                                     LDD_F_NEED_INDEX | LDD_F_WRITECONF |
 437                                     LDD_F_UPGRADE14);
 438                 mti->mti_flags |= LDD_F_REWRITE_LDD;
 439         }
 440
 441 out:
 442         mgs_revoke_lock(mgs, fsdb, CONFIG_T_CONFIG);
 443
 444 out_nolock:
 445         CDEBUG(D_MGS, "replying with %s, index=%d, rc=%d\n", mti->mti_svname,
 446                mti->mti_stripe_index, rc);
 447         req->rq_status = rc;
 448         if (rc)
 449                 /* we need an error flag to tell the target what's going on,
 450                  * instead of just doing it by error code only. */
 451                 mti->mti_flags |= LDD_F_ERROR;
 452
 453         rc = req_capsule_server_pack(&req->rq_pill);
 454         if (rc)
 455                 RETURN(rc);
 456
 457         /* send back the whole mti in the reply */
 458         rep_mti = req_capsule_server_get(&req->rq_pill, &RMF_MGS_TARGET_INFO);
 459         *rep_mti = *mti;
 460
 461         /* Flush logs to disk */
 462         dt_sync(req->rq_svc_thread->t_env, mgs->mgs_bottom);
 463         RETURN(rc);
 464 }
 465
 466 static int mgs_set_info_rpc(struct ptlrpc_request *req)
 467 {
 468         struct mgs_device *mgs = exp2mgs_dev(req->rq_export);
 469         struct lu_env     *env = req->rq_svc_thread->t_env;
 470         struct mgs_send_param *msp, *rep_msp;
 471         struct mgs_thread_info *mgi = mgs_env_info(env);
 472         int rc;
 473         struct lustre_cfg *lcfg;
 474         ENTRY;
 475
 476         msp = req_capsule_client_get(&req->rq_pill, &RMF_MGS_SEND_PARAM);
 477         LASSERT(msp);
 478
 479         /* Construct lustre_cfg structure to pass to function mgs_setparam */
 480         lustre_cfg_bufs_reset(&mgi->mgi_bufs, NULL);
 481         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, msp->mgs_param);
 482         lcfg = lustre_cfg_new(LCFG_PARAM, &mgi->mgi_bufs);
 483         if (IS_ERR(lcfg))
 484                 GOTO(out, rc = PTR_ERR(lcfg));
 485         rc = mgs_setparam(env, mgs, lcfg, mgi->mgi_fsname);
 486         if (rc) {
 487                 CERROR("Error %d in setting the parameter %s for fs %s\n",
 488                        rc, msp->mgs_param, mgi->mgi_fsname);
 489                 GOTO(out_cfg, rc);
 490         }
 491
 492         rc = req_capsule_server_pack(&req->rq_pill);
 493         if (rc == 0) {
 494                 rep_msp = req_capsule_server_get(&req->rq_pill, &RMF_MGS_SEND_PARAM);
 495                 rep_msp = msp;
 496         }
 497 out_cfg:
 498         lustre_cfg_free(lcfg);
 499 out:
 500         RETURN(rc);
 501 }
 502
 503 static int mgs_config_read(struct ptlrpc_request *req)
 504 {
 505         struct mgs_config_body *body;
 506         int rc;
 507         ENTRY;
 508
 509         body = req_capsule_client_get(&req->rq_pill, &RMF_MGS_CONFIG_BODY);
 510         if (body == NULL)
 511                 RETURN(-EINVAL);
 512
 513         switch (body->mcb_type) {
 514         case CONFIG_T_RECOVER:
 515                 rc = mgs_get_ir_logs(req);
 516                 break;
 517
 518         case CONFIG_T_CONFIG:
 519                 rc = -ENOTSUPP;
 520                 break;
 521
 522         default:
 523                 rc = -EINVAL;
 524                 break;
 525         }
 526
 527         RETURN(rc);
 528 }
 529
 530 /*
 531  * similar as in ost_connect_check_sptlrpc()
 532  */
 533 static int mgs_connect_check_sptlrpc(struct ptlrpc_request *req)
 534 {
 535         struct obd_export     *exp = req->rq_export;
 536         struct mgs_device     *mgs = exp2mgs_dev(req->rq_export);
 537         struct lu_env         *env = req->rq_svc_thread->t_env;
 538         struct fs_db          *fsdb;
 539         struct sptlrpc_flavor  flvr;
 540         int                    rc = 0;
 541
 542         if (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_INVALID) {
 543                 rc = mgs_find_or_make_fsdb(env, mgs, MGSSELF_NAME, &fsdb);
 544                 if (rc)
 545                         return rc;
 546
 547                 cfs_mutex_lock(&fsdb->fsdb_mutex);
 548                 if (sptlrpc_rule_set_choose(&fsdb->fsdb_srpc_gen,
 549                                             LUSTRE_SP_MGC, LUSTRE_SP_MGS,
 550                                             req->rq_peer.nid,
 551                                             &flvr) == 0) {
 552                         /* by defualt allow any flavors */
 553                         flvr.sf_rpc = SPTLRPC_FLVR_ANY;
 554                 }
 555                 cfs_mutex_unlock(&fsdb->fsdb_mutex);
 556
 557                 cfs_spin_lock(&exp->exp_lock);
 558
 559                 exp->exp_sp_peer = req->rq_sp_from;
 560                 exp->exp_flvr = flvr;
 561
 562                 if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_ANY &&
 563                     exp->exp_flvr.sf_rpc != req->rq_flvr.sf_rpc) {
 564                         CERROR("invalid rpc flavor %x, expect %x, from %s\n",
 565                                req->rq_flvr.sf_rpc, exp->exp_flvr.sf_rpc,
 566                                libcfs_nid2str(req->rq_peer.nid));
 567                         rc = -EACCES;
 568                 }
 569
 570                 cfs_spin_unlock(&exp->exp_lock);
 571         } else {
 572                 if (exp->exp_sp_peer != req->rq_sp_from) {
 573                         CERROR("RPC source %s doesn't match %s\n",
 574                                sptlrpc_part2name(req->rq_sp_from),
 575                                sptlrpc_part2name(exp->exp_sp_peer));
 576                         rc = -EACCES;
 577                 } else {
 578                         rc = sptlrpc_target_export_check(exp, req);
 579                 }
 580         }
 581
 582         return rc;
 583 }
 584
 585 /* Called whenever a target cleans up. */
 586 /* XXX - Currently unused */
 587 static int mgs_handle_target_del(struct ptlrpc_request *req)
 588 {
 589         ENTRY;
 590         mgs_counter_incr(req->rq_export, LPROC_MGS_TARGET_DEL);
 591         RETURN(0);
 592 }
 593
 594 /* XXX - Currently unused */
 595 static int mgs_handle_exception(struct ptlrpc_request *req)
 596 {
 597         ENTRY;
 598         mgs_counter_incr(req->rq_export, LPROC_MGS_EXCEPTION);
 599         RETURN(0);
 600 }
 601
 602 /*
 603  * For old clients there is no direct way of knowing which filesystems
 604  * a client is operating at the MGS side. But we need to pick up those
 605  * clients so that the MGS can mark the corresponding filesystem as
 606  * non-IR capable because old clients are not ready to be notified.
 607  *
 608  * This is why we have this _hack_ function. We detect the filesystem's
 609  * name by hacking llog operation which is currently used by the clients
 610  * to fetch configuration logs. At present this is fine because this is
 611  * the ONLY llog operation between mgc and the MGS.
 612  *
 613  * If extra llog operation is going to be added, this function needs fixing.
 614  *
 615  * If releases prior than 2.0 are not supported, we can remove this function.
 616  */
 617 static int mgs_handle_fslog_hack(struct ptlrpc_request *req)
 618 {
 619         char *logname;
 620         char fsname[16];
 621         char *ptr;
 622         int rc;
 623
 624         /* XXX: We suppose that llog at mgs is only used for
 625          * fetching file system log */
 626         logname = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
 627         if (logname == NULL) {
 628                 CERROR("No logname, is llog on MGS used for something else?\n");
 629                 return -EINVAL;
 630         }
 631
 632         ptr = strchr(logname, '-');
 633         rc = (int)(ptr - logname);
 634         if (ptr == NULL || rc >= sizeof(fsname)) {
 635                 CERROR("Invalid logname received: %s\n", logname);
 636                 return -EINVAL;
 637         }
 638
 639         strncpy(fsname, logname, rc);
 640         fsname[rc] = 0;
 641         rc = mgs_fsc_attach(req->rq_svc_thread->t_env, req->rq_export, fsname);
 642         if (rc < 0 && rc != -EEXIST)
 643                 CERROR("add fs client %s returns %d\n", fsname, rc);
 644
 645         return rc;
 646 }
 647
 648 /* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
 649 int mgs_handle(struct ptlrpc_request *req)
 650 {
 651         int fail = OBD_FAIL_MGS_ALL_REPLY_NET;
 652         int opc, rc = 0;
 653         ENTRY;
 654
 655         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
 656         CFS_FAIL_TIMEOUT_MS(OBD_FAIL_MGS_PAUSE_REQ, cfs_fail_val);
 657         if (CFS_FAIL_CHECK(OBD_FAIL_MGS_ALL_REQUEST_NET))
 658                 RETURN(0);
 659
 660         LASSERT(current->journal_info == NULL);
 661         opc = lustre_msg_get_opc(req->rq_reqmsg);
 662
 663         if (opc == SEC_CTX_INIT ||
 664             opc == SEC_CTX_INIT_CONT ||
 665             opc == SEC_CTX_FINI)
 666                 GOTO(out, rc = 0);
 667
 668         if (opc != MGS_CONNECT) {
 669                 if (!class_connected_export(req->rq_export)) {
 670                         DEBUG_REQ(D_MGS, req, "operation on unconnected MGS\n");
 671                         req->rq_status = -ENOTCONN;
 672                         GOTO(out, rc = -ENOTCONN);
 673                 }
 674         }
 675
 676         switch (opc) {
 677         case MGS_CONNECT:
 678                 DEBUG_REQ(D_MGS, req, "connect");
 679                 /* MGS and MDS have same request format for connect */
 680                 req_capsule_set(&req->rq_pill, &RQF_MDS_CONNECT);
 681                 rc = target_handle_connect(req);
 682                 if (rc == 0)
 683                         rc = mgs_connect_check_sptlrpc(req);
 684
 685                 if (!rc && (lustre_msg_get_conn_cnt(req->rq_reqmsg) > 1))
 686                         /* Make clients trying to reconnect after a MGS restart
 687                            happy; also requires obd_replayable */
 688                         lustre_msg_add_op_flags(req->rq_repmsg,
 689                                                 MSG_CONNECT_RECONNECT);
 690                 break;
 691         case MGS_DISCONNECT:
 692                 DEBUG_REQ(D_MGS, req, "disconnect");
 693                 /* MGS and MDS have same request format for disconnect */
 694                 req_capsule_set(&req->rq_pill, &RQF_MDS_DISCONNECT);
 695                 rc = target_handle_disconnect(req);
 696                 req->rq_status = rc;            /* superfluous? */
 697                 break;
 698         case MGS_EXCEPTION:
 699                 DEBUG_REQ(D_MGS, req, "exception");
 700                 rc = mgs_handle_exception(req);
 701                 break;
 702         case MGS_TARGET_REG:
 703                 DEBUG_REQ(D_MGS, req, "target add");
 704                 req_capsule_set(&req->rq_pill, &RQF_MGS_TARGET_REG);
 705                 rc = mgs_handle_target_reg(req);
 706                 break;
 707         case MGS_TARGET_DEL:
 708                 DEBUG_REQ(D_MGS, req, "target del");
 709                 rc = mgs_handle_target_del(req);
 710                 break;
 711         case MGS_SET_INFO:
 712                 DEBUG_REQ(D_MGS, req, "set_info");
 713                 req_capsule_set(&req->rq_pill, &RQF_MGS_SET_INFO);
 714                 rc = mgs_set_info_rpc(req);
 715                 break;
 716         case MGS_CONFIG_READ:
 717                 DEBUG_REQ(D_MGS, req, "read config");
 718                 req_capsule_set(&req->rq_pill, &RQF_MGS_CONFIG_READ);
 719                 rc = mgs_config_read(req);
 720                 break;
 721         case LDLM_ENQUEUE:
 722                 DEBUG_REQ(D_MGS, req, "enqueue");
 723                 req_capsule_set(&req->rq_pill, &RQF_LDLM_ENQUEUE);
 724                 rc = ldlm_handle_enqueue(req, ldlm_server_completion_ast,
 725                                          ldlm_server_blocking_ast, NULL);
 726                 break;
 727         case LDLM_BL_CALLBACK:
 728         case LDLM_CP_CALLBACK:
 729                 DEBUG_REQ(D_MGS, req, "callback");
 730                 CERROR("callbacks should not happen on MGS\n");
 731                 LBUG();
 732                 break;
 733
 734         case OBD_PING:
 735                 DEBUG_REQ(D_INFO, req, "ping");
 736                 req_capsule_set(&req->rq_pill, &RQF_OBD_PING);
 737                 rc = target_handle_ping(req);
 738                 break;
 739         case OBD_LOG_CANCEL:
 740                 DEBUG_REQ(D_MGS, req, "log cancel");
 741                 rc = -ENOTSUPP; /* la la la */
 742                 break;
 743
 744         case LLOG_ORIGIN_HANDLE_CREATE:
 745                 DEBUG_REQ(D_MGS, req, "llog_open");
 746                 req_capsule_set(&req->rq_pill, &RQF_LLOG_ORIGIN_HANDLE_CREATE);
 747                 rc = llog_origin_handle_open(req);
 748                 if (rc == 0)
 749                         (void)mgs_handle_fslog_hack(req);
 750                 break;
 751         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
 752                 DEBUG_REQ(D_MGS, req, "llog next block");
 753                 req_capsule_set(&req->rq_pill,
 754                                 &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
 755                 rc = llog_origin_handle_next_block(req);
 756                 break;
 757         case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
 758                 DEBUG_REQ(D_MGS, req, "llog prev block");
 759                 req_capsule_set(&req->rq_pill,
 760                                 &RQF_LLOG_ORIGIN_HANDLE_PREV_BLOCK);
 761                 rc = llog_origin_handle_prev_block(req);
 762                 break;
 763         case LLOG_ORIGIN_HANDLE_READ_HEADER:
 764                 DEBUG_REQ(D_MGS, req, "llog read header");
 765                 req_capsule_set(&req->rq_pill,
 766                                 &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER);
 767                 rc = llog_origin_handle_read_header(req);
 768                 break;
 769         case LLOG_ORIGIN_HANDLE_CLOSE:
 770                 DEBUG_REQ(D_MGS, req, "llog close");
 771                 rc = llog_origin_handle_close(req);
 772                 break;
 773         default:
 774                 rc = -EOPNOTSUPP;
 775         }
 776
 777         LASSERT(current->journal_info == NULL);
 778         if (rc) {
 779                 DEBUG_REQ(D_MGS, req, "MGS fail to handle opc = %d: rc = %d\n",
 780                           opc, rc);
 781                 req->rq_status = rc;
 782                 rc = ptlrpc_error(req);
 783                 RETURN(rc);
 784         }
 785 out:
 786         target_send_reply(req, rc, fail);
 787         RETURN(0);
 788 }
 789
 790 static inline int mgs_init_export(struct obd_export *exp)
 791 {
 792         struct mgs_export_data *data = &exp->u.eu_mgs_data;
 793
 794         /* init mgs_export_data for fsc */
 795         cfs_spin_lock_init(&data->med_lock);
 796         CFS_INIT_LIST_HEAD(&data->med_clients);
 797
 798         cfs_spin_lock(&exp->exp_lock);
 799         exp->exp_connecting = 1;
 800         cfs_spin_unlock(&exp->exp_lock);
 801
 802         /* self-export doesn't need client data and ldlm initialization */
 803         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
 804                                      &exp->exp_client_uuid)))
 805                 return 0;
 806         return ldlm_init_export(exp);
 807 }
 808
 809 static inline int mgs_destroy_export(struct obd_export *exp)
 810 {
 811         ENTRY;
 812
 813         target_destroy_export(exp);
 814         mgs_client_free(exp);
 815
 816         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
 817                                      &exp->exp_client_uuid)))
 818                 RETURN(0);
 819
 820         ldlm_destroy_export(exp);
 821
 822         RETURN(0);
 823 }
 824
 825 static int mgs_extract_fs_pool(char * arg, char *fsname, char *poolname)
 826 {
 827         char *ptr;
 828
 829         ENTRY;
 830         for (ptr = arg;  (*ptr != '\0') && (*ptr != '.'); ptr++ ) {
 831                 *fsname = *ptr;
 832                 fsname++;
 833         }
 834         if (*ptr == '\0')
 835                 return -EINVAL;
 836         *fsname = '\0';
 837         ptr++;
 838         strcpy(poolname, ptr);
 839
 840         RETURN(0);
 841 }
 842
 843 static int mgs_iocontrol_pool(const struct lu_env *env,
 844                               struct mgs_device *mgs,
 845                               struct obd_ioctl_data *data)
 846 {
 847         struct mgs_thread_info *mgi = mgs_env_info(env);
 848         int rc;
 849         struct lustre_cfg *lcfg = NULL;
 850         struct llog_rec_hdr rec;
 851         char *poolname = NULL;
 852         ENTRY;
 853
 854         OBD_ALLOC(poolname, LOV_MAXPOOLNAME + 1);
 855         if (poolname == NULL)
 856                 RETURN(-ENOMEM);
 857         rec.lrh_len = llog_data_len(data->ioc_plen1);
 858
 859         if (data->ioc_type == LUSTRE_CFG_TYPE) {
 860                 rec.lrh_type = OBD_CFG_REC;
 861         } else {
 862                 CERROR("unknown cfg record type:%d \n", data->ioc_type);
 863                 GOTO(out_pool, rc = -EINVAL);
 864         }
 865
 866         if (data->ioc_plen1 > CFS_PAGE_SIZE)
 867                 GOTO(out_pool, rc = -E2BIG);
 868
 869         OBD_ALLOC(lcfg, data->ioc_plen1);
 870         if (lcfg == NULL)
 871                 GOTO(out_pool, rc = -ENOMEM);
 872
 873         if (cfs_copy_from_user(lcfg, data->ioc_pbuf1, data->ioc_plen1))
 874                 GOTO(out_lcfg, rc = -EFAULT);
 875
 876         if (lcfg->lcfg_bufcount < 2)
 877                 GOTO(out_lcfg, rc = -EFAULT);
 878
 879         /* first arg is always <fsname>.<poolname> */
 880         rc = mgs_extract_fs_pool(lustre_cfg_string(lcfg, 1), mgi->mgi_fsname,
 881                                  poolname);
 882         if (rc)
 883                 GOTO(out_lcfg, rc);
 884
 885         switch (lcfg->lcfg_command) {
 886         case LCFG_POOL_NEW:
 887                 if (lcfg->lcfg_bufcount != 2)
 888                         GOTO(out_lcfg, rc = -EINVAL);
 889                 rc = mgs_pool_cmd(env, mgs, LCFG_POOL_NEW, mgi->mgi_fsname,
 890                                   poolname, NULL);
 891                 break;
 892         case LCFG_POOL_ADD:
 893                 if (lcfg->lcfg_bufcount != 3)
 894                         GOTO(out_lcfg, rc = -EINVAL);
 895                 rc = mgs_pool_cmd(env, mgs, LCFG_POOL_ADD, mgi->mgi_fsname,
 896                                   poolname, lustre_cfg_string(lcfg, 2));
 897                 break;
 898         case LCFG_POOL_REM:
 899                 if (lcfg->lcfg_bufcount != 3)
 900                         GOTO(out_lcfg, rc = -EINVAL);
 901                 rc = mgs_pool_cmd(env, mgs, LCFG_POOL_REM, mgi->mgi_fsname,
 902                                   poolname, lustre_cfg_string(lcfg, 2));
 903                 break;
 904         case LCFG_POOL_DEL:
 905                 if (lcfg->lcfg_bufcount != 2)
 906                         GOTO(out_lcfg, rc = -EINVAL);
 907                 rc = mgs_pool_cmd(env, mgs, LCFG_POOL_DEL, mgi->mgi_fsname,
 908                                   poolname, NULL);
 909                 break;
 910         default:
 911                  rc = -EINVAL;
 912         }
 913
 914         if (rc) {
 915                 CERROR("OBD_IOC_POOL err %d, cmd %X for pool %s.%s\n",
 916                        rc, lcfg->lcfg_command, mgi->mgi_fsname, poolname);
 917                 GOTO(out_lcfg, rc);
 918         }
 919
 920 out_lcfg:
 921         OBD_FREE(lcfg, data->ioc_plen1);
 922 out_pool:
 923         OBD_FREE(poolname, LOV_MAXPOOLNAME + 1);
 924         RETURN(rc);
 925 }
 926
 927 /* from mdt_iocontrol */
 928 int mgs_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
 929                   void *karg, void *uarg)
 930 {
 931         struct mgs_device *mgs = exp2mgs_dev(exp);
 932         struct obd_ioctl_data *data = karg;
 933         struct lvfs_run_ctxt saved;
 934         struct lu_env env;
 935         int rc = 0;
 936
 937         ENTRY;
 938         CDEBUG(D_IOCTL, "handling ioctl cmd %#x\n", cmd);
 939
 940         rc = lu_env_init(&env, LCT_MG_THREAD);
 941         if (rc)
 942                 RETURN(rc);
 943
 944         switch (cmd) {
 945
 946         case OBD_IOC_PARAM: {
 947                 struct mgs_thread_info *mgi = mgs_env_info(&env);
 948                 struct lustre_cfg *lcfg;
 949                 struct llog_rec_hdr rec;
 950
 951                 rec.lrh_len = llog_data_len(data->ioc_plen1);
 952
 953                 if (data->ioc_type == LUSTRE_CFG_TYPE) {
 954                         rec.lrh_type = OBD_CFG_REC;
 955                 } else {
 956                         CERROR("unknown cfg record type:%d \n", data->ioc_type);
 957                         GOTO(out, rc = -EINVAL);
 958                 }
 959
 960                 OBD_ALLOC(lcfg, data->ioc_plen1);
 961                 if (lcfg == NULL)
 962                         GOTO(out, rc = -ENOMEM);
 963                 if (cfs_copy_from_user(lcfg, data->ioc_pbuf1, data->ioc_plen1))
 964                         GOTO(out_free, rc = -EFAULT);
 965
 966                 if (lcfg->lcfg_bufcount < 1)
 967                         GOTO(out_free, rc = -EINVAL);
 968
 969                 rc = mgs_setparam(&env, mgs, lcfg, mgi->mgi_fsname);
 970                 if (rc)
 971                         CERROR("%s: setparam err: rc = %d\n",
 972                                exp->exp_obd->obd_name, rc);
 973 out_free:
 974                 OBD_FREE(lcfg, data->ioc_plen1);
 975                 break;
 976         }
 977
 978         case OBD_IOC_POOL:
 979                 rc = mgs_iocontrol_pool(&env, mgs, data);
 980                 break;
 981
 982         case OBD_IOC_DUMP_LOG: {
 983                 struct llog_ctxt *ctxt;
 984                 ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
 985                 push_ctxt(&saved, &mgs->mgs_obd->obd_lvfs_ctxt, NULL);
 986                 rc = class_config_dump_llog(ctxt, data->ioc_inlbuf1, NULL);
 987                 pop_ctxt(&saved, &mgs->mgs_obd->obd_lvfs_ctxt, NULL);
 988                 llog_ctxt_put(ctxt);
 989
 990                 break;
 991         }
 992
 993         case OBD_IOC_LLOG_CHECK:
 994         case OBD_IOC_LLOG_INFO:
 995         case OBD_IOC_LLOG_PRINT: {
 996                 struct llog_ctxt *ctxt;
 997                 ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
 998
 999                 push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
1000                 rc = llog_ioctl(ctxt, cmd, data);
1001                 pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
1002                 llog_ctxt_put(ctxt);
1003
1004                 break;
1005         }
1006
1007         default:
1008                 CDEBUG(D_INFO, "unknown command %x\n", cmd);
1009                 rc = -EINVAL;
1010                 break;
1011         }
1012 out:
1013         lu_env_fini(&env);
1014         RETURN(rc);
1015 }
1016
1017 static int mgs_connect_to_osd(struct mgs_device *m, const char *nextdev)
1018 {
1019         struct obd_connect_data *data = NULL;
1020         struct obd_device       *obd;
1021         int                      rc;
1022         ENTRY;
1023
1024         OBD_ALLOC_PTR(data);
1025         if (data == NULL)
1026                 RETURN(-ENOMEM);
1027
1028         obd = class_name2obd(nextdev);
1029         if (obd == NULL) {
1030                 CERROR("can't locate next device: %s\n", nextdev);
1031                 GOTO(out, rc = -ENOTCONN);
1032         }
1033
1034         data->ocd_version = LUSTRE_VERSION_CODE;
1035
1036         rc = obd_connect(NULL, &m->mgs_bottom_exp, obd,
1037                          &obd->obd_uuid, data, NULL);
1038         if (rc) {
1039                 CERROR("cannot connect to next dev %s (%d)\n", nextdev, rc);
1040                 GOTO(out, rc);
1041         }
1042
1043         m->mgs_bottom = lu2dt_dev(m->mgs_bottom_exp->exp_obd->obd_lu_dev);
1044         m->mgs_dt_dev.dd_lu_dev.ld_site = m->mgs_bottom->dd_lu_dev.ld_site;
1045         LASSERT(m->mgs_dt_dev.dd_lu_dev.ld_site);
1046 out:
1047         OBD_FREE_PTR(data);
1048         RETURN(rc);
1049 }
1050
1051
1052 static int mgs_init0(const struct lu_env *env, struct mgs_device *mgs,
1053                      struct lu_device_type *ldt, struct lustre_cfg *lcfg)
1054 {
1055         static struct ptlrpc_service_conf       conf;
1056         struct lprocfs_static_vars  lvars = { 0 };
1057         struct obd_device          *obd;
1058         struct lustre_mount_info   *lmi;
1059         int                         rc;
1060         ENTRY;
1061
1062         lmi = server_get_mount(lustre_cfg_string(lcfg, 0));
1063         if (lmi == NULL)
1064                 RETURN(-ENODEV);
1065
1066         mgs->mgs_dt_dev.dd_lu_dev.ld_ops = &mgs_lu_ops;
1067
1068         rc = mgs_connect_to_osd(mgs, lustre_cfg_string(lcfg, 3));
1069         if (rc)
1070                 RETURN(rc);
1071
1072         obd = class_name2obd(lustre_cfg_string(lcfg, 0));
1073         LASSERT(obd);
1074         mgs->mgs_obd = obd;
1075         mgs->mgs_obd->obd_lu_dev = &mgs->mgs_dt_dev.dd_lu_dev;
1076
1077         obd->u.obt.obt_magic = OBT_MAGIC;
1078         obd->u.obt.obt_instance = 0;
1079
1080         /* namespace for mgs llog */
1081         obd->obd_namespace = ldlm_namespace_new(obd ,"MGS",
1082                                                 LDLM_NAMESPACE_SERVER,
1083                                                 LDLM_NAMESPACE_MODEST,
1084                                                 LDLM_NS_TYPE_MGT);
1085         if (obd->obd_namespace == NULL)
1086                 GOTO(err_ops, rc = -ENOMEM);
1087
1088         /* ldlm setup */
1089         ptlrpc_init_client(LDLM_CB_REQUEST_PORTAL, LDLM_CB_REPLY_PORTAL,
1090                            "mgs_ldlm_client", &obd->obd_ldlm_client);
1091
1092         rc = mgs_fs_setup(env, mgs);
1093         if (rc) {
1094                 CERROR("%s: MGS filesystem method init failed: rc = %d\n",
1095                        obd->obd_name, rc);
1096                 GOTO(err_ns, rc);
1097         }
1098
1099         rc = obd_llog_init(obd, &obd->obd_olg, obd, NULL);
1100         if (rc)
1101                 GOTO(err_fs, rc);
1102
1103         /* No recovery for MGC's */
1104         obd->obd_replayable = 0;
1105
1106         /* Internal mgs setup */
1107         mgs_init_fsdb_list(mgs);
1108         cfs_mutex_init(&mgs->mgs_mutex);
1109         mgs->mgs_start_time = cfs_time_current_sec();
1110         cfs_spin_lock_init(&mgs->mgs_lock);
1111
1112         /* Setup proc */
1113         lprocfs_mgs_init_vars(&lvars);
1114         if (lprocfs_obd_setup(obd, lvars.obd_vars) == 0) {
1115                 lproc_mgs_setup(mgs);
1116                 rc = lprocfs_alloc_md_stats(obd, LPROC_MGS_LAST);
1117                 if (rc)
1118                         GOTO(err_llog, rc);
1119         }
1120
1121         conf = (typeof(conf)) {
1122                 .psc_name               = LUSTRE_MGS_NAME,
1123                 .psc_watchdog_factor    = MGS_SERVICE_WATCHDOG_FACTOR,
1124                 .psc_buf                = {
1125                         .bc_nbufs               = MGS_NBUFS,
1126                         .bc_buf_size            = MGS_BUFSIZE,
1127                         .bc_req_max_size        = MGS_MAXREQSIZE,
1128                         .bc_rep_max_size        = MGS_MAXREPSIZE,
1129                         .bc_req_portal          = MGS_REQUEST_PORTAL,
1130                         .bc_rep_portal          = MGC_REPLY_PORTAL,
1131                 },
1132                 .psc_thr                = {
1133                         .tc_thr_name            = "ll_mgs",
1134                         .tc_nthrs_init          = MGS_NTHRS_INIT,
1135                         .tc_nthrs_max           = MGS_NTHRS_MAX,
1136                         .tc_ctx_tags            = LCT_MG_THREAD,
1137                 },
1138                 .psc_ops                = {
1139                         .so_req_handler         = mgs_handle,
1140                         .so_req_printer         = target_print_req,
1141                 },
1142         };
1143         /* Start the service threads */
1144         mgs->mgs_service = ptlrpc_register_service(&conf, obd->obd_proc_entry);
1145         if (IS_ERR(mgs->mgs_service)) {
1146                 rc = PTR_ERR(mgs->mgs_service);
1147                 CERROR("failed to start service: %d\n", rc);
1148                 GOTO(err_lproc, rc);
1149         }
1150
1151         ping_evictor_start();
1152
1153         CDEBUG(D_INFO, "MGS %s started\n", obd->obd_name);
1154
1155         /* device stack is not yet fully setup to keep no objects behind */
1156         lu_site_purge(env, mgs2lu_dev(mgs)->ld_site, ~0);
1157         RETURN(0);
1158
1159 err_lproc:
1160         lproc_mgs_cleanup(mgs);
1161 err_llog:
1162 err_fs:
1163         /* No extra cleanup needed for llog_init_commit_thread() */
1164         mgs_fs_cleanup(env, mgs);
1165 err_ns:
1166         ldlm_namespace_free(obd->obd_namespace, NULL, 0);
1167         obd->obd_namespace = NULL;
1168 err_ops:
1169         lu_site_purge(env, mgs2lu_dev(mgs)->ld_site, ~0);
1170         if (!cfs_hash_is_empty(mgs2lu_dev(mgs)->ld_site->ls_obj_hash)) {
1171                 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
1172                 lu_site_print(env, mgs2lu_dev(mgs)->ld_site, &msgdata,
1173                                 lu_cdebug_printer);
1174         }
1175         obd_disconnect(mgs->mgs_bottom_exp);
1176         RETURN(rc);
1177 }
1178
1179 static struct lu_device *mgs_device_free(const struct lu_env *env,
1180                                          struct lu_device *lu)
1181 {
1182         struct mgs_device *mgs = lu2mgs_dev(lu);
1183         ENTRY;
1184
1185         dt_device_fini(&mgs->mgs_dt_dev);
1186         OBD_FREE_PTR(mgs);
1187         RETURN(NULL);
1188 }
1189
1190 static int mgs_process_config(const struct lu_env *env,
1191                               struct lu_device *dev,
1192                               struct lustre_cfg *lcfg)
1193 {
1194         LBUG();
1195         return 0;
1196 }
1197
1198 static int mgs_object_init(const struct lu_env *env, struct lu_object *o,
1199                            const struct lu_object_conf *unused)
1200 {
1201         struct mgs_device *d = lu2mgs_dev(o->lo_dev);
1202         struct lu_device  *under;
1203         struct lu_object  *below;
1204         int                rc = 0;
1205         ENTRY;
1206
1207         /* do no set .do_ops as mgs calls to bottom osd directly */
1208
1209         CDEBUG(D_INFO, "object init, fid = "DFID"\n",
1210                         PFID(lu_object_fid(o)));
1211
1212         under = &d->mgs_bottom->dd_lu_dev;
1213         below = under->ld_ops->ldo_object_alloc(env, o->lo_header, under);
1214         if (below != NULL)
1215                 lu_object_add(o, below);
1216         else
1217                 rc = -ENOMEM;
1218
1219         return 0;
1220 }
1221
1222 static void mgs_object_free(const struct lu_env *env, struct lu_object *o)
1223 {
1224         struct mgs_object *obj = lu2mgs_obj(o);
1225         struct lu_object_header *h = o->lo_header;
1226
1227         dt_object_fini(&obj->mgo_obj);
1228         lu_object_header_fini(h);
1229         OBD_FREE_PTR(obj);
1230 }
1231
1232 static int mgs_object_print(const struct lu_env *env, void *cookie,
1233                             lu_printer_t p, const struct lu_object *l)
1234 {
1235         const struct mgs_object *o = lu2mgs_obj((struct lu_object *) l);
1236
1237         return (*p)(env, cookie, LUSTRE_MGS_NAME"-object@%p", o);
1238 }
1239
1240 struct lu_object_operations mgs_lu_obj_ops = {
1241         .loo_object_init        = mgs_object_init,
1242         .loo_object_free        = mgs_object_free,
1243         .loo_object_print       = mgs_object_print,
1244 };
1245
1246 struct lu_object *mgs_object_alloc(const struct lu_env *env,
1247                                    const struct lu_object_header *hdr,
1248                                    struct lu_device *d)
1249 {
1250         struct lu_object_header *h;
1251         struct mgs_object       *o;
1252         struct lu_object        *l;
1253
1254         LASSERT(hdr == NULL);
1255
1256         OBD_ALLOC_PTR(o);
1257         if (o != NULL) {
1258                 l = &o->mgo_obj.do_lu;
1259                 h = &o->mgo_header;
1260
1261                 lu_object_header_init(h);
1262                 dt_object_init(&o->mgo_obj, h, d);
1263                 lu_object_add_top(h, l);
1264
1265                 l->lo_ops = &mgs_lu_obj_ops;
1266
1267                 return l;
1268         } else {
1269                 return NULL;
1270         }
1271 }
1272
1273 const struct lu_device_operations mgs_lu_ops = {
1274         .ldo_object_alloc       = mgs_object_alloc,
1275         .ldo_process_config     = mgs_process_config,
1276 };
1277
1278 static struct lu_device *mgs_device_alloc(const struct lu_env *env,
1279                                           struct lu_device_type *type,
1280                                           struct lustre_cfg *lcfg)
1281 {
1282         struct mgs_device *mgs;
1283         struct lu_device  *ludev;
1284
1285         OBD_ALLOC_PTR(mgs);
1286         if (mgs == NULL) {
1287                 ludev = ERR_PTR(-ENOMEM);
1288         } else {
1289                 int rc;
1290
1291                 ludev = mgs2lu_dev(mgs);
1292                 dt_device_init(&mgs->mgs_dt_dev, type);
1293                 rc = mgs_init0(env, mgs, type, lcfg);
1294                 if (rc != 0) {
1295                         mgs_device_free(env, ludev);
1296                         ludev = ERR_PTR(rc);
1297                 }
1298         }
1299         return ludev;
1300 }
1301
1302 static struct lu_device *mgs_device_fini(const struct lu_env *env,
1303                                          struct lu_device *d)
1304 {
1305         struct mgs_device *mgs = lu2mgs_dev(d);
1306         struct obd_device *obd = mgs->mgs_obd;
1307         ENTRY;
1308
1309         LASSERT(mgs->mgs_bottom);
1310
1311         ping_evictor_stop();
1312
1313         ptlrpc_unregister_service(mgs->mgs_service);
1314
1315         obd_exports_barrier(obd);
1316         obd_zombie_barrier();
1317
1318         mgs_cleanup_fsdb_list(mgs);
1319         lproc_mgs_cleanup(mgs);
1320
1321         obd_llog_finish(obd, 0);
1322
1323         mgs_fs_cleanup(env, mgs);
1324
1325         ldlm_namespace_free(obd->obd_namespace, NULL, 1);
1326         obd->obd_namespace = NULL;
1327
1328         lu_site_purge(env, d->ld_site, ~0);
1329         if (!cfs_hash_is_empty(d->ld_site->ls_obj_hash)) {
1330                 LIBCFS_DEBUG_MSG_DATA_DECL(msgdata, D_ERROR, NULL);
1331                 lu_site_print(env, d->ld_site, &msgdata, lu_cdebug_printer);
1332         }
1333
1334         LASSERT(mgs->mgs_bottom_exp);
1335         obd_disconnect(mgs->mgs_bottom_exp);
1336
1337         server_put_mount(obd->obd_name, NULL);
1338
1339         RETURN(NULL);
1340 }
1341
1342 /* context key constructor/destructor: mgs_key_init, mgs_key_fini */
1343 LU_KEY_INIT_FINI(mgs, struct mgs_thread_info);
1344
1345 LU_TYPE_INIT_FINI(mgs, &mgs_thread_key);
1346
1347 LU_CONTEXT_KEY_DEFINE(mgs, LCT_MG_THREAD);
1348
1349 static struct lu_device_type_operations mgs_device_type_ops = {
1350         .ldto_init              = mgs_type_init,
1351         .ldto_fini              = mgs_type_fini,
1352
1353         .ldto_start             = mgs_type_start,
1354         .ldto_stop              = mgs_type_stop,
1355
1356         .ldto_device_alloc      = mgs_device_alloc,
1357         .ldto_device_free       = mgs_device_free,
1358
1359         .ldto_device_fini       = mgs_device_fini
1360 };
1361
1362 static struct lu_device_type mgs_device_type = {
1363         .ldt_tags       = LU_DEVICE_DT,
1364         .ldt_name       = LUSTRE_MGS_NAME,
1365         .ldt_ops        = &mgs_device_type_ops,
1366         .ldt_ctx_tags   = LCT_MG_THREAD
1367 };
1368
1369
1370 /* use obd ops to offer management infrastructure */
1371 static struct obd_ops mgs_obd_ops = {
1372         .o_owner           = THIS_MODULE,
1373         .o_connect         = mgs_connect,
1374         .o_reconnect       = mgs_reconnect,
1375         .o_disconnect      = mgs_disconnect,
1376         .o_init_export     = mgs_init_export,
1377         .o_destroy_export  = mgs_destroy_export,
1378         .o_iocontrol       = mgs_iocontrol,
1379         .o_llog_init       = mgs_llog_init,
1380         .o_llog_finish     = mgs_llog_finish
1381 };
1382
1383 static int __init mgs_init(void)
1384 {
1385         struct lprocfs_static_vars lvars;
1386
1387         lprocfs_mgs_init_vars(&lvars);
1388         class_register_type(&mgs_obd_ops, NULL, lvars.module_vars,
1389                             LUSTRE_MGS_NAME, &mgs_device_type);
1390
1391         return 0;
1392 }
1393
1394 static void /*__exit*/ mgs_exit(void)
1395 {
1396         class_unregister_type(LUSTRE_MGS_NAME);
1397 }
1398
1399 MODULE_AUTHOR("Sun Microsystems, Inc. <http://www.lustre.org/>");
1400 MODULE_DESCRIPTION("Lustre  Management Server (MGS)");
1401 MODULE_LICENSE("GPL");
1402
1403 module_init(mgs_init);
1404 module_exit(mgs_exit);