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