[Cluster-devel] [PATCH] dlm: fix basts for granted PR waiting CW

[Cluster-devel] [PATCH] dlm: fix basts for granted PR waiting CW

[David Teigland]

Fix a long standing bug where a blocking callback would be missed
when there's a granted lock in PR mode and waiting locks in both
PR and CW modes (and the PR lock was added to the waiting queue
before the CW lock).  The logic simply compared the numerical values
of the modes to determine if a blocking callback was required, but in
the one case of PR and CW, the lower valued CW mode blocks the higher
valued PR mode.  We just need to add a special check for this PR/CW
case in the tests that decide when a blocking callback is needed.

Signed-off-by: David Teigland <teigland@redhat.com>

Index: linux-quilt/fs/dlm/lock.c
===================================================================
--- linux-quilt.orig/fs/dlm/lock.c
+++ linux-quilt/fs/dlm/lock.c
@@ -1670,9 +1670,10 @@ static int can_be_granted(struct dlm_rsb
    with a deadlk here, we'd have to generate something like grant_lock with
    the deadlk error.) */
 
-/* returns the highest requested mode of all blocked conversions */
+/* Returns the highest requested mode of all blocked conversions; sets
+   cw if there's a blocked conversion to DLM_LOCK_CW. */
 
-static int grant_pending_convert(struct dlm_rsb *r, int high)
+static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw)
 {
 	struct dlm_lkb *lkb, *s;
 	int hi, demoted, quit, grant_restart, demote_restart;
@@ -1709,6 +1710,9 @@ static int grant_pending_convert(struct 
 		}
 
 		hi = max_t(int, lkb->lkb_rqmode, hi);
+
+		if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
+			*cw = 1;
 	}
 
 	if (grant_restart)
@@ -1721,29 +1725,52 @@ static int grant_pending_convert(struct 
 	return max_t(int, high, hi);
 }
 
-static int grant_pending_wait(struct dlm_rsb *r, int high)
+static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw)
 {
 	struct dlm_lkb *lkb, *s;
 
 	list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
 		if (can_be_granted(r, lkb, 0, NULL))
 			grant_lock_pending(r, lkb);
-                else
+                else {
 			high = max_t(int, lkb->lkb_rqmode, high);
+			if (lkb->lkb_rqmode == DLM_LOCK_CW)
+				*cw = 1;
+		}
 	}
 
 	return high;
 }
 
