RxThreadPoolService.h

Go to the documentation of this file.

// Copyright (C) 2002-2024, Open Design Alliance (the "Alliance"). 
// All rights reserved. 
// 
// This software and its documentation and related materials are owned by 
// the Alliance. The software may only be incorporated into application 
// programs owned by members of the Alliance, subject to a signed 
// Membership Agreement and Supplemental Software License Agreement with the
// Alliance. The structure and organization of this software are the valuable  
// trade secrets of the Alliance and its suppliers. The software is also 
// protected by copyright law and international treaty provisions. Application  
// programs incorporating this software must include the following statement 
// with their copyright notices:
//   
//   This application incorporates Open Design Alliance software pursuant to a license 
//   agreement with Open Design Alliance.
//   Open Design Alliance Copyright (C) 2002-2024 by Open Design Alliance. 
//   All rights reserved.
//
// By use of this software, its documentation or related materials, you 
// acknowledge and accept the above terms.
 
#ifndef _ODRXTHREADPOOLSERVICE_INCLUDED_
#define _ODRXTHREADPOOLSERVICE_INCLUDED_ /* { Secret } **/
 
#include "TD_PackPush.h"
 
#include "RxModule.h"
#include "ThreadsCounter.h"
#include "OdArray.h"
#include "StaticRxObject.h"
 
#include <algorithm>
 
class OdRxThreadPoolService;
 
typedef ptrdiff_t OdApcParamType;
 
typedef void (*OdApcEntryPointVoidParam)( OdApcParamType parameter );
 
typedef void (*OdApcEntryPointRxObjParam)( OdRxObject* parameter );

class ODRX_ABSTRACT OdApcThread : public OdRxObject {
public:
  virtual void asyncProcCall( OdApcEntryPointVoidParam ep, OdApcParamType parameter ) = 0;

  virtual void asyncProcCall( OdApcEntryPointRxObjParam ep, OdRxObject* parameter ) = 0;

  virtual void wait(bool bNoThrow = false) = 0;

  virtual unsigned int getId() const = 0;

  virtual bool hasException() const = 0;

  virtual void processException(bool bReThrow = true, bool bClear = true) = 0;
};

typedef OdSmartPtr<OdApcThread> OdApcThreadPtr;
 

class ODRX_ABSTRACT OdApcAtom : public OdRxObject {
public:
  virtual void apcEntryPoint( OdRxObject* /*pMessage*/ ) {}

  virtual void apcEntryPoint( OdApcParamType /*pMessage*/ ) {}
};

typedef OdSmartPtr<OdApcAtom> OdApcAtomPtr;
 

class OdApcObjectPool : public OdRxObject {
public:
  virtual void reserve( OdUInt32 n ) = 0;

  virtual OdRxObjectPtr take() = 0;

  virtual void put( OdRxObject* pObj ) = 0;
};
 
typedef OdSmartPtr<OdApcObjectPool> OdApcObjectPoolPtr;
 
 
class OdApcObjectPool;

class ODRX_ABSTRACT OdApcQueue : public OdRxObject {
public:
  virtual OdRxThreadPoolService& framework() = 0;

  virtual void setAtomPoolRef( OdApcObjectPool* pAtomPool ) = 0;

  virtual void addEntryPoint( OdApcAtom* pRecipient, OdRxObject* pMessage = 0 ) = 0;

  virtual void addEntryPoint( OdApcAtom* pRecipient, OdApcParamType pMessage ) = 0;

  virtual void wait() = 0;

  virtual void executeMainThreadAction( MainThreadFunc mtFunc, void *pArg ) = 0;

  virtual int numThreads() const = 0;
};

typedef OdSmartPtr<OdApcQueue> OdApcQueuePtr;
 

class OdApcDataReadWriteDispatcher : public OdRxObject {
public:
  virtual void enter() = 0;

  virtual void leave() = 0;

  virtual void lock() = 0;

  virtual void unlock() = 0;

  virtual void lockFromInside() = 0;

  virtual void unlockFromInside() = 0;
};

typedef OdSmartPtr<OdApcDataReadWriteDispatcher> OdApcDataReadWriteDispatcherPtr;
 

class ODRX_ABSTRACT OdApcEvent : public OdRxObject {
public:
  virtual void set() = 0;

