X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdd%2Fmdd_device.c;h=acb5d37cb81ffb7574ace83c329147ffedf5088c;hp=e9d57c58530113ce431c555fefa5f2dd57f62710;hb=5c883ea2748ae9e430a9cd863a9b630b2a74440a;hpb=79d06b6fac3d18f2e5755a940e8afa42e70f3ba2 diff --git a/lustre/mdd/mdd_device.c b/lustre/mdd/mdd_device.c index e9d57c5..acb5d37 100644 --- a/lustre/mdd/mdd_device.c +++ b/lustre/mdd/mdd_device.c @@ -15,11 +15,7 @@ * * You should have received a copy of the GNU General Public License * version 2 along with this program; If not, see - * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf - * - * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara, - * CA 95054 USA or visit www.sun.com if you need additional information or - * have any questions. + * http://www.gnu.org/licenses/gpl-2.0.html * * GPL HEADER END */ @@ -27,7 +23,7 @@ * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved. * Use is subject to license terms. * - * Copyright (c) 2011, 2012, Intel Corporation. + * Copyright (c) 2011, 2017, Intel Corporation. */ /* * This file is part of Lustre, http://www.lustre.org/ @@ -43,23 +39,28 @@ #define DEBUG_SUBSYSTEM S_MDS #include +#include #include +#include #include #include #include -#include +#include #include +#include +#include #include "mdd_internal.h" -const struct md_device_operations mdd_ops; +static const struct md_device_operations mdd_ops; static struct lu_device_type mdd_device_type; static const char mdd_root_dir_name[] = "ROOT"; static const char mdd_obf_dir_name[] = "fid"; +static const char mdd_lpf_dir_name[] = "lost+found"; /* Slab for MDD object allocation */ -cfs_mem_cache_t *mdd_object_kmem; +struct kmem_cache *mdd_object_kmem; static struct lu_kmem_descr mdd_caches[] = { { @@ -117,19 +118,45 @@ out: static int mdd_init0(const struct lu_env *env, struct mdd_device *mdd, struct lu_device_type *t, struct lustre_cfg *lcfg) { - int rc; + int rc = -EINVAL; + const char *dev; ENTRY; + /* LU-8040 Set defaults here, before values configs */ + mdd->mdd_cl.mc_flags = 0; /* off by default */ + mdd->mdd_cl.mc_mask = CHANGELOG_DEFMASK; + mdd->mdd_cl.mc_deniednext = 60; /* 60 secs by default */ + + dev = lustre_cfg_string(lcfg, 0); + if (dev == NULL) + RETURN(rc); + + mdd->mdd_md_dev.md_lu_dev.ld_obd = class_name2obd(dev); + if (mdd->mdd_md_dev.md_lu_dev.ld_obd == NULL) + RETURN(rc); mdd->mdd_md_dev.md_lu_dev.ld_ops = &mdd_lu_ops; mdd->mdd_md_dev.md_ops = &mdd_ops; rc = mdd_connect_to_next(env, mdd, lustre_cfg_string(lcfg, 3)); - if (rc) + if (rc != 0) RETURN(rc); mdd->mdd_atime_diff = MAX_ATIME_DIFF; - /* sync permission changes */ - mdd->mdd_sync_permission = 1; + /* sync permission changes */ + mdd->mdd_sync_permission = 1; + /* enable changelog garbage collection */ + mdd->mdd_changelog_gc = 1; + /* with a significant amount of idle time */ + mdd->mdd_changelog_max_idle_time = CHLOG_MAX_IDLE_TIME; + /* or a significant amount of late indexes */ + mdd->mdd_changelog_max_idle_indexes = CHLOG_MAX_IDLE_INDEXES; + /* with a reasonable interval between each check */ + mdd->mdd_changelog_min_gc_interval = CHLOG_MIN_GC_INTERVAL; + /* with a very few number of free catalog entries */ + mdd->mdd_changelog_min_free_cat_entries = CHLOG_MIN_FREE_CAT_ENTRIES; + /* special default striping for files created with O_APPEND */ + mdd->mdd_append_stripe_count = 1; + mdd->mdd_append_pool[0] = '\0'; dt_conf_get(env, mdd->mdd_child, &mdd->mdd_dt_conf); @@ -146,94 +173,143 @@ static int mdd_init0(const struct lu_env *env, struct mdd_device *mdd, static struct lu_device *mdd_device_fini(const struct lu_env *env, struct lu_device *d) { - struct mdd_device *mdd = lu2mdd_dev(d); - int rc; + struct mdd_device *mdd = lu2mdd_dev(d); if (d->ld_site) lu_dev_del_linkage(d->ld_site, d); - rc = mdd_procfs_fini(mdd); - if (rc) { - CERROR("proc fini error %d \n", rc); - return ERR_PTR(rc); - } + mdd_procfs_fini(mdd); return NULL; } -static void mdd_changelog_fini(const struct lu_env *env, - struct mdd_device *mdd); - -static void mdd_device_shutdown(const struct lu_env *env, - struct mdd_device *m, struct lustre_cfg *cfg) -{ - ENTRY; - if (m->mdd_dot_lustre_objs.mdd_obf) - mdd_object_put(env, m->mdd_dot_lustre_objs.mdd_obf); - if (m->mdd_dot_lustre) - mdd_object_put(env, m->mdd_dot_lustre); - orph_index_fini(env, m); - if (m->mdd_capa != NULL) { - lu_object_put(env, &m->mdd_capa->do_lu); - m->mdd_capa = NULL; - } - lu_site_purge(env, m->mdd_md_dev.md_lu_dev.ld_site, -1); - /* remove upcall device*/ - md_upcall_fini(&m->mdd_md_dev); - - if (m->mdd_child_exp) - obd_disconnect(m->mdd_child_exp); - - EXIT; -} - static int changelog_init_cb(const struct lu_env *env, struct llog_handle *llh, struct llog_rec_hdr *hdr, void *data) { - struct mdd_device *mdd = (struct mdd_device *)data; - struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr; + struct mdd_device *mdd = (struct mdd_device *)data; + struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr; - LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN); - LASSERT(rec->cr_hdr.lrh_type == CHANGELOG_REC); + LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN); + LASSERT(rec->cr_hdr.lrh_type == CHANGELOG_REC); - CDEBUG(D_INFO, - "seeing record at index %d/%d/"LPU64" t=%x %.*s in log "LPX64"\n", - hdr->lrh_index, rec->cr_hdr.lrh_index, rec->cr.cr_index, - rec->cr.cr_type, rec->cr.cr_namelen, rec->cr.cr_name, - llh->lgh_id.lgl_oid); + CDEBUG(D_INFO, + "seeing record at index %d/%d/%llu t=%x %.*s in log" + DFID"\n", hdr->lrh_index, rec->cr_hdr.lrh_index, + rec->cr.cr_index, rec->cr.cr_type, rec->cr.cr_namelen, + changelog_rec_name(&rec->cr), PFID(&llh->lgh_id.lgl_oi.oi_fid)); - mdd->mdd_cl.mc_index = rec->cr.cr_index; - return LLOG_PROC_BREAK; + mdd->mdd_cl.mc_index = rec->cr.cr_index; + return LLOG_PROC_BREAK; } static int changelog_user_init_cb(const struct lu_env *env, struct llog_handle *llh, struct llog_rec_hdr *hdr, void *data) { - struct mdd_device *mdd = (struct mdd_device *)data; - struct llog_changelog_user_rec *rec = - (struct llog_changelog_user_rec *)hdr; + struct mdd_device *mdd = (struct mdd_device *)data; + struct llog_changelog_user_rec *rec = + (struct llog_changelog_user_rec *)hdr; - LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN); - LASSERT(rec->cur_hdr.lrh_type == CHANGELOG_USER_REC); + LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN); + LASSERT(rec->cur_hdr.lrh_type == CHANGELOG_USER_REC); - CDEBUG(D_INFO, "seeing user at index %d/%d id=%d endrec="LPU64 - " in log "LPX64"\n", hdr->lrh_index, rec->cur_hdr.lrh_index, - rec->cur_id, rec->cur_endrec, llh->lgh_id.lgl_oid); + CDEBUG(D_INFO, "seeing user at index %d/%d id=%d endrec=%llu" + " in log "DFID"\n", hdr->lrh_index, rec->cur_hdr.lrh_index, + rec->cur_id, rec->cur_endrec, PFID(&llh->lgh_id.lgl_oi.oi_fid)); spin_lock(&mdd->mdd_cl.mc_user_lock); mdd->mdd_cl.mc_lastuser = rec->cur_id; + mdd->mdd_cl.mc_users++; + if (rec->cur_endrec > mdd->mdd_cl.mc_index) + mdd->mdd_cl.mc_index = rec->cur_endrec; spin_unlock(&mdd->mdd_cl.mc_user_lock); return LLOG_PROC_BREAK; } +struct changelog_orphan_data { + __u64 index; + struct mdd_device *mdd; +}; + +/* find oldest changelog record index */ +static int changelog_detect_orphan_cb(const struct lu_env *env, + struct llog_handle *llh, + struct llog_rec_hdr *hdr, void *data) +{ + struct mdd_device *mdd = ((struct changelog_orphan_data *)data)->mdd; + struct llog_changelog_rec *rec = container_of(hdr, + struct llog_changelog_rec, + cr_hdr); + + LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN); + + if (rec->cr_hdr.lrh_type != CHANGELOG_REC) { + CWARN("%s: invalid record at index %d in log "DFID"\n", + mdd2obd_dev(mdd)->obd_name, hdr->lrh_index, + PFID(&llh->lgh_id.lgl_oi.oi_fid)); + /* try to find some next valid record and thus allow to recover + * from a corrupted LLOG, instead to assert and force a crash + */ + return 0; + } + + CDEBUG(D_INFO, "%s: seeing record at index %d/%d/%llu t=%x %.*s in log " + DFID"\n", mdd2obd_dev(mdd)->obd_name, hdr->lrh_index, + rec->cr_hdr.lrh_index, rec->cr.cr_index, rec->cr.cr_type, + rec->cr.cr_namelen, changelog_rec_name(&rec->cr), + PFID(&llh->lgh_id.lgl_oi.oi_fid)); + + ((struct changelog_orphan_data *)data)->index = rec->cr.cr_index; + return LLOG_PROC_BREAK; +} + +/* find oldest changelog user index */ +static int changelog_user_detect_orphan_cb(const struct lu_env *env, + struct llog_handle *llh, + struct llog_rec_hdr *hdr, void *data) +{ + struct mdd_device *mdd = ((struct changelog_orphan_data *)data)->mdd; + struct llog_changelog_user_rec *rec = container_of(hdr, + struct llog_changelog_user_rec, + cur_hdr); + + LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN); + + if (rec->cur_hdr.lrh_type != CHANGELOG_USER_REC) { + CWARN("%s: invalid user at index %d in log "DFID"\n", + mdd2obd_dev(mdd)->obd_name, hdr->lrh_index, + PFID(&llh->lgh_id.lgl_oi.oi_fid)); + /* try to find some next valid record and thus allow to recover + * from a corrupted LLOG, instead to assert and force a crash + */ + return 0; + } + + CDEBUG(D_INFO, "%s: seeing user at index %d/%d id=%d endrec=%llu in " + "log "DFID"\n", mdd2obd_dev(mdd)->obd_name, hdr->lrh_index, + rec->cur_hdr.lrh_index, rec->cur_id, rec->cur_endrec, + PFID(&llh->lgh_id.lgl_oi.oi_fid)); + + if (((struct changelog_orphan_data *)data)->index == 0 || + rec->cur_endrec < ((struct changelog_orphan_data *)data)->index) + ((struct changelog_orphan_data *)data)->index = rec->cur_endrec; + + return 0; +} + +struct changelog_cancel_cookie { + long long endrec; + struct mdd_device *mdd; +}; + static int llog_changelog_cancel_cb(const struct lu_env *env, struct llog_handle *llh, struct llog_rec_hdr *hdr, void *data) { struct llog_changelog_rec *rec = (struct llog_changelog_rec *)hdr; struct llog_cookie cookie; - long long endrec = *(long long *)data; + struct changelog_cancel_cookie *cl_cookie = + (struct changelog_cancel_cookie *)data; int rc; ENTRY; @@ -241,7 +317,26 @@ static int llog_changelog_cancel_cb(const struct lu_env *env, /* This is always a (sub)log, not the catalog */ LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN); - if (rec->cr.cr_index > endrec) + /* if current context is GC-thread allow it to stop upon umount + * remaining records cleanup will occur upon next mount + * + * also during testing, wait for GC-thread to be released + * + * XXX this requires the GC-thread to not fork a sub-thread via + * llog[_cat]_process_or_fork() and we may think to also implement + * this shutdown mechanism for manually started user unregister which + * can also take a long time if huge backlog of records + */ + if (unlikely(cl_cookie->mdd->mdd_cl.mc_gc_task == current)) { + /* wait to be released */ + while (CFS_FAIL_CHECK_QUIET(OBD_FAIL_FORCE_GC_THREAD)) + schedule(); + + if (kthread_should_stop()) + RETURN(LLOG_PROC_BREAK); + } + + if (rec->cr.cr_index > cl_cookie->endrec) /* records are in order, so we're done */ RETURN(LLOG_PROC_BREAK); @@ -258,8 +353,7 @@ static int llog_changelog_cancel_cb(const struct lu_env *env, static int llog_changelog_cancel(const struct lu_env *env, struct llog_ctxt *ctxt, - struct lov_stripe_md *lsm, int count, - struct llog_cookie *cookies, int flags) + struct changelog_cancel_cookie *cookie) { struct llog_handle *cathandle = ctxt->loc_handle; int rc; @@ -270,29 +364,81 @@ static int llog_changelog_cancel(const struct lu_env *env, LASSERT(cathandle->lgh_hdr->llh_flags & LLOG_F_IS_CAT); rc = llog_cat_process(env, cathandle, llog_changelog_cancel_cb, - (void *)cookies, 0, 0); + cookie, 0, 0); if (rc >= 0) /* 0 or 1 means we're done */ rc = 0; else - CERROR("%s: cancel idx %u of catalog "LPX64" rc=%d\n", + CERROR("%s: cancel idx %u of catalog "DFID": rc = %d\n", ctxt->loc_obd->obd_name, cathandle->lgh_last_idx, - cathandle->lgh_id.lgl_oid, rc); + PFID(&cathandle->lgh_id.lgl_oi.oi_fid), rc); RETURN(rc); } static struct llog_operations changelog_orig_logops; -int mdd_changelog_on(const struct lu_env *env, struct mdd_device *mdd, int on); +static int +mdd_changelog_write_header(const struct lu_env *env, struct mdd_device *mdd, + int markerflags); + +static int +mdd_changelog_on(const struct lu_env *env, struct mdd_device *mdd) +{ + int rc = 0; + + if ((mdd->mdd_cl.mc_flags & CLM_ON) != 0) + return rc; + + LCONSOLE_INFO("%s: changelog on\n", mdd2obd_dev(mdd)->obd_name); + if (mdd->mdd_cl.mc_flags & CLM_ERR) { + CERROR("Changelogs cannot be enabled due to error " + "condition (see %s log).\n", + mdd2obd_dev(mdd)->obd_name); + rc = -ESRCH; + } else { + spin_lock(&mdd->mdd_cl.mc_lock); + mdd->mdd_cl.mc_flags |= CLM_ON; + spin_unlock(&mdd->mdd_cl.mc_lock); + rc = mdd_changelog_write_header(env, mdd, CLM_START); + } + return rc; +} + +static int +mdd_changelog_off(const struct lu_env *env, struct mdd_device *mdd) +{ + int rc = 0; + + if ((mdd->mdd_cl.mc_flags & CLM_ON) != CLM_ON) + return rc; + + LCONSOLE_INFO("%s: changelog off\n", mdd2obd_dev(mdd)->obd_name); + rc = mdd_changelog_write_header(env, mdd, CLM_FINI); + spin_lock(&mdd->mdd_cl.mc_lock); + mdd->mdd_cl.mc_flags &= ~CLM_ON; + spin_unlock(&mdd->mdd_cl.mc_lock); + + return rc; +} static int mdd_changelog_llog_init(const struct lu_env *env, struct mdd_device *mdd) { struct obd_device *obd = mdd2obd_dev(mdd); struct llog_ctxt *ctxt = NULL, *uctxt = NULL; + struct changelog_orphan_data changelog_orphan = { .index = 0, + .mdd = mdd }, + user_orphan = { .index = 0, + .mdd = mdd }; int rc; + ENTRY; + + /* LU-2844 mdd setup failure should not cause umount oops */ + if (OBD_FAIL_CHECK(OBD_FAIL_MDS_CHANGELOG_INIT)) + RETURN(-EIO); + OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt); obd->obd_lvfs_ctxt.dt = mdd->mdd_bottom; rc = llog_setup(env, obd, &obd->obd_olg, LLOG_CHANGELOG_ORIG_CTXT, @@ -311,7 +457,7 @@ static int mdd_changelog_llog_init(const struct lu_env *env, if (rc) GOTO(out_cleanup, rc); - rc = llog_cat_init_and_process(env, ctxt->loc_handle); + rc = llog_init_handle(env, ctxt->loc_handle, LLOG_F_IS_CAT, NULL); if (rc) GOTO(out_close, rc); @@ -324,7 +470,7 @@ static int mdd_changelog_llog_init(const struct lu_env *env, GOTO(out_close, rc); } - CDEBUG(D_IOCTL, "changelog starting index="LPU64"\n", + CDEBUG(D_IOCTL, "changelog starting index=%llu\n", mdd->mdd_cl.mc_index); /* setup user changelog */ @@ -344,10 +490,7 @@ static int mdd_changelog_llog_init(const struct lu_env *env, if (rc) GOTO(out_ucleanup, rc); - uctxt->loc_handle->lgh_logops->lop_add = llog_cat_add_rec; - uctxt->loc_handle->lgh_logops->lop_declare_add = llog_cat_declare_add_rec; - - rc = llog_cat_init_and_process(env, uctxt->loc_handle); + rc = llog_init_handle(env, uctxt->loc_handle, LLOG_F_IS_CAT, NULL); if (rc) GOTO(out_uclose, rc); @@ -361,10 +504,55 @@ static int mdd_changelog_llog_init(const struct lu_env *env, /* If we have registered users, assume we want changelogs on */ if (mdd->mdd_cl.mc_lastuser > 0) { - rc = mdd_changelog_on(env, mdd, 1); + rc = mdd_changelog_on(env, mdd); if (rc < 0) GOTO(out_uclose, rc); } + + /* find and clear any orphan changelog records (1st record index < + * smallest of all users current index), likely to come from an + * interrupted manual or GC-thread purge, as its user record had + * been deleted first + * XXX we may wait for a still registered user clear operation to + * do the job, but it may then take a long time to reach the user's + * real targetted records if a huge purge backlog is still to be + * processed as a long time idle user record could have been deleted + * XXX we may need to run end of purge as a separate thread + */ + rc = llog_cat_process(env, ctxt->loc_handle, changelog_detect_orphan_cb, + &changelog_orphan, 0, 0); + if (rc < 0) { + CERROR("%s: changelog detect orphan failed: rc = %d\n", + obd->obd_name, rc); + GOTO(out_uclose, rc); + } + rc = llog_cat_process(env, uctxt->loc_handle, + changelog_user_detect_orphan_cb, + &user_orphan, 0, 0); + if (rc < 0) { + CERROR("%s: changelog user detect orphan failed: rc = %d\n", + obd->obd_name, rc); + GOTO(out_uclose, rc); + } + if (unlikely(changelog_orphan.index < user_orphan.index)) { + struct changelog_cancel_cookie cl_cookie = { + .endrec = user_orphan.index, + .mdd = mdd, + }; + + CWARN("%s : orphan changelog records found, starting from " + "index %llu to index %llu, being cleared now\n", + obd->obd_name, changelog_orphan.index, user_orphan.index); + + /* XXX we may need to run end of purge as a separate thread */ + rc = llog_changelog_cancel(env, ctxt, &cl_cookie); + if (rc < 0) { + CERROR("%s: purge of changelog orphan records failed: " + "rc = %d\n", obd->obd_name, rc); + GOTO(out_uclose, rc); + } + } + llog_ctxt_put(ctxt); llog_ctxt_put(uctxt); RETURN(0); @@ -386,12 +574,14 @@ static int mdd_changelog_init(const struct lu_env *env, struct mdd_device *mdd) mdd->mdd_cl.mc_index = 0; spin_lock_init(&mdd->mdd_cl.mc_lock); - mdd->mdd_cl.mc_starttime = cfs_time_current_64(); - mdd->mdd_cl.mc_flags = 0; /* off by default */ - mdd->mdd_cl.mc_mask = CHANGELOG_DEFMASK; + mdd->mdd_cl.mc_starttime = ktime_get(); spin_lock_init(&mdd->mdd_cl.mc_user_lock); mdd->mdd_cl.mc_lastuser = 0; + /* ensure a GC check will, and a thread run may, occur upon start */ + mdd->mdd_cl.mc_gc_time = 0; + mdd->mdd_cl.mc_gc_task = MDD_CHLG_GC_NONE; + rc = mdd_changelog_llog_init(env, mdd); if (rc) { CERROR("%s: changelog setup during init failed: rc = %d\n", @@ -410,6 +600,39 @@ static void mdd_changelog_fini(const struct lu_env *env, mdd->mdd_cl.mc_flags = 0; +again: + /* stop GC-thread if running */ + spin_lock(&mdd->mdd_cl.mc_lock); + if (likely(mdd->mdd_cl.mc_gc_task == MDD_CHLG_GC_NONE)) { + /* avoid any attempt to run a GC-thread */ + mdd->mdd_cl.mc_gc_task = current; + spin_unlock(&mdd->mdd_cl.mc_lock); + } else { + struct task_struct *gc_task; + + if (unlikely(mdd->mdd_cl.mc_gc_task == MDD_CHLG_GC_NEED || + mdd->mdd_cl.mc_gc_task == MDD_CHLG_GC_START)) { + /* need to wait for birthing GC-thread to be started + * and to have set mc_gc_task to itself + */ + spin_unlock(&mdd->mdd_cl.mc_lock); + /* Add a tiny sleep */ + schedule_timeout_uninterruptible(1); + /* go back to fully check if GC-thread has started or + * even already exited or if a new one is starting... + */ + goto again; + } + /* take a reference on task_struct to avoid it to be freed + * upon exit + */ + gc_task = mdd->mdd_cl.mc_gc_task; + get_task_struct(gc_task); + spin_unlock(&mdd->mdd_cl.mc_lock); + kthread_stop(gc_task); + put_task_struct(gc_task); + } + ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT); if (ctxt) { llog_cat_close(env, ctxt->loc_handle); @@ -422,49 +645,20 @@ static void mdd_changelog_fini(const struct lu_env *env, } } -int mdd_changelog_write_header(const struct lu_env *env, - struct mdd_device *mdd, int markerflags); - -/* Start / stop recording */ -int mdd_changelog_on(const struct lu_env *env, struct mdd_device *mdd, int on) -{ - int rc = 0; - - if ((on == 1) && ((mdd->mdd_cl.mc_flags & CLM_ON) == 0)) { - LCONSOLE_INFO("%s: changelog on\n", mdd2obd_dev(mdd)->obd_name); - if (mdd->mdd_cl.mc_flags & CLM_ERR) { - CERROR("Changelogs cannot be enabled due to error " - "condition (see %s log).\n", - mdd2obd_dev(mdd)->obd_name); - rc = -ESRCH; - } else { - spin_lock(&mdd->mdd_cl.mc_lock); - mdd->mdd_cl.mc_flags |= CLM_ON; - spin_unlock(&mdd->mdd_cl.mc_lock); - rc = mdd_changelog_write_header(env, mdd, CLM_START); - } - } else if ((on == 0) && ((mdd->mdd_cl.mc_flags & CLM_ON) == CLM_ON)) { - LCONSOLE_INFO("%s: changelog off\n",mdd2obd_dev(mdd)->obd_name); - rc = mdd_changelog_write_header(env, mdd, CLM_FINI); - spin_lock(&mdd->mdd_cl.mc_lock); - mdd->mdd_cl.mc_flags &= ~CLM_ON; - spin_unlock(&mdd->mdd_cl.mc_lock); - } - return rc; -} - /** Remove entries with indicies up to and including \a endrec from the * changelog * \param mdd * \param endrec * \retval 0 ok */ -int mdd_changelog_llog_cancel(const struct lu_env *env, - struct mdd_device *mdd, long long endrec) +static int +mdd_changelog_llog_cancel(const struct lu_env *env, struct mdd_device *mdd, + long long endrec) { struct obd_device *obd = mdd2obd_dev(mdd); struct llog_ctxt *ctxt; long long unsigned cur; + struct changelog_cancel_cookie cookie; int rc; ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT); @@ -495,9 +689,11 @@ int mdd_changelog_llog_cancel(const struct lu_env *env, /* Some records were purged, so reset repeat-access time (so we record new mtime update records, so users can see a file has been changed since the last purge) */ - mdd->mdd_cl.mc_starttime = cfs_time_current_64(); + mdd->mdd_cl.mc_starttime = ktime_get(); - rc = llog_cancel(env, ctxt, NULL, 1, (struct llog_cookie *)&endrec, 0); + cookie.endrec = endrec; + cookie.mdd = mdd; + rc = llog_changelog_cancel(env, ctxt, &cookie); out: llog_ctxt_put(ctxt); return rc; @@ -522,12 +718,12 @@ int mdd_changelog_write_header(const struct lu_env *env, ENTRY; if (mdd->mdd_cl.mc_mask & (1 << CL_MARK)) { - mdd->mdd_cl.mc_starttime = cfs_time_current_64(); + mdd->mdd_cl.mc_starttime = ktime_get(); RETURN(0); } reclen = llog_data_len(sizeof(*rec) + len); - buf = mdd_buf_alloc(env, reclen); + buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf, reclen); if (buf->lb_buf == NULL) RETURN(-ENOMEM); rec = buf->lb_buf; @@ -535,571 +731,372 @@ int mdd_changelog_write_header(const struct lu_env *env, rec->cr.cr_flags = CLF_VERSION; rec->cr.cr_type = CL_MARK; rec->cr.cr_namelen = len; - memcpy(rec->cr.cr_name, obd->obd_name, rec->cr.cr_namelen); + memcpy(changelog_rec_name(&rec->cr), obd->obd_name, rec->cr.cr_namelen); /* Status and action flags */ rec->cr.cr_markerflags = mdd->mdd_cl.mc_flags | markerflags; - rec->cr_hdr.lrh_len = llog_data_len(sizeof(*rec) + rec->cr.cr_namelen); + rec->cr_hdr.lrh_len = llog_data_len(changelog_rec_size(&rec->cr) + + rec->cr.cr_namelen); rec->cr_hdr.lrh_type = CHANGELOG_REC; rec->cr.cr_time = cl_time(); - spin_lock(&mdd->mdd_cl.mc_lock); - rec->cr.cr_index = ++mdd->mdd_cl.mc_index; - spin_unlock(&mdd->mdd_cl.mc_lock); ctxt = llog_get_context(obd, LLOG_CHANGELOG_ORIG_CTXT); LASSERT(ctxt); - rc = llog_cat_add(env, ctxt->loc_handle, &rec->cr_hdr, NULL, NULL); + rc = llog_cat_add(env, ctxt->loc_handle, &rec->cr_hdr, NULL); if (rc > 0) rc = 0; llog_ctxt_put(ctxt); /* assume on or off event; reset repeat-access time */ - mdd->mdd_cl.mc_starttime = cfs_time_current_64(); + mdd->mdd_cl.mc_starttime = ktime_get(); RETURN(rc); } /** - * Create ".lustre" directory. + * Lookup method for "fid" object. Only filenames with correct SEQ:OID format + * are valid. We also check if object with passed fid exists or not. */ -static int create_dot_lustre_dir(const struct lu_env *env, struct mdd_device *m) +static int obf_lookup(const struct lu_env *env, struct md_object *p, + const struct lu_name *lname, struct lu_fid *f, + struct md_op_spec *spec) { - struct lu_fid *fid = &mdd_env_info(env)->mti_fid; - struct md_object *mdo; - int rc; + char *name = (char *)lname->ln_name; + struct mdd_device *mdd = mdo2mdd(p); + struct mdd_object *child; + int rc = 0; - memcpy(fid, &LU_DOT_LUSTRE_FID, sizeof(struct lu_fid)); - mdo = llo_store_create_index(env, &m->mdd_md_dev, m->mdd_child, - mdd_root_dir_name, dot_lustre_name, - fid, &dt_directory_features); - /* .lustre dir may be already present */ - if (IS_ERR(mdo) && PTR_ERR(mdo) != -EEXIST) { - rc = PTR_ERR(mdo); - CERROR("creating obj [%s] fid = "DFID" rc = %d\n", - dot_lustre_name, PFID(fid), rc); - return rc; + while (*name == '[') + name++; + + sscanf(name, SFID, RFID(f)); + if (!fid_is_sane(f)) { + CWARN("%s: Trying to lookup invalid FID [%s] in %s/%s, FID " + "format should be "DFID"\n", mdd2obd_dev(mdd)->obd_name, + lname->ln_name, dot_lustre_name, mdd_obf_dir_name, + (__u64)FID_SEQ_NORMAL, 1, 0); + GOTO(out, rc = -EINVAL); } - if (!IS_ERR(mdo)) - lu_object_put(env, &mdo->mo_lu); + if (!fid_is_norm(f) && !fid_is_igif(f) && !fid_is_root(f) && + !fid_seq_is_dot(f->f_seq)) { + CWARN("%s: Trying to lookup invalid FID "DFID" in %s/%s, " + "sequence should be >= %#llx or within [%#llx," + "%#llx].\n", mdd2obd_dev(mdd)->obd_name, PFID(f), + dot_lustre_name, mdd_obf_dir_name, (__u64)FID_SEQ_NORMAL, + (__u64)FID_SEQ_IGIF, (__u64)FID_SEQ_IGIF_MAX); + GOTO(out, rc = -EINVAL); + } - return 0; -} + /* Check if object with this fid exists */ + child = mdd_object_find(env, mdd, f); + if (IS_ERR(child)) + GOTO(out, rc = PTR_ERR(child)); + if (mdd_object_exists(child) == 0) + rc = -ENOENT; -static int dot_lustre_mdd_permission(const struct lu_env *env, - struct md_object *pobj, - struct md_object *cobj, - struct md_attr *attr, int mask) -{ - if (mask & ~(MAY_READ | MAY_EXEC)) - return -EPERM; - else - return 0; -} + mdd_object_put(env, child); -static int dot_lustre_mdd_xattr_get(const struct lu_env *env, - struct md_object *obj, struct lu_buf *buf, - const char *name) -{ - return 0; +out: + return rc; } -static int dot_lustre_mdd_xattr_list(const struct lu_env *env, - struct md_object *obj, struct lu_buf *buf) +static int mdd_dummy_create(const struct lu_env *env, + struct md_object *pobj, + const struct lu_name *lname, + struct md_object *child, + struct md_op_spec *spec, + struct md_attr* ma) { - return 0; + return -EPERM; } -static int dot_lustre_mdd_xattr_set(const struct lu_env *env, - struct md_object *obj, - const struct lu_buf *buf, const char *name, - int fl) +static int mdd_dummy_rename(const struct lu_env *env, + struct md_object *src_pobj, + struct md_object *tgt_pobj, + const struct lu_fid *lf, + const struct lu_name *lsname, + struct md_object *tobj, + const struct lu_name *ltname, + struct md_attr *ma) { - return -EPERM; + return -EPERM; } -static int dot_lustre_mdd_xattr_del(const struct lu_env *env, - struct md_object *obj, - const char *name) +static int mdd_dummy_link(const struct lu_env *env, + struct md_object *tgt_obj, + struct md_object *src_obj, + const struct lu_name *lname, + struct md_attr *ma) { - return -EPERM; + return -EPERM; } -static int dot_lustre_mdd_swap_layouts(const struct lu_env *env, - struct md_object *obj1, - struct md_object *obj2, - __u64 flags) +static int mdd_dummy_unlink(const struct lu_env *env, + struct md_object *pobj, + struct md_object *cobj, + const struct lu_name *lname, + struct md_attr *ma, + int no_name) { return -EPERM; } -static int dot_lustre_mdd_readlink(const struct lu_env *env, - struct md_object *obj, struct lu_buf *buf) +int mdd_create(const struct lu_env *env, struct md_object *pobj, + const struct lu_name *lname, struct md_object *child, + struct md_op_spec *spec, struct md_attr *ma); +static int mdd_obf_create(const struct lu_env *env, struct md_object *pobj, + const struct lu_name *lname, struct md_object *child, + struct md_op_spec *spec, struct md_attr *ma) { - return 0; + if (spec->sp_cr_flags & MDS_OPEN_VOLATILE) + return mdd_create(env, pobj, lname, child, spec, ma); + RETURN(-EPERM); } -static int dot_lustre_mdd_object_create(const struct lu_env *env, - struct md_object *obj, - const struct md_op_spec *spec, - struct md_attr *ma) -{ - return -EPERM; -} +static struct md_dir_operations mdd_obf_dir_ops = { + .mdo_lookup = obf_lookup, + .mdo_create = mdd_obf_create, + .mdo_rename = mdd_dummy_rename, + .mdo_link = mdd_dummy_link, + .mdo_unlink = mdd_dummy_unlink +}; -static int dot_lustre_mdd_ref_add(const struct lu_env *env, - struct md_object *obj, - const struct md_attr *ma) -{ - return -EPERM; -} +static struct md_dir_operations mdd_lpf_dir_ops = { + .mdo_lookup = mdd_lookup, + .mdo_create = mdd_dummy_create, + .mdo_rename = mdd_dummy_rename, + .mdo_link = mdd_dummy_link, + .mdo_unlink = mdd_dummy_unlink +}; -static int dot_lustre_mdd_ref_del(const struct lu_env *env, - struct md_object *obj, - struct md_attr *ma) +static struct md_object *mdo_locate(const struct lu_env *env, + struct md_device *md, + const struct lu_fid *fid) { - return -EPERM; + struct lu_object *obj; + struct md_object *mdo; + + obj = lu_object_find(env, &md->md_lu_dev, fid, NULL); + if (!IS_ERR(obj)) { + obj = lu_object_locate(obj->lo_header, md->md_lu_dev.ld_type); + LASSERT(obj != NULL); + mdo = lu2md(obj); + } else { + mdo = ERR_PTR(PTR_ERR(obj)); + } + return mdo; } -static int dot_lustre_mdd_open(const struct lu_env *env, struct md_object *obj, - int flags) +static int mdd_lpf_setup(const struct lu_env *env, struct mdd_device *m) { - struct mdd_object *mdd_obj = md2mdd_obj(obj); + struct md_object *mdo; + struct mdd_object *mdd_lpf; + struct lu_fid fid = LU_LPF_FID; + int rc; + ENTRY; - mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD); - mdd_obj->mod_count++; - mdd_write_unlock(env, mdd_obj); + rc = mdd_local_file_create(env, m, mdd_object_fid(m->mdd_dot_lustre), + mdd_lpf_dir_name, S_IFDIR | S_IRUSR | S_IXUSR, + &fid); + if (rc != 0) + RETURN(rc); - return 0; -} + mdo = mdo_locate(env, &m->mdd_md_dev, &fid); + if (IS_ERR(mdo)) + RETURN(PTR_ERR(mdo)); -static int dot_lustre_mdd_close(const struct lu_env *env, struct md_object *obj, - struct md_attr *ma, int mode) -{ - struct mdd_object *mdd_obj = md2mdd_obj(obj); + LASSERT(lu_object_exists(&mdo->mo_lu)); - mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD); - mdd_obj->mod_count--; - mdd_write_unlock(env, mdd_obj); + mdd_lpf = md2mdd_obj(mdo); + mdd_lpf->mod_obj.mo_dir_ops = &mdd_lpf_dir_ops; + m->mdd_dot_lustre_objs.mdd_lpf = mdd_lpf; - return 0; + RETURN(0); } -static int dot_lustre_mdd_object_sync(const struct lu_env *env, - struct md_object *obj) +/** + * Create special in-memory "fid" object for open-by-fid. + */ +static int mdd_obf_setup(const struct lu_env *env, struct mdd_device *m) { - return -ENOSYS; -} + struct md_object *mdo; + struct mdd_object *mdd_obf; + struct lu_fid fid = LU_OBF_FID; + int rc; -static int dot_lustre_mdd_path(const struct lu_env *env, struct md_object *obj, - char *path, int pathlen, __u64 *recno, int *linkno) -{ - return -ENOSYS; + rc = mdd_local_file_create(env, m, mdd_object_fid(m->mdd_dot_lustre), + mdd_obf_dir_name, S_IFDIR | S_IXUSR, &fid); + if (rc < 0) + RETURN(rc); + + mdo = mdo_locate(env, &m->mdd_md_dev, &fid); + if (IS_ERR(mdo)) + RETURN(PTR_ERR(mdo)); + + LASSERT(lu_object_exists(&mdo->mo_lu)); + + mdd_obf = md2mdd_obj(mdo); + mdd_obf->mod_obj.mo_dir_ops = &mdd_obf_dir_ops; + m->mdd_dot_lustre_objs.mdd_obf = mdd_obf; + + return 0; } -static int dot_file_lock(const struct lu_env *env, struct md_object *obj, - struct lov_mds_md *lmm, struct ldlm_extent *extent, - struct lustre_handle *lockh) +static void mdd_dot_lustre_cleanup(const struct lu_env *env, + struct mdd_device *m) { - return -ENOSYS; + if (m->mdd_dot_lustre_objs.mdd_lpf != NULL) { + mdd_object_put(env, m->mdd_dot_lustre_objs.mdd_lpf); + m->mdd_dot_lustre_objs.mdd_lpf = NULL; + } + if (m->mdd_dot_lustre_objs.mdd_obf != NULL) { + mdd_object_put(env, m->mdd_dot_lustre_objs.mdd_obf); + m->mdd_dot_lustre_objs.mdd_obf = NULL; + } + if (m->mdd_dot_lustre != NULL) { + mdd_object_put(env, m->mdd_dot_lustre); + m->mdd_dot_lustre = NULL; + } } -static int dot_file_unlock(const struct lu_env *env, struct md_object *obj, - struct lov_mds_md *lmm, struct lustre_handle *lockh) +/** Setup ".lustre" directory object */ +static int mdd_dot_lustre_setup(const struct lu_env *env, struct mdd_device *m) { - return -ENOSYS; -} + struct md_object *mdo; + struct lu_fid fid; + int rc; -static struct md_object_operations mdd_dot_lustre_obj_ops = { - .moo_permission = dot_lustre_mdd_permission, - .moo_attr_get = mdd_attr_get, - .moo_attr_set = mdd_attr_set, - .moo_xattr_get = dot_lustre_mdd_xattr_get, - .moo_xattr_list = dot_lustre_mdd_xattr_list, - .moo_xattr_set = dot_lustre_mdd_xattr_set, - .moo_xattr_del = dot_lustre_mdd_xattr_del, - .moo_swap_layouts = dot_lustre_mdd_swap_layouts, - .moo_readpage = mdd_readpage, - .moo_readlink = dot_lustre_mdd_readlink, - .moo_object_create = dot_lustre_mdd_object_create, - .moo_ref_add = dot_lustre_mdd_ref_add, - .moo_ref_del = dot_lustre_mdd_ref_del, - .moo_open = dot_lustre_mdd_open, - .moo_close = dot_lustre_mdd_close, - .moo_capa_get = mdd_capa_get, - .moo_object_sync = dot_lustre_mdd_object_sync, - .moo_path = dot_lustre_mdd_path, - .moo_file_lock = dot_file_lock, - .moo_file_unlock = dot_file_unlock, -}; + ENTRY; + /* Create ".lustre" directory in ROOT. */ + fid = LU_DOT_LUSTRE_FID; + rc = mdd_local_file_create(env, m, &m->mdd_root_fid, + dot_lustre_name, + S_IFDIR | S_IRUGO | S_IWUSR | S_IXUGO, + &fid); + if (rc < 0) + RETURN(rc); + mdo = mdo_locate(env, &m->mdd_md_dev, &fid); + if (IS_ERR(mdo)) + RETURN(PTR_ERR(mdo)); + LASSERT(lu_object_exists(&mdo->mo_lu)); + m->mdd_dot_lustre = md2mdd_obj(mdo); -static int dot_lustre_mdd_lookup(const struct lu_env *env, struct md_object *p, - const struct lu_name *lname, struct lu_fid *f, - struct md_op_spec *spec) -{ - if (strcmp(lname->ln_name, mdd_obf_dir_name) == 0) { - *f = LU_OBF_FID; - return 0; + rc = mdd_obf_setup(env, m); + if (rc) { + CERROR("%s: error initializing \"fid\" object: rc = %d.\n", + mdd2obd_dev(m)->obd_name, rc); + GOTO(out, rc); } - return -ENOENT; -} - -static mdl_mode_t dot_lustre_mdd_lock_mode(const struct lu_env *env, - struct md_object *obj, - mdl_mode_t mode) -{ - return MDL_MINMODE; -} + rc = mdd_lpf_setup(env, m); + if (rc != 0) { + CERROR("%s: error initializing \"lost+found\": rc = %d.\n", + mdd2obd_dev(m)->obd_name, rc); + GOTO(out, rc); + } -static int dot_lustre_mdd_create(const struct lu_env *env, - struct md_object *pobj, - const struct lu_name *lname, - struct md_object *child, - struct md_op_spec *spec, - struct md_attr* ma) -{ - return -EPERM; -} + RETURN(0); -static int dot_lustre_mdd_create_data(const struct lu_env *env, - struct md_object *p, - struct md_object *o, - const struct md_op_spec *spec, - struct md_attr *ma) -{ - return -EPERM; -} +out: + mdd_dot_lustre_cleanup(env, m); -static int dot_lustre_mdd_rename(const struct lu_env *env, - struct md_object *src_pobj, - struct md_object *tgt_pobj, - const struct lu_fid *lf, - const struct lu_name *lsname, - struct md_object *tobj, - const struct lu_name *ltname, - struct md_attr *ma) -{ - return -EPERM; + return rc; } -static int dot_lustre_mdd_link(const struct lu_env *env, - struct md_object *tgt_obj, - struct md_object *src_obj, - const struct lu_name *lname, - struct md_attr *ma) +/** + * set llog methods and create LLOG_AGENT_ORIG_CTXT llog + * object in obd_device + */ +static int mdd_hsm_actions_llog_init(const struct lu_env *env, + struct mdd_device *m) { - return -EPERM; -} + struct obd_device *obd = mdd2obd_dev(m); + struct llog_ctxt *ctxt = NULL; + int rc; + ENTRY; -static int dot_lustre_mdd_unlink(const struct lu_env *env, - struct md_object *pobj, - struct md_object *cobj, - const struct lu_name *lname, - struct md_attr *ma) -{ - return -EPERM; -} + OBD_SET_CTXT_MAGIC(&obd->obd_lvfs_ctxt); + obd->obd_lvfs_ctxt.dt = m->mdd_bottom; -static int dot_lustre_mdd_name_insert(const struct lu_env *env, - struct md_object *obj, - const struct lu_name *lname, - const struct lu_fid *fid, - const struct md_attr *ma) -{ - return -EPERM; -} + rc = llog_setup(env, obd, &obd->obd_olg, LLOG_AGENT_ORIG_CTXT, + obd, &llog_common_cat_ops); + if (rc) { + CERROR("%s: hsm actions llog setup failed: rc = %d\n", + obd->obd_name, rc); + RETURN(rc); + } -static int dot_lustre_mdd_name_remove(const struct lu_env *env, - struct md_object *obj, - const struct lu_name *lname, - const struct md_attr *ma) -{ - return -EPERM; -} + ctxt = llog_get_context(obd, LLOG_AGENT_ORIG_CTXT); + LASSERT(ctxt); -static int dot_lustre_mdd_rename_tgt(const struct lu_env *env, - struct md_object *pobj, - struct md_object *tobj, - const struct lu_fid *fid, - const struct lu_name *lname, - struct md_attr *ma) -{ - return -EPERM; -} + rc = llog_open_create(env, ctxt, &ctxt->loc_handle, NULL, + HSM_ACTIONS); + if (rc) { + CERROR("%s: hsm actions llog open_create failed: rc = %d\n", + obd->obd_name, rc); + GOTO(out_cleanup, rc); + } + rc = llog_init_handle(env, ctxt->loc_handle, LLOG_F_IS_CAT, NULL); + if (rc) + GOTO(out_close, rc); -static struct md_dir_operations mdd_dot_lustre_dir_ops = { - .mdo_is_subdir = mdd_is_subdir, - .mdo_lookup = dot_lustre_mdd_lookup, - .mdo_lock_mode = dot_lustre_mdd_lock_mode, - .mdo_create = dot_lustre_mdd_create, - .mdo_create_data = dot_lustre_mdd_create_data, - .mdo_rename = dot_lustre_mdd_rename, - .mdo_link = dot_lustre_mdd_link, - .mdo_unlink = dot_lustre_mdd_unlink, - .mdo_name_insert = dot_lustre_mdd_name_insert, - .mdo_name_remove = dot_lustre_mdd_name_remove, - .mdo_rename_tgt = dot_lustre_mdd_rename_tgt, -}; + llog_ctxt_put(ctxt); + RETURN(0); -static int obf_attr_get(const struct lu_env *env, struct md_object *obj, - struct md_attr *ma) -{ - struct mdd_device *mdd = mdo2mdd(obj); +out_close: + llog_cat_close(env, ctxt->loc_handle); + ctxt->loc_handle = NULL; +out_cleanup: + llog_cleanup(env, ctxt); - /* "fid" is a virtual object and hence does not have any "real" - * attributes. So we reuse attributes of .lustre for "fid" dir */ - return mdd_attr_get(env, &mdd->mdd_dot_lustre->mod_obj, ma); + return rc; } -static int obf_attr_set(const struct lu_env *env, struct md_object *obj, - const struct md_attr *ma) +/** + * cleanup the context created by llog_setup_named() + */ +static int mdd_hsm_actions_llog_fini(const struct lu_env *env, + struct mdd_device *m) { - return -EPERM; -} + struct obd_device *obd = mdd2obd_dev(m); + struct llog_ctxt *lctxt; + ENTRY; -static int obf_xattr_list(const struct lu_env *env, - struct md_object *obj, struct lu_buf *buf) -{ - return 0; -} - -static int obf_xattr_get(const struct lu_env *env, - struct md_object *obj, struct lu_buf *buf, - const char *name) -{ - struct mdd_device *mdd = mdo2mdd(obj); - struct mdd_object *root; - int rc = 0; - - /* - * .lustre returns default striping which is 'stored' - * in the root - */ - if (strcmp(name, XATTR_NAME_LOV) == 0) { - root = mdd_object_find(env, mdd, &mdd->mdd_local_root_fid); - if (IS_ERR(root)) - return PTR_ERR(root); - rc = mdo_xattr_get(env, root, buf, name, - mdd_object_capa(env, md2mdd_obj(obj))); - mdd_object_put(env, root); + lctxt = llog_get_context(obd, LLOG_AGENT_ORIG_CTXT); + if (lctxt) { + llog_cat_close(env, lctxt->loc_handle); + lctxt->loc_handle = NULL; + llog_cleanup(env, lctxt); } - return rc; -} - -static int obf_xattr_set(const struct lu_env *env, - struct md_object *obj, - const struct lu_buf *buf, const char *name, - int fl) -{ - return -EPERM; -} - -static int obf_xattr_del(const struct lu_env *env, - struct md_object *obj, - const char *name) -{ - return -EPERM; -} - -static int obf_mdd_open(const struct lu_env *env, struct md_object *obj, - int flags) -{ - struct mdd_object *mdd_obj = md2mdd_obj(obj); - - mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD); - mdd_obj->mod_count++; - mdd_write_unlock(env, mdd_obj); - - return 0; -} - -static int obf_mdd_close(const struct lu_env *env, struct md_object *obj, - struct md_attr *ma, int mode) -{ - struct mdd_object *mdd_obj = md2mdd_obj(obj); - - mdd_write_lock(env, mdd_obj, MOR_TGT_CHILD); - mdd_obj->mod_count--; - mdd_write_unlock(env, mdd_obj); - - return 0; -} - -/** Nothing to list in "fid" directory */ -static int obf_mdd_readpage(const struct lu_env *env, struct md_object *obj, - const struct lu_rdpg *rdpg) -{ - return -EPERM; -} - -static int obf_path(const struct lu_env *env, struct md_object *obj, - char *path, int pathlen, __u64 *recno, int *linkno) -{ - return -ENOSYS; + RETURN(0); } -static struct md_object_operations mdd_obf_obj_ops = { - .moo_attr_get = obf_attr_get, - .moo_attr_set = obf_attr_set, - .moo_xattr_list = obf_xattr_list, - .moo_xattr_get = obf_xattr_get, - .moo_xattr_set = obf_xattr_set, - .moo_xattr_del = obf_xattr_del, - .moo_open = obf_mdd_open, - .moo_close = obf_mdd_close, - .moo_readpage = obf_mdd_readpage, - .moo_path = obf_path -}; - -/** - * Lookup method for "fid" object. Only filenames with correct SEQ:OID format - * are valid. We also check if object with passed fid exists or not. - */ -static int obf_lookup(const struct lu_env *env, struct md_object *p, - const struct lu_name *lname, struct lu_fid *f, - struct md_op_spec *spec) +static void mdd_device_shutdown(const struct lu_env *env, struct mdd_device *m, + struct lustre_cfg *cfg) { - char *name = (char *)lname->ln_name; - struct mdd_device *mdd = mdo2mdd(p); - struct mdd_object *child; - int rc = 0; - - while (*name == '[') - name++; - - sscanf(name, SFID, RFID(f)); - if (!fid_is_sane(f)) { - CWARN("%s: bad FID format [%s], should be "DFID"\n", - mdd2obd_dev(mdd)->obd_name, lname->ln_name, - (__u64)FID_SEQ_NORMAL, 1, 0); - GOTO(out, rc = -EINVAL); - } - - if (!fid_is_norm(f)) { - CWARN("%s: "DFID" is invalid, sequence should be " - ">= "LPX64"\n", mdd2obd_dev(mdd)->obd_name, PFID(f), - (__u64)FID_SEQ_NORMAL); - GOTO(out, rc = -EINVAL); + barrier_deregister(m->mdd_bottom); + lfsck_degister(env, m->mdd_bottom); + mdd_hsm_actions_llog_fini(env, m); + mdd_changelog_fini(env, m); + mdd_orphan_index_fini(env, m); + mdd_dot_lustre_cleanup(env, m); + if (mdd2obd_dev(m)->u.obt.obt_nodemap_config_file) { + nm_config_file_deregister_tgt(env, + mdd2obd_dev(m)->u.obt.obt_nodemap_config_file); + mdd2obd_dev(m)->u.obt.obt_nodemap_config_file = NULL; } + if (m->mdd_los != NULL) { + local_oid_storage_fini(env, m->mdd_los); + m->mdd_los = NULL; + } + lu_site_purge(env, mdd2lu_dev(m)->ld_site, ~0); - /* Check if object with this fid exists */ - child = mdd_object_find(env, mdd, f); - if (child == NULL) - GOTO(out, rc = 0); - if (IS_ERR(child)) - GOTO(out, rc = PTR_ERR(child)); - - if (mdd_object_exists(child) == 0) - rc = -ENOENT; - - mdd_object_put(env, child); - -out: - return rc; -} - -static int obf_create(const struct lu_env *env, struct md_object *pobj, - const struct lu_name *lname, struct md_object *child, - struct md_op_spec *spec, struct md_attr* ma) -{ - return -EPERM; -} - -static int obf_rename(const struct lu_env *env, - struct md_object *src_pobj, struct md_object *tgt_pobj, - const struct lu_fid *lf, const struct lu_name *lsname, - struct md_object *tobj, const struct lu_name *ltname, - struct md_attr *ma) -{ - return -EPERM; -} - -static int obf_link(const struct lu_env *env, struct md_object *tgt_obj, - struct md_object *src_obj, const struct lu_name *lname, - struct md_attr *ma) -{ - return -EPERM; -} - -static int obf_unlink(const struct lu_env *env, struct md_object *pobj, - struct md_object *cobj, const struct lu_name *lname, - struct md_attr *ma) -{ - return -EPERM; -} - -static struct md_dir_operations mdd_obf_dir_ops = { - .mdo_lookup = obf_lookup, - .mdo_create = obf_create, - .mdo_rename = obf_rename, - .mdo_link = obf_link, - .mdo_unlink = obf_unlink -}; - -/** - * Create special in-memory "fid" object for open-by-fid. - */ -static int mdd_obf_setup(const struct lu_env *env, struct mdd_device *m) -{ - struct mdd_object *mdd_obf; - struct lu_object *obf_lu_obj; - int rc = 0; - - m->mdd_dot_lustre_objs.mdd_obf = mdd_object_find(env, m, - &LU_OBF_FID); - if (m->mdd_dot_lustre_objs.mdd_obf == NULL || - IS_ERR(m->mdd_dot_lustre_objs.mdd_obf)) - GOTO(out, rc = -ENOENT); - - mdd_obf = m->mdd_dot_lustre_objs.mdd_obf; - mdd_obf->mod_obj.mo_dir_ops = &mdd_obf_dir_ops; - mdd_obf->mod_obj.mo_ops = &mdd_obf_obj_ops; - /* Don't allow objects to be created in "fid" dir */ - mdd_obf->mod_flags |= IMMUTE_OBJ; - - obf_lu_obj = mdd2lu_obj(mdd_obf); - obf_lu_obj->lo_header->loh_attr |= (LOHA_EXISTS | S_IFDIR); - -out: - return rc; -} - -/** Setup ".lustre" directory object */ -static int mdd_dot_lustre_setup(const struct lu_env *env, struct mdd_device *m) -{ - struct dt_object *dt_dot_lustre; - struct lu_fid *fid = &mdd_env_info(env)->mti_fid; - int rc; - ENTRY; - - rc = create_dot_lustre_dir(env, m); - if (rc) - return rc; - - dt_dot_lustre = dt_store_open(env, m->mdd_child, mdd_root_dir_name, - dot_lustre_name, fid); - if (IS_ERR(dt_dot_lustre)) { - rc = PTR_ERR(dt_dot_lustre); - GOTO(out, rc); - } - - /* references are released in mdd_device_shutdown() */ - m->mdd_dot_lustre = lu2mdd_obj(lu_object_locate(dt_dot_lustre->do_lu.lo_header, - &mdd_device_type)); - - m->mdd_dot_lustre->mod_obj.mo_dir_ops = &mdd_dot_lustre_dir_ops; - m->mdd_dot_lustre->mod_obj.mo_ops = &mdd_dot_lustre_obj_ops; - - rc = mdd_obf_setup(env, m); - if (rc) - CERROR("Error initializing \"fid\" object - %d.\n", rc); - -out: - RETURN(rc); + if (m->mdd_child_exp) + obd_disconnect(m->mdd_child_exp); } static int mdd_process_config(const struct lu_env *env, @@ -1112,28 +1109,30 @@ static int mdd_process_config(const struct lu_env *env, ENTRY; switch (cfg->lcfg_command) { - case LCFG_PARAM: { - struct lprocfs_static_vars lvars; - - lprocfs_mdd_init_vars(&lvars); - rc = class_process_proc_param(PARAM_MDD, lvars.obd_vars, cfg,m); - if (rc > 0 || rc == -ENOSYS) - /* we don't understand; pass it on */ - rc = next->ld_ops->ldo_process_config(env, next, cfg); - break; - } + case LCFG_PARAM: { + ssize_t count; + + count = class_modify_config(cfg, PARAM_MDD, &m->mdd_kobj); + rc = count > 0 ? 0 : count; + if (rc) + /* we don't understand; pass it on */ + rc = next->ld_ops->ldo_process_config(env, next, cfg); + break; + } case LCFG_SETUP: rc = next->ld_ops->ldo_process_config(env, next, cfg); if (rc) GOTO(out, rc); - dt->dd_ops->dt_conf_get(env, dt, &m->mdd_dt_conf); + dt_conf_get(env, dt, &m->mdd_dt_conf); break; - case LCFG_CLEANUP: - mdd_lfsck_cleanup(env, m); - mdd_changelog_fini(env, m); + case LCFG_PRE_CLEANUP: + rc = next->ld_ops->ldo_process_config(env, next, cfg); + mdd_generic_thread_stop(&m->mdd_orphan_cleanup_thread); + break; + case LCFG_CLEANUP: rc = next->ld_ops->ldo_process_config(env, next, cfg); lu_dev_del_linkage(d->ld_site, d); - mdd_device_shutdown(env, m, cfg); + mdd_device_shutdown(env, m, cfg); break; default: rc = next->ld_ops->ldo_process_config(env, next, cfg); @@ -1144,114 +1143,178 @@ out: } static int mdd_recovery_complete(const struct lu_env *env, - struct lu_device *d) + struct lu_device *d) { - struct mdd_device *mdd = lu2mdd_dev(d); + struct mdd_device *mdd = lu2mdd_dev(d); struct lu_device *next; - int rc; - ENTRY; + int rc; + ENTRY; - LASSERT(mdd != NULL); + LASSERT(mdd != NULL); next = &mdd->mdd_child->dd_lu_dev; - /* XXX: orphans handling. */ - __mdd_orphan_cleanup(env, mdd); - rc = next->ld_ops->ldo_recovery_complete(env, next); + if (!mdd->mdd_bottom->dd_rdonly) + mdd_orphan_cleanup(env, mdd); + rc = next->ld_ops->ldo_recovery_complete(env, next); - RETURN(rc); + RETURN(rc); } -static int mdd_find_or_create_root(const struct lu_env *env, - struct mdd_device *mdd) +int mdd_local_file_create(const struct lu_env *env, struct mdd_device *mdd, + const struct lu_fid *pfid, const char *name, + __u32 mode, struct lu_fid *fid) { - struct dt_object *root; - struct md_object *mroot; - struct lu_fid *fid = &mdd_env_info(env)->mti_fid; - int rc = 0; + struct dt_object *parent, *dto; + int rc = 0; ENTRY; - /* Check if the "ROOT" entry exists already */ - root = dt_store_open(env, mdd->mdd_child, "", mdd_root_dir_name, - fid); - if (!IS_ERR(root)) { - lu_object_put(env, &root->do_lu); - GOTO(out, rc = 0); - } - - lu_root_fid(fid); - /* New Filesystem, create /ROOT */ - mroot = llo_store_create_index(env, &mdd->mdd_md_dev, mdd->mdd_bottom, - "", mdd_root_dir_name, fid, - &dt_directory_features); - if (IS_ERR(mroot)) - GOTO(out, rc = PTR_ERR(mroot)); + LASSERT(!fid_is_zero(pfid)); + parent = dt_locate(env, mdd->mdd_bottom, pfid); + if (unlikely(IS_ERR(parent))) + RETURN(PTR_ERR(parent)); - lu_object_put(env, &mroot->mo_lu); -out: - if (rc == 0) - mdd->mdd_root_fid = *fid; + /* create local file/dir, if @fid is passed then try to use it */ + if (fid_is_zero(fid)) + dto = local_file_find_or_create(env, mdd->mdd_los, parent, + name, mode); + else + dto = local_file_find_or_create_with_fid(env, mdd->mdd_bottom, + fid, parent, name, + mode); + if (IS_ERR(dto)) + GOTO(out_put, rc = PTR_ERR(dto)); + *fid = *lu_object_fid(&dto->do_lu); + /* since stack is not fully set up the local_storage uses own stack + * and we should drop its object from cache */ + dt_object_put_nocache(env, dto); + EXIT; +out_put: + dt_object_put(env, parent); + return rc; +} - RETURN(rc); +static int mdd_lfsck_out_notify(const struct lu_env *env, void *data, + enum lfsck_events event) +{ + return 0; } static int mdd_prepare(const struct lu_env *env, struct lu_device *pdev, struct lu_device *cdev) { - struct mdd_device *mdd = lu2mdd_dev(cdev); - struct lu_device *next = &mdd->mdd_child->dd_lu_dev; - struct dt_object *root; - struct lu_fid *fid = &mdd_env_info(env)->mti_fid; - int rc; + struct mdd_device *mdd = lu2mdd_dev(cdev); + struct lu_device *next = &mdd->mdd_child->dd_lu_dev; + struct nm_config_file *nodemap_config; + struct obd_device_target *obt = &mdd2obd_dev(mdd)->u.obt; + struct lu_fid fid; + int rc; - ENTRY; + ENTRY; rc = next->ld_ops->ldo_prepare(env, cdev, next); if (rc) - GOTO(out, rc); + RETURN(rc); + + /* Setup local dirs */ + fid.f_seq = FID_SEQ_LOCAL_NAME; + fid.f_oid = 1; + fid.f_ver = 0; + rc = local_oid_storage_init(env, mdd->mdd_bottom, &fid, + &mdd->mdd_los); + if (rc) + RETURN(rc); rc = dt_root_get(env, mdd->mdd_child, &mdd->mdd_local_root_fid); - if (rc != 0) - GOTO(out, rc); + if (rc < 0) + GOTO(out_los, rc); + lu_root_fid(&fid); if (mdd_seq_site(mdd)->ss_node_id == 0) { - rc = mdd_find_or_create_root(env, mdd); + rc = mdd_local_file_create(env, mdd, &mdd->mdd_local_root_fid, + mdd_root_dir_name, S_IFDIR | + S_IRUGO | S_IWUSR | S_IXUGO, &fid); if (rc != 0) { CERROR("%s: create root fid failed: rc = %d\n", mdd2obd_dev(mdd)->obd_name, rc); - GOTO(out, rc); + GOTO(out_los, rc); } + mdd->mdd_root_fid = fid; rc = mdd_dot_lustre_setup(env, mdd); if (rc != 0) { CERROR("%s: initializing .lustre failed: rc = %d\n", mdd2obd_dev(mdd)->obd_name, rc); - GOTO(out, rc); + GOTO(out_los, rc); } + } else { + /* Normal client usually send root access to MDT0 directly, + * the root FID on non-MDT0 will only be used by echo client. */ + mdd->mdd_root_fid = fid; } - rc = orph_index_init(env, mdd); - if (rc != 0) - GOTO(out, rc); - - /* we use capa file to declare llog changes, - * will be fixed with new llog in 2.3 */ - root = dt_store_open(env, mdd->mdd_child, "", CAPA_KEYS, fid); - if (IS_ERR(root)) - GOTO(out, rc = PTR_ERR(root)); - - mdd->mdd_capa = root; + rc = mdd_orphan_index_init(env, mdd); + if (rc < 0) + GOTO(out_dot, rc); rc = mdd_changelog_init(env, mdd); - if (rc != 0) - GOTO(out, rc); + if (rc != 0) { + CERROR("%s: failed to initialize changelog: rc = %d\n", + mdd2obd_dev(mdd)->obd_name, rc); + GOTO(out_orph, rc); + } - rc = mdd_lfsck_setup(env, mdd); + rc = mdd_hsm_actions_llog_init(env, mdd); if (rc != 0) + GOTO(out_changelog, rc); + + nodemap_config = nm_config_file_register_tgt(env, mdd->mdd_bottom, + mdd->mdd_los); + if (IS_ERR(nodemap_config)) { + rc = PTR_ERR(nodemap_config); + if (rc != -EROFS) + GOTO(out_hsm, rc); + } else { + obt->obt_nodemap_config_file = nodemap_config; + } + + rc = lfsck_register(env, mdd->mdd_bottom, mdd->mdd_child, + mdd2obd_dev(mdd), mdd_lfsck_out_notify, + mdd, true); + if (rc != 0) { CERROR("%s: failed to initialize lfsck: rc = %d\n", mdd2obd_dev(mdd)->obd_name, rc); -out: + GOTO(out_nodemap, rc); + } + + rc = barrier_register(mdd->mdd_bottom, mdd->mdd_child); + if (rc) { + CERROR("%s: failed to register to barrier: rc = %d\n", + mdd2obd_dev(mdd)->obd_name, rc); + GOTO(out_lfsck, rc); + } + + RETURN(0); + +out_lfsck: + lfsck_degister(env, mdd->mdd_bottom); +out_nodemap: + nm_config_file_deregister_tgt(env, obt->obt_nodemap_config_file); + obt->obt_nodemap_config_file = NULL; +out_hsm: + mdd_hsm_actions_llog_fini(env, mdd); +out_changelog: + mdd_changelog_fini(env, mdd); +out_orph: + mdd_orphan_index_fini(env, mdd); +out_dot: + if (mdd_seq_site(mdd)->ss_node_id == 0) + mdd_dot_lustre_cleanup(env, mdd); +out_los: + local_oid_storage_fini(env, mdd->mdd_los); + mdd->mdd_los = NULL; + return rc; } @@ -1262,59 +1325,40 @@ const struct lu_device_operations mdd_lu_ops = { .ldo_prepare = mdd_prepare, }; -/* - * No permission check is needed. - */ static int mdd_root_get(const struct lu_env *env, - struct md_device *m, struct lu_fid *f) + struct md_device *m, struct lu_fid *f) { - struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev); + struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev); - ENTRY; - *f = mdd->mdd_root_fid; - RETURN(0); + ENTRY; + *f = mdd->mdd_root_fid; + RETURN(0); } /* * No permission check is needed. */ static int mdd_statfs(const struct lu_env *env, struct md_device *m, - struct obd_statfs *sfs) + struct obd_statfs *sfs) { - struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev); - int rc; - - ENTRY; - - rc = mdd_child_ops(mdd)->dt_statfs(env, mdd->mdd_child, sfs); + struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev); + int rc; - RETURN(rc); -} + ENTRY; -/* - * No permission check is needed. - */ -static int mdd_init_capa_ctxt(const struct lu_env *env, struct md_device *m, - int mode, unsigned long timeout, __u32 alg, - struct lustre_capa_key *keys) -{ - struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev); - int rc; - ENTRY; + rc = mdd_child_ops(mdd)->dt_statfs(env, mdd->mdd_child, sfs, NULL); - /* need barrier for mds_capa_keys access. */ + sfs->os_namelen = min_t(__u32, sfs->os_namelen, NAME_MAX); - rc = mdd_child_ops(mdd)->dt_init_capa_ctxt(env, mdd->mdd_child, mode, - timeout, alg, keys); - RETURN(rc); + RETURN(rc); } -static int mdd_update_capa_key(const struct lu_env *env, - struct md_device *m, - struct lustre_capa_key *key) +static const struct dt_device_param *mdd_dtconf_get(const struct lu_env *env, + struct md_device *m) { - /* we do not support capabilities ... */ - return -EINVAL; + struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev); + + return &mdd->mdd_dt_conf; } static int mdd_llog_ctxt_get(const struct lu_env *env, struct md_device *m, @@ -1332,7 +1376,7 @@ static struct lu_device *mdd_device_free(const struct lu_env *env, struct mdd_device *m = lu2mdd_dev(lu); ENTRY; - LASSERT(cfs_atomic_read(&lu->ld_ref) == 0); + LASSERT(atomic_read(&lu->ld_ref) == 0); md_device_fini(&m->mdd_md_dev); OBD_FREE_PTR(m); RETURN(NULL); @@ -1412,46 +1456,56 @@ static int mdd_obd_disconnect(struct obd_export *exp) RETURN(rc); } -static int mdd_obd_health_check(const struct lu_env *env, - struct obd_device *obd) +static int mdd_obd_get_info(const struct lu_env *env, struct obd_export *exp, + __u32 keylen, void *key, __u32 *vallen, void *val) { - struct mdd_device *mdd = lu2mdd_dev(obd->obd_lu_dev); - int rc; - ENTRY; + int rc = -EINVAL; + if (KEY_IS(KEY_OSP_CONNECTED)) { + struct obd_device *obd = exp->exp_obd; + struct mdd_device *mdd; + + if (!obd->obd_set_up || obd->obd_stopping) + RETURN(-EAGAIN); + + mdd = lu2mdd_dev(obd->obd_lu_dev); + LASSERT(mdd); + rc = obd_get_info(env, mdd->mdd_child_exp, keylen, key, vallen, + val); + RETURN(rc); + } + + RETURN(rc); +} + +static int mdd_obd_set_info_async(const struct lu_env *env, + struct obd_export *exp, + __u32 keylen, void *key, + __u32 vallen, void *val, + struct ptlrpc_request_set *set) +{ + struct obd_device *obd = exp->exp_obd; + struct mdd_device *mdd; + int rc; + + if (!obd->obd_set_up || obd->obd_stopping) + RETURN(-EAGAIN); + + mdd = lu2mdd_dev(obd->obd_lu_dev); LASSERT(mdd); - rc = obd_health_check(env, mdd->mdd_child_exp->exp_obd); + rc = obd_set_info_async(env, mdd->mdd_child_exp, keylen, key, + vallen, val, set); RETURN(rc); } -static struct obd_ops mdd_obd_device_ops = { +static const struct obd_ops mdd_obd_device_ops = { .o_owner = THIS_MODULE, .o_connect = mdd_obd_connect, .o_disconnect = mdd_obd_disconnect, - .o_health_check = mdd_obd_health_check -}; - -/* - * context key constructor/destructor: - * mdd_capainfo_key_init, mdd_capainfo_key_fini - */ -LU_KEY_INIT_FINI(mdd_capainfo, struct md_capainfo); - -struct lu_context_key mdd_capainfo_key = { - .lct_tags = LCT_SESSION, - .lct_init = mdd_capainfo_key_init, - .lct_fini = mdd_capainfo_key_fini + .o_get_info = mdd_obd_get_info, + .o_set_info_async = mdd_obd_set_info_async, }; -struct md_capainfo *md_capainfo(const struct lu_env *env) -{ - /* NB, in mdt_init0 */ - if (env->le_ses == NULL) - return NULL; - return lu_context_key_get(env->le_ses, &mdd_capainfo_key); -} -EXPORT_SYMBOL(md_capainfo); - static int mdd_changelog_user_register(const struct lu_env *env, struct mdd_device *mdd, int *id) { @@ -1471,10 +1525,7 @@ static int mdd_changelog_user_register(const struct lu_env *env, RETURN(-ENOMEM); } - /* Assume we want it on since somebody registered */ - rc = mdd_changelog_on(env, mdd, 1); - if (rc) - GOTO(out, rc); + CFS_RACE(CFS_FAIL_CHLOG_USER_REG_UNREG_RACE); rec->cur_hdr.lrh_len = sizeof(*rec); rec->cur_hdr.lrh_type = CHANGELOG_USER_REC; @@ -1485,150 +1536,303 @@ static int mdd_changelog_user_register(const struct lu_env *env, GOTO(out, rc = -EOVERFLOW); } *id = rec->cur_id = ++mdd->mdd_cl.mc_lastuser; + mdd->mdd_cl.mc_users++; rec->cur_endrec = mdd->mdd_cl.mc_index; + + rec->cur_time = (__u32)ktime_get_real_seconds(); + if (OBD_FAIL_CHECK(OBD_FAIL_TIME_IN_CHLOG_USER)) + rec->cur_time = 0; + spin_unlock(&mdd->mdd_cl.mc_user_lock); - rc = llog_cat_add(env, ctxt->loc_handle, &rec->cur_hdr, NULL, NULL); + rc = llog_cat_add(env, ctxt->loc_handle, &rec->cur_hdr, NULL); + if (rc) { + CWARN("%s: Failed to register changelog user %d: rc=%d\n", + mdd2obd_dev(mdd)->obd_name, *id, rc); + spin_lock(&mdd->mdd_cl.mc_user_lock); + mdd->mdd_cl.mc_users--; + spin_unlock(&mdd->mdd_cl.mc_user_lock); + GOTO(out, rc); + } CDEBUG(D_IOCTL, "Registered changelog user %d\n", *id); + + /* Assume we want it on since somebody registered */ + rc = mdd_changelog_on(env, mdd); + if (rc) + GOTO(out, rc); + out: OBD_FREE_PTR(rec); llog_ctxt_put(ctxt); RETURN(rc); } -struct mdd_changelog_user_data { - __u64 mcud_endrec; /**< purge record for this user */ - __u64 mcud_minrec; /**< lowest changelog recno still referenced */ - __u32 mcud_id; - __u32 mcud_minid; /**< user id with lowest rec reference */ - __u32 mcud_usercount; - int mcud_found:1; +struct mdd_changelog_user_purge { + struct mdd_device *mcup_mdd; + __u32 mcup_id; + __u32 mcup_usercount; + __u64 mcup_minrec; + bool mcup_found; }; -#define MCUD_UNREGISTER -1LL -/** Two things: - * 1. Find the smallest record everyone is willing to purge - * 2. Update the last purgeable record for this user +/** + * changelog_user_purge callback + * + * Is called once per user. + * + * Check to see the user requested is available from rec. + * Truncate the changelog. + * Keep track of the total number of users (calls). */ static int mdd_changelog_user_purge_cb(const struct lu_env *env, struct llog_handle *llh, struct llog_rec_hdr *hdr, void *data) { struct llog_changelog_user_rec *rec; - struct mdd_changelog_user_data *mcud = data; + struct mdd_changelog_user_purge *mcup = data; + struct llog_cookie cookie; int rc; - ENTRY; + ENTRY; - LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN); + if ((llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN) == 0) + RETURN(-ENXIO); - rec = (struct llog_changelog_user_rec *)hdr; + rec = container_of(hdr, struct llog_changelog_user_rec, cur_hdr); - mcud->mcud_usercount++; + mcup->mcup_usercount++; - /* If we have a new endrec for this id, use it for the following - min check instead of its old value */ - if (rec->cur_id == mcud->mcud_id) - rec->cur_endrec = max(rec->cur_endrec, mcud->mcud_endrec); + if (rec->cur_id != mcup->mcup_id) { + /* truncate to the lowest endrec that is not this user */ + mcup->mcup_minrec = min(mcup->mcup_minrec, + rec->cur_endrec); + RETURN(0); + } - /* Track the minimum referenced record */ - if (mcud->mcud_minid == 0 || mcud->mcud_minrec > rec->cur_endrec) { - mcud->mcud_minid = rec->cur_id; - mcud->mcud_minrec = rec->cur_endrec; - } + /* Unregister this user */ + cookie.lgc_lgl = llh->lgh_id; + cookie.lgc_index = hdr->lrh_index; + + rc = llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, + 1, &cookie); + if (rc == 0) { + mcup->mcup_found = true; + mcup->mcup_usercount--; + spin_lock(&mcup->mcup_mdd->mdd_cl.mc_user_lock); + mcup->mcup_mdd->mdd_cl.mc_users--; + spin_unlock(&mcup->mcup_mdd->mdd_cl.mc_user_lock); + } + + RETURN(rc); +} - if (rec->cur_id != mcud->mcud_id) - RETURN(0); +int mdd_changelog_user_purge(const struct lu_env *env, + struct mdd_device *mdd, __u32 id) +{ + struct mdd_changelog_user_purge mcup = { + .mcup_mdd = mdd, + .mcup_id = id, + .mcup_found = false, + .mcup_usercount = 0, + .mcup_minrec = ULLONG_MAX, + }; + struct llog_ctxt *ctxt; + int rc; - /* Update this user's record */ - mcud->mcud_found = 1; + ENTRY; - /* Special case: unregister this user */ - if (mcud->mcud_endrec == MCUD_UNREGISTER) { - struct llog_cookie cookie; + CDEBUG(D_IOCTL, "%s: Purge request: id=%u\n", + mdd2obd_dev(mdd)->obd_name, id); - cookie.lgc_lgl = llh->lgh_id; - cookie.lgc_index = hdr->lrh_index; + ctxt = llog_get_context(mdd2obd_dev(mdd), + LLOG_CHANGELOG_USER_ORIG_CTXT); + if (ctxt == NULL || + (ctxt->loc_handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT) == 0) + GOTO(out, rc = -ENXIO); - rc = llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, - 1, &cookie); - if (rc == 0) - mcud->mcud_usercount--; + rc = llog_cat_process(env, ctxt->loc_handle, + mdd_changelog_user_purge_cb, &mcup, + 0, 0); - RETURN(rc); - } + OBD_FAIL_TIMEOUT(OBD_FAIL_LLOG_PURGE_DELAY, cfs_fail_val); + + if ((rc == 0) && (mcup.mcup_usercount == 0)) { + spin_lock(&mdd->mdd_cl.mc_user_lock); + if (mdd->mdd_cl.mc_users == 0) { + /* No more users; turn changelogs off */ + CDEBUG(D_IOCTL, "turning off changelogs\n"); + rc = mdd_changelog_off(env, mdd); + } + spin_unlock(&mdd->mdd_cl.mc_user_lock); + } + + if ((rc == 0) && mcup.mcup_found) { + CDEBUG(D_IOCTL, "%s: Purging changelog entries for user %d " + "record=%llu\n", + mdd2obd_dev(mdd)->obd_name, id, mcup.mcup_minrec); + /* Cancelling record 0 destroys the entire changelog, make sure + we don't do that unless we mean it. */ + if (mcup.mcup_minrec != 0 || mcup.mcup_usercount == 0) { + rc = mdd_changelog_llog_cancel(env, mdd, + mcup.mcup_minrec); + } + } else { + CWARN("%s: No changelog for user %u; rc=%d\n", + mdd2obd_dev(mdd)->obd_name, id, rc); + GOTO(out, rc = -ENOENT); + } - /* Update the endrec */ - CDEBUG(D_IOCTL, "Rewriting changelog user %d endrec to "LPU64"\n", - mcud->mcud_id, rec->cur_endrec); + CFS_RACE(CFS_FAIL_CHLOG_USER_REG_UNREG_RACE); - /* hdr+1 is loc of data */ - hdr->lrh_len -= sizeof(*hdr) + sizeof(struct llog_rec_tail); - rc = llog_write(env, llh, hdr, NULL, 0, (void *)(hdr + 1), - hdr->lrh_index); + EXIT; +out: + if (ctxt != NULL) + llog_ctxt_put(ctxt); - RETURN(rc); + return rc; } -static int mdd_changelog_user_purge(const struct lu_env *env, - struct mdd_device *mdd, int id, - long long endrec) +struct mdd_changelog_user_clear { + __u64 mcuc_endrec; + __u64 mcuc_minrec; + __u32 mcuc_id; + bool mcuc_flush; +}; + +/** + * changelog_clear callback + * + * Is called once per user. + * + * Check to see the user requested is available from rec. + * Check the oldest (smallest) record for boundary conditions. + * Truncate the changelog. + */ +static int mdd_changelog_clear_cb(const struct lu_env *env, + struct llog_handle *llh, + struct llog_rec_hdr *hdr, + void *data) { - struct mdd_changelog_user_data data; - struct llog_ctxt *ctxt; - int rc; - ENTRY; + struct llog_changelog_user_rec *rec; + struct mdd_changelog_user_clear *mcuc = data; + int rc; + + ENTRY; + + if ((llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN) == 0) + RETURN(-ENXIO); + + rec = container_of(hdr, struct llog_changelog_user_rec, cur_hdr); + + /* Does the changelog id match the requested id? */ + if (rec->cur_id != mcuc->mcuc_id) { + mcuc->mcuc_minrec = min(mcuc->mcuc_minrec, + rec->cur_endrec); + RETURN(0); + } + + /* cur_endrec is the oldest purgeable record, make sure we're newer */ + if (rec->cur_endrec > mcuc->mcuc_endrec) { + CDEBUG(D_IOCTL, "Request %llu out of range: %llu\n", + mcuc->mcuc_endrec, rec->cur_endrec); + RETURN(-EINVAL); + } + + /* Flag that we've met all the range and user checks. + * We now know the record to flush. + */ + rec->cur_endrec = mcuc->mcuc_endrec; + + rec->cur_time = (__u32)ktime_get_real_seconds(); + if (OBD_FAIL_CHECK(OBD_FAIL_TIME_IN_CHLOG_USER)) + rec->cur_time = 0; + + mcuc->mcuc_flush = true; + + CDEBUG(D_IOCTL, "Rewriting changelog user %u endrec to %llu\n", + mcuc->mcuc_id, rec->cur_endrec); + + /* Update the endrec */ + rc = llog_write(env, llh, hdr, hdr->lrh_index); - CDEBUG(D_IOCTL, "Purge request: id=%d, endrec=%lld\n", id, endrec); + RETURN(rc); +} + +/** + * Clear a changelog up to entry specified by endrec for user id. + */ +static int mdd_changelog_clear(const struct lu_env *env, + struct mdd_device *mdd, __u32 id, + __u64 endrec) +{ + struct mdd_changelog_user_clear mcuc = { + .mcuc_id = id, + .mcuc_minrec = endrec, + .mcuc_flush = false, + }; + struct llog_ctxt *ctxt; + __u64 start_rec; + int rc; - data.mcud_id = id; - data.mcud_minid = 0; - data.mcud_minrec = 0; - data.mcud_usercount = 0; - data.mcud_endrec = endrec; + ENTRY; + + CDEBUG(D_IOCTL, "%s: Purge request: id=%u, endrec=%llu\n", + mdd2obd_dev(mdd)->obd_name, id, endrec); + /* start_rec is the newest (largest value) entry in the changelogs*/ spin_lock(&mdd->mdd_cl.mc_lock); - endrec = mdd->mdd_cl.mc_index; + start_rec = mdd->mdd_cl.mc_index; spin_unlock(&mdd->mdd_cl.mc_lock); - if ((data.mcud_endrec == 0) || - ((data.mcud_endrec > endrec) && - (data.mcud_endrec != MCUD_UNREGISTER))) - data.mcud_endrec = endrec; - ctxt = llog_get_context(mdd2obd_dev(mdd), - LLOG_CHANGELOG_USER_ORIG_CTXT); - if (ctxt == NULL) - return -ENXIO; + if (start_rec < endrec) { + CDEBUG(D_IOCTL, "%s: Could not clear changelog, requested "\ + "address out of range\n", mdd2obd_dev(mdd)->obd_name); + RETURN(-EINVAL); + } + + if (endrec == 0) { + mcuc.mcuc_endrec = start_rec; + mcuc.mcuc_minrec = ULLONG_MAX; + } else { + mcuc.mcuc_endrec = endrec; + } - LASSERT(ctxt->loc_handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT); + ctxt = llog_get_context(mdd2obd_dev(mdd), + LLOG_CHANGELOG_USER_ORIG_CTXT); + if (ctxt == NULL || + (ctxt->loc_handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT) == 0) + GOTO(out, rc = -ENXIO); rc = llog_cat_process(env, ctxt->loc_handle, - mdd_changelog_user_purge_cb, (void *)&data, + mdd_changelog_clear_cb, (void *)&mcuc, 0, 0); - if ((rc >= 0) && (data.mcud_minrec > 0)) { - CDEBUG(D_IOCTL, "Purging changelog entries up to "LPD64 - ", referenced by "CHANGELOG_USER_PREFIX"%d\n", - data.mcud_minrec, data.mcud_minid); - rc = mdd_changelog_llog_cancel(env, mdd, data.mcud_minrec); - } else { - CWARN("Could not determine changelog records to purge; rc=%d\n", - rc); - } - - llog_ctxt_put(ctxt); - if (!data.mcud_found) { - CWARN("No entry for user %d. Last changelog reference is " - LPD64" by changelog user %d\n", data.mcud_id, - data.mcud_minrec, data.mcud_minid); - rc = -ENOENT; - } + if (rc < 0) { + CWARN("%s: Failure to clear the changelog for user %d: %d\n", + mdd2obd_dev(mdd)->obd_name, id, rc); + } else if (mcuc.mcuc_flush) { + /* Cancelling record 0 destroys the entire changelog, make sure + we don't do that unless we mean it. */ + if (mcuc.mcuc_minrec != 0) { + CDEBUG(D_IOCTL, "%s: Purging changelog entries up "\ + "to %llu\n", mdd2obd_dev(mdd)->obd_name, + mcuc.mcuc_minrec); + + rc = mdd_changelog_llog_cancel(env, mdd, + mcuc.mcuc_minrec); + } + } else { + CWARN("%s: No entry for user %d\n", + mdd2obd_dev(mdd)->obd_name, id); + rc = -ENOENT; + } - if (!rc && data.mcud_usercount == 0) - /* No more users; turn changelogs off */ - rc = mdd_changelog_on(env, mdd, 0); + EXIT; +out: + if (ctxt != NULL) + llog_ctxt_put(ctxt); - RETURN (rc); + return rc; } /** mdd_iocontrol @@ -1642,77 +1846,91 @@ static int mdd_changelog_user_purge(const struct lu_env *env, static int mdd_iocontrol(const struct lu_env *env, struct md_device *m, unsigned int cmd, int len, void *karg) { - struct mdd_device *mdd; - struct obd_ioctl_data *data = karg; - int rc; - ENTRY; + struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev); + struct obd_ioctl_data *data = karg; + int rc; + ENTRY; - mdd = lu2mdd_dev(&m->md_lu_dev); + /* Doesn't use obd_ioctl_data */ + switch (cmd) { + case OBD_IOC_CHANGELOG_CLEAR: { + struct changelog_setinfo *cs = karg; - /* Doesn't use obd_ioctl_data */ - switch (cmd) { - case OBD_IOC_CHANGELOG_CLEAR: { - struct changelog_setinfo *cs = karg; - rc = mdd_changelog_user_purge(env, mdd, cs->cs_id, - cs->cs_recno); - RETURN(rc); - } - case OBD_IOC_GET_MNTOPT: { - mntopt_t *mntopts = (mntopt_t *)karg; - *mntopts = mdd->mdd_dt_conf.ddp_mntopts; - RETURN(0); - } + if (unlikely(!barrier_entry(mdd->mdd_bottom))) + RETURN(-EINPROGRESS); + + rc = mdd_changelog_clear(env, mdd, cs->cs_id, + cs->cs_recno); + barrier_exit(mdd->mdd_bottom); + RETURN(rc); + } case OBD_IOC_START_LFSCK: { - rc = mdd_lfsck_start(env, &mdd->mdd_lfsck, - (struct lfsck_start *)karg); + rc = lfsck_start(env, mdd->mdd_bottom, + (struct lfsck_start_param *)karg); RETURN(rc); } case OBD_IOC_STOP_LFSCK: { - rc = mdd_lfsck_stop(env, &mdd->mdd_lfsck, false); + rc = lfsck_stop(env, mdd->mdd_bottom, + (struct lfsck_stop *)karg); RETURN(rc); } - case OBD_IOC_PAUSE_LFSCK: { - rc = mdd_lfsck_stop(env, &mdd->mdd_lfsck, true); + case OBD_IOC_QUERY_LFSCK: { + rc = lfsck_query(env, mdd->mdd_bottom, NULL, NULL, + (struct lfsck_query *)karg); RETURN(rc); } - } + } - /* Below ioctls use obd_ioctl_data */ - if (len != sizeof(*data)) { - CERROR("Bad ioctl size %d\n", len); - RETURN(-EINVAL); - } - if (data->ioc_version != OBD_IOCTL_VERSION) { - CERROR("Bad magic %x != %x\n", data->ioc_version, - OBD_IOCTL_VERSION); - RETURN(-EINVAL); - } + /* Below ioctls use obd_ioctl_data */ + if (len != sizeof(*data)) { + CERROR("Bad ioctl size %d\n", len); + RETURN(-EINVAL); + } + if (data->ioc_version != OBD_IOCTL_VERSION) { + CERROR("Bad magic %x != %x\n", data->ioc_version, + OBD_IOCTL_VERSION); + RETURN(-EINVAL); + } - switch (cmd) { - case OBD_IOC_CHANGELOG_REG: - rc = mdd_changelog_user_register(env, mdd, &data->ioc_u32_1); - break; - case OBD_IOC_CHANGELOG_DEREG: - rc = mdd_changelog_user_purge(env, mdd, data->ioc_u32_1, - MCUD_UNREGISTER); - break; - default: - rc = -ENOTTY; - } + switch (cmd) { + case OBD_IOC_CHANGELOG_REG: + if (unlikely(!barrier_entry(mdd->mdd_bottom))) + RETURN(-EINPROGRESS); + + rc = mdd_changelog_user_register(env, mdd, &data->ioc_u32_1); + barrier_exit(mdd->mdd_bottom); + break; + case OBD_IOC_CHANGELOG_DEREG: + if (unlikely(!barrier_entry(mdd->mdd_bottom))) + RETURN(-EINPROGRESS); + + /* explicitly clear changelog first, to protect from crash in + * the middle of purge that would lead to unregistered consumer + * but pending changelog entries + */ + rc = mdd_changelog_clear(env, mdd, data->ioc_u32_1, 0); + if (!rc) + rc = mdd_changelog_user_purge(env, + mdd, data->ioc_u32_1); + + barrier_exit(mdd->mdd_bottom); + break; + default: + rc = -ENOTTY; + } - RETURN (rc); + RETURN(rc); } /* type constructor/destructor: mdd_type_init, mdd_type_fini */ -LU_TYPE_INIT_FINI(mdd, &mdd_thread_key, &mdd_capainfo_key); - -const struct md_device_operations mdd_ops = { - .mdo_statfs = mdd_statfs, - .mdo_root_get = mdd_root_get, - .mdo_init_capa_ctxt = mdd_init_capa_ctxt, - .mdo_update_capa_key= mdd_update_capa_key, - .mdo_llog_ctxt_get = mdd_llog_ctxt_get, - .mdo_iocontrol = mdd_iocontrol, +LU_TYPE_INIT_FINI(mdd, &mdd_thread_key); + +static const struct md_device_operations mdd_ops = { + .mdo_statfs = mdd_statfs, + .mdo_root_get = mdd_root_get, + .mdo_llog_ctxt_get = mdd_llog_ctxt_get, + .mdo_iocontrol = mdd_iocontrol, + .mdo_dtconf_get = mdd_dtconf_get, }; static struct lu_device_type_operations mdd_device_type_ops = { @@ -1739,88 +1957,77 @@ static struct lu_device_type mdd_device_type = { LU_KEY_INIT(mdd, struct mdd_thread_info); static void mdd_key_fini(const struct lu_context *ctx, - struct lu_context_key *key, void *data) + struct lu_context_key *key, void *data) { - struct mdd_thread_info *info = data; - if (info->mti_max_lmm != NULL) - OBD_FREE(info->mti_max_lmm, info->mti_max_lmm_size); - if (info->mti_max_cookie != NULL) - OBD_FREE(info->mti_max_cookie, info->mti_max_cookie_size); - mdd_buf_put(&info->mti_big_buf); - - OBD_FREE_PTR(info); + struct mdd_thread_info *info = data; + + lu_buf_free(&info->mti_big_buf); + lu_buf_free(&info->mti_link_buf); + lu_buf_free(&info->mti_xattr_buf); + + OBD_FREE_PTR(info); } /* context key: mdd_thread_key */ LU_CONTEXT_KEY_DEFINE(mdd, LCT_MD_THREAD); -static struct lu_local_obj_desc llod_capa_key = { - .llod_name = CAPA_KEYS, - .llod_oid = MDD_CAPA_KEYS_OID, - .llod_is_index = 0, -}; +int mdd_generic_thread_start(struct mdd_generic_thread *thread, + int (*func)(void *), void *data, char *name) +{ + struct task_struct *task; -static struct lu_local_obj_desc llod_mdd_orphan = { - .llod_name = orph_index_name, - .llod_oid = MDD_ORPHAN_OID, - .llod_is_index = 1, - .llod_feat = &dt_directory_features, -}; + LASSERT(thread->mgt_init == false); + init_completion(&thread->mgt_started); + init_completion(&thread->mgt_finished); + thread->mgt_data = data; + thread->mgt_abort = false; + thread->mgt_init = true; -static struct lu_local_obj_desc llod_lfsck_bookmark = { - .llod_name = lfsck_bookmark_name, - .llod_oid = LFSCK_BOOKMARK_OID, - .llod_is_index = 0, -}; + task = kthread_run(func, thread, name); + if (IS_ERR(task)) { + complete(&thread->mgt_finished); + return PTR_ERR(task); + } + wait_for_completion(&thread->mgt_started); + return 0; +} -static struct lu_local_obj_desc llod_lfsck_namespace = { - .llod_name = lfsck_namespace_name, - .llod_oid = LFSCK_NAMESPACE_OID, - .llod_is_index = 1, - .llod_feat = &dt_lfsck_features, -}; +void mdd_generic_thread_stop(struct mdd_generic_thread *thread) +{ + if (thread->mgt_init == true) { + thread->mgt_abort = true; + wait_for_completion(&thread->mgt_finished); + } +} -static int __init mdd_mod_init(void) +static int __init mdd_init(void) { - struct lprocfs_static_vars lvars; int rc; - lprocfs_mdd_init_vars(&lvars); - rc = lu_kmem_init(mdd_caches); if (rc) return rc; - changelog_orig_logops = llog_osd_ops; - changelog_orig_logops.lop_cancel = llog_changelog_cancel; - changelog_orig_logops.lop_add = llog_cat_add_rec; - changelog_orig_logops.lop_declare_add = llog_cat_declare_add_rec; + changelog_orig_logops = llog_common_cat_ops; + changelog_orig_logops.lop_write_rec = mdd_changelog_write_rec; - llo_local_obj_register(&llod_capa_key); - llo_local_obj_register(&llod_mdd_orphan); - llo_local_obj_register(&llod_lfsck_bookmark); - llo_local_obj_register(&llod_lfsck_namespace); - - rc = class_register_type(&mdd_obd_device_ops, NULL, lvars.module_vars, + rc = class_register_type(&mdd_obd_device_ops, NULL, false, NULL, LUSTRE_MDD_NAME, &mdd_device_type); if (rc) lu_kmem_fini(mdd_caches); return rc; } -static void __exit mdd_mod_exit(void) +static void __exit mdd_exit(void) { - llo_local_obj_unregister(&llod_capa_key); - llo_local_obj_unregister(&llod_mdd_orphan); - llo_local_obj_unregister(&llod_lfsck_bookmark); - llo_local_obj_unregister(&llod_lfsck_namespace); - class_unregister_type(LUSTRE_MDD_NAME); lu_kmem_fini(mdd_caches); } -MODULE_AUTHOR("Sun Microsystems, Inc. "); -MODULE_DESCRIPTION("Lustre Meta-data Device Prototype ("LUSTRE_MDD_NAME")"); +MODULE_AUTHOR("OpenSFS, Inc. "); +MODULE_DESCRIPTION("Lustre Meta-data Device Driver ("LUSTRE_MDD_NAME")"); +MODULE_VERSION(LUSTRE_VERSION_STRING); MODULE_LICENSE("GPL"); -cfs_module(mdd, "0.1.0", mdd_mod_init, mdd_mod_exit); +module_init(mdd_init); +module_exit(mdd_exit);