Fix transcoding startup bug.

9 files changed, 173 insertions, 47 deletions

diff --git a/.gitignore b/.gitignore
index 9fbd4b0..eaf2fa2 100644
--- a/.gitignore
+++ b/.gitignore
@@ -26,3 +26,13 @@ config.mk
 config.mk.bak
 example/bin
 example/obj
+transtreamproxy.IAB
+transtreamproxy.IAD
+transtreamproxy.IMB
+transtreamproxy.IMD
+transtreamproxy.PFI
+transtreamproxy.PO
+transtreamproxy.PR
+transtreamproxy.PS
+transtreamproxy.PRI
+transtreamproxy.WK3
diff --git a/src/Demuxer.cpp b/src/Demuxer.cpp
index d504b4e..345c830 100644
--- a/src/Demuxer.cpp
+++ b/src/Demuxer.cpp
@@ -19,10 +19,11 @@
 #include "Logger.h"
 #include "Demuxer.h"
 
+bool terminated();
+
 using namespace std;
 //-------------------------------------------------------------------------------
 
-bool terminated();
 std::string Demuxer::webif_reauest(std::string request) throw(http_trap)
 {
 	if ((sock = socket(PF_INET, SOCK_STREAM, 0)) < 0)
@@ -48,8 +49,9 @@ std::string Demuxer::webif_reauest(std::string request) throw(http_trap)
 
 		pollevt[0].revents = 0;
 		int poll_state = poll(pollevt, 1, 1000);
-		if (poll_state == 0)
+		if (poll_state == 0) { 
 			break;
+		}
 		else if (poll_state < 0) {
 			ERROR("webif receive poll error : %s (%d)", strerror(errno), errno);
 			throw(http_trap("webif response fail.", 502, "Bad Gateway, webif response error"));
@@ -207,6 +209,8 @@ void Demuxer::open() throw(http_trap)
 
 Demuxer::Demuxer(HttpHeader *header) throw(http_trap)
 {
+	SingleLock lock(&demux_mutex);
+
 	demux_id = pat_pid = fd = sock = -1;
 	pmt_pid = audio_pid = video_pid = 0;
 
@@ -222,7 +226,18 @@ Demuxer::Demuxer(HttpHeader *header) throw(http_trap)
 	}
 	webif_request += "\r\n";
 
-	std::string webif_response = webif_reauest(webif_request);
+	std::string webif_response = "";
+	for(int retry_count = 0; retry_count < 32; retry_count++) {
+		webif_response = webif_reauest(webif_request);
+		if (terminated()) {
+			return;
+		}
+		if(webif_response.length() && webif_response.find("-SERVICE ERROR:-") == string::npos){
+			DEBUG("webif_response recv success.");
+			break;
+		}
+		WARNING("webif_response fail, retry count : %d", retry_count);
+	}
 	DEBUG("webif response :\n%s", webif_response.c_str());
 
 	if (webif_response.find("WWW-Authenticate") != std::string::npos) {
@@ -238,6 +253,8 @@ Demuxer::Demuxer(HttpHeader *header) throw(http_trap)
 
 Demuxer::~Demuxer() throw()
 {
+	SingleLock lock(&demux_mutex);
+
 	std::vector<unsigned long>::iterator iter = pids.begin();
 	for (; iter != pids.end(); ++iter) {
 		unsigned long pid = *iter;
diff --git a/src/Demuxer.h b/src/Demuxer.h
index 11c0797..5370f44 100644
--- a/src/Demuxer.h
+++ b/src/Demuxer.h
@@ -16,15 +16,11 @@
 #include "Util.h"
 #include "Http.h"
 #include "Source.h"
+#include "Mutex.h"
 //----------------------------------------------------------------------
 
 class Demuxer : public Source
 {
-public:
-	int	pmt_pid;
-	int	video_pid;
-	int	audio_pid;
-
 private:
 	int fd;
 	int sock;
@@ -34,6 +30,8 @@ private:
 	std::vector<unsigned long> pids;
 	std::vector<unsigned long> new_pids;
 
+	Mutex demux_mutex;
+
 protected:
 	std::string webif_reauest(std::string request) throw(http_trap);
 	bool already_exist(std::vector<unsigned long> &pidlist, int pid);
diff --git a/src/Encoder.cpp b/src/Encoder.cpp
index 044e130..10142a1 100644
--- a/src/Encoder.cpp
+++ b/src/Encoder.cpp
@@ -18,11 +18,15 @@
 #include "Logger.h"
 #include "Encoder.h"
 
+bool terminated();
+
 using namespace std;
 //----------------------------------------------------------------------
 
 Encoder::Encoder() throw(trap)
 {
+	SingleLock lock(&encoder_mutex);
+
 	encoder_id = fd = -1;
 	max_encodr_count = state = ENCODER_STAT_INIT;
 
@@ -79,6 +83,8 @@ Encoder::Encoder() throw(trap)
 
 Encoder::~Encoder()
 {
+	SingleLock lock(&encoder_mutex);
+
 	Post();
 	encoder_close();
 }
@@ -99,6 +105,7 @@ void Encoder::encoder_close()
 
 bool Encoder::encoder_open()
 {
+	errno = 0;
 	std::string path = "/dev/bcm_enc" + Util::ultostr(encoder_id);
 	fd = ::open(path.c_str(), O_RDWR, 0);
 	if (fd >= 0) {
@@ -109,10 +116,9 @@ bool Encoder::encoder_open()
 }
 //----------------------------------------------------------------------
 
-bool terminated();
 bool Encoder::retry_open(int retry_count, int sleep_time)
 {
-	for (int i = 0; i < retry_count; ++i) {
+	for (int i = 0; i < retry_count*10; ++i) {
 		if (terminated()) {
 			break;
 		}
@@ -121,9 +127,7 @@ bool Encoder::retry_open(int retry_count, int sleep_time)
 			return true;
 		}
 		WARNING("encoder%d open fail, retry count : %d/%d", encoder_id, i, retry_count);
-		if (retry_count > 1) {
-			sleep(sleep_time);
-		}
+		usleep(sleep_time*100*1000); /*wait sleep_time ms*/
 	}
 	ERROR("encoder open fail : %s (%d)", strerror(errno), errno);
 	return false;
diff --git a/src/Encoder.h b/src/Encoder.h
index 3a69327..5831482 100644
--- a/src/Encoder.h
+++ b/src/Encoder.h
@@ -12,6 +12,7 @@
 
 #include "3rdparty/trap.h"
 
+#include "Mutex.h"
 #include "SharedMemory.h"
 //----------------------------------------------------------------------
 
@@ -25,6 +26,7 @@ class Encoder : public SharedMemory<Session>
 {
 private:
 	int fd;
+	Mutex encoder_mutex;
 
 public:
 	enum {
diff --git a/src/Mpeg.h b/src/Mpeg.h
index f3e643f..4c30e07 100644
--- a/src/Mpeg.h
+++ b/src/Mpeg.h
@@ -67,7 +67,6 @@ private:
 
 public:
 	off_t stream_length;
-	int pmt_pid, video_pid, audio_pid;
 
 	Mpeg(std::string filename, bool request_time_seek) throw (trap);
 	virtual ~Mpeg() throw () {}
diff --git a/src/Mutex.h b/src/Mutex.h
new file mode 100644
index 0000000..5bf9d2a
--- /dev/null
+++ b/src/Mutex.h
@@ -0,0 +1,66 @@
+/*

+ * Mutex.h

+ *

+ *  Created on: 2013. 1. 5.

+ *      Author: oskwon

+ */

+

+#ifndef MUTEX_H_

+#define MUTEX_H_

+

+#include <pthread.h>

+//----------------------------------------------------------------------

+

+class Mutex

+{

+protected:

+	pthread_mutex_t mLockHandle;

+

+public:

+	Mutex()

+	{

+		pthread_mutex_init(&mLockHandle, NULL);

+	}

+	virtual ~Mutex()

+	{

+		pthread_mutex_destroy(&mLockHandle);

+	}

+

+	int lock()

+	{

+		return pthread_mutex_lock(&mLockHandle);

+	}

+	int try_lock()

+	{

+		if (pthread_mutex_trylock(&mLockHandle) == 16)

+			return -1;

+		return 0;

+	}

+	int unlock()

+	{

+		return pthread_mutex_unlock(&mLockHandle);

+	}

+};

+//----------------------------------------------------------------------

+

+class SingleLock

+{

+protected:

+	Mutex* mMutexHandle;

+

+public:

+	SingleLock(Mutex* mutex)

+	{

+		mMutexHandle = mutex;

+		mMutexHandle->try_lock();

+	}

+

+	~SingleLock()

+	{

+		mMutexHandle->unlock();

+	}

+};

+//----------------------------------------------------------------------

+

+#endif

+

diff --git a/src/Source.h b/src/Source.h
index 81dbb13..2b90fa4 100644
--- a/src/Source.h
+++ b/src/Source.h
@@ -18,6 +18,11 @@ public:
 	virtual ~Source(){}
 	virtual int get_fd() const throw() = 0;
 	virtual bool is_initialized() = 0;
+	
+public:
+	int pmt_pid;
+	int video_pid;
+	int audio_pid;
 };
 //----------------------------------------------------------------------
 
diff --git a/src/main.cpp b/src/main.cpp
index fdf2263..b00c33e 100644
--- a/src/main.cpp
+++ b/src/main.cpp
@@ -58,20 +58,34 @@ bool terminated()
 }
 //----------------------------------------------------------------------
 
-void cbexit()
+inline void release_resource()
 {
-	INFO("release resource start");
-	if (encoder) { delete encoder; encoder = 0; }
-	if (source)  { delete source; source = 0; }
+	DEBUG("release resource start");
+	if (encoder) { 
+		delete encoder;
+		encoder = 0;
+		DEBUG("encoder release ok");
+	}
+	if (source) {
+		delete source;
+		source = 0;
+		DEBUG("source release ok");
+	}
+	DEBUG("release resource finish");
+}
+//----------------------------------------------------------------------
 
+void cbexit()
+{
+	release_resource();
 	char checker_filename[255] = {0};
 	if (tsp_pid) {
 		::sprintf(checker_filename, TSP_CHECKER_TEMPLETE, tsp_pid);
 		if (::access(checker_filename, F_OK) == 0) {
 			::unlink(checker_filename);
+			DEBUG("checker_file unlink ok - [%s]", checker_filename);
 		}
 	}
-	INFO("release resource finish");
 }
 //----------------------------------------------------------------------
 
@@ -91,12 +105,19 @@ inline int streaming_write(const char *buffer, size_t buffer_len, bool enable_lo
 	}
 //----------------------------------------------------------------------
 
-int send_signal(pid_t pid, int signal)
+inline int exist_process(pid_t pid)
 {
 	char process_path[255] = {0};
 	sprintf(process_path, "/proc/%d", pid);
+	if (access(process_path, F_OK) == 0)
+		return 1;
+	return 0;
+}
+//----------------------------------------------------------------------
 
-	if (access(process_path, F_OK) == 0) {
+inline int send_signal(pid_t pid, int signal)
+{
+	if (exist_process(pid)) {
 		kill(pid, signal);
 		DD_LOG("  >> run kill-pid : %ld -> %ld (%d)", getpid(), pid, signal);
 	}
@@ -110,10 +131,25 @@ void signal_handler_checker(int sig_no)
 }
 //----------------------------------------------------------------------
 
+int check_sleep(pid_t pid, int sleep_time)
+{
+	int sleep_count = 0;
+	int max_sleep_count = sleep_time * 10;
+
+	while (sleep_count++ < max_sleep_count) {
+		if (!exist_process(pid)) {
+			return 0;
+		}
+		usleep(100*1000);
+	}
+	return 1;
+}
+//----------------------------------------------------------------------
+
 int tsp_checker(pid_t pid)
 {
 	char check_filename[255] = {0};
-	sleep(1);
+	
 	sprintf(check_filename, TSP_CHECKER_TEMPLETE, ::getppid());
 
 	int timebase_count = 0, exit_count = 0;
@@ -150,10 +186,10 @@ int tsp_checker(pid_t pid)
 
 	DD_LOG("kill (%ld)", tsp_pid);
 
-	sleep(3);
-	send_signal(tsp_pid, SIGKILL);
-	sleep(2);
-
+	if (check_sleep(tsp_pid, 3)) {
+		send_signal(tsp_pid, SIGKILL);
+		check_sleep(tsp_pid, 2);
+	}
 	return 0;
 }
 //----------------------------------------------------------------------
@@ -173,6 +209,7 @@ int main(int argc, char **argv)
 	signal(SIGINT,  signal_handler);
 	signal(SIGSEGV, signal_handler);
 	signal(SIGUSR2, signal_handler);
+	signal(SIGPIPE, signal_handler);
 
 	atexit(cbexit);
 
@@ -205,11 +242,8 @@ int main(int argc, char **argv)
 		case HttpHeader::TRANSCODING_FILE:
 			try {
 				std::string uri = UriDecoder().decode(header.page_params["file"].c_str());
-				Mpeg *ts = new Mpeg(uri, false);
-				pmt_pid   = ts->pmt_pid;
-				video_pid = ts->video_pid;
-				audio_pid = ts->audio_pid;
-				source = ts;
+				encoder = new Encoder();
+				source = new Mpeg(uri, false);
 			}
 			catch (const trap &e) {
 				throw(http_trap(e.what(), 404, "Not Found"));
@@ -226,11 +260,8 @@ int main(int argc, char **argv)
 				sprintf(update_status_command, "touch "TSP_CHECKER_TEMPLETE, tsp_pid);
 				system(update_status_command);
 
-				Demuxer *dmx = new Demuxer(&header);
-				pmt_pid   = dmx->pmt_pid;
-				video_pid = dmx->video_pid;
-				audio_pid = dmx->audio_pid;
-				source = dmx;
+				encoder = new Encoder();
+				source = new Demuxer(&header);
 			}
 			catch (const http_trap &e) {
 				throw(e);
@@ -251,12 +282,11 @@ int main(int argc, char **argv)
 			throw(http_trap(std::string("not support source type : ") + Util::ultostr(header.type), 400, "Bad Request"));
 		}
 
-		encoder = new Encoder();
-		int encoder_retry_max_count = 1;
-		if (header.type == HttpHeader::TRANSCODING_FILE) {
-			encoder_retry_max_count = 2;
-		}
-		if (!encoder->retry_open(encoder_retry_max_count, 3)) {
+		pmt_pid   = source->pmt_pid;
+		video_pid = source->video_pid;
+		audio_pid = source->audio_pid;
+
+		if (!encoder->retry_open(2, 3)) {
 			throw(http_trap("encoder open fail.", 503, "Service Unavailable"));
 		}
 
@@ -494,11 +524,6 @@ void signal_handler(int sig_no)
 {
 	ERROR("signal no : %s (%d)", strsignal(sig_no), sig_no);
 	is_terminated = true;
-	cbexit();
-
-	if (sig_no == SIGSEGV) {
-		exit(0);
-	}
 }
 //----------------------------------------------------------------------