  virtual void reset() = 0;

  virtual void wait() = 0;

  virtual void waitAndReset() = 0;
};

typedef OdSmartPtr<OdApcEvent> OdApcEventPtr;
 

class ODRX_ABSTRACT OdApcGateway : public OdRxObject {
public:
  virtual void lockByMain( OdUInt32 numThreads ) = 0;

  virtual void waitByMain() = 0;

  virtual void unlockByMain() = 0;


  virtual void passBySecondary() = 0;
};

typedef OdSmartPtr<OdApcGateway> OdApcGatewayPtr;

class ODRX_ABSTRACT OdApcLoopedGateway : public OdRxObject {
public:

  virtual void init( OdUInt32 numThreads ) = 0;

  virtual void passByMain() = 0;

  virtual void waitByMain() = 0;

  virtual void passByMainNoWait() = 0;


  virtual void passBySecondary() = 0;
};

typedef OdSmartPtr<OdApcLoopedGateway> OdApcLoopedGatewayPtr;
 

enum OdApcStQueueFlags
{
  kStQueueNoFlags         = 0,        // Empty StQueue flags.
 
  kStQueueExecByMain      = (1 << 0), // Use main thread for queue tasks execution.
  kStQueueForceTopLevel   = (1 << 1), // Make StQueue top level even if other registered threads already run.
 
  kStQueueLastFlag        = kStQueueForceTopLevel
};
 

enum OdApcMtQueueFlags
{
  kMtQueueNoFlags         = 0,        // Empty MtQueue flags.
 
  kMtQueueForceNewThreads = (1 << 0), // Spawn new threads and add them to the pool if there are not enough free threads.
  kMtQueueAllowExecByMain = (1 << 1), // Allow to use main thread for execution if there are no free threads.
  kMtQueueForceTopLevel   = (1 << 2), // Make MtQueue top level even if other registered threads already run.
 
  kMtQueueLastFlag        = kMtQueueForceTopLevel
};
 

class FIRSTDLL_EXPORT ODRX_ABSTRACT OdRxThreadPoolService : public OdRxModule {
public:
  ODRX_DECLARE_MEMBERS(OdRxThreadPoolService);

  virtual int numCPUs() const = 0;

  virtual int numPhysicalCores() const = 0;

  virtual int numThreads() const = 0;

  virtual int numFreeThreads() const = 0;

  virtual OdApcThreadPtr newThread() = 0;

  virtual OdApcQueuePtr newSTQueue( unsigned nThreadAttributes = ThreadsCounter::kNoAttributes,
                                    OdUInt32 nFlags = kStQueueNoFlags ) = 0;

  virtual OdApcQueuePtr newMTQueue( unsigned nThreadAttributes = ThreadsCounter::kNoAttributes,
                                    int numThreads = 0, OdUInt32 nFlags = kMtQueueNoFlags ) = 0;

  virtual OdApcObjectPoolPtr newObjectPool() = 0;

  virtual OdApcEventPtr newEvent() = 0;

  virtual OdApcGatewayPtr newGateway() = 0;

  virtual OdApcLoopedGatewayPtr newLoopedGateway() = 0;

  virtual void executeMainThreadAction( MainThreadFunc mtFunc, void *pArg ) = 0;
 
  /* Support for external threads manager */

  virtual void registerExternalThreads( unsigned nThreads, const unsigned* aThreads, unsigned nThreadAttribs = ThreadsCounter::kNoAttributes ) = 0;
  virtual void unregisterExternalThreads( unsigned nThreads, const unsigned* aThreads ) = 0;
  virtual void externalThreadStart() = 0;
  virtual void externalThreadStop() = 0;
  virtual void setExternalMainThreadFunc( ExecuteMainThreadFunc execFunc ) = 0;
  virtual ExecuteMainThreadFunc getExternalMainThreadFunc() const = 0;

  virtual unsigned int getMainThreadId() const = 0;
  virtual unsigned int getCurrentThreadId() const = 0;
  inline bool isMainThread() const
  {
    return getMainThreadId() == getCurrentThreadId();
  }
 
  //FELIX_CHANGE_BEGIN
  virtual void init() = 0;
  //FELIX_CHANGE_END
};
 
