creaImageIOMultiThreadImageReader.cpp

Go to the documentation of this file.

#include <creaImageIOMultiThreadImageReader.h>
#include <creaImageIOImageReader.h>
#include <wx/utils.h>
#include <creaImageIOSystem.h>

#include <creaImageIOGimmick.h>
#ifdef _DEBUG
#define new DEBUG_NEW
#endif
namespace creaImageIO
{

  //=====================================================================
  void MultiThreadImageReaderUser::MultiThreadImageReaderSendEvent
  ( const std::string& filename,
    EventType type,
    vtkImageData* image)
  {
    wxMutexLocker lock(mMultiThreadImageReaderUserMutex);

    this->OnMultiThreadImageReaderEvent(filename,type,image);
  }
  //=====================================================================

  //=====================================================================
  class ThreadedImageReader: public wxThread
  {
  public:
    ThreadedImageReader(MultiThreadImageReader* tir) :
      mMultiThreadImageReader(tir)
    {}

    void* Entry();
    void  OnExit();

    vtkImageData* Read(const std::string& filename);
    
        struct deleter
        {
                void operator()(ThreadedImageReader* p)
                {
                        p->Delete();
                }
        };
        friend struct deleter;


  private:
    ImageReader mReader;
    MultiThreadImageReader* mMultiThreadImageReader;
        
  };

  //=====================================================================

  
  //=====================================================================
  MultiThreadImageReader::MultiThreadImageReader(int number_of_threads)
    : //mDoNotSignal(false),
      mReader(0),
      mTotalMem(0),
      mTotalMemMax(1000000)
  {
    //    std::cout << "#### MultiThreadImageReader::MultiThreadImageReader("
    //        << " #threads= " << number_of_threads <<" )"<<std::endl;

          mDone = false;
    // Create the threads
    for (int i=0; i<number_of_threads; i++) 
      {
                  //ThreadedImageReader* t = new ThreadedImageReader(this);
                  boost::shared_ptr<ThreadedImageReader> t(new ThreadedImageReader(this), ThreadedImageReader::deleter());
        mThreadedImageReaderList.push_back(t);
         std::cout << "  ===> Thread "<<i
                      <<" successfully added"<< std::endl;
      }
    mNumberOfThreadedReadersRunning = 0;
    // Init the queue
    mQueue.set(mComparator);
    mQueue.set(mIndexer);
    // 
    // no thread : alloc self reader
//    if (number_of_threads==0)
//      {
        mReader = new ImageReader();
//      }
  }
  //=====================================================================


  //=====================================================================
  bool MultiThreadImageReader::Start()
  {

    //    std::cout << "#### MultiThreadImageReader::Start()"
    //                <<std::endl;
          if (mNumberOfThreadedReadersRunning > 0) return true;
          
    ThreadedImageReaderListType::iterator i;
    for (i =mThreadedImageReaderList.begin();
         i!=mThreadedImageReaderList.end();
         i++)
      {
        (*i)->Create();
        if ( (*i)->Run() != wxTHREAD_NO_ERROR )
          {
            std::cout << "ERROR starting a thread"<< std::endl;
            return false;
          }
        else 
          {
                    std::cout << "  ===> Thread "<<(*i)->GetCurrentId()
                              <<" successfully created"<< std::endl;
            
          }
      }
    wxMutexLocker locker(GetMultiThreadImageReaderUserMutex());
    //    std::cout << "EO Start : #Threads running = "
    //                << mNumberOfThreadedReadersRunning<<std::endl;

    return true;
  }
  //=====================================================================

