skypiax: no more pipes for audio samples exchange between threads. Let's hope it…

skypiax: no more pipes for audio samples exchange between threads. Let's hope it does not breaks things around. Only other possible cause of audio delay is socket buffering. We'll check that too. 

git-svn-id: http://svn.freeswitch.org/svn/freeswitch/trunk@15541 d0543943-73ff-0310-b7d9-9358b9ac24b2

src/mod/endpoints/mod_skypiax/mod_skypiax.c

@@ -741,10 +741,22 @@ static switch_status_t channel_write_frame(switch_core_session_t *session, switc
 
 	sent = frame->datalen;
 #ifdef WIN32
-	switch_file_write(tech_pvt->audioskypepipe[1], frame->data, &sent);
+	//switch_file_write(tech_pvt->audiopipe_cli[1], frame->data, &sent);
 #else /* WIN32 */
-	sent = write(tech_pvt->audioskypepipe[1], frame->data, sent);
+	//sent = write(tech_pvt->audiopipe_cli[1], frame->data, sent);
 #endif /* WIN32 */
+					while(tech_pvt->flag_audio_cli == 1){
+						switch_sleep(100); //1 millisec
+						//NOTICA("write \n", SKYPIAX_P_LOG);
+						}
+						//WARNINGA("write \n", SKYPIAX_P_LOG);
+
+
+	//memcpy(tech_pvt->audiobuf_cli, frame->data, frame->datalen);
+	memcpy(tech_pvt->audiobuf_cli, frame->data, frame->datalen);
+	tech_pvt->flag_audio_cli = 1;
+	//NOTICA("write \n", SKYPIAX_P_LOG);
+
 	if (sent != frame->datalen && sent != -1) {
 		DEBUGA_SKYPE("CLI PIPE write %d\n", SKYPIAX_P_LOG, sent);
 	}
@@ -1420,7 +1432,7 @@ static switch_status_t load_config(int reload_type)
 static switch_status_t chat_send(const char *proto, const char *from, const char *to, const char *subject, const char *body, const char *type, const char *hint)
 {
 	//char *user, *host, *f_user = NULL, *ffrom = NULL, *f_host = NULL, *f_resource = NULL;
-	char *user, *host, *f_user = NULL, *f_host = NULL, *f_resource = NULL;
+	char *user=NULL, *host, *f_user = NULL, *f_host = NULL, *f_resource = NULL;
 	//mdl_profile_t *profile = NULL;
 	private_t * tech_pvt=NULL;
 	int i=0, found=0, tried=0;
@@ -1629,14 +1641,14 @@ SWITCH_MODULE_SHUTDOWN_FUNCTION(mod_skypiax_shutdown)
 			}
 #ifndef WIN32
 			WARNINGA("SHUTDOWN interface_id=%d\n", SKYPIAX_P_LOG, interface_id);
-			shutdown(tech_pvt->audioskypepipe[0], 2);
-			close(tech_pvt->audioskypepipe[0]);
-			shutdown(tech_pvt->audioskypepipe[1], 2);
-			close(tech_pvt->audioskypepipe[1]);
-			shutdown(tech_pvt->audiopipe[0], 2);
-			close(tech_pvt->audiopipe[0]);
-			shutdown(tech_pvt->audiopipe[1], 2);
-			close(tech_pvt->audiopipe[1]);
+			shutdown(tech_pvt->audiopipe_cli[0], 2);
+			close(tech_pvt->audiopipe_cli[0]);
+			shutdown(tech_pvt->audiopipe_cli[1], 2);
+			close(tech_pvt->audiopipe_cli[1]);
+			shutdown(tech_pvt->audiopipe_srv[0], 2);
+			close(tech_pvt->audiopipe_srv[0]);
+			shutdown(tech_pvt->audiopipe_srv[1], 2);
+			close(tech_pvt->audiopipe_srv[1]);
 			shutdown(tech_pvt->SkypiaxHandles.fdesc[0], 2);
 			close(tech_pvt->SkypiaxHandles.fdesc[0]);
 			shutdown(tech_pvt->SkypiaxHandles.fdesc[1], 2);

src/mod/endpoints/mod_skypiax/skypiax.h

@@ -229,12 +229,12 @@ struct private_object {
 	int skype;					/*!< \brief config flag, bool, Skype support on this interface (0 if false, -1 if true) */
 	int control_to_send;
 #ifdef WIN32
-	switch_file_t *audiopipe[2];
-	switch_file_t *audioskypepipe[2];
+	switch_file_t *audiopipe_srv[2];
+	switch_file_t *audiopipe_cli[2];
 	switch_file_t *skypiax_sound_capt_fd;	/*!< \brief file descriptor for sound capture dev */
 #else							/* WIN32 */
-	int audiopipe[2];
-	int audioskypepipe[2];
+	int audiopipe_srv[2];
+	int audiopipe_cli[2];
 	int skypiax_sound_capt_fd;	/*!< \brief file descriptor for sound capture dev */
 #endif							/* WIN32 */
 	switch_thread_t *tcp_srv_thread;
@@ -243,6 +243,12 @@ struct private_object {
 	switch_thread_t *skypiax_api_thread;
 	short audiobuf[SAMPLES_PER_FRAME];
 	int audiobuf_is_loaded;
+	short audiobuf_cli[SAMPLES_PER_FRAME];
+	switch_mutex_t *mutex_audio_cli;
+	int flag_audio_cli;
+	short audiobuf_srv[SAMPLES_PER_FRAME];
+	switch_mutex_t *mutex_audio_srv;
+	int flag_audio_srv;
 
 	//int phonebook_listing;
 	//int phonebook_querying;

src/mod/endpoints/mod_skypiax/skypiax_protocol.c

@@ -786,7 +786,18 @@ void *skypiax_do_tcp_srv_thread_func(void *obj)
 									i++;
 								}
 								/* send the complete frame through the pipe to our code waiting for incoming audio */
-								howmany = skypiax_pipe_write(tech_pvt->audiopipe[1], totalbuf, SAMPLES_PER_FRAME * sizeof(short));
+								//howmany = skypiax_pipe_write(tech_pvt->audiopipe_srv[1], totalbuf, SAMPLES_PER_FRAME * sizeof(short));
+					while(tech_pvt->flag_audio_srv == 1){
+						switch_sleep(100); //1 millisec
+						//NOTICA("write \n", SKYPIAX_P_LOG);
+						}
+						//WARNINGA("read \n", SKYPIAX_P_LOG);
+
+
+								howmany = SAMPLES_PER_FRAME * sizeof(short);
+								memcpy(tech_pvt->audiobuf_srv, totalbuf, SAMPLES_PER_FRAME * sizeof(short));
+								tech_pvt->flag_audio_srv = 1;
+								//NOTICA("read \n", SKYPIAX_P_LOG);
 								if (howmany != SAMPLES_PER_FRAME * sizeof(short)) {
 									ERRORA("howmany is %d, but was expected to be %d\n", SKYPIAX_P_LOG,
 										   howmany, (int) (SAMPLES_PER_FRAME * sizeof(short)));
@@ -811,15 +822,24 @@ void *skypiax_do_tcp_srv_thread_func(void *obj)
 
 				/* let's send some frame in the pipes, so both tcp_cli and tcp_srv will have an occasion to die */
 				kill_cli_size = SAMPLES_PER_FRAME * sizeof(short);
-				len = skypiax_pipe_write(tech_pvt->audiopipe[1], kill_cli_buff, kill_cli_size);
+				len = skypiax_pipe_write(tech_pvt->audiopipe_srv[1], kill_cli_buff, kill_cli_size);
 				kill_cli_size = SAMPLES_PER_FRAME * sizeof(short);
-				len = skypiax_pipe_write(tech_pvt->audioskypepipe[1], kill_cli_buff, kill_cli_size);
-				tech_pvt->interface_state = SKYPIAX_STATE_DOWN;
+				len = skypiax_pipe_write(tech_pvt->audiopipe_cli[1], kill_cli_buff, kill_cli_size);
+				//tech_pvt->interface_state = SKYPIAX_STATE_DOWN;
 				kill_cli_size = SAMPLES_PER_FRAME * sizeof(short);
-				len = skypiax_pipe_write(tech_pvt->audiopipe[1], kill_cli_buff, kill_cli_size);
+				len = skypiax_pipe_write(tech_pvt->audiopipe_srv[1], kill_cli_buff, kill_cli_size);
 				kill_cli_size = SAMPLES_PER_FRAME * sizeof(short);
-				len = skypiax_pipe_write(tech_pvt->audioskypepipe[1], kill_cli_buff, kill_cli_size);
+				len = skypiax_pipe_write(tech_pvt->audiopipe_cli[1], kill_cli_buff, kill_cli_size);
 
+				tech_pvt->flag_audio_cli = 1; //let's send some frame in the pipes, so both tcp_cli and tcp_srv will have an occasion to die
+				skypiax_sleep(200);
+				tech_pvt->flag_audio_srv = 1; //let's send some frame in the pipes, so both tcp_cli and tcp_srv will have an occasion to die
+				skypiax_sleep(200);
+				tech_pvt->interface_state = SKYPIAX_STATE_DOWN;
+				tech_pvt->flag_audio_cli = 1; //let's send some frame in the pipes, so both tcp_cli and tcp_srv will have an occasion to die
+				skypiax_sleep(200);
+				tech_pvt->flag_audio_srv = 1; //let's send some frame in the pipes, so both tcp_cli and tcp_srv will have an occasion to die
+				skypiax_sleep(200);
 				DEBUGA_SKYPE("Skype incoming audio GONE\n", SKYPIAX_P_LOG);
 				skypiax_close_socket(fd);
 			}
@@ -893,8 +913,8 @@ void *skypiax_do_tcp_cli_thread_func(void *obj)
 			while (s > 0 && (fd = accept(s, (struct sockaddr *) &remote_addr, &sin_size)) > 0) {
 				DEBUGA_SKYPE("ACCEPTED here you send me %d\n", SKYPIAX_P_LOG, tech_pvt->tcp_cli_port);
 #ifndef WIN32
-				fcntl(tech_pvt->audioskypepipe[0], F_SETFL, O_NONBLOCK);
-				fcntl(tech_pvt->audioskypepipe[1], F_SETFL, O_NONBLOCK);
+				fcntl(tech_pvt->audiopipe_cli[0], F_SETFL, O_NONBLOCK);
+				fcntl(tech_pvt->audiopipe_cli[1], F_SETFL, O_NONBLOCK);
 #endif //WIN32
 
 				if (!(running && tech_pvt->running))
@@ -918,22 +938,43 @@ void *skypiax_do_tcp_cli_thread_func(void *obj)
 					fdselect = fd;
 					FD_SET(fdselect, &fs);
 
-					rt = select(fdselect + 1, NULL, &fs, NULL, &to);
+					//rt = select(fdselect + 1, NULL, &fs, NULL, &to);
 #else
 /* on *unix and cygwin we select from the real pipe */
-					fdselect = tech_pvt->audioskypepipe[0];
-					FD_SET(fdselect, &fs);
+					//XXX fdselect = tech_pvt->audiopipe_cli[0];
+					//XXX FD_SET(fdselect, &fs);
 
-					rt = select(fdselect + 1, &fs, NULL, NULL, &to);
+					//rt = select(fdselect + 1, &fs, NULL, NULL, &to);
 #endif
+					fdselect = fd;
+					FD_SET(fdselect, &fs);
+
+					//XXX rt = select(fdselect + 1, NULL, &fs, NULL, &to);
+					while(tech_pvt->flag_audio_cli == 0){
+						skypiax_sleep(100); //1 millisec
+						//NOTICA("write \n", SKYPIAX_P_LOG);
+						}
+						//ERRORA("write \n", SKYPIAX_P_LOG);
+
+					rt = 1;
 
 					if (rt > 0) {
 						int counter;
 
 						/* until we drained the pipe to empty */
-						for (counter = 0; counter < 10; counter++) {
+						//for (counter = 0; counter < 10; counter++) {
+						for (counter = 0; counter < 1; counter++) {
 							/* read from the pipe the audio frame we are supposed to send out */
-							got = skypiax_pipe_read(tech_pvt->audioskypepipe[0], cli_in, SAMPLES_PER_FRAME * sizeof(short));
+							//got = skypiax_pipe_read(tech_pvt->audiopipe_cli[0], cli_in, SAMPLES_PER_FRAME * sizeof(short));
+
+
+							got = SAMPLES_PER_FRAME * sizeof(short);
+							memcpy(cli_in, tech_pvt->audiobuf_cli, SAMPLES_PER_FRAME * sizeof(short));
+							tech_pvt->flag_audio_cli = 0;
+
+
+
+
 							if (got == -1)
 								break;
 
@@ -1015,7 +1056,15 @@ int skypiax_audio_read(private_t * tech_pvt)
 {
 	unsigned int samples;
 
-	samples = skypiax_pipe_read(tech_pvt->audiopipe[0], tech_pvt->read_frame.data, SAMPLES_PER_FRAME * sizeof(short));
+	while(tech_pvt->flag_audio_srv == 0){
+		skypiax_sleep(100); //1 millisec
+		//NOTICA("read \n", SKYPIAX_P_LOG);
+		}
+		//ERRORA("read \n", SKYPIAX_P_LOG);
+	//samples = skypiax_pipe_read(tech_pvt->audiopipe_srv[0], tech_pvt->read_frame.data, SAMPLES_PER_FRAME * sizeof(short));
+	samples = SAMPLES_PER_FRAME * sizeof(short);
+	memcpy(tech_pvt->read_frame.data, tech_pvt->audiobuf_srv, SAMPLES_PER_FRAME * sizeof(short));
+	tech_pvt->flag_audio_srv = 0;
 
 	if (samples != SAMPLES_PER_FRAME * sizeof(short)) {
 		if (samples)
@@ -1100,8 +1149,8 @@ int skypiax_close_socket(unsigned int fd)
 int skypiax_audio_init(private_t * tech_pvt)
 {
 	switch_status_t rv;
-	rv = switch_file_pipe_create(&tech_pvt->audiopipe[0], &tech_pvt->audiopipe[1], skypiax_module_pool);
-	rv = switch_file_pipe_create(&tech_pvt->audioskypepipe[0], &tech_pvt->audioskypepipe[1], skypiax_module_pool);
+	rv = switch_file_pipe_create(&tech_pvt->audiopipe_srv[0], &tech_pvt->audiopipe_srv[1], skypiax_module_pool);
+	rv = switch_file_pipe_create(&tech_pvt->audiopipe_cli[0], &tech_pvt->audiopipe_cli[1], skypiax_module_pool);
 	return 0;
 }
 #else /* WIN32 */
@@ -1132,17 +1181,17 @@ int skypiax_close_socket(unsigned int fd)
 
 int skypiax_audio_init(private_t * tech_pvt)
 {
-	if (pipe(tech_pvt->audiopipe)) {
-		fcntl(tech_pvt->audiopipe[0], F_SETFL, O_NONBLOCK);
-		fcntl(tech_pvt->audiopipe[1], F_SETFL, O_NONBLOCK);
+	if (pipe(tech_pvt->audiopipe_srv)) {
+		fcntl(tech_pvt->audiopipe_srv[0], F_SETFL, O_NONBLOCK);
+		fcntl(tech_pvt->audiopipe_srv[1], F_SETFL, O_NONBLOCK);
 	}
-	if (pipe(tech_pvt->audioskypepipe)) {
-		fcntl(tech_pvt->audioskypepipe[0], F_SETFL, O_NONBLOCK);
-		fcntl(tech_pvt->audioskypepipe[1], F_SETFL, O_NONBLOCK);
+	if (pipe(tech_pvt->audiopipe_cli)) {
+		fcntl(tech_pvt->audiopipe_cli[0], F_SETFL, O_NONBLOCK);
+		fcntl(tech_pvt->audiopipe_cli[1], F_SETFL, O_NONBLOCK);
 	}
 
 /* this pipe is the audio fd for asterisk to poll on during a call. FS do not use it */
-	tech_pvt->skypiax_sound_capt_fd = tech_pvt->audiopipe[0];
+	tech_pvt->skypiax_sound_capt_fd = tech_pvt->audiopipe_srv[0];
 
 	return 0;
 }