Source/Contrib/BackgroundLoader/OSGBackgroundLoader.cpp

   1 #include "OSGBackgroundLoader.h"
   2
   3 #include "OSGThreadManager.h"
   4 #include "OSGBaseInitFunctions.h"
   5 #include <boost/lexical_cast.hpp>
   6 #include "OSGChangeList.h"
   7 #include <boost/format.hpp>
   8
   9 #include "OSGSingletonHolder.ins"
  10
  11 OSG_BEGIN_NAMESPACE
  12
  13 OSG_SINGLETON_INST(BackgroundLoaderBase, addPostFactoryExitFunction)
  14
  15 BackgroundLoaderBase::BackgroundLoaderBase(void) :
  16     mPendingRequests (     ),
  17     mFinishedRequests(     ),
  18     mLoadThreads     (     ),
  19     mFinishedLock    (NULL ),
  20     mLoadCondVar     (NULL ),
  21     mStopRunning     (false)
  22 {
  23     mStopRunning  = false;
  24     mFinishedLock = Lock   ::get("BackgroundLoader-FinishedLock", false);
  25     mLoadCondVar  = CondVar::get("BackgroundLoader-CondVar",      false);
  26 }
  27
  28 BackgroundLoaderBase::~BackgroundLoaderBase(void)
  29 {
  30     stop();
  31
  32     mFinishedLock = NULL;
  33     mLoadCondVar  = NULL;
  34 }
  35
  36 void BackgroundLoaderBase::stop(void)
  37 {
  38    if (mLoadThreads.size() > 0)
  39    {
  40       OSG_ASSERT(!mStopRunning &&
  41                  "BackgroundLoaderBase::stop() Already stopped, but thread "
  42                  "is not NULL");
  43
  44       mStopRunning = true;
  45
  46       mLoadCondVar->broadcast();
  47
  48       for(std::vector<ThreadRefPtr>::iterator itr  = mLoadThreads.begin();
  49                                               itr != mLoadThreads.end  ();
  50                                             ++itr                       )
  51       {
  52           Thread::join(*itr);
  53
  54           (*itr) = NULL;
  55       }
  56
  57       mLoadThreads.clear();
  58    }
  59 }
  60
  61 void BackgroundLoaderBase::addRequest(RequestPtr req)
  62 {
  63    mLoadCondVar->acquire();
  64
  65    mPendingRequests.push_back(req);
  66
  67    mLoadCondVar->release();
  68    mLoadCondVar->signal ();
  69 }
  70
  71 void BackgroundLoaderBase::removeRequest(RequestPtr req)
  72 {
  73    mLoadCondVar->acquire();
  74    request_queue_t::iterator found = std::find(mPendingRequests.begin(),
  75                                                mPendingRequests.end  (),
  76                                                req                     );
  77    if (found != mPendingRequests.end())
  78    {
  79        mPendingRequests.erase(found);
  80    }
  81
  82    mLoadCondVar->release();
  83    // XXX: We should not need this.
  84    mLoadCondVar->signal ();
  85 }
  86
  87 void BackgroundLoaderBase::start(const UInt16 numThreads)
  88 {
  89    OSG_ASSERT(0 == mLoadThreads.size() &&
  90               "Can't have threads before starting.");
  91
  92    for(UInt16 i = 0; i < numThreads; i++)
  93    {
  94        ThreadRefPtr new_thread =
  95            dynamic_pointer_cast<Thread>(ThreadManager::the()->getThread(NULL,
  96                                                                         false));
  97
  98        new_thread->runFunction(static_cast<Thread::ThreadFuncF>(loadProc),
  99                                Thread::getCurrentAspect(), this);
 100
 101        mLoadThreads.push_back(new_thread);
 102    }
 103 }
 104
 105 void BackgroundLoaderBase::processOne(void)
 106 {
 107    OSG_ASSERT(0 == mLoadThreads.size() &&
 108               "processOne can not be called with background thread active.");
 109
 110    mLoadCondVar->acquire();
 111
 112    if(! mPendingRequests.empty())
 113    {
 114       OSG_ASSERT(!mPendingRequests.empty());
 115
 116       // Get the pending request with the highest priority.
 117       RequestPtr req = RequestPtr();
 118
 119       request_queue_t::iterator next_i =
 120           std::min_element(mPendingRequests.begin(),
 121                            mPendingRequests.end  (),
 122                            Request::comparePriority);
 123
 124       req = *next_i;
 125
 126       // Erase our request from the pending list.
 127       mPendingRequests.erase(next_i);
 128
 129       if(NULL != req.get())
 130       {
 131          req->execute();
 132
 133          commitChanges();
 134
 135          mFinishedLock->acquire();
 136
 137          mFinishedRequests.push_back(req);
 138
 139          mFinishedLock->release();
 140       }
 141    }
 142
 143    mLoadCondVar->release();
 144 }
 145
 146 unsigned BackgroundLoaderBase::getNumPendingRequests(void)
 147 {
 148    unsigned ret_val(0);
 149
 150    mLoadCondVar->acquire();
 151
 152    ret_val = unsigned(mPendingRequests.size());
 153
 154    mLoadCondVar->release();
 155
 156    return ret_val;
 157 }
 158
 159 BackgroundLoaderBase::desc_list_t
 160     BackgroundLoaderBase::getPendingDescriptionList(void)
 161 {
 162     BackgroundLoaderBase::desc_list_t ret_list;
 163
 164     mLoadCondVar->acquire();
 165
 166     for(request_queue_t::iterator i  = mPendingRequests.begin();
 167                                   i != mPendingRequests.end  ();
 168                                 ++i                            )
 169    {
 170        std::string desc = boost::str(boost::format("[%4.2f]: %s") %
 171                                      (*i)->getPriority() %
 172                                      (*i)->getDescription());
 173
 174       ret_list.push_back(desc);
 175    }
 176
 177    mLoadCondVar->release();
 178
 179    return ret_list;
 180 }
 181
 182 void BackgroundLoaderBase::loadProc(void *arg)
 183 {
 184    // Get a pointer to the singleton object.
 185    BackgroundLoaderBase *the = reinterpret_cast<BackgroundLoaderBase *>(arg);
 186
 187    OSG_ASSERT(NULL != the && "Should have a pointer to BackgroundLoader.");
 188
 189    while(false == the->mStopRunning)
 190    {
 191       // Wait on a new request.
 192       the->mLoadCondVar->acquire();
 193
 194       while(false == the->mStopRunning && the->mPendingRequests.empty())
 195       {
 196           the->mLoadCondVar->wait();
 197       }
 198
 199       // Early out if we are exiting the loop.
 200       if(the->mStopRunning)
 201       {
 202           continue;
 203       }
 204
 205       OSG_ASSERT(!the->mPendingRequests.empty() &&
 206                  "Can't have empty pending list.");
 207
 208       // Get the pending request with the highest priority.
 209       RequestPtr req = RequestPtr();
 210
 211       request_queue_t::iterator next_i =
 212           std::min_element(the->mPendingRequests.begin(),
 213                            the->mPendingRequests.end  (),
 214                            Request::comparePriority     );
 215
 216       req = *next_i;
 217
 218       // Erase our request from the pending list.
 219       the->mPendingRequests.erase(next_i);
 220
 221       // Now that we have a request to process, release condition variable.
 222       the->mLoadCondVar->release();
 223
 224       // If we have a valid request, execute it and commit changes.
 225       if(NULL != req.get())
 226       {
 227           try
 228           {
 229               req->execute();
 230           }
 231           catch (std::exception& ex)
 232           {
 233               std::cout << "ERROR: Exception thrown in execute: "
 234                         << ex.what()
 235                         << std::endl;
 236          }
 237          catch(...)
 238          {
 239             std::cout << "ERROR: Unknown exception thrown in execute."
 240                       << std::endl;
 241          }
 242
 243          // Commit our changes to our aspect (shared) and clear out change list
 244          // since we don't need it around any more.
 245          // OSG::commitChanges();
 246          //OSG::Thread::getCurrentChangeList()->commitChangesAndClear();
 247
 248          the->mFinishedLock->acquire();
 249
 250          the->mFinishedRequests.push_back(req);
 251
 252          the->mFinishedLock->release();
 253       }
 254    }
 255 }
 256
 257 void BackgroundLoaderBase::sync(unsigned reqLimit)
 258 {
 259     mFinishedLock->acquire();
 260
 261     if(0 == reqLimit)
 262     {
 263         reqLimit = unsigned(mFinishedRequests.size());
 264     }
 265
 266    while((!mFinishedRequests.empty()) && (reqLimit-- != 0))
 267    {
 268        RequestPtr req = mFinishedRequests.front();
 269        mFinishedRequests.pop_front();
 270
 271        try
 272        {
 273            req->sync();
 274        }
 275        catch (std::exception& ex)
 276        {
 277            std::cout << "ERROR: Exception thrown in sync: "
 278                      << ex.what()
 279                      << std::endl;
 280        }
 281        catch(...)
 282        {
 283            std::cout << "ERROR: Unknown exception thrown in sync." << std::endl;
 284        }
 285
 286        req->mCompleted = true;
 287    }
 288
 289    // XXX: Don't know if this is needed to keep the change list from growing
 290    //out of control.
 291    //OSG::Thread::getCurrentChangeList()->commitChangesAndClear();
 292
 293    mFinishedLock->release();
 294 }
 295
 296 OSG_END_NAMESPACE