25 #include <qb/qbarray.h> 27 struct mainloop_child_s {
50 crm_trigger_prepare(GSource * source, gint * timeout)
77 crm_trigger_check(GSource * source)
85 crm_trigger_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
94 trig->trigger = FALSE;
97 rc = callback(trig->user_data);
99 crm_trace(
"Trigger handler %p not yet complete", trig);
100 trig->running = TRUE;
108 crm_trigger_finalize(GSource * source)
110 crm_trace(
"Trigger %p destroyed", source);
116 gpointer callback_data;
117 GSourceCallbackFuncs *callback_funcs;
119 const GSourceFuncs *source_funcs;
122 GMainContext *context;
139 g_source_refcount(GSource * source)
143 struct _GSourceCopy *evil = (
struct _GSourceCopy*)source;
144 return evil->ref_count;
149 static int g_source_refcount(GSource * source)
155 static GSourceFuncs crm_trigger_funcs = {
158 crm_trigger_dispatch,
159 crm_trigger_finalize,
163 mainloop_setup_trigger(GSource * source,
int priority,
int (*dispatch) (gpointer user_data),
171 trigger->trigger = FALSE;
172 trigger->user_data = userdata;
175 g_source_set_callback(source, dispatch, trigger, NULL);
178 g_source_set_priority(source, priority);
179 g_source_set_can_recurse(source, FALSE);
181 crm_trace(
"Setup %p with ref-count=%u", source, g_source_refcount(source));
182 trigger->id = g_source_attach(source, NULL);
183 crm_trace(
"Attached %p with ref-count=%u", source, g_source_refcount(source));
191 crm_trace(
"Trigger handler %p complete", trig);
192 trig->running = FALSE;
203 GSource *source = NULL;
206 source = g_source_new(&crm_trigger_funcs,
sizeof(
crm_trigger_t));
209 return mainloop_setup_trigger(source, priority, dispatch, userdata);
216 source->trigger = TRUE;
229 gs = (GSource *)source;
231 if(g_source_refcount(gs) > 2) {
232 crm_info(
"Trigger %p is still referenced %u times", gs, g_source_refcount(gs));
235 g_source_destroy(gs);
247 typedef struct signal_s {
249 void (*handler) (
int sig);
257 crm_signal_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
261 if(sig->signal != SIGCHLD) {
263 strsignal(sig->signal), sig->signal,
264 (sig->handler?
"invoking" :
"no"));
267 sig->trigger.trigger = FALSE;
269 sig->handler(sig->signal);
275 mainloop_signal_handler(
int sig)
277 if (sig > 0 && sig < NSIG && crm_signals[sig] != NULL) {
282 static GSourceFuncs crm_signal_funcs = {
286 crm_trigger_finalize,
294 struct sigaction old;
296 if (sigemptyset(&mask) < 0) {
297 crm_perror(LOG_ERR,
"Call to sigemptyset failed");
301 memset(&sa, 0,
sizeof(
struct sigaction));
302 sa.sa_handler = dispatch;
303 sa.sa_flags = SA_RESTART;
306 if (sigaction(sig, &sa, &old) < 0) {
307 crm_perror(LOG_ERR,
"Could not install signal handler for signal %d", sig);
315 mainloop_destroy_signal_entry(
int sig)
319 crm_signals[sig] = NULL;
328 GSource *source = NULL;
329 int priority = G_PRIORITY_HIGH - 1;
331 if (sig == SIGTERM) {
339 if (sig >= NSIG || sig < 0) {
340 crm_err(
"Signal %d is out of range", sig);
343 }
else if (crm_signals[sig] != NULL && crm_signals[sig]->handler == dispatch) {
344 crm_trace(
"Signal handler for %d is already installed", sig);
347 }
else if (crm_signals[sig] != NULL) {
348 crm_err(
"Different signal handler for %d is already installed", sig);
353 source = g_source_new(&crm_signal_funcs,
sizeof(
crm_signal_t));
355 crm_signals[sig] = (
crm_signal_t *) mainloop_setup_trigger(source, priority, NULL, NULL);
358 crm_signals[sig]->handler = dispatch;
359 crm_signals[sig]->signal = sig;
361 if (
crm_signal(sig, mainloop_signal_handler) == FALSE) {
362 mainloop_destroy_signal_entry(sig);
371 if (siginterrupt(sig, 1) < 0) {
372 crm_perror(LOG_INFO,
"Could not enable system call interruptions for signal %d", sig);
382 if (sig >= NSIG || sig < 0) {
383 crm_err(
"Signal %d is out of range", sig);
387 crm_perror(LOG_ERR,
"Could not uninstall signal handler for signal %d", sig);
390 }
else if (crm_signals[sig] == NULL) {
393 mainloop_destroy_signal_entry(sig);
397 static qb_array_t *gio_map = NULL;
403 qb_array_free(gio_map);
406 for (
int sig = 0; sig < NSIG; ++sig) {
407 mainloop_destroy_signal_entry(sig);
414 struct gio_to_qb_poll {
419 qb_ipcs_dispatch_fn_t fn;
420 enum qb_loop_priority p;
424 gio_read_socket(GIOChannel * gio, GIOCondition condition, gpointer
data)
426 struct gio_to_qb_poll *adaptor = (
struct gio_to_qb_poll *)data;
427 gint fd = g_io_channel_unix_get_fd(gio);
429 crm_trace(
"%p.%d %d", data, fd, condition);
435 return (adaptor->fn(fd, condition, adaptor->data) == 0);
439 gio_poll_destroy(gpointer data)
441 struct gio_to_qb_poll *adaptor = (
struct gio_to_qb_poll *)data;
446 if (adaptor->is_used == 0) {
447 crm_trace(
"Marking adaptor %p unused", adaptor);
453 gio_poll_dispatch_update(
enum qb_loop_priority p,
int32_t fd,
int32_t evts,
454 void *data, qb_ipcs_dispatch_fn_t fn,
int32_t add)
456 struct gio_to_qb_poll *adaptor;
460 res = qb_array_index(gio_map, fd, (
void **)&adaptor);
462 crm_err(
"Array lookup failed for fd=%d: %d", fd, res);
466 crm_trace(
"Adding fd=%d to mainloop as adaptor %p", fd, adaptor);
468 if (add && adaptor->source) {
469 crm_err(
"Adaptor for descriptor %d is still in-use", fd);
472 if (!add && !adaptor->is_used) {
473 crm_err(
"Adaptor for descriptor %d is not in-use", fd);
478 channel = g_io_channel_unix_new(fd);
480 crm_err(
"No memory left to add fd=%d", fd);
484 if (adaptor->source) {
485 g_source_remove(adaptor->source);
490 evts |= (G_IO_HUP | G_IO_NVAL | G_IO_ERR);
493 adaptor->events = evts;
494 adaptor->data =
data;
498 g_io_add_watch_full(channel, G_PRIORITY_DEFAULT, evts, gio_read_socket, adaptor,
510 g_io_channel_unref(channel);
512 crm_trace(
"Added to mainloop with gsource id=%d", adaptor->source);
513 if (adaptor->source > 0) {
521 gio_poll_dispatch_add(
enum qb_loop_priority p,
int32_t fd,
int32_t evts,
522 void *data, qb_ipcs_dispatch_fn_t fn)
524 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_TRUE);
528 gio_poll_dispatch_mod(
enum qb_loop_priority p,
int32_t fd,
int32_t evts,
529 void *data, qb_ipcs_dispatch_fn_t fn)
531 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_FALSE);
535 gio_poll_dispatch_del(
int32_t fd)
537 struct gio_to_qb_poll *adaptor;
540 if (qb_array_index(gio_map, fd, (
void **)&adaptor) == 0) {
541 if (adaptor->source) {
542 g_source_remove(adaptor->source);
551 .dispatch_add = gio_poll_dispatch_add,
552 .dispatch_mod = gio_poll_dispatch_mod,
553 .dispatch_del = gio_poll_dispatch_del,
556 static enum qb_ipc_type
557 pick_ipc_type(
enum qb_ipc_type requested)
559 const char *env = getenv(
"PCMK_ipc_type");
561 if (env && strcmp(
"shared-mem", env) == 0) {
563 }
else if (env && strcmp(
"socket", env) == 0) {
564 return QB_IPC_SOCKET;
565 }
else if (env && strcmp(
"posix", env) == 0) {
566 return QB_IPC_POSIX_MQ;
567 }
else if (env && strcmp(
"sysv", env) == 0) {
568 return QB_IPC_SYSV_MQ;
569 }
else if (requested == QB_IPC_NATIVE) {
583 struct qb_ipcs_service_handlers * callbacks)
586 qb_ipcs_service_t *server = NULL;
588 if (gio_map == NULL) {
589 gio_map = qb_array_create_2(64,
sizeof(
struct gio_to_qb_poll), 1);
593 server = qb_ipcs_create(name, 0, pick_ipc_type(type), callbacks);
595 #ifdef HAVE_IPCS_GET_BUFFER_SIZE 600 qb_ipcs_poll_handlers_set(server, &gio_poll_funcs);
602 rc = qb_ipcs_run(server);
615 qb_ipcs_destroy(server);
619 struct mainloop_io_s {
628 int (*dispatch_fn_ipc) (
const char *buffer, ssize_t length, gpointer userdata);
629 int (*dispatch_fn_io) (gpointer userdata);
630 void (*destroy_fn) (gpointer userdata);
635 mainloop_gio_callback(GIOChannel * gio, GIOCondition condition, gpointer data)
637 gboolean keep = TRUE;
640 CRM_ASSERT(client->fd == g_io_channel_unix_get_fd(gio));
642 if (condition & G_IO_IN) {
650 crm_trace(
"Message acquisition from %s[%p] failed: %s (%ld)",
653 }
else if (client->dispatch_fn_ipc) {
656 crm_trace(
"New message from %s[%p] = %ld (I/O condition=%d)", client->name, client, rc, condition);
657 if (client->dispatch_fn_ipc(buffer, rc, client->userdata) < 0) {
658 crm_trace(
"Connection to %s no longer required", client->name);
663 }
while (keep && rc > 0 && --max > 0);
666 crm_trace(
"New message from %s[%p] %u", client->name, client, condition);
667 if (client->dispatch_fn_io) {
668 if (client->dispatch_fn_io(client->userdata) < 0) {
669 crm_trace(
"Connection to %s no longer required", client->name);
677 crm_err(
"Connection to %s closed " CRM_XS "client=%p condition=%d",
678 client->name, client, condition);
681 }
else if (condition & (G_IO_HUP | G_IO_NVAL | G_IO_ERR)) {
682 crm_trace(
"The connection %s[%p] has been closed (I/O condition=%d)",
683 client->name, client, condition);
686 }
else if ((condition & G_IO_IN) == 0) {
714 crm_err(
"Strange condition: %d", condition);
724 mainloop_gio_destroy(gpointer c)
727 char *c_name = strdup(client->name);
732 crm_trace(
"Destroying client %s[%p]", c_name, c);
738 if (client->destroy_fn) {
739 void (*destroy_fn) (gpointer userdata) = client->destroy_fn;
741 client->destroy_fn = NULL;
742 destroy_fn(client->userdata);
752 crm_trace(
"Destroyed client %s[%p]", c_name, c);
754 free(client->name); client->name = NULL;
773 if (client == NULL) {
783 client->destroy_fn = callbacks->
destroy;
784 client->dispatch_fn_ipc = callbacks->
dispatch;
811 if (client == NULL) {
814 client->name = strdup(name);
815 client->userdata = userdata;
818 client->destroy_fn = callbacks->
destroy;
819 client->dispatch_fn_io = callbacks->
dispatch;
823 client->channel = g_io_channel_unix_new(fd);
825 g_io_add_watch_full(client->channel, priority,
826 (G_IO_IN | G_IO_HUP | G_IO_NVAL | G_IO_ERR), mainloop_gio_callback,
827 client, mainloop_gio_destroy);
838 g_io_channel_unref(client->channel);
839 crm_trace(
"Added connection %d for %s[%p].%d", client->source, client->name, client, fd);
850 if (client != NULL) {
851 crm_trace(
"Removing client %s[%p]", client->name, client);
852 if (client->source) {
856 g_source_remove(client->source);
878 return child->timeout;
884 return child->privatedata;
890 child->privatedata = NULL;
897 if (child->timerid != 0) {
898 crm_trace(
"Removing timer %d", child->timerid);
899 g_source_remove(child->timerid);
912 crm_debug(
"Kill pid %d only. leave group intact.", child->pid);
913 rc = kill(child->pid, SIGKILL);
915 crm_debug(
"Kill pid %d's group", child->pid);
916 rc = kill(-child->pid, SIGKILL);
920 if (errno != ESRCH) {
921 crm_perror(LOG_ERR,
"kill(%d, KILL) failed", child->pid);
929 child_timeout_callback(gpointer p)
935 if (child->timeout) {
936 crm_crit(
"%s process (PID %d) will not die!", child->desc, (
int)child->pid);
940 rc = child_kill_helper(child);
946 child->timeout = TRUE;
947 crm_warn(
"%s process (PID %d) timed out", child->desc, (
int)child->pid);
949 child->timerid = g_timeout_add(5000, child_timeout_callback, child);
962 rc = waitpid(child->pid, &status, flags);
964 crm_perror(LOG_DEBUG,
"wait(%d) = %d", child->pid, rc);
967 }
else if(rc != child->pid) {
971 crm_perror(LOG_ERR,
"Call to waitpid(%d) failed", child->pid);
974 crm_trace(
"Managed process %d exited: %p", child->pid, child);
976 if (WIFEXITED(status)) {
977 exitcode = WEXITSTATUS(status);
978 crm_trace(
"Managed process %d (%s) exited with rc=%d", child->pid, child->desc, exitcode);
980 }
else if (WIFSIGNALED(status)) {
981 signo = WTERMSIG(status);
982 crm_trace(
"Managed process %d (%s) exited with signal=%d", child->pid, child->desc, signo);
985 if (WCOREDUMP(status)) {
987 crm_err(
"Managed process %d (%s) dumped core", child->pid, child->desc);
992 if (child->callback) {
993 child->callback(child, child->pid, core, signo, exitcode);
999 child_death_dispatch(
int signal)
1007 exited = child_waitpid(child, WNOHANG);
1012 if (exited == FALSE) {
1015 crm_trace(
"Removing process entry %p for %d", child, child->pid);
1017 child_list = g_list_remove_link(child_list, saved);
1024 child_signal_init(gpointer p)
1031 child_death_dispatch(SIGCHLD);
1043 int waitflags = 0, rc = 0;
1045 for (iter = child_list; iter != NULL && match == NULL; iter = iter->next) {
1047 if (pid == child->pid) {
1052 if (match == NULL) {
1056 rc = child_kill_helper(match);
1067 crm_trace(
"Waiting for child %d to be reaped by child_death_dispatch()", match->pid);
1070 }
else if(rc != 0) {
1074 waitflags = WNOHANG;
1077 if (child_waitpid(match, waitflags) == FALSE) {
1082 child_list = g_list_remove(child_list, match);
1092 void (*callback) (
mainloop_child_t * p, pid_t pid,
int core,
int signo,
int exitcode))
1094 static bool need_init = TRUE;
1099 child->timeout = FALSE;
1100 child->privatedata = privatedata;
1101 child->callback = callback;
1102 child->flags =
flags;
1105 child->desc = strdup(desc);
1109 child->timerid = g_timeout_add(timeout, child_timeout_callback, child);
1112 child_list = g_list_append(child_list, child);
1120 g_timeout_add(1, child_signal_init, NULL);
1126 void (*callback) (
mainloop_child_t * p, pid_t pid,
int core,
int signo,
int exitcode))
1131 struct mainloop_timer_s {
1140 static gboolean mainloop_timer_cb(gpointer user_data)
1143 bool repeat = FALSE;
1144 struct mainloop_timer_s *t = user_data;
1154 crm_trace(
"Invoking callbacks for timer %s", t->name);
1156 if(t->cb(t->userdata) == FALSE) {
1157 crm_trace(
"Timer %s complete", t->name);
1172 if(t && t->id != 0) {
1181 if(t && t->period_ms > 0) {
1182 crm_trace(
"Starting timer %s", t->name);
1183 t->id = g_timeout_add(t->period_ms, mainloop_timer_cb, t);
1189 if(t && t->id != 0) {
1190 crm_trace(
"Stopping timer %s", t->name);
1191 g_source_remove(t->id);
1201 last = t->period_ms;
1202 t->period_ms = period_ms;
1205 if(t && t->id != 0 && last != t->period_ms) {
1224 t->period_ms = period_ms;
1227 t->userdata = userdata;
1228 crm_trace(
"Created timer %s with %p %p", t->name, userdata, t->userdata);
1237 crm_trace(
"Destroying timer %s", t->name);
1249 drain_timeout_cb(gpointer user_data)
1251 bool *timeout_popped = (
bool*) user_data;
1253 *timeout_popped = TRUE;
1272 bool timeout_popped = FALSE;
1274 GMainContext *ctx = NULL;
1278 ctx = g_main_loop_get_context(mloop);
1280 time_t start_time = time(NULL);
1282 timer = g_timeout_add(timer_ms, drain_timeout_cb, &timeout_popped);
1283 while (!timeout_popped
1284 && check(timer_ms - (time(NULL) - start_time) * 1000)) {
1285 g_main_context_iteration(ctx, TRUE);
1288 if (!timeout_popped && (timer > 0)) {
1289 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...)
const char * pcmk_strerror(int rc)
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)
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.
#define crm_err(fmt, args...)
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.
void crm_ipc_close(crm_ipc_t *client)
char * crm_strdup_printf(char const *format,...) __attribute__((__format__(__printf__
#define crm_info(fmt, args...)
int(* dispatch)(const char *buffer, ssize_t length, gpointer userdata)
enum crm_ais_msg_types type