OSG::ClusterServer Class Reference

Cluster rendering server. More...

#include <OSGClusterServer.h>


Public Member Functions
Constructors

	ClusterServer (WindowPtr window, const std::string &serviceName, const std::string &connectionType="StreamSock", const std::string &address="", UInt32 servicePort=8437, const std::string &serviceGroup="")
Destructor

virtual	~ClusterServer (void)
server actions

void	start (void)
void	stop (void)
void	doSync (bool applyToChangelist)
void	doSwap (void)
window access

WindowPtr	getClusterWindow (void)
WindowPtr	getServerWindow (void)
RemoteAspect *	getRemoteAspect (void) const
set

void	setInterface (const std::string &interf)
Protected Member Functions
ClusterWindow changed function

bool	windowChanged (const FieldContainerPtr &fcp, RemoteAspect *)
Protected Attributes
Member

WindowPtr	_window
PointConnection *	_connection
std::string	_requestAddress
std::string	_boundAddress
ClusterWindowPtr	_clusterWindow
RemoteAspect *	_aspect
std::string	_serviceName
std::string	_connectionType
UInt32	_servicePort
std::string	_serviceGroup
UInt32	_serverId
std::string	_interface
Private Member Functions
	ClusterServer (const ClusterServer &source)
	prohibit default function (move to 'public' if needed)
void	operator= (const ClusterServer &source)
	prohibit default function (move to 'public' if needed)
helper function

void	acceptClient ()

Detailed Description

A ClusterServer is responsible for syncronizing all client changes. Each cluster renderer can offer it's service by a symbolic name. So it is possible to have a server called "left" or "right". The server uses a local Qt or GLUT window for rendering.

 // create a server
 GLUTWindowPtr window=GLUTWindowcreate();
 server = new ClusterServer(window,"server1","Multicast");
 // wait for clients to connect
 server->init();
 ...
 // render
 server->render(ract);

Definition at line 54 of file OSGClusterServer.h.

Constructor & Destructor Documentation

ClusterServer::ClusterServer	(	WindowPtr	window,
		const std::string &	serviceName,
		const std::string &	connectionType = `"StreamSock"`,
		const std::string &	address = `""`,
		UInt32	servicePort = `8437`,
		const std::string &	serviceGroup = `""`
	)

Definition at line 97 of file OSGClusterServer.cpp.

References _requestAddress, _serviceName, _window, OSG::addRef(), and OSG::osgGetHostname().

00102                                                               :
00103     _window(window),
00104     _connection(NULL),
00105     _requestAddress(address),
00106     _boundAddress(""),
00107     _clusterWindow(),
00108     _aspect(NULL),
00109     _serviceName(serviceName),
00110     _connectionType(connectionType),
00111     _servicePort(servicePort),
00112     _serviceGroup(serviceGroup),
00113     _serverId(0),
00114     _interface("")
00115 {
00116     addRef(_window);
00117     char localhost[256];
00118 
00119     // default is hostname
00120     if(_serviceName.empty())
00121     {
00122         osgGetHostname(localhost,255);
00123         _serviceName = localhost;
00124     }
00125     // if service contains ":" than treat as address
00126     if(_requestAddress.empty())
00127     {
00128         if(strstr(_serviceName.c_str(),":"))
00129             _requestAddress = _serviceName;
00130     }
00131 }

ClusterServer::~ClusterServer ( void ) [virtual]

Definition at line 139 of file OSGClusterServer.cpp.

References _aspect, and _connection.

00140 {
00141     try
00142     {
00143         delete _connection;
00144         delete _aspect;
00145     }
00146 
00147     catch(...)
00148     {
00149     }
00150 }

OSG::ClusterServer::ClusterServer ( const ClusterServer & source ) [private]

Member Function Documentation

void ClusterServer::start ( void )

Definition at line 162 of file OSGClusterServer.cpp.

References _aspect, _connection, acceptClient(), OSG::ExceptionBinaryDataHandler::flush(), OSG::ClusterWindowBase::getClassType(), OSG::ExceptionBinaryDataHandler::getValue(), OSG::TypeBase::isDerivedFrom(), OSG::ExceptionBinaryDataHandler::putValue(), OSG::RemoteAspect::registerChanged(), OSG::Connection::selectChannel(), OSG::ExceptionBinaryDataHandler::setNetworkOrder(), OSG::SingletonHolder< SingletonT >::the(), and windowChanged().

