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
 152         ENTRY;
 153
 154         LASSERT(olg == &obd->obd_olg);
 155         rc = llog_setup(NULL, obd, olg, LLOG_CONFIG_ORIG_CTXT, obd,
 156                         &llog_lvfs_ops);
 157         RETURN(rc);
 158 }
 159
 160 static int mgs_llog_finish(struct obd_device *obd, int count)
 161 {
 162         struct llog_ctxt *ctxt;
 163
 164         ENTRY;
 165
 166         ctxt = llog_get_context(obd, LLOG_CONFIG_ORIG_CTXT);
 167         if (ctxt)
 168                 llog_cleanup(NULL, ctxt);
 169
 170         RETURN(0);
 171 }
 172
 173 static int mgs_completion_ast_config(struct ldlm_lock *lock, int flags,
 174                                      void *cbdata)
 175 {
 176         ENTRY;
 177
 178         if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
 179                        LDLM_FL_BLOCK_CONV))) {
 180                 struct fs_db *fsdb = (struct fs_db *)lock->l_ast_data;
 181                 struct lustre_handle lockh;
 182
 183                 /* clear the bit before lock put */
 184                 cfs_clear_bit(FSDB_REVOKING_LOCK, &fsdb->fsdb_flags);
 185
 186                 ldlm_lock2handle(lock, &lockh);
 187                 ldlm_lock_decref_and_cancel(&lockh, LCK_EX);
 188         }
 189
 190         RETURN(ldlm_completion_ast(lock, flags, cbdata));
 191 }
 192
 193 static int mgs_completion_ast_ir(struct ldlm_lock *lock, int flags,
 194                                  void *cbdata)
 195 {
 196         ENTRY;
 197
 198         if (!(flags & (LDLM_FL_BLOCK_WAIT | LDLM_FL_BLOCK_GRANTED |
 199                        LDLM_FL_BLOCK_CONV))) {
 200                 struct fs_db *fsdb;
 201
 202                 /* l_ast_data is used as a marker to avoid cancel ldlm lock
 203                  * twice. See LU-1259. */
 204                 lock_res_and_lock(lock);
 205                 fsdb = (struct fs_db *)lock->l_ast_data;
 206                 lock->l_ast_data = NULL;
 207                 unlock_res_and_lock(lock);
 208
 209                 if (fsdb != NULL) {
 210                         struct lustre_handle lockh;
 211
 212                         mgs_ir_notify_complete(fsdb);
 213
 214                         ldlm_lock2handle(lock, &lockh);
 215                         ldlm_lock_decref_and_cancel(&lockh, LCK_EX);
 216                 }
 217         }
 218
 219         RETURN(ldlm_completion_ast(lock, flags, cbdata));
 220 }
 221
 222 void mgs_revoke_lock(struct mgs_device *mgs, struct fs_db *fsdb, int type)
 223 {
 224         ldlm_completion_callback cp = NULL;
 225         struct lustre_handle     lockh = { 0 };
 226         struct ldlm_res_id       res_id;
 227         int flags = LDLM_FL_ATOMIC_CB;
 228         int rc;
 229         ENTRY;
 230
 231         LASSERT(fsdb->fsdb_name[0] != '\0');
 232         rc = mgc_fsname2resid(fsdb->fsdb_name, &res_id, type);
 233         LASSERT(rc == 0);
 234
 235         switch (type) {
 236         case CONFIG_T_CONFIG:
 237                 cp = mgs_completion_ast_config;
 238                 if (cfs_test_and_set_bit(FSDB_REVOKING_LOCK, &fsdb->fsdb_flags))
 239                         rc = -EALREADY;
 240                 break;
 241         case CONFIG_T_RECOVER:
 242                 cp = mgs_completion_ast_ir;
 243         default:
 244                 break;
 245         }
 246
 247         if (!rc) {
 248                 LASSERT(cp != NULL);
 249                 rc = ldlm_cli_enqueue_local(mgs->mgs_obd->obd_namespace,
 250                                             &res_id, LDLM_PLAIN, NULL, LCK_EX,
 251                                             &flags, ldlm_blocking_ast, cp,
 252                                             NULL, fsdb, 0, NULL, &lockh);
 253                 if (rc != ELDLM_OK) {
 254                         CERROR("can't take cfg lock for "LPX64"/"LPX64"(%d)\n",
 255                                le64_to_cpu(res_id.name[0]),
 256                                le64_to_cpu(res_id.name[1]), rc);
 257
 258                         if (type == CONFIG_T_CONFIG)
 259                                 cfs_clear_bit(FSDB_REVOKING_LOCK,
 260                                               &fsdb->fsdb_flags);
 261                 }
 262                 /* lock has been cancelled in completion_ast. */
 263         }
 264
 265         RETURN_EXIT;
 266 }
 267
 268 /* rc=0 means ok
 269       1 means update
 270      <0 means error */
 271 static int mgs_check_target(const struct lu_env *env,
 272                             struct mgs_device *mgs,
 273                             struct mgs_target_info *mti)
 274 {
 275         int rc;
 276         ENTRY;
 277
 278         rc = mgs_check_index(env, mgs, mti);
 279         if (rc == 0) {
 280                 LCONSOLE_ERROR_MSG(0x13b, "%s claims to have registered, but "
 281                                    "this MGS does not know about it, preventing "
 282                                    "registration.\n", mti->mti_svname);
 283                 rc = -ENOENT;
 284         } else if (rc == -1) {
 285                 LCONSOLE_ERROR_MSG(0x13c, "Client log %s-client has "
 286                                    "disappeared! Regenerating all logs.\n",
 287                                    mti->mti_fsname);
 288                 mti->mti_flags |= LDD_F_WRITECONF;
 289                 rc = 1;
 290         } else {
 291                 /* Index is correctly marked as used */
 292
 293                 /* If the logs don't contain the mti_nids then add
 294                    them as failover nids */
 295                 rc = mgs_check_failnid(env, mgs, mti);
 296         }
 297
 298         RETURN(rc);
 299 }
 300
 301 /* Ensure this is not a failover node that is connecting first*/
 302 static int mgs_check_failover_reg(struct mgs_target_info *mti)
 303 {
 304         lnet_nid_t nid;
 305         char *ptr;
 306         int i;
 307
 308         ptr = mti->mti_params;
 309         while (class_find_param(ptr, PARAM_FAILNODE, &ptr) == 0) {
 310                 while (class_parse_nid_quiet(ptr, &nid, &ptr) == 0) {
 311                         for (i = 0; i < mti->mti_nid_count; i++) {
 312                                 if (nid == mti->mti_nids[i]) {
 313                                         LCONSOLE_WARN("Denying initial registra"
 314                                                       "tion attempt from nid %s"
 315                                                       ", specified as failover"
 316                                                       "\n",libcfs_nid2str(nid));
 317                                         return -EADDRNOTAVAIL;
 318                                 }
 319                         }
 320                 }
 321         }
 322         return 0;
 323 }
 324
 325 /* Called whenever a target starts up.  Flags indicate first connect, etc. */
 326 static int mgs_handle_target_reg(struct ptlrpc_request *req)
 327 {
 328         struct obd_device *obd = req->rq_export->exp_obd;
 329         struct mgs_device *mgs = exp2mgs_dev(req->rq_export);
 330         struct lu_env     *env = req->rq_svc_thread->t_env;
 331         struct mgs_target_info *mti, *rep_mti;
 332         struct fs_db *fsdb;
 333         int opc;
 334         int rc = 0;
 335         ENTRY;
 336
 337         mgs_counter_incr(req->rq_export, LPROC_MGS_TARGET_REG);
 338
 339         mti = req_capsule_client_get(&req->rq_pill, &RMF_MGS_TARGET_INFO);
 340
 341         opc = mti->mti_flags & LDD_F_OPC_MASK;
 342         if (opc == LDD_F_OPC_READY) {
 343                 CDEBUG(D_MGS, "fs: %s index: %d is ready to reconnect.\n",
 344                        mti->mti_fsname, mti->mti_stripe_index);
 345                 rc = mgs_ir_update(env, mgs, mti);
 346                 if (rc) {
 347                         LASSERT(!(mti->mti_flags & LDD_F_IR_CAPABLE));
 348                         CERROR("Update IR return with %d(ignore and IR "
 349                                "disabled)\n", rc);
 350                 }
 351                 GOTO(out_nolock, rc);
 352         }
 353
 354         /* Do not support unregistering right now. */
 355         if (opc != LDD_F_OPC_REG)
 356                 GOTO(out_nolock, rc = -EINVAL);
 357
 358         CDEBUG(D_MGS, "fs: %s index: %d is registered to MGS.\n",
 359                mti->mti_fsname, mti->mti_stripe_index);
 360
 361         if (mti->mti_flags & LDD_F_NEED_INDEX)
 362                 mti->mti_flags |= LDD_F_WRITECONF;
 363
 364         if (!(mti->mti_flags & (LDD_F_WRITECONF | LDD_F_UPGRADE14 |
 365                                 LDD_F_UPDATE))) {
 366                 /* We're just here as a startup ping. */
 367                 CDEBUG(D_MGS, "Server %s is running on %s\n",
 368                        mti->mti_svname, obd_export_nid2str(req->rq_export));
 369                 rc = mgs_check_target(env, mgs, mti);
 370                 /* above will set appropriate mti flags */
 371                 if (rc <= 0)
 372                         /* Nothing wrong, or fatal error */
 373                         GOTO(out_nolock, rc);
 374         } else {
 375                 if (!(mti->mti_flags & LDD_F_NO_PRIMNODE)
 376                     && (rc = mgs_check_failover_reg(mti)))
 377                         GOTO(out_nolock, rc);
 378         }
 379
 380         OBD_FAIL_TIMEOUT(OBD_FAIL_MGS_PAUSE_TARGET_REG, 10);
 381
 382         if (mti->mti_flags & LDD_F_WRITECONF) {
 383                 if (mti->mti_flags & LDD_F_SV_TYPE_MDT &&
 384                     mti->mti_stripe_index == 0) {
 385                         rc = mgs_erase_logs(env, mgs, mti->mti_fsname);
 386                         LCONSOLE_WARN("%s: Logs for fs %s were removed by user "
 387                                       "request.  All servers must be restarted "
 388                                       "in order to regenerate the logs."
 389                                       "\n", obd->obd_name, mti->mti_fsname);
 390                 } else if (mti->mti_flags &
 391                            (LDD_F_SV_TYPE_OST | LDD_F_SV_TYPE_MDT)) {
 392                         rc = mgs_erase_log(env, mgs, mti->mti_svname);
 393                         LCONSOLE_WARN("%s: Regenerating %s log by user "
 394                                       "request.\n",
 395                                       obd->obd_name, mti->mti_svname);
 396                 }
 397                 mti->mti_flags |= LDD_F_UPDATE;
 398                 /* Erased logs means start from scratch. */
 399                 mti->mti_flags &= ~LDD_F_UPGRADE14;
 400                 if (rc)
 401                         GOTO(out_nolock, rc);
 402         }
 403
 404         rc = mgs_find_or_make_fsdb(env, mgs, mti->mti_fsname, &fsdb);
 405         if (rc) {
 406                 CERROR("Can't get db for %s: %d\n", mti->mti_fsname, rc);
 407                 GOTO(out_nolock, rc);
 408         }
 409
 410         /*
 411          * Log writing contention is handled by the fsdb_mutex.
 412          *
 413          * It should be alright if someone was reading while we were
 414          * updating the logs - if we revoke at the end they will just update
 415          * from where they left off.
 416          */
 417
 418         if (mti->mti_flags & LDD_F_UPGRADE14) {
 419                 CERROR("Can't upgrade from 1.4 (%d)\n", rc);
 420                 GOTO(out, rc);
 421         }
 422
 423         if (mti->mti_flags & LDD_F_UPDATE) {
 424                 CDEBUG(D_MGS, "updating %s, index=%d\n", mti->mti_svname,
 425                        mti->mti_stripe_index);
 426
 427                 /* create or update the target log
 428                    and update the client/mdt logs */
 429                 rc = mgs_write_log_target(env, mgs, mti, fsdb);
 430                 if (rc) {
 431                         CERROR("Failed to write %s log (%d)\n",
 432                                mti->mti_svname, rc);
 433                         GOTO(out, rc);
 434                 }
 435
 436                 mti->mti_flags &= ~(LDD_F_VIRGIN | LDD_F_UPDATE |
 437                                     LDD_F_NEED_INDEX | LDD_F_WRITECONF |
 438                                     LDD_F_UPGRADE14);
 439                 mti->mti_flags |= LDD_F_REWRITE_LDD;
 440         }
 441
 442 out:
 443         mgs_revoke_lock(mgs, fsdb, CONFIG_T_CONFIG);
 444
 445 out_nolock:
 446         CDEBUG(D_MGS, "replying with %s, index=%d, rc=%d\n", mti->mti_svname,
 447                mti->mti_stripe_index, rc);
 448         req->rq_status = rc;
 449         if (rc)
 450                 /* we need an error flag to tell the target what's going on,
 451                  * instead of just doing it by error code only. */
 452                 mti->mti_flags |= LDD_F_ERROR;
 453
 454         rc = req_capsule_server_pack(&req->rq_pill);
 455         if (rc)
 456                 RETURN(rc);
 457
 458         /* send back the whole mti in the reply */
 459         rep_mti = req_capsule_server_get(&req->rq_pill, &RMF_MGS_TARGET_INFO);
 460         *rep_mti = *mti;
 461
 462         /* Flush logs to disk */
 463         dt_sync(req->rq_svc_thread->t_env, mgs->mgs_bottom);
 464         RETURN(rc);
 465 }
 466
 467 static int mgs_set_info_rpc(struct ptlrpc_request *req)
 468 {
 469         struct mgs_device *mgs = exp2mgs_dev(req->rq_export);
 470         struct lu_env     *env = req->rq_svc_thread->t_env;
 471         struct mgs_send_param *msp, *rep_msp;
 472         struct mgs_thread_info *mgi = mgs_env_info(env);
 473         int rc;
 474         struct lustre_cfg *lcfg;
 475         ENTRY;
 476
 477         msp = req_capsule_client_get(&req->rq_pill, &RMF_MGS_SEND_PARAM);
 478         LASSERT(msp);
 479
 480         /* Construct lustre_cfg structure to pass to function mgs_setparam */
 481         lustre_cfg_bufs_reset(&mgi->mgi_bufs, NULL);
 482         lustre_cfg_bufs_set_string(&mgi->mgi_bufs, 1, msp->mgs_param);
 483         lcfg = lustre_cfg_new(LCFG_PARAM, &mgi->mgi_bufs);
 484         if (IS_ERR(lcfg))
 485                 GOTO(out, rc = PTR_ERR(lcfg));
 486         rc = mgs_setparam(env, mgs, lcfg, mgi->mgi_fsname);
 487         if (rc) {
 488                 CERROR("Error %d in setting the parameter %s for fs %s\n",
 489                        rc, msp->mgs_param, mgi->mgi_fsname);
 490                 GOTO(out_cfg, rc);
 491         }
 492
 493         rc = req_capsule_server_pack(&req->rq_pill);
 494         if (rc == 0) {
 495                 rep_msp = req_capsule_server_get(&req->rq_pill, &RMF_MGS_SEND_PARAM);
 496                 rep_msp = msp;
 497         }
 498 out_cfg:
 499         lustre_cfg_free(lcfg);
 500 out:
 501         RETURN(rc);
 502 }
 503
 504 static int mgs_config_read(struct ptlrpc_request *req)
 505 {
 506         struct mgs_config_body *body;
 507         int rc;
 508         ENTRY;
 509
 510         body = req_capsule_client_get(&req->rq_pill, &RMF_MGS_CONFIG_BODY);
 511         if (body == NULL)
 512                 RETURN(-EINVAL);
 513
 514         switch (body->mcb_type) {
 515         case CONFIG_T_RECOVER:
 516                 rc = mgs_get_ir_logs(req);
 517                 break;
 518
 519         case CONFIG_T_CONFIG:
 520                 rc = -ENOTSUPP;
 521                 break;
 522
 523         default:
 524                 rc = -EINVAL;
 525                 break;
 526         }
 527
 528         RETURN(rc);
 529 }
 530
 531 /*
 532  * similar as in ost_connect_check_sptlrpc()
 533  */
 534 static int mgs_connect_check_sptlrpc(struct ptlrpc_request *req)
 535 {
 536         struct obd_export     *exp = req->rq_export;
 537         struct mgs_device     *mgs = exp2mgs_dev(req->rq_export);
 538         struct lu_env         *env = req->rq_svc_thread->t_env;
 539         struct fs_db          *fsdb;
 540         struct sptlrpc_flavor  flvr;
 541         int                    rc = 0;
 542
 543         if (exp->exp_flvr.sf_rpc == SPTLRPC_FLVR_INVALID) {
 544                 rc = mgs_find_or_make_fsdb(env, mgs, MGSSELF_NAME, &fsdb);
 545                 if (rc)
 546                         return rc;
 547
 548                 cfs_mutex_lock(&fsdb->fsdb_mutex);
 549                 if (sptlrpc_rule_set_choose(&fsdb->fsdb_srpc_gen,
 550                                             LUSTRE_SP_MGC, LUSTRE_SP_MGS,
 551                                             req->rq_peer.nid,
 552                                             &flvr) == 0) {
 553                         /* by defualt allow any flavors */
 554                         flvr.sf_rpc = SPTLRPC_FLVR_ANY;
 555                 }
 556                 cfs_mutex_unlock(&fsdb->fsdb_mutex);
 557
 558                 cfs_spin_lock(&exp->exp_lock);
 559
 560                 exp->exp_sp_peer = req->rq_sp_from;
 561                 exp->exp_flvr = flvr;
 562
 563                 if (exp->exp_flvr.sf_rpc != SPTLRPC_FLVR_ANY &&
 564                     exp->exp_flvr.sf_rpc != req->rq_flvr.sf_rpc) {
 565                         CERROR("invalid rpc flavor %x, expect %x, from %s\n",
 566                                req->rq_flvr.sf_rpc, exp->exp_flvr.sf_rpc,
 567                                libcfs_nid2str(req->rq_peer.nid));
 568                         rc = -EACCES;
 569                 }
 570
 571                 cfs_spin_unlock(&exp->exp_lock);
 572         } else {
 573                 if (exp->exp_sp_peer != req->rq_sp_from) {
 574                         CERROR("RPC source %s doesn't match %s\n",
 575                                sptlrpc_part2name(req->rq_sp_from),
 576                                sptlrpc_part2name(exp->exp_sp_peer));
 577                         rc = -EACCES;
 578                 } else {
 579                         rc = sptlrpc_target_export_check(exp, req);
 580                 }
 581         }
 582
 583         return rc;
 584 }
 585
 586 /* Called whenever a target cleans up. */
 587 /* XXX - Currently unused */
 588 static int mgs_handle_target_del(struct ptlrpc_request *req)
 589 {
 590         ENTRY;
 591         mgs_counter_incr(req->rq_export, LPROC_MGS_TARGET_DEL);
 592         RETURN(0);
 593 }
 594
 595 /* XXX - Currently unused */
 596 static int mgs_handle_exception(struct ptlrpc_request *req)
 597 {
 598         ENTRY;
 599         mgs_counter_incr(req->rq_export, LPROC_MGS_EXCEPTION);
 600         RETURN(0);
 601 }
 602
 603 /*
 604  * For old clients there is no direct way of knowing which filesystems
 605  * a client is operating at the MGS side. But we need to pick up those
 606  * clients so that the MGS can mark the corresponding filesystem as
 607  * non-IR capable because old clients are not ready to be notified.
 608  *
 609  * This is why we have this _hack_ function. We detect the filesystem's
 610  * name by hacking llog operation which is currently used by the clients
 611  * to fetch configuration logs. At present this is fine because this is
 612  * the ONLY llog operation between mgc and the MGS.
 613  *
 614  * If extra llog operation is going to be added, this function needs fixing.
 615  *
 616  * If releases prior than 2.0 are not supported, we can remove this function.
 617  */
 618 static int mgs_handle_fslog_hack(struct ptlrpc_request *req)
 619 {
 620         char *logname;
 621         char fsname[16];
 622         char *ptr;
 623         int rc;
 624
 625         /* XXX: We suppose that llog at mgs is only used for
 626          * fetching file system log */
 627         logname = req_capsule_client_get(&req->rq_pill, &RMF_NAME);
 628         if (logname == NULL) {
 629                 CERROR("No logname, is llog on MGS used for something else?\n");
 630                 return -EINVAL;
 631         }
 632
 633         ptr = strchr(logname, '-');
 634         rc = (int)(ptr - logname);
 635         if (ptr == NULL || rc >= sizeof(fsname)) {
 636                 CERROR("Invalid logname received: %s\n", logname);
 637                 return -EINVAL;
 638         }
 639
 640         strncpy(fsname, logname, rc);
 641         fsname[rc] = 0;
 642         rc = mgs_fsc_attach(req->rq_svc_thread->t_env, req->rq_export, fsname);
 643         if (rc < 0 && rc != -EEXIST)
 644                 CERROR("add fs client %s returns %d\n", fsname, rc);
 645
 646         return rc;
 647 }
 648
 649 /* TODO: handle requests in a similar way as MDT: see mdt_handle_common() */
 650 int mgs_handle(struct ptlrpc_request *req)
 651 {
 652         int fail = OBD_FAIL_MGS_ALL_REPLY_NET;
 653         int opc, rc = 0;
 654         ENTRY;
 655
 656         req_capsule_init(&req->rq_pill, req, RCL_SERVER);
 657         CFS_FAIL_TIMEOUT_MS(OBD_FAIL_MGS_PAUSE_REQ, cfs_fail_val);
 658         if (CFS_FAIL_CHECK(OBD_FAIL_MGS_ALL_REQUEST_NET))
 659                 RETURN(0);
 660
 661         LASSERT(current->journal_info == NULL);
 662         opc = lustre_msg_get_opc(req->rq_reqmsg);
 663
 664         if (opc == SEC_CTX_INIT ||
 665             opc == SEC_CTX_INIT_CONT ||
 666             opc == SEC_CTX_FINI)
 667                 GOTO(out, rc = 0);
 668
 669         if (opc != MGS_CONNECT) {
 670                 if (!class_connected_export(req->rq_export)) {
 671                         DEBUG_REQ(D_MGS, req, "operation on unconnected MGS\n");
 672                         req->rq_status = -ENOTCONN;
 673                         GOTO(out, rc = -ENOTCONN);
 674                 }
 675         }
 676
 677         switch (opc) {
 678         case MGS_CONNECT:
 679                 DEBUG_REQ(D_MGS, req, "connect");
 680                 /* MGS and MDS have same request format for connect */
 681                 req_capsule_set(&req->rq_pill, &RQF_MDS_CONNECT);
 682                 rc = target_handle_connect(req);
 683                 if (rc == 0)
 684                         rc = mgs_connect_check_sptlrpc(req);
 685
 686                 if (!rc && (lustre_msg_get_conn_cnt(req->rq_reqmsg) > 1))
 687                         /* Make clients trying to reconnect after a MGS restart
 688                            happy; also requires obd_replayable */
 689                         lustre_msg_add_op_flags(req->rq_repmsg,
 690                                                 MSG_CONNECT_RECONNECT);
 691                 break;
 692         case MGS_DISCONNECT:
 693                 DEBUG_REQ(D_MGS, req, "disconnect");
 694                 /* MGS and MDS have same request format for disconnect */
 695                 req_capsule_set(&req->rq_pill, &RQF_MDS_DISCONNECT);
 696                 rc = target_handle_disconnect(req);
 697                 req->rq_status = rc;            /* superfluous? */
 698                 break;
 699         case MGS_EXCEPTION:
 700                 DEBUG_REQ(D_MGS, req, "exception");
 701                 rc = mgs_handle_exception(req);
 702                 break;
 703         case MGS_TARGET_REG:
 704                 DEBUG_REQ(D_MGS, req, "target add");
 705                 req_capsule_set(&req->rq_pill, &RQF_MGS_TARGET_REG);
 706                 rc = mgs_handle_target_reg(req);
 707                 break;
 708         case MGS_TARGET_DEL:
 709                 DEBUG_REQ(D_MGS, req, "target del");
 710                 rc = mgs_handle_target_del(req);
 711                 break;
 712         case MGS_SET_INFO:
 713                 DEBUG_REQ(D_MGS, req, "set_info");
 714                 req_capsule_set(&req->rq_pill, &RQF_MGS_SET_INFO);
 715                 rc = mgs_set_info_rpc(req);
 716                 break;
 717         case MGS_CONFIG_READ:
 718                 DEBUG_REQ(D_MGS, req, "read config");
 719                 req_capsule_set(&req->rq_pill, &RQF_MGS_CONFIG_READ);
 720                 rc = mgs_config_read(req);
 721                 break;
 722         case LDLM_ENQUEUE:
 723                 DEBUG_REQ(D_MGS, req, "enqueue");
 724                 req_capsule_set(&req->rq_pill, &RQF_LDLM_ENQUEUE);
 725                 rc = ldlm_handle_enqueue(req, ldlm_server_completion_ast,
 726                                          ldlm_server_blocking_ast, NULL);
 727                 break;
 728         case LDLM_BL_CALLBACK:
 729         case LDLM_CP_CALLBACK:
 730                 DEBUG_REQ(D_MGS, req, "callback");
 731                 CERROR("callbacks should not happen on MGS\n");
 732                 LBUG();
 733                 break;
 734
 735         case OBD_PING:
 736                 DEBUG_REQ(D_INFO, req, "ping");
 737                 req_capsule_set(&req->rq_pill, &RQF_OBD_PING);
 738                 rc = target_handle_ping(req);
 739                 break;
 740         case OBD_LOG_CANCEL:
 741                 DEBUG_REQ(D_MGS, req, "log cancel");
 742                 rc = -ENOTSUPP; /* la la la */
 743                 break;
 744
 745         case LLOG_ORIGIN_HANDLE_CREATE:
 746                 DEBUG_REQ(D_MGS, req, "llog_open");
 747                 req_capsule_set(&req->rq_pill, &RQF_LLOG_ORIGIN_HANDLE_CREATE);
 748                 rc = llog_origin_handle_open(req);
 749                 if (rc == 0)
 750                         (void)mgs_handle_fslog_hack(req);
 751                 break;
 752         case LLOG_ORIGIN_HANDLE_NEXT_BLOCK:
 753                 DEBUG_REQ(D_MGS, req, "llog next block");
 754                 req_capsule_set(&req->rq_pill,
 755                                 &RQF_LLOG_ORIGIN_HANDLE_NEXT_BLOCK);
 756                 rc = llog_origin_handle_next_block(req);
 757                 break;
 758         case LLOG_ORIGIN_HANDLE_PREV_BLOCK:
 759                 DEBUG_REQ(D_MGS, req, "llog prev block");
 760                 req_capsule_set(&req->rq_pill,
 761                                 &RQF_LLOG_ORIGIN_HANDLE_PREV_BLOCK);
 762                 rc = llog_origin_handle_prev_block(req);
 763                 break;
 764         case LLOG_ORIGIN_HANDLE_READ_HEADER:
 765                 DEBUG_REQ(D_MGS, req, "llog read header");
 766                 req_capsule_set(&req->rq_pill,
 767                                 &RQF_LLOG_ORIGIN_HANDLE_READ_HEADER);
 768                 rc = llog_origin_handle_read_header(req);
 769                 break;
 770         case LLOG_ORIGIN_HANDLE_CLOSE:
 771                 DEBUG_REQ(D_MGS, req, "llog close");
 772                 rc = llog_origin_handle_close(req);
 773                 break;
 774         default:
 775                 rc = -EOPNOTSUPP;
 776         }
 777
 778         LASSERT(current->journal_info == NULL);
 779         if (rc) {
 780                 DEBUG_REQ(D_MGS, req, "MGS fail to handle opc = %d: rc = %d\n",
 781                           opc, rc);
 782                 req->rq_status = rc;
 783                 rc = ptlrpc_error(req);
 784                 RETURN(rc);
 785         }
 786 out:
 787         target_send_reply(req, rc, fail);
 788         RETURN(0);
 789 }
 790
 791 static inline int mgs_init_export(struct obd_export *exp)
 792 {
 793         struct mgs_export_data *data = &exp->u.eu_mgs_data;
 794
 795         /* init mgs_export_data for fsc */
 796         cfs_spin_lock_init(&data->med_lock);
 797         CFS_INIT_LIST_HEAD(&data->med_clients);
 798
 799         cfs_spin_lock(&exp->exp_lock);
 800         exp->exp_connecting = 1;
 801         cfs_spin_unlock(&exp->exp_lock);
 802
 803         /* self-export doesn't need client data and ldlm initialization */
 804         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
 805                                      &exp->exp_client_uuid)))
 806                 return 0;
 807         return ldlm_init_export(exp);
 808 }
 809
 810 static inline int mgs_destroy_export(struct obd_export *exp)
 811 {
 812         ENTRY;
 813
 814         target_destroy_export(exp);
 815         mgs_client_free(exp);
 816
 817         if (unlikely(obd_uuid_equals(&exp->exp_obd->obd_uuid,
 818                                      &exp->exp_client_uuid)))
 819                 RETURN(0);
 820
 821         ldlm_destroy_export(exp);
 822
 823         RETURN(0);
 824 }
 825
 826 static int mgs_extract_fs_pool(char * arg, char *fsname, char *poolname)
 827 {
 828         char *ptr;
 829
 830         ENTRY;
 831         for (ptr = arg;  (*ptr != '\0') && (*ptr != '.'); ptr++ ) {
 832                 *fsname = *ptr;
 833                 fsname++;
 834         }
 835         if (*ptr == '\0')
 836                 return -EINVAL;
 837         *fsname = '\0';
 838         ptr++;
 839         strcpy(poolname, ptr);
 840
 841         RETURN(0);
 842 }
 843
 844 static int mgs_iocontrol_pool(const struct lu_env *env,
 845                               struct mgs_device *mgs,
 846                               struct obd_ioctl_data *data)
 847 {
 848         struct mgs_thread_info *mgi = mgs_env_info(env);
 849         int rc;
 850         struct lustre_cfg *lcfg = NULL;
 851         struct llog_rec_hdr rec;
 852         char *poolname = NULL;
 853         ENTRY;
 854
 855         OBD_ALLOC(poolname, LOV_MAXPOOLNAME + 1);
 856         if (poolname == NULL)
 857                 RETURN(-ENOMEM);
 858         rec.lrh_len = llog_data_len(data->ioc_plen1);
 859
 860         if (data->ioc_type == LUSTRE_CFG_TYPE) {
 861                 rec.lrh_type = OBD_CFG_REC;
 862         } else {
 863                 CERROR("unknown cfg record type:%d \n", data->ioc_type);
 864                 GOTO(out_pool, rc = -EINVAL);
 865         }
 866
 867         if (data->ioc_plen1 > CFS_PAGE_SIZE)
 868                 GOTO(out_pool, rc = -E2BIG);
 869
 870         OBD_ALLOC(lcfg, data->ioc_plen1);
 871         if (lcfg == NULL)
 872                 GOTO(out_pool, rc = -ENOMEM);
 873
 874         if (cfs_copy_from_user(lcfg, data->ioc_pbuf1, data->ioc_plen1))
 875                 GOTO(out_lcfg, rc = -EFAULT);
 876
 877         if (lcfg->lcfg_bufcount < 2)
 878                 GOTO(out_lcfg, rc = -EFAULT);
 879
 880         /* first arg is always <fsname>.<poolname> */
 881         rc = mgs_extract_fs_pool(lustre_cfg_string(lcfg, 1), mgi->mgi_fsname,
 882                                  poolname);
 883         if (rc)
 884                 GOTO(out_lcfg, rc);
 885
 886         switch (lcfg->lcfg_command) {
 887         case LCFG_POOL_NEW:
 888                 if (lcfg->lcfg_bufcount != 2)
 889                         GOTO(out_lcfg, rc = -EINVAL);
 890                 rc = mgs_pool_cmd(env, mgs, LCFG_POOL_NEW, mgi->mgi_fsname,
 891                                   poolname, NULL);
 892                 break;
 893         case LCFG_POOL_ADD:
 894                 if (lcfg->lcfg_bufcount != 3)
 895                         GOTO(out_lcfg, rc = -EINVAL);
 896                 rc = mgs_pool_cmd(env, mgs, LCFG_POOL_ADD, mgi->mgi_fsname,
 897                                   poolname, lustre_cfg_string(lcfg, 2));
 898                 break;
 899         case LCFG_POOL_REM:
 900                 if (lcfg->lcfg_bufcount != 3)
 901                         GOTO(out_lcfg, rc = -EINVAL);
 902                 rc = mgs_pool_cmd(env, mgs, LCFG_POOL_REM, mgi->mgi_fsname,
 903                                   poolname, lustre_cfg_string(lcfg, 2));
 904                 break;
 905         case LCFG_POOL_DEL:
 906                 if (lcfg->lcfg_bufcount != 2)
 907                         GOTO(out_lcfg, rc = -EINVAL);
 908                 rc = mgs_pool_cmd(env, mgs, LCFG_POOL_DEL, mgi->mgi_fsname,
 909                                   poolname, NULL);
 910                 break;
 911         default:
 912                  rc = -EINVAL;
 913         }
 914
 915         if (rc) {
 916                 CERROR("OBD_IOC_POOL err %d, cmd %X for pool %s.%s\n",
 917                        rc, lcfg->lcfg_command, mgi->mgi_fsname, poolname);
 918                 GOTO(out_lcfg, rc);
 919         }
 920
 921 out_lcfg:
 922         OBD_FREE(lcfg, data->ioc_plen1);
 923 out_pool:
 924         OBD_FREE(poolname, LOV_MAXPOOLNAME + 1);
 925         RETURN(rc);
 926 }
 927
 928 /* from mdt_iocontrol */
 929 int mgs_iocontrol(unsigned int cmd, struct obd_export *exp, int len,
 930                   void *karg, void *uarg)
 931 {
 932         struct mgs_device *mgs = exp2mgs_dev(exp);
 933         struct obd_ioctl_data *data = karg;
 934         struct lvfs_run_ctxt saved;
 935         struct lu_env env;
 936         int rc = 0;
 937
 938         ENTRY;
 939         CDEBUG(D_IOCTL, "handling ioctl cmd %#x\n", cmd);
 940
 941         rc = lu_env_init(&env, LCT_MG_THREAD);
 942         if (rc)
 943                 RETURN(rc);
 944
 945         switch (cmd) {
 946
 947         case OBD_IOC_PARAM: {
 948                 struct mgs_thread_info *mgi = mgs_env_info(&env);
 949                 struct lustre_cfg *lcfg;
 950                 struct llog_rec_hdr rec;
 951
 952                 rec.lrh_len = llog_data_len(data->ioc_plen1);
 953
 954                 if (data->ioc_type == LUSTRE_CFG_TYPE) {
 955                         rec.lrh_type = OBD_CFG_REC;
 956                 } else {
 957                         CERROR("unknown cfg record type:%d \n", data->ioc_type);
 958                         GOTO(out, rc = -EINVAL);
 959                 }
 960
 961                 OBD_ALLOC(lcfg, data->ioc_plen1);
 962                 if (lcfg == NULL)
 963                         GOTO(out, rc = -ENOMEM);
 964                 if (cfs_copy_from_user(lcfg, data->ioc_pbuf1, data->ioc_plen1))
 965                         GOTO(out_free, rc = -EFAULT);
 966
 967                 if (lcfg->lcfg_bufcount < 1)
 968                         GOTO(out_free, rc = -EINVAL);
 969
 970                 rc = mgs_setparam(&env, mgs, lcfg, mgi->mgi_fsname);
 971                 if (rc)
 972                         CERROR("%s: setparam err: rc = %d\n",
 973                                exp->exp_obd->obd_name, rc);
 974 out_free:
 975                 OBD_FREE(lcfg, data->ioc_plen1);
 976                 break;
 977         }
 978
 979         case OBD_IOC_POOL:
 980                 rc = mgs_iocontrol_pool(&env, mgs, data);
 981                 break;
 982
 983         case OBD_IOC_DUMP_LOG: {
 984                 struct llog_ctxt *ctxt;
 985                 ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
 986                 push_ctxt(&saved, &mgs->mgs_obd->obd_lvfs_ctxt, NULL);
 987                 rc = class_config_dump_llog(ctxt, data->ioc_inlbuf1, NULL);
 988                 pop_ctxt(&saved, &mgs->mgs_obd->obd_lvfs_ctxt, NULL);
 989                 llog_ctxt_put(ctxt);
 990
 991                 break;
 992         }
 993
 994         case OBD_IOC_LLOG_CHECK:
 995         case OBD_IOC_LLOG_INFO:
 996         case OBD_IOC_LLOG_PRINT: {
 997                 struct llog_ctxt *ctxt;
 998
 999                 ctxt = llog_get_context(mgs->mgs_obd, LLOG_CONFIG_ORIG_CTXT);
1000                 push_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
1001                 rc = llog_ioctl(NULL, ctxt, cmd, data);
1002                 pop_ctxt(&saved, &ctxt->loc_exp->exp_obd->obd_lvfs_ctxt, NULL);
1003                 llog_ctxt_put(ctxt);
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);