Whamcloud - gitweb
LU-12616 obclass: fix MDS start/stop race
[fs/lustre-release.git] / lustre / mdd / mdd_device.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.gnu.org/licenses/gpl-2.0.html
19  *
20  * GPL HEADER END
21  */
22 /*
23  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
24  * Use is subject to license terms.
25  *
26  * Copyright (c) 2011, 2017, Intel Corporation.
27  */
28 /*
29  * This file is part of Lustre, http://www.lustre.org/
30  * Lustre is a trademark of Sun Microsystems, Inc.
31  *
32  * lustre/mdd/mdd_device.c
33  *
34  * Lustre Metadata Server (mdd) routines
35  *
36  * Author: Wang Di <wangdi@clusterfs.com>
37  */
38
39 #define DEBUG_SUBSYSTEM S_MDS
40
41 #include <linux/module.h>
42 #include <linux/kthread.h>
43 #include <obd_class.h>
44 #include <uapi/linux/lustre/lustre_ioctl.h>
45 #include <lustre_mds.h>
46 #include <obd_support.h>
47 #include <lu_object.h>
48 #include <uapi/linux/lustre/lustre_param.h>
49 #include <lustre_fid.h>
50 #include <lustre_nodemap.h>
51 #include <lustre_barrier.h>
52
53 #include "mdd_internal.h"
54
55 static const struct md_device_operations mdd_ops;
56 static struct lu_device_type mdd_device_type;
57
58 static const char mdd_root_dir_name[] = "ROOT";
59 static const char mdd_obf_dir_name[] = "fid";
60 static const char mdd_lpf_dir_name[] = "lost+found";
61
62 /* Slab for MDD object allocation */
63 struct kmem_cache *mdd_object_kmem;
64
65 static struct lu_kmem_descr mdd_caches[] = {
66         {
67                 .ckd_cache = &mdd_object_kmem,
68                 .ckd_name  = "mdd_obj",
69                 .ckd_size  = sizeof(struct mdd_object)
70         },
71         {
72                 .ckd_cache = NULL
73         }
74 };
75
76 static int mdd_connect_to_next(const struct lu_env *env, struct mdd_device *m,
77                                const char *nextdev)
78 {
79         struct obd_connect_data *data = NULL;
80         struct lu_device        *lud = mdd2lu_dev(m);
81         struct obd_device       *obd;
82         int                      rc;
83         ENTRY;
84
85         LASSERT(m->mdd_child_exp == NULL);
86
87         OBD_ALLOC(data, sizeof(*data));
88         if (data == NULL)
89                 GOTO(out, rc = -ENOMEM);
90
91         obd = class_name2obd(nextdev);
92         if (obd == NULL) {
93                 CERROR("can't locate next device: %s\n", nextdev);
94                 GOTO(out, rc = -ENOTCONN);
95         }
96
97         data->ocd_connect_flags = OBD_CONNECT_VERSION;
98         data->ocd_version = LUSTRE_VERSION_CODE;
99
100         rc = obd_connect(NULL, &m->mdd_child_exp, obd, &obd->obd_uuid, data, NULL);
101         if (rc) {
102                 CERROR("cannot connect to next dev %s (%d)\n", nextdev, rc);
103                 GOTO(out, rc);
104         }
105
106         lud->ld_site = m->mdd_child_exp->exp_obd->obd_lu_dev->ld_site;
107         LASSERT(lud->ld_site);
108         m->mdd_child = lu2dt_dev(m->mdd_child_exp->exp_obd->obd_lu_dev);
109         m->mdd_bottom = lu2dt_dev(lud->ld_site->ls_bottom_dev);
110         lu_dev_add_linkage(lud->ld_site, lud);
111
112 out:
113         if (data)
114                 OBD_FREE(data, sizeof(*data));
115         RETURN(rc);
116 }
117
118 static int mdd_init0(const struct lu_env *env, struct mdd_device *mdd,
119                 struct lu_device_type *t, struct lustre_cfg *lcfg)
120 {
121         int rc = -EINVAL;
122         const char *dev;
123         ENTRY;
124
125         /* LU-8040 Set defaults here, before values configs */
126         mdd->mdd_cl.mc_flags = 0; /* off by default */
127         mdd->mdd_cl.mc_mask = CHANGELOG_DEFMASK;
128         mdd->mdd_cl.mc_deniednext = 60; /* 60 secs by default */
129
130         dev = lustre_cfg_string(lcfg, 0);
131         if (dev == NULL)
132                 RETURN(rc);
133
134         mdd->mdd_md_dev.md_lu_dev.ld_obd = class_name2obd(dev);
135         if (mdd->mdd_md_dev.md_lu_dev.ld_obd == NULL)
136                 RETURN(rc);
137         mdd->mdd_md_dev.md_lu_dev.ld_ops = &mdd_lu_ops;
138         mdd->mdd_md_dev.md_ops = &mdd_ops;
139
140         rc = mdd_connect_to_next(env, mdd, lustre_cfg_string(lcfg, 3));
141         if (rc != 0)
142                 RETURN(rc);
143
144         mdd->mdd_atime_diff = MAX_ATIME_DIFF;
145         /* sync permission changes */
146         mdd->mdd_sync_permission = 1;
147         /* enable changelog garbage collection */
148         mdd->mdd_changelog_gc = 0;
149         /* with a significant amount of idle time */
150         mdd->mdd_changelog_max_idle_time = CHLOG_MAX_IDLE_TIME;
151         /* or a significant amount of late indexes */
152         mdd->mdd_changelog_max_idle_indexes = CHLOG_MAX_IDLE_INDEXES;
153         /* with a reasonable interval between each check */
154         mdd->mdd_changelog_min_gc_interval = CHLOG_MIN_GC_INTERVAL;
155         /* with a very few number of free catalog entries */
156         mdd->mdd_changelog_min_free_cat_entries = CHLOG_MIN_FREE_CAT_ENTRIES;
157
158         dt_conf_get(env, mdd->mdd_child, &mdd->mdd_dt_conf);
159
160         /* we are using service name but not mdd obd name
161          * for compatibility reasons.
162          * It is passed from MDT in lustre_cfg[2] buffer */
163         rc = mdd_procfs_init(mdd, lustre_cfg_string(lcfg, 2));
164         if (rc < 0)
165                 obd_disconnect(mdd->mdd_child_exp);
166
167         RETURN(rc);
168 }
169
170 static struct lu_device *mdd_device_fini(const struct lu_env *env,
171                                          struct lu_device *d)
172 {
173         struct mdd_device *mdd = lu2mdd_dev(d);
174
175         if (d->ld_site)
176                 lu_dev_del_linkage(d->ld_site, d);
177
178         mdd_procfs_fini(mdd);
179         return NULL;
180 }
181
182 static int changelog_init_cb(const struct lu_env *env, struct llog_handle *llh,
183                              struct llog_rec_hdr *hdr, void *data)
184 {
185         struct mdd_device *mdd = (struct mdd_device *)data;
186         struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr;
187
188         LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
189         LASSERT(rec->cr_hdr.lrh_type == CHANGELOG_REC);
190
191         CDEBUG(D_INFO,
192                "seeing record at index %d/%d/%llu t=%x %.*s in log"
193                DFID"\n", hdr->lrh_index, rec->cr_hdr.lrh_index,
194                rec->cr.cr_index, rec->cr.cr_type, rec->cr.cr_namelen,
195                changelog_rec_name(&rec->cr), PFID(&llh->lgh_id.lgl_oi.oi_fid));
196
197         mdd->mdd_cl.mc_index = rec->cr.cr_index;
198         return LLOG_PROC_BREAK;
199 }
200
201 static int changelog_user_init_cb(const struct lu_env *env,
202                                   struct llog_handle *llh,
203                                   struct llog_rec_hdr *hdr, void *data)
204 {
205         struct mdd_device *mdd = (struct mdd_device *)data;
206         struct llog_changelog_user_rec *rec =
207                 (struct llog_changelog_user_rec *)hdr;
208
209         LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
210         LASSERT(rec->cur_hdr.lrh_type == CHANGELOG_USER_REC);
211
212         CDEBUG(D_INFO, "seeing user at index %d/%d id=%d endrec=%llu"
213                " in log "DFID"\n", hdr->lrh_index, rec->cur_hdr.lrh_index,
214                rec->cur_id, rec->cur_endrec, PFID(&llh->lgh_id.lgl_oi.oi_fid));
215
216         spin_lock(&mdd->mdd_cl.mc_user_lock);
217         mdd->mdd_cl.mc_lastuser = rec->cur_id;
218         mdd->mdd_cl.mc_users++;
219         if (rec->cur_endrec > mdd->mdd_cl.mc_index)
220                 mdd->mdd_cl.mc_index = rec->cur_endrec;
221         spin_unlock(&mdd->mdd_cl.mc_user_lock);
222
223         return LLOG_PROC_BREAK;
224 }
225
226 struct changelog_orphan_data {
227         __u64 index;
228         struct mdd_device *mdd;
229 };
230
231 /* find oldest changelog record index */
232 static int changelog_detect_orphan_cb(const struct lu_env *env,
233                                       struct llog_handle *llh,
234                                       struct llog_rec_hdr *hdr, void *data)
235 {
236         struct mdd_device *mdd = ((struct changelog_orphan_data *)data)->mdd;
237         struct llog_changelog_rec *rec = container_of(hdr,
238                                                       struct llog_changelog_rec,
239                                                       cr_hdr);
240
241         LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
242
243         if (rec->cr_hdr.lrh_type != CHANGELOG_REC) {
244                 CWARN("%s: invalid record at index %d in log "DFID"\n",
245                       mdd2obd_dev(mdd)->obd_name, hdr->lrh_index,
246                       PFID(&llh->lgh_id.lgl_oi.oi_fid));
247                 /* try to find some next valid record and thus allow to recover
248                  * from a corrupted LLOG, instead to assert and force a crash
249                  */
250                 return 0;
251         }
252
253         CDEBUG(D_INFO, "%s: seeing record at index %d/%d/%llu t=%x %.*s in log "
254                DFID"\n", mdd2obd_dev(mdd)->obd_name, hdr->lrh_index,
255                rec->cr_hdr.lrh_index, rec->cr.cr_index, rec->cr.cr_type,
256                rec->cr.cr_namelen, changelog_rec_name(&rec->cr),
257                PFID(&llh->lgh_id.lgl_oi.oi_fid));
258
259         ((struct changelog_orphan_data *)data)->index = rec->cr.cr_index;
260         return LLOG_PROC_BREAK;
261 }
262
263 /* find oldest changelog user index */
264 static int changelog_user_detect_orphan_cb(const struct lu_env *env,
265                                            struct llog_handle *llh,
266                                            struct llog_rec_hdr *hdr, void *data)
267 {
268         struct mdd_device *mdd = ((struct changelog_orphan_data *)data)->mdd;
269         struct llog_changelog_user_rec *rec = container_of(hdr,
270                                                 struct llog_changelog_user_rec,
271                                                 cur_hdr);
272
273         LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
274
275         if (rec->cur_hdr.lrh_type != CHANGELOG_USER_REC) {
276                 CWARN("%s: invalid user at index %d in log "DFID"\n",
277                       mdd2obd_dev(mdd)->obd_name, hdr->lrh_index,
278                       PFID(&llh->lgh_id.lgl_oi.oi_fid));
279                 /* try to find some next valid record and thus allow to recover
280                  * from a corrupted LLOG, instead to assert and force a crash
281                  */
282                 return 0;
283         }
284
285         CDEBUG(D_INFO, "%s: seeing user at index %d/%d id=%d endrec=%llu in "
286                "log "DFID"\n", mdd2obd_dev(mdd)->obd_name, hdr->lrh_index,
287                rec->cur_hdr.lrh_index, rec->cur_id, rec->cur_endrec,
288                PFID(&llh->lgh_id.lgl_oi.oi_fid));
289
290         if (((struct changelog_orphan_data *)data)->index == 0 ||
291             rec->cur_endrec < ((struct changelog_orphan_data *)data)->index)
292                 ((struct changelog_orphan_data *)data)->index = rec->cur_endrec;
293
294         return 0;
295 }
296
297 struct changelog_cancel_cookie {
298         long long endrec;
299         struct mdd_device *mdd;
300 };
301
302 static int llog_changelog_cancel_cb(const struct lu_env *env,
303                                     struct llog_handle *llh,
304                                     struct llog_rec_hdr *hdr, void *data)
305 {
306         struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr;
307         struct llog_cookie       cookie;
308         struct changelog_cancel_cookie *cl_cookie =
309                 (struct changelog_cancel_cookie *)data;
310         int                      rc;
311
312         ENTRY;
313
314         /* This is always a (sub)log, not the catalog */
315         LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
316
317         /* if current context is GC-thread allow it to stop upon umount
318          * remaining records cleanup will occur upon next mount
319          *
320          * also during testing, wait for GC-thread to be released
321          *
322          * XXX this requires the GC-thread to not fork a sub-thread via
323          * llog[_cat]_process_or_fork() and we may think to also implement
324          * this shutdown mechanism for manually started user unregister which
325          * can also take a long time if huge backlog of records
326          */
327         if (unlikely(cl_cookie->mdd->mdd_cl.mc_gc_task == current)) {
328                 /* wait to be released */
329                 while (CFS_FAIL_CHECK_QUIET(OBD_FAIL_FORCE_GC_THREAD))
330                         schedule();
331
332                 if (kthread_should_stop())
333                         RETURN(LLOG_PROC_BREAK);
334         }
335
336         if (rec->cr.cr_index > cl_cookie->endrec)
337                 /* records are in order, so we're done */
338                 RETURN(LLOG_PROC_BREAK);
339
340         cookie.lgc_lgl = llh->lgh_id;
341         cookie.lgc_index = hdr->lrh_index;
342
343         /* cancel them one at a time.  I suppose we could store up the cookies
344          * and cancel them all at once; probably more efficient, but this is
345          * done as a user call, so who cares... */
346         rc = llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, 1,
347                                      &cookie);
348         RETURN(rc < 0 ? rc : 0);
349 }
350
351 static int llog_changelog_cancel(const struct lu_env *env,
352                                  struct llog_ctxt *ctxt,
353                                  struct changelog_cancel_cookie *cookie)
354 {
355         struct llog_handle      *cathandle = ctxt->loc_handle;
356         int                      rc;
357
358         ENTRY;
359
360         /* This should only be called with the catalog handle */
361         LASSERT(cathandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT);
362
363         rc = llog_cat_process(env, cathandle, llog_changelog_cancel_cb,
364                               cookie, 0, 0);
365         if (rc >= 0)
366                 /* 0 or 1 means we're done */
367                 rc = 0;
368         else
369                 CERROR("%s: cancel idx %u of catalog "DFID": rc = %d\n",
370                        ctxt->loc_obd->obd_name, cathandle->lgh_last_idx,
371                        PFID(&cathandle->lgh_id.lgl_oi.oi_fid), rc);
372
373         RETURN(rc);
374 }
375
376 static int
377 mdd_changelog_write_header(const struct lu_env *env, struct mdd_device *mdd,
378                            int markerflags);
379
380 static int
381 mdd_changelog_on(const struct lu_env *env, struct mdd_device *mdd)
382 {
383         int rc = 0;
384
385         if ((mdd->mdd_cl.mc_flags & CLM_ON) != 0)
386                 return rc;
387
388         LCONSOLE_INFO("%s: changelog on\n", mdd2obd_dev(mdd)->obd_name);
389         if (mdd->mdd_cl.mc_flags & CLM_ERR) {
390                 CERROR("Changelogs cannot be enabled due to error "
391                        "condition (see %s log).\n",
392                        mdd2obd_dev(mdd)->obd_name);
393                 rc = -ESRCH;
394         } else {
395                 spin_lock(&mdd->mdd_cl.mc_lock);
396                 mdd->mdd_cl.mc_flags |= CLM_ON;
397                 spin_unlock(&mdd->mdd_cl.mc_lock);
398                 rc = mdd_changelog_write_header(env, mdd, CLM_START);
399         }
400         return rc;
401 }
402
403 static int
404 mdd_changelog_off(const struct lu_env *env, struct mdd_device *mdd)
405 {
406         int rc = 0;
407
408         if ((mdd->mdd_cl.mc_flags & CLM_ON) != CLM_ON)
409                 return rc;
410
411         LCONSOLE_INFO("%s: changelog off\n", mdd2obd_dev(mdd)->obd_name);
412         rc = mdd_changelog_write_header(env, mdd, CLM_FINI);
413         spin_lock(&mdd->mdd_cl.mc_lock);
414         mdd->mdd_cl.mc_flags &= ~CLM_ON;
415         spin_unlock(&mdd->mdd_cl.mc_lock);
416
417         return rc;
418 }
419
420 static int mdd_changelog_llog_init(const struct lu_env *env,
421                                    struct mdd_device *mdd)
422 {
423         struct obd_device       *obd = mdd2obd_dev(mdd);
424         struct llog_ctxt        *ctxt = NULL, *uctxt = NULL;
425         struct changelog_orphan_data changelog_orphan = { .index = 0,
426                                                           .mdd = mdd },
427                                      user_orphan = { .index = 0,
428                                                      .mdd = mdd };
429         int                      rc;
430
431         ENTRY;
432
433         /* LU-2844 mdd setup failure should not cause umount oops */
434         if (OBD_FAIL_CHECK(OBD_FAIL_MDS_CHANGELOG_INIT))
435                 RETURN(-EIO);
436
437         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
438         obd->obd_lvfs_ctxt.dt = mdd->mdd_bottom;
439         rc = llog_setup(env, obd, &obd->obd_olg, LLOG_CHANGELOG_ORIG_CTXT,
440                         obd, &llog_common_cat_ops);
441         if (rc) {
442                 CERROR("%s: changelog llog setup failed: rc = %d\n",
443                        obd->obd_name, rc);
444                 RETURN(rc);
445         }
446
447         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
448         LASSERT(ctxt);
449
450         rc = llog_open_create(env, ctxt, &ctxt->loc_handle, NULL,
451                               CHANGELOG_CATALOG);
452         if (rc)
453                 GOTO(out_cleanup, rc);
454
455         rc = llog_init_handle(env, ctxt->loc_handle, LLOG_F_IS_CAT, NULL);
456         if (rc)
457                 GOTO(out_close, rc);
458
459         rc = llog_cat_reverse_process(env, ctxt->loc_handle,
460                                       changelog_init_cb, mdd);
461
462         if (rc < 0) {
463                 CERROR("%s: changelog init failed: rc = %d\n", obd->obd_name,
464                        rc);
465                 GOTO(out_close, rc);
466         }
467
468         CDEBUG(D_IOCTL, "changelog starting index=%llu\n",
469                mdd->mdd_cl.mc_index);
470
471         /* setup user changelog */
472         rc = llog_setup(env, obd, &obd->obd_olg, LLOG_CHANGELOG_USER_ORIG_CTXT,
473                         obd, &llog_common_cat_ops);
474         if (rc) {
475                 CERROR("%s: changelog users llog setup failed: rc = %d\n",
476                        obd->obd_name, rc);
477                 GOTO(out_close, rc);
478         }
479
480         uctxt = llog_get_context(obd, LLOG_CHANGELOG_USER_ORIG_CTXT);
481         LASSERT(ctxt);
482
483         rc = llog_open_create(env, uctxt, &uctxt->loc_handle, NULL,
484                               CHANGELOG_USERS);
485         if (rc)
486                 GOTO(out_ucleanup, rc);
487
488         rc = llog_init_handle(env, uctxt->loc_handle, LLOG_F_IS_CAT, NULL);
489         if (rc)
490                 GOTO(out_uclose, rc);
491
492         rc = llog_cat_reverse_process(env, uctxt->loc_handle,
493                                       changelog_user_init_cb, mdd);
494         if (rc < 0) {
495                 CERROR("%s: changelog user init failed: rc = %d\n",
496                        obd->obd_name, rc);
497                 GOTO(out_uclose, rc);
498         }
499
500         /* If we have registered users, assume we want changelogs on */
501         if (mdd->mdd_cl.mc_lastuser > 0) {
502                 rc = mdd_changelog_on(env, mdd);
503                 if (rc < 0)
504                         GOTO(out_uclose, rc);
505         }
506
507         /* find and clear any orphan changelog records (1st record index <
508          * smallest of all users current index), likely to come from an
509          * interrupted manual or GC-thread purge, as its user record had
510          * been deleted first
511          * XXX we may wait for a still registered user clear operation to
512          * do the job, but it may then take a long time to reach the user's
513          * real targetted records if a huge purge backlog is still to be
514          * processed as a long time idle user record could have been deleted
515          * XXX we may need to run end of purge as a separate thread
516          */
517         rc = llog_cat_process(env, ctxt->loc_handle, changelog_detect_orphan_cb,
518                               &changelog_orphan, 0, 0);
519         if (rc < 0) {
520                 CERROR("%s: changelog detect orphan failed: rc = %d\n",
521                        obd->obd_name, rc);
522                 GOTO(out_uclose, rc);
523         }
524         rc = llog_cat_process(env, uctxt->loc_handle,
525                               changelog_user_detect_orphan_cb,
526                               &user_orphan, 0, 0);
527         if (rc < 0) {
528                 CERROR("%s: changelog user detect orphan failed: rc = %d\n",
529                        obd->obd_name, rc);
530                 GOTO(out_uclose, rc);
531         }
532         if (unlikely(changelog_orphan.index < user_orphan.index)) {
533                 struct changelog_cancel_cookie cl_cookie = {
534                         .endrec = user_orphan.index,
535                         .mdd = mdd,
536                 };
537
538                 CWARN("%s : orphan changelog records found, starting from "
539                       "index %llu to index %llu, being cleared now\n",
540                       obd->obd_name, changelog_orphan.index, user_orphan.index);
541
542                 /* XXX we may need to run end of purge as a separate thread */
543                 rc = llog_changelog_cancel(env, ctxt, &cl_cookie);
544                 if (rc < 0) {
545                         CERROR("%s: purge of changelog orphan records failed: "
546                                "rc = %d\n", obd->obd_name, rc);
547                         GOTO(out_uclose, rc);
548                 }
549         }
550
551         llog_ctxt_put(ctxt);
552         llog_ctxt_put(uctxt);
553         RETURN(0);
554 out_uclose:
555         llog_cat_close(env, uctxt->loc_handle);
556 out_ucleanup:
557         llog_cleanup(env, uctxt);
558 out_close:
559         llog_cat_close(env, ctxt->loc_handle);
560 out_cleanup:
561         llog_cleanup(env, ctxt);
562         return rc;
563 }
564
565 static int mdd_changelog_init(const struct lu_env *env, struct mdd_device *mdd)
566 {
567         struct obd_device       *obd = mdd2obd_dev(mdd);
568         int                      rc;
569
570         mdd->mdd_cl.mc_index = 0;
571         spin_lock_init(&mdd->mdd_cl.mc_lock);
572         mdd->mdd_cl.mc_starttime = ktime_get();
573         spin_lock_init(&mdd->mdd_cl.mc_user_lock);
574         mdd->mdd_cl.mc_lastuser = 0;
575
576         /* ensure a GC check will, and a thread run may, occur upon start */
577         mdd->mdd_cl.mc_gc_time = 0;
578         mdd->mdd_cl.mc_gc_task = MDD_CHLG_GC_NONE;
579
580         rc = mdd_changelog_llog_init(env, mdd);
581         if (rc) {
582                 CERROR("%s: changelog setup during init failed: rc = %d\n",
583                        obd->obd_name, rc);
584                 mdd->mdd_cl.mc_flags |= CLM_ERR;
585         }
586
587         return rc;
588 }
589
590 static void mdd_changelog_fini(const struct lu_env *env,
591                                struct mdd_device *mdd)
592 {
593         struct obd_device       *obd = mdd2obd_dev(mdd);
594         struct llog_ctxt        *ctxt;
595
596         mdd->mdd_cl.mc_flags = 0;
597
598 again:
599         /* stop GC-thread if running */
600         spin_lock(&mdd->mdd_cl.mc_lock);
601         if (likely(mdd->mdd_cl.mc_gc_task == MDD_CHLG_GC_NONE)) {
602                 /* avoid any attempt to run a GC-thread */
603                 mdd->mdd_cl.mc_gc_task = current;
604                 spin_unlock(&mdd->mdd_cl.mc_lock);
605         } else {
606                 struct task_struct *gc_task;
607
608                 if (unlikely(mdd->mdd_cl.mc_gc_task == MDD_CHLG_GC_NEED ||
609                              mdd->mdd_cl.mc_gc_task == MDD_CHLG_GC_START)) {
610                         /* need to wait for birthing GC-thread to be started
611                          * and to have set mc_gc_task to itself
612                          */
613                         spin_unlock(&mdd->mdd_cl.mc_lock);
614                         schedule_timeout(usecs_to_jiffies(10));
615                         /* go back to fully check if GC-thread has started or
616                          * even already exited or if a new one is starting...
617                          */
618                         goto again;
619                 }
620                 /* take a reference on task_struct to avoid it to be freed
621                  * upon exit
622                  */
623                 gc_task = mdd->mdd_cl.mc_gc_task;
624                 get_task_struct(gc_task);
625                 spin_unlock(&mdd->mdd_cl.mc_lock);
626                 kthread_stop(gc_task);
627                 put_task_struct(gc_task);
628         }
629
630         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
631         if (ctxt) {
632                 llog_cat_close(env, ctxt->loc_handle);
633                 llog_cleanup(env, ctxt);
634         }
635         ctxt = llog_get_context(obd, LLOG_CHANGELOG_USER_ORIG_CTXT);
636         if (ctxt) {
637                 llog_cat_close(env, ctxt->loc_handle);
638                 llog_cleanup(env, ctxt);
639         }
640 }
641
642 /** Remove entries with indicies up to and including \a endrec from the
643  *  changelog
644  * \param mdd
645  * \param endrec
646  * \retval 0 ok
647  */
648 static int
649 mdd_changelog_llog_cancel(const struct lu_env *env, struct mdd_device *mdd,
650                           long long endrec)
651 {
652         struct obd_device *obd = mdd2obd_dev(mdd);
653         struct llog_ctxt *ctxt;
654         long long unsigned cur;
655         struct changelog_cancel_cookie cookie;
656         int rc;
657
658         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
659         if (ctxt == NULL)
660                 return -ENXIO;
661
662         spin_lock(&mdd->mdd_cl.mc_lock);
663         cur = (long long)mdd->mdd_cl.mc_index;
664         spin_unlock(&mdd->mdd_cl.mc_lock);
665         if (endrec > cur)
666                 endrec = cur;
667
668         /* purge to "0" is shorthand for everything */
669         if (endrec == 0)
670                 endrec = cur;
671
672         /* If purging all records, write a header entry so we don't have an
673            empty catalog and we're sure to have a valid starting index next
674            time.  In case of crash, we just restart with old log so we're
675            allright. */
676         if (endrec == cur) {
677                 /* XXX: transaction is started by llog itself */
678                 rc = mdd_changelog_write_header(env, mdd, CLM_PURGE);
679                 if (rc)
680                       goto out;
681         }
682
683         /* Some records were purged, so reset repeat-access time (so we
684            record new mtime update records, so users can see a file has been
685            changed since the last purge) */
686         mdd->mdd_cl.mc_starttime = ktime_get();
687
688         cookie.endrec = endrec;
689         cookie.mdd = mdd;
690         rc = llog_changelog_cancel(env, ctxt, &cookie);
691 out:
692         llog_ctxt_put(ctxt);
693         return rc;
694 }
695
696 /** Add a CL_MARK record to the changelog
697  * \param mdd
698  * \param markerflags - CLM_*
699  * \retval 0 ok
700  */
701 int mdd_changelog_write_header(const struct lu_env *env,
702                                struct mdd_device *mdd, int markerflags)
703 {
704         struct obd_device               *obd = mdd2obd_dev(mdd);
705         struct llog_changelog_rec       *rec;
706         struct lu_buf                   *buf;
707         struct llog_ctxt                *ctxt;
708         int                              reclen;
709         int                              len = strlen(obd->obd_name);
710         int                              rc;
711
712         ENTRY;
713
714         if (mdd->mdd_cl.mc_mask & (1 << CL_MARK)) {
715                 mdd->mdd_cl.mc_starttime = ktime_get();
716                 RETURN(0);
717         }
718
719         reclen = llog_data_len(sizeof(*rec) + len);
720         buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf, reclen);
721         if (buf->lb_buf == NULL)
722                 RETURN(-ENOMEM);
723         rec = buf->lb_buf;
724
725         rec->cr.cr_flags = CLF_VERSION;
726         rec->cr.cr_type = CL_MARK;
727         rec->cr.cr_namelen = len;
728         memcpy(changelog_rec_name(&rec->cr), obd->obd_name, rec->cr.cr_namelen);
729         /* Status and action flags */
730         rec->cr.cr_markerflags = mdd->mdd_cl.mc_flags | markerflags;
731         rec->cr_hdr.lrh_len = llog_data_len(changelog_rec_size(&rec->cr) +
732                                             rec->cr.cr_namelen);
733         rec->cr_hdr.lrh_type = CHANGELOG_REC;
734         rec->cr.cr_time = cl_time();
735         spin_lock(&mdd->mdd_cl.mc_lock);
736         rec->cr.cr_index = ++mdd->mdd_cl.mc_index;
737         spin_unlock(&mdd->mdd_cl.mc_lock);
738
739         ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT);
740         LASSERT(ctxt);
741
742         rc = llog_cat_add(env, ctxt->loc_handle, &rec->cr_hdr, NULL);
743         if (rc > 0)
744                 rc = 0;
745         llog_ctxt_put(ctxt);
746
747         /* assume on or off event; reset repeat-access time */
748         mdd->mdd_cl.mc_starttime = ktime_get();
749         RETURN(rc);
750 }
751
752 /**
753  * Lookup method for "fid" object. Only filenames with correct SEQ:OID format
754  * are valid. We also check if object with passed fid exists or not.
755  */
756 static int obf_lookup(const struct lu_env *env, struct md_object *p,
757                       const struct lu_name *lname, struct lu_fid *f,
758                       struct md_op_spec *spec)
759 {
760         char *name = (char *)lname->ln_name;
761         struct mdd_device *mdd = mdo2mdd(p);
762         struct mdd_object *child;
763         int rc = 0;
764
765         while (*name == '[')
766                 name++;
767
768         sscanf(name, SFID, RFID(f));
769         if (!fid_is_sane(f)) {
770                 CWARN("%s: Trying to lookup invalid FID [%s] in %s/%s, FID "
771                       "format should be "DFID"\n", mdd2obd_dev(mdd)->obd_name,
772                       lname->ln_name, dot_lustre_name, mdd_obf_dir_name,
773                       (__u64)FID_SEQ_NORMAL, 1, 0);
774                 GOTO(out, rc = -EINVAL);
775         }
776
777         if (!fid_is_norm(f) && !fid_is_igif(f) && !fid_is_root(f) &&
778             !fid_seq_is_dot(f->f_seq)) {
779                 CWARN("%s: Trying to lookup invalid FID "DFID" in %s/%s, "
780                       "sequence should be >= %#llx or within [%#llx,"
781                       "%#llx].\n", mdd2obd_dev(mdd)->obd_name, PFID(f),
782                       dot_lustre_name, mdd_obf_dir_name, (__u64)FID_SEQ_NORMAL,
783                       (__u64)FID_SEQ_IGIF, (__u64)FID_SEQ_IGIF_MAX);
784                 GOTO(out, rc = -EINVAL);
785         }
786
787         /* Check if object with this fid exists */
788         child = mdd_object_find(env, mdd, f);
789         if (IS_ERR(child))
790                 GOTO(out, rc = PTR_ERR(child));
791
792         if (mdd_object_exists(child) == 0)
793                 rc = -ENOENT;
794
795         mdd_object_put(env, child);
796
797 out:
798         return rc;
799 }
800
801 static int mdd_dummy_create(const struct lu_env *env,
802                             struct md_object *pobj,
803                             const struct lu_name *lname,
804                             struct md_object *child,
805                             struct md_op_spec *spec,
806                             struct md_attr* ma)
807 {
808         return -EPERM;
809 }
810
811 static int mdd_dummy_rename(const struct lu_env *env,
812                             struct md_object *src_pobj,
813                             struct md_object *tgt_pobj,
814                             const struct lu_fid *lf,
815                             const struct lu_name *lsname,
816                             struct md_object *tobj,
817                             const struct lu_name *ltname,
818                             struct md_attr *ma)
819 {
820         return -EPERM;
821 }
822
823 static int mdd_dummy_link(const struct lu_env *env,
824                           struct md_object *tgt_obj,
825                           struct md_object *src_obj,
826                           const struct lu_name *lname,
827                           struct md_attr *ma)
828 {
829         return -EPERM;
830 }
831
832 static int mdd_dummy_unlink(const struct lu_env *env,
833                             struct md_object *pobj,
834                             struct md_object *cobj,
835                             const struct lu_name *lname,
836                             struct md_attr *ma,
837                             int no_name)
838 {
839         return -EPERM;
840 }
841
842 int mdd_create(const struct lu_env *env, struct md_object *pobj,
843                       const struct lu_name *lname, struct md_object *child,
844                       struct md_op_spec *spec, struct md_attr *ma);
845 static int mdd_obf_create(const struct lu_env *env, struct md_object *pobj,
846                       const struct lu_name *lname, struct md_object *child,
847                       struct md_op_spec *spec, struct md_attr *ma)
848 {
849         if (spec->sp_cr_flags & MDS_OPEN_VOLATILE)
850                 return mdd_create(env, pobj, lname, child, spec, ma);
851         RETURN(-EPERM);
852 }
853
854 static struct md_dir_operations mdd_obf_dir_ops = {
855         .mdo_lookup = obf_lookup,
856         .mdo_create = mdd_obf_create,
857         .mdo_rename = mdd_dummy_rename,
858         .mdo_link   = mdd_dummy_link,
859         .mdo_unlink = mdd_dummy_unlink
860 };
861
862 static struct md_dir_operations mdd_lpf_dir_ops = {
863         .mdo_lookup = mdd_lookup,
864         .mdo_create = mdd_dummy_create,
865         .mdo_rename = mdd_dummy_rename,
866         .mdo_link   = mdd_dummy_link,
867         .mdo_unlink = mdd_dummy_unlink
868 };
869
870 static struct md_object *mdo_locate(const struct lu_env *env,
871                                     struct md_device *md,
872                                     const struct lu_fid *fid)
873 {
874         struct lu_object *obj;
875         struct md_object *mdo;
876
877         obj = lu_object_find(env, &md->md_lu_dev, fid, NULL);
878         if (!IS_ERR(obj)) {
879                 obj = lu_object_locate(obj->lo_header, md->md_lu_dev.ld_type);
880                 LASSERT(obj != NULL);
881                 mdo = lu2md(obj);
882         } else {
883                 mdo = ERR_PTR(PTR_ERR(obj));
884         }
885         return mdo;
886 }
887
888 static int mdd_lpf_setup(const struct lu_env *env, struct mdd_device *m)
889 {
890         struct md_object        *mdo;
891         struct mdd_object       *mdd_lpf;
892         struct lu_fid            fid    = LU_LPF_FID;
893         int                      rc;
894         ENTRY;
895
896         rc = mdd_local_file_create(env, m, mdd_object_fid(m->mdd_dot_lustre),
897                                    mdd_lpf_dir_name, S_IFDIR | S_IRUSR | S_IXUSR,
898                                    &fid);
899         if (rc != 0)
900                 RETURN(rc);
901
902         mdo = mdo_locate(env, &m->mdd_md_dev, &fid);
903         if (IS_ERR(mdo))
904                 RETURN(PTR_ERR(mdo));
905
906         LASSERT(lu_object_exists(&mdo->mo_lu));
907
908         mdd_lpf = md2mdd_obj(mdo);
909         mdd_lpf->mod_obj.mo_dir_ops = &mdd_lpf_dir_ops;
910         m->mdd_dot_lustre_objs.mdd_lpf = mdd_lpf;
911
912         RETURN(0);
913 }
914
915 /**
916  * Create special in-memory "fid" object for open-by-fid.
917  */
918 static int mdd_obf_setup(const struct lu_env *env, struct mdd_device *m)
919 {
920         struct md_object        *mdo;
921         struct mdd_object       *mdd_obf;
922         struct lu_fid            fid = LU_OBF_FID;
923         int                      rc;
924
925         rc = mdd_local_file_create(env, m, mdd_object_fid(m->mdd_dot_lustre),
926                                    mdd_obf_dir_name, S_IFDIR | S_IXUSR, &fid);
927         if (rc < 0)
928                 RETURN(rc);
929
930         mdo = mdo_locate(env, &m->mdd_md_dev, &fid);
931         if (IS_ERR(mdo))
932                 RETURN(PTR_ERR(mdo));
933
934         LASSERT(lu_object_exists(&mdo->mo_lu));
935
936         mdd_obf = md2mdd_obj(mdo);
937         mdd_obf->mod_obj.mo_dir_ops = &mdd_obf_dir_ops;
938         m->mdd_dot_lustre_objs.mdd_obf = mdd_obf;
939
940         return 0;
941 }
942
943 static void mdd_dot_lustre_cleanup(const struct lu_env *env,
944                                    struct mdd_device *m)
945 {
946         if (m->mdd_dot_lustre_objs.mdd_lpf != NULL) {
947                 mdd_object_put(env, m->mdd_dot_lustre_objs.mdd_lpf);
948                 m->mdd_dot_lustre_objs.mdd_lpf = NULL;
949         }
950         if (m->mdd_dot_lustre_objs.mdd_obf != NULL) {
951                 mdd_object_put(env, m->mdd_dot_lustre_objs.mdd_obf);
952                 m->mdd_dot_lustre_objs.mdd_obf = NULL;
953         }
954         if (m->mdd_dot_lustre != NULL) {
955                 mdd_object_put(env, m->mdd_dot_lustre);
956                 m->mdd_dot_lustre = NULL;
957         }
958 }
959
960 /** Setup ".lustre" directory object */
961 static int mdd_dot_lustre_setup(const struct lu_env *env, struct mdd_device *m)
962 {
963         struct md_object        *mdo;
964         struct lu_fid            fid;
965         int                      rc;
966
967         ENTRY;
968         /* Create ".lustre" directory in ROOT. */
969         fid = LU_DOT_LUSTRE_FID;
970         rc = mdd_local_file_create(env, m, &m->mdd_root_fid,
971                                    dot_lustre_name,
972                                    S_IFDIR | S_IRUGO | S_IWUSR | S_IXUGO,
973                                    &fid);
974         if (rc < 0)
975                 RETURN(rc);
976         mdo = mdo_locate(env, &m->mdd_md_dev, &fid);
977         if (IS_ERR(mdo))
978                 RETURN(PTR_ERR(mdo));
979         LASSERT(lu_object_exists(&mdo->mo_lu));
980
981         m->mdd_dot_lustre = md2mdd_obj(mdo);
982
983         rc = mdd_obf_setup(env, m);
984         if (rc) {
985                 CERROR("%s: error initializing \"fid\" object: rc = %d.\n",
986                        mdd2obd_dev(m)->obd_name, rc);
987                 GOTO(out, rc);
988         }
989
990         rc = mdd_lpf_setup(env, m);
991         if (rc != 0) {
992                 CERROR("%s: error initializing \"lost+found\": rc = %d.\n",
993                        mdd2obd_dev(m)->obd_name, rc);
994                 GOTO(out, rc);
995         }
996
997         RETURN(0);
998
999 out:
1000         mdd_dot_lustre_cleanup(env, m);
1001
1002         return rc;
1003 }
1004
1005 /**
1006  * set llog methods and create LLOG_AGENT_ORIG_CTXT llog
1007  * object in obd_device
1008  */
1009 static int mdd_hsm_actions_llog_init(const struct lu_env *env,
1010                                      struct mdd_device *m)
1011 {
1012         struct obd_device       *obd = mdd2obd_dev(m);
1013         struct llog_ctxt        *ctxt = NULL;
1014         int                      rc;
1015         ENTRY;
1016
1017         OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt);
1018         obd->obd_lvfs_ctxt.dt = m->mdd_bottom;
1019
1020         rc = llog_setup(env, obd, &obd->obd_olg, LLOG_AGENT_ORIG_CTXT,
1021                         obd, &llog_common_cat_ops);
1022         if (rc) {
1023                 CERROR("%s: hsm actions llog setup failed: rc = %d\n",
1024                         obd->obd_name, rc);
1025                 RETURN(rc);
1026         }
1027
1028         ctxt = llog_get_context(obd, LLOG_AGENT_ORIG_CTXT);
1029         LASSERT(ctxt);
1030
1031         rc = llog_open_create(env, ctxt, &ctxt->loc_handle, NULL,
1032                               HSM_ACTIONS);
1033         if (rc) {
1034                 CERROR("%s: hsm actions llog open_create failed: rc = %d\n",
1035                         obd->obd_name, rc);
1036                 GOTO(out_cleanup, rc);
1037         }
1038
1039         rc = llog_init_handle(env, ctxt->loc_handle, LLOG_F_IS_CAT, NULL);
1040         if (rc)
1041                 GOTO(out_close, rc);
1042
1043         llog_ctxt_put(ctxt);
1044         RETURN(0);
1045
1046 out_close:
1047         llog_cat_close(env, ctxt->loc_handle);
1048         ctxt->loc_handle = NULL;
1049 out_cleanup:
1050         llog_cleanup(env, ctxt);
1051
1052         return rc;
1053 }
1054
1055 /**
1056  * cleanup the context created by llog_setup_named()
1057  */
1058 static int mdd_hsm_actions_llog_fini(const struct lu_env *env,
1059                                      struct mdd_device *m)
1060 {
1061         struct obd_device       *obd = mdd2obd_dev(m);
1062         struct llog_ctxt        *lctxt;
1063         ENTRY;
1064
1065         lctxt = llog_get_context(obd, LLOG_AGENT_ORIG_CTXT);
1066         if (lctxt) {
1067                 llog_cat_close(env, lctxt->loc_handle);
1068                 lctxt->loc_handle = NULL;
1069                 llog_cleanup(env, lctxt);
1070         }
1071
1072         RETURN(0);
1073 }
1074
1075 static void mdd_device_shutdown(const struct lu_env *env, struct mdd_device *m,
1076                                 struct lustre_cfg *cfg)
1077 {
1078         barrier_deregister(m->mdd_bottom);
1079         lfsck_degister(env, m->mdd_bottom);
1080         mdd_hsm_actions_llog_fini(env, m);
1081         mdd_changelog_fini(env, m);
1082         mdd_orphan_index_fini(env, m);
1083         mdd_dot_lustre_cleanup(env, m);
1084         if (mdd2obd_dev(m)->u.obt.obt_nodemap_config_file) {
1085                 nm_config_file_deregister_tgt(env,
1086                                 mdd2obd_dev(m)->u.obt.obt_nodemap_config_file);
1087                 mdd2obd_dev(m)->u.obt.obt_nodemap_config_file = NULL;
1088         }
1089         if (m->mdd_los != NULL) {
1090                 local_oid_storage_fini(env, m->mdd_los);
1091                 m->mdd_los = NULL;
1092         }
1093         lu_site_purge(env, mdd2lu_dev(m)->ld_site, ~0);
1094
1095         if (m->mdd_child_exp)
1096                 obd_disconnect(m->mdd_child_exp);
1097 }
1098
1099 static int mdd_process_config(const struct lu_env *env,
1100                               struct lu_device *d, struct lustre_cfg *cfg)
1101 {
1102         struct mdd_device *m    = lu2mdd_dev(d);
1103         struct dt_device  *dt   = m->mdd_child;
1104         struct lu_device  *next = &dt->dd_lu_dev;
1105         int rc;
1106         ENTRY;
1107
1108         switch (cfg->lcfg_command) {
1109         case LCFG_PARAM: {
1110                 ssize_t count;
1111
1112                 count = class_modify_config(cfg, PARAM_MDD, &m->mdd_kobj);
1113                 rc = count > 0 ? 0 : count;
1114                 if (rc)
1115                         /* we don't understand; pass it on */
1116                         rc = next->ld_ops->ldo_process_config(env, next, cfg);
1117                 break;
1118         }
1119         case LCFG_SETUP:
1120                 rc = next->ld_ops->ldo_process_config(env, next, cfg);
1121                 if (rc)
1122                         GOTO(out, rc);
1123                 dt_conf_get(env, dt, &m->mdd_dt_conf);
1124                 break;
1125         case LCFG_PRE_CLEANUP:
1126                 rc = next->ld_ops->ldo_process_config(env, next, cfg);
1127                 mdd_generic_thread_stop(&m->mdd_orphan_cleanup_thread);
1128                 break;
1129         case LCFG_CLEANUP:
1130                 rc = next->ld_ops->ldo_process_config(env, next, cfg);
1131                 lu_dev_del_linkage(d->ld_site, d);
1132                 mdd_device_shutdown(env, m, cfg);
1133                 break;
1134         default:
1135                 rc = next->ld_ops->ldo_process_config(env, next, cfg);
1136                 break;
1137         }
1138 out:
1139         RETURN(rc);
1140 }
1141
1142 static int mdd_recovery_complete(const struct lu_env *env,
1143                                  struct lu_device *d)
1144 {
1145         struct mdd_device *mdd = lu2mdd_dev(d);
1146         struct lu_device *next;
1147         int rc;
1148         ENTRY;
1149
1150         LASSERT(mdd != NULL);
1151         next = &mdd->mdd_child->dd_lu_dev;
1152
1153         if (!mdd->mdd_bottom->dd_rdonly)
1154                 mdd_orphan_cleanup(env, mdd);
1155         rc = next->ld_ops->ldo_recovery_complete(env, next);
1156
1157         RETURN(rc);
1158 }
1159
1160 int mdd_local_file_create(const struct lu_env *env, struct mdd_device *mdd,
1161                           const struct lu_fid *pfid, const char *name,
1162                           __u32 mode, struct lu_fid *fid)
1163 {
1164         struct dt_object *parent, *dto;
1165         int rc = 0;
1166
1167         ENTRY;
1168
1169         LASSERT(!fid_is_zero(pfid));
1170         parent = dt_locate(env, mdd->mdd_bottom, pfid);
1171         if (unlikely(IS_ERR(parent)))
1172                 RETURN(PTR_ERR(parent));
1173
1174         /* create local file/dir, if @fid is passed then try to use it */
1175         if (fid_is_zero(fid))
1176                 dto = local_file_find_or_create(env, mdd->mdd_los, parent,
1177                                                 name, mode);
1178         else
1179                 dto = local_file_find_or_create_with_fid(env, mdd->mdd_bottom,
1180                                                          fid, parent, name,
1181                                                          mode);
1182         if (IS_ERR(dto))
1183                 GOTO(out_put, rc = PTR_ERR(dto));
1184         *fid = *lu_object_fid(&dto->do_lu);
1185         /* since stack is not fully set up the local_storage uses own stack
1186          * and we should drop its object from cache */
1187         dt_object_put_nocache(env, dto);
1188         EXIT;
1189 out_put:
1190         dt_object_put(env, parent);
1191         return rc;
1192 }
1193
1194 static int mdd_lfsck_out_notify(const struct lu_env *env, void *data,
1195                                 enum lfsck_events event)
1196 {
1197         return 0;
1198 }
1199
1200 static int mdd_prepare(const struct lu_env *env,
1201                        struct lu_device *pdev,
1202                        struct lu_device *cdev)
1203 {
1204         struct mdd_device *mdd = lu2mdd_dev(cdev);
1205         struct lu_device *next = &mdd->mdd_child->dd_lu_dev;
1206         struct nm_config_file *nodemap_config;
1207         struct obd_device_target *obt = &mdd2obd_dev(mdd)->u.obt;
1208         struct lu_fid fid;
1209         int rc;
1210
1211         ENTRY;
1212
1213         rc = next->ld_ops->ldo_prepare(env, cdev, next);
1214         if (rc)
1215                 RETURN(rc);
1216
1217         /* Setup local dirs */
1218         fid.f_seq = FID_SEQ_LOCAL_NAME;
1219         fid.f_oid = 1;
1220         fid.f_ver = 0;
1221         rc = local_oid_storage_init(env, mdd->mdd_bottom, &fid,
1222                                     &mdd->mdd_los);
1223         if (rc)
1224                 RETURN(rc);
1225
1226         rc = dt_root_get(env, mdd->mdd_child, &mdd->mdd_local_root_fid);
1227         if (rc < 0)
1228                 GOTO(out_los, rc);
1229
1230         lu_root_fid(&fid);
1231         if (mdd_seq_site(mdd)->ss_node_id == 0) {
1232                 rc = mdd_local_file_create(env, mdd, &mdd->mdd_local_root_fid,
1233                                            mdd_root_dir_name, S_IFDIR |
1234                                            S_IRUGO | S_IWUSR | S_IXUGO, &fid);
1235                 if (rc != 0) {
1236                         CERROR("%s: create root fid failed: rc = %d\n",
1237                                mdd2obd_dev(mdd)->obd_name, rc);
1238                         GOTO(out_los, rc);
1239                 }
1240
1241                 mdd->mdd_root_fid = fid;
1242                 rc = mdd_dot_lustre_setup(env, mdd);
1243                 if (rc != 0) {
1244                         CERROR("%s: initializing .lustre failed: rc = %d\n",
1245                                mdd2obd_dev(mdd)->obd_name, rc);
1246                         GOTO(out_los, rc);
1247                 }
1248         } else {
1249                 /* Normal client usually send root access to MDT0 directly,
1250                  * the root FID on non-MDT0 will only be used by echo client. */
1251                 mdd->mdd_root_fid = fid;
1252         }
1253
1254         rc = mdd_orphan_index_init(env, mdd);
1255         if (rc < 0)
1256                 GOTO(out_dot, rc);
1257
1258         rc = mdd_changelog_init(env, mdd);
1259         if (rc != 0) {
1260                 CERROR("%s: failed to initialize changelog: rc = %d\n",
1261                        mdd2obd_dev(mdd)->obd_name, rc);
1262                 GOTO(out_orph, rc);
1263         }
1264
1265         rc = mdd_hsm_actions_llog_init(env, mdd);
1266         if (rc != 0)
1267                 GOTO(out_changelog, rc);
1268
1269         nodemap_config = nm_config_file_register_tgt(env, mdd->mdd_bottom,
1270                                                      mdd->mdd_los);
1271         if (IS_ERR(nodemap_config)) {
1272                 rc = PTR_ERR(nodemap_config);
1273                 if (rc != -EROFS)
1274                         GOTO(out_hsm, rc);
1275         } else {
1276                 obt->obt_nodemap_config_file = nodemap_config;
1277         }
1278
1279         rc = lfsck_register(env, mdd->mdd_bottom, mdd->mdd_child,
1280                             mdd2obd_dev(mdd), mdd_lfsck_out_notify,
1281                             mdd, true);
1282         if (rc != 0) {
1283                 CERROR("%s: failed to initialize lfsck: rc = %d\n",
1284                        mdd2obd_dev(mdd)->obd_name, rc);
1285                 GOTO(out_nodemap, rc);
1286         }
1287
1288         rc = barrier_register(mdd->mdd_bottom, mdd->mdd_child);
1289         if (rc) {
1290                 CERROR("%s: failed to register to barrier: rc = %d\n",
1291                        mdd2obd_dev(mdd)->obd_name, rc);
1292                 GOTO(out_lfsck, rc);
1293         }
1294
1295         RETURN(0);
1296
1297 out_lfsck:
1298         lfsck_degister(env, mdd->mdd_bottom);
1299 out_nodemap:
1300         nm_config_file_deregister_tgt(env, obt->obt_nodemap_config_file);
1301         obt->obt_nodemap_config_file = NULL;
1302 out_hsm:
1303         mdd_hsm_actions_llog_fini(env, mdd);
1304 out_changelog:
1305         mdd_changelog_fini(env, mdd);
1306 out_orph:
1307         mdd_orphan_index_fini(env, mdd);
1308 out_dot:
1309         if (mdd_seq_site(mdd)->ss_node_id == 0)
1310                 mdd_dot_lustre_cleanup(env, mdd);
1311 out_los:
1312         local_oid_storage_fini(env, mdd->mdd_los);
1313         mdd->mdd_los = NULL;
1314
1315         return rc;
1316 }
1317
1318 const struct lu_device_operations mdd_lu_ops = {
1319         .ldo_object_alloc      = mdd_object_alloc,
1320         .ldo_process_config    = mdd_process_config,
1321         .ldo_recovery_complete = mdd_recovery_complete,
1322         .ldo_prepare           = mdd_prepare,
1323 };
1324
1325 static int mdd_root_get(const struct lu_env *env,
1326                         struct md_device *m, struct lu_fid *f)
1327 {
1328         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
1329
1330         ENTRY;
1331         *f = mdd->mdd_root_fid;
1332         RETURN(0);
1333 }
1334
1335 /*
1336  * No permission check is needed.
1337  */
1338 static int mdd_statfs(const struct lu_env *env, struct md_device *m,
1339                       struct obd_statfs *sfs)
1340 {
1341         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
1342         int rc;
1343
1344         ENTRY;
1345
1346         rc = mdd_child_ops(mdd)->dt_statfs(env, mdd->mdd_child, sfs, NULL);
1347
1348         sfs->os_namelen = min_t(__u32, sfs->os_namelen, NAME_MAX);
1349
1350         RETURN(rc);
1351 }
1352
1353 static const struct dt_device_param *mdd_dtconf_get(const struct lu_env *env,
1354                                                     struct md_device *m)
1355 {
1356         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
1357
1358         return &mdd->mdd_dt_conf;
1359 }
1360
1361 static int mdd_llog_ctxt_get(const struct lu_env *env, struct md_device *m,
1362                              int idx, void **h)
1363 {
1364         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
1365
1366         *h = llog_group_get_ctxt(&mdd2obd_dev(mdd)->obd_olg, idx);
1367         return (*h == NULL ? -ENOENT : 0);
1368 }
1369
1370 static struct lu_device *mdd_device_free(const struct lu_env *env,
1371                                          struct lu_device *lu)
1372 {
1373         struct mdd_device *m = lu2mdd_dev(lu);
1374         ENTRY;
1375
1376         LASSERT(atomic_read(&lu->ld_ref) == 0);
1377         md_device_fini(&m->mdd_md_dev);
1378         OBD_FREE_PTR(m);
1379         RETURN(NULL);
1380 }
1381
1382 static struct lu_device *mdd_device_alloc(const struct lu_env *env,
1383                                           struct lu_device_type *t,
1384                                           struct lustre_cfg *lcfg)
1385 {
1386         struct lu_device  *l;
1387         struct mdd_device *m;
1388
1389         OBD_ALLOC_PTR(m);
1390         if (m == NULL) {
1391                 l = ERR_PTR(-ENOMEM);
1392         } else {
1393                 int rc;
1394
1395                 l = mdd2lu_dev(m);
1396                 md_device_init(&m->mdd_md_dev, t);
1397                 rc = mdd_init0(env, m, t, lcfg);
1398                 if (rc != 0) {
1399                         mdd_device_free(env, l);
1400                         l = ERR_PTR(rc);
1401                 }
1402         }
1403
1404         return l;
1405 }
1406
1407 /*
1408  * we use exports to track all mdd users
1409  */
1410 static int mdd_obd_connect(const struct lu_env *env, struct obd_export **exp,
1411                            struct obd_device *obd, struct obd_uuid *cluuid,
1412                            struct obd_connect_data *data, void *localdata)
1413 {
1414         struct mdd_device    *mdd = lu2mdd_dev(obd->obd_lu_dev);
1415         struct lustre_handle  conn;
1416         int                   rc;
1417         ENTRY;
1418
1419         CDEBUG(D_CONFIG, "connect #%d\n", mdd->mdd_connects);
1420
1421         rc = class_connect(&conn, obd, cluuid);
1422         if (rc)
1423                 RETURN(rc);
1424
1425         *exp = class_conn2export(&conn);
1426
1427         /* Why should there ever be more than 1 connect? */
1428         LASSERT(mdd->mdd_connects == 0);
1429         mdd->mdd_connects++;
1430
1431         RETURN(0);
1432 }
1433
1434 /*
1435  * once last export (we don't count self-export) disappeared
1436  * mdd can be released
1437  */
1438 static int mdd_obd_disconnect(struct obd_export *exp)
1439 {
1440         struct obd_device *obd = exp->exp_obd;
1441         struct mdd_device *mdd = lu2mdd_dev(obd->obd_lu_dev);
1442         int                rc, release = 0;
1443         ENTRY;
1444
1445         mdd->mdd_connects--;
1446         if (mdd->mdd_connects == 0)
1447                 release = 1;
1448
1449         rc = class_disconnect(exp);
1450
1451         if (rc == 0 && release)
1452                 class_manual_cleanup(obd);
1453         RETURN(rc);
1454 }
1455
1456 static int mdd_obd_get_info(const struct lu_env *env, struct obd_export *exp,
1457                             __u32 keylen, void *key, __u32 *vallen, void *val)
1458 {
1459         int rc = -EINVAL;
1460
1461         if (KEY_IS(KEY_OSP_CONNECTED)) {
1462                 struct obd_device       *obd = exp->exp_obd;
1463                 struct mdd_device       *mdd;
1464
1465                 if (!obd->obd_set_up || obd->obd_stopping)
1466                         RETURN(-EAGAIN);
1467
1468                 mdd = lu2mdd_dev(obd->obd_lu_dev);
1469                 LASSERT(mdd);
1470                 rc = obd_get_info(env, mdd->mdd_child_exp, keylen, key, vallen,
1471                                   val);
1472                 RETURN(rc);
1473         }
1474
1475         RETURN(rc);
1476 }
1477
1478 static int mdd_obd_set_info_async(const struct lu_env *env,
1479                                   struct obd_export *exp,
1480                                   __u32 keylen, void *key,
1481                                   __u32 vallen, void *val,
1482                                   struct ptlrpc_request_set *set)
1483 {
1484         struct obd_device       *obd = exp->exp_obd;
1485         struct mdd_device       *mdd;
1486         int                      rc;
1487
1488         if (!obd->obd_set_up || obd->obd_stopping)
1489                 RETURN(-EAGAIN);
1490
1491         mdd = lu2mdd_dev(obd->obd_lu_dev);
1492         LASSERT(mdd);
1493         rc = obd_set_info_async(env, mdd->mdd_child_exp, keylen, key,
1494                                 vallen, val, set);
1495         RETURN(rc);
1496 }
1497
1498 static struct obd_ops mdd_obd_device_ops = {
1499         .o_owner        = THIS_MODULE,
1500         .o_connect      = mdd_obd_connect,
1501         .o_disconnect   = mdd_obd_disconnect,
1502         .o_get_info     = mdd_obd_get_info,
1503         .o_set_info_async = mdd_obd_set_info_async,
1504 };
1505
1506 static int mdd_changelog_user_register(const struct lu_env *env,
1507                                        struct mdd_device *mdd, int *id)
1508 {
1509         struct llog_ctxt *ctxt;
1510         struct llog_changelog_user_rec *rec;
1511         int rc;
1512         ENTRY;
1513
1514         ctxt = llog_get_context(mdd2obd_dev(mdd),
1515                                 LLOG_CHANGELOG_USER_ORIG_CTXT);
1516         if (ctxt == NULL)
1517                 RETURN(-ENXIO);
1518
1519         OBD_ALLOC_PTR(rec);
1520         if (rec == NULL) {
1521                 llog_ctxt_put(ctxt);
1522                 RETURN(-ENOMEM);
1523         }
1524
1525         CFS_RACE(CFS_FAIL_CHLOG_USER_REG_UNREG_RACE);
1526
1527         rec->cur_hdr.lrh_len = sizeof(*rec);
1528         rec->cur_hdr.lrh_type = CHANGELOG_USER_REC;
1529         spin_lock(&mdd->mdd_cl.mc_user_lock);
1530         if (mdd->mdd_cl.mc_lastuser == (unsigned int)(-1)) {
1531                 spin_unlock(&mdd->mdd_cl.mc_user_lock);
1532                 CERROR("Maximum number of changelog users exceeded!\n");
1533                 GOTO(out, rc = -EOVERFLOW);
1534         }
1535         *id = rec->cur_id = ++mdd->mdd_cl.mc_lastuser;
1536         mdd->mdd_cl.mc_users++;
1537         rec->cur_endrec = mdd->mdd_cl.mc_index;
1538
1539         rec->cur_time = (__u32)ktime_get_real_seconds();
1540         if (OBD_FAIL_CHECK(OBD_FAIL_TIME_IN_CHLOG_USER))
1541                 rec->cur_time = 0;
1542
1543         spin_unlock(&mdd->mdd_cl.mc_user_lock);
1544
1545         rc = llog_cat_add(env, ctxt->loc_handle, &rec->cur_hdr, NULL);
1546         if (rc) {
1547                 CWARN("%s: Failed to register changelog user %d: rc=%d\n",
1548                       mdd2obd_dev(mdd)->obd_name, *id, rc);
1549                 spin_lock(&mdd->mdd_cl.mc_user_lock);
1550                 mdd->mdd_cl.mc_users--;
1551                 spin_unlock(&mdd->mdd_cl.mc_user_lock);
1552                 GOTO(out, rc);
1553         }
1554
1555         CDEBUG(D_IOCTL, "Registered changelog user %d\n", *id);
1556
1557         /* Assume we want it on since somebody registered */
1558         rc = mdd_changelog_on(env, mdd);
1559         if (rc)
1560                 GOTO(out, rc);
1561
1562 out:
1563         OBD_FREE_PTR(rec);
1564         llog_ctxt_put(ctxt);
1565         RETURN(rc);
1566 }
1567
1568 struct mdd_changelog_user_purge {
1569         struct mdd_device *mcup_mdd;
1570         __u32 mcup_id;
1571         __u32 mcup_usercount;
1572         __u64 mcup_minrec;
1573         bool mcup_found;
1574 };
1575
1576 /**
1577  * changelog_user_purge callback
1578  *
1579  * Is called once per user.
1580  *
1581  * Check to see the user requested is available from rec.
1582  * Truncate the changelog.
1583  * Keep track of the total number of users (calls).
1584  */
1585 static int mdd_changelog_user_purge_cb(const struct lu_env *env,
1586                                        struct llog_handle *llh,
1587                                        struct llog_rec_hdr *hdr, void *data)
1588 {
1589         struct llog_changelog_user_rec  *rec;
1590         struct mdd_changelog_user_purge *mcup = data;
1591         struct llog_cookie               cookie;
1592         int                              rc;
1593
1594         ENTRY;
1595
1596         if ((llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN) == 0)
1597                 RETURN(-ENXIO);
1598
1599         rec = container_of(hdr, struct llog_changelog_user_rec, cur_hdr);
1600
1601         mcup->mcup_usercount++;
1602
1603         if (rec->cur_id != mcup->mcup_id) {
1604                 /* truncate to the lowest endrec that is not this user */
1605                 mcup->mcup_minrec = min(mcup->mcup_minrec,
1606                                         rec->cur_endrec);
1607                 RETURN(0);
1608         }
1609
1610         /* Unregister this user */
1611         cookie.lgc_lgl = llh->lgh_id;
1612         cookie.lgc_index = hdr->lrh_index;
1613
1614         rc = llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle,
1615                                      1, &cookie);
1616         if (rc == 0) {
1617                 mcup->mcup_found = true;
1618                 mcup->mcup_usercount--;
1619                 spin_lock(&mcup->mcup_mdd->mdd_cl.mc_user_lock);
1620                 mcup->mcup_mdd->mdd_cl.mc_users--;
1621                 spin_unlock(&mcup->mcup_mdd->mdd_cl.mc_user_lock);
1622         }
1623
1624         RETURN(rc);
1625 }
1626
1627 int mdd_changelog_user_purge(const struct lu_env *env,
1628                              struct mdd_device *mdd, __u32 id)
1629 {
1630         struct mdd_changelog_user_purge mcup = {
1631                 .mcup_mdd = mdd,
1632                 .mcup_id = id,
1633                 .mcup_found = false,
1634                 .mcup_usercount = 0,
1635                 .mcup_minrec = ULLONG_MAX,
1636         };
1637         struct llog_ctxt *ctxt;
1638         int rc;
1639
1640         ENTRY;
1641
1642         CDEBUG(D_IOCTL, "%s: Purge request: id=%u\n",
1643                mdd2obd_dev(mdd)->obd_name, id);
1644
1645         ctxt = llog_get_context(mdd2obd_dev(mdd),
1646                                 LLOG_CHANGELOG_USER_ORIG_CTXT);
1647         if (ctxt == NULL ||
1648             (ctxt->loc_handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT) == 0)
1649                 GOTO(out, rc = -ENXIO);
1650
1651         rc = llog_cat_process(env, ctxt->loc_handle,
1652                               mdd_changelog_user_purge_cb, &mcup,
1653                               0, 0);
1654
1655         OBD_FAIL_TIMEOUT(OBD_FAIL_LLOG_PURGE_DELAY, cfs_fail_val);
1656
1657         if ((rc == 0) && (mcup.mcup_usercount == 0)) {
1658                 spin_lock(&mdd->mdd_cl.mc_user_lock);
1659                 if (mdd->mdd_cl.mc_users == 0) {
1660                         /* No more users; turn changelogs off */
1661                         CDEBUG(D_IOCTL, "turning off changelogs\n");
1662                         rc = mdd_changelog_off(env, mdd);
1663                 }
1664                 spin_unlock(&mdd->mdd_cl.mc_user_lock);
1665         }
1666
1667         if ((rc == 0) && mcup.mcup_found) {
1668                 CDEBUG(D_IOCTL, "%s: Purging changelog entries for user %d "
1669                        "record=%llu\n",
1670                        mdd2obd_dev(mdd)->obd_name, id, mcup.mcup_minrec);
1671                 /* Cancelling record 0 destroys the entire changelog, make sure
1672                    we don't do that unless we mean it. */
1673                 if (mcup.mcup_minrec != 0 || mcup.mcup_usercount == 0) {
1674                         rc = mdd_changelog_llog_cancel(env, mdd,
1675                                                        mcup.mcup_minrec);
1676                 }
1677         } else {
1678                 CWARN("%s: No changelog for user %u; rc=%d\n",
1679                       mdd2obd_dev(mdd)->obd_name, id, rc);
1680                 GOTO(out, rc = -ENOENT);
1681         }
1682
1683         CFS_RACE(CFS_FAIL_CHLOG_USER_REG_UNREG_RACE);
1684
1685         EXIT;
1686 out:
1687         if (ctxt != NULL)
1688                 llog_ctxt_put(ctxt);
1689
1690         return rc;
1691 }
1692
1693 struct mdd_changelog_user_clear {
1694         __u64 mcuc_endrec;
1695         __u64 mcuc_minrec;
1696         __u32 mcuc_id;
1697         bool mcuc_flush;
1698 };
1699
1700 /**
1701  * changelog_clear callback
1702  *
1703  * Is called once per user.
1704  *
1705  * Check to see the user requested is available from rec.
1706  * Check the oldest (smallest) record for boundary conditions.
1707  * Truncate the changelog.
1708  */
1709 static int mdd_changelog_clear_cb(const struct lu_env *env,
1710                                   struct llog_handle *llh,
1711                                   struct llog_rec_hdr *hdr,
1712                                   void *data)
1713 {
1714         struct llog_changelog_user_rec *rec;
1715         struct mdd_changelog_user_clear *mcuc = data;
1716         int rc;
1717
1718         ENTRY;
1719
1720         if ((llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN) == 0)
1721                 RETURN(-ENXIO);
1722
1723         rec = container_of(hdr, struct llog_changelog_user_rec, cur_hdr);
1724
1725         /* Does the changelog id match the requested id? */
1726         if (rec->cur_id != mcuc->mcuc_id) {
1727                 mcuc->mcuc_minrec = min(mcuc->mcuc_minrec,
1728                                         rec->cur_endrec);
1729                 RETURN(0);
1730         }
1731
1732         /* cur_endrec is the oldest purgeable record, make sure we're newer */
1733         if (rec->cur_endrec > mcuc->mcuc_endrec) {
1734                 CDEBUG(D_IOCTL, "Request %llu out of range: %llu\n",
1735                        mcuc->mcuc_endrec, rec->cur_endrec);
1736                 RETURN(-EINVAL);
1737         }
1738
1739         /* Flag that we've met all the range and user checks.
1740          * We now know the record to flush.
1741          */
1742         rec->cur_endrec = mcuc->mcuc_endrec;
1743
1744         rec->cur_time = (__u32)ktime_get_real_seconds();
1745         if (OBD_FAIL_CHECK(OBD_FAIL_TIME_IN_CHLOG_USER))
1746                 rec->cur_time = 0;
1747
1748         mcuc->mcuc_flush = true;
1749
1750         CDEBUG(D_IOCTL, "Rewriting changelog user %u endrec to %llu\n",
1751                mcuc->mcuc_id, rec->cur_endrec);
1752
1753         /* Update the endrec */
1754         rc = llog_write(env, llh, hdr, hdr->lrh_index);
1755
1756         RETURN(rc);
1757 }
1758
1759 /**
1760  * Clear a changelog up to entry specified by endrec for user id.
1761  */
1762 static int mdd_changelog_clear(const struct lu_env *env,
1763                                struct mdd_device *mdd, __u32 id,
1764                                __u64 endrec)
1765 {
1766         struct mdd_changelog_user_clear mcuc = {
1767                 .mcuc_id = id,
1768                 .mcuc_minrec = endrec,
1769                 .mcuc_flush = false,
1770         };
1771         struct llog_ctxt *ctxt;
1772         __u64 start_rec;
1773         int rc;
1774
1775         ENTRY;
1776
1777         CDEBUG(D_IOCTL, "%s: Purge request: id=%u, endrec=%llu\n",
1778                mdd2obd_dev(mdd)->obd_name, id, endrec);
1779         /* start_rec is the newest (largest value) entry in the changelogs*/
1780         spin_lock(&mdd->mdd_cl.mc_lock);
1781         start_rec = mdd->mdd_cl.mc_index;
1782         spin_unlock(&mdd->mdd_cl.mc_lock);
1783
1784         if (start_rec < endrec) {
1785                 CDEBUG(D_IOCTL, "%s: Could not clear changelog, requested "\
1786                        "address out of range\n", mdd2obd_dev(mdd)->obd_name);
1787                 RETURN(-EINVAL);
1788         }
1789
1790         if (endrec == 0) {
1791                 mcuc.mcuc_endrec = start_rec;
1792                 mcuc.mcuc_minrec = ULLONG_MAX;
1793         } else {
1794                 mcuc.mcuc_endrec = endrec;
1795         }
1796
1797         ctxt = llog_get_context(mdd2obd_dev(mdd),
1798                                 LLOG_CHANGELOG_USER_ORIG_CTXT);
1799         if (ctxt == NULL ||
1800             (ctxt->loc_handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT) == 0)
1801                 GOTO(out, rc = -ENXIO);
1802
1803         rc = llog_cat_process(env, ctxt->loc_handle,
1804                               mdd_changelog_clear_cb, (void *)&mcuc,
1805                               0, 0);
1806
1807         if (rc < 0) {
1808                 CWARN("%s: Failure to clear the changelog for user %d: %d\n",
1809                       mdd2obd_dev(mdd)->obd_name, id, rc);
1810         } else if (mcuc.mcuc_flush) {
1811                 /* Cancelling record 0 destroys the entire changelog, make sure
1812                    we don't do that unless we mean it. */
1813                 if (mcuc.mcuc_minrec != 0) {
1814                         CDEBUG(D_IOCTL, "%s: Purging changelog entries up "\
1815                                "to %llu\n", mdd2obd_dev(mdd)->obd_name,
1816                                mcuc.mcuc_minrec);
1817
1818                         rc = mdd_changelog_llog_cancel(env, mdd,
1819                                                       mcuc.mcuc_minrec);
1820                 }
1821         } else {
1822                 CWARN("%s: No entry for user %d\n",
1823                       mdd2obd_dev(mdd)->obd_name, id);
1824                 rc = -ENOENT;
1825         }
1826
1827         EXIT;
1828 out:
1829         if (ctxt != NULL)
1830                 llog_ctxt_put(ctxt);
1831
1832         return rc;
1833 }
1834
1835 /** mdd_iocontrol
1836  * May be called remotely from mdt_iocontrol_handle or locally from
1837  * mdt_iocontrol. Data may be freeform - remote handling doesn't enforce
1838  * an obd_ioctl_data format (but local ioctl handler does).
1839  * \param cmd - ioc
1840  * \param len - data len
1841  * \param karg - ioctl data, in kernel space
1842  */
1843 static int mdd_iocontrol(const struct lu_env *env, struct md_device *m,
1844                          unsigned int cmd, int len, void *karg)
1845 {
1846         struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
1847         struct obd_ioctl_data *data = karg;
1848         int rc;
1849         ENTRY;
1850
1851         /* Doesn't use obd_ioctl_data */
1852         switch (cmd) {
1853         case OBD_IOC_CHANGELOG_CLEAR: {
1854                 struct changelog_setinfo *cs = karg;
1855
1856                 if (unlikely(!barrier_entry(mdd->mdd_bottom)))
1857                         RETURN(-EINPROGRESS);
1858
1859                 rc = mdd_changelog_clear(env, mdd, cs->cs_id,
1860                                          cs->cs_recno);
1861                 barrier_exit(mdd->mdd_bottom);
1862                 RETURN(rc);
1863         }
1864         case OBD_IOC_START_LFSCK: {
1865                 rc = lfsck_start(env, mdd->mdd_bottom,
1866                                  (struct lfsck_start_param *)karg);
1867                 RETURN(rc);
1868         }
1869         case OBD_IOC_STOP_LFSCK: {
1870                 rc = lfsck_stop(env, mdd->mdd_bottom,
1871                                 (struct lfsck_stop *)karg);
1872                 RETURN(rc);
1873         }
1874         case OBD_IOC_QUERY_LFSCK: {
1875                 rc = lfsck_query(env, mdd->mdd_bottom, NULL, NULL,
1876                                  (struct lfsck_query *)karg);
1877                 RETURN(rc);
1878         }
1879         }
1880
1881         /* Below ioctls use obd_ioctl_data */
1882         if (len != sizeof(*data)) {
1883                 CERROR("Bad ioctl size %d\n", len);
1884                 RETURN(-EINVAL);
1885         }
1886         if (data->ioc_version != OBD_IOCTL_VERSION) {
1887                 CERROR("Bad magic %x != %x\n", data->ioc_version,
1888                        OBD_IOCTL_VERSION);
1889                 RETURN(-EINVAL);
1890         }
1891
1892         switch (cmd) {
1893         case OBD_IOC_CHANGELOG_REG:
1894                 if (unlikely(!barrier_entry(mdd->mdd_bottom)))
1895                         RETURN(-EINPROGRESS);
1896
1897                 rc = mdd_changelog_user_register(env, mdd, &data->ioc_u32_1);
1898                 barrier_exit(mdd->mdd_bottom);
1899                 break;
1900         case OBD_IOC_CHANGELOG_DEREG:
1901                 if (unlikely(!barrier_entry(mdd->mdd_bottom)))
1902                         RETURN(-EINPROGRESS);
1903
1904                 /* explicitly clear changelog first, to protect from crash in
1905                  * the middle of purge that would lead to unregistered consumer
1906                  * but pending changelog entries
1907                  */
1908                 rc = mdd_changelog_clear(env, mdd, data->ioc_u32_1, 0);
1909                 if (!rc)
1910                         rc = mdd_changelog_user_purge(env,
1911                                                       mdd, data->ioc_u32_1);
1912
1913                 barrier_exit(mdd->mdd_bottom);
1914                 break;
1915         default:
1916                 rc = -ENOTTY;
1917         }
1918
1919         RETURN(rc);
1920 }
1921
1922 /* type constructor/destructor: mdd_type_init, mdd_type_fini */
1923 LU_TYPE_INIT_FINI(mdd, &mdd_thread_key);
1924
1925 static const struct md_device_operations mdd_ops = {
1926         .mdo_statfs         = mdd_statfs,
1927         .mdo_root_get       = mdd_root_get,
1928         .mdo_llog_ctxt_get  = mdd_llog_ctxt_get,
1929         .mdo_iocontrol      = mdd_iocontrol,
1930         .mdo_dtconf_get     = mdd_dtconf_get,
1931 };
1932
1933 static struct lu_device_type_operations mdd_device_type_ops = {
1934         .ldto_init = mdd_type_init,
1935         .ldto_fini = mdd_type_fini,
1936
1937         .ldto_start = mdd_type_start,
1938         .ldto_stop  = mdd_type_stop,
1939
1940         .ldto_device_alloc = mdd_device_alloc,
1941         .ldto_device_free  = mdd_device_free,
1942
1943         .ldto_device_fini    = mdd_device_fini
1944 };
1945
1946 static struct lu_device_type mdd_device_type = {
1947         .ldt_tags     = LU_DEVICE_MD,
1948         .ldt_name     = LUSTRE_MDD_NAME,
1949         .ldt_ops      = &mdd_device_type_ops,
1950         .ldt_ctx_tags = LCT_MD_THREAD
1951 };
1952
1953 /* context key constructor: mdd_key_init */
1954 LU_KEY_INIT(mdd, struct mdd_thread_info);
1955
1956 static void mdd_key_fini(const struct lu_context *ctx,
1957                          struct lu_context_key *key, void *data)
1958 {
1959         struct mdd_thread_info *info = data;
1960
1961         lu_buf_free(&info->mti_big_buf);
1962         lu_buf_free(&info->mti_link_buf);
1963         lu_buf_free(&info->mti_xattr_buf);
1964
1965         OBD_FREE_PTR(info);
1966 }
1967
1968 /* context key: mdd_thread_key */
1969 LU_CONTEXT_KEY_DEFINE(mdd, LCT_MD_THREAD);
1970
1971 int mdd_generic_thread_start(struct mdd_generic_thread *thread,
1972                              int (*func)(void *), void *data, char *name)
1973 {
1974         struct task_struct      *task;
1975
1976         LASSERT(thread->mgt_init == false);
1977         init_completion(&thread->mgt_started);
1978         init_completion(&thread->mgt_finished);
1979         thread->mgt_data = data;
1980         thread->mgt_abort = false;
1981         thread->mgt_init = true;
1982
1983         task = kthread_run(func, thread, name);
1984         if (IS_ERR(task)) {
1985                 complete(&thread->mgt_finished);
1986                 return PTR_ERR(task);
1987         }
1988         wait_for_completion(&thread->mgt_started);
1989         return 0;
1990 }
1991
1992 void mdd_generic_thread_stop(struct mdd_generic_thread *thread)
1993 {
1994         if (thread->mgt_init == true) {
1995                 thread->mgt_abort = true;
1996                 wait_for_completion(&thread->mgt_finished);
1997         }
1998 }
1999
2000 static int __init mdd_init(void)
2001 {
2002         int rc;
2003
2004         rc = lu_kmem_init(mdd_caches);
2005         if (rc)
2006                 return rc;
2007
2008         rc = class_register_type(&mdd_obd_device_ops, NULL, false, NULL,
2009                                  LUSTRE_MDD_NAME, &mdd_device_type);
2010         if (rc)
2011                 lu_kmem_fini(mdd_caches);
2012         return rc;
2013 }
2014
2015 static void __exit mdd_exit(void)
2016 {
2017         class_unregister_type(LUSTRE_MDD_NAME);
2018         lu_kmem_fini(mdd_caches);
2019 }
2020
2021 MODULE_AUTHOR("OpenSFS, Inc. <http://www.lustre.org/>");
2022 MODULE_DESCRIPTION("Lustre Meta-data Device Driver ("LUSTRE_MDD_NAME")");
2023 MODULE_VERSION(LUSTRE_VERSION_STRING);
2024 MODULE_LICENSE("GPL");
2025
2026 module_init(mdd_init);
2027 module_exit(mdd_exit);