OSG::BackgroundLoaderBase Class Reference

00018    : mLoadThreads()
00019    , mFinishedLock(NULL)
00020    , mLoadCondVar(NULL)
00021    , mStopRunning(false)
00022 {
00023    mStopRunning = false;
00024    mFinishedLock = Lock::get("BackgroundLoader-FinishedLock");
00025    mLoadCondVar = CondVar::get("BackgroundLoader-CondVar");
00026 }

00029 {
00030    stop();
00031 }

00073 {
00074    OSG_ASSERT(0 == mLoadThreads.size() && "Can't have threads before starting.");
00075    for (OSG::UInt16 i = 0; i < numThreads; i++)
00076    {
00077       OSG::Thread* new_thread = dynamic_cast<Thread *>(ThreadManager::the()->getThread(NULL));
00078       new_thread->runFunction((Thread::ThreadFuncF)loadProc, Thread::getCurrentAspect(), this);
00079       mLoadThreads.push_back(new_thread);
00080    }
00081 }

00034 {
00035    if (mLoadThreads.size() > 0)
00036    {
00037       OSG_ASSERT(!mStopRunning && "BackgroundLoaderBase::stop() Already stopped, but thread is not NULL");
00038       mStopRunning = true;
00039       mLoadCondVar->broadcast();
00040 
00041       for (std::vector<OSG::Thread*>::iterator itr = mLoadThreads.begin();
00042            itr != mLoadThreads.end(); itr++)
00043       {
00044          OSG::Thread::join(*itr);
00045          (*itr) = NULL;
00046       }
00047       mLoadThreads.clear();
00048    }
00049 }

00052 {
00053    mLoadCondVar->acquire();
00054    mPendingRequests.push_back(req);
00055    mLoadCondVar->release();
00056    mLoadCondVar->signal();
00057 }

00060 {
00061    mLoadCondVar->acquire();
00062    request_queue_t::iterator found = std::find(mPendingRequests.begin(), mPendingRequests.end(), req);
00063    if (found != mPendingRequests.end())
00064    {
00065       mPendingRequests.erase(found);
00066    }
00067    mLoadCondVar->release();
00068    // XXX: We should not need this.
00069    mLoadCondVar->signal();
00070 }

00203 {
00204    mFinishedLock->acquire();
00205 
00206    if (0 == reqLimit)
00207    { reqLimit = mFinishedRequests.size(); }
00208 
00209    while((!mFinishedRequests.empty()) && (reqLimit-- >= 0))
00210    {
00211       RequestPtr req = mFinishedRequests.front();
00212       mFinishedRequests.pop_front();
00213 
00214       try
00215       {
00216          req->sync();
00217       }
00218       catch (std::exception& ex)
00219       {
00220          std::cout << "ERROR: Exception thrown in sync: " << ex.what() << std::endl;
00221       }
00222       catch(...)
00223       {
00224          std::cout << "ERROR: Unknown exception thrown in sync." << std::endl;
00225       }
00226 
00227       req->mCompleted = true;
00228    }
00229 
00230    // XXX: Don't know if this is needed to keep the change list from growing out of control.
00231    //OSG::Thread::getCurrentChangeList()->commitChangesAndClear();
00232    mFinishedLock->release();
00233 }

00084 {
00085    OSG_ASSERT(0 == mLoadThreads.size() && "processOne can not be called with background thread active.");
00086 
00087    mLoadCondVar->acquire();
00088    if (!mPendingRequests.empty())
00089    {
00090       OSG_ASSERT(!mPendingRequests.empty());
00091 
00092       // Get the pending request with the highest priority.
00093       RequestPtr req = RequestPtr();
00094       request_queue_t::iterator next_i = 
00095          std::min_element(mPendingRequests.begin(), mPendingRequests.end(), Request::comparePriority);
00096       req = *next_i;
00097 
00098       // Erase our request from the pending list.
00099       mPendingRequests.erase(next_i);
00100 
00101       if (NULL != req.get())
00102       {
00103          req->execute();
00104          OSG::commitChanges();
00105 
00106          mFinishedLock->acquire();
00107          mFinishedRequests.push_back(req);
00108          mFinishedLock->release();
00109       }
00110    }
00111    mLoadCondVar->release();
00112 }

00115 {
00116    unsigned ret_val(0);
00117    mLoadCondVar->acquire();
00118    ret_val = mPendingRequests.size();
00119    mLoadCondVar->release();
00120 
00121    return ret_val;
00122 }

00125 {
00126    BackgroundLoaderBase::desc_list_t ret_list;
00127 
00128    mLoadCondVar->acquire();
00129    for(request_queue_t::iterator i = mPendingRequests.begin(); i!=mPendingRequests.end(); i++)
00130    {
00131       std::string desc = boost::str(boost::format("[%4.2f]: %s") % (*i)->getPriority() % (*i)->getDescription());
00132       ret_list.push_back(desc);
00133    }
00134    mLoadCondVar->release();
00135 
00136    return ret_list;
00137 }

00140 {
00141    // Get a pointer to the singleton object.
00142    BackgroundLoaderBase* the = reinterpret_cast<BackgroundLoaderBase*>(arg);
00143    OSG_ASSERT(NULL != the && "Should have a pointer to BackgroundLoader.");
00144 
00145    while(false == the->mStopRunning)
00146    {
00147       // Wait on a new request.
00148       the->mLoadCondVar->acquire();
00149       while(false == the->mStopRunning &&
00150             the->mPendingRequests.empty())
00151       {
00152          the->mLoadCondVar->wait();
00153       }
00154 
00155       // Early out if we are exiting the loop.
00156       if (the->mStopRunning)
00157       { continue; }
00158 
00159       OSG_ASSERT(!the->mPendingRequests.empty() && "Can't have empty pending list.");
00160 
00161       // Get the pending request with the highest priority.
00162       RequestPtr req = RequestPtr();
00163       request_queue_t::iterator next_i = 
00164          std::min_element(the->mPendingRequests.begin(), the->mPendingRequests.end(), Request::comparePriority);
00165       req = *next_i;
00166 
00167       // Erase our request from the pending list.
00168       the->mPendingRequests.erase(next_i);
00169 
00170       // Now that we have a request to process, release condition variable.
00171       the->mLoadCondVar->release();
00172 
00173       // If we have a valid request, execute it and commit changes.
00174       if (NULL != req.get())
00175       {
00176          try
00177          {
00178             req->execute();
00179             OSG::commitChanges();
00180          }
00181          catch (std::exception& ex)
00182          {
00183             std::cout << "ERROR: Exception thrown in execute: " << ex.what() << std::endl;
00184          }
00185          catch(...)
00186          {
00187             std::cout << "ERROR: Unknown exception thrown in execute." << std::endl;
00188          }
00189 
00190          // Commit our changes to our aspect (shared) and clear out change list 
00191          // since we don't need it around any more.
00192          // OSG::commitChanges();
00193          //OSG::Thread::getCurrentChangeList()->commitChangesAndClear();
00194 
00195          the->mFinishedLock->acquire();
00196          the->mFinishedRequests.push_back(req);
00197          the->mFinishedLock->release();
00198       }
00199    }
00200 }