comparison data/fileio/FileReadThread.cpp @ 408:115f60df1e4d

* Speed up spectrogram painting by releasing mutex in FFTDataServer
while calculating data prior to writing it, and by adding whole-column
value query methods to FFT objects
* Add paint cache to Thumbwheel -- repaints of this widget were slowing
down the whole spectrogram repaint
* More uses of MutexLocker (named and with debug) and more profile
points
* Make startup much quicker some of the time, with OSC server in place

 */
 
 #include "FileReadThread.h"
 
 #include "base/Profiler.h"
+#include "base/Thread.h"
 
 #include <iostream>
 #include <unistd.h>
 
 //#define DEBUG_FILE_READ_THREAD 1

 }
 
 void
 FileReadThread::run()
 {
-    m_mutex.lock();
+    MutexLocker locker(&m_mutex, "FileReadThread::run::m_mutex");
 
     while (!m_exiting) {
         if (m_queue.empty()) {
             m_condition.wait(&m_mutex, 1000);
         } else {

         }
         notifyCancelled();
     }
 
     notifyCancelled();
-    m_mutex.unlock();
 
 #ifdef DEBUG_FILE_READ_THREAD
     std::cerr << "FileReadThread::run() exiting" << std::endl;
 #endif
 }

 {
 #ifdef DEBUG_FILE_READ_THREAD
     std::cerr << "FileReadThread::finish()" << std::endl;
 #endif
 
-    m_mutex.lock();
-    while (!m_queue.empty()) {
-        m_cancelledRequests[m_queue.begin()->first] = m_queue.begin()->second;
-        m_newlyCancelled.insert(m_queue.begin()->first);
-        m_queue.erase(m_queue.begin());
-    }
-
-    m_exiting = true;
-    m_mutex.unlock();
+    {
+        MutexLocker locker(&m_mutex, "FileReadThread::finish::m_mutex");
+
+        while (!m_queue.empty()) {
+            m_cancelledRequests[m_queue.begin()->first] = m_queue.begin()->second;
+            m_newlyCancelled.insert(m_queue.begin()->first);
+            m_queue.erase(m_queue.begin());
+        }
+        
+        m_exiting = true;
+    }
 
     m_condition.wakeAll();
 
 #ifdef DEBUG_FILE_READ_THREAD
     std::cerr << "FileReadThread::finish() exiting" << std::endl;

 }
 
 int
 FileReadThread::request(const Request &request)
 {
-    m_mutex.lock();
-    
-    int token = m_nextToken++;
-    m_queue[token] = request;
-
-    m_mutex.unlock();
+    int token;
+
+    {
+        MutexLocker locker(&m_mutex, "FileReadThread::request::m_mutex");
+    
+        token = m_nextToken++;
+        m_queue[token] = request;
+    }
+
     m_condition.wakeAll();
 
     return token;
 }
 
 void
 FileReadThread::cancel(int token)
 {
-    m_mutex.lock();
-
-    if (m_queue.find(token) != m_queue.end()) {
-        m_cancelledRequests[token] = m_queue[token];
-        m_queue.erase(token);
-        m_newlyCancelled.insert(token);
-    } else if (m_readyRequests.find(token) != m_readyRequests.end()) {
-        m_cancelledRequests[token] = m_readyRequests[token];
-        m_readyRequests.erase(token);
-    } else {
-        std::cerr << "WARNING: FileReadThread::cancel: token " << token << " not found" << std::endl;
-    }
-
-    m_mutex.unlock();
+    {
+        MutexLocker locker(&m_mutex, "FileReadThread::cancel::m_mutex");
+
+        if (m_queue.find(token) != m_queue.end()) {
+            m_cancelledRequests[token] = m_queue[token];
+            m_queue.erase(token);
+            m_newlyCancelled.insert(token);
+        } else if (m_readyRequests.find(token) != m_readyRequests.end()) {
+            m_cancelledRequests[token] = m_readyRequests[token];
+            m_readyRequests.erase(token);
+        } else {
+            std::cerr << "WARNING: FileReadThread::cancel: token " << token << " not found" << std::endl;
+        }
+    }
 
 #ifdef DEBUG_FILE_READ_THREAD
     std::cerr << "FileReadThread::cancel(" << token << ") waking condition" << std::endl;
 #endif
 

 }
 
 bool
 FileReadThread::isReady(int token)
 {
-    m_mutex.lock();
+    MutexLocker locker(&m_mutex, "FileReadThread::isReady::m_mutex");
 
     bool ready = m_readyRequests.find(token) != m_readyRequests.end();
 
-    m_mutex.unlock();
     return ready;
 }
 
 bool
 FileReadThread::isCancelled(int token)
 {
-    m_mutex.lock();
+    MutexLocker locker(&m_mutex, "FileReadThread::isCancelled::m_mutex");
 
     bool cancelled = 
         m_cancelledRequests.find(token) != m_cancelledRequests.end() &&
         m_newlyCancelled.find(token) == m_newlyCancelled.end();
 
-    m_mutex.unlock();
     return cancelled;
 }
 
 bool
 FileReadThread::getRequest(int token, Request &request)
 {
-    m_mutex.lock();
+    MutexLocker locker(&m_mutex, "FileReadThread::getRequest::m_mutex");
 
     bool found = false;
 
     if (m_queue.find(token) != m_queue.end()) {
         request = m_queue[token];

     } else if (m_readyRequests.find(token) != m_readyRequests.end()) {
         request = m_readyRequests[token];
         found = true;
     }
 
-    m_mutex.unlock();
-    
     return found;
 }
 
 void
 FileReadThread::done(int token)
 {
-    m_mutex.lock();
+    MutexLocker locker(&m_mutex, "FileReadThread::done::m_mutex");
 
     bool found = false;
 
     if (m_cancelledRequests.find(token) != m_cancelledRequests.end()) {
         m_cancelledRequests.erase(token);

         found = true;
     } else if (m_queue.find(token) != m_queue.end()) {
         std::cerr << "WARNING: FileReadThread::done(" << token << "): request is still in queue (wait or cancel it)" << std::endl;
     }
 
-    m_mutex.unlock();
-
     if (!found) {
         std::cerr << "WARNING: FileReadThread::done(" << token << "): request not found" << std::endl;
     }
 }
 
 void
 FileReadThread::process()
 {
     // entered with m_mutex locked and m_queue non-empty
 
-#ifdef DEBUG_FILE_READ_THREAD
-    Profiler profiler("FileReadThread::process()", true);
-#endif
+    Profiler profiler("FileReadThread::process", true);
 
     int token = m_queue.begin()->first;
     Request request = m_queue.begin()->second;
 
     m_mutex.unlock();

 
     bool successful = false;
     bool seekFailed = false;
     ssize_t r = 0;
 
-    if (request.mutex) request.mutex->lock();
-
-    if (::lseek(request.fd, request.start, SEEK_SET) == (off_t)-1) {
-        seekFailed = true;
-    } else {
-        
-        // if request.size is large, we want to avoid making a single
-        // system call to read it all as it may block too much
-
-        static const size_t blockSize = 256 * 1024;
-        
-        size_t size = request.size;
-        char *destination = request.data;
-
-        while (size > 0) {
-            size_t readSize = size;
-            if (readSize > blockSize) readSize = blockSize;
-            ssize_t br = ::read(request.fd, destination, readSize);
-            if (br < 0) { 
-                r = br;
-                break;
-            } else {
-                r += br;
-                if (br < ssize_t(readSize)) break;
+    { 
+        MutexLocker rlocker(request.mutex, "FileReadThread::process::request.mutex");
+
+        if (::lseek(request.fd, request.start, SEEK_SET) == (off_t)-1) {
+            seekFailed = true;
+        } else {
+        
+            // if request.size is large, we want to avoid making a single
+            // system call to read it all as it may block too much
+            
+            static const size_t blockSize = 256 * 1024;
+            
+            size_t size = request.size;
+            char *destination = request.data;
+            
+            while (size > 0) {
+                size_t readSize = size;
+                if (readSize > blockSize) readSize = blockSize;
+                ssize_t br = ::read(request.fd, destination, readSize);
+                if (br < 0) { 
+                    r = br;
+                    break;
+                } else {
+                    r += br;
+                    if (br < ssize_t(readSize)) break;
+                }
+                destination += readSize;
+                size -= readSize;
             }
-            destination += readSize;
-            size -= readSize;
-        }
-    }
-
-    if (request.mutex) request.mutex->unlock();
+        }
+    }
 
     if (seekFailed) {
         ::perror("Seek failed");
         std::cerr << "ERROR: FileReadThread::process: seek to "
                   << request.start << " failed" << std::endl;