Linux DVB output for STBs based on Broadcom - replace active pooling by select (ready to write)

Signed-off-by: max_10 <max_10@gmx.de>

libeplayer3-arm/include/playback.h

@@ -2,6 +2,10 @@
 #define PLAYBACK_H_
 #include <sys/types.h>
 #include <stdint.h>
+#include <stdbool.h>
+
+typedef void( * PlaybackDieNowCallback )();
+bool PlaybackDieNowRegisterCallback(PlaybackDieNowCallback callback);
 
 typedef enum {
 	PLAYBACK_OPEN,

libeplayer3-arm/include/writer.h

@@ -4,6 +4,7 @@
 #include <sys/uio.h>
 #include <stdio.h>
 #include <stdint.h>
+#include "common.h"
 
 typedef enum { eNone, eAudio, eVideo} eWriterType_t;
 
@@ -87,5 +88,6 @@ Writer_t *getDefaultVideoWriter();
 Writer_t *getDefaultAudioWriter();
 ssize_t write_with_retry(int fd, const void *buf, size_t size);
 ssize_t writev_with_retry(int fd, const struct iovec *iov, size_t ic);
+ssize_t WriteWithRetry(Context_t *context, int pipefd, int fd, const void *buf, size_t size);
 
 #endif

libeplayer3-arm/main/exteplayer.c

@@ -84,10 +84,15 @@ static void TerminateAllSockets(void)
 	}
 }
 
-static int g_pfd[2] = {-1, -1}; /* Used to wake terminate thread */
+static int g_pfd[2] = {-1, -1}; /* Used to wake terminate thread and kbhit */
 static int isPlaybackStarted = 0;
 static pthread_mutex_t playbackStartMtx;
 
