Changeset 605 for trunk/avango-core

Timestamp:

01/10/12 17:36:46 (4 months ago)

Author:

acbernst

Message:

added ZMQ_DISTRIBUTION_SUPPORT

File:

• trunk/avango-core/src/avango/network/NetNode.cpp (modified) (12 diffs)

trunk/avango-core/src/avango/network/NetNode.cpp

@@ -38 +38 @@
 #include "NetGroup.h"
 
+#include <avango/Config.h> 
+
+#if defined(ZMQ_DISTRIBUTION_SUPPORT)
+#include "NetNodeClient.h"
+#include "NetNodeServer.h"
+#include <boost/tokenizer.hpp>
+#endif
+
 namespace
 {
   av::Logger &logger(av::getLogger("av::NetNode"));
+
+#if defined(ZMQ_DISTRIBUTION_SUPPORT)
+  av::NetNodeClient* gClient = 0;
+  av::NetNodeServer* gServer = 0;
+  const std::string  gClientEndpoint("{Endpt:127.0.0.2:34818:215:0}");
+  const std::string  gServerEndpoint("{Endpt:127.0.0.2:34818:215:0}");
+#endif
 }
 
@@ -68 +83 @@
 #endif
 
+#ifdef ZMQ_DISTRIBUTION_SUPPORT
+  bool isServer;
+  std::string hostName;
+  std::string port;
+  uint64_t serverHWM = 2;
+  {
+    // from http://stackoverflow.com/questions/541561/using-boost-tokenizer-escaped-list-separator-with-different-parameters
+    std::string gn(groupName);
+    
+    typedef boost::tokenizer< boost::escaped_list_separator<char> > Tokenizer;
+    boost::escaped_list_separator<char> Separator( ' ', '|' );
+    Tokenizer tok( gn, Separator );
+    unsigned numTokens = 0;
+    for (Tokenizer::iterator iter = tok.begin(); iter != tok.end(); ++iter) {
+      ++numTokens;
+    }
+    if (3 == numTokens) {
+      Tokenizer::iterator iter = tok.begin();
+      isServer = ("AVSERVER" == *iter);
+      ++iter;
+      hostName = *iter;
+      ++iter;
+      port = *iter;
+    } else if (4 == numTokens) {
+      Tokenizer::iterator iter = tok.begin();
+      isServer = ("AVSERVER" == *iter);
+      ++iter;
+      hostName = *iter;
+      ++iter;
+      port = *iter;
+      serverHWM = atoi((*iter).c_str());
+    } else {
+      std::stringstream msg;
+      msg << "ERROR in av::NetNode::join(const std::string& groupName), could not join - invalid CONFIG";
+      throw (std::runtime_error(msg.str()));
+    }
+
+  }
+  
+  if (isServer) {
+    gServer = new NetNodeServer(hostName,port, this, gClientEndpoint, gServerEndpoint, serverHWM);
+    joined(gServerEndpoint);
+
+    av::Msg av_msg;
+    getStateFragment(gServerEndpoint,av_msg);
+    setStateFragment(gClientEndpoint,av_msg);
+  } else {
+    gClient = new NetNodeClient(hostName,port, this, gClientEndpoint, gServerEndpoint);
+    joined(gClientEndpoint);
+    gClient->start();
+  }
+  //handleNetworkSends();
+  //handleNetworkReceives();
+
+  mIdCounter = 0;
+
+#else
   leave();
 
@@ -129 +201 @@
   logger.debug() << "av::NetNode::join: completed.";
 #endif
+
+#endif // #ifdef ZMQ_DISTRIBUTION_SUPPORT
 }
 
@@ -148 +222 @@
 {
   assert(onAir());
+#ifdef ZMQ_DISTRIBUTION_SUPPORT
+  if(gServer)
+     return NetID(gServerEndpoint, ++mIdCounter);
+  else
+     return NetID(gClientEndpoint, ++mIdCounter);
+#else
   return NetID(netEID(), ++mIdCounter);
+#endif
   assert(mIdCounter > 0); // detect overflow
 }
@@ -155 +236 @@
 av::NetNode::onAir() const
 {
+#ifdef ZMQ_DISTRIBUTION_SUPPORT
+  return true;
+#else
   return (mMember != 0);
+#endif
 }
 
@@ -170 +255 @@
 {
   // consume_received_messages();
-
+#ifdef ZMQ_DISTRIBUTION_SUPPORT
+  if (0 != gServer) {
+    notifyGroup();
+  }
+#else
   if (!mBlocked)
     notifyGroup();
+#endif
 }
 
@@ -178 +268 @@
 av::NetNode::handleNetworkReceives()
 {
+#ifdef ZMQ_DISTRIBUTION_SUPPORT
+  if (gClient) {
+    consumeReceivedMessages();
+  }
+#else
   while (mMember && mMember->upcallSerializer().handleNextUpcall(this))
     ;
+#endif
 }
 
@@ -420 +516 @@
   av_pushMsg(p_msg, sPackedMsg);
 
+#ifdef ZMQ_DISTRIBUTION_SUPPORT
+  gServer->cast(p_msg);
+#else
   mMember->cast(p_msg);
+#endif
 }
 
@@ -459 +559 @@
   av_pushMsg(p_msg, sPackedMsg);
 
+#ifdef ZMQ_DISTRIBUTION_SUPPORT
+  gServer->cast(p_msg);
+#else
   mMember->cast(p_msg);
+#endif
 }
 
@@ -497 +601 @@
   av_pushMsg(p_msg, sPackedMsg);
 
+#ifdef ZMQ_DISTRIBUTION_SUPPORT
+  gServer->cast(p_msg);
+#else
   mMember->cast(p_msg);
+#endif
 }
 
@@ -714 +822 @@
 
 void
-  av::NetNode::receiveMessage(const std::string &/*origin*/, av::Msg& msg)
+av::NetNode::receiveMessage(const std::string &origin, av::Msg& msg)
 {
   // stack up the received messages in any case
@@ -720 +828 @@
   // which received it.
   boost::mutex::scoped_lock lock(mMessageMutex);
+
+#ifdef ZMQ_DISTRIBUTION_SUPPORT
+  // we need to reset our old state fragment before we receive the new state fragment
+  if (!mObjectMap.slotExists(origin))
+  {
+    logger.debug() << "ALARM: in av::NetNode::receiveMessage !!!!!!!!!!!!!!!! slot was empty for origin " << origin;
+    mObjectMap.addSlot(origin);
+  }
+#endif
 
   Msg av_msg(msg);