parallelTickManager.h

Go to the documentation of this file.

/*******************************************************************************
** Copyright (c) 2014 MAK Technologies, Inc.
** All rights reserved.
*******************************************************************************/



#pragma once

#include "vrfcore/vrfcoreDefines.h"

#include "vrfutil/vrfCallbackQueue.h"
#include "vrfutil/profiler.h"

#include <boost/noncopyable.hpp>
#include <boost/unordered/unordered_map.hpp>
#include <boost/ptr_container/ptr_vector.hpp>

#include <tbb/mutex.h>
#include <tbb/tbb_thread.h>
#include <tbb/compat/condition_variable>
#include <tbb/enumerable_thread_specific.h>
#include <tbb/atomic.h>
#include <tbb/concurrent_queue.h>

#include <vector>

#include "vrfutil/itt.h"

class DtConsoleCommandOwner;
class DtSubCommandManager;
class DtVrfSimOptionsInterface;
class DtParallelTickManager;
class DtCriticalSection;
class DtSimManager;

class DT_DLL_vrfcore DtParallelTickManager : boost::noncopyable
{
public:
   enum AccessType
   {
      EXCLUSIVE,
      SHARED,
      UNLOCKED,
      DEFERRED    
   };

   class DT_DLL_vrfcore CallbackInterface : private DtVrfCallbackQueue::CallbackInterface
   {
      friend class DtParallelTickManager;

      void run();

   public:
      virtual ~CallbackInterface();

      virtual void runCallback() = 0;
   };

   typedef boost::unordered_map<std::string, bool> EnabledMap;

   explicit DtParallelTickManager(DtVrfSimOptionsInterface* options);
   virtual ~DtParallelTickManager();

   virtual void setCallbackQueue(DtVrfCallbackQueue*);

   virtual DtVrfCallbackQueue* queue() const;


   virtual void flushQueue(bool useThisThread);
   virtual void flushQueue();

   virtual void clearDeferredJobs();


   virtual bool mainThreadWait() const;
   virtual void setMainThreadWait(bool);


   virtual unsigned maxSpinCount() const;
   virtual void setMaxSpinCount(unsigned);


   virtual bool active() const;
   virtual void setActive(bool);

   virtual bool enabled(const std::string&, bool def) const;

   virtual bool executeCommand(const std::string&, bool on);

   virtual void printInfo(std::ostream&) const;


   virtual void addCallback(CallbackInterface*);
   virtual void addCallback(CallbackInterface&);

   virtual void addCallback(CallbackInterface*, const std::string& tag, bool def=true);
   virtual void addCallback(CallbackInterface&, const std::string& tag, bool def=true);

   void addCallback(CallbackInterface*, const char* tag, bool def=true);
   void addCallback(CallbackInterface&, const char* tag, bool def=true);

   template <typename Container>
   void addCallbacks(Container& cont)
   {
      typedef typename Container::iterator iterator;
      for (iterator i=cont.begin(), end=cont.end(); i != end; ++i)
         addCallback(*i);
   }

   template <typename Container>
   void addCallbacks(Container& cont, const std::string& tag, bool def=true)
   {
      bool runNow = !enabled(tag, def);
      typedef typename Container::iterator iterator;
      if (runNow)
      {
         for (iterator i = cont.begin(), end = cont.end(); i != end; ++i)
         {
            RunJob(*i);
         }
      }
      else
      {
         for (iterator i = cont.begin(), end = cont.end(); i != end; ++i)
         {
            addCallback(*i);
         }
      }
   }

   template <typename Container>
   void addCallbacks(Container& cont, const char* tag, bool def=true)
   {
      addCallbacks(cont, std::string(tag), def);
   }


   virtual bool lock();
   virtual bool unlock();
   virtual bool readLock();
   virtual bool readUnlock();
   virtual bool upgradeToWriter();
   virtual bool downgradeToReader();

   virtual bool changeAccess(AccessType from, AccessType to);



   enum JobClass
   {
      JobClass_VRLINK,
      JobClass_MAX
   };

   virtual void configureDeferredJobClass(
      JobClass jobclass,
      AccessType access,
      unsigned maxConcurrent);

   virtual void addDeferredJob(JobClass jobclass, const boost::function<void ()>& callback);


   virtual void addVRLinkDeferredJob(const boost::function<void()>& callback);

   virtual void setDeferJobs(bool);

   class ThreadLocalInfo;

   virtual ThreadLocalInfo& threadLocalInfo();

private:
   static void RunJob(CallbackInterface*);
   static void RunJob(CallbackInterface&);

   virtual void executeDeferredJobs(AccessType access);
   virtual void flushDeferredJobs();

   tbb::atomic<int> myAccessCount;

   typedef tbb::cache_aligned_allocator<ThreadLocalInfo*> ThreadLocalAllocator;
   //typedef tbb::enumerable_thread_specific<
   typedef tbb::enumerable_thread_specific<
      ThreadLocalInfo*,
      ThreadLocalAllocator,
      tbb::ets_key_per_instance> ThreadLocalContainer;

   ThreadLocalContainer myThreadLocals;

   DtVrfSimOptionsInterface* myOptions;
   DtVrfCallbackQueue* myCallbackQueue;

   mutable EnabledMap myEnabledMap;
   bool myWaitFlag;
   unsigned myMaxSpinCount;

   struct ClassInfo
   {
      typedef tbb::concurrent_queue< boost::function<void()> > JobQueue;

      JobQueue queue;
      tbb::atomic<unsigned> jobCount;
      tbb::atomic<unsigned> runCount;
      unsigned max;
      AccessType access;

      explicit ClassInfo(
         AccessType access = EXCLUSIVE,
         unsigned max = 1);
   };

   bool myDeferJobsFlag;
   boost::ptr_vector<ClassInfo> myDeferredJobs;
   tbb::atomic<unsigned> myDeferredJobCount;

   DtSubCommandManager* myCommandManager;
   DtConsoleCommandOwner* myCommandOwner;

   __itt_domain* myLockITTDomain;

   __itt_string_handle* mySharedLockEvent;
   __itt_string_handle* myExclusiveLockEvent;
   __itt_string_handle* mySharedAccessEvent;
   __itt_string_handle* myExclusiveAccessEvent;
   __itt_string_handle* myDeferredJobTask;
};

class DT_DLL_vrfcore DtCriticalSection : boost::noncopyable
{
public:
   typedef DtParallelTickManager::AccessType AccessType;

   explicit DtCriticalSection(DtSimManager*, AccessType initState);

   explicit DtCriticalSection(DtSimManager*);


   explicit DtCriticalSection(DtCriticalSection&, AccessType to);
   explicit DtCriticalSection(DtCriticalSection&);

   ~DtCriticalSection();

   void reset(AccessType access);

   void unlock();

   void lockShared();

   void lock();

   void release()
   {
      reset(from);
   }

protected:
   explicit DtCriticalSection(DtParallelTickManager*, AccessType to);
   explicit DtCriticalSection(DtParallelTickManager*);

   friend class DtParallelTickManager;

   DtParallelTickManager* mgr;
   DtParallelTickManager::ThreadLocalInfo* info;

   AccessType from;
   AccessType to;
};

class DT_DLL_vrfcore DtParallelTickManager::ThreadLocalInfo : boost::noncopyable
{
public:
   explicit ThreadLocalInfo(DtParallelTickManager* m);

   DtParallelTickManager::AccessType state() const;

private:
   friend class DtCriticalSection;

   void push(DtCriticalSection& cs);
   void pop(DtCriticalSection& cs);

   typedef std::vector<DtCriticalSection*> Stack;
   Stack myLocks;
};