+/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
+   on either the convert or waiting queue.
+   high is the largest rqmode of all locks blocked on the convert or
+   waiting queue. */
+
+static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
+{
+	if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
+		if (gr->lkb_highbast < DLM_LOCK_EX)
+			return 1;
+		return 0;
+	}
+
+	if (gr->lkb_highbast < high &&
+	    !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
+		return 1;
+	return 0;
+}
+
 static void grant_pending_locks(struct dlm_rsb *r)
 {
 	struct dlm_lkb *lkb, *s;
 	int high = DLM_LOCK_IV;
+	int cw = 0;
 
 	DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
 
-	high = grant_pending_convert(r, high);
-	high = grant_pending_wait(r, high);
+	high = grant_pending_convert(r, high, &cw);
+	high = grant_pending_wait(r, high, &cw);
 
 	if (high == DLM_LOCK_IV)
 		return;
@@ -1751,27 +1778,38 @@ static void grant_pending_locks(struct d
 	/*
 	 * If there are locks left on the wait/convert queue then send blocking
 	 * ASTs to granted locks based on the largest requested mode (high)
-	 * found above. FIXME: highbast < high comparison not valid for PR/CW.
+	 * found above.
 	 */
 
 	list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
-		if (lkb->lkb_bastaddr && (lkb->lkb_highbast < high) &&
-		    !__dlm_compat_matrix[lkb->lkb_grmode+1][high+1]) {
+		if (lkb->lkb_bastaddr && lock_requires_bast(lkb, high, cw)) {
 			queue_bast(r, lkb, high);
 			lkb->lkb_highbast = high;
 		}
 	}
 }
 
+static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
+{
+	if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
+	    (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
+		if (gr->lkb_highbast < DLM_LOCK_EX)
+			return 1;
+		return 0;
+	}
+
+	if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
+		return 1;
+	return 0;
+}
+
 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
 			    struct dlm_lkb *lkb)
 {
 	struct dlm_lkb *gr;
 
 	list_for_each_entry(gr, head, lkb_statequeue) {
-		if (gr->lkb_bastaddr &&
-		    gr->lkb_highbast < lkb->lkb_rqmode &&
-		    !modes_compat(gr, lkb)) {
+		if (gr->lkb_bastaddr && modes_require_bast(gr, lkb)) {
 			queue_bast(r, gr, lkb->lkb_rqmode);
 			gr->lkb_highbast = lkb->lkb_rqmode;
 		}
@@ -2235,7 +2273,7 @@ static int do_convert(struct dlm_rsb *r,
 	   before we try again to grant this one. */
 
 	if (is_demoted(lkb)) {
-		grant_pending_convert(r, DLM_LOCK_IV);
+		grant_pending_convert(r, DLM_LOCK_IV, NULL);
 		if (_can_be_granted(r, lkb, 1)) {
 			grant_lock(r, lkb);
 			queue_cast(r, lkb, 0);

[Cluster-devel] [PATCH] dlm: fix basts for granted PR waiting CW

[David Teigland]

On Wed, Jul 25, 2007 at 11:01:05AM -0500, David Teigland wrote:
> Fix a long standing bug where a blocking callback would be missed
> when there's a granted lock in PR mode and waiting locks in both
> PR and CW modes (and the PR lock was added to the waiting queue
> before the CW lock).  The logic simply compared the numerical values
> of the modes to determine if a blocking callback was required, but in
> the one case of PR and CW, the lower valued CW mode blocks the higher
> valued PR mode.  We just need to add a special check for this PR/CW
> case in the tests that decide when a blocking callback is needed.

Attached is a test I used in the past to trigger this bug (and others as
well).  I'd run two instances of the test on each of four nodes, like
this:

node01: rand_direct -d4 -f10     rand_direct -d8 -f10
node02: rand_direct -d4 -f10 -u  rand_direct -d8 -f10
node03: rand_direct -d4 -f10     rand_direct -d8 -f10
node04: rand_direct -d4 -f10     rand_direct -d8 -f10

Dave

-------------- next part --------------
/* tests to run:

   on two nodes: prealloc file, rand_direct -d1 -f1

   on two nodes: prealloc file, rand_direct -d1 -f1 -b

*/

#define _GNU_SOURCE
#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <signal.h>
#include <string.h>
#include <errno.h>
#include <ctype.h>
#include <limits.h>
#include <time.h>
#include <sys/stat.h>
#include <sys/file.h>
#include <sys/types.h>
#include <sys/wait.h>
#include <sys/times.h>
#include <sys/uio.h>
#include <sys/time.h>
#include <fcntl.h>

#define ONEMB 1048576

char *prog_name;
int quiet = 0;
int verbose = 0;
int delay = 0;
int enable_seeks = 0;
int enable_unlinks = 0;
int enable_buffered = 0;
int iterations = 0; /* forever */
int num_dirs = 10;
int num_files = 100;
int iobuf_size = ONEMB;
char *iobuf, **p_iobuf;

#define die(fmt, args...) \
{ \
  fprintf(stderr, "%s: ", prog_name); \
  fprintf(stderr, fmt, ##args); \
  exit(EXIT_FAILURE); \
}

void usage(void)
{
	printf("Usage:\n");
	printf("\n");
	printf("%s [options]\n", prog_name);
	printf("\n");
	printf("Options:\n");
	printf("\n");
	printf("  -b		random mix of buffered and direct io\n");
	printf("  -s		enable seeks before reads and writes\n");
	printf("  -u		enable unlinks\n");
	printf("  -i <n>	iterations, default 0 (forever)\n");
	printf("  -d <n>	number of dirs, default 10\n");
	printf("  -f <n>	number of files per dir, default 100\n");
}

int rand_int(int a, int b)
{
	return a + (int) (((float)(b - a + 1)) * random() / (RAND_MAX+1.0)); 
}

int do_read(int fd)
{
	read(fd, iobuf, iobuf_size);
	return 0;
}

int do_write(int fd)
{
	int rv;

	memset(iobuf, 0x55, iobuf_size);

	rv = write(fd, iobuf, iobuf_size);
	if (rv != iobuf_size)
		die("write size %d vs %d\n", rv, iobuf_size);
	return 0;
}

int do_seek(int fd)
{
	int off, rv;

	off = rand_int(0, ONEMB*100);
	off -= (off % 4096);
	
	rv = lseek(fd, off, SEEK_SET);

	return 0;
}

int rand_dio(void)
{
	if (rand_int(0, 1))
		return O_DIRECT;
	return 0;
}

int main(int argc, char *argv[])
{
	int i, j, flags, rv, c, fd, op, dir, file, ops = 0;
	char path[64];

	srandom(time(NULL));
	prog_name = argv[0];

	while ((c = getopt(argc, argv, "bqvsui:d:f:y:")) != -1) {
                switch (c) {
			case 'q':
				quiet = 1;
				break;
			case 'v':
				verbose = 1;
				break;
			case 'b':
				enable_buffered = 1;
				break;
			case 's':
				enable_seeks = 1;
				break;
			case 'u':
				enable_unlinks = 1;
				break;
			case 'i':
				iterations = atoi(optarg);
				break;
			case 'd':
				num_dirs = atoi(optarg);
				break;
			case 'f':
				num_files = atoi(optarg);
				break;
			case 'y':
				delay = atoi(optarg);
				break;
			case 'h':
			default:
				usage();
				exit(2);
		}
	}

	for (i = 0; i < num_dirs; i++) {
		memset(path, 0, sizeof(path));
		sprintf(path, "dir%.10u", i);
		mkdir(path, 0755);
	}

	p_iobuf = &iobuf;

	rv = posix_memalign((void *)p_iobuf, ONEMB, iobuf_size);
	if (rv) {
		perror("memalign error");
		exit(-2);
	}

	while (1) {
		op = rand_int(0, 5);
		dir = rand_int(0, num_dirs-1);
		file = rand_int(0, num_files-1);

		memset(path, 0, sizeof(path));
		sprintf(path, "dir%.10u/file%.10u", dir, file);

		if (verbose)
			printf("%s %d\n", path, op);

		if (op == 0) {
			/* open (write, append, create)
			   close */

			fd = open(path, O_WRONLY|O_CREAT|O_APPEND|O_DIRECT, 0644);
			if (fd < 0)
				die("open %s error %d errno %d\n",
				    path, fd, errno);
			close(fd);
			ops++;

		} else if (op == 1 || op == 2) {
			/* open (read, write)
			   write
			   close */

			if (enable_buffered)
				flags = rand_dio();
			else
				flags = O_DIRECT;

			fd = open(path, O_RDWR | flags);
			if (fd > 0) {
				if (enable_seeks)
					do_seek(fd);
				do_write(fd);
				close(fd);
				ops++;
			}

		} else if (op == 3 || op == 4) {
			/* open (read)
			   read
			   close */

			if (enable_buffered)
				flags = rand_dio();
			else
				flags = O_DIRECT;

			fd = open(path, O_RDONLY | flags);
			if (fd > 0) {
				if (enable_seeks)
					do_seek(fd);
				do_read(fd);
				close(fd);
				ops++;
			}

		} else if (op == 5) {
			if (enable_unlinks) {
				unlink(path);
				ops++;
			}

		} else {
			exit(-1);
		}

		if (!quiet)
			printf("%u:%d\n", ops, op);

		if (delay)
			sleep(delay);

		if (iterations && ops == iterations)
			break;
	}

	exit(EXIT_SUCCESS);
}

[Cluster-devel] Re: [PATCH] dlm: fix basts for granted PR waiting CW

[Steven Whitehouse]

Hi,

Now in the GFS2 -nmw git tree. Thanks,

Steve.

On Wed, 2007-07-25 at 11:01 -0500, David Teigland wrote:
> Fix a long standing bug where a blocking callback would be missed
> when there's a granted lock in PR mode and waiting locks in both
> PR and CW modes (and the PR lock was added to the waiting queue
> before the CW lock).  The logic simply compared the numerical values
> of the modes to determine if a blocking callback was required, but in
> the one case of PR and CW, the lower valued CW mode blocks the higher
> valued PR mode.  We just need to add a special check for this PR/CW
> case in the tests that decide when a blocking callback is needed.
> 
> Signed-off-by: David Teigland <teigland@redhat.com>
> 
> Index: linux-quilt/fs/dlm/lock.c
> ===================================================================
> --- linux-quilt.orig/fs/dlm/lock.c
> +++ linux-quilt/fs/dlm/lock.c
> @@ -1670,9 +1670,10 @@ static int can_be_granted(struct dlm_rsb
>     with a deadlk here, we'd have to generate something like grant_lock with
>     the deadlk error.) */
>  
> -/* returns the highest requested mode of all blocked conversions */
> +/* Returns the highest requested mode of all blocked conversions; sets
> +   cw if there's a blocked conversion to DLM_LOCK_CW. */
>  
> -static int grant_pending_convert(struct dlm_rsb *r, int high)
> +static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw)
>  {
>  	struct dlm_lkb *lkb, *s;
>  	int hi, demoted, quit, grant_restart, demote_restart;
> @@ -1709,6 +1710,9 @@ static int grant_pending_convert(struct 
>  		}
>  
>  		hi = max_t(int, lkb->lkb_rqmode, hi);
> +
> +		if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
> +			*cw = 1;
>  	}
>  
>  	if (grant_restart)
> @@ -1721,29 +1725,52 @@ static int grant_pending_convert(struct 
>  	return max_t(int, high, hi);
>  }
>  
> -static int grant_pending_wait(struct dlm_rsb *r, int high)
> +static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw)
>  {
>  	struct dlm_lkb *lkb, *s;
>  
>  	list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
>  		if (can_be_granted(r, lkb, 0, NULL))
>  			grant_lock_pending(r, lkb);
> -                else
> +                else {
>  			high = max_t(int, lkb->lkb_rqmode, high);
> +			if (lkb->lkb_rqmode == DLM_LOCK_CW)
> +				*cw = 1;
> +		}
>  	}
>  
>  	return high;
>  }
>  
> +/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
> +   on either the convert or waiting queue.
> +   high is the largest rqmode of all locks blocked on the convert or
> +   waiting queue. */
> +
> +static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
> +{
> +	if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
> +		if (gr->lkb_highbast < DLM_LOCK_EX)
> +			return 1;
> +		return 0;
> +	}
> +
> +	if (gr->lkb_highbast < high &&
> +	    !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
> +		return 1;
> +	return 0;
> +}
> +
>  static void grant_pending_locks(struct dlm_rsb *r)
>  {
>  	struct dlm_lkb *lkb, *s;
>  	int high = DLM_LOCK_IV;
> +	int cw = 0;
>  
>  	DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
>  
> -	high = grant_pending_convert(r, high);
> -	high = grant_pending_wait(r, high);
> +	high = grant_pending_convert(r, high, &cw);
> +	high = grant_pending_wait(r, high, &cw);
>  
>  	if (high == DLM_LOCK_IV)
>  		return;
> @@ -1751,27 +1778,38 @@ static void grant_pending_locks(struct d
>  	/*
>  	 * If there are locks left on the wait/convert queue then send blocking
>  	 * ASTs to granted locks based on the largest requested mode (high)
> -	 * found above. FIXME: highbast < high comparison not valid for PR/CW.
> +	 * found above.
>  	 */
>  
>  	list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
> -		if (lkb->lkb_bastaddr && (lkb->lkb_highbast < high) &&
> -		    !__dlm_compat_matrix[lkb->lkb_grmode+1][high+1]) {
> +		if (lkb->lkb_bastaddr && lock_requires_bast(lkb, high, cw)) {
>  			queue_bast(r, lkb, high);
>  			lkb->lkb_highbast = high;
>  		}
>  	}
>  }
>  
> +static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
> +{
> +	if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
> +	    (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
> +		if (gr->lkb_highbast < DLM_LOCK_EX)
> +			return 1;
> +		return 0;
> +	}
> +
> +	if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
> +		return 1;
> +	return 0;
> +}
> +
>  static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
>  			    struct dlm_lkb *lkb)
>  {
>  	struct dlm_lkb *gr;
>  
>  	list_for_each_entry(gr, head, lkb_statequeue) {
> -		if (gr->lkb_bastaddr &&
> -		    gr->lkb_highbast < lkb->lkb_rqmode &&
> -		    !modes_compat(gr, lkb)) {
> +		if (gr->lkb_bastaddr && modes_require_bast(gr, lkb)) {
>  			queue_bast(r, gr, lkb->lkb_rqmode);
>  			gr->lkb_highbast = lkb->lkb_rqmode;
>  		}
> @@ -2235,7 +2273,7 @@ static int do_convert(struct dlm_rsb *r,
>  	   before we try again to grant this one. */
>  
>  	if (is_demoted(lkb)) {
> -		grant_pending_convert(r, DLM_LOCK_IV);
> +		grant_pending_convert(r, DLM_LOCK_IV, NULL);
>  		if (_can_be_granted(r, lkb, 1)) {
>  			grant_lock(r, lkb);
>  			queue_cast(r, lkb, 0);

[Cluster-devel] [PATCH] dlm: fix basts for granted PR waiting CW -v2

[David Teigland]

Fix a long standing bug where a blocking callback would be missed
when there's a granted lock in PR mode and waiting locks in both
PR and CW modes (and the PR lock was added to the waiting queue
before the CW lock).  The logic simply compared the numerical values
of the modes to determine if a blocking callback was required, but in
the one case of PR and CW, the lower valued CW mode blocks the higher
valued PR mode.  We just need to add a special check for this PR/CW
case in the tests that decide when a blocking callback is needed.

The first version of this patch would sometimes send an incorrect bast
mode of PR when it should have been CW.

Signed-off-by: David Teigland <teigland@redhat.com>

Index: linux-quilt/fs/dlm/lock.c
===================================================================
--- linux-quilt.orig/fs/dlm/lock.c
+++ linux-quilt/fs/dlm/lock.c
@@ -1670,9 +1670,10 @@ static int can_be_granted(struct dlm_rsb
    with a deadlk here, we'd have to generate something like grant_lock with
    the deadlk error.) */
 
-/* returns the highest requested mode of all blocked conversions */
+/* Returns the highest requested mode of all blocked conversions; sets
+   cw if there's a blocked conversion to DLM_LOCK_CW. */
 
-static int grant_pending_convert(struct dlm_rsb *r, int high)
+static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw)
 {
 	struct dlm_lkb *lkb, *s;
 	int hi, demoted, quit, grant_restart, demote_restart;
@@ -1709,6 +1710,9 @@ static int grant_pending_convert(struct 
 		}
 
 		hi = max_t(int, lkb->lkb_rqmode, hi);
+
+		if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
+			*cw = 1;
 	}
 
 	if (grant_restart)
@@ -1721,29 +1725,52 @@ static int grant_pending_convert(struct 
 	return max_t(int, high, hi);
 }
 
-static int grant_pending_wait(struct dlm_rsb *r, int high)
+static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw)
 {
 	struct dlm_lkb *lkb, *s;
 
 	list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
 		if (can_be_granted(r, lkb, 0, NULL))
 			grant_lock_pending(r, lkb);
-                else
+                else {
 			high = max_t(int, lkb->lkb_rqmode, high);
+			if (lkb->lkb_rqmode == DLM_LOCK_CW)
+				*cw = 1;
+		}
 	}
 
 	return high;
 }
 
+/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
+   on either the convert or waiting queue.
+   high is the largest rqmode of all locks blocked on the convert or
+   waiting queue. */
+
+static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
+{
+	if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
+		if (gr->lkb_highbast < DLM_LOCK_EX)
+			return 1;
+		return 0;
+	}
+
+	if (gr->lkb_highbast < high &&
+	    !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
+		return 1;
+	return 0;
+}
+
 static void grant_pending_locks(struct dlm_rsb *r)
 {
 	struct dlm_lkb *lkb, *s;
 	int high = DLM_LOCK_IV;
+	int cw = 0;
 
 	DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
 
-	high = grant_pending_convert(r, high);
-	high = grant_pending_wait(r, high);
+	high = grant_pending_convert(r, high, &cw);
+	high = grant_pending_wait(r, high, &cw);
 
 	if (high == DLM_LOCK_IV)
 		return;
@@ -1751,27 +1778,41 @@ static void grant_pending_locks(struct d
 	/*
 	 * If there are locks left on the wait/convert queue then send blocking
 	 * ASTs to granted locks based on the largest requested mode (high)
-	 * found above. FIXME: highbast < high comparison not valid for PR/CW.
+	 * found above.
 	 */
 
 	list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
-		if (lkb->lkb_bastaddr && (lkb->lkb_highbast < high) &&
-		    !__dlm_compat_matrix[lkb->lkb_grmode+1][high+1]) {
-			queue_bast(r, lkb, high);
+		if (lkb->lkb_bastaddr && lock_requires_bast(lkb, high, cw)) {
+			if (cw && high == DLM_LOCK_PR)
+				queue_bast(r, lkb, DLM_LOCK_CW);
+			else
+				queue_bast(r, lkb, high);
 			lkb->lkb_highbast = high;
 		}
 	}
 }
 
+static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
+{
+	if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
+	    (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
+		if (gr->lkb_highbast < DLM_LOCK_EX)
+			return 1;
+		return 0;
+	}
+
+	if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
+		return 1;
+	return 0;
+}
+
 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
 			    struct dlm_lkb *lkb)
 {
 	struct dlm_lkb *gr;
 
 	list_for_each_entry(gr, head, lkb_statequeue) {
-		if (gr->lkb_bastaddr &&
-		    gr->lkb_highbast < lkb->lkb_rqmode &&
-		    !modes_compat(gr, lkb)) {
+		if (gr->lkb_bastaddr && modes_require_bast(gr, lkb)) {
 			queue_bast(r, gr, lkb->lkb_rqmode);
 			gr->lkb_highbast = lkb->lkb_rqmode;
 		}
@@ -2235,7 +2276,7 @@ static int do_convert(struct dlm_rsb *r,
 	   before we try again to grant this one. */
 
 	if (is_demoted(lkb)) {
-		grant_pending_convert(r, DLM_LOCK_IV);
+		grant_pending_convert(r, DLM_LOCK_IV, NULL);
 		if (_can_be_granted(r, lkb, 1)) {
 			grant_lock(r, lkb);
 			queue_cast(r, lkb, 0);

[Cluster-devel] [PATCH] [DLM] Clear othercon pointers when a connection is closed

[swhiteho]

From: Patrick Caulfield <pcaulfie@redhat.com>

This patch clears the othercon pointer and frees the memory when a connnection
is closed. This could cause a small memory leak when nodes leave the cluster.

Signed-Off-By: Patrick Caulfield <pcaulfie@redhat.com>
Signed-off-by: Steven Whitehouse <swhiteho@redhat.com>

diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c
index dd36273..d15fd5f 100644
--- a/fs/dlm/lowcomms.c
+++ b/fs/dlm/lowcomms.c
@@ -327,6 +327,8 @@ static void close_connection(struct connection *con, bool and_other)
 	if (con->othercon && and_other) {
 		/* Will only re-enter once. */
 		close_connection(con->othercon, false);
+		kmem_cache_free(con_cache, con->othercon);
+		con->othercon = NULL;
 	}
 	if (con->rx_page) {
 		__free_page(con->rx_page);
-- 
1.5.1.2

[Cluster-devel] [PATCH] [DLM] fix basts for granted PR waiting CW

[swhiteho]

From: David Teigland <teigland@redhat.com>

Fix a long standing bug where a blocking callback would be missed
when there's a granted lock in PR mode and waiting locks in both
PR and CW modes (and the PR lock was added to the waiting queue
before the CW lock).  The logic simply compared the numerical values
of the modes to determine if a blocking callback was required, but in
the one case of PR and CW, the lower valued CW mode blocks the higher
valued PR mode.  We just need to add a special check for this PR/CW
case in the tests that decide when a blocking callback is needed.

Signed-off-by: David Teigland <teigland@redhat.com>
Signed-off-by: Steven Whitehouse <swhiteho@redhat.com>

diff --git a/fs/dlm/lock.c b/fs/dlm/lock.c
index b455919..2082daf 100644
--- a/fs/dlm/lock.c
+++ b/fs/dlm/lock.c
@@ -1670,9 +1670,10 @@ static int can_be_granted(struct dlm_rsb *r, struct dlm_lkb *lkb, int now,
    with a deadlk here, we'd have to generate something like grant_lock with
    the deadlk error.) */
 
-/* returns the highest requested mode of all blocked conversions */
+/* Returns the highest requested mode of all blocked conversions; sets
+   cw if there's a blocked conversion to DLM_LOCK_CW. */
 
-static int grant_pending_convert(struct dlm_rsb *r, int high)
+static int grant_pending_convert(struct dlm_rsb *r, int high, int *cw)
 {
 	struct dlm_lkb *lkb, *s;
 	int hi, demoted, quit, grant_restart, demote_restart;
@@ -1709,6 +1710,9 @@ static int grant_pending_convert(struct dlm_rsb *r, int high)
 		}
 
 		hi = max_t(int, lkb->lkb_rqmode, hi);
+
+		if (cw && lkb->lkb_rqmode == DLM_LOCK_CW)
+			*cw = 1;
 	}
 
 	if (grant_restart)
@@ -1721,29 +1725,52 @@ static int grant_pending_convert(struct dlm_rsb *r, int high)
 	return max_t(int, high, hi);
 }
 
-static int grant_pending_wait(struct dlm_rsb *r, int high)
+static int grant_pending_wait(struct dlm_rsb *r, int high, int *cw)
 {
 	struct dlm_lkb *lkb, *s;
 
 	list_for_each_entry_safe(lkb, s, &r->res_waitqueue, lkb_statequeue) {
 		if (can_be_granted(r, lkb, 0, NULL))
 			grant_lock_pending(r, lkb);
-                else
+                else {
 			high = max_t(int, lkb->lkb_rqmode, high);
+			if (lkb->lkb_rqmode == DLM_LOCK_CW)
+				*cw = 1;
+		}
 	}
 
 	return high;
 }
 
+/* cw of 1 means there's a lock with a rqmode of DLM_LOCK_CW that's blocked
+   on either the convert or waiting queue.
+   high is the largest rqmode of all locks blocked on the convert or
+   waiting queue. */
+
+static int lock_requires_bast(struct dlm_lkb *gr, int high, int cw)
+{
+	if (gr->lkb_grmode == DLM_LOCK_PR && cw) {
+		if (gr->lkb_highbast < DLM_LOCK_EX)
+			return 1;
+		return 0;
+	}
+
+	if (gr->lkb_highbast < high &&
+	    !__dlm_compat_matrix[gr->lkb_grmode+1][high+1])
+		return 1;
+	return 0;
+}
+
 static void grant_pending_locks(struct dlm_rsb *r)
 {
 	struct dlm_lkb *lkb, *s;
 	int high = DLM_LOCK_IV;
+	int cw = 0;
 
 	DLM_ASSERT(is_master(r), dlm_dump_rsb(r););
 
-	high = grant_pending_convert(r, high);
-	high = grant_pending_wait(r, high);
+	high = grant_pending_convert(r, high, &cw);
+	high = grant_pending_wait(r, high, &cw);
 
 	if (high == DLM_LOCK_IV)
 		return;
@@ -1751,27 +1778,41 @@ static void grant_pending_locks(struct dlm_rsb *r)
 	/*
 	 * If there are locks left on the wait/convert queue then send blocking
 	 * ASTs to granted locks based on the largest requested mode (high)
-	 * found above. FIXME: highbast < high comparison not valid for PR/CW.
+	 * found above.
 	 */
 
 	list_for_each_entry_safe(lkb, s, &r->res_grantqueue, lkb_statequeue) {
-		if (lkb->lkb_bastaddr && (lkb->lkb_highbast < high) &&
-		    !__dlm_compat_matrix[lkb->lkb_grmode+1][high+1]) {
-			queue_bast(r, lkb, high);
+		if (lkb->lkb_bastaddr && lock_requires_bast(lkb, high, cw)) {
+			if (cw && high == DLM_LOCK_PR)
+				queue_bast(r, lkb, DLM_LOCK_CW);
+			else
+				queue_bast(r, lkb, high);
 			lkb->lkb_highbast = high;
 		}
 	}
 }
 
+static int modes_require_bast(struct dlm_lkb *gr, struct dlm_lkb *rq)
+{
+	if ((gr->lkb_grmode == DLM_LOCK_PR && rq->lkb_rqmode == DLM_LOCK_CW) ||
+	    (gr->lkb_grmode == DLM_LOCK_CW && rq->lkb_rqmode == DLM_LOCK_PR)) {
+		if (gr->lkb_highbast < DLM_LOCK_EX)
+			return 1;
+		return 0;
+	}
+
+	if (gr->lkb_highbast < rq->lkb_rqmode && !modes_compat(gr, rq))
+		return 1;
+	return 0;
+}
+
 static void send_bast_queue(struct dlm_rsb *r, struct list_head *head,
 			    struct dlm_lkb *lkb)
 {
 	struct dlm_lkb *gr;
 
 	list_for_each_entry(gr, head, lkb_statequeue) {
-		if (gr->lkb_bastaddr &&
-		    gr->lkb_highbast < lkb->lkb_rqmode &&
-		    !modes_compat(gr, lkb)) {
+		if (gr->lkb_bastaddr && modes_require_bast(gr, lkb)) {
 			queue_bast(r, gr, lkb->lkb_rqmode);
 			gr->lkb_highbast = lkb->lkb_rqmode;
 		}
@@ -2235,7 +2276,7 @@ static int do_convert(struct dlm_rsb *r, struct dlm_lkb *lkb)
 	   before we try again to grant this one. */
 
 	if (is_demoted(lkb)) {
-		grant_pending_convert(r, DLM_LOCK_IV);
+		grant_pending_convert(r, DLM_LOCK_IV, NULL);
 		if (_can_be_granted(r, lkb, 1)) {
 			grant_lock(r, lkb);
 			queue_cast(r, lkb, 0);
-- 
1.5.1.2