This source file includes following definitions.
- crm_trigger_prepare
- crm_trigger_check
- crm_trigger_dispatch
- crm_trigger_finalize
- mainloop_setup_trigger
- mainloop_trigger_complete
- mainloop_add_trigger
- mainloop_set_trigger
- mainloop_destroy_trigger
- crm_signal_dispatch
- mainloop_signal_handler
- crm_signal_handler
- mainloop_destroy_signal_entry
- mainloop_add_signal
- mainloop_destroy_signal
- mainloop_cleanup
- gio_read_socket
- gio_poll_destroy
- conv_prio_libqb2glib
- conv_libqb_prio2ratelimit
- gio_poll_dispatch_update
- gio_poll_dispatch_add
- gio_poll_dispatch_mod
- gio_poll_dispatch_del
- pick_ipc_type
- mainloop_add_ipc_server
- mainloop_add_ipc_server_with_prio
- mainloop_del_ipc_server
- mainloop_gio_callback
- mainloop_gio_destroy
- pcmk__add_mainloop_ipc
- pcmk__mainloop_timer_get_period
- mainloop_add_ipc_client
- mainloop_del_ipc_client
- mainloop_get_ipc_client
- mainloop_add_fd
- mainloop_del_fd
- mainloop_child_pid
- mainloop_child_name
- mainloop_child_timeout
- mainloop_child_userdata
- mainloop_clear_child_userdata
- child_free
- child_kill_helper
- child_timeout_callback
- child_waitpid
- child_death_dispatch
- child_signal_init
- mainloop_child_kill
- mainloop_child_add_with_flags
- mainloop_child_add
- mainloop_timer_cb
- mainloop_timer_running
- mainloop_timer_start
- mainloop_timer_stop
- mainloop_timer_set_period
- mainloop_timer_add
- mainloop_timer_del
- drain_timeout_cb
- pcmk_quit_main_loop
- pcmk_drain_main_loop
- crm_signal
1
2
3
4
5
6
7
8
9
10 #include <crm_internal.h>
11
12 #ifndef _GNU_SOURCE
13 # define _GNU_SOURCE
14 #endif
15
16 #include <stdlib.h>
17 #include <string.h>
18 #include <signal.h>
19 #include <errno.h>
20
21 #include <sys/wait.h>
22
23 #include <crm/crm.h>
24 #include <crm/common/xml.h>
25 #include <crm/common/mainloop.h>
26 #include <crm/common/ipc_internal.h>
27
28 #include <qb/qbarray.h>
29
30 struct mainloop_child_s {
31 pid_t pid;
32 char *desc;
33 unsigned timerid;
34 gboolean timeout;
35 void *privatedata;
36
37 enum mainloop_child_flags flags;
38
39
40 void (*callback) (mainloop_child_t * p, pid_t pid, int core, int signo, int exitcode);
41 };
42
43 struct trigger_s {
44 GSource source;
45 gboolean running;
46 gboolean trigger;
47 void *user_data;
48 guint id;
49
50 };
51
52 struct mainloop_timer_s {
53 guint id;
54 guint period_ms;
55 bool repeat;
56 char *name;
57 GSourceFunc cb;
58 void *userdata;
59 };
60
61 static gboolean
62 crm_trigger_prepare(GSource * source, gint * timeout)
63 {
64 crm_trigger_t *trig = (crm_trigger_t *) source;
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83 *timeout = 500;
84
85 return trig->trigger;
86 }
87
88 static gboolean
89 crm_trigger_check(GSource * source)
90 {
91 crm_trigger_t *trig = (crm_trigger_t *) source;
92
93 return trig->trigger;
94 }
95
96
97
98
99
100
101
102
103
104
105
106 static gboolean
107 crm_trigger_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
108 {
109 gboolean rc = G_SOURCE_CONTINUE;
110 crm_trigger_t *trig = (crm_trigger_t *) source;
111
112 if (trig->running) {
113
114 return G_SOURCE_CONTINUE;
115 }
116 trig->trigger = FALSE;
117
118 if (callback) {
119 int callback_rc = callback(trig->user_data);
120
121 if (callback_rc < 0) {
122 crm_trace("Trigger handler %p not yet complete", trig);
123 trig->running = TRUE;
124 } else if (callback_rc == 0) {
125 rc = G_SOURCE_REMOVE;
126 }
127 }
128 return rc;
129 }
130
131 static void
132 crm_trigger_finalize(GSource * source)
133 {
134 crm_trace("Trigger %p destroyed", source);
135 }
136
137 static GSourceFuncs crm_trigger_funcs = {
138 crm_trigger_prepare,
139 crm_trigger_check,
140 crm_trigger_dispatch,
141 crm_trigger_finalize,
142 };
143
144 static crm_trigger_t *
145 mainloop_setup_trigger(GSource * source, int priority, int (*dispatch) (gpointer user_data),
146 gpointer userdata)
147 {
148 crm_trigger_t *trigger = NULL;
149
150 trigger = (crm_trigger_t *) source;
151
152 trigger->id = 0;
153 trigger->trigger = FALSE;
154 trigger->user_data = userdata;
155
156 if (dispatch) {
157 g_source_set_callback(source, dispatch, trigger, NULL);
158 }
159
160 g_source_set_priority(source, priority);
161 g_source_set_can_recurse(source, FALSE);
162
163 trigger->id = g_source_attach(source, NULL);
164 return trigger;
165 }
166
167 void
168 mainloop_trigger_complete(crm_trigger_t * trig)
169 {
170 crm_trace("Trigger handler %p complete", trig);
171 trig->running = FALSE;
172 }
173
174
175
176
177
178
179
180
181
182
183
184
185 crm_trigger_t *
186 mainloop_add_trigger(int priority, int (*dispatch) (gpointer user_data), gpointer userdata)
187 {
188 GSource *source = NULL;
189
190 CRM_ASSERT(sizeof(crm_trigger_t) > sizeof(GSource));
191 source = g_source_new(&crm_trigger_funcs, sizeof(crm_trigger_t));
192 CRM_ASSERT(source != NULL);
193
194 return mainloop_setup_trigger(source, priority, dispatch, userdata);
195 }
196
197 void
198 mainloop_set_trigger(crm_trigger_t * source)
199 {
200 if(source) {
201 source->trigger = TRUE;
202 }
203 }
204
205 gboolean
206 mainloop_destroy_trigger(crm_trigger_t * source)
207 {
208 GSource *gs = NULL;
209
210 if(source == NULL) {
211 return TRUE;
212 }
213
214 gs = (GSource *)source;
215
216 g_source_destroy(gs);
217 g_source_unref(gs);
218
219
220
221
222
223
224
225 return TRUE;
226 }
227
228
229
230
231 typedef struct signal_s {
232 crm_trigger_t trigger;
233 void (*handler) (int sig);
234 int signal;
235 } crm_signal_t;
236
237
238 static crm_signal_t *crm_signals[NSIG];
239
240
241
242
243
244
245
246
247
248
249
250
251 static gboolean
252 crm_signal_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
253 {
254 crm_signal_t *sig = (crm_signal_t *) source;
255
256 if(sig->signal != SIGCHLD) {
257 crm_notice("Caught '%s' signal "CRM_XS" %d (%s handler)",
258 strsignal(sig->signal), sig->signal,
259 (sig->handler? "invoking" : "no"));
260 }
261
262 sig->trigger.trigger = FALSE;
263 if (sig->handler) {
264 sig->handler(sig->signal);
265 }
266 return TRUE;
267 }
268
269
270
271
272
273
274
275
276
277
278 static void
279 mainloop_signal_handler(int sig)
280 {
281 if (sig > 0 && sig < NSIG && crm_signals[sig] != NULL) {
282 mainloop_set_trigger((crm_trigger_t *) crm_signals[sig]);
283 }
284 }
285
286
287 static GSourceFuncs crm_signal_funcs = {
288 crm_trigger_prepare,
289 crm_trigger_check,
290 crm_signal_dispatch,
291 crm_trigger_finalize,
292 };
293
294
295
296
297
298
299
300
301
302
303
304
305
306 sighandler_t
307 crm_signal_handler(int sig, sighandler_t dispatch)
308 {
309 sigset_t mask;
310 struct sigaction sa;
311 struct sigaction old;
312
313 if (sigemptyset(&mask) < 0) {
314 crm_err("Could not set handler for signal %d: %s",
315 sig, pcmk_strerror(errno));
316 return SIG_ERR;
317 }
318
319 memset(&sa, 0, sizeof(struct sigaction));
320 sa.sa_handler = dispatch;
321 sa.sa_flags = SA_RESTART;
322 sa.sa_mask = mask;
323
324 if (sigaction(sig, &sa, &old) < 0) {
325 crm_err("Could not set handler for signal %d: %s",
326 sig, pcmk_strerror(errno));
327 return SIG_ERR;
328 }
329 return old.sa_handler;
330 }
331
332 static void
333 mainloop_destroy_signal_entry(int sig)
334 {
335 crm_signal_t *tmp = crm_signals[sig];
336
337 crm_signals[sig] = NULL;
338
339 crm_trace("Destroying signal %d", sig);
340 mainloop_destroy_trigger((crm_trigger_t *) tmp);
341 }
342
343
344
345
346
347
348
349
350
351
352
353
354 gboolean
355 mainloop_add_signal(int sig, void (*dispatch) (int sig))
356 {
357 GSource *source = NULL;
358 int priority = G_PRIORITY_HIGH - 1;
359
360 if (sig == SIGTERM) {
361
362
363
364
365 priority--;
366 }
367
368 if (sig >= NSIG || sig < 0) {
369 crm_err("Signal %d is out of range", sig);
370 return FALSE;
371
372 } else if (crm_signals[sig] != NULL && crm_signals[sig]->handler == dispatch) {
373 crm_trace("Signal handler for %d is already installed", sig);
374 return TRUE;
375
376 } else if (crm_signals[sig] != NULL) {
377 crm_err("Different signal handler for %d is already installed", sig);
378 return FALSE;
379 }
380
381 CRM_ASSERT(sizeof(crm_signal_t) > sizeof(GSource));
382 source = g_source_new(&crm_signal_funcs, sizeof(crm_signal_t));
383
384 crm_signals[sig] = (crm_signal_t *) mainloop_setup_trigger(source, priority, NULL, NULL);
385 CRM_ASSERT(crm_signals[sig] != NULL);
386
387 crm_signals[sig]->handler = dispatch;
388 crm_signals[sig]->signal = sig;
389
390 if (crm_signal_handler(sig, mainloop_signal_handler) == SIG_ERR) {
391 mainloop_destroy_signal_entry(sig);
392 return FALSE;
393 }
394 #if 0
395
396
397
398
399
400 if (siginterrupt(sig, 1) < 0) {
401 crm_perror(LOG_INFO, "Could not enable system call interruptions for signal %d", sig);
402 }
403 #endif
404
405 return TRUE;
406 }
407
408 gboolean
409 mainloop_destroy_signal(int sig)
410 {
411 if (sig >= NSIG || sig < 0) {
412 crm_err("Signal %d is out of range", sig);
413 return FALSE;
414
415 } else if (crm_signal_handler(sig, NULL) == SIG_ERR) {
416 crm_perror(LOG_ERR, "Could not uninstall signal handler for signal %d", sig);
417 return FALSE;
418
419 } else if (crm_signals[sig] == NULL) {
420 return TRUE;
421 }
422 mainloop_destroy_signal_entry(sig);
423 return TRUE;
424 }
425
426 static qb_array_t *gio_map = NULL;
427
428 void
429 mainloop_cleanup(void)
430 {
431 if (gio_map) {
432 qb_array_free(gio_map);
433 }
434
435 for (int sig = 0; sig < NSIG; ++sig) {
436 mainloop_destroy_signal_entry(sig);
437 }
438 }
439
440
441
442
443 struct gio_to_qb_poll {
444 int32_t is_used;
445 guint source;
446 int32_t events;
447 void *data;
448 qb_ipcs_dispatch_fn_t fn;
449 enum qb_loop_priority p;
450 };
451
452 static gboolean
453 gio_read_socket(GIOChannel * gio, GIOCondition condition, gpointer data)
454 {
455 struct gio_to_qb_poll *adaptor = (struct gio_to_qb_poll *)data;
456 gint fd = g_io_channel_unix_get_fd(gio);
457
458 crm_trace("%p.%d %d", data, fd, condition);
459
460
461
462 CRM_ASSERT(adaptor->is_used > 0);
463
464 return (adaptor->fn(fd, condition, adaptor->data) == 0);
465 }
466
467 static void
468 gio_poll_destroy(gpointer data)
469 {
470 struct gio_to_qb_poll *adaptor = (struct gio_to_qb_poll *)data;
471
472 adaptor->is_used--;
473 CRM_ASSERT(adaptor->is_used >= 0);
474
475 if (adaptor->is_used == 0) {
476 crm_trace("Marking adaptor %p unused", adaptor);
477 adaptor->source = 0;
478 }
479 }
480
481
482
483
484
485
486
487
488
489 static gint
490 conv_prio_libqb2glib(enum qb_loop_priority prio)
491 {
492 gint ret = G_PRIORITY_DEFAULT;
493 switch (prio) {
494 case QB_LOOP_LOW:
495 ret = G_PRIORITY_LOW;
496 break;
497 case QB_LOOP_HIGH:
498 ret = G_PRIORITY_HIGH;
499 break;
500 default:
501 crm_trace("Invalid libqb's loop priority %d, assuming QB_LOOP_MED",
502 prio);
503
504 case QB_LOOP_MED:
505 break;
506 }
507 return ret;
508 }
509
510
511
512
513
514
515
516
517
518 static enum qb_ipcs_rate_limit
519 conv_libqb_prio2ratelimit(enum qb_loop_priority prio)
520 {
521
522 enum qb_ipcs_rate_limit ret = QB_IPCS_RATE_NORMAL;
523 switch (prio) {
524 case QB_LOOP_LOW:
525 ret = QB_IPCS_RATE_SLOW;
526 break;
527 case QB_LOOP_HIGH:
528 ret = QB_IPCS_RATE_FAST;
529 break;
530 default:
531 crm_trace("Invalid libqb's loop priority %d, assuming QB_LOOP_MED",
532 prio);
533
534 case QB_LOOP_MED:
535 break;
536 }
537 return ret;
538 }
539
540 static int32_t
541 gio_poll_dispatch_update(enum qb_loop_priority p, int32_t fd, int32_t evts,
542 void *data, qb_ipcs_dispatch_fn_t fn, int32_t add)
543 {
544 struct gio_to_qb_poll *adaptor;
545 GIOChannel *channel;
546 int32_t res = 0;
547
548 res = qb_array_index(gio_map, fd, (void **)&adaptor);
549 if (res < 0) {
550 crm_err("Array lookup failed for fd=%d: %d", fd, res);
551 return res;
552 }
553
554 crm_trace("Adding fd=%d to mainloop as adaptor %p", fd, adaptor);
555
556 if (add && adaptor->source) {
557 crm_err("Adaptor for descriptor %d is still in-use", fd);
558 return -EEXIST;
559 }
560 if (!add && !adaptor->is_used) {
561 crm_err("Adaptor for descriptor %d is not in-use", fd);
562 return -ENOENT;
563 }
564
565
566 channel = g_io_channel_unix_new(fd);
567 if (!channel) {
568 crm_err("No memory left to add fd=%d", fd);
569 return -ENOMEM;
570 }
571
572 if (adaptor->source) {
573 g_source_remove(adaptor->source);
574 adaptor->source = 0;
575 }
576
577
578 evts |= (G_IO_HUP | G_IO_NVAL | G_IO_ERR);
579
580 adaptor->fn = fn;
581 adaptor->events = evts;
582 adaptor->data = data;
583 adaptor->p = p;
584 adaptor->is_used++;
585 adaptor->source =
586 g_io_add_watch_full(channel, conv_prio_libqb2glib(p), evts,
587 gio_read_socket, adaptor, gio_poll_destroy);
588
589
590
591
592
593
594
595
596
597
598 g_io_channel_unref(channel);
599
600 crm_trace("Added to mainloop with gsource id=%d", adaptor->source);
601 if (adaptor->source > 0) {
602 return 0;
603 }
604
605 return -EINVAL;
606 }
607
608 static int32_t
609 gio_poll_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t evts,
610 void *data, qb_ipcs_dispatch_fn_t fn)
611 {
612 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_TRUE);
613 }
614
615 static int32_t
616 gio_poll_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t evts,
617 void *data, qb_ipcs_dispatch_fn_t fn)
618 {
619 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_FALSE);
620 }
621
622 static int32_t
623 gio_poll_dispatch_del(int32_t fd)
624 {
625 struct gio_to_qb_poll *adaptor;
626
627 crm_trace("Looking for fd=%d", fd);
628 if (qb_array_index(gio_map, fd, (void **)&adaptor) == 0) {
629 if (adaptor->source) {
630 g_source_remove(adaptor->source);
631 adaptor->source = 0;
632 }
633 }
634 return 0;
635 }
636
637 struct qb_ipcs_poll_handlers gio_poll_funcs = {
638 .job_add = NULL,
639 .dispatch_add = gio_poll_dispatch_add,
640 .dispatch_mod = gio_poll_dispatch_mod,
641 .dispatch_del = gio_poll_dispatch_del,
642 };
643
644 static enum qb_ipc_type
645 pick_ipc_type(enum qb_ipc_type requested)
646 {
647 const char *env = getenv("PCMK_ipc_type");
648
649 if (env && strcmp("shared-mem", env) == 0) {
650 return QB_IPC_SHM;
651 } else if (env && strcmp("socket", env) == 0) {
652 return QB_IPC_SOCKET;
653 } else if (env && strcmp("posix", env) == 0) {
654 return QB_IPC_POSIX_MQ;
655 } else if (env && strcmp("sysv", env) == 0) {
656 return QB_IPC_SYSV_MQ;
657 } else if (requested == QB_IPC_NATIVE) {
658
659
660
661
662
663
664 return QB_IPC_SHM;
665 }
666 return requested;
667 }
668
669 qb_ipcs_service_t *
670 mainloop_add_ipc_server(const char *name, enum qb_ipc_type type,
671 struct qb_ipcs_service_handlers *callbacks)
672 {
673 return mainloop_add_ipc_server_with_prio(name, type, callbacks, QB_LOOP_MED);
674 }
675
676 qb_ipcs_service_t *
677 mainloop_add_ipc_server_with_prio(const char *name, enum qb_ipc_type type,
678 struct qb_ipcs_service_handlers *callbacks,
679 enum qb_loop_priority prio)
680 {
681 int rc = 0;
682 qb_ipcs_service_t *server = NULL;
683
684 if (gio_map == NULL) {
685 gio_map = qb_array_create_2(64, sizeof(struct gio_to_qb_poll), 1);
686 }
687
688 server = qb_ipcs_create(name, 0, pick_ipc_type(type), callbacks);
689
690 if (server == NULL) {
691 crm_err("Could not create %s IPC server: %s (%d)", name, pcmk_strerror(rc), rc);
692 return NULL;
693 }
694
695 if (prio != QB_LOOP_MED) {
696 qb_ipcs_request_rate_limit(server, conv_libqb_prio2ratelimit(prio));
697 }
698
699
700 qb_ipcs_enforce_buffer_size(server, crm_ipc_default_buffer_size());
701 qb_ipcs_poll_handlers_set(server, &gio_poll_funcs);
702
703 rc = qb_ipcs_run(server);
704 if (rc < 0) {
705 crm_err("Could not start %s IPC server: %s (%d)", name, pcmk_strerror(rc), rc);
706 return NULL;
707 }
708
709 return server;
710 }
711
712 void
713 mainloop_del_ipc_server(qb_ipcs_service_t * server)
714 {
715 if (server) {
716 qb_ipcs_destroy(server);
717 }
718 }
719
720 struct mainloop_io_s {
721 char *name;
722 void *userdata;
723
724 int fd;
725 guint source;
726 crm_ipc_t *ipc;
727 GIOChannel *channel;
728
729 int (*dispatch_fn_ipc) (const char *buffer, ssize_t length, gpointer userdata);
730 int (*dispatch_fn_io) (gpointer userdata);
731 void (*destroy_fn) (gpointer userdata);
732
733 };
734
735
736
737
738
739
740
741
742
743
744
745 static gboolean
746 mainloop_gio_callback(GIOChannel * gio, GIOCondition condition, gpointer data)
747 {
748 gboolean rc = G_SOURCE_CONTINUE;
749 mainloop_io_t *client = data;
750
751 CRM_ASSERT(client->fd == g_io_channel_unix_get_fd(gio));
752
753 if (condition & G_IO_IN) {
754 if (client->ipc) {
755 long read_rc = 0L;
756 int max = 10;
757
758 do {
759 read_rc = crm_ipc_read(client->ipc);
760 if (read_rc <= 0) {
761 crm_trace("Could not read IPC message from %s: %s (%ld)",
762 client->name, pcmk_strerror(read_rc), read_rc);
763
764 } else if (client->dispatch_fn_ipc) {
765 const char *buffer = crm_ipc_buffer(client->ipc);
766
767 crm_trace("New %ld-byte IPC message from %s "
768 "after I/O condition %d",
769 read_rc, client->name, (int) condition);
770 if (client->dispatch_fn_ipc(buffer, read_rc, client->userdata) < 0) {
771 crm_trace("Connection to %s no longer required", client->name);
772 rc = G_SOURCE_REMOVE;
773 }
774 }
775
776 } while ((rc == G_SOURCE_CONTINUE) && (read_rc > 0) && --max > 0);
777
778 } else {
779 crm_trace("New I/O event for %s after I/O condition %d",
780 client->name, (int) condition);
781 if (client->dispatch_fn_io) {
782 if (client->dispatch_fn_io(client->userdata) < 0) {
783 crm_trace("Connection to %s no longer required", client->name);
784 rc = G_SOURCE_REMOVE;
785 }
786 }
787 }
788 }
789
790 if (client->ipc && crm_ipc_connected(client->ipc) == FALSE) {
791 crm_err("Connection to %s closed " CRM_XS "client=%p condition=%d",
792 client->name, client, condition);
793 rc = G_SOURCE_REMOVE;
794
795 } else if (condition & (G_IO_HUP | G_IO_NVAL | G_IO_ERR)) {
796 crm_trace("The connection %s[%p] has been closed (I/O condition=%d)",
797 client->name, client, condition);
798 rc = G_SOURCE_REMOVE;
799
800 } else if ((condition & G_IO_IN) == 0) {
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828 crm_err("Strange condition: %d", condition);
829 }
830
831
832
833
834 return rc;
835 }
836
837 static void
838 mainloop_gio_destroy(gpointer c)
839 {
840 mainloop_io_t *client = c;
841 char *c_name = strdup(client->name);
842
843
844
845
846 crm_trace("Destroying client %s[%p]", c_name, c);
847
848 if (client->ipc) {
849 crm_ipc_close(client->ipc);
850 }
851
852 if (client->destroy_fn) {
853 void (*destroy_fn) (gpointer userdata) = client->destroy_fn;
854
855 client->destroy_fn = NULL;
856 destroy_fn(client->userdata);
857 }
858
859 if (client->ipc) {
860 crm_ipc_t *ipc = client->ipc;
861
862 client->ipc = NULL;
863 crm_ipc_destroy(ipc);
864 }
865
866 crm_trace("Destroyed client %s[%p]", c_name, c);
867
868 free(client->name); client->name = NULL;
869 free(client);
870
871 free(c_name);
872 }
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892 int
893 pcmk__add_mainloop_ipc(crm_ipc_t *ipc, int priority, void *userdata,
894 struct ipc_client_callbacks *callbacks,
895 mainloop_io_t **source)
896 {
897 CRM_CHECK((ipc != NULL) && (callbacks != NULL), return EINVAL);
898
899 if (!crm_ipc_connect(ipc)) {
900 int rc = errno;
901 crm_debug("Connection to %s failed: %d", crm_ipc_name(ipc), errno);
902 return rc;
903 }
904 *source = mainloop_add_fd(crm_ipc_name(ipc), priority, crm_ipc_get_fd(ipc),
905 userdata, NULL);
906 if (*source == NULL) {
907 int rc = errno;
908
909 crm_ipc_close(ipc);
910 return rc;
911 }
912 (*source)->ipc = ipc;
913 (*source)->destroy_fn = callbacks->destroy;
914 (*source)->dispatch_fn_ipc = callbacks->dispatch;
915 return pcmk_rc_ok;
916 }
917
918
919
920
921
922
923
924
925 guint
926 pcmk__mainloop_timer_get_period(mainloop_timer_t *timer)
927 {
928 if (timer) {
929 return timer->period_ms;
930 }
931 return 0;
932 }
933
934 mainloop_io_t *
935 mainloop_add_ipc_client(const char *name, int priority, size_t max_size,
936 void *userdata, struct ipc_client_callbacks *callbacks)
937 {
938 crm_ipc_t *ipc = crm_ipc_new(name, max_size);
939 mainloop_io_t *source = NULL;
940 int rc = pcmk__add_mainloop_ipc(ipc, priority, userdata, callbacks,
941 &source);
942
943 if (rc != pcmk_rc_ok) {
944 if (crm_log_level == LOG_STDOUT) {
945 fprintf(stderr, "Connection to %s failed: %s",
946 name, pcmk_rc_str(rc));
947 }
948 crm_ipc_destroy(ipc);
949 if (rc > 0) {
950 errno = rc;
951 } else {
952 errno = ENOTCONN;
953 }
954 return NULL;
955 }
956 return source;
957 }
958
959 void
960 mainloop_del_ipc_client(mainloop_io_t * client)
961 {
962 mainloop_del_fd(client);
963 }
964
965 crm_ipc_t *
966 mainloop_get_ipc_client(mainloop_io_t * client)
967 {
968 if (client) {
969 return client->ipc;
970 }
971 return NULL;
972 }
973
974 mainloop_io_t *
975 mainloop_add_fd(const char *name, int priority, int fd, void *userdata,
976 struct mainloop_fd_callbacks * callbacks)
977 {
978 mainloop_io_t *client = NULL;
979
980 if (fd >= 0) {
981 client = calloc(1, sizeof(mainloop_io_t));
982 if (client == NULL) {
983 return NULL;
984 }
985 client->name = strdup(name);
986 client->userdata = userdata;
987
988 if (callbacks) {
989 client->destroy_fn = callbacks->destroy;
990 client->dispatch_fn_io = callbacks->dispatch;
991 }
992
993 client->fd = fd;
994 client->channel = g_io_channel_unix_new(fd);
995 client->source =
996 g_io_add_watch_full(client->channel, priority,
997 (G_IO_IN | G_IO_HUP | G_IO_NVAL | G_IO_ERR), mainloop_gio_callback,
998 client, mainloop_gio_destroy);
999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009 g_io_channel_unref(client->channel);
1010 crm_trace("Added connection %d for %s[%p].%d", client->source, client->name, client, fd);
1011 } else {
1012 errno = EINVAL;
1013 }
1014
1015 return client;
1016 }
1017
1018 void
1019 mainloop_del_fd(mainloop_io_t * client)
1020 {
1021 if (client != NULL) {
1022 crm_trace("Removing client %s[%p]", client->name, client);
1023 if (client->source) {
1024
1025
1026
1027 g_source_remove(client->source);
1028 }
1029 }
1030 }
1031
1032 static GList *child_list = NULL;
1033
1034 pid_t
1035 mainloop_child_pid(mainloop_child_t * child)
1036 {
1037 return child->pid;
1038 }
1039
1040 const char *
1041 mainloop_child_name(mainloop_child_t * child)
1042 {
1043 return child->desc;
1044 }
1045
1046 int
1047 mainloop_child_timeout(mainloop_child_t * child)
1048 {
1049 return child->timeout;
1050 }
1051
1052 void *
1053 mainloop_child_userdata(mainloop_child_t * child)
1054 {
1055 return child->privatedata;
1056 }
1057
1058 void
1059 mainloop_clear_child_userdata(mainloop_child_t * child)
1060 {
1061 child->privatedata = NULL;
1062 }
1063
1064
1065 static void
1066 child_free(mainloop_child_t *child)
1067 {
1068 if (child->timerid != 0) {
1069 crm_trace("Removing timer %d", child->timerid);
1070 g_source_remove(child->timerid);
1071 child->timerid = 0;
1072 }
1073 free(child->desc);
1074 free(child);
1075 }
1076
1077
1078 static int
1079 child_kill_helper(mainloop_child_t *child)
1080 {
1081 int rc;
1082 if (child->flags & mainloop_leave_pid_group) {
1083 crm_debug("Kill pid %d only. leave group intact.", child->pid);
1084 rc = kill(child->pid, SIGKILL);
1085 } else {
1086 crm_debug("Kill pid %d's group", child->pid);
1087 rc = kill(-child->pid, SIGKILL);
1088 }
1089
1090 if (rc < 0) {
1091 if (errno != ESRCH) {
1092 crm_perror(LOG_ERR, "kill(%d, KILL) failed", child->pid);
1093 }
1094 return -errno;
1095 }
1096 return 0;
1097 }
1098
1099 static gboolean
1100 child_timeout_callback(gpointer p)
1101 {
1102 mainloop_child_t *child = p;
1103 int rc = 0;
1104
1105 child->timerid = 0;
1106 if (child->timeout) {
1107 crm_warn("%s process (PID %d) will not die!", child->desc, (int)child->pid);
1108 return FALSE;
1109 }
1110
1111 rc = child_kill_helper(child);
1112 if (rc == -ESRCH) {
1113
1114 return FALSE;
1115 }
1116
1117 child->timeout = TRUE;
1118 crm_debug("%s process (PID %d) timed out", child->desc, (int)child->pid);
1119
1120 child->timerid = g_timeout_add(5000, child_timeout_callback, child);
1121 return FALSE;
1122 }
1123
1124 static bool
1125 child_waitpid(mainloop_child_t *child, int flags)
1126 {
1127 int rc = 0;
1128 int core = 0;
1129 int signo = 0;
1130 int status = 0;
1131 int exitcode = 0;
1132 bool callback_needed = true;
1133
1134 rc = waitpid(child->pid, &status, flags);
1135 if (rc == 0) {
1136 crm_trace("Child process %d (%s) still active",
1137 child->pid, child->desc);
1138 callback_needed = false;
1139
1140 } else if (rc != child->pid) {
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150 signo = SIGCHLD;
1151 exitcode = 1;
1152 crm_notice("Wait for child process %d (%s) interrupted: %s",
1153 child->pid, child->desc, pcmk_strerror(errno));
1154
1155 } else if (WIFEXITED(status)) {
1156 exitcode = WEXITSTATUS(status);
1157 crm_trace("Child process %d (%s) exited with status %d",
1158 child->pid, child->desc, exitcode);
1159
1160 } else if (WIFSIGNALED(status)) {
1161 signo = WTERMSIG(status);
1162 crm_trace("Child process %d (%s) exited with signal %d (%s)",
1163 child->pid, child->desc, signo, strsignal(signo));
1164
1165 #ifdef WCOREDUMP
1166 } else if (WCOREDUMP(status)) {
1167 core = 1;
1168 crm_err("Child process %d (%s) dumped core",
1169 child->pid, child->desc);
1170 #endif
1171
1172 } else {
1173 crm_trace("Child process %d (%s) stopped or continued",
1174 child->pid, child->desc);
1175 callback_needed = false;
1176 }
1177
1178 if (callback_needed && child->callback) {
1179 child->callback(child, child->pid, core, signo, exitcode);
1180 }
1181 return callback_needed;
1182 }
1183
1184 static void
1185 child_death_dispatch(int signal)
1186 {
1187 for (GList *iter = child_list; iter; ) {
1188 GList *saved = iter;
1189 mainloop_child_t *child = iter->data;
1190
1191 iter = iter->next;
1192 if (child_waitpid(child, WNOHANG)) {
1193 crm_trace("Removing completed process %d from child list",
1194 child->pid);
1195 child_list = g_list_remove_link(child_list, saved);
1196 g_list_free(saved);
1197 child_free(child);
1198 }
1199 }
1200 }
1201
1202 static gboolean
1203 child_signal_init(gpointer p)
1204 {
1205 crm_trace("Installed SIGCHLD handler");
1206
1207 mainloop_add_signal(SIGCHLD, child_death_dispatch);
1208
1209
1210 child_death_dispatch(SIGCHLD);
1211 return FALSE;
1212 }
1213
1214 gboolean
1215 mainloop_child_kill(pid_t pid)
1216 {
1217 GList *iter;
1218 mainloop_child_t *child = NULL;
1219 mainloop_child_t *match = NULL;
1220
1221
1222 int waitflags = 0, rc = 0;
1223
1224 for (iter = child_list; iter != NULL && match == NULL; iter = iter->next) {
1225 child = iter->data;
1226 if (pid == child->pid) {
1227 match = child;
1228 }
1229 }
1230
1231 if (match == NULL) {
1232 return FALSE;
1233 }
1234
1235 rc = child_kill_helper(match);
1236 if(rc == -ESRCH) {
1237
1238
1239
1240
1241
1242 crm_trace("Waiting for signal that child process %d completed",
1243 match->pid);
1244 return TRUE;
1245
1246 } else if(rc != 0) {
1247
1248
1249
1250 waitflags = WNOHANG;
1251 }
1252
1253 if (!child_waitpid(match, waitflags)) {
1254
1255 return FALSE;
1256 }
1257
1258 child_list = g_list_remove(child_list, match);
1259 child_free(match);
1260 return TRUE;
1261 }
1262
1263
1264
1265
1266
1267
1268
1269
1270 void
1271 mainloop_child_add_with_flags(pid_t pid, int timeout, const char *desc, void *privatedata, enum mainloop_child_flags flags,
1272 void (*callback) (mainloop_child_t * p, pid_t pid, int core, int signo, int exitcode))
1273 {
1274 static bool need_init = TRUE;
1275 mainloop_child_t *child = g_new(mainloop_child_t, 1);
1276
1277 child->pid = pid;
1278 child->timerid = 0;
1279 child->timeout = FALSE;
1280 child->privatedata = privatedata;
1281 child->callback = callback;
1282 child->flags = flags;
1283
1284 if(desc) {
1285 child->desc = strdup(desc);
1286 }
1287
1288 if (timeout) {
1289 child->timerid = g_timeout_add(timeout, child_timeout_callback, child);
1290 }
1291
1292 child_list = g_list_append(child_list, child);
1293
1294 if(need_init) {
1295 need_init = FALSE;
1296
1297
1298
1299
1300 g_timeout_add(1, child_signal_init, NULL);
1301 }
1302 }
1303
1304 void
1305 mainloop_child_add(pid_t pid, int timeout, const char *desc, void *privatedata,
1306 void (*callback) (mainloop_child_t * p, pid_t pid, int core, int signo, int exitcode))
1307 {
1308 mainloop_child_add_with_flags(pid, timeout, desc, privatedata, 0, callback);
1309 }
1310
1311 static gboolean
1312 mainloop_timer_cb(gpointer user_data)
1313 {
1314 int id = 0;
1315 bool repeat = FALSE;
1316 struct mainloop_timer_s *t = user_data;
1317
1318 CRM_ASSERT(t != NULL);
1319
1320 id = t->id;
1321 t->id = 0;
1322
1323
1324
1325 if(t->cb) {
1326 crm_trace("Invoking callbacks for timer %s", t->name);
1327 repeat = t->repeat;
1328 if(t->cb(t->userdata) == FALSE) {
1329 crm_trace("Timer %s complete", t->name);
1330 repeat = FALSE;
1331 }
1332 }
1333
1334 if(repeat) {
1335
1336 t->id = id;
1337 }
1338
1339 return repeat;
1340 }
1341
1342 bool
1343 mainloop_timer_running(mainloop_timer_t *t)
1344 {
1345 if(t && t->id != 0) {
1346 return TRUE;
1347 }
1348 return FALSE;
1349 }
1350
1351 void
1352 mainloop_timer_start(mainloop_timer_t *t)
1353 {
1354 mainloop_timer_stop(t);
1355 if(t && t->period_ms > 0) {
1356 crm_trace("Starting timer %s", t->name);
1357 t->id = g_timeout_add(t->period_ms, mainloop_timer_cb, t);
1358 }
1359 }
1360
1361 void
1362 mainloop_timer_stop(mainloop_timer_t *t)
1363 {
1364 if(t && t->id != 0) {
1365 crm_trace("Stopping timer %s", t->name);
1366 g_source_remove(t->id);
1367 t->id = 0;
1368 }
1369 }
1370
1371 guint
1372 mainloop_timer_set_period(mainloop_timer_t *t, guint period_ms)
1373 {
1374 guint last = 0;
1375
1376 if(t) {
1377 last = t->period_ms;
1378 t->period_ms = period_ms;
1379 }
1380
1381 if(t && t->id != 0 && last != t->period_ms) {
1382 mainloop_timer_start(t);
1383 }
1384 return last;
1385 }
1386
1387 mainloop_timer_t *
1388 mainloop_timer_add(const char *name, guint period_ms, bool repeat, GSourceFunc cb, void *userdata)
1389 {
1390 mainloop_timer_t *t = calloc(1, sizeof(mainloop_timer_t));
1391
1392 if(t) {
1393 if(name) {
1394 t->name = crm_strdup_printf("%s-%u-%d", name, period_ms, repeat);
1395 } else {
1396 t->name = crm_strdup_printf("%p-%u-%d", t, period_ms, repeat);
1397 }
1398 t->id = 0;
1399 t->period_ms = period_ms;
1400 t->repeat = repeat;
1401 t->cb = cb;
1402 t->userdata = userdata;
1403 crm_trace("Created timer %s with %p %p", t->name, userdata, t->userdata);
1404 }
1405 return t;
1406 }
1407
1408 void
1409 mainloop_timer_del(mainloop_timer_t *t)
1410 {
1411 if(t) {
1412 crm_trace("Destroying timer %s", t->name);
1413 mainloop_timer_stop(t);
1414 free(t->name);
1415 free(t);
1416 }
1417 }
1418
1419
1420
1421
1422
1423 static gboolean
1424 drain_timeout_cb(gpointer user_data)
1425 {
1426 bool *timeout_popped = (bool*) user_data;
1427
1428 *timeout_popped = TRUE;
1429 return FALSE;
1430 }
1431
1432
1433
1434
1435
1436
1437
1438 void
1439 pcmk_quit_main_loop(GMainLoop *mloop, unsigned int n)
1440 {
1441 if ((mloop != NULL) && g_main_loop_is_running(mloop)) {
1442 GMainContext *ctx = g_main_loop_get_context(mloop);
1443
1444
1445
1446
1447 for (int i = 0; (i < n) && g_main_context_pending(ctx); ++i) {
1448 g_main_context_dispatch(ctx);
1449 }
1450 g_main_loop_quit(mloop);
1451 }
1452 }
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466 void
1467 pcmk_drain_main_loop(GMainLoop *mloop, guint timer_ms, bool (*check)(guint))
1468 {
1469 bool timeout_popped = FALSE;
1470 guint timer = 0;
1471 GMainContext *ctx = NULL;
1472
1473 CRM_CHECK(mloop && check, return);
1474
1475 ctx = g_main_loop_get_context(mloop);
1476 if (ctx) {
1477 time_t start_time = time(NULL);
1478
1479 timer = g_timeout_add(timer_ms, drain_timeout_cb, &timeout_popped);
1480 while (!timeout_popped
1481 && check(timer_ms - (time(NULL) - start_time) * 1000)) {
1482 g_main_context_iteration(ctx, TRUE);
1483 }
1484 }
1485 if (!timeout_popped && (timer > 0)) {
1486 g_source_remove(timer);
1487 }
1488 }
1489
1490
1491
1492
1493 #include <crm/common/mainloop_compat.h>
1494
1495 gboolean
1496 crm_signal(int sig, void (*dispatch) (int sig))
1497 {
1498 return crm_signal_handler(sig, dispatch) != SIG_ERR;
1499 }
1500
1501
1502