[PATCH 1/5] LOOP: prevent timers from deleting them selves.

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ipcs.c         |    2 +-
 lib/loop_poll.c    |  112 ++++++++++++++++++++++++++++++++++++++++-----------
 tests/check_loop.c |   63 +++++++++++++++++++++++++++--
 3 files changed, 147 insertions(+), 30 deletions(-)

diff --git a/lib/ipcs.c b/lib/ipcs.c
index e744f51..962bd09 100644
--- a/lib/ipcs.c
+++ b/lib/ipcs.c
@@ -176,7 +176,7 @@ void qb_ipcs_unref(struct qb_ipcs_service *s)
 			}
 			qb_ipcs_disconnect(c);
 		}
-		qb_ipcs_us_withdraw(s);
+		(void)qb_ipcs_us_withdraw(s);
 		free(s);
 	}
 }
diff --git a/lib/loop_poll.c b/lib/loop_poll.c
index 2128dd0..8e667f9 100644
--- a/lib/loop_poll.c
+++ b/lib/loop_poll.c
@@ -55,6 +55,14 @@ enum qb_poll_type {
 	QB_JOB,
 };
 
+enum qb_poll_entry_state {
+	QB_POLL_ENTRY_EMPTY,
+	QB_POLL_ENTRY_EXPIRED,
+	QB_POLL_ENTRY_DELETED,
+	QB_POLL_ENTRY_ACTIVE,
+};
+
+
 struct qb_poll_entry;
 
 typedef int32_t (*qb_poll_add_to_jobs_fn) (struct qb_loop* l, struct qb_poll_entry* pe);
