framework.h

/* BEGIN_COPYRIGHT                                                        */
/*                                                                        */
/* Open Diameter: Open-source software for the Diameter and               */
/*                Diameter related protocols                              */
/*                                                                        */
/* Copyright (C) 2002-2004 Open Diameter Project                          */
/*                                                                        */
/* This library is free software; you can redistribute it and/or modify   */
/* it under the terms of the GNU Lesser General Public License as         */
/* published by the Free Software Foundation; either version 2.1 of the   */
/* License, or (at your option) any later version.                        */
/*                                                                        */
/* This library is distributed in the hope that it will be useful,        */
/* but WITHOUT ANY WARRANTY; without even the implied warranty of         */
/* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU      */
/* Lesser General Public License for more details.                        */
/*                                                                        */
/* You should have received a copy of the GNU Lesser General Public       */
/* License along with this library; if not, write to the Free Software    */
/* Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307    */
/* USA.                                                                   */
/*                                                                        */
/* In addition, when you copy and redistribute some or the entire part of */
/* the source code of this software with or without modification, you     */
/* MUST include this copyright notice in each copy.                       */
/*                                                                        */
/* If you make any changes that are appeared to be useful, please send    */
/* sources that include the changed part to                               */
/* diameter-developers@lists.sourceforge.net so that we can reflect your  */
/* changes to one unified version of this software.                       */
/*                                                                        */
/* END_COPYRIGHT                                                          */
// $Id: framework.h,v 1.55 2004/06/17 21:03:14 yohba Exp $
// framework.h: Protocol Framework API based on ACE.
// Written by Yoshihiro Ohba

#ifndef __FRAMEWORK_H__
#define __FRAMEWORK_H__

#include <list>
#include <map>
#include <string>
#include <ace/Task.h>
#include <ace/Task_T.h>
#include <ace/Thread.h>
#include <ace/Thread_Manager.h>
#include <ace/Event_Handler.h>
#include <ace/Synch.h>
#include <ace/Reactor.h>
#include <ace/Thread_Exit.h>
#include <ace/Time_Value.h>
#include <ace/Singleton.h>
#include <ace/Token.h>
#include <ace/Log_Msg.h>
#include <ace/Atomic_Op_T.h>
#include <boost/shared_ptr.hpp>
#include <boost/scoped_array.hpp>
#include <boost/shared_array.hpp>

#define FRAMEWORK_LOG ACE_Log_Msg::instance()->log

template <class T> struct Type2Type
{
  typedef T OriginalType; 
};

enum AAA_Error { 
                NoInitialState, 
                FoundDuplicateStateTableEntry,
                ActivationFailure,
                TaskNotRunning,
                NoData
        };


template <class LOCK>
class AAA_ScopeLock
{
 public:
  AAA_ScopeLock(LOCK &l) : lock(l) {
    lock.acquire();
  }
  ~AAA_ScopeLock() {
    lock.release();
  }

 private:
  LOCK &lock;
};

typedef AAA_ScopeLock<ACE_Token> AAA_TokenScopeLock;

typedef AAA_ScopeLock<ACE_Mutex> AAA_MutexScopeLock;

typedef AAA_ScopeLock<ACE_Thread_Mutex> AAA_ThreadMutexScopeLock;

class AAA_JobData {};

class AAA_JobDeleter;
class AAA_Task;



class AAA_Job
{
  friend class AAA_JobDeleter;

 public:
  AAA_Job(AAA_JobData *d=0, char* name=0) : data(d), priority(1), weight(1)
  {
    if (name) this->name = std::string(name);
    refcount = 1;
  }

  virtual int Serve()=0;

  virtual int Schedule(AAA_Job*, size_t backlogSize=1)=0;

  void Delete() { if (--refcount <=0 ) delete this; }

  virtual bool ExistBacklog() { return false; }

  virtual size_t BacklogSize() { return 0; }

  inline std::string& Name() { return name; }

  inline AAA_JobData* Data() { return data; }

  template <class T> inline T* Data(Type2Type<T>) { return (T*)data; }