  //=====================================================================
  void MultiThreadImageReader::Stop()
  { 
//                  std::cout << "#### MultiThreadImageReader::Stop()"
//            <<std::endl;
  //  std::cout << "Sending stop order to the threads..."<<std::endl;
          if (mDone) return;

    ThreadedImageReaderListType::iterator i;
    for (i =mThreadedImageReaderList.begin();
         i!=mThreadedImageReaderList.end();
         i++)
      { std::cout << "  ===> Thread "<<(*i)->GetCurrentId()
                              <<" successfully stopped"<< std::endl;
                  if((*i)->IsAlive())
                  {(*i)->Pause();
                          (*i).reset();
                         //                       (*i)->Delete();
                  }
      }
   mThreadedImageReaderList.clear();
    // Wait a little to be sure that all threads have stopped
    // A better way to do this ?
    //    wxMilliSleep(1000);
    // New method : the threads generate a stop event when they have finished
    // We wait until all threads have stopped
//        std::cout << "Waiting for stop signals..."<<std::endl;
    do 
      {
        // Sleep a little
                wxMilliSleep(10);
        // Lock
        {
          wxMutexLocker locker(GetMultiThreadImageReaderUserMutex());
//                std::cout << "#Threads running = "
//                          << mNumberOfThreadedReadersRunning<<std::endl;
          // Break if all readers have stopped
          if (mNumberOfThreadedReadersRunning <= 0) 
            {
              break;
            }
        }
      } 
    while (true);
//        std::cout << "All threads stopped : OK "<<std::endl;

    ImageMapType::iterator j;
    for (j =mImages.begin();
         j!=mImages.end();
         ++j)

      {
        delete j->first;
      }
    mImages.clear();
        mDone = true;
  }
  //=====================================================================

  //=====================================================================
  MultiThreadImageReader::~MultiThreadImageReader()
  {
    //    std::cout << "#### MultiThreadImageReader::~MultiThreadImageReader()"
    //        <<std::endl;
    Stop();
    if (mReader) delete mReader;
        mThreadedImageReaderList.clear();
  }
  //=====================================================================

  //=====================================================================
  void MultiThreadImageReader::UpdateUnloadPriority(ImageToLoadPtr p, 
                                                    int priority)
  {
    // not in unload queue : ciao
    if (p->UnloadIndex()<0) return;
    int old_prio = p->GetPriority();
    if (priority > old_prio) 
      {
        p->SetPriority(priority);
        mUnloadQueue.downsort(p->UnloadIndex());
      }
    else if ( old_prio > priority )
      {
        p->SetPriority(priority);
        mUnloadQueue.upsort(p->UnloadIndex());
     }
  }
  //=====================================================================
  // function to read attributes for a file
  void MultiThreadImageReader::getAttributes(const std::string filename, 
          std::map <std::string , std::string> &infos,std::vector<std::string> i_attr)
  {
          mReader->getAttributes(filename, infos, i_attr);
  }

  //=====================================================================
  void MultiThreadImageReader::Request( MultiThreadImageReaderUser* user,
                                        const std::string& filename, 
                                        int priority )
  {
        wxMutexLocker lock(GetMultiThreadImageReaderUserMutex()); //mMutex);

          if (mNumberOfThreadedReadersRunning==0)
//    if (mThreadedImageReaderList.size()==0) 
      {
        // no detached reader : use self reader
        ImageToLoad itl(user,filename);
        ImageMapType::iterator i = mImages.find(&itl);
        if (i!=mImages.end())
          {
            ImageToLoadPtr pitl = const_cast<ImageToLoadPtr>(i->first);
            // Already inserted
            if (pitl->GetImage() != 0)
              {
                // Already read
                pitl->SetUser(user);
                UpdateUnloadPriority(pitl,priority);
                SignalImageRead(pitl,false);
                return; // pitl->GetImage();
              }
          }
        ImageToLoadPtr pitl = new ImageToLoad(user,filename,0);
        mImages[pitl] = 0;
        pitl->SetImage(mReader->ReadImage(filename));
        UpdateUnloadPriority(pitl,priority);
        SignalImageRead(pitl,true);
        //      return pitl->GetImage();
        return;
      }

    ImageToLoad itl(user,filename);
    ImageMapType::iterator i = mImages.find(&itl);
    if (i!=mImages.end())
      {
        // Already inserted
        if (i->first->GetImage() != 0)
          {
            // Already read : ok :signal the user
            UpdateUnloadPriority(i->first,priority);
            SignalImageRead(i->first,false);
            return;
          }
        ImageToLoadPtr pitl = const_cast<ImageToLoadPtr>(i->first);
        pitl->SetPriority(priority);
        // Already in queue
        if (pitl->Index()>=0) 
          {
            // Re-sort the queue
            mQueue.upsort(pitl->Index());
          }
        // Not read but not in queue = being read = ok
        else 
          {
            
          }
      }
    else 
      {
        // Never requested before or unloaded 
        ImageToLoadPtr pitl = new ImageToLoad(user,filename,priority);
        mImages[pitl] = 0;
        mQueue.insert(pitl);
      }
  }
  //=====================================================================

