OSGClusterServer.cpp

Go to the documentation of this file.

/*---------------------------------------------------------------------------*\
 *                                OpenSG                                     *
 *                                                                           *
 *                                                                           *
 *             Copyright (C) 2000-2002 by the OpenSG Forum                   *
 *                                                                           *
 *                            www.opensg.org                                 *
 *                                                                           *
 *   contact: dirk@opensg.org, gerrit.voss@vossg.org, jbehr@zgdv.de          *
 *                                                                           *
\*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*\
 *                                License                                    *
 *                                                                           *
 * This library is free software; you can redistribute it and/or modify it   *
 * under the terms of the GNU Library General Public License as published    *
 * by the Free Software Foundation, version 2.                               *
 *                                                                           *
 * This library is distributed in the hope that it will be useful, but       *
 * WITHOUT ANY WARRANTY; without even the implied warranty of                *
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU         *
 * Library General Public License for more details.                          *
 *                                                                           *
 * You should have received a copy of the GNU Library General Public         *
 * License along with this library; if not, write to the Free Software       *
 * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.                 *
 *                                                                           *
\*---------------------------------------------------------------------------*/
/*---------------------------------------------------------------------------*\
 *                                Changes                                    *
 *                                                                           *
 *                                                                           *
 *                                                                           *
 *                                                                           *
 *                                                                           *
 *                                                                           *
\*---------------------------------------------------------------------------*/

#include <stdlib.h>
#include <stdio.h>

#include "OSGConfig.h"
#include "OSGThread.h"
#include "OSGThreadManager.h"
#include "OSGClusterServer.h"
#include "OSGPointConnection.h"
#include "OSGConnectionFactory.h"
#include "OSGDgramSocket.h"
#include "OSGClusterWindow.h"
#include "OSGBinaryMessage.h"
#include "OSGClusterNetwork.h"
#include "OSGSHLChunk.h"

OSG_USING_NAMESPACE

/*-------------------------------------------------------------------------*/
/*                            Constructors                                 */

ClusterServer::ClusterServer(        WindowPtr    window,
                             const std::string    &serviceName,
                             const std::string    &connectionType,
                             const std::string    &address,
                                     UInt32       servicePort,
                             const std::string    &serviceGroup):
    _window(window),
    _connection(NULL),
    _requestAddress(address),
    _boundAddress(""),
    _clusterWindow(),
    _aspect(NULL),
    _serviceName(serviceName),
    _connectionType(connectionType),
    _servicePort(servicePort),
    _serviceGroup(serviceGroup),
    _serverId(0),
    _interface("")
{
    char localhost[256];

    // default is hostname
    if(_serviceName.empty())
    {
        gethostname(localhost,255);
        _serviceName = localhost;
    }
    // if service contains ":" than treat as address
    if(_requestAddress.empty())
    {
        if(strstr(_serviceName.c_str(),":"))
            _requestAddress = _serviceName;
    }
}

/*-------------------------------------------------------------------------*/
/*                             Destructor                                  */

ClusterServer::~ClusterServer(void)
{
    try
    {
        if(_connection)
            delete _connection;
        if(_aspect)
            delete _aspect;
    }
    catch(...)
    {
    }
}

/*-------------------------------------------------------------------------*/
/*                             Class specific                              */


void ClusterServer::start()
{
    OSG::FieldContainerType *fct;

    // reset conneciton
    if(_connection)
        delete _connection;
    _connection = NULL;

    // create aspect
    _aspect = new RemoteAspect();

    // register interrest for all changed cluster windows
    for(UInt32 i = 0; i < OSG::TypeFactory::the()->getNumTypes(); ++i)
    {
        fct=OSG::FieldContainerFactory::the()->findType(i);
        if(fct && fct->isDerivedFrom(ClusterWindow::getClassType()))
        {
            _aspect->registerChanged(
                *fct,
                osgTypedMethodFunctor2ObjPtrCPtrRef
                <
                bool,
                ClusterServer,
                FieldContainerPtr,
                RemoteAspect     *
                >(this,&ClusterServer::windowChanged));
        }
    }
    // accept incomming connections
    try {
        UInt8                    forceNetworkOrder;
#if BYTE_ORDER == LITTLE_ENDIAN
        UInt8                    littleEndian = true;
#else
        UInt8                    littleEndian = false;
#endif

        // accept
        acceptClient();
        // determine network order
        _connection->putValue(littleEndian);
        _connection->flush();
        _connection->selectChannel();
        _connection->getValue(forceNetworkOrder);
        _connection->setNetworkOrder((forceNetworkOrder != 0));
    } 
    catch(...)
    {
        throw;
    }
}

