vrfCallbackQueue.h

Go to the documentation of this file.

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


#pragma once

#include "vrfutil/vrfutilDefines.h"
#include <vrfutil/objectCounter.h>
#include <atomic>
#include <boost/function.hpp>
#include <boost/noncopyable.hpp>
#include <tbb/tick_count.h>
#include <thread>

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

#include <condition_variable>

class DtConsoleCommandManager;

class DT_DLL_vrfutil DtVrfCallbackQueue : boost::noncopyable
{
   struct ThreadInfo 
#ifdef DtObjectDebugger_Enable
      : public DtObjectDebugger<ThreadInfo>
#endif
   {
      typedef std::shared_ptr<ThreadInfo> Ptr;
      typedef std::list<Ptr> List;

      static Ptr Make(DtVrfCallbackQueue&, unsigned threadnum);

      bool quitFlag;
      std::string name;
      std::thread thread;

   protected:
      explicit ThreadInfo(DtVrfCallbackQueue&, const std::string& name);
   };

   friend struct ThreadInfo;

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

   class DT_DLL_vrfutil CallbackInterface
#ifdef DtObjectDebugger_Enable
      : public DtObjectDebugger<CallbackInterface>
#endif
   {
   public:
      typedef std::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 DtVrfCallbackQueue(
      const std::string& label,
      unsigned maxTasks);

   virtual ~DtVrfCallbackQueue();

   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 
#ifdef DtObjectDebugger_Enable
      : public DtObjectDebugger<Record>
#endif
   {
      typedef std::list<Record> List;

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

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

   typedef std::mutex Mutex;
   typedef std::condition_variable Condition;
   typedef std::unique_lock<Mutex> Lock;

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

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

   bool myShutdownFlag;
   size_t myTotalNumCallbacks;

   std::atomic<size_t> myNumExecuting;

   DtConsoleCommandManager* myConsoleManager;

   std::string myLabel;

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

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

protected:
   DtVrfCallbackQueue::CallbackInterface* myCallback;
};

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

protected:
   DtVrfCallbackQueue::CallbackInterface& myCallback;
};

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