CubeDataThread.cpp

 
/* SPDX-License-Identifier: CC0-1.0 */
#include "IsisDebug.h"
#include "CubeDataThread.h"
 
#include <QApplication>
#include <QEventLoop>
#include <QMap>
#include <QMutex>
#include <QPair>
#include <QReadWriteLock>
#include <QString>
 
#include <iomanip>
#include <iostream>
 
#include "Brick.h"
#include "Cube.h"
#include "FileName.h"
#include "IException.h"
#include "IString.h"
#include "UniversalGroundMap.h"
 
namespace Isis {
  CubeDataThread::CubeDataThread() {
    p_managedCubes = NULL;
    p_managedData = NULL;
    p_threadSafeMutex = NULL;
    p_managedDataSources = NULL;
 
    p_managedCubes = new QMap< int, QPair< bool, Cube * > > ;
    p_managedData = new QList< QPair< QReadWriteLock *, Brick * > > ;
    p_threadSafeMutex = new QMutex();
    p_managedDataSources = new QList< int > ;
 
    p_numChangeListeners = 0;
    p_currentLocksWaiting = 0;
    p_currentId = 1; // start with ID == 1
    p_stopping = false;
 
    // Start this thread's event loop and make it so the slots are contained
    //   within this thread (this class automatically runs in its own thread)
    start();
    moveToThread(this);
 
  }
 
 
  CubeDataThread::~CubeDataThread() {
    // Shutdown the event loop(s)
    QThread::exit(0);
 
    p_stopping = true;
 
    while (!isFinished()) {
      QThread::yieldCurrentThread();
    }
 
    // Destroy the bricks still in memory
    if (p_managedData) {
      for (int i = p_managedData->size() - 1; i >= 0; i--) {
        delete (*p_managedData)[i].first;
        delete (*p_managedData)[i].second;
        p_managedData->removeAt(i);
      }
 
      delete p_managedData;
      p_managedData = NULL;
    }
 
    // Destroy the cubes still in memory
    if (p_managedCubes) {
      for (int i = p_managedCubes->size() - 1; i >= 0; i--) {
        if ((p_managedCubes->end() - 1)->first) // only delete if we own it!
          delete (p_managedCubes->end() - 1).value().second;
        p_managedCubes->erase(p_managedCubes->end() - 1);
      }
    }
 
    // Destroy the mutex that controls access to the cubes
    if (p_threadSafeMutex) {
      delete p_threadSafeMutex;
      p_threadSafeMutex = NULL;
    }
 
    // Destroy the data sources vector
    if (p_managedDataSources) {
      delete p_managedDataSources;
      p_managedDataSources = NULL;
    }
  }
 
  int CubeDataThread::AddCube(const FileName &fileName,
                              bool mustOpenReadWrite) {
    Cube *newCube = new Cube();
 
    try {
      newCube->open(fileName.expanded(), "rw");
    }
    catch (IException &e) {
      if (!mustOpenReadWrite) {
        newCube->open(fileName.expanded(), "r");
      }
      else {
        throw;
      }
    }
 
    p_threadSafeMutex->lock();
 
    int newId = p_currentId;
    p_currentId++;
 
    QPair< bool, Cube * > newEntry;
    newEntry.first = true; // we own this!
    newEntry.second = newCube;
    p_managedCubes->insert(newId, newEntry);
 
    p_threadSafeMutex->unlock();
 
    return newId;
  }
 
  int CubeDataThread::AddCube(Cube *cube) {
    p_threadSafeMutex->lock();
 
    int newId = p_currentId;
    p_currentId++;
 
    QPair< bool, Cube * > newEntry;
    newEntry.first = false; // we don't own this!
    newEntry.second = cube;
    p_managedCubes->insert(newId, newEntry);
 
    p_threadSafeMutex->unlock();
 
    return newId;
  }
 
 
  void CubeDataThread::RemoveCube(int cubeId) {
    p_threadSafeMutex->lock();
 
    QMap< int, QPair< bool, Cube * > >::iterator i;
    i = p_managedCubes->find(cubeId);
 
    if (i == p_managedCubes->end()) {
      p_threadSafeMutex->unlock();
      QString msg = "CubeDataThread::RemoveCube failed because cube ID [";
      msg += QString::number(cubeId);
      msg += "] not found";
      throw IException(IException::Programmer, msg, _FILEINFO_);
    }
 
    if (p_managedDataSources->contains(cubeId)) {
      p_threadSafeMutex->unlock();
      QString msg = "CubeDataThread::RemoveCube failed cube ID [";
      msg += QString::number(cubeId);
      msg += "] has requested Bricks";
      throw IException(IException::Programmer, msg, _FILEINFO_);
    }
 
    // if we have ownership of the cube then me must delete it
    if (i.value().first)
      delete i.value().second;
    i.value().second = NULL;
 
    p_managedCubes->remove(i.key());
 
    p_threadSafeMutex->unlock();
 
  }
 
 
  void CubeDataThread::AddChangeListener() {
    p_numChangeListeners++;
  }
 