  unsigned& Priority() { return priority; }

  unsigned& Weight() { return weight; }

  void Acquire() { ++refcount; }

  int Release(int n=1) 
  { 
    refcount -= n;
    if (refcount<0) {
      delete this;
      return -1;
    }
    return 0;
  }

  virtual bool Running() { return true; }

  bool operator==(AAA_Job* job) { return (this==job); }

  bool operator!=(AAA_Job* job) { return (this!=job); }

 protected:

  virtual ~AAA_Job() {}

 private:  

  std::string name;

  AAA_JobData* data;

  ACE_Atomic_Op<ACE_Thread_Mutex, int> refcount;

  unsigned priority;

  unsigned weight;
};

template <class JOB>
class AAA_JobHandle : public boost::shared_ptr<JOB>
{
 public:
  AAA_JobHandle(JOB* job) : boost::shared_ptr<JOB>(job, AAA_JobDeleter()) {}

  JOB& Job() { return (JOB&)*get(); }
};

class AAA_JobDeleter
{
 public:
  void operator()(AAA_Job *job) { job->Delete(); }
  static void Delete(AAA_Job *job) { job->Delete(); }
};

const unsigned maxNumPriority = 3;
const unsigned maxMaxWeight = 3;

template <class T, class LOCK = ACE_Thread_Mutex>
  class AAA_QueueJob : public AAA_Job
{
 public:
  AAA_QueueJob(AAA_JobData *d=0, char* name=0, 
               unsigned numPriority=1, unsigned maxWeight=1) throw (int)
 : AAA_Job(d, name), 
 numPriority(numPriority), maxWeight(maxWeight), total(0),
 cond(std::auto_ptr< ACE_Condition<LOCK> >
      (new ACE_Condition<LOCK>(lock))),
 signaled(false)

 {
   if (numPriority > maxNumPriority)
     {
       FRAMEWORK_LOG(LM_ERROR, "[%N] Too many priorities.\n");
       throw 1;
     }
   if (maxWeight > maxMaxWeight)
     {
       FRAMEWORK_LOG(LM_ERROR, "[%N] Too large weight.\n");
       throw 1;
     }
   indexQueue = boost::shared_array< std::list<unsigned> >
   (new std::list<unsigned>[numPriority]);

   dataQueue = boost::shared_array< std::list<T> >
   (new std::list<T>[numPriority*maxWeight]);
 }

  virtual ~AAA_QueueJob() {}

  virtual int Serve()=0;

  virtual int Schedule(AAA_Job*, size_t backlogSize=1)=0;

 inline bool ExistBacklog() 
 { 
   // Lock is not needed here, because this does not change the state
   // of the object.
   return (total>0);
 }

 inline size_t BacklogSize() { return total; }

 inline int Enqueue(T entry, bool blocking=false,
                    unsigned priority=1, unsigned weight=1) 
 { 
   AAA_ScopeLock<LOCK> g(lock); 

   // In the case where one or more threads can enter here
   // simultaneously (this is the case when Dequeue is called by a
   // task), conditional mutex is needed to avoid possible blocking
   // on Enqueue() when the queue is full.

   while (!signaled && total==MaxSize()) 
   { 
     if (!blocking)
       {
         FRAMEWORK_LOG(LM_DEBUG, "[%N] Enqueue is blocked.\n");
         return 0;
       }
     cond->wait(); 
   }

   if (signaled && blocking)
      return -1;

   unsigned pIndex = Index(priority);
   unsigned dIndex = Index(priority, weight);


   // The enqueueing algorithm is as follows.  

   // 1) Compare the weight of the job with the queue length of
   // the data queue where the job is enqueued.

   // 2) If the queue length of the data queue is less than the
   // weight, then enqueue the index of the data queue to the index
   // queue that is corresponding to the the priority of the job.

   // 3) Enqueue the job to the data queue.

   // Step (2) means that the larger the weight, the more the job
   // gets the chance to be served.  

   if (dataQueue[dIndex].size() < weight)
     indexQueue[pIndex].push_back(dIndex);

   dataQueue[dIndex].push_back(entry); 
   total++;

   // Wake up waiting threads.  This signal() call can be outside of
   // the scope lock.
   cond->signal();
   return total;
  }