  //=====================================================================
  void MultiThreadImageReader::OnMultiThreadImageReaderEvent
  (const std::string& filename,
   MultiThreadImageReaderUser::EventType e,
   vtkImageData* image)
  {
    if ((e==MultiThreadImageReaderUser::ImageLoaded) &&
        (filename == mRequestedFilename))
      {
        mRequestedImage = image;
      }
    else if (e==MultiThreadImageReaderUser::ThreadedReaderStarted)
      {
        mNumberOfThreadedReadersRunning++;
        //      std::cout << "#TR=" << mNumberOfThreadedReadersRunning << std::endl;
      }
    else if (e==MultiThreadImageReaderUser::ThreadedReaderStopped)
      {
        
                 mNumberOfThreadedReadersRunning--;
        //      std::cout << "#TR=" << mNumberOfThreadedReadersRunning << std::endl;
      }
  }
  //=====================================================================

  //=====================================================================
  vtkImageData* MultiThreadImageReader::GetImage(const std::string& filename)
  {
         // Start();
    //       std::cout << "** MultiThreadImageReader::GetImage('"<<filename<<"')"
    //           <<std::endl;
    
    do 
      {
        //      wxMutexLocker lock(GetMultiThreadImageReaderUserMutex()); //mMutex);
                
        //     std::cout << "** MultiThreadImageReader::GetImage('"<<filename
        //             <<"') lock ok"
        //               <<std::endl;
    
        //                if (mNumberOfThreadedReadersRunning==0)
        //      if (mThreadedImageReaderList.size()==0)
        if (true)
          {
            ImageToLoad itl(this,filename);
            ImageMapType::iterator i = mImages.find(&itl);
            if (i!=mImages.end())
              {
                ImageToLoadPtr pitl = const_cast<ImageToLoadPtr>(i->first);
                // Already inserted
                if (pitl->GetImage() != 0)
                  {
                    // Already read
                    UpdateUnloadPriority(pitl,
                                         GetMaximalPriorityWithoutLocking()+1);
                    return pitl->GetImage();
                  }
              }
            ImageToLoadPtr pitl = new ImageToLoad(this,filename,0);
            mImages[pitl] = 0;
            pitl->SetImage(mReader->ReadImage(filename));
            UpdateUnloadPriority(pitl,
                                 GetMaximalPriorityWithoutLocking()+1);
            return pitl->GetImage();
          }

        /*      
        mRequestedFilename = filename;
        mRequestedImage = 0;
        ImageToLoad itl(this,filename);
        ImageMapType::iterator i = mImages.find(&itl);
        if (i!=mImages.end())
          {
            // Already inserted in queue
            if (i->first->GetImage() != 0)
              {
                // Already read : ok : return it 
                return i->first->GetImage();
              }
              ImageToLoadPtr pitl = const_cast<ImageToLoadPtr>(i->first);
              pitl->SetPriority( GetMaximalPriorityWithoutLocking() + 1 );
              pitl->SetUser( this );
              // Already in queue
              if (pitl->Index()>=0) 
                {
                  // Re-sort the queue
                  mQueue.upsort(pitl->Index());
                }
              // Not read but not in queue = being read = ok
              else 
                {
                  pitl->SetUser( this );
                }
          }
        else 
          {
            
            // Never requested before or unloaded 
            ImageToLoadPtr pitl = 
              new ImageToLoad(this,filename,
                              GetMaximalPriorityWithoutLocking() + 1);
            mImages[pitl] = 0;
            mQueue.insert(pitl);
          }
        */
      }
    while (0);

    //    std::cout << "Waiting..."<<std::endl;

    /*
    // Waiting that it is read
    int n = 0;
    do 
      {
        //      std::cout << n++ << std::endl;
        wxMilliSleep(10);
        do 
          {
            //      wxMutexLocker lock(mMutex);
            wxMutexLocker lock(GetMultiThreadImageReaderUserMutex());
            if (mRequestedImage!=0) 
              {
                return mRequestedImage;
              } 
          }
        while (0);
      }
    while (true);
    // 
    */
  }
  //=====================================================================
  
