add pmxcfs restart detection heuristic for IPCC

[pve-cluster.git] / data / PVE / IPCC.xs

diff --git a/data/PVE/IPCC.xs b/data/PVE/IPCC.xs
index ccadc33288e58334d068afad0293afc5b0cc6287..decc60b721bb9ba8c92fadccf2140294872a78a1 100644 (file)
--- a/data/PVE/IPCC.xs
+++ b/data/PVE/IPCC.xs
@@ -8,29 +8,86 @@
 
 #include "ppport.h"
 
+/* sendfd: BSD style file descriptor passing over unix domain sockets
+ *  Richard Stevens: Unix Network Programming, Prentice Hall, 1990;
+ */
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/uio.h>
+#include <unistd.h>
+#include <time.h>
+#include <errno.h>
+
+#ifndef SCM_RIGHTS
+#error "SCM_RIGHTS undefined"
+#endif 
+
+/* interface to pmxcfs (libqb) */
 #include <sys/syslog.h>
 #include <qb/qbdefs.h>
 #include <qb/qbutil.h>
+#include <qb/qblog.h>
 #include <qb/qbipcc.h>
 
+#define RESTART_FLAG_FILE "/run/pve-cluster/cfs-restart-flag"
+#define RESTART_GRACE_PERIOD 5
+
 #define PCS_SOCKET_NAME "pve2"
 
 #define PCS_SERVICE1 1
 #define MAX_MSG_SIZE (8192*128)
 
 static qb_ipcc_connection_t *conn;
+static pid_t conn_pid;
+
 static char ipcbuffer[MAX_MSG_SIZE];
 
-static void libqb_log_writer(const char *file_name,
-                            int32_t file_line,
-                            int32_t severity, const char *msg)
-{
-       if (severity == LOG_DEBUG)
-               return;
+static qb_ipcc_connection_t *init_connection() {
+
+       static qb_ipcc_connection_t *connection = NULL;
+       struct timespec retry_timeout, now;
+       int cfs_restart_flag_fd = -1;
+
+       // check if pmxcfs is currently restarting
+       if ((cfs_restart_flag_fd = open(RESTART_FLAG_FILE, 0)) > 0) {
+               clock_gettime(CLOCK_MONOTONIC, &retry_timeout);
+               retry_timeout.tv_sec += RESTART_GRACE_PERIOD;
+       }
+
+       qb_log_init("IPCC.xs", LOG_USER, LOG_EMERG);
+       qb_log_ctl(QB_LOG_SYSLOG, QB_LOG_CONF_ENABLED, QB_TRUE);
+
+retry_connection:
+       connection = qb_ipcc_connect(PCS_SOCKET_NAME, MAX_MSG_SIZE);
+
+       if (!connection) {
+               if (cfs_restart_flag_fd >= 0) {
+                       // cfs restarting and hopefully back soon, poll
+                       clock_gettime(CLOCK_MONOTONIC, &now);
 
-       warn("libqb: %s:%d [%d] %s\n", file_name, file_line, severity, msg);
+                       if (now.tv_sec < retry_timeout.tv_sec ||
+                          (now.tv_sec == retry_timeout.tv_sec &&
+                           now.tv_nsec < retry_timeout.tv_nsec)) {
+
+                               usleep(100 * 1000);
+                               goto retry_connection;
+
+                       } else {
+                               // timeout: cleanup flag file if still the same
+                               struct stat s;
+                               fstat(cfs_restart_flag_fd, &s);
+                               if (s.st_nlink > 0)
+                                       unlink(RESTART_FLAG_FILE);
+                       }
+               }
+       }
+
+       if (cfs_restart_flag_fd >= 0) close(cfs_restart_flag_fd);
+
+       return connection;
 }
 
+
 MODULE = PVE::IPCC             PACKAGE = PVE::IPCC             
 
 SV *
@@ -40,12 +97,24 @@ SV * data;
 PROTOTYPE: $;$
 CODE:
 {
+       uint8_t retried_cache_connection = 0;
+       pid_t cpid = getpid();
+
+       /* Each process needs its own ipcc connection,
+        * else the shared memory buffer gets corrupted.
+        */ 
+       if (conn && conn_pid != cpid) {
+               conn = NULL;
+       }
+
        if (conn == NULL) {
-               qb_util_set_log_function(libqb_log_writer);
-               conn = qb_ipcc_connect(PCS_SOCKET_NAME, MAX_MSG_SIZE);
+recache_connection:
+               conn = init_connection();
 
                if (!conn)
                        XSRETURN_UNDEF;
+
+               conn_pid = cpid;
        }
 
        size_t len = 0;
@@ -71,6 +140,11 @@ CODE:
        if (res < 0) {
                qb_ipcc_disconnect(conn);
                conn = NULL;
+               // requests during cfs restart and the first thereafter will fail, retry
+               if (!retried_cache_connection) {
+                       retried_cache_connection = 1;
+                       goto recache_connection;
+               }
                errno = -res;
                XSRETURN_UNDEF;
        }
@@ -84,13 +158,69 @@ CODE:
                errno = -res_header->error;
                XSRETURN_UNDEF;
        } else {
+               errno = 0;
                if (dsize > 0) {
                        RETVAL = newSVpv(ipcbuffer + sizeof(struct qb_ipc_response_header), dsize);
                } else {
-                       errno = 0;
                        XSRETURN_UNDEF;
                }
        }
 }
 OUTPUT: RETVAL
 
+# helper to pass SCM ACCESS RIGHTS
+
+int
+sendfd(sock_fd, send_me_fd, data=NULL)
+int sock_fd
+int send_me_fd
+SV * data;
+CODE:
+{
+       int ret = 0;
+       struct iovec  iov[1];
+       struct msghdr msg;
+       memset(&msg, 0, sizeof(msg));
+
+       size_t len = 0;
+       char *dataptr = NULL;
+       if (data && SvPOK(data))
+               dataptr = SvPV(data, len);
+       
+       iov[0].iov_base = dataptr;
+       iov[0].iov_len = len;   
+       msg.msg_iov = iov;
+       msg.msg_iovlen = 1;
+       msg.msg_name = 0;
+       msg.msg_namelen = 0;
+
+       char control[CMSG_SPACE(sizeof(int))];
+       memset(control, 0, sizeof(control));
+
+       msg.msg_control = control;
+       msg.msg_controllen = sizeof(control);
+       msg.msg_flags = 0;
+               
+       struct cmsghdr* h = CMSG_FIRSTHDR(&msg);
+       h->cmsg_len = CMSG_LEN(sizeof(int));
+       h->cmsg_level= SOL_SOCKET;
+       h->cmsg_type = SCM_RIGHTS;
+       *((int*)CMSG_DATA(h)) = send_me_fd;
+
+       int repeat;
+       do {
+               repeat = 0;
+               ret = sendmsg(sock_fd, &msg, 0);
+               if (ret < 0) {
+                       if (errno == EINTR) {
+                               repeat = 1;
+                       } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
+                               repeat = 1;
+                               usleep(1000);
+                       }
+               }
+       } while (repeat);
+       
+       RETVAL = ret;
+}
+OUTPUT: RETVAL