 int Dequeue(T &entry, bool blocking=false) throw(int)
  { 
    do {
      AAA_ScopeLock<LOCK> g(lock);
      // In the case where one or more threads can enter here
      // simultaneously (this is the case when Dequeue is called by a
      // task), conditional mutex is needed to avoid possible blocking
      // on Dequeue() when the queue is empty.

      while (!signaled && total==0) 
        { 
          if (!blocking)
            return 0;
          cond->wait(); 
        }

      if (signaled && blocking)
        return -1;

      // The serving algorithm is as follows:

      // 1) Find the highest priority class that has a backlog.

      // 2) Dequeue an index from the index queue of the found
      // class. The dequeued index points the data queue to serve.  

      // 3) Dequeue an entry from the data queue corresponding to the
      // index.

      // 4) After step (3), if the queue length of the data queue is
      // not less than the weight given to that queue, then re-enqueue
      // the index to the index queue of the same priority class.

      // Step 2) of Enqueue() and Step 4) of Dequeue are complement
      // operations, which provides both "work conserving" and "O(1)
      // complexity" characteristics.
      
      do {
        bool found=false;
        for (unsigned p=0; p<numPriority; p++)
          {
            if (indexQueue[p].empty())
              continue;

            unsigned dIndex = indexQueue[p].front();
            indexQueue[p].pop_front();
            entry = dataQueue[dIndex].front();
            dataQueue[dIndex].pop_front();
            if (dataQueue[dIndex].size() >= (dIndex % maxWeight) + 1)
              indexQueue[p].push_back(dIndex);
            found=true;
            total--;
            break;
          }

        if (!found)
          {
            FRAMEWORK_LOG(LM_ERROR, "[%N] BUG (dequeue error).\n");
            return 0;
          }

      } while (0);
    } while (0);
    if (total>0) 
      cond->signal();
    return 1;
  }

 inline int Remove(T entry)
 { 
   AAA_ScopeLock<LOCK> g(lock); 
   int totalRemove=0;
   for (unsigned i=0; i<numPriority; i++)
     for (unsigned j=i*maxWeight; j<(i+1)*maxWeight; j++)
     {
       unsigned preSize = (unsigned)dataQueue[j].size();
       dataQueue[j].remove(entry);
       unsigned postSize = (unsigned)dataQueue[j].size();
       if (preSize != postSize)
         {
           totalRemove =+ preSize - postSize;
           total -= preSize - postSize;
           // Adjust the index queue so that the index for the removed entry 
           AdjustIndexQueue(i, j);
         }
     }
   if (totalRemove>0) 
     cond->signal();
   return totalRemove;
 }

  virtual void Flush() 
  { 
    AAA_ScopeLock<LOCK> g(lock); 
    for (unsigned i=0; i<numPriority; indexQueue[i++].clear());
    for (unsigned i=0; i<numPriority*maxWeight; dataQueue[i++].clear());
    total=0;
    cond->signal();
  }

 public:
 void Signal() 
 { 
   signaled=true;
   cond->broadcast(); 
 }

 inline size_t MaxSize() { return (size_t)-1; }

 protected:

  LOCK lock;

 private:

  inline unsigned Index(unsigned p)
    {
      p = (p>maxNumPriority) ? maxNumPriority : p;
      return p-1;
    }

 inline unsigned Index(unsigned p, unsigned w)
    {
      p = (p>maxNumPriority) ? maxNumPriority : p;
      w = (w>maxMaxWeight) ? maxMaxWeight : w;
      return (p-1)*maxWeight + (w-1);
    }

 void AdjustIndexQueue(unsigned pIndex, unsigned dIndex)
 {
   std::list<unsigned> &iQueue = indexQueue[pIndex];
   std::list<unsigned>::iterator i;
   unsigned j=0;
   for (i=iQueue.begin(); i!=iQueue.end(); )
     {
       if ((*i) == dIndex)
         i = (++j > dataQueue[dIndex].size()) ? iQueue.erase(i) : i++;
       else
         i++;
     }
 }

