Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com>
---
include/qb/qbloop.h | 55 ++++++++++-
lib/loop.c | 8 +-
lib/loop_int.h | 5 +
lib/loop_poll.c | 279 +++++++++++++++++++++++++++++++++++++++++++++++---
tests/check_loop.c | 13 +--
tests/loop.c | 21 ++++
6 files changed, 352 insertions(+), 29 deletions(-)
diff --git a/include/qb/qbloop.h b/include/qb/qbloop.h
index 026b2be..f8d0b55 100644
--- a/include/qb/qbloop.h
+++ b/include/qb/qbloop.h
@@ -21,6 +21,8 @@
#ifndef QB_LOOP_DEFINED
#define QB_LOOP_DEFINED
+#include <signal.h>
+
/**
* @file qbloop.h
*
@@ -42,12 +44,13 @@ enum qb_loop_priority {
*/
typedef struct qb_loop qb_loop_t;
-typedef void *qb_loop_job_handle;
typedef void *qb_loop_timer_handle;
+typedef void *qb_loop_signal_handle;
typedef int32_t (*qb_loop_poll_dispatch_fn) (int32_t fd, int32_t revents, void *data);
typedef void (*qb_loop_job_dispatch_fn)(void *data);
typedef void (*qb_loop_timer_dispatch_fn)(void *data);
+typedef int32_t (*qb_loop_signal_dispatch_fn)(int32_t rsignal, void *data);
typedef void (*qb_loop_poll_low_fds_event_fn) (int32_t not_enough, int32_t
fds_available);
@@ -141,8 +144,6 @@ uint64_t qb_loop_timer_expire_time_get(struct qb_loop *l,
qb_loop_timer_handle t
int32_t qb_loop_poll_low_fds_event_set(qb_loop_t *l,
qb_loop_poll_low_fds_event_fn fn);
-
-
/**
* Add a poll job to the mainloop.
* @note it is a re-occuring job.
@@ -189,5 +190,53 @@ int32_t qb_loop_poll_mod(qb_loop_t *l,
*/
int32_t qb_loop_poll_del(qb_loop_t *l, int32_t fd);
+/**
+ * Add a signal job.
+ *
+ * Get a callback on this signal (not in the context of the signal).
+ *
+ * @param l pointer to the loop instance
+ * @param p the priority
+ * @param signal (SIGHUP or SIGINT) etc ....
+ * @param data user data passed into the dispatch function
+ * @param dispatch_fn callback function
+ * @param handle (out) a reference to the signal job
+ * @return status (0 == ok, -errno == failure)
+ */
+int32_t qb_loop_signal_add(qb_loop_t *l,
+ enum qb_loop_priority p,
+ int32_t signal,
+ void *data,
+ qb_loop_signal_dispatch_fn dispatch_fn,
+ qb_loop_signal_handle *handle);
+
+/**
+ * Modify the signal job
+ *
+ * @param l pointer to the loop instance
+ * @param p the priority
+ * @param signal (SIGHUP or SIGINT) etc ....
+ * @param data user data passed into the dispatch function
+ * @param dispatch_fn callback function
+ * @param handle (in) a reference to the signal job
+ * @return status (0 == ok, -errno == failure)
+ */
+int32_t qb_loop_signal_mod(qb_loop_t *l,
+ enum qb_loop_priority p,
+ int32_t signal,
+ void *data,
+ qb_loop_signal_dispatch_fn dispatch_fn,
+ qb_loop_signal_handle handle);
+
+/**
+ * Delete the signal job.
+ *
+ * @param l pointer to the loop instance
+ * @param handle (in) a reference to the signal job
+ * @return status (0 == ok, -errno == failure)
+ */
+int32_t qb_loop_signal_del(qb_loop_t *l,
+ qb_loop_signal_handle handle);
+
#endif /* QB_LOOP_DEFINED */
diff --git a/lib/loop.c b/lib/loop.c
index 7f7cf74..6eb802d 100644
--- a/lib/loop.c
+++ b/lib/loop.c
@@ -70,6 +70,7 @@ struct qb_loop * qb_loop_create(void)
l->timer_source = qb_loop_timer_create(l);
l->job_source = qb_loop_jobs_create(l);
l->fd_source = qb_loop_poll_create(l);
+ l->signal_source = qb_loop_signals_create(l);
return l;
}
@@ -79,6 +80,7 @@ void qb_loop_destroy(struct qb_loop * l)
qb_loop_timer_destroy(l);
qb_loop_jobs_destroy(l);
qb_loop_poll_destroy(l);
+ qb_loop_signals_destroy(l);
free(l);
}
@@ -101,8 +103,10 @@ void qb_loop_run(struct qb_loop *l)
p_stop--;
}
- todo += l->job_source->poll(l->job_source, 0);
- if (l->timer_source) {
+ if (l->job_source && l->job_source->poll) {
+ todo += l->job_source->poll(l->job_source, 0);
+ }
+ if (l->timer_source && l->timer_source->poll) {
todo += l->timer_source->poll(l->timer_source, 0);
}
if (todo > 0) {
diff --git a/lib/loop_int.h b/lib/loop_int.h
index d3146e0..e9da182 100644
--- a/lib/loop_int.h
+++ b/lib/loop_int.h
@@ -54,6 +54,7 @@ struct qb_loop {
struct qb_loop_source * timer_source;
struct qb_loop_source * job_source;
struct qb_loop_source * fd_source;
+ struct qb_loop_source * signal_source;
};
struct qb_loop_source *
@@ -65,12 +66,16 @@ qb_loop_timer_create(struct qb_loop *l);
struct qb_loop_source*
qb_loop_poll_create(struct qb_loop *l);
+struct qb_loop_source *
+qb_loop_signals_create(struct qb_loop *l);
+
void qb_loop_jobs_destroy(struct qb_loop *l);
void qb_loop_timer_destroy(struct qb_loop *l);
void qb_loop_poll_destroy(struct qb_loop *l);
+void qb_loop_signals_destroy(struct qb_loop *l);
int32_t qb_loop_timer_msec_duration_to_expire(struct qb_loop_source *timer_source);
diff --git a/lib/loop_poll.c b/lib/loop_poll.c
index 2067f9c..d74f296 100644
--- a/lib/loop_poll.c
+++ b/lib/loop_poll.c
@@ -30,6 +30,7 @@
#include <sys/timerfd.h>
#endif /* HAVE_SYS_TIMERFD_H */
#endif /* S_SPLINT_S */
+#include <signal.h>
#include <qb/qbdefs.h>
#include <qb/qblist.h>
@@ -46,17 +47,23 @@
enum qb_poll_type {
QB_POLL,
QB_TIMER,
+ QB_SIGNAL,
+ QB_JOB,
};
+struct qb_poll_entry;
+
+typedef int32_t (*qb_poll_add_to_jobs_fn) (struct qb_loop* l, struct qb_poll_entry* pe);
struct qb_poll_entry {
struct qb_loop_item item;
- struct pollfd ufd;
+ enum qb_poll_type type;
qb_loop_poll_dispatch_fn poll_dispatch_fn;
qb_loop_timer_dispatch_fn timer_dispatch_fn;
enum qb_loop_priority p;
int32_t install_pos;
- enum qb_poll_type type;
+ struct pollfd ufd;
+ qb_poll_add_to_jobs_fn add_to_jobs;
};
struct qb_poll_source {
@@ -100,7 +107,7 @@ static int32_t epoll_to_poll_event(int32_t event)
#endif /* HAVE_EPOLL */
static void poll_dispatch_and_take_back(struct qb_loop_item * item,
- enum qb_loop_priority p)
+ enum qb_loop_priority p)
{
struct qb_poll_entry *pe = (struct qb_poll_entry *)item;
int32_t res;
@@ -207,9 +214,8 @@ static int32_t poll_and_add_to_jobs(struct qb_loop_source* src,
int32_t ms_timeo
continue;
}
pe->ufd.revents = epoll_to_poll_event(events[i].events);
- qb_list_init(&pe->item.list);
- qb_list_add_tail(&pe->item.list, &s->s.l->level[pe->p].job_head);
- new_jobs++;
+
+ new_jobs += pe->add_to_jobs(src->l, pe);
}
return new_jobs;
@@ -249,9 +255,7 @@ static int32_t poll_and_add_to_jobs(struct qb_loop_source* src,
int32_t ms_timeo
continue;
}
pe->ufd.revents = s->ufds[i].revents;
- qb_list_init(&pe->item.list);
- qb_list_add_tail(&pe->item.list, &s->s.l->level[pe->p].job_head);
- new_jobs++;
+ new_jobs += pe->add_to_jobs(src->l, pe);
}
return new_jobs;
@@ -286,14 +290,16 @@ void qb_loop_poll_destroy(struct qb_loop *l)
struct qb_poll_source * s = (struct qb_poll_source *)l->fd_source;
qb_array_free(s->poll_entries);
#ifdef HAVE_EPOLL
- close(s->epollfd);
+ if (s->epollfd != -1) {
+ close(s->epollfd);
+ s->epollfd = -1;
+ }
#endif /* HAVE_EPOLL */
free(s);
}
-int32_t qb_loop_poll_low_fds_event_set(
- qb_loop_t *l,
- qb_loop_poll_low_fds_event_fn fn)
+int32_t qb_loop_poll_low_fds_event_set(struct qb_loop *l,
+ qb_loop_poll_low_fds_event_fn fn)
{
struct qb_poll_source * s = (struct qb_poll_source *)l->fd_source;
s->low_fds_event_fn = fn;
@@ -301,7 +307,7 @@ int32_t qb_loop_poll_low_fds_event_set(
return 0;
}
-static int32_t new_array_position_get(struct qb_poll_source * s)
+static int32_t _get_empty_array_position_(struct qb_poll_source * s)
{
int32_t found = 0;
int32_t install_pos;
@@ -377,7 +383,7 @@ static int32_t _poll_add_(struct qb_loop *l,
s = (struct qb_poll_source *)l->fd_source;
- install_pos = new_array_position_get(s);
+ install_pos = _get_empty_array_position_(s);
assert(qb_array_index(s->poll_entries, install_pos, (void**)&pe) == 0);
pe->install_pos = install_pos;
@@ -402,6 +408,13 @@ static int32_t _poll_add_(struct qb_loop *l,
return (res);
}
+static int32_t _qb_poll_add_to_jobs_(struct qb_loop* l, struct qb_poll_entry* pe)
+{
+ qb_list_init(&pe->item.list);
+ qb_list_add_tail(&pe->item.list, &l->level[pe->p].job_head);
+ return 1;
+}
+
int32_t qb_loop_poll_add(struct qb_loop *l,
enum qb_loop_priority p,
int32_t fd,
@@ -413,6 +426,7 @@ int32_t qb_loop_poll_add(struct qb_loop *l,
int32_t res = _poll_add_(l, p, fd, events, data, &pe);
pe->poll_dispatch_fn = dispatch_fn;
pe->type = QB_POLL;
+ pe->add_to_jobs = _qb_poll_add_to_jobs_;
return res;
}
@@ -495,7 +509,7 @@ int32_t qb_loop_poll_del(struct qb_loop *l, int32_t fd)
#ifdef HAVE_TIMERFD
int32_t qb_loop_timer_msec_duration_to_expire(struct qb_loop_source *timer_source)
{
- return 0;
+ return -1;
}
struct qb_loop_source*
@@ -557,6 +571,7 @@ int32_t qb_loop_timer_add(struct qb_loop *l,
}
pe->timer_dispatch_fn = timer_fn;
pe->type = QB_TIMER;
+ pe->add_to_jobs = _qb_poll_add_to_jobs_;
*timer_handle_out = pe;
return res;
@@ -596,7 +611,9 @@ int32_t qb_loop_timer_del(struct qb_loop *l, qb_loop_timer_handle th)
s->ufds[pe->install_pos].events = 0;
s->ufds[pe->install_pos].revents = 0;
#endif /* HAVE_EPOLL */
- close(pe->ufd.fd);
+ if (pe->ufd.fd != -1) {
+ close(pe->ufd.fd);
+ }
pe->ufd.fd = -1;
pe->ufd.events = 0;
@@ -624,3 +641,231 @@ uint64_t qb_loop_timer_expire_time_get(struct qb_loop *l,
qb_loop_timer_handle t
#endif /* HAVE_TIMERFD */
+
+static int32_t pipe_fds[2] = {-1, -1};
+
+struct qb_signal_source {
+ struct qb_loop_source s;
+ struct qb_list_head sig_head;
+ sigset_t signal_superset;
+};
+
+struct qb_loop_sig {
+ struct qb_loop_item item;
+ int32_t signal;
+ enum qb_loop_priority p;
+ qb_loop_signal_dispatch_fn dispatch_fn;
+ struct qb_loop_sig *cloned_from;
+};
+
+static void _handle_real_signal_(int signal_num, siginfo_t * si, void *context)
+{
+ int32_t sig = signal_num;
+ if (pipe_fds[1] > 0) {
+ (void)write(pipe_fds[1], &sig, sizeof(int32_t));
+ }
+}
+
+static void signal_dispatch_and_take_back(struct qb_loop_item * item,
+ enum qb_loop_priority p)
+{
+ struct qb_loop_sig *sig = (struct qb_loop_sig *)item;
+ int32_t res;
+
+ res = sig->dispatch_fn(sig->signal, sig->item.user_data);
+ if (res != 0) {
+ qb_list_del(&sig->cloned_from->item.list);
+ free(sig->cloned_from);
+ }
+ free(sig);
+}
+
+
+struct qb_loop_source *
+qb_loop_signals_create(struct qb_loop *l)
+{
+ struct qb_signal_source *s = calloc(1, sizeof(struct qb_signal_source));
+ s->s.l = l;
+ s->s.dispatch_and_take_back = signal_dispatch_and_take_back;
+ s->s.poll = NULL;
+ qb_list_init(&s->sig_head);
+ sigemptyset(&s->signal_superset);
+
+ return (struct qb_loop_source *)s;
+}
+
+void qb_loop_signals_destroy(struct qb_loop *l)
+{
+ close(pipe_fds[0]);
+ pipe_fds[0] = -1;
+ close(pipe_fds[1]);
+ pipe_fds[1] = -1;
+ free(l->signal_source);
+}
+
+static int32_t _qb_signal_add_to_jobs_(struct qb_loop* l,
+ struct qb_poll_entry* pe)
+{
+ struct qb_signal_source *s = (struct qb_signal_source *)l->signal_source;
+ struct qb_list_head *list;
+ struct qb_loop_sig *sig;
+ struct qb_loop_item *item;
+ struct qb_loop_sig *new_sig_job;
+ int32_t the_signal;
+ ssize_t res;
+ int32_t jobs_added = 0;
+
+ res = read(pipe_fds[0], &the_signal, sizeof(int32_t));
+ if (res != sizeof(int32_t)) {
+ res = -errno;
+ qb_util_log(LOG_ERR, "failed to read pipe: %s", strerror(errno));
+ return 0;
+ }
+ pe->ufd.revents = 0;
+
+ qb_list_for_each(list, &s->sig_head) {
+ item = qb_list_entry(list, struct qb_loop_item, list);
+ sig = (struct qb_loop_sig *)item;
+ if (sig->signal == the_signal) {
+ new_sig_job = calloc(1, sizeof(struct qb_loop_sig));
+ memcpy(new_sig_job, sig, sizeof(struct qb_loop_sig));
+
+ new_sig_job->cloned_from = sig;
+ qb_list_init(&new_sig_job->item.list);
+ qb_list_add_tail(&new_sig_job->item.list,
+ &l->level[pe->p].job_head);
+ jobs_added++;
+ }
+ }
+ return jobs_added;
+}
+
+static void _adjust_sigactions_(struct qb_signal_source *s)
+{
+ struct qb_loop_sig *sig;
+ struct qb_loop_item *item;
+ struct sigaction sa;
+ int32_t i;
+ int32_t needed;
+
+ sa.sa_flags = SA_SIGINFO;
+ sa.sa_sigaction = _handle_real_signal_;
+ sigemptyset(&s->signal_superset);
+ sigemptyset(&sa.sa_mask);
+
+ /* re-set to default */
+ for (i = 0; i < 30; i++) {
+ needed = QB_FALSE;
+ qb_list_for_each_entry(item, &s->sig_head, list) {
+ sig = (struct qb_loop_sig *)item;
+ if (i == sig->signal) {
+ needed = QB_TRUE;
+ break;
+ }
+ }
+ if (needed) {
+ sigaddset(&s->signal_superset, i);
+ sigaction(i, &sa, NULL);
+ } else {
+ (void)signal(i, SIG_DFL);
+ }
+ }
+}
+
+int32_t qb_loop_signal_add(qb_loop_t *l,
+ enum qb_loop_priority p,
+ int32_t the_sig,
+ void *data,
+ qb_loop_signal_dispatch_fn dispatch_fn,
+ qb_loop_signal_handle *handle)
+{
+ struct qb_loop_sig *sig;
+ struct qb_poll_entry *pe;
+ struct qb_signal_source *s;
+ int32_t res = 0;
+
+ if (l == NULL || dispatch_fn == NULL) {
+ return -EINVAL;
+ }
+ if (p < QB_LOOP_LOW || p > QB_LOOP_HIGH) {
+ return -EINVAL;
+ }
+ s = (struct qb_signal_source *)l->signal_source;
+ sig = calloc(1, sizeof(struct qb_loop_sig));
+
+ sig->dispatch_fn = dispatch_fn;
+ sig->p = p;
+ sig->signal = the_sig;
+ sig->item.user_data = data;
+ sig->item.source = l->signal_source;
+
+ qb_list_init(&sig->item.list);
+ qb_list_add_tail(&sig->item.list, &s->sig_head);
+
+ if (pipe_fds[0] < 0) {
+ pipe(pipe_fds);
+ res = _poll_add_(l, QB_LOOP_HIGH,
+ pipe_fds[0], POLLIN,
+ NULL, &pe);
+ if (res == 0) {
+ pe->poll_dispatch_fn = NULL;
+ pe->type = QB_SIGNAL;
+ pe->add_to_jobs = _qb_signal_add_to_jobs_;
+ } else {
+ qb_util_log(LOG_ERR,
+ "failed to setup pipe: %s",
+ strerror(-res));
+ }
+ }
+
+ if (sigismember(&s->signal_superset, the_sig) != 1) {
+ _adjust_sigactions_(s);
+ }
+ if (handle) {
+ *handle = sig;
+ }
+
+ return 0;
+}
+
+int32_t qb_loop_signal_mod(qb_loop_t *l,
+ enum qb_loop_priority p,
+ int32_t the_sig,
+ void *data,
+ qb_loop_signal_dispatch_fn dispatch_fn,
+ qb_loop_signal_handle handle)
+{
+ struct qb_signal_source *s;
+ struct qb_loop_sig *sig = (struct qb_loop_sig *)handle;
+
+ if (l == NULL || dispatch_fn == NULL || handle == NULL) {
+ return -EINVAL;
+ }
+ if (p < QB_LOOP_LOW || p > QB_LOOP_HIGH) {
+ return -EINVAL;
+ }
+ s = (struct qb_signal_source *)l->signal_source;
+
+ sig->item.user_data = data;
+ sig->dispatch_fn = dispatch_fn;
+ sig->p = p;
+
+ if (sig->signal != the_sig) {
+ sig->signal = the_sig;
+ _adjust_sigactions_(s);
+ }
+
+ return 0;
+}
+
+int32_t qb_loop_signal_del(qb_loop_t *l,
+ qb_loop_signal_handle handle)
+{
+ struct qb_loop_sig *sig = (struct qb_loop_sig *)handle;
+
+ qb_list_del(&sig->item.list);
+ free(sig);
+ return 0;
+}
+
+
diff --git a/tests/check_loop.c b/tests/check_loop.c
index 0f82b30..f34e026 100644
--- a/tests/check_loop.c
+++ b/tests/check_loop.c
@@ -205,13 +205,13 @@ struct qb_stop_watch {
int32_t ms_timer;
int64_t total;
int32_t count;
+ qb_loop_timer_handle th;
};
static void stop_watch_tmo(void*data)
{
- qb_loop_timer_handle th;
struct qb_stop_watch *sw = (struct qb_stop_watch *)data;
- int64_t per;
+ float per;
sw->end = qb_util_nano_current_get();
sw->total += sw->end - sw->start;
@@ -219,10 +219,10 @@ static void stop_watch_tmo(void*data)
sw->start = sw->end;
sw->count++;
if (sw->count < 50) {
- qb_loop_timer_add(sw->l, QB_LOOP_LOW, sw->ms_timer, data, stop_watch_tmo,
&th);
+ qb_loop_timer_add(sw->l, QB_LOOP_LOW, sw->ms_timer, data, stop_watch_tmo,
&sw->th);
} else {
- per = (sw->total / sw->count) * 100 / (sw->ms_timer * QB_TIME_NS_IN_MSEC);
- printf("average error for %d ms timer is %"PRIi64" (ns)
(%"PRIi64"%%)\n",
+ per = ((sw->total * 100) / sw->count) / (sw->ms_timer *
(float)QB_TIME_NS_IN_MSEC);
+ printf("average error for %d ms timer is %"PRIi64" (ns) (%f)\n",
sw->ms_timer,
sw->total/sw->count, per);
if (sw->ms_timer == 100) {
@@ -233,7 +233,6 @@ static void stop_watch_tmo(void*data)
static void start_timer(qb_loop_t *l, struct qb_stop_watch *sw, int32_t timeout)
{
- qb_loop_timer_handle th;
int32_t res;
sw->l = l;
@@ -241,7 +240,7 @@ static void start_timer(qb_loop_t *l, struct qb_stop_watch *sw,
int32_t timeout)
sw->total = 0;
sw->ms_timer = timeout;
sw->start = qb_util_nano_current_get();
- res = qb_loop_timer_add(sw->l, QB_LOOP_LOW, sw->ms_timer, sw, stop_watch_tmo,
&th);
+ res = qb_loop_timer_add(sw->l, QB_LOOP_LOW, sw->ms_timer, sw, stop_watch_tmo,
&sw->th);
ck_assert_int_eq(res, 0);
}
diff --git a/tests/loop.c b/tests/loop.c
index 38c959e..f920ffb 100644
--- a/tests/loop.c
+++ b/tests/loop.c
@@ -19,6 +19,8 @@
* along with libqb. If not, see <
http://www.gnu.org/licenses/>.
*/
#include "os_base.h"
+#include <signal.h>
+
#include <sys/poll.h>
#include <qb/qbloop.h>
@@ -45,6 +47,19 @@ static void more_important_jobs(void *data)
qb_loop_job_add(l, QB_LOOP_HIGH, NULL, job_1_9);
}
+static int32_t handle_reconf_signal(int32_t sig, void *data)
+{
+ printf("%s(%d) \n", __func__, sig);
+ return 0;
+}
+
+static int32_t handle_exit_signal(int32_t sig, void *data)
+{
+ printf("%s(%d) exiting ... bye\n", __func__, sig);
+ qb_loop_stop(l);
+ return -1;
+}
+
static void more_jobs(void *data)
{
printf("%s\n", __func__);
@@ -82,6 +97,8 @@ static void libqb_log_fn(const char *file_name,
int main(int argc, char * argv[])
{
+ qb_loop_signal_handle sh;
+
qb_util_set_log_function(libqb_log_fn);
l = qb_loop_create();
@@ -97,6 +114,10 @@ int main(int argc, char * argv[])
qb_loop_poll_add(l, QB_LOOP_LOW, 0, POLLIN | POLLPRI | POLLNVAL,
NULL, read_stdin);
+ qb_loop_signal_add(l, QB_LOOP_MED, SIGINT, NULL, handle_exit_signal, &sh);
+ qb_loop_signal_add(l, QB_LOOP_MED, SIGSEGV, NULL, handle_exit_signal, &sh);
+ qb_loop_signal_add(l, QB_LOOP_MED, SIGHUP, NULL, handle_reconf_signal, &sh);
+
qb_loop_run(l);
return 0;
}
--
1.7.3.1