callbackQueue.h

Go to the documentation of this file.

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


#pragma once

#include "vrfutil/vrfutilDefines.h"
#include "vrfutil/objectCounter.h"

#include <boost/function.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/noncopyable.hpp>

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

#include <list>
#include <map>
#include <iosfwd>

class DtConsoleCommandManager;

class DT_DLL_vrfutil DtCallbackQueue : boost::noncopyable
{
   struct ThreadInfo : public DtObjectDebugger<ThreadInfo>
   {
      typedef boost::shared_ptr<ThreadInfo> Ptr;
      typedef std::list<Ptr> List;

      static Ptr Make(DtCallbackQueue&);

      bool quitFlag;
      tbb::tbb_thread thread;

   protected:
      explicit ThreadInfo(DtCallbackQueue&);
   };

   friend struct ThreadInfo;

public:
   typedef boost::function<void ()> Callback;

   class DT_DLL_vrfutil CallbackInterface
      : public DtObjectDebugger<CallbackInterface>
   {
   public:
      typedef boost::shared_ptr<CallbackInterface> Ptr;

      class Wrapper;
      class RefWrapper;

      virtual ~CallbackInterface();

      virtual std::string label() const;

      virtual void run() = 0;

      virtual void cancel();

      virtual bool valid() const;
   };

   explicit DtCallbackQueue(
      const std::string& label,
      unsigned maxTasks);

   virtual ~DtCallbackQueue();

   virtual std::string label() const;

   static void cleanup();

   virtual void setWarnTime(double seconds);
   virtual void setWarnTime(double seconds, std::ostream& output);

   virtual void printInfo(std::ostream&, size_t width=0) const;

   virtual size_t totalNumberOfCallbacks() const;

   virtual unsigned size() const;

   virtual size_t numExecuting() const;

   virtual unsigned maxTasks() const;

   virtual void setMaxTasks(unsigned tasks, bool wait=true);

   void shutdown(bool wait = true);

   virtual void clear(bool wait = true);

   virtual void add_back(const Callback&, const std::string& label = "");

   virtual void add_front(const Callback&, const std::string& label = "");

   virtual void add_back(CallbackInterface*);

   virtual void add_front(CallbackInterface*);

   virtual void add_back(const CallbackInterface::Ptr&);

   virtual void add_front(const CallbackInterface::Ptr&);

   virtual void add_back(CallbackInterface&);

   virtual void add_front(CallbackInterface&);

   virtual bool runOne(bool block=false, ThreadInfo* info=0);

   virtual void runAll(bool useThisThread=false);

private:
   virtual bool pop(
      CallbackInterface::Ptr&, bool block=true, ThreadInfo* info=0);

   void execute(ThreadInfo*);

   struct Record : public DtObjectDebugger<Record>
   {
      typedef std::list<Record> List;

      const CallbackInterface::Ptr callback;
      const tbb::tick_count timeAdded;

      explicit Record(CallbackInterface*);
      explicit Record(const CallbackInterface::Ptr&);
   };

   typedef tbb::mutex Mutex;
   typedef tbb::interface5::condition_variable Condition;
   typedef tbb::interface5::unique_lock<Mutex> Lock;

   mutable Mutex myMutex;
   mutable Condition myCondition;
   mutable Condition myDoneCondition;

   ThreadInfo::List myThreads;
   Record::List myCallbacks;

   bool myShutdownFlag;
   size_t myTotalNumCallbacks;

   tbb::atomic<size_t> myNumExecuting;

   DtConsoleCommandManager* myConsoleManager;

   std::string myLabel;

   typedef std::multimap<double, std::ostream*, std::greater<double> > StreamMap;
   StreamMap myWarnTimes;
};

class DtCallbackQueue::CallbackInterface::Wrapper
   : public DtCallbackQueue::CallbackInterface
{
public:
   explicit Wrapper(DtCallbackQueue::CallbackInterface* i) : myCallback(i) { }
   ~Wrapper()                { delete myCallback; }
   std::string label() const { return myCallback->label(); }
   void run()                { myCallback->run(); }
   void cancel()             { myCallback->cancel(); }

protected:
   DtCallbackQueue::CallbackInterface* myCallback;
};

class DtCallbackQueue::CallbackInterface::RefWrapper
   : public DtCallbackQueue::CallbackInterface
{
public:
   explicit RefWrapper(DtCallbackQueue::CallbackInterface& i) : myCallback(i) { }
   std::string label() const { return myCallback.label(); }
   void run()                { myCallback.run(); }
   void cancel()             { myCallback.cancel(); }

protected:
   DtCallbackQueue::CallbackInterface& myCallback;
};

DT_DEFINE_OBJECT_DEBUGGER_NAME(DtCallbackQueue::ThreadInfo);
DT_DEFINE_OBJECT_DEBUGGER_NAME(DtCallbackQueue::CallbackInterface);
DT_DEFINE_OBJECT_DEBUGGER_NAME(DtCallbackQueue::Record);