scoutfs: dlmglue/lock counters and tracing

We can use the excellent code in counters.h to easily place a whole set
of useful counters in dlmglue:

- one for every kind of wait in cluster_lock (blocked, busy, etc)
- one for each type of dlm operation (lock/unlock requests,
  converts, etc)
- one for each type of downconvert (cw/pr/ex)

These will give us a decent idea of the amount and type of lock traffic a
given node is seeing.

In addition, we add a second trace at the bottom of invalidate_caches.
By turning both traces in invalidate_caches on, we can look at our
trace log to see how long a given locks downconvert took.

Signed-off-by: Mark Fasheh <mfasheh@versity.com>

kmod/src/counters.h

@@ -38,6 +38,13 @@
 	EXPAND_COUNTER(data_invalidatepage)	\
 	EXPAND_COUNTER(data_writepage)		\
 	EXPAND_COUNTER(data_end_writeback_page)	\
+	EXPAND_COUNTER(dlm_cancel_convert)	\
+	EXPAND_COUNTER(dlm_convert_request)	\
+	EXPAND_COUNTER(dlm_cw_downconvert)	\
+	EXPAND_COUNTER(dlm_ex_downconvert)	\
+	EXPAND_COUNTER(dlm_lock_request)	\
+	EXPAND_COUNTER(dlm_pr_downconvert)	\
+	EXPAND_COUNTER(dlm_unlock_request)	\
 	EXPAND_COUNTER(item_alloc)		\
 	EXPAND_COUNTER(item_free)		\
 	EXPAND_COUNTER(item_create)		\
@@ -58,10 +65,14 @@
 	EXPAND_COUNTER(item_shrink_small_split)	\
 	EXPAND_COUNTER(item_shrink)		\
 	EXPAND_COUNTER(lock_alloc)		\
-	EXPAND_COUNTER(lock_free)
+	EXPAND_COUNTER(lock_blocked_wait)	\
+	EXPAND_COUNTER(lock_busy_wait)		\
+	EXPAND_COUNTER(lock_free)		\
+	EXPAND_COUNTER(lock_incompat_wait)
+
 
 #define FIRST_COUNTER alloc_alloc
-#define LAST_COUNTER lock_free
+#define LAST_COUNTER lock_incompat_wait
 
 #undef EXPAND_COUNTER
 #define EXPAND_COUNTER(which) struct percpu_counter which;

kmod/src/dlmglue.c

@@ -36,6 +36,7 @@
 #include <linux/sched.h>
 #include <linux/dlm.h>
 
+#include "counters.h"
 #include "dlmglue.h"
 
 #include "scoutfs_trace.h"
@@ -1143,6 +1144,7 @@ again:
 		 * them. */
 		lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BUSY, 0);
 		wait = 1;
+		scoutfs_inc_counter(osb->sb, lock_busy_wait);
 		goto unlock;
 	}
 
@@ -1171,6 +1173,7 @@ again:
 		 * another node */
 		lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BLOCKED, 0);
 		wait = 1;
+		scoutfs_inc_counter(osb->sb, lock_blocked_wait);
 		goto unlock;
 	}
 
@@ -1187,6 +1190,7 @@ again:
 		set_lock_blocking(lockres, DLM_LOCK_EX);
 		lockres_add_mask_waiter(lockres, &mw, OCFS2_LOCK_BLOCKED, 0);
 		wait = 1;
+		scoutfs_inc_counter(osb->sb, lock_incompat_wait);
 		goto unlock;
 	}
 
@@ -1216,6 +1220,11 @@ again:
 		gen = lockres_set_pending(lockres);
 		spin_unlock_irqrestore(&lockres->l_lock, flags);
 
+		if (lkm_flags & DLM_LKF_CONVERT)
+			scoutfs_inc_counter(osb->sb, dlm_convert_request);
+		else
+			scoutfs_inc_counter(osb->sb, dlm_lock_request);
+
 		BUG_ON(level == DLM_LOCK_IV);
 		BUG_ON(level == DLM_LOCK_NL);
 
@@ -1974,8 +1983,9 @@ static void ocfs2_do_node_down(int node_num, void *data)
 {
 }
 