  //=====================================================================
  void MultiThreadImageReader::SignalImageRead(ImageToLoadPtr p, 
                                               bool purge)
  {
    
//    std::cout << "MultiThreadImageReader::SignalImageRead" <<std::endl;
    //    std::cout << "this="<<this <<std::endl;
    //    std::cout << "user="<<p->GetUser() <<std::endl;

    if ( p->GetUser() == this ) 
      GetMultiThreadImageReaderUserMutex().Unlock();

    p->GetUser()->MultiThreadImageReaderSendEvent
      (p->GetFilename(),
       MultiThreadImageReaderUser::ImageLoaded,
       p->GetImage());

    /*
      AN ATTEMPT TO UNLOAD OLDEST IMAGE IF EXCEEDED A CERTAIN MEMORY QUOTA
      BUGGY : TO FIX 
    */
    if (!purge)  return;
    GimmickMessage(5,"Image '"<<p->GetFilename()<<"' read"<<std::endl);

    //    wxMutexLocker lock(GetMultiThreadImageReaderUserMutex());
           
    mUnloadQueue.insert(p);
    p->GetImage()->UpdateInformation();
    p->GetImage()->PropagateUpdateExtent();
    long ImMem = p->GetImage()->GetEstimatedMemorySize();
    mTotalMem += ImMem;

    GimmickMessage(5,"==> Image in memory = "<<mUnloadQueue.size()<<std::endl);
    GimmickMessage(5,"==> Total mem       = "<<mTotalMem<<" Ko"<<std::endl);

    //  return;

    while (mTotalMem > mTotalMemMax)
      {
        GimmickMessage(5,
                       "   ! Exceeded max of "
                       << mTotalMemMax << " Ko : unloading oldest image ... "
                       << std::endl);
        if ( mUnloadQueue.size() <= 1 ) 
          {
             GimmickMessage(5,
                            "   Only one image : cannot load AND unload it !!"
                            <<std::endl);
            break; 
            
          }
        ImageToLoadPtr unload = mUnloadQueue.remove_top();
        MultiThreadImageReaderUser* user = unload->GetUser();

        /*
        if ((user!=0)&&(user!=this)) 
          {
            user->GetMultiThreadImageReaderUserMutex().Lock();
          }
        */

        std::string filename = unload->GetFilename();

        GimmickMessage(5,"'" << filename << "'" << std::endl);
        mTotalMem -= unload->GetImage()->GetEstimatedMemorySize();

        GimmickMessage(5," ==> Total mem = "<<mTotalMem<<" Ko "<<std::endl);

        if (user!=0) 
          {
            //      std::cout << "unlock..."<<std::endl;
            //   user->GetMultiThreadImageReaderUserMutex().Unlock();
            //      std::cout << "event"<<std::endl;
            user->MultiThreadImageReaderSendEvent
              (filename,
               MultiThreadImageReaderUser::ImageUnloaded,
               0);
            //      std::cout << "event ok"<<std::endl;
          }     

        if (unload->Index()>=0)
          {
            // GimmickMessage(5,"still in queue"<<std::endl);
          }
        unload->Index() = -1;


        ImageMapType::iterator it = mImages.find(unload);
        if (it!=mImages.end())
          {
            mImages.erase(it);
          }
        //          std::cout << "delete..."<<std::endl;
        delete unload;
        //          std::cout << "delete ok."<<std::endl;

      }
  }
  //=====================================================================

