freetdm: remove busy-waiting and use ftdm interrupt to wait for state change completion

libs/freetdm/src/ftdm_state.c

@@ -101,8 +101,10 @@ FT_DECLARE(ftdm_status_t) _ftdm_channel_complete_state(const char *file, const c
 			ftdm_channel_state2str(fchan->last_state), ftdm_channel_state2str(state), diff);
 	
 
-	/* FIXME: broadcast condition to wake up anyone waiting on state completion if the channel 
-	 * is blocking (FTDM_CHANNEL_NONBLOCK is not set) */
+	if (ftdm_test_flag(fchan, FTDM_CHANNEL_BLOCKING)) {
+		ftdm_clear_flag(fchan, FTDM_CHANNEL_BLOCKING);
+		ftdm_interrupt_signal(fchan->state_change_notify);
+	}
 
 	return FTDM_SUCCESS;
 }
@@ -169,6 +171,7 @@ static int ftdm_parse_state_map(ftdm_channel_t *ftdmchan, ftdm_channel_state_t s
 #define DEFAULT_WAIT_TIME 1000
 FT_DECLARE(ftdm_status_t) ftdm_channel_set_state(const char *file, const char *func, int line, ftdm_channel_t *ftdmchan, ftdm_channel_state_t state, int waitrq)
 {
+	ftdm_status_t status;
 	int ok = 1;
 	int waitms = DEFAULT_WAIT_TIME;	
 
@@ -191,6 +194,16 @@ FT_DECLARE(ftdm_status_t) ftdm_channel_set_state(const char *file, const char *f
 		return FTDM_FAIL;
 	}
 
+	if (!ftdmchan->state_change_notify) {
+		status = ftdm_interrupt_create(&ftdmchan->state_change_notify, FTDM_INVALID_SOCKET);
+		if (status != FTDM_SUCCESS) {
+			ftdm_log_chan_ex(ftdmchan, file, func, line, FTDM_LOG_LEVEL_CRIT, 
+					"Failed to create state change interrupt when moving from %s to %s\n", ftdm_channel_state2str(ftdmchan->state), ftdm_channel_state2str(state));
+			return status;
+		}
+	}
+
+
 	if (ftdmchan->span->state_map) {
 		ok = ftdm_parse_state_map(ftdmchan, state, ftdmchan->span->state_map);
 		goto end;
@@ -276,67 +289,54 @@ FT_DECLARE(ftdm_status_t) ftdm_channel_set_state(const char *file, const char *f
 
 end:
 
-	if (ok) {
-		ftdm_log_chan_ex(ftdmchan, file, func, line, FTDM_LOG_LEVEL_DEBUG, "Changed state from %s to %s\n", ftdm_channel_state2str(ftdmchan->state), ftdm_channel_state2str(state));
-		ftdmchan->last_state = ftdmchan->state; 
-		ftdmchan->state = state;
-		ftdmchan->state_status = FTDM_STATE_STATUS_NEW;
-		ftdmchan->history[ftdmchan->hindex].file = file;
-		ftdmchan->history[ftdmchan->hindex].func = func;
-		ftdmchan->history[ftdmchan->hindex].line = line;
-		ftdmchan->history[ftdmchan->hindex].state = ftdmchan->state;
-		ftdmchan->history[ftdmchan->hindex].last_state = ftdmchan->last_state;
-		ftdmchan->history[ftdmchan->hindex].time = ftdm_current_time_in_ms();
-		ftdmchan->history[ftdmchan->hindex].end_time = 0;
-		ftdmchan->hindex++;
-		if (ftdmchan->hindex == ftdm_array_len(ftdmchan->history)) {
-			ftdmchan->hindex = 0;
-		}
-		ftdm_set_flag(ftdmchan, FTDM_CHANNEL_STATE_CHANGE);
-
-		ftdm_mutex_lock(ftdmchan->span->mutex);
-		ftdm_set_flag(ftdmchan->span, FTDM_SPAN_STATE_CHANGE);
-		if (ftdmchan->span->pendingchans) {
-			ftdm_queue_enqueue(ftdmchan->span->pendingchans, ftdmchan);
-		}
-		ftdm_mutex_unlock(ftdmchan->span->mutex);
-	} else {
+	if (!ok) {
 		ftdm_log_chan_ex(ftdmchan, file, func, line, FTDM_LOG_LEVEL_WARNING, "VETO state change from %s to %s\n", ftdm_channel_state2str(ftdmchan->state), ftdm_channel_state2str(state));
 		goto done;
 	}
 
+	ftdm_log_chan_ex(ftdmchan, file, func, line, FTDM_LOG_LEVEL_DEBUG, "Changed state from %s to %s\n", ftdm_channel_state2str(ftdmchan->state), ftdm_channel_state2str(state));
+	ftdmchan->last_state = ftdmchan->state; 
+	ftdmchan->state = state;
+	ftdmchan->state_status = FTDM_STATE_STATUS_NEW;
+	ftdmchan->history[ftdmchan->hindex].file = file;
+	ftdmchan->history[ftdmchan->hindex].func = func;
+	ftdmchan->history[ftdmchan->hindex].line = line;
+	ftdmchan->history[ftdmchan->hindex].state = ftdmchan->state;
+	ftdmchan->history[ftdmchan->hindex].last_state = ftdmchan->last_state;
+	ftdmchan->history[ftdmchan->hindex].time = ftdm_current_time_in_ms();
+	ftdmchan->history[ftdmchan->hindex].end_time = 0;
+	ftdmchan->hindex++;
+	if (ftdmchan->hindex == ftdm_array_len(ftdmchan->history)) {
+		ftdmchan->hindex = 0;
+	}
+	ftdm_set_flag(ftdmchan, FTDM_CHANNEL_STATE_CHANGE);
+
+	ftdm_mutex_lock(ftdmchan->span->mutex);
+	ftdm_set_flag(ftdmchan->span, FTDM_SPAN_STATE_CHANGE);
+	if (ftdmchan->span->pendingchans) {
+		ftdm_queue_enqueue(ftdmchan->span->pendingchans, ftdmchan);
+	}
+	ftdm_mutex_unlock(ftdmchan->span->mutex);
+
 	if (ftdm_test_flag(ftdmchan, FTDM_CHANNEL_NONBLOCK)) {
 		/* the channel should not block waiting for state processing */
 		goto done;
 	}
 
-	/* there is an inherent race here between set and check of the change flag but we do not care because
-	 * the flag should never last raised for more than a few ms for any state change */
-	while (waitrq && waitms > 0) {
-		/* give a chance to the signaling stack to process it */
-		ftdm_mutex_unlock(ftdmchan->mutex);
+	ftdm_set_flag(ftdmchan, FTDM_CHANNEL_BLOCKING);
 
-		ftdm_sleep(10);
-		waitms -= 10;
+	ftdm_mutex_unlock(ftdmchan->mutex);
 
-		ftdm_mutex_lock(ftdmchan->mutex);
-		
-		/* if the flag is no longer set, the state change was processed (or is being processed) */
-		if (!ftdm_test_flag(ftdmchan, FTDM_CHANNEL_STATE_CHANGE)) {
-			break;
-		}
+	status = ftdm_interrupt_wait(ftdmchan->state_change_notify, waitms);
 
-		/* if the state is no longer what we set, the state change was 
-		 * obviously processed (and the current state change flag is for other state change) */
-		if (ftdmchan->state != state) {
-			break;
-		}
-	}
+	ftdm_mutex_lock(ftdmchan->mutex);
 
-	if (waitms <= 0) {
-		ftdm_log_chan_ex(ftdmchan, file, func, line, FTDM_LOG_LEVEL_WARNING, "state change from %s to %s was most likely not processed after aprox %dms\n",
+	if (status != FTDM_SUCCESS) {
+		ftdm_log_chan_ex(ftdmchan, file, func, line, 
+				FTDM_LOG_LEVEL_WARNING, "state change from %s to %s was most likely not processed after aprox %dms\n",
 				ftdm_channel_state2str(ftdmchan->last_state), ftdm_channel_state2str(state), DEFAULT_WAIT_TIME);
 		ok = 0;
+		goto done;
 	}
 done:
 	return ok ? FTDM_SUCCESS : FTDM_FAIL;

libs/freetdm/src/ftdm_threadmutex.c

@@ -56,7 +56,11 @@ struct ftdm_interrupt {
 	/* for generic interruption */
 	HANDLE event;
 #else
-	/* for generic interruption */
+	/* In theory we could be using thread conditions for generic interruption,
+	 * however, Linux does not have a primitive like Windows WaitForMultipleObjects
+	 * to wait for both thread condition and file descriptors, therefore we decided
+	 * to use a dummy pipe for generic interruption/condition logic
+	 * */
 	int readfd;
 	int writefd;
 #endif
@@ -243,6 +247,7 @@ FT_DECLARE(ftdm_status_t) _ftdm_mutex_unlock(ftdm_mutex_t *mutex)
 
 FT_DECLARE(ftdm_status_t) ftdm_interrupt_create(ftdm_interrupt_t **ininterrupt, ftdm_socket_t device)
 {
+	ftdm_status_t status = FTDM_SUCCESS;
 	ftdm_interrupt_t *interrupt = NULL;
 #ifndef WIN32
 	int fds[2];
@@ -253,7 +258,7 @@ FT_DECLARE(ftdm_status_t) ftdm_interrupt_create(ftdm_interrupt_t **ininterrupt,
 	interrupt = ftdm_calloc(1, sizeof(*interrupt));
 	if (!interrupt) {
 		ftdm_log(FTDM_LOG_ERROR, "Failed to allocate interrupt memory\n");
-		return FTDM_FAIL;
+		return FTDM_ENOMEM;
 	}
 
 	interrupt->device = device;
@@ -261,11 +266,13 @@ FT_DECLARE(ftdm_status_t) ftdm_interrupt_create(ftdm_interrupt_t **ininterrupt,
 	interrupt->event = CreateEvent(NULL, FALSE, FALSE, NULL);
 	if (!interrupt->event) {
 		ftdm_log(FTDM_LOG_ERROR, "Failed to allocate interrupt event\n");
+		status = FTDM_ENOMEM;
 		goto failed;
 	}
 #else
 	if (pipe(fds)) {
 		ftdm_log(FTDM_LOG_ERROR, "Failed to allocate interrupt pipe: %s\n", strerror(errno));
+		status = FTDM_FAIL;
 		goto failed;
 	}
 	interrupt->readfd = fds[0];
@@ -287,7 +294,7 @@ failed:
 #endif
 		ftdm_safe_free(interrupt);
 	}
-	return FTDM_FAIL;
+	return status;
 }
 
 #define ONE_BILLION 1000000000

libs/freetdm/src/include/private/ftdm_core.h

@@ -462,6 +462,7 @@ struct ftdm_channel {
 	ftdm_dtmf_debug_t dtmfdbg;
 	ftdm_io_dump_t rxdump;
 	ftdm_io_dump_t txdump;
+	ftdm_interrupt_t *state_change_notify; /*!< Notify when a state change is terminated */
 	int32_t txdrops;
 	int32_t rxdrops;
 };

libs/freetdm/src/include/private/ftdm_types.h

@@ -250,6 +250,8 @@ typedef enum {
 #define FTDM_CHANNEL_NONBLOCK        (1ULL << 33)
 /*!< There is a pending acknowledge for an indication */
 #define FTDM_CHANNEL_IND_ACK_PENDING (1ULL << 34)
+/*!< There is someone blocking in the channel waiting for state completion */
+#define FTDM_CHANNEL_BLOCKING        (1ULL << 35)
 
 #include "ftdm_state.h"