comparison data/fft/FFTDataServer.cpp @ 549:388afa99d537

* More careful (I hope!) locking

     if (m_fillThread) {
         m_fillThread->wait();
         delete m_fillThread;
     }
 
-    MutexLocker locker(&m_writeMutex,
-                       "FFTDataServer::~FFTDataServer::m_writeMutex");
+//    MutexLocker locker(&m_writeMutex,
+//                       "FFTDataServer::~FFTDataServer::m_writeMutex");
+
+    QMutexLocker mlocker(&m_fftBuffersLock);
+    QWriteLocker wlocker(&m_cacheVectorLock);
 
     for (CacheVector::iterator i = m_caches.begin(); i != m_caches.end(); ++i) {
-
         if (*i) {
             delete *i;
         }
     }
 

 #ifdef DEBUG_FFT_SERVER
     std::cerr << "FFTDataServer(" << this << " [" << (void *)QThread::currentThreadId() << "]): suspend" << std::endl;
 #endif
     Profiler profiler("FFTDataServer::suspend", false);
 
-    MutexLocker locker(&m_writeMutex,
-                       "FFTDataServer::suspend::m_writeMutex");
+    QMutexLocker locker(&m_fftBuffersLock);
     m_suspended = true;
 }
 
 void
 FFTDataServer::suspendWrites()

         Profiler profiler("FFTDataServer::getMagnitudeAt: filling");
 #ifdef DEBUG_FFT_SERVER
         std::cerr << "FFTDataServer::getMagnitudeAt: calling fillColumn(" 
                   << x << ")" << std::endl;
 #endif
-        // hold mutex so that write thread doesn't mess with class
-        // member data in fillColumn
-        MutexLocker locker(&m_writeMutex,
-                           "FFTDataServer::getMagnitudeAt: m_writeMutex");
-        fillColumn(x, true);
+        fillColumn(x, cache);
     }
     return cache->getMagnitudeAt(col, y);
 }
 
 bool

     if (!cache) return false;
 
     //!!! n.b. can throw
     if (!cache->haveSetColumnAt(col)) {
         Profiler profiler("FFTDataServer::getMagnitudesAt: filling");
-        MutexLocker locker(&m_writeMutex,
-                           "FFTDataServer::getMagnitudesAt: m_writeMutex");
-        fillColumn(x, true);
+        fillColumn(x, cache);
     }
 
     cache->getMagnitudesAt(col, values, minbin, count, step);
 
     return true;

     if (!cache) return 0;
 
     //!!! n.b. can throw
     if (!cache->haveSetColumnAt(col)) {
         Profiler profiler("FFTDataServer::getNormalizedMagnitudeAt: filling");
-        // hold mutex so that write thread doesn't mess with class
-        // member data in fillColumn
-        MutexLocker locker(&m_writeMutex,
-                           "FFTDataServer::getNormalizedMagnitudeAt: m_writeMutex");
-        fillColumn(x, true);
+        fillColumn(x, cache);
     }
     return cache->getNormalizedMagnitudeAt(col, y);
 }
 
 bool

     if (!cache) return false;
 
     //!!! n.b. can throw
     if (!cache->haveSetColumnAt(col)) {
         Profiler profiler("FFTDataServer::getNormalizedMagnitudesAt: filling");
-        MutexLocker locker(&m_writeMutex,
-                           "FFTDataServer::getNormalizedMagnitudesAt: m_writeMutex");
-        fillColumn(x, true);
+        fillColumn(x, cache);
     }
 
     for (size_t i = 0; i < count; ++i) {
         values[i] = cache->getNormalizedMagnitudeAt(col, i * step + minbin);
     }

     if (!cache) return 0;
 
     //!!! n.b. can throw
     if (!cache->haveSetColumnAt(col)) {
         Profiler profiler("FFTDataServer::getMaximumMagnitudeAt: filling");
-        // hold mutex so that write thread doesn't mess with class
-        // member data in fillColumn
-        MutexLocker locker(&m_writeMutex,
-                           "FFTDataServer::getMaximumMagnitudeAt: m_writeMutex");
-        fillColumn(x, true);
+        fillColumn(x, cache);
     }
     return cache->getMaximumMagnitudeAt(col);
 }
 
 float

     if (!cache) return 0;
 
     //!!! n.b. can throw
     if (!cache->haveSetColumnAt(col)) {
         Profiler profiler("FFTDataServer::getPhaseAt: filling");
-        // hold mutex so that write thread doesn't mess with class
-        // member data in fillColumn
-        MutexLocker locker(&m_writeMutex,
-                           "FFTDataServer::getPhaseAt: m_writeMutex");
-        fillColumn(x, true);
+        fillColumn(x, cache);
     }
     return cache->getPhaseAt(col, y);
 }
 
 bool

     if (!cache) return false;
 
     //!!! n.b. can throw
     if (!cache->haveSetColumnAt(col)) {
         Profiler profiler("FFTDataServer::getPhasesAt: filling");
-        MutexLocker locker(&m_writeMutex,
-                           "FFTDataServer::getPhasesAt: m_writeMutex");
-        fillColumn(x, true);
+        fillColumn(x, cache);
     }
 
     for (size_t i = 0; i < count; ++i) {
         values[i] = cache->getPhaseAt(col, i * step + minbin);
     }

     if (!cache->haveSetColumnAt(col)) {
         Profiler profiler("FFTDataServer::getValuesAt: filling");
 #ifdef DEBUG_FFT_SERVER
         std::cerr << "FFTDataServer::getValuesAt(" << x << ", " << y << "): filling" << std::endl;
 #endif
-        // hold mutex so that write thread doesn't mess with class
-        // member data in fillColumn
-        MutexLocker locker(&m_writeMutex,
-                           "FFTDataServer::getValuesAt: m_writeMutex");
-        fillColumn(x, true);
+        fillColumn(x, cache);
     }        
 
     cache->getValuesAt(col, y, real, imaginary);
 }
 

     //!!! n.b. can throw
     return cache->haveSetColumnAt(col);
 }    
 
 void
