ACE/protocols/ace/TMCast/MTQueue.hpp

   1 // author    : Boris Kolpackov <boris@dre.vanderbilt.edu>
   2
   3 #ifndef TMCAST_MT_QUEUE_HPP
   4 #define TMCAST_MT_QUEUE_HPP
   5
   6 #include "ace/Unbounded_Set.h"
   7 #include "ace/Unbounded_Queue.h"
   8 #include "ace/os_include/sys/os_types.h"
   9 #include "ace/Condition_T.h"
  10 #include <memory>
  11
  12 namespace ACE_TMCast
  13 {
  14   template <typename T,
  15             typename M,
  16             typename C,
  17             typename Q = ACE_Unbounded_Queue<T> >
  18   class MTQueue
  19   {
  20   public:
  21     typedef T ElementType;
  22     typedef M MutexType;
  23     typedef C ConditionalType;
  24     typedef Q QueueType;
  25
  26   public:
  27
  28     MTQueue ()
  29         : mutexp_ (new MutexType),
  30           mutex_ (*mutexp_),
  31           queue_ (),
  32           signal_ (false)
  33     {
  34     }
  35
  36     MTQueue (MutexType& mutex)
  37         : mutexp_ (),
  38           mutex_ (mutex),
  39           queue_ (),
  40           signal_ (false)
  41     {
  42     }
  43
  44   public:
  45     bool
  46     empty () const
  47     {
  48       return queue_.is_empty ();
  49     }
  50
  51     size_t
  52     size () const
  53     {
  54       return queue_.size ();
  55     }
  56
  57     // typedef typename QueueType::Empty Empty;
  58
  59     class Empty {};
  60
  61     T&
  62     front ()
  63     {
  64       ACE_Unbounded_Queue_Iterator<T> f (queue_);
  65       T* tmp;
  66       if (!f.next (tmp)) throw Empty ();
  67
  68       return *tmp;
  69     }
  70
  71
  72     T const&
  73     front () const
  74     {
  75       ACE_Unbounded_Queue_Const_Iterator<T> f (queue_);
  76       T* tmp;
  77       if (!f.next (tmp)) throw Empty ();
  78
  79       return *tmp;
  80     }
  81
  82     /*
  83       T&
  84       back ()
  85       {
  86       return queue_.back ();
  87       }
  88
  89
  90       T const&
  91       back () const
  92       {
  93       return queue_.back ();
  94       }
  95     */
  96
  97     void
  98     push (T const& t)
  99     {
 100       signal_ = empty ();
 101       queue_.enqueue_tail (t);
 102     }
 103
 104     void
 105     pop ()
 106     {
 107       T junk;
 108       queue_.dequeue_head (junk);
 109     }
 110
 111   public:
 112     void
 113     lock () const
 114     {
 115       mutex_.acquire ();
 116     }
 117
 118     void
 119     unlock () const
 120     {
 121       if (signal_)
 122       {
 123         signal_ = false;
 124
 125         for (ConditionalSetConstIterator_ i (cond_set_);
 126              !i.done ();
 127              i.advance ())
 128         {
 129           ConditionalType** c = 0;
 130
 131           i.next (c);
 132
 133           (*c)->signal ();
 134         }
 135       }
 136
 137       mutex_.release ();
 138     }
 139
 140     void
 141     subscribe (ConditionalType& c)
 142     {
 143       //@@ should check for duplicates
 144       //
 145       cond_set_.insert (&c);
 146     }
 147
 148     void
 149     unsubscribe (ConditionalType& c)
 150     {
 151       //@@ should check for absence
 152       //
 153       cond_set_.remove (&c);
 154     }
 155
 156   private:
 157     std::unique_ptr<MutexType> mutexp_;
 158     MutexType& mutex_;
 159     QueueType queue_;
 160
 161     typedef
 162     ACE_Unbounded_Set<ConditionalType*>
 163     ConditionalSet_;
 164
 165     typedef
 166     ACE_Unbounded_Set_Const_Iterator<ConditionalType*>
 167     ConditionalSetConstIterator_;
 168
 169     ConditionalSet_ cond_set_;
 170
 171     mutable bool signal_;
 172   };
 173 }
 174
 175 #endif // TMCAST_MT_QUEUE_HPP