00163 {
00164     OSG::FieldContainerType *fct;
00165 
00166     // reset conneciton
00167 
00168     delete _connection;
00169 
00170     _connection = NULL;
00171 
00172     // create aspect
00173     _aspect = new RemoteAspect();
00174 
00175     // register interrest for all changed cluster windows
00176     for(UInt32 i = 1; 
00177                i < OSG::TypeFactory::the()->getNumTypes(); 
00178              ++i)
00179     {
00180         fct = OSG::FieldContainerFactory::the()->findType(i);
00181 
00182         if(fct && fct->isDerivedFrom(ClusterWindow::getClassType()))
00183         {
00184             _aspect->registerChanged(
00185                 *fct,
00186                 boost::bind(&ClusterServer::windowChanged, this, _1, _2));
00187         }
00188     }
00189 
00190     // accept incomming connections
00191     try 
00192     {
00193         UInt8                    forceNetworkOrder;
00194 #if BYTE_ORDER == LITTLE_ENDIAN
00195         UInt8                    littleEndian = true;
00196 #else
00197         UInt8                    littleEndian = false;
00198 #endif
00199 
00200         // accept
00201         acceptClient();
00202 
00203         // determine network order
00204         _connection->putValue(littleEndian);
00205         _connection->flush();
00206         _connection->selectChannel();
00207         _connection->getValue(forceNetworkOrder);
00208         _connection->setNetworkOrder((forceNetworkOrder != 0));
00209     } 
00210     catch(...)
00211     {
00212         throw;
00213     }
00214 }

void ClusterServer::stop ( void )

Stop cluster server, remove current remote aspect and all its field containers.

Definition at line 220 of file OSGClusterServer.cpp.

References _aspect, _clusterWindow, _connection, and NullFC.

00221 {
00222     // get aspect ownership
00223     if(_clusterWindow != NullFC)
00224     {
00225         _aspect = _clusterWindow->getNetwork()->getAspect();
00226 
00227         _clusterWindow->getNetwork()->setAspect(NULL);
00228     }
00229     // destroy connection
00230 
00231     try
00232     {
00233         delete _connection;
00234         _connection = NULL;
00235     }
00236     catch(...)
00237     {
00238     }
00239 
00240     // destroy aspect
00241     delete _aspect;
00242 
00243     // reset 
00244     _connection   = NULL;
00245     _aspect       = NULL;
00246     _clusterWindow= NullFC;
00247 }

void ClusterServer::doSync ( bool applyToChangelist )

Synchronize all field containers with the client and call serverInit, serverRender and serverSwap for the cluster window. The cluster server uses the first synced ClusterWindow that contains the name of this server. serverInit is called after the first ClusterWindow sync.

todo: Sync RenderAciton contents

Definition at line 287 of file OSGClusterServer.cpp.

References _aspect, _clusterWindow, _connection, _serverId, _serviceName, _window, OSG::commitChanges(), OSG::commitChangesAndClear(), NullFC, OSG::RemoteAspect::receiveSync(), SINFO, and size.

00288 {
00289     // do we have a cluster window?
00290     if(_clusterWindow==NullFC)
00291     {
00292         do
00293         {
00294             // recive 
00295             _aspect->receiveSync(*_connection,applyToChangelist);
00296         }
00297         while(_clusterWindow==NullFC);
00298 
00299         // get server id
00300         for(_serverId = 0;
00301              (_clusterWindow->getServers()[_serverId] != _serviceName) &&
00302              (_serverId < _clusterWindow->getServers().size());
00303             _serverId++);
00304 
00305         // server connected and cluster window found
00306         SINFO << "Start server " << _serviceName 
00307               << " with id "     << _serverId 
00308               << std::endl;
00309   
00310       // now the window is responsible for connection and aspect
00311 
00312         _clusterWindow->getNetwork()->setMainConnection(_connection);
00313         _clusterWindow->getNetwork()->setAspect        (_aspect);
00314 
00315         _connection = NULL;
00316         _aspect     = NULL;
00317 
00318         _clusterWindow->setDrawerId(_serverId);
00319         _clusterWindow->serverInit(_window,_serverId);
00320     }
00321 
00322     RemoteAspect *aspect     = _clusterWindow->getNetwork()->getAspect();
00323     Connection   *connection = 
00324         _clusterWindow->getNetwork()->getMainConnection();
00325 
00326     // sync with render clinet
00327     aspect->receiveSync(*connection, applyToChangelist);
00328 
00329     // sync with render client
00330     if(_clusterWindow->getInterleave())
00331     {
00332         // if the reminder of the division of interleave and 
00333         // framecount is equal to the servers id, the right
00334         // sync point for the current render frame is reached
00335         while( ( _clusterWindow->getFrameCount() %
00336                  _clusterWindow->getInterleave() )  != 
00337                (_serverId%_clusterWindow->getInterleave()) ) 
00338         {
00339             aspect->receiveSync(*connection, applyToChangelist);
00340         }
00341     }
00342 
00343     if(applyToChangelist)
00344     {
00345         commitChanges();
00346     }
00347     else
00348     {
00349         commitChangesAndClear();
00350     }
00351 }