-FFTDataServer::fillColumn(size_t x, bool lockHeld)
+FFTDataServer::fillColumn(size_t x, FFTCacheReader *tester)
 {
     Profiler profiler("FFTDataServer::fillColumn", false);
 
     if (!m_model->isReady()) {
         std::cerr << "WARNING: FFTDataServer::fillColumn(" 
                   << x << "): model not yet ready" << std::endl;
         return;
     }
-
+/*
     if (!m_fftInput) {
         std::cerr << "WARNING: FFTDataServer::fillColumn(" << x << "): "
                   << "input has already been completed and discarded?"
                   << std::endl;
         return;
     }
-
+*/
     if (x >= m_width) {
         std::cerr << "WARNING: FFTDataServer::fillColumn(" << x << "): "
                   << "x > width (" << x << " > " << m_width << ")"
                   << std::endl;
         return;

     int startFrame = m_windowIncrement * x;
     int endFrame = startFrame + m_windowSize;
 
     startFrame -= winsize / 2;
     endFrame   -= winsize / 2;
-
-    for (int i = 0; i < off; ++i) {
-        m_fftInput[i] = 0.0;
-    }
-
-    for (int i = 0; i < off; ++i) {
-        m_fftInput[fftsize - i - 1] = 0.0;
-    }
-
-    if (startFrame < 0) {
-	pfx = -startFrame;
-	for (int i = 0; i < pfx; ++i) {
-	    m_fftInput[off + i] = 0.0;
-	}
-    }
 
 #ifdef DEBUG_FFT_SERVER_FILL
     std::cerr << "FFTDataServer::fillColumn: requesting frames "
               << startFrame + pfx << " -> " << endFrame << " ( = "
               << endFrame - (startFrame + pfx) << ") at index "
               << off + pfx << " in buffer of size " << m_fftSize
               << " with window size " << m_windowSize 
               << " from channel " << m_channel << std::endl;
 #endif
 
+    QMutexLocker locker(&m_fftBuffersLock);
+
+    if (tester) {
+        // We are being called from a function that wanted to obtain a
+        // column using an FFTCacheReader.  Before calling us, it
+        // checked whether the column was available already, and the
+        // reader reported that it wasn't.  Now we test again with the
+        // mutex held, to avoid a race condition in case another
+        // thread has called fillColumn at the same time.
+        if (tester->haveSetColumnAt(x & m_cacheWidthMask)) {
+            return;
+        }
+    }
+
+    if (!m_fftInput) {
+        std::cerr << "WARNING: FFTDataServer::fillColumn(" << x << "): "
+                  << "input has already been completed and discarded?"
+                  << std::endl;
+        return;
+    }
+
+    for (int i = 0; i < off; ++i) {
+        m_fftInput[i] = 0.0;
+    }
+
+    for (int i = 0; i < off; ++i) {
+        m_fftInput[fftsize - i - 1] = 0.0;
+    }
+
+    if (startFrame < 0) {
+	pfx = -startFrame;
+	for (int i = 0; i < pfx; ++i) {
+	    m_fftInput[off + i] = 0.0;
+	}
+    }
+
     int count = 0;
     if (endFrame > startFrame + pfx) count = endFrame - (startFrame + pfx);
 
     int got = m_model->getData(m_channel, startFrame + pfx,
                                count, m_fftInput + off + pfx);

 	m_fftInput[i + hs] = temp;
     }
 
     fftf_execute(m_fftPlan);
 
-    // If our cache uses polar storage, it's more friendly for us to
-    // do the conversion before taking the write mutex
-
     float factor = 0.f;
 
     if (cache->getStorageType() == FFTCache::Compact ||
         cache->getStorageType() == FFTCache::Polar) {
 

         }
     }
 
     Profiler subprof("FFTDataServer::fillColumn: set to cache");
 
