This commit was manufactured by cvs2svn to create branchunlabeled-1.4.2

'unlabeled-1.4.2'.

1 files changed, 508 insertions, 0 deletions

diff --git a/protocols/ace/TMCast/Group.cpp b/protocols/ace/TMCast/Group.cpp
new file mode 100644
index 00000000000..901aa7ba650
--- /dev/null
+++ b/protocols/ace/TMCast/Group.cpp
@@ -0,0 +1,508 @@
+// file      : TMCast/Group.cpp
+// author    : Boris Kolpackov <boris@dre.vanderbilt.edu>
+// cvs-id    : $Id$
+
+#include "Group.hpp"
+
+#include <typeinfo>
+
+// OS primitives
+#include <ace/OS.h>
+#include <ace/Synch.h>
+#include <ace/Time_Value.h>
+#include <ace/SOCK_Dgram_Mcast.h>
+
+#include "Messaging.hpp"
+
+#include "Protocol.hpp"
+
+// Components
+
+#include "LinkListener.hpp"
+#include "FaultDetector.hpp"
+#include "TransactionController.hpp"
+
+namespace TMCast
+{
+  bool
+  operator== (std::type_info const* pa, std::type_info const& b)
+  {
+    return *pa == b;
+  }
+
+  //
+  //
+  //
+  class Terminate : public virtual Message {};
+
+
+  //
+  //
+  //
+  class Failure : public virtual Message {};
+
+
+  //
+  //
+  //
+  class Scheduler
+  {
+  public:
+    Scheduler (ACE_INET_Addr const& addr,
+               char const* id,
+               MessageQueue& out_send_data,
+               MessageQueue& out_recv_data,
+               MessageQueue& out_control)
+
+        : cond_ (mutex_),
+
+          addr_ (addr),
+          sock_ (),
+
+          out_control_ (out_control),
+
+          in_data_ (mutex_),
+          in_link_data_(mutex_),
+          in_control_ (mutex_),
+
+          sync_schedule (ACE_OS::gettimeofday ()),
+
+          transaction_controller_ (in_data_, out_send_data, out_recv_data)
+    {
+      ACE_OS::strncpy (id_, id, Protocol::MEMBER_ID_LENGTH);
+      id_[Protocol::MEMBER_ID_LENGTH - 1] = '\0';
+
+      sock_.set_option (IP_MULTICAST_TTL, 32); // @@ ttl is hardcoded
+
+      in_data_.subscribe (cond_);
+      in_link_data_.subscribe (cond_);
+      in_control_.subscribe (cond_);
+
+      if (ACE_OS::thr_create (&thread_thunk,
+                              this,
+                              THR_JOINABLE,
+                              &thread_) != 0) ::abort ();
+    }
+
+    ~Scheduler ()
+    {
+      {
+        MessageQueueAutoLock lock (in_control_);
+
+        in_control_.push (MessagePtr (new Terminate));
+      }
+
+      if (ACE_OS::thr_join (thread_, &thread_, 0) != 0) ::abort ();
+
+      // cerr << "Scheduler is down." << endl;
+    }
+
+  public:
+    MessageQueue&
+    in_data ()
+    {
+      return in_data_;
+    }
+
+  private:
+    static ACE_THR_FUNC_RETURN
+    thread_thunk (void* arg)
+    {
+      Scheduler* obj = reinterpret_cast<Scheduler*> (arg);
+
+      obj->execute ();
+      return 0;
+    }
+
+    void
+    execute ()
+    {
+      try
+      {
+        sock_.join (addr_);
+
+        auto_ptr<LinkListener> ll (new LinkListener (sock_, in_link_data_));
+
+        {
+          AutoLock lock (mutex_);
+
+          // Loop
+          //
+          //
+
+          while (true)
+          {
+            cond_.wait (&sync_schedule);
+
+            // "Loop of Fairness"
+
+            bool done = false;
+
+            do
+            {
+              // control message
+              //
+              //
+              if (!in_control_.empty ())
+              {
+                done = true;
+                break;
+              }
+
+
+              // outsync
+              //
+              //
+              if (sync_schedule < ACE_OS::gettimeofday ())
+              {
+                // OUTSYNC
+
+                outsync ();
+
+                // schedule next outsync
+
+                sync_schedule =
+                  ACE_OS::gettimeofday () +
+                  ACE_Time_Value (0, Protocol::SYNC_PERIOD);
+              }
+
+              // link message
+              //
+              //
+              if (!in_link_data_.empty ())
+              {
+                MessagePtr m (in_link_data_.front ());
+                in_link_data_.pop ();
+
+                std::type_info const* exp = &typeid (*m);
+
+                if (exp == typeid (LinkFailure))
+                {
+                  // cerr << "link failure" << endl;
+                  throw false;
+                }
+                else if (exp == typeid (LinkData))
+                {
+
+                  LinkData* data = dynamic_cast<LinkData*> (m.get ());
+
+                  // INSYNC, TL, CT
+
+                  // Filter out loopback.
+                  //
+                  if (ACE_OS::strcmp (data->header().member_id.id, id_) != 0)
+                  {
+                    insync ();
+                    transaction_list ();
+                    current_transaction (data->header().current,
+                                         data->payload (),
+                                         data->size ());
+                  }
+                }
+                else
+                {
+                  // cerr << "unknown message type from link listener: "
+                  //      << typeid (*m).name () << endl;
+                  abort ();
+                }
+              }
+
+              // api message
+              //
+              //
+              if (!in_data_.empty ())
+              {
+                // API
+
+                api ();
+              }
+
+            } while (!in_link_data_.empty() ||
+                     sync_schedule < ACE_OS::gettimeofday ());
+
+            if (done) break;
+          }
+        }
+      }
+      catch (...)
+      {
+        // cerr << "Exception in scheduler loop." << endl;
+
+        MessageQueueAutoLock lock (out_control_);
+        out_control_.push (MessagePtr (new Failure));
+      }
+    }
+
+    // Events
+    //
+    // Order:
+    //
+    // INSYNC, TSL, VOTE, BEGIN
+    // API
+    // OUTSYNC
+    //
+
+    void
+    insync ()
+    {
+      fault_detector_.insync ();
+    }
+
+    void
+    outsync ()
+    {
+      char buf[Protocol::MAX_MESSAGE_SIZE];
+
+      Protocol::MessageHeader* hdr =
+        reinterpret_cast<Protocol::MessageHeader*> (buf);
+
+      void* data = buf + sizeof (Protocol::MessageHeader);
+
+      hdr->length = sizeof (Protocol::MessageHeader);
+      hdr->check_sum = 0;
+
+      ACE_OS::strcpy (hdr->member_id.id, id_);
+
+      size_t size (0);
+
+      transaction_controller_.outsync (hdr->current, data, size);
+
+      hdr->length += size;
+
+      fault_detector_.outsync ();
+
+      // sock_.send (buf, hdr->length, addr_);
+
+      sock_.send (buf, hdr->length);
+    }
+
+    void
+    transaction_list ()
+    {
+    }
+
+    void
+    current_transaction (Protocol::Transaction const& t,
+                         void const* payload,
+                         size_t size)
+    {
+      transaction_controller_.current_transaction (t, payload, size);
+    }
+
+    void
+    api ()
+    {
+      transaction_controller_.api ();
+    }
+
+  private:
+    ACE_thread_t thread_;
+
+    ACE_Thread_Mutex mutex_;
+    ACE_Condition<ACE_Thread_Mutex> cond_;
+
+    typedef ACE_Guard<ACE_Thread_Mutex> AutoLock;
+
+    char id_[Protocol::MEMBER_ID_LENGTH];
+
+    ACE_INET_Addr addr_;
+    ACE_SOCK_Dgram_Mcast sock_;
+
+    MessageQueue& out_control_;
+
+    MessageQueue  in_data_;
+    MessageQueue  in_link_data_;
+    MessageQueue  in_control_;
+
+    // Protocol state
+    //
+    //
+
+    ACE_Time_Value sync_schedule;
+
+    FaultDetector fault_detector_;
+    TransactionController transaction_controller_;
+  };
+
+
+  //
+  //
+  //
+  class Group::GroupImpl
+  {
+  public:
+    ~GroupImpl ()
+    {
+    }
+
+    GroupImpl (ACE_INET_Addr const& addr, char const* id)
+      throw (Group::Failed)
+        : send_cond_ (mutex_),
+          recv_cond_ (mutex_),
+          failed_ (false),
+          in_send_data_ (mutex_),
+          in_recv_data_ (mutex_),
+          in_control_ (mutex_),
+          scheduler_ (new  Scheduler (addr,
+                                      id,
+                                      in_send_data_,
+                                      in_recv_data_,
+                                      in_control_)),
+          out_data_ (scheduler_->in_data ())
+    {
+      in_send_data_.subscribe (send_cond_);
+      in_recv_data_.subscribe (recv_cond_);
+
+      in_control_.subscribe (send_cond_);
+      in_control_.subscribe (recv_cond_);
+    }
+
+    void
+    send (void const* msg, size_t size)
+      throw (Group::InvalidArg, Group::Failed, Group::Aborted)
+    {
+      if (size > Protocol::MAX_PAYLOAD_SIZE) throw InvalidArg ();
+
+      // Note the potential deadlock if I lock mutex_ and out_data_ in
+      // reverse order.
+
+      MessageQueueAutoLock l1 (out_data_);
+      AutoLock l2 (mutex_);
+
+      throw_if_failed ();
+
+      out_data_.push (MessagePtr (new Send (msg, size)));
+
+      l1.unlock (); // no need to keep it locked
+
+      while (true)
+      {
+        throw_if_failed ();
+
+        if (!in_send_data_.empty ())
+        {
+          MessagePtr m (in_send_data_.front ());
+          in_send_data_.pop ();
+
+          std::type_info const* exp = &typeid (*m);
+
+          if (exp == typeid (TMCast::Aborted))
+          {
+            throw Group::Aborted ();
+          }
+          else if (exp == typeid (Commited))
+          {
+            return;
+          }
+          else
+          {
+            // cerr << "send: group-scheduler messaging protocol violation; "
+            //     << "unexpected message " << typeid (*m).name ()
+            //     << " " << typeid (Aborted).name () << endl;
+
+            abort ();
+          }
+        }
+
+        // cerr << "send: waiting on condition" << endl;
+        send_cond_.wait ();
+        // cerr << "send: wokeup on condition" << endl;
+      }
+    }
+
+
+
+    size_t
+    recv (void* msg, size_t size) throw (Group::Failed, Group::InsufficienSpace)
+    {
+      AutoLock lock (mutex_);
+
+      while (true)
+      {
+        throw_if_failed ();
+
+        if (!in_recv_data_.empty ())
+        {
+          MessagePtr m (in_recv_data_.front ());
+          in_recv_data_.pop ();
+
+          std::type_info const* exp = &typeid (*m);
+
+          if (exp == typeid (Recv))
+          {
+            Recv* data = dynamic_cast<Recv*> (m.get ());
+
+            if (size < data->size ()) throw Group::InsufficienSpace ();
+
+            memcpy (msg, data->payload (), data->size ());
+
+            return data->size ();
+          }
+          else
+          {
+            // cerr << "recv: group-scheduler messaging protocol violation. "
+            //     << "unexpected message " << typeid (*m).name () << endl;
+
+            abort ();
+          }
+        }
+
+        recv_cond_.wait ();
+      }
+    }
+
+  private:
+    void
+    throw_if_failed ()
+    {
+      if (!failed_ && !in_control_.empty ()) failed_ = true;
+
+      if (failed_) throw Group::Failed ();
+    }
+
+  private:
+    ACE_Thread_Mutex mutex_;
+    ACE_Condition<ACE_Thread_Mutex> send_cond_;
+    ACE_Condition<ACE_Thread_Mutex> recv_cond_;
+
+    typedef ACE_Guard<ACE_Thread_Mutex> AutoLock;
+
+    bool failed_;
+
+    MessageQueue  in_send_data_;
+    MessageQueue  in_recv_data_;
+    MessageQueue  in_control_;
+
+    auto_ptr<Scheduler> scheduler_;
+
+    MessageQueue& out_data_;
+  };
+
+
+  // Group
+  //
+  //
+  Group::
+  Group (ACE_INET_Addr const& addr, char const* id)
+    throw (Group::Failed)
+      : pimpl_ (new GroupImpl (addr, id))
+  {
+  }
+
+  Group::
+  ~Group ()
+  {
+  }
+
+  void 
+  Group::send (void const* msg, size_t size) throw (Group::InvalidArg, Group::Failed, Group::Aborted)
+  {
+    pimpl_->send (msg, size);
+  }
+
+  size_t 
+  Group::recv (void* msg, size_t size) throw (Group::Failed, Group::InsufficienSpace)
+  {
+    return pimpl_->recv (msg, size);
+  }
+}