Moved mutex into queue (made it thread safe), to avoid mutex locking in player, encoder and decoder.

4 files changed, 58 insertions, 33 deletions

diff --git a/src/decoder.cc b/src/decoder.cc
index 3792daf..35ba5ff 100644
--- a/src/decoder.cc
+++ b/src/decoder.cc
@@ -73,10 +73,8 @@ void Decoder::decode()
     Frame *pframe = new Frame(ptr, DVPACKAGE_SIZE);
     free(ptr);
 
-    pthread_mutex_lock(mutex);
     encode_queue->push(eframe);
     player_queue->push(pframe);
-    pthread_mutex_unlock(mutex);
     
     sem_post(encode_sem);
     
diff --git a/src/encoder.cc b/src/encoder.cc
index 878c97d..52a7f93 100644
--- a/src/encoder.cc
+++ b/src/encoder.cc
@@ -73,12 +73,9 @@ void Encoder::encode()
   while(*running) {
     sem_wait(sem);
 
-    pthread_mutex_lock(mutex);
     frame = queue->pop();
-    //    queue->plength();
-    //    pthread_mutex_unlock(mutex);
     
-    while(frame) {
+    if(frame) {
       if(record || 
          (freeze_request != freeze_value) || 
          (shoot_request != shoot_value)) {
@@ -97,12 +94,7 @@ void Encoder::encode()
         n->sendPackage(&h, frame->data, frame->size);
       }
       if(frame) delete frame;
-
-      //      pthread_mutex_lock(mutex);
-      frame = queue->pop();
-      //      queue->plength();
     }
-    pthread_mutex_unlock(mutex);
   }
 }
 
diff --git a/src/player.cc b/src/player.cc
index d57f831..963f6e7 100644
--- a/src/player.cc
+++ b/src/player.cc
@@ -129,9 +129,7 @@ void Player::player()
       break;
       
     case SDL_USEREVENT:
-      pthread_mutex_lock(mutex);
       frame = queue->pop();
-      pthread_mutex_unlock(mutex);
       if(!frame) break;
       
       if(first) {
@@ -139,8 +137,6 @@ void Player::player()
         pitches[1] = overlay->pitches[1];
         pitches[2] = overlay->pitches[2];
 
-        //        fprintf(stderr, "pitch[0]: %d - pitch[1]: %d - pitch[2]: %d\n", pitches[0], pitches[1], pitches[2]); fflush(stderr);
-
         dv_parse_header(decoder, frame->data);
         //dv_parse_packs(decoder, frame->data); // Not needed anyway!
 
@@ -151,8 +147,6 @@ void Player::player()
         first = false;
       }
       
-      fprintf(stderr, "[%d]", dv_is_PAL(decoder)); fflush(stderr);
-
       SDL_LockYUVOverlay(overlay);	
 
       // libdv img decode to yuv
@@ -184,10 +178,8 @@ void Player::player()
   ts.tv_nsec = 100000000L;	// 100ms
   nanosleep(&ts, NULL);
 
-  pthread_mutex_lock(mutex);
   frame = queue->pop();
-  pthread_mutex_unlock(mutex);
-  //  if(frame) delete frame;
+  if(frame) delete frame;
 }
 
 void Player::run()
diff --git a/src/queue.h b/src/queue.h
index 1cd0a3c..5e6d0f8 100644
--- a/src/queue.h
+++ b/src/queue.h
@@ -1,3 +1,4 @@
+/* -*- Mode: C++; tab-width: 2; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
 /*
  * RTVideoRec Realtime video recoder and encoder for Linux
  *
@@ -30,6 +31,7 @@
 #include <avformat.h>
 #include <avcodec.h>
 
+#include "thread.h"
 #include "util.h"
 
 typedef struct __buf_t {
@@ -41,7 +43,7 @@ typedef struct __buf_t {
 
 template<typename T>
 class Queue {
- public:
+public:
   Queue(int glimit = 0);
   ~Queue();
 
@@ -50,49 +52,56 @@ class Queue {
   
   void plength();
 
- private:
+private:
   int limit;
   buf_t *head;
   buf_t *tail;
   int count;
   pthread_mutex_t mutex;
+  T *_pop();
 };
 
-template<typename T>
-void Queue<T>::plength()
-{
-  fprintf(stderr, "[ql: %d]", count);
-  fflush(stderr);
-}
-
+/**
+ * Initialize queue
+ */
 template<typename T>
 Queue<T>::Queue(int glimit)
 {
+  pthread_mutex_init (&mutex, NULL);
   limit = glimit;
   count = 0;
   head = NULL;
   tail = NULL;
 }
 
+/**
+ * Clean up queue.
+ */
 template<typename T>
 Queue<T>::~Queue()
-{
+{ 
   if(count != 0) {
     fprintf(stderr, "Queue not empty (%d)\n", count);
-    while(T *t = pop()) delete t;
+    while(T *t = _pop()) delete t;
   }
+  pthread_mutex_destroy(&mutex);
 }
 
+/**
+ * Push element on queue.
+ */
 template<typename T>
 void Queue<T>::push(T *t)
 {
+  pthread_mutex_lock(&mutex);
+
   buf_t *b = (buf_t*)xmalloc(sizeof(*b));
   b->data = (void*)t;
 
   assert(b != NULL);
   
   if(limit && count > 0) {
-    T* tmp = (T*)pop();
+    T* tmp = (T*)_pop();
     delete tmp;
   }
   
@@ -100,6 +109,7 @@ void Queue<T>::push(T *t)
     head = tail = b;
     b->next = b->prev = NULL;
     count = 1;
+    pthread_mutex_unlock(&mutex);
     return;
   }
   
@@ -109,18 +119,38 @@ void Queue<T>::push(T *t)
     tail->prev = b;
   tail = b;
   count++;
+  
+  pthread_mutex_unlock(&mutex);
 }
 
+/**
+ * Pop element from queue.
+ * If queue is empty, NULL is returned.
+ */
 template<typename T>
 T *Queue<T>::pop()
 {
+  pthread_mutex_lock(&mutex);
+  T *d = _pop();
+  pthread_mutex_unlock(&mutex);
+  return d;
+}
+
+/**
+ * Pop helper method
+ * If queue is empty, NULL is returned.
+ */
+template<typename T>
+T *Queue<T>::_pop()
+{
   T *d;
   buf_t *b;
 
   assert(count >= 0);
   
-  if(count == 0)
+  if(count == 0) {
     return NULL;
+  }
   
   b = head;
   if(b->prev)
@@ -132,8 +162,21 @@ T *Queue<T>::pop()
   
   d = (T*)b->data;
   free(b);
+  
   return d;
 }
 
+/**
+ * Print current length of queue
+ */
+template<typename T>
+void Queue<T>::plength()
+{
+  pthread_mutex_lock(&mutex);
+  fprintf(stderr, "[ql: %d]", count);
+  fflush(stderr);
+  pthread_mutex_unlock(&mutex);
+}
+
 #endif