28 #include <qb/qbarray.h> 30 struct mainloop_child_s {
52 struct mainloop_timer_s {
62 crm_trigger_prepare(GSource * source, gint *
timeout)
89 crm_trigger_check(GSource * source)
107 crm_trigger_dispatch(GSource *source, GSourceFunc callback, gpointer userdata)
109 gboolean rc = G_SOURCE_CONTINUE;
114 return G_SOURCE_CONTINUE;
116 trig->trigger = FALSE;
119 int callback_rc = callback(trig->user_data);
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;
132 crm_trigger_finalize(GSource * source)
134 crm_trace(
"Trigger %p destroyed", source);
137 static GSourceFuncs crm_trigger_funcs = {
140 crm_trigger_dispatch,
141 crm_trigger_finalize,
145 mainloop_setup_trigger(GSource * source,
int priority,
int (*dispatch) (gpointer user_data),
153 trigger->trigger = FALSE;
154 trigger->user_data = userdata;
157 g_source_set_callback(source, dispatch, trigger, NULL);
160 g_source_set_priority(source, priority);
161 g_source_set_can_recurse(source, FALSE);
163 trigger->id = g_source_attach(source, NULL);
170 crm_trace(
"Trigger handler %p complete", trig);
171 trig->running = FALSE;
190 GSource *source = NULL;
193 source = g_source_new(&crm_trigger_funcs,
sizeof(
crm_trigger_t));
195 return mainloop_setup_trigger(source, priority, dispatch, userdata);
202 source->trigger = TRUE;
215 gs = (GSource *)source;
217 g_source_destroy(gs);
232 typedef struct signal_s {
234 void (*handler) (
int sig);
253 crm_signal_dispatch(GSource *source, GSourceFunc callback, gpointer userdata)
257 if(sig->signal != SIGCHLD) {
259 strsignal(sig->signal), sig->signal,
260 (sig->handler?
"invoking" :
"no"));
263 sig->trigger.trigger = FALSE;
265 sig->handler(sig->signal);
280 mainloop_signal_handler(
int sig)
282 if (sig > 0 && sig < NSIG && crm_signals[sig] != NULL) {
288 static GSourceFuncs crm_signal_funcs = {
292 crm_trigger_finalize,
312 struct sigaction old;
314 if (sigemptyset(&mask) < 0) {
315 crm_err(
"Could not set handler for signal %d: %s",
320 memset(&sa, 0,
sizeof(
struct sigaction));
321 sa.sa_handler = dispatch;
322 sa.sa_flags = SA_RESTART;
325 if (sigaction(sig, &sa, &old) < 0) {
326 crm_err(
"Could not set handler for signal %d: %s",
330 return old.sa_handler;
334 mainloop_destroy_signal_entry(
int sig)
338 crm_signals[sig] = NULL;
358 GSource *source = NULL;
359 int priority = G_PRIORITY_HIGH - 1;
361 if (sig == SIGTERM) {
369 if (sig >= NSIG || sig < 0) {
370 crm_err(
"Signal %d is out of range", sig);
373 }
else if (crm_signals[sig] != NULL && crm_signals[sig]->handler == dispatch) {
374 crm_trace(
"Signal handler for %d is already installed", sig);
377 }
else if (crm_signals[sig] != NULL) {
378 crm_err(
"Different signal handler for %d is already installed", sig);
383 source = g_source_new(&crm_signal_funcs,
sizeof(
crm_signal_t));
385 crm_signals[sig] = (
crm_signal_t *) mainloop_setup_trigger(source, priority, NULL, NULL);
388 crm_signals[sig]->handler = dispatch;
389 crm_signals[sig]->signal = sig;
392 mainloop_destroy_signal_entry(sig);
402 if (sig >= NSIG || sig < 0) {
403 crm_err(
"Signal %d is out of range", sig);
407 crm_perror(LOG_ERR,
"Could not uninstall signal handler for signal %d", sig);
410 }
else if (crm_signals[sig] == NULL) {
413 mainloop_destroy_signal_entry(sig);
417 static qb_array_t *gio_map = NULL;
423 qb_array_free(gio_map);
426 for (
int sig = 0; sig < NSIG; ++sig) {
427 mainloop_destroy_signal_entry(sig);
434 struct gio_to_qb_poll {
439 qb_ipcs_dispatch_fn_t fn;
440 enum qb_loop_priority p;
444 gio_read_socket(GIOChannel * gio, GIOCondition condition, gpointer
data)
446 struct gio_to_qb_poll *adaptor = (
struct gio_to_qb_poll *)
data;
447 gint fd = g_io_channel_unix_get_fd(gio);
455 return (adaptor->fn(fd, condition, adaptor->data) == 0);
459 gio_poll_destroy(gpointer
data)
461 struct gio_to_qb_poll *adaptor = (
struct gio_to_qb_poll *)
data;
466 if (adaptor->is_used == 0) {
467 crm_trace(
"Marking adaptor %p unused", adaptor);
481 conv_prio_libqb2glib(
enum qb_loop_priority prio)
484 case QB_LOOP_LOW:
return G_PRIORITY_LOW;
485 case QB_LOOP_HIGH:
return G_PRIORITY_HIGH;
486 default:
return G_PRIORITY_DEFAULT;
499 static enum qb_ipcs_rate_limit
500 conv_libqb_prio2ratelimit(
enum qb_loop_priority prio)
503 case QB_LOOP_LOW:
return QB_IPCS_RATE_SLOW;
504 case QB_LOOP_HIGH:
return QB_IPCS_RATE_FAST;
505 default:
return QB_IPCS_RATE_NORMAL;
510 gio_poll_dispatch_update(
enum qb_loop_priority p, int32_t fd, int32_t evts,
511 void *
data, qb_ipcs_dispatch_fn_t fn, int32_t add)
513 struct gio_to_qb_poll *adaptor;
517 res = qb_array_index(gio_map, fd, (
void **)&adaptor);
519 crm_err(
"Array lookup failed for fd=%d: %d", fd, res);
523 crm_trace(
"Adding fd=%d to mainloop as adaptor %p", fd, adaptor);
525 if (add && adaptor->source) {
526 crm_err(
"Adaptor for descriptor %d is still in-use", fd);
529 if (!add && !adaptor->is_used) {
530 crm_err(
"Adaptor for descriptor %d is not in-use", fd);
535 channel = g_io_channel_unix_new(fd);
537 crm_err(
"No memory left to add fd=%d", fd);
541 if (adaptor->source) {
542 g_source_remove(adaptor->source);
547 evts |= (G_IO_HUP | G_IO_NVAL | G_IO_ERR);
550 adaptor->events = evts;
551 adaptor->data =
data;
555 g_io_add_watch_full(channel, conv_prio_libqb2glib(p), evts,
556 gio_read_socket, adaptor, gio_poll_destroy);
567 g_io_channel_unref(channel);
569 crm_trace(
"Added to mainloop with gsource id=%d", adaptor->source);
570 if (adaptor->source > 0) {
578 gio_poll_dispatch_add(
enum qb_loop_priority p, int32_t fd, int32_t evts,
579 void *
data, qb_ipcs_dispatch_fn_t fn)
581 return gio_poll_dispatch_update(p, fd, evts,
data, fn, QB_TRUE);
585 gio_poll_dispatch_mod(
enum qb_loop_priority p, int32_t fd, int32_t evts,
586 void *
data, qb_ipcs_dispatch_fn_t fn)
588 return gio_poll_dispatch_update(p, fd, evts,
data, fn, QB_FALSE);
592 gio_poll_dispatch_del(int32_t fd)
594 struct gio_to_qb_poll *adaptor;
597 if (qb_array_index(gio_map, fd, (
void **)&adaptor) == 0) {
598 if (adaptor->source) {
599 g_source_remove(adaptor->source);
608 .dispatch_add = gio_poll_dispatch_add,
609 .dispatch_mod = gio_poll_dispatch_mod,
610 .dispatch_del = gio_poll_dispatch_del,
613 static enum qb_ipc_type
614 pick_ipc_type(
enum qb_ipc_type requested)
618 if (env && strcmp(
"shared-mem", env) == 0) {
620 }
else if (env && strcmp(
"socket", env) == 0) {
621 return QB_IPC_SOCKET;
622 }
else if (env && strcmp(
"posix", env) == 0) {
623 return QB_IPC_POSIX_MQ;
624 }
else if (env && strcmp(
"sysv", env) == 0) {
625 return QB_IPC_SYSV_MQ;
626 }
else if (requested == QB_IPC_NATIVE) {
640 struct qb_ipcs_service_handlers *callbacks)
647 struct qb_ipcs_service_handlers *callbacks,
648 enum qb_loop_priority prio)
651 qb_ipcs_service_t *server = NULL;
653 if (gio_map == NULL) {
654 gio_map = qb_array_create_2(64,
sizeof(
struct gio_to_qb_poll), 1);
657 server = qb_ipcs_create(
name, 0, pick_ipc_type(
type), callbacks);
659 if (server == NULL) {
660 crm_err(
"Could not create %s IPC server: %s (%d)",
665 if (prio != QB_LOOP_MED) {
666 qb_ipcs_request_rate_limit(server, conv_libqb_prio2ratelimit(prio));
673 rc = qb_ipcs_run(server);
686 qb_ipcs_destroy(server);
690 struct mainloop_io_s {
699 int (*dispatch_fn_ipc) (
const char *buffer, ssize_t length, gpointer userdata);
700 int (*dispatch_fn_io) (gpointer userdata);
701 void (*destroy_fn) (gpointer userdata);
716 mainloop_gio_callback(GIOChannel *gio, GIOCondition condition, gpointer
data)
718 gboolean rc = G_SOURCE_CONTINUE;
721 CRM_ASSERT(client->fd == g_io_channel_unix_get_fd(gio));
723 if (condition & G_IO_IN) {
731 crm_trace(
"Could not read IPC message from %s: %s (%ld)",
734 }
else if (client->dispatch_fn_ipc) {
737 crm_trace(
"New %ld-byte IPC message from %s " 738 "after I/O condition %d",
739 read_rc, client->name, (
int) condition);
740 if (client->dispatch_fn_ipc(buffer, read_rc, client->userdata) < 0) {
741 crm_trace(
"Connection to %s no longer required", client->name);
742 rc = G_SOURCE_REMOVE;
746 }
while ((rc == G_SOURCE_CONTINUE) && (read_rc > 0) && --max > 0);
749 crm_trace(
"New I/O event for %s after I/O condition %d",
750 client->name, (
int) condition);
751 if (client->dispatch_fn_io) {
752 if (client->dispatch_fn_io(client->userdata) < 0) {
753 crm_trace(
"Connection to %s no longer required", client->name);
754 rc = G_SOURCE_REMOVE;
761 crm_err(
"Connection to %s closed " CRM_XS "client=%p condition=%d",
762 client->name, client, condition);
763 rc = G_SOURCE_REMOVE;
765 }
else if (condition & (G_IO_HUP | G_IO_NVAL | G_IO_ERR)) {
766 crm_trace(
"The connection %s[%p] has been closed (I/O condition=%d)",
767 client->name, client, condition);
768 rc = G_SOURCE_REMOVE;
770 }
else if ((condition & G_IO_IN) == 0) {
798 crm_err(
"Strange condition: %d", condition);
808 mainloop_gio_destroy(gpointer c)
811 char *c_name = strdup(client->name);
816 crm_trace(
"Destroying client %s[%p]", c_name, c);
822 if (client->destroy_fn) {
823 void (*destroy_fn) (gpointer userdata) = client->destroy_fn;
825 client->destroy_fn = NULL;
826 destroy_fn(client->userdata);
836 crm_trace(
"Destroyed client %s[%p]", c_name, c);
838 free(client->name); client->name = NULL;
869 const char *ipc_name = NULL;
871 CRM_CHECK((ipc != NULL) && (callbacks != NULL),
return EINVAL);
882 crm_debug(
"Could not obtain file descriptor for %s IPC: %s",
889 if (*source == NULL) {
895 (*source)->ipc = ipc;
896 (*source)->destroy_fn = callbacks->
destroy;
897 (*source)->dispatch_fn_ipc = callbacks->
dispatch;
912 return timer->period_ms;
928 fprintf(stderr,
"Connection to %s failed: %s",
965 if (client == NULL) {
968 client->name = strdup(
name);
969 client->userdata = userdata;
972 client->destroy_fn = callbacks->
destroy;
973 client->dispatch_fn_io = callbacks->
dispatch;
977 client->channel = g_io_channel_unix_new(fd);
979 g_io_add_watch_full(client->channel, priority,
980 (G_IO_IN | G_IO_HUP | G_IO_NVAL | G_IO_ERR), mainloop_gio_callback,
981 client, mainloop_gio_destroy);
992 g_io_channel_unref(client->channel);
993 crm_trace(
"Added connection %d for %s[%p].%d", client->source, client->name, client, fd);
1004 if (client != NULL) {
1005 crm_trace(
"Removing client %s[%p]", client->name, client);
1006 if (client->source) {
1010 g_source_remove(client->source);
1015 static GList *child_list = NULL;
1032 return child->timeout;
1038 return child->privatedata;
1044 child->privatedata = NULL;
1051 if (child->timerid != 0) {
1052 crm_trace(
"Removing timer %d", child->timerid);
1053 g_source_remove(child->timerid);
1066 crm_debug(
"Kill pid %d only. leave group intact.", child->pid);
1067 rc = kill(child->pid, SIGKILL);
1069 crm_debug(
"Kill pid %d's group", child->pid);
1070 rc = kill(-child->pid, SIGKILL);
1074 if (errno != ESRCH) {
1075 crm_perror(LOG_ERR,
"kill(%d, KILL) failed", child->pid);
1083 child_timeout_callback(gpointer p)
1089 if (child->timeout) {
1090 crm_warn(
"%s process (PID %d) will not die!", child->desc, (
int)child->pid);
1094 rc = child_kill_helper(child);
1100 child->timeout = TRUE;
1101 crm_debug(
"%s process (PID %d) timed out", child->desc, (
int)child->pid);
1103 child->timerid = g_timeout_add(5000, child_timeout_callback, child);
1115 bool callback_needed =
true;
1117 rc = waitpid(child->pid, &status,
flags);
1119 crm_trace(
"Child process %d (%s) still active",
1120 child->pid, child->desc);
1121 callback_needed =
false;
1123 }
else if (rc != child->pid) {
1135 crm_notice(
"Wait for child process %d (%s) interrupted: %s",
1138 }
else if (WIFEXITED(status)) {
1139 exitcode = WEXITSTATUS(status);
1140 crm_trace(
"Child process %d (%s) exited with status %d",
1141 child->pid, child->desc, exitcode);
1143 }
else if (WIFSIGNALED(status)) {
1144 signo = WTERMSIG(status);
1145 crm_trace(
"Child process %d (%s) exited with signal %d (%s)",
1146 child->pid, child->desc, signo, strsignal(signo));
1148 #ifdef WCOREDUMP // AIX, SunOS, maybe others 1149 }
else if (WCOREDUMP(status)) {
1151 crm_err(
"Child process %d (%s) dumped core",
1152 child->pid, child->desc);
1156 crm_trace(
"Child process %d (%s) stopped or continued",
1157 child->pid, child->desc);
1158 callback_needed =
false;
1161 if (callback_needed && child->callback) {
1162 child->callback(child, child->pid, core, signo, exitcode);
1164 return callback_needed;
1168 child_death_dispatch(
int signal)
1170 for (GList *iter = child_list; iter; ) {
1171 GList *saved = iter;
1175 if (child_waitpid(child, WNOHANG)) {
1176 crm_trace(
"Removing completed process %d from child list",
1178 child_list = g_list_remove_link(child_list, saved);
1186 child_signal_init(gpointer p)
1193 child_death_dispatch(SIGCHLD);
1205 int waitflags = 0, rc = 0;
1207 for (iter = child_list; iter != NULL && match == NULL; iter = iter->next) {
1209 if (
pid == child->pid) {
1214 if (match == NULL) {
1218 rc = child_kill_helper(match);
1225 crm_trace(
"Waiting for signal that child process %d completed",
1229 }
else if(rc != 0) {
1233 waitflags = WNOHANG;
1236 if (!child_waitpid(match, waitflags)) {
1241 child_list = g_list_remove(child_list, match);
1257 static bool need_init = TRUE;
1262 child->timeout = FALSE;
1263 child->privatedata = privatedata;
1264 child->callback = callback;
1265 child->flags =
flags;
1269 child->timerid = g_timeout_add(
timeout, child_timeout_callback, child);
1272 child_list = g_list_append(child_list, child);
1280 g_timeout_add(1, child_signal_init, NULL);
1292 mainloop_timer_cb(gpointer user_data)
1295 bool repeat = FALSE;
1296 struct mainloop_timer_s *t = user_data;
1306 crm_trace(
"Invoking callbacks for timer %s", t->name);
1308 if(t->cb(t->userdata) == FALSE) {
1309 crm_trace(
"Timer %s complete", t->name);
1325 if(t && t->id != 0) {
1335 if(t && t->period_ms > 0) {
1336 crm_trace(
"Starting timer %s", t->name);
1337 t->id = g_timeout_add(t->period_ms, mainloop_timer_cb, t);
1344 if(t && t->id != 0) {
1345 crm_trace(
"Stopping timer %s", t->name);
1346 g_source_remove(t->id);
1357 last = t->period_ms;
1358 t->period_ms = period_ms;
1361 if(t && t->id != 0 && last != t->period_ms) {
1378 t->period_ms = period_ms;
1381 t->userdata = userdata;
1382 crm_trace(
"Created timer %s with %p %p", t->name, userdata, t->userdata);
1390 crm_trace(
"Destroying timer %s", t->name);
1402 drain_timeout_cb(gpointer user_data)
1404 bool *timeout_popped = (
bool*) user_data;
1406 *timeout_popped = TRUE;
1419 if ((mloop != NULL) && g_main_loop_is_running(mloop)) {
1420 GMainContext *ctx = g_main_loop_get_context(mloop);
1425 for (
int i = 0; (i < n) && g_main_context_pending(ctx); ++i) {
1426 g_main_context_dispatch(ctx);
1428 g_main_loop_quit(mloop);
1448 bool timeout_popped = FALSE;
1450 GMainContext *ctx = NULL;
1454 ctx = g_main_loop_get_context(mloop);
1456 time_t start_time = time(NULL);
1458 timer = g_timeout_add(timer_ms, drain_timeout_cb, &timeout_popped);
1459 while (!timeout_popped
1460 && check(timer_ms - (time(NULL) - start_time) * 1000)) {
1461 g_main_context_iteration(ctx, TRUE);
1464 if (!timeout_popped && (timer > 0)) {
1465 g_source_remove(timer);
void * mainloop_child_userdata(mainloop_child_t *child)
#define CRM_CHECK(expr, failure_action)
pid_t mainloop_child_pid(mainloop_child_t *child)
bool mainloop_timer_running(mainloop_timer_t *t)
#define crm_notice(fmt, args...)
const char * pcmk_strerror(int rc)
struct signal_s crm_signal_t
void mainloop_del_fd(mainloop_io_t *client)
void mainloop_del_ipc_server(qb_ipcs_service_t *server)
gboolean mainloop_destroy_signal(int sig)
crm_trigger_t * mainloop_add_trigger(int priority, int(*dispatch)(gpointer user_data), gpointer userdata)
Create a trigger to be used as a mainloop source.
void(* destroy)(gpointer userdata)
Destroy function for mainloop file descriptor client data.
Deprecated Pacemaker mainloop API.
void mainloop_trigger_complete(crm_trigger_t *trig)
const char * mainloop_child_name(mainloop_child_t *child)
void pcmk_quit_main_loop(GMainLoop *mloop, unsigned int n)
Drain some remaining main loop events then quit it.
struct mainloop_timer_s mainloop_timer_t
struct mainloop_io_s mainloop_io_t
struct mainloop_child_s mainloop_child_t
gboolean crm_signal(int sig, void(*dispatch)(int sig))
long crm_ipc_read(crm_ipc_t *client)
mainloop_timer_t * mainloop_timer_add(const char *name, guint period_ms, bool repeat, GSourceFunc cb, void *userdata)
enum crm_ais_msg_types type
const char * pcmk_rc_str(int rc)
Get a user-friendly description of a return code.
mainloop_io_t * mainloop_add_ipc_client(const char *name, int priority, size_t max_size, void *userdata, struct ipc_client_callbacks *callbacks)
struct qb_ipcs_poll_handlers gio_poll_funcs
const char * pcmk__env_option(const char *option)
guint pcmk__mainloop_timer_get_period(const mainloop_timer_t *timer)
Get period for mainloop timer.
guint mainloop_timer_set_period(mainloop_timer_t *t, guint period_ms)
Wrappers for and extensions to glib mainloop.
void mainloop_timer_del(mainloop_timer_t *t)
crm_ipc_t * mainloop_get_ipc_client(mainloop_io_t *client)
const char * crm_ipc_buffer(crm_ipc_t *client)
void mainloop_del_ipc_client(mainloop_io_t *client)
struct trigger_s crm_trigger_t
int pcmk__connect_generic_ipc(crm_ipc_t *ipc)
#define PCMK__ENV_IPC_TYPE
int(* dispatch)(gpointer userdata)
Dispatch function for mainloop file descriptor with data ready.
#define crm_warn(fmt, args...)
int mainloop_child_timeout(mainloop_child_t *child)
#define crm_debug(fmt, args...)
struct crm_ipc_s crm_ipc_t
void mainloop_set_trigger(crm_trigger_t *source)
void(* destroy)(gpointer userdata)
Destroy function for mainloop IPC connection client data.
#define crm_trace(fmt, args...)
mainloop_io_t * mainloop_add_fd(const char *name, int priority, int fd, void *userdata, struct mainloop_fd_callbacks *callbacks)
void mainloop_timer_start(mainloop_timer_t *t)
char * crm_strdup_printf(char const *format,...) G_GNUC_PRINTF(1
qb_ipcs_service_t * mainloop_add_ipc_server(const char *name, enum qb_ipc_type type, struct qb_ipcs_service_handlers *callbacks)
Wrappers for and extensions to libxml2.
void mainloop_clear_child_userdata(mainloop_child_t *child)
gboolean mainloop_child_kill(pid_t pid)
void mainloop_timer_stop(mainloop_timer_t *t)
void mainloop_child_add_with_flags(pid_t pid, int timeout, const char *desc, void *privatedata, enum mainloop_child_flags flags, void(*callback)(mainloop_child_t *p, pid_t pid, int core, int signo, int exitcode))
unsigned int crm_ipc_default_buffer_size(void)
Return pacemaker's default IPC buffer size.
void crm_ipc_destroy(crm_ipc_t *client)
void mainloop_child_add(pid_t pid, int timeout, const char *desc, void *privatedata, void(*callback)(mainloop_child_t *p, pid_t pid, int core, int signo, int exitcode))
#define pcmk__str_copy(str)
bool crm_ipc_connected(crm_ipc_t *client)
unsigned int crm_log_level
void(* sighandler_t)(int)
const char * crm_ipc_name(crm_ipc_t *client)
#define crm_perror(level, fmt, args...)
Send a system error message to both the log and stderr.
qb_ipcs_service_t * mainloop_add_ipc_server_with_prio(const char *name, enum qb_ipc_type type, struct qb_ipcs_service_handlers *callbacks, enum qb_loop_priority prio)
Start server-side API end-point, hooked into the internal event loop.
#define crm_err(fmt, args...)
int pcmk__ipc_fd(crm_ipc_t *ipc, int *fd)
crm_ipc_t * crm_ipc_new(const char *name, size_t max_size)
Create a new (legacy) object for using Pacemaker daemon IPC.
void mainloop_cleanup(void)
gboolean mainloop_add_signal(int sig, void(*dispatch)(int sig))
int pcmk__add_mainloop_ipc(crm_ipc_t *ipc, int priority, void *userdata, const struct ipc_client_callbacks *callbacks, mainloop_io_t **source)
Connect to IPC and add it as a main loop source.
gboolean mainloop_destroy_trigger(crm_trigger_t *source)
void pcmk_drain_main_loop(GMainLoop *mloop, guint timer_ms, bool(*check)(guint))
Process main loop events while a certain condition is met.
#define pcmk__assert_alloc(nmemb, size)
sighandler_t crm_signal_handler(int sig, sighandler_t dispatch)
void crm_ipc_close(crm_ipc_t *client)
int(* dispatch)(const char *buffer, ssize_t length, gpointer userdata)
Dispatch function for an IPC connection used as mainloop source.