path: root/dlm/dlm_load.c

/*
 * Copyright 2010-2011 David Teigland
 *
 * 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.
 */

#include <unistd.h>
#include <stdio.h>
#include <stdlib.h>
#include <stdint.h>
#include <stddef.h>
#include <string.h>
#include <fcntl.h>
#include <errno.h>
#include <time.h>
#include <signal.h>
#include <syslog.h>
#include <sys/time.h>
#include <asm/types.h>
#include <sys/socket.h>
#include <sys/poll.h>
#include <sys/un.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/errno.h>

#include "libdlm.h"

#define LKM_IVMODE -1

#define MAX_CLIENTS 4
#define DEFAULT_LOCKS 16
#define DEFAULT_RESOURCES 16

#define DEFAULT_SLEEP_ITER 10
#define DEFAULT_SLEEP_TIME 200000 /* usecs */

static dlm_lshandle_t *dh;
static int libdlm_fd;
static uint32_t iterations = 0;
static int maxn = DEFAULT_LOCKS;
static int maxr = DEFAULT_RESOURCES;
static int openclose = 0;
static int quiet = 0;
static int do_log_pre = 1;
static int exit_on_error = 1;
static int user_sh = 0;
static int sleep_iter = DEFAULT_SLEEP_ITER;
static int sleep_time = DEFAULT_SLEEP_TIME;
static int timewarn = 0;
static uint64_t our_xid = 0;
static uint32_t ast_count = 0;

struct client {
	int fd;
	char type[32];
};

static int client_size = MAX_CLIENTS;
static struct client client[MAX_CLIENTS];
static struct pollfd pollfd[MAX_CLIENTS];

enum {
	Op_request = 1,
	Op_convert,
	Op_unlock,
	Op_cancel,
};

struct lk {
	int num;
	int locked_stable;
	int unlocked_stable;
	int wait_request;
	int wait_convert;
	int wait_unlock;
	int wait_cancel;
	int rqmode;
	int grmode;
	int lastop;
	int last_status;
	int bast;
	int cancel_busy;
	time_t wait_start;
	struct dlm_lksb lksb;
};

static struct lk *locks;

