data/PVE/IPCC.xs

   1 #ifdef HAVE_CONFIG_H
   2 #include <config.h>
   3 #endif /* HAVE_CONFIG_H */
   4
   5 #include "EXTERN.h"
   6 #include "perl.h"
   7 #include "XSUB.h"
   8
   9 #include "ppport.h"
  10
  11 /* sendfd: BSD style file descriptor passing over unix domain sockets
  12  *  Richard Stevens: Unix Network Programming, Prentice Hall, 1990;
  13  */
  14 #include <sys/types.h>
  15 #include <sys/socket.h>
  16 #include <sys/uio.h>
  17 #include <unistd.h>
  18 #include <time.h>
  19 #include <errno.h>
  20
  21 #ifndef SCM_RIGHTS
  22 #error "SCM_RIGHTS undefined"
  23 #endif
  24
  25 /* interface to pmxcfs (libqb) */
  26 #include <sys/syslog.h>
  27 #include <qb/qbdefs.h>
  28 #include <qb/qbutil.h>
  29 #include <qb/qblog.h>
  30 #include <qb/qbipcc.h>
  31
  32 #define RESTART_FLAG_FILE "/run/pve-cluster/cfs-restart-flag"
  33 #define RESTART_GRACE_PERIOD 10
  34
  35 #define PCS_SOCKET_NAME "pve2"
  36
  37 #define PCS_SERVICE1 1
  38 #define MAX_MSG_SIZE (8192*128)
  39
  40 static qb_ipcc_connection_t *conn;
  41 static pid_t conn_pid;
  42
  43 static char ipcbuffer[MAX_MSG_SIZE];
  44
  45 static qb_ipcc_connection_t *init_connection() {
  46
  47         static qb_ipcc_connection_t *connection = NULL;
  48         struct timespec retry_timeout, now;
  49         int cfs_restart_flag_fd = -1;
  50
  51         // check if pmxcfs is currently restarting
  52         if ((cfs_restart_flag_fd = open(RESTART_FLAG_FILE, 0)) > 0) {
  53                 clock_gettime(CLOCK_MONOTONIC, &retry_timeout);
  54                 retry_timeout.tv_sec += RESTART_GRACE_PERIOD;
  55         }
  56
  57         qb_log_init("IPCC.xs", LOG_USER, LOG_EMERG);
  58         qb_log_ctl(QB_LOG_SYSLOG, QB_LOG_CONF_ENABLED, QB_TRUE);
  59
  60 retry_connection:
  61         connection = qb_ipcc_connect(PCS_SOCKET_NAME, MAX_MSG_SIZE);
  62
  63         if (!connection) {
  64                 if (cfs_restart_flag_fd >= 0) {
  65                         // cfs restarting and hopefully back soon, poll
  66                         clock_gettime(CLOCK_MONOTONIC, &now);
  67
  68                         if (now.tv_sec < retry_timeout.tv_sec ||
  69                            (now.tv_sec == retry_timeout.tv_sec &&
  70                             now.tv_nsec < retry_timeout.tv_nsec)) {
  71
  72                                 usleep(100 * 1000);
  73                                 goto retry_connection;
  74
  75                         } else {
  76                                 // timeout: cleanup flag file if still the same
  77                                 struct stat s;
  78                                 fstat(cfs_restart_flag_fd, &s);
  79                                 if (s.st_nlink > 0)
  80                                         unlink(RESTART_FLAG_FILE);
  81                         }
  82                 }
  83         }
  84
  85         if (cfs_restart_flag_fd >= 0) close(cfs_restart_flag_fd);
  86
  87         return connection;
  88 }
  89
  90
  91 MODULE = PVE::IPCC              PACKAGE = PVE::IPCC
  92
  93 SV *
  94 ipcc_send_rec(msgid, data=NULL)
  95 I32 msgid;
  96 SV * data;
  97 PROTOTYPE: $;$
  98 CODE:
  99 {
 100         uint8_t retried_cache_connection = 0;
 101         pid_t cpid = getpid();
 102
 103         /* Each process needs its own ipcc connection,
 104          * else the shared memory buffer gets corrupted.
 105          */
 106         if (conn && conn_pid != cpid) {
 107                 conn = NULL;
 108         }
 109
 110         if (conn == NULL) {
 111 recache_connection:
 112                 conn = init_connection();
 113
 114                 if (!conn)
 115                         XSRETURN_UNDEF;
 116
 117                 conn_pid = cpid;
 118         }
 119
 120         size_t len = 0;
 121         char *dataptr = NULL;
 122         if (data && SvPOK(data))
 123                 dataptr = SvPV(data, len);
 124
 125         int iov_len = 2;
 126         struct iovec iov[iov_len];
 127
 128         struct qb_ipc_request_header req_header;
 129
 130         req_header.id = msgid;
 131         req_header.size = sizeof(req_header) + len;
 132
 133         iov[0].iov_base = (char *)&req_header;
 134         iov[0].iov_len = sizeof(req_header);
 135         iov[1].iov_base = dataptr;
 136         iov[1].iov_len = len;
 137
 138         int32_t ms_timeout = -1; // fixme:
 139         int res = qb_ipcc_sendv_recv(conn, iov, iov_len, ipcbuffer, sizeof(ipcbuffer), ms_timeout);
 140         if (res < 0) {
 141                 qb_ipcc_disconnect(conn);
 142                 conn = NULL;
 143                 // requests during cfs restart and the first thereafter will fail, retry
 144                 if (!retried_cache_connection) {
 145                         retried_cache_connection = 1;
 146                         goto recache_connection;
 147                 }
 148                 errno = -res;
 149                 XSRETURN_UNDEF;
 150         }
 151
 152         struct qb_ipc_response_header *res_header;
 153
 154         res_header = (struct qb_ipc_response_header *)ipcbuffer;
 155         int dsize = res_header->size - sizeof(struct qb_ipc_response_header);
 156
 157         if (res_header->error < 0) {
 158                 errno = -res_header->error;
 159                 XSRETURN_UNDEF;
 160         } else {
 161                 errno = 0;
 162                 if (dsize > 0) {
 163                         RETVAL = newSVpv(ipcbuffer + sizeof(struct qb_ipc_response_header), dsize);
 164                 } else {
 165                         XSRETURN_UNDEF;
 166                 }
 167         }
 168 }
 169 OUTPUT: RETVAL
 170
 171 # helper to pass SCM ACCESS RIGHTS
 172
 173 int
 174 sendfd(sock_fd, send_me_fd, data=NULL)
 175 int sock_fd
 176 int send_me_fd
 177 SV * data;
 178 CODE:
 179 {
 180         int ret = 0;
 181         struct iovec  iov[1];
 182         struct msghdr msg;
 183         memset(&msg, 0, sizeof(msg));
 184
 185         size_t len = 0;
 186         char *dataptr = NULL;
 187         if (data && SvPOK(data))
 188                 dataptr = SvPV(data, len);
 189
 190         iov[0].iov_base = dataptr;
 191         iov[0].iov_len = len;
 192         msg.msg_iov = iov;
 193         msg.msg_iovlen = 1;
 194         msg.msg_name = 0;
 195         msg.msg_namelen = 0;
 196
 197         char control[CMSG_SPACE(sizeof(int))];
 198         memset(control, 0, sizeof(control));
 199
 200         msg.msg_control = control;
 201         msg.msg_controllen = sizeof(control);
 202         msg.msg_flags = 0;
 203
 204         struct cmsghdr* h = CMSG_FIRSTHDR(&msg);
 205         h->cmsg_len = CMSG_LEN(sizeof(int));
 206         h->cmsg_level= SOL_SOCKET;
 207         h->cmsg_type = SCM_RIGHTS;
 208         *((int*)CMSG_DATA(h)) = send_me_fd;
 209
 210         int repeat;
 211         do {
 212                 repeat = 0;
 213                 ret = sendmsg(sock_fd, &msg, 0);
 214                 if (ret < 0) {
 215                         if (errno == EINTR) {
 216                                 repeat = 1;
 217                         } else if (errno == EAGAIN || errno == EWOULDBLOCK) {
 218                                 repeat = 1;
 219                                 usleep(1000);
 220                         }
 221                 }
 222         } while (repeat);
 223
 224         RETVAL = ret;
 225 }
 226 OUTPUT: RETVAL