36 #include <qb/qbarray.h> 38 struct mainloop_child_s {
62 crm_trigger_prepare(GSource * source, gint * timeout)
89 crm_trigger_check(GSource * source)
97 crm_trigger_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
106 trig->trigger = FALSE;
109 rc = callback(trig->user_data);
111 crm_trace(
"Trigger handler %p not yet complete", trig);
112 trig->running = TRUE;
120 crm_trigger_finalize(GSource * source)
122 crm_trace(
"Trigger %p destroyed", source);
128 gpointer callback_data;
129 GSourceCallbackFuncs *callback_funcs;
131 const GSourceFuncs *source_funcs;
134 GMainContext *context;
151 g_source_refcount(GSource * source)
155 struct _GSourceCopy *evil = (
struct _GSourceCopy*)source;
156 return evil->ref_count;
161 static int g_source_refcount(GSource * source)
167 static GSourceFuncs crm_trigger_funcs = {
170 crm_trigger_dispatch,
171 crm_trigger_finalize,
175 mainloop_setup_trigger(GSource * source,
int priority,
int (*dispatch) (gpointer user_data),
183 trigger->trigger = FALSE;
184 trigger->user_data = userdata;
187 g_source_set_callback(source, dispatch, trigger, NULL);
190 g_source_set_priority(source, priority);
191 g_source_set_can_recurse(source, FALSE);
193 crm_trace(
"Setup %p with ref-count=%u", source, g_source_refcount(source));
194 trigger->id = g_source_attach(source, NULL);
195 crm_trace(
"Attached %p with ref-count=%u", source, g_source_refcount(source));
203 crm_trace(
"Trigger handler %p complete", trig);
204 trig->running = FALSE;
215 GSource *source = NULL;
218 source = g_source_new(&crm_trigger_funcs,
sizeof(
crm_trigger_t));
221 return mainloop_setup_trigger(source, priority, dispatch, userdata);
228 source->trigger = TRUE;
241 gs = (GSource *)source;
243 if(g_source_refcount(gs) > 2) {
244 crm_info(
"Trigger %p is still referenced %u times", gs, g_source_refcount(gs));
247 g_source_destroy(gs);
259 typedef struct signal_s {
261 void (*handler) (
int sig);
269 crm_signal_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
273 if(sig->signal != SIGCHLD) {
275 strsignal(sig->signal), sig->signal,
276 (sig->handler?
"invoking" :
"no"));
279 sig->trigger.trigger = FALSE;
281 sig->handler(sig->signal);
287 mainloop_signal_handler(
int sig)
289 if (sig > 0 && sig < NSIG && crm_signals[sig] != NULL) {
294 static GSourceFuncs crm_signal_funcs = {
298 crm_trigger_finalize,
306 struct sigaction old;
308 if (sigemptyset(&mask) < 0) {
309 crm_perror(LOG_ERR,
"Call to sigemptyset failed");
313 memset(&sa, 0,
sizeof(
struct sigaction));
314 sa.sa_handler = dispatch;
315 sa.sa_flags = SA_RESTART;
318 if (sigaction(sig, &sa, &old) < 0) {
319 crm_perror(LOG_ERR,
"Could not install signal handler for signal %d", sig);
329 GSource *source = NULL;
330 int priority = G_PRIORITY_HIGH - 1;
332 if (sig == SIGTERM) {
340 if (sig >= NSIG || sig < 0) {
341 crm_err(
"Signal %d is out of range", sig);
344 }
else if (crm_signals[sig] != NULL && crm_signals[sig]->handler == dispatch) {
345 crm_trace(
"Signal handler for %d is already installed", sig);
348 }
else if (crm_signals[sig] != NULL) {
349 crm_err(
"Different signal handler for %d is already installed", sig);
354 source = g_source_new(&crm_signal_funcs,
sizeof(
crm_signal_t));
356 crm_signals[sig] = (
crm_signal_t *) mainloop_setup_trigger(source, priority, NULL, NULL);
359 crm_signals[sig]->handler = dispatch;
360 crm_signals[sig]->signal = sig;
362 if (
crm_signal(sig, mainloop_signal_handler) == FALSE) {
365 crm_signals[sig] = NULL;
376 if (siginterrupt(sig, 1) < 0) {
377 crm_perror(LOG_INFO,
"Could not enable system call interruptions for signal %d", sig);
389 if (sig >= NSIG || sig < 0) {
390 crm_err(
"Signal %d is out of range", sig);
394 crm_perror(LOG_ERR,
"Could not uninstall signal handler for signal %d", sig);
397 }
else if (crm_signals[sig] == NULL) {
402 tmp = crm_signals[sig];
403 crm_signals[sig] = NULL;
408 static qb_array_t *gio_map = NULL;
414 qb_array_free(gio_map);
421 struct gio_to_qb_poll {
426 qb_ipcs_dispatch_fn_t fn;
427 enum qb_loop_priority p;
431 gio_read_socket(GIOChannel * gio, GIOCondition condition, gpointer
data)
433 struct gio_to_qb_poll *adaptor = (
struct gio_to_qb_poll *)data;
434 gint fd = g_io_channel_unix_get_fd(gio);
436 crm_trace(
"%p.%d %d", data, fd, condition);
442 return (adaptor->fn(fd, condition, adaptor->data) == 0);
446 gio_poll_destroy(gpointer data)
448 struct gio_to_qb_poll *adaptor = (
struct gio_to_qb_poll *)data;
453 if (adaptor->is_used == 0) {
454 crm_trace(
"Marking adaptor %p unused", adaptor);
468 conv_prio_libqb2glib(
enum qb_loop_priority prio)
470 gint ret = G_PRIORITY_DEFAULT;
473 ret = G_PRIORITY_LOW;
476 ret = G_PRIORITY_HIGH;
479 crm_trace(
"Invalid libqb's loop priority %d, assuming QB_LOOP_MED",
496 static enum qb_ipcs_rate_limit
497 conv_libqb_prio2ratelimit(
enum qb_loop_priority prio)
500 enum qb_ipcs_rate_limit ret = QB_IPCS_RATE_NORMAL;
503 ret = QB_IPCS_RATE_SLOW;
506 ret = QB_IPCS_RATE_FAST;
509 crm_trace(
"Invalid libqb's loop priority %d, assuming QB_LOOP_MED",
519 gio_poll_dispatch_update(
enum qb_loop_priority p,
int32_t fd,
int32_t evts,
520 void *data, qb_ipcs_dispatch_fn_t fn,
int32_t add)
522 struct gio_to_qb_poll *adaptor;
526 res = qb_array_index(gio_map, fd, (
void **)&adaptor);
528 crm_err(
"Array lookup failed for fd=%d: %d", fd, res);
532 crm_trace(
"Adding fd=%d to mainloop as adaptor %p", fd, adaptor);
534 if (add && adaptor->source) {
535 crm_err(
"Adaptor for descriptor %d is still in-use", fd);
538 if (!add && !adaptor->is_used) {
539 crm_err(
"Adaptor for descriptor %d is not in-use", fd);
544 channel = g_io_channel_unix_new(fd);
546 crm_err(
"No memory left to add fd=%d", fd);
550 if (adaptor->source) {
551 g_source_remove(adaptor->source);
556 evts |= (G_IO_HUP | G_IO_NVAL | G_IO_ERR);
559 adaptor->events = evts;
560 adaptor->data =
data;
564 g_io_add_watch_full(channel, conv_prio_libqb2glib(p), evts,
565 gio_read_socket, adaptor, gio_poll_destroy);
576 g_io_channel_unref(channel);
578 crm_trace(
"Added to mainloop with gsource id=%d", adaptor->source);
579 if (adaptor->source > 0) {
587 gio_poll_dispatch_add(
enum qb_loop_priority p,
int32_t fd,
int32_t evts,
588 void *data, qb_ipcs_dispatch_fn_t fn)
590 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_TRUE);
594 gio_poll_dispatch_mod(
enum qb_loop_priority p,
int32_t fd,
int32_t evts,
595 void *data, qb_ipcs_dispatch_fn_t fn)
597 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_FALSE);
601 gio_poll_dispatch_del(
int32_t fd)
603 struct gio_to_qb_poll *adaptor;
606 if (qb_array_index(gio_map, fd, (
void **)&adaptor) == 0) {
607 if (adaptor->source) {
608 g_source_remove(adaptor->source);
617 .dispatch_add = gio_poll_dispatch_add,
618 .dispatch_mod = gio_poll_dispatch_mod,
619 .dispatch_del = gio_poll_dispatch_del,
622 static enum qb_ipc_type
623 pick_ipc_type(
enum qb_ipc_type requested)
625 const char *env = getenv(
"PCMK_ipc_type");
627 if (env && strcmp(
"shared-mem", env) == 0) {
629 }
else if (env && strcmp(
"socket", env) == 0) {
630 return QB_IPC_SOCKET;
631 }
else if (env && strcmp(
"posix", env) == 0) {
632 return QB_IPC_POSIX_MQ;
633 }
else if (env && strcmp(
"sysv", env) == 0) {
634 return QB_IPC_SYSV_MQ;
635 }
else if (requested == QB_IPC_NATIVE) {
649 struct qb_ipcs_service_handlers *callbacks)
656 struct qb_ipcs_service_handlers *callbacks,
657 enum qb_loop_priority prio)
660 qb_ipcs_service_t *server = NULL;
662 if (gio_map == NULL) {
663 gio_map = qb_array_create_2(64,
sizeof(
struct gio_to_qb_poll), 1);
667 server = qb_ipcs_create(name, 0, pick_ipc_type(type), callbacks);
669 if (server == NULL) {
674 if (prio != QB_LOOP_MED) {
675 qb_ipcs_request_rate_limit(server, conv_libqb_prio2ratelimit(prio));
678 #ifdef HAVE_IPCS_GET_BUFFER_SIZE 683 qb_ipcs_poll_handlers_set(server, &gio_poll_funcs);
685 rc = qb_ipcs_run(server);
698 qb_ipcs_destroy(server);
702 struct mainloop_io_s {
711 int (*dispatch_fn_ipc) (
const char *buffer, ssize_t length, gpointer userdata);
712 int (*dispatch_fn_io) (gpointer userdata);
713 void (*destroy_fn) (gpointer userdata);
718 mainloop_gio_callback(GIOChannel * gio, GIOCondition condition, gpointer data)
720 gboolean keep = TRUE;
723 CRM_ASSERT(client->fd == g_io_channel_unix_get_fd(gio));
725 if (condition & G_IO_IN) {
733 crm_trace(
"Message acquisition from %s[%p] failed: %s (%ld)",
736 }
else if (client->dispatch_fn_ipc) {
739 crm_trace(
"New message from %s[%p] = %ld (I/O condition=%d)", client->name, client, rc, condition);
740 if (client->dispatch_fn_ipc(buffer, rc, client->userdata) < 0) {
741 crm_trace(
"Connection to %s no longer required", client->name);
746 }
while (keep && rc > 0 && --max > 0);
749 crm_trace(
"New message from %s[%p] %u", client->name, client, condition);
750 if (client->dispatch_fn_io) {
751 if (client->dispatch_fn_io(client->userdata) < 0) {
752 crm_trace(
"Connection to %s no longer required", client->name);
760 crm_err(
"Connection to %s[%p] closed (I/O condition=%d)", client->name, client, condition);
763 }
else if (condition & (G_IO_HUP | G_IO_NVAL | G_IO_ERR)) {
764 crm_trace(
"The connection %s[%p] has been closed (I/O condition=%d)",
765 client->name, client, condition);
768 }
else if ((condition & G_IO_IN) == 0) {
796 crm_err(
"Strange condition: %d", condition);
806 mainloop_gio_destroy(gpointer c)
809 char *c_name = strdup(client->name);
814 crm_trace(
"Destroying client %s[%p]", c_name, c);
820 if (client->destroy_fn) {
821 void (*destroy_fn) (gpointer userdata) = client->destroy_fn;
823 client->destroy_fn = NULL;
824 destroy_fn(client->userdata);
834 crm_trace(
"Destroyed client %s[%p]", c_name, c);
836 free(client->name); client->name = NULL;
855 if (client == NULL) {
865 client->destroy_fn = callbacks->
destroy;
866 client->dispatch_fn_ipc = callbacks->
dispatch;
893 if (client == NULL) {
896 client->name = strdup(name);
897 client->userdata = userdata;
900 client->destroy_fn = callbacks->
destroy;
901 client->dispatch_fn_io = callbacks->
dispatch;
905 client->channel = g_io_channel_unix_new(fd);
907 g_io_add_watch_full(client->channel, priority,
908 (G_IO_IN | G_IO_HUP | G_IO_NVAL | G_IO_ERR), mainloop_gio_callback,
909 client, mainloop_gio_destroy);
920 g_io_channel_unref(client->channel);
921 crm_trace(
"Added connection %d for %s[%p].%d", client->source, client->name, client, fd);
932 if (client != NULL) {
933 crm_trace(
"Removing client %s[%p]", client->name, client);
934 if (client->source) {
938 g_source_remove(client->source);
960 return child->timeout;
966 return child->privatedata;
972 child->privatedata = NULL;
979 if (child->timerid != 0) {
980 crm_trace(
"Removing timer %d", child->timerid);
981 g_source_remove(child->timerid);
994 crm_debug(
"Kill pid %d only. leave group intact.", child->pid);
995 rc = kill(child->pid, SIGKILL);
997 crm_debug(
"Kill pid %d's group", child->pid);
998 rc = kill(-child->pid, SIGKILL);
1002 if (errno != ESRCH) {
1003 crm_perror(LOG_ERR,
"kill(%d, KILL) failed", child->pid);
1011 child_timeout_callback(gpointer p)
1017 if (child->timeout) {
1018 crm_crit(
"%s process (PID %d) will not die!", child->desc, (
int)child->pid);
1022 rc = child_kill_helper(child);
1028 child->timeout = TRUE;
1029 crm_warn(
"%s process (PID %d) timed out", child->desc, (
int)child->pid);
1031 child->timerid = g_timeout_add(5000, child_timeout_callback, child);
1044 rc = waitpid(child->pid, &status, flags);
1046 crm_perror(LOG_DEBUG,
"wait(%d) = %d", child->pid, rc);
1049 }
else if(rc != child->pid) {
1053 crm_perror(LOG_ERR,
"Call to waitpid(%d) failed", child->pid);
1056 crm_trace(
"Managed process %d exited: %p", child->pid, child);
1058 if (WIFEXITED(status)) {
1059 exitcode = WEXITSTATUS(status);
1060 crm_trace(
"Managed process %d (%s) exited with rc=%d", child->pid, child->desc, exitcode);
1062 }
else if (WIFSIGNALED(status)) {
1063 signo = WTERMSIG(status);
1064 crm_trace(
"Managed process %d (%s) exited with signal=%d", child->pid, child->desc, signo);
1067 if (WCOREDUMP(status)) {
1069 crm_err(
"Managed process %d (%s) dumped core", child->pid, child->desc);
1074 if (child->callback) {
1075 child->callback(child, child->pid, core, signo, exitcode);
1081 child_death_dispatch(
int signal)
1089 exited = child_waitpid(child, WNOHANG);
1094 if (exited == FALSE) {
1097 crm_trace(
"Removing process entry %p for %d", child, child->pid);
1099 child_list = g_list_remove_link(child_list, saved);
1106 child_signal_init(gpointer p)
1113 child_death_dispatch(SIGCHLD);
1125 int waitflags = 0, rc = 0;
1127 for (iter = child_list; iter != NULL && match == NULL; iter = iter->next) {
1129 if (pid == child->pid) {
1134 if (match == NULL) {
1138 rc = child_kill_helper(match);
1149 crm_trace(
"Waiting for child %d to be reaped by child_death_dispatch()", match->pid);
1152 }
else if(rc != 0) {
1156 waitflags = WNOHANG;
1159 if (child_waitpid(match, waitflags) == FALSE) {
1164 child_list = g_list_remove(child_list, match);
1174 void (*callback) (
mainloop_child_t * p, pid_t pid,
int core,
int signo,
int exitcode))
1176 static bool need_init = TRUE;
1181 child->timeout = FALSE;
1182 child->privatedata = privatedata;
1183 child->callback = callback;
1184 child->flags =
flags;
1187 child->desc = strdup(desc);
1191 child->timerid = g_timeout_add(timeout, child_timeout_callback, child);
1194 child_list = g_list_append(child_list, child);
1202 g_timeout_add(1, child_signal_init, NULL);
1208 void (*callback) (
mainloop_child_t * p, pid_t pid,
int core,
int signo,
int exitcode))
1213 struct mainloop_timer_s {
1224 static gboolean mainloop_timer_cb(gpointer user_data)
1227 bool repeat = FALSE;
1228 struct mainloop_timer_s *t = user_data;
1238 crm_trace(
"Invoking callbacks for timer %s", t->name);
1240 if(t->cb(t->userdata) == FALSE) {
1241 crm_trace(
"Timer %s complete", t->name);
1256 if(t && t->id != 0) {
1265 if(t && t->period_ms > 0) {
1266 crm_trace(
"Starting timer %s", t->name);
1267 t->id = g_timeout_add(t->period_ms, mainloop_timer_cb, t);
1273 if(t && t->id != 0) {
1274 crm_trace(
"Stopping timer %s", t->name);
1275 g_source_remove(t->id);
1285 last = t->period_ms;
1286 t->period_ms = period_ms;
1289 if(t && t->id != 0 && last != t->period_ms) {
1308 t->period_ms = period_ms;
1311 t->userdata = userdata;
1312 crm_trace(
"Created timer %s with %p %p", t->name, userdata, t->userdata);
1321 crm_trace(
"Destroying timer %s", t->name);
1333 drain_timeout_cb(gpointer user_data)
1335 bool *timeout_popped = (
bool*) user_data;
1337 *timeout_popped = TRUE;
1356 bool timeout_popped = FALSE;
1358 GMainContext *ctx = NULL;
1362 ctx = g_main_loop_get_context(mloop);
1364 time_t start_time = time(NULL);
1366 timer = g_timeout_add(timer_ms, drain_timeout_cb, &timeout_popped);
1367 while (!timeout_popped
1368 && check(timer_ms - (time(NULL) - start_time) * 1000)) {
1369 g_main_context_iteration(ctx, TRUE);
1372 if (!timeout_popped && (timer > 0)) {
1373 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)
bool crm_ipc_connect(crm_ipc_t *client)
Establish an IPC connection to a Pacemaker component.
void(* destroy)(gpointer)
#define crm_notice(fmt, args...)
struct signal_s crm_signal_t
#define crm_crit(fmt, args...)
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)
void(* destroy)(gpointer userdata)
int crm_ipc_get_fd(crm_ipc_t *client)
const char * pcmk_strerror(int rc)
void mainloop_trigger_complete(crm_trigger_t *trig)
const char * mainloop_child_name(mainloop_child_t *child)
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)
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
guint mainloop_timer_set_period(mainloop_timer_t *t, guint period_ms)
Wrappers for and extensions to glib mainloop.
void crm_client_init(void)
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(* dispatch)(gpointer userdata)
#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)
#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)
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)
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)
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))
bool crm_ipc_connected(crm_ipc_t *client)
#define crm_perror(level, fmt, args...)
Log a system error message.
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...)
struct mainloop_timer_s mainloop
crm_ipc_t * crm_ipc_new(const char *name, size_t max_size)
void mainloop_cleanup(void)
gboolean mainloop_add_signal(int sig, void(*dispatch)(int sig))
int mainloop_child_kill(pid_t pid)
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.
char * crm_strdup_printf(char const *format,...) __attribute__((__format__(__printf__
void crm_ipc_close(crm_ipc_t *client)
#define crm_info(fmt, args...)
int(* dispatch)(const char *buffer, ssize_t length, gpointer userdata)
enum crm_ais_msg_types type