4:43 p.m.
sanlock_reg_lockspace() returns an fd that the caller should
use in poll(2), and when ready should call sanlock_get_callback()
to read the callback.
Currently, the only callback type is host message. When a host
receives a message from another host, the host message is sent
to the callback fd.
Signed-off-by: David Teigland <teigland(a)redhat.com>
---
src/client.c | 80 +++++++++++++++++++++++++++++++++++
src/cmd.c | 31 ++++++++++++++
src/lockspace.c | 48 ++++++++++++++++++++-
src/lockspace.h | 7 ++--
src/main.c | 1 +
src/resource.c | 51 ++++++++++++++++++++++-
src/resource.h | 2 +-
src/sanlock.h | 19 +++++++++
src/sanlock_admin.h | 4 ++
src/sanlock_internal.h | 1 +
src/sanlock_sock.h | 3 ++
tests/sanlk_cb.c | 110 +++++++++++++++++++++++++++++++++++++++++++++++++
12 files changed, 350 insertions(+), 7 deletions(-)
create mode 100644 tests/sanlk_cb.c
diff --git a/src/client.c b/src/client.c
index 01d56b22706d..9e53f8b7aa2d 100644
--- a/src/client.c
+++ b/src/client.c
@@ -844,6 +844,86 @@ int sanlock_set_message(const char *ls_name, uint32_t flags,
return rv;
}
+int sanlock_reg_lockspace(struct sanlk_lockspace *ls, uint32_t flags)
+{
+ struct sm_header h;
+ int rv, fd;
+
+ if (!ls->name[0])
+ return -EINVAL;
+
+ rv = connect_socket(&fd);
+ if (rv < 0)
+ return rv;
+
+ rv = send_header(fd, SM_CMD_REG_LOCKSPACE, flags, 0, 0, 0);
+ if (rv < 0) {
+ close(fd);
+ return rv;
+ }
+
+ rv = send(fd, ls, sizeof(struct sanlk_lockspace), 0);
+ if (rv < 0) {
+ close(fd);
+ return rv;
+ }
+
+ memset(&h, 0, sizeof(struct sm_header));
+
+ rv = recv(fd, &h, sizeof(h), MSG_WAITALL);
+ if (rv < 0) {
+ rv = -errno;
+ close(fd);
+ return rv;
+ }
+
+ if (rv != sizeof(h)) {
+ rv = -1;
+ close(fd);
+ return rv;
+ }
+
+ rv = (int)h.data;
+ if (rv < 0) {
+ close(fd);
+ return rv;
+ }
+
+ return fd;
+}
+
+int sanlock_get_callback(int fd, uint32_t flags GNUC_UNUSED,
+ struct sanlk_callback *cb, int cb_size)
+{
+ struct sm_header h;
+ struct sanlk_callback callback;
+ int rv, len;
+
+ rv = recv(fd, &h, sizeof(h), MSG_WAITALL);
+ if (rv < 0)
+ return -errno;
+
+ if (rv != sizeof(h))
+ return -1;
+
+ len = h.length - sizeof(h);
+
+ if (len > sizeof(callback))
+ return -EINVAL;
+
+ memset(&callback, 0, sizeof(callback));
+
+ rv = recv(fd, &callback, len, MSG_WAITALL);
+ if (rv < 0)
+ return -errno;
+
+ if (len > cb_size)
+ len = cb_size;
+
+ memcpy(cb, &callback, len);
+ return 0;
+}
+
/* old api */
int sanlock_init(struct sanlk_lockspace *ls,
struct sanlk_resource *res,
diff --git a/src/cmd.c b/src/cmd.c
index 727d782e4764..a22fec5f8b6c 100644
--- a/src/cmd.c
+++ b/src/cmd.c
@@ -2443,6 +2443,33 @@ static void cmd_version(int ci GNUC_UNUSED, int fd, struct
sm_header *h_recv)
send(fd, h_recv, sizeof(struct sm_header), MSG_NOSIGNAL);
}
+static void cmd_reg_lockspace(int fd, struct sm_header *h_recv)
+{
+ struct sm_header h;
+ struct sanlk_lockspace lockspace;
+ int rv;
+
+ log_debug("cmd_reg_lockspace fd %d", fd);
+
+ memcpy(&h, h_recv, sizeof(struct sm_header));
+ h.version = SM_PROTO;
+ h.length = sizeof(struct sm_header);
+
+ rv = recv(fd, &lockspace, sizeof(struct sanlk_lockspace), MSG_WAITALL);
+ if (rv != sizeof(struct sanlk_lockspace)) {
+ h.data = -ENOTCONN;
+ goto out;
+ }
+
+ rv = register_lockspace_fd(&lockspace, fd);
+
+ h.data = rv;
+ h.data2 = fd;
+out:
+ log_debug("cmd_reg_lockspace fd %d rv %d", fd, rv);
+ send(fd, &h, sizeof(struct sm_header), MSG_NOSIGNAL);
+}
+
static int get_peer_pid(int fd, int *pid)
{
struct ucred cred;
@@ -2531,6 +2558,10 @@ void call_cmd_daemon(int ci, struct sm_header *h_recv, int
client_maxi)
strcpy(client[ci].owner_name, "get_hosts");
cmd_get_hosts(fd, h_recv);
break;
+ case SM_CMD_REG_LOCKSPACE:
+ strcpy(client[ci].owner_name, "reg_lockspace");
+ cmd_reg_lockspace(fd, h_recv);
+ break;
};
if (auto_close)
diff --git a/src/lockspace.c b/src/lockspace.c
index c57497fa4a03..28181a475e07 100644
--- a/src/lockspace.c
+++ b/src/lockspace.c
@@ -111,6 +111,23 @@ int lockspace_info(const char *space_name, struct space_info *spi)
return rv;
}
+int lockspace_callback_fd(int space_id)
+{
+ struct space *sp;
+ int fd = -ENOENT;
+
+ pthread_mutex_lock(&spaces_mutex);
+ list_for_each_entry(sp, &spaces, list) {
+ if (sp->space_id != space_id)
+ continue;
+ fd = sp->callback_fd;
+ break;
+ }
+ pthread_mutex_unlock(&spaces_mutex);
+
+ return fd;
+}
+
int lockspace_disk(char *space_name, struct sync_disk *disk)
{
struct space *sp;
@@ -371,7 +388,7 @@ void check_other_leases(struct space *sp, char *buf)
(unsigned long long)hs->owner_generation,
hm.msg, hm.seq);
- set_message_examine(sp->space_name, &hm,
+ set_message_examine(sp->space_id, &hm,
hs->owner_id, hs->owner_generation);
@@ -670,6 +687,9 @@ static void *lockspace_thread(void *arg_in)
if (opened)
close(sp->host_id_disk.fd);
+ if (sp->callback_fd != -1)
+ close(sp->callback_fd);
+
close_task_aio(&task);
return NULL;
}
@@ -705,6 +725,7 @@ int add_lockspace_start(struct sanlk_lockspace *ls, uint32_t
io_timeout, struct
sp->host_id_disk.fd = -1;
sp->host_id = ls->host_id;
sp->io_timeout = io_timeout;
+ sp->callback_fd = -1;
pthread_mutex_init(&sp->mutex, NULL);
pthread_mutex_lock(&spaces_mutex);
@@ -1246,6 +1267,31 @@ int set_lockspace_message(struct sanlk_lockspace *ls, struct
sanlk_host_message
return 0;
}
+int register_lockspace_fd(struct sanlk_lockspace *ls, int fd)
+{
+ struct space *sp;
+
+ if (!ls->name[0])
+ return -EINVAL;
+
+ pthread_mutex_lock(&spaces_mutex);
+ sp = _search_space(ls->name, NULL, 0, &spaces, NULL, NULL, NULL);
+ if (!sp) {
+ pthread_mutex_unlock(&spaces_mutex);
+ return -ENOENT;
+ }
+ pthread_mutex_unlock(&spaces_mutex);
+
+ pthread_mutex_lock(&sp->mutex);
+ if (sp->callback_fd != -1)
+ close(sp->callback_fd);
+ sp->callback_fd = dup(fd);
+ log_space(sp, "register cb fd %d from fd %d", sp->callback_fd, fd);
+ pthread_mutex_unlock(&sp->mutex);
+
+ return 0;
+}
+
/*
* we call stop_host_id() when all pids are gone and we're in a safe state, so
* it's safe to unlink the watchdog right away here. We want to sp the unlink
diff --git a/src/lockspace.h b/src/lockspace.h
index 0189d00fd524..2b6996ee2ff7 100644
--- a/src/lockspace.h
+++ b/src/lockspace.h
@@ -6,8 +6,8 @@
* of the GNU General Public License v2 or (at your option) any later version.
*/
-#ifndef __HOST_ID_H__
-#define __HOST_ID__H__
+#ifndef __LOCKSPACE_H__
+#define __LOCKSPACE_H__
struct space *find_lockspace(const char *name);
int _lockspace_info(const char *space_name, struct space_info *spi);
@@ -29,6 +29,7 @@ int get_lockspaces(char *buf, int *len, int *count, int maxlen);
int get_hosts(struct sanlk_lockspace *ls, char *buf, int *len, int *count, int maxlen);
void host_message_from_extra(struct sanlk_host_message *hm, struct delta_extra *extra);
int set_lockspace_message(struct sanlk_lockspace *ls, struct sanlk_host_message *hm);
-
+int register_lockspace_fd(struct sanlk_lockspace *ls, int fd);
+int lockspace_callback_fd(int space_id);
#endif
diff --git a/src/main.c b/src/main.c
index a693652a85fc..279c95214368 100644
--- a/src/main.c
+++ b/src/main.c
@@ -1178,6 +1178,7 @@ static void process_connection(int ci)
case SM_CMD_LOG_DUMP:
case SM_CMD_GET_LOCKSPACES:
case SM_CMD_GET_HOSTS:
+ case SM_CMD_REG_LOCKSPACE:
call_cmd_daemon(ci, &h, client_maxi);
break;
case SM_CMD_ADD_LOCKSPACE:
diff --git a/src/resource.c b/src/resource.c
index 3c499b5afeda..467b72c3977b 100644
--- a/src/resource.c
+++ b/src/resource.c
@@ -22,8 +22,11 @@
#include <signal.h>
#include <sys/types.h>
#include <sys/time.h>
+#include <sys/socket.h>
+#include <sys/un.h>
#include "sanlock_internal.h"
+#include "sanlock_sock.h"
#include "diskio.h"
#include "log.h"
#include "paxos_lease.h"
@@ -1697,13 +1700,13 @@ int set_resource_examine(char *space_name, char *res_name)
struct recv_message {
struct list_head list;
+ uint32_t space_id;
uint64_t from_host_id;
uint64_t from_generation;
struct sanlk_host_message hm;
};
-void set_message_examine(char *space_name GNUC_UNUSED,
- struct sanlk_host_message *hm,
+void set_message_examine(uint32_t space_id, struct sanlk_host_message *hm,
uint64_t from_host_id, uint64_t from_generation)
{
struct recv_message *rm;
@@ -1716,6 +1719,7 @@ void set_message_examine(char *space_name GNUC_UNUSED,
memset(rm, 0, sizeof(struct recv_message));
memcpy(&rm->hm, hm, sizeof(struct sanlk_host_message));
+ rm->space_id = space_id;
rm->from_host_id = from_host_id;
rm->from_generation = from_generation;
@@ -1726,6 +1730,47 @@ void set_message_examine(char *space_name GNUC_UNUSED,
pthread_mutex_unlock(&resource_mutex);
}
+static void send_callback(struct recv_message *rm)
+{
+ struct sm_header h;
+ struct sanlk_callback_hm cb;
+ int fd, rv;
+
+ fd = lockspace_callback_fd(rm->space_id);
+ if (fd < 0)
+ return;
+
+ memset(&h, 0, sizeof(h));
+ memset(&cb, 0, sizeof(cb));
+
+ h.magic = SM_MAGIC;
+ h.version = SM_CB_PROTO;
+ h.cmd = SM_CMD_CALLBACK;
+ h.length = sizeof(h) + sizeof(cb);
+
+ cb.type = SANLK_CB_HOST_MESSAGE;
+ cb.from_host_id = rm->from_host_id;
+ cb.from_generation = rm->from_generation;
+ cb.host_id = rm->hm.host_id;
+ cb.generation = rm->hm.generation;
+ cb.msg = rm->hm.msg;
+ cb.seq = rm->hm.seq;
+
+ rv = send(fd, &h, sizeof(h), MSG_NOSIGNAL | MSG_DONTWAIT);
+ if (rv == -EAGAIN) {
+ log_error("close callback fd %d sp %d", fd, rm->space_id);
+ close(fd);
+ return;
+ }
+
+ rv = send(fd, &cb, sizeof(cb), MSG_NOSIGNAL | MSG_DONTWAIT);
+ if (rv == -EAGAIN) {
+ log_error("close callback fd %d sp %d", fd, rm->space_id);
+ close(fd);
+ return;
+ }
+}
+
static void resource_thread_messages(void)
{
struct recv_message *rm, *safe;
@@ -1738,6 +1783,8 @@ static void resource_thread_messages(void)
(unsigned long long)rm->from_generation,
rm->hm.msg, rm->hm.seq);
+ send_callback(rm);
+
if (rm->hm.msg & SANLK_HM_WD_RESET) {
log_error("Host reset request from host_id %llu gen %llu",
(unsigned long long)rm->from_host_id,
diff --git a/src/resource.h b/src/resource.h
index ce97884b1c90..17b6376a18b6 100644
--- a/src/resource.h
+++ b/src/resource.h
@@ -27,7 +27,7 @@ int request_token(struct task *task, struct token *token, uint32_t
force_mode,
int set_resource_examine(char *space_name, char *res_name);
-void set_message_examine(char *space_name, struct sanlk_host_message *hm,
+void set_message_examine(uint32_t space_id, struct sanlk_host_message *hm,
uint64_t from_host_id, uint64_t from_generation);
int res_set_lvb(struct sanlk_resource *res, char *lvb, int lvblen);
diff --git a/src/sanlock.h b/src/sanlock.h
index b95d010ac97b..8e662d5b4a96 100644
--- a/src/sanlock.h
+++ b/src/sanlock.h
@@ -115,6 +115,25 @@ struct sanlk_host_message {
uint32_t ack_seq;
};
+#define SANLK_CB_HOST_MESSAGE 0x00000001
+
+struct sanlk_callback_hm {
+ uint32_t type;
+ uint32_t flags;
+ uint32_t msg;
+ uint32_t seq;
+ uint64_t host_id;
+ uint64_t generation;
+ uint64_t from_host_id;
+ uint64_t from_generation;
+};
+
+struct sanlk_callback {
+ union {
+ struct sanlk_callback_hm hm;
+ };
+};
+
size_t sanlock_path_export(char *dst, const char *src, size_t dstlen);
size_t sanlock_path_import(char *dst, const char *src, size_t dstlen);
diff --git a/src/sanlock_admin.h b/src/sanlock_admin.h
index ee00e478027b..8464a7ce7b2f 100644
--- a/src/sanlock_admin.h
+++ b/src/sanlock_admin.h
@@ -32,6 +32,10 @@
#define SANLK_HOST_DEAD 0x00000005
#define SANLK_HOST_MASK 0x0000000F /* select SANLK_HOST_ from flags */
+int sanlock_reg_lockspace(struct sanlk_lockspace *ls, uint32_t flags);
+
+int sanlock_get_callback(int fd, uint32_t flags, struct sanlk_callback *cb, int
cb_size);
+
/*
* add_lockspace returns:
* 0: the lockspace has been added successfully
diff --git a/src/sanlock_internal.h b/src/sanlock_internal.h
index 414d3f841eb0..966395902508 100644
--- a/src/sanlock_internal.h
+++ b/src/sanlock_internal.h
@@ -170,6 +170,7 @@ struct space {
int external_remove;
int thread_stop;
int wd_fd;
+ int callback_fd;
pthread_t thread;
pthread_mutex_t mutex; /* protects lease_status, thread_stop */
struct lease_status lease_status;
diff --git a/src/sanlock_sock.h b/src/sanlock_sock.h
index b61023720130..42c8d45cad65 100644
--- a/src/sanlock_sock.h
+++ b/src/sanlock_sock.h
@@ -15,6 +15,7 @@
#define SM_MAGIC 0x04282010
#define SM_PROTO 0x00000001
+#define SM_CB_PROTO 0x00000001
#define MAX_CLIENT_MSG (1024 * 1024) /* TODO: this is random */
@@ -49,6 +50,8 @@ enum {
SM_CMD_VERSION = 28,
SM_CMD_SET_MESSAGE = 29,
SM_CMD_READ_LOCKSPACE_MESSAGE = 30,
+ SM_CMD_REG_LOCKSPACE = 31,
+ SM_CMD_CALLBACK = 32,
};
struct sm_header {
diff --git a/tests/sanlk_cb.c b/tests/sanlk_cb.c
new file mode 100644
index 000000000000..d2961b14344f
--- /dev/null
+++ b/tests/sanlk_cb.c
@@ -0,0 +1,110 @@
+#include <inttypes.h>
+#include <unistd.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <stdint.h>
+#include <stddef.h>
+#include <fcntl.h>
+#include <string.h>
+#include <errno.h>
+#include <limits.h>
+#include <time.h>
+#include <signal.h>
+#include <poll.h>
+#include <sys/types.h>
+#include <sys/wait.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#include "sanlock.h"
+#include "sanlock_admin.h"
+#include "../src/sanlock_sock.h"
+
+int main(int argc, char *argv[])
+{
+ struct sanlk_lockspace ls;
+ struct sm_header h;
+ struct sanlk_callback cb;
+ struct pollfd pollfd;
+ int fd, rv;
+
+ if (argc < 2) {
+ printf("sanlk_cb <lockspace_name>\n");
+ return -1;
+ }
+
+ memset(&ls, 0, sizeof(ls));
+
+ strcpy(ls.name, argv[1]);
+
+ fd = sanlock_reg_lockspace(&ls, 0);
+ if (fd < 0) {
+ printf("reg error %d\n", fd);
+ return -1;
+ }
+
+ printf("sanlock_reg_lockspace fd %d\n", fd);
+
+ memset(&pollfd, 0, sizeof(pollfd));
+ pollfd.fd = fd;
+ pollfd.events = POLLIN;
+
+ while (1) {
+ rv = poll(&pollfd, 1, -1);
+ if (rv == -1 && errno == EINTR)
+ continue;
+
+ if (rv < 0) {
+ printf("poll error %d\n", rv);
+ exit(0);
+ }
+
+ if (pollfd.revents & POLLIN) {
+#if 0
+ rv = recv(fd, &h, sizeof(h), MSG_WAITALL);
+ if (rv < 0) {
+ printf("recv h %d %d\n", rv, errno);
+ return -errno;
+ }
+ if (rv != sizeof(h)) {
+ printf("recv rv %d\n", rv);
+ return -1;
+ }
+
+ printf("h m %x v %x c %d l %d\n",
+ h.magic, h.version, h.cmd, h.length);
+
+ rv = recv(fd, &cb, sizeof(cb), MSG_WAITALL);
+ if (rv < 0) {
+ printf("recv cb %d %d\n", rv, errno);
+ return -errno;
+ }
+ if (rv != sizeof(cb)) {
+ printf("recv rv %d\n", rv);
+ return -1;
+ }
+#endif
+
+ rv = sanlock_get_callback(fd, 0, &cb, sizeof(cb));
+
+ if (cb.hm.type != SANLK_CB_HOST_MESSAGE) {
+ printf("unknown cb type %d\n", cb.hm.type);
+ continue;
+ }
+
+ printf("host message from host_id %llu gen %llu\n",
+ (unsigned long long)cb.hm.from_host_id,
+ (unsigned long long)cb.hm.from_generation);
+ printf("msg 0x%08x seq 0x%08x\n",
+ cb.hm.msg, cb.hm.seq);
+ }
+
+ if (pollfd.revents & (POLLERR | POLLHUP | POLLNVAL)) {
+ printf("poll revents %x\n", pollfd.revents);
+ exit(0);
+ }
+ }
+
+ return 0;
+}
+
--
1.8.3.1