ACE/protocols/ace/RMCast/Acknowledge.h

   1 // author    : Boris Kolpackov <boris@kolpackov.net>
   2
   3 #ifndef ACE_RMCAST_ACKNOWLEDGE_H
   4 #define ACE_RMCAST_ACKNOWLEDGE_H
   5
   6 #include "ace/Hash_Map_Manager.h"
   7 #include "ace/Thread_Manager.h"
   8
   9 #include "Stack.h"
  10 #include "Protocol.h"
  11 #include "Bits.h"
  12 #include "Parameters.h"
  13
  14 #if !defined (ACE_RMCAST_DEFAULT_MAP_SIZE)
  15 #define ACE_RMCAST_DEFAULT_MAP_SIZE 10
  16 #endif /* ACE_RMCAST_DEFAULT_MAP_SIZE */
  17
  18 #if !defined (ACE_RMCAST_DEFAULT_QUEUE_SIZE)
  19 #define ACE_RMCAST_DEFAULT_QUEUE_SIZE 10
  20 #endif /* ACE_RMCAST_DEFAULT_QUEUE_SIZE */
  21
  22 namespace ACE_RMCast
  23 {
  24   class Acknowledge : public Element
  25   {
  26   public:
  27     Acknowledge (Parameters const& params);
  28
  29     virtual void
  30     in_start (In_Element* in);
  31
  32     virtual void
  33     out_start (Out_Element* out);
  34
  35     virtual void
  36     out_stop ();
  37
  38   public:
  39     virtual void
  40     recv (Message_ptr m);
  41
  42     virtual void
  43     send (Message_ptr m);
  44
  45   private:
  46     struct Descr
  47     {
  48       //@@ There should be no default c-tor.
  49       //
  50       Descr ()
  51           : nak_count_ (0), timer_ (1)
  52       {
  53       }
  54
  55       Descr (unsigned long timer)
  56           : nak_count_ (0), timer_ (timer)
  57       {
  58       }
  59
  60       Descr (Message_ptr m)
  61           : m_ (m)
  62       {
  63       }
  64
  65     public:
  66       bool
  67       lost () const
  68       {
  69         return m_.get () == 0;
  70       }
  71
  72     public:
  73       Message_ptr
  74       msg ()
  75       {
  76         return m_;
  77       }
  78
  79       void
  80       msg (Message_ptr m)
  81       {
  82         m_ = m;
  83       }
  84
  85     public:
  86       unsigned long
  87       nak_count () const
  88       {
  89         return nak_count_;
  90       }
  91
  92       void
  93       nak_count (unsigned long v)
  94       {
  95         nak_count_ = v;
  96       }
  97
  98       unsigned long
  99       timer () const
 100       {
 101         return timer_;
 102       }
 103
 104       void
 105       timer (unsigned long v)
 106       {
 107         timer_ = v;
 108       }
 109
 110     private:
 111       Message_ptr m_;
 112
 113       unsigned long nak_count_;
 114       unsigned long timer_;
 115     };
 116
 117   private:
 118     struct Queue : ACE_Hash_Map_Manager<u64, Descr, ACE_Null_Mutex>
 119     {
 120       typedef ACE_Hash_Map_Manager<u64, Descr, ACE_Null_Mutex> Base;
 121
 122       // Should never be here but required by ACE_Hash_Blah_Blah.
 123       //
 124       Queue ()
 125           : Base (ACE_RMCAST_DEFAULT_MAP_SIZE), sn_ (0), max_sn_ (0)
 126       {
 127       }
 128
 129       Queue (u64 sn)
 130         : Base (ACE_RMCAST_DEFAULT_MAP_SIZE), sn_ (sn), max_sn_ (sn)
 131       {
 132       }
 133
 134       Queue (Queue const& q)
 135         : Base (ACE_RMCAST_DEFAULT_MAP_SIZE), sn_ (q.sn_), max_sn_ (sn_)
 136       {
 137         for (Queue::const_iterator i (q), e (q, 1); i != e; ++i)
 138         {
 139           bind ((*i).ext_id_, (*i).int_id_);
 140         }
 141       }
 142
 143     public:
 144       int
 145       bind (u64 sn, Descr const& d)
 146       {
 147         int r (Base::bind (sn, d));
 148
 149         if (r == 0 && sn > max_sn_) max_sn_ = sn;
 150
 151         return r;
 152       }
 153
 154       int
 155       rebind (u64 sn, Descr const& d)
 156       {
 157         int r (Base::rebind (sn, d));
 158
 159         if (r == 0 && sn > max_sn_) max_sn_ = sn;
 160
 161         return r;
 162       }
 163
 164       int
 165       unbind (u64 sn)
 166       {
 167         int r (Base::unbind (sn));
 168
 169         if (r == 0 && sn == max_sn_)
 170         {
 171           for (--max_sn_; max_sn_ >= sn_; --max_sn_)
 172           {
 173             if (find (max_sn_) == 0) break;
 174           }
 175         }
 176
 177         return r;
 178       }
 179
 180     public:
 181       u64
 182       sn () const
 183       {
 184         return sn_;
 185       }
 186
 187       void
 188       sn (u64 sn)
 189       {
 190         sn_ = sn;
 191       }
 192
 193       u64
 194       max_sn () const
 195       {
 196         if (current_size () == 0) return sn_;
 197
 198         return max_sn_;
 199       }
 200
 201     private:
 202       u64 sn_, max_sn_;
 203     };
 204
 205     typedef
 206     ACE_Hash_Map_Manager_Ex<Address,
 207                             Queue,
 208                             AddressHasher,
 209                             ACE_Equal_To<Address>,
 210                             ACE_Null_Mutex>
 211     Map;
 212
 213   private:
 214     void
 215     collapse (Queue& q);
 216
 217     void
 218     track ();
 219
 220     void
 221     track_queue (Address const& addr, Queue& q, Messages& msgs);
 222
 223     Profile_ptr
 224     create_nrtm (u32 max_elem);
 225
 226     static ACE_THR_FUNC_RETURN
 227     track_thunk (void* obj);
 228
 229   private:
 230     Parameters const& params_;
 231
 232     Map hold_;
 233     Mutex mutex_;
 234     Condition cond_;
 235
 236     unsigned long nrtm_timer_;
 237
 238     bool stop_;
 239     ACE_Thread_Manager tracker_mgr_;
 240   };
 241 }
 242
 243 #endif  // ACE_RMCAST_ACKNOWLEDGE_H