1 /* $NetBSD: clvmd.c,v 1.1.1.3 2009/12/02 00:27:05 haad Exp $ */
2
3 /*
4 * Copyright (C) 2002-2004 Sistina Software, Inc. All rights reserved.
5 * Copyright (C) 2004-2009 Red Hat, Inc. All rights reserved.
6 *
7 * This file is part of LVM2.
8 *
9 * This copyrighted material is made available to anyone wishing to use,
10 * modify, copy, or redistribute it subject to the terms and conditions
11 * of the GNU General Public License v.2.
12 *
13 * You should have received a copy of the GNU General Public License
14 * along with this program; if not, write to the Free Software Foundation,
15 * Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
16 */
17
18 /*
19 * CLVMD: Cluster LVM daemon
20 */
21
22 #define _GNU_SOURCE
23 #define _FILE_OFFSET_BITS 64
24
25 #include <configure.h>
26 #include <libdevmapper.h>
27
28 #include <pthread.h>
29 #include <sys/types.h>
30 #include <sys/stat.h>
31 #include <sys/socket.h>
32 #include <sys/uio.h>
33 #include <sys/un.h>
34 #include <sys/time.h>
35 #include <sys/ioctl.h>
36 #include <sys/utsname.h>
37 #include <netinet/in.h>
38 #include <stdio.h>
39 #include <stdlib.h>
40 #include <stddef.h>
41 #include <stdarg.h>
42 #include <signal.h>
43 #include <unistd.h>
44 #include <fcntl.h>
45 #include <getopt.h>
46 #include <syslog.h>
47 #include <errno.h>
48 #include <limits.h>
49 #ifdef HAVE_COROSYNC_CONFDB_H
50 #include <corosync/confdb.h>
51 #endif
52
53 #include "clvmd-comms.h"
54 #include "lvm-functions.h"
55 #include "clvm.h"
56 #include "lvm-version.h"
57 #include "clvmd.h"
58 #include "refresh_clvmd.h"
59 #include "lvm-logging.h"
60
61 #ifndef TRUE
62 #define TRUE 1
63 #endif
64 #ifndef FALSE
65 #define FALSE 0
66 #endif
67
68 #define MAX_RETRIES 4
69
70 #define ISLOCAL_CSID(c) (memcmp(c, our_csid, max_csid_len) == 0)
71
72 /* Head of the fd list. Also contains
73 the cluster_socket details */
74 static struct local_client local_client_head;
75
76 static unsigned short global_xid = 0; /* Last transaction ID issued */
77
78 struct cluster_ops *clops = NULL;
79
80 static char our_csid[MAX_CSID_LEN];
81 static unsigned max_csid_len;
82 static unsigned max_cluster_message;
83 static unsigned max_cluster_member_name_len;
84
85 /* Structure of items on the LVM thread list */
86 struct lvm_thread_cmd {
87 struct dm_list list;
88
89 struct local_client *client;
90 struct clvm_header *msg;
91 char csid[MAX_CSID_LEN];
92 int remote; /* Flag */
93 int msglen;
94 unsigned short xid;
95 };
96
97 debug_t debug;
98 static pthread_t lvm_thread;
99 static pthread_mutex_t lvm_thread_mutex;
100 static pthread_cond_t lvm_thread_cond;
101 static pthread_mutex_t lvm_start_mutex;
102 static struct dm_list lvm_cmd_head;
103 static volatile sig_atomic_t quit = 0;
104 static volatile sig_atomic_t reread_config = 0;
105 static int child_pipe[2];
106
107 /* Reasons the daemon failed initialisation */
108 #define DFAIL_INIT 1
109 #define DFAIL_LOCAL_SOCK 2
110 #define DFAIL_CLUSTER_IF 3
111 #define DFAIL_MALLOC 4
112 #define DFAIL_TIMEOUT 5
113 #define SUCCESS 0
114
115 typedef enum {IF_AUTO, IF_CMAN, IF_GULM, IF_OPENAIS, IF_COROSYNC} if_type_t;
116
117 typedef void *(lvm_pthread_fn_t)(void*);
118
119 /* Prototypes for code further down */
120 static void sigusr2_handler(int sig);
121 static void sighup_handler(int sig);
122 static void sigterm_handler(int sig);
123 static void send_local_reply(struct local_client *client, int status,
124 int clientid);
125 static void free_reply(struct local_client *client);
126 static void send_version_message(void);
127 static void *pre_and_post_thread(void *arg);
128 static int send_message(void *buf, int msglen, const char *csid, int fd,
129 const char *errtext);
130 static int read_from_local_sock(struct local_client *thisfd);
131 static int process_local_command(struct clvm_header *msg, int msglen,
132 struct local_client *client,
133 unsigned short xid);
134 static void process_remote_command(struct clvm_header *msg, int msglen, int fd,
135 const char *csid);
136 static int process_reply(const struct clvm_header *msg, int msglen,
137 const char *csid);
138 static int open_local_sock(void);
139 static int check_local_clvmd(void);
140 static struct local_client *find_client(int clientid);
141 static void main_loop(int local_sock, int cmd_timeout);
142 static void be_daemon(int start_timeout);
143 static int check_all_clvmds_running(struct local_client *client);
144 static int local_rendezvous_callback(struct local_client *thisfd, char *buf,
145 int len, const char *csid,
146 struct local_client **new_client);
147 static void lvm_thread_fn(void *) __attribute__ ((noreturn));
148 static int add_to_lvmqueue(struct local_client *client, struct clvm_header *msg,
149 int msglen, const char *csid);
150 static int distribute_command(struct local_client *thisfd);
151 static void hton_clvm(struct clvm_header *hdr);
152 static void ntoh_clvm(struct clvm_header *hdr);
153 static void add_reply_to_list(struct local_client *client, int status,
154 const char *csid, const char *buf, int len);
155 static if_type_t parse_cluster_interface(char *ifname);
156 static if_type_t get_cluster_type(void);
157
usage(char * prog,FILE * file)158 static void usage(char *prog, FILE *file)
159 {
160 fprintf(file, "Usage:\n");
161 fprintf(file, "%s [Vhd]\n", prog);
162 fprintf(file, "\n");
163 fprintf(file, " -V Show version of clvmd\n");
164 fprintf(file, " -h Show this help information\n");
165 fprintf(file, " -d Set debug level\n");
166 fprintf(file, " If starting clvmd then don't fork, run in the foreground\n");
167 fprintf(file, " -R Tell all running clvmds in the cluster to reload their device cache\n");
168 fprintf(file, " -C Sets debug level (from -d) on all clvmd instances clusterwide\n");
169 fprintf(file, " -t<secs> Command timeout (default 60 seconds)\n");
170 fprintf(file, " -T<secs> Startup timeout (default none)\n");
171 fprintf(file, " -I<cmgr> Cluster manager (default: auto)\n");
172 fprintf(file, " Available cluster managers: ");
173 #ifdef USE_COROSYNC
174 fprintf(file, "corosync ");
175 #endif
176 #ifdef USE_CMAN
177 fprintf(file, "cman ");
178 #endif
179 #ifdef USE_OPENAIS
180 fprintf(file, "openais ");
181 #endif
182 #ifdef USE_GULM
183 fprintf(file, "gulm ");
184 #endif
185 fprintf(file, "\n");
186 }
187
188 /* Called to signal the parent how well we got on during initialisation */
child_init_signal(int status)189 static void child_init_signal(int status)
190 {
191 if (child_pipe[1]) {
192 write(child_pipe[1], &status, sizeof(status));
193 close(child_pipe[1]);
194 }
195 if (status)
196 exit(status);
197 }
198
199
debuglog(const char * fmt,...)200 void debuglog(const char *fmt, ...)
201 {
202 time_t P;
203 va_list ap;
204 static int syslog_init = 0;
205
206 if (debug == DEBUG_STDERR) {
207 va_start(ap,fmt);
208 time(&P);
209 fprintf(stderr, "CLVMD[%x]: %.15s ", (int)pthread_self(), ctime(&P)+4 );
210 vfprintf(stderr, fmt, ap);
211 va_end(ap);
212 }
213 if (debug == DEBUG_SYSLOG) {
214 if (!syslog_init) {
215 openlog("clvmd", LOG_PID, LOG_DAEMON);
216 syslog_init = 1;
217 }
218
219 va_start(ap,fmt);
220 vsyslog(LOG_DEBUG, fmt, ap);
221 va_end(ap);
222 }
223 }
224
decode_cmd(unsigned char cmdl)225 static const char *decode_cmd(unsigned char cmdl)
226 {
227 static char buf[128];
228 const char *command;
229
230 switch (cmdl) {
231 case CLVMD_CMD_TEST:
232 command = "TEST";
233 break;
234 case CLVMD_CMD_LOCK_VG:
235 command = "LOCK_VG";
236 break;
237 case CLVMD_CMD_LOCK_LV:
238 command = "LOCK_LV";
239 break;
240 case CLVMD_CMD_REFRESH:
241 command = "REFRESH";
242 break;
243 case CLVMD_CMD_SET_DEBUG:
244 command = "SET_DEBUG";
245 break;
246 case CLVMD_CMD_GET_CLUSTERNAME:
247 command = "GET_CLUSTERNAME";
248 break;
249 case CLVMD_CMD_VG_BACKUP:
250 command = "VG_BACKUP";
251 break;
252 case CLVMD_CMD_REPLY:
253 command = "REPLY";
254 break;
255 case CLVMD_CMD_VERSION:
256 command = "VERSION";
257 break;
258 case CLVMD_CMD_GOAWAY:
259 command = "GOAWAY";
260 break;
261 case CLVMD_CMD_LOCK:
262 command = "LOCK";
263 break;
264 case CLVMD_CMD_UNLOCK:
265 command = "UNLOCK";
266 break;
267 case CLVMD_CMD_LOCK_QUERY:
268 command = "LOCK_QUERY";
269 break;
270 default:
271 command = "unknown";
272 break;
273 }
274
275 sprintf(buf, "%s (0x%x)", command, cmdl);
276
277 return buf;
278 }
279
main(int argc,char * argv[])280 int main(int argc, char *argv[])
281 {
282 int local_sock;
283 struct local_client *newfd;
284 struct utsname nodeinfo;
285 signed char opt;
286 int cmd_timeout = DEFAULT_CMD_TIMEOUT;
287 int start_timeout = 0;
288 if_type_t cluster_iface = IF_AUTO;
289 sigset_t ss;
290 int using_gulm = 0;
291 int debug_opt = 0;
292 int clusterwide_opt = 0;
293
294 /* Deal with command-line arguments */
295 opterr = 0;
296 optind = 0;
297 while ((opt = getopt(argc, argv, "?vVhd::t:RT:CI:")) != EOF) {
298 switch (opt) {
299 case 'h':
300 usage(argv[0], stdout);
301 exit(0);
302
303 case '?':
304 usage(argv[0], stderr);
305 exit(0);
306
307 case 'R':
308 return refresh_clvmd()==1?0:1;
309
310 case 'C':
311 clusterwide_opt = 1;
312 break;
313
314 case 'd':
315 debug_opt = 1;
316 if (optarg)
317 debug = atoi(optarg);
318 else
319 debug = DEBUG_STDERR;
320 break;
321
322 case 't':
323 cmd_timeout = atoi(optarg);
324 if (!cmd_timeout) {
325 fprintf(stderr, "command timeout is invalid\n");
326 usage(argv[0], stderr);
327 exit(1);
328 }
329 break;
330 case 'I':
331 cluster_iface = parse_cluster_interface(optarg);
332 break;
333 case 'T':
334 start_timeout = atoi(optarg);
335 if (start_timeout <= 0) {
336 fprintf(stderr, "startup timeout is invalid\n");
337 usage(argv[0], stderr);
338 exit(1);
339 }
340 break;
341
342 case 'V':
343 printf("Cluster LVM daemon version: %s\n", LVM_VERSION);
344 printf("Protocol version: %d.%d.%d\n",
345 CLVMD_MAJOR_VERSION, CLVMD_MINOR_VERSION,
346 CLVMD_PATCH_VERSION);
347 exit(1);
348 break;
349
350 }
351 }
352
353 /* Setting debug options on an existing clvmd */
354 if (debug_opt && !check_local_clvmd()) {
355
356 /* Sending to stderr makes no sense for a detached daemon */
357 if (debug == DEBUG_STDERR)
358 debug = DEBUG_SYSLOG;
359 return debug_clvmd(debug, clusterwide_opt)==1?0:1;
360 }
361
362 /* Fork into the background (unless requested not to) */
363 if (debug != DEBUG_STDERR) {
364 be_daemon(start_timeout);
365 }
366
367 DEBUGLOG("CLVMD started\n");
368
369 /* Open the Unix socket we listen for commands on.
370 We do this before opening the cluster socket so that
371 potential clients will block rather than error if we are running
372 but the cluster is not ready yet */
373 local_sock = open_local_sock();
374 if (local_sock < 0)
375 child_init_signal(DFAIL_LOCAL_SOCK);
376
377 /* Set up signal handlers, USR1 is for cluster change notifications (in cman)
378 USR2 causes child threads to exit.
379 HUP causes gulm version to re-read nodes list from CCS.
380 PIPE should be ignored */
381 signal(SIGUSR2, sigusr2_handler);
382 signal(SIGHUP, sighup_handler);
383 signal(SIGPIPE, SIG_IGN);
384
385 /* Block SIGUSR2/SIGINT/SIGTERM in process */
386 sigemptyset(&ss);
387 sigaddset(&ss, SIGUSR2);
388 sigaddset(&ss, SIGINT);
389 sigaddset(&ss, SIGTERM);
390 sigprocmask(SIG_BLOCK, &ss, NULL);
391
392 /* Initialise the LVM thread variables */
393 dm_list_init(&lvm_cmd_head);
394 pthread_mutex_init(&lvm_thread_mutex, NULL);
395 pthread_cond_init(&lvm_thread_cond, NULL);
396 pthread_mutex_init(&lvm_start_mutex, NULL);
397 init_lvhash();
398
399 /* Start the cluster interface */
400 if (cluster_iface == IF_AUTO)
401 cluster_iface = get_cluster_type();
402
403 #ifdef USE_CMAN
404 if ((cluster_iface == IF_AUTO || cluster_iface == IF_CMAN) && (clops = init_cman_cluster())) {
405 max_csid_len = CMAN_MAX_CSID_LEN;
406 max_cluster_message = CMAN_MAX_CLUSTER_MESSAGE;
407 max_cluster_member_name_len = CMAN_MAX_NODENAME_LEN;
408 syslog(LOG_NOTICE, "Cluster LVM daemon started - connected to CMAN");
409 }
410 #endif
411 #ifdef USE_GULM
412 if (!clops)
413 if ((cluster_iface == IF_AUTO || cluster_iface == IF_GULM) && (clops = init_gulm_cluster())) {
414 max_csid_len = GULM_MAX_CSID_LEN;
415 max_cluster_message = GULM_MAX_CLUSTER_MESSAGE;
416 max_cluster_member_name_len = GULM_MAX_CLUSTER_MEMBER_NAME_LEN;
417 using_gulm = 1;
418 syslog(LOG_NOTICE, "Cluster LVM daemon started - connected to GULM");
419 }
420 #endif
421 #ifdef USE_COROSYNC
422 if (!clops)
423 if (((cluster_iface == IF_AUTO || cluster_iface == IF_COROSYNC) && (clops = init_corosync_cluster()))) {
424 max_csid_len = COROSYNC_CSID_LEN;
425 max_cluster_message = COROSYNC_MAX_CLUSTER_MESSAGE;
426 max_cluster_member_name_len = COROSYNC_MAX_CLUSTER_MEMBER_NAME_LEN;
427 syslog(LOG_NOTICE, "Cluster LVM daemon started - connected to Corosync");
428 }
429 #endif
430 #ifdef USE_OPENAIS
431 if (!clops)
432 if ((cluster_iface == IF_AUTO || cluster_iface == IF_OPENAIS) && (clops = init_openais_cluster())) {
433 max_csid_len = OPENAIS_CSID_LEN;
434 max_cluster_message = OPENAIS_MAX_CLUSTER_MESSAGE;
435 max_cluster_member_name_len = OPENAIS_MAX_CLUSTER_MEMBER_NAME_LEN;
436 syslog(LOG_NOTICE, "Cluster LVM daemon started - connected to OpenAIS");
437 }
438 #endif
439
440 if (!clops) {
441 DEBUGLOG("Can't initialise cluster interface\n");
442 log_error("Can't initialise cluster interface\n");
443 child_init_signal(DFAIL_CLUSTER_IF);
444 }
445 DEBUGLOG("Cluster ready, doing some more initialisation\n");
446
447 /* Save our CSID */
448 uname(&nodeinfo);
449 clops->get_our_csid(our_csid);
450
451 /* Initialise the FD list head */
452 local_client_head.fd = clops->get_main_cluster_fd();
453 local_client_head.type = CLUSTER_MAIN_SOCK;
454 local_client_head.callback = clops->cluster_fd_callback;
455
456 /* Add the local socket to the list */
457 newfd = malloc(sizeof(struct local_client));
458 if (!newfd)
459 child_init_signal(DFAIL_MALLOC);
460
461 newfd->fd = local_sock;
462 newfd->removeme = 0;
463 newfd->type = LOCAL_RENDEZVOUS;
464 newfd->callback = local_rendezvous_callback;
465 newfd->next = local_client_head.next;
466 local_client_head.next = newfd;
467
468 /* This needs to be started after cluster initialisation
469 as it may need to take out locks */
470 DEBUGLOG("starting LVM thread\n");
471
472 /* Don't let anyone else to do work until we are started */
473 pthread_mutex_lock(&lvm_start_mutex);
474 pthread_create(&lvm_thread, NULL, (lvm_pthread_fn_t*)lvm_thread_fn,
475 (void *)(long)using_gulm);
476
477 /* Tell the rest of the cluster our version number */
478 /* CMAN can do this immediately, gulm needs to wait until
479 the core initialisation has finished and the node list
480 has been gathered */
481 if (clops->cluster_init_completed)
482 clops->cluster_init_completed();
483
484 DEBUGLOG("clvmd ready for work\n");
485 child_init_signal(SUCCESS);
486
487 /* Try to shutdown neatly */
488 signal(SIGTERM, sigterm_handler);
489 signal(SIGINT, sigterm_handler);
490
491 /* Do some work */
492 main_loop(local_sock, cmd_timeout);
493
494 destroy_lvm();
495
496 return 0;
497 }
498
499 /* Called when the GuLM cluster layer has completed initialisation.
500 We send the version message */
clvmd_cluster_init_completed()501 void clvmd_cluster_init_completed()
502 {
503 send_version_message();
504 }
505
506 /* Data on a connected socket */
local_sock_callback(struct local_client * thisfd,char * buf,int len,const char * csid,struct local_client ** new_client)507 static int local_sock_callback(struct local_client *thisfd, char *buf, int len,
508 const char *csid,
509 struct local_client **new_client)
510 {
511 *new_client = NULL;
512 return read_from_local_sock(thisfd);
513 }
514
515 /* Data on a connected socket */
local_rendezvous_callback(struct local_client * thisfd,char * buf,int len,const char * csid,struct local_client ** new_client)516 static int local_rendezvous_callback(struct local_client *thisfd, char *buf,
517 int len, const char *csid,
518 struct local_client **new_client)
519 {
520 /* Someone connected to our local socket, accept it. */
521
522 struct sockaddr_un socka;
523 struct local_client *newfd;
524 socklen_t sl = sizeof(socka);
525 int client_fd = accept(thisfd->fd, (struct sockaddr *) &socka, &sl);
526
527 if (client_fd == -1 && errno == EINTR)
528 return 1;
529
530 if (client_fd >= 0) {
531 newfd = malloc(sizeof(struct local_client));
532 if (!newfd) {
533 close(client_fd);
534 return 1;
535 }
536 newfd->fd = client_fd;
537 newfd->type = LOCAL_SOCK;
538 newfd->xid = 0;
539 newfd->removeme = 0;
540 newfd->callback = local_sock_callback;
541 newfd->bits.localsock.replies = NULL;
542 newfd->bits.localsock.expected_replies = 0;
543 newfd->bits.localsock.cmd = NULL;
544 newfd->bits.localsock.in_progress = FALSE;
545 newfd->bits.localsock.sent_out = FALSE;
546 newfd->bits.localsock.threadid = 0;
547 newfd->bits.localsock.finished = 0;
548 newfd->bits.localsock.pipe_client = NULL;
549 newfd->bits.localsock.private = NULL;
550 newfd->bits.localsock.all_success = 1;
551 DEBUGLOG("Got new connection on fd %d\n", newfd->fd);
552 *new_client = newfd;
553 }
554 return 1;
555 }
556
local_pipe_callback(struct local_client * thisfd,char * buf,int maxlen,const char * csid,struct local_client ** new_client)557 static int local_pipe_callback(struct local_client *thisfd, char *buf,
558 int maxlen, const char *csid,
559 struct local_client **new_client)
560 {
561 int len;
562 char buffer[PIPE_BUF];
563 struct local_client *sock_client = thisfd->bits.pipe.client;
564 int status = -1; /* in error by default */
565
566 len = read(thisfd->fd, buffer, sizeof(int));
567 if (len == -1 && errno == EINTR)
568 return 1;
569
570 if (len == sizeof(int)) {
571 memcpy(&status, buffer, sizeof(int));
572 }
573
574 DEBUGLOG("read on PIPE %d: %d bytes: status: %d\n",
575 thisfd->fd, len, status);
576
577 /* EOF on pipe or an error, close it */
578 if (len <= 0) {
579 int jstat;
580 void *ret = &status;
581 close(thisfd->fd);
582
583 /* Clear out the cross-link */
584 if (thisfd->bits.pipe.client != NULL)
585 thisfd->bits.pipe.client->bits.localsock.pipe_client =
586 NULL;
587
588 /* Reap child thread */
589 if (thisfd->bits.pipe.threadid) {
590 jstat = pthread_join(thisfd->bits.pipe.threadid, &ret);
591 thisfd->bits.pipe.threadid = 0;
592 if (thisfd->bits.pipe.client != NULL)
593 thisfd->bits.pipe.client->bits.localsock.
594 threadid = 0;
595 }
596 return -1;
597 } else {
598 DEBUGLOG("background routine status was %d, sock_client=%p\n",
599 status, sock_client);
600 /* But has the client gone away ?? */
601 if (sock_client == NULL) {
602 DEBUGLOG
603 ("Got PIPE response for dead client, ignoring it\n");
604 } else {
605 /* If error then just return that code */
606 if (status)
607 send_local_reply(sock_client, status,
608 sock_client->fd);
609 else {
610 if (sock_client->bits.localsock.state ==
611 POST_COMMAND) {
612 send_local_reply(sock_client, 0,
613 sock_client->fd);
614 } else // PRE_COMMAND finished.
615 {
616 if (
617 (status =
618 distribute_command(sock_client)) !=
619 0) send_local_reply(sock_client,
620 EFBIG,
621 sock_client->
622 fd);
623 }
624 }
625 }
626 }
627 return len;
628 }
629
630 /* If a noed is up, look for it in the reply array, if it's not there then
631 add one with "ETIMEDOUT".
632 NOTE: This won't race with real replies because they happen in the same thread.
633 */
timedout_callback(struct local_client * client,const char * csid,int node_up)634 static void timedout_callback(struct local_client *client, const char *csid,
635 int node_up)
636 {
637 if (node_up) {
638 struct node_reply *reply;
639 char nodename[max_cluster_member_name_len];
640
641 clops->name_from_csid(csid, nodename);
642 DEBUGLOG("Checking for a reply from %s\n", nodename);
643 pthread_mutex_lock(&client->bits.localsock.reply_mutex);
644
645 reply = client->bits.localsock.replies;
646 while (reply && strcmp(reply->node, nodename) != 0) {
647 reply = reply->next;
648 }
649
650 pthread_mutex_unlock(&client->bits.localsock.reply_mutex);
651
652 if (!reply) {
653 DEBUGLOG("Node %s timed-out\n", nodename);
654 add_reply_to_list(client, ETIMEDOUT, csid,
655 "Command timed out", 18);
656 }
657 }
658 }
659
660 /* Called when the request has timed out on at least one node. We fill in
661 the remaining node entries with ETIMEDOUT and return.
662
663 By the time we get here the node that caused
664 the timeout could have gone down, in which case we will never get the expected
665 number of replies that triggers the post command so we need to do it here
666 */
request_timed_out(struct local_client * client)667 static void request_timed_out(struct local_client *client)
668 {
669 DEBUGLOG("Request timed-out. padding\n");
670 clops->cluster_do_node_callback(client, timedout_callback);
671
672 if (client->bits.localsock.num_replies !=
673 client->bits.localsock.expected_replies) {
674 /* Post-process the command */
675 if (client->bits.localsock.threadid) {
676 pthread_mutex_lock(&client->bits.localsock.mutex);
677 client->bits.localsock.state = POST_COMMAND;
678 pthread_cond_signal(&client->bits.localsock.cond);
679 pthread_mutex_unlock(&client->bits.localsock.mutex);
680 }
681 }
682 }
683
684 /* This is where the real work happens */
main_loop(int local_sock,int cmd_timeout)685 static void main_loop(int local_sock, int cmd_timeout)
686 {
687 DEBUGLOG("Using timeout of %d seconds\n", cmd_timeout);
688
689 sigset_t ss;
690 sigemptyset(&ss);
691 sigaddset(&ss, SIGINT);
692 sigaddset(&ss, SIGTERM);
693 pthread_sigmask(SIG_UNBLOCK, &ss, NULL);
694 /* Main loop */
695 while (!quit) {
696 fd_set in;
697 int select_status;
698 struct local_client *thisfd;
699 struct timeval tv = { cmd_timeout, 0 };
700 int quorate = clops->is_quorate();
701
702 /* Wait on the cluster FD and all local sockets/pipes */
703 local_client_head.fd = clops->get_main_cluster_fd();
704 FD_ZERO(&in);
705 for (thisfd = &local_client_head; thisfd != NULL;
706 thisfd = thisfd->next) {
707
708 if (thisfd->removeme)
709 continue;
710
711 /* if the cluster is not quorate then don't listen for new requests */
712 if ((thisfd->type != LOCAL_RENDEZVOUS &&
713 thisfd->type != LOCAL_SOCK) || quorate)
714 FD_SET(thisfd->fd, &in);
715 }
716
717 select_status = select(FD_SETSIZE, &in, NULL, NULL, &tv);
718
719 if (reread_config) {
720 int saved_errno = errno;
721
722 reread_config = 0;
723 if (clops->reread_config)
724 clops->reread_config();
725 errno = saved_errno;
726 }
727
728 if (select_status > 0) {
729 struct local_client *lastfd = NULL;
730 char csid[MAX_CSID_LEN];
731 char buf[max_cluster_message];
732
733 for (thisfd = &local_client_head; thisfd != NULL;
734 thisfd = thisfd->next) {
735
736 if (thisfd->removeme) {
737 struct local_client *free_fd;
738 lastfd->next = thisfd->next;
739 free_fd = thisfd;
740 thisfd = lastfd;
741
742 DEBUGLOG("removeme set for fd %d\n", free_fd->fd);
743
744 /* Queue cleanup, this also frees the client struct */
745 add_to_lvmqueue(free_fd, NULL, 0, NULL);
746 break;
747 }
748
749 if (FD_ISSET(thisfd->fd, &in)) {
750 struct local_client *newfd = NULL;
751 int ret;
752
753 /* Do callback */
754 ret =
755 thisfd->callback(thisfd, buf,
756 sizeof(buf), csid,
757 &newfd);
758 /* Ignore EAGAIN */
759 if (ret < 0 && (errno == EAGAIN ||
760 errno == EINTR)) continue;
761
762 /* Got error or EOF: Remove it from the list safely */
763 if (ret <= 0) {
764 struct local_client *free_fd;
765 int type = thisfd->type;
766
767 /* If the cluster socket shuts down, so do we */
768 if (type == CLUSTER_MAIN_SOCK ||
769 type == CLUSTER_INTERNAL)
770 goto closedown;
771
772 DEBUGLOG("ret == %d, errno = %d. removing client\n",
773 ret, errno);
774 lastfd->next = thisfd->next;
775 free_fd = thisfd;
776 thisfd = lastfd;
777 close(free_fd->fd);
778
779 /* Queue cleanup, this also frees the client struct */
780 add_to_lvmqueue(free_fd, NULL, 0, NULL);
781 break;
782 }
783
784 /* New client...simply add it to the list */
785 if (newfd) {
786 newfd->next = thisfd->next;
787 thisfd->next = newfd;
788 break;
789 }
790 }
791 lastfd = thisfd;
792 }
793 }
794
795 /* Select timed out. Check for clients that have been waiting too long for a response */
796 if (select_status == 0) {
797 time_t the_time = time(NULL);
798
799 for (thisfd = &local_client_head; thisfd != NULL;
800 thisfd = thisfd->next) {
801 if (thisfd->type == LOCAL_SOCK
802 && thisfd->bits.localsock.sent_out
803 && thisfd->bits.localsock.sent_time +
804 cmd_timeout < the_time
805 && thisfd->bits.localsock.
806 expected_replies !=
807 thisfd->bits.localsock.num_replies) {
808 /* Send timed out message + replies we already have */
809 DEBUGLOG
810 ("Request timed-out (send: %ld, now: %ld)\n",
811 thisfd->bits.localsock.sent_time,
812 the_time);
813
814 thisfd->bits.localsock.all_success = 0;
815
816 request_timed_out(thisfd);
817 }
818 }
819 }
820 if (select_status < 0) {
821 if (errno == EINTR)
822 continue;
823
824 #ifdef DEBUG
825 perror("select error");
826 exit(-1);
827 #endif
828 }
829 }
830
831 closedown:
832 clops->cluster_closedown();
833 close(local_sock);
834 }
835
wait_for_child(int c_pipe,int timeout)836 static __attribute__ ((noreturn)) void wait_for_child(int c_pipe, int timeout)
837 {
838 int child_status;
839 int sstat;
840 fd_set fds;
841 struct timeval tv = {timeout, 0};
842
843 FD_ZERO(&fds);
844 FD_SET(c_pipe, &fds);
845
846 sstat = select(c_pipe+1, &fds, NULL, NULL, timeout? &tv: NULL);
847 if (sstat == 0) {
848 fprintf(stderr, "clvmd startup timed out\n");
849 exit(DFAIL_TIMEOUT);
850 }
851 if (sstat == 1) {
852 if (read(c_pipe, &child_status, sizeof(child_status)) !=
853 sizeof(child_status)) {
854
855 fprintf(stderr, "clvmd failed in initialisation\n");
856 exit(DFAIL_INIT);
857 }
858 else {
859 switch (child_status) {
860 case SUCCESS:
861 break;
862 case DFAIL_INIT:
863 fprintf(stderr, "clvmd failed in initialisation\n");
864 break;
865 case DFAIL_LOCAL_SOCK:
866 fprintf(stderr, "clvmd could not create local socket\n");
867 fprintf(stderr, "Another clvmd is probably already running\n");
868 break;
869 case DFAIL_CLUSTER_IF:
870 fprintf(stderr, "clvmd could not connect to cluster manager\n");
871 fprintf(stderr, "Consult syslog for more information\n");
872 break;
873 case DFAIL_MALLOC:
874 fprintf(stderr, "clvmd failed, not enough memory\n");
875 break;
876 default:
877 fprintf(stderr, "clvmd failed, error was %d\n", child_status);
878 break;
879 }
880 exit(child_status);
881 }
882 }
883 fprintf(stderr, "clvmd startup, select failed: %s\n", strerror(errno));
884 exit(DFAIL_INIT);
885 }
886
887 /*
888 * Fork into the background and detach from our parent process.
889 * In the interests of user-friendliness we wait for the daemon
890 * to complete initialisation before returning its status
891 * the the user.
892 */
be_daemon(int timeout)893 static void be_daemon(int timeout)
894 {
895 pid_t pid;
896 int devnull = open("/dev/null", O_RDWR);
897 if (devnull == -1) {
898 perror("Can't open /dev/null");
899 exit(3);
900 }
901
902 pipe(child_pipe);
903
904 switch (pid = fork()) {
905 case -1:
906 perror("clvmd: can't fork");
907 exit(2);
908
909 case 0: /* Child */
910 close(child_pipe[0]);
911 break;
912
913 default: /* Parent */
914 close(child_pipe[1]);
915 wait_for_child(child_pipe[0], timeout);
916 }
917
918 /* Detach ourself from the calling environment */
919 if (close(0) || close(1) || close(2)) {
920 perror("Error closing terminal FDs");
921 exit(4);
922 }
923 setsid();
924
925 if (dup2(devnull, 0) < 0 || dup2(devnull, 1) < 0
926 || dup2(devnull, 2) < 0) {
927 perror("Error setting terminal FDs to /dev/null");
928 log_error("Error setting terminal FDs to /dev/null: %m");
929 exit(5);
930 }
931 if (chdir("/")) {
932 log_error("Error setting current directory to /: %m");
933 exit(6);
934 }
935
936 }
937
938 /* Called when we have a read from the local socket.
939 was in the main loop but it's grown up and is a big girl now */
read_from_local_sock(struct local_client * thisfd)940 static int read_from_local_sock(struct local_client *thisfd)
941 {
942 int len;
943 int argslen;
944 int missing_len;
945 char buffer[PIPE_BUF];
946
947 len = read(thisfd->fd, buffer, sizeof(buffer));
948 if (len == -1 && errno == EINTR)
949 return 1;
950
951 DEBUGLOG("Read on local socket %d, len = %d\n", thisfd->fd, len);
952
953 /* EOF or error on socket */
954 if (len <= 0) {
955 int *status;
956 int jstat;
957
958 DEBUGLOG("EOF on local socket: inprogress=%d\n",
959 thisfd->bits.localsock.in_progress);
960
961 thisfd->bits.localsock.finished = 1;
962
963 /* If the client went away in mid command then tidy up */
964 if (thisfd->bits.localsock.in_progress) {
965 pthread_kill(thisfd->bits.localsock.threadid, SIGUSR2);
966 pthread_mutex_lock(&thisfd->bits.localsock.mutex);
967 thisfd->bits.localsock.state = POST_COMMAND;
968 pthread_cond_signal(&thisfd->bits.localsock.cond);
969 pthread_mutex_unlock(&thisfd->bits.localsock.mutex);
970
971 /* Free any unsent buffers */
972 free_reply(thisfd);
973 }
974
975 /* Kill the subthread & free resources */
976 if (thisfd->bits.localsock.threadid) {
977 DEBUGLOG("Waiting for child thread\n");
978 pthread_mutex_lock(&thisfd->bits.localsock.mutex);
979 thisfd->bits.localsock.state = PRE_COMMAND;
980 pthread_cond_signal(&thisfd->bits.localsock.cond);
981 pthread_mutex_unlock(&thisfd->bits.localsock.mutex);
982
983 jstat =
984 pthread_join(thisfd->bits.localsock.threadid,
985 (void **) &status);
986 DEBUGLOG("Joined child thread\n");
987
988 thisfd->bits.localsock.threadid = 0;
989 pthread_cond_destroy(&thisfd->bits.localsock.cond);
990 pthread_mutex_destroy(&thisfd->bits.localsock.mutex);
991
992 /* Remove the pipe client */
993 if (thisfd->bits.localsock.pipe_client != NULL) {
994 struct local_client *newfd;
995 struct local_client *lastfd = NULL;
996 struct local_client *free_fd = NULL;
997
998 close(thisfd->bits.localsock.pipe_client->fd); /* Close pipe */
999 close(thisfd->bits.localsock.pipe);
1000
1001 /* Remove pipe client */
1002 for (newfd = &local_client_head; newfd != NULL;
1003 newfd = newfd->next) {
1004 if (thisfd->bits.localsock.
1005 pipe_client == newfd) {
1006 thisfd->bits.localsock.
1007 pipe_client = NULL;
1008
1009 lastfd->next = newfd->next;
1010 free_fd = newfd;
1011 newfd->next = lastfd;
1012 free(free_fd);
1013 break;
1014 }
1015 lastfd = newfd;
1016 }
1017 }
1018 }
1019
1020 /* Free the command buffer */
1021 free(thisfd->bits.localsock.cmd);
1022
1023 /* Clear out the cross-link */
1024 if (thisfd->bits.localsock.pipe_client != NULL)
1025 thisfd->bits.localsock.pipe_client->bits.pipe.client =
1026 NULL;
1027
1028 close(thisfd->fd);
1029 return 0;
1030 } else {
1031 int comms_pipe[2];
1032 struct local_client *newfd;
1033 char csid[MAX_CSID_LEN];
1034 struct clvm_header *inheader;
1035 int status;
1036
1037 inheader = (struct clvm_header *) buffer;
1038
1039 /* Fill in the client ID */
1040 inheader->clientid = htonl(thisfd->fd);
1041
1042 /* If we are already busy then return an error */
1043 if (thisfd->bits.localsock.in_progress) {
1044 struct clvm_header reply;
1045 reply.cmd = CLVMD_CMD_REPLY;
1046 reply.status = EBUSY;
1047 reply.arglen = 0;
1048 reply.flags = 0;
1049 send_message(&reply, sizeof(reply), our_csid,
1050 thisfd->fd,
1051 "Error sending EBUSY reply to local user");
1052 return len;
1053 }
1054
1055 /* Free any old buffer space */
1056 free(thisfd->bits.localsock.cmd);
1057
1058 /* See if we have the whole message */
1059 argslen =
1060 len - strlen(inheader->node) - sizeof(struct clvm_header);
1061 missing_len = inheader->arglen - argslen;
1062
1063 if (missing_len < 0)
1064 missing_len = 0;
1065
1066 /* Save the message */
1067 thisfd->bits.localsock.cmd = malloc(len + missing_len);
1068
1069 if (!thisfd->bits.localsock.cmd) {
1070 struct clvm_header reply;
1071 reply.cmd = CLVMD_CMD_REPLY;
1072 reply.status = ENOMEM;
1073 reply.arglen = 0;
1074 reply.flags = 0;
1075 send_message(&reply, sizeof(reply), our_csid,
1076 thisfd->fd,
1077 "Error sending ENOMEM reply to local user");
1078 return 0;
1079 }
1080 memcpy(thisfd->bits.localsock.cmd, buffer, len);
1081 thisfd->bits.localsock.cmd_len = len + missing_len;
1082 inheader = (struct clvm_header *) thisfd->bits.localsock.cmd;
1083
1084 /* If we don't have the full message then read the rest now */
1085 if (missing_len) {
1086 char *argptr =
1087 inheader->node + strlen(inheader->node) + 1;
1088
1089 while (missing_len > 0 && len >= 0) {
1090 DEBUGLOG
1091 ("got %d bytes, need another %d (total %d)\n",
1092 argslen, missing_len, inheader->arglen);
1093 len = read(thisfd->fd, argptr + argslen,
1094 missing_len);
1095 if (len >= 0) {
1096 missing_len -= len;
1097 argslen += len;
1098 }
1099 }
1100 }
1101
1102 /* Initialise and lock the mutex so the subthread will wait after
1103 finishing the PRE routine */
1104 if (!thisfd->bits.localsock.threadid) {
1105 pthread_mutex_init(&thisfd->bits.localsock.mutex, NULL);
1106 pthread_cond_init(&thisfd->bits.localsock.cond, NULL);
1107 pthread_mutex_init(&thisfd->bits.localsock.reply_mutex, NULL);
1108 }
1109
1110 /* Only run the command if all the cluster nodes are running CLVMD */
1111 if (((inheader->flags & CLVMD_FLAG_LOCAL) == 0) &&
1112 (check_all_clvmds_running(thisfd) == -1)) {
1113 thisfd->bits.localsock.expected_replies = 0;
1114 thisfd->bits.localsock.num_replies = 0;
1115 send_local_reply(thisfd, EHOSTDOWN, thisfd->fd);
1116 return len;
1117 }
1118
1119 /* Check the node name for validity */
1120 if (inheader->node[0] && clops->csid_from_name(csid, inheader->node)) {
1121 /* Error, node is not in the cluster */
1122 struct clvm_header reply;
1123 DEBUGLOG("Unknown node: '%s'\n", inheader->node);
1124
1125 reply.cmd = CLVMD_CMD_REPLY;
1126 reply.status = ENOENT;
1127 reply.flags = 0;
1128 reply.arglen = 0;
1129 send_message(&reply, sizeof(reply), our_csid,
1130 thisfd->fd,
1131 "Error sending ENOENT reply to local user");
1132 thisfd->bits.localsock.expected_replies = 0;
1133 thisfd->bits.localsock.num_replies = 0;
1134 thisfd->bits.localsock.in_progress = FALSE;
1135 thisfd->bits.localsock.sent_out = FALSE;
1136 return len;
1137 }
1138
1139 /* If we already have a subthread then just signal it to start */
1140 if (thisfd->bits.localsock.threadid) {
1141 pthread_mutex_lock(&thisfd->bits.localsock.mutex);
1142 thisfd->bits.localsock.state = PRE_COMMAND;
1143 pthread_cond_signal(&thisfd->bits.localsock.cond);
1144 pthread_mutex_unlock(&thisfd->bits.localsock.mutex);
1145 return len;
1146 }
1147
1148 /* Create a pipe and add the reading end to our FD list */
1149 pipe(comms_pipe);
1150 newfd = malloc(sizeof(struct local_client));
1151 if (!newfd) {
1152 struct clvm_header reply;
1153 close(comms_pipe[0]);
1154 close(comms_pipe[1]);
1155
1156 reply.cmd = CLVMD_CMD_REPLY;
1157 reply.status = ENOMEM;
1158 reply.arglen = 0;
1159 reply.flags = 0;
1160 send_message(&reply, sizeof(reply), our_csid,
1161 thisfd->fd,
1162 "Error sending ENOMEM reply to local user");
1163 return len;
1164 }
1165 DEBUGLOG("creating pipe, [%d, %d]\n", comms_pipe[0],
1166 comms_pipe[1]);
1167 newfd->fd = comms_pipe[0];
1168 newfd->removeme = 0;
1169 newfd->type = THREAD_PIPE;
1170 newfd->callback = local_pipe_callback;
1171 newfd->next = thisfd->next;
1172 newfd->bits.pipe.client = thisfd;
1173 newfd->bits.pipe.threadid = 0;
1174 thisfd->next = newfd;
1175
1176 /* Store a cross link to the pipe */
1177 thisfd->bits.localsock.pipe_client = newfd;
1178
1179 thisfd->bits.localsock.pipe = comms_pipe[1];
1180
1181 /* Make sure the thread has a copy of it's own ID */
1182 newfd->bits.pipe.threadid = thisfd->bits.localsock.threadid;
1183
1184 /* Run the pre routine */
1185 thisfd->bits.localsock.in_progress = TRUE;
1186 thisfd->bits.localsock.state = PRE_COMMAND;
1187 DEBUGLOG("Creating pre&post thread\n");
1188 status = pthread_create(&thisfd->bits.localsock.threadid, NULL,
1189 pre_and_post_thread, thisfd);
1190 DEBUGLOG("Created pre&post thread, state = %d\n", status);
1191 }
1192 return len;
1193 }
1194
1195 /* Add a file descriptor from the cluster or comms interface to
1196 our list of FDs for select
1197 */
add_client(struct local_client * new_client)1198 int add_client(struct local_client *new_client)
1199 {
1200 new_client->next = local_client_head.next;
1201 local_client_head.next = new_client;
1202
1203 return 0;
1204 }
1205
1206 /* Called when the pre-command has completed successfully - we
1207 now execute the real command on all the requested nodes */
distribute_command(struct local_client * thisfd)1208 static int distribute_command(struct local_client *thisfd)
1209 {
1210 struct clvm_header *inheader =
1211 (struct clvm_header *) thisfd->bits.localsock.cmd;
1212 int len = thisfd->bits.localsock.cmd_len;
1213
1214 thisfd->xid = global_xid++;
1215 DEBUGLOG("distribute command: XID = %d\n", thisfd->xid);
1216
1217 /* Forward it to other nodes in the cluster if needed */
1218 if (!(inheader->flags & CLVMD_FLAG_LOCAL)) {
1219 /* if node is empty then do it on the whole cluster */
1220 if (inheader->node[0] == '\0') {
1221 thisfd->bits.localsock.expected_replies =
1222 clops->get_num_nodes();
1223 thisfd->bits.localsock.num_replies = 0;
1224 thisfd->bits.localsock.sent_time = time(NULL);
1225 thisfd->bits.localsock.in_progress = TRUE;
1226 thisfd->bits.localsock.sent_out = TRUE;
1227
1228 /* Do it here first */
1229 add_to_lvmqueue(thisfd, inheader, len, NULL);
1230
1231 DEBUGLOG("Sending message to all cluster nodes\n");
1232 inheader->xid = thisfd->xid;
1233 send_message(inheader, len, NULL, -1,
1234 "Error forwarding message to cluster");
1235 } else {
1236 /* Do it on a single node */
1237 char csid[MAX_CSID_LEN];
1238
1239 if (clops->csid_from_name(csid, inheader->node)) {
1240 /* This has already been checked so should not happen */
1241 return 0;
1242 } else {
1243 /* OK, found a node... */
1244 thisfd->bits.localsock.expected_replies = 1;
1245 thisfd->bits.localsock.num_replies = 0;
1246 thisfd->bits.localsock.in_progress = TRUE;
1247
1248 /* Are we the requested node ?? */
1249 if (memcmp(csid, our_csid, max_csid_len) == 0) {
1250 DEBUGLOG("Doing command on local node only\n");
1251 add_to_lvmqueue(thisfd, inheader, len, NULL);
1252 } else {
1253 DEBUGLOG("Sending message to single node: %s\n",
1254 inheader->node);
1255 inheader->xid = thisfd->xid;
1256 send_message(inheader, len,
1257 csid, -1,
1258 "Error forwarding message to cluster node");
1259 }
1260 }
1261 }
1262 } else {
1263 /* Local explicitly requested, ignore nodes */
1264 thisfd->bits.localsock.in_progress = TRUE;
1265 thisfd->bits.localsock.expected_replies = 1;
1266 thisfd->bits.localsock.num_replies = 0;
1267 add_to_lvmqueue(thisfd, inheader, len, NULL);
1268 }
1269 return 0;
1270 }
1271
1272 /* Process a command from a remote node and return the result */
process_remote_command(struct clvm_header * msg,int msglen,int fd,const char * csid)1273 static void process_remote_command(struct clvm_header *msg, int msglen, int fd,
1274 const char *csid)
1275 {
1276 char *replyargs;
1277 char nodename[max_cluster_member_name_len];
1278 int replylen = 0;
1279 int buflen = max_cluster_message - sizeof(struct clvm_header) - 1;
1280 int status;
1281 int msg_malloced = 0;
1282
1283 /* Get the node name as we /may/ need it later */
1284 clops->name_from_csid(csid, nodename);
1285
1286 DEBUGLOG("process_remote_command %s for clientid 0x%x XID %d on node %s\n",
1287 decode_cmd(msg->cmd), msg->clientid, msg->xid, nodename);
1288
1289 /* Check for GOAWAY and sulk */
1290 if (msg->cmd == CLVMD_CMD_GOAWAY) {
1291
1292 DEBUGLOG("Told to go away by %s\n", nodename);
1293 log_error("Told to go away by %s\n", nodename);
1294 exit(99);
1295 }
1296
1297 /* Version check is internal - don't bother exposing it in
1298 clvmd-command.c */
1299 if (msg->cmd == CLVMD_CMD_VERSION) {
1300 int version_nums[3];
1301 char node[256];
1302
1303 memcpy(version_nums, msg->args, sizeof(version_nums));
1304
1305 clops->name_from_csid(csid, node);
1306 DEBUGLOG("Remote node %s is version %d.%d.%d\n",
1307 node,
1308 ntohl(version_nums[0]),
1309 ntohl(version_nums[1]), ntohl(version_nums[2]));
1310
1311 if (ntohl(version_nums[0]) != CLVMD_MAJOR_VERSION) {
1312 struct clvm_header byebyemsg;
1313 DEBUGLOG
1314 ("Telling node %s to go away because of incompatible version number\n",
1315 node);
1316 log_notice
1317 ("Telling node %s to go away because of incompatible version number %d.%d.%d\n",
1318 node, ntohl(version_nums[0]),
1319 ntohl(version_nums[1]), ntohl(version_nums[2]));
1320
1321 byebyemsg.cmd = CLVMD_CMD_GOAWAY;
1322 byebyemsg.status = 0;
1323 byebyemsg.flags = 0;
1324 byebyemsg.arglen = 0;
1325 byebyemsg.clientid = 0;
1326 clops->cluster_send_message(&byebyemsg, sizeof(byebyemsg),
1327 our_csid,
1328 "Error Sending GOAWAY message");
1329 } else {
1330 clops->add_up_node(csid);
1331 }
1332 return;
1333 }
1334
1335 /* Allocate a default reply buffer */
1336 replyargs = malloc(max_cluster_message - sizeof(struct clvm_header));
1337
1338 if (replyargs != NULL) {
1339 /* Run the command */
1340 status =
1341 do_command(NULL, msg, msglen, &replyargs, buflen,
1342 &replylen);
1343 } else {
1344 status = ENOMEM;
1345 }
1346
1347 /* If it wasn't a reply, then reply */
1348 if (msg->cmd != CLVMD_CMD_REPLY) {
1349 char *aggreply;
1350
1351 aggreply =
1352 realloc(replyargs, replylen + sizeof(struct clvm_header));
1353 if (aggreply) {
1354 struct clvm_header *agghead =
1355 (struct clvm_header *) aggreply;
1356
1357 replyargs = aggreply;
1358 /* Move it up so there's room for a header in front of the data */
1359 memmove(aggreply + offsetof(struct clvm_header, args),
1360 replyargs, replylen);
1361
1362 agghead->xid = msg->xid;
1363 agghead->cmd = CLVMD_CMD_REPLY;
1364 agghead->status = status;
1365 agghead->flags = 0;
1366 agghead->clientid = msg->clientid;
1367 agghead->arglen = replylen;
1368 agghead->node[0] = '\0';
1369 send_message(aggreply,
1370 sizeof(struct clvm_header) +
1371 replylen, csid, fd,
1372 "Error sending command reply");
1373 } else {
1374 struct clvm_header head;
1375
1376 DEBUGLOG("Error attempting to realloc return buffer\n");
1377 /* Return a failure response */
1378 head.cmd = CLVMD_CMD_REPLY;
1379 head.status = ENOMEM;
1380 head.flags = 0;
1381 head.clientid = msg->clientid;
1382 head.arglen = 0;
1383 head.node[0] = '\0';
1384 send_message(&head, sizeof(struct clvm_header), csid,
1385 fd, "Error sending ENOMEM command reply");
1386 return;
1387 }
1388 }
1389
1390 /* Free buffer if it was malloced */
1391 if (msg_malloced) {
1392 free(msg);
1393 }
1394 free(replyargs);
1395 }
1396
1397 /* Add a reply to a command to the list of replies for this client.
1398 If we have got a full set then send them to the waiting client down the local
1399 socket */
add_reply_to_list(struct local_client * client,int status,const char * csid,const char * buf,int len)1400 static void add_reply_to_list(struct local_client *client, int status,
1401 const char *csid, const char *buf, int len)
1402 {
1403 struct node_reply *reply;
1404
1405 pthread_mutex_lock(&client->bits.localsock.reply_mutex);
1406
1407 /* Add it to the list of replies */
1408 reply = malloc(sizeof(struct node_reply));
1409 if (reply) {
1410 reply->status = status;
1411 clops->name_from_csid(csid, reply->node);
1412 DEBUGLOG("Reply from node %s: %d bytes\n", reply->node, len);
1413
1414 if (len > 0) {
1415 reply->replymsg = malloc(len);
1416 if (!reply->replymsg) {
1417 reply->status = ENOMEM;
1418 } else {
1419 memcpy(reply->replymsg, buf, len);
1420 }
1421 } else {
1422 reply->replymsg = NULL;
1423 }
1424 /* Hook it onto the reply chain */
1425 reply->next = client->bits.localsock.replies;
1426 client->bits.localsock.replies = reply;
1427 } else {
1428 /* It's all gone horribly wrong... */
1429 pthread_mutex_unlock(&client->bits.localsock.reply_mutex);
1430 send_local_reply(client, ENOMEM, client->fd);
1431 return;
1432 }
1433 DEBUGLOG("Got %d replies, expecting: %d\n",
1434 client->bits.localsock.num_replies + 1,
1435 client->bits.localsock.expected_replies);
1436
1437 /* If we have the whole lot then do the post-process */
1438 if (++client->bits.localsock.num_replies ==
1439 client->bits.localsock.expected_replies) {
1440 /* Post-process the command */
1441 if (client->bits.localsock.threadid) {
1442 pthread_mutex_lock(&client->bits.localsock.mutex);
1443 client->bits.localsock.state = POST_COMMAND;
1444 pthread_cond_signal(&client->bits.localsock.cond);
1445 pthread_mutex_unlock(&client->bits.localsock.mutex);
1446 }
1447 }
1448 pthread_mutex_unlock(&client->bits.localsock.reply_mutex);
1449 }
1450
1451 /* This is the thread that runs the PRE and post commands for a particular connection */
pre_and_post_thread(void * arg)1452 static __attribute__ ((noreturn)) void *pre_and_post_thread(void *arg)
1453 {
1454 struct local_client *client = (struct local_client *) arg;
1455 int status;
1456 int write_status;
1457 sigset_t ss;
1458 int pipe_fd = client->bits.localsock.pipe;
1459
1460 DEBUGLOG("in sub thread: client = %p\n", client);
1461
1462 /* Don't start until the LVM thread is ready */
1463 pthread_mutex_lock(&lvm_start_mutex);
1464 pthread_mutex_unlock(&lvm_start_mutex);
1465 DEBUGLOG("Sub thread ready for work.\n");
1466
1467 /* Ignore SIGUSR1 (handled by master process) but enable
1468 SIGUSR2 (kills subthreads) */
1469 sigemptyset(&ss);
1470 sigaddset(&ss, SIGUSR1);
1471 pthread_sigmask(SIG_BLOCK, &ss, NULL);
1472
1473 sigdelset(&ss, SIGUSR1);
1474 sigaddset(&ss, SIGUSR2);
1475 pthread_sigmask(SIG_UNBLOCK, &ss, NULL);
1476
1477 /* Loop around doing PRE and POST functions until the client goes away */
1478 while (!client->bits.localsock.finished) {
1479 /* Execute the code */
1480 status = do_pre_command(client);
1481
1482 if (status)
1483 client->bits.localsock.all_success = 0;
1484
1485 DEBUGLOG("Writing status %d down pipe %d\n", status, pipe_fd);
1486
1487 /* Tell the parent process we have finished this bit */
1488 do {
1489 write_status = write(pipe_fd, &status, sizeof(int));
1490 if (write_status == sizeof(int))
1491 break;
1492 if (write_status < 0 &&
1493 (errno == EINTR || errno == EAGAIN))
1494 continue;
1495 log_error("Error sending to pipe: %m\n");
1496 break;
1497 } while(1);
1498
1499 if (status) {
1500 client->bits.localsock.state = POST_COMMAND;
1501 goto next_pre;
1502 }
1503
1504 /* We may need to wait for the condition variable before running the post command */
1505 pthread_mutex_lock(&client->bits.localsock.mutex);
1506 DEBUGLOG("Waiting to do post command - state = %d\n",
1507 client->bits.localsock.state);
1508
1509 if (client->bits.localsock.state != POST_COMMAND) {
1510 pthread_cond_wait(&client->bits.localsock.cond,
1511 &client->bits.localsock.mutex);
1512 }
1513 pthread_mutex_unlock(&client->bits.localsock.mutex);
1514
1515 DEBUGLOG("Got post command condition...\n");
1516
1517 /* POST function must always run, even if the client aborts */
1518 status = 0;
1519 do_post_command(client);
1520
1521 do {
1522 write_status = write(pipe_fd, &status, sizeof(int));
1523 if (write_status == sizeof(int))
1524 break;
1525 if (write_status < 0 &&
1526 (errno == EINTR || errno == EAGAIN))
1527 continue;
1528 log_error("Error sending to pipe: %m\n");
1529 break;
1530 } while(1);
1531 next_pre:
1532 DEBUGLOG("Waiting for next pre command\n");
1533
1534 pthread_mutex_lock(&client->bits.localsock.mutex);
1535 if (client->bits.localsock.state != PRE_COMMAND &&
1536 !client->bits.localsock.finished) {
1537 pthread_cond_wait(&client->bits.localsock.cond,
1538 &client->bits.localsock.mutex);
1539 }
1540 pthread_mutex_unlock(&client->bits.localsock.mutex);
1541
1542 DEBUGLOG("Got pre command condition...\n");
1543 }
1544 DEBUGLOG("Subthread finished\n");
1545 pthread_exit((void *) 0);
1546 }
1547
1548 /* Process a command on the local node and store the result */
process_local_command(struct clvm_header * msg,int msglen,struct local_client * client,unsigned short xid)1549 static int process_local_command(struct clvm_header *msg, int msglen,
1550 struct local_client *client,
1551 unsigned short xid)
1552 {
1553 char *replybuf = malloc(max_cluster_message);
1554 int buflen = max_cluster_message - sizeof(struct clvm_header) - 1;
1555 int replylen = 0;
1556 int status;
1557
1558 DEBUGLOG("process_local_command: %s msg=%p, msglen =%d, client=%p\n",
1559 decode_cmd(msg->cmd), msg, msglen, client);
1560
1561 if (replybuf == NULL)
1562 return -1;
1563
1564 status = do_command(client, msg, msglen, &replybuf, buflen, &replylen);
1565
1566 if (status)
1567 client->bits.localsock.all_success = 0;
1568
1569 /* If we took too long then discard the reply */
1570 if (xid == client->xid) {
1571 add_reply_to_list(client, status, our_csid, replybuf, replylen);
1572 } else {
1573 DEBUGLOG
1574 ("Local command took too long, discarding xid %d, current is %d\n",
1575 xid, client->xid);
1576 }
1577
1578 free(replybuf);
1579 return status;
1580 }
1581
process_reply(const struct clvm_header * msg,int msglen,const char * csid)1582 static int process_reply(const struct clvm_header *msg, int msglen, const char *csid)
1583 {
1584 struct local_client *client = NULL;
1585
1586 client = find_client(msg->clientid);
1587 if (!client) {
1588 DEBUGLOG("Got message for unknown client 0x%x\n",
1589 msg->clientid);
1590 log_error("Got message for unknown client 0x%x\n",
1591 msg->clientid);
1592 return -1;
1593 }
1594
1595 if (msg->status)
1596 client->bits.localsock.all_success = 0;
1597
1598 /* Gather replies together for this client id */
1599 if (msg->xid == client->xid) {
1600 add_reply_to_list(client, msg->status, csid, msg->args,
1601 msg->arglen);
1602 } else {
1603 DEBUGLOG("Discarding reply with old XID %d, current = %d\n",
1604 msg->xid, client->xid);
1605 }
1606 return 0;
1607 }
1608
1609 /* Send an aggregated reply back to the client */
send_local_reply(struct local_client * client,int status,int fd)1610 static void send_local_reply(struct local_client *client, int status, int fd)
1611 {
1612 struct clvm_header *clientreply;
1613 struct node_reply *thisreply = client->bits.localsock.replies;
1614 char *replybuf;
1615 char *ptr;
1616 int message_len = 0;
1617
1618 DEBUGLOG("Send local reply\n");
1619
1620 /* Work out the total size of the reply */
1621 while (thisreply) {
1622 if (thisreply->replymsg)
1623 message_len += strlen(thisreply->replymsg) + 1;
1624 else
1625 message_len++;
1626
1627 message_len += strlen(thisreply->node) + 1 + sizeof(int);
1628
1629 thisreply = thisreply->next;
1630 }
1631
1632 /* Add in the size of our header */
1633 message_len = message_len + sizeof(struct clvm_header) + 1;
1634 replybuf = malloc(message_len);
1635
1636 clientreply = (struct clvm_header *) replybuf;
1637 clientreply->status = status;
1638 clientreply->cmd = CLVMD_CMD_REPLY;
1639 clientreply->node[0] = '\0';
1640 clientreply->flags = 0;
1641
1642 ptr = clientreply->args;
1643
1644 /* Add in all the replies, and free them as we go */
1645 thisreply = client->bits.localsock.replies;
1646 while (thisreply) {
1647 struct node_reply *tempreply = thisreply;
1648
1649 strcpy(ptr, thisreply->node);
1650 ptr += strlen(thisreply->node) + 1;
1651
1652 if (thisreply->status)
1653 clientreply->flags |= CLVMD_FLAG_NODEERRS;
1654
1655 memcpy(ptr, &thisreply->status, sizeof(int));
1656 ptr += sizeof(int);
1657
1658 if (thisreply->replymsg) {
1659 strcpy(ptr, thisreply->replymsg);
1660 ptr += strlen(thisreply->replymsg) + 1;
1661 } else {
1662 ptr[0] = '\0';
1663 ptr++;
1664 }
1665 thisreply = thisreply->next;
1666
1667 free(tempreply->replymsg);
1668 free(tempreply);
1669 }
1670
1671 /* Terminate with an empty node name */
1672 *ptr = '\0';
1673
1674 clientreply->arglen = ptr - clientreply->args + 1;
1675
1676 /* And send it */
1677 send_message(replybuf, message_len, our_csid, fd,
1678 "Error sending REPLY to client");
1679 free(replybuf);
1680
1681 /* Reset comms variables */
1682 client->bits.localsock.replies = NULL;
1683 client->bits.localsock.expected_replies = 0;
1684 client->bits.localsock.in_progress = FALSE;
1685 client->bits.localsock.sent_out = FALSE;
1686 }
1687
1688 /* Just free a reply chain baceuse it wasn't used. */
free_reply(struct local_client * client)1689 static void free_reply(struct local_client *client)
1690 {
1691 /* Add in all the replies, and free them as we go */
1692 struct node_reply *thisreply = client->bits.localsock.replies;
1693 while (thisreply) {
1694 struct node_reply *tempreply = thisreply;
1695
1696 thisreply = thisreply->next;
1697
1698 free(tempreply->replymsg);
1699 free(tempreply);
1700 }
1701 client->bits.localsock.replies = NULL;
1702 }
1703
1704 /* Send our version number to the cluster */
send_version_message()1705 static void send_version_message()
1706 {
1707 char message[sizeof(struct clvm_header) + sizeof(int) * 3];
1708 struct clvm_header *msg = (struct clvm_header *) message;
1709 int version_nums[3];
1710
1711 msg->cmd = CLVMD_CMD_VERSION;
1712 msg->status = 0;
1713 msg->flags = 0;
1714 msg->clientid = 0;
1715 msg->arglen = sizeof(version_nums);
1716
1717 version_nums[0] = htonl(CLVMD_MAJOR_VERSION);
1718 version_nums[1] = htonl(CLVMD_MINOR_VERSION);
1719 version_nums[2] = htonl(CLVMD_PATCH_VERSION);
1720
1721 memcpy(&msg->args, version_nums, sizeof(version_nums));
1722
1723 hton_clvm(msg);
1724
1725 clops->cluster_send_message(message, sizeof(message), NULL,
1726 "Error Sending version number");
1727 }
1728
1729 /* Send a message to either a local client or another server */
send_message(void * buf,int msglen,const char * csid,int fd,const char * errtext)1730 static int send_message(void *buf, int msglen, const char *csid, int fd,
1731 const char *errtext)
1732 {
1733 int len = 0;
1734 int saved_errno = 0;
1735 struct timespec delay;
1736 struct timespec remtime;
1737
1738 int retry_cnt = 0;
1739
1740 /* Send remote messages down the cluster socket */
1741 if (csid == NULL || !ISLOCAL_CSID(csid)) {
1742 hton_clvm((struct clvm_header *) buf);
1743 return clops->cluster_send_message(buf, msglen, csid, errtext);
1744 } else {
1745 int ptr = 0;
1746
1747 /* Make sure it all goes */
1748 do {
1749 if (retry_cnt > MAX_RETRIES)
1750 {
1751 errno = saved_errno;
1752 log_error("%s", errtext);
1753 errno = saved_errno;
1754 break;
1755 }
1756
1757 len = write(fd, buf + ptr, msglen - ptr);
1758
1759 if (len <= 0) {
1760 if (errno == EINTR)
1761 continue;
1762 if (errno == EAGAIN ||
1763 errno == EIO ||
1764 errno == ENOSPC) {
1765 saved_errno = errno;
1766 retry_cnt++;
1767
1768 delay.tv_sec = 0;
1769 delay.tv_nsec = 100000;
1770 remtime.tv_sec = 0;
1771 remtime.tv_nsec = 0;
1772 (void) nanosleep (&delay, &remtime);
1773
1774 continue;
1775 }
1776 log_error("%s", errtext);
1777 break;
1778 }
1779 ptr += len;
1780 } while (ptr < msglen);
1781 }
1782 return len;
1783 }
1784
process_work_item(struct lvm_thread_cmd * cmd)1785 static int process_work_item(struct lvm_thread_cmd *cmd)
1786 {
1787 /* If msg is NULL then this is a cleanup request */
1788 if (cmd->msg == NULL) {
1789 DEBUGLOG("process_work_item: free fd %d\n", cmd->client->fd);
1790 cmd_client_cleanup(cmd->client);
1791 free(cmd->client);
1792 return 0;
1793 }
1794
1795 if (!cmd->remote) {
1796 DEBUGLOG("process_work_item: local\n");
1797 process_local_command(cmd->msg, cmd->msglen, cmd->client,
1798 cmd->xid);
1799 } else {
1800 DEBUGLOG("process_work_item: remote\n");
1801 process_remote_command(cmd->msg, cmd->msglen, cmd->client->fd,
1802 cmd->csid);
1803 }
1804 return 0;
1805 }
1806
1807 /*
1808 * Routine that runs in the "LVM thread".
1809 */
lvm_thread_fn(void * arg)1810 static void lvm_thread_fn(void *arg)
1811 {
1812 struct dm_list *cmdl, *tmp;
1813 sigset_t ss;
1814 int using_gulm = (int)(long)arg;
1815
1816 DEBUGLOG("LVM thread function started\n");
1817
1818 /* Ignore SIGUSR1 & 2 */
1819 sigemptyset(&ss);
1820 sigaddset(&ss, SIGUSR1);
1821 sigaddset(&ss, SIGUSR2);
1822 pthread_sigmask(SIG_BLOCK, &ss, NULL);
1823
1824 /* Initialise the interface to liblvm */
1825 init_lvm(using_gulm);
1826
1827 /* Allow others to get moving */
1828 pthread_mutex_unlock(&lvm_start_mutex);
1829
1830 /* Now wait for some actual work */
1831 for (;;) {
1832 DEBUGLOG("LVM thread waiting for work\n");
1833
1834 pthread_mutex_lock(&lvm_thread_mutex);
1835 if (dm_list_empty(&lvm_cmd_head))
1836 pthread_cond_wait(&lvm_thread_cond, &lvm_thread_mutex);
1837
1838 dm_list_iterate_safe(cmdl, tmp, &lvm_cmd_head) {
1839 struct lvm_thread_cmd *cmd;
1840
1841 cmd =
1842 dm_list_struct_base(cmdl, struct lvm_thread_cmd, list);
1843 dm_list_del(&cmd->list);
1844 pthread_mutex_unlock(&lvm_thread_mutex);
1845
1846 process_work_item(cmd);
1847 free(cmd->msg);
1848 free(cmd);
1849
1850 pthread_mutex_lock(&lvm_thread_mutex);
1851 }
1852 pthread_mutex_unlock(&lvm_thread_mutex);
1853 }
1854 }
1855
1856 /* Pass down some work to the LVM thread */
add_to_lvmqueue(struct local_client * client,struct clvm_header * msg,int msglen,const char * csid)1857 static int add_to_lvmqueue(struct local_client *client, struct clvm_header *msg,
1858 int msglen, const char *csid)
1859 {
1860 struct lvm_thread_cmd *cmd;
1861
1862 cmd = malloc(sizeof(struct lvm_thread_cmd));
1863 if (!cmd)
1864 return ENOMEM;
1865
1866 if (msglen) {
1867 cmd->msg = malloc(msglen);
1868 if (!cmd->msg) {
1869 log_error("Unable to allocate buffer space\n");
1870 free(cmd);
1871 return -1;
1872 }
1873 memcpy(cmd->msg, msg, msglen);
1874 }
1875 else {
1876 cmd->msg = NULL;
1877 }
1878 cmd->client = client;
1879 cmd->msglen = msglen;
1880 cmd->xid = client->xid;
1881
1882 if (csid) {
1883 memcpy(cmd->csid, csid, max_csid_len);
1884 cmd->remote = 1;
1885 } else {
1886 cmd->remote = 0;
1887 }
1888
1889 DEBUGLOG
1890 ("add_to_lvmqueue: cmd=%p. client=%p, msg=%p, len=%d, csid=%p, xid=%d\n",
1891 cmd, client, msg, msglen, csid, cmd->xid);
1892 pthread_mutex_lock(&lvm_thread_mutex);
1893 dm_list_add(&lvm_cmd_head, &cmd->list);
1894 pthread_cond_signal(&lvm_thread_cond);
1895 pthread_mutex_unlock(&lvm_thread_mutex);
1896
1897 return 0;
1898 }
1899
1900 /* Return 0 if we can talk to an existing clvmd */
check_local_clvmd(void)1901 static int check_local_clvmd(void)
1902 {
1903 int local_socket;
1904 struct sockaddr_un sockaddr;
1905 int ret = 0;
1906
1907 /* Open local socket */
1908 if ((local_socket = socket(PF_UNIX, SOCK_STREAM, 0)) < 0) {
1909 return -1;
1910 }
1911
1912 memset(&sockaddr, 0, sizeof(sockaddr));
1913 memcpy(sockaddr.sun_path, CLVMD_SOCKNAME, sizeof(CLVMD_SOCKNAME));
1914 sockaddr.sun_family = AF_UNIX;
1915
1916 if (connect(local_socket,(struct sockaddr *) &sockaddr,
1917 sizeof(sockaddr))) {
1918 ret = -1;
1919 }
1920
1921 close(local_socket);
1922 return ret;
1923 }
1924
1925
1926 /* Open the local socket, that's the one we talk to libclvm down */
open_local_sock()1927 static int open_local_sock()
1928 {
1929 int local_socket;
1930 struct sockaddr_un sockaddr;
1931
1932 /* Open local socket */
1933 if (CLVMD_SOCKNAME[0] != '\0')
1934 unlink(CLVMD_SOCKNAME);
1935 local_socket = socket(PF_UNIX, SOCK_STREAM, 0);
1936 if (local_socket < 0) {
1937 log_error("Can't create local socket: %m");
1938 return -1;
1939 }
1940 /* Set Close-on-exec & non-blocking */
1941 fcntl(local_socket, F_SETFD, 1);
1942 fcntl(local_socket, F_SETFL, fcntl(local_socket, F_GETFL, 0) | O_NONBLOCK);
1943
1944 memset(&sockaddr, 0, sizeof(sockaddr));
1945 memcpy(sockaddr.sun_path, CLVMD_SOCKNAME, sizeof(CLVMD_SOCKNAME));
1946 sockaddr.sun_family = AF_UNIX;
1947 if (bind(local_socket, (struct sockaddr *) &sockaddr, sizeof(sockaddr))) {
1948 log_error("can't bind local socket: %m");
1949 close(local_socket);
1950 return -1;
1951 }
1952 if (listen(local_socket, 1) != 0) {
1953 log_error("listen local: %m");
1954 close(local_socket);
1955 return -1;
1956 }
1957 if (CLVMD_SOCKNAME[0] != '\0')
1958 chmod(CLVMD_SOCKNAME, 0600);
1959
1960 return local_socket;
1961 }
1962
process_message(struct local_client * client,const char * buf,int len,const char * csid)1963 void process_message(struct local_client *client, const char *buf, int len,
1964 const char *csid)
1965 {
1966 struct clvm_header *inheader;
1967
1968 inheader = (struct clvm_header *) buf;
1969 ntoh_clvm(inheader); /* Byteswap fields */
1970 if (inheader->cmd == CLVMD_CMD_REPLY)
1971 process_reply(inheader, len, csid);
1972 else
1973 add_to_lvmqueue(client, inheader, len, csid);
1974 }
1975
1976
check_all_callback(struct local_client * client,const char * csid,int node_up)1977 static void check_all_callback(struct local_client *client, const char *csid,
1978 int node_up)
1979 {
1980 if (!node_up)
1981 add_reply_to_list(client, EHOSTDOWN, csid, "CLVMD not running",
1982 18);
1983 }
1984
1985 /* Check to see if all CLVMDs are running (ie one on
1986 every node in the cluster).
1987 If not, returns -1 and prints out a list of errant nodes */
check_all_clvmds_running(struct local_client * client)1988 static int check_all_clvmds_running(struct local_client *client)
1989 {
1990 DEBUGLOG("check_all_clvmds_running\n");
1991 return clops->cluster_do_node_callback(client, check_all_callback);
1992 }
1993
1994 /* Return a local_client struct given a client ID.
1995 client IDs are in network byte order */
find_client(int clientid)1996 static struct local_client *find_client(int clientid)
1997 {
1998 struct local_client *thisfd;
1999 for (thisfd = &local_client_head; thisfd != NULL; thisfd = thisfd->next) {
2000 if (thisfd->fd == ntohl(clientid))
2001 return thisfd;
2002 }
2003 return NULL;
2004 }
2005
2006 /* Byte-swapping routines for the header so we
2007 work in a heterogeneous environment */
hton_clvm(struct clvm_header * hdr)2008 static void hton_clvm(struct clvm_header *hdr)
2009 {
2010 hdr->status = htonl(hdr->status);
2011 hdr->arglen = htonl(hdr->arglen);
2012 hdr->xid = htons(hdr->xid);
2013 /* Don't swap clientid as it's only a token as far as
2014 remote nodes are concerned */
2015 }
2016
ntoh_clvm(struct clvm_header * hdr)2017 static void ntoh_clvm(struct clvm_header *hdr)
2018 {
2019 hdr->status = ntohl(hdr->status);
2020 hdr->arglen = ntohl(hdr->arglen);
2021 hdr->xid = ntohs(hdr->xid);
2022 }
2023
2024 /* Handler for SIGUSR2 - sent to kill subthreads */
sigusr2_handler(int sig)2025 static void sigusr2_handler(int sig)
2026 {
2027 DEBUGLOG("SIGUSR2 received\n");
2028 return;
2029 }
2030
sigterm_handler(int sig)2031 static void sigterm_handler(int sig)
2032 {
2033 DEBUGLOG("SIGTERM received\n");
2034 quit = 1;
2035 return;
2036 }
2037
sighup_handler(int sig)2038 static void sighup_handler(int sig)
2039 {
2040 DEBUGLOG("got SIGHUP\n");
2041 reread_config = 1;
2042 }
2043
sync_lock(const char * resource,int mode,int flags,int * lockid)2044 int sync_lock(const char *resource, int mode, int flags, int *lockid)
2045 {
2046 return clops->sync_lock(resource, mode, flags, lockid);
2047 }
2048
sync_unlock(const char * resource,int lockid)2049 int sync_unlock(const char *resource, int lockid)
2050 {
2051 return clops->sync_unlock(resource, lockid);
2052 }
2053
parse_cluster_interface(char * ifname)2054 static if_type_t parse_cluster_interface(char *ifname)
2055 {
2056 if_type_t iface = IF_AUTO;
2057
2058 if (!strcmp(ifname, "auto"))
2059 iface = IF_AUTO;
2060 if (!strcmp(ifname, "cman"))
2061 iface = IF_CMAN;
2062 if (!strcmp(ifname, "gulm"))
2063 iface = IF_GULM;
2064 if (!strcmp(ifname, "openais"))
2065 iface = IF_OPENAIS;
2066 if (!strcmp(ifname, "corosync"))
2067 iface = IF_COROSYNC;
2068
2069 return iface;
2070 }
2071
2072 /*
2073 * Try and find a cluster system in corosync's objdb, if it is running. This is
2074 * only called if the command-line option is not present, and if it fails
2075 * we still try the interfaces in order.
2076 */
get_cluster_type()2077 static if_type_t get_cluster_type()
2078 {
2079 #ifdef HAVE_COROSYNC_CONFDB_H
2080 confdb_handle_t handle;
2081 if_type_t type = IF_AUTO;
2082 int result;
2083 char buf[255];
2084 size_t namelen = sizeof(buf);
2085 hdb_handle_t cluster_handle;
2086 hdb_handle_t clvmd_handle;
2087 confdb_callbacks_t callbacks = {
2088 .confdb_key_change_notify_fn = NULL,
2089 .confdb_object_create_change_notify_fn = NULL,
2090 .confdb_object_delete_change_notify_fn = NULL
2091 };
2092
2093 result = confdb_initialize (&handle, &callbacks);
2094 if (result != CS_OK)
2095 return type;
2096
2097 result = confdb_object_find_start(handle, OBJECT_PARENT_HANDLE);
2098 if (result != CS_OK)
2099 goto out;
2100
2101 result = confdb_object_find(handle, OBJECT_PARENT_HANDLE, (void *)"cluster", strlen("cluster"), &cluster_handle);
2102 if (result != CS_OK)
2103 goto out;
2104
2105 result = confdb_object_find_start(handle, cluster_handle);
2106 if (result != CS_OK)
2107 goto out;
2108
2109 result = confdb_object_find(handle, cluster_handle, (void *)"clvmd", strlen("clvmd"), &clvmd_handle);
2110 if (result != CS_OK)
2111 goto out;
2112
2113 result = confdb_key_get(handle, clvmd_handle, (void *)"interface", strlen("interface"), buf, &namelen);
2114 if (result != CS_OK)
2115 goto out;
2116
2117 buf[namelen] = '\0';
2118 type = parse_cluster_interface(buf);
2119 DEBUGLOG("got interface type '%s' from confdb\n", buf);
2120 out:
2121 confdb_finalize(handle);
2122 return type;
2123 #else
2124 return IF_AUTO;
2125 #endif
2126 }
2127