 unsigned numPriority;  // number of priority
 unsigned maxWeight;    // maximum weight value

 // This queue stores indexes for the dataQueue per priority.
 boost::shared_array< std::list<unsigned> > indexQueue;

 // This queue stores data for per priority and per weight 
 boost::shared_array< std::list<T> > dataQueue;

 unsigned total;

 std::auto_ptr< ACE_Condition<LOCK> > cond;

 bool signaled;
};

typedef int AAA_Event;



typedef AAA_QueueJob<AAA_Event> AAA_EventQueueJob;

typedef AAA_QueueJob<AAA_Job*> AAA_JobQueueJob;

#define AAA_SCHED_FIFO      0  
#define AAA_SCHED_WFQ       1
#define AAA_SCHED_PRIORITY  2

typedef ACE_UINT16 AAA_SchedulingPolicy;


class AAA_GroupedJob : public AAA_QueueJob<AAA_Job*, ACE_Thread_Mutex>
{
 public:
  static AAA_GroupedJob* 
    Create(AAA_Job &parent, AAA_JobData* d, 
           char *name=0, 
           AAA_SchedulingPolicy policy=AAA_SCHED_FIFO,
           bool blocking=false,
           unsigned numPriority=1, unsigned maxWeight=1) throw(AAA_Error)
  {
    if (d==0)
      {
        FRAMEWORK_LOG(LM_ERROR, "AAA_JobData must be non-null.\n");
        throw NoData;
      }
    return new AAA_GroupedJob(parent, d, name, policy, blocking,
                              numPriority, maxWeight);
  }

  static AAA_GroupedJob* 
    Create(AAA_JobData* d, char *name=0, 
           AAA_SchedulingPolicy policy=AAA_SCHED_FIFO,
           bool blocking=false,
           unsigned numPriority=1, unsigned maxWeight=1) throw(AAA_Error)
  {
    if (d==0)
      {
        FRAMEWORK_LOG(LM_ERROR, "AAA_JobData must be non-null.\n");
        throw NoData;
      }
    return new AAA_GroupedJob(d, name, policy, blocking,
                              numPriority, maxWeight);
  }

  void Start() throw(AAA_Error) { running=true; }

  void Stop() 
  { 
    running=false; Flush(); 
    Signal();
  }
  
  void Flush()
  {
    while (ExistBacklog())
      {
        AAA_Job* job;
        Dequeue(job);   // Non-blocking dequeue.
        job->Release();
      }
  }

  void Remove(AAA_Job *job)
  {
    // Root job relies on Acquire/Release mechinism instead of
    // Remove().
    int n = AAA_QueueJob<AAA_Job*, ACE_Thread_Mutex>::Remove(job);
    job->Release(n);
  }
        
  int Schedule(AAA_Job *job , size_t backlogSize=1)
  {
    if (!Running())
      return (-1);

    job->Acquire();

    unsigned priority, weight = 0;
    EnforceSchedulingPolicy(job, priority, weight);

    // If there is already a backlog for this session, do nothing.
    // Job weight is taken into account here.
    if (backlogSize > weight)
      return (0);

    // Since weight is taken into account by above backlog-weight
    // comparison, just specify the priority in Enqueue().
    do {
      int result = Enqueue(job, blocking);
      if (result <= 0)
        return (-1);
    } while (0);

    //    Enqueue(job, priority);

    // If this is not a root job, then ask the parent to schedule me.
    return (!IsRoot()) ? parent.Schedule(this) : 0;
  }