  //=====================================================================
  int MultiThreadImageReader::GetMaximalPriority()
  { 
    wxMutexLocker lock(GetMultiThreadImageReaderUserMutex()); //mMutex);
    return GetMaximalPriorityWithoutLocking();
  }
  //=====================================================================


  //=====================================================================
  int MultiThreadImageReader::GetMaximalPriorityWithoutLocking()
  { 
    long max = 0;
    if (mQueue.size()>0) 
      {
        max = mQueue.top()->GetPriority();
      }
    if (mUnloadQueue.size()>0)
      {
        int max2 = mUnloadQueue.top()->GetPriority();
        if (max2>max) max=max2;
      }
    return max;
  }
  //=====================================================================


  //=====================================================================
  //=====================================================================
  //=====================================================================
  //=====================================================================

  //=====================================================================
  void*  ThreadedImageReader::Entry()
  {
    //    std::cout << "### Thread "<<GetCurrentId()<<"::Entry()"
    //                << std::endl;

    mMultiThreadImageReader->MultiThreadImageReaderSendEvent
      ("",
       MultiThreadImageReaderUser::ThreadedReaderStarted,
       0);

    // While was not deleted 
    while (!TestDestroy())
      {
                //std::cout << "### Thread "<<GetCurrentId()<<" still alive"  << std::endl;
          
        // Lock the mutex
        mMultiThreadImageReader->MultiThreadImageReaderEventLock();
        //mMutex.Lock();
        // If image in queue
        if (mMultiThreadImageReader->mQueue.size()>0)
          {
            MultiThreadImageReader::ImageToLoadPtr i = 
              mMultiThreadImageReader->mQueue.remove_top();

            mMultiThreadImageReader->MultiThreadImageReaderEventUnlock();
            //mMutex.Unlock();

            
            //      std::cout << "### Thread "<<GetCurrentId()<<" : reading '"
            //                << i->GetFilename() << "'" << std::endl;
            
            // Do the job
            vtkImageData* im = Read(i->GetFilename());

            // Store it in the map
            mMultiThreadImageReader->MultiThreadImageReaderEventLock();
            //mMutex.Lock();
            MultiThreadImageReader::ImageToLoad itl(0,i->GetFilename());
            MultiThreadImageReader::ImageMapType::iterator it = 
              mMultiThreadImageReader->mImages.find(&itl);
            MultiThreadImageReader::ImageToLoadPtr 
              pitl = const_cast<MultiThreadImageReader::ImageToLoadPtr>
              (it->first);
            pitl->SetImage(im);
            mMultiThreadImageReader->SignalImageRead(pitl,true);//i->GetFilename());
            mMultiThreadImageReader->MultiThreadImageReaderEventUnlock();           //mMutex.Unlock();
            
            //      std::cout << "### Thread "<<GetCurrentId()<<" : reading '"
            //                << i->GetFilename() << "' : DONE" << std::endl;
            
          }
        else 
          {
            mMultiThreadImageReader->MultiThreadImageReaderEventUnlock();
            //mMutex.Unlock();
            // Wait a little to avoid blocking 
            Sleep(10);
          }
      };
    //    std::cout << "### Thread "<<GetCurrentId()<<" stopping"
    //                << std::endl;
       
    return 0;
  }
  //=====================================================================

  //=====================================================================
  void ThreadedImageReader::OnExit()
  {
    mMultiThreadImageReader->MultiThreadImageReaderSendEvent
      ("",
       MultiThreadImageReaderUser::ThreadedReaderStopped,
       0);
  }
  //=====================================================================

  //=====================================================================
  vtkImageData* ThreadedImageReader::Read(const std::string& filename)
  {
    return mReader.ReadImage(filename);
  }
  //=====================================================================

} // namespace creaImageIO