dlm: master - dlm_controld: new plock state transfer

David Teigland teigland at fedoraproject.org
Fri Sep 23 20:50:57 UTC 2011


Gitweb:        http://git.fedorahosted.org/git/dlm.git?p=dlm.git;a=commitdiff;h=fffedc10167085cb2668e5b86ad4eee38288637e
Commit:        fffedc10167085cb2668e5b86ad4eee38288637e
Parent:        a6e07ad59a764effe13bd02bbc22c0e25b2e225b
Author:        David Teigland <teigland at redhat.com>
AuthorDate:    Fri Jul 8 16:36:20 2011 -0500
Committer:     David Teigland <teigland at redhat.com>
CommitterDate: Fri Sep 23 15:50:12 2011 -0500

dlm_controld: new plock state transfer

- use cpg messages to transfer plock state to new nodes
  instead of openais checkpoints
- split transfer of many plocks on single resource into
  multiple sends
- set default plock_ownership to 0 (like previous versions)
- set daemon protocol version to 2 (not compat with 1)
- don't compile deadlock code (also used ckpts)
- don't compile netlink code (only used for deadlk)
- add new logging routines to make plock debugging easier

Signed-off-by: David Teigland <teigland at redhat.com>
---
 dlm/libdlmcontrol/main.c        |   36 +-
 dlm/tool/main.c                 |    8 +
 group/dlm_controld/Makefile.am  |    6 +-
 group/dlm_controld/config.h     |    2 +-
 group/dlm_controld/cpg.c        |  173 ++++-----
 group/dlm_controld/dlm_daemon.h |   96 ++----
 group/dlm_controld/logging.c    |  120 ++++++
 group/dlm_controld/main.c       |  141 ++-----
 group/dlm_controld/plock.c      |  814 ++++++++++++++-------------------------
 9 files changed, 579 insertions(+), 817 deletions(-)

diff --git a/dlm/libdlmcontrol/main.c b/dlm/libdlmcontrol/main.c
index 0e5c076..a4cf500 100644
--- a/dlm/libdlmcontrol/main.c
+++ b/dlm/libdlmcontrol/main.c
@@ -88,22 +88,16 @@ static void init_header(struct dlmc_header *h, int cmd, char *name,
 		strncpy(h->name, name, DLM_LOCKSPACE_LEN);
 }
 
+static char copy_buf[DLMC_DUMP_SIZE];
+
 static int do_dump(int cmd, char *name, char *buf)
 {
-	struct dlmc_header h, *rh;
-	char *reply;
-	int reply_len;
-	int fd, rv;
+	struct dlmc_header h;
+	int fd, rv, len;
 
-	init_header(&h, cmd, name, 0);
+	memset(copy_buf, 0, DLMC_DUMP_SIZE);
 
-	reply_len = sizeof(struct dlmc_header) + DLMC_DUMP_SIZE;
-	reply = malloc(reply_len);
-	if (!reply) {
-		rv = -1;
-		goto out;
-	}
-	memset(reply, 0, reply_len);
+	init_header(&h, cmd, name, 0);
 
 	fd = do_connect(DLMC_QUERY_SOCK_PATH);
 	if (fd < 0) {
@@ -115,16 +109,22 @@ static int do_dump(int cmd, char *name, char *buf)
 	if (rv < 0)
 		goto out_close;
 
-	/* won't always get back the full reply_len */
-	do_read(fd, reply, reply_len);
+	memset(&h, 0, sizeof(h));
 
-	rh = (struct dlmc_header *)reply;
-	rv = rh->data;
+	rv = do_read(fd, &h, sizeof(h));
+	if (rv < 0)
+		goto out_close;
+
+	len = h.len - sizeof(h);
+
+	if (len <= 0 || len > DLMC_DUMP_SIZE)
+		goto out_close;
+
+	rv = do_read(fd, copy_buf, len);
 	if (rv < 0)
 		goto out_close;
 
-	memcpy(buf, (char *)reply + sizeof(struct dlmc_header),
-	       DLMC_DUMP_SIZE);
+	memcpy(buf, copy_buf, len);
  out_close:
 	close(fd);
  out:
diff --git a/dlm/tool/main.c b/dlm/tool/main.c
index e98a3cc..236463f 100644
--- a/dlm/tool/main.c
+++ b/dlm/tool/main.c
@@ -1242,6 +1242,8 @@ static void do_plocks(char *name)
 
 	dlmc_dump_plocks(name, buf);
 
+	buf[DLMC_DUMP_SIZE-1] = '\0';
+
 	do_write(STDOUT_FILENO, buf, strlen(buf));
 }
 
@@ -1253,7 +1255,10 @@ static void do_dump(void)
 
 	dlmc_dump_debug(buf);
 
+	buf[DLMC_DUMP_SIZE-1] = '\0';
+
 	do_write(STDOUT_FILENO, buf, strlen(buf));
+	printf("\n");
 }
 
 static void do_log_plock(void)
@@ -1264,7 +1269,10 @@ static void do_log_plock(void)
 
 	dlmc_dump_log_plock(buf);
 
+	buf[DLMC_DUMP_SIZE-1] = '\0';
+
 	do_write(STDOUT_FILENO, buf, strlen(buf));
+	printf("\n");
 }
 
 int main(int argc, char **argv)
diff --git a/group/dlm_controld/Makefile.am b/group/dlm_controld/Makefile.am
index 65a7a55..aa68000 100644
--- a/group/dlm_controld/Makefile.am
+++ b/group/dlm_controld/Makefile.am
@@ -4,8 +4,8 @@ sbin_PROGRAMS		= dlm_controld
 
 noinst_HEADERS		= config.h dlm_controld.h dlm_daemon.h
 
-dlm_controld_SOURCES	= action.c cpg.c crc.c deadlock.c main.c \
-			  netlink.c plock.c config.c member_cman.c \
+dlm_controld_SOURCES	= action.c cpg.c crc.c main.c \
+			  plock.c config.c member_cman.c \
 			  logging.c rbtree.c
 
 dlm_controld_CPPFLAGS	= -I$(top_srcdir)/dlm/libdlm \
@@ -14,7 +14,6 @@ dlm_controld_CPPFLAGS	= -I$(top_srcdir)/dlm/libdlm \
 
 dlm_controld_CFLAGS	= $(logt_CFLAGS) \
 			  $(cpg_CFLAGS) \
-			  $(sackpt_CFLAGS) \
 			  $(confdb_CFLAGS)
 			  $(ccs_CFLAGS) \
 			  $(fenced_CFLAGS) \
@@ -23,7 +22,6 @@ dlm_controld_CFLAGS	= $(logt_CFLAGS) \
 
 dlm_controld_LDFLAGS	= $(logt_LIBS) \
 			  $(cpg_LIBS) \
-			  $(sackpt_LIBS) \
 			  $(confdb_LIBS) \
 			  $(ccs_LIBS) \
 			  $(fenced_LIBS) \
diff --git a/group/dlm_controld/config.h b/group/dlm_controld/config.h
index 9a643d6..e575023 100644
--- a/group/dlm_controld/config.h
+++ b/group/dlm_controld/config.h
@@ -11,7 +11,7 @@
 #define DEFAULT_ENABLE_PLOCK 1
 #define DEFAULT_PLOCK_DEBUG 0
 #define DEFAULT_PLOCK_RATE_LIMIT 0
-#define DEFAULT_PLOCK_OWNERSHIP 1
+#define DEFAULT_PLOCK_OWNERSHIP 0
 #define DEFAULT_DROP_RESOURCES_TIME 10000 /* 10 sec */
 #define DEFAULT_DROP_RESOURCES_COUNT 10
 #define DEFAULT_DROP_RESOURCES_AGE 10000 /* 10 sec */
diff --git a/group/dlm_controld/cpg.c b/group/dlm_controld/cpg.c
index 12cb202..f4be240 100644
--- a/group/dlm_controld/cpg.c
+++ b/group/dlm_controld/cpg.c
@@ -201,8 +201,10 @@ const char *msg_name(int type)
 		return "plock_sync_lock";
 	case DLM_MSG_PLOCK_SYNC_WAITER:
 		return "plock_sync_waiter";
-	case DLM_MSG_PLOCKS_STORED:
-		return "plocks_stored";
+	case DLM_MSG_PLOCKS_DATA:
+		return "plocks_data";
+	case DLM_MSG_PLOCKS_DONE:
+		return "plocks_done";
 	case DLM_MSG_DEADLK_CYCLE_START:
 		return "deadlk_cycle_start";
 	case DLM_MSG_DEADLK_CYCLE_END:
@@ -850,7 +852,7 @@ static void cleanup_changes(struct lockspace *ls)
    (although it may be able to reuse the ckpt if no plock state has changed).
 */
 
-static void set_plock_ckpt_node(struct lockspace *ls)
+static void set_plock_data_node(struct lockspace *ls)
 {
 	struct change *cg = list_first_entry(&ls->changes, struct change, list);
 	struct member *memb;
@@ -864,20 +866,10 @@ static void set_plock_ckpt_node(struct lockspace *ls)
 			low = memb->nodeid;
 	}
 
-	log_group(ls, "set_plock_ckpt_node from %d to %d",
-		  ls->plock_ckpt_node, low);
+	log_dlock(ls, "set_plock_data_node from %d to %d",
+		  ls->plock_data_node, low);
 
-	if (ls->plock_ckpt_node == our_nodeid && low != our_nodeid) {
-		/* Close ckpt so it will go away when the new ckpt_node
-		   unlinks it prior to creating a new one; if we fail
-		   our open ckpts are automatically closed.  At this point
-		   the ckpt has not been unlinked, but won't be held open by
-		   anyone.  We use the max "retentionDuration" to stop the
-		   system from cleaning up ckpts that are open by no one. */
-		close_plock_checkpoint(ls);
-	}
-
-	ls->plock_ckpt_node = low;
+	ls->plock_data_node = low;
 }
 
 static struct id_info *get_id_struct(struct id_info *ids, int count, int size,
@@ -1089,32 +1081,25 @@ static void receive_start(struct lockspace *ls, struct dlm_header *hd, int len)
 	memb->start = 1;
 }
 