  int Serve()
  {
    AAA_Job* job;

    do {
      int result = Dequeue(job, blocking);
      if (result <= 0)
        return result;
    } while (0);

    if (job->Release() == -1)
      return (int)BacklogSize();

    unsigned childBacklogSize = job->Serve();

    unsigned priority, weight = 0;
    EnforceSchedulingPolicy(job, priority, weight);

    // If there is already a backlog for this session, do nothing.
    // Job weight is taken into account here.
    if (childBacklogSize < weight)
      return (int)BacklogSize();

    job->Acquire();

    // Since weight is taken into account by above backlog-weight
    // comparison, just specify the priority in Enqueue().
    return Enqueue(job, priority ? true : false);
  }

 protected:

 private:
   AAA_GroupedJob(AAA_Job &parent, 
                  AAA_JobData* d, 
                  char *name=0,
                  AAA_SchedulingPolicy policy=AAA_SCHED_FIFO,
                  bool blocking=false, 
                  unsigned numPriority=1, unsigned maxWeight=1)
     : AAA_QueueJob<AAA_Job*, ACE_Thread_Mutex>(d, name, 
                                                numPriority, maxWeight), 
    parent(parent), policy(policy), running(false), blocking(blocking)
  {
    // Increment the reference counter of the parent.
    parent.Acquire();
  }

  AAA_GroupedJob(AAA_JobData* d, char *name=0,
                 AAA_SchedulingPolicy policy=AAA_SCHED_FIFO, 
                 bool blocking=false,
                 unsigned numPriority=1, unsigned maxWeight=1)
    : AAA_QueueJob<AAA_Job*, ACE_Thread_Mutex>(d, name,
                                               numPriority, maxWeight), 
    parent(*this), policy(policy), running(false), blocking(blocking)
  {}

  ~AAA_GroupedJob() { 
    Stop(); 
    if (!IsRoot())
      parent.Release(); 
  }

  inline bool IsRoot() { return (parent == this); }
  

  void EnforceSchedulingPolicy(AAA_Job *job, 
                               unsigned &priority,
                               unsigned &weight)
  {
    const int infinity = 100000;
 
    priority = job->Priority();
    weight = job->Weight();

    // Adjust the priority and weight to the scheduling policy.
    if (policy != AAA_SCHED_WFQ)
      weight = infinity;
    
    if (policy != AAA_SCHED_PRIORITY)
      priority = 1;
  }

  AAA_Job& parent;

  unsigned policy;

  bool running;

  bool blocking;
};



class AAA_Task : public ACE_Task<ACE_MT_SYNCH>
{
public:

  AAA_Task(AAA_SchedulingPolicy policy=AAA_SCHED_FIFO, char *name=0,
           unsigned numPriority=1, unsigned maxWeight=1) 
    : handle
    (AAA_JobHandle<AAA_GroupedJob>
     (AAA_GroupedJob::Create((AAA_JobData*)this, name, policy, true,
                             numPriority, maxWeight))),
    cond(std::auto_ptr< ACE_Condition<ACE_Mutex> >
         (new ACE_Condition<ACE_Mutex>(mutex)))
    {
      reactor(0);
    }

  virtual ~AAA_Task() { Stop(); }
  
  void Start(int n) throw(AAA_Error)
  {
    if (Running())
      return;

    if (activate(THR_NEW_LWP|THR_JOINABLE, n+1) == -1)
      {
        throw ActivationFailure;
      }

    // Wait until the timer thread is up and running.
    do {
      AAA_MutexScopeLock guard(mutex);
      while (!Running()) { cond->wait(); }
    } while (0);

    Job().Start();
  }
    
    
  void Stop()
  {
    if (!Running())
      return;

    // Send delete signal to threads waiting on scheduling queue.
    Job().Stop();

    // Send signal to delete the timer thread.
    reactor()->schedule_timer(this, (const void*)0, ACE_Time_Value(0));
        
    // Wait for all activated threads to complete.
    wait(); 
  }

  inline long ScheduleTimer(ACE_Event_Handler* eventHandler, const void* arg, 
                            ACE_Time_Value delay, ACE_Time_Value interval=0)
  {
    if (Running())
      return reactor()->schedule_timer(eventHandler, arg, delay, interval);
    return (-1);
  }