  void CubeDataThread::RemoveChangeListener() {
    p_numChangeListeners--;
  }
 
  void CubeDataThread::GetCubeData(int cubeId, int ss, int sl, int es, int el,
                                   int band, void *caller, bool sharedLock) {
 
    Brick *requestedBrick = NULL;
 
    p_threadSafeMutex->lock();
    requestedBrick = new Brick(*p_managedCubes->value(cubeId).second, es
                                     - ss + 1, el - sl + 1, 1);
    requestedBrick->SetBasePosition(ss, sl, band);
    p_threadSafeMutex->unlock();
 
    // See if we already have this brick
    int instance = 0;
    int exactIndex = -1;
    bool exactMatch = false;
    int index = OverlapIndex(requestedBrick, cubeId, instance, exactMatch);
 
    // while overlaps are found
    while (index != -1) {
      if (sharedLock) {
        // make sure we can get read locks on exact overlaps
        //  We need to try to get the lock to verify partial overlaps not
        //   write locked and only keep read locks on exact matches.
        AcquireLock((*p_managedData)[index].first, true);
        if (!exactMatch) {
          (*p_managedData)[index].first->unlock();
        }
      }
      else {
        AcquireLock((*p_managedData)[index].first, false);
        // we arent actually writing to this, but now we know we can delete it
        (*p_managedData)[index].first->unlock();
 
        exactMatch = false;
 
        // destroy things that overlap but arent the same when asking to write
        if (FreeBrick(index)) {
          instance--;
        }
      }
 
      if (exactMatch) {
        exactIndex = index;
      }
 
      instance++;
      index = OverlapIndex(requestedBrick, cubeId, instance, exactMatch);
    }
 
    if(p_stopping) return;
 
    if (exactIndex == -1) {
      p_threadSafeMutex->lock();
 
      p_managedCubes->value(cubeId).second->read(*requestedBrick);
 
      QPair< QReadWriteLock *, Brick * > managedDataEntry;
 
      managedDataEntry.first = new QReadWriteLock();
 
      AcquireLock(managedDataEntry.first, sharedLock);
 
      managedDataEntry.second = requestedBrick;
 
      p_managedData->append(managedDataEntry);
      p_managedDataSources->append(cubeId);
 
      exactIndex = p_managedData->size() - 1;
 
      p_threadSafeMutex->unlock();
    }
 
    if (el - sl + 1 != (*p_managedData)[exactIndex].second->LineDimension())
    {
      //abort();
    }
 
    if (sharedLock) {
      emit ReadReady(caller, cubeId, (*p_managedData)[exactIndex].second);
    }
    else {
      emit ReadWriteReady(caller, cubeId, (*p_managedData)[exactIndex].second);
    }
  }
 
 
  int CubeDataThread::FindCubeId(const Cube * cubeToFind) const {
    QMapIterator< int, QPair< bool, Cube * > > i(*p_managedCubes);
    while (i.hasNext()) {
      i.next();
      if (i.value().second == cubeToFind)
        return i.key();
    }
    throw IException(IException::Programmer,
                     "Cube does not exist in this CubeDataThread", _FILEINFO_);
  }
 
 
  void CubeDataThread::AcquireLock(QReadWriteLock *lockObject, bool readLock) {
    // This method can be called recursively (sorta).  In the "recursive"
    // cases p_currentLockWaiting is > 0.  This guarantees that locks are
    // aquired in the reverse order as they are requested, which isn't
    // particularly helpful, but it is consistent :)
 
    if (readLock) {
      while (!lockObject->tryLockForRead()) {
        // while we can't get the lock, allow other processing to happen for
        //  brief periods of time
        //
        // Give time for things to happen in other threads
        QThread::yieldCurrentThread();
 
        p_currentLocksWaiting++;
        qApp->sendPostedEvents(this, 0);
        p_currentLocksWaiting--;
 
        if(p_stopping) return;
      }
    }
    else {
      while (!lockObject->tryLockForWrite()) {
        // while we can't get the lock, allow other processing to happen for
        //  brief periods of time
        //
        // Give time for things to happen in other threads
        QThread::yieldCurrentThread();
 
        p_currentLocksWaiting++;
        qApp->sendPostedEvents(this, 0);
        p_currentLocksWaiting--;
 
        if(p_stopping) return;
      }
    }
  }
 
