Prevent res_timing_pthread from blocking callers

There were several reports of deadlock when using
res_timing_pthread. Backtraces indicated that one thread was blocked
waiting for the write to the pipe to complete and this thread held
the container lock for the timers.  Therefore any thread that wanted
to create a new timer or read an existing timer would block waiting
for either the timer lock or the container lock and deadlock ensued.

This patch changes the way the pipe is used to eliminate this source
of deadlocks:

1) The pipe is placed in non-blocking mode so that it would never
block even if the following changes someone fail...

2) Instead of writing bytes into the pipe for each "tick" that's
fired the pipe now has two states--signaled and unsignaled. If
signaled, the pipe is hot and any pollers of the read side
filedescriptor will be woken up. If unsigned the pipe is idle. This
eliminates even the chance of filling up the pipe and reduces the
potential overhead of calling unnecessary writes.

3) Since we're tracking the signaled / unsignaled state, we can
eliminate the exta poll system call for every firing because we know
that there is data to be read.

(closes issue ASTERISK-21389)
Reported by: Matt Jordan
Tested by: Shaun Ruffell, Matt Jordan, Tony Lewis
patches:
  0001-res_timing_pthread-Reduce-probability-of-deadlocking.patch uploaded by sruffell (License 5417)

(closes issue ASTERISK-19754)
Reported by: Nikola Ciprich

(closes issue ASTERISK-20577)
Reported by: Kien Kennedy

(closes issue ASTERISK-17436)
Reported by: Henry Fernandes

(closes issue ASTERISK-17467)
Reported by: isrl

(closes issue ASTERISK-17458)
Reported by: isrl

Review: https://reviewboard.asterisk.org/r/2441/
........

Merged revisions 386109 from http://svn.asterisk.org/svn/asterisk/branches/1.8
........

Merged revisions 386159 from http://svn.asterisk.org/svn/asterisk/branches/11


git-svn-id: https://origsvn.digium.com/svn/asterisk/trunk@386160 65c4cc65-6c06-0410-ace0-fbb531ad65f3

1 files changed, 58 insertions, 63 deletions

diff --git a/res/res_timing_pthread.c b/res/res_timing_pthread.c
index aa37885e2..55ed6acfb 100644
--- a/res/res_timing_pthread.c
+++ b/res/res_timing_pthread.c
@@ -31,8 +31,10 @@
 
 ASTERISK_FILE_VERSION(__FILE__, "$Revision$");
 
+#include <stdbool.h>
 #include <math.h>
-#include <sys/select.h>
+#include <unistd.h>
+#include <fcntl.h>
 
 #include "asterisk/module.h"
 #include "asterisk/timing.h"
@@ -40,7 +42,6 @@ ASTERISK_FILE_VERSION(__FILE__, "$Revision$");
 #include "asterisk/astobj2.h"
 #include "asterisk/time.h"
 #include "asterisk/lock.h"
-#include "asterisk/poll-compat.h"
 
 static void *timing_funcs_handle;
 
@@ -91,13 +92,15 @@ struct pthread_timer {
 	unsigned int tick_count;
 	unsigned int pending_ticks;
 	struct timeval start;
-	unsigned int continuous:1;
+	bool continuous:1;
+	bool pipe_signaled:1;
 };
 
 static void pthread_timer_destructor(void *obj);
 static struct pthread_timer *find_timer(int handle, int unlinkobj);
-static void write_byte(struct pthread_timer *timer);
-static int read_pipe(struct pthread_timer *timer, unsigned int num);
+static void signal_pipe(struct pthread_timer *timer);
+static void unsignal_pipe(struct pthread_timer *timer);
+static void ack_ticks(struct pthread_timer *timer, unsigned int num);
 
 /*!
  * \brief Data for the timing thread
@@ -113,6 +116,7 @@ static int pthread_timer_open(void)
 {
 	struct pthread_timer *timer;
 	int fd;
+	int i;
 
 	if (!(timer = ao2_alloc(sizeof(*timer), pthread_timer_destructor))) {
 		errno = ENOMEM;
@@ -127,6 +131,12 @@ static int pthread_timer_open(void)
 		return -1;
 	}
 
+	for (i = 0; i < ARRAY_LEN(timer->pipe); ++i) {
+		int flags = fcntl(timer->pipe[i], F_GETFL);
+		flags |= O_NONBLOCK;
+		fcntl(timer->pipe[i], F_SETFL, flags);
+	}
+	
 	ao2_lock(pthread_timers);
 	if (!ao2_container_count(pthread_timers)) {
 		ast_mutex_lock(&timing_thread.lock);
@@ -193,7 +203,6 @@ static int pthread_timer_set_rate(int handle, unsigned int rate)
 static int pthread_timer_ack(int handle, unsigned int quantity)
 {
 	struct pthread_timer *timer;
-	int res;
 
 	ast_assert(quantity > 0);
 
@@ -202,12 +211,12 @@ static int pthread_timer_ack(int handle, unsigned int quantity)
 	}
 
 	ao2_lock(timer);
-	res = read_pipe(timer, quantity);
+	ack_ticks(timer, quantity);
 	ao2_unlock(timer);
 
 	ao2_ref(timer, -1);
 
-	return res;
+	return 0;
 }
 
 static int pthread_timer_enable_continuous(int handle)
@@ -221,8 +230,8 @@ static int pthread_timer_enable_continuous(int handle)
 
 	ao2_lock(timer);
 	if (!timer->continuous) {
-		timer->continuous = 1;
-		write_byte(timer);
+		timer->continuous = true;
+		signal_pipe(timer);
 	}
 	ao2_unlock(timer);
 
@@ -242,13 +251,8 @@ static int pthread_timer_disable_continuous(int handle)
 
 	ao2_lock(timer);
 	if (timer->continuous) {
-		timer->continuous = 0;
-		if (read_pipe(timer, 1) != 0) {
-			/* Let the errno from read_pipe propagate up */
-			ao2_unlock(timer);
-			ao2_ref(timer, -1);
-			return -1;
-		}
+		timer->continuous = false;
+		unsignal_pipe(timer);
 	}
 	ao2_unlock(timer);
 
