1 /*
2 * Copyright 2004-2026 the Pacemaker project contributors
3 *
4 * The version control history for this file may have further details.
5 *
6 * This source code is licensed under the GNU Lesser General Public License
7 * version 2.1 or later (LGPLv2.1+) WITHOUT ANY WARRANTY.
8 */
9
10 #include <crm_internal.h>
11
12 #include <stdbool.h>
13 #include <stdlib.h>
14 #include <string.h>
15 #include <signal.h>
16 #include <errno.h>
17
18 #include <sys/wait.h>
19
20 #include <crm/crm.h>
21 #include <crm/common/xml.h>
22 #include <crm/common/mainloop.h>
23
24 #include <qb/qbarray.h>
25
26 struct trigger_s {
27 GSource source;
28 gboolean running;
29 gboolean trigger;
30 void *user_data;
31 unsigned int id;
32 };
33
34 struct mainloop_timer_s {
35 unsigned int id;
36 unsigned int period_ms;
37 bool repeat;
38 char *name;
39 GSourceFunc cb;
40 void *userdata;
41 };
42
43 static GList *child_list = NULL;
44 static qb_array_t *gio_map = NULL;
45
46 static void
47 child_free(mainloop_child_t *child)
48 {
49 if (child->timerid != 0) {
50 pcmk__trace("Removing timer %d", child->timerid);
51 g_source_remove(child->timerid);
52 child->timerid = 0;
53 }
54
55 free(child->desc);
56 free(child);
57 }
58
59 static gboolean
60 crm_trigger_prepare(GSource *source, int *timeout)
61 {
62 crm_trigger_t *trig = (crm_trigger_t *) source;
63
64 /* cluster-glue's FD and IPC related sources make use of
65 * g_source_add_poll() but do not set a timeout in their prepare
66 * functions
67 *
68 * This means mainloop's poll() will block until an event for one
69 * of these sources occurs - any /other/ type of source, such as
70 * this one or g_idle_*, that doesn't use g_source_add_poll() is
71 * S-O-L and won't be processed until there is something fd-based
72 * happens.
73 *
74 * Luckily the timeout we can set here affects all sources and
75 * puts an upper limit on how long poll() can take.
76 *
77 * So unconditionally set a small-ish timeout, not too small that
78 * we're in constant motion, which will act as an upper bound on
79 * how long the signal handling might be delayed for.
80 */
81 *timeout = 500; /* Timeout in ms */
82
83 return trig->trigger;
84 }
85
86 static gboolean
87 crm_trigger_check(GSource * source)
88 {
89 crm_trigger_t *trig = (crm_trigger_t *) source;
90
91 return trig->trigger;
92 }
93
94 /*!
95 * \internal
96 * \brief GSource dispatch function for crm_trigger_t
97 *
98 * \param[in] source crm_trigger_t being dispatched
99 * \param[in] callback Callback passed at source creation
100 * \param[in,out] userdata User data passed at source creation
101 *
102 * \return G_SOURCE_REMOVE to remove source, G_SOURCE_CONTINUE to keep it
103 */
104 static gboolean
105 crm_trigger_dispatch(GSource *source, GSourceFunc callback, void *userdata)
106 {
107 gboolean rc = G_SOURCE_CONTINUE;
108 crm_trigger_t *trig = (crm_trigger_t *) source;
109
110 if (trig->running) {
111 /* Wait until the existing job is complete before starting the next one */
112 return G_SOURCE_CONTINUE;
113 }
114 trig->trigger = FALSE;
115
116 if (callback) {
117 int callback_rc = callback(trig->user_data);
118
119 if (callback_rc < 0) {
120 pcmk__trace("Trigger handler %p not yet complete", trig);
121 trig->running = TRUE;
122 } else if (callback_rc == 0) {
123 rc = G_SOURCE_REMOVE;
124 }
125 }
126 return rc;
127 }
128
129 static void
130 crm_trigger_finalize(GSource * source)
131 {
132 pcmk__trace("Trigger %p destroyed", source);
133 }
134
135 static GSourceFuncs crm_trigger_funcs = {
136 crm_trigger_prepare,
137 crm_trigger_check,
138 crm_trigger_dispatch,
139 crm_trigger_finalize,
140 };
141
142 static crm_trigger_t *
143 mainloop_setup_trigger(GSource * source, int priority,
144 int (*dispatch)(void *user_data), void *userdata)
145 {
146 crm_trigger_t *trigger = NULL;
147
148 trigger = (crm_trigger_t *) source;
149
150 trigger->id = 0;
151 trigger->trigger = FALSE;
152 trigger->user_data = userdata;
153
154 if (dispatch) {
155 g_source_set_callback(source, dispatch, trigger, NULL);
156 }
157
158 g_source_set_priority(source, priority);
159 g_source_set_can_recurse(source, FALSE);
160
161 trigger->id = g_source_attach(source, NULL);
162 return trigger;
163 }
164
165 void
166 mainloop_trigger_complete(crm_trigger_t * trig)
167 {
168 pcmk__trace("Trigger handler %p complete", trig);
169 trig->running = FALSE;
170 }
171
172 /*!
173 * \brief Create a trigger to be used as a mainloop source
174 *
175 * \param[in] priority Relative priority of source (lower number is higher priority)
176 * \param[in] dispatch Trigger dispatch function (should return 0 to remove the
177 * trigger from the mainloop, -1 if the trigger should be
178 * kept but the job is still running and not complete, and
179 * 1 if the trigger should be kept and the job is complete)
180 * \param[in] userdata Pointer to pass to \p dispatch
181 *
182 * \return Newly allocated mainloop source for trigger
183 */
184 crm_trigger_t *
185 mainloop_add_trigger(int priority, int (*dispatch) (void *user_data),
186 void *userdata)
187 {
188 GSource *source = NULL;
189
190 pcmk__assert(sizeof(crm_trigger_t) > sizeof(GSource));
191 source = g_source_new(&crm_trigger_funcs, sizeof(crm_trigger_t));
192
193 return mainloop_setup_trigger(source, priority, dispatch, userdata);
194 }
195
196 void
197 mainloop_set_trigger(crm_trigger_t * source)
198 {
199 if(source) {
200 source->trigger = TRUE;
201 }
202 }
203
204 gboolean
205 mainloop_destroy_trigger(crm_trigger_t * source)
206 {
207 GSource *gs = NULL;
208
209 if(source == NULL) {
210 return TRUE;
211 }
212
213 gs = (GSource *)source;
214
215 g_source_destroy(gs); /* Remove from mainloop, ref_count-- */
216 g_source_unref(gs); /* The caller no longer carries a reference to source
217 *
218 * At this point the source should be free'd,
219 * unless we're currently processing said
220 * source, in which case mainloop holds an
221 * additional reference and it will be free'd
222 * once our processing completes
223 */
224 return TRUE;
225 }
226
227 // Define a custom glib source for signal handling
228
229 // Data structure for custom glib source
230 typedef struct {
231 crm_trigger_t trigger; // trigger that invoked source (must be first)
232 void (*handler) (int sig); // signal handler
233 int signal; // signal that was received
234 } crm_signal_t;
235
236 // Table to associate signal handlers with signal numbers
237 static crm_signal_t *crm_signals[NSIG];
238
239 /*!
240 * \internal
241 * \brief Dispatch an event from custom glib source for signals
242 *
243 * Given an signal event, clear the event trigger and call any registered
244 * signal handler.
245 *
246 * \param[in] source glib source that triggered this dispatch
247 * \param[in] callback (ignored)
248 * \param[in] userdata (ignored)
249 */
250 static gboolean
251 crm_signal_dispatch(GSource *source, GSourceFunc callback, void *userdata)
252 {
253 crm_signal_t *sig = (crm_signal_t *) source;
254
255 if(sig->signal != SIGCHLD) {
256 pcmk__notice("Caught '%s' signal " QB_XS " %d (%s handler)",
257 strsignal(sig->signal), sig->signal,
258 ((sig->handler != NULL)? "invoking" : "no"));
259 }
260
261 sig->trigger.trigger = FALSE;
262 if (sig->handler) {
263 sig->handler(sig->signal);
264 }
265 return TRUE;
266 }
267
268 /*!
269 * \internal
270 * \brief Handle a signal by setting a trigger for signal source
271 *
272 * \param[in] sig Signal number that was received
273 *
274 * \note This is the true signal handler for the mainloop signal source, and
275 * must be async-safe.
276 */
277 static void
278 mainloop_signal_handler(int sig)
279 {
280 if (sig > 0 && sig < NSIG && crm_signals[sig] != NULL) {
281 mainloop_set_trigger((crm_trigger_t *) crm_signals[sig]);
282 }
283 }
284
285 // Functions implementing our custom glib source for signal handling
286 static GSourceFuncs crm_signal_funcs = {
287 crm_trigger_prepare,
288 crm_trigger_check,
289 crm_signal_dispatch,
290 crm_trigger_finalize,
291 };
292
293 /*!
294 * \internal
295 * \brief Set a true signal handler
296 *
297 * signal()-like interface to sigaction()
298 *
299 * \param[in] sig Signal number to register handler for
300 * \param[in] dispatch Signal handler
301 *
302 * \return The previous value of the signal handler, or SIG_ERR on error
303 * \note The dispatch function must be async-safe.
304 */
305 sighandler_t
306 crm_signal_handler(int sig, sighandler_t dispatch)
307 {
308 sigset_t mask;
309 struct sigaction sa;
310 struct sigaction old;
311
312 if (sigemptyset(&mask) < 0) {
313 pcmk__err("Could not %sset handler for signal %d: %s",
314 ((dispatch == NULL)? "un" : ""), sig, strerror(errno));
315 return SIG_ERR;
316 }
317
318 memset(&sa, 0, sizeof(struct sigaction));
319 sa.sa_handler = dispatch;
320 sa.sa_flags = SA_RESTART;
321 sa.sa_mask = mask;
322
323 if (sigaction(sig, &sa, &old) < 0) {
324 pcmk__err("Could not %sset handler for signal %d: %s",
325 ((dispatch == NULL)? "un" : ""), sig, strerror(errno));
326 return SIG_ERR;
327 }
328 return old.sa_handler;
329 }
330
331 static void
332 mainloop_destroy_signal_entry(int sig)
333 {
334 crm_signal_t *tmp = crm_signals[sig];
335
336 if (tmp != NULL) {
337 crm_signals[sig] = NULL;
338 pcmk__trace("Unregistering mainloop handler for signal %d", sig);
339 mainloop_destroy_trigger((crm_trigger_t *) tmp);
340 }
341 }
342
343 /*!
344 * \internal
345 * \brief Add a signal handler to a mainloop
346 *
347 * \param[in] sig Signal number to handle
348 * \param[in] dispatch Signal handler function (\c NULL to ignore the signal)
349 *
350 * \note The true signal handler merely sets a mainloop trigger to call this
351 * dispatch function via the mainloop. Therefore, the dispatch function
352 * does not need to be async-safe.
353 * \note The added signal handler gets freed by \c mainloop_cleanup() if it is
354 * not freed manually using \c mainloop_destroy_signal().
355 */
356 gboolean
357 mainloop_add_signal(int sig, void (*dispatch) (int sig))
358 {
359 GSource *source = NULL;
360 int priority = G_PRIORITY_HIGH - 1;
361
362 if (sig == SIGTERM) {
363 /* TERM is higher priority than other signals,
364 * signals are higher priority than other ipc.
365 * Yes, minus: smaller is "higher"
366 */
367 priority--;
368 }
369
370 if (sig >= NSIG || sig < 0) {
371 pcmk__err("Signal %d is out of range", sig);
372 return FALSE;
373
374 } else if (crm_signals[sig] != NULL && crm_signals[sig]->handler == dispatch) {
375 pcmk__trace("Signal handler for %d is already installed", sig);
376 return TRUE;
377
378 } else if (crm_signals[sig] != NULL) {
379 pcmk__err("Different signal handler for %d is already installed", sig);
380 return FALSE;
381 }
382
383 pcmk__assert(sizeof(crm_signal_t) > sizeof(GSource));
384 source = g_source_new(&crm_signal_funcs, sizeof(crm_signal_t));
385
386 crm_signals[sig] = (crm_signal_t *) mainloop_setup_trigger(source, priority, NULL, NULL);
387 pcmk__assert(crm_signals[sig] != NULL);
388
389 crm_signals[sig]->handler = dispatch;
390 crm_signals[sig]->signal = sig;
391
392 if (crm_signal_handler(sig, mainloop_signal_handler) == SIG_ERR) {
393 mainloop_destroy_signal_entry(sig);
394 return FALSE;
395 }
396
397 return TRUE;
398 }
399
400 gboolean
401 mainloop_destroy_signal(int sig)
402 {
403 if (sig >= NSIG || sig < 0) {
404 pcmk__err("Signal %d is out of range", sig);
405 return FALSE;
406
407 } else if (crm_signal_handler(sig, NULL) == SIG_ERR) {
408 // Error already logged
409 return FALSE;
410
411 } else if (crm_signals[sig] == NULL) {
412 return TRUE;
413 }
414 mainloop_destroy_signal_entry(sig);
415 return TRUE;
416 }
417
418 /*!
419 * \internal
420 * \brief Free data structures used for the mainloop
421 *
422 * \todo This is incomplete. Free other data structures created in this file.
423 */
424 void
425 mainloop_cleanup(void)
426 {
427 g_list_free_full(child_list, (GDestroyNotify) child_free);
428 child_list = NULL;
429
430 g_clear_pointer(&gio_map, qb_array_free);
431
432 for (int sig = 0; sig < NSIG; ++sig) {
433 mainloop_destroy_signal_entry(sig);
434 }
435 }
436
437 /*
438 * libqb...
439 */
440 struct gio_to_qb_poll {
441 int32_t is_used;
442 unsigned int source;
443 int32_t events;
444 void *data;
445 qb_ipcs_dispatch_fn_t fn;
446 enum qb_loop_priority p;
447 };
448
449 static gboolean
450 gio_read_socket(GIOChannel * gio, GIOCondition condition, void *data)
451 {
452 struct gio_to_qb_poll *adaptor = (struct gio_to_qb_poll *)data;
453 int fd = g_io_channel_unix_get_fd(gio);
454
455 pcmk__trace("%p.%d %d", data, fd, condition);
456
457 /* if this assert get's hit, then there is a race condition between
458 * when we destroy a fd and when mainloop actually gives it up */
459 pcmk__assert(adaptor->is_used > 0);
460
461 return (adaptor->fn(fd, condition, adaptor->data) == 0);
462 }
463
464 static void
465 gio_poll_destroy(void *data)
466 {
467 struct gio_to_qb_poll *adaptor = (struct gio_to_qb_poll *)data;
468
469 adaptor->is_used--;
470 pcmk__assert(adaptor->is_used >= 0);
471
472 if (adaptor->is_used == 0) {
473 pcmk__trace("Marking adaptor %p unused", adaptor);
474 adaptor->source = 0;
475 }
476 }
477
478 /*!
479 * \internal
480 * \brief Convert libqb's poll priority into GLib's one
481 *
482 * \param[in] prio libqb's poll priority (#QB_LOOP_MED assumed as fallback)
483 *
484 * \return best matching GLib's priority
485 */
486 static int
487 conv_prio_libqb2glib(enum qb_loop_priority prio)
488 {
489 switch (prio) {
490 case QB_LOOP_LOW: return G_PRIORITY_LOW;
491 case QB_LOOP_HIGH: return G_PRIORITY_HIGH;
492 default: return G_PRIORITY_DEFAULT; // QB_LOOP_MED
493 }
494 }
495
496 /*!
497 * \internal
498 * \brief Convert libqb's poll priority to rate limiting spec
499 *
500 * \param[in] prio libqb's poll priority (#QB_LOOP_MED assumed as fallback)
501 *
502 * \return best matching rate limiting spec
503 * \note This is the inverse of libqb's qb_ipcs_request_rate_limit().
504 */
505 static enum qb_ipcs_rate_limit
506 conv_libqb_prio2ratelimit(enum qb_loop_priority prio)
507 {
508 switch (prio) {
509 case QB_LOOP_LOW: return QB_IPCS_RATE_SLOW;
510 case QB_LOOP_HIGH: return QB_IPCS_RATE_FAST;
511 default: return QB_IPCS_RATE_NORMAL; // QB_LOOP_MED
512 }
513 }
514
515 static int32_t
516 gio_poll_dispatch_update(enum qb_loop_priority p, int32_t fd, int32_t evts,
517 void *data, qb_ipcs_dispatch_fn_t fn, int32_t add)
518 {
519 struct gio_to_qb_poll *adaptor;
520 GIOChannel *channel;
521 int32_t res = 0;
522
523 res = qb_array_index(gio_map, fd, (void **)&adaptor);
524 if (res < 0) {
525 pcmk__err("Array lookup failed for fd=%d: %d", fd, res);
526 return res;
527 }
528
529 pcmk__trace("Adding fd=%d to mainloop as adaptor %p", fd, adaptor);
530
531 if (add && adaptor->source) {
532 pcmk__err("Adaptor for descriptor %d is still in-use", fd);
533 return -EEXIST;
534 }
535 if (!add && !adaptor->is_used) {
536 pcmk__err("Adaptor for descriptor %d is not in-use", fd);
537 return -ENOENT;
538 }
539
540 /* channel is created with ref_count = 1 */
541 channel = g_io_channel_unix_new(fd);
542 if (!channel) {
543 pcmk__err("No memory left to add fd=%d", fd);
544 return -ENOMEM;
545 }
546
547 if (adaptor->source) {
548 g_source_remove(adaptor->source);
549 adaptor->source = 0;
550 }
551
552 /* Because unlike the poll() API, glib doesn't tell us about HUPs by default */
553 evts |= (G_IO_HUP | G_IO_NVAL | G_IO_ERR);
554
555 adaptor->fn = fn;
556 adaptor->events = evts;
557 adaptor->data = data;
558 adaptor->p = p;
559 adaptor->is_used++;
560 adaptor->source =
561 g_io_add_watch_full(channel, conv_prio_libqb2glib(p), evts,
562 gio_read_socket, adaptor, gio_poll_destroy);
563
564 /* Now that mainloop now holds a reference to channel,
565 * thanks to g_io_add_watch_full(), drop ours from g_io_channel_unix_new().
566 *
567 * This means that channel will be free'd by:
568 * g_main_context_dispatch()
569 * -> g_source_destroy_internal()
570 * -> g_source_callback_unref()
571 * shortly after gio_poll_destroy() completes
572 */
573 g_io_channel_unref(channel);
574
575 pcmk__trace("Added to mainloop with gsource id=%d", adaptor->source);
576 if (adaptor->source > 0) {
577 return 0;
578 }
579
580 return -EINVAL;
581 }
582
583 static int32_t
584 gio_poll_dispatch_add(enum qb_loop_priority p, int32_t fd, int32_t evts,
585 void *data, qb_ipcs_dispatch_fn_t fn)
586 {
587 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_TRUE);
588 }
589
590 static int32_t
591 gio_poll_dispatch_mod(enum qb_loop_priority p, int32_t fd, int32_t evts,
592 void *data, qb_ipcs_dispatch_fn_t fn)
593 {
594 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_FALSE);
595 }
596
597 static int32_t
598 gio_poll_dispatch_del(int32_t fd)
599 {
600 struct gio_to_qb_poll *adaptor;
601
602 pcmk__trace("Looking for fd=%d", fd);
603 if (qb_array_index(gio_map, fd, (void **)&adaptor) == 0) {
604 if (adaptor->source) {
605 g_source_remove(adaptor->source);
606 adaptor->source = 0;
607 }
608 }
609 return 0;
610 }
611
612 struct qb_ipcs_poll_handlers gio_poll_funcs = {
613 .job_add = NULL,
614 .dispatch_add = gio_poll_dispatch_add,
615 .dispatch_mod = gio_poll_dispatch_mod,
616 .dispatch_del = gio_poll_dispatch_del,
617 };
618
619 qb_ipcs_service_t *
620 mainloop_add_ipc_server(const char *name, enum qb_ipc_type type,
621 struct qb_ipcs_service_handlers *callbacks)
622 {
623 return mainloop_add_ipc_server_with_prio(name, type, callbacks, QB_LOOP_MED);
624 }
625
626 qb_ipcs_service_t *
627 mainloop_add_ipc_server_with_prio(const char *name, enum qb_ipc_type type,
628 struct qb_ipcs_service_handlers *callbacks,
629 enum qb_loop_priority prio)
630 {
631 int rc = 0;
632 qb_ipcs_service_t *server = NULL;
633
634 if (gio_map == NULL) {
635 gio_map = qb_array_create_2(64, sizeof(struct gio_to_qb_poll), 1);
636 }
637
638 server = qb_ipcs_create(name, 0, QB_IPC_SHM, callbacks);
639
640 if (server == NULL) {
641 pcmk__err("Could not create %s IPC server: %s (%d)", name,
642 pcmk_rc_str(errno), errno);
643 return NULL;
644 }
645
646 if (prio != QB_LOOP_MED) {
647 qb_ipcs_request_rate_limit(server, conv_libqb_prio2ratelimit(prio));
648 }
649
650 // Enforce a minimum IPC buffer size on all clients
651 qb_ipcs_enforce_buffer_size(server, crm_ipc_default_buffer_size());
652 qb_ipcs_poll_handlers_set(server, &gio_poll_funcs);
653
654 rc = qb_ipcs_run(server);
655 if (rc < 0) {
656 pcmk__err("Could not start %s IPC server: %s (%d)", name,
657 pcmk_strerror(rc), rc);
658 return NULL; // qb_ipcs_run() destroys server on failure
659 }
660
661 return server;
662 }
663
664 void
665 mainloop_del_ipc_server(qb_ipcs_service_t * server)
666 {
667 if (server) {
668 qb_ipcs_destroy(server);
669 }
670 }
671
672 /*!
673 * \internal
674 * \brief I/O watch callback function (GIOFunc)
675 *
676 * \param[in] gio I/O channel being watched
677 * \param[in] condition I/O condition satisfied
678 * \param[in] data User data passed when source was created
679 *
680 * \return G_SOURCE_REMOVE to remove source, G_SOURCE_CONTINUE to keep it
681 */
682 static gboolean
683 mainloop_gio_callback(GIOChannel *gio, GIOCondition condition, void *data)
684 {
685 gboolean rc = G_SOURCE_CONTINUE;
686 mainloop_io_t *client = data;
687
688 pcmk__assert(client->fd == g_io_channel_unix_get_fd(gio));
689
690 if (condition & G_IO_IN) {
691 if (client->ipc) {
692 long read_rc = 0L;
693 int max = 10;
694
695 do {
696 read_rc = crm_ipc_read(client->ipc);
697 if (read_rc <= 0) {
698 pcmk__trace("Could not read IPC message from %s: %s (%ld)",
699 client->name, pcmk_strerror(read_rc), read_rc);
700
701 if (read_rc == -EAGAIN) {
702 continue;
703 }
704
705 } else if (client->dispatch_fn_ipc) {
706 const char *buffer = crm_ipc_buffer(client->ipc);
707
708 pcmk__trace("New %ld-byte IPC message from %s after I/O "
709 "condition %d",
710 read_rc, client->name, (int) condition);
711 if (client->dispatch_fn_ipc(buffer, read_rc, client->userdata) < 0) {
712 pcmk__trace("Connection to %s no longer required",
713 client->name);
714 rc = G_SOURCE_REMOVE;
715 }
716 }
717
718 pcmk__ipc_free_client_buffer(client->ipc);
719
720 } while ((rc == G_SOURCE_CONTINUE) && (--max > 0)
721 && ((read_rc > 0) || (read_rc == -EAGAIN)));
722
723 } else {
724 pcmk__trace("New I/O event for %s after I/O condition %d",
725 client->name, (int) condition);
726 if (client->dispatch_fn_io) {
727 if (client->dispatch_fn_io(client->userdata) < 0) {
728 pcmk__trace("Connection to %s no longer required",
729 client->name);
730 rc = G_SOURCE_REMOVE;
731 }
732 }
733 }
734 }
735
736 if (client->ipc && !crm_ipc_connected(client->ipc)) {
737 pcmk__err("Connection to %s closed " QB_XS " client=%p condition=%d",
738 client->name, client, condition);
739 rc = G_SOURCE_REMOVE;
740
741 } else if (condition & (G_IO_HUP | G_IO_NVAL | G_IO_ERR)) {
742 pcmk__trace("The connection %s[%p] has been closed (I/O condition=%d)",
743 client->name, client, condition);
744 rc = G_SOURCE_REMOVE;
745
746 } else if ((condition & G_IO_IN) == 0) {
747 /*
748 #define GLIB_SYSDEF_POLLIN =1
749 #define GLIB_SYSDEF_POLLPRI =2
750 #define GLIB_SYSDEF_POLLOUT =4
751 #define GLIB_SYSDEF_POLLERR =8
752 #define GLIB_SYSDEF_POLLHUP =16
753 #define GLIB_SYSDEF_POLLNVAL =32
754
755 typedef enum
756 {
757 G_IO_IN GLIB_SYSDEF_POLLIN,
758 G_IO_OUT GLIB_SYSDEF_POLLOUT,
759 G_IO_PRI GLIB_SYSDEF_POLLPRI,
760 G_IO_ERR GLIB_SYSDEF_POLLERR,
761 G_IO_HUP GLIB_SYSDEF_POLLHUP,
762 G_IO_NVAL GLIB_SYSDEF_POLLNVAL
763 } GIOCondition;
764
765 A bitwise combination representing a condition to watch for on an event source.
766
767 G_IO_IN There is data to read.
768 G_IO_OUT Data can be written (without blocking).
769 G_IO_PRI There is urgent data to read.
770 G_IO_ERR Error condition.
771 G_IO_HUP Hung up (the connection has been broken, usually for pipes and sockets).
772 G_IO_NVAL Invalid request. The file descriptor is not open.
773 */
774 pcmk__err("Strange condition: %d", condition);
775 }
776
777 /* G_SOURCE_REMOVE results in mainloop_gio_destroy() being called
778 * just before the source is removed from mainloop
779 */
780 return rc;
781 }
782
783 static void
|
(3) Event deallocator: |
Deallocator for "struct mainloop_io_s". |
| Also see events: |
[allocation][allocation] |
784 mainloop_gio_destroy(void *c)
785 {
786 mainloop_io_t *client = c;
787 char *c_name = strdup(client->name);
788
789 /* client->source is valid but about to be destroyed (ref_count == 0) in gmain.c
790 * client->channel will still have ref_count > 0... should be == 1
791 */
792 pcmk__trace("Destroying client %s[%p]", c_name, c);
793
794 if (client->ipc) {
795 crm_ipc_close(client->ipc);
796 }
797
798 if (client->destroy_fn) {
799 void (*destroy_fn)(void *userdata) = client->destroy_fn;
800
801 client->destroy_fn = NULL;
802 destroy_fn(client->userdata);
803 }
804
805 if (client->ipc) {
806 crm_ipc_t *ipc = client->ipc;
807
808 client->ipc = NULL;
809 crm_ipc_destroy(ipc);
810 }
811
812 pcmk__trace("Destroyed client %s[%p]", c_name, c);
813
814 free(client->name);
815 free(client);
816
817 free(c_name);
818 }
819
820 /*!
821 * \brief Connect to IPC and add it as a main loop source
822 *
823 * \param[in,out] ipc IPC connection to add
824 * \param[in] priority Event source priority to use for connection
825 * \param[in] userdata Data to register with callbacks
826 * \param[in] callbacks Dispatch and destroy callbacks for connection
827 * \param[out] source Newly allocated event source
828 *
829 * \return Standard Pacemaker return code
830 *
831 * \note On failure, the caller is still responsible for ipc. On success, the
832 * caller should call mainloop_del_ipc_client() when source is no longer
833 * needed, which will lead to the disconnection of the IPC later in the
834 * main loop if it is connected. However the IPC disconnects,
835 * mainloop_gio_destroy() will free ipc and source after calling the
836 * destroy callback.
837 */
838 int
839 pcmk__add_mainloop_ipc(crm_ipc_t *ipc, int priority, void *userdata,
840 const struct ipc_client_callbacks *callbacks,
841 mainloop_io_t **source)
842 {
843 int rc = pcmk_rc_ok;
844 int fd = -1;
845 const char *ipc_name = NULL;
846
847 CRM_CHECK((ipc != NULL) && (callbacks != NULL), return EINVAL);
848
849 ipc_name = pcmk__s(crm_ipc_name(ipc), "Pacemaker");
850 rc = pcmk__connect_generic_ipc(ipc);
851 if (rc != pcmk_rc_ok) {
852 pcmk__debug("Connection to %s failed: %s", ipc_name, pcmk_rc_str(rc));
853 return rc;
854 }
855
856 rc = pcmk__ipc_fd(ipc, &fd);
857 if (rc != pcmk_rc_ok) {
858 pcmk__debug("Could not obtain file descriptor for %s IPC: %s", ipc_name,
859 pcmk_rc_str(rc));
860 crm_ipc_close(ipc);
861 return rc;
862 }
863
864 *source = mainloop_add_fd(ipc_name, priority, fd, userdata, NULL);
865 if (*source == NULL) {
866 rc = errno;
867 crm_ipc_close(ipc);
868 return rc;
869 }
870
871 (*source)->ipc = ipc;
872 (*source)->destroy_fn = callbacks->destroy;
873 (*source)->dispatch_fn_ipc = callbacks->dispatch;
874 return pcmk_rc_ok;
875 }
876
877 /*!
878 * \brief Get period for mainloop timer
879 *
880 * \param[in] timer Timer
881 *
882 * \return Period in ms
883 */
884 unsigned int
885 pcmk__mainloop_timer_get_period(const mainloop_timer_t *timer)
886 {
887 if (timer) {
888 return timer->period_ms;
889 }
890 return 0;
891 }
892
893 mainloop_io_t *
894 mainloop_add_ipc_client(const char *name, int priority, size_t max_size,
895 void *userdata, struct ipc_client_callbacks *callbacks)
896 {
897 crm_ipc_t *ipc = crm_ipc_new(name, 0);
898 mainloop_io_t *source = NULL;
899 int rc = pcmk__add_mainloop_ipc(ipc, priority, userdata, callbacks,
900 &source);
901
902 if (rc != pcmk_rc_ok) {
903 if (crm_log_level == PCMK__LOG_STDOUT) {
904 fprintf(stderr, "Connection to %s failed: %s",
905 name, pcmk_rc_str(rc));
906 }
907 crm_ipc_destroy(ipc);
908 if (rc > 0) {
909 errno = rc;
910 } else {
911 errno = ENOTCONN;
912 }
913 return NULL;
914 }
915 return source;
916 }
917
918 void
919 mainloop_del_ipc_client(mainloop_io_t * client)
920 {
921 mainloop_del_fd(client);
922 }
923
924 crm_ipc_t *
925 mainloop_get_ipc_client(mainloop_io_t * client)
926 {
927 if (client) {
928 return client->ipc;
929 }
930 return NULL;
931 }
932
933 mainloop_io_t *
934 mainloop_add_fd(const char *name, int priority, int fd, void *userdata,
935 struct mainloop_fd_callbacks * callbacks)
936 {
937 mainloop_io_t *client = NULL;
938 const GIOCondition condition = G_IO_IN|G_IO_HUP|G_IO_NVAL|G_IO_ERR;
939
940 if (fd < 0) {
941 errno = EINVAL;
942 return NULL;
943 }
944
945 client = pcmk__assert_alloc(1, sizeof(mainloop_io_t));
946 client->name = pcmk__str_copy(name);
947 client->userdata = userdata;
948
949 if (callbacks != NULL) {
950 client->destroy_fn = callbacks->destroy;
951 client->dispatch_fn_io = callbacks->dispatch;
952 }
953
954 client->fd = fd;
|
CID (unavailable; MK=b13c37cce9c4cfa0c98303204ab57333) (#1 of 1): Resource not released (INCOMPLETE_DEALLOCATOR): |
|
(1) Event allocation: |
Memory is allocated. |
|
(2) Event allocation: |
The field "client->channel" is allocated, but not released in the identified deallocator. |
| Also see events: |
[deallocator] |
955 client->channel = g_io_channel_unix_new(fd);
956 client->source = g_io_add_watch_full(client->channel, priority, condition,
957 mainloop_gio_callback, client,
958 mainloop_gio_destroy);
959
960 /* Now that mainloop now holds a reference to channel, thanks to
961 * g_io_add_watch_full(), drop ours from g_io_channel_unix_new().
962 *
963 * This means that channel will be free'd by:
964 * g_main_context_dispatch() or g_source_remove()
965 * -> g_source_destroy_internal()
966 * -> g_source_callback_unref()
967 * shortly after mainloop_gio_destroy() completes
968 */
969 g_io_channel_unref(client->channel);
970
971 pcmk__trace("Added connection %d for %s[%p].%d", client->source,
972 client->name, client, fd);
973 return client;
974 }
975
976 void
977 mainloop_del_fd(mainloop_io_t *client)
978 {
979 if ((client == NULL) || (client->source == 0)) {
980 return;
981 }
982
983 pcmk__trace("Removing client %s[%p]", client->name, client);
984
985 /* g_source_remove() marks the source as destroyed, unsets the source
986 * callback (mainloop_gio_callback()), and destroys the callback data (the
987 * client) via the notify function (mainloop_gio_destroy()). We can rely on
988 * mainloop_gio_callback() not getting called again for this source, and on
989 * the client being destroyed.
990 */
991 g_source_remove(client->source);
992 }
993
994 pid_t
995 mainloop_child_pid(mainloop_child_t * child)
996 {
997 return child->pid;
998 }
999
1000 const char *
1001 mainloop_child_name(mainloop_child_t * child)
1002 {
1003 return child->desc;
1004 }
1005
1006 int
1007 mainloop_child_timeout(mainloop_child_t * child)
1008 {
1009 return child->timeout;
1010 }
1011
1012 void *
1013 mainloop_child_userdata(mainloop_child_t * child)
1014 {
1015 return child->privatedata;
1016 }
1017
1018 void
1019 mainloop_clear_child_userdata(mainloop_child_t * child)
1020 {
1021 child->privatedata = NULL;
1022 }
1023
1024 static int
1025 child_kill_helper(const mainloop_child_t *child)
1026 {
1027 int rc = 0;
1028
1029 if (pcmk__is_set(child->flags, mainloop_leave_pid_group)) {
1030 pcmk__debug("Killing PID %lld only. Leaving its process group intact.",
1031 (long long) child->pid);
1032 rc = kill(child->pid, SIGKILL);
1033
1034 } else {
1035 pcmk__debug("Killing PID %lld's entire process group",
1036 (long long) child->pid);
1037 rc = kill(-child->pid, SIGKILL);
1038 }
1039
1040 if (rc == 0) {
1041 return pcmk_rc_ok;
1042 }
1043
1044 rc = errno;
1045 if (rc == ESRCH) {
1046 return rc;
1047 }
1048
1049 pcmk__err("kill(%lld, KILL) failed: %s", (long long) child->pid,
1050 strerror(rc));
1051 return rc;
1052 }
1053
1054 static gboolean
1055 child_timeout_callback(void *p)
1056 {
1057 mainloop_child_t *child = p;
1058 int rc = pcmk_rc_ok;
1059
1060 child->timerid = 0;
1061 if (child->timeout) {
1062 pcmk__warn("%s process (PID %lld) will not die!", child->desc,
1063 (long long) child->pid);
1064 return FALSE;
1065 }
1066
1067 rc = child_kill_helper(child);
1068 if (rc == ESRCH) {
1069 /* Nothing left to do. pid doesn't exist */
1070 return FALSE;
1071 }
1072
1073 child->timeout = TRUE;
1074 pcmk__debug("%s process (PID %lld) timed out", child->desc,
1075 (long long) child->pid);
1076
1077 child->timerid = pcmk__create_timer(5000, child_timeout_callback, child);
1078 return FALSE;
1079 }
1080
1081 static bool
1082 child_waitpid(mainloop_child_t *child, int flags)
1083 {
1084 int rc = 0;
1085 int core = 0;
1086 int signo = 0;
1087 int status = 0;
1088 int exitcode = 0;
1089
1090 rc = waitpid(child->pid, &status, flags);
1091
1092 if (rc == 0) { // WNOHANG in flags, and child status is not available
1093 pcmk__trace("Child process %lld (%s) still active",
1094 (long long) child->pid, child->desc);
1095 return false;
1096 }
1097
1098 if (rc != child->pid) {
1099 /* According to POSIX, possible conditions:
1100 * - child->pid was non-positive (process group or any child),
1101 * and rc is specific child
1102 * - errno ECHILD (pid does not exist or is not child)
1103 * - errno EINVAL (invalid flags)
1104 * - errno EINTR (caller interrupted by signal)
1105 *
1106 * @TODO Handle these cases more specifically.
1107 */
1108 signo = SIGCHLD;
1109 exitcode = 1;
1110 pcmk__notice("Wait for child process %lld (%s) interrupted: %s",
1111 (long long) child->pid, child->desc, strerror(errno));
1112
1113 } else if (WIFEXITED(status)) {
1114 exitcode = WEXITSTATUS(status);
1115 pcmk__trace("Child process %lld (%s) exited with status %d",
1116 (long long) child->pid, child->desc, exitcode);
1117
1118 } else if (WIFSIGNALED(status)) {
1119 signo = WTERMSIG(status);
1120 pcmk__trace("Child process %lld (%s) exited with signal %d (%s)",
1121 (long long) child->pid, child->desc, signo,
1122 strsignal(signo));
1123
1124 #ifdef WCOREDUMP // AIX, SunOS, maybe others
1125 } else if (WCOREDUMP(status)) {
1126 core = 1;
1127 pcmk__err("Child process %lld (%s) dumped core", (long long) child->pid,
1128 child->desc);
1129 #endif
1130
1131 } else { // flags must contain WUNTRACED and/or WCONTINUED to reach this
1132 pcmk__trace("Child process %lld (%s) stopped or continued",
1133 (long long) child->pid, child->desc);
1134 return false;
1135 }
1136
1137 if (child->exit_fn != NULL) {
1138 child->exit_fn(child, core, signo, exitcode);
1139 }
1140
1141 return true;
1142 }
1143
1144 static void
1145 child_death_dispatch(int signal)
1146 {
1147 for (GList *iter = child_list; iter; ) {
1148 GList *saved = iter;
1149 mainloop_child_t *child = iter->data;
1150
1151 iter = iter->next;
1152 if (child_waitpid(child, WNOHANG)) {
1153 pcmk__trace("Removing completed process %lld from child list",
1154 (long long) child->pid);
1155 child_list = g_list_remove_link(child_list, saved);
1156 g_list_free(saved);
1157 child_free(child);
1158 }
1159 }
1160 }
1161
1162 static gboolean
1163 child_signal_init(void *p)
1164 {
1165 pcmk__trace("Installed SIGCHLD handler");
1166 /* Do NOT use g_child_watch_add() and friends, they rely on pthreads */
1167 mainloop_add_signal(SIGCHLD, child_death_dispatch);
1168
1169 /* In case they terminated before the signal handler was installed */
1170 child_death_dispatch(SIGCHLD);
1171 return FALSE;
1172 }
1173
1174 gboolean
1175 mainloop_child_kill(pid_t pid)
1176 {
1177 GList *iter;
1178 mainloop_child_t *child = NULL;
1179 mainloop_child_t *match = NULL;
1180 /* It is impossible to block SIGKILL, this allows us to
1181 * call waitpid without WNOHANG flag.*/
1182 int waitflags = 0;
1183 int rc = pcmk_rc_ok;
1184
1185 for (iter = child_list; iter != NULL && match == NULL; iter = iter->next) {
1186 child = iter->data;
1187 if (pid == child->pid) {
1188 match = child;
1189 }
1190 }
1191
1192 if (match == NULL) {
1193 return FALSE;
1194 }
1195
1196 rc = child_kill_helper(match);
1197 if (rc == ESRCH) {
1198 /* It's gone, but hasn't shown up in waitpid() yet. Wait until we get
1199 * SIGCHLD and let handler clean it up as normal (so we get the correct
1200 * return code/status). The blocking alternative would be to call
1201 * child_waitpid(match, 0).
1202 */
1203 pcmk__trace("Waiting for signal that child process %lld completed",
1204 (long long) match->pid);
1205 return TRUE;
1206 }
1207
1208 if (rc != pcmk_rc_ok) {
1209 /* If kill() failed for some other reason, set the WNOHANG flag, since
1210 * we can't be certain what happened.
1211 */
1212 waitflags = WNOHANG;
1213 }
1214
1215 if (!child_waitpid(match, waitflags)) {
1216 /* not much we can do if this occurs */
1217 return FALSE;
1218 }
1219
1220 child_list = g_list_remove(child_list, match);
1221 child_free(match);
1222 return TRUE;
1223 }
1224
1225 /* Create/Log a new tracked process
1226 * To track a process group, use -pid
1227 *
1228 * @TODO Using a non-positive pid (i.e. any child, or process group) would
1229 * likely not be useful since we will free the child after the first
1230 * completed process.
1231 */
1232 void
1233 mainloop_child_add_with_flags(pid_t pid, int timeout, const char *desc,
1234 void *privatedata,
1235 enum mainloop_child_flags flags,
1236 pcmk__mainloop_child_exit_fn_t exit_fn)
1237 {
1238 static bool need_init = TRUE;
1239 mainloop_child_t *child = pcmk__assert_alloc(1, sizeof(mainloop_child_t));
1240
1241 child->pid = pid;
1242 child->timerid = 0;
1243 child->timeout = FALSE;
1244 child->privatedata = privatedata;
1245 child->exit_fn = exit_fn;
1246 child->flags = flags;
1247 child->desc = pcmk__str_copy(desc);
1248
1249 if (timeout) {
1250 child->timerid = pcmk__create_timer(timeout, child_timeout_callback, child);
1251 }
1252
1253 child_list = g_list_append(child_list, child);
1254
1255 if(need_init) {
1256 need_init = FALSE;
1257 /* SIGCHLD processing has to be invoked from mainloop.
1258 * We do not want it to be possible to both add a child pid
1259 * to mainloop, and have the pid's exit callback invoked within
1260 * the same callstack. */
1261 pcmk__create_timer(1, child_signal_init, NULL);
1262 }
1263 }
1264
1265 void
1266 mainloop_child_add(pid_t pid, int timeout, const char *desc, void *privatedata,
1267 pcmk__mainloop_child_exit_fn_t exit_fn)
1268 {
1269 mainloop_child_add_with_flags(pid, timeout, desc, privatedata, 0, exit_fn);
1270 }
1271
1272 static gboolean
1273 mainloop_timer_cb(void *user_data)
1274 {
1275 int id = 0;
1276 bool repeat = FALSE;
1277 struct mainloop_timer_s *t = user_data;
1278
1279 pcmk__assert(t != NULL);
1280
1281 id = t->id;
1282 t->id = 0; /* Ensure it's unset during callbacks so that
1283 * mainloop_timer_running() works as expected
1284 */
1285
1286 if(t->cb) {
1287 pcmk__trace("Invoking callbacks for timer %s", t->name);
1288 repeat = t->repeat;
1289 if(t->cb(t->userdata) == FALSE) {
1290 pcmk__trace("Timer %s complete", t->name);
1291 repeat = FALSE;
1292 }
1293 }
1294
1295 if(repeat) {
1296 /* Restore if repeating */
1297 t->id = id;
1298 }
1299
1300 return repeat;
1301 }
1302
1303 bool
1304 mainloop_timer_running(mainloop_timer_t *t)
1305 {
1306 if(t && t->id != 0) {
1307 return TRUE;
1308 }
1309 return FALSE;
1310 }
1311
1312 void
1313 mainloop_timer_start(mainloop_timer_t *t)
1314 {
1315 mainloop_timer_stop(t);
1316 if(t && t->period_ms > 0) {
1317 pcmk__trace("Starting timer %s", t->name);
1318 t->id = pcmk__create_timer(t->period_ms, mainloop_timer_cb, t);
1319 }
1320 }
1321
1322 void
1323 mainloop_timer_stop(mainloop_timer_t *t)
1324 {
1325 if(t && t->id != 0) {
1326 pcmk__trace("Stopping timer %s", t->name);
1327 g_source_remove(t->id);
1328 t->id = 0;
1329 }
1330 }
1331
1332 unsigned int
1333 mainloop_timer_set_period(mainloop_timer_t *t, unsigned int period_ms)
1334 {
1335 unsigned int last = 0;
1336
1337 if(t) {
1338 last = t->period_ms;
1339 t->period_ms = period_ms;
1340 }
1341
1342 if(t && t->id != 0 && last != t->period_ms) {
1343 mainloop_timer_start(t);
1344 }
1345 return last;
1346 }
1347
1348 mainloop_timer_t *
1349 mainloop_timer_add(const char *name, unsigned int period_ms, bool repeat,
1350 GSourceFunc cb, void *userdata)
1351 {
1352 mainloop_timer_t *t = pcmk__assert_alloc(1, sizeof(mainloop_timer_t));
1353
1354 if (name != NULL) {
1355 t->name = pcmk__assert_asprintf("%s-%u-%d", name, period_ms, repeat);
1356 } else {
1357 t->name = pcmk__assert_asprintf("%p-%u-%d", t, period_ms, repeat);
1358 }
1359 t->id = 0;
1360 t->period_ms = period_ms;
1361 t->repeat = repeat;
1362 t->cb = cb;
1363 t->userdata = userdata;
1364 pcmk__trace("Created timer %s with %p %p", t->name, userdata, t->userdata);
1365 return t;
1366 }
1367
1368 void
1369 mainloop_timer_del(mainloop_timer_t *t)
1370 {
1371 if(t) {
1372 pcmk__trace("Destroying timer %s", t->name);
1373 mainloop_timer_stop(t);
1374 free(t->name);
1375 free(t);
1376 }
1377 }
1378
1379 /*
1380 * Helpers to make sure certain events aren't lost at shutdown
1381 */
1382
1383 static gboolean
1384 drain_timeout_cb(void *user_data)
1385 {
1386 bool *timeout_popped = (bool*) user_data;
1387
1388 *timeout_popped = TRUE;
1389 return FALSE;
1390 }
1391
1392 /*!
1393 * \brief Drain some remaining main loop events then quit it
1394 *
1395 * \param[in,out] mloop Main loop to drain and quit
1396 * \param[in] n Drain up to this many pending events
1397 */
1398 void
1399 pcmk_quit_main_loop(GMainLoop *mloop, unsigned int n)
1400 {
1401 if ((mloop != NULL) && g_main_loop_is_running(mloop)) {
1402 GMainContext *ctx = g_main_loop_get_context(mloop);
1403
1404 /* Drain up to n events in case some memory clean-up is pending
1405 * (helpful to reduce noise in valgrind output).
1406 */
1407 for (int i = 0; (i < n) && g_main_context_pending(ctx); ++i) {
1408 g_main_context_dispatch(ctx);
1409 }
1410 g_main_loop_quit(mloop);
1411 }
1412 }
1413
1414 /*!
1415 * \brief Process main loop events while a certain condition is met
1416 *
1417 * \param[in,out] mloop Main loop to process
1418 * \param[in] timer_ms Don't process longer than this amount of time
1419 * \param[in] check Function that returns true if events should be
1420 * processed
1421 *
1422 * \note This function is intended to be called at shutdown if certain important
1423 * events should not be missed. The caller would likely quit the main loop
1424 * or exit after calling this function. The check() function will be
1425 * passed the remaining timeout in milliseconds.
1426 */
1427 void
1428 pcmk_drain_main_loop(GMainLoop *mloop, unsigned int timer_ms,
1429 bool (*check)(unsigned int))
1430 {
1431 bool timeout_popped = FALSE;
1432 unsigned int timer = 0;
1433 GMainContext *ctx = NULL;
1434
1435 CRM_CHECK(mloop && check, return);
1436
1437 ctx = g_main_loop_get_context(mloop);
1438 if (ctx) {
1439 time_t start_time = time(NULL);
1440
1441 timer = pcmk__create_timer(timer_ms, drain_timeout_cb, &timeout_popped);
1442 while (!timeout_popped
1443 && check(timer_ms - (time(NULL) - start_time) * 1000)) {
1444 g_main_context_iteration(ctx, TRUE);
1445 }
1446 }
1447 if (!timeout_popped && (timer > 0)) {
1448 g_source_remove(timer);
1449 }
1450 }
1451