cluster: RHEL6 - Cleanup and exit if we receive a SIGTERM
by Ryan McCabe
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: 88f822170144e8cb0420298a0f4994c78cc44e00
Parent: d5dc6b1364b754a695a1eaacaac328b73cc3e151
Author: Ryan McCabe <rmccabe(a)redhat.com>
AuthorDate: Mon Apr 23 16:46:33 2012 -0400
Committer: Ryan McCabe <rmccabe(a)redhat.com>
CommitterDate: Tue May 1 10:57:10 2012 -0400
Cleanup and exit if we receive a SIGTERM
Signed-off-by: Ryan McCabe <rmccabe(a)redhat.com>
Reviewed-by: Lon Hohberger <lhh(a)redhat.com>
Reviewed-by: Fabio M. Di Nitto <fdinitto(a)redhat.com>
---
rgmanager/src/daemons/cpglockd.c | 103 ++++++++++++++++++++++++-------------
1 files changed, 67 insertions(+), 36 deletions(-)
diff --git a/rgmanager/src/daemons/cpglockd.c b/rgmanager/src/daemons/cpglockd.c
index 2163f82..0cba6f3 100644
--- a/rgmanager/src/daemons/cpglockd.c
+++ b/rgmanager/src/daemons/cpglockd.c
@@ -1,5 +1,6 @@
#include <stdio.h>
#include <signal.h>
+#include <signals.h>
#include <sys/types.h>
#include <sys/time.h>
#include <sys/select.h>
@@ -71,7 +72,8 @@ static int local_lockid = 0;
static int message_count = 0;
static int joined = 0;
static int cluster_quorate = 0;
-static int shutdown_requested = 0;
+static int cman_shutdown_requested = 0;
+static int shutdown_pending = 0;
static cman_node_t cman_nodes[CPG_MEMBERS_MAX];
static int cman_node_count;
@@ -79,6 +81,12 @@ static int cman_node_count;
static cman_node_t old_cman_nodes[CPG_MEMBERS_MAX];
static int old_cman_node_count;
+static void
+flag_shutdown(int __attribute__ ((unused)) sig)
+{
+ shutdown_pending = 1;
+}
+
static int
cman_nodes_lost(cman_node_t *old_nodes,
size_t old_node_len,
@@ -211,7 +219,7 @@ cman_callback(cman_handle_t ch, void *privdata, int reason, int arg)
}
}
} else if (reason == CMAN_REASON_TRY_SHUTDOWN) {
- shutdown_requested = 1;
+ cman_shutdown_requested = 1;
}
}
@@ -220,7 +228,7 @@ wait_for_fencing_join(int nodeid)
{
if (node_has_fencing(nodeid)) {
cpgl_debug("Waiting for fence domain join operation to complete\n");
- while (!fence_domain_joined())
+ while (!shutdown_pending && !fence_domain_joined())
sleep(1);
cpgl_debug("Fence domain joined\n");
} else
@@ -230,7 +238,7 @@ wait_for_fencing_join(int nodeid)
static void
wait_for_quorum_formation(cman_handle_t ch) {
cpgl_debug("Waiting for quorum to form\n");
- while (!(cluster_quorate = cman_is_quorate(ch)))
+ while (!shutdown_pending && !(cluster_quorate = cman_is_quorate(ch)))
sleep(1);
cpgl_debug("Quorum formed\n");
}
@@ -243,7 +251,7 @@ cman_connect(cman_handle_t *ch)
*ch = cman_init(NULL);
if (!*ch) {
cpgl_debug("Waiting for CMAN to start\n");
- while (!(*ch = cman_init(NULL)))
+ while (!shutdown_pending && !(*ch = cman_init(NULL)))
sleep(1);
cpgl_debug("CMAN started\n");
}
@@ -407,7 +415,7 @@ send_lock_msg(struct cpg_lock_msg *m)
cpgl_debug("send_lock_msg() failed\n");
usleep(250000);
}
- } while (ret != CPG_OK);
+ } while (ret != CPG_OK && !shutdown_pending);
return 0;
}
@@ -1325,6 +1333,7 @@ main(int argc, char **argv)
int wait_for_fencing = 1;
int opt;
int ret;
+ int exit_status = 0;
struct cpg_lock_msg m;
struct client_node *client;
cman_handle_t cman_handle = NULL;
@@ -1354,29 +1363,43 @@ main(int argc, char **argv)
}
}
- signal(SIGPIPE, SIG_IGN);
+ if (!nofork)
+ daemon_init((char *) "cpglockd");
+
+ setup_signal(SIGPIPE, SIG_IGN);
+ setup_signal(SIGTERM, flag_shutdown);
cman_connect(&cman_handle);
if (cman_handle == NULL) {
fprintf(stderr, "Unable to connect to cman\n");
- return -1;
+ exit_status = -1;
+ goto out4;
}
+ if (shutdown_pending)
+ goto out3;
+
if (wait_for_quorum)
wait_for_quorum_formation(cman_handle);
+ if (shutdown_pending)
+ goto out3;
+
memset(&my_node, 0, sizeof(my_node));
cman_get_node(cman_handle, CMAN_NODEID_US, &my_node);
if (my_node.cn_nodeid == 0) {
fprintf(stderr, "Unable to get our cluster node ID\n");
- cman_finish(cman_handle);
- return -1;
+ exit_status = -1;
+ goto out3;
}
if (wait_for_fencing)
wait_for_fencing_join(my_node.cn_nodeid);
+ if (shutdown_pending)
+ goto out3;
+
memset(&cman_nodes, 0, sizeof(cman_nodes));
cman_node_count = 0;
@@ -1385,24 +1408,23 @@ main(int argc, char **argv)
&cman_node_count, cman_nodes);
if (ret < 0) {
fprintf(stderr, "Unable to get cman nodes list\n");
- cman_finish(cman_handle);
- return -1;
+ exit_status = -1;
+ goto out3;
}
cman_fd = cman_get_fd(cman_handle);
if (cman_fd < 0) {
fprintf(stderr, "Error: cman fd is %d\n", cman_fd);
- cman_finish(cman_handle);
- return -1;
+ exit_status = -1;
+ goto out3;
}
cman_start_notification(cman_handle, cman_callback);
if (cpg_init() < 0) {
fprintf(stderr, "Unable to join CPG group\n");
- cman_stop_notification(cman_handle);
- cman_finish(cman_handle);
- return -1;
+ exit_status = -1;
+ goto out2;
}
assert(my_node.cn_nodeid == my_node_id);
@@ -1411,21 +1433,17 @@ main(int argc, char **argv)
if (fd < 0) {
fprintf(stderr, "Error connecting to %s: %s\n",
CPG_LOCKD_SOCK, strerror(errno));
- cman_stop_notification(cman_handle);
- cman_finish(cman_handle);
- cpg_fin();
- return -1;
+ exit_status = -1;
+ goto out1;
}
- if (!nofork)
- daemon_init((char *) "cpglockd");
-
- cpg_local_get(cpg, &my_node_id);
cpg_fd_get(cpg, &cpgfd);
- if (send_join() < 0)
- return -1;
+ if (cpgfd < 0 || send_join() < 0) {
+ exit_status = -1;
+ goto out;
+ }
- while (1) {
+ while (!shutdown_pending) {
struct timeval tv;
struct pending_fence_node *pf_node;
@@ -1448,22 +1466,24 @@ main(int argc, char **argv)
n = select_retry(x+1, &rfds, NULL, NULL, &tv);
if (n < 0) {
fprintf(stderr, "Error: select: %s\n", strerror(errno));
- return -1;
+ exit_status = -1;
+ goto out;
}
if (FD_ISSET(cman_fd, &rfds)) {
if (cman_dispatch(cman_handle, CMAN_DISPATCH_ALL) < 0) {
fprintf(stderr, "Fatal: cman_dispatch() failed: %s\n",
strerror(errno));
- return -1;
+ exit_status = -1;
+ goto out;
}
--n;
}
- if (shutdown_requested) {
- cman_replyto_shutdown(cman_handle, 1);
+ if (cman_shutdown_requested) {
fprintf(stderr, "Fatal: cman requested shutdown.\n");
- return -1;
+ cman_replyto_shutdown(cman_handle, 1);
+ goto out;
}
fence_check:
@@ -1526,6 +1546,9 @@ main(int argc, char **argv)
}
}
+ if (shutdown_pending)
+ goto out;
+
/* While fencing is pending for any nodes pause lock activity. */
if (pending_fencing != NULL) {
usleep(500000);
@@ -1590,9 +1613,17 @@ main(int argc, char **argv)
} while (n);
}
+out:
+ close(fd);
+ unlink(CPG_LOCKD_SOCK);
+out1:
+ cpg_fin();
+out2:
cman_stop_notification(cman_handle);
+out3:
cman_finish(cman_handle);
- cpg_fin();
-
- return 0;
+out4:
+ if (!nofork)
+ daemon_cleanup();
+ return exit_status;
}
12 years, 1 month
cluster: RHEL6 - Add a simple init script for cpglockd. The rgmanager init script still needs to be updated to call this when RRP mode is enabled and CPG locks are not explicitly disabled.
by Ryan McCabe
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: a3efdc591c273acde2e790241b3200d79eb711f1
Parent: 88f822170144e8cb0420298a0f4994c78cc44e00
Author: Ryan McCabe <rmccabe(a)redhat.com>
AuthorDate: Mon Apr 23 18:25:40 2012 -0400
Committer: Ryan McCabe <rmccabe(a)redhat.com>
CommitterDate: Tue May 1 10:57:14 2012 -0400
Add a simple init script for cpglockd. The rgmanager init script still needs to be updated to call this when RRP mode is enabled and CPG locks are not explicitly disabled.
Signed-off-by: Ryan McCabe <rmccabe(a)redhat.com>
Reviewed-by: Lon Hohberger <lhh(a)redhat.com>
Reviewed-by: Fabio M. Di Nitto <fdinitto(a)redhat.com>
---
rgmanager/init.d/cpglockd.in | 142 ++++++++++++++++++++++++++++++++++++++++++
1 files changed, 142 insertions(+), 0 deletions(-)
diff --git a/rgmanager/init.d/cpglockd.in b/rgmanager/init.d/cpglockd.in
new file mode 100644
index 0000000..9fbb43c
--- /dev/null
+++ b/rgmanager/init.d/cpglockd.in
@@ -0,0 +1,142 @@
+#!/bin/bash
+#
+# chkconfig: - 99 01
+# description: Starts and stops the CPG lock daemon
+#
+#
+### BEGIN INIT INFO
+# Provides: cpglockd
+# Required-Start: cman
+# Required-Stop: cman
+# Default-Start:
+# Default-Stop:
+# Short-Description: Starts and stops the CPG lock daemon
+# Description: Starts and stops the CPG lock daemon
+### END INIT INFO
+
+ID="CPG lock daemon"
+CPGLOCKD="cpglockd"
+
+# set secure PATH
+PATH="/sbin:/bin:/usr/sbin:/usr/bin:@SBINDIR@"
+
+success()
+{
+ echo -ne "[ OK ]\r"
+}
+
+failure()
+{
+ echo -ne "[FAILED]\r"
+}
+
+status()
+{
+ pid=$(pidof $1 2>/dev/null)
+ statusrtrn=$?
+ if [ $statusrtrn -ne 0 ]; then
+ echo "$1 is stopped"
+ else
+ echo "$1 (pid $pid) is running..."
+ fi
+ return $statusrtrn
+}
+
+# rpm based distros
+if [ -d /etc/sysconfig ]; then
+ [ -f @INITDDIR@/functions ] && . @INITDDIR@/functions
+ [ -f /etc/sysconfig/cluster ] && . /etc/sysconfig/cluster
+ [ -f /etc/sysconfig/rgmanager ] && . /etc/sysconfig/rgmanager
+ [ -f /etc/sysconfig/cpglockd ] && . /etc/sysconfig/cpglockd
+ [ -z "$LOCK_FILE" ] && LOCK_FILE="/var/lock/subsys/cpglockd"
+fi
+
+# deb based distros
+if [ -d /etc/default ]; then
+ [ -f /etc/default/cluster ] && . /etc/default/cluster
+ [ -f /etc/default/rgmanager ] && . /etc/default/rgmanager
+ [ -f /etc/default/cpglockd ] && . /etc/default/cpglockd
+ [ -z "$LOCK_FILE" ] && LOCK_FILE="/var/lock/cpglockd"
+fi
+
+#
+# Stop cpglockd
+#
+stop_cpglockd()
+{
+ kill -TERM $(pidof $CPGLOCKD)
+}
+
+rtrn=0
+
+if [ "$EUID" != "0" ]; then
+ echo "Only root can execute $0 script"
+ exit 4
+fi
+
+case "$1" in
+start)
+ echo -n "Starting $ID: "
+
+ # most recent distributions use tmpfs for /var/run
+ # to avoid to clean it up on every boot.
+ # they also assume that init scripts will create
+ # required subdirectories for proper operations
+ mkdir -p /var/run/cluster
+
+ if status $CPGLOCKD > /dev/null 2>&1; then
+ success
+ else
+ if $CPGLOCKD $CPGLOCKD_OPTS; then
+ touch $LOCK_FILE
+ success
+ else
+ failure
+ rtrn=1
+ fi
+ fi
+ echo
+;;
+restart)
+ $0 stop
+ $0 start
+;;
+condrestart|try-restart)
+ if status $CPGLOCKD > /dev/null 2>&1; then
+ $0 stop
+ $0 start
+ rtrn=$?
+ fi
+;;
+reload|force-reload)
+ # not required anymore
+ # return not implemented
+ rtrn=3
+;;
+status)
+ status $CPGLOCKD
+ rtrn=$?
+;;
+stop)
+ echo -n "Stopping $ID: "
+
+ if status $CPGLOCKD > /dev/null 2>&1; then
+ if stop_cpglockd; then
+ success
+ else
+ failure
+ rtrn=1
+ fi
+ else
+ success
+ fi
+ echo
+ rm -f $LOCK_FILE
+;;
+*)
+ echo "usage: $0 {start|stop|restart|condrestart|try-restart|reload|force-reload|status}"
+ rtrn=2
+;;
+esac
+
+exit $rtrn
12 years, 1 month
cluster: RHEL6 - Allow -C to take a 1 or 0 argument to respectively force enable and force disable of CPG locks.
by Ryan McCabe
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: d5dc6b1364b754a695a1eaacaac328b73cc3e151
Parent: f4bc657f47270e15f2156cc28a32effc3f1c9af3
Author: Ryan McCabe <rmccabe(a)redhat.com>
AuthorDate: Mon Apr 23 14:48:35 2012 -0400
Committer: Ryan McCabe <rmccabe(a)redhat.com>
CommitterDate: Tue May 1 10:57:06 2012 -0400
Allow -C to take a 1 or 0 argument to respectively force enable and force disable of CPG locks.
Signed-off-by: Ryan McCabe <rmccabe(a)redhat.com>
Reviewed-by: Lon Hohberger <lhh(a)redhat.com>
Reviewed-by: Fabio M. Di Nitto <fdinitto(a)redhat.com>
---
rgmanager/src/daemons/main.c | 27 +++++++++++++++++++++++----
1 files changed, 23 insertions(+), 4 deletions(-)
diff --git a/rgmanager/src/daemons/main.c b/rgmanager/src/daemons/main.c
index 87ea6e7..3dc0e8a 100644
--- a/rgmanager/src/daemons/main.c
+++ b/rgmanager/src/daemons/main.c
@@ -1053,14 +1053,15 @@ int
main(int argc, char **argv)
{
int rv, do_init = 1;
- char foreground = 0, wd = 1, cpg_locks = 0;
+ int cpg_locks = 0;
+ char foreground = 0, wd = 1, cpg_lock_opt = 0;
cman_node_t me;
msgctx_t *cluster_ctx;
msgctx_t *local_ctx;
pthread_t th;
cman_handle_t clu = NULL;
- while ((rv = getopt(argc, argv, "wfdNqC")) != EOF) {
+ while ((rv = getopt(argc, argv, "wfdNqC::")) != EOF) {
switch (rv) {
case 'w':
wd = 0;
@@ -1078,7 +1079,22 @@ main(int argc, char **argv)
rgm_dbus_notify = 0;
break;
case 'C':
- cpg_locks = 1;
+ if (optarg) {
+ if (!strcmp(optarg, "0")) {
+ /* force disable */
+ cpg_lock_opt = -1;
+ } else if (!strcmp(optarg, "1")) {
+ /* force enable */
+ cpg_lock_opt = 1;
+ } else {
+ fprintf(stderr,
+ "Unknown argument for -C: argument must be 1 or 0\n");
+ return -1;
+ }
+ } else {
+ /* auto handling: enable when RRP mode is enabled */
+ cpg_lock_opt = 0;
+ }
break;
default:
return 1;
@@ -1127,8 +1143,11 @@ main(int argc, char **argv)
}
xmlInitParser();
- if (cpg_locks || cpglockd_needed() == 1)
+ if (cpg_lock_opt == 1 || (cpg_lock_opt != -1 && cpglockd_needed() == 1)) {
cpglockd_start();
+ cpg_locks = 1;
+ } else
+ cpg_locks = 0;
if (!cpg_locks) {
if (clu_lock_init(rgmanager_lsname) != 0) {
12 years, 1 month
cluster: RHEL6 - rgmanager: Sync cpglock bits with upstream
by Ryan McCabe
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: c4e7ff456bd1eeab2fc5c092188084d4ca354113
Parent: 35b7d83225dbc53fbe6ba106eb0a878cf1989c1e
Author: Ryan McCabe <rmccabe(a)redhat.com>
AuthorDate: Mon Apr 23 10:44:26 2012 -0400
Committer: Ryan McCabe <rmccabe(a)redhat.com>
CommitterDate: Tue May 1 10:56:47 2012 -0400
rgmanager: Sync cpglock bits with upstream
Signed-off-by: Ryan McCabe <rmccabe(a)redhat.com>
Reviewed-by: Lon Hohberger <lhh(a)redhat.com>
Reviewed-by: Fabio M. Di Nitto <fdinitto(a)redhat.com>
---
rgmanager/src/clulib/libcpglock.c | 4 +-
rgmanager/src/daemons/Makefile | 5 +-
rgmanager/src/daemons/cpglockd.c | 423 +++++++++++++++++++++++++++++++++++--
3 files changed, 413 insertions(+), 19 deletions(-)
diff --git a/rgmanager/src/clulib/libcpglock.c b/rgmanager/src/clulib/libcpglock.c
index 2db65a5..9fe5ead 100644
--- a/rgmanager/src/clulib/libcpglock.c
+++ b/rgmanager/src/clulib/libcpglock.c
@@ -250,8 +250,10 @@ cpg_lock_dump(FILE *fp)
memset(&l, 0, sizeof(l));
l.request = MSG_DUMP;
- if (write_retry(fd, &l, sizeof(l), 0) < 0)
+ if (write_retry(fd, &l, sizeof(l), 0) < 0) {
+ close(fd);
return -1;
+ }
while (read_retry(fd, &c, 1, 0) == 1)
fprintf(fp, "%c", c);
diff --git a/rgmanager/src/daemons/Makefile b/rgmanager/src/daemons/Makefile
index dbd75ea..61ae6e4 100644
--- a/rgmanager/src/daemons/Makefile
+++ b/rgmanager/src/daemons/Makefile
@@ -61,6 +61,7 @@ CMAN_LDFLAGS += -L${cmanlibdir} -lcman
LOGSYS_LDFLAGS += -L${logtlibdir} -llogthread
DLM_LDFLAGS += -L${dlmlibdir} -ldlm
CPG_LDFLAGS += -lcpg
+FENCED_LDFLAGS += -lfenced
XML2_LDFLAGS += `xml2-config --libs`
SLANG_LDFLAGS += -L${slanglibdir} -lslang
EXTRA_LDFLAGS += -lpthread
@@ -101,8 +102,8 @@ ${TARGET3}: ${TARGET1}
ln -sf ${TARGET1} ${TARGET3}
${TARGET4}: ${OBJS4} ${LDDEPS}
- $(CC) -o $@ $^ $(CPG_LDFLAGS) $(EXTRA_LDFLAGS) $(LDFLAGS)
-
+ $(CC) -o $@ $^ $(CPG_LDFLAGS) $(CMAN_LDFLAGS) $(CCS_LDFLAGS) \
+ $(FENCED_LDFLAGS) $(EXTRA_LDFLAGS) $(LDFLAGS)
check: rg_test
cd tests && ./runtests.sh
diff --git a/rgmanager/src/daemons/cpglockd.c b/rgmanager/src/daemons/cpglockd.c
index 80d05a1..2163f82 100644
--- a/rgmanager/src/daemons/cpglockd.c
+++ b/rgmanager/src/daemons/cpglockd.c
@@ -5,19 +5,25 @@
#include <sys/select.h>
#include <errno.h>
#include <unistd.h>
+#include <stdlib.h>
#include <malloc.h>
#include <string.h>
#include <time.h>
+#include <assert.h>
#include <sys/uio.h>
+
#include <corosync/cpg.h>
+#include <ccs.h>
+#include <libcman.h>
+#include <libfenced.h>
+
#include "daemon_init.h"
#include "sock.h"
#include "cpglock.h"
#include "cpglock-internal.h"
#include "list.h"
-
struct request_node {
list_head();
struct cpg_lock l;
@@ -28,6 +34,12 @@ struct lock_node {
struct cpg_lock l;
};
+struct pending_fence_node {
+ list_head();
+ int nodeid;
+ uint64_t fail_time;
+};
+
struct client_node {
list_head();
int fd;
@@ -48,6 +60,7 @@ struct msg_node {
/* Local vars */
static cpg_handle_t cpg;
static uint32_t my_node_id = 0;
+static struct pending_fence_node *pending_fencing = NULL;
static struct request_node *requests = NULL;
static struct lock_node *locks = NULL;
static struct client_node *clients = NULL;
@@ -57,7 +70,186 @@ static int total_members = 0;
static int local_lockid = 0;
static int message_count = 0;
static int joined = 0;
+static int cluster_quorate = 0;
+static int shutdown_requested = 0;
+
+static cman_node_t cman_nodes[CPG_MEMBERS_MAX];
+static int cman_node_count;
+
+static cman_node_t old_cman_nodes[CPG_MEMBERS_MAX];
+static int old_cman_node_count;
+
+static int
+cman_nodes_lost(cman_node_t *old_nodes,
+ size_t old_node_len,
+ cman_node_t *new_nodes,
+ size_t new_node_len,
+ cman_node_t *lost_nodes,
+ size_t lost_nodes_size)
+{
+ int i;
+ size_t lost_nodes_len = 0;
+
+ if (lost_nodes_size < old_node_len) {
+ cpgl_debug("Lost nodes array smaller than old nodes array");
+ return -1;
+ }
+
+ for (i = 0 ; i < old_node_len ; i++) {
+ if (old_nodes[i].cn_member) {
+ int present_in_new_list = 0;
+ int j;
+
+ for (j = 0 ; j < new_node_len ; j++) {
+ if (new_nodes[j].cn_nodeid == old_nodes[i].cn_nodeid) {
+ present_in_new_list = 1;
+ if (!new_nodes[j].cn_member) {
+ memcpy(&lost_nodes[lost_nodes_len++],
+ &old_nodes[i], sizeof(*lost_nodes));
+ break;
+ }
+ }
+ }
+
+ if (!present_in_new_list) {
+ memcpy(&lost_nodes[lost_nodes_len++],
+ &old_nodes[i], sizeof(*lost_nodes));
+ }
+ }
+ }
+
+ return lost_nodes_len;
+}
+
+static int
+node_has_fencing(int nodeid)
+{
+ int ccs_desc;
+ char *val = NULL;
+ char buf[1024];
+ int ret = 1;
+
+ ccs_desc = ccs_connect();
+ if (ccs_desc < 0) {
+ cpgl_debug("Unable to connect to ccsd\n");
+ /* Assume node has fencing */
+ return 1;
+ }
+
+ snprintf(buf, sizeof(buf),
+ "/cluster/clusternodes/clusternode[@nodeid=\"%d\"]"
+ "/fence/method/device/@name", nodeid);
+
+ if (ccs_get(ccs_desc, buf, &val) != 0)
+ ret = 0;
+ if (val)
+ free(val);
+ ccs_disconnect(ccs_desc);
+ return ret;
+}
+
+static int
+fence_domain_joined(void) {
+ int ret;
+ struct fenced_node fn;
+
+ ret = fenced_node_info(FENCED_NODEID_US, &fn);
+ if (ret < 0) {
+ cpgl_debug("Unable to determine fence domain membership\n");
+ return 0;
+ }
+
+ return fn.member;
+}
+
+static void
+cman_callback(cman_handle_t ch, void *privdata, int reason, int arg)
+{
+ if (reason == CMAN_REASON_STATECHANGE) {
+ int i;
+ int ret;
+ int nl;
+ time_t cur_time = time(NULL);
+ size_t nodes_elem = sizeof(cman_nodes) / sizeof(cman_nodes[0]);
+ cman_node_t lost_nodes[nodes_elem];
+
+ cluster_quorate = arg;
+
+ memcpy(&old_cman_nodes, &cman_nodes, sizeof(old_cman_nodes));
+ old_cman_node_count = cman_node_count;
+
+ memset(&cman_nodes, 0, sizeof(cman_nodes));
+ cman_node_count = 0;
+
+ ret = cman_get_nodes(ch, nodes_elem, &cman_node_count, cman_nodes);
+ if (ret < 0) {
+ cpgl_debug("Unable to get cman nodes list\n");
+ return;
+ }
+
+ nl = cman_nodes_lost(old_cman_nodes, old_cman_node_count,
+ cman_nodes, cman_node_count, lost_nodes, nodes_elem);
+ if (nl < 0) {
+ cpgl_debug("Unable to get list of lost nodes");
+ return;
+ }
+
+ for (i = 0 ; i < nl ; i++) {
+ int cur_nodeid = lost_nodes[i].cn_nodeid;
+
+ if (node_has_fencing(cur_nodeid)) {
+ struct pending_fence_node *pf;
+
+ cpgl_debug("Lost node %d\n", cur_nodeid);
+ pf = do_alloc(sizeof(*pf));
+ pf->nodeid = cur_nodeid;
+ pf->fail_time = cur_time;
+ list_append(&pending_fencing, pf);
+ } else {
+ cpgl_debug("Lost node %d but fencing not configured\n",
+ cur_nodeid);
+ }
+ }
+ } else if (reason == CMAN_REASON_TRY_SHUTDOWN) {
+ shutdown_requested = 1;
+ }
+}
+
+static void
+wait_for_fencing_join(int nodeid)
+{
+ if (node_has_fencing(nodeid)) {
+ cpgl_debug("Waiting for fence domain join operation to complete\n");
+ while (!fence_domain_joined())
+ sleep(1);
+ cpgl_debug("Fence domain joined\n");
+ } else
+ cpgl_debug("No fencing is configured, not waiting\n");
+}
+
+static void
+wait_for_quorum_formation(cman_handle_t ch) {
+ cpgl_debug("Waiting for quorum to form\n");
+ while (!(cluster_quorate = cman_is_quorate(ch)))
+ sleep(1);
+ cpgl_debug("Quorum formed\n");
+}
+
+static int
+cman_connect(cman_handle_t *ch)
+{
+ assert(ch != NULL);
+
+ *ch = cman_init(NULL);
+ if (!*ch) {
+ cpgl_debug("Waiting for CMAN to start\n");
+ while (!(*ch = cman_init(NULL)))
+ sleep(1);
+ cpgl_debug("CMAN started\n");
+ }
+ return 0;
+}
static const char *
ls2str(int x)
@@ -209,9 +401,14 @@ send_lock_msg(struct cpg_lock_msg *m)
iov.iov_base = m;
iov.iov_len = sizeof (*m);
- ret = cpg_mcast_joined(cpg, CPG_TYPE_AGREED, &iov, 1);
- if (ret != CPG_OK)
- return -1;
+ do {
+ ret = cpg_mcast_joined(cpg, CPG_TYPE_AGREED, &iov, 1);
+ if (ret != CPG_OK) {
+ cpgl_debug("send_lock_msg() failed\n");
+ usleep(250000);
+ }
+ } while (ret != CPG_OK);
+
return 0;
}
@@ -397,8 +594,23 @@ del_node(uint32_t nodeid)
struct lock_node *l;
int x, recovered = 0, granted = 0;
- if (group_members->nodeid != my_node_id)
+ if (group_members->nodeid != my_node_id) {
+ /*
+ ** Update the owner and pid of any locks owned by the deleted
+ ** node to those of the oldest node in the group.
+ */
+ list_for(&locks, l, x) {
+ if (l->l.owner_nodeid == nodeid) {
+ cpgl_debug("RECOVERY: LOCK UPDATED: %s [%d:%d]=>[%d:%d]\n",
+ l->l.resource,
+ l->l.owner_nodeid, l->l.owner_pid,
+ group_members->nodeid, group_members->pid);
+ l->l.owner_nodeid = group_members->nodeid;
+ l->l.owner_pid = group_members->pid;
+ }
+ }
return;
+ }
cpgl_debug("RECOVERY: I am oldest node in the group, recovering locks\n");
@@ -415,19 +627,34 @@ del_node(uint32_t nodeid)
send_lock_msg(&m);
list_for(&locks, l, x) {
- if (l->l.owner_nodeid == nodeid &&
- l->l.state == LOCK_HELD) {
+ if (l->l.owner_nodeid == nodeid && l->l.state == LOCK_HELD) {
cpgl_debug("RECOVERY: Releasing %s held by dead node %d\n",
l->l.resource, nodeid);
l->l.state = LOCK_FREE;
strncpy(m.resource, l->l.resource, sizeof(m.resource));
- if (grant_next(&m) == 0)
+ if (grant_next(&m) == 0) {
+ cpgl_debug("RECOVERY: HELD LOCK UPDATED: %s [%d:%d]=>[%d:%d]\n",
+ l->l.resource,
+ l->l.owner_nodeid, l->l.owner_pid,
+ my_node_id, getpid());
+ l->l.owner_nodeid = my_node_id;
+ l->l.owner_pid = getpid();
+ m.owner_nodeid = my_node_id;
+ m.owner_pid = getpid();
send_unlock(&m);
+ }
++recovered;
- } else if (l->l.state == LOCK_FREE) {
- if (grant_next(&m) == 0)
- send_unlock(&m);
+ } else if (l->l.owner_nodeid == nodeid && l->l.state == LOCK_FREE) {
+ strncpy(m.resource, l->l.resource, sizeof(m.resource));
+ if (grant_next(&m) == 0) {
+ cpgl_debug("RECOVERY: FREE LOCK UPDATED: %s [%d:%d]=>[%d:%d]\n",
+ l->l.resource,
+ l->l.owner_nodeid, l->l.owner_pid,
+ my_node_id, getpid());
+ l->l.owner_nodeid = my_node_id;
+ l->l.owner_pid = getpid();
+ }
++granted;
}
}
@@ -968,6 +1195,10 @@ cpg_config_change(cpg_handle_t h,
list_for(&group_members, n, y) {
if (n->nodeid == left[x].nodeid) {
if (n->pid == left[x].pid) {
+ if (n->nodeid == my_node_id) {
+ cpgl_debug("Received DELETE message for self. Exiting.");
+ exit(0);
+ }
list_remove(&group_members, n);
cpgl_debug("DELETE: node %d.%u\n", n->nodeid, n->pid);
del_node(n->nodeid);
@@ -1087,37 +1318,101 @@ main(int argc, char **argv)
int fd;
int cpgfd;
int afd = -1;
+ int cman_fd;
int n,x;
int nofork = 0;
+ int wait_for_quorum = 1;
+ int wait_for_fencing = 1;
int opt;
+ int ret;
struct cpg_lock_msg m;
struct client_node *client;
+ cman_handle_t cman_handle = NULL;
+ cman_node_t my_node;
- while ((opt = getopt(argc, argv, "fh")) != EOF) {
+ while ((opt = getopt(argc, argv, "FQfh")) != EOF) {
switch (opt) {
+ case 'F':
+ wait_for_fencing = 0;
+ break;
+ case 'Q':
+ wait_for_quorum = 0;
+ break;
case 'f':
nofork = 1;
break;
case 'h':
printf("Usage: %s [options]\n\
+ -F Don't wait for the current node to join the fencing domain at startup\n\
+ -Q Don't wait for quorum formation at startup\n\
-f Don't daemonize\n\
-h Print this help message\n",
argv[0]);
return 0;
+ default:
+ return -1;
}
}
signal(SIGPIPE, SIG_IGN);
+ cman_connect(&cman_handle);
+ if (cman_handle == NULL) {
+ fprintf(stderr, "Unable to connect to cman\n");
+ return -1;
+ }
+
+ if (wait_for_quorum)
+ wait_for_quorum_formation(cman_handle);
+
+ memset(&my_node, 0, sizeof(my_node));
+ cman_get_node(cman_handle, CMAN_NODEID_US, &my_node);
+
+ if (my_node.cn_nodeid == 0) {
+ fprintf(stderr, "Unable to get our cluster node ID\n");
+ cman_finish(cman_handle);
+ return -1;
+ }
+
+ if (wait_for_fencing)
+ wait_for_fencing_join(my_node.cn_nodeid);
+
+ memset(&cman_nodes, 0, sizeof(cman_nodes));
+ cman_node_count = 0;
+
+ ret = cman_get_nodes(cman_handle,
+ sizeof(cman_nodes) / sizeof(cman_nodes[0]),
+ &cman_node_count, cman_nodes);
+ if (ret < 0) {
+ fprintf(stderr, "Unable to get cman nodes list\n");
+ cman_finish(cman_handle);
+ return -1;
+ }
+
+ cman_fd = cman_get_fd(cman_handle);
+ if (cman_fd < 0) {
+ fprintf(stderr, "Error: cman fd is %d\n", cman_fd);
+ cman_finish(cman_handle);
+ return -1;
+ }
+
+ cman_start_notification(cman_handle, cman_callback);
+
if (cpg_init() < 0) {
fprintf(stderr, "Unable to join CPG group\n");
+ cman_stop_notification(cman_handle);
+ cman_finish(cman_handle);
return -1;
}
+ assert(my_node.cn_nodeid == my_node_id);
+
fd = sock_listen(CPG_LOCKD_SOCK);
if (fd < 0) {
fprintf(stderr, "Error connecting to %s: %s\n",
CPG_LOCKD_SOCK, strerror(errno));
+ cman_stop_notification(cman_handle);
+ cman_finish(cman_handle);
cpg_fin();
return -1;
}
@@ -1131,6 +1426,9 @@ main(int argc, char **argv)
return -1;
while (1) {
+ struct timeval tv;
+ struct pending_fence_node *pf_node;
+
FD_ZERO(&rfds);
x = client_fdset(&rfds);
FD_SET(fd, &rfds);
@@ -1139,13 +1437,101 @@ main(int argc, char **argv)
FD_SET(cpgfd, &rfds);
if (cpgfd > x)
x = cpgfd;
-
- n = select_retry(x+1, &rfds, NULL, NULL, NULL);
+ cman_fd = cman_get_fd(cman_handle);
+ FD_SET(cman_fd, &rfds);
+ if (cman_fd > x)
+ x = cman_fd;
+
+ tv.tv_sec = 0;
+ tv.tv_usec = 500000;
+
+ n = select_retry(x+1, &rfds, NULL, NULL, &tv);
if (n < 0) {
- perror("select");
+ fprintf(stderr, "Error: select: %s\n", strerror(errno));
return -1;
}
+ if (FD_ISSET(cman_fd, &rfds)) {
+ if (cman_dispatch(cman_handle, CMAN_DISPATCH_ALL) < 0) {
+ fprintf(stderr, "Fatal: cman_dispatch() failed: %s\n",
+ strerror(errno));
+ return -1;
+ }
+ --n;
+ }
+
+ if (shutdown_requested) {
+ cman_replyto_shutdown(cman_handle, 1);
+ fprintf(stderr, "Fatal: cman requested shutdown.\n");
+ return -1;
+ }
+
+ fence_check:
+ list_for(&pending_fencing, pf_node, x) {
+ int victim;
+ uint64_t lft;
+ struct fenced_node fn;
+
+ ret = fenced_node_info(pf_node->nodeid, &fn);
+ if (ret < 0) {
+ cpgl_debug("Unable to get last fenced time for node %d\n",
+ pf_node->nodeid);
+ victim = 1;
+ lft = 0;
+ } else {
+ victim = fn.victim;
+ lft = fn.last_fenced_time;
+ }
+
+ if (lft > pf_node->fail_time) {
+ cpgl_debug("Fencing for node %d finished at %ld (>%ld)\n",
+ pf_node->nodeid, lft, pf_node->fail_time);
+ list_remove(&pending_fencing, pf_node);
+ free(pf_node);
+ goto fence_check;
+ }
+
+ /*
+ ** If node A either fails or leaves the cluster cleanly while
+ ** fencing is outstanding on another node (or nodes), we aren't
+ ** able to determine whether A has failed or left cleanly until
+ ** fencing has completed for the nodes that have left or failed
+ ** prior to A while any fencing is outstanding. The first entry
+ ** in this list will be the node that has left or failed longer
+ ** ago than any others in the list. If fenced has not set
+ ** victim to 1 by now, we can deduce it has left cleanly, and we
+ ** don't need to wait for it.
+ */
+ if (!victim && !x) {
+ int retries = 0;
+ /* Wait up to 1s for fenced to set victim */
+ do {
+ usleep(250000);
+ if (fenced_node_info(pf_node->nodeid, &fn) < 0) {
+ cpgl_debug("Unable to get fenced data for node %d\n",
+ pf_node->nodeid);
+ } else
+ victim = fn.victim;
+ } while (!victim && retries++ < 4);
+
+ if (!victim) {
+ cpgl_debug("First entry in list and victim == 0, removing %d from pending fencing\n", pf_node->nodeid);
+ list_remove(&pending_fencing, pf_node);
+ free(pf_node);
+ } else {
+ cpgl_debug("fenced_node.victim flipped to 1 for %d\n",
+ pf_node->nodeid);
+ }
+ goto fence_check;
+ }
+ }
+
+ /* While fencing is pending for any nodes pause lock activity. */
+ if (pending_fencing != NULL) {
+ usleep(500000);
+ continue;
+ }
+
if (FD_ISSET(fd, &rfds)) {
afd = accept(fd, NULL, NULL);
insert_client(afd);
@@ -1153,7 +1539,10 @@ main(int argc, char **argv)
}
if (FD_ISSET(cpgfd, &rfds)) {
- cpg_dispatch(cpg, CPG_DISPATCH_ALL);
+ if (cpg_dispatch(cpg, CPG_DISPATCH_ALL) != CPG_OK) {
+ fprintf(stderr, "Fatal: Lost CPG connection.\n");
+ return -1;
+ }
--n;
}
@@ -1201,6 +1590,8 @@ main(int argc, char **argv)
} while (n);
}
+ cman_stop_notification(cman_handle);
+ cman_finish(cman_handle);
cpg_fin();
return 0;
12 years, 1 month
cluster: RHEL6 - rgmanager: Add a utility to dump the cpglockd state
by Ryan McCabe
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: f4bc657f47270e15f2156cc28a32effc3f1c9af3
Parent: 9a583d796f1f13b6b836eb140bdbda05d3c95a01
Author: Ryan McCabe <rmccabe(a)redhat.com>
AuthorDate: Mon Apr 23 10:55:58 2012 -0400
Committer: Ryan McCabe <rmccabe(a)redhat.com>
CommitterDate: Tue May 1 10:56:58 2012 -0400
rgmanager: Add a utility to dump the cpglockd state
Signed-off-by: Ryan McCabe <rmccabe(a)redhat.com>
Reviewed-by: Lon Hohberger <lhh(a)redhat.com>
Reviewed-by: Fabio M. Di Nitto <fdinitto(a)redhat.com>
---
rgmanager/src/daemons/Makefile | 10 ++++++++--
rgmanager/src/daemons/cpglockdump.c | 8 ++++++++
2 files changed, 16 insertions(+), 2 deletions(-)
diff --git a/rgmanager/src/daemons/Makefile b/rgmanager/src/daemons/Makefile
index 61ae6e4..336b6cf 100644
--- a/rgmanager/src/daemons/Makefile
+++ b/rgmanager/src/daemons/Makefile
@@ -2,11 +2,12 @@ TARGET1= rgmanager
TARGET2= rg_test
TARGET3= clurgmgrd
TARGET4= cpglockd
+TARGET5= cpglockdump
-SBINDIRT=$(TARGET1) $(TARGET2) $(TARGET4)
+SBINDIRT=$(TARGET1) $(TARGET2) $(TARGET4) $(TARGET5)
SBINSYMT=$(TARGET3)
-all: depends ${TARGET1} ${TARGET2} ${TARGET3} $(TARGET4)
+all: depends ${TARGET1} ${TARGET2} ${TARGET3} $(TARGET4) $(TARGET5)
include ../../../make/defines.mk
include $(OBJDIR)/make/cobj.mk
@@ -44,6 +45,8 @@ OBJS2= test-noccs.o \
OBJS4= cpglockd.o
+OBJS5= cpglockdump.o
+
CFLAGS += -DSHAREDIR=\"${sharedir}\" -D_GNU_SOURCE
CFLAGS += -fPIC
CFLAGS += -I${ccsincdir} -I${cmanincdir} -I${dlmincdir} -I${logtincdir}
@@ -105,6 +108,9 @@ ${TARGET4}: ${OBJS4} ${LDDEPS}
$(CC) -o $@ $^ $(CPG_LDFLAGS) $(CMAN_LDFLAGS) $(CCS_LDFLAGS) \
$(FENCED_LDFLAGS) $(EXTRA_LDFLAGS) $(LDFLAGS)
+${TARGET5}: ${OBJS5} ${LDDEPS}
+ $(CC) -o $@ $^ $(CPG_LDFLAGS)
+
check: rg_test
cd tests && ./runtests.sh
diff --git a/rgmanager/src/daemons/cpglockdump.c b/rgmanager/src/daemons/cpglockdump.c
new file mode 100644
index 0000000..2750e4a
--- /dev/null
+++ b/rgmanager/src/daemons/cpglockdump.c
@@ -0,0 +1,8 @@
+#include <stdio.h>
+#include <cpglock.h>
+
+int
+main(int argc, char **argv)
+{
+ return cpg_lock_dump(stdout);
+}
12 years, 1 month
cluster: RHEL6 - rgmanager: Start cpglockd automatically when RRP mode is enabled
by Ryan McCabe
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: 9a583d796f1f13b6b836eb140bdbda05d3c95a01
Parent: c4e7ff456bd1eeab2fc5c092188084d4ca354113
Author: Ryan McCabe <rmccabe(a)redhat.com>
AuthorDate: Mon Apr 23 10:49:29 2012 -0400
Committer: Ryan McCabe <rmccabe(a)redhat.com>
CommitterDate: Tue May 1 10:56:53 2012 -0400
rgmanager: Start cpglockd automatically when RRP mode is enabled
Signed-off-by: Ryan McCabe <rmccabe(a)redhat.com>
Reviewed-by: Lon Hohberger <lhh(a)redhat.com>
Reviewed-by: Fabio M. Di Nitto <fdinitto(a)redhat.com>
---
rgmanager/include/cpglock-internal.h | 4 ++
rgmanager/src/daemons/main.c | 75 +++++++++++++++++++++++++++++++++-
2 files changed, 78 insertions(+), 1 deletions(-)
diff --git a/rgmanager/include/cpglock-internal.h b/rgmanager/include/cpglock-internal.h
index d6734eb..27917c7 100644
--- a/rgmanager/include/cpglock-internal.h
+++ b/rgmanager/include/cpglock-internal.h
@@ -11,6 +11,10 @@
#define CPG_LOCKD_SOCK "/var/run/cpglockd.sk"
#endif
+#ifndef CPGLOCKD_BIN_PATH
+# define CPGLOCKD_BIN_PATH "/usr/sbin/cpglockd"
+#endif
+
#include <stdint.h>
typedef enum {
diff --git a/rgmanager/src/daemons/main.c b/rgmanager/src/daemons/main.c
index 29af894..87ea6e7 100644
--- a/rgmanager/src/daemons/main.c
+++ b/rgmanager/src/daemons/main.c
@@ -25,6 +25,8 @@
#include <groups.h>
#include <rg_dbus.h>
+#include <cpglock-internal.h>
+
#ifdef WRAP_THREADS
void dump_thread_states(FILE *);
#endif
@@ -792,6 +794,74 @@ statedump(int __attribute__ ((unused)) sig)
signalled++;
}
+static int
+cpglockd_needed(void)
+{
+ char *v;
+ int ccsfd;
+ int start_cpglockd = 0;
+
+ ccsfd = ccs_force_connect(NULL, 0);
+ if (ccsfd < 0)
+ return -1;
+
+ if (ccs_get(ccsfd, "/cluster/cman/altmulticast/@addr", &v) == 0) {
+ if (v != NULL) {
+ int ret;
+ free(v);
+
+ ret = ccs_get(ccsfd,
+ "/cluster/clusternodes/clusternode/altname/@name", &v);
+ if (ret == 0) {
+ if (v != NULL) {
+ free(v);
+ start_cpglockd = 1;
+ }
+ }
+ }
+ }
+
+ if (start_cpglockd) {
+ if (ccs_get(ccsfd, "/cluster/totem/@rrp_mode", &v) == 0) {
+ if (v != NULL) {
+ if (!strcasecmp(v, "none"))
+ start_cpglockd = 0;
+ free(v);
+ }
+ }
+ }
+
+ ccs_disconnect(ccsfd);
+ return start_cpglockd;
+}
+
+static int
+cpglockd_start(void) {
+ int pid;
+ int status;
+
+ pid = fork();
+ if (pid == -1)
+ return -1;
+
+ if (pid) {
+ waitpid(pid, &status, 0);
+ if (!WIFEXITED(status) || WEXITSTATUS(status)) {
+ logt_print(LOG_NOTICE, "Unable to start cpglockd\n");
+ return -1;
+ }
+ } else {
+ int i;
+
+ for (i = 3 ; i < __FD_SETSIZE ; i++)
+ close(i);
+
+ execl(CPGLOCKD_BIN_PATH, "cpglockd", NULL);
+ exit(-1);
+ }
+
+ return 0;
+}
static int
rgmanager_disabled(int ccsfd)
@@ -1056,6 +1126,10 @@ main(int argc, char **argv)
return -1;
}
+ xmlInitParser();
+ if (cpg_locks || cpglockd_needed() == 1)
+ cpglockd_start();
+
if (!cpg_locks) {
if (clu_lock_init(rgmanager_lsname) != 0) {
printf("Locks not working!\n");
@@ -1092,7 +1166,6 @@ main(int argc, char **argv)
We know we're quorate. At this point, we need to
read the resource group trees from ccsd.
*/
- xmlInitParser();
configure_rgmanager(-1, debug, &cluster_timeout);
if (shutdown_pending == 1)
goto out_ls;
12 years, 1 month
cluster: RHEL6 - rgmanager: Sync cpglock bits with upstream
by Ryan McCabe
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: 35b7d83225dbc53fbe6ba106eb0a878cf1989c1e
Parent: b1cc2fd5ce82d5c65ce979c4138aa9087aaf984c
Author: Ryan McCabe <rmccabe(a)redhat.com>
AuthorDate: Wed Mar 28 14:17:12 2012 -0400
Committer: Ryan McCabe <rmccabe(a)redhat.com>
CommitterDate: Tue May 1 10:56:39 2012 -0400
rgmanager: Sync cpglock bits with upstream
Signed-off-by: Ryan McCabe <rmccabe(a)redhat.com>
Reviewed-by: Lon Hohberger <lhh(a)redhat.com>
Reviewed-by: Fabio M. Di Nitto <fdinitto(a)redhat.com>
---
rgmanager/include/cpglock-internal.h | 6 +
rgmanager/src/clulib/libcpglock.c | 11 +-
rgmanager/src/daemons/cpglockd.c | 239 +++++++++++++++++++++-------------
3 files changed, 164 insertions(+), 92 deletions(-)
diff --git a/rgmanager/include/cpglock-internal.h b/rgmanager/include/cpglock-internal.h
index 1c322ae..d6734eb 100644
--- a/rgmanager/include/cpglock-internal.h
+++ b/rgmanager/include/cpglock-internal.h
@@ -1,6 +1,12 @@
#ifndef _CPGLOCK_INT_H
#define _CPGLOCK_INT_H
+#ifdef CPGLOCK_DEBUG
+# define cpgl_debug(format, args...) do { printf(format, ##args); } while (0)
+#else
+# define cpgl_debug(format, args...) do { } while (0)
+#endif
+
#ifndef CPG_LOCKD_SOCK
#define CPG_LOCKD_SOCK "/var/run/cpglockd.sk"
#endif
diff --git a/rgmanager/src/clulib/libcpglock.c b/rgmanager/src/clulib/libcpglock.c
index 8e5c7ff..2db65a5 100644
--- a/rgmanager/src/clulib/libcpglock.c
+++ b/rgmanager/src/clulib/libcpglock.c
@@ -97,7 +97,7 @@ cpg_lock(void *handle, const char *resource, lock_flag_t flags, struct cpg_lock
if (!h) {
errno = EINVAL;
- goto out;
+ return -1;
}
pthread_mutex_lock(&h->mutex);
@@ -145,7 +145,8 @@ cpg_lock(void *handle, const char *resource, lock_flag_t flags, struct cpg_lock
}
if (strcmp(r.resource, l.resource)) {
- printf("NOTE: msg for wrong lock want: %s got: %s\n", l.resource, r.resource);
+ cpgl_debug("NOTE: msg for wrong lock want: %s got: %s\n",
+ l.resource, r.resource);
add_pending(&h->pending, &r);
pthread_mutex_unlock(&h->mutex);
usleep(random() & 16383);
@@ -153,7 +154,8 @@ cpg_lock(void *handle, const char *resource, lock_flag_t flags, struct cpg_lock
continue;
}
if (r.owner_tid != l.owner_tid) {
- printf("NOTE: msg for wrong seq want: %d got: %d\n", l.owner_tid, r.owner_tid );
+ cpgl_debug("NOTE: msg for wrong seq want: %d got: %d\n",
+ l.owner_tid, r.owner_tid );
add_pending(&h->pending, &r);
pthread_mutex_unlock(&h->mutex);
usleep(random() & 16383);
@@ -169,7 +171,8 @@ cpg_lock(void *handle, const char *resource, lock_flag_t flags, struct cpg_lock
if (r.request == MSG_NAK) {
errno = EAGAIN;
- return -1;
+ ret = -1;
+ goto out;
}
if (r.request != MSG_GRANT) {
diff --git a/rgmanager/src/daemons/cpglockd.c b/rgmanager/src/daemons/cpglockd.c
index 3ed132b..80d05a1 100644
--- a/rgmanager/src/daemons/cpglockd.c
+++ b/rgmanager/src/daemons/cpglockd.c
@@ -11,6 +11,7 @@
#include <sys/uio.h>
#include <corosync/cpg.h>
+#include "daemon_init.h"
#include "sock.h"
#include "cpglock.h"
#include "cpglock-internal.h"
@@ -36,6 +37,7 @@ struct client_node {
struct member_node {
list_head();
int nodeid;
+ uint32_t pid;
};
struct msg_node {
@@ -104,7 +106,7 @@ dump_state(FILE *fp)
if (group_members) {
fprintf(fp, "Participants:");
list_for(&group_members, m, x) {
- fprintf(fp, " %d", m->nodeid);
+ fprintf(fp, " %d.%u", m->nodeid, m->pid);
}
fprintf(fp, "\n");
}
@@ -202,11 +204,15 @@ static int
send_lock_msg(struct cpg_lock_msg *m)
{
struct iovec iov;
+ int ret;
iov.iov_base = m;
iov.iov_len = sizeof (*m);
- return cpg_mcast_joined(cpg, CPG_TYPE_AGREED, &iov, 1);
+ ret = cpg_mcast_joined(cpg, CPG_TYPE_AGREED, &iov, 1);
+ if (ret != CPG_OK)
+ return -1;
+ return 0;
}
@@ -225,7 +231,7 @@ send_grant(struct request_node *n)
{
struct cpg_lock_msg m;
- printf("-> sending grant %s to %d:%d:%d\n",
+ cpgl_debug("-> sending grant %s to %d:%d:%d\n",
n->l.resource, n->l.owner_nodeid, n->l.owner_pid, n->l.owner_tid);
memset(&m, 0, sizeof(m));
@@ -284,7 +290,7 @@ grant_next(struct cpg_lock_msg *m)
/* Send grant */
if (r->l.state == LOCK_PENDING) {
- printf("LOCK %s: grant to %d:%d:%d\n", m->resource,
+ cpgl_debug("LOCK %s: grant to %d:%d:%d\n", m->resource,
r->l.owner_nodeid, r->l.owner_pid, r->l.owner_tid);
/* don't send dup grants */
r->l.state = LOCK_HELD;
@@ -321,9 +327,9 @@ purge_requests(uint32_t nodeid, uint32_t pid)
if (count) {
if (pid) {
- printf("RECOVERY: purged %d requests from %d:%d\n", count, nodeid, pid);
+ cpgl_debug("RECOVERY: purged %d requests from %d:%d\n", count, nodeid, pid);
} else {
- printf("RECOVERY: purged %d requests from node %d\n", count, nodeid);
+ cpgl_debug("RECOVERY: purged %d requests from node %d\n", count, nodeid);
}
}
}
@@ -351,11 +357,12 @@ del_client(int fd)
if (!pid)
return;
- printf("RECOVERY: Looking for locks held by PID %d\n", pid);
+ cpgl_debug("RECOVERY: Looking for locks held by PID %d\n", pid);
/* This may not be needed */
purge_requests(my_node_id, pid);
+ memset(&m, 0, sizeof(m));
m.request = MSG_PURGE;
m.owner_nodeid = my_node_id;
m.owner_pid = pid;
@@ -368,7 +375,7 @@ del_client(int fd)
l->l.state != LOCK_HELD)
continue;
- printf("RECOVERY: Releasing %s \n", l->l.resource);
+ cpgl_debug("RECOVERY: Releasing %s \n", l->l.resource);
l->l.state = LOCK_FREE;
strncpy(m.resource, l->l.resource, sizeof(m.resource));
@@ -377,9 +384,9 @@ del_client(int fd)
}
if (recovered) {
- printf("RECOVERY: %d locks from local PID %d\n", recovered, pid);
+ cpgl_debug("RECOVERY: %d locks from local PID %d\n", recovered, pid);
}
- printf("RECOVERY: Complete\n");
+ cpgl_debug("RECOVERY: Complete\n");
}
@@ -393,13 +400,14 @@ del_node(uint32_t nodeid)
if (group_members->nodeid != my_node_id)
return;
- printf("RECOVERY: I am oldest node in the group, recovering locks\n");
+ cpgl_debug("RECOVERY: I am oldest node in the group, recovering locks\n");
/* pass 1: purge outstanding requests from this node. */
/* This may not be needed */
purge_requests(nodeid, 0);
+ memset(&m, 0, sizeof(m));
m.request = MSG_PURGE;
m.owner_nodeid = nodeid;
m.owner_pid = 0;
@@ -409,8 +417,8 @@ del_node(uint32_t nodeid)
list_for(&locks, l, x) {
if (l->l.owner_nodeid == nodeid &&
l->l.state == LOCK_HELD) {
- printf("RECOVERY: Releasing %s held by dead node %d\n", l->l.resource,
- nodeid);
+ cpgl_debug("RECOVERY: Releasing %s held by dead node %d\n",
+ l->l.resource, nodeid);
l->l.state = LOCK_FREE;
strncpy(m.resource, l->l.resource, sizeof(m.resource));
@@ -425,13 +433,13 @@ del_node(uint32_t nodeid)
}
if (recovered) {
- printf("RECOVERY: %d locks from node %d\n", recovered, nodeid);
+ cpgl_debug("RECOVERY: %d locks from node %d\n", recovered, nodeid);
}
if (granted) {
- printf("RECOVERY: %d pending locks granted\n", granted);
+ cpgl_debug("RECOVERY: %d pending locks granted\n", granted);
}
- printf("RECOVERY: Complete\n");
+ cpgl_debug("RECOVERY: Complete\n");
}
@@ -493,7 +501,7 @@ grant_client(struct lock_node *l)
struct client_node *c;
struct cpg_lock_msg m;
- //memset(&m, 0, sizeof(m));
+ memset(&m, 0, sizeof(m));
strncpy(m.resource, l->l.resource, sizeof(m.resource));
m.request = MSG_GRANT;
m.owner_pid = l->l.owner_pid;
@@ -504,19 +512,19 @@ grant_client(struct lock_node *l)
c = find_client(l->l.owner_pid);
if (!c) {
- printf("can't find client for pid %d\n", l->l.owner_pid);
+ cpgl_debug("can't find client for pid %d\n", l->l.owner_pid);
return 1;
}
if (c->fd < 0) {
- printf(" Client has bad fd\n");
+ cpgl_debug(" Client has bad fd\n");
return -1;
}
if (write_retry(c->fd, &m, sizeof(m), NULL) < 0) {
/* no client anymore; drop and send to next guy XXX */
/* This should be handled by our main loop */
- //printf("Failed to notify client!\n");
+ //cpgl_debug("Failed to notify client!\n");
}
return 0;
@@ -529,6 +537,7 @@ nak_client(struct request_node *l)
struct client_node *c;
struct cpg_lock_msg m;
+ memset(&m, 0, sizeof(m));
strncpy(m.resource, l->l.resource, sizeof(m.resource));
m.request = MSG_NAK;
m.owner_pid = l->l.owner_pid;
@@ -537,19 +546,19 @@ nak_client(struct request_node *l)
c = find_client(l->l.owner_pid);
if (!c) {
- printf("can't find client for pid %d\n", l->l.owner_pid);
+ cpgl_debug("can't find client for pid %d\n", l->l.owner_pid);
return 1;
}
if (c->fd < 0) {
- printf(" Client has bad fd\n");
+ cpgl_debug(" Client has bad fd\n");
return -1;
}
if (write_retry(c->fd, &m, sizeof(m), NULL) < 0) {
/* no client anymore; drop and send to next guy XXX */
/* This should be handled by our main loop */
- //printf("Failed to notify client!\n");
+ //cpgl_debug("Failed to notify client!\n");
}
return 0;
@@ -581,7 +590,7 @@ process_lock(struct cpg_lock_msg *m)
if (!joined)
return 0;
- printf("LOCK %s: queue for %d:%d:%d\n", m->resource,
+ cpgl_debug("LOCK %s: queue for %d:%d:%d\n", m->resource,
m->owner_nodeid, m->owner_pid, m->owner_tid);
queue_request(m);
@@ -657,14 +666,14 @@ process_grant(struct cpg_lock_msg *m, uint32_t nodeid)
if (l->l.state == LOCK_HELD) {
if (m->owner_pid == 0 ||
m->owner_nodeid == 0) {
- printf("GRANT averted\n");
+ cpgl_debug("GRANT averted\n");
return 0;
}
} else {
l->l.state = LOCK_HELD;
}
- printf("GRANT %s: to %d:%d:%d\n",
+ cpgl_debug("GRANT %s: to %d:%d:%d\n",
m->resource, m->owner_nodeid,
m->owner_pid, m->owner_tid);
@@ -705,7 +714,8 @@ process_grant(struct cpg_lock_msg *m, uint32_t nodeid)
if (group_members->nodeid == my_node_id &&
!is_member(l->l.owner_nodeid)) {
- printf("GRANT to non-member %d; giving to next requestor\n", l->l.owner_nodeid);
+ cpgl_debug("GRANT to non-member %d; giving to next requestor\n",
+ l->l.owner_nodeid);
l->l.state = LOCK_FREE;
if (grant_next(m) == 0)
@@ -717,7 +727,7 @@ process_grant(struct cpg_lock_msg *m, uint32_t nodeid)
/* Record lock state since we now know it */
/* Allocate a lock structure */
- l = do_alloc(sizeof(*m));
+ l = do_alloc(sizeof(*l));
strncpy(l->l.resource, m->resource, sizeof(l->l.resource));
l->l.state = LOCK_HELD;
l->l.owner_nodeid = m->owner_nodeid;
@@ -779,10 +789,13 @@ process_unlock(struct cpg_lock_msg *m, uint32_t nodeid)
GRANT to the first node on the request queue */
if (l->l.owner_nodeid == m->owner_nodeid &&
l->l.owner_pid == m->owner_pid) {
- printf("UNLOCK %s: %d:%d:%d\n", m->resource, m->owner_nodeid, m->owner_pid, m->owner_tid);
+ cpgl_debug("UNLOCK %s: %d:%d:%d\n",
+ m->resource, m->owner_nodeid, m->owner_pid, m->owner_tid);
l->l.state = LOCK_FREE;
- //if (grant_next(m) != 0)
- //l->l.state = LOCK_PENDING;
+ if (l->l.owner_nodeid == my_node_id) {
+ if (grant_next(m) != 0)
+ l->l.state = LOCK_PENDING;
+ }
}
}
@@ -802,7 +815,7 @@ find_lock(struct cpg_lock_msg *m)
list_for(&locks, l, x) {
if (m->lockid == l->l.local_id) {
strncpy(m->resource, l->l.resource, sizeof(m->resource));
- printf("LOCK %d -> %s\n", m->lockid, m->resource);
+ cpgl_debug("LOCK %d -> %s\n", m->lockid, m->resource);
m->owner_nodeid = l->l.owner_nodeid;
m->owner_pid = l->l.owner_pid;
m->owner_tid = l->l.owner_tid;
@@ -816,29 +829,29 @@ find_lock(struct cpg_lock_msg *m)
static int
-process_join(struct cpg_lock_msg *m, uint32_t nodeid)
+process_join(struct cpg_lock_msg *m, uint32_t nodeid, uint32_t pid)
{
struct member_node *n;
int x;
list_for(&group_members, n, x) {
if (n->nodeid == nodeid) {
- list_remove(&group_members, n);
- list_append(&group_members, n);
- printf("JOIN: moving %d to back\n", nodeid);
+ cpgl_debug("IGNORING JOIN from existing member %d.%d (%d.%d)\n",
+ nodeid, pid, nodeid, n->pid);
return 0;
}
}
n = do_alloc(sizeof(*n));
n->nodeid = nodeid;
- printf("JOIN: node %d", n->nodeid);
+ n->pid = pid;
+ cpgl_debug("JOIN: node %d.%u", n->nodeid, n->pid);
if (nodeid == my_node_id) {
- printf(" (self)");
+ cpgl_debug(" (self)");
joined = 1;
}
total_members++;
- printf("\n");
+ cpgl_debug("\n");
list_insert(&group_members, n);
return 0;
@@ -846,10 +859,10 @@ process_join(struct cpg_lock_msg *m, uint32_t nodeid)
static int
-process_request(struct cpg_lock_msg *m, uint32_t nodeid)
+process_request(struct cpg_lock_msg *m, uint32_t nodeid, uint32_t pid)
{
if (m->request == MSG_HALT) {
- printf("FAULT: Halting operations; see node %d\n", m->owner_nodeid);
+ cpgl_debug("FAULT: Halting operations; see node %d\n", m->owner_nodeid);
while (1)
sleep(30);
}
@@ -873,7 +886,7 @@ process_request(struct cpg_lock_msg *m, uint32_t nodeid)
purge_requests(m->owner_nodeid, m->owner_pid);
break;
case MSG_JOIN:
- process_join(m, nodeid);
+ process_join(m, nodeid, pid);
break;
}
@@ -891,10 +904,10 @@ cpg_deliver_func(cpg_handle_t h,
{
if (msglen != sizeof(struct cpg_lock_msg)) {
- printf("Invalid message size %d\n", (int)msglen);
+ cpgl_debug("Invalid message size %d\n", (int)msglen);
}
- process_request((struct cpg_lock_msg *)msg, nodeid);
+ process_request((struct cpg_lock_msg *)msg, nodeid, pid);
}
@@ -908,7 +921,7 @@ cpg_config_change(cpg_handle_t h,
struct member_node *n;
size_t x, y;
struct cpg_lock_msg m;
-
+ int cpglock_members_removed = 0;
memset(&m, 0, sizeof(m));
strncpy(m.resource, "(none)", sizeof(m.resource));
@@ -917,8 +930,7 @@ cpg_config_change(cpg_handle_t h,
old_msg(&m);
if (total_members == 0) {
-
- printf("JOIN: Setting up initial node list\n");
+ cpgl_debug("JOIN: Setting up initial node list\n");
for (x = 0; x < memberlen; x++) {
for (y = 0; y < joinlen; y++) {
if (join[y].nodeid == members[x].nodeid)
@@ -928,16 +940,17 @@ cpg_config_change(cpg_handle_t h,
n = do_alloc(sizeof(*n));
n->nodeid = members[x].nodeid;
- printf("JOIN: node %d\n", n->nodeid);
+ n->pid = members[x].pid;
+ cpgl_debug("JOIN: node %d.%u\n", n->nodeid, n->pid);
list_insert(&group_members, n);
}
}
- printf("JOIN: Done\n");
+ cpgl_debug("JOIN: Done\n");
total_members = memberlen;
}
- //printf("members %d now, %d joined, %d left\n", memberlen, joinlen, leftlen);
+ //cpgl_debug("members %d now, %d joined, %d left\n", memberlen, joinlen, leftlen);
#if 0
/* XXX process join on receipt of JOIN message rather than here
@@ -946,34 +959,39 @@ cpg_config_change(cpg_handle_t h,
for (x = 0; x < joinlen; x++) {
n = do_alloc(sizeof(*n));
n->nodeid = join[x].nodeid;
- printf("ADD: node %d\n", n->nodeid);
+ cpgl_debug("ADD: node %d\n", n->nodeid);
list_insert(&group_members, n);
}
#endif
for (x = 0; x < leftlen; x++) {
-
list_for(&group_members, n, y) {
if (n->nodeid == left[x].nodeid) {
- list_remove(&group_members, n);
- printf("DELETE: node %d\n", n->nodeid);
- del_node(n->nodeid);
- free(n);
- break;
+ if (n->pid == left[x].pid) {
+ list_remove(&group_members, n);
+ cpgl_debug("DELETE: node %d.%u\n", n->nodeid, n->pid);
+ del_node(n->nodeid);
+ cpglock_members_removed++;
+ free(n);
+ break;
+ } else {
+ cpgl_debug("DUPE NODE %d LEFT (%u != %u)\n",
+ n->nodeid, n->pid, left[x].pid);
+ }
}
}
-
- total_members -= leftlen;
- if (total_members < 0)
- total_members = 0;
}
+ total_members -= cpglock_members_removed;
+ if (total_members < 0)
+ total_members = 0;
+
#if 0
- printf("MEMBERS:");
+ cpgl_debug("MEMBERS:");
list_for(&group_members, n, y) {
- printf(" %d", n->nodeid);
+ cpgl_debug(" %d.%u", n->nodeid, n->pid);
}
- printf("\n");
+ cpgl_debug("\n");
#endif
return;
@@ -985,9 +1003,8 @@ static cpg_callbacks_t my_callbacks = {
.cpg_confchg_fn = cpg_config_change
};
-
static int
-cpg_init(void)
+cpg_fin(void)
{
struct cpg_name gname;
@@ -1004,27 +1021,19 @@ cpg_init(void)
if (gname.length <= 0)
return -1;
- memset(&cpg, 0, sizeof(cpg));
- if (cpg_initialize(&cpg, &my_callbacks) != CPG_OK) {
- perror("cpg_initialize");
- return -1;
- }
-
- if (cpg_join(cpg, &gname) != CPG_OK) {
- perror("cpg_join");
- return -1;
- }
-
- cpg_local_get(cpg, &my_node_id);
+ cpg_leave(cpg, &gname);
+ cpg_finalize(cpg);
return 0;
}
-
static int
-cpg_fin(void)
+cpg_init(void)
{
struct cpg_name gname;
+ struct cpg_address member_list[64];
+ int cpg_member_list_len = 0;
+ int i, ret;
errno = EINVAL;
@@ -1039,13 +1048,38 @@ cpg_fin(void)
if (gname.length <= 0)
return -1;
- cpg_leave(cpg, &gname);
- cpg_finalize(cpg);
+ memset(&cpg, 0, sizeof(cpg));
+ if (cpg_initialize(&cpg, &my_callbacks) != CPG_OK) {
+ perror("cpg_initialize");
+ return -1;
+ }
+
+ if (cpg_join(cpg, &gname) != CPG_OK) {
+ perror("cpg_join");
+ return -1;
+ }
+
+ cpg_local_get(cpg, &my_node_id);
+
+ ret = cpg_membership_get(cpg, &gname, member_list, &cpg_member_list_len);
+ if (ret != CPG_OK) {
+ fprintf(stderr, "cpg_membership_get() failed: %s", strerror(errno));
+ cpg_fin();
+ return -1;
+ }
+
+ for (i = 0 ; i < cpg_member_list_len ; i++) {
+ if (member_list[i].nodeid == my_node_id) {
+ fprintf(stderr, "nodeid %d already in group with PID %u\n",
+ member_list[i].nodeid, member_list[i].pid);
+ cpg_fin();
+ return -1;
+ }
+ }
return 0;
}
-
int
main(int argc, char **argv)
{
@@ -1054,14 +1088,43 @@ main(int argc, char **argv)
int cpgfd;
int afd = -1;
int n,x;
-
+ int nofork = 0;
+ int opt;
struct cpg_lock_msg m;
struct client_node *client;
+ while ((opt = getopt(argc, argv, "fh")) != EOF) {
+ switch (opt) {
+ case 'f':
+ nofork = 1;
+ break;
+ case 'h':
+ printf("Usage: %s [options]\n\
+ -f Don't daemonize\n\
+ -h Print this help message\n",
+ argv[0]);
+ return 0;
+ }
+ }
+
signal(SIGPIPE, SIG_IGN);
+ if (cpg_init() < 0) {
+ fprintf(stderr, "Unable to join CPG group\n");
+ return -1;
+ }
+
fd = sock_listen(CPG_LOCKD_SOCK);
- cpg_init();
+ if (fd < 0) {
+ fprintf(stderr, "Error connecting to %s: %s\n",
+ CPG_LOCKD_SOCK, strerror(errno));
+ cpg_fin();
+ return -1;
+ }
+
+ if (!nofork)
+ daemon_init((char *) "cpglockd");
+
cpg_local_get(cpg, &my_node_id);
cpg_fd_get(cpg, &cpgfd);
if (send_join() < 0)
@@ -1090,7 +1153,7 @@ main(int argc, char **argv)
}
if (FD_ISSET(cpgfd, &rfds)) {
- cpg_dispatch(cpg, CPG_DISPATCH_ONE);
+ cpg_dispatch(cpg, CPG_DISPATCH_ALL);
--n;
}
@@ -1103,7 +1166,7 @@ main(int argc, char **argv)
continue;
--n;
if (read_retry(client->fd, &m, sizeof(m), NULL) < 0) {
- printf("Closing client fd %d pid %d: %d\n",
+ cpgl_debug("Closing client fd %d pid %d: %d\n",
client->fd, client->pid, errno);
del_client(client->fd);
@@ -1118,7 +1181,7 @@ main(int argc, char **argv)
}
if (m.request == MSG_UNLOCK) {
- //printf("Unlock from fd %d\n", client->fd);
+ //cpgl_debug("Unlock from fd %d\n", client->fd);
find_lock(&m);
if (grant_next(&m) == 0)
send_unlock(&m);
12 years, 1 month
cluster: RHEL6 - rgmanager: Merge upstream cpglockd
by Ryan McCabe
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: b1cc2fd5ce82d5c65ce979c4138aa9087aaf984c
Parent: 8c2898779b82f4d09949b6fb78a7d4715125810d
Author: Lon Hohberger <lhh(a)redhat.com>
AuthorDate: Wed Mar 21 16:35:10 2012 -0400
Committer: Ryan McCabe <rmccabe(a)redhat.com>
CommitterDate: Tue May 1 10:55:43 2012 -0400
rgmanager: Merge upstream cpglockd
Signed-off-by: Lon Hohberger <lhh(a)redhat.com>
Reviewed-by: Fabio M. Di Nitto <fdinitto(a)redhat.com>
Reviewed-by: Ryan McCabe <rmccabe(a)redhat.com>
---
rgmanager/src/daemons/cpglockd.c | 9 +++------
1 files changed, 3 insertions(+), 6 deletions(-)
diff --git a/rgmanager/src/daemons/cpglockd.c b/rgmanager/src/daemons/cpglockd.c
index 55d832a..3ed132b 100644
--- a/rgmanager/src/daemons/cpglockd.c
+++ b/rgmanager/src/daemons/cpglockd.c
@@ -176,7 +176,7 @@ old_msg(struct cpg_lock_msg *m)
n = do_alloc(sizeof(*n));
memcpy(&n->m, m, sizeof(n->m));
list_append(&messages, n);
- if (message_count < 10) {
+ if (message_count < 20) {
++message_count;
} else {
n = messages;
@@ -781,8 +781,8 @@ process_unlock(struct cpg_lock_msg *m, uint32_t nodeid)
l->l.owner_pid == m->owner_pid) {
printf("UNLOCK %s: %d:%d:%d\n", m->resource, m->owner_nodeid, m->owner_pid, m->owner_tid);
l->l.state = LOCK_FREE;
- if (grant_next(m) != 0)
- l->l.state = LOCK_PENDING;
+ //if (grant_next(m) != 0)
+ //l->l.state = LOCK_PENDING;
}
}
@@ -1055,12 +1055,9 @@ main(int argc, char **argv)
int afd = -1;
int n,x;
- struct cpg_lock l;
struct cpg_lock_msg m;
struct client_node *client;
- l.state = 0;
-
signal(SIGPIPE, SIG_IGN);
fd = sock_listen(CPG_LOCKD_SOCK);
12 years, 1 month
cluster: RHEL6 - rgmanager: Add simple locking over CPG
by Ryan McCabe
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: 8c2898779b82f4d09949b6fb78a7d4715125810d
Parent: 24d54d2abb33c858b28e6d53d03e1b243881657b
Author: Lon Hohberger <lhh(a)redhat.com>
AuthorDate: Wed Mar 21 16:29:12 2012 -0400
Committer: Ryan McCabe <rmccabe(a)redhat.com>
CommitterDate: Tue May 1 10:54:16 2012 -0400
rgmanager: Add simple locking over CPG
Signed-off-by: Lon Hohberger <lhh(a)redhat.com>
Reviewed-by: Ryan McCabe <rmccabe(a)redhat.com>
Reviewed-by: Fabio M. Di Nitto <fdinitto(a)redhat.com>
Conflicts:
rgmanager/src/clulib/lock.c
---
rgmanager/include/cpglock-internal.h | 38 ++
rgmanager/include/cpglock.h | 40 ++
rgmanager/include/list.h | 1 +
rgmanager/include/lock.h | 15 +-
rgmanager/include/sock.h | 18 +
rgmanager/src/clulib/Makefile | 2 +-
rgmanager/src/clulib/cpg_lock.c | 104 +++
rgmanager/src/clulib/dlm_lock.c | 301 +++++++++
rgmanager/src/clulib/libcpglock.c | 286 +++++++++
rgmanager/src/clulib/lock.c | 297 +---------
rgmanager/src/clulib/sock.c | 354 +++++++++++
rgmanager/src/daemons/Makefile | 12 +-
rgmanager/src/daemons/cpglockd.c | 1147 ++++++++++++++++++++++++++++++++++
rgmanager/src/daemons/main.c | 24 +-
14 files changed, 2326 insertions(+), 313 deletions(-)
diff --git a/rgmanager/include/cpglock-internal.h b/rgmanager/include/cpglock-internal.h
new file mode 100644
index 0000000..1c322ae
--- /dev/null
+++ b/rgmanager/include/cpglock-internal.h
@@ -0,0 +1,38 @@
+#ifndef _CPGLOCK_INT_H
+#define _CPGLOCK_INT_H
+
+#ifndef CPG_LOCKD_SOCK
+#define CPG_LOCKD_SOCK "/var/run/cpglockd.sk"
+#endif
+
+#include <stdint.h>
+
+typedef enum {
+ MSG_LOCK = 1,
+ MSG_NAK = 2,
+ MSG_GRANT = 3,
+ MSG_UNLOCK = 4,
+ MSG_PURGE = 5,
+ MSG_CONFCHG= 6,
+ MSG_JOIN = 7,
+ MSG_DUMP = 998,
+ MSG_HALT = 999
+} cpg_lock_req_t;
+
+/* Mixed architecture not supported yet */
+struct __attribute__((packed)) cpg_lock_msg {
+ char resource[96];
+ int32_t request;
+ uint32_t owner_nodeid;
+ uint32_t owner_pid;
+ uint32_t flags;
+ uint32_t lockid;
+ uint32_t owner_tid;
+ char pad[8];
+}; /* 128 */
+
+pid_t _gettid (void);
+
+#define CPG_LOCKD_NAME "cpglockd"
+
+#endif
diff --git a/rgmanager/include/cpglock.h b/rgmanager/include/cpglock.h
new file mode 100644
index 0000000..e451d2f
--- /dev/null
+++ b/rgmanager/include/cpglock.h
@@ -0,0 +1,40 @@
+#ifndef _CPGLOCK_H
+#define _CPGLOCK_H
+
+typedef enum {
+ LOCK_FREE = 0,
+ LOCK_PENDING = 1,
+ LOCK_HELD = 2
+} lock_state_t;
+
+typedef enum {
+ FL_TRY = 0x1
+} lock_flag_t;
+
+struct cpg_lock {
+ char resource[96];
+ int local_fd;
+ int local_id;
+ lock_state_t state;
+ int owner_nodeid;
+ int owner_pid;
+ int owner_tid;
+};
+
+typedef void * cpg_lock_handle_t;
+
+int cpg_lock_init(cpg_lock_handle_t *h);
+
+/* 0 if successful, -1 if error */
+int cpg_lock(cpg_lock_handle_t h, const char *resource,
+ lock_flag_t flags, struct cpg_lock *lock);
+
+int cpg_unlock(cpg_lock_handle_t h,
+ struct cpg_lock *lock);
+
+/* Warning: drops all locks with this client */
+int cpg_lock_fin(cpg_lock_handle_t h);
+
+int cpg_lock_dump(FILE *fp);
+
+#endif
diff --git a/rgmanager/include/list.h b/rgmanager/include/list.h
index 0dac6e8..0d1c1d5 100644
--- a/rgmanager/include/list.h
+++ b/rgmanager/include/list.h
@@ -36,6 +36,7 @@ do { \
*list = newnode; \
} while (0)
+#define list_append list_insert
#define list_remove(list, oldnode) \
do { \
diff --git a/rgmanager/include/lock.h b/rgmanager/include/lock.h
index 7941304..82a1faf 100644
--- a/rgmanager/include/lock.h
+++ b/rgmanager/include/lock.h
@@ -6,17 +6,12 @@
#include <stdlib.h>
#include <libdlm.h>
-int clu_ls_lock(dlm_lshandle_t ls, int mode, struct dlm_lksb *lksb,
- int options, const char *resource);
-dlm_lshandle_t clu_open_lockspace(const char *lsname);
-int clu_ls_unlock(dlm_lshandle_t ls, struct dlm_lksb *lksb);
-int clu_close_lockspace(dlm_lshandle_t ls, const char *lsname);
-
/* Default lockspace wrappers */
int clu_lock_init(const char *default_lsname);
-int clu_lock(int mode, struct dlm_lksb *lksb,
- int options, const char *resource);
-int clu_unlock(struct dlm_lksb *lksb);
-void clu_lock_finished(const char *default_lsname);
+int cpg_lock_initialize(void);
+
+extern int (*clu_lock)(int, struct dlm_lksb *, int, const char *);
+extern int (*clu_unlock)(struct dlm_lksb *lksb);
+extern void (*clu_lock_finished)(const char *);
#endif
diff --git a/rgmanager/include/sock.h b/rgmanager/include/sock.h
new file mode 100644
index 0000000..6cd2331
--- /dev/null
+++ b/rgmanager/include/sock.h
@@ -0,0 +1,18 @@
+#ifndef _SOCK_H
+#define _SOCK_H
+
+ssize_t read_retry(int sockfd, void *buf, size_t count,
+ struct timeval * timeout);
+int select_retry(int fdmax, fd_set * rfds, fd_set * wfds, fd_set * xfds,
+ struct timeval *timeout);
+ssize_t write_retry(int fd, void *buf, size_t count,
+ struct timeval *timeout);
+
+int sock_listen(const char *sockpath);
+int sock_connect(const char *sockpath, int tout);
+int sock_accept(int fd);
+void hexdump(const void *buf, size_t len);
+
+void *do_alloc(size_t);
+
+#endif
diff --git a/rgmanager/src/clulib/Makefile b/rgmanager/src/clulib/Makefile
index 34f58fb..adb5578 100644
--- a/rgmanager/src/clulib/Makefile
+++ b/rgmanager/src/clulib/Makefile
@@ -12,7 +12,7 @@ include $(OBJDIR)/make/uninstall.mk
OBJS1= logging.o daemon_init.o signals.o msgsimple.o \
gettid.o rg_strings.o message.o members.o fdops.o \
lock.o cman.o vft.o msg_cluster.o msg_socket.o \
- wrap_lock.o sets.o
+ wrap_lock.o sets.o sock.o libcpglock.o dlm_lock.o cpg_lock.o
OBJS2= msgtest.o
diff --git a/rgmanager/src/clulib/cpg_lock.c b/rgmanager/src/clulib/cpg_lock.c
new file mode 100644
index 0000000..a747419
--- /dev/null
+++ b/rgmanager/src/clulib/cpg_lock.c
@@ -0,0 +1,104 @@
+/** @file
+ * Locking.
+ */
+#include <errno.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <string.h>
+#include <stdio.h>
+#include <assert.h>
+#include <sys/ioctl.h>
+#include <lock.h>
+#include <sys/types.h>
+#include <sys/select.h>
+#include <pthread.h>
+
+#include <cpglock.h>
+
+/* Default lockspace stuff */
+static cpg_lock_handle_t _cpgh = NULL;
+static pthread_mutex_t _default_lock = PTHREAD_MUTEX_INITIALIZER;
+
+static void
+dlm2cpg(struct dlm_lksb *dlm, struct cpg_lock *cpg)
+{
+ memset(cpg, 0, sizeof(*cpg));
+ cpg->local_id = dlm->sb_lkid;
+ cpg->state = dlm->sb_status;
+}
+
+static void
+cpg2dlm(struct cpg_lock *cpg, struct dlm_lksb *dlm)
+{
+ memset(dlm, 0, sizeof(*dlm));
+ dlm->sb_status = cpg->state;
+ dlm->sb_lkid = cpg->local_id;
+}
+
+
+static int
+_cpg_lock(int mode,
+ struct dlm_lksb *lksb,
+ int options,
+ const char *resource)
+{
+ int ret = 0;
+
+ struct cpg_lock l;
+
+ if (options == LKF_NOQUEUE)
+ ret = cpg_lock(_cpgh, resource, 1, &l);
+ else
+ ret = cpg_lock(_cpgh, resource, 0, &l);
+
+ if (ret == 0) {
+ cpg2dlm(&l, lksb);
+ }
+
+ return ret;
+}
+
+
+static int
+_cpg_unlock(struct dlm_lksb *lksb)
+{
+ struct cpg_lock l;
+
+ dlm2cpg(lksb, &l);
+ return cpg_unlock(_cpgh, &l);
+}
+
+
+static void
+_cpg_lock_finished(const char *name)
+{
+ pthread_mutex_lock(&_default_lock);
+ cpg_lock_fin(_cpgh);
+ pthread_mutex_unlock(&_default_lock);
+}
+
+
+int
+cpg_lock_initialize(void)
+{
+ int ret, err;
+
+ pthread_mutex_lock(&_default_lock);
+ if (_cpgh) {
+ pthread_mutex_unlock(&_default_lock);
+ return 0;
+ }
+
+ cpg_lock_init(&_cpgh);
+ ret = (_cpgh == NULL);
+ err = errno;
+ pthread_mutex_unlock(&_default_lock);
+
+ /* Set up function pointers */
+ clu_lock = _cpg_lock;
+ clu_unlock = _cpg_unlock;
+ clu_lock_finished = _cpg_lock_finished;
+
+ errno = err;
+ return ret;
+}
diff --git a/rgmanager/src/clulib/dlm_lock.c b/rgmanager/src/clulib/dlm_lock.c
new file mode 100644
index 0000000..d3f4250
--- /dev/null
+++ b/rgmanager/src/clulib/dlm_lock.c
@@ -0,0 +1,301 @@
+/** @file
+ * Locking.
+ */
+#include <errno.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <string.h>
+#include <stdio.h>
+#include <assert.h>
+#include <sys/ioctl.h>
+#include <lock.h>
+#include <sys/types.h>
+#include <sys/select.h>
+#include <pthread.h>
+
+/* Default lockspace stuff */
+static dlm_lshandle_t _default_ls = NULL;
+static pthread_mutex_t _default_lock = PTHREAD_MUTEX_INITIALIZER;
+
+
+static void
+ast_function(void * __attribute__ ((unused)) arg)
+{
+}
+
+
+static int
+wait_for_dlm_event(dlm_lshandle_t ls)
+{
+ fd_set rfds;
+ int fd = dlm_ls_get_fd(ls);
+
+ FD_ZERO(&rfds);
+ FD_SET(fd, &rfds);
+
+ if (select(fd + 1, &rfds, NULL, NULL, NULL) == 1)
+ return dlm_dispatch(fd);
+
+ return -1;
+}
+
+
+static int
+clu_ls_lock(dlm_lshandle_t ls,
+ int mode,
+ struct dlm_lksb *lksb,
+ int options,
+ const char *resource)
+{
+ int ret;
+
+ if (!ls || !lksb || !resource || !strlen(resource)) {
+ printf("%p %p %p %d\n", ls, lksb, resource,
+ (int)strlen(resource));
+ printf("INVAL...\n");
+ errno = EINVAL;
+ return -1;
+ }
+
+ ret = dlm_ls_lock(ls, mode, lksb, options, resource,
+ strlen(resource), 0, ast_function, lksb,
+ NULL, NULL);
+
+ if (ret < 0) {
+ if (errno == ENOENT)
+ assert(0);
+
+ return -1;
+ }
+
+ if ((ret = (wait_for_dlm_event(ls) < 0))) {
+ fprintf(stderr, "wait_for_dlm_event: %d / %d\n",
+ ret, errno);
+ return -1;
+ }
+
+ if (lksb->sb_status == 0)
+ return 0;
+
+ errno = lksb->sb_status;
+ return -1;
+}
+
+
+static dlm_lshandle_t
+clu_open_lockspace(const char *lsname)
+{
+ dlm_lshandle_t ls = NULL;
+
+ //printf("opening lockspace %s\n", lsname);
+
+ while (!ls) {
+ ls = dlm_open_lockspace(lsname);
+ if (ls)
+ break;
+
+ /*
+ printf("Failed to open: %s; trying create.\n",
+ strerror(errno));
+ */
+
+ ls = dlm_create_lockspace(lsname, 0644);
+ if (ls)
+ break;
+
+ /* Work around race: Someone was closing lockspace as
+ we were trying to open it. Retry. */
+ if (errno == ENOENT)
+ continue;
+
+ fprintf(stderr, "failed acquiring lockspace: %s\n",
+ strerror(errno));
+
+ return NULL;
+ }
+
+ return ls;
+}
+
+
+static int
+clu_ls_unlock(dlm_lshandle_t ls, struct dlm_lksb *lksb)
+{
+ int ret;
+
+ if (!ls || !lksb) {
+ errno = EINVAL;
+ return -1;
+ }
+
+ ret = dlm_ls_unlock(ls, lksb->sb_lkid, 0, lksb, NULL);
+
+ if (ret != 0)
+ return ret;
+
+ /* lksb->sb_status should be EINPROG at this point */
+
+ if (wait_for_dlm_event(ls) < 0) {
+ errno = lksb->sb_status;
+ return -1;
+ }
+
+ return 0;
+}
+
+
+static int
+clu_close_lockspace(dlm_lshandle_t ls, const char *name)
+{
+ return dlm_release_lockspace(name, ls, 1);
+}
+
+
+static int
+_clu_lock(int mode,
+ struct dlm_lksb *lksb,
+ int options,
+ const char *resource)
+{
+ int ret = 0, block = 0, conv = 0, err;
+
+ block = !(options & LKF_NOQUEUE);
+
+ errno = EINVAL;
+ if (!lksb)
+ return -1;
+
+ memset(lksb, 0, sizeof(struct dlm_lksb));
+
+ /*
+ Try to use a conversion lock mechanism when possible
+ If the caller calls explicitly with a NULL lock, then
+ assume the caller knows what it is doing.
+
+ Only take the NULL lock if:
+ (a) the user isn't specifying CONVERT; if they are, they
+ know what they are doing.
+
+ ...and one of...
+
+ (b) This is a blocking call, or
+ (c) The user requested a NULL lock explicitly. In this case,
+ short-out early; there's no reason to convert a NULL lock
+ to a NULL lock.
+ */
+ if (!(options & LKF_CONVERT) &&
+ (block || (mode == LKM_NLMODE))) {
+ /* Acquire NULL lock */
+ pthread_mutex_lock(&_default_lock);
+ ret = clu_ls_lock(_default_ls, LKM_NLMODE, lksb,
+ (options & ~LKF_NOQUEUE),
+ resource);
+ err = errno;
+ pthread_mutex_unlock(&_default_lock);
+ if (ret == 0) {
+ if (mode == LKM_NLMODE) {
+ /* User only wanted a NULL lock... */
+ return 0;
+ }
+ /*
+ Ok, NULL lock was taken, rest of blocking
+ call should be done using lock conversions.
+ */
+ options |= LKF_CONVERT;
+ conv = 1;
+ } else {
+ switch(err) {
+ case EINVAL:
+ /* Oops, null locks don't work on this
+ plugin; use normal spam mode */
+ break;
+ default:
+ errno = err;
+ return -1;
+ }
+ }
+ }
+
+ while (1) {
+ pthread_mutex_lock(&_default_lock);
+ ret = clu_ls_lock(_default_ls, mode, lksb,
+ (options | LKF_NOQUEUE),
+ resource);
+ err = errno;
+ pthread_mutex_unlock(&_default_lock);
+
+ if ((ret != 0) && (err == EAGAIN) && block) {
+ usleep(random()&32767);
+ continue;
+ }
+
+ break;
+ }
+
+ if (ret != 0 && conv) {
+ /* If we get some other error, release the NL lock we
+ took so we don't leak locks*/
+ pthread_mutex_lock(&_default_lock);
+ clu_ls_unlock(_default_ls, lksb);
+ pthread_mutex_unlock(&_default_lock);
+ errno = err;
+ }
+
+ return ret;
+}
+
+
+static int
+_clu_unlock(struct dlm_lksb *lksb)
+{
+ int ret, err;
+ pthread_mutex_lock(&_default_lock);
+ ret = clu_ls_unlock(_default_ls, lksb);
+ err = errno;
+ pthread_mutex_unlock(&_default_lock);
+
+ usleep(random()&32767);
+ errno = err;
+ return ret;
+}
+
+
+static void
+_clu_lock_finished(const char *name)
+{
+ pthread_mutex_lock(&_default_lock);
+ if (_default_ls)
+ clu_close_lockspace(_default_ls, name);
+ pthread_mutex_unlock(&_default_lock);
+}
+
+
+int
+clu_lock_init(const char *dflt_lsname)
+{
+ int ret, err;
+
+ pthread_mutex_lock(&_default_lock);
+ if (_default_ls) {
+ pthread_mutex_unlock(&_default_lock);
+ return 0;
+ }
+
+ if (!dflt_lsname || !strlen(dflt_lsname)) {
+ pthread_mutex_unlock(&_default_lock);
+ errno = EINVAL;
+ return -1;
+ }
+
+ _default_ls = clu_open_lockspace(dflt_lsname);
+ ret = (_default_ls == NULL);
+ err = errno;
+ pthread_mutex_unlock(&_default_lock);
+
+ clu_lock = _clu_lock;
+ clu_unlock = _clu_unlock;
+ clu_lock_finished = _clu_lock_finished;
+
+ errno = err;
+ return ret;
+}
diff --git a/rgmanager/src/clulib/libcpglock.c b/rgmanager/src/clulib/libcpglock.c
new file mode 100644
index 0000000..8e5c7ff
--- /dev/null
+++ b/rgmanager/src/clulib/libcpglock.c
@@ -0,0 +1,286 @@
+#include <stdlib.h>
+#include <sys/socket.h>
+#include <errno.h>
+#include <stdio.h>
+#include <string.h>
+#include <pthread.h>
+#include <unistd.h>
+
+#include "cpglock.h"
+#include "cpglock-internal.h"
+#include "list.h"
+#include "sock.h"
+
+
+struct pending_node {
+ list_head();
+ struct cpg_lock_msg m;
+};
+
+
+struct cpg_lock_handle {
+ pthread_mutex_t mutex;
+ struct pending_node *pending;
+ int fd;
+ int pid;
+ int seq;
+};
+
+
+static int
+check_pending(struct pending_node **l,
+ struct cpg_lock_msg *exp,
+ struct cpg_lock_msg *ret)
+{
+ struct pending_node *p;
+ int x;
+
+ list_for(l, p, x) {
+ if (!strcmp(exp->resource, p->m.resource) &&
+ exp->owner_tid == p->m.owner_tid) {
+ list_remove(l, p);
+ memcpy(ret, &p->m, sizeof(*ret));
+ free(p);
+ return 0;
+ }
+ }
+
+ return 1;
+}
+
+
+static void
+add_pending(struct pending_node **l,
+ struct cpg_lock_msg *m)
+{
+ struct pending_node *p = do_alloc(sizeof(*p));
+
+ memcpy(&p->m, m, sizeof(p->m));
+ list_insert(l, p);
+}
+
+
+/* Not thread safe */
+int
+cpg_lock_init(void **handle)
+{
+ struct cpg_lock_handle *h;
+ int esv;
+
+ h = do_alloc(sizeof (*h));
+ if (!h)
+ return -1;
+
+ h->fd = sock_connect(CPG_LOCKD_SOCK, 3);
+ if (h->fd < 0) {
+ esv = errno;
+ free(h);
+ errno = esv;
+ return -1;
+ }
+
+ h->pid = getpid();
+ pthread_mutex_init(&h->mutex, NULL);
+
+ *handle = (void *)h;
+ return 0;
+}
+
+
+int
+cpg_lock(void *handle, const char *resource, lock_flag_t flags, struct cpg_lock *lock)
+{
+ struct cpg_lock_handle *h = handle;
+ struct cpg_lock_msg l, r;
+ struct timeval tv;
+ int ret = -1;
+
+ if (!h) {
+ errno = EINVAL;
+ goto out;
+ }
+
+ pthread_mutex_lock(&h->mutex);
+
+ if (h->pid != (int)getpid()) {
+ errno = EBADF;
+ goto out;
+ }
+
+ if (strlen(resource) > sizeof(l.resource)-1) {
+ errno = ENAMETOOLONG;
+ goto out;
+ }
+
+ memset(&l, 0, sizeof(l));
+ memset(&r, 0, sizeof(r));
+ strncpy(l.resource, resource, sizeof(l.resource));
+ strncpy(lock->resource, resource, sizeof(lock->resource));
+ l.owner_pid = h->pid;
+ ++h->seq;
+ l.owner_tid = h->seq;
+ l.request = MSG_LOCK;
+ l.flags = (uint32_t)flags;
+
+ if (write_retry(h->fd, &l, sizeof(l), 0) < 0)
+ goto out;
+
+ /* Thread concurrency: in case multiple threads wake up
+ from select, peek at the message to see if it's ours */
+ do {
+ if (check_pending(&h->pending, &l, &r) == 0)
+ break;
+
+ tv.tv_sec = 0;
+ tv.tv_usec = random() & 16383;
+
+ if (read_retry(h->fd, &r, sizeof(r), &tv) < 0) {
+ if (errno == ETIMEDOUT) {
+ pthread_mutex_unlock(&h->mutex);
+ usleep(random() & 16383);
+ pthread_mutex_lock(&h->mutex);
+ continue;
+ }
+ goto out;
+ }
+
+ if (strcmp(r.resource, l.resource)) {
+ printf("NOTE: msg for wrong lock want: %s got: %s\n", l.resource, r.resource);
+ add_pending(&h->pending, &r);
+ pthread_mutex_unlock(&h->mutex);
+ usleep(random() & 16383);
+ pthread_mutex_lock(&h->mutex);
+ continue;
+ }
+ if (r.owner_tid != l.owner_tid) {
+ printf("NOTE: msg for wrong seq want: %d got: %d\n", l.owner_tid, r.owner_tid );
+ add_pending(&h->pending, &r);
+ pthread_mutex_unlock(&h->mutex);
+ usleep(random() & 16383);
+ pthread_mutex_lock(&h->mutex);
+ continue;
+ }
+ break;
+ } while (1);
+ /* locked */
+
+ if (r.owner_nodeid == 0)
+ goto out;
+
+ if (r.request == MSG_NAK) {
+ errno = EAGAIN;
+ return -1;
+ }
+
+ if (r.request != MSG_GRANT) {
+ //ret = -1;
+ goto out;
+ }
+
+ lock->state = LOCK_HELD;
+ lock->owner_nodeid = r.owner_nodeid;
+ lock->owner_pid = h->pid; /* XXX */
+ lock->local_id = r.lockid;
+
+ ret = 0;
+
+out:
+ pthread_mutex_unlock(&h->mutex);
+ return ret;
+}
+
+
+int
+cpg_unlock(void *handle, struct cpg_lock *lock)
+{
+ struct cpg_lock_handle *h = handle;
+ struct cpg_lock_msg l;
+ int ret = -1;
+
+ if (!h) {
+ errno = EINVAL;
+ goto out;
+ }
+
+ /* Only block on lock requests, not unlock */
+ if (h->pid != (int)getpid()) {
+ errno = EBADF;
+ goto out;
+ }
+
+ if (lock->state != LOCK_HELD) {
+ errno = EINVAL;
+ goto out;
+ }
+
+ if (!lock->local_id) {
+ errno = EINVAL;
+ goto out;
+ }
+
+ strncpy(l.resource, lock->resource, sizeof(l.resource));
+ l.request = MSG_UNLOCK;
+ l.owner_nodeid = lock->owner_nodeid;
+ l.owner_pid = lock->owner_pid;
+ l.lockid = lock->local_id;
+ //l.owner_tid = _gettid();
+ l.owner_tid = 0;
+ lock->state = LOCK_FREE;
+
+ ret = write_retry(h->fd, &l, sizeof(l), 0);
+out:
+ return ret;
+}
+
+
+int
+cpg_lock_dump(FILE *fp)
+{
+ struct cpg_lock_msg l;
+ int fd;
+ char c;
+
+ fd = sock_connect(CPG_LOCKD_SOCK, 3);
+ if (fd < 0)
+ return -1;
+
+ memset(&l, 0, sizeof(l));
+ l.request = MSG_DUMP;
+
+ if (write_retry(fd, &l, sizeof(l), 0) < 0)
+ return -1;
+
+ while (read_retry(fd, &c, 1, 0) == 1)
+ fprintf(fp, "%c", c);
+
+ close(fd);
+ return 0;
+}
+
+
+
+/* Not thread safe */
+int
+cpg_lock_fin(void *handle)
+{
+ struct cpg_lock_handle *h = handle;
+
+ if (!h) {
+ errno = EINVAL;
+ return -1;
+ }
+
+ pthread_mutex_lock(&h->mutex);
+
+ if (h->pid != (int)getpid()) {
+ errno = EBADF;
+ return -1;
+ }
+
+ close(h->fd);
+
+ pthread_mutex_destroy(&h->mutex);
+ free(h);
+ return 0;
+}
+
diff --git a/rgmanager/src/clulib/lock.c b/rgmanager/src/clulib/lock.c
index be542c1..74cd8b1 100644
--- a/rgmanager/src/clulib/lock.c
+++ b/rgmanager/src/clulib/lock.c
@@ -1,297 +1,6 @@
-/** @file
- * Locking.
- */
-#include <errno.h>
-#include <stdlib.h>
-#include <unistd.h>
-#include <string.h>
-#include <stdio.h>
-#include <assert.h>
-#include <sys/ioctl.h>
#include <lock.h>
-#include <sys/types.h>
-#include <sys/select.h>
-#include <pthread.h>
-/* Default lockspace stuff */
-static dlm_lshandle_t _default_ls = NULL;
-static pthread_mutex_t _default_lock = PTHREAD_MUTEX_INITIALIZER;
+int (*clu_lock)(int, struct dlm_lksb *, int, const char *) = NULL;
+int (*clu_unlock)(struct dlm_lksb *lksb) = NULL;
+void (*clu_lock_finished)(const char *) = NULL;
-
-static void
-ast_function(void * __attribute__ ((unused)) arg)
-{
-}
-
-
-static int
-wait_for_dlm_event(dlm_lshandle_t ls)
-{
- fd_set rfds;
- int fd = dlm_ls_get_fd(ls);
-
- FD_ZERO(&rfds);
- FD_SET(fd, &rfds);
-
- if (select(fd + 1, &rfds, NULL, NULL, NULL) == 1)
- return dlm_dispatch(fd);
-
- return -1;
-}
-
-
-int
-clu_ls_lock(dlm_lshandle_t ls,
- int mode,
- struct dlm_lksb *lksb,
- int options,
- const char *resource)
-{
- int ret;
-
- if (!ls || !lksb || !resource || !strlen(resource)) {
- printf("%p %p %p %d\n", ls, lksb, resource,
- (int)strlen(resource));
- printf("INVAL...\n");
- errno = EINVAL;
- return -1;
- }
-
- ret = dlm_ls_lock(ls, mode, lksb, options, resource,
- strlen(resource), 0, ast_function, lksb,
- NULL, NULL);
-
- if (ret < 0) {
- if (errno == ENOENT)
- assert(0);
-
- return -1;
- }
-
- if ((ret = (wait_for_dlm_event(ls) < 0))) {
- fprintf(stderr, "wait_for_dlm_event: %d / %d\n",
- ret, errno);
- return -1;
- }
-
- if (lksb->sb_status == 0)
- return 0;
-
- errno = lksb->sb_status;
- return -1;
-}
-
-
-
-dlm_lshandle_t
-clu_open_lockspace(const char *lsname)
-{
- dlm_lshandle_t ls = NULL;
-
- //printf("opening lockspace %s\n", lsname);
-
- while (!ls) {
- ls = dlm_open_lockspace(lsname);
- if (ls)
- break;
-
- /*
- printf("Failed to open: %s; trying create.\n",
- strerror(errno));
- */
-
- ls = dlm_create_lockspace(lsname, 0644);
- if (ls)
- break;
-
- /* Work around race: Someone was closing lockspace as
- we were trying to open it. Retry. */
- if (errno == ENOENT)
- continue;
-
- fprintf(stderr, "failed acquiring lockspace: %s\n",
- strerror(errno));
-
- return NULL;
- }
-
- return ls;
-}
-
-
-int
-clu_ls_unlock(dlm_lshandle_t ls, struct dlm_lksb *lksb)
-{
- int ret;
-
- if (!ls || !lksb) {
- errno = EINVAL;
- return -1;
- }
-
- ret = dlm_ls_unlock(ls, lksb->sb_lkid, 0, lksb, NULL);
-
- if (ret != 0)
- return ret;
-
- /* lksb->sb_status should be EINPROG at this point */
-
- if (wait_for_dlm_event(ls) < 0) {
- errno = lksb->sb_status;
- return -1;
- }
-
- return 0;
-}
-
-
-int
-clu_close_lockspace(dlm_lshandle_t ls, const char *name)
-{
- return dlm_release_lockspace(name, ls, 1);
-}
-
-
-int
-clu_lock(int mode,
- struct dlm_lksb *lksb,
- int options,
- const char *resource)
-{
- int ret = 0, block = 0, conv = 0, err;
-
- block = !(options & LKF_NOQUEUE);
-
- errno = EINVAL;
- if (!lksb)
- return -1;
-
- memset(lksb, 0, sizeof(struct dlm_lksb));
-
- /*
- Try to use a conversion lock mechanism when possible
- If the caller calls explicitly with a NULL lock, then
- assume the caller knows what it is doing.
-
- Only take the NULL lock if:
- (a) the user isn't specifying CONVERT; if they are, they
- know what they are doing.
-
- ...and one of...
-
- (b) This is a blocking call, or
- (c) The user requested a NULL lock explicitly. In this case,
- short-out early; there's no reason to convert a NULL lock
- to a NULL lock.
- */
- if (!(options & LKF_CONVERT) &&
- (block || (mode == LKM_NLMODE))) {
- /* Acquire NULL lock */
- pthread_mutex_lock(&_default_lock);
- ret = clu_ls_lock(_default_ls, LKM_NLMODE, lksb,
- (options & ~LKF_NOQUEUE),
- resource);
- err = errno;
- pthread_mutex_unlock(&_default_lock);
- if (ret == 0) {
- if (mode == LKM_NLMODE) {
- /* User only wanted a NULL lock... */
- return 0;
- }
- /*
- Ok, NULL lock was taken, rest of blocking
- call should be done using lock conversions.
- */
- options |= LKF_CONVERT;
- conv = 1;
- } else {
- switch(err) {
- case EINVAL:
- /* Oops, null locks don't work on this
- plugin; use normal spam mode */
- break;
- default:
- errno = err;
- return -1;
- }
- }
- }
-
- while (1) {
- pthread_mutex_lock(&_default_lock);
- ret = clu_ls_lock(_default_ls, mode, lksb,
- (options | LKF_NOQUEUE),
- resource);
- err = errno;
- pthread_mutex_unlock(&_default_lock);
-
- if ((ret != 0) && (err == EAGAIN) && block) {
- usleep(random()&32767);
- continue;
- }
-
- break;
- }
-
- if (ret != 0 && conv) {
- /* If we get some other error, release the NL lock we
- took so we don't leak locks*/
- pthread_mutex_lock(&_default_lock);
- clu_ls_unlock(_default_ls, lksb);
- pthread_mutex_unlock(&_default_lock);
- errno = err;
- }
-
- return ret;
-}
-
-
-int
-clu_unlock(struct dlm_lksb *lksb)
-{
- int ret, err;
- pthread_mutex_lock(&_default_lock);
- ret = clu_ls_unlock(_default_ls, lksb);
- err = errno;
- pthread_mutex_unlock(&_default_lock);
-
- usleep(random()&32767);
- errno = err;
- return ret;
-}
-
-
-int
-clu_lock_init(const char *dflt_lsname)
-{
- int ret, err;
-
- pthread_mutex_lock(&_default_lock);
- if (_default_ls) {
- pthread_mutex_unlock(&_default_lock);
- return 0;
- }
-
- if (!dflt_lsname || !strlen(dflt_lsname)) {
- pthread_mutex_unlock(&_default_lock);
- errno = EINVAL;
- return -1;
- }
-
- _default_ls = clu_open_lockspace(dflt_lsname);
- ret = (_default_ls == NULL);
- err = errno;
- pthread_mutex_unlock(&_default_lock);
-
- errno = err;
- return ret;
-}
-
-void
-clu_lock_finished(const char *name)
-{
- pthread_mutex_lock(&_default_lock);
- if (_default_ls)
- clu_close_lockspace(_default_ls, name);
- pthread_mutex_unlock(&_default_lock);
-}
diff --git a/rgmanager/src/clulib/sock.c b/rgmanager/src/clulib/sock.c
new file mode 100644
index 0000000..3965a38
--- /dev/null
+++ b/rgmanager/src/clulib/sock.c
@@ -0,0 +1,354 @@
+/*
+ Copyright Red Hat, Inc. 2002-2003, 2012
+ Copyright Mission Critical Linux, 2000
+
+ This program is free software; you can redistribute it and/or modify it
+ under the terms of the GNU General Public License as published by the
+ Free Software Foundation; either version 2, or (at your option) any
+ later version.
+
+ This program is distributed in the hope that it will be useful, but
+ WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ General Public License for more details.
+
+ You should have received a copy of the GNU General Public License
+ along with this program; see the file COPYING. If not, write to the
+ Free Software Foundation, Inc., 675 Mass Ave, Cambridge,
+ MA 02139, USA.
+*/
+#include <stdio.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <sys/errno.h>
+#include <time.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+#include <fcntl.h>
+
+#include "sock.h"
+
+void *
+do_alloc(size_t n)
+{
+ void *p;
+
+ do {
+ p = malloc(n);
+ if (!p) {
+ usleep(10000);
+ }
+ } while(!p);
+
+ memset(p, 0, n);
+ return p;
+}
+
+
+/**
+ * This is a wrapper around select which will retry in the case we receive
+ * EINTR. This is necessary for read_retry, since it wouldn't make sense
+ * to have read_retry terminate if and only if two EINTRs were received
+ * in a row - one during the read() call, one during the select call...
+ *
+ * See select(2) for description of parameters.
+ */
+int
+select_retry(int fdmax, fd_set * rfds, fd_set * wfds, fd_set * xfds,
+ struct timeval *timeout)
+{
+ int rv;
+
+ while (1) {
+ rv = select(fdmax, rfds, wfds, xfds, timeout);
+ if ((rv == -1) && (errno == EINTR))
+ /* return on EBADF/EINVAL/ENOMEM; continue on EINTR */
+ continue;
+ return rv;
+ }
+}
+
+/**
+ * Retries a write in the event of a non-blocked interrupt signal.
+ *
+ * @param fd File descriptor to which we are writing.
+ * @param buf Data buffer to send.
+ * @param count Number of bytes in buf to send.
+ * @param timeout (struct timeval) telling us how long we should retry.
+ * @return The number of bytes written to the file descriptor,
+ * or -1 on error (with errno set appropriately).
+ */
+ssize_t
+write_retry(int fd, void *buf, size_t count, struct timeval * timeout)
+{
+ int n, rv = 0;
+ ssize_t total = 0, remain = count;
+ fd_set wfds, xfds;
+
+ while (total < count) {
+
+ /* Create the write FD set of 1... */
+ FD_ZERO(&wfds);
+ FD_SET(fd, &wfds);
+ FD_ZERO(&xfds);
+ FD_SET(fd, &xfds);
+
+ /* wait for the fd to be available for writing */
+ rv = select_retry(fd + 1, NULL, &wfds, &xfds, timeout);
+ if (rv == -1)
+ return -1;
+ else if (rv == 0) {
+ errno = ETIMEDOUT;
+ return -1;
+ }
+
+ if (FD_ISSET(fd, &xfds)) {
+ errno = EPIPE;
+ return -1;
+ }
+
+ /*
+ * Attempt to write to fd
+ */
+ n = write(fd, (char *)buf + (off_t) total, remain);
+
+ /*
+ * When we know our fd was select()ed and we receive 0 bytes
+ * when we write, the fd was closed.
+ */
+ if ((n == 0) && (rv == 1)) {
+ errno = EPIPE;
+ return -1;
+ }
+
+ if (n == -1) {
+ if ((errno == EAGAIN) || (errno == EINTR)) {
+ /*
+ * Not ready?
+ */
+ continue;
+ }
+
+ /* Other errors: EIO, EINVAL, etc */
+ return -1;
+ }
+
+ total += n;
+ remain -= n;
+ }
+
+ return total;
+}
+
+/**
+ * Retry reads until we (a) time out or (b) get our data. Of course, if
+ * timeout is NULL, it'll wait forever.
+ *
+ * @param sockfd File descriptor we want to read from.
+ * @param buf Preallocated buffer into which we will read data.
+ * @param count Number of bytes to read.
+ * @param timeout (struct timeval) describing how long we should retry.
+ * @return The number of bytes read on success, or -1 on failure.
+ Note that we will always return (count) or (-1).
+ */
+ssize_t
+read_retry(int sockfd, void *buf, size_t count, struct timeval * timeout)
+{
+ int n, rv = 0;
+ ssize_t total = 0, remain = count;
+ fd_set rfds, xfds;
+
+ memset(buf, 0, count);
+
+ while (total < count) {
+ FD_ZERO(&rfds);
+ FD_SET(sockfd, &rfds);
+ FD_ZERO(&xfds);
+ FD_SET(sockfd, &xfds);
+
+ /*
+ * Select on the socket, in case it closes while we're not
+ * looking...
+ */
+ rv = select_retry(sockfd + 1, &rfds, NULL, &xfds, timeout);
+ if (rv == -1)
+ return -1;
+ else if (rv == 0) {
+ errno = ETIMEDOUT;
+ return -1;
+ }
+
+ if (FD_ISSET(sockfd, &xfds)) {
+ errno = EPIPE;
+ return -1;
+ }
+
+ /*
+ * Attempt to read off the socket
+ */
+ n = read(sockfd, (char *)buf + (off_t) total, remain);
+
+ /*
+ * When we know our socket was select()ed and we receive 0 bytes
+ * when we read, the socket was closed.
+ */
+ if ((n == 0) && (rv == 1)) {
+ errno = EPIPE;
+ return -1;
+ }
+
+ if (n == -1) {
+ if ((errno == EAGAIN) || (errno == EINTR)) {
+ /*
+ * Not ready? Wait for data to become available
+ */
+ continue;
+ }
+
+ /* Other errors: EPIPE, EINVAL, etc */
+ return -1;
+ }
+
+ total += n;
+ remain -= n;
+
+ //printf("read-retry %d/%d remain %d \n", total, count, remain);
+ }
+
+ return total;
+}
+
+
+int
+sock_listen(const char *sockpath)
+{
+ int sock = -1;
+ struct sockaddr_un su;
+ mode_t om;
+
+ sock = socket(PF_LOCAL, SOCK_STREAM, 0);
+ if (sock < 0)
+ goto fail;
+
+ su.sun_family = PF_LOCAL;
+ snprintf(su.sun_path, (sizeof(su.sun_path)), "%s", sockpath);
+
+ unlink(su.sun_path);
+ om = umask(077);
+
+ if (bind(sock, (struct sockaddr *)&su, sizeof(su)) < 0) {
+ umask(om);
+ goto fail;
+ }
+ umask(om);
+
+ if (listen(sock, SOMAXCONN) < 0)
+ goto fail;
+ return sock;
+fail:
+ if (sock >= 0)
+ close(sock);
+ return -1;
+}
+
+
+int
+sock_connect(const char *sockpath, int tout)
+{
+ struct timeval timeout = {tout, 0};
+ int sock, flags, error, ret;
+ socklen_t len;
+ struct sockaddr_un sun;
+ fd_set rfds, wfds;
+
+ sock = socket(PF_LOCAL, SOCK_STREAM, 0);
+ if (sock < 0)
+ return -1;
+
+ sun.sun_family = PF_LOCAL;
+ snprintf(sun.sun_path, (sizeof(sun.sun_path)), "%s", sockpath);
+
+ flags = fcntl(sock, F_GETFL, 0);
+ fcntl(sock, F_SETFL, flags | O_NONBLOCK);
+
+ ret = connect(sock, (struct sockaddr *) &sun, sizeof(sun));
+
+ if (ret < 0 && (errno != EINPROGRESS)) {
+ close(sock);
+ return -1;
+ }
+
+ if (ret == 0)
+ goto done;
+
+ FD_ZERO(&rfds);
+ FD_SET(sock, &rfds);
+ wfds = rfds;
+
+ ret = select(sock + 1, &rfds, &wfds, NULL, &timeout);
+ if (ret == 0) {
+ close(sock);
+ errno = ETIMEDOUT;
+ return -1;
+ }
+
+ if (FD_ISSET(sock, &rfds) || FD_ISSET(sock, &wfds)) {
+ len = sizeof (error);
+ if (getsockopt(sock, SOL_SOCKET, SO_ERROR, &error, &len) < 0) {
+ close(sock);
+ return -1;
+ }
+ } else {
+ close(sock);
+ return -1;
+ }
+
+done:
+
+ return sock;
+}
+
+
+int
+sock_accept(int sockfd)
+{
+ int acceptfd;
+
+ if (sockfd < 0) {
+ errno = EBADF;
+ return -1;
+ }
+
+ while ((acceptfd =
+ accept(sockfd, (struct sockaddr *) NULL, NULL)) < 0) {
+ if (errno == EINTR) {
+ continue;
+ }
+
+ return -1;
+ }
+
+ return acceptfd;
+}
+
+
+void
+hexdump(const void *buf, size_t len)
+{
+ size_t x;
+
+ printf("%d bytes @ %p \n", (int)len, buf);
+
+ for (x = 0; x < len; x++) {
+ printf(" %02x", (((char *)buf)[x])&0xff);
+ if (((x+1) % 16) == 0)
+ printf("\n");
+ }
+
+ printf("\n");
+}
+
+
+
diff --git a/rgmanager/src/daemons/Makefile b/rgmanager/src/daemons/Makefile
index 7218c53..dbd75ea 100644
--- a/rgmanager/src/daemons/Makefile
+++ b/rgmanager/src/daemons/Makefile
@@ -1,11 +1,12 @@
TARGET1= rgmanager
TARGET2= rg_test
TARGET3= clurgmgrd
+TARGET4= cpglockd
-SBINDIRT=$(TARGET1) $(TARGET2)
+SBINDIRT=$(TARGET1) $(TARGET2) $(TARGET4)
SBINSYMT=$(TARGET3)
-all: depends ${TARGET1} ${TARGET2} ${TARGET3}
+all: depends ${TARGET1} ${TARGET2} ${TARGET3} $(TARGET4)
include ../../../make/defines.mk
include $(OBJDIR)/make/cobj.mk
@@ -41,6 +42,8 @@ OBJS2= test-noccs.o \
rg_locks-noccs.o \
event_config-noccs.o
+OBJS4= cpglockd.o
+
CFLAGS += -DSHAREDIR=\"${sharedir}\" -D_GNU_SOURCE
CFLAGS += -fPIC
CFLAGS += -I${ccsincdir} -I${cmanincdir} -I${dlmincdir} -I${logtincdir}
@@ -57,6 +60,7 @@ CCS_LDFLAGS += -L${ccslibdir} -lccs
CMAN_LDFLAGS += -L${cmanlibdir} -lcman
LOGSYS_LDFLAGS += -L${logtlibdir} -llogthread
DLM_LDFLAGS += -L${dlmlibdir} -ldlm
+CPG_LDFLAGS += -lcpg
XML2_LDFLAGS += `xml2-config --libs`
SLANG_LDFLAGS += -L${slanglibdir} -lslang
EXTRA_LDFLAGS += -lpthread
@@ -96,6 +100,10 @@ ${TARGET2}: ${OBJS2} ${LDDEPS}
${TARGET3}: ${TARGET1}
ln -sf ${TARGET1} ${TARGET3}
+${TARGET4}: ${OBJS4} ${LDDEPS}
+ $(CC) -o $@ $^ $(CPG_LDFLAGS) $(EXTRA_LDFLAGS) $(LDFLAGS)
+
+
check: rg_test
cd tests && ./runtests.sh
diff --git a/rgmanager/src/daemons/cpglockd.c b/rgmanager/src/daemons/cpglockd.c
new file mode 100644
index 0000000..55d832a
--- /dev/null
+++ b/rgmanager/src/daemons/cpglockd.c
@@ -0,0 +1,1147 @@
+#include <stdio.h>
+#include <signal.h>
+#include <sys/types.h>
+#include <sys/time.h>
+#include <sys/select.h>
+#include <errno.h>
+#include <unistd.h>
+#include <malloc.h>
+#include <string.h>
+#include <time.h>
+#include <sys/uio.h>
+#include <corosync/cpg.h>
+
+#include "sock.h"
+#include "cpglock.h"
+#include "cpglock-internal.h"
+#include "list.h"
+
+
+struct request_node {
+ list_head();
+ struct cpg_lock l;
+};
+
+struct lock_node {
+ list_head();
+ struct cpg_lock l;
+};
+
+struct client_node {
+ list_head();
+ int fd;
+ int pid;
+};
+
+struct member_node {
+ list_head();
+ int nodeid;
+};
+
+struct msg_node {
+ list_head();
+ struct cpg_lock_msg m;
+};
+
+/* Local vars */
+static cpg_handle_t cpg;
+static uint32_t my_node_id = 0;
+static struct request_node *requests = NULL;
+static struct lock_node *locks = NULL;
+static struct client_node *clients = NULL;
+static struct member_node *group_members = NULL;
+static struct msg_node *messages = NULL;
+static int total_members = 0;
+static int local_lockid = 0;
+static int message_count = 0;
+static int joined = 0;
+
+
+static const char *
+ls2str(int x)
+{
+ switch(x){
+ case LOCK_FREE: return "FREE";
+ case LOCK_HELD: return "HELD";
+ case LOCK_PENDING: return "PENDING";
+ }
+ return "unknown";
+}
+
+
+static const char *
+rq2str(int x)
+{
+ switch(x){
+ case MSG_LOCK: return "LOCK";
+ case MSG_UNLOCK: return "UNLOCK";
+ case MSG_GRANT: return "GRANT";
+ case MSG_NAK: return "NAK";
+ case MSG_PURGE: return "PURGE";
+ case MSG_CONFCHG: return "CONFCHG";
+ case MSG_JOIN: return "JOIN";
+ case MSG_HALT: return "HALT";
+ }
+ return "unknown";
+}
+
+
+static void
+dump_state(FILE *fp)
+{
+ struct request_node *r = NULL;
+ struct lock_node *l = NULL;
+ struct client_node *c = NULL;
+ struct member_node *m = NULL;
+ struct msg_node *s = NULL;
+ int x;
+
+ fprintf(fp, "cpglockd state\n");
+ fprintf(fp, "======== =====\n");
+
+ fprintf(fp, "Node ID: %d\n", my_node_id);
+
+ if (group_members) {
+ fprintf(fp, "Participants:");
+ list_for(&group_members, m, x) {
+ fprintf(fp, " %d", m->nodeid);
+ }
+ fprintf(fp, "\n");
+ }
+ if (clients) {
+ fprintf(fp, "Clients:");
+ list_for(&clients, c, x) {
+ fprintf(fp, " %d.%d", c->pid, c->fd );
+ }
+ fprintf(fp, "\n");
+ }
+ fprintf(fp, "\n");
+
+ if (locks) {
+ fprintf(fp, "Locks\n");
+ fprintf(fp, "=====\n");
+ list_for(&locks, l, x) {
+ fprintf(fp, " %s: %s", l->l.resource, ls2str(l->l.state));
+ if (l->l.owner_nodeid) {
+ fprintf(fp, ", owner %d:%d:%d", l->l.owner_nodeid,l->l.owner_pid, l->l.owner_tid);
+ if (l->l.owner_nodeid == my_node_id &&
+ l->l.state == LOCK_HELD)
+ fprintf(fp, ", Local ID %d", l->l.local_id);
+ }
+ fprintf(fp, "\n");
+ }
+ fprintf(fp, "\n");
+ }
+ if (requests) {
+ fprintf(fp, "Requests\n");
+ fprintf(fp, "========\n");
+ list_for(&requests, r, x) {
+ fprintf(fp, " %s: %s", r->l.resource, rq2str(r->l.state));
+ if (r->l.owner_nodeid) {
+ fprintf(fp, ", from %d:%d:%d", r->l.owner_nodeid,r->l.owner_pid, r->l.owner_tid);
+ }
+ fprintf(fp, "\n");
+ }
+ fprintf(fp, "\n");
+ }
+ if (messages) {
+ fprintf(fp, "Message History\n");
+ fprintf(fp, "======= =======\n");
+ list_for(&messages, s, x) {
+ switch(s->m.request) {
+ case MSG_CONFCHG:
+ fprintf(fp, " CONFIG CHANGE\n");
+ break;
+ case MSG_PURGE:
+ fprintf(fp, " PURGE for %d:%d\n", s->m.owner_nodeid, s->m.owner_pid);
+ break;
+ case MSG_JOIN:
+ fprintf(fp, " JOIN %d\n", s->m.owner_nodeid);
+ break;
+ default:
+ fprintf(fp, " %s: %s %d:%d:%d\n", rq2str(s->m.request), s->m.resource, s->m.owner_nodeid, s->m.owner_pid, s->m.owner_tid);
+ break;
+ }
+ }
+ fprintf(fp, "\n");
+ }
+}
+
+
+static void
+old_msg(struct cpg_lock_msg *m)
+{
+ struct msg_node *n;
+
+ n = do_alloc(sizeof(*n));
+ memcpy(&n->m, m, sizeof(n->m));
+ list_append(&messages, n);
+ if (message_count < 10) {
+ ++message_count;
+ } else {
+ n = messages;
+ list_remove(&messages, n);
+ free(n);
+ }
+}
+
+
+static void
+insert_client(int fd)
+{
+ struct client_node *n = NULL;
+
+ n = do_alloc(sizeof(*n));
+ n->fd = fd;
+ list_append(&clients, n);
+}
+
+
+/* forward request from client */
+static int
+send_lock_msg(struct cpg_lock_msg *m)
+{
+ struct iovec iov;
+
+ iov.iov_base = m;
+ iov.iov_len = sizeof (*m);
+
+ return cpg_mcast_joined(cpg, CPG_TYPE_AGREED, &iov, 1);
+}
+
+
+/* forward request from client */
+static int
+send_lock(struct cpg_lock_msg *m)
+{
+ m->owner_nodeid = my_node_id;
+
+ return send_lock_msg(m);
+}
+
+
+static int
+send_grant(struct request_node *n)
+{
+ struct cpg_lock_msg m;
+
+ printf("-> sending grant %s to %d:%d:%d\n",
+ n->l.resource, n->l.owner_nodeid, n->l.owner_pid, n->l.owner_tid);
+
+ memset(&m, 0, sizeof(m));
+ strncpy(m.resource, n->l.resource, sizeof(m.resource));
+ m.request = MSG_GRANT;
+ m.owner_nodeid = n->l.owner_nodeid;
+ m.owner_pid = n->l.owner_pid;
+ m.owner_tid = n->l.owner_tid;
+
+ return send_lock_msg(&m);
+}
+
+
+static int
+send_nak(struct cpg_lock_msg *m)
+{
+ m->request = MSG_NAK;
+
+ return send_lock_msg(m);
+}
+
+
+static int
+send_join(void)
+{
+ struct cpg_lock_msg m;
+
+ m.request = MSG_JOIN;
+ m.owner_nodeid = my_node_id;
+ return send_lock_msg(&m);
+}
+
+
+
+static int
+send_unlock(struct cpg_lock_msg *m)
+{
+ m->request = MSG_UNLOCK;
+ return send_lock_msg(m);
+}
+
+
+/*
+ * Grant the lock in this request node to the next
+ * waiting client.
+ */
+static int
+grant_next(struct cpg_lock_msg *m)
+{
+ struct request_node *r;
+ int x;
+
+ list_for(&requests, r, x) {
+ if (strcmp(m->resource, r->l.resource))
+ continue;
+
+ /* Send grant */
+ if (r->l.state == LOCK_PENDING) {
+ printf("LOCK %s: grant to %d:%d:%d\n", m->resource,
+ r->l.owner_nodeid, r->l.owner_pid, r->l.owner_tid);
+ /* don't send dup grants */
+ r->l.state = LOCK_HELD;
+ send_grant(r);
+ }
+ return 1;
+ }
+
+ return 0;
+}
+
+
+static void
+purge_requests(uint32_t nodeid, uint32_t pid)
+{
+ struct request_node *r;
+ int found = 0, count = 0, x = 0;
+
+ do {
+ found = 0;
+ list_for(&requests, r, x) {
+ if (r->l.owner_nodeid != nodeid ||
+ (pid &&
+ r->l.owner_pid != pid))
+ continue;
+
+ list_remove(&requests, r);
+ free(r);
+ found = 1;
+ ++count;
+ break;
+ }
+ } while (found);
+
+ if (count) {
+ if (pid) {
+ printf("RECOVERY: purged %d requests from %d:%d\n", count, nodeid, pid);
+ } else {
+ printf("RECOVERY: purged %d requests from node %d\n", count, nodeid);
+ }
+ }
+}
+
+
+
+static void
+del_client(int fd)
+{
+ struct cpg_lock_msg m;
+ struct client_node *n;
+ struct lock_node *l;
+ int x, pid = 0, recovered = 0;
+
+ list_for(&clients, n, x) {
+ if (n->fd == fd) {
+ list_remove(&clients, n);
+ close(n->fd);
+ pid = n->pid;
+ free(n);
+ break;
+ }
+ }
+
+ if (!pid)
+ return;
+
+ printf("RECOVERY: Looking for locks held by PID %d\n", pid);
+
+ /* This may not be needed */
+ purge_requests(my_node_id, pid);
+
+ m.request = MSG_PURGE;
+ m.owner_nodeid = my_node_id;
+ m.owner_pid = pid;
+
+ send_lock_msg(&m);
+
+ list_for(&locks, l, x) {
+ if (l->l.owner_nodeid != my_node_id ||
+ l->l.owner_pid != pid ||
+ l->l.state != LOCK_HELD)
+ continue;
+
+ printf("RECOVERY: Releasing %s \n", l->l.resource);
+
+ l->l.state = LOCK_FREE;
+ strncpy(m.resource, l->l.resource, sizeof(m.resource));
+ if (grant_next(&m) == 0)
+ send_unlock(&m);
+ }
+
+ if (recovered) {
+ printf("RECOVERY: %d locks from local PID %d\n", recovered, pid);
+ }
+ printf("RECOVERY: Complete\n");
+}
+
+
+static void
+del_node(uint32_t nodeid)
+{
+ struct cpg_lock_msg m;
+ struct lock_node *l;
+ int x, recovered = 0, granted = 0;
+
+ if (group_members->nodeid != my_node_id)
+ return;
+
+ printf("RECOVERY: I am oldest node in the group, recovering locks\n");
+
+ /* pass 1: purge outstanding requests from this node. */
+
+ /* This may not be needed */
+ purge_requests(nodeid, 0);
+
+ m.request = MSG_PURGE;
+ m.owner_nodeid = nodeid;
+ m.owner_pid = 0;
+
+ send_lock_msg(&m);
+
+ list_for(&locks, l, x) {
+ if (l->l.owner_nodeid == nodeid &&
+ l->l.state == LOCK_HELD) {
+ printf("RECOVERY: Releasing %s held by dead node %d\n", l->l.resource,
+ nodeid);
+
+ l->l.state = LOCK_FREE;
+ strncpy(m.resource, l->l.resource, sizeof(m.resource));
+ if (grant_next(&m) == 0)
+ send_unlock(&m);
+ ++recovered;
+ } else if (l->l.state == LOCK_FREE) {
+ if (grant_next(&m) == 0)
+ send_unlock(&m);
+ ++granted;
+ }
+ }
+
+ if (recovered) {
+ printf("RECOVERY: %d locks from node %d\n", recovered, nodeid);
+ }
+ if (granted) {
+ printf("RECOVERY: %d pending locks granted\n", granted);
+ }
+
+ printf("RECOVERY: Complete\n");
+}
+
+
+static int
+client_fdset(fd_set *set)
+{
+ int max = -1, x = 0;
+ struct client_node *n;
+
+ FD_ZERO(set);
+
+ list_for(&clients, n, x) {
+ FD_SET(n->fd, set);
+ if (n->fd > max)
+ max = n->fd;
+ }
+
+ if (!x)
+ return 0;
+
+ return max;
+}
+
+
+static struct client_node *
+find_client(int pid)
+{
+ int x;
+ struct client_node *n;
+
+ list_for(&clients, n, x) {
+ if (n->pid == pid)
+ return n;
+ }
+
+ return NULL;
+}
+
+
+#if 0
+static void
+send_fault(const char *resource)
+{
+ struct cpg_lock_msg m;
+
+ strncpy(m.resource, resource, sizeof(m.resource));
+ m.request = MSG_HALT;
+ m.owner_pid = 0;
+ m.owner_nodeid = my_node_id;
+
+ send_lock_msg(&m);
+}
+#endif
+
+
+static int
+grant_client(struct lock_node *l)
+{
+ struct client_node *c;
+ struct cpg_lock_msg m;
+
+ //memset(&m, 0, sizeof(m));
+ strncpy(m.resource, l->l.resource, sizeof(m.resource));
+ m.request = MSG_GRANT;
+ m.owner_pid = l->l.owner_pid;
+ m.owner_tid = l->l.owner_tid;
+ l->l.local_id = ++local_lockid;
+ m.lockid = l->l.local_id;
+ m.owner_nodeid = my_node_id;
+
+ c = find_client(l->l.owner_pid);
+ if (!c) {
+ printf("can't find client for pid %d\n", l->l.owner_pid);
+ return 1;
+ }
+
+ if (c->fd < 0) {
+ printf(" Client has bad fd\n");
+ return -1;
+ }
+
+ if (write_retry(c->fd, &m, sizeof(m), NULL) < 0) {
+ /* no client anymore; drop and send to next guy XXX */
+ /* This should be handled by our main loop */
+ //printf("Failed to notify client!\n");
+ }
+
+ return 0;
+}
+
+
+static int
+nak_client(struct request_node *l)
+{
+ struct client_node *c;
+ struct cpg_lock_msg m;
+
+ strncpy(m.resource, l->l.resource, sizeof(m.resource));
+ m.request = MSG_NAK;
+ m.owner_pid = l->l.owner_pid;
+ m.owner_tid = l->l.owner_tid;
+ m.owner_nodeid = my_node_id;
+
+ c = find_client(l->l.owner_pid);
+ if (!c) {
+ printf("can't find client for pid %d\n", l->l.owner_pid);
+ return 1;
+ }
+
+ if (c->fd < 0) {
+ printf(" Client has bad fd\n");
+ return -1;
+ }
+
+ if (write_retry(c->fd, &m, sizeof(m), NULL) < 0) {
+ /* no client anymore; drop and send to next guy XXX */
+ /* This should be handled by our main loop */
+ //printf("Failed to notify client!\n");
+ }
+
+ return 0;
+}
+
+static int
+queue_request(struct cpg_lock_msg *m)
+{
+ struct request_node *r;
+
+ r = do_alloc(sizeof(*r));
+ strncpy(r->l.resource, m->resource, sizeof(r->l.resource));
+ r->l.owner_nodeid = m->owner_nodeid;
+ r->l.owner_pid = m->owner_pid;
+ r->l.owner_tid = m->owner_tid;
+ r->l.state = LOCK_PENDING;
+
+ list_insert(&requests, r);
+ return 0;
+}
+
+
+static int
+process_lock(struct cpg_lock_msg *m)
+{
+ struct lock_node *l;
+ int x;
+
+ if (!joined)
+ return 0;
+
+ printf("LOCK %s: queue for %d:%d:%d\n", m->resource,
+ m->owner_nodeid, m->owner_pid, m->owner_tid);
+ queue_request(m);
+
+ list_for(&locks, l, x) {
+ if (strcmp(m->resource, l->l.resource))
+ continue;
+
+ /* if it's owned locally, we need send a
+ GRANT to the first node on the request queue */
+ if (l->l.owner_nodeid == my_node_id) {
+ if (l->l.state == LOCK_FREE) {
+ /* Set local state to PENDING to avoid double-grants */
+ l->l.state = LOCK_PENDING;
+ grant_next(m);
+ } else {
+ /* state is PENDING or HELD */
+ if (m->flags & FL_TRY) {
+ /* nack to client if needed */
+ send_nak(m);
+ }
+ }
+ }
+
+
+ return 0;
+ }
+
+ l = do_alloc(sizeof(*l));
+ strncpy(l->l.resource, m->resource, sizeof(l->l.resource));
+ l->l.state = LOCK_FREE;
+ list_insert(&locks, l);
+
+ if (group_members->nodeid == my_node_id) {
+ /* Allocate a lock structure and immediately grant */
+ l->l.state = LOCK_PENDING;
+ if (grant_next(m) == 0)
+ l->l.state = LOCK_FREE;
+ }
+
+ return 0;
+}
+
+
+static int
+is_member(uint32_t nodeid)
+{
+ struct member_node *n;
+ int x;
+
+ list_for(&group_members, n, x) {
+ if (n->nodeid == nodeid)
+ return 1;
+ }
+
+ return 0;
+}
+
+
+static int
+process_grant(struct cpg_lock_msg *m, uint32_t nodeid)
+{
+ struct lock_node *l;
+ struct request_node *r;
+ int x, y;
+
+ if (!joined)
+ return 0;
+
+ list_for(&locks, l, x) {
+ if (strcmp(m->resource, l->l.resource))
+ continue;
+
+ if (l->l.state == LOCK_HELD) {
+ if (m->owner_pid == 0 ||
+ m->owner_nodeid == 0) {
+ printf("GRANT averted\n");
+ return 0;
+ }
+ } else {
+ l->l.state = LOCK_HELD;
+ }
+
+ printf("GRANT %s: to %d:%d:%d\n",
+ m->resource, m->owner_nodeid,
+ m->owner_pid, m->owner_tid);
+
+ l->l.owner_nodeid = m->owner_nodeid;
+ l->l.owner_pid = m->owner_pid;
+ l->l.owner_tid = m->owner_tid;
+
+ list_for(&requests, r, y) {
+ if (strcmp(r->l.resource, m->resource))
+ continue;
+
+ if (r->l.owner_nodeid == m->owner_nodeid &&
+ r->l.owner_pid == m->owner_pid &&
+ r->l.owner_tid == m->owner_tid) {
+ list_remove(&requests, r);
+ free(r);
+ break;
+ }
+ }
+
+ /* granted lock */
+ if (l->l.owner_nodeid == my_node_id) {
+ if (grant_client(l) != 0) {
+ /* Grant to a nonexistent PID can
+ happen because we may have a pending
+ request after a fd was closed.
+ since we process on delivery, we
+ now simply make an unlock request
+ and move on */
+ purge_requests(my_node_id, l->l.owner_pid);
+ if (grant_next(m) == 0)
+ send_unlock(m);
+ return 0;
+ }
+ }
+
+ /* What if node has died with a GRANT in flight? */
+ if (group_members->nodeid == my_node_id &&
+ !is_member(l->l.owner_nodeid)) {
+
+ printf("GRANT to non-member %d; giving to next requestor\n", l->l.owner_nodeid);
+
+ l->l.state = LOCK_FREE;
+ if (grant_next(m) == 0)
+ send_unlock(m);
+ return 0;
+ }
+ return 0;
+ }
+
+ /* Record lock state since we now know it */
+ /* Allocate a lock structure */
+ l = do_alloc(sizeof(*m));
+ strncpy(l->l.resource, m->resource, sizeof(l->l.resource));
+ l->l.state = LOCK_HELD;
+ l->l.owner_nodeid = m->owner_nodeid;
+ l->l.owner_pid = m->owner_pid;
+ l->l.owner_tid = m->owner_tid;
+ list_insert(&locks, l);
+
+ return 0;
+}
+
+
+static int
+process_nak(struct cpg_lock_msg *m, uint32_t nodeid)
+{
+ struct request_node *r = NULL;
+ int y;
+
+ if (!joined)
+ return 0;
+
+ list_for(&requests, r, y) {
+ if (strcmp(r->l.resource, m->resource))
+ continue;
+
+ if (r->l.owner_nodeid == m->owner_nodeid &&
+ r->l.owner_pid == m->owner_pid &&
+ r->l.owner_tid == m->owner_tid) {
+ list_remove(&requests, r);
+ if (r->l.owner_nodeid == my_node_id) {
+ if (nak_client(r) != 0) {
+ purge_requests(my_node_id, r->l.owner_pid);
+ }
+ }
+ free(r);
+ break;
+ }
+ }
+
+ return 0;
+}
+
+
+static int
+process_unlock(struct cpg_lock_msg *m, uint32_t nodeid)
+{
+ struct lock_node *l;
+ int x;
+
+ if (!joined)
+ return 0;
+
+ list_for(&locks, l, x) {
+ if (l->l.state != LOCK_HELD)
+ continue;
+ if (strcmp(m->resource, l->l.resource))
+ continue;
+
+ /* Held lock... if it's local, we need send a
+ GRANT to the first node on the request queue */
+ if (l->l.owner_nodeid == m->owner_nodeid &&
+ l->l.owner_pid == m->owner_pid) {
+ printf("UNLOCK %s: %d:%d:%d\n", m->resource, m->owner_nodeid, m->owner_pid, m->owner_tid);
+ l->l.state = LOCK_FREE;
+ if (grant_next(m) != 0)
+ l->l.state = LOCK_PENDING;
+ }
+ }
+
+ return 0;
+}
+
+
+static int
+find_lock(struct cpg_lock_msg *m)
+{
+ struct lock_node *l;
+ int x;
+
+ if (m->resource[0] != 0)
+ return 0;
+
+ list_for(&locks, l, x) {
+ if (m->lockid == l->l.local_id) {
+ strncpy(m->resource, l->l.resource, sizeof(m->resource));
+ printf("LOCK %d -> %s\n", m->lockid, m->resource);
+ m->owner_nodeid = l->l.owner_nodeid;
+ m->owner_pid = l->l.owner_pid;
+ m->owner_tid = l->l.owner_tid;
+ m->lockid = 0;
+ return 0;
+ }
+ }
+
+ return 1;
+}
+
+
+static int
+process_join(struct cpg_lock_msg *m, uint32_t nodeid)
+{
+ struct member_node *n;
+ int x;
+
+ list_for(&group_members, n, x) {
+ if (n->nodeid == nodeid) {
+ list_remove(&group_members, n);
+ list_append(&group_members, n);
+ printf("JOIN: moving %d to back\n", nodeid);
+ return 0;
+ }
+ }
+
+ n = do_alloc(sizeof(*n));
+ n->nodeid = nodeid;
+ printf("JOIN: node %d", n->nodeid);
+ if (nodeid == my_node_id) {
+ printf(" (self)");
+ joined = 1;
+ }
+ total_members++;
+ printf("\n");
+ list_insert(&group_members, n);
+
+ return 0;
+}
+
+
+static int
+process_request(struct cpg_lock_msg *m, uint32_t nodeid)
+{
+ if (m->request == MSG_HALT) {
+ printf("FAULT: Halting operations; see node %d\n", m->owner_nodeid);
+ while (1)
+ sleep(30);
+ }
+
+ old_msg(m);
+
+ switch (m->request) {
+ case MSG_LOCK:
+ process_lock(m);
+ break;
+ case MSG_NAK:
+ process_nak(m, nodeid);
+ break;
+ case MSG_GRANT:
+ process_grant(m, nodeid);
+ break;
+ case MSG_UNLOCK:
+ process_unlock(m, nodeid);
+ break;
+ case MSG_PURGE:
+ purge_requests(m->owner_nodeid, m->owner_pid);
+ break;
+ case MSG_JOIN:
+ process_join(m, nodeid);
+ break;
+ }
+
+ return 0;
+}
+
+
+static void
+cpg_deliver_func(cpg_handle_t h,
+ const struct cpg_name *group_name,
+ uint32_t nodeid,
+ uint32_t pid,
+ void *msg,
+ size_t msglen)
+{
+
+ if (msglen != sizeof(struct cpg_lock_msg)) {
+ printf("Invalid message size %d\n", (int)msglen);
+ }
+
+ process_request((struct cpg_lock_msg *)msg, nodeid);
+}
+
+
+static void
+cpg_config_change(cpg_handle_t h,
+ const struct cpg_name *group_name,
+ const struct cpg_address *members, size_t memberlen,
+ const struct cpg_address *left, size_t leftlen,
+ const struct cpg_address *join, size_t joinlen)
+{
+ struct member_node *n;
+ size_t x, y;
+ struct cpg_lock_msg m;
+
+
+ memset(&m, 0, sizeof(m));
+ strncpy(m.resource, "(none)", sizeof(m.resource));
+ m.request = MSG_CONFCHG;
+
+ old_msg(&m);
+
+ if (total_members == 0) {
+
+ printf("JOIN: Setting up initial node list\n");
+ for (x = 0; x < memberlen; x++) {
+ for (y = 0; y < joinlen; y++) {
+ if (join[y].nodeid == members[x].nodeid)
+ continue;
+ if (members[x].nodeid == my_node_id)
+ continue;
+
+ n = do_alloc(sizeof(*n));
+ n->nodeid = members[x].nodeid;
+ printf("JOIN: node %d\n", n->nodeid);
+ list_insert(&group_members, n);
+ }
+ }
+ printf("JOIN: Done\n");
+
+ total_members = memberlen;
+ }
+
+ //printf("members %d now, %d joined, %d left\n", memberlen, joinlen, leftlen);
+#if 0
+
+ /* XXX process join on receipt of JOIN message rather than here
+ since ordered delivery is agreed, this prevents >1 member from
+ believing it is the oldest host */
+ for (x = 0; x < joinlen; x++) {
+ n = do_alloc(sizeof(*n));
+ n->nodeid = join[x].nodeid;
+ printf("ADD: node %d\n", n->nodeid);
+ list_insert(&group_members, n);
+ }
+#endif
+
+ for (x = 0; x < leftlen; x++) {
+
+ list_for(&group_members, n, y) {
+ if (n->nodeid == left[x].nodeid) {
+ list_remove(&group_members, n);
+ printf("DELETE: node %d\n", n->nodeid);
+ del_node(n->nodeid);
+ free(n);
+ break;
+ }
+ }
+
+ total_members -= leftlen;
+ if (total_members < 0)
+ total_members = 0;
+ }
+
+#if 0
+ printf("MEMBERS:");
+ list_for(&group_members, n, y) {
+ printf(" %d", n->nodeid);
+ }
+ printf("\n");
+#endif
+
+ return;
+}
+
+
+static cpg_callbacks_t my_callbacks = {
+ .cpg_deliver_fn = cpg_deliver_func,
+ .cpg_confchg_fn = cpg_config_change
+};
+
+
+static int
+cpg_init(void)
+{
+ struct cpg_name gname;
+
+ errno = EINVAL;
+
+ gname.length = snprintf(gname.value,
+ sizeof(gname.value),
+ CPG_LOCKD_NAME);
+ if (gname.length >= sizeof(gname.value)) {
+ errno = ENAMETOOLONG;
+ return -1;
+ }
+
+ if (gname.length <= 0)
+ return -1;
+
+ memset(&cpg, 0, sizeof(cpg));
+ if (cpg_initialize(&cpg, &my_callbacks) != CPG_OK) {
+ perror("cpg_initialize");
+ return -1;
+ }
+
+ if (cpg_join(cpg, &gname) != CPG_OK) {
+ perror("cpg_join");
+ return -1;
+ }
+
+ cpg_local_get(cpg, &my_node_id);
+
+ return 0;
+}
+
+
+static int
+cpg_fin(void)
+{
+ struct cpg_name gname;
+
+ errno = EINVAL;
+
+ gname.length = snprintf(gname.value,
+ sizeof(gname.value),
+ CPG_LOCKD_NAME);
+ if (gname.length >= sizeof(gname.value)) {
+ errno = ENAMETOOLONG;
+ return -1;
+ }
+
+ if (gname.length <= 0)
+ return -1;
+
+ cpg_leave(cpg, &gname);
+ cpg_finalize(cpg);
+
+ return 0;
+}
+
+
+int
+main(int argc, char **argv)
+{
+ fd_set rfds;
+ int fd;
+ int cpgfd;
+ int afd = -1;
+ int n,x;
+
+ struct cpg_lock l;
+ struct cpg_lock_msg m;
+ struct client_node *client;
+
+ l.state = 0;
+
+ signal(SIGPIPE, SIG_IGN);
+
+ fd = sock_listen(CPG_LOCKD_SOCK);
+ cpg_init();
+ cpg_local_get(cpg, &my_node_id);
+ cpg_fd_get(cpg, &cpgfd);
+ if (send_join() < 0)
+ return -1;
+
+ while (1) {
+ FD_ZERO(&rfds);
+ x = client_fdset(&rfds);
+ FD_SET(fd, &rfds);
+ if (fd > x)
+ x = fd;
+ FD_SET(cpgfd, &rfds);
+ if (cpgfd > x)
+ x = cpgfd;
+
+ n = select_retry(x+1, &rfds, NULL, NULL, NULL);
+ if (n < 0) {
+ perror("select");
+ return -1;
+ }
+
+ if (FD_ISSET(fd, &rfds)) {
+ afd = accept(fd, NULL, NULL);
+ insert_client(afd);
+ --n;
+ }
+
+ if (FD_ISSET(cpgfd, &rfds)) {
+ cpg_dispatch(cpg, CPG_DISPATCH_ONE);
+ --n;
+ }
+
+ if (n <= 0)
+ continue;
+
+ do {
+ list_for(&clients, client, x) {
+ if (!FD_ISSET(client->fd, &rfds))
+ continue;
+ --n;
+ if (read_retry(client->fd, &m, sizeof(m), NULL) < 0) {
+ printf("Closing client fd %d pid %d: %d\n",
+ client->fd, client->pid, errno);
+
+ del_client(client->fd);
+ break;
+ }
+
+ /* send lock request */
+ /* XXX check for dup connection */
+ if (m.request == MSG_LOCK) {
+ client->pid = m.owner_pid;
+ send_lock(&m);
+ }
+
+ if (m.request == MSG_UNLOCK) {
+ //printf("Unlock from fd %d\n", client->fd);
+ find_lock(&m);
+ if (grant_next(&m) == 0)
+ send_unlock(&m);
+ }
+
+ if (m.request == MSG_DUMP) {
+ FILE *fp = fdopen(client->fd, "w");
+
+ list_remove(&clients, client);
+ dump_state(fp);
+ fclose(fp);
+ close(client->fd);
+ free(client);
+ break;
+ }
+ }
+ } while (n);
+ }
+
+ cpg_fin();
+
+ return 0;
+}
diff --git a/rgmanager/src/daemons/main.c b/rgmanager/src/daemons/main.c
index 94047c3..29af894 100644
--- a/rgmanager/src/daemons/main.c
+++ b/rgmanager/src/daemons/main.c
@@ -983,14 +983,14 @@ int
main(int argc, char **argv)
{
int rv, do_init = 1;
- char foreground = 0, wd = 1;
+ char foreground = 0, wd = 1, cpg_locks = 0;
cman_node_t me;
msgctx_t *cluster_ctx;
msgctx_t *local_ctx;
pthread_t th;
cman_handle_t clu = NULL;
- while ((rv = getopt(argc, argv, "wfdNq")) != EOF) {
+ while ((rv = getopt(argc, argv, "wfdNqC")) != EOF) {
switch (rv) {
case 'w':
wd = 0;
@@ -1007,6 +1007,9 @@ main(int argc, char **argv)
case 'q':
rgm_dbus_notify = 0;
break;
+ case 'C':
+ cpg_locks = 1;
+ break;
default:
return 1;
break;
@@ -1053,10 +1056,19 @@ main(int argc, char **argv)
return -1;
}
- if (clu_lock_init(rgmanager_lsname) != 0) {
- printf("Locks not working!\n");
- cman_finish(clu);
- return -1;
+ if (!cpg_locks) {
+ if (clu_lock_init(rgmanager_lsname) != 0) {
+ printf("Locks not working!\n");
+ cman_finish(clu);
+ return -1;
+ }
+ } else {
+ if (cpg_lock_initialize() != 0) {
+ printf("Locks not working!\n");
+ cman_finish(clu);
+ return -1;
+ }
+ logt_print(LOG_INFO, "Using CPG for locking (EXPERIMENTAL)\n");
}
memset(&me, 0, sizeof(me));
12 years, 1 month
cluster: STABLE32 - rgmanager: Add man page for cpglockd; update rgmanager.8
by Lon Hohberger
Gitweb: http://git.fedorahosted.org/git/cluster.git?p=cluster.git;a=commitdiff;h=...
Commit: 7e44784a83001516ab02ec66495fc1b83255ea3a
Parent: 26c3db58dac85cabef6001b9b9abc3a089b77c26
Author: Lon Hohberger <lhh(a)redhat.com>
AuthorDate: Tue May 1 10:18:58 2012 -0400
Committer: Lon Hohberger <lhh(a)redhat.com>
CommitterDate: Tue May 1 10:28:00 2012 -0400
rgmanager: Add man page for cpglockd; update rgmanager.8
Signed-off-by: Lon Hohberger <lhh(a)redhat.com>
---
rgmanager/man/cpglockd.8 | 21 +++++++++++++++++++++
rgmanager/man/rgmanager.8 | 7 ++++++-
2 files changed, 27 insertions(+), 1 deletions(-)
diff --git a/rgmanager/man/cpglockd.8 b/rgmanager/man/cpglockd.8
new file mode 100644
index 0000000..8084d96
--- /dev/null
+++ b/rgmanager/man/cpglockd.8
@@ -0,0 +1,21 @@
+.TH "cpglockd" "8" "May 2012" "" "Red Hat High Availability"
+.SH "NAME"
+cpglockd \- CPG lock server for rgmanager
+.SH "DESCRIPTION"
+.PP
+.B cpglockd
+utilizes the extended virtual synchrony features of the Corosync
+Cluster Engine to implement a simplistic, distributed lock server
+for rgmanager.
+.SH "COMMAND LINE OPTIONS"
+.IP \-F
+Don't wait for the current node to join the fencing domain at startup
+.IP \-Q
+Don't wait for quorum formation at startup
+.IP \-f
+Operate in the foreground mode; do not daemonize
+.IP \-h
+Print command line usage.
+
+.SH "SEE ALSO"
+rgmanager(8), corosync(8)
diff --git a/rgmanager/man/rgmanager.8 b/rgmanager/man/rgmanager.8
index 2b018d2..bab52ef 100644
--- a/rgmanager/man/rgmanager.8
+++ b/rgmanager/man/rgmanager.8
@@ -380,8 +380,13 @@ Do not perform stop-before-start. Combined with the
.I -Z
flag to clusvcadm, this can be used to allow rgmanager to be upgraded
without stopping a given user service or set of services.
+.IP \-C [0|1]
+Explicitly disable or enable CPG-based locking. The default is to
+enable this when RRP is turned on (which requires a cluster outage).
+This option MUST be the same on all hosts in the cluster and must
+only be enabled or disabled with all instances of rgmanager turned off.
.SH "SEE ALSO"
http://sources.redhat.com/cluster/wiki/RGManager
-clusvcadm(8), cluster.conf(5)
+clusvcadm(8), cluster.conf(5), cpglockd(8)
12 years, 1 month