Changeset 69 for osgVisual/src/cluster

Timestamp:

Jul 20, 2010, 10:44:13 AM (14 years ago)

Author:

Torben Dannhauer

Message:

Network sync:
now works:

• serialization of transport container
• transport via ENet UDP
• de-serialization of the transport container.
• transport of viewmatrix and framenumber to the slave.

ToDo?: apply viewmatrix on slave still do not work.

Location:

osgVisual/src/cluster

Files:

• dataIO_clusterENet.cpp (modified) (10 diffs)
• dataIO_clusterENet_implementation.cpp (modified) (4 diffs)
• dataIO_clusterUDP.cpp (modified) (1 diff)

osgVisual/src/cluster/dataIO_clusterENet.cpp

@@ -23 +23 @@
         OSG_NOTIFY( osg::ALWAYS ) << "clusterENet constructed" << std::endl;
         serverToConnect = "unknown";
+        hardSync = false;       // integrate into init()
+        port = 12345;   // integrate into init()
 }
 
@@ -32 +34 @@
 
 
-void dataIO_clusterENet::init( osg::ArgumentParser& arguments_, clustermode clusterMode_, osgVisual::dataIO_transportContainer* sendContainer_, bool compressionEnabled_, bool asAscii_ )
+void dataIO_clusterENet::init( osg::ArgumentParser& arguments_, osgViewer::Viewer* viewer_, clustermode clusterMode_, osgVisual::dataIO_transportContainer* sendContainer_, bool compressionEnabled_, bool asAscii_ )
 {
         OSG_NOTIFY( osg::ALWAYS ) << "clusterENet init();" << std::endl;
         
+        // Store viewer
+        viewer = viewer_;
+
         // Configure the clustermode
         clusterMode = clusterMode_;
@@ -41 +46 @@
         // store sendContainer
         sendContainer = sendContainer_;
+
+        // Configure Compression and instantiate read/write-options
+        std::string readOptionString = "";
+        std::string writeOptionString = "";
+        if(asAscii_)
+        {
+                readOptionString = "Ascii";
+                writeOptionString = "Ascii";
+        }
+        if (compressionEnabled_)
+                writeOptionString+=" Compressor=zlib";
+        readOptions = new osgDB::Options( readOptionString.c_str() );
+        writeOptions = new osgDB::Options( writeOptionString.c_str() );
+
+        // Get ReaderWriter
+        rw = osgDB::Registry::instance()->getReaderWriterForExtension("osgb");  
         
         // create ENet implementation object.
-        enet_impl = new osgVisual::dataIO_clusterENet_implementation();
+        enet_impl = new osgVisual::dataIO_clusterENet_implementation(receivedTransportContainer);
 
         // initialize ENet implementation
         if(clusterMode == MASTER)
         {
+                std::cout << "Init dataIO_cluster_ENet as Server on port " << port << std::endl;
                 enet_impl->init(dataIO_clusterENet_implementation::SERVER, port);
 
@@ -81 +103 @@
                 {
                         initialized = false;
-                        std::cout << "Failed to establish connection to server " << serverToConnect << std::endl;
+                        std::cout << "Finally failed to establish connection to server " << serverToConnect << std::endl;
+                        exit(-1);
                 }
         }       // IF SLAVE END
@@ -95 +118 @@
 bool dataIO_clusterENet::sendTO_OBJvaluesToSlaves()
 {
-        OSG_NOTIFY( osg::ALWAYS ) << "clusterENet sendTO_OBJvaluesToSlaves()" << std::endl;
-
+        //OSG_NOTIFY( osg::ALWAYS ) << "clusterENet sendTO_OBJvaluesToSlaves()" << std::endl;
+
+        if(sendContainer.valid())
+        {
+                // Pack FrameID & Viewmatrix
+                sendContainer->setFrameID(viewer->getFrameStamp()->getFrameNumber());
+                sendContainer->setViewMatrix(viewer->getCamera()->getViewMatrix());
+
+                // Writing node to stream
+                std::stringstream myOstream;
+                if ( rw )
+                {
+                        osgDB::ReaderWriter::WriteResult wr = rw->writeObject( *sendContainer.get(), myOstream, writeOptions );
+                        if (wr.success() )                      
+                        {
+                                // Send Data via ENet:
+                                OSG_NOTIFY( osg::ALWAYS ) << "dataIO_clusterUDP::sendTO_OBJvaluesToSlaves() - Bytes to send: " << myOstream.str().length() << std::endl;
+                                //sOSG_NOTIFY( osg::ALWAYS ) << "Send: " << myOstream.str() << std::endl;
+                                ENetPacket * packet = enet_packet_create (myOstream.str().c_str(), 
+                                                                                                  myOstream.str().size(), 
+                                                                                                  ENET_PACKET_FLAG_RELIABLE);
+                
+                                // Send data
+                                enet_impl->sendPacket( packet, 0, 0, true);
+                        }
+                        else OSG_NOTIFY( osg::WARN ) << "ERROR: dataIO_clusterENet::sendTO_OBJvaluesToSlaves() :: Save failed: " << wr.message() << std::endl;
+                }
+                else OSG_NOTIFY( osg::WARN ) << "ERROR: dataIO_clusterENet::sendTO_OBJvaluesToSlaves() :: Unable to get readerWriter for osgb" << std::endl;
+        }
+        else OSG_NOTIFY( osg::WARN ) << "ERROR: dataIO_clusterENet::sendTO_OBJvaluesToSlaves() :: Invalid transportContainer" << std::endl;
+
+
+        enet_impl->processEvents();     // As Master: process events AFTER doing anything to have up to have the "sent" commands in queue.
         return true;
 }
@@ -103 +157 @@
 bool dataIO_clusterENet::readTO_OBJvaluesFromMaster()
 {
-        OSG_NOTIFY( osg::ALWAYS ) << "clusterENet readTO_OBJvaluesFromMaster()" << std::endl;
+        //OSG_NOTIFY( osg::ALWAYS ) << "clusterENet readTO_OBJvaluesFromMaster()" << std::endl;
+        enet_impl->processEvents();     // As Slave: process events BEFORE doing anything to have up to date values.
+
+        int bytes_received = receivedTransportContainer.size();
+        if (bytes_received > 0 )
+        {
+                OSG_NOTIFY( osg::ALWAYS ) << "dataIO_clusterENet::readTO_OBJvaluesFromMaster() - Bytes received: " << bytes_received << std::endl;
+                //OSG_NOTIFY( osg::ALWAYS ) << "Received: " << std::endl << receivedTransportContainer << std::endl;
+
+                // Unserialize data
+                if ( rw )
+                {
+                        std::stringstream test;
+                        test << receivedTransportContainer;
+                        osgDB::ReaderWriter::ReadResult rr = rw->readObject( test, readOptions );
+                        if (rr.success())
+                        {
+                                sendContainer = dynamic_cast<osgVisual::dataIO_transportContainer*>(rr.takeObject());
+                                if (sendContainer)
+                                {
+                                        OSG_NOTIFY( osg::ALWAYS ) << "Received::FrameID is: " << sendContainer->getFrameID() << std::endl;
+                                        // Restore Viewmatrix
+                                        viewer->getCamera()->setViewMatrix(sendContainer->getViewMatrix());
+                                }
+                                else
+                                        OSG_NOTIFY( osg::WARN ) << "ERROR: dataIO_clusterENet::readTO_OBJvaluesFromMaster() - Unable to cast converted node to transportContainer" << std::endl;
+                        }
+                        else
+                                OSG_NOTIFY( osg::WARN ) << "ERROR: dataIO_clusterENet::readTO_OBJvaluesFromMaster() - Unable to convert stream to node" << std::endl;
+                }
+                else
+                        OSG_NOTIFY( osg::WARN ) << "ERROR: dataIO_clusterENet::readTO_OBJvaluesFromMaster() - Unable to get readerWriter for osgb" << std::endl;
+        }       // IF bytes recv > 0 END
+
+
 
         return true;
@@ -111 +199 @@
 void dataIO_clusterENet::reportAsReadyToSwap()
 {
+        if(!hardSync)
+                return;
+
         OSG_NOTIFY( osg::ALWAYS ) << "clusterENet reportAsReadyToSwap()" << std::endl;
 }
@@ -116 +207 @@
 bool dataIO_clusterENet::waitForSwap()
 {
+        if(!hardSync)
+                return true;
+
         OSG_NOTIFY( osg::ALWAYS ) << "clusterENet waitForSwap()" << std::endl;
 
@@ -124 +218 @@
 bool dataIO_clusterENet::waitForAllReadyToSwap()
 {
+        if(!hardSync)
+                return true;
+
         OSG_NOTIFY( osg::ALWAYS ) << "clusterENet waitForAllReadyToSwap()" << std::endl;
 
@@ -132 +229 @@
 bool dataIO_clusterENet::sendSwapCommand()
 {
+        if(!hardSync)
+                return true;
+
         OSG_NOTIFY( osg::ALWAYS ) << "clusterENet sendSwapCommand()" << std::endl;
 

osgVisual/src/cluster/dataIO_clusterENet_implementation.cpp

@@ -21 +21 @@
 int dataIO_clusterENet_implementation::activeENetInstances = 0;
 
-dataIO_clusterENet_implementation::dataIO_clusterENet_implementation()
+dataIO_clusterENet_implementation::dataIO_clusterENet_implementation(std::string& receivedTransportContainer_)
+: receivedTransportContainer(receivedTransportContainer_)
 {
         std::cout << "Instantiated server class# "<< activeENetInstances << std::endl;
@@ -142 +143 @@
         if( peerList.size() == 0 )
         {
-                std::cout << "dataIO_clusterENet_implementation::sendPacket() - ERROR: No connected peer available!" << std::endl;
+                //std::cout << "dataIO_clusterENet_implementation::sendPacket() - ERROR: No connected peer available!" << std::endl;
                 return;
         }
@@ -215 +216 @@
 {
         if(currentRole != dataIO_clusterENet_implementation::CLIENT)
+        {
+                std::cout << "dataIO_clusterENet_implementation::connectTo() : ERROR: ENet does not work as client - ignoring to connect!" << std::endl;
                 return false;
+        }
 
         ENetAddress address;
@@ -257 +261 @@
 void dataIO_clusterENet_implementation::onReceivePacket(ENetEvent* event_)
 {
-                std::string datastring;
-                datastring.assign((char*)(event_->packet->data), event_->packet->dataLength);   
-                std::cout << "A packet of length "<<event_->packet->dataLength<<" containing "<<datastring<<" was received from "<<datastring<<" on channel "<<(int)(event_->channelID)<<std::endl;
-                datastring+=std::string("answer");
-
-                ENetPacket * packet = enet_packet_create (datastring.c_str(), 
-                                                                                                  datastring.size(), 
-                                                                                                  ENET_PACKET_FLAG_RELIABLE);
-                
-                // Send answer
-                sendPacket( packet, 0, 0, true);
+                receivedTransportContainer.assign((char*)(event_->packet->data), event_->packet->dataLength);   
+                //std::cout << "A packet of length "<<event_->packet->dataLength<<" containing "<<receivedTransportContainer<<" was received from "<<receivedTransportContainer<<" on channel "<<(int)(event_->channelID)<<std::endl;
 }
 

osgVisual/src/cluster/dataIO_clusterUDP.cpp

@@ -128 +128 @@
                 else
                         OSG_NOTIFY( osg::WARN ) << "ERROR: dataIO_clusterUDP::readTO_OBJvaluesFromMaster() - Unable to get readerWriter for osgb" << std::endl;
-        }
+        }       // IF bytes recv >0