+static void TerminateWakeUp()
+{
+	write(g_pfd[1], "x", 1);
+}
+
 static void *TermThreadFun(void *arg __attribute__((unused)))
 {
 	const char *socket_path = "/tmp/iptvplayer_extplayer_term_fd";
@@ -126,13 +131,6 @@ static void *TermThreadFun(void *arg __attribute__((unused)))
 	}
 	if (FD_ISSET(fd, &readfds))
 	{
-		/*
-		if ((cl = accept(fd, NULL, NULL)) == -1)
-		{
-		    perror("TermThreadFun accept error");
-		    goto finish;
-		}
-		*/
 		pthread_mutex_lock(&playbackStartMtx);
 		PlaybackDieNow(1);
 		if (isPlaybackStarted)
@@ -171,16 +169,17 @@ static void map_inter_file_path(char *filename)
 static int kbhit(void)
 {
 	struct timeval tv;
-	fd_set read_fd;
+	fd_set readfds;
 	tv.tv_sec = 1;
 	tv.tv_usec = 0;
-	FD_ZERO(&read_fd);
-	FD_SET(0, &read_fd);
-	if (-1 == select(1, &read_fd, NULL, NULL, &tv))
+	FD_ZERO(&readfds);
+	FD_SET(0,&readfds);
+	FD_SET(g_pfd[0], &readfds);
+	if(-1 == select(g_pfd[0] + 1, &readfds, NULL, NULL, &tv))
 	{
 		return 0;
 	}
-	if (FD_ISSET(0, &read_fd))
+	if (FD_ISSET(0, &readfds))
 	{
 		return 1;
 	}
@@ -639,7 +638,7 @@ int main(int argc, char *argv[])
 	memset(argvBuff, '\0', sizeof(argvBuff));
 	int commandRetVal = -1;
 	/* inform client that we can handle additional commands */
-	fprintf(stderr, "{\"EPLAYER3_EXTENDED\":{\"version\":%d}}\n", 41);
+	fprintf(stderr, "{\"EPLAYER3_EXTENDED\":{\"version\":%d}}\n", 42);
 	if (0 != ParseParams(argc, argv, file, audioFile, &audioTrackIdx, &subtitleTrackIdx, &linuxDvbBufferSizeMB))
 	{
 		printf("Usage: exteplayer3 filePath [-u user-agent] [-c cookies] [-h headers] [-p prio] [-a] [-d] [-w] [-l] [-s] [-i] [-t audioTrackId] [-9 subtitleTrackId] [-x separateAudioUri] plabackUri\n");
@@ -745,6 +744,7 @@ int main(int argc, char *argv[])
 		fprintf(stderr, "{\"PLAYBACK_PLAY\":{\"sts\":%d}}\n", commandRetVal);
 		if (g_player->playback->isPlaying)
 		{
+			PlaybackDieNowRegisterCallback(TerminateWakeUp);
 			HandleTracks(g_player->manager->video, (PlaybackCmd_t) - 1, "vc");
 			HandleTracks(g_player->manager->audio, (PlaybackCmd_t) - 1, "al");
 			if (audioTrackIdx >= 0)
@@ -763,7 +763,7 @@ int main(int argc, char *argv[])
 			}
 			HandleTracks(g_player->manager->subtitle, (PlaybackCmd_t) - 1, "sc");
 		}
-		while (g_player->playback->isPlaying)
+		while (g_player->playback->isPlaying && 0 == PlaybackDieNow(0))
 		{
 			/* we made fgets non blocking */
 			if (NULL == fgets(argvBuff, sizeof(argvBuff) - 1, stdin))
@@ -994,6 +994,5 @@ int main(int argc, char *argv[])
 	pthread_mutex_destroy(&playbackStartMtx);
 	close(g_pfd[0]);
 	close(g_pfd[1]);
-	//printOutputCapabilities();
 	exit(0);
 }

libeplayer3-arm/output/linuxdvb_buffering.c

@@ -25,6 +25,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <unistd.h>
+#include <fcntl.h>
 #include <memory.h>
 #include <string.h>
 #include <errno.h>
@@ -98,6 +99,7 @@ static uint32_t bufferingDataSize = 0;
 
 static int videofd = -1;
 static int audiofd = -1;
+static int g_pfd[2] = {-1, -1};
 
 /* ***************************** */
 /* Prototypes                    */
@@ -106,14 +108,46 @@ static int audiofd = -1;
 /* ***************************** */
 /* MISC Functions                */
 /* ***************************** */
+static void WriteWakeUp()
+{
+	write(g_pfd[1], "x", 1);
+}
 
 /* **************************** */
 /* Worker Thread                */
 /* **************************** */
 static void LinuxDvbBuffThread(Context_t *context) 
 {
+	int flags = 0;
 	static BufferingNode_t *nodePtr = NULL;
 	buff_printf(20, "ENTER\n");
+	
+	if (pipe(g_pfd) == -1)
+	{
+		buff_err("critical error\n");
+	}
+	/* Make read and write ends of pipe nonblocking */
+	if ((flags = fcntl(g_pfd[0], F_GETFL)) == -1)
+	{
+		buff_err("critical error\n");
+	}
+	/* Make read end nonblocking */
+	flags |= O_NONBLOCK;
+	if (fcntl(g_pfd[0], F_SETFL, flags) == -1)
+	{
+		buff_err("critical error\n");
+	}
+	if ((flags = fcntl(g_pfd[1], F_GETFL)) == -1)
+	{
+		buff_err("critical error\n");
+	}
+	/* Make write end nonblocking */
+	flags |= O_NONBLOCK;
+	if (fcntl(g_pfd[1], F_SETFL, flags) == -1)
+	{
+		buff_err("critical error\n");
+	}
+	PlaybackDieNowRegisterCallback(WriteWakeUp);
 	while (0 == PlaybackDieNow(0))
 	{
 		pthread_mutex_lock(&bufferingMtx);
@@ -165,9 +199,9 @@ static void LinuxDvbBuffThread(Context_t *context)
 			/* Write data to valid output */
 			uint8_t *dataPtr = (uint8_t *)nodePtr + sizeof(BufferingNode_t);
 			int fd = nodePtr->dataType == OUTPUT_VIDEO ? videofd : audiofd;
-			if (0 != write_with_retry(fd, dataPtr, nodePtr->dataSize))
+			if (0 != WriteWithRetry(context, g_pfd[0], fd, dataPtr, nodePtr->dataSize))
 			{
-				printf("Something is WRONG\n");
+				buff_err("Something is WRONG\n");
 			}
 		}
 	}
@@ -178,6 +212,10 @@ static void LinuxDvbBuffThread(Context_t *context)
 	
 	buff_printf(20, "EXIT\n");
 	hasBufferingThreadStarted = false;
+	close(g_pfd[0]);
+	close(g_pfd[1]);
+	g_pfd[0] = -1;
+	g_pfd[1] = -1;
 }
 
 int32_t WriteSetBufferingSize(const uint32_t bufferSize)
@@ -240,7 +278,7 @@ int32_t LinuxDvbBuffOpen(Context_t *context, char *type, int outfd)
 	return ret;
 }
 