  inline int CancelTimer(ACE_UINT32 timerHandle, const void** arg)
  {
    if (Running())
      return reactor()->cancel_timer(timerHandle, arg);
    return (-1);
  }

  inline bool Running() { return reactor() ? true : false; }

  inline AAA_GroupedJob& Job() { return handle.Job(); }

  inline AAA_JobHandle<AAA_GroupedJob>& JobHandle() { return handle; }

 protected:

 private:

  int svc()
  {
#if !defined(OS_FREEBSD) && !defined(WIN32)
    std::auto_ptr<ACE_Thread_Exit> threadExit(new ACE_Thread_Exit);
    ACE_TSS<ACE_Thread_Exit> threadExitTSS(threadExit.get());
#endif
    mutex.acquire();
    if (!reactor())
      {
        std::auto_ptr<ACE_Reactor> r(new ACE_Reactor);
        reactor(r.get());
        mutex.release();
        cond->signal();
        while (1) // Loop for timer thread loop.
          if (reactor()->handle_events() == -1)
            FRAMEWORK_LOG(LM_ERROR, "[%N] handle_events().\n", 
                          Job().Name().c_str());
        return 0;
      }
    mutex.release();
    while (1)  // Loop for serving threads.
      {
        if (Job().Serve()<0)
          return 0;
      }
    return 0;
  }
  
  int close(u_long flags=0) { 
    return 0; 
  }

  int handle_timeout(const ACE_Time_Value &tv, const void *arg)
  {
    reactor(0);
    ACE_Thread::exit();
    return 0;
  }

  AAA_JobHandle<AAA_GroupedJob> handle;

  std::auto_ptr< ACE_Condition<ACE_Mutex> > cond;

  ACE_Mutex mutex;
};


typedef int AAA_State;

template <class ARG> class AAA_Action 
{
 public:
  virtual void operator()(ARG&)=0;
 protected:
  virtual ~AAA_Action() {}
  AAA_Action() {}
};

template <class ARG> class AAA_NullAction_S : public AAA_Action<ARG>
{
  friend class ACE_Singleton<AAA_NullAction_S<ARG>, 
    ACE_Recursive_Thread_Mutex>;
 public:
  void operator()(ARG&) {}
 private:
  AAA_NullAction_S() {}
  ~AAA_NullAction_S() {}
};

template <class ARG> class AAA_NullAction 
{
 public:
  AAA_Action<ARG>& operator()() 
  { 
    return *ACE_Singleton<AAA_NullAction_S<ARG>, ACE_Recursive_Thread_Mutex>::
      instance();
  }
};



template <class ARG> class AAA_StateTableEntry 
{
public:
  AAA_StateTableEntry(AAA_State st1, AAA_Event ev, AAA_State st2, 
                      AAA_Action<ARG>& ac=AAA_NullAction<ARG>()()) : 
  prevState(st1), event(ev), nextState(st2), action(ac), isWildcardEvent(false)
  {}
  AAA_StateTableEntry(AAA_State st1, AAA_State st2, 
                      AAA_Action<ARG>& ac=AAA_NullAction<ARG>()()) :
  prevState(st1), nextState(st2), action(ac), isWildcardEvent(true)
  {}

  ~AAA_StateTableEntry() {}

  inline AAA_State PrevState() { return prevState; }
  inline AAA_Event Event() { return event; }
  inline AAA_State NextState() { return nextState; }
  inline AAA_Action<ARG>& Action() { return action; }
  inline bool IsWildcardEvent() { return isWildcardEvent; }

 protected:
  AAA_State   prevState;   
  AAA_Event   event;       
  AAA_State   nextState;
  AAA_Action<ARG>& action;
  bool isWildcardEvent;
};


