scoutfs: use dlm for locking

To actually use it, we first have to copy symbols over from the dlm build
into the scoutfs source directory. Make that happen automatically for us in
the Makefile.

The only users of locking at the moment are mount, unmount and xattr
read/write. Adding more locking calls should be a straight-forward endeavor.

The LVB based server ip communication didn't work out, and LVBS as they are
written don't make sense in a range locking world. So instead, we record the
server ip address in the superblock. This is protected by the listen lock,
which also arbitrates which node will be the manifest server.

We take and drop the dlm lock on each lock/unlock call. Lock caching will
come in a future patch.

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

kmod/Makefile

@@ -12,14 +12,15 @@ else
 SP = @:
 endif
 
-SCOUTFS_ARGS := CONFIG_SCOUTFS_FS=m -C $(SK_KSRC) M=$(CURDIR)/src
+SCOUTFS_ARGS := CONFIG_SCOUTFS_FS=m -C $(SK_KSRC) -I $(CURDIR)/dlm/include M=$(CURDIR)/src
 DLM_ARGS := CONFIG_DLM=m CONFIG_DLM_DEBUG=y -C $(SK_KSRC) M=$(CURDIR)/dlm
 
 all: module
 
 module:
-	make $(SCOUTFS_ARGS)
 	make $(DLM_ARGS)
+	cp $(CURDIR)/dlm/Module.symvers $(CURDIR)/src/
+	make $(SCOUTFS_ARGS)
 	$(SP) make C=2 CF="-D__CHECK_ENDIAN__" $(SCOUTFS_ARGS)
 # Do not enable until we can clean up some warnings
 #	$(SP) make C=2 CF="-D__CHECK_ENDIAN__" $(DLM_ARGS)

kmod/dlm/lock.c

@@ -2552,6 +2552,11 @@ static int _can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
 	 * order.
 	 */
 
+	/*
+	 * XXX: Right now scoutfs uses NOORDER but if that changes
+	 * we'll have to replace the list_empty() checks below with
+	 * tree searches.
+	 */
 	if (lkb->lkb_exflags & DLM_LKF_NOORDER)
 		return 1;
 

kmod/src/Makefile

@@ -3,5 +3,5 @@ obj-$(CONFIG_SCOUTFS_FS) := scoutfs.o
 CFLAGS_scoutfs_trace.o = -I$(src) # define_trace.h double include
  
 scoutfs-y += alloc.o bio.o compact.o counters.o data.o dir.o kvec.o inode.o \
-	     ioctl.o item.o key.o lock.o manifest.o msg.o net.o ring.o seg.o \
-	     scoutfs_trace.o super.o trans.o xattr.o
+	     ioctl.o item.o key.o lock.o manifest.o msg.o net.o  options.o \
+	     ring.o seg.o scoutfs_trace.o super.o trans.o xattr.o

kmod/src/format.h

@@ -276,6 +276,13 @@ struct scoutfs_inode_index_key {
 
 #define SCOUTFS_UUID_BYTES 16
 
+/* XXX ipv6 */
+struct scoutfs_inet_addr {
+	__le32 addr;
+	__le16 port;
+} __packed;
+
+#define SCOUTFS_DEFAULT_PORT 12345
 
 /*
  * The ring fields describe the statically allocated ring log.  The
@@ -298,6 +305,7 @@ struct scoutfs_super_block {
 	__le64 next_seg_seq;
 	struct scoutfs_ring_descriptor alloc_ring;
 	struct scoutfs_manifest manifest;
+	struct scoutfs_inet_addr server_addr;
 } __packed;
 
 #define SCOUTFS_ROOT_INO 1
@@ -397,12 +405,6 @@ enum {
  * messages over the wire.
  */
 
-/* XXX ipv6 */
-struct scoutfs_inet_addr {
-	__le32 addr;
-	__le16 port;
-} __packed;
-
 /*
  * This header precedes and describes all network messages sent over
  * sockets.  The id is set by the request and sent in the reply.  The

kmod/src/lock.c

@@ -19,13 +19,9 @@
 #include "lock.h"
 #include "item.h"
 #include "scoutfs_trace.h"
+#include "msg.h"
 
-/*
- * This is meant to be simple and correct, not performant.
- */
-
-static DECLARE_RWSEM(global_rwsem);
-static LIST_HEAD(global_super_list);
+#include "linux/dlm.h"
 
 /*
  * Allocated once and pointed to by the lock info of all the supers with
@@ -34,13 +30,8 @@ static LIST_HEAD(global_super_list);
 struct held_locks {
 	spinlock_t lock;
 	struct list_head list;
+	unsigned int seq_cnt;
 	wait_queue_head_t waitq;
-
-	/* super hacky fake lvb that only allows one specific key */
-	char fake_lvb[sizeof(struct scoutfs_inet_addr)];
-	struct scoutfs_key_buf fake_lvb_key;
-	char fake_lvb_key_data[SCOUTFS_MAX_KEY_SIZE];
-
 };
 
 /*
@@ -50,62 +41,25 @@ struct held_locks {
  */
 struct lock_info {
 	struct super_block *sb;
+	dlm_lockspace_t *ls;
+	char ls_name[DLM_LOCKSPACE_LEN];
 	bool shutdown;
 	struct held_locks *held;
 	struct list_head id_head;
-	struct list_head global_head;
 };
 