@@ -68,6 +76,9 @@ struct qb_poll_entry {
 	int32_t install_pos;
 	struct pollfd ufd;
 	qb_poll_add_to_jobs_fn add_to_jobs;
+	uint32_t runs;
+	enum qb_poll_entry_state state;
+	qb_loop_timer_handle handle_addr;
 };
 
 struct qb_poll_source {
@@ -118,30 +129,48 @@ static void poll_dispatch_and_take_back(struct qb_loop_item * item,
 #ifdef DEBUG_DISPATCH_TIME
 	uint64_t start;
 	uint64_t stop;
+	int32_t log_warn = QB_FALSE;
 
 	start = qb_util_nano_current_get();
 #endif /* DEBUG_DISPATCH_TIME */
 	if (pe->type == QB_POLL) {
 		res = pe->poll_dispatch_fn(pe->ufd.fd, pe->ufd.revents,
pe->item.user_data);
 		if (res < 0) {
+			pe->state = QB_POLL_ENTRY_EXPIRED;
 			pe->ufd.fd = -1; /* empty entry */
 		}
-		pe->ufd.revents = 0;
 	} else {
+		pe->state = QB_POLL_ENTRY_EXPIRED;
+		memset(pe->handle_addr, 0, sizeof(struct qb_poll_entry *));
 		pe->timer_dispatch_fn(pe->item.user_data);
-		if (pe->ufd.fd != -1) {
-			close(pe->ufd.fd);
-			pe->ufd.fd = -1; /* empty entry */
-			pe->ufd.revents = 0;
-		}
 	}
+	if (pe->state == QB_POLL_ENTRY_ACTIVE) {
+		pe->ufd.revents = 0;
 #ifdef DEBUG_DISPATCH_TIME
-	stop = qb_util_nano_current_get();
-	if ((stop - start) > (10 * QB_TIME_NS_IN_MSEC)) {
-		qb_util_log(LOG_WARNING, " dispatch function \"%p\" took %d ms",
-			    pe->poll_dispatch_fn, (int32_t)((stop - start)/QB_TIME_NS_IN_MSEC));
-	}
+		pe->runs++;
+		if ((pe->runs % 50) == 0) {
+			log_warn = QB_TRUE;
+		}
+		stop = qb_util_nano_current_get();
+		if ((stop - start) > (10 * QB_TIME_NS_IN_MSEC)) {
+			log_warn = QB_TRUE;
+		}
+
+		if (log_warn && pe->type == QB_POLL) {
+			qb_util_log(LOG_INFO,
+				    "[fd:%d] dispatch:%p runs:%d duration:%d ms",
+				    pe->ufd.fd, pe->poll_dispatch_fn,
+				    pe->runs,
+				    (int32_t)((stop - start)/QB_TIME_NS_IN_MSEC));
+		} else if (log_warn && pe->type == QB_TIMER) {
+			qb_util_log(LOG_INFO,
+				    "[timer] dispatch:%p runs:%d duration:%d ms",
+				    pe->timer_dispatch_fn,
+				    pe->runs,
+				    (int32_t)((stop - start)/QB_TIME_NS_IN_MSEC));
+		}
 #endif /* DEBUG_DISPATCH_TIME */
+	}
 }
 
 static void poll_fds_usage_check(struct qb_poll_source *s)
@@ -170,7 +199,7 @@ static void poll_fds_usage_check(struct qb_poll_source *s)
 
 	for (i = 0; i < s->poll_entry_count; i++) {
 		assert(qb_array_index(s->poll_entries, i, (void**)&pe) == 0);
-		if (pe->ufd.fd != -1) {
+		if (pe->state == QB_POLL_ENTRY_ACTIVE && pe->ufd.fd != -1) {
 			socks_used++;
 		}
 	}
@@ -197,7 +226,7 @@ static void poll_fds_usage_check(struct qb_poll_source *s)
 }
 
 #ifdef HAVE_EPOLL
-#define MAX_EVENTS 100
+#define MAX_EVENTS 12
 static int32_t poll_and_add_to_jobs(struct qb_loop_source* src, int32_t ms_timeout)
 {
 	int32_t i;
@@ -209,6 +238,14 @@ static int32_t poll_and_add_to_jobs(struct qb_loop_source* src,
int32_t ms_timeo
 
 	poll_fds_usage_check(s);
 
+	for (i = 0; i < s->poll_entry_count; i++) {
+		assert(qb_array_index(s->poll_entries, i, (void**)&pe) == 0);
+		if (pe->state == QB_POLL_ENTRY_DELETED ||
+		    pe->state == QB_POLL_ENTRY_EXPIRED) {
+		    pe->state = QB_POLL_ENTRY_EMPTY;
+		}
+	}
+
  retry_poll:
 
 	res = epoll_wait(s->epollfd, events, MAX_EVENTS, ms_timeout);
@@ -250,6 +287,11 @@ static int32_t poll_and_add_to_jobs(struct qb_loop_source* src,
int32_t ms_timeo
 	for (i = 0; i < s->poll_entry_count; i++) {
 		assert(qb_array_index(s->poll_entries, i, (void**)&pe) == 0);
 		memcpy(&s->ufds[i], &pe->ufd, sizeof(struct pollfd));
+
+		if (pe->state == QB_POLL_ENTRY_DELETED ||
+		    pe->state == QB_POLL_ENTRY_EXPIRED) {
+		    pe->state = QB_POLL_ENTRY_EMPTY;
+		}
 	}
 
  retry_poll:
@@ -339,7 +381,7 @@ static int32_t _get_empty_array_position_(struct qb_poll_source * s)
 	for (found = 0, install_pos = 0;
 	     install_pos < s->poll_entry_count; install_pos++) {
 		assert(qb_array_index(s->poll_entries, install_pos, (void**)&pe) == 0);
-		if (pe->ufd.fd == -1) {
+		if (pe->state == QB_POLL_ENTRY_EMPTY) {
 			found = 1;
 			break;
 		}
@@ -402,6 +444,7 @@ static int32_t _poll_add_(struct qb_loop *l,
 	install_pos = _get_empty_array_position_(s);
 
 	assert(qb_array_index(s->poll_entries, install_pos, (void**)&pe) == 0);
+	pe->state = QB_POLL_ENTRY_ACTIVE;
 	pe->install_pos = install_pos;
 	pe->ufd.fd = fd;
 	pe->ufd.events = events;
@@ -409,6 +452,7 @@ static int32_t _poll_add_(struct qb_loop *l,
 	pe->item.user_data = data;
 	pe->item.source = (struct qb_loop_source*)l->fd_source;
 	pe->p = p;
+	pe->runs = 0;
 #ifdef HAVE_EPOLL
 	ev = &s->events[install_pos];
 	ev->events = poll_to_epoll_event(events);
@@ -424,6 +468,23 @@ static int32_t _poll_add_(struct qb_loop *l,
 	return (res);
 }
 
+static int32_t _qb_timer_add_to_jobs_(struct qb_loop* l, struct qb_poll_entry* pe)
+{
+	uint64_t expired = 0;
+
+	if (pe->ufd.revents == POLLIN) {
+		(void)read(pe->ufd.fd, &expired, sizeof(expired));
+		qb_list_init(&pe->item.list);
+		qb_list_add_tail(&pe->item.list, &l->level[pe->p].job_head);
+	} else {
+		qb_util_log(LOG_ERR, "timer revents: %d expected %d",
+			    pe->ufd.revents, POLLIN);
+	}
+	close(pe->ufd.fd);
+	pe->ufd.fd = -1;
+	return 1;
+}
+
 static int32_t _qb_poll_add_to_jobs_(struct qb_loop* l, struct qb_poll_entry* pe)
 {
 	qb_list_init(&pe->item.list);
@@ -503,6 +564,7 @@ int32_t qb_loop_poll_del(struct qb_loop *l, int32_t fd)
 			continue;
 		}
 		pe->ufd.fd = -1;
+		pe->state = QB_POLL_ENTRY_DELETED;
 		pe->ufd.events = 0;
 		pe->ufd.revents = 0;
 #ifdef HAVE_EPOLL
@@ -579,16 +641,16 @@ int32_t qb_loop_timer_add(struct qb_loop *l,
 	}
 
 	res = _poll_add_(l, p, fd, POLLIN, data, &pe);
-	if (res == -1) {
-		res = -errno;
+	if (res != 0) {
 		qb_util_log(LOG_ERR, "_poll_add_() : %s",
 				    strerror(-res));
 		goto close_and_return;
 	}
 	pe->timer_dispatch_fn = timer_fn;
 	pe->type = QB_TIMER;
-	pe->add_to_jobs = _qb_poll_add_to_jobs_;
+	pe->add_to_jobs = _qb_timer_add_to_jobs_;
 	*timer_handle_out = pe;
+	pe->handle_addr = timer_handle_out;
 
 	return res;
 
@@ -614,27 +676,29 @@ int32_t qb_loop_timer_del(struct qb_loop *l, qb_loop_timer_handle
th)
 	if (pe->type != QB_TIMER) {
 		return -EINVAL;
 	}
+	memset(pe->handle_addr, 0, sizeof(struct qb_poll_entry *));
+	if (pe->state != QB_POLL_ENTRY_ACTIVE &&
+	    pe->state != QB_POLL_ENTRY_EXPIRED) {
+		return -EINVAL;
+	}
 	if (pe->ufd.fd != -1) {
 #ifdef HAVE_EPOLL
 		if (epoll_ctl(s->epollfd, EPOLL_CTL_DEL, pe->ufd.fd, NULL) == -1) {
 			res = -errno;
 			qb_util_log(LOG_ERR, "epoll_ctl(del:%d) : %s",
 				    pe->ufd.fd, strerror(-res));
-			return res;
 		}
 #else
 		s->ufds[pe->install_pos].fd = -1;
 		s->ufds[pe->install_pos].events = 0;
 		s->ufds[pe->install_pos].revents = 0;
 #endif /* HAVE_EPOLL */
-		if (pe->ufd.fd != -1) {
-			close(pe->ufd.fd);
-		}
-
+		close(pe->ufd.fd);
 		pe->ufd.fd = -1;
-		pe->ufd.events = 0;
-		pe->ufd.revents = 0;
 	}
+	pe->ufd.events = 0;
+	pe->ufd.revents = 0;
+	pe->state = QB_POLL_ENTRY_DELETED;
 
 	return 0;
 }
diff --git a/tests/check_loop.c b/tests/check_loop.c
index 18edbcc..274da97 100644
--- a/tests/check_loop.c
+++ b/tests/check_loop.c
@@ -180,17 +180,17 @@ static Suite *loop_job_suite(void)
  * -----------------------------------------------------------------------
  *  Timers
  */
+static qb_loop_timer_handle test_th;
 
 START_TEST(test_loop_timer_input)
 {
 	int32_t res;
-	qb_loop_timer_handle th;
 	qb_loop_t *l = qb_loop_create();
 	fail_if(l == NULL);
 
-	res = qb_loop_timer_add(NULL, QB_LOOP_LOW, 5, NULL, job_2, &th);
+	res = qb_loop_timer_add(NULL, QB_LOOP_LOW, 5, NULL, job_2, &test_th);
 	ck_assert_int_eq(res, -EINVAL);
-	res = qb_loop_timer_add(l, QB_LOOP_LOW, 5, l, NULL, &th);
+	res = qb_loop_timer_add(l, QB_LOOP_LOW, 5, l, NULL, &test_th);
 	ck_assert_int_eq(res, -EINVAL);
 	res = qb_loop_timer_add(l, QB_LOOP_LOW, 5, l, job_1, NULL);
 	ck_assert_int_eq(res, -ENOENT);
@@ -198,6 +198,52 @@ START_TEST(test_loop_timer_input)
 }
 END_TEST
 
+static void one_shot_tmo(void*data)
+{
+	static int32_t been_here = QB_FALSE;
+	ck_assert_int_eq(been_here, QB_FALSE);
+	been_here = QB_TRUE;
+}
+
+static qb_loop_timer_handle reset_th;
+static int32_t reset_timer_step = 0;
+static void reset_one_shot_tmo(void*data)
+{
+	int32_t res;
+	qb_loop_t *l = data;
+	if (reset_timer_step == 0) {
+		res = qb_loop_timer_del(l, reset_th);
+		ck_assert_int_eq(res, -EINVAL);
+		res = qb_loop_timer_add(l, QB_LOOP_LOW, 8, l, reset_one_shot_tmo, &reset_th);
+		ck_assert_int_eq(res, 0);
+	}
+
+	reset_timer_step++;
+}
+
+START_TEST(test_loop_timer_basic)
+{
+	int32_t res;
+	qb_loop_t *l = qb_loop_create();
+	fail_if(l == NULL);
+
+	res = qb_loop_timer_add(l, QB_LOOP_LOW, 5, l, one_shot_tmo, &test_th);
+	ck_assert_int_eq(res, 0);
+
+	res = qb_loop_timer_add(l, QB_LOOP_LOW, 7, l, reset_one_shot_tmo, &reset_th);
+	ck_assert_int_eq(res, 0);
+
+	res = qb_loop_timer_add(l, QB_LOOP_LOW, 60, l, job_stop, &test_th);
+	ck_assert_int_eq(res, 0);
+
+	qb_loop_run(l);
+
+	ck_assert_int_eq(reset_timer_step, 2);
+
+	qb_loop_destroy(l);
+}
+END_TEST
+
 struct qb_stop_watch {
 	uint64_t start;
 	uint64_t end;
@@ -217,8 +263,10 @@ static void stop_watch_tmo(void*data)
 	sw->end = qb_util_nano_current_get();
 
 	diff = sw->end - sw->start;
+	if (diff < (sw->ms_timer * QB_TIME_NS_IN_MSEC)) {
+		printf("timer expired early! by %lld\n", (sw->ms_timer *
QB_TIME_NS_IN_MSEC) - diff);
+	}
 	ck_assert(diff >= (sw->ms_timer * QB_TIME_NS_IN_MSEC));
-
 	sw->total += diff;
 	sw->total -= sw->ms_timer * QB_TIME_NS_IN_MSEC;
 	sw->start = sw->end;
@@ -251,7 +299,7 @@ static void start_timer(qb_loop_t *l, struct qb_stop_watch *sw,
int32_t timeout)
 }
 
 
-START_TEST(test_loop_timer_basic)
+START_TEST(test_loop_timer_precision)
 {
 	int32_t i;
 	int32_t tmo;
@@ -285,6 +333,11 @@ static Suite *loop_timer_suite(void)
 	tcase_set_timeout(tc, 30);
 	suite_add_tcase(s, tc);
 
+	tc = tcase_create("precision");
+	tcase_add_test(tc, test_loop_timer_precision);
+	tcase_set_timeout(tc, 30);
+	suite_add_tcase(s, tc);
+
 	return s;
 }
 
-- 
1.7.3.1

[PATCH 2/5] IPC: remove try again log message

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ipcs.c |    4 ++--
 1 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/lib/ipcs.c b/lib/ipcs.c
index 962bd09..37d950b 100644
--- a/lib/ipcs.c
+++ b/lib/ipcs.c
@@ -247,7 +247,7 @@ ssize_t qb_ipcs_event_send(struct qb_ipcs_connection *c, const void
*data,
 				res2 = qb_ipc_us_send(&c->setup, &res, 1);
 			} while (res2 == -EAGAIN);
 		}
-	} else {
+	} else if (res != -EAGAIN) {
 		qb_util_log(LOG_ERR,
 			    "failed to send event : %s",
 			    strerror(-res));
@@ -281,7 +281,7 @@ ssize_t qb_ipcs_event_sendv(struct qb_ipcs_connection *c, const struct
iovec * i
 				res2 = qb_ipc_us_send(&c->setup, &res, 1);
 			} while (res2 == -EAGAIN);
 		}
-	} else {
+	} else if (res != -EAGAIN) {
 		qb_util_log(LOG_ERR,
 			    "failed to send event : %s",
 			    strerror(-res));
-- 
1.7.3.1

[PATCH 3/5] Fix some error handling in RB & IPC

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ipc_shm.c    |   18 +++++++++++-------
 lib/ipc_us.c     |   17 ++++++++++-------
 lib/ringbuffer.c |   21 +++++++++++++++------
 lib/util.c       |    2 ++
 4 files changed, 38 insertions(+), 20 deletions(-)

diff --git a/lib/ipc_shm.c b/lib/ipc_shm.c
index 220586c..e9b60d5 100644
--- a/lib/ipc_shm.c
+++ b/lib/ipc_shm.c
@@ -149,14 +149,15 @@ int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection *c,
 					 QB_RB_FLAG_SHARED_PROCESS,
 					 sizeof(int32_t));
 	if (c->request.u.shm.rb == NULL) {
-		perror("qb_rb_open:REQUEST");
-		return -errno;
+		res = -errno;
+		goto return_error;
 	}
 	c->response.u.shm.rb = qb_rb_open(response->response,
 					  c->response.max_msg_size,
 					  QB_RB_FLAG_SHARED_PROCESS, 0);
 
 	if (c->response.u.shm.rb == NULL) {
+		res = -errno;
 		perror("qb_rb_open:RESPONSE");
 		goto cleanup_request;
 	}
@@ -171,13 +172,15 @@ int32_t qb_ipcc_shm_connect(struct qb_ipcc_connection *c,
 	}
 	return 0;
 
-cleanup_request_response:
+ cleanup_request_response:
 	qb_rb_close(c->response.u.shm.rb);
 
-cleanup_request:
+ cleanup_request:
 	qb_rb_close(c->request.u.shm.rb);
 
-	qb_util_log(LOG_DEBUG, "connection failed %d\n", res);
+ return_error:
+	qb_util_log(LOG_ERR, "connection failed %s\n",
+		    strerror(-res));
 
 	return res;
 }
@@ -226,7 +229,6 @@ static int32_t qb_ipcs_shm_connect(struct qb_ipcs_service *s,
 	snprintf(r->response, NAME_MAX, "qb-%s-response-%d-%d", s->name,
c->pid, c->setup.u.us.sock);
 	snprintf(r->event, NAME_MAX, "qb-%s-event-%d-%d", s->name, c->pid,
c->setup.u.us.sock);
 
-	qb_util_log(LOG_DEBUG, "rb_open:%s", r->request);
 	c->request.u.shm.rb = qb_rb_open(r->request,
 					 c->request.max_msg_size,
 					 QB_RB_FLAG_CREATE |
@@ -257,7 +259,7 @@ static int32_t qb_ipcs_shm_connect(struct qb_ipcs_service *s,
 
 	if (c->event.u.shm.rb == NULL) {
 		res = -errno;
-		perror("mq_open:EVENT");
+		perror("qb_rb_open:EVENT");
 		goto cleanup_request_response;
 	}
 	res = qb_rb_chown(c->event.u.shm.rb, c->euid, c->egid);
@@ -273,6 +275,8 @@ cleanup_request:
 
 cleanup:
 	r->hdr.error = res;
+	qb_util_log(LOG_ERR, "shm connection FAILED [%s]\n",
+		    strerror(-res));
 
 	return res;
 }
diff --git a/lib/ipc_us.c b/lib/ipc_us.c
index cf55b22..2f11daf 100644
--- a/lib/ipc_us.c
+++ b/lib/ipc_us.c
@@ -567,7 +567,6 @@ static int32_t handle_new_connection(struct qb_ipcs_service *s,
 	struct qb_ipc_connection_request *req = msg;
 	int32_t res = auth_result;
 	struct qb_ipc_connection_response response;
-	char error_str[100];
 
 	if (res != 0) {
 		goto send_response;
@@ -638,8 +637,13 @@ send_response:
 		s->stats.active_connections++;
 	}
 
-	if (qb_ipc_us_send(&c->setup, &response, response.hdr.size) < 0) {
-		qb_util_log(LOG_ERR, "Error send connection response.");
+	res = qb_ipc_us_send(&c->setup, &response, response.hdr.size);
+	if (res == response.hdr.size) {
+		res = 0;
+	}
+	if (res < 0) {
+		qb_util_log(LOG_ERR, "Error sending connection response: %s",
+			    strerror(-res));
 	}
 
 	if (res == 0) {
@@ -649,9 +653,8 @@ send_response:
 	} else if (res == -EACCES) {
 		qb_util_log(LOG_ERR, "Invalid IPC credentials.");
 	} else {
-		strerror_r(-response.hdr.error, error_str, 100);
 		qb_util_log(LOG_ERR, "Error in connection setup: %s.",
-			    error_str);
+			    strerror(-res));
 	}
 	if (res != 0 && c) {
 		qb_ipcs_disconnect(c);
@@ -806,10 +809,10 @@ retry_accept:
 		return -1;
 	}
 	if (new_fd == -1) {
-		strerror_r(errno, error_str, 100);
+	res = -errno;
 		qb_util_log(LOG_ERR,
 			    "Could not accept Library connection: [%d] %s\n",
-			    errno, error_str);
+			    errno, strerror(-res));
 		return 0;	/* This is an error, but -1 would indicate disconnect from poll loop */
 	}
 
diff --git a/lib/ringbuffer.c b/lib/ringbuffer.c
index bc57a7d..17365e5 100644
--- a/lib/ringbuffer.c
+++ b/lib/ringbuffer.c
@@ -91,6 +91,7 @@ qb_ringbuffer_t *qb_rb_open(const char *name, size_t size, uint32_t
flags,
 	uint32_t file_flags = O_RDWR;
 	size_t shared_size = sizeof(struct qb_ringbuffer_shared_s);
 	char filename[PATH_MAX];
+	int32_t error = 0;
 
 	shared_size += shared_user_data_size;
 
@@ -105,15 +106,17 @@ qb_ringbuffer_t *qb_rb_open(const char *name, size_t size, uint32_t
flags,
 					shared_size,
 					file_flags);
 	if (fd_hdr < 0) {
+		error = fd_hdr;
 		qb_util_log(LOG_ERR, "couldn't create file for mmap");
 		return NULL;
 	}
 
 	rb->shared_hdr = mmap(0,
-			      sizeof(struct qb_ringbuffer_shared_s),
+			      shared_size,
 			      PROT_READ | PROT_WRITE, MAP_SHARED, fd_hdr, 0);
 
 	if (rb->shared_hdr == MAP_FAILED) {
+		error = -errno;
 		qb_util_log(LOG_ERR, "couldn't create mmap for header");
 		goto cleanup_hdr;
 	}
@@ -131,9 +134,10 @@ qb_ringbuffer_t *qb_rb_open(const char *name, size_t size, uint32_t
flags,
 		rb->shared_hdr->read_pt = 0;
 		strncpy(rb->shared_hdr->hdr_path, path, PATH_MAX);
 	}
-	if (qb_rb_sem_create(rb, flags) < 0) {
+	error = qb_rb_sem_create(rb, flags);
+	if (error < 0) {
 		qb_util_log(LOG_ERR, "couldn't get a semaphore %s",
-			    strerror(errno));
+			    strerror(-error));
 		goto cleanup_hdr;
 	}
 
@@ -152,16 +156,20 @@ qb_ringbuffer_t *qb_rb_open(const char *name, size_t size, uint32_t
flags,
 						 real_size, file_flags);
 	}
 	if (fd_data < 0) {
+		error = fd_data;
 		qb_util_log(LOG_ERR, "couldn't create file for mmap");
 		goto cleanup_hdr;
 	}
 
 	qb_util_log(LOG_DEBUG,
-		    "shm \n size:%zd\n real_size:%zd\n rb->size:%d\n", size,
+		    "shm size:%zd; real_size:%zd; rb->size:%d", size,
 		    real_size, rb->shared_hdr->size);
 
-	if (qb_util_circular_mmap(fd_data,
-				  (void **)&rb->shared_data, real_size) != 0) {
+	error = qb_util_circular_mmap(fd_data,
+				  (void **)&rb->shared_data, real_size);
+	if (error != 0) {
+		qb_util_log(LOG_ERR, "couldn't create circular mmap on %s",
+			    rb->shared_hdr->data_path);
 		goto cleanup_data;
 	}
 
@@ -193,6 +201,7 @@ cleanup_hdr:
 		munmap(rb->shared_hdr, sizeof(struct qb_ringbuffer_shared_s));
 	}
 	free(rb);
+	errno = -error;
 	return NULL;
 }
 
diff --git a/lib/util.c b/lib/util.c
index 0452354..0ab5f1a 100644
--- a/lib/util.c
+++ b/lib/util.c
@@ -288,6 +288,7 @@ int32_t qb_util_mmap_file_open(char *path, const char *file, size_t
bytes,
 		long page_size = sysconf(_SC_PAGESIZE);
 		buffer = calloc(1, page_size);
 		if (buffer == NULL) {
+			res = -ENOMEM;
 			goto unlink_exit;
 		}
 		for (i = 0; i < (bytes / page_size); i++) {
@@ -297,6 +298,7 @@ retry_write:
 				goto retry_write;
 			}
 			if (written != page_size) {
+				res = -ENOSPC;
 				free (buffer);
 				goto unlink_exit;
 			}
-- 
1.7.3.1

[PATCH 4/5] RB: fix munmap size (else shared mem not freed correctly)

[Angus Salkeld]

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ringbuffer.c |    2 +-
 1 files changed, 1 insertions(+), 1 deletions(-)

diff --git a/lib/ringbuffer.c b/lib/ringbuffer.c
index 17365e5..ad3c50d 100644
--- a/lib/ringbuffer.c
+++ b/lib/ringbuffer.c
@@ -217,7 +217,7 @@ void qb_rb_close(qb_ringbuffer_t * rb)
 	(void)rb->sem_destroy_fn(rb);
 	unlink(rb->shared_hdr->data_path);
 	unlink(rb->shared_hdr->hdr_path);
-	munmap(rb->shared_data, rb->shared_hdr->size);
+	munmap(rb->shared_data, (rb->shared_hdr->size * sizeof(uint32_t)) << 1);
 	munmap(rb->shared_hdr, sizeof(struct qb_ringbuffer_shared_s));
 	free(rb);
 }
-- 
1.7.3.1

[PATCH 5/5] LOOP: move pipe creation into qb_loop_signals_create()

[Angus Salkeld]

Also make sure the pipe is non blocking.

Signed-off-by: Angus Salkeld <asalkeld(a)redhat.com&gt;
---
 lib/ipc_us.c    |   56 ++--------------------------------------------
 lib/loop_poll.c |   66 +++++++++++++++++++++++++++++++++++++++---------------
 lib/util.c      |   33 +++++++++++++++++++++++++++
 lib/util_int.h  |   12 ++++++++-
 4 files changed, 93 insertions(+), 74 deletions(-)

diff --git a/lib/ipc_us.c b/lib/ipc_us.c
index 2f11daf..7b75522 100644
--- a/lib/ipc_us.c
+++ b/lib/ipc_us.c
@@ -73,43 +73,6 @@ static void socket_nosigpipe(int32_t s)
 }
 #endif
 
-static int32_t set_cloexec_flag(int32_t fd)
-{
-	int32_t res;
-	char error_str[100];
-	int32_t oldflags = fcntl(fd, F_GETFD, 0);
-
-	if (oldflags < 0) {
-		oldflags = 0;
-	}
-	oldflags |= FD_CLOEXEC;
-	res = fcntl(fd, F_SETFD, oldflags);
-	if (res == -1) {
-		res = -errno;
-		strerror_r(errno, error_str, 100);
-		qb_util_log(LOG_CRIT,
-			    "Could not set close-on-exit operation on socket: %s\n",
-			    error_str);
-	}
-	return res;
-}
-
-static int32_t set_nonblock_flag(int32_t fd)
-{
-	int32_t res;
-	char error_str[100];
-
-	res = fcntl(fd, F_SETFL, O_NONBLOCK);
-	if (res == -1) {
-		res = -errno;
-		strerror_r(errno, error_str, 100);
-		qb_util_log(LOG_CRIT,
-			    "Could not set non-blocking operation on socket: %s\n",
-			    error_str);
-	}
-	return res;
-}
-
 #ifndef MSG_NOSIGNAL
 #define MSG_NOSIGNAL 0
 #endif
@@ -302,11 +265,7 @@ static int32_t qb_ipcc_us_sock_connect(const char *socket_name,
int32_t * sock_p
 #ifdef SO_NOSIGPIPE
 	socket_nosigpipe(request_fd);
 #endif /* SO_NOSIGPIPE */
-	res = set_cloexec_flag(request_fd);
-	if (res < 0) {
-		goto error_connect;
-	}
-	res = set_nonblock_flag(request_fd);
+	res = qb_util_fd_nonblock_cloexec_set(request_fd);
 	if (res < 0) {
 		goto error_connect;
 	}
@@ -481,11 +440,7 @@ int32_t qb_ipcs_us_publish(struct qb_ipcs_service * s)
 		return res;
 	}
 
-	res = set_cloexec_flag(s->server_sock);
-	if (res < 0) {
-		goto error_close;
-	}
-	res = set_nonblock_flag(s->server_sock);
+	res = qb_util_fd_nonblock_cloexec_set(s->server_sock);
 	if (res < 0) {
 		goto error_close;
 	}
@@ -816,12 +771,7 @@ retry_accept:
 		return 0;	/* This is an error, but -1 would indicate disconnect from poll loop */
 	}
 
-	res = set_cloexec_flag(new_fd);
-	if (res < 0) {
-		close(new_fd);
-		return 0;	/* This is an error, but -1 would indicate disconnect from poll loop */
-	}
-	res = set_nonblock_flag(new_fd);
+	res = qb_util_fd_nonblock_cloexec_set(new_fd);
 	if (res < 0) {
 		close(new_fd);
 		return 0;	/* This is an error, but -1 would indicate disconnect from poll loop */
diff --git a/lib/loop_poll.c b/lib/loop_poll.c
index 8e667f9..12f317b 100644
--- a/lib/loop_poll.c
+++ b/lib/loop_poll.c
@@ -95,6 +95,8 @@ struct qb_poll_source {
 	int32_t not_enough_fds;
 };
 
+static int32_t _qb_signal_add_to_jobs_(struct qb_loop* l,
+				       struct qb_poll_entry* pe);
 
 #ifdef HAVE_EPOLL
 static int32_t poll_to_epoll_event(int32_t event)
@@ -741,8 +743,17 @@ struct qb_loop_sig {
 static void _handle_real_signal_(int signal_num, siginfo_t * si, void *context)
 {
 	int32_t sig = signal_num;
+	int32_t res = 0;
+
 	if (pipe_fds[1] > 0) {
-		(void)write(pipe_fds[1], &sig, sizeof(int32_t));
+ try_again:
+		res = write(pipe_fds[1], &sig, sizeof(int32_t));
+		if (res == -1 && errno == EAGAIN) {
+			goto try_again;
+		} else if (res != sizeof(int32_t)) {
+			qb_util_log(LOG_ERR, "failed to write signal to pipe [%d]",
+				    res);
+		}
 	}
 }
 
@@ -764,14 +775,49 @@ static void signal_dispatch_and_take_back(struct qb_loop_item *
item,
 struct qb_loop_source *
 qb_loop_signals_create(struct qb_loop *l)
 {
+	int32_t res = 0;
+	struct qb_poll_entry *pe;
 	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);
 
+	if (pipe_fds[0] < 0) {
+		res = pipe(pipe_fds);
+		if (res == -1) {
+			res = -errno;
+			qb_util_log(LOG_ERR,
+				    "Can't light pipe: %s",
+				    strerror(-res));
+			goto error_exit;
+		}
+		(void)qb_util_fd_nonblock_cloexec_set(pipe_fds[0]);
+		(void)qb_util_fd_nonblock_cloexec_set(pipe_fds[1]);
+
+		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,
+				    "Can't smoke pipe: %s",
+				    strerror(-res));
+			goto error_exit;
+		}
+	}
+
 	return (struct qb_loop_source *)s;
+
+ error_exit:
+	free(s);
+	errno = -res;
+	return NULL;
 }
 
 void qb_loop_signals_destroy(struct qb_loop *l)
@@ -860,9 +906,7 @@ int32_t qb_loop_signal_add(qb_loop_t *l,
 			   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;
@@ -882,22 +926,6 @@ int32_t qb_loop_signal_add(qb_loop_t *l,
 	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);
 	}
diff --git a/lib/util.c b/lib/util.c
index 0ab5f1a..5238d4b 100644
--- a/lib/util.c
+++ b/lib/util.c
@@ -353,3 +353,36 @@ int32_t qb_util_circular_mmap(int32_t fd, void **buf, size_t bytes)
 	*buf = addr_orig;
 	return (0);
 }
+
+int32_t qb_util_fd_nonblock_cloexec_set(int32_t fd)
+{
+	int32_t res;
+	char error_str[100];
+	int32_t oldflags = fcntl(fd, F_GETFD, 0);
+
+	if (oldflags < 0) {
+		oldflags = 0;
+	}
+	oldflags |= FD_CLOEXEC;
+	res = fcntl(fd, F_SETFD, oldflags);
+	if (res == -1) {
+		res = -errno;
+		strerror_r(errno, error_str, 100);
+		qb_util_log(LOG_CRIT,
+			    "Could not set close-on-exit operation on fd: %s\n",
+			    error_str);
+		return res;
+	}
+
+	res = fcntl(fd, F_SETFL, O_NONBLOCK);
+	if (res == -1) {
+		res = -errno;
+		strerror_r(errno, error_str, 100);
+		qb_util_log(LOG_CRIT,
+			    "Could not set non-blocking operation on fd: %s\n",
+			    error_str);
+	}
+	return res;
+}
+
+
diff --git a/lib/util_int.h b/lib/util_int.h
index 749937a..3e1f641 100644
--- a/lib/util_int.h
+++ b/lib/util_int.h
@@ -40,7 +40,7 @@ void _qb_util_log(const char *file_name,
  * @param file (in) the name of the file to be used.
  * @param bytes the size to truncate the file to.
  * @param file_flags same as passed into open()
- * @return 0 (success) or -1 (error)
+ * @return 0 (success) or -errno
  */
 int32_t qb_util_mmap_file_open(char *path, const char *file, size_t bytes,
 			       uint32_t file_flags);
@@ -51,8 +51,16 @@ int32_t qb_util_mmap_file_open(char *path, const char *file, size_t
bytes,
  * @param fd an open file to use to back the shared memory.
  * @param buf (out) the pointer to the start of the memory.
  * @param bytes the size of the shared memory.
- * @return 0 (success) or -1 (error)
+ * @return 0 (success) or -errno
  */
 int32_t qb_util_circular_mmap(int32_t fd, void **buf, size_t bytes);
 
+
+/**
+ * Set O_NONBLOCK and FD_CLOEXEC on a file descriptor.
+ * @param fd the file descriptor.
+ * @return 0 (success) or -errno
+ */
+int32_t qb_util_fd_nonblock_cloexec_set(int32_t fd);
+
 #endif /* QB_UTIL_INT_H_DEFINED */
-- 
1.7.3.1