---

PsnMultiThreadedScheduler.cxx

Go to the documentation of this file.

/*
 * This file is part of openMask © INRIA, CNRS, Universite de Rennes 1 1993-2002, thereinafter the Software
 * 
 * The Software has been developped within the Siames Project. 
 * INRIA, the University of Rennes 1 and CNRS jointly hold intellectual property rights
 * 
 * The Software has been registered with the Agence pour la Protection des
 * Programmes (APP) under registration number IDDN.FR.001.510008.00.S.P.2001.000.41200
 *  
 * This file may be distributed under the terms of the Q Public License
 * version 1.0 as defined by Trolltech AS of Norway and appearing in the file
 * LICENSE.QPL included in the packaging of this file.
 *
 * Licensees holding valid specific licenses issued by INRIA, CNRS or Université de Rennes 1 
 * for the software may use this file in accordance with that specific license
 *
 */
#include "PsnMultiThreadedScheduler.h"

#include "PsnFrameScheduler.h"
#include "PsnBenchScheduler.h"
#include "PsnBarrier.h"
#include "PsController.h"
#include "PsDuplicatedObject.h"
#include "PsnThreadMemoryManager.h"
#include "PsnDescriptorMemoryManager.h"
#include "PsnNew.h"
#include "PsnSystemMemoryManager.h"
#include <unistd.h>

PsnMultiThreadedScheduler::PsnMultiThreadedScheduler (unsigned int numberOfSteps, unsigned int numberOfThreads, bool createThreadAndLocalSchedulers ) :
   PsnScheduler(numberOfSteps),
   _numberOfThreads(numberOfThreads),
   _currentStep ( 0 ),
   _nextThreadForScheduling ( 0 )
{

   _barrier = new PsnBarrier ( _numberOfThreads + 1) ;

   _threadIds = new pthread_t [ _numberOfThreads ] ;

   //onlu used when profiling
   _profilingTimer = new itimerval  [ _numberOfThreads ] ;

   _threadData = new vector <pair<PsnScheduler *, unsigned int> * > ( _numberOfThreads ) ;

   for (int i = 0 ; i < _numberOfThreads ; i++ ) 
      {
         (*_threadData)[i] = new pair<PsnScheduler *, unsigned int> (this, i) ;

         // get the data associated to the timer used for profiling
         getitimer(ITIMER_PROF, &(_profilingTimer[i] ));
      }


   _localSchedulers = new vector <PsnScheduler *> ( _numberOfThreads ) ;

   int ret;

#ifdef _SGI
   //for better use of the cpus available
   pthread_setconcurrency(_numberOfThreads) ;
#endif

   if ( createThreadAndLocalSchedulers )
      {
         
         _duplicatedObjectScheduler = new PsnFrameScheduler( numberOfSteps ) ;
         
         for (int i = 0 ; i < _numberOfThreads ; i++ ) 
            {
               (*_localSchedulers)[i] = new PsnFrameScheduler( numberOfSteps ) ;
            }
            
         for (int i = 0 ; i < _numberOfThreads ; i++ ) 
            {
               ret = pthread_create( & (_threadIds [i]), 
                                     0,
                                     (void *(*)(void *))_threadContainer,
                                     (*_threadData)[i]);
               if (ret) 
                  {
                     PsController::error("Thread creation impossible") ;
                  }
               
            }
      }
}


PsnMultiThreadedScheduler::~PsnMultiThreadedScheduler() {
   

   for (int i = 0 ; i < _numberOfThreads ; i++ ) {
      
      delete (*_localSchedulers)[i] ;
      
      delete (*_threadData)[i] ;

   }

   delete _profilingTimer ;

   delete _localSchedulers ;

   delete _barrier ;

   delete [] _threadIds ;

   delete _threadData ;

   delete _duplicatedObjectScheduler ;
}



void PsnMultiThreadedScheduler::runStep (unsigned int which) 
{

   _currentStep = which ;

   _duplicatedObjectScheduler->runStep (which) ;

#if defined (_DEBUGSCHEDULER)   
   cerr<<"Execution par block pour debug "<<endl;
   for ( int i = 0 ; i < _numberOfThreads ; i++ ) {
      cerr<<"Lot "<<i<<" ***********************************"<<endl;
      (*_localSchedulers)[ i ]->runStep (which) ;
   }
#else
   // let the thread do their work
   _barrier->Sync() ;


   // wait for the threads to finish their work
   _barrier->Sync() ;
   //let the controller do it's work

   //purge the memory manager from memory that couldn't be freed efficiently
   PsnThreadMemoryManager::freeLeakedMemoryBlocks() ;
#endif
}



void PsnMultiThreadedScheduler::addToScheduable (PsnReferenceObjectHandle * ref, unsigned int when, unsigned int whichThread) {
   //cerr<<"PsnMultiThreadedScheduler::addToScheduable"<<((PsnReferenceObjectHandle *)ref)->objetSimul()->Nom()<<endl;
   if (dynamic_cast<const PsDuplicatedObject *>(&((PsnReferenceObjectHandle *)ref)->getSimulatedObject())) {
      _duplicatedObjectScheduler->addToScheduable ( ref , when ) ;
   }
   else {
      (*_localSchedulers)[ whichThread ] -> addToScheduable( ref , when ) ;
   }

}



