ACE/protocols/ace/TMCast/Group.cpp

   1 // author    : Boris Kolpackov <boris@dre.vanderbilt.edu>
   2
   3 #include "Group.hpp"
   4
   5 #include <typeinfo>
   6
   7 // OS primitives
   8 #include <ace/OS_NS_stdlib.h>
   9 #include <ace/Synch.h>
  10 #include <ace/Time_Value.h>
  11 #include <ace/SOCK_Dgram_Mcast.h>
  12 #include <ace/Condition_T.h>
  13 #include <ace/OS_NS_sys_time.h>
  14
  15 #include "Messaging.hpp"
  16
  17 #include "Protocol.hpp"
  18
  19 // Components
  20
  21 #include "LinkListener.hpp"
  22 #include "FaultDetector.hpp"
  23 #include "TransactionController.hpp"
  24
  25 namespace ACE_TMCast
  26 {
  27   bool
  28   operator== (std::type_info const* pa, std::type_info const& b)
  29   {
  30     return *pa == b;
  31   }
  32
  33   //
  34   //
  35   //
  36   class Terminate : public virtual Message {};
  37
  38
  39   //
  40   //
  41   //
  42   class Failure : public virtual Message {};
  43
  44
  45   //
  46   //
  47   //
  48   class Scheduler
  49   {
  50   public:
  51     Scheduler (ACE_INET_Addr const& addr,
  52                char const* id,
  53                MessageQueue& out_send_data,
  54                MessageQueue& out_recv_data,
  55                MessageQueue& out_control)
  56
  57         : cond_ (mutex_),
  58
  59           addr_ (addr),
  60           sock_ (),
  61
  62           out_control_ (out_control),
  63
  64           in_data_ (mutex_),
  65           in_link_data_(mutex_),
  66           in_control_ (mutex_),
  67
  68           sync_schedule (ACE_OS::gettimeofday ()),
  69
  70           transaction_controller_ (in_data_, out_send_data, out_recv_data)
  71     {
  72       ACE_OS::strncpy (id_, id, Protocol::MEMBER_ID_LENGTH);
  73       id_[Protocol::MEMBER_ID_LENGTH - 1] = '\0';
  74
  75       sock_.set_option (IP_MULTICAST_TTL, 32); // @@ ttl is hardcoded
  76
  77       in_data_.subscribe (cond_);
  78       in_link_data_.subscribe (cond_);
  79       in_control_.subscribe (cond_);
  80
  81       ACE_thread_t unused;
  82       if (ACE_OS::thr_create (&thread_thunk,
  83                               this,
  84                               THR_JOINABLE,
  85                               &unused,
  86                               &thread_) != 0) ACE_OS::abort ();
  87     }
  88
  89     virtual ~Scheduler ()
  90     {
  91       {
  92         MessageQueueAutoLock lock (in_control_);
  93
  94         in_control_.push (MessagePtr (new Terminate));
  95       }
  96
  97       if (ACE_OS::thr_join (thread_, 0) != 0) ACE_OS::abort ();
  98
  99       // cerr << "Scheduler is down." << endl;
 100     }
 101
 102   public:
 103     MessageQueue&
 104     in_data ()
 105     {
 106       return in_data_;
 107     }
 108
 109   private:
 110     static ACE_THR_FUNC_RETURN
 111     thread_thunk (void* arg)
 112     {
 113       Scheduler* obj = reinterpret_cast<Scheduler*> (arg);
 114       obj->execute ();
 115       return 0;
 116     }
 117
 118     void
 119     execute ()
 120     {
 121       try
 122       {
 123         sock_.join (addr_);
 124         std::unique_ptr<LinkListener> ll (new LinkListener (sock_, in_link_data_));
 125
 126         {
 127           AutoLock lock (mutex_);
 128
 129           // Loop
 130           //
 131           //
 132
 133           while (true)
 134           {
 135             cond_.wait (&sync_schedule);
 136
 137             // "Loop of Fairness"
 138
 139             bool done = false;
 140
 141             do
 142             {
 143               // control message
 144               //
 145               //
 146               if (!in_control_.empty ())
 147               {
 148                 done = true;
 149                 break;
 150               }
 151
 152               // outsync
 153               //
 154               //
 155               if (sync_schedule < ACE_OS::gettimeofday ())
 156               {
 157                 // OUTSYNC
 158
 159                 outsync ();
 160
 161                 // schedule next outsync
 162                 sync_schedule =
 163                   ACE_OS::gettimeofday () +
 164                   ACE_Time_Value (0, Protocol::SYNC_PERIOD);
 165               }
 166
 167               // link message
 168               //
 169               //
 170               if (!in_link_data_.empty ())
 171               {
 172                 MessagePtr m (in_link_data_.front ());
 173                 in_link_data_.pop ();
 174
 175                 LinkData* data;
 176
 177                 if (dynamic_cast<LinkFailure*>(m.get()))
 178                 {
 179                   // cerr << "link failure" << endl;
 180                   throw false;
 181                 }
 182                 else if ((data = dynamic_cast<LinkData*> (m.get ())))
 183                 {
 184                   // INSYNC, TL, CT
 185
 186                   // Filter out loopback.
 187                   //
 188                   if (ACE_OS::strcmp (data->header().member_id.id, id_) != 0)
 189                   {
 190                     insync ();
 191                     transaction_list ();
 192                     current_transaction (data->header().current,
 193                                          data->payload (),
 194                                          data->size ());
 195                   }
 196                 }
 197                 else
 198                 {
 199                   // cerr << "unknown message type from link listener: "
 200                   //      << typeid (*m).name () << endl;
 201                   ACE_OS::abort ();
 202                 }
 203               }
 204
 205               // api message
 206               //
 207               //
 208               if (!in_data_.empty ())
 209               {
 210                 // API
 211
 212                 api ();
 213               }
 214
 215             } while (!in_link_data_.empty() ||
 216                      sync_schedule < ACE_OS::gettimeofday ());
 217
 218             if (done) break;
 219           }
 220         }
 221       }
 222       catch (...)
 223       {
 224         // cerr << "Exception in scheduler loop." << endl;
 225         MessageQueueAutoLock lock (out_control_);
 226         out_control_.push (MessagePtr (new Failure));
 227       }
 228     }
 229
 230     // Events
 231     //
 232     // Order:
 233     //
 234     // INSYNC, TSL, VOTE, BEGIN
 235     // API
 236     // OUTSYNC
 237     //
 238
 239     void
 240     insync ()
 241     {
 242       fault_detector_.insync ();
 243     }
 244
 245     void
 246     outsync ()
 247     {
 248       char buf[Protocol::MAX_MESSAGE_SIZE];
 249
 250       Protocol::MessageHeader* hdr =
 251         reinterpret_cast<Protocol::MessageHeader*> (buf);
 252
 253       void* data = buf + sizeof (Protocol::MessageHeader);
 254
 255       hdr->length = sizeof (Protocol::MessageHeader);
 256       hdr->check_sum = 0;
 257
 258       ACE_OS::strcpy (hdr->member_id.id, id_);
 259
 260       size_t size (0);
 261
 262       transaction_controller_.outsync (hdr->current, data, size);
 263
 264       hdr->length += size;
 265
 266       fault_detector_.outsync ();
 267
 268       // sock_.send (buf, hdr->length, addr_);
 269       sock_.send (buf, hdr->length);
 270     }
 271
 272     void
 273     transaction_list ()
 274     {
 275     }
 276
 277     void
 278     current_transaction (Protocol::Transaction const& t,
 279                          void const* payload,
 280                          size_t size)
 281     {
 282       transaction_controller_.current_transaction (t, payload, size);
 283     }
 284
 285     void
 286     api ()
 287     {
 288       transaction_controller_.api ();
 289     }
 290
 291   private:
 292     ACE_hthread_t thread_;
 293
 294     ACE_Thread_Mutex mutex_;
 295     ACE_Condition<ACE_Thread_Mutex> cond_;
 296
 297 // FUZZ: disable check_for_ACE_Guard
 298     typedef ACE_Guard<ACE_Thread_Mutex> AutoLock;
 299 // FUZZ: enable check_for_ACE_Guard
 300
 301     char id_[Protocol::MEMBER_ID_LENGTH];
 302
 303     ACE_INET_Addr addr_;
 304     ACE_SOCK_Dgram_Mcast sock_;
 305
 306     MessageQueue& out_control_;
 307
 308     MessageQueue  in_data_;
 309     MessageQueue  in_link_data_;
 310     MessageQueue  in_control_;
 311
 312     // Protocol state
 313     //
 314     //
 315
 316     ACE_Time_Value sync_schedule;
 317
 318     FaultDetector fault_detector_;
 319     TransactionController transaction_controller_;
 320   };
 321
 322
 323   //
 324   //
 325   //
 326   class Group::GroupImpl
 327   {
 328   public:
 329     virtual ~GroupImpl ()
 330     {
 331     }
 332
 333     GroupImpl (ACE_INET_Addr const& addr, char const* id)
 334         : send_cond_ (mutex_),
 335           recv_cond_ (mutex_),
 336           failed_ (false),
 337           in_send_data_ (mutex_),
 338           in_recv_data_ (mutex_),
 339           in_control_ (mutex_),
 340           scheduler_ (new  Scheduler (addr,
 341                                       id,
 342                                       in_send_data_,
 343                                       in_recv_data_,
 344                                       in_control_)),
 345           out_data_ (scheduler_->in_data ())
 346     {
 347       in_send_data_.subscribe (send_cond_);
 348       in_recv_data_.subscribe (recv_cond_);
 349
 350       in_control_.subscribe (send_cond_);
 351       in_control_.subscribe (recv_cond_);
 352     }
 353
 354     //FUZZ: disable check_for_lack_ACE_OS
 355     void send (void const* msg, size_t size)
 356     {
 357     //FUZZ: enable check_for_lack_ACE_OS
 358
 359       if (size > Protocol::MAX_PAYLOAD_SIZE) throw InvalidArg ();
 360
 361       // Note the potential deadlock if I lock mutex_ and out_data_ in
 362       // reverse order.
 363
 364       MessageQueueAutoLock l1 (out_data_);
 365       AutoLock l2 (mutex_);
 366
 367       throw_if_failed ();
 368
 369       out_data_.push (MessagePtr (new Send (msg, size)));
 370
 371       l1.unlock (); // no need to keep it locked
 372
 373       while (true)
 374       {
 375         throw_if_failed ();
 376
 377         if (!in_send_data_.empty ())
 378         {
 379           MessagePtr m (in_send_data_.front ());
 380           in_send_data_.pop ();
 381
 382           if (dynamic_cast<ACE_TMCast::Aborted*>(m.get()))
 383           {
 384             throw Group::Aborted ();
 385           }
 386           else if (dynamic_cast<Commited*>(m.get()))
 387           {
 388             return;
 389           }
 390           else
 391           {
 392             // cerr << "send: group-scheduler messaging protocol violation; "
 393             //     << "unexpected message " << typeid (*m).name ()
 394             //     << " " << typeid (Aborted).name () << endl;
 395
 396             ACE_OS::abort ();
 397           }
 398         }
 399
 400         // cerr << "send: waiting on condition" << endl;
 401         send_cond_.wait ();
 402         // cerr << "send: wokeup on condition" << endl;
 403       }
 404     }
 405
 406     //FUZZ: disable check_for_lack_ACE_OS
 407     size_t recv (void* msg, size_t size)
 408     {
 409     //FUZZ: enable check_for_lack_ACE_OS
 410
 411       AutoLock lock (mutex_);
 412
 413       while (true)
 414       {
 415         throw_if_failed ();
 416
 417         if (!in_recv_data_.empty ())
 418         {
 419           MessagePtr m (in_recv_data_.front ());
 420           in_recv_data_.pop ();
 421
 422           Recv* data = dynamic_cast<Recv*> (m.get ());
 423           if (data)
 424           {
 425             Recv* data = dynamic_cast<Recv*> (m.get ());
 426
 427             if (size < data->size ())
 428               throw Group::InsufficienSpace ();
 429
 430             ACE_OS::memcpy (msg, data->payload (), data->size ());
 431
 432             return data->size ();
 433           }
 434           else
 435           {
 436             // cerr << "recv: group-scheduler messaging protocol violation. "
 437             //     << "unexpected message " << typeid (*m).name () << endl;
 438
 439             ACE_OS::abort ();
 440           }
 441         }
 442
 443         recv_cond_.wait ();
 444       }
 445     }
 446
 447   private:
 448     void
 449     throw_if_failed ()
 450     {
 451       if (!failed_ && !in_control_.empty ()) failed_ = true;
 452
 453       if (failed_) throw Group::Failed ();
 454     }
 455
 456   private:
 457     ACE_Thread_Mutex mutex_;
 458     ACE_Condition<ACE_Thread_Mutex> send_cond_;
 459     ACE_Condition<ACE_Thread_Mutex> recv_cond_;
 460
 461     // FUZZ: disable check_for_ACE_Guard
 462     typedef ACE_Guard<ACE_Thread_Mutex> AutoLock;
 463     // FUZZ: enable check_for_ACE_Guard
 464
 465     bool failed_;
 466
 467     MessageQueue  in_send_data_;
 468     MessageQueue  in_recv_data_;
 469     MessageQueue  in_control_;
 470
 471     std::unique_ptr<Scheduler> scheduler_;
 472
 473     MessageQueue& out_data_;
 474   };
 475
 476
 477   // Group
 478   //
 479   //
 480   Group::
 481   Group (ACE_INET_Addr const& addr, char const* id)
 482       : pimpl_ (new GroupImpl (addr, id))
 483   {
 484   }
 485
 486   Group::
 487   ~Group ()
 488   {
 489   }
 490
 491   void
 492   Group::send (void const* msg, size_t size)
 493   {
 494     pimpl_->send (msg, size);
 495   }
 496
 497   size_t
 498   Group::recv (void* msg, size_t size)
 499   {
 500     return pimpl_->recv (msg, size);
 501   }
 502 }
 503