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