typedef OdSmartPtr<OdRxThreadPoolService> OdRxThreadPoolServicePtr;
 

class OdApcQueueHelper : public OdSmartPtr<OdApcQueue> {
public:
  OdApcQueueHelper( ) {}
  OdApcQueueHelper( const OdApcQueue* pObject, OdRxObjMod m ) : OdSmartPtr<OdApcQueue>( pObject, m ) {}
  OdApcQueueHelper( const OdApcQueue* pObject ) : OdSmartPtr<OdApcQueue>( pObject ) {}
  OdApcQueueHelper( const OdRxObject* pObject ) : OdSmartPtr<OdApcQueue>( pObject ) {}
  OdApcQueueHelper( OdRxObject* pObject, OdRxObjMod m ) : OdSmartPtr<OdApcQueue>( pObject, m ) {}
  OdApcQueueHelper( const OdSmartPtr<OdApcQueue>& pObject ) : OdSmartPtr<OdApcQueue>( pObject ) {}
  OdApcQueueHelper( const OdRxObjectPtr& pObject ) : OdSmartPtr<OdApcQueue>( pObject ) {}
  OdApcQueueHelper( const OdBaseObjectPtr& pObject ) : OdSmartPtr<OdApcQueue>( pObject ) {}
 
  //FELIX_CHANGE_BEGIN
  OdApcQueueHelper( const OdApcQueueHelper& ) = default;
  OdApcQueueHelper( OdApcQueueHelper&& ) = default;
  OdApcQueueHelper& operator = ( const OdApcQueueHelper& ) = default;
  OdApcQueueHelper& operator = ( OdApcQueueHelper&& ) = default;
  //FELIX_CHANGE_END
 
  using OdSmartPtr<OdApcQueue>::operator =;
 
  void initST( OdRxThreadPoolService* pThreadPool ) {
    *this = pThreadPool ? pThreadPool->newSTQueue().get() : 0;
  }
 
  void initMT( OdRxThreadPoolService* pThreadPool ) {
    *this = pThreadPool ? pThreadPool->newMTQueue().get() : 0;
  }
 
  void call( OdApcAtom* pAction, OdRxObject* pParam = 0 ) {
    if( m_pObject ) {
      get()->addEntryPoint( pAction, pParam );
    }
    else {
      pAction->apcEntryPoint( pParam );
    }
  }
 
  void call( OdApcAtom* pAction, OdApcParamType param ) {
    if( m_pObject ) {
      get()->addEntryPoint( pAction, param );
    }
    else {
      pAction->apcEntryPoint( param );
    }
  }
 
  void setAtomPoolRef( OdApcObjectPool* pAtomPool ) {
    if( m_pObject ) {
      get()->setAtomPoolRef( pAtomPool );
    }
  }
 
  void wait() {
    if( m_pObject ) {
      get()->wait();
    }
  }
};
 

 

class OdApcQueueHelperArray : public OdArray<OdApcQueueHelper> {
public:
  void initST( int n, OdRxThreadPoolService* pTP ) {
    resize( n );
    if ( pTP ) {
      for( int i=0; i<n; ++i ) {
        setAt( i, pTP->newSTQueue() );
      }
    }
  }
  void wait() {
    size_type n = size();
    for( size_type i=0; i<n; ++i ) {
      at( i ).wait();
    }
  }
};
 

class ODRX_ABSTRACT OdAsyncForEachBase : public OdApcAtom {
  void* sync_next( OdMutex* mutex, OdUInt32 threadIndex, OdUInt32& itemIndex ) {
    OdMutexAutoLock lock( *mutex );
    return next( threadIndex, itemIndex );
  }
protected:
  virtual OdMutex* mutexForNext() = 0;
 
  virtual void* next( OdUInt32 threadIndex, OdUInt32& itemIndex ) = 0;
 
  virtual void doAction( OdUInt32 threadIndex, OdUInt32 itemIndex, void* pItem ) = 0;
 