  void CubeDataThread::ReadCube(int cubeId, int startSample, int startLine,
                                int endSample, int endLine, int band,
                                void *caller) {
 
    if(!p_managedCubes->contains(cubeId)) {
      IString msg = "cube ID [";
      msg += IString(cubeId);
      msg += "] is not a valid cube ID";
      throw IException(IException::Programmer, msg, _FILEINFO_);
    }
 
    GetCubeData(cubeId, startSample, startLine, endSample, endLine, band,
                caller, true);
  }
 
  void CubeDataThread::ReadWriteCube(int cubeId, int startSample,
                                     int startLine, int endSample, int endLine,
                                     int band, void *caller) {
    if(!p_managedCubes->contains(cubeId)) {
      IString msg = "cube ID [";
      msg += IString(cubeId);
      msg += "] is not a valid cube ID";
      throw IException(IException::Programmer, msg, _FILEINFO_);
    }
 
    GetCubeData(cubeId, startSample, startLine, endSample, endLine, band,
                caller, false);
  }
 
  int CubeDataThread::OverlapIndex(const Brick *overlapping, int cubeId,
                                   int instanceNum, bool &exact) {
    exact = false;
 
    // Start with extracting the range of the input (search) brick
    int startSample = overlapping->Sample(0);
    int endSample = overlapping->Sample(overlapping->size() - 1);
    int startLine = overlapping->Line(0);
    int endLine = overlapping->Line(overlapping->size() - 1);
    int startBand = overlapping->Band(0);
    int endBand = overlapping->Band(overlapping->size() - 1);
 
    // Now let's search for overlaps
    ASSERT(p_managedData->size() == p_managedDataSources->size());
    for (int knownBrick = 0; knownBrick < p_managedData->size(); knownBrick++) {
      int sourceCube = (*p_managedDataSources)[knownBrick];
 
      // Ignore other cubes; they can't overlap
      if (sourceCube != cubeId)
        continue;
 
      QPair< QReadWriteLock *, Brick * > &managedBrick =
        (*p_managedData)[knownBrick];
 
      Brick &brick = *managedBrick.second;
 
      // Get the range of this brick we've found in memory to see if any overlap
      //   exists
      int compareSampStart = brick.Sample(0);
      int compareSampEnd = brick.Sample(brick.size() - 1);
      int compareLineStart = brick.Line(0);
      int compareLineEnd = brick.Line(brick.size() - 1);
      int compareBandStart = brick.Band(0);
      int compareBandEnd = brick.Band(brick.size() - 1);
 
      bool overlap = false;
 
      // sample start is inside our sample range
      if (compareSampStart >= startSample && compareSampStart <= endSample) {
        overlap = true;
      }
 
      // sample end is inside our sample range
      if (compareSampEnd >= startSample && compareSampEnd <= endSample) {
        overlap = true;
      }
 
      // line start is in our line range
      if (compareLineStart >= startLine && compareLineStart <= endLine) {
        overlap = true;
      }
 
      // line end is in our line range
      if (compareLineEnd >= startLine && compareLineEnd <= endLine) {
        overlap = true;
      }
 
      // band start is in our line range
      if (compareBandStart >= startBand && compareBandStart <= endBand) {
        overlap = true;
      }
 
      // band end is in our line range
      if (compareBandEnd >= startBand && compareBandEnd <= endBand) {
        overlap = true;
      }
 
      exact = false;
      if (compareSampStart == startSample && compareSampEnd == endSample
          && compareLineStart == startLine && compareLineEnd == endLine
          && compareBandStart == startBand && compareBandEnd == endBand) {
        exact = true;
      }
 
      // If we have overlap, and we're at the requested instance of overlap,
      //   return it.
      if (overlap) {
        instanceNum--;
 
        if (instanceNum < 0) {
          return knownBrick;
        }
      }
    }
 
    // None found at this instance
    return -1;
  }
 