void ClusterServer::doSwap ( void )

swap server window

Definition at line 408 of file OSGClusterServer.cpp.

References _clusterWindow, _serverId, and _window.

00409 {
00410     _clusterWindow->serverSwap(_window, _serverId);
00411 }

WindowPtr ClusterServer::getClusterWindow ( void )

Definition at line 416 of file OSGClusterServer.cpp.

References _clusterWindow.

00417 {
00418     return _clusterWindow;
00419 }

WindowPtr ClusterServer::getServerWindow ( void )

return the window used for rendering

Definition at line 424 of file OSGClusterServer.cpp.

References _window.

00425 {
00426     return _window;
00427 }

RemoteAspect * OSG::ClusterServer::getRemoteAspect ( void ) const [inline]

Definition at line 43 of file OSGClusterServer.inl.

References _aspect.

00044 {
00045     return _aspect;
00046 }

void OSG::ClusterServer::setInterface ( const std::string & interf ) [inline]

Definition at line 49 of file OSGClusterServer.inl.

References _interface.

00050 {
00051     _interface = interf;
00052 }

bool ClusterServer::windowChanged	(	const FieldContainerPtr &	fcp,
		RemoteAspect *
	)			`[protected]`

Definition at line 433 of file OSGClusterServer.cpp.

References _clusterWindow, _serviceName, NullFC, and SWARNING.

Referenced by start().

00435 {
00436     if(_clusterWindow != NullFC)
00437         return true;
00438     
00439     ClusterWindowPtr window = dynamic_cast<ClusterWindowPtr>(fcp);
00440 
00441     if(window->getServers().size())
00442     {
00443         if(window->getServers().find(_serviceName) == 
00444            window->getServers().end())
00445         {
00446             SWARNING << "wrong window" << std::endl;
00447         }
00448         else
00449         {
00450             _clusterWindow = window;
00451         }
00452     }
00453 
00454     return true;
00455 }

void ClusterServer::acceptClient ( ) [private]

Definition at line 462 of file OSGClusterServer.cpp.

References _boundAddress, _connection, _connectionType, _interface, _requestAddress, _serviceGroup, _serviceName, _servicePort, OSG::PointConnection::acceptGroup(), OSG::SocketAddress::ANY, OSG::Socket::bind(), OSG::Connection::bind(), OSG::DgramSocket::close(), OSG::DgramSocket::join(), OSG::DgramSocket::open(), OSG::DgramSocket::recvFrom(), OSG::DgramSocket::sendTo(), OSG::Connection::setInterface(), OSG::Socket::setReusePort(), SINFO, SWARNING, OSG::Socket::waitReadable(), and OSG::Exception::what().

Referenced by start().