template <class ARG> class AAA_StateTable : 
public std::list< AAA_StateTableEntry <ARG> * >
{
public:
  AAA_StateTable() : isInitialStateSet(false) {}

  virtual ~AAA_StateTable() 
  { 
    for (typename std::list< AAA_StateTableEntry <ARG> * >::iterator 
           i=begin(); i!=end(); delete *(i++));
  }

  
  AAA_State InitialState() throw (AAA_Error)
  {
    if (! isInitialStateSet)
      throw NoInitialState;
    return initialState; 
  }

  void InitialState(AAA_State st) 
  {
    initialState = st; 
    isInitialStateSet = true;
  }

  bool FindStateTableEntry(AAA_State st, AAA_Event ev, 
                           AAA_StateTableEntry<ARG>*& entry)
  {
    bool found=false;
    for (typename std::list< AAA_StateTableEntry <ARG> * >::iterator 
           i=begin(); i!=end(); i++)
      {
        AAA_StateTableEntry<ARG> *e = *i;
        if (e->PrevState() != st)
          continue;

        if (e->IsWildcardEvent())
          {
            entry = e;
            found=true;
            continue;
          }

        if (e->Event() == ev)
          {
            entry = e;
            return true;
          }
      }
    return found;
  }

  bool FindStateTableEntry(AAA_State st, AAA_StateTableEntry<ARG>*& entry)
  {
    for (typename std::list< AAA_StateTableEntry <ARG> * >::iterator 
           i=begin(); i!=end(); i++)
      {
        entry  = *i;
        if (entry->PrevState() != st)
          continue;
        if (entry->IsWildcardEvent())
          return true;
      }
    return false;
  }

protected:
  void AddStateTableEntry(AAA_State pSt, AAA_Event ev, 
                          AAA_State nSt, 
                          AAA_Action<ARG>& ac= AAA_NullAction<ARG>()())
    throw (AAA_Error)
  {
    AAA_StateTableEntry<ARG> *dummy;
    if (FindStateTableEntry(pSt, ev, dummy))
      {
        throw FoundDuplicateStateTableEntry;
      }
    push_back(new AAA_StateTableEntry<ARG>(pSt, ev, nSt, ac));
  }

  void AddWildcardStateTableEntry(AAA_State pSt, AAA_State nSt, 
                                  AAA_Action<ARG>& ac= AAA_NullAction<ARG>()())
      throw (AAA_Error)
  {
    AAA_StateTableEntry<ARG> *dummy;
    if (FindStateTableEntry(pSt, dummy))
      throw FoundDuplicateStateTableEntry;

    push_back(new AAA_StateTableEntry<ARG>(pSt,nSt,ac));
  }
private:
  AAA_State initialState;
  bool isInitialStateSet;
};


class AAA_StateMachineBase
{
public:
  virtual void Start()=0;

  virtual void Stop()=0;

  virtual void Restart() { Stop(); Start(); }

  virtual bool Running()=0;

  virtual void Event(AAA_Event ev)=0;

  AAA_StateMachineBase& operator=(AAA_StateMachineBase& sm) { return sm; }

  std::string& Name() { return name; }

protected:
  AAA_StateMachineBase(char *name=0) 
  {
    if (name)
      this->name = std::string(name);
  }

  virtual ~AAA_StateMachineBase() {}

 private:
  std::string name;
};

template <class ARG> 
class AAA_StateMachine : public AAA_StateMachineBase
{
public:
  virtual ~AAA_StateMachine() {}

  virtual void Start() throw(AAA_Error)
  { 
    state = stateTable.InitialState();
    running = true;
  }

  inline void Stop() { running=false; }

  inline bool Running() { return running; }


  void Event(AAA_Event ev)
  {
    // Throw an exception if state machine is not started.
    if (!running)
      {
        FRAMEWORK_LOG(LM_ERROR, 
                      "StateMachine[%s] state machine is not running.\n", 
                      Name().c_str());
        return;
      }

    AAA_StateTableEntry<ARG>* entry;

    // Search state table for StateTableEntry containing the event
    // defined in the current state.
    if (!stateTable.FindStateTableEntry(state, ev, entry))
      {
        // Cannot found state table that accepts event.
        FRAMEWORK_LOG
          (LM_ERROR, 
           "StateMachine[%s] cannot accept event %d at state %d.\n", 
           Name().c_str(), ev, state);
        return;
      }

    // Change the state.
    state = entry->NextState();

    // Execute the action for the current StateTableEntry.
    entry->Action()(actionArg);
  }