-    {
-        MutexLocker locker(lockHeld ? 0 : &m_writeMutex,
-                           "FFTDataServer::fillColumn: m_writeMutex [2]");
-
-        if (cache->getStorageType() == FFTCache::Compact ||
-            cache->getStorageType() == FFTCache::Polar) {
+    if (cache->getStorageType() == FFTCache::Compact ||
+        cache->getStorageType() == FFTCache::Polar) {
             
-            cache->setColumnAt(col,
-                               m_workbuffer,
-                               m_workbuffer + hs + 1,
-                               factor);
-
-        } else {
-
-            cache->setColumnAt(col,
-                               m_workbuffer,
-                               m_workbuffer + hs + 1);
-        }
+        cache->setColumnAt(col,
+                           m_workbuffer,
+                           m_workbuffer + hs + 1,
+                           factor);
+
+    } else {
+
+        cache->setColumnAt(col,
+                           m_workbuffer,
+                           m_workbuffer + hs + 1);
     }
 
     if (m_suspended) {
 //        std::cerr << "FFTDataServer::fillColumn(" << x << "): calling resume" << std::endl;
 //        resume();

 
     if (m_fillFrom > start) {
 
         for (size_t f = m_fillFrom; f < end; f += m_server.m_windowIncrement) {
 	    
-            m_server.fillColumn(int((f - start) / m_server.m_windowIncrement),
-                                false);
+            m_server.fillColumn(int((f - start) / m_server.m_windowIncrement));
 
             if (m_server.m_exiting) return;
 
             while (m_server.m_suspended) {
 #ifdef DEBUG_FFT_SERVER
                 std::cerr << "FFTDataServer(" << this << " [" << (void *)QThread::currentThreadId() << "]): suspended, waiting..." << std::endl;
 #endif
-                {
-                    MutexLocker locker(&m_server.m_writeMutex,
-                                       "FFTDataServer::run::m_writeMutex [1]");
-                    m_server.m_condition.wait(&m_server.m_writeMutex, 10000);
+                MutexLocker locker(&m_server.m_fftBuffersLock,
+                                   "FFTDataServer::run::m_fftBuffersLock [1]");
+                if (m_server.m_suspended && !m_server.m_exiting) {
+                    m_server.m_condition.wait(&m_server.m_fftBuffersLock, 10000);
                 }
 #ifdef DEBUG_FFT_SERVER
                 std::cerr << "FFTDataServer(" << this << " [" << (void *)QThread::currentThreadId() << "]): waited" << std::endl;
 #endif
                 if (m_server.m_exiting) return;

 
     size_t baseCompletion = m_completion;
 
     for (size_t f = start; f < remainingEnd; f += m_server.m_windowIncrement) {
 
-        m_server.fillColumn(int((f - start) / m_server.m_windowIncrement),
-                            false);
+        m_server.fillColumn(int((f - start) / m_server.m_windowIncrement));
 
         if (m_server.m_exiting) return;
 
         while (m_server.m_suspended) {
 #ifdef DEBUG_FFT_SERVER
             std::cerr << "FFTDataServer(" << this << " [" << (void *)QThread::currentThreadId() << "]): suspended, waiting..." << std::endl;
 #endif
             {
-                MutexLocker locker(&m_server.m_writeMutex,
-                                   "FFTDataServer::run::m_writeMutex [2]");
-                m_server.m_condition.wait(&m_server.m_writeMutex, 10000);
+                MutexLocker locker(&m_server.m_fftBuffersLock,
+                                   "FFTDataServer::run::m_fftBuffersLock [2]");
+                if (m_server.m_suspended && !m_server.m_exiting) {
+                    m_server.m_condition.wait(&m_server.m_fftBuffersLock, 10000);
+                }
             }
             if (m_server.m_exiting) return;
         }
 		    
         if (++counter == updateAt) {