-static void receive_plocks_stored(struct lockspace *ls, struct dlm_header *hd,
-				  int len)
+static void receive_plocks_done(struct lockspace *ls, struct dlm_header *hd,
+				int len)
 {
 	struct ls_info *li;
 	struct id_info *ids;
-	uint32_t sig;
-
-	log_group(ls, "receive_plocks_stored %d:%u flags %x sig %x "
-		  "need_plocks %d", hd->nodeid, hd->msgdata, hd->flags,
-		  hd->msgdata2, ls->need_plocks);
 
-	log_plock(ls, "receive_plocks_stored %d:%u flags %x sig %x "
-		  "need_plocks %d", hd->nodeid, hd->msgdata, hd->flags,
-		  hd->msgdata2, ls->need_plocks);
-
-	ls->last_plock_sig = hd->msgdata2;
+	log_dlock(ls, "receive_plocks_done %d:%u flags %x plocks_data %u need %d save %d",
+		  hd->nodeid, hd->msgdata, hd->flags, hd->msgdata2,
+		  ls->need_plocks, ls->save_plocks);
 
 	if (!ls->need_plocks)
 		return;
 
-	/* a confchg arrived between the last start and the plocks_stored msg,
-	   so we ignore this plocks_stored msg and wait to read the ckpt until
-	   the next plocks_stored msg following the current start */
-   
-	if (!list_empty(&ls->changes) || !ls->started_change) {
-		log_group(ls, "receive_plocks_stored %d:%u ignore",
+	if (ls->need_plocks && !ls->save_plocks)
+		return;
+
+	if (!ls->started_change) {
+		/* don't think this should happen */
+		log_elock(ls, "receive_plocks_done %d:%u no started_change",
 			  hd->nodeid, hd->msgdata);
 		return;
 	}
@@ -1125,26 +1110,27 @@ static void receive_plocks_stored(struct lockspace *ls, struct dlm_header *hd,
 	ids_in(li, ids);
 
 	if (!match_change(ls, ls->started_change, hd, li, ids)) {
-		log_group(ls, "receive_plocks_stored %d:%u ignore no match",
+		/* don't think this should happen */
+		log_elock(ls, "receive_plocks_done %d:%u no match_change",
 			  hd->nodeid, hd->msgdata);
+
+		/* remove/free anything we've saved from
+		   receive_plocks_data messages that weren't for us */
+		clear_plocks_data(ls);
 		return;
 	}
 
-	retrieve_plocks(ls, &sig);
-
-	if ((hd->flags & DLM_MFLG_PLOCK_SIG) && (sig != hd->msgdata2)) {
-		log_error("lockspace %s plock disabled our sig %x "
-			  "nodeid %d sig %x", ls->name, sig, hd->nodeid,
-			  hd->msgdata2);
-		ls->disable_plock = 1;
-		ls->need_plocks = 1; /* don't set HAVEPLOCK */
-		ls->save_plocks = 0;
-		return;
+	if (ls->recv_plocks_data_count != hd->msgdata2) {
+		log_elock(ls, "receive_plocks_done plocks_data %u recv %u",
+			  hd->msgdata2, ls->recv_plocks_data_count);
 	}
 
 	process_saved_plocks(ls);
 	ls->need_plocks = 0;
 	ls->save_plocks = 0;
+
+	log_dlock(ls, "receive_plocks_done %d:%u plocks_data_count %u",
+		  hd->nodeid, hd->msgdata, ls->recv_plocks_data_count);
 }
 
 static void send_info(struct lockspace *ls, struct change *cg, int type,
@@ -1203,29 +1189,29 @@ static void send_info(struct lockspace *ls, struct change *cg, int type,
 		id++;
 	}
 
-	log_group(ls, "send_%s cg %u flags %x data2 %x counts %u %d %d %d %d",
-		  type == DLM_MSG_START ? "start" : "plocks_stored",
-		  cg->seq, hd->flags, hd->msgdata2, ls->started_count,
-		  cg->member_count, cg->joined_count, cg->remove_count,
-		  cg->failed_count);
-
 	dlm_send_message(ls, buf, len);
 
 	free(buf);
 }
 
-static void send_start(struct lockspace *ls)
+static void send_start(struct lockspace *ls, struct change *cg)
 {
-	struct change *cg = list_first_entry(&ls->changes, struct change, list);
+	log_group(ls, "send_start %d:%u counts %u %d %d %d %d",
+		  our_nodeid, cg->seq, ls->started_count,
+		  cg->member_count, cg->joined_count, cg->remove_count,
+		  cg->failed_count);
 
 	send_info(ls, cg, DLM_MSG_START, 0, 0);
 }
 
-static void send_plocks_stored(struct lockspace *ls, uint32_t sig)
+static void send_plocks_done(struct lockspace *ls, struct change *cg, uint32_t plocks_data)
 {
-	struct change *cg = list_first_entry(&ls->changes, struct change, list);
+	log_dlock(ls, "send_plocks_done %d:%u counts %u %d %d %d %d plocks_data %u",
+		  our_nodeid, cg->seq, ls->started_count,
+		  cg->member_count, cg->joined_count, cg->remove_count,
+		  cg->failed_count, plocks_data);
 
-	send_info(ls, cg, DLM_MSG_PLOCKS_STORED, DLM_MFLG_PLOCK_SIG, sig);
+	send_info(ls, cg, DLM_MSG_PLOCKS_DONE, 0, plocks_data);
 }
 
 static int same_members(struct change *cg1, struct change *cg2)
@@ -1272,74 +1258,58 @@ static void prepare_plocks(struct lockspace *ls)
 {
 	struct change *cg = list_first_entry(&ls->changes, struct change, list);
 	struct member *memb;
-	uint32_t sig;
-
-	log_plock(ls, "prepare_plocks");
+	uint32_t plocks_data;
 
 	if (!cfgd_enable_plock || ls->disable_plock)
 		return;
 
-	/* if we're the only node in the lockspace, then we are the ckpt_node
+	log_dlock(ls, "prepare_plocks");
+
+	/* if we're the only node in the lockspace, then we are the data_node
 	   and we don't need plocks */
 
 	if (cg->member_count == 1) {
 		list_for_each_entry(memb, &cg->members, list) {
 			if (memb->nodeid != our_nodeid) {
-				log_error("prepare_plocks other member %d",
+				log_elock(ls, "prepare_plocks other member %d",
 					  memb->nodeid);
 			}
 		}
-		ls->plock_ckpt_node = our_nodeid;
+		ls->plock_data_node = our_nodeid;
 		ls->need_plocks = 0;
 		return;
 	}
 
 	/* the low node that indicated it had plock state in its last
-	   start message is the ckpt_node */
+	   start message is the data_node */
 
-	set_plock_ckpt_node(ls);
+	set_plock_data_node(ls);
 
 	/* there is no node with plock state, so there's no syncing to do */
 
-	if (!ls->plock_ckpt_node) {
+	if (!ls->plock_data_node) {
 		ls->need_plocks = 0;
 		ls->save_plocks = 0;
 		return;
 	}
 
-	/* We save all plock messages from the time that the low node saves
-	   existing plock state in the ckpt to the time that we read that state
-	   from the ckpt. */
+	/* We save all plock messages received after our own confchg and
+	   apply them after we receive the plocks_done message from the
+	   data_node. */
 
 	if (ls->need_plocks) {
+		log_dlock(ls, "save_plocks start");
 		ls->save_plocks = 1;
 		return;
 	}
 
-	if (ls->plock_ckpt_node != our_nodeid)
+	if (ls->plock_data_node != our_nodeid)
 		return;
 
-	/* At each start, a ckpt is written if there have been nodes added
-	   since the last start/ckpt.  If no nodes have been added, no one
-	   does anything with ckpts.  If the node that wrote the last ckpt
-	   is no longer the ckpt_node, the new ckpt_node will unlink and
-	   write a new one.  If the node that wrote the last ckpt is still
-	   the ckpt_node and no plock state has changed since the last ckpt,
-	   it will just leave the old ckpt and not write a new one.
-	 
-	   A new ckpt_node will send a stored message even if it doesn't
-	   write a ckpt because new nodes in the previous start may be
-	   waiting to read the ckpt from the previous ckpt_node after ignoring
-	   the previous stored message.  They will read the ckpt from the
-	   previous ckpt_node upon receiving the stored message from us. */
-
-	if (nodes_added(ls)) {
-		store_plocks(ls, &sig);
-		ls->last_plock_sig = sig;
-	} else {
-		sig = ls->last_plock_sig;
-	}
-	send_plocks_stored(ls, sig);
+	if (nodes_added(ls))
+		send_all_plocks_data(ls, cg->seq, &plocks_data);
+
+	send_plocks_done(ls, cg, plocks_data);
 }
 
 static void apply_changes(struct lockspace *ls)
@@ -1355,7 +1325,7 @@ static void apply_changes(struct lockspace *ls)
 	case CGST_WAIT_CONDITIONS:
 		if (wait_conditions_done(ls)) {
 			send_nacks(ls, cg);
-			send_start(ls);
+			send_start(ls, cg);
 			cg->state = CGST_WAIT_MESSAGES;
 		}
 		break;
@@ -1556,10 +1526,11 @@ static void confchg_cb(cpg_handle_t handle,
 
 	apply_changes(ls);
 
+#if 0
 	deadlk_confchg(ls, member_list, member_list_entries,
 		       left_list, left_list_entries,
 		       joined_list, joined_list_entries);
-
+#endif
 }
 
 static void dlm_header_in(struct dlm_header *hd)
@@ -1687,16 +1658,27 @@ static void deliver_cb(cpg_handle_t handle,
 				  cfgd_plock_ownership);
 		break;
 
-	case DLM_MSG_PLOCKS_STORED:
+	case DLM_MSG_PLOCKS_DATA:
+		if (ls->disable_plock)
+			break;
+		if (cfgd_enable_plock)
+			receive_plocks_data(ls, hd, len);
+		else
+			log_error("msg %d nodeid %d enable_plock %d",
+				  hd->type, nodeid, cfgd_enable_plock);
+		break;
+
+	case DLM_MSG_PLOCKS_DONE:
 		if (ls->disable_plock)
 			break;
 		if (cfgd_enable_plock)
-			receive_plocks_stored(ls, hd, len);
+			receive_plocks_done(ls, hd, len);
 		else
 			log_error("msg %d nodeid %d enable_plock %d",
 				  hd->type, nodeid, cfgd_enable_plock);
 		break;
 
+#if 0
 	case DLM_MSG_DEADLK_CYCLE_START:
 		if (cfgd_enable_deadlk)
 			receive_cycle_start(ls, hd, len);
@@ -1728,6 +1710,7 @@ static void deliver_cb(cpg_handle_t handle,
 			log_error("msg %d nodeid %d enable_deadlk %d",
 				  hd->type, nodeid, cfgd_enable_deadlk);
 		break;
+#endif
 
 	default:
 		log_error("unknown msg type %d", hd->type);
@@ -2328,7 +2311,7 @@ int setup_cpg_daemon(void)
 	INIT_LIST_HEAD(&daemon_nodes);
 
 	memset(&our_protocol, 0, sizeof(our_protocol));
-	our_protocol.daemon_max[0] = 1;
+	our_protocol.daemon_max[0] = 2;
 	our_protocol.daemon_max[1] = 1;
 	our_protocol.daemon_max[2] = 1;
 	our_protocol.kernel_max[0] = 1;
diff --git a/group/dlm_controld/dlm_daemon.h b/group/dlm_controld/dlm_daemon.h
index 264d068..cb98b3d 100644
--- a/group/dlm_controld/dlm_daemon.h
+++ b/group/dlm_controld/dlm_daemon.h
@@ -22,6 +22,7 @@
 #include <stdlib.h>
 #include <stddef.h>
 #include <stdint.h>
+#include <stdarg.h>
 #include <fcntl.h>
 #include <netdb.h>
 #include <limits.h>
@@ -32,8 +33,6 @@
 #include <signal.h>
 #include <sys/time.h>
 #include <dirent.h>
-#include <openais/saAis.h>
-#include <openais/saCkpt.h>
 
 #include <corosync/cpg.h>
 #include <liblogthread.h>
@@ -83,66 +82,24 @@ extern int plock_ci;
 extern struct list_head lockspaces;
 extern int cluster_quorate;
 extern int our_nodeid;
-extern char plock_dump_buf[DLMC_DUMP_SIZE];
-extern int plock_dump_len;
 extern uint32_t control_minor;
 extern uint32_t monitor_minor;
 extern uint32_t plock_minor;
 extern uint32_t old_plock_minor;
 
-/* circular buffer of log_debug and log_error messages */
-extern char daemon_debug_buf[256];
-extern char dump_buf[DLMC_DUMP_SIZE];
-extern int dump_point;
-extern int dump_wrap;
-
-/* circular buffer of log_plock messages */
-extern char log_plock_line[256];
-extern char log_plock_buf[DLMC_DUMP_SIZE];
-extern int log_plock_point;
-extern int log_plock_wrap;
-
-void daemon_dump_save(void);
-void log_plock_save(void);
-
-#define log_level(lvl, fmt, args...) \
-do { \
-	snprintf(daemon_debug_buf, 255, "%ld " fmt "\n", time(NULL), ##args); \
-	daemon_dump_save(); \
-	logt_print(lvl, fmt "\n", ##args); \
-	if (daemon_debug_opt) \
-		fprintf(stderr, "%s", daemon_debug_buf); \
-} while (0)
-
-#define log_debug(fmt, args...) log_level(LOG_DEBUG, fmt, ##args)
-#define log_error(fmt, args...) log_level(LOG_ERR, fmt, ##args)
-
-#define log_group(ls, fmt, args...) \
-do { \
-	snprintf(daemon_debug_buf, 255, "%ld %s " fmt "\n", time(NULL), \
-		 (ls)->name, ##args); \
-	daemon_dump_save(); \
-	logt_print(LOG_DEBUG, "%s " fmt "\n", (ls)->name, ##args); \
-	if (daemon_debug_opt) \
-		fprintf(stderr, "%s", daemon_debug_buf); \
-} while (0)
-
-#define log_plock(ls, fmt, args...) \
-do { \
-	snprintf(log_plock_line, 255, "%ld %s " fmt "\n", time(NULL), \
-		 (ls)->name, ##args); \
-	log_plock_save(); \
-	if (daemon_debug_opt && cfgd_plock_debug) \
-		fprintf(stderr, "%s", log_plock_line); \
-} while (0)
-
-#define log_plock_error(ls, fmt, args...) \
-do { \
-	log_level(LOG_ERR, fmt, ##args); \
-	snprintf(log_plock_line, 255, "%ld %s " fmt "\n", time(NULL), \
-		 (ls)->name, ##args); \
-	log_plock_save(); \
-} while (0)
+#define LOG_DUMP_SIZE DLMC_DUMP_SIZE
+
+#define LOG_PLOCK 0x00010000
+
+void log_level(char *name_in, uint32_t level_in, const char *fmt, ...);
+
+#define log_error(fmt, args...) log_level(NULL, LOG_ERR, fmt, ##args)
+#define log_debug(fmt, args...) log_level(NULL, LOG_DEBUG, fmt, ##args)
+#define log_group(ls, fmt, args...) log_level((ls)->name, LOG_DEBUG, fmt, ##args)
+
+#define log_plock(ls, fmt, args...) log_level((ls)->name, LOG_PLOCK, fmt, ##args)
+#define log_dlock(ls, fmt, args...) log_level((ls)->name, LOG_PLOCK|LOG_DEBUG, fmt, ##args)
+#define log_elock(ls, fmt, args...) log_level((ls)->name, LOG_PLOCK|LOG_ERR, fmt, ##args)
 
 /* dlm_header types */
 enum {
@@ -153,7 +110,8 @@ enum {
 	DLM_MSG_PLOCK_DROP,
 	DLM_MSG_PLOCK_SYNC_LOCK,
 	DLM_MSG_PLOCK_SYNC_WAITER,
-	DLM_MSG_PLOCKS_STORED,
+	DLM_MSG_PLOCKS_DONE,
+	DLM_MSG_PLOCKS_DATA,
 	DLM_MSG_DEADLK_CYCLE_START,
 	DLM_MSG_DEADLK_CYCLE_END,
 	DLM_MSG_DEADLK_CHECKPOINT_READY,
@@ -202,10 +160,11 @@ struct lockspace {
 
 	/* plock stuff */
 
-	int			plock_ckpt_node;
+	int			plock_data_node;
 	int			need_plocks;
 	int			save_plocks;
 	int			disable_plock;
+	uint32_t		recv_plocks_data_count;
 	uint32_t		associated_mg_id;
 	struct list_head	saved_messages;
 	struct list_head	plock_resources;
@@ -213,13 +172,8 @@ struct lockspace {
 	time_t			last_checkpoint_time;
 	time_t			last_plock_time;
 	struct timeval		drop_resources_last;
-	uint64_t		plock_ckpt_handle;
-	uint64_t		checkpoint_r_num_first;
-	uint64_t		checkpoint_r_num_last;
-	uint32_t		checkpoint_r_count;
-	uint32_t		checkpoint_p_count;
-	uint32_t		last_plock_sig;
 
+#if 0
 	/* deadlock stuff */
 
 	int			deadlk_low_nodeid;
@@ -233,6 +187,7 @@ struct lockspace {
 	struct timeval		last_send_cycle_start;
 	int			cycle_running;
 	int			all_checkpoints_ready;
+#endif
 };
 
 /* action.c */
@@ -334,17 +289,20 @@ void receive_own(struct lockspace *ls, struct dlm_header *hd, int len);
 void receive_sync(struct lockspace *ls, struct dlm_header *hd, int len);
 void receive_drop(struct lockspace *ls, struct dlm_header *hd, int len);
 void process_saved_plocks(struct lockspace *ls);
-void close_plock_checkpoint(struct lockspace *ls);
-void store_plocks(struct lockspace *ls, uint32_t *sig);
-void retrieve_plocks(struct lockspace *ls, uint32_t *sig);
 void purge_plocks(struct lockspace *ls, int nodeid, int unmount);
-int fill_plock_dump_buf(struct lockspace *ls);
+int copy_plock_state(struct lockspace *ls, char *buf, int *len_out);
+
+void send_all_plocks_data(struct lockspace *ls, uint32_t seq, uint32_t *plocks_data);
+void receive_plocks_data(struct lockspace *ls, struct dlm_header *hd, int len);
+void clear_plocks_data(struct lockspace *ls);
 
 /* logging.c */
 
 void init_logging(void);
 void setup_logging(void);
 void close_logging(void);
+void copy_log_dump(char *buf, int *len);
+void copy_log_dump_plock(char *buf, int *len);
 
 /* crc.c */
 uint32_t cpgname_to_crc(const char *data, int len);
diff --git a/group/dlm_controld/logging.c b/group/dlm_controld/logging.c
index cfcacb0..04f48c7 100644
--- a/group/dlm_controld/logging.c
+++ b/group/dlm_controld/logging.c
@@ -59,3 +59,123 @@ void close_logging(void)
 	logt_exit();
 }
 
+#define NAME_ID_SIZE 32
+#define LOG_STR_LEN 512
+static char log_str[LOG_STR_LEN];
+
+static char log_dump[LOG_DUMP_SIZE];
+static unsigned int log_point;
+static unsigned int log_wrap;
+
+static char log_dump_plock[LOG_DUMP_SIZE];
+static unsigned int log_point_plock;
+static unsigned int log_wrap_plock;
+
+static void log_copy(char *buf, int *len, char *log_buf,
+		     unsigned int *point, unsigned int *wrap)
+{
+	unsigned int p = *point;
+	unsigned int w = *wrap;
+	int tail_len;
+
+	if (!w && !p) {
+		*len = 0;
+	} else if (*wrap) {
+		tail_len = LOG_DUMP_SIZE - p;
+		memcpy(buf, log_buf + p, tail_len);
+		if (p)
+			memcpy(buf+tail_len, log_buf, p);
+		*len = LOG_DUMP_SIZE;
+	} else {
+		memcpy(buf, log_buf, p-1);
+		*len = p-1;
+	}
+}
+
+void copy_log_dump(char *buf, int *len)
+{
+	log_copy(buf, len, log_dump, &log_point, &log_wrap);
+}
+
+void copy_log_dump_plock(char *buf, int *len)
+{
+	log_copy(buf, len, log_dump_plock, &log_point_plock, &log_wrap_plock);
+}
+
+static void log_save_str(int level, int len, char *log_buf,
+			 unsigned int *point, unsigned int *wrap)
+{
+	unsigned int p = *point;
+	unsigned int w = *wrap;
+	int i;
+
+	if (len < LOG_DUMP_SIZE - p) {
+		memcpy(log_buf + p, log_str, len);
+		p += len;
+
+		if (p == LOG_DUMP_SIZE) {
+			p = 0;
+			w = 1;
+		}
+		goto out;
+	}
+
+	for (i = 0; i < len; i++) {
+		log_buf[p++] = log_str[i];
+
+		if (p == LOG_DUMP_SIZE) {
+			p = 0;
+			w = 1;
+		}
+	}
+ out:
+	*point = p;
+	*wrap = w;
+}
+
+void log_level(char *name_in, uint32_t level_in, const char *fmt, ...)
+{
+	va_list ap;
+	char name[NAME_ID_SIZE + 1];
+	uint32_t level = level_in & 0x0000FFFF;
+	uint32_t extra = level_in & 0xFFFF0000;
+	int ret, pos = 0;
+	int len = LOG_STR_LEN - 2;
+	int plock = extra & LOG_PLOCK;
+
+	memset(name, 0, sizeof(name));
+
+	if (name_in)
+		snprintf(name, NAME_ID_SIZE, "%s ", name_in);
+
+	ret = snprintf(log_str + pos, len - pos, "%llu %s",
+		       (unsigned long long)time(NULL), name);
+
+	pos += ret;
+
+	va_start(ap, fmt);
+	ret = vsnprintf(log_str + pos, len - pos, fmt, ap);
+	va_end(ap);
+
+	if (ret >= len - pos)
+		pos = len - 1;
+	else
+		pos += ret;
+
+	log_str[pos++] = '\n';
+	log_str[pos++] = '\0';
+
+	if (level)
+		log_save_str(level, pos - 1, log_dump, &log_point, &log_wrap);
+	if (plock)
+		log_save_str(level, pos - 1, log_dump_plock, &log_point_plock, &log_wrap_plock);
+	if (level)
+		logt_print(level, "%s", log_str);
+
+	if (!daemon_debug_opt)
+		return;
+
+	if (level || (plock && cfgd_plock_debug))
+		fprintf(stderr, "%s", log_str);
+}
+
diff --git a/group/dlm_controld/main.c b/group/dlm_controld/main.c
index afc22bd..2f023ee 100644
--- a/group/dlm_controld/main.c
+++ b/group/dlm_controld/main.c
@@ -165,9 +165,11 @@ static struct lockspace *create_ls(char *name)
 	INIT_LIST_HEAD(&ls->saved_messages);
 	INIT_LIST_HEAD(&ls->plock_resources);
 	ls->plock_resources_root = RB_ROOT;
+#if 0
 	INIT_LIST_HEAD(&ls->deadlk_nodes);
 	INIT_LIST_HEAD(&ls->transactions);
 	INIT_LIST_HEAD(&ls->resources);
+#endif
  out:
 	return ls;
 }
@@ -408,85 +410,56 @@ static void init_header(struct dlmc_header *h, int cmd, char *name, int result,
 		strncpy(h->name, name, DLM_LOCKSPACE_LEN);
 }
 
+static char copy_buf[LOG_DUMP_SIZE];
+
 static void query_dump_debug(int fd)
 {
 	struct dlmc_header h;
-	int extra_len;
-	int len;
-
-	/* in the case of dump_wrap, extra_len will go in two writes,
-	   first the log tail, then the log head */
-	if (dump_wrap)
-		extra_len = DLMC_DUMP_SIZE;
-	else
-		extra_len = dump_point;
-
-	init_header(&h, DLMC_CMD_DUMP_DEBUG, NULL, 0, extra_len);
-	do_write(fd, &h, sizeof(h));
-
-	if (dump_wrap) {
-		len = DLMC_DUMP_SIZE - dump_point;
-		do_write(fd, dump_buf + dump_point, len);
-		len = dump_point;
-	} else
-		len = dump_point;
-
-	/* NUL terminate the debug string */
-	dump_buf[dump_point] = '\0';
-
-	do_write(fd, dump_buf, len);
+	int len = 0;
+
+	copy_log_dump(copy_buf, &len);
+
+	init_header(&h, DLMC_CMD_DUMP_DEBUG, NULL, 0, len);
+	send(fd, &h, sizeof(h), MSG_NOSIGNAL);
+
+	if (len)
+		send(fd, copy_buf, len, MSG_NOSIGNAL);
 }
 
 static void query_dump_log_plock(int fd)
 {
 	struct dlmc_header h;
-	int extra_len;
-	int len;
-
-	/* in the case of dump_wrap, extra_len will go in two writes,
-	   first the log tail, then the log head */
-	if (log_plock_wrap)
-		extra_len = DLMC_DUMP_SIZE;
-	else
-		extra_len = log_plock_point;
-
-	init_header(&h, DLMC_CMD_DUMP_LOG_PLOCK, NULL, 0, extra_len);
-	do_write(fd, &h, sizeof(h));
-
-	if (log_plock_wrap) {
-		len = DLMC_DUMP_SIZE - log_plock_point;
-		do_write(fd, log_plock_buf + log_plock_point, len);
-		len = log_plock_point;
-	} else
-		len = log_plock_point;
-
-	/* NUL terminate the debug string */
-	log_plock_buf[log_plock_point] = '\0';
-
-	do_write(fd, log_plock_buf, len);
+	int len = 0;
+
+	copy_log_dump_plock(copy_buf, &len);
+
+	init_header(&h, DLMC_CMD_DUMP_DEBUG, NULL, 0, len);
+	send(fd, &h, sizeof(h), MSG_NOSIGNAL);
+
+	if (len)
+		send(fd, copy_buf, len, MSG_NOSIGNAL);
 }
 
 static void query_dump_plocks(int fd, char *name)
 {
 	struct lockspace *ls;
 	struct dlmc_header h;
+	int len = 0;
 	int rv;
 
 	ls = find_ls(name);
 	if (!ls) {
-		plock_dump_len = 0;
 		rv = -ENOENT;
-	} else {
-		/* writes to plock_dump_buf and sets plock_dump_len */
-		rv = fill_plock_dump_buf(ls);
+		goto out;
 	}
 
-	init_header(&h, DLMC_CMD_DUMP_PLOCKS, name, rv, plock_dump_len);
-
-	do_write(fd, &h, sizeof(h));
+	rv = copy_plock_state(ls, copy_buf, &len);
+ out:
+	init_header(&h, DLMC_CMD_DUMP_PLOCKS, name, rv, len);
+	send(fd, &h, sizeof(h), MSG_NOSIGNAL);
 
-	if (plock_dump_len)
-		do_write(fd, plock_dump_buf, plock_dump_len);
+	if (len)
+		send(fd, copy_buf, len, MSG_NOSIGNAL);
 }
 
 /* combines a header and the data and sends it back to the client in
@@ -690,13 +663,14 @@ static void process_connection(int ci)
 			 h.data, NULL, 0);
 		break;
 
+#if 0
 	case DLMC_CMD_DEADLOCK_CHECK:
 		ls = find_ls(h.name);
 		if (ls)
 			send_cycle_start(ls);
 		client_dead(ci);
 		break;
-
+#endif
 	default:
 		log_error("process_connection %d unknown command %d",
 			  ci, h.command);
@@ -932,6 +906,7 @@ static void loop(void)
 	if (rv < 0)
 		goto out;
 
+#if 0
 	if (cfgd_enable_deadlk) {
 		rv = setup_netlink();
 		if (rv < 0)
@@ -940,6 +915,7 @@ static void loop(void)
 
 		setup_deadlock();
 	}
+#endif
 
 	rv = setup_plocks();
 	if (rv < 0)
@@ -1079,8 +1055,10 @@ static void print_usage(void)
 	printf("		Default is %d\n", DEFAULT_ENABLE_FENCING);
 	printf("  -q <num>	Enable (1) or disable (0) quorum recovery dependency\n");
 	printf("		Default is %d\n", DEFAULT_ENABLE_QUORUM);
+#if 0
 	printf("  -d <num>	Enable (1) or disable (0) deadlock detection code\n");
 	printf("		Default is %d\n", DEFAULT_ENABLE_DEADLK);
+#endif
 	printf("  -p <num>	Enable (1) or disable (0) plock code for cluster fs\n");
 	printf("		Default is %d\n", DEFAULT_ENABLE_PLOCK);
 	printf("  -P		Enable plock debugging\n");
@@ -1098,7 +1076,7 @@ static void print_usage(void)
 	printf("  -V		Print program version information, then exit\n");
 }
 
-#define OPTION_STRING "LDKf:q:d:p:Pl:o:t:c:a:hVr:"
+#define OPTION_STRING "LDKf:q:p:Pl:o:t:c:a:hVr:"
 
 static void read_arguments(int argc, char **argv)
 {
@@ -1109,7 +1087,6 @@ static void read_arguments(int argc, char **argv)
 		optchar = getopt(argc, argv, OPTION_STRING);
 
 		switch (optchar) {
-
 		case 'D':
 			daemon_debug_opt = 1;
 			break;
@@ -1247,7 +1224,7 @@ int main(int argc, char **argv)
 	}
 	lockfile();
 	init_logging();
-	log_level(LOG_INFO, "dlm_controld %s started", VERSION);
+	log_level(NULL, LOG_INFO, "dlm_controld %s started", VERSION);
 	signal(SIGTERM, sigterm_handler);
 	set_scheduler();
 
@@ -1256,38 +1233,6 @@ int main(int argc, char **argv)
 	return 0;
 }
 
-void daemon_dump_save(void)
-{
-	int len, i;
-
-	len = strlen(daemon_debug_buf);
-
-	for (i = 0; i < len; i++) {
-		dump_buf[dump_point++] = daemon_debug_buf[i];
-
-		if (dump_point == DLMC_DUMP_SIZE) {
-			dump_point = 0;
-			dump_wrap = 1;
-		}
-	}
-}
-
-void log_plock_save(void)
-{
-	int len, i;
-
-	len = strlen(log_plock_line);
-
-	for (i = 0; i < len; i++) {
-		log_plock_buf[log_plock_point++] = log_plock_line[i];
-
-		if (log_plock_point == DLMC_DUMP_SIZE) {
-			log_plock_point = 0;
-			log_plock_wrap = 1;
-		}
-	}
-}
-
 int daemon_debug_opt;
 int daemon_quit;
 int cluster_down;
@@ -1301,20 +1246,10 @@ int plock_ci;
 struct list_head lockspaces;
 int cluster_quorate;
 int our_nodeid;
-char plock_dump_buf[DLMC_DUMP_SIZE];
-int plock_dump_len;
 uint32_t control_minor;
 uint32_t monitor_minor;
 uint32_t plock_minor;
 uint32_t old_plock_minor;
-char daemon_debug_buf[256];
-char dump_buf[DLMC_DUMP_SIZE];
-int dump_point;
-int dump_wrap;
-char log_plock_line[256];
-char log_plock_buf[DLMC_DUMP_SIZE];
-int log_plock_point;
-int log_plock_wrap;
 
 /* was a config value set on command line?, 0 or 1.
    optk is a kernel option, optd is a daemon option */
diff --git a/group/dlm_controld/plock.c b/group/dlm_controld/plock.c
index f8d6b02..308eeae 100644
--- a/group/dlm_controld/plock.c
+++ b/group/dlm_controld/plock.c
@@ -20,16 +20,21 @@ static struct timeval plock_recv_time;
 static struct timeval plock_rate_last;
 
 static int plock_device_fd = -1;
-static SaCkptHandleT system_ckpt_handle;
-static SaCkptCallbacksT callbacks = { 0, 0 };
-static SaVersionT version = { 'B', 1, 1 };
-static char section_buf[1024 * 1024];
-static uint32_t section_len;
 static int need_fsid_translation = 0;
 
 extern int message_flow_control_on;
 
-struct pack_plock {
+#define RD_CONTINUE 0x00000001
+
+struct resource_data {
+	uint64_t number;
+	int      owner;
+	uint32_t lock_count;
+	uint32_t flags;
+	uint32_t pad;
+};
+
+struct plock_data {
 	uint64_t start;
 	uint64_t end;
 	uint64_t owner;
@@ -161,8 +166,6 @@ static const char *ex_str(int optype, int ex)
 
 int setup_plocks(void)
 {
-	SaAisErrorT err;
-
 	plock_read_count = 0;
 	plock_recv_count = 0;
 	plock_rate_delays = 0;
@@ -170,15 +173,6 @@ int setup_plocks(void)
 	gettimeofday(&plock_recv_time, NULL);
 	gettimeofday(&plock_rate_last, NULL);
 
-	err = saCkptInitialize(&system_ckpt_handle, &callbacks, &version);
-	if (err != SA_AIS_OK) {
-		log_error("ckpt init error %d", err);
-		cfgd_enable_plock = 0;
-
-		/* still try to open and read the control device so that we can
-		   send ENOSYS back to the kernel if it tries to do a plock */
-	}
-
 	if (plock_minor) {
 		plock_device_fd = open("/dev/misc/dlm_plock", O_RDWR);
 	} else if (old_plock_minor) {
@@ -202,8 +196,6 @@ int setup_plocks(void)
 
 void close_plocks(void)
 {
-	if (system_ckpt_handle)
-		saCkptFinalize(system_ckpt_handle);
 	if (plock_device_fd > 0)
 		close(plock_device_fd);
 }
@@ -321,7 +313,7 @@ static int find_resource(struct lockspace *ls, uint64_t number, int create,
 
 	r = malloc(sizeof(struct resource));
 	if (!r) {
-		log_plock_error(ls, "find_resource no memory %d", errno);
+		log_elock(ls, "find_resource no memory %d", errno);
 		rv = -ENOMEM;
 		goto out;
 	}
@@ -699,12 +691,12 @@ static void clear_waiters(struct lockspace *ls, struct resource *r,
 
 		list_del(&w->list);
 
-		log_plock_error(ls, "clear waiter %llx %llx-%llx %d/%u/%llx",
-				(unsigned long long)in->number,
-				(unsigned long long)in->start,
-				(unsigned long long)in->end,
-				in->nodeid, in->pid,
-				(unsigned long long)in->owner);
+		log_elock(ls, "clear waiter %llx %llx-%llx %d/%u/%llx",
+			  (unsigned long long)in->number,
+			  (unsigned long long)in->start,
+			  (unsigned long long)in->end,
+			  in->nodeid, in->pid,
+			  (unsigned long long)in->owner);
 		free(w);
 	}
 }
@@ -864,8 +856,8 @@ static void __receive_plock(struct lockspace *ls, struct dlm_plock_info *in,
 		do_get(ls, in, r);
 		break;
 	default:
-		log_plock_error(ls, "receive_plock error from %d optype %d",
-				from, in->optype);
+		log_elock(ls, "receive_plock error from %d optype %d",
+			  from, in->optype);
 		if (from == our_nodeid)
 			write_result(ls, in, -EINVAL);
 	}
@@ -914,8 +906,8 @@ static void _receive_plock(struct lockspace *ls, struct dlm_header *hd, int len)
 		return;
 
 	if (from != hd->nodeid || from != info.nodeid) {
-		log_plock_error(ls, "receive_plock error from %d header %d info %d",
-				from, hd->nodeid, info.nodeid);
+		log_elock(ls, "receive_plock error from %d header %d info %d",
+			  from, hd->nodeid, info.nodeid);
 		return;
 	}
 
@@ -946,8 +938,8 @@ static void _receive_plock(struct lockspace *ls, struct dlm_header *hd, int len)
 		/* r not found, rv is -ENOENT, this shouldn't happen because
 		   process_plocks() creates a resource for every op */
 
-		log_plock_error(ls, "receive_plock error from %d no r %llx %d",
-				from, (unsigned long long)info.number, rv);
+		log_elock(ls, "receive_plock error from %d no r %llx %d",
+			  from, (unsigned long long)info.number, rv);
 		return;
 	}
 
@@ -1036,7 +1028,7 @@ static int send_struct_info(struct lockspace *ls, struct dlm_plock_info *in,
 	free(buf);
  out:
 	if (rv)
-		log_plock_error(ls, "send_struct_info error %d", rv);
+		log_elock(ls, "send_struct_info error %d", rv);
 	return rv;
 }
 
@@ -1129,7 +1121,7 @@ static void save_pending_plock(struct lockspace *ls, struct resource *r,
 
 	w = malloc(sizeof(struct lock_waiter));
 	if (!w) {
-		log_plock_error(ls, "save_pending_plock no mem");
+		log_elock(ls, "save_pending_plock no mem");
 		return;
 	}
 	memcpy(&w->info, in, sizeof(struct dlm_plock_info));
@@ -1280,10 +1272,10 @@ static void _receive_own(struct lockspace *ls, struct dlm_header *hd, int len)
 	}
 
 	if (should_not_happen) {
-		log_plock_error(ls, "receive_own error from %u %llx "
-				"info nodeid %d r owner %d",
-			  	from, (unsigned long long)r->number,
-				info.nodeid, r->owner);
+		log_elock(ls, "receive_own error from %u %llx "
+			  "info nodeid %d r owner %d",
+			  from, (unsigned long long)r->number,
+			  info.nodeid, r->owner);
 	}
 }
 
@@ -1329,14 +1321,13 @@ static void clear_syncing_flag(struct lockspace *ls, struct resource *r,
 		}
 	}
 
-	log_plock_error(ls, "clear_syncing error %llx no match %s %llx-%llx "
-			"%d/%u/%llx",
-			(unsigned long long)r->number,
-			in->ex ? "WR" : "RD", 
-			(unsigned long long)in->start,
-			(unsigned long long)in->end,
-			in->nodeid, in->pid,
-			(unsigned long long)in->owner);
+	log_elock(ls, "clear_syncing error %llx no match %s %llx-%llx %d/%u/%llx",
+		  (unsigned long long)r->number,
+		  in->ex ? "WR" : "RD", 
+		  (unsigned long long)in->start,
+		  (unsigned long long)in->end,
+		  in->nodeid, in->pid,
+		  (unsigned long long)in->owner);
 }
 
 static void _receive_sync(struct lockspace *ls, struct dlm_header *hd, int len)
@@ -1358,8 +1349,8 @@ static void _receive_sync(struct lockspace *ls, struct dlm_header *hd, int len)
 
 	rv = find_resource(ls, info.number, 0, &r);
 	if (rv) {
-		log_plock_error(ls, "receive_sync error no r %llx from %d",
-				info.number, from);
+		log_elock(ls, "receive_sync error no r %llx from %d",
+			  info.number, from);
 		return;
 	}
 
@@ -1422,8 +1413,8 @@ static void _receive_drop(struct lockspace *ls, struct dlm_header *hd, int len)
 
 	if (!list_empty(&r->pending)) {
 		/* shouldn't happen */
-		log_plock_error(ls, "receive_drop error from %d r %llx pending op",
-				from, (unsigned long long)r->number);
+		log_elock(ls, "receive_drop error from %d r %llx pending op",
+			  from, (unsigned long long)r->number);
 		return;
 	}
 
@@ -1661,7 +1652,7 @@ void process_saved_plocks(struct lockspace *ls)
 	struct dlm_header *hd;
 	int count = 0;
 
-	log_plock(ls, "process_saved_plocks begin");
+	log_dlock(ls, "process_saved_plocks begin");
 
 	if (list_empty(&ls->saved_messages))
 		goto out;
@@ -1692,31 +1683,59 @@ void process_saved_plocks(struct lockspace *ls)
 		count++;
 	}
  out:
-	log_plock(ls, "process_saved_plocks %d end", count);
+	log_dlock(ls, "process_saved_plocks %d done", count);
 }
 
 /* locks still marked SYNCING should not go into the ckpt; the new node
    will get those locks by receiving PLOCK_SYNC messages */
 
-static void pack_section_buf(struct lockspace *ls, struct resource *r,
-			     int owner)
+#define MAX_SEND_SIZE 1024 /* 1024 holds 24 plock_data */
+
+static char send_buf[MAX_SEND_SIZE];
+
+static int pack_send_buf(struct lockspace *ls, struct resource *r, int owner,
+			 int full, int *count_out, void **last)
 {
-	struct pack_plock *pp;
+	struct resource_data *rd;
+	struct plock_data *pp;
 	struct posix_lock *po;
 	struct lock_waiter *w;
 	int count = 0;
+	int find = 0;
+	int len;
+
+	/* N.B. owner not always equal to r->owner */
+	rd = (struct resource_data *)(send_buf + sizeof(struct dlm_header));
+	rd->number = cpu_to_le64(r->number);
+	rd->owner = cpu_to_le32(owner);
 
-	/* plocks on owned resources are not replicated on other nodes;
-	   N.B. owner not always equal to r->owner */
+	if (full) {
+		rd->flags = RD_CONTINUE;
+		find = 1;
+	}
 
+	/* plocks not replicated for owned resources */
 	if (cfgd_plock_ownership && (owner == our_nodeid))
-		return;
+		goto done;
 
-	pp = (struct pack_plock *) &section_buf;
+	len = sizeof(struct dlm_header) + sizeof(struct resource_data);
+
+	pp = (struct plock_data *)(send_buf + sizeof(struct dlm_header) + sizeof(struct resource_data));
 
 	list_for_each_entry(po, &r->locks, list) {
+		if (find && *last != po)
+			continue;
+		find = 0;
+
 		if (po->flags & P_SYNCING)
 			continue;
+
+		if (len + sizeof(struct plock_data) > sizeof(send_buf)) {
+			*last = po;
+			goto full;
+		}
+		len += sizeof(struct plock_data);
+
 		pp->start	= cpu_to_le64(po->start);
 		pp->end		= cpu_to_le64(po->end);
 		pp->owner	= cpu_to_le64(po->owner);
@@ -1729,8 +1748,19 @@ static void pack_section_buf(struct lockspace *ls, struct resource *r,
 	}
 
 	list_for_each_entry(w, &r->waiters, list) {
+		if (find && *last != w)
+			continue;
+		find = 0;
+
 		if (w->flags & P_SYNCING)
 			continue;
+
+		if (len + sizeof(struct plock_data) > sizeof(send_buf)) {
+			*last = w;
+			goto full;
+		}
+		len += sizeof(struct plock_data);
+
 		pp->start	= cpu_to_le64(w->info.start);
 		pp->end		= cpu_to_le64(w->info.end);
 		pp->owner	= cpu_to_le64(w->info.owner);
@@ -1741,210 +1771,18 @@ static void pack_section_buf(struct lockspace *ls, struct resource *r,
 		pp++;
 		count++;
 	}
-
-	section_len = count * sizeof(struct pack_plock);
-}
-
-static int unpack_section_buf(struct lockspace *ls, char *numbuf, int buflen,
-			      uint64_t *r_num, int *lock_count)
-{
-	struct pack_plock *pp;
-	struct posix_lock *po;
-	struct lock_waiter *w;
-	struct resource *r;
-	int count = section_len / sizeof(struct pack_plock);
-	int i, owner = 0;
-	unsigned long long num;
-	struct timeval now;
-
-	gettimeofday(&now, NULL);
-
-	sscanf(numbuf, "r%llu.%d", &num, &owner);
-
-#if 0
-	/* would be nice to always compile this, but it adds a lot of time */
-	r = search_resource(ls, num);
-	if (r) {
-		log_error("unpack %llu duplicate", num);
-		return -1;
-	}
-#endif
-
-	r = malloc(sizeof(struct resource));
-	if (!r)
-		return -ENOMEM;
-	memset(r, 0, sizeof(struct resource));
-	INIT_LIST_HEAD(&r->locks);
-	INIT_LIST_HEAD(&r->waiters);
-	INIT_LIST_HEAD(&r->pending);
-
-	if (!cfgd_plock_ownership) {
-		if (owner) {
-			log_error("unpack %llu bad owner %d count %d",
-				  (unsigned long long)num, owner, count);
-			free(r);
-			return -1;
-		}
-	} else {
-		if (!owner)
-			r->flags |= R_GOT_UNOWN;
-
-		/* no locks should be included for owned resources */
-
-		if (owner && count) {
-			log_error("unpack %llu owner %d bad count %d",
-				  (unsigned long long)num, owner, count);
-			free(r);
-			return -1;
-		}
-	}
-
-	r->number = num;
-	r->owner = owner;
-	r->last_access = now;
-
-	*r_num = num;
-
-	pp = (struct pack_plock *) &section_buf;
-
-	for (i = 0; i < count; i++) {
-		if (!pp->waiter) {
-			po = malloc(sizeof(struct posix_lock));
-			if (!po)
-				return -ENOMEM;
-			po->start	= le64_to_cpu(pp->start);
-			po->end		= le64_to_cpu(pp->end);
-			po->owner	= le64_to_cpu(pp->owner);
-			po->pid		= le32_to_cpu(pp->pid);
-			po->nodeid	= le32_to_cpu(pp->nodeid);
-			po->ex		= pp->ex;
-			list_add_tail(&po->list, &r->locks);
-		} else {
-			w = malloc(sizeof(struct lock_waiter));
-			if (!w)
-				return -ENOMEM;
-			w->info.start	= le64_to_cpu(pp->start);
-			w->info.end	= le64_to_cpu(pp->end);
-			w->info.owner	= le64_to_cpu(pp->owner);
-			w->info.pid	= le32_to_cpu(pp->pid);
-			w->info.nodeid	= le32_to_cpu(pp->nodeid);
-			w->info.ex	= pp->ex;
-			list_add_tail(&w->list, &r->waiters);
-		}
-		pp++;
-	}
-
-	list_add_tail(&r->list, &ls->plock_resources);
-	rb_insert_plock_resource(ls, r);
-	*lock_count = count;
+ done:
+	rd->lock_count = cpu_to_le32(count);
+	*count_out = count;
+	*last = NULL;
 	return 0;
-}
-
-/* If we are the new ckpt_node, we'll be unlinking a ckpt that we don't
-   have open, which was created by the previous ckpt_node.  The previous
-   ckpt_node should have closed the ckpt in set_plock_ckpt_node() so it
-   will go away when we unlink it here. */
-
-static int _unlink_checkpoint(struct lockspace *ls, SaNameT *name)
-{
-	SaCkptCheckpointHandleT h;
-	SaCkptCheckpointDescriptorT s;
-	SaAisErrorT rv;
-	int ret = 0;
-
-	h = (SaCkptCheckpointHandleT) ls->plock_ckpt_handle;
-	log_group(ls, "unlink ckpt %llx", (unsigned long long)h);
-
- unlink_retry:
-	rv = saCkptCheckpointUnlink(system_ckpt_handle, name);
-	if (rv == SA_AIS_ERR_TRY_AGAIN) {
-		log_group(ls, "unlink ckpt retry");
-		sleep(1);
-		goto unlink_retry;
-	}
-	if (rv == SA_AIS_OK)
-		goto out_close;
 
-	log_group(ls, "unlink ckpt error %d %s", rv, ls->name);
-	ret = -1;
-
- status_retry:
-	rv = saCkptCheckpointStatusGet(h, &s);
-	if (rv == SA_AIS_ERR_TRY_AGAIN) {
-		log_group(ls, "unlink ckpt status retry");
-		sleep(1);
-		goto status_retry;
-	}
-	if (rv != SA_AIS_OK) {
-		log_group(ls, "unlink ckpt status error %d %s", rv, ls->name);
-		goto out_close;
-	}
-
-	log_group(ls, "unlink ckpt status: size %llu, max sections %u, "
-		      "max section size %llu, section count %u, mem %u",
-		 (unsigned long long)s.checkpointCreationAttributes.checkpointSize,
-		 s.checkpointCreationAttributes.maxSections,
-		 (unsigned long long)s.checkpointCreationAttributes.maxSectionSize,
-		 s.numberOfSections, s.memoryUsed);
-
- out_close:
-	if (!h)
-		goto out;
-
-	rv = saCkptCheckpointClose(h);
-	if (rv == SA_AIS_ERR_TRY_AGAIN) {
-		log_group(ls, "unlink ckpt close retry");
-		sleep(1);
-		goto out_close;
-	}
-	if (rv != SA_AIS_OK) {
-		log_error("unlink ckpt %llx close err %d %s",
-			  (unsigned long long)h, rv, ls->name);
-		/* should we return an error here and possibly cause
-		   store_plocks() to fail on this? */
-		/* ret = -1; */
-	}
- out:
-	ls->plock_ckpt_handle = 0;
-	return ret;
-}
-
-void close_plock_checkpoint(struct lockspace *ls)
-{
-	SaCkptCheckpointHandleT h;
-	SaAisErrorT rv;
-
-	h = (SaCkptCheckpointHandleT) ls->plock_ckpt_handle;
-	if (!h)
-		return;
- retry:
-	rv = saCkptCheckpointClose(h);
-	if (rv == SA_AIS_ERR_TRY_AGAIN) {
-		log_group(ls, "close_plock_checkpoint retry");
-		sleep(1);
-		goto retry;
-	}
-	if (rv != SA_AIS_OK) {
-		log_error("close_plock_checkpoint %llx err %d %s",
-			  (unsigned long long)h, rv, ls->name);
-	}
-
-	ls->plock_ckpt_handle = 0;
+ full:
+	rd->lock_count = cpu_to_le32(count);
+	*count_out = count;
+	return 1;
 }
 
-/*
- * section id is r<inodenum>.<owner>, the maximum string length is:
- * "r" prefix       =  1    strlen("r")
- * max uint64       = 20    strlen("18446744073709551615")
- * "." before owner =  1    strlen(".")
- * max int          = 11    strlen("-2147483647")
- * \0 at end        =  1
- * ---------------------
- *                    34    SECTION_NAME_LEN
- */
-
-#define SECTION_NAME_LEN 34
-
 /* Copy all plock state into a checkpoint so new node can retrieve it.  The
    node creating the ckpt for the mounter needs to be the same node that's
    sending the mounter its journals message (i.e. the low nodeid).  The new
@@ -1956,104 +1794,30 @@ void close_plock_checkpoint(struct lockspace *ls)
    it.  The ckpt should then disappear and the new node can create a new ckpt
    for the next mounter. */
 
-void store_plocks(struct lockspace *ls, uint32_t *sig)
+static int send_plocks_data(struct lockspace *ls, uint32_t seq, char *buf, int len)
 {
-	SaCkptCheckpointCreationAttributesT attr;
-	SaCkptCheckpointHandleT h;
-	SaCkptSectionIdT section_id;
-	SaCkptSectionCreationAttributesT section_attr;
-	SaCkptCheckpointOpenFlagsT flags;
-	SaNameT name;
-	SaAisErrorT rv;
-	char buf[SECTION_NAME_LEN];
-	struct resource *r;
-	struct posix_lock *po;
-	struct lock_waiter *w;
-	int total_size, section_size, max_section_size;
-	int len, owner;
-	uint32_t r_count = 0, p_count = 0;
-	uint64_t r_num_first = 0, r_num_last = 0;
-
-	if (!cfgd_enable_plock || ls->disable_plock)
-		return;
-
-	/* no change to plock state since we created the last checkpoint */
-	if (ls->last_checkpoint_time > ls->last_plock_time) {
-		log_group(ls, "store_plocks saved ckpt uptodate");
-		r_num_first = ls->checkpoint_r_num_first;
-		r_num_last = ls->checkpoint_r_num_last;
-		r_count = ls->checkpoint_r_count;
-		p_count = ls->checkpoint_p_count;
-		goto out;
-	}
-	ls->last_checkpoint_time = time(NULL);
-
-	len = snprintf((char *)name.value, SA_MAX_NAME_LENGTH, "dlmplock.%s",
-		       ls->name);
-	name.length = len;
+	struct dlm_header *hd;
 
-	_unlink_checkpoint(ls, &name);
+	hd = (struct dlm_header *)buf;
+	hd->type = DLM_MSG_PLOCKS_DATA;
+	hd->msgdata = seq;
 
-	/* loop through all plocks to figure out sizes to set in
-	   the attr fields */
+	dlm_send_message(ls, buf, len);
 
-	r_count = 0;
-	p_count = 0;
-	total_size = 0;
-	max_section_size = 0;
+	return 0;
+}
 
-	list_for_each_entry(r, &ls->plock_resources, list) {
-		if (r->owner == -1)
-			continue;
+void send_all_plocks_data(struct lockspace *ls, uint32_t seq, uint32_t *plocks_data)
+{
+	struct resource *r;
+	void *last;
+	int owner, count, len, full;
+	uint32_t send_count = 0;
 
-		r_count++;
-		section_size = 0;
-		list_for_each_entry(po, &r->locks, list) {
-			section_size += sizeof(struct pack_plock);
-			p_count++;
-		}
-		list_for_each_entry(w, &r->waiters, list) {
-			section_size += sizeof(struct pack_plock);
-			p_count++;
-		}
-		total_size += section_size;
-		if (section_size > max_section_size)
-			max_section_size = section_size;
-	}
-
-	log_group(ls, "store_plocks r_count %u p_count %u "
-		  "total_size %d max_section_size %d",
-		  r_count, p_count, total_size, max_section_size);
-	log_plock(ls, "store_plocks r_count %u p_count %u "
-		  "total_size %d max_section_size %d",
-		  r_count, p_count, total_size, max_section_size);
-
-	attr.creationFlags = SA_CKPT_WR_ALL_REPLICAS;
-	attr.checkpointSize = total_size;
-	attr.retentionDuration = SA_TIME_MAX;
-	attr.maxSections = r_count + 1;      /* don't know why we need +1 */
-	attr.maxSectionSize = max_section_size;
-	attr.maxSectionIdSize = SECTION_NAME_LEN;
-
-	flags = SA_CKPT_CHECKPOINT_READ |
-		SA_CKPT_CHECKPOINT_WRITE |
-		SA_CKPT_CHECKPOINT_CREATE;
-
- open_retry:
-	rv = saCkptCheckpointOpen(system_ckpt_handle, &name,&attr,flags,0,&h);
-	if (rv == SA_AIS_ERR_TRY_AGAIN) {
-		log_group(ls, "store_plocks ckpt open retry");
-		sleep(1);
-		goto open_retry;
-	}
-	if (rv != SA_AIS_OK) {
-		log_error("store_plocks ckpt open error %d %s", rv, ls->name);
-		goto fail;
-	}
+	if (!cfgd_enable_plock || ls->disable_plock)
+		return;
 
-	log_group(ls, "store_plocks open ckpt handle %llx",
-		  (unsigned long long)h);
-	ls->plock_ckpt_handle = (uint64_t) h;
+	log_dlock(ls, "send_all_plocks_data %d:%u", our_nodeid, seq);
 
 	/* - If r owner is -1, ckpt nothing.
 	   - If r owner is us, ckpt owner of us and no plocks.
@@ -2079,215 +1843,214 @@ void store_plocks(struct lockspace *ls, uint32_t *sig)
 		else if (!r->owner)
 			owner = 0;
 		else {
-			log_plock_error(ls, "store_plocks error owner %d r %llx",
-					r->owner, (unsigned long long)r->number);
+			log_elock(ls, "send_all_plocks_data error owner %d r %llx",
+				  r->owner, (unsigned long long)r->number);
 			continue;
 		}
 
-		memset(&buf, 0, sizeof(buf));
-		len = snprintf(buf, SECTION_NAME_LEN, "r%llu.%d",
-			       (unsigned long long)r->number, owner);
+		memset(&send_buf, 0, sizeof(send_buf));
+		count = 0;
+		full = 0;
+		last = NULL;
 
-		section_id.id = (void *)buf;
-		section_id.idLen = len + 1;
-		section_attr.sectionId = &section_id;
-		section_attr.expirationTime = SA_TIME_END;
+		do {
+			full = pack_send_buf(ls, r, owner, full, &count, &last);
 
-		memset(&section_buf, 0, sizeof(section_buf));
-		section_len = 0;
+			len = sizeof(struct dlm_header) +
+			      sizeof(struct resource_data) +
+			      sizeof(struct plock_data) * count;
 
-		pack_section_buf(ls, r, owner);
+			log_plock(ls, "send_plocks_data %d:%u n %llu o %d locks %d len %d",
+				  our_nodeid, seq, (unsigned long long)r->number, r->owner,
+				  count, len);
 
-		if (!r_num_first)
-			r_num_first = r->number;
-		r_num_last = r->number;
+			send_plocks_data(ls, seq, send_buf, len);
 
-		log_plock(ls, "wr sect ro %d rf %x len %u \"%s\"",
-			  r->owner, r->flags, section_len, buf);
+			send_count++;
 
-	 create_retry:
-		rv = saCkptSectionCreate(h, &section_attr, &section_buf,
-					 section_len);
-		if (rv == SA_AIS_ERR_TRY_AGAIN) {
-			log_group(ls, "store_plocks ckpt create retry");
-			sleep(1);
-			goto create_retry;
-		}
-		if (rv != SA_AIS_OK) {
-			log_error("store_plocks ckpt section create err %d %s",
-				  rv, ls->name);
-			goto fail;
-		}
+		} while (full);
 	}
- out:
-	*sig = (0xFFFFFFFF & r_num_first) ^ (0xFFFFFFFF & r_num_last) ^ r_count;
-
-	log_group(ls, "store_plocks first %llu last %llu r_count %u "
-		  "p_count %u sig %x",
-		  (unsigned long long)r_num_first,
-		  (unsigned long long)r_num_last,
-		  r_count, p_count, *sig);
-	log_plock(ls, "store_plocks first %llu last %llu r_count %u "
-		  "p_count %u sig %x",
-		  (unsigned long long)r_num_first,
-		  (unsigned long long)r_num_last,
-		  r_count, p_count, *sig);
-
-	ls->checkpoint_r_num_first = r_num_first;
-	ls->checkpoint_r_num_last = r_num_last;
-	ls->checkpoint_r_count = r_count;
-	ls->checkpoint_p_count = p_count;
-	return;
 
- fail:
-	ls->disable_plock = 1;
-	/* force the node receiving plocks to fail sig check and disable
-	   plocks as well */
-	*sig = 0xF0000000;
+	*plocks_data = send_count;
+
+	log_dlock(ls, "send_all_plocks_data %d:%u %u done",
+		  our_nodeid, seq, send_count);
 }
 
-/* called by a node that's just been added to the group to get existing plock
-   state */
+static void free_r_lists(struct resource *r)
+{
+	struct posix_lock *po, *po2;
+	struct lock_waiter *w, *w2;
+
+	list_for_each_entry_safe(po, po2, &r->locks, list) {
+		list_del(&po->list);
+		free(po);
+	}
+
+	list_for_each_entry_safe(w, w2, &r->waiters, list) {
+		list_del(&w->list);
+		free(w);
+	}
+}
 
-void retrieve_plocks(struct lockspace *ls, uint32_t *sig)
+void receive_plocks_data(struct lockspace *ls, struct dlm_header *hd, int len)
 {
-	SaCkptCheckpointHandleT h;
-	SaCkptSectionIterationHandleT itr;
-	SaCkptSectionDescriptorT desc;
-	SaCkptIOVectorElementT iov;
-	SaNameT name;
-	SaAisErrorT rv;
-	char buf[SECTION_NAME_LEN];
-	int len, lock_count, error;
-	uint32_t r_count = 0, p_count = 0;
-	uint64_t r_num, r_num_first = 0, r_num_last = 0;
+	struct resource_data *rd;
+	struct plock_data *pp;
+	struct posix_lock *po;
+	struct lock_waiter *w;
+	struct resource *r;
+	uint64_t num;
+	uint32_t count;
+	uint32_t flags;
+	int owner;
+	int i;
 
 	if (!cfgd_enable_plock || ls->disable_plock)
 		return;
 
-	log_group(ls, "retrieve_plocks");
+	if (!ls->need_plocks)
+		return;
 
-	len = snprintf((char *)name.value, SA_MAX_NAME_LENGTH, "dlmplock.%s",
-		       ls->name);
-	name.length = len;
+	if (!ls->save_plocks)
+		return;
 
- open_retry:
-	rv = saCkptCheckpointOpen(system_ckpt_handle, &name, NULL,
-				  SA_CKPT_CHECKPOINT_READ, 0, &h);
-	if (rv == SA_AIS_ERR_TRY_AGAIN) {
-		log_group(ls, "retrieve_plocks ckpt open retry");
-		sleep(1);
-		goto open_retry;
-	}
-	if (rv != SA_AIS_OK) {
-		log_error("retrieve_plocks ckpt open error %d %s",
-			  rv, ls->name);
+	ls->recv_plocks_data_count++;
+
+	if (len < sizeof(struct dlm_header) + sizeof(struct resource_data)) {
+		log_elock(ls, "recv_plocks_data %d:%u bad len %d",
+			  hd->nodeid, hd->msgdata, len);
 		return;
 	}
 
- init_retry:
-	rv = saCkptSectionIterationInitialize(h, SA_CKPT_SECTIONS_ANY, 0, &itr);
-	if (rv == SA_AIS_ERR_TRY_AGAIN) {
-		log_group(ls, "retrieve_plocks ckpt iterinit retry");
-		sleep(1);
-		goto init_retry;
+	rd = (struct resource_data *)((char *)hd + sizeof(struct dlm_header));
+	num = le64_to_cpu(rd->number);
+	owner = le32_to_cpu(rd->owner);
+	count = le32_to_cpu(rd->lock_count);
+	flags = le32_to_cpu(rd->flags);
+
+	if (flags & RD_CONTINUE) {
+		r = search_resource(ls, num);
+		if (!r) {
+			log_elock(ls, "recv_plocks_data %d:%u n %llu not found",
+				  hd->nodeid, hd->msgdata, (unsigned long long)num);
+			return;
+		}
+		log_plock(ls, "recv_plocks_data %d:%u n %llu continue",
+			  hd->nodeid, hd->msgdata, (unsigned long long)num);
+		goto unpack;
 	}
-	if (rv != SA_AIS_OK) {
-		log_error("retrieve_plocks ckpt iterinit error %d %s",
-			  rv, ls->name);
-		goto out;
+
+	r = malloc(sizeof(struct resource));
+	if (!r) {
+		log_elock(ls, "recv_plocks_data %d:%u n %llu no mem",
+			  hd->nodeid, hd->msgdata, (unsigned long long)num);
+		return;
 	}
+	memset(r, 0, sizeof(struct resource));
+	INIT_LIST_HEAD(&r->locks);
+	INIT_LIST_HEAD(&r->waiters);
+	INIT_LIST_HEAD(&r->pending);
 
-	while (1) {
-	 next_retry:
-		rv = saCkptSectionIterationNext(itr, &desc);
-		if (rv == SA_AIS_ERR_NO_SECTIONS)
-			break;
-		if (rv == SA_AIS_ERR_TRY_AGAIN) {
-			log_group(ls, "retrieve_plocks ckpt iternext retry");
-			sleep(1);
-			goto next_retry;
-		}
-		if (rv != SA_AIS_OK) {
-			log_error("retrieve_plocks ckpt iternext error %d %s",
-				  rv, ls->name);
-			goto out_it;
+	if (!cfgd_plock_ownership) {
+		if (owner) {
+			log_elock(ls, "recv_plocks_data %d:%u n %llu bad owner %d",
+				  hd->nodeid, hd->msgdata, (unsigned long long)num,
+				  owner);
+			goto fail_free;
 		}
+	} else {
+		if (!owner)
+			r->flags |= R_GOT_UNOWN;
 
-		if (!desc.sectionId.idLen)
-			continue;
+		/* no locks should be included for owned resources */
 
-		iov.sectionId = desc.sectionId;
-		iov.dataBuffer = &section_buf;
-		iov.dataSize = desc.sectionSize;
-		iov.dataOffset = 0;
-
-		/* for debug print */
-		memset(&buf, 0, sizeof(buf));
-		snprintf(buf, SECTION_NAME_LEN, "%s", desc.sectionId.id);
-
-	 read_retry:
-		rv = saCkptCheckpointRead(h, &iov, 1, NULL);
-		if (rv == SA_AIS_ERR_TRY_AGAIN) {
-			log_group(ls, "retrieve_plocks ckpt read retry");
-			sleep(1);
-			goto read_retry;
-		}
-		if (rv != SA_AIS_OK) {
-			log_error("retrieve_plocks ckpt read error %d %s",
-				  rv, ls->name);
-			goto out_it;
+		if (owner && count) {
+			log_elock(ls, "recv_plocks_data %d:%u n %llu o %d bad count %u",
+				  (unsigned long long)num, owner, count);
+			goto fail_free;
 		}
+	}
 
-		/* we'll get empty (zero length) sections for resources with
-		   no locks, which exist in ownership mode; the resource
-		   name and owner come from the section id */
+	r->number = num;
+	r->owner = owner;
 
-		log_plock(ls, "rd sect len %llu \"%s\"",
-			  (unsigned long long)iov.readSize, buf);
-				
-		section_len = iov.readSize;
+ unpack:
+	if (len < sizeof(struct dlm_header) +
+		  sizeof(struct resource_data) +
+		  sizeof(struct plock_data) * count) {
+		log_elock(ls, "recv_plocks_data %d:%u count %u bad len %d",
+			  hd->nodeid, hd->msgdata, count, len);
+		goto fail_free;
+	}
 
-		if (section_len % sizeof(struct pack_plock)) {
-			log_error("retrieve_plocks bad section len %d %s",
-				  section_len, ls->name);
-			continue;
-		}
+	pp = (struct plock_data *)((char *)rd + sizeof(struct resource_data));
 
-		r_num = 0;
-		lock_count = 0;
+	for (i = 0; i < count; i++) {
+		if (!pp->waiter) {
+			po = malloc(sizeof(struct posix_lock));
+			if (!po)
+				goto fail_free;
+			po->start	= le64_to_cpu(pp->start);
+			po->end		= le64_to_cpu(pp->end);
+			po->owner	= le64_to_cpu(pp->owner);
+			po->pid		= le32_to_cpu(pp->pid);
+			po->nodeid	= le32_to_cpu(pp->nodeid);
+			po->ex		= pp->ex;
+			list_add_tail(&po->list, &r->locks);
+		} else {
+			w = malloc(sizeof(struct lock_waiter));
+			if (!w)
+				goto fail_free;
+			w->info.start	= le64_to_cpu(pp->start);
+			w->info.end	= le64_to_cpu(pp->end);
+			w->info.owner	= le64_to_cpu(pp->owner);
+			w->info.pid	= le32_to_cpu(pp->pid);
+			w->info.nodeid	= le32_to_cpu(pp->nodeid);
+			w->info.ex	= pp->ex;
+			list_add_tail(&w->list, &r->waiters);
+		}
+		pp++;
+	}
 
-		error = unpack_section_buf(ls, (char *)desc.sectionId.id,
-					   desc.sectionId.idLen, &r_num,
-					   &lock_count);
-		if (error < 0)
-			continue;
+	log_plock(ls, "recv_plocks_data %d:%u n %llu o %d locks %d len %d",
+		  hd->nodeid, hd->msgdata, (unsigned long long)r->number,
+		  r->owner, count, len);
 
-		r_count++;
-		p_count += lock_count;
+	if (!(flags & RD_CONTINUE)) {
+		list_add_tail(&r->list, &ls->plock_resources);
+		rb_insert_plock_resource(ls, r);
+	}
+	return;
 
-		if (!r_num_first)
-			r_num_first = r_num;
-		r_num_last = r_num;
+ fail_free:
+	if (!(flags & RD_CONTINUE)) {
+		free_r_lists(r);
+		free(r);
 	}
+	return;
+}
 
- out_it:
-	saCkptSectionIterationFinalize(itr);
- out:
-	saCkptCheckpointClose(h);
+void clear_plocks_data(struct lockspace *ls)
+{
+	struct resource *r, *r2;
+	uint32_t count = 0;
 
-	*sig = (0xFFFFFFFF & r_num_first) ^ (0xFFFFFFFF & r_num_last) ^ r_count;
+	if (!cfgd_enable_plock || ls->disable_plock)
+		return;
 
-	log_group(ls, "retrieve_plocks first %llu last %llu r_count %u "
-		  "p_count %u sig %x",
-		  (unsigned long long)r_num_first,
-		  (unsigned long long)r_num_last,
-		  r_count, p_count, *sig);
-	log_plock(ls, "retrieve_plocks first %llu last %llu r_count %u "
-		  "p_count %u sig %x",
-		  (unsigned long long)r_num_first,
-		  (unsigned long long)r_num_last,
-		  r_count, p_count, *sig);
+	list_for_each_entry_safe(r, r2, &ls->plock_resources, list) {
+		free_r_lists(r);
+		rb_del_plock_resource(ls, r);
+		list_del(&r->list);
+		free(r);
+		count++;
+	}
+
+	log_dlock(ls, "clear_plocks_data done %u recv_plocks_data_count %u",
+		  count, ls->recv_plocks_data_count);
+
+	ls->recv_plocks_data_count = 0;
 }
 
 /* Called when a node has failed, or we're unmounting.  For a node failure, we
@@ -2346,10 +2109,10 @@ void purge_plocks(struct lockspace *ls, int nodeid, int unmount)
 	if (purged)
 		ls->last_plock_time = time(NULL);
 
-	log_plock(ls, "purged %d plocks for %d", purged, nodeid);
+	log_dlock(ls, "purged %d plocks for %d", purged, nodeid);
 }
 
-int fill_plock_dump_buf(struct lockspace *ls)
+int copy_plock_state(struct lockspace *ls, char *buf, int *len_out)
 {
 	struct posix_lock *po;
 	struct lock_waiter *w;
@@ -2358,9 +2121,6 @@ int fill_plock_dump_buf(struct lockspace *ls)
 	int rv = 0;
 	int len = DLMC_DUMP_SIZE, pos = 0, ret;
 
-	memset(plock_dump_buf, 0, sizeof(plock_dump_buf));
-	plock_dump_len = 0;
-
 	gettimeofday(&now, NULL);
 
 	list_for_each_entry(r, &ls->plock_resources, list) {
@@ -2368,7 +2128,7 @@ int fill_plock_dump_buf(struct lockspace *ls)
 		if (list_empty(&r->locks) &&
 		    list_empty(&r->waiters) &&
 		    list_empty(&r->pending)) {
-			ret = snprintf(plock_dump_buf + pos, len - pos,
+			ret = snprintf(buf + pos, len - pos,
 			      "%llu rown %d unused_ms %llu\n",
 			      (unsigned long long)r->number, r->owner,
 			      (unsigned long long)time_diff_ms(&r->last_access,
@@ -2382,7 +2142,7 @@ int fill_plock_dump_buf(struct lockspace *ls)
 		}
 
 		list_for_each_entry(po, &r->locks, list) {
-			ret = snprintf(plock_dump_buf + pos, len - pos,
+			ret = snprintf(buf + pos, len - pos,
 			      "%llu %s %llu-%llu nodeid %d pid %u owner %llx rown %d\n",
 			      (unsigned long long)r->number,
 			      po->ex ? "WR" : "RD",
@@ -2399,7 +2159,7 @@ int fill_plock_dump_buf(struct lockspace *ls)
 		}
 
 		list_for_each_entry(w, &r->waiters, list) {
-			ret = snprintf(plock_dump_buf + pos, len - pos,
+			ret = snprintf(buf + pos, len - pos,
 			      "%llu %s %llu-%llu nodeid %d pid %u owner %llx rown %d WAITING\n",
 			      (unsigned long long)r->number,
 			      w->info.ex ? "WR" : "RD",
@@ -2416,7 +2176,7 @@ int fill_plock_dump_buf(struct lockspace *ls)
 		}
 
 		list_for_each_entry(w, &r->pending, list) {
-			ret = snprintf(plock_dump_buf + pos, len - pos,
+			ret = snprintf(buf + pos, len - pos,
 			      "%llu %s %llu-%llu nodeid %d pid %u owner %llx rown %d PENDING\n",
 			      (unsigned long long)r->number,
 			      w->info.ex ? "WR" : "RD",
@@ -2433,7 +2193,7 @@ int fill_plock_dump_buf(struct lockspace *ls)
 		}
 	}
  out:
-	plock_dump_len = pos;
+	*len_out = pos;
 	return rv;
 }
 


More information about the cluster-commits mailing list