#define log_debug(fmt, args...) \
do { \
	if (!quiet) \
		printf(fmt "\n", ##args); \
} while (0)

#define log_pre(fmt, args...) \
do { \
	if (!quiet && do_log_pre) \
		printf(fmt "\n", ##args); \
} while (0)

#define log_error(fmt, args...) \
do { \
	printf("ERROR " fmt "\n", ##args); \
	if (exit_on_error) \
		exit(-1); \
} while (0)

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

static const char *status_str(int status)
{
	static char sts_str[8];

	switch (status) {
	case 0:
		return "0      ";
	case EUNLOCK:
		return "U      ";
	case ECANCEL:
		return "ECANCEL";
	case EAGAIN:
		return "EAGAIN ";
	case EBUSY:
		return "EBUSY  ";
	case ETIMEDOUT:
		return "ETIMEDO";
	case EDEADLK:
		return "EDEADLK";
	default:
		snprintf(sts_str, 8, "%8x", status);
		return sts_str;
	}
}

static const char *op_str(int op)
{
	switch (op) {
	case Op_request:
		return "request";
	case Op_convert:
		return "convert";
	case Op_unlock:
		return "unlock ";
	case Op_cancel:
		return "cancel ";
	default:
		return "unknown";
	}
}

static const char *num_str(int num)
{
	static char num_str[8];

	if (num == -1)
		return "-";

	snprintf(num_str, 8, "%d", num);
	return num_str;
}

static struct lk *get_lock(int i)
{
	if (i < 0)
		return NULL;
	if (i >= maxn)
		return NULL;
	return &locks[i];
}

static void dump_lk(int num)
{
	struct lk *lk = get_lock(num);

	if (!lk)
		return;

	printf("lk %03u id %08x gr %s rq %s op %s ",
	       num, lk->lksb.sb_lkid, num_str(lk->grmode), num_str(lk->rqmode),
	       op_str(lk->lastop));

	if (lk->wait_request || lk->wait_convert ||
	    lk->wait_unlock || lk->wait_cancel) {
		printf("wait r%d c%d u%d c%d\n",
			lk->wait_request,
			lk->wait_convert,
			lk->wait_unlock,
			lk->wait_cancel);
	} else {
		printf("done\n");
	}

	fflush(stdout);
}

static void dump(void)
{
	int i;

	for (i = 0; i < maxn; i++)
		dump_lk(i);
}

static void bastfn(void *arg)
{
	struct lk *lk = arg;
	lk->bast = 1;

	ast_count++;

	printf("lk %03u bast\n", lk->num);
	fflush(stdout);
}

static void astfn(void *arg)
{
	struct lk *lk = arg;
	int status = lk->lksb.sb_status;
	int num = lk->num;

	ast_count++;

	switch (status) {
	case 0:
	case EAGAIN:
		if (lk->wait_request || lk->wait_convert) {
			if (lk->wait_request && (status == EAGAIN)) {
				lk->locked_stable = 0;
				lk->unlocked_stable = 1;
			} else {
				lk->locked_stable = 1;
				lk->unlocked_stable = 0;
			}

			lk->wait_request = 0;
			lk->wait_convert = 0;

			if (!status) {
				lk->grmode = lk->rqmode;
				lk->rqmode = LKM_IVMODE;
			} else {
				lk->rqmode = LKM_IVMODE;
			}
		} else {
			/* wait state error */
			log_error("lk %03u cast %04u %s %08x "
				  "wait r%d c%d u%d c%d "
		  		  "gr %2d rq %2d last %s wait state error 1",
				  num, ast_count, status_str(status),
				  lk->lksb.sb_lkid,
				  lk->wait_request, lk->wait_convert,
				  lk->wait_unlock, lk->wait_cancel,
				  lk->grmode, lk->rqmode,
				  op_str(lk->lastop));
		}

		if (lk->wait_cancel) {
			lk->wait_cancel = 0;
		}

		if (lk->wait_unlock) {
			/* wait state error */
			log_error("lk %03u cast %04u %s %08x "
				  "wait r%d c%d u%d c%d "
		  		  "gr %2d rq %2d last %s wait state error 2",
				  num, ast_count, status_str(status),
				  lk->lksb.sb_lkid,
				  lk->wait_request, lk->wait_convert,
				  lk->wait_unlock, lk->wait_cancel,
				  lk->grmode, lk->rqmode,
				  op_str(lk->lastop));
		}
		break;

	case EUNLOCK:
		if (lk->wait_unlock) {
			lk->locked_stable = 0;
			lk->unlocked_stable = 1;

			lk->wait_unlock = 0;

			lk->grmode = LKM_IVMODE;
			lk->bast = 0;
		} else {
			/* wait state error */
			log_error("lk %03u cast %04u %s %08x "
				  "wait r%d c%d u%d c%d "
		  		  "gr %2d rq %2d last %s wait state error 3",
				  num, ast_count, status_str(status),
				  lk->lksb.sb_lkid,
				  lk->wait_request, lk->wait_convert,
				  lk->wait_unlock, lk->wait_cancel,
				  lk->grmode, lk->rqmode,
				  op_str(lk->lastop));
		}

		if (lk->wait_request || lk->wait_convert || lk->wait_cancel) {
			/* wait state error */
			log_error("lk %03u cast %04u %s %08x "
				  "wait r%d c%d u%d c%d "
		  		  "gr %2d rq %2d last %s wait state error 4",
				  num, ast_count, status_str(status),
				  lk->lksb.sb_lkid,
				  lk->wait_request, lk->wait_convert,
				  lk->wait_unlock, lk->wait_cancel,
				  lk->grmode, lk->rqmode,
				  op_str(lk->lastop));
		}
		break;

	case ECANCEL:
		if (lk->wait_cancel) {
			if (lk->grmode > LKM_IVMODE) {
				lk->locked_stable = 1;
				lk->unlocked_stable = 0;
			} else {
				lk->locked_stable = 0;
				lk->unlocked_stable = 1;
			}

			lk->wait_cancel = 0;

			lk->rqmode = LKM_IVMODE;
		} else {
			/* wait state error */
			log_error("lk %03u cast %04u %s %08x "
				  "wait r%d c%d u%d c%d "
		  		  "gr %2d rq %2d last %s wait state error 5",
				  num, ast_count, status_str(status),
				  lk->lksb.sb_lkid,
				  lk->wait_request, lk->wait_convert,
				  lk->wait_unlock, lk->wait_cancel,
				  lk->grmode, lk->rqmode,
				  op_str(lk->lastop));
		}

		if (lk->wait_request || lk->wait_convert) {
			lk->wait_request = 0;
			lk->wait_convert = 0;
		}

		if (lk->wait_unlock) {
			/* wait state error */
			log_error("lk %03u cast %04u %s %08x "
				  "wait r%d c%d u%d c%d "
		  		  "gr %2d rq %2d last %s wait state error 6",
				  num, ast_count, status_str(status),
				  lk->lksb.sb_lkid,
				  lk->wait_request, lk->wait_convert,
				  lk->wait_unlock, lk->wait_cancel,
				  lk->grmode, lk->rqmode,
				  op_str(lk->lastop));
		}
		break;

	case ETIMEDOUT:
		log_error("lk %03u cast %04u %s %08x not using timeouts",
			  num, ast_count, status_str(status),
			  lk->lksb.sb_lkid);
		break;

	case EDEADLK:
		if (lk->wait_convert) {
			lk->locked_stable = 1;
			lk->unlocked_stable = 0;

			lk->wait_convert = 0;

			lk->rqmode = LKM_IVMODE;
		} else {
			/* wait state error */
			log_error("lk %03u cast %04u %s %08x "
				  "wait r%d c%d u%d c%d "
		  		  "gr %2d rq %2d last %s wait state error 7",
				  num, ast_count, status_str(status),
				  lk->lksb.sb_lkid,
				  lk->wait_request, lk->wait_convert,
				  lk->wait_unlock, lk->wait_cancel,
				  lk->grmode, lk->rqmode,
				  op_str(lk->lastop));
		}

		if (lk->wait_request || lk->wait_unlock || lk->wait_cancel) {
			/* wait state error */
			log_error("lk %03u cast %04u %s %08x "
				  "wait r%d c%d u%d c%d "
		  		  "gr %2d rq %2d last %s wait state error 8",
				  num, ast_count, status_str(status),
				  lk->lksb.sb_lkid,
				  lk->wait_request, lk->wait_convert,
				  lk->wait_unlock, lk->wait_cancel,
				  lk->grmode, lk->rqmode,
				  op_str(lk->lastop));
		}

		break;

	default:
		log_error("lk %03u cast %04u %d %08x error unexpected status",
			  num, ast_count, status, lk->lksb.sb_lkid);
	};

	printf("lk %03u id %08x gr %s rq %s op %s cast %s\n",
	       num, lk->lksb.sb_lkid, num_str(lk->grmode), num_str(lk->rqmode),
	       op_str(lk->lastop), status_str(status));
	fflush(stdout);

}

/* FIXME: ran two dlm_load's on four nodes over the weekend, and one instance
   of dlm_load hit:
   lk 007 unlock
   lk 007 cast 3000665 EUNLOCK 022c0912 wait r0 c0 u1 c0 gr  2 rq -1 last unlock
   lk 007 request mode 2 noqueue 1
   lk 007 cast 3000705 EAGAIN  00680909 wait r1 c0 u0 c0 gr -1 rq  2 last request
   lk 007 request mode 0 noqueue 1
   lk 007 cast 3000732 EAGAIN  017b090e wait r1 c0 u0 c0 gr -1 rq  0 last request
   lk 007 request mode 6 noqueue 0
   ERROR lk 007 request errno 22 wait r0 c0 u0 c0 lastop request
*/

static int do_request(int num, int mode, int noqueue)
{
	struct lk *lk;
	char name[DLM_RESNAME_MAXLEN];
	uint64_t *timeout_arg = NULL;
	uint32_t flags = 0;
	int rv;

	lk = get_lock(num);
	if (!lk) {
		printf("unknown lk %d\n", num);
		return;
	}

	if (noqueue)
		flags |= LKF_NOQUEUE;

	memset(name, 0, sizeof(name));
	snprintf(name, sizeof(name), "test%d", (num % maxr));

	log_pre("lk %03u request mode %d noqueue %d", num, mode, noqueue);

	rv = dlm_ls_lockx(dh, mode, &lk->lksb, flags, name, strlen(name), 0,
			  astfn, (void *)lk, bastfn, &our_xid, timeout_arg);

	if (rv) {
		log_error("lk %03u request errno %d mode %d noqueue %d",
			  num, errno, mode, noqueue);
		dump_lk(num);
	} else {
		lk->lastop = Op_request;
		lk->rqmode = mode;
		lk->wait_start = time(NULL);
		lk->wait_request = 1;
		lk->locked_stable = 0;
		lk->unlocked_stable = 0;
	}
	return rv;
}

static int do_convert(int num, int mode, int noqueue, int quecvt)
{
	struct lk *lk;
	char name[DLM_RESNAME_MAXLEN];
	uint64_t *timeout_arg = NULL;
	uint32_t flags = LKF_CONVERT;
	int rv;

	lk = get_lock(num);
	if (!lk) {
		printf("unknown lk %d\n", num);
		return;
	}

	if (noqueue)
		flags |= LKF_NOQUEUE;
	if (quecvt)
		flags |= LKF_QUECVT;

	memset(name, 0, sizeof(name));
	snprintf(name, sizeof(name), "test%d", (num % maxr));

	log_pre("lk %03u convert mode %d %x noqueue %d quecvt %d",
		num, mode, flags, noqueue, quecvt);

	rv = dlm_ls_lockx(dh, mode, &lk->lksb, flags, name, strlen(name), 0,
			  astfn, (void *)lk, bastfn, &our_xid, timeout_arg);

	/* Unfortunately we don't have a way to know when the dlm will be
	   done dealing with a cancel and won't return -EBUSY for ops.
	   (We don't get a callback for the cancel when got a callback for
	   the original op and the cancel did nothing.) */

	if (rv == -1 && errno == EBUSY && lk->lastop == Op_cancel) {
		lk->cancel_busy++;
		log_debug("lk %03u convert cancel_busy %d", lk->cancel_busy);
		goto out;
	} else {
		lk->cancel_busy = 0;
	}

	if (rv) {
		log_error("lk %03u convert errno %d mode %d noqueue %d",
			  num, errno, mode, noqueue);
		dump_lk(num);
	} else {
		lk->lastop = Op_convert;
		lk->rqmode = mode;
		lk->bast = 0;
		lk->wait_start = time(NULL);
		lk->wait_convert = 1;
		lk->locked_stable = 0;
		lk->unlocked_stable = 0;
	}
 out:
	return rv;
}

static int do_unlock(int num)
{
	struct lk *lk;
	uint32_t flags = 0;
	int rv;

	lk = get_lock(num);
	if (!lk) {
		printf("unknown lk %d\n", num);
		return;
	}

	log_pre("lk %03u unlock", num);

	rv = dlm_ls_unlock(dh, lk->lksb.sb_lkid, flags, &lk->lksb, lk);

	/* See comment above. */

	if (rv == -1 && errno == EBUSY && lk->lastop == Op_cancel) {
		lk->cancel_busy++;
		log_debug("lk %03u unlock cancel_busy %d", lk->cancel_busy);
		goto out;
	} else {
		lk->cancel_busy = 0;
	}

	if (rv) {
		log_error("lk %03u unlock errno %d", num, errno);
		dump_lk(num);
	} else {
		lk->lastop = Op_unlock;
		lk->rqmode = LKM_IVMODE;
		lk->bast = 0;
		lk->wait_unlock = 1;
		lk->locked_stable = 0;
		lk->unlocked_stable = 0;
	}
 out:
	return rv;
}

/* The op following a cancel can return -EBUSY.  I don't think there is any
   way to know in this case when we can do the next op without getting
   EBUSY back.  See lk->cancel_busy.  Without the cancel_busy check we get
   this error:
   lk 025 request mode 4 noqueue 0
   lk 025 busy locked 0 unlocked 0 wait r1 c0 u0 c0
   lk 025 cancel
   lk 025 busy locked 0 unlocked 0 wait r1 c0 u0 c1
   lk 025 cast 0706 0       01060001 wait r1 c0 u0 c1 gr -1 rq  4 last cancel
   lk 025 bast 0707
   lk 025 convert mode 0 noqueue 0
   ERROR lk 025 convert errno 16 wait r0 c0 u0 c0 lastop cancel
*/

static int do_cancel(int num)
{
	struct lk *lk;
	uint32_t flags = LKF_CANCEL;
	int rv;

	lk = get_lock(num);
	if (!lk) {
		printf("unknown lk %d\n", num);
		return;
	}

	log_pre("lk %03u cancel", num);

	rv = dlm_ls_unlock(dh, lk->lksb.sb_lkid, flags, &lk->lksb, lk);

	if (rv) {
		log_error("lk %03u cancel errno %d", num, errno);
		dump_lk(num);
	} else {
		lk->lastop = Op_cancel;
		lk->wait_cancel = 1;
		lk->locked_stable = 0;
		lk->unlocked_stable = 0;
	}
	return rv;
}

static void rand_loop(int iter)
{
	struct lk *lk;
	uint32_t n = 0;
	int num, noqueue, quecvt, rqmode, rv;

	while (1) {
		dlm_dispatch(libdlm_fd);

		n++;
		if (iter && n == iter+1)
			break;

		if (sleep_iter && !(n % sleep_iter))
			usleep(sleep_time);

		num = rand_int(0, maxn - 1);
		noqueue = rand_int(0, 1);
		rqmode = rand_int(0, LKM_EXMODE);
		quecvt = 0;

		lk = get_lock(num);
		if (!lk)
			continue;

		if (lk->unlocked_stable) {
			rv = do_request(num, rqmode, noqueue);
			continue;
		}

		if (lk->locked_stable && lk->bast) {
			if (rand_int(0, 1)) {
				rv = do_convert(num, LKM_NLMODE, 0, 0);
			} else {
				rv = do_unlock(num);
			}
			continue;
		}

		if (lk->locked_stable) {
			if (rqmode > lk->grmode)
				quecvt = rand_int(0, 1);

			if (rand_int(0, 1)) {
				rv = do_convert(num, rqmode, noqueue, quecvt);
			} else {
				rv = do_unlock(num);
			}
			continue;
		}

		if ((lk->wait_request || lk->wait_convert) && !lk->wait_cancel) {
			if (time(NULL) - lk->wait_start > 2) {
				rv = do_cancel(num);
			}
		}

		log_debug("lk %03u id %08x gr %s rq %s op %s skip "
			  "ls %d us %d wait r%d c%d u%d c%d",
			  num,
			  lk->lksb.sb_lkid,
			  num_str(lk->grmode),
			  num_str(lk->rqmode),
			  op_str(lk->lastop),
			  lk->locked_stable,
			  lk->unlocked_stable,
			  lk->wait_request,
			  lk->wait_convert,
			  lk->wait_unlock,
			  lk->wait_cancel);
	}
}

static int client_add(int fd, int *maxi)
{
	int i;

	for (i = 0; i < client_size; i++) {
		if (client[i].fd == -1) {
			client[i].fd = fd;
			pollfd[i].fd = fd;
			pollfd[i].events = POLLIN;
			if (i > *maxi)
				*maxi = i;
			log_debug("client %d fd %d added", i, fd);
			return i;
		}
	}
	log_error("client add failed");
	return -1;
}

static void client_dead(int ci)
{
	log_debug("client %d fd %d dead", ci, client[ci].fd);
	close(client[ci].fd);
	client[ci].fd = -1;
	pollfd[ci].fd = -1;
}

static void client_init(void)
{
	int i;

	for (i = 0; i < client_size; i++)
		client[i].fd = -1;
}

static void print_help(void)
{
	printf("quit\n");
	printf("help\n");
	printf("rand num - run num random ops on random lks\n");
	printf("ls [lk]  - list one or all locks\n");
	printf("lock    lk mode noqueue\n");
	printf("convert lk mode noqueue\n");
	printf("quecvt  lk mode\n");
	printf("unlock  lk\n");
	printf("cancel  lk\n");
}

static void user_cmd(int *quit)
{
	char inbuf[132];
	char cmd[132];
	int rv, num = 0, x = 0, y = 0;

	memset(inbuf, 0, sizeof(inbuf));
	memset(cmd, 0, sizeof(cmd));

	fgets(inbuf, sizeof(inbuf), stdin);
	rv = sscanf(inbuf, "%s %d %d %d", cmd, &num, &x, &y);

	if (!rv || !strcmp(cmd, "help")) {
		print_help();
		return;
	}

	if (!strcmp(cmd, "quit")) {
		*quit = 1;
		return;
	}

	if (!strcmp(cmd, "ls") && rv == 1) {
		dump();
		return;
	}

	else if (!strcmp(cmd, "lock")) {
		do_request(num, x, y);
	}

	else if (!strcmp(cmd, "convert")) {
		do_convert(num, x, y, 0);
	}

	else if (!strcmp(cmd, "quecvt")) {
		do_convert(num, x, y, 1);
	}

	else if (!strcmp(cmd, "unlock")) {
		do_unlock(num);
	}
	
	else if (!strcmp(cmd, "cancel")) {
		do_cancel(num);
	}

	else if (!strcmp(cmd, "rand")) {
		rand_loop(num);
	}

	else if (!strcmp(cmd, "ls")) {
		dump_lk(num);
	}

	else if (strlen(cmd) > 0) {
		printf("unknown cmd %s\n", cmd);
		print_help();
	}

	else {
		print_help();
	}
}

static void user_loop(void)
{
	int quit = 0;
	int maxi = 0;
	int rv, i;

	exit_on_error = 0;
	do_log_pre = 0;

	client_add(libdlm_fd, &maxi);
	client_add(STDIN_FILENO, &maxi);

	printf("Type exit to quit, help for usage\n");

	while (!quit) {
		rv = poll(pollfd, maxi + 1, -1);
		if (rv < 0 && errno == EINTR)
			continue;

		for (i = 0; i <= maxi; i++) {
			if (client[i].fd < 0)
				continue;

			if (pollfd[i].revents & POLLIN) {
				if (pollfd[i].fd == libdlm_fd)
					dlm_dispatch(libdlm_fd);
				else if (pollfd[i].fd == STDIN_FILENO) {
					user_cmd(&quit);
				}
			}

			if (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
				client_dead(i);
		}
	}
}

static void print_usage(void)
{
	printf("Options:\n");
	printf("\n");
	printf("  -n <num>     The number of locks to work with, default %d\n", DEFAULT_LOCKS);
	printf("  -r <num>     The number of resources to work with, default %d\n", DEFAULT_RESOURCES);
	printf("  -i <num>     Iterations in looping stress test, default 0 is no limit\n");
	printf("  -S <num>     sleep every <num> iterations, default %d\n", DEFAULT_SLEEP_ITER);
	printf("  -s <num>     sleep <num> usecs, default %d\n", DEFAULT_SLEEP_TIME);
	printf("  -o           Open/close existing lockspace\n");
	printf("  -q           Quiet output, only print errors\n");
	printf("  -u           User command input\n");
}

static void decode_arguments(int argc, char **argv)
{
	int cont = 1;
	int optchar;

	while (cont) {
		optchar = getopt(argc, argv, "n:r:i:s:S:uqoh");

		switch (optchar) {

		case 'n':
			maxn = atoi(optarg);
			break;

		case 'r':
			maxr = atoi(optarg);
			break;

		case 'i':
			iterations = atoi(optarg);
			break;

		case 'o':
			openclose = 1;
			break;

		case 'q':
			quiet = 1;
			break;

		case 's':
			sleep_time = atoi(optarg);
			break;

		case 'S':
			sleep_iter = atoi(optarg);
			break;

		case 'u':
			user_sh = 1;
			break;

		case 'h':
			print_usage();
			exit(EXIT_SUCCESS);
			break;

		case 'V':
			printf("%s (built %s %s)\n", argv[0], __DATE__, __TIME__);
			exit(EXIT_SUCCESS);
			break;

		case ':':
		case '?':
			fprintf(stderr, "Please use '-h' for usage.\n");
			exit(EXIT_FAILURE);
			break;

		case EOF:
			cont = 0;
			break;

		default:
			fprintf(stderr, "unknown option: %c\n", optchar);
			exit(EXIT_FAILURE);
			break;
		};
	}
}

int main(int argc, char *argv[])
{
	uint32_t major, minor, patch;
	struct lk *lk;
	int i, rv, quit = 0;

	srandom(time(NULL));

	decode_arguments(argc, argv);

	if (maxn < maxr) {
		log_error("number of resources must be >= number of locks");
		return -1;
	}
	if (maxn % maxr) {
		log_error("number of locks must be multiple of number of resources");
		return -1;
	}

	log_debug("maxn = %d", maxn);
	log_debug("maxr = %d", maxr);
	log_debug("locks per resource = %d", maxn / maxr);

	client_init();

	locks = malloc(maxn * sizeof(struct lk));
	if (!locks) {
		log_error("no mem for %d locks", maxn);
		return 0;
	}
	memset(locks, 0, sizeof(*locks));

	lk = locks;
	for (i = 0; i < maxn; i++) {
		lk->num = i;
		lk->grmode = -1;
		lk->rqmode = -1;
		lk->unlocked_stable = 1;
		lk++;
	}

	rv = dlm_kernel_version(&major, &minor, &patch);
	if (rv < 0) {
		log_error("can't detect dlm in kernel %d", errno);
		return -1;
	}
	log_debug("dlm kernel version: %u.%u.%u", major, minor, patch);
	dlm_library_version(&major, &minor, &patch);
	log_debug("dlm library version: %u.%u.%u", major, minor, patch);

	if (openclose) {
		log_debug("dlm_open_lockspace...");

		dh = dlm_open_lockspace("test");
		if (!dh) {
			log_error("dlm_open_lockspace error %lu %d",
				  (unsigned long)dh, errno);
			return -ENOTCONN;
		}
	} else {
		log_debug("dlm_new_lockspace...");

		dh = dlm_new_lockspace("test", 0600,
				       timewarn ? DLM_LSFL_TIMEWARN : 0);
		if (!dh) {
			log_error("dlm_new_lockspace error %lu %d",
				  (unsigned long)dh, errno);
			return -ENOTCONN;
		}
	}

	rv = dlm_ls_get_fd(dh);
	if (rv < 0) {
		log_error("dlm_ls_get_fd error %d %d", rv, errno);
		dlm_release_lockspace("test", dh, 1);
		return rv;
	}
	libdlm_fd = rv;

	if (user_sh)
		user_loop();
	else
		rand_loop(iterations);

	if (openclose) {
		log_debug("dlm_close_lockspace");

		rv = dlm_close_lockspace(dh);
		if (rv < 0)
			log_error("dlm_close_lockspace error %d %d",
				  rv, errno);
	} else {
		log_debug("dlm_release_lockspace");

		rv = dlm_release_lockspace("test", dh, 1);
		if (rv < 0)
			log_error("dlm_release_lockspace error %d %d",
				  rv, errno);
	}

	return 0;
}