-int ocfs2_dlm_init(struct ocfs2_super *osb, char *cluster_stack,
-		   char *cluster_name, char *ls_name, struct dentry *debug_root)
+int ocfs2_dlm_init(struct ocfs2_super *osb, struct super_block *sb,
+		   char *cluster_stack, char *cluster_name, char *ls_name,
+		   struct dentry *debug_root)
 {
 	int status = 0;
 	struct ocfs2_cluster_connection *conn = NULL;
@@ -1986,6 +1996,7 @@ int ocfs2_dlm_init(struct ocfs2_super *osb, char *cluster_stack,
 		goto local;
 	}
 #endif
+	osb->sb = sb;
 
 	status = ocfs2_dlm_init_debug(osb, debug_root);
 	if (status < 0) {
@@ -2149,6 +2160,8 @@ static int ocfs2_drop_lock(struct ocfs2_super *osb,
 	mlog(0, "lock %s, successful return from ocfs2_dlm_unlock\n",
 	     lockres->l_name);
 
+	scoutfs_inc_counter(osb->sb, dlm_unlock_request);
+
 	ocfs2_wait_on_busy_lock(lockres);
 out:
 	return 0;
@@ -2352,6 +2365,8 @@ static int ocfs2_cancel_convert(struct ocfs2_super *osb,
 
 	mlog(ML_BASTS, "lockres %s\n", lockres->l_name);
 
+	scoutfs_inc_counter(osb->sb, dlm_cancel_convert);
+
 	return ret;
 }
 
@@ -2501,11 +2516,6 @@ recheck:
 		goto leave_requeue;
 	}
 
-	/* If we get here, then we know that there are no more
-	 * incompatible holders (and anyone asking for an incompatible
-	 * lock is blocked). We can now downconvert the lock */
-	if (!lockres->l_ops->downconvert_worker)
-		goto downconvert;
 
 	/* Some lockres types want to do a bit of work before
 	 * downconverting a lock. Allow that here. The worker function
@@ -2513,6 +2523,13 @@ recheck:
 	 * it may change while we're not holding the spin lock. */
 	blocking = lockres->l_blocking;
 	level = lockres->l_level;
+
+	/* If we get here, then we know that there are no more
+	 * incompatible holders (and anyone asking for an incompatible
+	 * lock is blocked). We can now downconvert the lock */
+	if (!lockres->l_ops->downconvert_worker)
+		goto downconvert;
+
 	spin_unlock_irqrestore(&lockres->l_lock, flags);
 
 	ctl->unblock_action = lockres->l_ops->downconvert_worker(lockres, blocking);
@@ -2552,6 +2569,19 @@ downconvert:
 
 	gen = ocfs2_prepare_downconvert(lockres, new_level);
 	spin_unlock_irqrestore(&lockres->l_lock, flags);
+
+	switch (level) {
+	case DLM_LOCK_EX:
+		scoutfs_inc_counter(osb->sb, dlm_ex_downconvert);
+		break;
+	case DLM_LOCK_PR:
+		scoutfs_inc_counter(osb->sb, dlm_pr_downconvert);
+		break;
+	case DLM_LOCK_CW:
+		scoutfs_inc_counter(osb->sb, dlm_cw_downconvert);
+		break;
+	}
+
 	ret = ocfs2_downconvert_lock(osb, lockres, new_level, set_lvb,
 				     gen);
 

kmod/src/dlmglue.h

@@ -176,6 +176,10 @@ struct ocfs2_super
 	atomic64_t refresh_gen;
 
 	unsigned long s_mount_opt;
+
+	/* sb is for use with scoutfs counter macros only. Eventually
+	 * we'll roll our own counters code in dlmglue. */
+	struct super_block *sb;
 };
 /* For s_mount_opt */
 #define OCFS2_MOUNT_NOINTR (1 << 2)
@@ -318,8 +322,9 @@ void ocfs2_cluster_unlock(struct ocfs2_super *osb,
 			  struct ocfs2_lock_res *lockres, int level);
 
 int ocfs2_init_super(struct ocfs2_super *osb, int flags);
-int ocfs2_dlm_init(struct ocfs2_super *osb, char *cluster_stack,
-		   char *cluster_name, char *ls_name, struct dentry *debug_root);
+int ocfs2_dlm_init(struct ocfs2_super *osb, struct super_block *sb,
+		   char *cluster_stack, char *cluster_name, char *ls_name,
+		   struct dentry *debug_root);
 void ocfs2_dlm_shutdown(struct ocfs2_super *osb, int hangup_pending);
 void ocfs2_lock_res_init_once(struct ocfs2_lock_res *res);
 void ocfs2_lock_res_init_common(struct ocfs2_super *osb,

kmod/src/lock.c

@@ -223,7 +223,6 @@ static int invalidate_caches(struct super_block *sb, int mode,
 	if (ret)
 		return ret;
 
-
 	if (mode == DLM_LOCK_EX ||
 	    (mode == DLM_LOCK_PR && lock->lockres.l_level == DLM_LOCK_CW)) {
 		if (lock->lock_name.zone == SCOUTFS_FS_ZONE) {
@@ -238,6 +237,12 @@ static int invalidate_caches(struct super_block *sb, int mode,
 		ret = scoutfs_item_invalidate(sb, start, end);
 	}
 
+	/*
+	 * Not really tracing the return value here, we're mostly
+	 * interested in elapsed time between the top trace and this one.
+	 */
+	trace_scoutfs_lock_invalidate_ret(sb, lock);
+
 	return ret;
 }
 
@@ -1123,8 +1128,9 @@ int scoutfs_lock_setup(struct super_block *sb)
 		goto out;
 	}
 
-	ret = ocfs2_dlm_init(&linfo->dlmglue, "null", sbi->opts.cluster_name,
-			     linfo->ls_name, sbi->debug_root);
+	ret = ocfs2_dlm_init(&linfo->dlmglue, sb, "null",
+			     sbi->opts.cluster_name, linfo->ls_name,
+			     sbi->debug_root);
 	if (ret)
 		goto out;
 	linfo->dlmglue_online = true;

kmod/src/scoutfs_trace.h

@@ -1654,6 +1654,11 @@ DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_invalidate,
        TP_ARGS(sb, lck)
 );
 
+DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_invalidate_ret,
+       TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
+       TP_ARGS(sb, lck)
+);
+
 DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_reclaim,
        TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
        TP_ARGS(sb, lck)