+#define	RANGE_LOCK_RESOURCE	"fs_range"
+#define	RANGE_LOCK_RESOURCE_LEN	(strlen(RANGE_LOCK_RESOURCE))
+
+
 #define DECLARE_LOCK_INFO(sb, name) \
 	struct lock_info *name = SCOUTFS_SB(sb)->lock_info
 
 /*
- * locks are compatible if they're from the same super, or are both reads,
- * or don't overlap.
- */
-static bool compatible_locks(struct scoutfs_lock *a, struct scoutfs_lock *b)
-{
-	return a->sb == b->sb ||
-	       (a->mode == SCOUTFS_LOCK_MODE_READ &&
-	        b->mode == SCOUTFS_LOCK_MODE_READ) ||
-	       scoutfs_key_compare_ranges(a->start, a->end, b->start, b->end);
-}
-
-/* also returns true if we're shutting down, caller tests after waiting */
-static bool lock_added(struct lock_info *linf, struct scoutfs_lock *add)
-{
-	struct held_locks *held = linf->held;
-	struct scoutfs_lock *lck;
-	bool added = true;
-
-	spin_lock(&held->lock);
-
-	if (linf->shutdown) {
-		added = true;
-		goto out;
-	}
-
-	list_for_each_entry(lck, &held->list, head) {
-		if (!compatible_locks(lck, add)) {
-			added = false;
-			break;
-		}
-	}
-
-	if (added)
-		list_add(&add->head, &held->list);
-
-out:
-	spin_unlock(&held->lock);
-
-	return added;
-}
-
-/*
- * Invalidate caches on this super because another super has acquired
- * a lock with the given mode and range.  We always have to write out
- * dirty overlapping items.  If they're writing then we need to also
- * invalidate all cached overlapping structures.
+ * Invalidate caches on this because another node wants a lock
+ * with the a lock with the given mode and range. We always have to
+ * write out dirty overlapping items.  If they're writing then we need
+ * to also invalidate all cached overlapping structures.
  */
 static int invalidate_caches(struct super_block *sb, int mode,
 			     struct scoutfs_key_buf *start,
@@ -119,120 +73,81 @@ static int invalidate_caches(struct super_block *sb, int mode,
 	if (ret)
 		return ret;
 
-	if (mode == SCOUTFS_LOCK_MODE_WRITE) {
+	if (mode == SCOUTFS_LOCK_MODE_WRITE)
 		ret = scoutfs_item_invalidate(sb, start, end);
-#if 0
-		scoutfs_dir_invalidate(sb, start, end) ?:
-		scoutfs_inode_invalidate(sb, start, end) ?:
-		scoutfs_data_invalidate(sb, start, end);
-#endif
-	}
 
 	return ret;
 }
 
-#define for_each_other_linf(linf, from_linf)				  \
-	for (linf = list_entry(from_linf->id_head.next, struct lock_info, \
-			       id_head);				  \
-	     linf != from_linf;						  \
-	     linf = list_entry(linf->id_head.next, struct lock_info,	  \
-			       id_head))
-
-static int invalidate_others(struct super_block *from, int mode,
-			     struct scoutfs_key_buf *start,
-			     struct scoutfs_key_buf *end)
-{
-	DECLARE_LOCK_INFO(from, from_linf);
-	struct lock_info *linf;
-	int ret = 0;
-
-	down_read(&global_rwsem);
-
-	for_each_other_linf(linf, from_linf) {
-		ret = invalidate_caches(linf->sb, mode, start, end);
-		if (ret)
-			break;
-	}
-
-	up_read(&global_rwsem);
-
-	return ret;
-}
-
-static void unlock(struct held_locks *held, struct scoutfs_lock *lck)
+static void uninit_scoutfs_lock(struct held_locks *held,
+				struct scoutfs_lock *lck)
 {
 	spin_lock(&held->lock);
+	lck->rqmode = SCOUTFS_LOCK_MODE_IV;
 	list_del_init(&lck->head);
 	spin_unlock(&held->lock);
+	lck->sequence = 0;
+}
+
+static void init_scoutfs_lock(struct super_block *sb, struct scoutfs_lock *lck,
+			      struct scoutfs_key_buf *start,
+			      struct scoutfs_key_buf *end)
+{
+	DECLARE_LOCK_INFO(sb, linfo);
+	struct held_locks *held = linfo->held;
+
+	memset(lck, 0, sizeof(*lck));
+	INIT_LIST_HEAD(&lck->head);
+	lck->sb = sb;
+	lck->mode = SCOUTFS_LOCK_MODE_IV;
+
+	if (start) {
+		lck->start = start;
+		lck->dlm_start.val = start->data;
+		lck->dlm_start.len = start->key_len;
+	}
+	if (end) {
+		lck->end = end;
+		lck->dlm_end.val = end->data;
+		lck->dlm_end.len = end->key_len;
+	}
+
+	spin_lock(&held->lock);
+	lck->sequence = ++held->seq_cnt;
+	spin_unlock(&held->lock);
+}
+
+static void scoutfs_ast(void *astarg)
+{
+	struct scoutfs_lock *lck = astarg;
+	DECLARE_LOCK_INFO(lck->sb, linfo);
+	struct held_locks *held = linfo->held;
+
+	trace_scoutfs_ast(lck->sb, lck);
+
+	spin_lock(&held->lock);
+	lck->mode = lck->rqmode;
+	lck->rqmode = SCOUTFS_LOCK_MODE_IV;
+	spin_unlock(&held->lock);
 
 	wake_up(&held->waitq);
 }
 
-static void assert_fake_lvb(struct held_locks *held,
-			    struct scoutfs_key_buf *start,
-			    struct scoutfs_key_buf *end, unsigned lvb_len)
+static void scoutfs_rbast(void *astarg, int mode,
+			 struct dlm_key *start, struct dlm_key *end)
 {
 
-	BUG_ON(scoutfs_key_compare(start, end));
-	BUG_ON(lvb_len != sizeof(held->fake_lvb));
-	BUG_ON(held->fake_lvb_key.key_len &&
-	       scoutfs_key_compare(&held->fake_lvb_key, start));
 }
 
-/*
- * Acquire a coherent lock on the given range of keys.  While the lock
- * is held other lockers are serialized.  Cache coherency is maintained
- * by the locking infrastructure.  Lock acquisition causes writeout from
- * or invalidation of other caches.
- *
- * The caller provides the opaque lock structure used for storage and
- * their start and end pointers will be accessed while the lock is held.
- */
-int scoutfs_lock_range_lvb(struct super_block *sb, int mode,
-			   struct scoutfs_key_buf *start,
-			   struct scoutfs_key_buf *end,
-			   void *caller_lvb, unsigned lvb_len,
-			   struct scoutfs_lock *lck)
+static int lock_granted(struct held_locks *held, struct scoutfs_lock *lck,
+			int mode)
 {
-	DECLARE_LOCK_INFO(sb, linf);
-	struct held_locks *held = linf->held;
 	int ret;
 
-	INIT_LIST_HEAD(&lck->head);
-	lck->sb = sb;
-	lck->start = start;
-	lck->end = end;
-	lck->mode = mode;
+	spin_lock(&held->lock);
+	ret = !!(mode == lck->mode);
+	spin_unlock(&held->lock);
 
-	trace_scoutfs_lock_range(sb, lck);
-
-	ret = wait_event_interruptible(held->waitq, lock_added(linf, lck));
-	if (ret)
-		goto out;
-
-	if (linf->shutdown) {
-		ret = -ESHUTDOWN;
-		goto out;
-	}
-
-	ret = invalidate_others(sb, mode, start, end);
-	if (ret)
-		goto out;
-
-	if (caller_lvb) {
-		assert_fake_lvb(held, start, end, lvb_len);
-		if (mode == SCOUTFS_LOCK_MODE_WRITE) {
-			memcpy(held->fake_lvb, caller_lvb, lvb_len);
-			scoutfs_key_copy(&held->fake_lvb_key, start);
-		} else {
-			memcpy(caller_lvb, held->fake_lvb, lvb_len);
-		}
-	}
-	ret = 0;
-
-out:
-	if (ret < 0 && !list_empty(&lck->head))
-		unlock(held, lck);
 	return ret;
 }
 
@@ -250,81 +165,124 @@ int scoutfs_lock_range(struct super_block *sb, int mode,
 		       struct scoutfs_key_buf *end,
 		       struct scoutfs_lock *lck)
 {
-	return scoutfs_lock_range_lvb(sb, mode, start, end, NULL, 0, lck);
+	DECLARE_LOCK_INFO(sb, linfo);
+	struct held_locks *held = linfo->held;
+	int ret;
+
+	init_scoutfs_lock(sb, lck, start, end);
+
+	trace_scoutfs_lock_range(sb, lck);
+
+	spin_lock(&held->lock);
+	if (linfo->shutdown) {
+		spin_unlock(&held->lock);
+		return -ESHUTDOWN;
+	}
+
+	list_add(&lck->head, &held->list);
+	spin_unlock(&held->lock);
+
+	lck->rqmode = mode;
+	ret = dlm_lock_range(linfo->ls, mode, &lck->dlm_start, &lck->dlm_end,
+			     &lck->lksb, DLM_LKF_NOORDER, RANGE_LOCK_RESOURCE,
+			     RANGE_LOCK_RESOURCE_LEN, 0, scoutfs_ast, lck,
+			     scoutfs_rbast);
+	if (ret) {
+		scoutfs_err(sb, "Error %d locking %s\n", ret,
+			    RANGE_LOCK_RESOURCE);
+		uninit_scoutfs_lock(held, lck);
+		return ret;
+	}
+
+	wait_event(held->waitq, lock_granted(held, lck, mode));
+
+	return 0;
 }
 
 void scoutfs_unlock_range(struct super_block *sb, struct scoutfs_lock *lck)
 {
-	DECLARE_LOCK_INFO(sb, linf);
-	struct held_locks *held = linf->held;
+	DECLARE_LOCK_INFO(sb, linfo);
+	struct held_locks *held = linfo->held;
+	int ret;
 
 	trace_scoutfs_unlock_range(sb, lck);
 
-	unlock(held, lck);
+	BUG_ON(!lck->sequence);
+
+	/*
+	 * Use write mode to invalidate all since we are completely
+	 * dropping the lock. Once we keep the locks around then we
+	 * can invalidate based on what level we're downconverting to
+	 * (PR, NL).
+	 */
+	invalidate_caches(sb, SCOUTFS_LOCK_MODE_WRITE, lck->start, lck->end);
+
+	lck->rqmode = DLM_LOCK_IV;
+	ret = dlm_unlock(linfo->ls, lck->lksb.sb_lkid, 0, &lck->lksb, lck);
+	if (ret) {
+		scoutfs_err(sb, "Error %d unlocking %s\n", ret,
+			    RANGE_LOCK_RESOURCE);
+		goto out;
+	}
+
+	wait_event(held->waitq, lock_granted(held, lck, DLM_LOCK_IV));
+out:
+	uninit_scoutfs_lock(held, lck);
+	/* lock was removed from held list, wake up umount process */
+	wake_up(&held->waitq);
 }
 
 /*
  * The moment this is done we can have other mounts start asking
  * us to write back and invalidate, so do this very very late.
  */
-int scoutfs_lock_setup(struct super_block *sb)
+static int init_lock_info(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
-	struct scoutfs_sb_info *other_sbi;
-	struct lock_info *other_linf;
 	struct held_locks *held;
-	struct lock_info *linf;
+	struct lock_info *linfo;
 
-	linf = kmalloc(sizeof(struct lock_info), GFP_KERNEL);
-	if (!linf)
+	linfo = kzalloc(sizeof(struct lock_info), GFP_KERNEL);
+	if (!linfo)
 		return -ENOMEM;
 
 	held = kzalloc(sizeof(struct held_locks), GFP_KERNEL);
 	if (!held) {
-		kfree(linf);
+		kfree(linfo);
 		return -ENOMEM;
 	}
 
 	spin_lock_init(&held->lock);
 	INIT_LIST_HEAD(&held->list);
 	init_waitqueue_head(&held->waitq);
-	scoutfs_key_init_buf_len(&held->fake_lvb_key, &held->fake_lvb_key_data,
-				 0, sizeof(held->fake_lvb_key_data));
 
-	linf->sb = sb;
-	linf->shutdown = false;
-	linf->held = held;
-	INIT_LIST_HEAD(&linf->id_head);
-	INIT_LIST_HEAD(&linf->global_head);
+	linfo->sb = sb;
+	linfo->shutdown = false;
+	linfo->held = held;
+	INIT_LIST_HEAD(&linfo->id_head);
+	linfo->ls = NULL;
 
-	sbi->lock_info = linf;
+	snprintf(linfo->ls_name, DLM_LOCKSPACE_LEN, "%llx",
+		 le64_to_cpu(sbi->super.hdr.fsid));
 
-	trace_printk("sb %p id %016llx allocated linf %p held %p\n",
-		     sb, le64_to_cpu(sbi->super.id), linf, held);
+	sbi->lock_info = linfo;
 
-	down_write(&global_rwsem);
-
-	list_for_each_entry(other_linf, &global_super_list, global_head) {
-		other_sbi = SCOUTFS_SB(other_linf->sb);
-		if (other_sbi->super.id == sbi->super.id) {
-			list_add(&linf->id_head, &other_linf->id_head);
-			linf->held = other_linf->held;
-			trace_printk("sharing held %p\n", linf->held);
-			break;
-		}
-	}
-
-	/* add to global list after walking so we don't see ourselves */
-	list_add(&linf->global_head, &global_super_list);
-
-	up_write(&global_rwsem);
-
-	if (linf->held != held)
-		kfree(held);
+	trace_printk("sb %p id %016llx allocated linfo %p held %p\n",
+		     sb, le64_to_cpu(sbi->super.id), linfo, held);
 
 	return 0;
 }
 
+static int can_complete_shutdown(struct held_locks *held)
+{
+	int ret;
+
+	spin_lock(&held->lock);
+	ret = !!list_empty(&held->list);
+	spin_unlock(&held->lock);
+	return ret;
+}
+
 /*
  * Cause all lock attempts from our super to fail, waking anyone who is
  * currently blocked attempting to lock.  Now that locks can't block we
@@ -333,13 +291,13 @@ int scoutfs_lock_setup(struct super_block *sb)
  */
 void scoutfs_lock_shutdown(struct super_block *sb)
 {
-	DECLARE_LOCK_INFO(sb, linf);
-	struct held_locks *held = linf->held;
+	DECLARE_LOCK_INFO(sb, linfo);
+	struct held_locks *held = linfo->held;
 
-	if (linf) {
-		held = linf->held;
+	if (linfo) {
+		held = linfo->held;
 		spin_lock(&held->lock);
-		linf->shutdown = true;
+		linfo->shutdown = true;
 		spin_unlock(&held->lock);
 
 		wake_up(&held->waitq);
@@ -349,27 +307,50 @@ void scoutfs_lock_shutdown(struct super_block *sb)
 void scoutfs_lock_destroy(struct super_block *sb)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
-	DECLARE_LOCK_INFO(sb, linf);
+	DECLARE_LOCK_INFO(sb, linfo);
 	struct held_locks *held;
+	int ret;
 
-	if (linf) {
-		down_write(&global_rwsem);
+	if (linfo) {
+		held = linfo->held;
+		wait_event(held->waitq, can_complete_shutdown(held));
 
-		list_del_init(&linf->global_head);
+		ret = dlm_release_lockspace(linfo->ls, 2);
+		if (ret)
+			scoutfs_info(sb, "Error %d releasing lockspace %s\n",
+				     ret, linfo->ls_name);
 
-		if (!list_empty(&linf->id_head)) {
-			list_del_init(&linf->id_head);
-			held = NULL;
-		} else {
-			held = linf->held;
-		}
+		sbi->lock_info = NULL;
 
-		up_write(&global_rwsem);
-
-		trace_printk("sb %p id %016llx freeing linf %p held %p\n",
-			     sb, le64_to_cpu(sbi->super.id), linf, held);
+		trace_printk("sb %p id %016llx freeing linfo %p held %p\n",
+			     sb, le64_to_cpu(sbi->super.id), linfo, held);
 
 		kfree(held);
-		kfree(linf);
+		kfree(linfo);
 	}
 }
+
+int scoutfs_lock_setup(struct super_block *sb)
+{
+	struct lock_info *linfo;
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	int ret;
+
+	ret = init_lock_info(sb);
+	if (ret)
+		return ret;
+
+	linfo = sbi->lock_info;
+	/*
+	 * Open coded '64' here is for lvb_len. We never use the LVB
+	 * flag so this doesn't matter, but the dlm needs a non-zero
+	 * multiple of 8
+	 */
+	ret = dlm_new_lockspace(linfo->ls_name, sbi->opts.cluster_name,
+				DLM_LSFL_FS|DLM_LSFL_NEWEXCL, 64, NULL,
+				NULL, NULL, &linfo->ls);
+	if (ret)
+		scoutfs_lock_destroy(sb);
+
+	return ret;
+}

kmod/src/lock.h

@@ -1,30 +1,38 @@
 #ifndef _SCOUTFS_LOCK_H_
 #define _SCOUTFS_LOCK_H_
 
+#include "../dlm/include/linux/dlm.h"
+
 struct scoutfs_lock {
 	struct list_head head;
 	struct super_block *sb;
 	struct scoutfs_key_buf *start;
 	struct scoutfs_key_buf *end;
 	int mode;
+	int rqmode;
+	struct dlm_lksb lksb;
+	struct dlm_key dlm_start;
+	struct dlm_key dlm_end;
+	unsigned int sequence; /* for debugging and sanity checks */
 };
 
 enum {
-	SCOUTFS_LOCK_MODE_READ,
-	SCOUTFS_LOCK_MODE_WRITE,
+	SCOUTFS_LOCK_MODE_IV = DLM_LOCK_IV,
+	SCOUTFS_LOCK_MODE_READ = DLM_LOCK_PR,
+	SCOUTFS_LOCK_MODE_WRITE = DLM_LOCK_EX,
 };
 
 int scoutfs_lock_range(struct super_block *sb, int mode,
 		       struct scoutfs_key_buf *start,
 		       struct scoutfs_key_buf *end,
 		       struct scoutfs_lock *lck);
-int scoutfs_lock_range_lvb(struct super_block *sb, int mode,
-			   struct scoutfs_key_buf *start,
-			   struct scoutfs_key_buf *end,
-			   void *caller_lvb, unsigned lvb_len,
-			   struct scoutfs_lock *lck);
 void scoutfs_unlock_range(struct super_block *sb, struct scoutfs_lock *lck);
 
+int scoutfs_lock_addr(struct super_block *sb, int wanted_mode,
+		      void *caller_lvb, unsigned lvb_len);
+void scoutfs_unlock_addr(struct super_block *sb, void *caller_lvb,
+			 unsigned lvb_len);
+
 int scoutfs_lock_setup(struct super_block *sb);
 void scoutfs_lock_shutdown(struct super_block *sb);
 void scoutfs_lock_destroy(struct super_block *sb);

kmod/src/net.c

@@ -15,6 +15,7 @@
 #include <linux/slab.h>
 #include <asm/ioctls.h>
 #include <linux/net.h>
+#include <linux/inet.h>
 #include <linux/in.h>
 #include <net/sock.h>
 #include <net/tcp.h>
@@ -30,6 +31,7 @@
 #include "seg.h"
 #include "compact.h"
 #include "scoutfs_trace.h"
+#include "msg.h"
 
 /*
  * scoutfs mounts use a simple client-server model to send and process
@@ -889,7 +891,7 @@ static void scoutfs_net_proc_func(struct work_struct *work)
 	while (!nti->server_loaded) {
 		mutex_lock(&nti->mutex);
 		if (!nti->server_loaded) {
-			ret = scoutfs_read_supers(sb) ?:
+			ret = scoutfs_read_supers(sb, &SCOUTFS_SB(sb)->super) ?:
 			      scoutfs_manifest_setup(sb) ?:
 			      scoutfs_alloc_setup(sb) ?:
 			      scoutfs_compact_setup(sb);
@@ -1108,18 +1110,29 @@ static void set_sock_callbacks(struct sock_info *sinf)
 
 }
 
-/* get or set the address of the listening server depending on mode */
-static int lock_addr_lvb(struct super_block *sb, int mode,
-			   struct scoutfs_inet_addr *addr)
+static int write_server_addr(struct super_block *sb,
+			     struct scoutfs_inet_addr *addr)
+{
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
+	struct scoutfs_super_block *super = &sbi->super;
+
+	super->server_addr.addr = addr->addr;
+	super->server_addr.port = addr->port;
+
+	return scoutfs_write_dirty_super(sb);
+}
+
+static int read_server_addr(struct super_block *sb,
+			    struct scoutfs_inet_addr *addr)
 {
-	struct scoutfs_lock lck;
 	int ret;
+	struct scoutfs_super_block stack;
 
-	ret = scoutfs_lock_range_lvb(sb, mode, &addr_key, &addr_key,
-				     addr, sizeof(*addr), &lck);
-	if (ret == 0)
-		scoutfs_unlock_range(sb, &lck);
-
+	ret = scoutfs_read_supers(sb, &stack);
+	if (ret == 0) {
+		addr->addr = stack.server_addr.addr;
+		addr->port = stack.server_addr.port;
+	}
 	return ret;
 }
 
@@ -1177,6 +1190,7 @@ static void scoutfs_net_shutdown_func(struct work_struct *work)
 	struct super_block *sb = sinf->sb;
 	DECLARE_NET_INFO(sb, nti);
 	struct socket *sock = sinf->sock;
+	int ret;
 
 	trace_printk("sinf %p sock %p shutting_down %d\n",
 		     sinf, sock, sinf->shutting_down);
@@ -1198,9 +1212,13 @@ static void scoutfs_net_shutdown_func(struct work_struct *work)
 		destroy_server_state(sb);
 		nti->server_loaded = false;
 
-		/* clear addr lvb and try to reacquire lock and listen */
+		/* clear addr, try to reacquire lock and listen */
 		memset(&sinf->addr, 0, sizeof(sinf->addr));
-		lock_addr_lvb(sb, SCOUTFS_LOCK_MODE_WRITE, &sinf->addr);
+		ret = write_server_addr(sb, &sinf->addr);
+		if (ret)
+			scoutfs_err(sb,
+				    "Non-fatal error %d while writing server "
+				    "address\n", ret);
 		scoutfs_unlock_range(sb, &sinf->listen_lck);
 		queue_delayed_work(nti->proc_wq, &nti->server_work, 0);
 
@@ -1883,6 +1901,7 @@ static void scoutfs_net_listen_func(struct work_struct *work)
 	struct sock_info *sinf = container_of(work, struct sock_info,
 					      listen_work);
 	struct super_block *sb = sinf->sb;
+	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_inet_addr addr;
 	struct sockaddr_in sin;
 	struct socket *sock;
@@ -1890,10 +1909,9 @@ static void scoutfs_net_listen_func(struct work_struct *work)
 	int optval;
 	int ret;
 
-	/* XXX option to set listening address */
 	sin.sin_family = AF_INET;
-	sin.sin_addr.s_addr = cpu_to_be32(INADDR_LOOPBACK);
-	sin.sin_port = 0;
+	sin.sin_addr.s_addr = cpu_to_be32(le32_to_cpu(sbi->opts.listen_addr.addr));
+	sin.sin_port = cpu_to_be16(le16_to_cpu(sbi->opts.listen_addr.port));
 
 	trace_printk("binding to %pIS:%u\n",
 		     &sin, be16_to_cpu(sin.sin_port));
@@ -1931,10 +1949,17 @@ static void scoutfs_net_listen_func(struct work_struct *work)
 
 	set_sock_callbacks(sinf);
 
-	ret = kernel_listen(sock, 255) ?:
-	      lock_addr_lvb(sb, SCOUTFS_LOCK_MODE_WRITE, &addr);
-	if (ret == 0)
-		queue_sock_work(sinf, &sinf->accept_work);
+	ret = kernel_listen(sock, 255);
+	if (ret)
+		goto out;
+
+	scoutfs_advance_dirty_super(sb);
+	ret = write_server_addr(sb, &addr);
+	if (ret)
+		goto out;
+	scoutfs_advance_dirty_super(sb);
+
+	queue_sock_work(sinf, &sinf->accept_work);
 
 out:
 	if (ret) {
@@ -2012,7 +2037,7 @@ static void scoutfs_net_client_func(struct work_struct *work)
 	INIT_WORK(&sinf->send_work, scoutfs_net_send_func);
 	INIT_WORK(&sinf->recv_work, scoutfs_net_recv_func);
 
-	ret = lock_addr_lvb(sb, SCOUTFS_LOCK_MODE_READ, &sinf->addr);
+	ret = read_server_addr(sb, &sinf->addr);
 	if (ret == 0 && sinf->addr.addr == cpu_to_le32(INADDR_ANY))
 		ret = -ENOENT;
 	if (ret < 0) {

kmod/src/options.c

@@ -0,0 +1,81 @@
+/*
+ * Copyright (C) 2017 Versity Software, Inc.  All rights reserved.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU General Public
+ * License v2 as published by the Free Software Foundation.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License for more details.
+ */
+#include <linux/kernel.h>
+#include <linux/module.h>
+#include <linux/fs.h>
+#include <linux/types.h>
+
+#include <linux/parser.h>
+#include <linux/inet.h>
+#include <linux/string.h>
+#include <linux/in.h>
+
+#include "msg.h"
+#include "options.h"
+
+enum {
+	Opt_listen = 0,
+	Opt_cluster,
+	Opt_err,
+};
+
+static const match_table_t tokens = {
+	{Opt_listen, "listen=%s"},
+	{Opt_cluster, "cluster=%s"},
+	{Opt_err, NULL}
+};
+
+int scoutfs_parse_options(struct super_block *sb, char *options,
+			  struct mount_options *parsed)
+{
+	char ipstr[INET_ADDRSTRLEN + 1];
+	substring_t args[MAX_OPT_ARGS];
+	int token, len;
+	__be32 addr;
+	char *p;
+
+	/* Set defaults */
+	memset(parsed, 0, sizeof(*parsed));
+	strcpy(parsed->cluster_name, "scoutfs");
+
+	while ((p = strsep(&options, ",")) != NULL) {
+		if (!*p)
+			continue;
+
+		token = match_token(p, tokens, args);
+		switch (token) {
+		case Opt_listen:
+			match_strlcpy(ipstr, args, ARRAY_SIZE(ipstr));
+			addr = in_aton(ipstr);
+			if (ipv4_is_multicast(addr) || ipv4_is_lbcast(addr) ||
+			    ipv4_is_zeronet(addr) || ipv4_is_local_multicast(addr))
+				return -EINVAL;
+			parsed->listen_addr.addr =
+				cpu_to_le32(be32_to_cpu(addr));
+			break;
+		case Opt_cluster:
+			len = args[0].to - args[0].from;
+			if (len == 0 || len > (MAX_CLUSTER_NAME_LEN - 1))
+				return -EINVAL;
+			match_strlcpy(parsed->cluster_name, args,
+				      MAX_CLUSTER_NAME_LEN);
+			break;
+		default:
+			scoutfs_err(sb, "Unknown or malformed option, \"%s\"\n",
+				    p);
+			break;
+		}
+	}
+
+	return 0;
+}

kmod/src/options.h

@@ -0,0 +1,17 @@
+#ifndef	_SCOUTFS_OPTIONS_H_
+#define	_SCOUTFS_OPTIONS_H_
+
+#include <linux/fs.h>
+#include "format.h"
+
+#define MAX_CLUSTER_NAME_LEN 17
+struct mount_options
+{
+	struct scoutfs_inet_addr	listen_addr;
+	char				cluster_name[MAX_CLUSTER_NAME_LEN];
+};
+
+int scoutfs_parse_options(struct super_block *sb, char *options,
+			  struct mount_options *parsed);
+
+#endif	/* _SCOUTFS_OPTIONS_H_ */

kmod/src/scoutfs_trace.h

@@ -267,16 +267,21 @@ DECLARE_EVENT_CLASS(scoutfs_lock_class,
         TP_ARGS(sb, lck),
         TP_STRUCT__entry(
 		__field(int, mode)
+		__field(int, rqmode)
+		__field(unsigned int, seq)
                 __dynamic_array(char, start, scoutfs_key_str(NULL, lck->start))
                 __dynamic_array(char, end, scoutfs_key_str(NULL, lck->end))
         ),
         TP_fast_assign(
 		__entry->mode = lck->mode;
+		__entry->rqmode = lck->rqmode;
+		__entry->seq = lck->sequence;
 		scoutfs_key_str(__get_dynamic_array(start), lck->start);
 		scoutfs_key_str(__get_dynamic_array(end), lck->end);
         ),
-        TP_printk("mode %s start %s end %s",
-		  lock_mode(__entry->mode), __get_str(start), __get_str(end))
+        TP_printk("seq %u mode %s rqmode %s start %s end %s",
+		  __entry->seq, lock_mode(__entry->mode),
+		  lock_mode(__entry->rqmode), __get_str(start), __get_str(end))
 );
 
 DEFINE_EVENT(scoutfs_lock_class, scoutfs_lock_range,
@@ -289,6 +294,11 @@ DEFINE_EVENT(scoutfs_lock_class, scoutfs_unlock_range,
        TP_ARGS(sb, lck)
 );
 
+DEFINE_EVENT(scoutfs_lock_class, scoutfs_ast,
+       TP_PROTO(struct super_block *sb, struct scoutfs_lock *lck),
+       TP_ARGS(sb, lck)
+);
+
 TRACE_EVENT(scoutfs_lock_invalidate_sb,
         TP_PROTO(struct super_block *sb, int mode,
 		 struct scoutfs_key_buf *start, struct scoutfs_key_buf *end),

kmod/src/super.c

@@ -18,6 +18,7 @@
 #include <linux/magic.h>
 #include <linux/random.h>
 #include <linux/statfs.h>
+#include <linux/sched.h>
 
 #include "super.h"
 #include "format.h"
@@ -36,6 +37,7 @@
 #include "data.h"
 #include "lock.h"
 #include "net.h"
+#include "options.h"
 #include "scoutfs_trace.h"
 
 static struct kset *scoutfs_kset;
@@ -139,7 +141,8 @@ int scoutfs_write_dirty_super(struct super_block *sb)
  * to re-read the super every time it comes up so that it can work from
  * the most recent persistent state.
  */
-int scoutfs_read_supers(struct super_block *sb)
+int scoutfs_read_supers(struct super_block *sb,
+			struct scoutfs_super_block *local)
 {
 	struct scoutfs_sb_info *sbi = SCOUTFS_SB(sb);
 	struct scoutfs_super_block *super;
@@ -147,6 +150,7 @@ int scoutfs_read_supers(struct super_block *sb)
 	int found = -1;
 	int ret;
 	int i;
+	u64 seq = 0;
 
 	page = alloc_page(GFP_KERNEL);
 	if (!page)
@@ -168,9 +172,9 @@ int scoutfs_read_supers(struct super_block *sb)
 			continue;
 		}
 
-		if (found < 0 || (le64_to_cpu(super->hdr.seq) >
-				le64_to_cpu(sbi->super.hdr.seq))) {
-			sbi->super = *super;
+		if (found < 0 || (le64_to_cpu(super->hdr.seq) > seq)) {
+			*local = *super;
+			seq = le64_to_cpu((*local).hdr.seq);
 			found = i;
 		}
 	}
@@ -191,6 +195,7 @@ int scoutfs_read_supers(struct super_block *sb)
 static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 {
 	struct scoutfs_sb_info *sbi;
+	struct mount_options opts;
 	struct inode *inode;
 	int ret;
 
@@ -221,8 +226,14 @@ static int scoutfs_fill_super(struct super_block *sb, void *data, int silent)
 	if (!sbi->kset)
 		return -ENOMEM;
 
+	ret = scoutfs_parse_options(sb, data, &opts);
+	if (ret)
+		return ret;
+
+	sbi->opts = opts;
+
 	ret = scoutfs_setup_counters(sb) ?:
-	      scoutfs_read_supers(sb) ?:
+	      scoutfs_read_supers(sb, &SCOUTFS_SB(sb)->super) ?:
 	      scoutfs_seg_setup(sb) ?:
 	      scoutfs_item_setup(sb) ?:
 	      scoutfs_inode_setup(sb) ?:

kmod/src/super.h

@@ -5,6 +5,7 @@
 #include <linux/rbtree.h>
 
 #include "format.h"
+#include "options.h"
 
 struct scoutfs_counters;
 struct item_cache;
@@ -54,6 +55,8 @@ struct scoutfs_sb_info {
 	struct kset *kset;
 
 	struct scoutfs_counters *counters;
+
+	struct mount_options opts;
 };
 
 static inline struct scoutfs_sb_info *SCOUTFS_SB(struct super_block *sb)
@@ -61,7 +64,8 @@ static inline struct scoutfs_sb_info *SCOUTFS_SB(struct super_block *sb)
 	return sb->s_fs_info;
 }
 
-int scoutfs_read_supers(struct super_block *sb);
+int scoutfs_read_supers(struct super_block *sb,
+			struct scoutfs_super_block *local);
 void scoutfs_advance_dirty_super(struct super_block *sb);
 int scoutfs_write_dirty_super(struct super_block *sb);