dlm: fix basts for granted PR waiting CW
David Teigland
maintainerPATCH 1/1 dlm: fix basts for granted PR waiting CW
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.
fs/dlm/lock.c +51 -13
@@ -1670,9 +1670,10 @@ static int can_be_granted(struct dlm_rsbwith 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(structreturn 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 );
David Teigland
maintainerPATCH 1/1 dlm: fix basts for granted PR waiting CW
Quoted 8 lines · 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.
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);
}Hi,
Now in the GFS2 -nmw git tree. Thanks,
Steve.
Quoted 149 lines · 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 <[email protected]>Index:
linux-quilt/fs/dlm/lock.c
===================================================================fs/dlm/lock.c +51
-13@@ -1670,9 +1670,10 @@ static int can_be_granted(struct dlm_rsbwith 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(structreturn 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 );
David Teigland
maintainerPATCH 1/1 dlm: fix basts for granted PR waiting CW -v2
In reply to David TeiglandFix 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.
fs/dlm/lock.c +55 -14
@@ -1670,9 +1670,10 @@ static int can_be_granted(struct dlm_rsbwith 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(structreturn 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 );