  void CubeDataThread::DoneWithData(int cubeId, const Isis::Brick *brickDone) {
    ASSERT(brickDone != NULL);
    int instance = 0;
    bool exactMatch = false;
    bool writeLock = false;
 
    int index = OverlapIndex(brickDone, cubeId, instance, exactMatch);
    ASSERT(p_managedData->size() == p_managedDataSources->size());
    while (index != -1) {
      // If this isn't the data they're finished with, we don't care about it
      if (!exactMatch) {
        instance++;
        index = OverlapIndex(brickDone, cubeId, instance, exactMatch);
        continue;
      }
 
      // Test if we had a write lock (tryLockForRead will fail). If we had a
      //   write lock make note of it.
      if (!(*p_managedData)[index].first->tryLockForRead()) {
        if (writeLock) {
          IString msg = "Overlapping data had write locks";
          throw IException(IException::Programmer, msg, _FILEINFO_);
        }
 
        writeLock = true;
      }
      // A read lock was in place, undo the lock we just made
      else {
        if (writeLock) {
          IString msg = "Overlapping data had write locks";
          throw IException(IException::Programmer, msg, _FILEINFO_);
        }
 
        // Unlock the lock we just made
        (*p_managedData)[index].first->unlock();
      }
 
      // If we had a write lock we need to write the data to the file and
      //   notify others of the change if we have listeners.
      if (writeLock) {
        p_threadSafeMutex->lock();
        Brick cpy(*brickDone);
        p_managedCubes->value(cubeId).second->write(cpy);
        p_threadSafeMutex->unlock();
 
        // Unlock the existing lock
        (*p_managedData)[index].first->unlock();
 
        // No listeners? Remove this entry
        if (p_numChangeListeners == 0) {
          if (FreeBrick(index)) {
            // We've freed the one and only match, nobody wants to know about
            //   it, so we're done
            break;
          }
        }
        // We have listeners, lock the data the appropriate number of times and
        //   then emit the BrickChanged with a pointer
        else {
          // notify others of this change
          for (int i = 0; i < p_numChangeListeners; i++) {
            AcquireLock((*p_managedData)[index].first, true);
          }
          emit BrickChanged((*p_managedDataSources)[index],
                            (*p_managedData)[index].second);
        }
      }
      // if we had a read lock and no longer have any locks, remove data from
      //   list
      else {
        // We had the one and only (hopefully!) exact match, let's free it if
        //   we can get a write lock and be done.
        // Free original read lock
        (*p_managedData)[index].first->unlock();
 
        if ((*p_managedData)[index].first->tryLockForWrite()) {
          (*p_managedData)[index].first->unlock();
          FreeBrick(index);
        }
 
        break;
      }
 
      instance++;
      index = OverlapIndex(brickDone, cubeId, instance, exactMatch);
    }
 
    ASSERT(p_managedData->size() == p_managedDataSources->size());
  }
 
  bool CubeDataThread::FreeBrick(int brickIndex) {
    ASSERT(p_managedData->size() == p_managedDataSources->size());
 
    // make sure brick is not still being used!
    if (!(*p_managedData)[brickIndex].first->tryLockForWrite()) {
      throw IException(IException::Programmer,
                       "CubeDataThread::FreeBrick called on a locked brick",
                       _FILEINFO_);
    }
    else {
      (*p_managedData)[brickIndex].first->unlock();
    }
 
    // make sure no one is looking through p_managedData in order to delete it
    p_threadSafeMutex->lock();
 
    if (p_currentLocksWaiting == 0) {
      delete (*p_managedData)[brickIndex].first;
      delete (*p_managedData)[brickIndex].second;
 
      p_managedData->removeAt(brickIndex);
      p_managedDataSources->removeAt(brickIndex);
 
      // Try to free any leftover bricks too
      for (int i = 0; i < p_managedData->size(); i++) {
        if ((*p_managedData)[i].first->tryLockForWrite()) {
          delete (*p_managedData)[i].first;
          delete (*p_managedData)[i].second;
          p_managedData->removeAt(i);
          p_managedDataSources->removeAt(i);
          i--;
        }
      }
 
      p_threadSafeMutex->unlock();
      return true;
    }
 
    p_threadSafeMutex->unlock();
 
    // no actual free was done
    return false;
  }
 
  int CubeDataThread::BricksInMemory() {
    p_threadSafeMutex->lock();
    int numBricksInMemory = p_managedData->size();
    p_threadSafeMutex->unlock();
 
    return numBricksInMemory;
  }
 
  UniversalGroundMap *CubeDataThread::GetUniversalGroundMap(int cubeId) const {
    if (!p_managedCubes->contains(cubeId)) {
      throw IException(IException::Programmer,
                       "Invalid Cube ID [" + IString(cubeId) + "]",
                       _FILEINFO_);
    }
 
    return new UniversalGroundMap(*(*p_managedCubes)[cubeId].second);
  }
 
  const Cube *CubeDataThread::GetCube(int cubeId) const {
    if (!p_managedCubes->contains(cubeId)) {
      throw IException(IException::Programmer,
                       "Invalid Cube ID [" + IString(cubeId) + "]",
                       _FILEINFO_);
    }
 
    return (*p_managedCubes)[cubeId].second;
  }
}