liz/rawTherapee
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Threading cleanups

Browse Source
This commit is contained in:
Oliver Duis
2011-06-21 21:59:51 +02:00
parent 0583d2bc39
commit c1a8e6bbbd
13 changed files with 260 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files

316
rtgui/batchqueue.cc
View File

@@ -122,68 +122,72 @@ bool BatchQueue::saveBatchQueue( )
return true;
}
bool BatchQueue::loadBatchQueue( )
void BatchQueue::loadBatchQueue( )
{
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
{
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
Glib::ustring savedQueueFile;
savedQueueFile = options.rtdir+"/batch/queue";
FILE *f = safe_g_fopen (savedQueueFile, "rt");
Glib::ustring savedQueueFile;
savedQueueFile = options.rtdir+"/batch/queue";
FILE *f = safe_g_fopen (savedQueueFile, "rt");
if (f==NULL)
return false;
char *buffer = new char[1024];
unsigned numLoaded=0;
while (fgets (buffer, 1024, f)){
char *p = strchr(buffer,';' );
if( p ){
char *le = buffer + strlen(buffer);
while( --le > buffer && (*le == '\n' || *le == '\r') );
Glib::ustring source(buffer, p-buffer );
Glib::ustring paramsFile(p+1, (le +1)- (p+1) );
if (f!=NULL) {
char *buffer = new char[1024];
unsigned numLoaded=0;
while (fgets (buffer, 1024, f)){
char *p = strchr(buffer,';' );
if( p ){
char *le = buffer + strlen(buffer);
while( --le > buffer && (*le == '\n' || *le == '\r') );
Glib::ustring source(buffer, p-buffer );
Glib::ustring paramsFile(p+1, (le +1)- (p+1) );
rtengine::procparams::ProcParams pparams;
if( pparams.load( paramsFile ) )
continue;
rtengine::procparams::ProcParams pparams;
if( pparams.load( paramsFile ) )
continue;
::Thumbnail *thumb = cacheMgr->getEntry( source );
if( thumb ){
rtengine::ProcessingJob* job = rtengine::ProcessingJob::create(source, thumb->getType() == FT_Raw, pparams);
::Thumbnail *thumb = cacheMgr->getEntry( source );
if( thumb ){
rtengine::ProcessingJob* job = rtengine::ProcessingJob::create(source, thumb->getType() == FT_Raw, pparams);
int prevh = options.maxThumbnailHeight;
int prevw = prevh;
guint8* prev = NULL;
double tmpscale;
rtengine::IImage8* img = thumb->processThumbImage(pparams, options.maxThumbnailHeight, tmpscale);
if (img) {
prevw = img->getWidth();
prevh = img->getHeight();
prev = new guint8[prevw * prevh * 3];
memcpy(prev, img->getData(), prevw * prevh * 3);
img->free();
}
BatchQueueEntry *entry = new BatchQueueEntry(job, pparams,source, prev, prevw, prevh, thumb);
entry->setParent(this);
entry->resize(options.thumbSize);
entry->savedParamsFile = paramsFile;
entry->selected = false;
fd.push_back(entry);
int prevh = options.maxThumbnailHeight;
int prevw = prevh;
guint8* prev = NULL;
double tmpscale;
rtengine::IImage8* img = thumb->processThumbImage(pparams, options.maxThumbnailHeight, tmpscale);
if (img) {
prevw = img->getWidth();
prevh = img->getHeight();
prev = new guint8[prevw * prevh * 3];
memcpy(prev, img->getData(), prevw * prevh * 3);
img->free();
}
BatchQueueEntry *entry = new BatchQueueEntry(job, pparams,source, prev, prevw, prevh, thumb);
entry->setParent(this);
entry->resize(options.thumbSize);
entry->savedParamsFile = paramsFile;
entry->selected = false;
fd.push_back(entry);
BatchQueueButtonSet* bqbs = new BatchQueueButtonSet(entry);
bqbs->setButtonListener(this);
entry->addButtonSet(bqbs);
numLoaded++;
}
}
BatchQueueButtonSet* bqbs = new BatchQueueButtonSet(entry);
bqbs->setButtonListener(this);
entry->addButtonSet(bqbs);
numLoaded++;
}
}
}
delete [] buffer;
fclose(f);
}
}
delete [] buffer;
fclose(f);
arrangeFiles ();
queue_draw ();
return numLoaded > 0;
notifyListener();
}
Glib::ustring BatchQueue::getTempFilenameForParams( const Glib::ustring filename )
@@ -214,95 +218,99 @@ int deleteitem (void* data)
void BatchQueue::cancelItems (std::vector<ThumbBrowserEntryBase*>* items) {
{
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
for (int i=0; i<items->size(); i++) {
BatchQueueEntry* entry = (BatchQueueEntry*)(*items)[i];
if (entry->processing)
continue;
std::vector<ThumbBrowserEntryBase*>::iterator pos = std::find (fd.begin(), fd.end(), entry);
if (pos!=fd.end()) {
fd.erase (pos);
rtengine::ProcessingJob::destroy (entry->job);
if (entry->thumbnail)
entry->thumbnail->imageRemovedFromQueue ();
g_idle_add (deleteitem, entry);
for (int i=0; i<items->size(); i++) {
BatchQueueEntry* entry = (BatchQueueEntry*)(*items)[i];
if (entry->processing)
continue;
std::vector<ThumbBrowserEntryBase*>::iterator pos = std::find (fd.begin(), fd.end(), entry);
if (pos!=fd.end()) {
fd.erase (pos);
rtengine::ProcessingJob::destroy (entry->job);
if (entry->thumbnail)
entry->thumbnail->imageRemovedFromQueue ();
g_idle_add (deleteitem, entry);
}
}
}
for (int i=0; i<fd.size(); i++)
fd[i]->selected = false;
lastClicked = NULL;
selected.clear ();
for (int i=0; i<fd.size(); i++)
fd[i]->selected = false;
lastClicked = NULL;
selected.clear ();
saveBatchQueue( );
}
saveBatchQueue( );
}
redraw ();
notifyListener ();
}
void BatchQueue::headItems (std::vector<ThumbBrowserEntryBase*>* items) {
{
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
for (int i=items->size()-1; i>=0; i--) {
BatchQueueEntry* entry = (BatchQueueEntry*)(*items)[i];
if (entry->processing)
continue;
std::vector<ThumbBrowserEntryBase*>::iterator pos = std::find (fd.begin(), fd.end(), entry);
if (pos!=fd.end() && pos!=fd.begin()) {
fd.erase (pos);
// find the first item that is not under processing
for (pos=fd.begin(); pos!=fd.end(); pos++)
if (!(*pos)->processing) {
fd.insert (pos, entry);
break;
}
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
for (int i=items->size()-1; i>=0; i--) {
BatchQueueEntry* entry = (BatchQueueEntry*)(*items)[i];
if (entry->processing)
continue;
std::vector<ThumbBrowserEntryBase*>::iterator pos = std::find (fd.begin(), fd.end(), entry);
if (pos!=fd.end() && pos!=fd.begin()) {
fd.erase (pos);
// find the first item that is not under processing
for (pos=fd.begin(); pos!=fd.end(); pos++)
if (!(*pos)->processing) {
fd.insert (pos, entry);
break;
}
}
}
saveBatchQueue( );
}
saveBatchQueue( );
}
redraw ();
}
void BatchQueue::tailItems (std::vector<ThumbBrowserEntryBase*>* items) {
{
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
for (int i=0; i<items->size(); i++) {
BatchQueueEntry* entry = (BatchQueueEntry*)(*items)[i];
if (entry->processing)
continue;
std::vector<ThumbBrowserEntryBase*>::iterator pos = std::find (fd.begin(), fd.end(), entry);
if (pos!=fd.end()) {
fd.erase (pos);
fd.push_back (entry);
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
for (int i=0; i<items->size(); i++) {
BatchQueueEntry* entry = (BatchQueueEntry*)(*items)[i];
if (entry->processing)
continue;
std::vector<ThumbBrowserEntryBase*>::iterator pos = std::find (fd.begin(), fd.end(), entry);
if (pos!=fd.end()) {
fd.erase (pos);
fd.push_back (entry);
}
}
saveBatchQueue( );
}
saveBatchQueue( );
}
redraw ();
}
void BatchQueue::selectAll () {
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
{// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
lastClicked = NULL;
selected.clear ();
for (int i=0; i<fd.size(); i++) {
if (fd[i]->processing)
continue;
fd[i]->selected = true;
selected.push_back (fd[i]);
lastClicked = NULL;
selected.clear ();
for (int i=0; i<fd.size(); i++) {
if (fd[i]->processing)
continue;
fd[i]->selected = true;
selected.push_back (fd[i]);
}
}
queue_draw ();
}
@@ -387,46 +395,46 @@ rtengine::ProcessingJob* BatchQueue::imageReady (rtengine::IImage16* img) {
delete processing;
processing = NULL;
{
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
// TODO: Check for Linux
#ifdef WIN32
Glib::Mutex::Lock lock(entryMutex);
#endif
fd.erase (fd.begin());
fd.erase (fd.begin());
// return next job
if (fd.size()==0) {
if (listener)
listener->queueEmpty ();
}
else if (listener && listener->canStartNext ()) {
BatchQueueEntry* next = (BatchQueueEntry*)fd[0];
// tag it as selected
next->processing = true;
processing = next;
// remove from selection
if (processing->selected) {
std::vector<ThumbBrowserEntryBase*>::iterator pos = std::find (selected.begin(), selected.end(), processing);
if (pos!=selected.end())
selected.erase (pos);
processing->selected = false;
// return next job
if (fd.size()==0) {
if (listener)
listener->queueEmpty ();
}
else if (listener && listener->canStartNext ()) {
BatchQueueEntry* next = (BatchQueueEntry*)fd[0];
// tag it as selected
next->processing = true;
processing = next;
// remove from selection
if (processing->selected) {
std::vector<ThumbBrowserEntryBase*>::iterator pos = std::find (selected.begin(), selected.end(), processing);
if (pos!=selected.end())
selected.erase (pos);
processing->selected = false;
}
// remove button set
next->removeButtonSet ();
}
if( saveBatchQueue( ) ){
safe_g_remove( processedParams );
// Delete all files in directory \batch when finished, just to be sure to remove zombies
if( fd.size()==0 ){
std::vector<Glib::ustring> names;
Glib::ustring batchdir = options.rtdir+"/batch/";
Glib::RefPtr<Gio::File> dir = Gio::File::create_for_path (batchdir);
safe_build_file_list (dir, names, batchdir);
for(std::vector<Glib::ustring>::iterator iter=names.begin(); iter != names.end();iter++ )
safe_g_remove( *iter );
}
}
// remove button set
next->removeButtonSet ();
}
if( saveBatchQueue( ) ){
safe_g_remove( processedParams );
// Delete all files in directory \batch when finished, just to be sure to remove zombies
if( fd.size()==0 ){
std::vector<Glib::ustring> names;
Glib::ustring batchdir = options.rtdir+"/batch/";
Glib::RefPtr<Gio::File> dir = Gio::File::create_for_path (batchdir);
safe_build_file_list (dir, names, batchdir);
for(std::vector<Glib::ustring>::iterator iter=names.begin(); iter != names.end();iter++ )
safe_g_remove( *iter );
}
}
}
redraw ();
notifyListener ();

5
rtgui/batchqueue.h
View File

@@ -56,6 +56,7 @@ class BatchQueue : public ThumbBrowserBase,
Glib::ustring autoCompleteFileName (const Glib::ustring& fileName, const Glib::ustring& format);
Glib::ustring getTempFilenameForParams( const Glib::ustring filename );
bool saveBatchQueue( );
void notifyListener ();
public:
BatchQueue ();
@@ -78,8 +79,8 @@ class BatchQueue : public ThumbBrowserBase,
void redrawNeeded (LWButton* button);
void setBatchQueueListener (BatchQueueListener* l) { listener = l; }
void notifyListener ();
bool loadBatchQueue ();
void loadBatchQueue ();
static Glib::ustring calcAutoFileNameBase (const Glib::ustring& origFileName);
};

3
rtgui/batchqueueentry.cc
View File

@@ -56,8 +56,7 @@ void BatchQueueEntry::refreshThumbnailImage () {
if (!opreview)
return;
batchQueueEntryUpdater.add (opreview, origpw, origph, preh, this);
batchQueueEntryUpdater.process ();
batchQueueEntryUpdater.process (opreview, origpw, origph, preh, this);
}
void BatchQueueEntry::calcThumbnailSize () {

1
rtgui/batchqueuepanel.cc
View File

@@ -133,7 +133,6 @@ BatchQueuePanel::BatchQueuePanel () {
show_all ();
batchQueue->loadBatchQueue ();
batchQueue->notifyListener ();
}

81
rtgui/bqentryupdater.cc
View File

@@ -26,8 +26,7 @@ BatchQueueEntryUpdater::BatchQueueEntryUpdater ()
: tostop(false), stopped(true), qMutex(NULL) {
}
void BatchQueueEntryUpdater::add (guint8* oimg, int ow, int oh, int newh, BQEntryUpdateListener* listener) {
void BatchQueueEntryUpdater::process (guint8* oimg, int ow, int oh, int newh, BQEntryUpdateListener* listener) {
if (!qMutex)
qMutex = new Glib::Mutex ();
@@ -42,6 +41,7 @@ void BatchQueueEntryUpdater::add (guint8* oimg, int ow, int oh, int newh, BQEntr
i->listener = listener;
break;
}
// not found, create and append new job
if (i==jqueue.end ()) {
Job j;
@@ -53,33 +53,30 @@ void BatchQueueEntryUpdater::add (guint8* oimg, int ow, int oh, int newh, BQEntr
jqueue.push_back (j);
}
qMutex->unlock ();
}
void BatchQueueEntryUpdater::process () {
// Start thread if not running yet
if (stopped) {
stopped = false;
tostop = false;
if (stopped){
#undef THREAD_PRIORITY_LOW
thread = Glib::Thread::create(sigc::mem_fun(*this, &BatchQueueEntryUpdater::process_), (unsigned long int)0, true, true, Glib::THREAD_PRIORITY_LOW);
thread = Glib::Thread::create(sigc::mem_fun(*this, &BatchQueueEntryUpdater::processThread), (unsigned long int)0, true, true, Glib::THREAD_PRIORITY_LOW);
}
}
void BatchQueueEntryUpdater::process_ () {
stopped = false;
tostop = false;
// TODO: process visible jobs first
bool isEmpty=false;
void BatchQueueEntryUpdater::processThread () {
// TODO: process visible jobs first
bool isEmpty=false;
while (!tostop && !isEmpty) {
qMutex->lock ();
isEmpty=jqueue.empty (); // do NOT put into while() since it must be within mutex section
Job current;
if (!isEmpty) {
current = jqueue.front ();
jqueue.pop_front ();
}
isEmpty=jqueue.empty (); // do NOT put into while() since it must be within mutex section
Job current;
if (!isEmpty) {
current = jqueue.front ();
jqueue.pop_front ();
}
qMutex->unlock ();
if (!isEmpty && current.listener) {
@@ -89,38 +86,13 @@ void BatchQueueEntryUpdater::process_ () {
current.listener->updateImage (img, neww, current.newh);
}
}
stopped = true;
}
void BatchQueueEntryUpdater::stop () {
if (stopped) {
tostop = true;
return; }
gdk_threads_leave();
tostop = true;
Glib::Thread::self()->yield();
if (!stopped)
thread->join ();
gdk_threads_enter();
}
void BatchQueueEntryUpdater::removeJobs () {
if (!qMutex)
return;
qMutex->lock ();
while (!jqueue.empty())
jqueue.pop_front ();
qMutex->unlock ();
}
void BatchQueueEntryUpdater::removeJobs (BQEntryUpdateListener* listener) {
if (!qMutex)
return;
if (!qMutex) return;
qMutex->lock ();
bool ready = false;
@@ -138,9 +110,22 @@ void BatchQueueEntryUpdater::removeJobs (BQEntryUpdateListener* listener) {
}
void BatchQueueEntryUpdater::terminate () {
// never started or currently not running?
if (!qMutex || stopped) return;
stop ();
removeJobs ();
if (!stopped) {
// Yield to currenly running thread and wait till it's finished
gdk_threads_leave();
tostop = true;
Glib::Thread::self()->yield();
if (!stopped) thread->join ();
gdk_threads_enter();
}
// Remove remaining jobs
qMutex->lock ();
while (!jqueue.empty()) jqueue.pop_front ();
qMutex->unlock ();
}

7
rtgui/bqentryupdater.h
View File

@@ -47,14 +47,11 @@ class BatchQueueEntryUpdater {
public:
BatchQueueEntryUpdater ();
void add (guint8* oimg, int ow, int oh, int newh, BQEntryUpdateListener* listener);
void process ();
void stop ();
void removeJobs ();
void process (guint8* oimg, int ow, int oh, int newh, BQEntryUpdateListener* listener);
void removeJobs (BQEntryUpdateListener* listener);
void terminate ();
void process_ ();
void processThread ();
};
extern BatchQueueEntryUpdater batchQueueEntryUpdater;

26
rtgui/previewhandler.cc
View File

@@ -48,7 +48,7 @@ struct iaimgpar {
CropParams cp;
};
int iasetimage (void* data) {
int setImageThread (void* data) {
gdk_threads_enter ();
@@ -66,13 +66,14 @@ int iasetimage (void* data) {
}
if (pih->phandler->image) {
IImage8* temp = pih->phandler->image;
temp->getMutex().lock ();
IImage8* oldImg = pih->phandler->image;
oldImg->getMutex().lock ();
pih->phandler->image = iap->image;
temp->getMutex().unlock ();
oldImg->getMutex().unlock ();
}
else
pih->phandler->image = iap->image;
pih->phandler->cropParams = iap->cp;
pih->phandler->previewScale = iap->scale;
pih->pending--;
@@ -93,10 +94,11 @@ void PreviewHandler::setImage (rtengine::IImage8* i, double scale, rtengine::pro
iap->scale = scale;
iap->cp = cp;
g_idle_add (iasetimage, iap);
g_idle_add (setImageThread, iap);
}
int iadelimage (void* data) {
int delImageThread (void* data) {
gdk_threads_enter ();
@@ -114,10 +116,10 @@ int iadelimage (void* data) {
}
if (pih->phandler->image) {
IImage8* temp = pih->phandler->image;
temp->getMutex().lock ();
IImage8* oldImg = pih->phandler->image;
oldImg->getMutex().lock ();
pih->phandler->image = NULL;
temp->getMutex().unlock ();
oldImg->getMutex().unlock ();
}
iap->image->free ();
pih->phandler->previewImgMutex.lock ();
@@ -140,10 +142,10 @@ void PreviewHandler::delImage (IImage8* i) {
iap->image = i;
iap->pih = pih;
g_idle_add (iadelimage, iap);
g_idle_add (delImageThread, iap);
}
int imready (void* data) {
int imageReadyThread (void* data) {
gdk_threads_enter ();
@@ -179,7 +181,7 @@ void PreviewHandler::imageReady (CropParams cp) {
iaimgpar* iap = new iaimgpar;
iap->pih = pih;
iap->cp = cp;
g_idle_add (imready, iap);
g_idle_add (imageReadyThread, iap);
}
Glib::RefPtr<Gdk::Pixbuf> PreviewHandler::getRoughImage (int x, int y, int w, int h, double zoom) {

6
rtgui/previewhandler.h
View File

@@ -38,9 +38,9 @@ struct PreviewHandlerIdleHelper {
class PreviewHandler : public rtengine::PreviewImageListener {
friend int iasetimage (void* data);
friend int iadelimage (void* data);
friend int imready (void* data);
friend int setImageThread (void* data);
friend int delImageThread (void* data);
friend int imageReadyThread (void* data);
protected:
rtengine::IImage8* image;

54
rtgui/previewloader.cc
View File

@@ -75,15 +75,11 @@ public:
}
Glib::ThreadPool* threadPool_;
Glib::Mutex mutex_;
JobSet jobs_;
gint nConcurrentThreads;
void
processNextJob(void)
void processNextJob()
{
Job j;
{
@@ -102,10 +98,12 @@ public:
DEBUG("processing %s",j.dir_entry_.c_str());
DEBUG("%d job(s) remaining",jobs_.size());
}
g_atomic_int_inc (&nConcurrentThreads);
g_atomic_int_inc (&nConcurrentThreads); // to detect when last thread in pool has run out
// unlock and do processing; will relock on block exit, then call listener
// if something got
try{
try {
Thumbnail* tmb = 0;
{
if (safe_file_test(j.dir_entry_, Glib::FILE_TEST_EXISTS))
@@ -114,22 +112,17 @@ public:
}
}
// we got something so notify listener
// we got something, so notify listener
if ( tmb )
{
j.listener_->previewReady(j.dir_id_,new FileBrowserEntry(tmb,j.dir_entry_));
}
}catch(Glib::Error){
} catch (Glib::Error){} catch(...){}
}catch(...){}
bool last = g_atomic_int_dec_and_test(& nConcurrentThreads);
bool last = g_atomic_int_dec_and_test (&nConcurrentThreads);
// signal at end
if ( jobs_.empty() )
{
if(last)
j.listener_->previewsFinished(j.dir_id_);
}
if (last && jobs_.empty()) j.listener_->previewsFinished(j.dir_id_);
}
};
@@ -138,29 +131,33 @@ PreviewLoader::PreviewLoader():
{
}
PreviewLoader*
PreviewLoader::getInstance(void)
PreviewLoader* PreviewLoader::getInstance(void)
{
// this will not be deleted...
static PreviewLoader* instance_ = 0;
if ( instance_ == 0 )
static PreviewLoader* instance_ = NULL;
if ( instance_ == NULL )
{
instance_ = new PreviewLoader();
static Glib::Mutex smutex_;
Glib::Mutex::Lock lock(smutex_);
if ( instance_ == NULL ) instance_ = new PreviewLoader();
}
return instance_;
}
void
PreviewLoader::add(int dir_id, const Glib::ustring& dir_entry, PreviewLoaderListener* l)
void PreviewLoader::add(int dir_id, const Glib::ustring& dir_entry, PreviewLoaderListener* l)
{
// somebody listening?
if ( l != 0 )
{
Glib::Mutex::Lock lock(impl_->mutex_);
{
Glib::Mutex::Lock lock(impl_->mutex_);
// create a new job and append to queue
DEBUG("saving job %s",dir_entry.c_str());
impl_->jobs_.insert(Impl::Job(dir_id,dir_entry,l));
// create a new job and append to queue
DEBUG("saving job %s",dir_entry.c_str());
impl_->jobs_.insert(Impl::Job(dir_id,dir_entry,l));
}
// queue a run request
DEBUG("adding run request %s",dir_entry.c_str());
@@ -168,8 +165,7 @@ PreviewLoader::add(int dir_id, const Glib::ustring& dir_entry, PreviewLoaderList
}
}
void
PreviewLoader::removeAllJobs(void)
void PreviewLoader::removeAllJobs(void)
{
DEBUG("stop %d",impl_->nConcurrentThreads);
Glib::Mutex::Lock lock(impl_->mutex_);