etevent.h

Go to the documentation of this file.

/*
* Copyright (c) 2019 Sprint
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
*    http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

#ifndef __ETEVENT_H
#define __ETEVENT_H

#include <unistd.h>
#include <sys/syscall.h>

#include "ebase.h"
#include "etbasic.h"
#include "eerror.h"
#include "egetopt.h"
#include "eshmem.h"
#include "esynch.h"
#include "esynch2.h"
#include "etime.h"
#include "etimer.h"



DECLARE_ERROR(EThreadQueueBaseError_NotOpenForWriting);
DECLARE_ERROR(EThreadQueueBaseError_NotOpenForReading);
DECLARE_ERROR(EThreadQueueBaseError_MultipleReadersNotAllowed);
DECLARE_ERROR(EThreadQueuePublicError_UnInitialized);

DECLARE_ERROR_ADVANCED(EThreadTimerError_UnableToInitialize);
DECLARE_ERROR_ADVANCED(EThreadTimerError_NotInitialized);
DECLARE_ERROR_ADVANCED(EThreadTimerError_AlreadyInitialized);
DECLARE_ERROR_ADVANCED(EThreadTimerError_UnableToStart);
DECLARE_ERROR_ADVANCED(EThreadTimerError_UnableToRegisterTimerHandler);



class EThreadEventMessageDataBase
{
public:
   EThreadEventMessageDataBase() : m_msgid() {}
   EThreadEventMessageDataBase(UInt msgid) : m_msgid(msgid) {}
   virtual ~EThreadEventMessageDataBase() {}

   UInt getMessageId() { return m_msgid; }
   EThreadEventMessageDataBase &setMessageId(UInt msgid) { m_msgid = msgid; return *this; }
   virtual pVoid getVoidPtr() = 0;
   virtual Void setVoidPtr(pVoid p) = 0;

   ETimer &getTimer() { return m_timer; }

private:   
   ETimer m_timer;
   UInt m_msgid;
};

class EThreadEventMessageData : public EThreadEventMessageDataBase
{
public:
   typedef union {
      pVoid       voidptr;
      LongLong    int64;
      Long        int32[sizeof(pVoid) / sizeof(Long)];
      Short       int16[sizeof(pVoid) / sizeof(Short)];
      Char        int8[sizeof(pVoid) / sizeof(Char)];
      ULongLong   uint64;
      ULong       uint32[sizeof(pVoid) / sizeof(ULong)];
      UShort      uint16[sizeof(pVoid) / sizeof(UShort)];
      UChar       uint8[sizeof(pVoid) / sizeof(UChar)];
   } DataUnion;

   EThreadEventMessageData() : EThreadEventMessageDataBase(), m_data() {}
   EThreadEventMessageData(UInt msgid) : EThreadEventMessageDataBase(msgid), m_data() {}
   EThreadEventMessageData(UInt msgid, EThreadEventMessageData &data) : EThreadEventMessageDataBase(msgid), m_data(data.data()) {}
   EThreadEventMessageData(UInt msgid, pVoid v) : EThreadEventMessageDataBase(msgid), m_data{.voidptr=v} {}
   EThreadEventMessageData(UInt msgid, LongLong v) : EThreadEventMessageDataBase(msgid), m_data{.int64=v} {}
   EThreadEventMessageData(UInt msgid, ULongLong v) : EThreadEventMessageDataBase(msgid), m_data{.uint64=v} {}
   EThreadEventMessageData(UInt msgid, Long v1, Long v2) : EThreadEventMessageDataBase(msgid), m_data{.int32={v1,v2}} {}
   EThreadEventMessageData(UInt msgid, ULong v1, ULong v2) : EThreadEventMessageDataBase(msgid), m_data{.uint32={v1,v2}} {}
   EThreadEventMessageData(UInt msgid, Short v1, Short v2, Short v3, Short v4) : EThreadEventMessageDataBase(msgid), m_data{.int16={v1,v2,v3,v4}} {}
   EThreadEventMessageData(UInt msgid, UShort v1, UShort v2, UShort v3, UShort v4) : EThreadEventMessageDataBase(msgid), m_data{.uint16={v1,v2,v3,v4}} {}
   EThreadEventMessageData(UInt msgid, Char v1, Char v2, Char v3, Char v4, Char v5, Char v6, Char v7, Char v8) : EThreadEventMessageDataBase(msgid), m_data{.int8={v1,v2,v3,v4,v5,v6,v7,v8}} {}
   EThreadEventMessageData(UInt msgid, UChar v1, UChar v2, UChar v3, UChar v4, UChar v5, UChar v6, UChar v7, UChar v8) : EThreadEventMessageDataBase(msgid), m_data{.uint8={v1,v2,v3,v4,v5,v6,v7,v8}} {}
   virtual ~EThreadEventMessageData() {}

   DataUnion &data() { return m_data; }

   pVoid getVoidPtr() { return m_data.voidptr; }
   Void setVoidPtr(pVoid p) { m_data.voidptr = p; }

private:
   DataUnion m_data;
};

class _EThreadEventMessageBase
{
public:
   virtual ~_EThreadEventMessageBase() {}
};

template <class T>
class EThreadEventMessageBase : public _EThreadEventMessageBase
{
public:
   EThreadEventMessageBase() : m_data() {}
   EThreadEventMessageBase(const T &data) : m_data( data ) {}
   virtual ~EThreadEventMessageBase() {}
   T &operator=(const T &data)
   {
      m_data = data;
      return *this;
   }
   T &data()
   {
      return m_data;
   }
   UInt getMessageId() { return m_data.getMessageId(); }
   EThreadEventMessageBase &setMessageId(UInt msgid) { m_data.setMessageid( msgid ); return *this; }
   pVoid getVoidPtr() { return m_data.getVoidPtr(); }
   Void setVoidPtr(pVoid p) { m_data.setVoidPtr(p); }
   ETimer &getTimer() { return m_data.getTimer(); }

private:
   T m_data;
};

typedef EThreadEventMessageBase<EThreadEventMessageData> _EThreadMessage;

class EThreadMessage : public _EThreadMessage
{
public:
   EThreadMessage() : _EThreadMessage() {}
   EThreadMessage(UInt msgid) : _EThreadMessage(EThreadEventMessageData(msgid)) {}
   EThreadMessage(UInt msgid, EThreadEventMessageData &data) : _EThreadMessage(data) {}
   EThreadMessage(UInt msgid, pVoid v) : _EThreadMessage(EThreadEventMessageData(msgid, v)) {}
   EThreadMessage(UInt msgid, LongLong v) : _EThreadMessage(EThreadEventMessageData(msgid, v)) {}
   EThreadMessage(UInt msgid, ULongLong v) : _EThreadMessage(EThreadEventMessageData(msgid, v)) {}
   EThreadMessage(UInt msgid, Long v1, Long v2=0) : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2)) {}
   EThreadMessage(UInt msgid, ULong v1, ULong v2=0) : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2)) {}
   EThreadMessage(UInt msgid, Short v1, Short v2=0, Short v3=0, Short v4=0) : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2, v3, v4)) {}
   EThreadMessage(UInt msgid, UShort v1, UShort v2=0, UShort v3=0, UShort v4=0) : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2, v3, v4)) {}
   EThreadMessage(UInt msgid, Char v1, Char v2=0, Char v3=0, Char v4=0, Char v5=0, Char v6=0, Char v7=0, Char v8=0)
      : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2, v3, v4, v5, v6, v7, v8)) {}
   EThreadMessage(UInt msgid, UChar v1, UChar v2=0, UChar v3=0, UChar v4=0, UChar v5=0, UChar v6=0, UChar v7=0, UChar v8=0)
      : _EThreadMessage(EThreadEventMessageData(msgid, v1, v2, v3, v4, v5, v6, v7, v8)) {}
   virtual ~EThreadMessage() {}
};


enum class EThreadQueueMode
{
   ReadOnly,
   WriteOnly,
   ReadWrite
};

template <class T>
class EThreadQueueBase
{
public:
   Int queueSize() const { return msgCnt(); }

   Bool push(const T &msg, Bool wait = True)
   {
      if (m_mode == EThreadQueueMode::ReadOnly)
         throw EThreadQueueBaseError_NotOpenForWriting();

      if (!semFree().Decrement(wait))
         return False;

      {
         EMutexLock l(mutex());

         data()[msgHead()] = msg;
         data()[msgHead()].data().getTimer().Start();

         msgHead()++;

         if (msgHead() >= msgCnt())
            msgHead() = 0;
      }

      semMsgs().Increment();

      return True;
   }
   Bool pop(T &msg, Bool wait = True)
   {
      if (m_mode == EThreadQueueMode::WriteOnly)
         throw EThreadQueueBaseError_NotOpenForReading();

      if (!semMsgs().Decrement(wait))
         return False;

      {
         EMutexLock l(mutex(),False);

         if (multipleReaders())
            l.acquire();

         msg = data()[msgTail()++];

         if (msgTail() >= msgCnt())
            msgTail() = 0;

         semFree().Increment();
      }

      return True;
   }

   Bool isInitialized() { return m_initialized; }
   EThreadQueueMode mode() { return m_mode; }

protected:

   virtual Bool isPublic() = 0;
   virtual Int &msgCnt() = 0;
   virtual Int &msgHead() = 0;
   virtual Int &msgTail() = 0;
   virtual Bool &multipleReaders() = 0;
   virtual Bool &multipleWriters() = 0;
   virtual Int &numReaders() = 0;
   virtual Int &numWriters() = 0;
   virtual Int &refCnt() = 0;
   virtual T *data() = 0;
   virtual Void allocDataSpace(cpStr sFile, Char cId, Int nSize) = 0;
   virtual Void initMutex() = 0;
   virtual Void initSemFree(UInt initialCount) = 0;
   virtual Void initSemMsgs(UInt initialCount) = 0;

   virtual EMutexData &mutex() = 0;
   virtual ESemaphoreData &semMsgs() = 0;
   virtual ESemaphoreData &semFree() = 0;

   virtual int *getBumpPipe() = 0;

   EThreadQueueBase()
   {
      m_initialized = False;
   }
   ~EThreadQueueBase()
   {
   }

   Void init(Int nMsgCnt, Int threadId, Bool bMultipleWriters, EThreadQueueMode eMode, Bool bMultipleReaders = False)
   {
      m_mode = eMode;

      // construct the shared memory name
      Char szName[EPC_FILENAME_MAX];
      epc_sprintf_s(szName, sizeof(szName), "%d", threadId);

      // calcuate the space required
      int nSize = sizeof(T) * nMsgCnt;

      // initialize the shared memory
      allocDataSpace(szName, 'A', nSize);

      // initialize the control block values
      if (refCnt() == 0)
      {
         refCnt() = 0;
         numReaders() = 0;
         numWriters() = 0;
         multipleReaders() = bMultipleReaders;
         multipleWriters() = bMultipleWriters;
         msgCnt() = nMsgCnt;
         msgHead() = 0;
         msgTail() = 0;

         // initialize the control mutex and semaphores
         initMutex();
         initSemFree(msgCnt());
         initSemMsgs(0);
      }

      attach(eMode);

      m_initialized = True;
   }

   Void attach(EThreadQueueMode eMode)
   {
      EMutexLock l(mutex());

      if (!multipleReaders() && numReaders() > 0 &&
          (eMode == EThreadQueueMode::ReadOnly || eMode == EThreadQueueMode::ReadWrite))
      {
         throw EThreadQueueBaseError_MultipleReadersNotAllowed();
      }

      refCnt()++;
      numReaders() += (eMode == EThreadQueueMode::ReadOnly || eMode == EThreadQueueMode::ReadWrite) ? 1 : 0;
      numWriters() += (eMode == EThreadQueueMode::WriteOnly || eMode == EThreadQueueMode::ReadWrite) ? 1 : 0;
   }

   Void destroy()
   {
      Bool destroyMutex = False;

      if (m_initialized)
      {
         EMutexLock l(mutex());

         if (refCnt() == 1)
         {
            semFree().destroy();
            semMsgs().destroy();

            destroyMutex = True;

            m_initialized = False;
         }
         else
         {
            refCnt()--;
            numReaders() -= (m_mode == EThreadQueueMode::ReadOnly || m_mode == EThreadQueueMode::ReadWrite) ? 1 : 0;
            numWriters() -= (m_mode == EThreadQueueMode::WriteOnly || m_mode == EThreadQueueMode::ReadWrite) ? 1 : 0;
         }
      }

      if (destroyMutex)
         mutex().destroy();
   }

private:
   Bool _push(const _EThreadEventMessageBase &msg, Bool wait)
   {
      return push( (const T &)msg, wait );
   }

   Bool m_initialized;
   EThreadQueueMode m_mode;
};


template <class T>
class EThreadQueuePublic : public EThreadQueueBase<T>
{
   template <class TQueue, class TMessage> friend class EThreadEvent;
   template <class TQueue, class TMessage> friend class EThreadEventWorker;
   template <class TQueue, class TMessage, class TWorker> friend class EThreadEventWorkGroup;
public:
   EThreadQueuePublic()
   {
      m_pCtrl = NULL;
      m_pData = NULL;
   }
   ~EThreadQueuePublic()
   {
      EThreadQueueBase<T>::destroy();
   }
   Void init(Int nMsgCnt, Int threadId, Bool bMultipleWriters,
             EThreadQueueMode eMode)
   {
      EThreadQueueBase<T>::init(nMsgCnt, threadId, bMultipleWriters, eMode);
   }

protected:
   Void init(Int nMsgCnt, Int threadId, Bool bMultipleWriters,
             EThreadQueueMode eMode, Bool bMultipleReaders)
   {
      EThreadQueueBase<T>::init(nMsgCnt, threadId, bMultipleWriters, eMode, bMultipleReaders);
   }

   Bool isPublic() { return True; }
   Int &msgCnt() { return m_pCtrl->m_msgCnt; }
   Int &msgHead() { return m_pCtrl->m_head; }
   Int &msgTail() { return m_pCtrl->m_tail; }
   Bool &multipleReaders() { return m_pCtrl->m_multipleReaders; }
   Bool &multipleWriters() { return m_pCtrl->m_multipleWriters; }
   Int &numReaders() { return m_pCtrl->m_numReaders; }
   Int &numWriters() { return m_pCtrl->m_numWriters; }
   Int &refCnt() { return m_pCtrl->m_refCnt; }
   T *data() { return m_pData; }
   Void allocDataSpace(cpStr sFile, Char cId, Int nSize)
   {
      m_sharedmem.init(sFile, cId, nSize + sizeof(ethreadmessagequeue_ctrl_t));
      m_pCtrl = (ethreadmessagequeue_ctrl_t *)m_sharedmem.getDataPtr();
      m_pData = (T *)(((pChar)m_sharedmem.getDataPtr()) + sizeof(ethreadmessagequeue_ctrl_t));
   }
   Void initMutex()
   {
      EMutexPublic m;
      m_pCtrl->m_mutexid = m.mutexId();
      m.detach();
   }
   Void initSemFree(UInt initialCount)
   {
      ESemaphorePublic s(initialCount);
      m_pCtrl->m_freeSemId = s.semIndex();
      s.detach();
   }
   Void initSemMsgs(UInt initialCount)
   {
      ESemaphorePublic s(initialCount);
      m_pCtrl->m_msgsSemId = s.semIndex();
      s.detach();
   }

   EMutexData &mutex() { return ESynchObjects::getMutex(m_pCtrl->m_mutexid); }
   ESemaphoreData &semFree() { return ESynchObjects::getSemaphore(m_pCtrl->m_freeSemId); }
   ESemaphoreData &semMsgs() { return ESynchObjects::getSemaphore(m_pCtrl->m_msgsSemId); }

   int *getBumpPipe() { return m_pCtrl->m_bumppipe; }

private:
   typedef struct
   {
      Int m_refCnt;
      Int m_numReaders;
      Int m_numWriters;
      Bool m_multipleReaders;
      Bool m_multipleWriters;

      Int m_msgCnt;
      Int m_head; // next location to write
      Int m_tail; // next location to read

      Int m_mutexid;
      Int m_freeSemId;
      Int m_msgsSemId;

      int m_bumppipe[2];
   } ethreadmessagequeue_ctrl_t;

   ESharedMemory m_sharedmem;
   ethreadmessagequeue_ctrl_t *m_pCtrl;
   T *m_pData;
};


template <class T>
class EThreadQueuePrivate : public EThreadQueueBase<T>
{
   template <class TQueue, class TMessage> friend class EThreadEvent;
   template <class TQueue, class TMessage> friend class EThreadEventWorker;
   template <class TQueue, class TMessage, class TWorker> friend class EThreadEventWorkGroup;
public:
   EThreadQueuePrivate()
      : m_mutex(False), m_semFree(0, False), m_semMsgs(0, False)
   {
      m_refCnt = 0;
      m_numReaders = 0;
      m_numWriters = 0;
      m_multipleReaders = False;
      m_multipleWriters = False;

      m_msgCnt = 0;
      m_head = 0; // next location to write
      m_tail = 0; // next location to read
      m_pData = NULL;
   }
   ~EThreadQueuePrivate()
   {
      EThreadQueueBase<T>::destroy();
      if (m_pData && refCnt() == 1)
         delete[](pChar) m_pData;
      m_pData = NULL;
   }
   Void init(Int nMsgCnt, Int threadId, Bool bMultipleWriters,
             EThreadQueueMode eMode)
   {
      EThreadQueueBase<T>::init(nMsgCnt, threadId, bMultipleWriters, eMode);
   }

protected:
   Void init(Int nMsgCnt, Int threadId, Bool bMultipleWriters,
             EThreadQueueMode eMode, Bool bMultipleReaders)
   {
      EThreadQueueBase<T>::init(nMsgCnt, threadId, bMultipleWriters, eMode, bMultipleReaders);
   }

   Bool isPublic() { return False; }
   Int &msgCnt() { return m_msgCnt; }
   Int &msgHead() { return m_head; }
   Int &msgTail() { return m_tail; }
   Bool &multipleReaders() { return m_multipleReaders; }
   Bool &multipleWriters() { return m_multipleWriters; }
   Int &numReaders() { return m_numReaders; }
   Int &numWriters() { return m_numWriters; }
   Int &refCnt() { return m_refCnt; }
   T *data() { return m_pData; }
   Void allocDataSpace(cpStr sFile, Char cId, Int nSize)
   {
      if (!m_pData)
      {
         m_pData = (EThreadMessage *)new Char[nSize];
         memset((pChar)m_pData, 0, nSize);
      }
   }
   Void initMutex()
   {
      m_mutex.init();
   }
   Void initSemFree(UInt initialCount)
   {
      m_semFree.init(initialCount);
   }
   Void initSemMsgs(UInt initialCount)
   {
      m_semMsgs.init(initialCount);
   }

   EMutexData &mutex() { return m_mutex; }
   ESemaphoreData &semFree() { return m_semFree; }
   ESemaphoreData &semMsgs() { return m_semMsgs; }

   int *getBumpPipe() { return m_bumppipe; }

private:
   Int m_refCnt;
   Int m_numReaders;
   Int m_numWriters;
   Bool m_multipleReaders;
   Bool m_multipleWriters;

   Int m_msgCnt;
   Int m_head; // next location to write
   Int m_tail; // next location to read

   EMutexPrivate m_mutex;
   ESemaphorePrivate m_semFree;
   ESemaphorePrivate m_semMsgs;

   int m_bumppipe[2];

   T *m_pData;
};


#define EM_INIT 1
#define EM_QUIT 2
#define EM_SUSPEND 3
#define EM_TIMER 4
#define EM_SOCKETSELECT_READ 5
#define EM_SOCKETSELECT_WRITE 6
#define EM_SOCKETSELECT_ERROR 7
#define EM_SOCKETSELECT_EXCEPTION 8
#define EM_USER 10000


class _EThreadEventNotification
{
public:
   virtual Bool _sendTimerExpiration(const _EThreadEventMessageBase &msg, Bool wait = True) = 0;
   virtual Bool _sendThreadMessage(const _EThreadEventMessageBase &msg, Bool wait = True) = 0;
};

class EThreadEventTimer : public EStatic
{
   friend class EThreadEventTimerHandler;
   template <class TQueue, class TMessage> friend class EThreadEvent;
   template <class TQueue, class TMessage, class TWorker> friend class EThreadEventWorkGroup;
protected:
   Void init(_EThreadEventNotification *notify, _EThreadEventMessageBase *msg)
   {
      if (isInitialized())
         throw EThreadTimerError_AlreadyInitialized();

      m_notify = notify;
      m_msg = msg;

      struct sigevent sev = {};
      sev.sigev_notify = SIGEV_SIGNAL;
      sev.sigev_signo = SIGRTMIN;
      sev.sigev_value.sival_ptr = this;
      if (timer_create(CLOCK_REALTIME, &sev, &m_timer) == -1)
         throw EThreadTimerError_UnableToInitialize();
      m_initialized = True;
   }

public:
   EThreadEventTimer()
   {
      m_initialized = False;
      // assign the id
      m_id = atomic_inc(m_nextid);
      // m_thread = NULL;
      m_notify = NULL;
      m_msg = NULL;
      m_interval = 0;
      m_oneshot = True;
      m_timer = NULL;
   }
   EThreadEventTimer(Long milliseconds, Bool oneshot = False)
   {
      m_initialized = False;
      // assign the id
      m_id = atomic_inc(m_nextid);
      // m_thread = NULL;
      m_notify = NULL;
      m_msg = NULL;
      m_interval = milliseconds;
      m_oneshot = oneshot;
      m_timer = NULL;
   }      
   ~EThreadEventTimer()
   {
      destroy();
   }
   Void destroy()
   {
      if (isInitialized())
      {
         stop();
         timer_delete(m_timer);
         m_timer = NULL;
         m_initialized = False;
      }
      if (m_msg)
      {
         delete m_msg;
         m_msg = NULL;
      }
      m_notify = NULL;
   }
   Void start()
   {
      if (!isInitialized())
         throw EThreadTimerError_NotInitialized();

      struct itimerspec its;
      its.it_value.tv_sec = m_interval / 1000;              // seconds
      its.it_value.tv_nsec = (m_interval % 1000) * 1000000; // nano-seconds
      its.it_interval.tv_sec = m_oneshot ? 0 : its.it_value.tv_sec;
      its.it_interval.tv_nsec = m_oneshot ? 0 : its.it_value.tv_nsec;
      if (timer_settime(m_timer, 0, &its, NULL) == -1)
         throw EThreadTimerError_UnableToStart();
   }
   Void stop()
   {
      if (isInitialized())
      {
         struct itimerspec its;
         its.it_value.tv_sec = 0;  // seconds
         its.it_value.tv_nsec = 0; // nano-seconds
         its.it_interval.tv_sec = its.it_value.tv_sec;
         its.it_interval.tv_nsec = its.it_value.tv_nsec;
         timer_settime(m_timer, 0, &its, NULL);
      }
   }
   Long getInterval() { return m_interval; }
   Void setInterval(LongLong interval) { m_interval = interval; }
   Void setInterval(const ETime &t)
   {
      ETime dur = t >= ETime::Now() ? t - ETime::Now() : t;
      m_interval = dur.getTimeVal().tv_sec * 1000 + dur.getTimeVal().tv_usec / 1000;
   }
   Void setOneShot(Bool oneshot) { m_oneshot = oneshot; }
   Long getId() { return m_id; }
   Bool isInitialized() { return m_initialized; }

protected:
   static void _timerHandler(int signo, siginfo_t *pinfo, void *pcontext)
   {
      EThreadEventTimer *timer = (EThreadEventTimer*)pinfo->si_value.sival_ptr;
      if (timer)
         timer->m_notify->_sendTimerExpiration(*timer->m_msg);
   }

private:
   static Long m_nextid;

   Bool m_initialized;
   Long m_id;
   _EThreadEventNotification *m_notify;
   _EThreadEventMessageBase *m_msg;
   Bool m_oneshot;
   LongLong m_interval;
   timer_t m_timer;
};

class EThreadEventTimerHandler : EStatic
{
public:
   EThreadEventTimerHandler() {}
   ~EThreadEventTimerHandler() {}

   virtual Int getInitType() { return STATIC_INIT_TYPE_THREADS; }
   Void init(EGetOpt &options)
   {
      struct sigaction sa;
      sa.sa_flags = SA_SIGINFO;
      sa.sa_sigaction = EThreadEventTimer::_timerHandler;
      sigemptyset(&sa.sa_mask);
      int signo = SIGRTMIN;
      if (sigaction(signo, &sa, NULL) == -1)
         throw EThreadTimerError_UnableToRegisterTimerHandler();
   }
   Void uninit()
   {
   }
};


#define DECLARE_MESSAGE_MAP()                                  \
protected:                                                     \
   static const msgmap_t *GetThisMessageMap();                 \
   virtual const msgmap_t *GetMessageMap() const;



#define BEGIN_MESSAGE_MAP(theClass, baseClass)                 \
   const theClass::msgmap_t *theClass::GetMessageMap() const   \
   {                                                           \
      return GetThisMessageMap();                              \
   }                                                           \
   const theClass::msgmap_t *theClass::GetThisMessageMap()     \
   {                                                           \
      typedef baseClass TheBaseClass;                          \
      _Pragma("GCC diagnostic push")                           \
      _Pragma("GCC diagnostic ignored \"-Wpmf-conversions\"")  \
      static const msgentry_t _msgEntries[] =                  \
      {

#define ON_MESSAGE(id, memberFxn)                              \
         {id, (msgfxn_t)&memberFxn},

#define END_MESSAGE_MAP()                                      \
         {0, (msgfxn_t)NULL}                                   \
      };                                                       \
      _Pragma("GCC diagnostic pop")                            \
      static const msgmap_t msgMap =                           \
         {&TheBaseClass::GetThisMessageMap, &_msgEntries[0]};  \
      return &msgMap;                                          \
   }

template <class TQueue, class TMessage>
class EThreadEvent : public EThreadBasic, public _EThreadEventNotification
{
public:
   typedef Void (EThreadEvent::*msgfxn_t)(TMessage &);
   typedef struct
   {
      UInt     nMessage; // message
      msgfxn_t pFn; // routine to call (or special value)
   } msgentry_t;

   struct msgmap_t
   {
      const msgmap_t *(*pfnGetBaseMap)();
      const msgentry_t *lpEntries;
   };

   EThreadEvent()
      : EThreadBasic(),
        m_tid(-1),
        m_arg(NULL),
        m_stacksize(0),
        m_suspendCnt(0),
        m_suspendSem(0)
   {
   }
   ~EThreadEvent()
   {
   }

   Bool sendMessage(UInt message, Bool wait = True)
   {
      TMessage msg(message);
      Bool result = m_queue.push(msg, wait);
      if (result)
         onMessageQueued(msg);
      return result;
   }
   Bool sendMessage(UInt message, pVoid voidptr, Bool wait = True)
   {
      TMessage msg(message);
      msg.setVoidPtr(voidptr);
      Bool result = m_queue.push(msg, wait);
      if (result)
         onMessageQueued(msg);
      return result;
   }
   Bool sendMessage(const TMessage &msg, Bool wait = True)
   {
      Bool result = m_queue.push(msg, wait);
      if (result)
         onMessageQueued(msg);
      return result;
   }

   virtual Void init(Short appId, UShort threadId, pVoid arg, Int queueSize = 16384, Bool suspended = False, Dword stackSize = 0)
   {
      m_appId = appId;
      m_threadId = threadId;
      m_queueSize = queueSize;
      m_stacksize = stackSize;

      long id = m_appId * 100000 + 10000 + m_threadId;

      m_queue.init(m_queueSize, id, True, EThreadQueueMode::ReadWrite);

      if (!suspended)
         start();
   }

   Void quit()
   {
      sendMessage(EM_QUIT);
   }
   Void start()
   {
      if (!isInitialized())
         EThreadBasic::init(m_arg, m_stacksize);
   }
   Void suspend()
   {
      if (atomic_inc(m_suspendCnt) == 1)
         sendMessage(EM_SUSPEND);
   }
   Void resume()
   {
      if (atomic_dec(m_suspendCnt) == 0)
         m_suspendSem.Increment();
   }
   virtual Void onInit()
   {
   }
   virtual Void onQuit()
   {
   }
   virtual Void onSuspend()
   {
   }
   virtual Void onTimer(EThreadEventTimer *ptimer)
   {
   }
   Void initTimer(EThreadEventTimer &t)
   {
      TMessage *msg = new TMessage(EM_TIMER);
      msg->setVoidPtr(&t);
      t.init(this, msg);
   }
   ESemaphoreData &getMsgSemaphore()
   {
      return m_queue.semMsgs();
   }

protected:
   virtual const msgmap_t *GetMessageMap() const
   {
      return GetThisMessageMap();
   }
   static const msgmap_t *GetThisMessageMap()
   {
      return NULL;
   }
   int *getBumpPipe()
   {
      return m_queue.getBumpPipe();
   }
   Bool _sendTimerExpiration(const _EThreadEventMessageBase &msg, Bool wait = True)
   {
      Bool result = m_queue.push(static_cast<const TMessage &>(msg), wait);
      if (result)
         onMessageQueued(static_cast<const TMessage &>(msg));
      return result;
   }
   Bool _sendThreadMessage(const _EThreadEventMessageBase &msg, Bool wait = True)
   {
      Bool result = m_queue.push(static_cast<const TMessage &>(msg), wait);
      if (result)
         onMessageQueued(static_cast<const TMessage &>(msg));
      return result;
   }

   virtual Void onMessageQueued(const TMessage &msg)
   {
   }
   Bool pumpMessage(TMessage &msg, Bool wait = true)
   {
      Bool bMsg = m_queue.pop(msg, wait);
      if (bMsg)
         dispatch(msg);

      return bMsg;
   }
   virtual Void pumpMessages()
   {
      TMessage msg;

      onInit();

      try
      {
         while (True)
         {
            if (pumpMessage(msg))
            {
               if (msg.getMessageId() == EM_QUIT)
                  break;
               if (msg.getMessageId() == EM_SUSPEND)
                  m_suspendSem.Decrement();
            }
         }
      }
      catch (EError &e)
      {
         throw;
      }
      catch (...)
      {
         throw;
      }
   }
   virtual Void defaultMessageHandler(TMessage &msg)
   {
   }
   pid_t getThreadId()
   {
      if (m_tid == -1)
         m_tid = syscall(SYS_gettid);
      return m_tid;
   }

private:
   Dword threadProc(pVoid arg)
   {
      pumpMessages();
      return 0;
   }

   Bool dispatch(TMessage &msg)
   {
      Bool keepgoing = True;
      const msgmap_t *pMap;
      const msgentry_t *pEntries;

      if (msg.getMessageId() >= EM_USER)
      {
         // interate through each map
         for (pMap = GetMessageMap(); keepgoing && pMap && pMap->pfnGetBaseMap != NULL; pMap = (*pMap->pfnGetBaseMap)())
         {
            // interate through each entry for the map
            for (pEntries = pMap->lpEntries; pEntries->nMessage; pEntries++)
            {
               if (pEntries->nMessage == msg.getMessageId())
               {
                  (this->*pEntries->pFn)(msg);
                  keepgoing = False;
                  break;
               }
            }
         }

         if (pMap == NULL)
            defaultMessageHandler(msg);
      }
      else
      {
         switch (msg.getMessageId())
         {
            case EM_INIT:     { onInit(); break; }
            case EM_QUIT:     { onQuit(); break; }
            case EM_SUSPEND:  { onSuspend(); break; }
            case EM_TIMER:    { onTimer( (EThreadEventTimer*)msg.getVoidPtr() ); break; }
            default:          { break; }
         }
      }

      return keepgoing;
   }

   pid_t m_tid;
   pVoid m_arg;
   size_t m_stacksize;
   Int m_suspendCnt;
   ESemaphorePrivate m_suspendSem;

   Short m_appId;
   UShort m_threadId;
   Int m_queueSize;
   TQueue m_queue;
};

typedef EThreadEvent<EThreadQueuePublic<EThreadMessage>,EThreadMessage> EThreadPublic;
typedef EThreadEvent<EThreadQueuePrivate<EThreadMessage>,EThreadMessage> EThreadPrivate;


class EThreadEventWorkerBase
{
public:
   // typedef Void (*msgfxn_t)(TMessage &);
   typedef Void (EThreadEventWorkerBase::*msgfxnvoid_t)();
   typedef struct
   {
      UInt     nMessage; // message
      msgfxnvoid_t pFn; // routine to call (or special value)
   } msgentry_t;

   struct msgmap_t
   {
      // const _msgmap_t *(*pfnGetBaseMap)();
      const msgmap_t *(*pfnGetBaseMap)();
      const msgentry_t *lpEntries;
   };
};

#define BEGIN_MESSAGE_MAP2(theClass, baseClass)                            \
   virtual const EThreadEventWorkerBase::msgmap_t *GetMessageMap() const   \
   {                                                                       \
      return GetThisMessageMap();                                          \
   }                                                                       \
   static const EThreadEventWorkerBase::msgmap_t *GetThisMessageMap()      \
   {                                                                       \
      typedef baseClass TheBaseClass;                                      \
      _Pragma("GCC diagnostic push")                                       \
      _Pragma("GCC diagnostic ignored \"-Wpmf-conversions\"")              \
      static const EThreadEventWorkerBase::msgentry_t _msgEntries[] =      \
      {

#define ON_MESSAGE2(id, memberFxn)                                         \
         {id, (EThreadEventWorkerBase::msgfxnvoid_t)&memberFxn},

#define END_MESSAGE_MAP2()                                                 \
         {0, (EThreadEventWorkerBase::msgfxnvoid_t)NULL}                   \
      };                                                                   \
      _Pragma("GCC diagnostic pop")                                        \
      static const EThreadEventWorkerBase::msgmap_t msgMap =               \
         {&TheBaseClass::GetThisMessageMap, &_msgEntries[0]};              \
      return &msgMap;                                                      \
   }

template <class TQueue, class TMessage>
class EThreadEventWorker : public EThreadBasic, public EThreadEventWorkerBase
{
   template <class T1, class T2, class T3> friend class EThreadEventWorkGroup;
public:
   Int workerId() const
   {
      return m_workerid;
   }

   virtual Void onInit()
   {
   }
   virtual Void onQuit()
   {
   }
   virtual Void onTimer(EThreadEventTimer *ptimer)
   {
   }
   ESemaphoreData &getMsgSemaphore()
   {
      return m_queue->semMsgs();
   }

   Bool sendMessage(UInt message, Bool wait = True)
   {
      TMessage msg(message);
      Bool result = m_queue->push(msg, wait);
      if (result)
         onMessageQueued(msg);
      return result;
   }
   Bool sendMessage(UInt message, pVoid voidptr, Bool wait = True)
   {
      TMessage msg(message);
      msg.setVoidPtr(voidptr);
      Bool result = m_queue->push(msg, wait);
      if (result)
         onMessageQueued(msg);
      return result;
   }
   Bool sendMessage(const TMessage &msg, Bool wait = True)
   {
      Bool result = m_queue->push(msg, wait);
      if (result)
         onMessageQueued(msg);
      return result;
   }

protected:
   EThreadEventWorker()
      : EThreadBasic(),
        m_workerid(0),
        m_queue(nullptr),
        m_arg(nullptr),
        m_stacksize(0),
        m_tid(-1)
   {
   }
   ~EThreadEventWorker()
   {
   }

   virtual Void onMessageQueued(const TMessage &msg)
   {
   }

   virtual Void init(TQueue &queue, Int workerid, pVoid arg, Dword stackSize = 0)
   {
      m_queue = &queue;
      m_workerid = workerid;
      m_stacksize = stackSize;
      m_arg = arg;
      m_queue->attach(EThreadQueueMode::ReadWrite);
      start();
   }

   Void start()
   {
      if (!isInitialized())
         EThreadBasic::init(m_arg, m_stacksize);
   }

   virtual const msgmap_t *GetMessageMap() const
   {
      return GetThisMessageMap();
   }
   static const msgmap_t *GetThisMessageMap()
   {
      return NULL;
   }

   Bool pumpMessage(TMessage &msg, Bool wait = true)
   {
      Bool bMsg = m_queue->pop(msg, wait);
      if (bMsg)
         dispatch(msg);

      return bMsg;
   }
   virtual Void pumpMessages()
   {
      TMessage msg;

      onInit();

      try
      {
         while (True)
         {
            if (pumpMessage(msg))
            {
               if (msg.getMessageId() == EM_QUIT)
                  break;
            }
         }
      }
      catch (EError &e)
      {
         throw;
      }
      catch (...)
      {
         throw;
      }
   }
   virtual Void defaultMessageHandler(TMessage &msg)
   {
   }
   pid_t getThreadId()
   {
      if (m_tid == -1)
         m_tid = syscall(SYS_gettid);
      return m_tid;
   }

private:
   Dword threadProc(pVoid arg)
   {
      pumpMessages();
      return 0;
   }

   Bool dispatch(TMessage &msg)
   {
      Bool keepgoing = True;
      const msgmap_t *pMap;
      const msgentry_t *pEntries;

      if (msg.getMessageId() >= EM_USER)
      {
         // interate through each map
         for (pMap = (msgmap_t*)GetMessageMap(); keepgoing && pMap && pMap->pfnGetBaseMap != NULL; pMap = (msgmap_t*)(*pMap->pfnGetBaseMap)())
         {
            // interate through each entry for the map
            for (pEntries = pMap->lpEntries; pEntries->nMessage; pEntries++)
            {
               if (pEntries->nMessage == msg.getMessageId())
               {
                  (this->*((void(EThreadEventWorkerBase::*)(TMessage&))pEntries->pFn))(msg);
                  keepgoing = False;
                  break;
               }
            }
         }

         if (pMap == NULL)
            defaultMessageHandler(msg);
      }
      else
      {
         switch (msg.getMessageId())
         {
            case EM_QUIT:     { onQuit(); break; }
            case EM_TIMER:    { onTimer( (EThreadEventTimer*)msg.getVoidPtr() ); break; }
            default:          { break; }
         }
      }

      return keepgoing;
   }

   Int m_workerid;
   TQueue *m_queue;
   pVoid m_arg;
   size_t m_stacksize;
   pid_t m_tid;
};


template <class TQueue, class TMessage, class TWorker>
class EThreadEventWorkGroup : public _EThreadEventNotification
{
public:
   EThreadEventWorkGroup()
      : m_initialized(False),
        m_arg(NULL),
        m_stacksize(0),
        m_appId(0),
        m_workGroupId(0),
        m_queueSize(0),
        m_minWorkers(0),
        m_maxWorkers(0),
        m_actvWorkers(0)
   {
   }
   ~EThreadEventWorkGroup()
   {
      for (int i=0; i<m_actvWorkers; i++)
      {
         delete m_workers[i];
         m_workers[i] = nullptr;
      }
   }

   Bool isInitialized() { return m_initialized; }

   Bool sendMessage(UInt message, Bool wait = True)
   {
      TMessage msg(message);
      Bool result = m_queue.push(msg, wait);
      if (result)
         onMessageQueued(msg);
      return result;
   }
   Bool sendMessage(UInt message, pVoid voidptr, Bool wait = True)
   {
      TMessage msg(message);
      msg.setVoidPtr(voidptr);
      Bool result = m_queue.push(msg, wait);
      if (result)
         onMessageQueued(msg);
      return result;
   }
   Bool sendMessage(const TMessage &msg, Bool wait = True)
   {
      Bool result = m_queue.push(msg, wait);
      if (result)
         onMessageQueued(msg);
      return result;
   }

   virtual Void init(Short appId, UShort workGroupId, Int minWorkers, Int maxWorkers = -1,
      Int queueSize = 16384, pVoid arg = nullptr, Bool suspended = False, Dword stackSize = 0)
   {
      m_appId = appId;
      m_workGroupId = workGroupId;
      m_arg = arg;
      m_queueSize = queueSize;
      m_stacksize = stackSize;
      m_minWorkers = minWorkers;
      m_maxWorkers = maxWorkers < minWorkers ? minWorkers : maxWorkers;

      long id = m_appId * 100000 + 20000 + m_workGroupId;

      m_queue.init(m_queueSize, id, True, EThreadQueueMode::ReadWrite, True);

      if (!suspended)
         start();
   }

   Void join()
   {
      for (auto worker : m_workers)
         worker->join();
   }
   Void quit()
   {
      for (int i=0; i<m_actvWorkers; i++)
         sendMessage(EM_QUIT);
   }
   Void start()
   {
      if (!isInitialized())
      {
         m_workers.resize(m_maxWorkers);
         for (Int i=0; i<m_minWorkers; i++)
         {
            m_workers[i] = nullptr;
            if (i < m_minWorkers)
               addWorker();
         }
      }
   }

   Bool addWorker()
   {
      EMutexLock l(m_mutex);
      if (m_actvWorkers >= m_maxWorkers)
         return False;
      addWorker(m_actvWorkers++);
      return True;
   }

   Void initTimer(EThreadEventTimer &t)
   {
      TMessage *msg = new TMessage(EM_TIMER);
      msg->setVoidPtr(&t);
      t.init(this, msg);
   }
   ESemaphoreData &getMsgSemaphore()
   {
      return m_queue.semMsgs();
   }

protected:
   virtual Void onMessageQueued(const TMessage &msg)
   {
   }
   virtual Void onCreateWorker(TWorker &worker)
   {
   }

private:
   Bool _sendTimerExpiration(const _EThreadEventMessageBase &msg, Bool wait = True)
   {
      Bool result = m_queue.push(static_cast<const TMessage &>(msg), wait);
      if (result)
         onMessageQueued(static_cast<const TMessage &>(msg));
      return result;
   }
   Bool _sendThreadMessage(const _EThreadEventMessageBase &msg, Bool wait = True)
   {
      Bool result = m_queue.push(static_cast<const TMessage &>(msg), wait);
      if (result)
         onMessageQueued(static_cast<const TMessage &>(msg));
      return result;
   }

   Void addWorker(Int idx)
   {
      TWorker *worker = new TWorker();
      // worker->setGroup(*this);
      m_workers[idx] = worker;
      worker->init(m_queue, idx + 1, m_arg, m_stacksize);
      onCreateWorker(*worker);
   }

   EMutexPrivate m_mutex;
   Bool m_initialized;
   pVoid m_arg;
   size_t m_stacksize;

   Short m_appId;
   UShort m_workGroupId;
   Int m_queueSize;
   TQueue m_queue;

   Int m_minWorkers;
   Int m_maxWorkers;
   Int m_actvWorkers;
   std::vector<TWorker*> m_workers;
};

using EThreadWorkerPublic = EThreadEventWorker<EThreadQueuePublic<EThreadMessage>,EThreadMessage>;
template <class TWorker> using EThreadWorkGroupPublic = EThreadEventWorkGroup<EThreadQueuePublic<EThreadMessage>,EThreadMessage,TWorker>;

using EThreadWorkerPrivate = EThreadEventWorker<EThreadQueuePrivate<EThreadMessage>,EThreadMessage>;
template <class TWorker> using EThreadWorkGroupPrivate = EThreadEventWorkGroup<EThreadQueuePrivate<EThreadMessage>,EThreadMessage,TWorker>;


#endif // #ifndef __ETEVENT_H