@@ -267,7 +271,7 @@ static enum ast_timer_event pthread_timer_get_event(int handle)
 	}
 
 	ao2_lock(timer);
-	if (timer->continuous && timer->pending_ticks == 1) {
+	if (timer->continuous) {
 		res = AST_TIMING_EVENT_CONTINUOUS;
 	}
 	ao2_unlock(timer);
@@ -366,79 +370,69 @@ static int check_timer(struct pthread_timer *timer)
 /*!
  * \internal
  * \pre timer is locked
- * \retval 0 if nothing to read or read success
- * \retval -1 on error
  */
-static int read_pipe(struct pthread_timer *timer, unsigned int quantity)
+static void ack_ticks(struct pthread_timer *timer, unsigned int quantity)
 {
-	int rd_fd = timer->pipe[PIPE_READ];
 	int pending_ticks = timer->pending_ticks;
 
 	ast_assert(quantity);
 
-	if (timer->continuous && pending_ticks) {
-		pending_ticks--;
-	}
-
 	if (quantity > pending_ticks) {
 		quantity = pending_ticks;
 	}
 
 	if (!quantity) {
-		return 0;
+		return;
 	}
 
-	do {
-		unsigned char buf[1024];
-		ssize_t res;
-		struct pollfd pfd = {
-			.fd = rd_fd,
-			.events = POLLIN,
-		};
-
-		if (ast_poll(&pfd, 1, 0) != 1) {
-			ast_debug(1, "Reading not available on timing pipe, "
-					"quantity: %u\n", quantity);
-			return -1;
-		}
+	timer->pending_ticks -= quantity;
 
-		res = read(rd_fd, buf,
-			(quantity < sizeof(buf)) ? quantity : sizeof(buf));
+	if ((0 == timer->pending_ticks) && !timer->continuous) {
+		unsignal_pipe(timer);
+	}
+}
 
-		if (res == -1) {
-			if (errno == EAGAIN) {
-				continue;
-			}
-			ast_log(LOG_ERROR, "read failed on timing pipe: %s\n",
-					strerror(errno));
-			return -1;
-		}
+/*!
+ * \internal
+ * \pre timer is locked
+ */
+static void signal_pipe(struct pthread_timer *timer)
+{
+	ssize_t res;
+	unsigned char x = 42;
 
-		quantity -= res;
-		timer->pending_ticks -= res;
-	} while (quantity);
+	if (timer->pipe_signaled) {
+		return;
+	}
 
-	return 0;
+	res = write(timer->pipe[PIPE_WRITE], &x, 1);
+	if (-1 == res) {
+		ast_log(LOG_ERROR, "Error writing to timing pipe: %s\n",
+				strerror(errno));
+	} else {
+		timer->pipe_signaled = true;
+	}
 }
 
 /*!
  * \internal
  * \pre timer is locked
  */
-static void write_byte(struct pthread_timer *timer)
+static void unsignal_pipe(struct pthread_timer *timer)
 {
 	ssize_t res;
-	unsigned char x = 42;
+	unsigned long buffer;
 
-	do {
-		res = write(timer->pipe[PIPE_WRITE], &x, 1);
-	} while (res == -1 && errno == EAGAIN);
+	if (!timer->pipe_signaled) {
+		return;
+	}
 
-	if (res == -1) {
-		ast_log(LOG_ERROR, "Error writing to timing pipe: %s\n",
+	res = read(timer->pipe[PIPE_READ], &buffer, sizeof(buffer));
+	if (-1 == res) {
+		ast_log(LOG_ERROR, "Error reading from pipe: %s\n",
 				strerror(errno));
 	} else {
-		timer->pending_ticks++;
+		timer->pipe_signaled = false;
 	}
 }
 
@@ -452,7 +446,8 @@ static int run_timer(void *obj, void *arg, int flags)
 
 	ao2_lock(timer);
 	if (check_timer(timer)) {
-		write_byte(timer);
+		timer->pending_ticks++;
+		signal_pipe(timer);
 	}
 	ao2_unlock(timer);