  virtual void apcEntryPoint( OdApcParamType threadIndex ) {
    OdMutex* mutex = mutexForNext();
    OdUInt32 itemIndex;
    while( void* pItem = sync_next( mutex, (OdUInt32)threadIndex, itemIndex ) ) {
      doAction( (OdUInt32)threadIndex, itemIndex, pItem );
    }
  }
public:
  void for_each( OdApcQueue* pQueue, OdUInt32 nThreads = 0 ) {
    if ( pQueue!=0 ) {
      OdUInt32 n = nThreads==0 ? pQueue->framework().numCPUs() : nThreads;
      if( n > 1 ) {
        while( --n ) {
          pQueue->addEntryPoint( this, (OdApcParamType)n );
        }
        apcEntryPoint( (OdApcParamType)n );
        pQueue->wait();
        return;
      }
    }
    apcEntryPoint( 0 );
  }
};

template< class It, class Fn >
class OdAsyncForEach : public OdStaticRxObject<OdApcAtom> {
  OdMutex                 m_mutex;
  It                      m_cur, m_last;
  Fn                      m_fn;
 
  bool next( It& cur ) {
    OdMutexAutoLock lock( m_mutex );
    if ( m_cur < m_last ) {
      cur = m_cur;
      ++m_cur;
      return true;
    }
    return false;
  }
 
  void apcEntryPoint( OdRxObject* ) {
    It cur;
    if( next( cur ) ) {
      do {
        m_fn( *cur );
      }
      while( next( cur ) );
    }
  }
public:
  void for_each( OdApcQueue* pQueue, It first, It last, Fn fn ) {
    unsigned long n = static_cast<unsigned long>(last-first);
    if( pQueue && n>1 ) {
      m_cur = first;
      m_last = last;
      m_fn = fn;
 
      unsigned long numCPUs = pQueue->framework().numCPUs();
      n = ( n < numCPUs ) ? n : numCPUs;
      while( n-- ) {
        pQueue->addEntryPoint( this );
      }
      pQueue->wait();
    }
    else {
      std::for_each( first, last, fn );
    }
  }
 
};
 
//DOM-IGNORE-BEGIN
template< class It, class Fn >
void od_async_for_each( OdApcQueue* pQueue, It first, It last, Fn fn ) {
  ODA_ASSERT_ONCE( first <= last );
  if( first < last ) {
    OdAsyncForEach< It, Fn >().for_each( pQueue, first, last, fn );
  }
}
//DOM-IGNORE-END

template< class TObject >
struct OdApcObjectPoolHelperDummyInitFn {
  void operator () ( TObject& ) const {
    // does nothing
  }
};

template< class TObject, class TInitFn = OdApcObjectPoolHelperDummyInitFn<TObject> >
class OdApcObjectPoolHelper : public OdApcObjectPoolPtr {
  TObject* m_pEntries;
  OdUInt32 m_nCount;
public:
  OdApcObjectPoolHelper() : m_pEntries( NULL ), m_nCount(0) {}
  ~OdApcObjectPoolHelper() { release(); ::delete [] m_pEntries; m_pEntries = NULL; m_nCount=0; }
  void init( OdRxThreadPoolService* pTP, int n = 0, const TInitFn* pInitFn = 0 ) {
    if ( pTP ) {
      if( n==0 ) {
        n = pTP->numCPUs();
      }
      m_nCount = n;
      m_pEntries = ::new TObject[n];
      (*(OdApcObjectPoolPtr*)this) = pTP->newObjectPool();
      while( --n ) {
        TObject& obj = m_pEntries[ n ];
        if( pInitFn ) {
          (*pInitFn)( obj );
        }
        get()->put( &obj );
      }
    }
    else {
      m_nCount = 1;
      m_pEntries = ::new TObject[1];
      if( pInitFn ) {
        (*pInitFn)( *m_pEntries );
      }
    }
  }
 
  TObject* take() {
    if( m_pObject ) {
      return static_cast<TObject*>( get()->take().get() );
    }
    return m_pEntries;
  }
 
  OdUInt32 size() const {
    return m_nCount;
  }
 
  TObject& at( OdUInt32 i ) const {
    if( i < m_nCount ) {
      return m_pEntries[i];
    }
    throw OdError_InvalidIndex();
  }
 
  operator OdApcObjectPool* () { return get(); }
  operator const OdApcObjectPool* () const { return get(); }
};
 
#include "TD_PackPop.h"
 
#endif //_ODRXTHREADPOOLSERVICE_INCLUDED_