void PsnMultiThreadedScheduler::addToScheduable (PsnReferenceObjectHandle * ref, unsigned int when) {
   //cerr<<"PsnMultiThreadedScheduler::addToScheduable "<<((PsnReferenceObjectHandle *)ref)->objetSimul()->Nom()<<endl;

   if (dynamic_cast<const PsDuplicatedObject *>(&((PsnReferenceObjectHandle *)ref)->getSimulatedObject())) {
      _duplicatedObjectScheduler->addToScheduable ( ref , when ) ;
   }
   else {
      assert ( ref != NULL ) ;
      assert ( ref->getSimulatedObject().getFather () != NULL );
      map<PsnReferenceObjectHandle *, unsigned int>::iterator i = _whichThread.find((PsnReferenceObjectHandle *)ref->getSimulatedObject().getFather()->getObjectHandle() ) ;
      if ( i==_whichThread.end() ){
         
         (*_localSchedulers)[ _nextThreadForScheduling ] -> addToScheduable( ref , when ) ;
#ifdef _DEBUGSCHEDULER
         cout<<ref->getSimulatedObject().getObjectDescriptor().getName()<< " goes to thread "<<_nextThreadForScheduling<<endl;
#endif
         _whichThread [ ref ] = _nextThreadForScheduling ;

         _nextThreadForScheduling = ( _nextThreadForScheduling + 1 ) % _numberOfThreads ;
         
      }
      else {//schedule on the same thread as ref's father
         
         (*_localSchedulers)[ (*i).second ] -> addToScheduable( ref , when ) ;
#ifdef _DEBUGSCHEDULER
         cout<<ref->getSimulatedObject().getObjectDescriptor().getName()<< " goes to thread "<<(*i).second<<endl;
#endif
         
         _whichThread [ ref ] = (*i).second ;
         
      }
   }
}


 
void PsnMultiThreadedScheduler::removeFromScheduable (PsnReferenceObjectHandle * ref) {

   if (dynamic_cast<const PsDuplicatedObject *>(&((PsnReferenceObjectHandle *)ref)->getSimulatedObject())) {
      _duplicatedObjectScheduler->removeFromScheduable ( ref ) ;
   }
   else {
      (*_localSchedulers) [ _whichThread [ ref ] ] ->removeFromScheduable (ref ) ;
   }
}



void PsnMultiThreadedScheduler::schedule(PsnReferenceObjectHandle * ref) {

   if (dynamic_cast<const PsDuplicatedObject *>(&((PsnReferenceObjectHandle *)ref)->getSimulatedObject())) {
      _duplicatedObjectScheduler->schedule ( ref ) ;
   }
   else {
      (*_localSchedulers) [ _whichThread [ ref ] ] -> schedule (ref) ;
   }
}



void PsnMultiThreadedScheduler::unschedule(PsnReferenceObjectHandle * ref) {

   if (dynamic_cast<const PsDuplicatedObject *>(&((PsnReferenceObjectHandle *)ref)->getSimulatedObject())) {
      _duplicatedObjectScheduler->unschedule ( ref ) ;
   }
   else {
      (*_localSchedulers) [ _whichThread [ ref ] ] -> unschedule (ref) ;
   }
}



void * PsnMultiThreadedScheduler::_threadContainer (void * myThreadData) 
{
   // install the memory managers
   HeapStackTop systemMemoryManager ( HeapStackTop::getSystemMemoryManager() ) ;
  
   PsnMemoryManager * currentMemoryManager = HeapStackTop::getSystemMemoryManager() ;

   assert ( currentMemoryManager != 0 ) ;

   size_t initialAllocatatedMemory = 1024 * 1024 ;

   void * startMemory = currentMemoryManager->allocateSizeRemembered ( initialAllocatatedMemory ) ;

   PsnDescriptorMemoryManager * descriptorMemoryManager = new PsnDescriptorMemoryManager ( *currentMemoryManager ) ;

   descriptorMemoryManager->init() ;
   
   PsnThreadMemoryManager * memoryManagerForThisThread ;
   
   memoryManagerForThisThread = new PsnThreadMemoryManager ( startMemory, 
                                                             initialAllocatatedMemory,
                                                             *currentMemoryManager,
                                                             *descriptorMemoryManager ) ;
   memoryManagerForThisThread -> init() ;

   HeapStackTop allocationContextForThisThread ( memoryManagerForThisThread ) ;


   // get the data specific to this thread
   pair<PsnMultiThreadedScheduler *, unsigned int> * threadData = ( pair<PsnMultiThreadedScheduler *, unsigned int> * ) myThreadData ;

   // get the eventual profiler to profile this thread
   setitimer(ITIMER_PROF, &(( * threadData ).first->_profilingTimer[( * threadData ).second]), NULL);

   cerr<<"Thread now profiled ?"<<endl;

   ( * threadData ).first->run( ( * threadData ).second ) ;

   return threadData ;
}


void PsnMultiThreadedScheduler::run( unsigned int whichFrameScheduler ) 
{

   while ( true ) {
      
      _barrier->Sync() ;
      

      (*_localSchedulers)[ whichFrameScheduler ]->runStep ( _currentStep ) ;


      _barrier->Sync() ;


   }
}

---