00463 {
00464     BinaryMessage  msg;
00465     DgramSocket    serviceSock;
00466     SocketAddress  addr;
00467     std::string    service;
00468     std::string    connectionType;
00469     UInt32         readable;
00470     bool           connected=false;
00471     std::string    address;
00472     bool           bound = false;
00473 
00474     SINFO << "Waiting for request of "
00475           << _serviceName
00476           << std::endl;
00477 
00478     try
00479     {
00480         if(!_requestAddress.empty())
00481         {            
00482             // create connection
00483 
00484             _connection = ConnectionFactory::the()->createPoint(
00485                 _connectionType);
00486 
00487             if(_connection)
00488             {
00489                 // set interface
00490                 _connection->setInterface(_interface);
00491                 // bind connection
00492                 try 
00493                 {
00494                     // bind to requested address
00495                     _boundAddress = _connection->bind(_requestAddress);
00496                     bound = true;
00497                 }
00498                 catch(...)
00499                 {
00500                     SINFO << "Unable to bind, use name as symbolic "
00501                           << "service name" 
00502                           << std::endl;
00503                 }
00504             }
00505         }
00506 
00507         serviceSock.open();
00508         serviceSock.setReusePort(true);
00509 
00510         // join to multicast group
00511         if(!_serviceGroup.empty())
00512         {
00513             SocketAddress groupAddress = 
00514                 SocketAddress(_serviceGroup.c_str(),
00515                               _servicePort);
00516 
00517             if(groupAddress.isMulticast())
00518             {
00519                 SINFO << "wait for request on multicast:" 
00520                       << _serviceGroup << std::endl;
00521 
00522                 serviceSock.bind(SocketAddress(SocketAddress::ANY,
00523                                                _servicePort));
00524                 serviceSock.join(SocketAddress(groupAddress));
00525             }
00526             else
00527             {
00528                 SINFO << "wait for request by broadcast:"
00529                       << _serviceGroup << std::endl;
00530                 serviceSock.bind(SocketAddress(groupAddress));
00531             }
00532         }
00533         else
00534         {
00535             SINFO << "wait for request by broadcast" << std::endl;
00536             serviceSock.bind(SocketAddress(SocketAddress::ANY,
00537                                            _servicePort));
00538         }
00539 
00540         while(!connected)
00541         {
00542             try
00543             {
00544                 if(_connection) 
00545                     readable = serviceSock.waitReadable(.01);
00546                 else
00547                     readable = true;
00548 
00549                 if(readable)
00550                 {
00551                     serviceSock.recvFrom(msg,addr);
00552 
00553                     service        = msg.getString();
00554                     connectionType = msg.getString();
00555 
00556                     SINFO << "Request for " 
00557                           << service << " " 
00558                           << connectionType 
00559                           << std::endl;
00560                     
00561                     if(service == _serviceName)
00562                     {
00563                         // remove old connection if typename missmaches
00564                         if(_connection && 
00565                            _connection->getType()->getName() != connectionType)
00566                         {
00567                             delete _connection;
00568                             _connection = NULL;
00569                         }
00570 
00571                         // try to create connection
00572                         if(!_connection)
00573                         {
00574                             // create connection
00575                             _connection = 
00576                                 ConnectionFactory::the()->createPoint(
00577                                     connectionType);
00578 
00579                             if(_connection)
00580                             {
00581                                 // set interface
00582                                 _connection->setInterface(_interface);
00583                                 // bind connection
00584                                 _boundAddress = _connection->bind(
00585                                     _requestAddress);
00586 
00587                                 bound = true;
00588                             } 
00589                             else
00590                             {
00591                                 SINFO << "Unknown connection type '" 
00592                                       << connectionType << "'" << std::endl;
00593                             }
00594                         }
00595 
00596                         if(_connection)
00597                         {
00598                             msg.clear    (             );
00599                             msg.putString(_serviceName );
00600                             msg.putString(_boundAddress);
00601                             serviceSock.sendTo(msg, addr);
00602 
00603                             SINFO << "Response " 
00604                                   << connectionType << ":"
00605                                   << _boundAddress 
00606                                   << std::endl;
00607                         }
00608                     }
00609                 }
00610             }
00611 
00612             catch(SocketConnReset &e)
00613             {
00614                 // ignore if there is a connection. This can happen, if
00615                 // a client has send a request. The server has send an
00616                 // answer meanwile the client has send a second request
00617                 // the client gets the answer to the first request and
00618                 // the server tries to send a second answer. The second
00619                 // answer can not be delivered because the client has
00620                 // closed its service port. This is a win-socket problem.
00621 
00622                 SWARNING << e.what() << std::endl;
00623 
00624                 // if there is no connection, then its a real problem
00625                 if(!_connection)
00626                     throw;
00627             }
00628             catch(OSG_STDEXCEPTION_NAMESPACE::exception &e)
00629             {
00630                 SWARNING << e.what() << std::endl;
00631             }
00632             try 
00633             {
00634                 // try to accept
00635                 if(bound && _connection && _connection->acceptGroup(0.2) >= 0)
00636                 {
00637                     connected = true;
00638                     SINFO << "Connection accepted "
00639                           << _boundAddress 
00640                           << std::endl;
00641                 }
00642             }
00643 
00644             catch(OSG_STDEXCEPTION_NAMESPACE::exception &e)
00645             {
00646                 SWARNING << e.what() << std::endl;
00647             }
00648         }
00649 
00650         serviceSock.close();
00651     }
00652 
00653     catch(OSG_STDEXCEPTION_NAMESPACE::exception &e)
00654     {
00655         throw;
00656     }
00657 }