36 struct mainloop_child_s {
60 crm_trigger_prepare(GSource * source, gint * timeout)
87 crm_trigger_check(GSource * source)
95 crm_trigger_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
104 trig->trigger = FALSE;
107 rc = callback(trig->user_data);
109 crm_trace(
"Trigger handler %p not yet complete", trig);
110 trig->running = TRUE;
118 crm_trigger_finalize(GSource * source)
120 crm_trace(
"Trigger %p destroyed", source);
126 gpointer callback_data;
127 GSourceCallbackFuncs *callback_funcs;
129 const GSourceFuncs *source_funcs;
132 GMainContext *context;
149 g_source_refcount(GSource * source)
153 struct _GSourceCopy *evil = (
struct _GSourceCopy*)source;
154 return evil->ref_count;
159 static int g_source_refcount(GSource * source)
165 static GSourceFuncs crm_trigger_funcs = {
168 crm_trigger_dispatch,
169 crm_trigger_finalize,
173 mainloop_setup_trigger(GSource * source,
int priority,
int (*dispatch) (gpointer user_data),
181 trigger->trigger = FALSE;
182 trigger->user_data = userdata;
185 g_source_set_callback(source, dispatch, trigger, NULL);
188 g_source_set_priority(source, priority);
189 g_source_set_can_recurse(source, FALSE);
191 crm_trace(
"Setup %p with ref-count=%u", source, g_source_refcount(source));
192 trigger->id = g_source_attach(source, NULL);
193 crm_trace(
"Attached %p with ref-count=%u", source, g_source_refcount(source));
201 crm_trace(
"Trigger handler %p complete", trig);
202 trig->running = FALSE;
213 GSource *source = NULL;
216 source = g_source_new(&crm_trigger_funcs,
sizeof(
crm_trigger_t));
219 return mainloop_setup_trigger(source, priority, dispatch, userdata);
226 source->trigger = TRUE;
239 gs = (GSource *)source;
241 if(g_source_refcount(gs) > 2) {
242 crm_info(
"Trigger %p is still referenced %u times", gs, g_source_refcount(gs));
245 g_source_destroy(gs);
257 typedef struct signal_s {
259 void (*handler) (
int sig);
267 crm_signal_dispatch(GSource * source, GSourceFunc callback, gpointer userdata)
271 if(sig->signal != SIGCHLD) {
272 crm_notice(
"Invoking handler for signal %d: %s", sig->signal, strsignal(sig->signal));
275 sig->trigger.trigger = FALSE;
277 sig->handler(sig->signal);
283 mainloop_signal_handler(
int sig)
285 if (sig > 0 && sig < NSIG && crm_signals[sig] != NULL) {
290 static GSourceFuncs crm_signal_funcs = {
294 crm_trigger_finalize,
302 struct sigaction old;
304 if (sigemptyset(&mask) < 0) {
305 crm_perror(LOG_ERR,
"Call to sigemptyset failed");
309 memset(&sa, 0,
sizeof(
struct sigaction));
310 sa.sa_handler = dispatch;
311 sa.sa_flags = SA_RESTART;
314 if (sigaction(sig, &sa, &old) < 0) {
315 crm_perror(LOG_ERR,
"Could not install signal handler for signal %d", sig);
325 GSource *source = NULL;
326 int priority = G_PRIORITY_HIGH - 1;
328 if (sig == SIGTERM) {
336 if (sig >= NSIG || sig < 0) {
337 crm_err(
"Signal %d is out of range", sig);
340 }
else if (crm_signals[sig] != NULL && crm_signals[sig]->handler == dispatch) {
341 crm_trace(
"Signal handler for %d is already installed", sig);
344 }
else if (crm_signals[sig] != NULL) {
345 crm_err(
"Different signal handler for %d is already installed", sig);
350 source = g_source_new(&crm_signal_funcs,
sizeof(
crm_signal_t));
352 crm_signals[sig] = (
crm_signal_t *) mainloop_setup_trigger(source, priority, NULL, NULL);
355 crm_signals[sig]->handler = dispatch;
356 crm_signals[sig]->signal = sig;
358 if (
crm_signal(sig, mainloop_signal_handler) == FALSE) {
361 crm_signals[sig] = NULL;
372 if (siginterrupt(sig, 1) < 0) {
373 crm_perror(LOG_INFO,
"Could not enable system call interruptions for signal %d", sig);
385 if (sig >= NSIG || sig < 0) {
386 crm_err(
"Signal %d is out of range", sig);
390 crm_perror(LOG_ERR,
"Could not uninstall signal handler for signal %d", sig);
393 }
else if (crm_signals[sig] == NULL) {
398 tmp = crm_signals[sig];
399 crm_signals[sig] = NULL;
404 static qb_array_t *gio_map = NULL;
410 qb_array_free(gio_map);
417 struct gio_to_qb_poll {
422 qb_ipcs_dispatch_fn_t fn;
423 enum qb_loop_priority p;
427 gio_read_socket(GIOChannel * gio, GIOCondition condition, gpointer
data)
429 struct gio_to_qb_poll *adaptor = (
struct gio_to_qb_poll *)data;
430 gint fd = g_io_channel_unix_get_fd(gio);
432 crm_trace(
"%p.%d %d", data, fd, condition);
438 return (adaptor->fn(fd, condition, adaptor->data) == 0);
442 gio_poll_destroy(gpointer data)
444 struct gio_to_qb_poll *adaptor = (
struct gio_to_qb_poll *)data;
449 if (adaptor->is_used == 0) {
450 crm_trace(
"Marking adaptor %p unused", adaptor);
456 gio_poll_dispatch_update(
enum qb_loop_priority p,
int32_t fd,
int32_t evts,
457 void *data, qb_ipcs_dispatch_fn_t fn,
int32_t add)
459 struct gio_to_qb_poll *adaptor;
463 res = qb_array_index(gio_map, fd, (
void **)&adaptor);
465 crm_err(
"Array lookup failed for fd=%d: %d", fd, res);
469 crm_trace(
"Adding fd=%d to mainloop as adaptor %p", fd, adaptor);
471 if (add && adaptor->source) {
472 crm_err(
"Adaptor for descriptor %d is still in-use", fd);
475 if (!add && !adaptor->is_used) {
476 crm_err(
"Adaptor for descriptor %d is not in-use", fd);
481 channel = g_io_channel_unix_new(fd);
483 crm_err(
"No memory left to add fd=%d", fd);
487 if (adaptor->source) {
488 g_source_remove(adaptor->source);
493 evts |= (G_IO_HUP | G_IO_NVAL | G_IO_ERR);
496 adaptor->events = evts;
497 adaptor->data =
data;
501 g_io_add_watch_full(channel, G_PRIORITY_DEFAULT, evts, gio_read_socket, adaptor,
513 g_io_channel_unref(channel);
515 crm_trace(
"Added to mainloop with gsource id=%d", adaptor->source);
516 if (adaptor->source > 0) {
524 gio_poll_dispatch_add(
enum qb_loop_priority p,
int32_t fd,
int32_t evts,
525 void *data, qb_ipcs_dispatch_fn_t fn)
527 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_TRUE);
531 gio_poll_dispatch_mod(
enum qb_loop_priority p,
int32_t fd,
int32_t evts,
532 void *data, qb_ipcs_dispatch_fn_t fn)
534 return gio_poll_dispatch_update(p, fd, evts, data, fn, QB_FALSE);
538 gio_poll_dispatch_del(
int32_t fd)
540 struct gio_to_qb_poll *adaptor;
543 if (qb_array_index(gio_map, fd, (
void **)&adaptor) == 0) {
544 if (adaptor->source) {
545 g_source_remove(adaptor->source);
554 .dispatch_add = gio_poll_dispatch_add,
555 .dispatch_mod = gio_poll_dispatch_mod,
556 .dispatch_del = gio_poll_dispatch_del,
559 static enum qb_ipc_type
560 pick_ipc_type(
enum qb_ipc_type requested)
562 const char *env = getenv(
"PCMK_ipc_type");
564 if (env && strcmp(
"shared-mem", env) == 0) {
566 }
else if (env && strcmp(
"socket", env) == 0) {
567 return QB_IPC_SOCKET;
568 }
else if (env && strcmp(
"posix", env) == 0) {
569 return QB_IPC_POSIX_MQ;
570 }
else if (env && strcmp(
"sysv", env) == 0) {
571 return QB_IPC_SYSV_MQ;
572 }
else if (requested == QB_IPC_NATIVE) {
586 struct qb_ipcs_service_handlers * callbacks)
589 qb_ipcs_service_t *server = NULL;
591 if (gio_map == NULL) {
592 gio_map = qb_array_create_2(64,
sizeof(
struct gio_to_qb_poll), 1);
596 server = qb_ipcs_create(name, 0, pick_ipc_type(type), callbacks);
598 #ifdef HAVE_IPCS_GET_BUFFER_SIZE 603 qb_ipcs_poll_handlers_set(server, &gio_poll_funcs);
605 rc = qb_ipcs_run(server);
618 qb_ipcs_destroy(server);
622 struct mainloop_io_s {
631 int (*dispatch_fn_ipc) (
const char *buffer, ssize_t length, gpointer userdata);
632 int (*dispatch_fn_io) (gpointer userdata);
633 void (*destroy_fn) (gpointer userdata);
638 mainloop_gio_callback(GIOChannel * gio, GIOCondition condition, gpointer data)
640 gboolean keep = TRUE;
643 CRM_ASSERT(client->fd == g_io_channel_unix_get_fd(gio));
645 if (condition & G_IO_IN) {
653 crm_trace(
"Message acquisition from %s[%p] failed: %s (%ld)",
656 }
else if (client->dispatch_fn_ipc) {
659 crm_trace(
"New message from %s[%p] = %d", client->name, client, rc, condition);
660 if (client->dispatch_fn_ipc(buffer, rc, client->userdata) < 0) {
661 crm_trace(
"Connection to %s no longer required", client->name);
666 }
while (keep && rc > 0 && --max > 0);
669 crm_trace(
"New message from %s[%p] %u", client->name, client, condition);
670 if (client->dispatch_fn_io) {
671 if (client->dispatch_fn_io(client->userdata) < 0) {
672 crm_trace(
"Connection to %s no longer required", client->name);
680 crm_err(
"Connection to %s[%p] closed (I/O condition=%d)", client->name, client, condition);
683 }
else if (condition & (G_IO_HUP | G_IO_NVAL | G_IO_ERR)) {
684 crm_trace(
"The connection %s[%p] has been closed (I/O condition=%d)",
685 client->name, client, condition);
688 }
else if ((condition & G_IO_IN) == 0) {
716 crm_err(
"Strange condition: %d", condition);
726 mainloop_gio_destroy(gpointer c)
729 char *c_name = strdup(client->name);
734 crm_trace(
"Destroying client %s[%p]", c_name, c);
740 if (client->destroy_fn) {
741 void (*destroy_fn) (gpointer userdata) = client->destroy_fn;
743 client->destroy_fn = NULL;
744 destroy_fn(client->userdata);
754 crm_trace(
"Destroyed client %s[%p]", c_name, c);
756 free(client->name); client->name = NULL;
775 if (client == NULL) {
785 client->destroy_fn = callbacks->
destroy;
786 client->dispatch_fn_ipc = callbacks->
dispatch;
813 if (client == NULL) {
816 client->name = strdup(name);
817 client->userdata = userdata;
820 client->destroy_fn = callbacks->
destroy;
821 client->dispatch_fn_io = callbacks->
dispatch;
825 client->channel = g_io_channel_unix_new(fd);
827 g_io_add_watch_full(client->channel, priority,
828 (G_IO_IN | G_IO_HUP | G_IO_NVAL | G_IO_ERR), mainloop_gio_callback,
829 client, mainloop_gio_destroy);
840 g_io_channel_unref(client->channel);
841 crm_trace(
"Added connection %d for %s[%p].%d", client->source, client->name, client, fd);
852 if (client != NULL) {
853 crm_trace(
"Removing client %s[%p]", client->name, client);
854 if (client->source) {
858 g_source_remove(client->source);
880 return child->timeout;
886 return child->privatedata;
892 child->privatedata = NULL;
899 if (child->timerid != 0) {
900 crm_trace(
"Removing timer %d", child->timerid);
901 g_source_remove(child->timerid);
914 crm_debug(
"Kill pid %d only. leave group intact.", child->pid);
915 rc = kill(child->pid, SIGKILL);
917 crm_debug(
"Kill pid %d's group", child->pid);
918 rc = kill(-child->pid, SIGKILL);
922 if (errno != ESRCH) {
923 crm_perror(LOG_ERR,
"kill(%d, KILL) failed", child->pid);
931 child_timeout_callback(gpointer p)
937 if (child->timeout) {
938 crm_crit(
"%s process (PID %d) will not die!", child->desc, (
int)child->pid);
942 rc = child_kill_helper(child);
948 child->timeout = TRUE;
949 crm_warn(
"%s process (PID %d) timed out", child->desc, (
int)child->pid);
951 child->timerid = g_timeout_add(5000, child_timeout_callback, child);
964 rc = waitpid(child->pid, &status, flags);
966 crm_perror(LOG_DEBUG,
"wait(%d) = %d", child->pid, rc);
969 }
else if(rc != child->pid) {
973 crm_perror(LOG_ERR,
"Call to waitpid(%d) failed", child->pid);
976 crm_trace(
"Managed process %d exited: %p", child->pid, child);
978 if (WIFEXITED(status)) {
979 exitcode = WEXITSTATUS(status);
980 crm_trace(
"Managed process %d (%s) exited with rc=%d", child->pid, child->desc, exitcode);
982 }
else if (WIFSIGNALED(status)) {
983 signo = WTERMSIG(status);
984 crm_trace(
"Managed process %d (%s) exited with signal=%d", child->pid, child->desc, signo);
987 if (WCOREDUMP(status)) {
989 crm_err(
"Managed process %d (%s) dumped core", child->pid, child->desc);
994 if (child->callback) {
995 child->callback(child, child->pid, core, signo, exitcode);
1001 child_death_dispatch(
int signal)
1009 exited = child_waitpid(child, WNOHANG);
1014 if (exited == FALSE) {
1017 crm_trace(
"Removing process entry %p for %d", child, child->pid);
1019 child_list = g_list_remove_link(child_list, saved);
1026 child_signal_init(gpointer p)
1033 child_death_dispatch(SIGCHLD);
1045 int waitflags = 0, rc = 0;
1047 for (iter = child_list; iter != NULL && match == NULL; iter = iter->next) {
1049 if (pid == child->pid) {
1054 if (match == NULL) {
1058 rc = child_kill_helper(match);
1069 crm_trace(
"Waiting for child %d to be reaped by child_death_dispatch()", match->pid);
1072 }
else if(rc != 0) {
1076 waitflags = WNOHANG;
1079 if (child_waitpid(match, waitflags) == FALSE) {
1084 child_list = g_list_remove(child_list, match);
1094 void (*callback) (
mainloop_child_t * p, pid_t pid,
int core,
int signo,
int exitcode))
1096 static bool need_init = TRUE;
1101 child->timeout = FALSE;
1102 child->privatedata = privatedata;
1103 child->callback = callback;
1104 child->flags =
flags;
1107 child->desc = strdup(desc);
1111 child->timerid = g_timeout_add(timeout, child_timeout_callback, child);
1114 child_list = g_list_append(child_list, child);
1122 g_timeout_add(1, child_signal_init, NULL);
1128 void (*callback) (
mainloop_child_t * p, pid_t pid,
int core,
int signo,
int exitcode))
1133 struct mainloop_timer_s {
1144 static gboolean mainloop_timer_cb(gpointer user_data)
1147 bool repeat = FALSE;
1148 struct mainloop_timer_s *t = user_data;
1158 crm_trace(
"Invoking callbacks for timer %s", t->name);
1160 if(t->cb(t->userdata) == FALSE) {
1161 crm_trace(
"Timer %s complete", t->name);
1176 if(t && t->id != 0) {
1185 if(t && t->period_ms > 0) {
1186 crm_trace(
"Starting timer %s", t->name);
1187 t->id = g_timeout_add(t->period_ms, mainloop_timer_cb, t);
1193 if(t && t->id != 0) {
1194 crm_trace(
"Stopping timer %s", t->name);
1195 g_source_remove(t->id);
1205 last = t->period_ms;
1206 t->period_ms = period_ms;
1209 if(t && t->id != 0 && last != t->period_ms) {
1228 t->period_ms = period_ms;
1231 t->userdata = userdata;
1232 crm_trace(
"Created timer %s with %p %p", t->name, userdata, t->userdata);
1241 crm_trace(
"Destroying timer %s", t->name);
void * mainloop_child_userdata(mainloop_child_t *child)
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.
#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)
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