-int32_t LinuxDvbBuffClose(Context_t *context)
+int32_t LinuxDvbBuffClose(Context_t *context __attribute__((unused)))
 {
 	int32_t ret = 0;
 	
@@ -251,6 +289,10 @@ int32_t LinuxDvbBuffClose(Context_t *context)
 	if (hasBufferingThreadStarted) 
 	{
 		struct timespec max_wait = {0, 0};
+		
+		/* WakeUp if we are waiting in the write */ 
+		WriteWakeUp();
+		
 		pthread_mutex_lock(&bufferingMtx);
 		/* wait for thread end */
 		clock_gettime(CLOCK_REALTIME, &max_wait);
@@ -278,10 +320,14 @@ int32_t LinuxDvbBuffClose(Context_t *context)
 	return ret;
 }
 
-int32_t LinuxDvbBuffFlush(Context_t *context)
+int32_t LinuxDvbBuffFlush(Context_t *context __attribute__((unused)))
 {
 	static BufferingNode_t *nodePtr = NULL;
 	buff_printf(40, "ENTER bufferingQueueHead[%p]\n", bufferingQueueHead);
+	
+	/* signal if we are waiting for write to DVB decoders */
+	WriteWakeUp();
+
 	pthread_mutex_lock(&bufferingMtx);
 	while (bufferingQueueHead)
 	{

libeplayer3-arm/output/writer/mipsel/writer.c

@@ -29,6 +29,7 @@
 
 #include "misc.h"
 #include "writer.h"
+#include "common.h"
 
 /* ***************************** */
 /* Makros/Constants              */
@@ -100,6 +101,69 @@ static Writer_t *AvailableWriter[] =
 /*  Functions                    */
 /* ***************************** */
 
+ssize_t WriteWithRetry(Context_t *context, int pipefd, int fd, const void *buf, size_t size)
+{
+	fd_set rfds;
+	fd_set wfds;
+	
+	ssize_t ret;
+	int retval = -1;
+	int maxFd = pipefd > fd ? pipefd : fd;
+	
+	while(size > 0 && 0 == PlaybackDieNow(0) && !context->playback->isSeeking)
+	{
+		FD_ZERO(&rfds);
+		FD_ZERO(&wfds);
+
+		FD_SET(pipefd, &rfds);
+		FD_SET(fd, &wfds);
+
+		retval = select(maxFd + 1, &rfds, &wfds, NULL, NULL);
+		if (retval < 0)
+		{
+			break;
+		}
+		
+		if(FD_ISSET(pipefd, &rfds))
+		{
+			char tmp;
+			/* flush pipefd pipe */
+			while(1 == read(pipefd, &tmp, 1));
+			break;
+		}
+		
+		if(FD_ISSET(fd, &wfds))
+		{
+			ret = write(fd, buf, size);
+			if (ret < 0)
+			{
+				switch(errno)
+				{
+					case EINTR:
+					case EAGAIN:
+						continue;
+					default:
+						retval = -3;
+						break;
+				}
+				if (retval < 0)
+				{
+					break;
+				}
+			}
+				
+			if (ret < 0)
+			{
+				return ret;
+			}
+			
+			size -= ret;
+			buf += ret;
+		}
+	}
+	return 0;
+}
+
 ssize_t write_with_retry(int fd, const void *buf, size_t size)
 {
 	ssize_t ret;

libeplayer3-arm/playback/playback.c

@@ -57,6 +57,8 @@ if (debug_level >= level) printf("[%s:%s] " fmt, __FILE__, __FUNCTION__, ## x);
 #define cMaxSpeed_ff   128  /* fixme: revise */
 #define cMaxSpeed_fr   -320 /* fixme: revise */
 
+#define MAX_PLAYBACK_DIE_NOW_CALLBACKS 10
+
 /* ***************************** */
 /* Varaibles                     */
 /* ***************************** */
@@ -67,17 +69,30 @@ static int hasThreadStarted = 0;
 /* ***************************** */
 /* Prototypes                    */
 /* ***************************** */
+
 static int32_t PlaybackTerminate(Context_t *context);
+static int8_t dieNow = 0;
+static PlaybackDieNowCallback playbackDieNowCallbacks[MAX_PLAYBACK_DIE_NOW_CALLBACKS] = {NULL};
 
 /* ***************************** */
 /* MISC Functions                */
 /* ***************************** */
+
 int8_t PlaybackDieNow(int8_t val)
 {
-	static int8_t dieNow = 0;
-	if (val == 1)
+	if (val == 1 && dieNow == 0)
 	{
+		uint32_t i = 0;
 		dieNow = 1;
+		while (i < MAX_PLAYBACK_DIE_NOW_CALLBACKS)
+		{
+			if (playbackDieNowCallbacks[i] == NULL)
+			{
+				break;
+			}
+			playbackDieNowCallbacks[i]();
+			i += 1;
+		}
 	}
 	else if (val == 2)
 	{
@@ -86,6 +101,32 @@ int8_t PlaybackDieNow(int8_t val)
 	return dieNow;
 }
 
+bool PlaybackDieNowRegisterCallback(PlaybackDieNowCallback callback)
+{
+	bool ret = false;
+	if (callback)
+	{
+		uint32_t i = 0;
+		while (i < MAX_PLAYBACK_DIE_NOW_CALLBACKS)
+		{
+			if (playbackDieNowCallbacks[i] == callback)
+			{
+				ret = true;
+				break;
+			}
+			
+			if (playbackDieNowCallbacks[i] == NULL)
+			{
+				playbackDieNowCallbacks[i] = callback;
+				ret = true;
+				break;
+			}
+			i += 1;
+		}
+	}
+	return ret;
+}
+
 /* **************************** */
 /* Supervisor Thread            */
 /* **************************** */