X-Git-Url: https://git.whamcloud.com/?p=fs%2Flustre-release.git;a=blobdiff_plain;f=lustre%2Fmdd%2Fmdd_device.c;h=f24c2086c7b5fcf4d973b986335061b495a1ad38;hp=d43bf3b0cf2f973b0c63f69e1c7706e871461a44;hb=ffe259f81cda5b5cd9241362308ec26ebad194b8;hpb=a9e0b853aabc1c817e3f4c250714ed86bb76a1ee

diff --git a/lustre/mdd/mdd_device.c b/lustre/mdd/mdd_device.c
index d43bf3b..f24c208 100644
--- a/lustre/mdd/mdd_device.c
+++ b/lustre/mdd/mdd_device.c
@@ -23,11 +23,10 @@
  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  * Use is subject to license terms.
  *
- * Copyright (c) 2011, 2016, Intel Corporation.
+ * Copyright (c) 2011, 2017, Intel Corporation.
  */
 /*
  * This file is part of Lustre, http://www.lustre.org/
- * Lustre is a trademark of Sun Microsystems, Inc.
  *
  * lustre/mdd/mdd_device.c
  *
@@ -120,11 +119,16 @@ static int mdd_init0(const struct lu_env *env, struct mdd_device *mdd,
 {
 	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;
+	/* per-server mask is set via parameters if needed */
+	mdd->mdd_cl.mc_proc_mask = CHANGELOG_MINMASK;
+	/* current mask is calculated from mask above and users masks */
+	mdd->mdd_cl.mc_current_mask = CHANGELOG_MINMASK;
+	mdd->mdd_cl.mc_deniednext = 60; /* 60 secs by default */
 
 	dev = lustre_cfg_string(lcfg, 0);
 	if (dev == NULL)
@@ -141,8 +145,21 @@ static int mdd_init0(const struct lu_env *env, struct mdd_device *mdd,
 		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);
 
@@ -182,67 +199,226 @@ static int changelog_init_cb(const struct lu_env *env, struct llog_handle *llh,
 	       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));
-
+	spin_lock(&mdd->mdd_cl.mc_lock);
 	mdd->mdd_cl.mc_index = rec->cr.cr_index;
+	spin_unlock(&mdd->mdd_cl.mc_lock);
 	return LLOG_PROC_BREAK;
 }
 
+char *mdd_chlg_username(struct llog_changelog_user_rec2 *rec, char *buf,
+			size_t len)
+{
+	if (rec->cur_hdr.lrh_type == CHANGELOG_USER_REC2 &&
+	    rec->cur_name[0])
+		snprintf(buf, len, "%s%u-%s",  CHANGELOG_USER_PREFIX,
+			 rec->cur_id, rec->cur_name);
+	else
+		snprintf(buf, len, "%s%u", CHANGELOG_USER_PREFIX, rec->cur_id);
+	return buf;
+}
+
+__u32 mdd_chlg_usermask(struct llog_changelog_user_rec2 *rec)
+{
+	return rec->cur_hdr.lrh_type == CHANGELOG_USER_REC2 ?
+	       rec->cur_mask : 0;
+}
+
 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 = data;
+	struct llog_changelog_user_rec2 *rec;
+	char user_name[CHANGELOG_USER_NAMELEN_FULL];
 
 	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=%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));
+	rec = container_of(hdr, typeof(*rec), cur_hdr);
+	if (rec->cur_hdr.lrh_type != CHANGELOG_USER_REC &&
+	    rec->cur_hdr.lrh_type != CHANGELOG_USER_REC2) {
+		CWARN("%s: unknown user type %x at index %u in log "DFID"\n",
+		      mdd2obd_dev(mdd)->obd_name, hdr->lrh_index,
+		      rec->cur_hdr.lrh_type, PFID(&llh->lgh_id.lgl_oi.oi_fid));
+
+		return 0;
+	}
+
+	CDEBUG(D_INFO, "%s: user %s at index %u/%u endrec=%llu in log "DFID"\n",
+	       mdd2obd_dev(mdd)->obd_name, mdd_chlg_username(rec, user_name,
+							     sizeof(user_name)),
+	       hdr->lrh_index, rec->cur_hdr.lrh_index, 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_hdr.lrh_type == CHANGELOG_USER_REC2 && rec->cur_mask)
+		mdd->mdd_cl.mc_current_mask |= rec->cur_mask;
+	else if (mdd->mdd_cl.mc_proc_mask == CHANGELOG_MINMASK)
+		mdd->mdd_cl.mc_current_mask |= CHANGELOG_DEFMASK;
+	spin_unlock(&mdd->mdd_cl.mc_user_lock);
+	spin_lock(&mdd->mdd_cl.mc_lock);
 	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);