void ClusterServer::stop()
{
    // get aspect ownership
    if(_clusterWindow != NullFC)
    {
        _aspect=_clusterWindow->getNetwork()->getAspect();
        _clusterWindow->getNetwork()->setAspect(NULL);
    }
    // destroy connection
    try
    {
        if(_connection)
            delete _connection;
        _connection = NULL;
    }
    catch(...)
    {
    }
    // destroy aspect
    if(_aspect)
        delete _aspect;
    // reset 
    _connection=NULL;
    _aspect=NULL;
    _clusterWindow=NullFC;
}

void ClusterServer::render(RenderActionBase *action)
{
    doSync(false);
    doRender(action);
    doSwap();
}

void ClusterServer::doSync(bool applyToChangelist)
{
    // do we have a cluster window?
    if(_clusterWindow==NullFC)
    {
        do
        {
            // recive 
            _aspect->receiveSync(*_connection,applyToChangelist);
        }
        while(_clusterWindow==NullFC);
        // get server id
        for(_serverId=0;
            _clusterWindow->getServers()[_serverId] != _serviceName &&
                _serverId<_clusterWindow->getServers().size();
            _serverId++);
        // server connected and cluster window found
        SINFO << "Start server " << _serviceName 
              << " with id "     << _serverId 
              << std::endl;
        // now the window is responsible for connection and aspect
        _clusterWindow->getNetwork()->setMainConnection(_connection);
        _clusterWindow->getNetwork()->setAspect        (_aspect);
        _connection=NULL;
        _aspect=NULL;
        SHLChunk::setClusterId(Int32(_serverId));
        _clusterWindow->serverInit(_window,_serverId);
    }

    RemoteAspect *aspect=_clusterWindow->getNetwork()->getAspect();
    Connection   *connection=_clusterWindow->getNetwork()->getMainConnection();

    // sync with render clinet
    aspect->receiveSync(*connection,applyToChangelist);

    // sync with render client
    if(_clusterWindow->getInterleave())
    {
        // if the reminder of the division of interleave and 
        // framecount is equal to the servers id, the right
        // sync point for the current render frame is reached
        while( (_clusterWindow->getFrameCount()%
                _clusterWindow->getInterleave())
               !=
               (_serverId%_clusterWindow->getInterleave()) ) 
        {
            aspect->receiveSync(*connection,applyToChangelist);
        }
    }
}

void ClusterServer::doRender(RenderActionBase *action)
{
    _clusterWindow->serverRender( _window,_serverId,action );
}

void ClusterServer::doSwap(void)
{
    _clusterWindow->serverSwap  ( _window,_serverId );
}

WindowPtr ClusterServer::getClusterWindow (void)
{
    return _clusterWindow;
}

WindowPtr ClusterServer::getServerWindow  (void)
{
    return _window;
}

bool ClusterServer::windowChanged(FieldContainerPtr& fcp,
                                  RemoteAspect *)
{
    if(_clusterWindow != NullFC)
        return true;

    ClusterWindowPtr window=ClusterWindowPtr::dcast(fcp);

    if(window->getServers().size())
    {
        if(window->getServers().find(_serviceName) == 
           window->getServers().end())
        {
            SWARNING << "wrong window" << std::endl;
        }
        else
        {
            _clusterWindow=window;
        }
    }
    return true;
}

