ACE/protocols/ace/TMCast/TransactionController.hpp

   1 // author    : Boris Kolpackov <boris@dre.vanderbilt.edu>
   2
   3 #include "ace/OS_NS_string.h"
   4 #include "ace/OS_NS_stdlib.h"
   5 #include "ace/Synch.h"
   6 #include "ace/Bound_Ptr.h"
   7
   8 #include "Protocol.hpp"
   9 #include "Messaging.hpp"
  10
  11 #include <typeinfo>
  12
  13 namespace ACE_TMCast
  14 {
  15
  16   // Messages
  17   //
  18   //
  19   class Send : public virtual Message
  20   {
  21   public:
  22     Send (void const* msg, size_t size)
  23         : size_ (size)
  24     {
  25       ACE_OS::memcpy (payload_, msg, size_);
  26     }
  27
  28     void const*
  29     payload () const
  30     {
  31       return payload_;
  32     }
  33
  34     size_t
  35     size () const
  36     {
  37       return size_;
  38     }
  39
  40   private:
  41     size_t size_;
  42     char payload_[Protocol::MAX_PAYLOAD_SIZE];
  43   };
  44
  45   typedef
  46   ACE_Strong_Bound_Ptr<Send, ACE_SYNCH_MUTEX>
  47   SendPtr;
  48
  49
  50   class Recv : public virtual Message
  51   {
  52   public:
  53     Recv (void const* msg, size_t size)
  54         : size_ (size)
  55     {
  56       ACE_OS::memcpy (payload_, msg, size_);
  57     }
  58
  59     void const*
  60     payload () const
  61     {
  62       return payload_;
  63     }
  64
  65     size_t
  66     size () const
  67     {
  68       return size_;
  69     }
  70
  71   private:
  72     size_t size_;
  73     char payload_[Protocol::MAX_PAYLOAD_SIZE];
  74   };
  75
  76   typedef
  77   ACE_Strong_Bound_Ptr<Recv, ACE_SYNCH_MUTEX>
  78   RecvPtr;
  79
  80   class Aborted : public virtual Message {};
  81
  82   class Commited : public virtual Message {};
  83
  84
  85   //
  86   //
  87   //
  88   class TransactionController
  89   {
  90   public:
  91     TransactionController (MessageQueue& in,
  92                            MessageQueue& send_out,
  93                            MessageQueue& recv_out)
  94       : //trace_ (false),
  95           voting_duration_ (0),
  96           separation_duration_ (0),
  97           in_ (in),
  98           send_out_ (send_out),
  99           recv_out_ (recv_out)
 100     {
 101       current_.id = 0;
 102       current_.status = Protocol::TS_COMMITED;
 103     }
 104
 105   public:
 106     class Failure {};
 107
 108
 109     void
 110     outsync (Protocol::Transaction& c, void* payload, size_t& size)
 111     {
 112       if (current_.status == Protocol::TS_COMMIT ||
 113           current_.status == Protocol::TS_ABORT)
 114       {
 115         if (++voting_duration_ >= Protocol::VOTING_FRAME)
 116         {
 117           // end of voting frame
 118
 119           if (current_.status == Protocol::TS_COMMIT)
 120           {
 121             {
 122               if (initiated_)
 123               {
 124                 MessageQueueAutoLock lock (send_out_);
 125                 send_out_.push (MessagePtr (new Commited));
 126               }
 127               else // joined transaction
 128               {
 129                 MessageQueueAutoLock lock (recv_out_);
 130                 recv_out_.push (MessagePtr (recv_));
 131                 recv_ = RecvPtr ();
 132               }
 133             }
 134
 135             current_.status = Protocol::TS_COMMITED;
 136
 137             // if (trace_) cerr << "commited transaction with id "
 138             //                  << current_.id << endl;
 139           }
 140           else // TS_ABORT
 141           {
 142             if (initiated_)
 143             {
 144               MessageQueueAutoLock lock (send_out_);
 145               send_out_.push (MessagePtr (new Aborted));
 146             }
 147             else
 148             {
 149               // free revc_ buffer if necessary
 150               //
 151               if (recv_.get ()) recv_ = RecvPtr ();
 152             }
 153
 154
 155             current_.status = Protocol::TS_ABORTED;
 156
 157             // if (trace_) cerr << "aborted transaction with id "
 158             //                  << current_.id << endl;
 159           }
 160
 161           // start transaction separation frame (counts down)
 162           // +1 because it will be decremented on this iteration
 163           separation_duration_ = Protocol::SEPARATION_FRAME + 1;
 164         }
 165       }
 166
 167       // Set current outsync info
 168
 169       c.id = current_.id;
 170       c.status = current_.status;
 171
 172
 173       // Do some post-processing
 174
 175       switch (current_.status)
 176       {
 177       case Protocol::TS_COMMITED:
 178       case Protocol::TS_ABORTED:
 179         {
 180           if (separation_duration_ > 0) --separation_duration_;
 181           break;
 182         }
 183       case Protocol::TS_BEGIN:
 184         {
 185           // transfer payload
 186
 187           size = send_->size ();
 188           ACE_OS::memcpy (payload, send_->payload (), size);
 189
 190           send_ = SendPtr ();
 191
 192           // get redy to vote for 'commit'
 193
 194           current_.status = Protocol::TS_COMMIT;
 195           voting_duration_ = 0;
 196         }
 197       }
 198     }
 199
 200     void
 201     current_transaction (Protocol::Transaction const& t,
 202                          void const* payload,
 203                          size_t size)
 204     {
 205       Protocol::TransactionId& id = current_.id;
 206       Protocol::TransactionStatus& s = current_.status;
 207
 208       if (id == 0 && t.id != 0) // catch up
 209       {
 210         switch (t.status)
 211         {
 212         case Protocol::TS_BEGIN:
 213         case Protocol::TS_COMMIT:
 214         case Protocol::TS_ABORT:
 215           {
 216             id = t.id - 1;
 217             s = Protocol::TS_COMMITED;
 218             break;
 219           }
 220         case Protocol::TS_ABORTED:
 221         case Protocol::TS_COMMITED:
 222           {
 223             id = t.id;
 224             s = t.status;
 225             break;
 226           }
 227         }
 228
 229         // if (trace_) cerr << "caught up with id " << id << endl;
 230       }
 231
 232       bool stable (s == Protocol::TS_COMMITED || s == Protocol::TS_ABORTED);
 233
 234       switch (t.status)
 235       {
 236       case Protocol::TS_BEGIN:
 237         {
 238           if (!stable || t.id != id + 1)
 239           {
 240             // Transaction is in progress or hole in transaction id's
 241
 242             // cerr << "unexpected request to join " << t
 243             //     << " while on " << current_ << endl;
 244
 245             // if (!stable) cerr << "voting progress is " << voting_duration_
 246             //                  << "/" << Protocol::VOTING_FRAME << endl;
 247
 248             if (t.id == id) // collision
 249             {
 250               if (!stable && s != Protocol::TS_ABORT)
 251               {
 252                 // abort both
 253                 // cerr << "aborting both transactions" << endl;
 254
 255                 s = Protocol::TS_ABORT;
 256                 voting_duration_ = 0; //@@ reset voting frame
 257               }
 258             }
 259             else
 260             {
 261               // @@ delicate case. need to think more
 262
 263               // cerr << "Declaring node failed." << endl;
 264               throw Failure ();
 265             }
 266           }
 267           else
 268           {
 269             // join the transaction
 270
 271             initiated_ = false;
 272
 273             recv_ = RecvPtr (new Recv (payload, size));
 274
 275             id = t.id;
 276             s = Protocol::TS_COMMIT;
 277             voting_duration_ = 0;
 278
 279             // if (trace_) cerr << "joining-for-commit transaction with id "
 280             //                  << id << endl;
 281           }
 282           break;
 283         }
 284       case Protocol::TS_COMMIT:
 285         {
 286           if (stable && id == t.id - 1)
 287           {
 288             // not begin and and we haven't joined
 289
 290             // join for abort
 291
 292             initiated_ = false;
 293
 294             current_.id = t.id;
 295             current_.status = Protocol::TS_ABORT;
 296             voting_duration_ = 0;
 297
 298             // if (trace_) cerr << "joining-for-abort transaction with id "
 299             //                  << current_.id << endl;
 300           }
 301           break;
 302         }
 303       case Protocol::TS_ABORT:
 304         {
 305           if ((!stable && id == t.id && s == Protocol::TS_COMMIT) ||
 306               (stable && id == t.id - 1)) // abort current || new transaction
 307           {
 308             // if (trace_) cerr << "voting-for-abort on transaction with id "
 309             //                  << current_.id << endl;
 310
 311             id = t.id;
 312             s = Protocol::TS_ABORT;
 313
 314             voting_duration_ = 0; //@@ reseting voting_duration_
 315           }
 316           else
 317           {
 318           }
 319
 320           break;
 321         }
 322       case Protocol::TS_ABORTED:
 323       case Protocol::TS_COMMITED:
 324         {
 325           // nothing for now
 326           break;
 327         }
 328       }
 329     }
 330
 331     void
 332     api ()
 333     {
 334       if ((current_.status == Protocol::TS_COMMITED ||
 335            current_.status == Protocol::TS_ABORTED) &&
 336           separation_duration_ == 0) // no transaction in progress
 337       {
 338         // start new transaction
 339
 340         // Note that in_ is already locked by Scheduler
 341
 342         MessagePtr m (in_.front ());
 343         in_.pop ();
 344
 345         Message const &msg = *m; // avoid warning for side-effects in typeid
 346         if (typeid (msg) == typeid (Send))
 347         {
 348           send_ = SendPtr (m);
 349         }
 350         else
 351         {
 352           // cerr << "Expecting Send but received " << typeid (*m).name ()
 353           //      << endl;
 354
 355           ACE_OS::abort ();
 356         }
 357
 358         current_.id++;
 359         current_.status = Protocol::TS_BEGIN;
 360
 361         initiated_ = true;
 362
 363         // if (trace_) cerr << "starting transaction with id " << current_.id
 364         //                  << endl;
 365       }
 366     }
 367
 368   private:
 369     // FUZZ: disable check_for_ACE_Guard
 370     typedef ACE_Guard<ACE_Thread_Mutex> AutoLock;
 371     // FUZZ: enable check_for_ACE_Guard
 372
 373     // bool trace_;
 374
 375     Protocol::Transaction current_;
 376
 377     bool initiated_;
 378
 379     unsigned short voting_duration_;
 380     unsigned short separation_duration_;
 381
 382     MessageQueue& in_;
 383     MessageQueue& send_out_;
 384     MessageQueue& recv_out_;
 385
 386     SendPtr send_;
 387     RecvPtr recv_;
 388   };
 389 }