scoutfs: invalidate cache when we free locks

We weren't invalidating our cache before freeing locks due to memory
pressure. This would cause stale data on the node which originally held
the lock. Fix this by firing a callback from dlmglue before we free a
lock from the system. On the scoutfs side, the callback is wired to
call our invalidate function. This will ensure that the right data and
metadata hit disk before another node is allowed to acquire that lock.

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

kmod/src/dlmglue.c

@@ -2255,6 +2255,10 @@ void ocfs2_simple_drop_lockres(struct ocfs2_super *osb,
 	trace_ocfs2_simple_drop_lockres(osb, lockres);
 
 	ocfs2_mark_lockres_freeing(osb, lockres);
+
+	if (lockres->l_ops->drop_worker)
+		lockres->l_ops->drop_worker(lockres);
+
 	ret = ocfs2_drop_lock(osb, lockres);
 	if (ret)
 		mlog_errno(ret);

kmod/src/dlmglue.h

@@ -272,6 +272,18 @@ struct ocfs2_lock_res_ops {
 	 */
 	int (*downconvert_worker)(struct ocfs2_lock_res *, int);
 
+	/*
+	 * Called before we free a lock from the system. This allows
+	 * the filesystem to sync and invalidate caches before that
+	 * happens. The concept is identical to ->downconvert_worker
+	 * except for two exceptions:
+	 *  - The FS must do the full downconvert work - as if it were
+	 *    blocking an EX.
+	 * - We do not return an ocfs2_unblock_action - this worker is not
+	 *    allowed to delay dropping of the lock.
+	 */
+	void (*drop_worker)(struct ocfs2_lock_res *);
+
 	/*
 	 * Optional: pretty print the lockname into a buffer
 	 */

kmod/src/lock.c

@@ -325,6 +325,21 @@ static int ino_lock_downconvert(struct ocfs2_lock_res *lockres, int blocking)
 	return UNBLOCK_CONTINUE;
 }
 
+static void ino_lock_drop(struct ocfs2_lock_res *lockres)
+{
+	struct scoutfs_lock *lock = lockres->l_priv;
+	struct super_block *sb = lock->sb;
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+
+	/*
+	 * Locks get shut down near the end of our unmount process. By
+	 * now everything that needs to be synced or invalidated, has
+	 * been.
+	 */
+	if (!sbi->shutdown)
+		invalidate_caches(sb, DLM_LOCK_EX, lock);
+}
+
 static void lock_name_string(struct ocfs2_lock_res *lockres, char *buf,
 			     unsigned int len)
 {
@@ -362,6 +377,7 @@ static void count_idx_lock_event(struct ocfs2_lock_res *lockres,
 static struct ocfs2_lock_res_ops scoufs_ino_lops = {
 	.get_osb 		= get_ino_lock_osb,
 	.downconvert_worker 	= ino_lock_downconvert,
+	.drop_worker		= ino_lock_drop,
 	/* XXX: .check_downconvert that queries the item cache for dirty items */
 	.print			= lock_name_string,
 	.notify_event		= count_ino_lock_event,
@@ -371,6 +387,7 @@ static struct ocfs2_lock_res_ops scoufs_ino_lops = {
 static struct ocfs2_lock_res_ops scoufs_ino_index_lops = {
 	.get_osb 		= get_ino_lock_osb,
 	.downconvert_worker 	= ino_lock_downconvert,
+	.drop_worker		= ino_lock_drop,
 	.notify_event		= count_idx_lock_event,
 	/* XXX: .check_downconvert that queries the item cache for dirty items */
 	.print			= lock_name_string,
@@ -387,6 +404,7 @@ static struct ocfs2_lock_res_ops scoutfs_node_id_lops = {
 	.get_osb		= get_ino_lock_osb,
 	/* XXX: .check_downconvert that queries the item cache for dirty items */
 	.downconvert_worker 	= ino_lock_downconvert,
+	.drop_worker		= ino_lock_drop,
 	.print			= lock_name_string,
 	.flags			= 0,
 };

kmod/src/super.c

@@ -119,6 +119,8 @@ static void scoutfs_put_super(struct super_block *sb)
 
 	trace_scoutfs_put_super(sb);
 
+	sbi->shutdown = true;
+
 	scoutfs_unlock_flags(sb, sbi->node_id_lock, DLM_LOCK_EX,
 			     SCOUTFS_LKF_NO_TASK_REF);
 	sbi->node_id_lock = NULL;

kmod/src/super.h

@@ -66,6 +66,8 @@ struct scoutfs_sb_info {
 	struct mount_options opts;
 
 	struct dentry *debug_root;
+
+	bool shutdown;
 };
 
 static inline struct scoutfs_sb_info *SCOUTFS_SB(struct super_block *sb)