void ClusterServer::acceptClient()
{
    BinaryMessage  msg;
    DgramSocket    serviceSock;
    SocketAddress  addr;
    std::string    service;
    std::string    connectionType;
    UInt32         readable;
    bool           connected=false;
    std::string    address;
    bool           bound = false;

    SINFO << "Waiting for request of "
          << _serviceName
          << std::endl;

    try
    {
        if(!_requestAddress.empty())
        {            
            // create connection
            _connection = ConnectionFactory::the().
                createPoint(_connectionType);
            if(_connection)
            {
                // set interface
                _connection->setInterface(_interface);
                // bind connection
                try 
                {
                    // bind to requested address
                    _boundAddress = _connection->bind(_requestAddress);
                    bound = true;
                }
                catch(...)
                {
                    SINFO << "Unable to bind, use name as symbolic service name" 
                          << std::endl;
                }
            }
        }
        serviceSock.open();
        serviceSock.setReusePort(true);
        // join to multicast group
        if(!_serviceGroup.empty())
        {
            SocketAddress groupAddress = SocketAddress(
                _serviceGroup.c_str(),
                _servicePort);
            if(groupAddress.isMulticast())
            {
                SINFO << "wait for request on multicast:" << 
                    _serviceGroup << std::endl;
                serviceSock.bind(SocketAddress(SocketAddress::ANY,
                                               _servicePort));
                serviceSock.join(SocketAddress(groupAddress));
            }
            else
            {
                SINFO << "wait for request by broadcast:" << 
                    _serviceGroup << std::endl;
                serviceSock.bind(SocketAddress(groupAddress));
            }
        }
        else
        {
            SINFO << "wait for request by broadcast" << std::endl;
            serviceSock.bind(SocketAddress(SocketAddress::ANY,
                                           _servicePort));
        }

        while(!connected)
        {
            try
            {
                if(_connection) 
                    readable = serviceSock.waitReadable(.01);
                else
                    readable = true;
                if(readable)
                {
                    serviceSock.recvFrom(msg,addr);

                    service        = msg.getString();
                    connectionType = msg.getString();

                    SINFO << "Request for " 
                          << service << " " 
                          << connectionType 
                          << std::endl;
                    
                    if(service == _serviceName)
                    {
                        // remove old connection if typename missmaches
                        if(_connection && 
                           _connection->getType()->getName() != connectionType)
                        {
                            delete _connection;
                            _connection = NULL;
                        }
                        // try to create connection
                        if(!_connection)
                        {
                            // create connection
                            _connection = ConnectionFactory::the().
                                createPoint(connectionType);
                            if(_connection)
                            {
                                // set interface
                                _connection->setInterface(_interface);
                                // bind connection
                                _boundAddress = _connection->bind(_requestAddress);
                                bound = true;
                            } 
                            else
                            {
                                SINFO << "Unknown connection type '" 
                                      << connectionType << "'" << std::endl;
                            }
                        }
                        if(_connection)
                        {
                            msg.clear    (             );
                            msg.putString(_serviceName );
                            msg.putString(_boundAddress);
                            serviceSock.sendTo(msg, addr);
                            SINFO << "Response " 
                                  << connectionType << ":"
                                  << _boundAddress 
                                  << std::endl;
                        }
                    }
                }
            }
            catch(SocketConnReset &e)
            {
                // ignore if there is a connection. This can happen, if
                // a client has send a request. The server has send an
                // answer meanwile the client has send a second request
                // the client gets the answer to the first request and
                // the server tries to send a second answer. The second
                // answer can not be delivered because the client has
                // closed its service port. This is a win-socket problem.

                SWARNING << e.what() << std::endl;

                // if there is no connection, then its a real problem
                if(!_connection)
                    throw;
            }
            catch(OSG_STDEXCEPTION_NAMESPACE::exception &e)
            {
                SWARNING << e.what() << std::endl;
            }
            try 
            {
                // try to accept
                if(bound && _connection && _connection->acceptGroup(0.2) >= 0)
                {
                    connected = true;
                    SINFO << "Connection accepted " << _boundAddress << std::endl;
                }
            }
            catch(OSG_STDEXCEPTION_NAMESPACE::exception &e)
            {
                SWARNING << e.what() << std::endl;
            }
        }
        serviceSock.close();
    }
    catch(OSG_STDEXCEPTION_NAMESPACE::exception &e)
    {
        throw;
    }
}


/*-------------------------------------------------------------------------*/
/*                              cvs id's                                   */

#ifdef __sgi
#pragma set woff 1174
#endif

#ifdef OSG_LINUX_ICC
#pragma warning( disable : 177 )
#endif

namespace
{
    static Char8 cvsid_cpp[] = "@(#)$Id:$";
    static Char8 cvsid_hpp[] = OSG_CLUSTERSERVERHEADER_CVSID;
}

---