  AAA_StateMachine<ARG>& operator=(AAA_StateMachine<ARG>& sm) { return sm; }

protected:
  AAA_StateMachine(ARG &arg, AAA_StateTable<ARG> &table, char *name=0)
    : AAA_StateMachineBase(name),
    stateTable(table), actionArg(arg), running(true)
  {}

  AAA_StateTable<ARG>& stateTable;
  AAA_State state;
  ARG& actionArg;
private:
  bool running;
};


class AAA_TimerTypeAllocator_S
{
  friend class ACE_Singleton<AAA_TimerTypeAllocator_S, 
    ACE_Recursive_Thread_Mutex>;
 public:
  inline int Allocate() { return ++lastAllocatedType.value_i(); }
 private:
  AAA_TimerTypeAllocator_S() {}
  ~AAA_TimerTypeAllocator_S() {}
  ACE_Atomic_Op<ACE_Thread_Mutex, int> lastAllocatedType;
};

typedef ACE_Singleton<AAA_TimerTypeAllocator_S, ACE_Recursive_Thread_Mutex> 
AAA_TimerTypeAllocator;

typedef std::map<int,ACE_UINT32> TimerHandleMap;



template <class ARG>
class AAA_StateMachineWithTimer : public AAA_StateMachine<ARG>
{
public:
  void ScheduleTimer(AAA_Event ev, ACE_UINT32 sec, 
                     ACE_UINT32 usec=0, int type=0)
  {
    AAA_Event *eventP = new AAA_Event(ev);
    ACE_UINT32 timerHandle;
    // Cancel the already scheduled timer for this type.
    CancelTimer(type);
    timerHandle =  reactor.schedule_timer(
          (ACE_Event_Handler*)&timerEventHandler, 
          (const void*)eventP, 
          ACE_Time_Value(sec, usec), 
          ACE_Time_Value(sec, usec));
    timerHandleMap.insert(std::pair<int, ACE_UINT32>(type, timerHandle));
  }

  void CancelTimer(int type=0) 
  {
    AAA_Event *ev;

    ACE_UINT32 timerHandle;
    TimerHandleMap::iterator i;
    i = timerHandleMap.find(type);
    if (i == timerHandleMap.end())
      return;

    timerHandle = i->second;
    timerHandleMap.erase(i);
    if (reactor.cancel_timer(timerHandle, (const void**)&ev) == 1)
      delete ev;
  }

  void CancelAllTimer()
  {
    while (! timerHandleMap.empty()) {
      TimerHandleMap::iterator i = timerHandleMap.begin();
      CancelTimer(i->first);  
    }
  }

 protected:
  // separate class derivation for ACE_Event_Handler
  // to protect against changes to ACE_Event_Handler
  class AAA_FsmTimerEventHandler : public ACE_Event_Handler
  {
  public:
    AAA_FsmTimerEventHandler(AAA_StateMachineWithTimer<ARG> &fsm)
                             : fsm_ref(fsm) { }

  private:
    int handle_timeout(const ACE_Time_Value &tv, const void *arg)
    {
      AAA_Event event = *(AAA_Event*)arg;

      fsm_ref.Timeout(event);
      return (tv ? 0 : 0);
    }
    AAA_StateMachineWithTimer<ARG> &fsm_ref;
  };
    
  AAA_StateMachineWithTimer(ARG &arg, AAA_StateTable<ARG> &table, 
                            ACE_Reactor &r, char *name=0)
      : AAA_StateMachine<ARG>(arg, table, name), reactor(r),
        timerEventHandler(*this)        
  {}

  virtual ~AAA_StateMachineWithTimer()  { CancelAllTimer(); }

  virtual void Timeout(AAA_Event ev)=0;

private:
  ACE_Reactor& reactor;
  TimerHandleMap timerHandleMap;
  AAA_FsmTimerEventHandler timerEventHandler;
};

#endif // __FRAMEWORK_H__

---