+	spin_unlock(&mdd->mdd_cl.mc_lock);
+
+	return LLOG_PROC_BREAK;
+}
+
+struct changelog_orphan_data {
+	__u64			clod_index;
+	struct mdd_device	*clod_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 changelog_orphan_data *clod = data;
+	struct mdd_device *mdd = clod->clod_mdd;
+	struct llog_changelog_rec *rec;
+
+	LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
+
+	rec = container_of(hdr, typeof(*rec), cr_hdr);
+	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: 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));
+
+	clod->clod_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 changelog_orphan_data *clod = data;
+	struct mdd_device *mdd = clod->clod_mdd;
+	struct llog_changelog_user_rec2 *rec;
+	char user_name[CHANGELOG_USER_NAMELEN_FULL];
+
+	LASSERT(llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN);
+
+	rec = container_of(hdr, typeof(*rec), cur_hdr);
+	if (rec->cur_hdr.lrh_type != CHANGELOG_USER_REC &&
+	    rec->cur_hdr.lrh_type != CHANGELOG_USER_REC2) {
+		CWARN("%s: unknown user type %u at index %u in log "DFID"\n",
+		      mdd2obd_dev(mdd)->obd_name, hdr->lrh_index,
+		      rec->cur_hdr.lrh_type, 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: user %s at index %u/%u endrec=%llu in log "DFID"\n",
+	       mdd2obd_dev(mdd)->obd_name, mdd_chlg_username(rec, user_name,
+							     sizeof(user_name)),
+	       hdr->lrh_index, rec->cur_hdr.lrh_index,
+	       rec->cur_endrec, PFID(&llh->lgh_id.lgl_oi.oi_fid));
+
+	clod->clod_index = min_t(__u64, clod->clod_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;
-	int			 rc;
+	struct changelog_cancel_cookie *cl_cookie =
+		(struct changelog_cancel_cookie *)data;
 
 	ENTRY;
 
 	/* 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);
 
-	cookie.lgc_lgl = llh->lgh_id;
-	cookie.lgc_index = hdr->lrh_index;
+	if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_CHANGELOG_RACE))) {
+		if (cfs_fail_val == 0)
+			cfs_fail_val = hdr->lrh_index;
+		if (cfs_fail_val == hdr->lrh_index)
+			OBD_RACE(OBD_FAIL_MDS_CHANGELOG_RACE);
+	}
+
+	/* Records folow one by one, cr_index++. We could calculate the
+	 * last cr_index at this plain llog. And if it less then cookie endrec
+	 * cancel the whole file.
+	 */
+	if ((LLOG_HDR_BITMAP_SIZE(llh->lgh_hdr) - hdr->lrh_index +
+	     rec->cr.cr_index) < cl_cookie->endrec) {
+		int rc;
+
+		if (unlikely(OBD_FAIL_PRECHECK(OBD_FAIL_MDS_CHANGELOG_DEL))) {
+			if (cfs_fail_val == 0) {
+				cfs_fail_val = (unsigned long)llh & 0xFFFFFFFF;
+				OBD_RACE(OBD_FAIL_MDS_CHANGELOG_DEL);
+			}
+		}
+		rc = llog_destroy(env, llh);
+		if (!rc) {
+			CDEBUG(D_HA, "Changelog destroyed plain "DFID"\n",
+			       PFID(&llh->lgh_id.lgl_oi.oi_fid));
+			RETURN(LLOG_DEL_PLAIN);
+		}
+	}
 
 	/* cancel them one at a time.  I suppose we could store up the cookies
 	 * and cancel them all at once; probably more efficient, but this is
 	 * done as a user call, so who cares... */
-	rc = llog_cat_cancel_records(env, llh->u.phd.phd_cat_handle, 1,
-				     &cookie);
-	RETURN(rc < 0 ? rc : 0);
+
+	RETURN(LLOG_DEL_RECORD);
 }
 
 static int llog_changelog_cancel(const struct lu_env *env,
 				 struct llog_ctxt *ctxt,
-				 struct llog_cookie *cookies, int flags)
+				 struct changelog_cancel_cookie *cookie)
 {
 	struct llog_handle	*cathandle = ctxt->loc_handle;
 	int			 rc;
@@ -253,7 +429,7 @@ 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;
@@ -314,9 +490,16 @@ mdd_changelog_off(const struct lu_env *env, struct mdd_device *mdd)
 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;
-	int			 rc;
+	struct obd_device *obd = mdd2obd_dev(mdd);
+	struct llog_ctxt *ctxt = NULL, *uctxt = NULL;
+	struct changelog_orphan_data clod = {
+		.clod_mdd = mdd,
+		.clod_index = -1,
+	}, user_orphan = {
+		.clod_mdd = mdd,
+		.clod_index = -1,
+	};
+	int rc;
 
 	ENTRY;
 
@@ -375,9 +558,6 @@ 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_init_handle(env, uctxt->loc_handle, LLOG_F_IS_CAT, NULL);
 	if (rc)
 		GOTO(out_uclose, rc);
@@ -390,12 +570,60 @@ static int mdd_changelog_llog_init(const struct lu_env *env,
 		GOTO(out_uclose, rc);
 	}
 
+	/* Finally apply per-server mask */
+	mdd->mdd_cl.mc_current_mask |= mdd->mdd_cl.mc_proc_mask;
+
 	/* If we have registered users, assume we want changelogs on */
 	if (mdd->mdd_cl.mc_lastuser > 0) {
 		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,
+			      &clod, 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(clod.clod_index < user_orphan.clod_index)) {
+		struct changelog_cancel_cookie cl_cookie = {
+			.endrec = user_orphan.clod_index,
+			.mdd = mdd,
+		};
+
+		CWARN("%s : orphan changelog records found, starting from "
+		      "index %llu to index %llu, being cleared now\n",
+		      obd->obd_name, clod.clod_index, user_orphan.clod_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);
@@ -421,6 +649,10 @@ static int mdd_changelog_init(const struct lu_env *env, struct mdd_device *mdd)
 	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",
@@ -437,7 +669,42 @@ static void mdd_changelog_fini(const struct lu_env *env,
 	struct obd_device	*obd = mdd2obd_dev(mdd);
 	struct llog_ctxt	*ctxt;
 
-	mdd->mdd_cl.mc_flags = 0;
+	if (mdd->mdd_cl.mc_flags & CLM_CLEANUP_DONE)
+		return;
+	mdd->mdd_cl.mc_flags = CLM_CLEANUP_DONE;
+
+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) {
@@ -464,6 +731,7 @@ mdd_changelog_llog_cancel(const struct lu_env *env, struct mdd_device *mdd,
         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);
@@ -496,7 +764,9 @@ mdd_changelog_llog_cancel(const struct lu_env *env, struct mdd_device *mdd,
            changed since the last purge) */
 	mdd->mdd_cl.mc_starttime = ktime_get();
 
-	rc = llog_cancel(env, ctxt, (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;
@@ -520,13 +790,13 @@ int mdd_changelog_write_header(const struct lu_env *env,
 
 	ENTRY;
 
-	if (mdd->mdd_cl.mc_mask & (1 << CL_MARK)) {
+	if (mdd->mdd_cl.mc_current_mask & BIT(CL_MARK)) {
 		mdd->mdd_cl.mc_starttime = ktime_get();
 		RETURN(0);
 	}
 
 	reclen = llog_data_len(sizeof(*rec) + len);
-	buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mti_big_buf, reclen);
+	buf = lu_buf_check_and_alloc(&mdd_env_info(env)->mdi_chlg_buf, reclen);
 	if (buf->lb_buf == NULL)
 		RETURN(-ENOMEM);
 	rec = buf->lb_buf;
@@ -541,9 +811,6 @@ int mdd_changelog_write_header(const struct lu_env *env,
 					    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);
@@ -575,23 +842,12 @@ static int obf_lookup(const struct lu_env *env, struct md_object *p,
                 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 (!fid_is_sane(f))
+		GOTO(out, rc = -ENOENT);
 
 	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);
-	}
+	    !fid_seq_is_dot(f->f_seq))
+		GOTO(out, rc = -ENOENT);
 
         /* Check if object with this fid exists */
         child = mdd_object_find(env, mdd, f);
@@ -648,15 +904,27 @@ static int mdd_dummy_unlink(const struct lu_env *env,
 	return -EPERM;
 }
 
-static struct md_dir_operations mdd_obf_dir_ops = {
+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)
+{
+	if (spec->sp_cr_flags & MDS_OPEN_VOLATILE)
+		return mdd_create(env, pobj, lname, child, spec, ma);
+	RETURN(-EPERM);
+}
+
+static const struct md_dir_operations mdd_obf_dir_ops = {
 	.mdo_lookup = obf_lookup,
-	.mdo_create = mdd_dummy_create,
+	.mdo_create = mdd_obf_create,
 	.mdo_rename = mdd_dummy_rename,
 	.mdo_link   = mdd_dummy_link,
 	.mdo_unlink = mdd_dummy_unlink
 };
 
-static struct md_dir_operations mdd_lpf_dir_ops = {
+static const struct md_dir_operations mdd_lpf_dir_ops = {
 	.mdo_lookup = mdd_lookup,
 	.mdo_create = mdd_dummy_create,
 	.mdo_rename = mdd_dummy_rename,
@@ -677,7 +945,7 @@ static struct md_object *mdo_locate(const struct lu_env *env,
 		LASSERT(obj != NULL);
 		mdo = lu2md(obj);
 	} else {
-		mdo = ERR_PTR(PTR_ERR(obj));
+		mdo = ERR_CAST(obj);
 	}
 	return mdo;
 }
@@ -799,9 +1067,6 @@ out:
 	return rc;
 }
 
-
-static struct llog_operations hsm_actions_logops;
-
 /**
  * set llog methods and create LLOG_AGENT_ORIG_CTXT llog
  * object in obd_device
@@ -818,7 +1083,7 @@ static int mdd_hsm_actions_llog_init(const struct lu_env *env,
 	obd->obd_lvfs_ctxt.dt = m->mdd_bottom;
 
 	rc = llog_setup(env, obd, &obd->obd_olg, LLOG_AGENT_ORIG_CTXT,
-			obd, &hsm_actions_logops);
+			obd, &llog_common_cat_ops);
 	if (rc) {
 		CERROR("%s: hsm actions llog setup failed: rc = %d\n",
 			obd->obd_name, rc);
@@ -879,7 +1144,7 @@ static void mdd_device_shutdown(const struct lu_env *env, struct mdd_device *m,
 	lfsck_degister(env, m->mdd_bottom);
 	mdd_hsm_actions_llog_fini(env, m);
 	mdd_changelog_fini(env, m);
-	orph_index_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,
@@ -907,10 +1172,11 @@ static int mdd_process_config(const struct lu_env *env,
 
         switch (cfg->lcfg_command) {
 	case LCFG_PARAM: {
-		struct obd_device *obd = mdd2obd_dev(m);
+		ssize_t count;
 
-		rc = class_process_proc_param(PARAM_MDD, obd->obd_vars, cfg, m);
-		if (rc > 0 || rc == -ENOSYS)
+		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;
@@ -923,7 +1189,7 @@ static int mdd_process_config(const struct lu_env *env,
                 break;
 	case LCFG_PRE_CLEANUP:
 		rc = next->ld_ops->ldo_process_config(env, next, cfg);
-		mdd_generic_thread_stop(&m->mdd_orph_cleanup_thread);
+		mdd_generic_thread_stop(&m->mdd_orphan_cleanup_thread);
 		break;
 	case LCFG_CLEANUP:
 		rc = next->ld_ops->ldo_process_config(env, next, cfg);
@@ -939,22 +1205,21 @@ 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. */
 	if (!mdd->mdd_bottom->dd_rdonly)
 		mdd_orphan_cleanup(env, mdd);
-        rc = next->ld_ops->ldo_recovery_complete(env, next);
+	rc = next->ld_ops->ldo_recovery_complete(env, next);
 
-        RETURN(rc);
+	RETURN(rc);
 }
 
 int mdd_local_file_create(const struct lu_env *env, struct mdd_device *mdd,
@@ -1005,6 +1270,8 @@ static int mdd_prepare(const struct lu_env *env,
 	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 dt_object *root = NULL;
+	struct thandle *th = NULL;
 	struct lu_fid fid;
 	int rc;
 
@@ -1038,6 +1305,58 @@ static int mdd_prepare(const struct lu_env *env,
 			GOTO(out_los, rc);
 		}
 
+		/* store a default directory layout on the root directory if
+		 * it doesn't already exist to improve MDT space balance.
+		 */
+		root = dt_locate(env, mdd->mdd_bottom, &fid);
+		if (unlikely(IS_ERR(root)))
+			GOTO(out_los, rc = PTR_ERR(root));
+
+		rc = dt_xattr_get(env, root, &LU_BUF_NULL,
+				  XATTR_NAME_DEFAULT_LMV);
+		if (rc == -ENODATA) {
+			struct lu_buf buf;
+			struct lmv_user_md lmv_default = {
+				.lum_magic		= LMV_USER_MAGIC,
+				.lum_stripe_count	= 1,
+				.lum_stripe_offset	= LMV_OFFSET_DEFAULT,
+				.lum_max_inherit	= LMV_INHERIT_UNLIMITED,
+				.lum_max_inherit_rr	= LMV_INHERIT_RR_ROOT,
+			};
+
+			th = dt_trans_create(env, mdd->mdd_bottom);
+			if (IS_ERR(th))
+				GOTO(out_root_put, rc = PTR_ERR(th));
+
+			buf.lb_buf = &lmv_default;
+			buf.lb_len = sizeof(lmv_default);
+			rc = dt_declare_xattr_set(env, root, &buf,
+						  XATTR_NAME_DEFAULT_LMV, 0,
+						  th);
+			if (rc)
+				GOTO(out_trans_stop, rc);
+
+			rc = dt_trans_start_local(env, mdd->mdd_bottom, th);
+			if (rc)
+				GOTO(out_trans_stop, rc);
+
+			rc = dt_xattr_set(env, root, &buf,
+					  XATTR_NAME_DEFAULT_LMV, 0, th);
+			if (rc)
+				GOTO(out_trans_stop, rc);
+
+			dt_trans_stop(env, mdd->mdd_bottom, th);
+			th = NULL;
+		} else if (rc < 0 && rc != -ERANGE) {
+			CERROR("%s: get default LMV of root failed: rc = %d\n",
+			       mdd2obd_dev(mdd)->obd_name, rc);
+
+			GOTO(out_root_put, rc);
+		}
+
+		dt_object_put(env, root);
+		root = NULL;
+
 		mdd->mdd_root_fid = fid;
 		rc = mdd_dot_lustre_setup(env, mdd);
 		if (rc != 0) {
@@ -1051,7 +1370,7 @@ static int mdd_prepare(const struct lu_env *env,
 		mdd->mdd_root_fid = fid;
 	}
 
-	rc = orph_index_init(env, mdd);
+	rc = mdd_orphan_index_init(env, mdd);
 	if (rc < 0)
 		GOTO(out_dot, rc);
 
@@ -1104,10 +1423,16 @@ out_hsm:
 out_changelog:
 	mdd_changelog_fini(env, mdd);
 out_orph:
-	orph_index_fini(env, mdd);
+	mdd_orphan_index_fini(env, mdd);
 out_dot:
 	if (mdd_seq_site(mdd)->ss_node_id == 0)
 		mdd_dot_lustre_cleanup(env, mdd);
+out_trans_stop:
+	if (th != NULL)
+		dt_trans_stop(env, mdd->mdd_bottom, th);
+out_root_put:
+	if (root != NULL)
+		dt_object_put(env, root);
 out_los:
 	local_oid_storage_fini(env, mdd->mdd_los);
 	mdd->mdd_los = NULL;
@@ -1115,11 +1440,30 @@ out_los:
 	return rc;
 }
 
+/**
+ * Implementation of lu_device_operations::ldo_fid_alloc() for MDD.
+ *
+ * Find corresponding device by passed parent and name, and allocate FID from
+ * there.
+ *
+ * see include/lu_object.h for the details.
+ */
+static int mdd_fid_alloc(const struct lu_env *env, struct lu_device *d,
+			 struct lu_fid *fid, struct lu_object *parent,
+			 const struct lu_name *name)
+{
+	struct mdd_device *mdd = lu2mdd_dev(d);
+	struct lu_object *o = lu_object_next(parent);
+
+	return dt_fid_alloc(env, mdd->mdd_child, fid, o, name);
+}
+
 const struct lu_device_operations mdd_lu_ops = {
         .ldo_object_alloc      = mdd_object_alloc,
         .ldo_process_config    = mdd_process_config,
         .ldo_recovery_complete = mdd_recovery_complete,
         .ldo_prepare           = mdd_prepare,
+	.ldo_fid_alloc	       = mdd_fid_alloc,
 };
 
 static int mdd_root_get(const struct lu_env *env,
@@ -1143,22 +1487,19 @@ static int mdd_statfs(const struct lu_env *env, struct md_device *m,
 
 	ENTRY;
 
-	rc = mdd_child_ops(mdd)->dt_statfs(env, mdd->mdd_child, sfs);
+	rc = mdd_child_ops(mdd)->dt_statfs(env, mdd->mdd_child, sfs, NULL);
 
 	sfs->os_namelen = min_t(__u32, sfs->os_namelen, NAME_MAX);
 
 	RETURN(rc);
 }
 
-static int mdd_maxeasize_get(const struct lu_env *env, struct md_device *m,
-				int *easize)
+static const struct dt_device_param *mdd_dtconf_get(const struct lu_env *env,
+						    struct md_device *m)
 {
 	struct mdd_device *mdd = lu2mdd_dev(&m->md_lu_dev);
-	ENTRY;
 
-	*easize = mdd->mdd_dt_conf.ddp_max_ea_size;
-
-	RETURN(0);
+	return &mdd->mdd_dt_conf;
 }
 
 static int mdd_llog_ctxt_get(const struct lu_env *env, struct md_device *m,
@@ -1298,7 +1639,7 @@ static int mdd_obd_set_info_async(const struct lu_env *env,
 	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,
@@ -1306,52 +1647,256 @@ static struct obd_ops mdd_obd_device_ops = {
 	.o_set_info_async = mdd_obd_set_info_async,
 };
 
+struct mdd_changelog_name_check_data {
+	const char *mcnc_name;
+	__u32	    mcnc_id;
+};
+
+/**
+ * changelog_recalc_mask callback
+ *
+ * Is is called per each registered user and calculates combined mask of
+ * all registered users.
+ */
+static int mdd_changelog_name_check_cb(const struct lu_env *env,
+				       struct llog_handle *llh,
+				       struct llog_rec_hdr *hdr, void *data)
+{
+	struct llog_changelog_user_rec2 *rec;
+	struct mdd_changelog_name_check_data *mcnc = data;
+
+	rec = container_of(hdr, typeof(*rec), cur_hdr);
+	if (rec->cur_hdr.lrh_type == CHANGELOG_USER_REC2 &&
+	    !strncmp(rec->cur_name, mcnc->mcnc_name, sizeof(rec->cur_name))) {
+		mcnc->mcnc_id = rec->cur_id;
+		return -EEXIST;
+	}
+	return 0;
+}
+
+static int mdd_changelog_name_check(const struct lu_env *env,
+				    struct llog_ctxt *ctxt,
+				    struct mdd_device *mdd, const char *name)
+{
+	struct mdd_changelog_name_check_data mcnc = { .mcnc_name = name, };
+	int chr = 0;
+	int rc;
+
+	ENTRY;
+
+	/* first symbol is a letter */
+	if (!isalpha(name[0])) {
+		rc = -EINVAL;
+		CERROR("%s: first char '%c' in '%s' is not letter: rc = %d\n",
+		       mdd2obd_dev(mdd)->obd_name, name[0], name, rc);
+		RETURN(rc);
+	}
+
+	/* name is valid: contains letters, numbers and '-', '_' only */
+	while (name[++chr]) {
+		if (!(isalnum(name[chr]) || name[chr] == '_' ||
+		      name[chr] == '-')) {
+			rc = -EINVAL;
+			CERROR("%s: wrong char '%c' in name '%s': rc = %d\n",
+			       mdd2obd_dev(mdd)->obd_name, name[chr], name, rc);
+			RETURN(rc);
+		}
+	}
+
+	if (chr > CHANGELOG_USER_NAMELEN) {
+		rc = -ENAMETOOLONG;
+		CERROR("%s: name '%s' is over %d symbols limit: rc = %d\n",
+		       mdd2obd_dev(mdd)->obd_name, name,
+		       CHANGELOG_USER_NAMELEN, rc);
+		RETURN(rc);
+	}
+
+	rc = llog_cat_process(env, ctxt->loc_handle,
+			      mdd_changelog_name_check_cb, &mcnc, 0, 0);
+	if (rc == -EEXIST)
+		CWARN("%s: changelog name %s exists already: rc = %d\n",
+		      mdd2obd_dev(mdd)->obd_name, name, rc);
+	else if (rc < 0)
+		CWARN("%s: failed user changelog processing: rc = %d\n",
+		      mdd2obd_dev(mdd)->obd_name, rc);
+	RETURN(rc);
+}
+
 static int mdd_changelog_user_register(const struct lu_env *env,
-				       struct mdd_device *mdd, int *id)
+				       struct mdd_device *mdd, int *id,
+				       const char *name, const char *mask)
 {
-        struct llog_ctxt *ctxt;
-        struct llog_changelog_user_rec *rec;
-        int rc;
-        ENTRY;
+	struct llog_ctxt *ctxt;
+	struct llog_changelog_user_rec2 *rec;
+	char user_name[CHANGELOG_USER_NAMELEN_FULL];
+	int rc;
+
+	ENTRY;
 
-        ctxt = llog_get_context(mdd2obd_dev(mdd),
+	ctxt = llog_get_context(mdd2obd_dev(mdd),
 				LLOG_CHANGELOG_USER_ORIG_CTXT);
-        if (ctxt == NULL)
-                RETURN(-ENXIO);
+	if (ctxt == NULL)
+		RETURN(-ENXIO);
 
-        OBD_ALLOC_PTR(rec);
-        if (rec == NULL) {
-                llog_ctxt_put(ctxt);
-                RETURN(-ENOMEM);
-        }
+	OBD_ALLOC_PTR(rec);
+	if (rec == NULL) {
+		llog_ctxt_put(ctxt);
+		RETURN(-ENOMEM);
+	}
 
-	/* Assume we want it on since somebody registered */
-	rc = mdd_changelog_on(env, mdd);
-	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;
+	rec->cur_hdr.lrh_len = sizeof(*rec);
+	/* keep old record type for users without mask/name for
+	 * compatibility needs
+	 */
+	if (mask || (name && name[0]))
+		rec->cur_hdr.lrh_type = CHANGELOG_USER_REC2;
+	else
+		rec->cur_hdr.lrh_type = CHANGELOG_USER_REC;
 	spin_lock(&mdd->mdd_cl.mc_user_lock);
 	if (mdd->mdd_cl.mc_lastuser == (unsigned int)(-1)) {
 		spin_unlock(&mdd->mdd_cl.mc_user_lock);
-		CERROR("Maximum number of changelog users exceeded!\n");
-		GOTO(out, rc = -EOVERFLOW);
+		rc = -EOVERFLOW;
+		CERROR("%s: registering %s user: max ID is exceeded: rc = %d\n",
+		       mdd2obd_dev(mdd)->obd_name,
+		       (name && name[0]) ? name : "new", rc);
+		GOTO(out, rc);
 	}
 	*id = rec->cur_id = ++mdd->mdd_cl.mc_lastuser;
-	rec->cur_endrec = mdd->mdd_cl.mc_index;
+	mdd->mdd_cl.mc_users++;
 	spin_unlock(&mdd->mdd_cl.mc_user_lock);
 
+	rec->cur_time = (__u32)ktime_get_real_seconds();
+	if (OBD_FAIL_CHECK(OBD_FAIL_TIME_IN_CHLOG_USER))
+		rec->cur_time = 0;
+
+	spin_lock(&mdd->mdd_cl.mc_lock);
+	rec->cur_endrec = mdd->mdd_cl.mc_index;
+	spin_unlock(&mdd->mdd_cl.mc_lock);
+
+	if (mask) {
+		/* if user will use relative mask apply it on default one */
+		rec->cur_mask = CHANGELOG_DEFMASK;
+		rc = cfs_str2mask(mask, changelog_type2str, &rec->cur_mask,
+				  CHANGELOG_MINMASK, CHANGELOG_ALLMASK);
+		if (rc)
+			GOTO(out_users, rc);
+	} else if (mdd->mdd_cl.mc_proc_mask == CHANGELOG_MINMASK) {
+		/* a maskless users means default mask but only if server has
+		 * no specific mask set
+		 */
+		rec->cur_mask = CHANGELOG_DEFMASK;
+	}
+
+	if (name && name[0]) {
+		rc = mdd_changelog_name_check(env, ctxt, mdd, name);
+		if (rc)
+			GOTO(out_users, rc);
+		strlcpy(rec->cur_name, name, sizeof(rec->cur_name));
+	}
+	mdd_chlg_username(rec, user_name, sizeof(user_name));
+
 	rc = llog_cat_add(env, ctxt->loc_handle, &rec->cur_hdr, NULL);
+	if (rc) {
+		CWARN("%s: failed to register changelog user %s: rc = %d\n",
+		      mdd2obd_dev(mdd)->obd_name, user_name, rc);
+		GOTO(out_users, rc);
+	}
 
-        CDEBUG(D_IOCTL, "Registered changelog user %d\n", *id);
+	/* apply user mask finally */
+	spin_lock(&mdd->mdd_cl.mc_user_lock);
+	mdd->mdd_cl.mc_current_mask |= rec->cur_mask;
+	spin_unlock(&mdd->mdd_cl.mc_user_lock);
+
+	CDEBUG(D_IOCTL, "%s: registered changelog user '%s', mask %#x\n",
+	       mdd2obd_dev(mdd)->obd_name, user_name, rec->cur_mask);
+
+	/* Assume we want it on since somebody registered */
+	rc = mdd_changelog_on(env, mdd);
+	if (rc)
+		/* record is added, so don't decrement users on error */
+		GOTO(out, rc);
+out_users:
+	if (rc) {
+		spin_lock(&mdd->mdd_cl.mc_user_lock);
+		mdd->mdd_cl.mc_users--;
+		spin_unlock(&mdd->mdd_cl.mc_user_lock);
+	}
 out:
-        OBD_FREE_PTR(rec);
-        llog_ctxt_put(ctxt);
-        RETURN(rc);
+	OBD_FREE_PTR(rec);
+	llog_ctxt_put(ctxt);
+	RETURN(rc);
+}
+
+struct mdd_changelog_recalc_mask_data {
+	struct mdd_device *mcrm_mdd;
+	__u32		   mcrm_mask;
+};
+
+/**
+ * changelog_recalc_mask callback
+ *
+ * Is is called per each registered user and calculates combined mask of
+ * all registered users.
+ */
+static int mdd_changelog_recalc_mask_cb(const struct lu_env *env,
+				       struct llog_handle *llh,
+				       struct llog_rec_hdr *hdr, void *data)
+{
+	struct llog_changelog_user_rec2 *rec;
+	struct mdd_changelog_recalc_mask_data *mcrm = data;
+
+	rec = container_of(hdr, typeof(*rec), cur_hdr);
+	if (rec->cur_hdr.lrh_type == CHANGELOG_USER_REC2 && rec->cur_mask)
+		mcrm->mcrm_mask |= rec->cur_mask;
+	else if (mcrm->mcrm_mdd->mdd_cl.mc_proc_mask == CHANGELOG_MINMASK)
+		mcrm->mcrm_mask |= CHANGELOG_DEFMASK;
+
+	return 0;
+}
+
+int mdd_changelog_recalc_mask(const struct lu_env *env, struct mdd_device *mdd)
+{
+	struct llog_ctxt *ctxt;
+	struct mdd_changelog_recalc_mask_data mcrm = {
+		.mcrm_mdd = mdd,
+		.mcrm_mask = mdd->mdd_cl.mc_proc_mask,
+	};
+	int rc;
+
+	ENTRY;
+
+	ctxt = llog_get_context(mdd2obd_dev(mdd),
+				LLOG_CHANGELOG_USER_ORIG_CTXT);
+	if (!ctxt)
+		RETURN(-ENXIO);
+
+	if (!(ctxt->loc_handle->lgh_hdr->llh_flags & LLOG_F_IS_CAT))
+		GOTO(out, rc = -ENXIO);
+
+	rc = llog_cat_process(env, ctxt->loc_handle,
+			      mdd_changelog_recalc_mask_cb, &mcrm, 0, 0);
+	if (rc < 0)
+		CWARN("%s: failed user changelog processing: rc = %d\n",
+		      mdd2obd_dev(mdd)->obd_name, rc);
+
+	spin_lock(&mdd->mdd_cl.mc_user_lock);
+	CDEBUG(D_INFO, "%s: recalc changelog mask: %#x -> %#x\n",
+	       mdd2obd_dev(mdd)->obd_name, mdd->mdd_cl.mc_current_mask,
+	       mcrm.mcrm_mask);
+	mdd->mdd_cl.mc_current_mask = mcrm.mcrm_mask;
+	spin_unlock(&mdd->mdd_cl.mc_user_lock);
+
+	EXIT;
+out:
+	llog_ctxt_put(ctxt);
+
+	return rc;
 }
 
 struct mdd_changelog_user_purge {
+	struct mdd_device *mcup_mdd;
 	__u32 mcup_id;
 	__u32 mcup_usercount;
 	__u64 mcup_minrec;
@@ -1371,24 +1916,23 @@ 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_purge	*mcup = data;
-	struct llog_cookie               cookie;
-	int				 rc;
+	struct llog_changelog_user_rec2 *rec;
+	struct mdd_changelog_user_purge *mcup = data;
+	struct llog_cookie cookie;
+	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);
+	rec = container_of(hdr, typeof(*rec), cur_hdr);
 
 	mcup->mcup_usercount++;
 
 	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);
+		mcup->mcup_minrec = min(mcup->mcup_minrec, rec->cur_endrec);
 		RETURN(0);
 	}
 
@@ -1401,15 +1945,19 @@ static int mdd_changelog_user_purge_cb(const struct lu_env *env,
 	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);
 }
 
-static int mdd_changelog_user_purge(const struct lu_env *env,
-				    struct mdd_device *mdd, __u32 id)
+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,
@@ -1433,6 +1981,18 @@ static int mdd_changelog_user_purge(const struct lu_env *env,
 			      mdd_changelog_user_purge_cb, &mcup,
 			      0, 0);
 
+	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",
@@ -1449,11 +2009,7 @@ static int mdd_changelog_user_purge(const struct lu_env *env,
 		GOTO(out, rc = -ENOENT);
 	}
 
-	if ((rc == 0) && (mcup.mcup_usercount == 0)) {
-		/* No more users; turn changelogs off */
-		CDEBUG(D_IOCTL, "turning off changelogs\n");
-		rc = mdd_changelog_off(env, mdd);
-	}
+	CFS_RACE(CFS_FAIL_CHLOG_USER_REG_UNREG_RACE);
 
 	EXIT;
 out:
@@ -1468,6 +2024,7 @@ struct mdd_changelog_user_clear {
 	__u64 mcuc_minrec;
 	__u32 mcuc_id;
 	bool mcuc_flush;
+	struct mdd_device *mcuc_mdd;
 };
 
 /**
@@ -1484,8 +2041,10 @@ static int mdd_changelog_clear_cb(const struct lu_env *env,
 				  struct llog_rec_hdr *hdr,
 				  void *data)
 {
-	struct llog_changelog_user_rec *rec;
+	struct llog_changelog_user_rec2 *rec;
 	struct mdd_changelog_user_clear *mcuc = data;
+	char user_name[CHANGELOG_USER_NAMELEN_FULL];
+	struct mdd_device *mdd = mcuc->mcuc_mdd;
 	int rc;
 
 	ENTRY;
@@ -1493,8 +2052,7 @@ static int mdd_changelog_clear_cb(const struct lu_env *env,
 	if ((llh->lgh_hdr->llh_flags & LLOG_F_IS_PLAIN) == 0)
 		RETURN(-ENXIO);
 
-	rec = container_of(hdr, struct llog_changelog_user_rec, cur_hdr);
-
+	rec = container_of(hdr, typeof(*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,
@@ -1504,19 +2062,31 @@ static int mdd_changelog_clear_cb(const struct lu_env *env,
 
 	/* 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);
+		rc = -EINVAL;
+		CDEBUG(D_IOCTL,
+		       "%s: request %llu > endrec %llu for user %s: rc = %d\n",
+		       mdd2obd_dev(mdd)->obd_name,
+		       mcuc->mcuc_endrec, rec->cur_endrec,
+		       mdd_chlg_username(rec, user_name, sizeof(user_name)),
+		       rc);
+		RETURN(rc);
 	}
 
 	/* 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);
+	CDEBUG(D_IOCTL, "%s: rewriting changelog user %s endrec = %llu\n",
+	       mdd2obd_dev(mdd)->obd_name,
+	       mdd_chlg_username(rec, user_name, sizeof(user_name)),
+	       rec->cur_endrec);
 
 	/* Update the endrec */
 	rc = llog_write(env, llh, hdr, hdr->lrh_index);
@@ -1535,6 +2105,7 @@ static int mdd_changelog_clear(const struct lu_env *env,
 		.mcuc_id = id,
 		.mcuc_minrec = endrec,
 		.mcuc_flush = false,
+		.mcuc_mdd = mdd,
 	};
 	struct llog_ctxt *ctxt;
 	__u64 start_rec;
@@ -1572,7 +2143,11 @@ static int mdd_changelog_clear(const struct lu_env *env,
 			      mdd_changelog_clear_cb, (void *)&mcuc,
 			      0, 0);
 
-	if (rc < 0) {
+	if (rc == -EINVAL) {
+		CDEBUG(D_IOCTL, "%s: No changelog recnum <= %llu to clear\n",
+		       mdd2obd_dev(mdd)->obd_name, (unsigned long long) endrec);
+		RETURN(-EINVAL);
+	} else 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) {
@@ -1587,7 +2162,7 @@ static int mdd_changelog_clear(const struct lu_env *env,
 						      mcuc.mcuc_minrec);
 		}
 	} else {
-		CWARN("%s: No entry for user %d\n",
+		CDEBUG(D_IOCTL, "%s: No entry for user %d\n",
 		      mdd2obd_dev(mdd)->obd_name, id);
 		rc = -ENOENT;
 	}
@@ -1600,6 +2175,51 @@ out:
 	return rc;
 }
 
+static int mdd_changelog_user_deregister(const struct lu_env *env,
+				       struct mdd_device *mdd, int *id,
+				       const char *name)
+{
+	struct llog_ctxt *ctxt;
+	struct mdd_changelog_name_check_data mcnc = {
+		.mcnc_name = name,
+		.mcnc_id = 0,
+	};
+	int rc;
+
+	ENTRY;
+
+	if (name) {
+		ctxt = llog_get_context(mdd2obd_dev(mdd),
+					LLOG_CHANGELOG_USER_ORIG_CTXT);
+		if (!ctxt)
+			RETURN(-ENXIO);
+
+		rc = llog_cat_process(env, ctxt->loc_handle,
+			      mdd_changelog_name_check_cb, &mcnc, 0, 0);
+		llog_ctxt_put(ctxt);
+
+		if (rc != -EEXIST) {
+			CDEBUG(D_IOCTL, "%s: no entry for username %s\n",
+			       mdd2obd_dev(mdd)->obd_name, name);
+			RETURN(-ENOENT);
+		}
+		*id = mcnc.mcnc_id;
+	}
+
+	/* 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, *id, 0);
+	if (!rc)
+		rc = mdd_changelog_user_purge(env, mdd, *id);
+
+	/* recalc changelog current mask */
+	mdd_changelog_recalc_mask(env, mdd);
+
+	RETURN(rc);
+}
+
 /** mdd_iocontrol
  * May be called remotely from mdt_iocontrol_handle or locally from
  * mdt_iocontrol. Data may be freeform - remote handling doesn't enforce
@@ -1629,11 +2249,6 @@ static int mdd_iocontrol(const struct lu_env *env, struct md_device *m,
 		barrier_exit(mdd->mdd_bottom);
 		RETURN(rc);
 	}
-	case OBD_IOC_GET_MNTOPT: {
-		mntopt_t *mntopts = (mntopt_t *)karg;
-		*mntopts = mdd->mdd_dt_conf.ddp_mntopts;
-		RETURN(0);
-	}
 	case OBD_IOC_START_LFSCK: {
 		rc = lfsck_start(env, mdd->mdd_bottom,
 				 (struct lfsck_start_param *)karg);
@@ -1652,10 +2267,6 @@ static int mdd_iocontrol(const struct lu_env *env, struct md_device *m,
 	}
 
 	/* 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);
@@ -1667,14 +2278,18 @@ static int mdd_iocontrol(const struct lu_env *env, struct md_device *m,
 		if (unlikely(!barrier_entry(mdd->mdd_bottom)))
 			RETURN(-EINPROGRESS);
 
-		rc = mdd_changelog_user_register(env, mdd, &data->ioc_u32_1);
+		rc = mdd_changelog_user_register(env, mdd, &data->ioc_u32_1,
+						 data->ioc_inlbuf1,
+						 data->ioc_inlbuf2);
 		barrier_exit(mdd->mdd_bottom);
 		break;
 	case OBD_IOC_CHANGELOG_DEREG:
 		if (unlikely(!barrier_entry(mdd->mdd_bottom)))
 			RETURN(-EINPROGRESS);
 
-		rc = mdd_changelog_user_purge(env, mdd, data->ioc_u32_1);
+		rc = mdd_changelog_user_deregister(env, mdd, &data->ioc_u32_1,
+						   data->ioc_inlbuf1);
+
 		barrier_exit(mdd->mdd_bottom);
 		break;
 	default:
@@ -1692,20 +2307,20 @@ static const struct md_device_operations mdd_ops = {
 	.mdo_root_get	    = mdd_root_get,
 	.mdo_llog_ctxt_get  = mdd_llog_ctxt_get,
 	.mdo_iocontrol      = mdd_iocontrol,
-	.mdo_maxeasize_get  = mdd_maxeasize_get,
+	.mdo_dtconf_get     = mdd_dtconf_get,
 };
 
-static struct lu_device_type_operations mdd_device_type_ops = {
-        .ldto_init = mdd_type_init,
-        .ldto_fini = mdd_type_fini,
+static const struct lu_device_type_operations mdd_device_type_ops = {
+	.ldto_init		= mdd_type_init,
+	.ldto_fini		= mdd_type_fini,
 
-        .ldto_start = mdd_type_start,
-        .ldto_stop  = mdd_type_stop,
+	.ldto_start		= mdd_type_start,
+	.ldto_stop		= mdd_type_stop,
 
-        .ldto_device_alloc = mdd_device_alloc,
-        .ldto_device_free  = mdd_device_free,
+	.ldto_device_alloc	= mdd_device_alloc,
+	.ldto_device_free	= mdd_device_free,
 
-        .ldto_device_fini    = mdd_device_fini
+	.ldto_device_fini	= mdd_device_fini
 };
 
 static struct lu_device_type mdd_device_type = {
@@ -1723,9 +2338,10 @@ static void mdd_key_fini(const struct lu_context *ctx,
 {
 	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);
+	lu_buf_free(&info->mdi_big_buf);
+	lu_buf_free(&info->mdi_link_buf);
+	lu_buf_free(&info->mdi_xattr_buf);
+	lu_buf_free(&info->mdi_chlg_buf);
 
 	OBD_FREE_PTR(info);
 }
@@ -1770,16 +2386,10 @@ static int __init mdd_init(void)
 	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;
-
-	hsm_actions_logops = llog_osd_ops;
-	hsm_actions_logops.lop_add = llog_cat_add_rec;
-	hsm_actions_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;
 
-	rc = class_register_type(&mdd_obd_device_ops, NULL, true, NULL,
+	rc = class_register_type(&mdd_obd_device_ops, NULL, false,
 				 LUSTRE_MDD_NAME, &mdd_device_type);
 	if (rc)
 		lu_kmem_fini(mdd_caches);