Improve thumbnail parallelism

rtgui/filebrowserentry.cc

@@ -56,14 +56,20 @@ FileBrowserEntry::FileBrowserEntry (Thumbnail* thm, const Glib::ustring& fname)
 
 FileBrowserEntry::~FileBrowserEntry () {
 
-    thumbImageUpdater.removeJobs (this);
+	// so jobs arriving now do nothing
+    if (feih->pending)
+	{
+        feih->destroyed = true;
+	}
+    else
+	{
+        delete feih;
+		feih = 0;
+	}
+
+    thumbImageUpdater->removeJobs (this);
     if (thumbnail)
         thumbnail->removeThumbnailListener (this);
-
-    if (feih->pending)
-        feih->destroyed = true;
-    else
-        delete feih;
 }
 
 void FileBrowserEntry::refreshThumbnailImage () {
@@ -71,8 +77,7 @@ void FileBrowserEntry::refreshThumbnailImage () {
     if (!thumbnail)
         return;
 
-    thumbImageUpdater.add (thumbnail, thumbnail->getProcParams(), preh, &updatepriority, this);    
+    thumbImageUpdater->add (thumbnail, thumbnail->getProcParams(), preh, &updatepriority, this);    
-    thumbImageUpdater.process ();
 }
 
 void FileBrowserEntry::calcThumbnailSize () {
@@ -134,10 +139,11 @@ struct tiupdate {
 
 int fbeupdate (void* data) {
     
-    gdk_threads_enter ();
     tiupdate* params = (tiupdate*)data;
     FileBrowserEntryIdleHelper* feih = params->feih;
 
+	GThreadLock lock;
+
     if (feih->destroyed) {
         if (feih->pending == 1)
             delete feih;
@@ -145,14 +151,12 @@ int fbeupdate (void* data) {
             feih->pending--;
         params->img->free ();
         delete params;
-        gdk_threads_leave ();
         return 0;
     }
     
     feih->fbentry->_updateImage (params->img, params->scale, params->cropParams);
     feih->pending--;
 
-    gdk_threads_leave ();
     delete params;
     
     return 0;
@@ -160,8 +164,19 @@ int fbeupdate (void* data) {
 
 void FileBrowserEntry::updateImage (rtengine::IImage8* img, double scale, rtengine::procparams::CropParams cropParams) {
 
-    redrawRequests++;
+	{
-    feih->pending++;
+		GThreadLock lock;
+
+		if ( feih == 0 ||
+				feih->destroyed )
+		{
+			return;
+		}
+
+		redrawRequests++;
+		feih->pending++;
+	}
+
     tiupdate* param = new tiupdate ();
     param->feih = feih;
     param->img = img;

rtgui/filecatalog.cc

@@ -228,7 +228,7 @@ void FileCatalog::closeDir () {
     previewLoader.terminate ();
 
     // terminate thumbnail updater
-    thumbImageUpdater.terminate ();
+    thumbImageUpdater->removeAllJobs ();
 
     // remove entries
     fileBrowser->close ();  
@@ -427,14 +427,12 @@ void FileCatalog::_openImage (std::vector<Thumbnail*> tmb) {
 
     if (enabled && listener!=NULL) {
         previewLoader.stop ();
-        thumbImageUpdater.stop ();
         for (size_t i=0; i<tmb.size(); i++) {
             if (editedFiles.find (tmb[i]->getFileName())==editedFiles.end())
                 listener->fileSelected (tmb[i]);
             tmb[i]->decreaseRef ();
         }
         previewLoader.process ();
-        thumbImageUpdater.process ();
     }
 }
 
@@ -496,7 +494,6 @@ void FileCatalog::deleteRequested  (std::vector<FileBrowserEntry*> tbe) {
 void FileCatalog::developRequested (std::vector<FileBrowserEntry*> tbe) {
 
     if (listener) {
-        thumbImageUpdater.stop ();
         for (size_t i=0; i<tbe.size(); i++) {
             rtengine::procparams::ProcParams params = tbe[i]->thumbnail->getProcParams();
             rtengine::ProcessingJob* pjob = rtengine::ProcessingJob::create (tbe[i]->filename, tbe[i]->thumbnail->getType()==FT_Raw, params);
@@ -516,7 +513,6 @@ void FileCatalog::developRequested (std::vector<FileBrowserEntry*> tbe) {
                 listener->addBatchQueueJob (new BatchQueueEntry (pjob, params, tbe[i]->filename, NULL, pw, ph, tbe[i]->thumbnail));
             }
         }
-        thumbImageUpdater.process ();
     }
 }
 

rtgui/guiutils.h

@@ -16,8 +16,8 @@
  *  You should have received a copy of the GNU General Public License
  *  along with RawTherapee.  If not, see <http://www.gnu.org/licenses/>.
  */
-#ifndef __UTILS_
+#ifndef __GUI_UTILS_
-#define __UTILS_
+#define __GUI_UTILS_
 
 #include <gtkmm.h>
 #include <rtengine.h>
@@ -28,4 +28,47 @@ Glib::ustring removeExtension (const Glib::ustring& filename);
 Glib::ustring getExtension (const Glib::ustring& filename);
 void drawCrop (Cairo::RefPtr<Cairo::Context> cr, int imx, int imy, int imw, int imh, int startx, int starty, double scale, const rtengine::procparams::CropParams& cparams);
 
+/** 
+ * @brief Lock GTK for critical section.
+ *
+ * Will unlock on destruction. To use:
+ *
+ *   <code>
+ *     {
+ *       GThreadLock lock;
+ *       // critical code
+ *     }
+ *   </code>
+ */
+class GThreadLock
+{
+public:
+	GThreadLock()
+	{
+		gdk_threads_enter();
+	}
+	~GThreadLock()
+	{
+		gdk_threads_leave();
+	}
+};
+
+/** 
+ * @brief Unlock GTK critical section.
+ *
+ * Will relock on destruction.
+ */
+class GThreadUnLock
+{
+public:
+	GThreadUnLock()
+	{
+		gdk_threads_leave();
+	}
+	~GThreadUnLock()
+	{
+		gdk_threads_enter();
+	}
+};
+
 #endif

rtgui/thumbimageupdater.cc

@@ -16,150 +16,240 @@
  *  You should have received a copy of the GNU General Public License
  *  along with RawTherapee.  If not, see <http://www.gnu.org/licenses/>.
  */
+
+#include <set>
 #include <thumbimageupdater.h>
 #include <gtkmm.h>
+#include <guiutils.h>
 
-#define threadNum 1
+#define THREAD_NUM 4
-ThumbImageUpdater thumbImageUpdater;
 
-ThumbImageUpdater::ThumbImageUpdater () 
+#define DEBUG(format,args...) 
-    : tostop(false), stopped(true), qMutex(NULL), startMutex(NULL), threadPool(NULL) {
+//#define DEBUG(format,args...) printf("ThumbImageUpdate::%s: " format "\n", __FUNCTION__, ## args)
-        
-}
 
-ThumbImageUpdater::~ThumbImageUpdater ()
+struct 
+Job
 {
-    delete threadPool;
+	Job(Thumbnail* thumbnail, const rtengine::procparams::ProcParams& pparams,
+				int height, bool* priority,
+				ThumbImageUpdateListener* listener):
+		thumbnail_(thumbnail),
+		pparams_(pparams),
+		height_(height),
+		priority_(priority),
+		listener_(listener)
+	{}
+
+	Job():
+		thumbnail_(0),
+		listener_(0)
+	{}
+
+	Thumbnail* thumbnail_;
+	rtengine::procparams::ProcParams pparams_;
+	int height_;
+	bool* priority_;
+	ThumbImageUpdateListener* listener_;
+};
+
+typedef std::list<Job> JobList;
+
+class
+ThumbImageUpdater::Impl
+{
+public:
+	Impl():
+		threadPool_(new Glib::ThreadPool(THREAD_NUM,0)),
+		active_(0),
+		inactive_waiting_(false)
+	{}
+
+	Glib::ThreadPool* threadPool_;
+
+	Glib::Mutex mutex_;
+
+	JobList jobs_;
+
+	unsigned int active_;
+
+	bool inactive_waiting_;
+
+	Glib::Cond inactive_;
+
+	void
+	processNextJob(void)
+	{ 
+		Job j;
+
+		{
+			Glib::Mutex::Lock lock(mutex_);
+
+			// nothing to do; could be jobs have been removed
+			if ( jobs_.empty() )
+			{
+				DEBUG("processing: nothing to do (%d,%d)",paused_,jobs_.empty());
+				return;
+			}
+
+			JobList::iterator i;
+
+			// see if any priority jobs exist
+			for ( i = jobs_.begin(); i != jobs_.end(); ++i)
+			{
+				if ( *(i->priority_) )
+				{
+					DEBUG("processing(priority) %s",i->thumbnail_->getFileName().c_str());
+					break;
+				}
+			}
+
+			// if none, then use first
+			if ( i == jobs_.end() )
+			{
+				i = jobs_.begin();
+				DEBUG("processing(first) %s",i->thumbnail_->getFileName().c_str());
+			}
+
+			// copy found job
+			j = *i;
+
+			// remove so not run again
+			jobs_.erase(i);
+			DEBUG("%d job(s) remaining",jobs_.size());
+
+			++active_;
+		}
+
+		// unlock and do processing; will relock on block exit, then call listener
+		double scale = 1.0;
+		rtengine::IImage8* img = 0;
+		{
+			img = j.thumbnail_->processThumbImage(j.pparams_, j.height_, scale);
+		}
+
+		if (img)
+		{
+			DEBUG("pushing image %s",j.thumbnail_->getFileName().c_str());
+			j.listener_->updateImage(img, scale, j.pparams_.crop);
+		}
+
+		{
+			Glib::Mutex::Lock lock(mutex_);
+
+			if ( --active_ == 0 &&
+					inactive_waiting_ )
+			{
+				inactive_waiting_ = false;
+				inactive_.signal();
+			}
+		}
+	}
+};
+
+ThumbImageUpdater*
+ThumbImageUpdater::getInstance(void)
+{
+	// this will not be deleted...
+	static ThumbImageUpdater* instance_ = 0;
+	if ( instance_ == 0 )
+	{
+		instance_ = new ThumbImageUpdater();
+	}
+	return instance_;
 }
 
-void ThumbImageUpdater::add (Thumbnail* t, const rtengine::procparams::ProcParams& params, int height, bool* priority, ThumbImageUpdateListener* l) {
+ThumbImageUpdater::ThumbImageUpdater():
+	impl_(new Impl())
+{
+}
 
-    if (!qMutex)
+void 
-        qMutex = new Glib::Mutex ();
+ThumbImageUpdater::add(Thumbnail* t, const rtengine::procparams::ProcParams& params,
-    if (!startMutex)
+							int height, bool* priority, ThumbImageUpdateListener* l) 
-        startMutex = new Glib::Mutex ();
+{
+	// nobody listening?
+	if ( l == 0 )
+	{
+		return;
+	}
+
+	Glib::Mutex::Lock lock(impl_->mutex_);
 
-    qMutex->lock ();
     // look up if an older version is in the queue
-    std::list<Job>::iterator i;
+    JobList::iterator i(impl_->jobs_.begin());
-    for (i=jqueue.begin(); i!=jqueue.end(); i++)
+	for ( ; i != impl_->jobs_.end(); ++i )
-        if (i->thumbnail==t && i->listener==l) {
+	{
-            i->pparams = params;
+		if ( i->thumbnail_ == t &&
-            i->height = height;
+				i->listener_ == l )
-            i->priority = priority;
+		{
-            break;
+			DEBUG("updating job %s",t->getFileName().c_str());
-        }
+			// we have one, update queue entry, will be picked up by thread when processed
-    // not found, create and append new job
+			i->pparams_ = params;
-    if (i==jqueue.end ()) {
+			i->height_ = height;
-        Job j;
+			i->priority_ = priority;
-        j.thumbnail = t;
+			return;
-        j.pparams = params;
+		}
-        j.height = height;
+	}
-        j.listener = l;
-        j.priority = priority;
-        jqueue.push_back (j);
-    }
-    qMutex->unlock ();
-}
 
-void ThumbImageUpdater::process () {
+	// create a new job and append to queue
+	DEBUG("queing job %s",t->getFileName().c_str());
+	impl_->jobs_.push_back(Job(t,params,height,priority,l));
 
-    if (stopped) {
+	DEBUG("adding run request %s",t->getFileName().c_str());
-        stopped = false;
+	impl_->threadPool_->push(sigc::mem_fun(*impl_, &ThumbImageUpdater::Impl::processNextJob));
-
-        if(!threadPool)
-            threadPool = new Glib::ThreadPool(threadNum,1);
-     
-        //thread = Glib::Thread::create (sigc::mem_fun(*this, &ThumbImageUpdater::process_), (unsigned long int)0, true, true, Glib::THREAD_PRIORITY_NORMAL);
-        process_();
-    }
-}
-
-void ThumbImageUpdater::process_ () { 
-
-    stopped = false;
-    tostop = false;
-
-    while (!tostop && !jqueue.empty ()) {
-
-            std::list<Job>::iterator i;
-            for (i=jqueue.begin (); i!=jqueue.end(); i++)
-                if (*(i->priority))
-                    break;
-
-            if (i==jqueue.end())
-                i = jqueue.begin();
-
-            Job current = *i;         
-            if (current.listener)                    
-                    threadPool->push(sigc::bind(sigc::mem_fun(*this, &ThumbImageUpdater::processJob), current));
-
-            jqueue.erase (i);
-    }
-    
-    stopped = true;
-    //printf("Threads # %d \n", threadPool->get_num_threads());
-
-}
-
-void ThumbImageUpdater::processJob (Job current) {
-
-    if (current.listener) {
-        double scale = 1.0;
-        rtengine::IImage8* img = current.thumbnail->processThumbImage (current.pparams, current.height, scale);
-        if (img)
-            current.listener->updateImage (img, scale, current.pparams.crop);
-    }
-
-}
-
-void ThumbImageUpdater::stop () {
-        
-    gdk_threads_leave(); 
-    tostop = true;
-
-    if (threadPool)    {
-        threadPool->shutdown(TRUE);
-        threadPool = NULL;
-    }
-    gdk_threads_enter();
-}
-
-void ThumbImageUpdater::removeJobs () {
-
-    if (!qMutex)
-        return;
-
-    qMutex->lock ();
-    while (!jqueue.empty()) 
-        jqueue.pop_front (); 
-    qMutex->unlock ();
-}
-
-void ThumbImageUpdater::removeJobs (ThumbImageUpdateListener* listener) {
-
-    if (!qMutex)
-        return;
-
-    qMutex->lock ();
-    bool ready = false;
-    while (!ready) {
-        ready = true;
-        std::list<Job>::iterator i;
-        for (i=jqueue.begin(); i!=jqueue.end(); i++)
-            if (i->listener == listener) {
-                jqueue.erase (i);
-                ready = false;
-                break;
-            }
-    }
-    qMutex->unlock ();
-}
-
-void ThumbImageUpdater::terminate  () { 
-
-    stop (); 
-    removeJobs (); 
 }
 
 
+void 
+ThumbImageUpdater::removeJobs(ThumbImageUpdateListener* listener)
+{
+	DEBUG("removeJobs(%p)",listener);
+
+	Glib::Mutex::Lock lock(impl_->mutex_);
+
+	for( JobList::iterator i(impl_->jobs_.begin()); i != impl_->jobs_.end(); )
+	{
+		if (i->listener_ == listener)
+		{
+			DEBUG("erasing specific job");
+			JobList::iterator e(i++);
+			impl_->jobs_.erase(e);
+		}
+		else
+		{
+			++i;
+		}
+	}
+
+	while ( impl_->active_ != 0 )
+	{
+		// XXX this is nasty... it would be nicer if we weren't called with
+		// this lock held
+		GThreadUnLock unlock;
+		DEBUG("waiting for running jobs1");
+		impl_->inactive_waiting_ = true;
+		impl_->inactive_.wait(impl_->mutex_);
+	}
+}
+
+void 
+ThumbImageUpdater::removeAllJobs(void) 
+{ 
+	DEBUG("stop");
+
+
+	Glib::Mutex::Lock lock(impl_->mutex_);
+
+	impl_->jobs_.clear();
+
+	while ( impl_->active_ != 0 )
+	{
+		// XXX this is nasty... it would be nicer if we weren't called with
+		// this lock held
+		GThreadUnLock unlock;
+		DEBUG("waiting for running jobs2");
+		impl_->inactive_waiting_ = true;
+		impl_->inactive_.wait(impl_->mutex_);
+	}
+}
+

rtgui/thumbimageupdater.h

@@ -26,45 +26,76 @@
 
 class ThumbImageUpdateListener {
 
-    public:
+public:
-        virtual void updateImage (rtengine::IImage8* img, double scale, rtengine::procparams::CropParams cropParams) {}
+
+	/** 
+	 * @brief Called when thumbnail image is update
+	 * 
+	 * @param img new thumbnail image
+	 * @param scale scale (??)
+	 * @param cropParams how it was cropped (??)
+	 *
+	 * @note no locks are held when called back
+	 */
+	virtual void updateImage (rtengine::IImage8* img, double scale, rtengine::procparams::CropParams cropParams) {}
 };
 
 class ThumbImageUpdater {
 
-    struct Job {
-        Thumbnail* thumbnail;
-        rtengine::procparams::ProcParams pparams;
-        int height;
-        bool* priority;
-        ThumbImageUpdateListener* listener;
-    };
-
-  protected:
-    bool tostop;
-    bool stopped;
-    std::list<Job> jqueue;
-    Glib::Thread* thread;
-    Glib::Mutex* qMutex;
-    Glib::Mutex* startMutex;
-    //Glib::Thread **threadPool;
-    Glib::ThreadPool * threadPool;
-
   public:
-    ThumbImageUpdater ();
-    ~ThumbImageUpdater ();
 
-    void add        (Thumbnail* t, const rtengine::procparams::ProcParams& params, int height, bool* priority, ThumbImageUpdateListener* l);
+	/** 
-    void process    ();
+	 * @brief Singleton entry point.
-    void stop       ();
+	 * 
-    void removeJobs ();
+	 * @return Pointer to thumbnail image updater.
-    void removeJobs (ThumbImageUpdateListener* listener);
+	 */
-    void terminate  ();
+	static ThumbImageUpdater* getInstance(void);
 
-    void process_   ();
+	/** 
-    void processJob (Job current);
+	 * @brief Add an thumbnail image update request.
+	 *
+	 * Code will add the request to the queue and, if needed, start a pool
+	 * thread to process it.
+	 * 
+	 * @param t thumbnail
+	 * @param params processing params (?)
+	 * @param height how big
+	 * @param priority if \c true then run as soon as possible
+	 * @param l listener waiting on update
+	 */
+    void add(Thumbnail* t, const rtengine::procparams::ProcParams& params,
+					int height, bool* priority, ThumbImageUpdateListener* l);
+
+	/** 
+	 * @brief Remove jobs associated with listener \c l.
+	 * 
+	 * Jobs being processed will be finished. Will not return till all jobs for
+	 * \c l have been completed.
+	 *
+	 * @param listener jobs associated with this will be stopped
+	 */
+    void removeJobs(ThumbImageUpdateListener* listener);
+
+	/** 
+	 * @brief Stop processing and remove all jobs.
+	 *
+	 * Will not return till all running jobs have completed.
+	 */
+    void removeAllJobs(void);
+
+  private:
+
+	ThumbImageUpdater();
+
+	class Impl;
+	Impl* impl_;
 };
 
-extern ThumbImageUpdater thumbImageUpdater;
+/** 
+ * @brief Singleton boiler plate.
+ *
+ * To use: \c thumbImageUpdater->start() ,
+ */
+#define thumbImageUpdater ThumbImageUpdater::getInstance()
 
 #endif