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