Source/System/Cluster/Server/OSGClusterServer.cpp

   1 /*---------------------------------------------------------------------------*\
   2  *                                OpenSG                                     *
   3  *                                                                           *
   4  *                                                                           *
   5  *             Copyright (C) 2000-2002 by the OpenSG Forum                   *
   6  *                                                                           *
   7  *                            www.opensg.org                                 *
   8  *                                                                           *
   9  *   contact: dirk@opensg.org, gerrit.voss@vossg.org, jbehr@zgdv.de          *
  10  *                                                                           *
  11 \*---------------------------------------------------------------------------*/
  12 /*---------------------------------------------------------------------------*\
  13  *                                License                                    *
  14  *                                                                           *
  15  * This library is free software; you can redistribute it and/or modify it   *
  16  * under the terms of the GNU Library General Public License as published    *
  17  * by the Free Software Foundation, version 2.                               *
  18  *                                                                           *
  19  * This library is distributed in the hope that it will be useful, but       *
  20  * WITHOUT ANY WARRANTY; without even the implied warranty of                *
  21  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU         *
  22  * Library General Public License for more details.                          *
  23  *                                                                           *
  24  * You should have received a copy of the GNU Library General Public         *
  25  * License along with this library; if not, write to the Free Software       *
  26  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.                 *
  27  *                                                                           *
  28 \*---------------------------------------------------------------------------*/
  29 /*---------------------------------------------------------------------------*\
  30  *                                Changes                                    *
  31  *                                                                           *
  32  *                                                                           *
  33  *                                                                           *
  34  *                                                                           *
  35  *                                                                           *
  36  *                                                                           *
  37 \*---------------------------------------------------------------------------*/
  38
  39 #include <stdlib.h>
  40 #include <stdio.h>
  41 #include <boost/bind.hpp>
  42
  43 #include "OSGConfig.h"
  44 #include "OSGThread.h"
  45 #include "OSGThreadManager.h"
  46 #include "OSGClusterServer.h"
  47 #include "OSGPointConnection.h"
  48 #include "OSGConnectionFactory.h"
  49 #include "OSGDgramSocket.h"
  50 #include "OSGClusterWindow.h"
  51 #include "OSGBinaryMessage.h"
  52 #include "OSGClusterNetwork.h"
  53 #include "OSGRemoteAspect.h"
  54 #include "OSGTypeFactory.h"
  55
  56 #include "OSGImageComposer.h"
  57
  58 // testing
  59 //#include "OSGMultiDisplayWindow.h"
  60
  61 #include "OSGOSGWriter.h"
  62
  63 OSG_USING_NAMESPACE
  64
  65 /*! \class OSG::ClusterServer
  66     \brief Cluster rendering server
  67
  68    A ClusterServer is responsible for syncronizing all client changes.
  69    Each cluster renderer can offer it's service by a symbolic name.
  70    So it is possible to have a server called "left" or "right".
  71    The server uses a local Qt or GLUT window for rendering.
  72    <pre>
  73    // create a server
  74    GLUTWindowPtr window=GLUTWindow::create();
  75    server = new ClusterServer(window,"server1","Multicast");
  76    // wait for clients to connect
  77    server->init();
  78    ...
  79    // render
  80    server->render(ract);
  81    </pre>
  82  */
  83
  84 std::vector<InitFuncF> *ClusterServer::osgInitFunctions = NULL;
  85 std::vector<ExitFuncF> *ClusterServer::osgExitFunctions = NULL;
  86
  87 void ClusterServer::addInitFunction(InitFuncF initFunc)
  88 {
  89     if(osgInitFunctions == NULL)
  90     {
  91         osgInitFunctions = new std::vector<InitFuncF>(0);
  92     }
  93
  94     osgInitFunctions->push_back(initFunc);
  95 }
  96
  97 void ClusterServer::addExitFunction(ExitFuncF exitFunc)
  98 {
  99     if(osgExitFunctions == NULL)
 100     {
 101         osgExitFunctions = new std::vector<ExitFuncF>(0);
 102     }
 103
 104     osgExitFunctions->push_back(exitFunc);
 105 }
 106
 107 bool ClusterServer::init(Int32,
 108                          Char8 **)
 109 {
 110     bool returnValue = true;
 111
 112     if(GlobalSystemState != Running)
 113     {
 114         FFATAL(("ClusterServer::init: System not initialized; calls is "
 115                 "NOT allowed.\n"));
 116
 117         returnValue = false;
 118     }
 119
 120     if(osgInitFunctions != NULL)
 121     {
 122         for(UInt32 i = 0; i < osgInitFunctions->size(); i++)
 123         {
 124             returnValue &= (*osgInitFunctions)[i]();
 125
 126             if(returnValue == false)
 127                 break;
 128         }
 129
 130         osgInitFunctions->clear();
 131     }
 132
 133     return returnValue;
 134 }
 135
 136 bool ClusterServer::exit(void)
 137 {
 138     bool returnValue = true;
 139
 140     if(GlobalSystemState != Running)
 141     {
 142          return true;
 143     }
 144
 145     if(osgExitFunctions != NULL)
 146     {
 147         for(PtrDiffT i = osgExitFunctions->size() - 1; i >= 0; i--)
 148         {
 149             returnValue &= (*osgExitFunctions)[i]();
 150
 151             if(returnValue == false)
 152                 break;
 153         }
 154     }
 155
 156     delete osgExitFunctions;
 157
 158     return returnValue;
 159 }
 160
 161 /*-------------------------------------------------------------------------*/
 162 /*                            Constructors                                 */
 163
 164 /*! Constructor
 165  *
 166  * \param window          rendering window. e.g. a Qt or GLUT window
 167  * \param serviceName     wait for connections that request this name
 168  * \param connectionType  network type. e.g. "Multicast"
 169  * \param address         address to wait for connections
 170  * \param servicePort     port to wait for connections
 171  * \param serviceGroup    service group
 172  *
 173  */
 174 ClusterServer::ClusterServer(           Window *window,
 175                              const std::string &serviceName,
 176                              const std::string &connectionType,
 177                              const std::string &address,
 178                                         UInt32  servicePort,
 179                              const std::string &serviceGroup):
 180     _window(window),
 181     _connection(NULL),
 182     _requestAddress(address),
 183     _boundAddress(""),
 184     _clusterWindow(),
 185     _aspect(NULL),
 186     _serviceName(serviceName),
 187     _connectionType(connectionType),
 188     _servicePort(servicePort),
 189     _serviceGroup(serviceGroup),
 190     _serverId(0),
 191     _interface("")
 192 {
 193     char localhost[256];
 194
 195     // default is hostname
 196     if(_serviceName.empty())
 197     {
 198         osgGetHostname(localhost,255);
 199         _serviceName = localhost;
 200     }
 201     // if service contains ":" than treat as address
 202     if(_requestAddress.empty())
 203     {
 204         if(strstr(_serviceName.c_str(),":"))
 205             _requestAddress = _serviceName;
 206     }
 207 }
 208
 209 /*-------------------------------------------------------------------------*/
 210 /*                             Destructor                                  */
 211
 212 /*! Destructor. Disconnect from all connected rendering servers
 213  */
 214
 215 ClusterServer::~ClusterServer(void)
 216 {
 217     _window = NULL;
 218
 219     try
 220     {
 221         delete _connection;
 222         delete _aspect;
 223     }
 224
 225     catch(...)
 226     {
 227     }
 228 }
 229
 230 /*-------------------------------------------------------------------------*/
 231 /*                             Class specific                              */
 232
 233
 234 /*! start server
 235  *
 236  * Start cluster server and wait for a client to connect. This method
 237  * will return after a client connection or an error situation.
 238  */
 239
 240 void ClusterServer::start(void)
 241 {
 242     OSG::FieldContainerType *fct;
 243
 244     // reset conneciton
 245
 246     delete _connection;
 247
 248     _connection = NULL;
 249
 250     // create aspect
 251     _aspect = new RemoteAspect();
 252
 253     // register interrest for all changed cluster windows
 254     for(UInt32 i = 1;
 255                i < OSG::TypeFactory::the()->getNumTypes();
 256              ++i)
 257     {
 258         fct = OSG::FieldContainerFactory::the()->findType(i);
 259
 260         if(fct && fct->isDerivedFrom(ClusterWindow::getClassType()))
 261         {
 262             _aspect->registerChanged(
 263                 *fct,
 264                 boost::bind(&ClusterServer::windowChanged, this, _1, _2));
 265         }
 266     }
 267
 268     // accept incomming connections
 269     try
 270     {
 271         UInt8                    forceNetworkOrder;
 272 #if BYTE_ORDER == LITTLE_ENDIAN
 273         UInt8                    littleEndian = true;
 274 #else
 275         UInt8                    littleEndian = false;
 276 #endif
 277
 278         // accept
 279         acceptClient();
 280
 281         // determine network order
 282         _connection->putValue(littleEndian);
 283         _connection->flush();
 284         _connection->selectChannel();
 285         _connection->getValue(forceNetworkOrder);
 286         _connection->setNetworkOrder((forceNetworkOrder != 0));
 287     }
 288     catch(...)
 289     {
 290         throw;
 291     }
 292 }
 293
 294 /*! Stop cluster server, remove current remote aspect and all its
 295     field containers.
 296  */
 297
 298 void ClusterServer::stop()
 299 {
 300     // get aspect ownership
 301     if(_clusterWindow != NULL)
 302     {
 303         _aspect = _clusterWindow->getNetwork()->getAspect();
 304
 305         _clusterWindow->getNetwork()->setAspect(NULL);
 306
 307         // That's the app one we will never receive as without the
 308         // app window it can not be send.
 309         _clusterWindow->subReferenceUnresolved();
 310         _clusterWindow = NULL;
 311     }
 312
 313     if(_window != NULL)
 314         _window->resolveLinks();
 315
 316     // destroy connection
 317
 318     try
 319     {
 320         delete _connection;
 321         _connection = NULL;
 322     }
 323     catch(...)
 324     {
 325     }
 326
 327     // destroy aspect
 328     delete _aspect;
 329
 330     // reset
 331     _connection   = NULL;
 332     _aspect       = NULL;
 333 }
 334
 335 #ifdef OSG_OLD_RENDER_ACTION
 336
 337 /*! sync with client and render scenegraph
 338  */
 339
 340 void ClusterServer::render(DrawActionBase *action)
 341 {
 342     doSync  (false );
 343     doRender(action);
 344     doSwap  (      );
 345 }
 346
 347 #endif
 348
 349 /*! sync with client and render scenegraph
 350  */
 351
 352 void ClusterServer::render(RenderActionBase *action)
 353 {
 354     doSync  (false );
 355     doRender(action);
 356     doSwap  (      );
 357 }
 358
 359
 360 /*! Synchronize all field containers with the client and call
 361  *  <code>serverInit</code>, <code>serverRender</code> and
 362  *  <code>serverSwap</code> for the cluster window.
 363  *  The cluster server uses the first synced ClusterWindow that
 364  *  contains the name of this server. <code>serverInit</code> is
 365  *  called after the first ClusterWindow sync.
 366  *
 367  *  todo: Sync RenderAciton contents
 368  */
 369
 370 void ClusterServer::doSync(bool applyToChangelist)
 371 {
 372     // do we have a cluster window?
 373     if(_clusterWindow == NULL)
 374     {
 375         do
 376         {
 377             // recive
 378             _aspect->receiveSync(*_connection,applyToChangelist);
 379         }
 380         while(_clusterWindow == NULL);
 381
 382         // get server id
 383         for(_serverId = 0;
 384              (_clusterWindow->getServers(_serverId) != _serviceName) &&
 385              (_serverId < _clusterWindow->getMFServers()->size());
 386             _serverId++) ;
 387
 388         // server connected and cluster window found
 389         SINFO << "Start server " << _serviceName
 390               << " with id "     << _serverId
 391               << std::endl;
 392
 393       // now the window is responsible for connection and aspect
 394
 395         _clusterWindow->getNetwork()->setMainConnection(_connection);
 396         _clusterWindow->getNetwork()->setAspect        (_aspect);
 397
 398         _connection = NULL;
 399         _aspect     = NULL;
 400
 401         _clusterWindow->setDrawerId(_serverId);
 402         _clusterWindow->serverInit(_window,_serverId);
 403     }
 404
 405     RemoteAspect *aspect     = _clusterWindow->getNetwork()->getAspect();
 406     Connection   *connection =
 407         _clusterWindow->getNetwork()->getMainConnection();
 408
 409     // sync with render clinet
 410     aspect->receiveSync(*connection, applyToChangelist);
 411
 412     // sync with render client
 413     if(_clusterWindow->getInterleave())
 414     {
 415         // if the reminder of the division of interleave and
 416         // framecount is equal to the servers id, the right
 417         // sync point for the current render frame is reached
 418         while( ( _clusterWindow->getFrameCount() %
 419                  _clusterWindow->getInterleave() )  !=
 420                (_serverId%_clusterWindow->getInterleave()) )
 421         {
 422             aspect->receiveSync(*connection, applyToChangelist);
 423         }
 424     }
 425
 426     if(applyToChangelist)
 427     {
 428         commitChanges();
 429     }
 430     else
 431     {
 432         commitChangesAndClear();
 433     }
 434 }
 435
 436 /*! render server window
 437  */
 438
 439 void ClusterServer::doRender(RenderActionBase *action)
 440 {
 441 #if 0
 442     OSG::IndentFileOutStream outFileStream("/tmp/cluster.osg");
 443
 444     if(outFileStream)
 445     {
 446         //std::cerr << "STARTING PRINTOUT:" << std::endl;
 447
 448         OSG::OSGWriter writer(outFileStream, 4);
 449
 450         writer.write(_clusterWindow);
 451
 452         outFileStream.close();
 453     }
 454 #endif
 455
 456     _clusterWindow->serverRender(_window, _serverId, action);
 457 }
 458
 459 /*! swap server window
 460  */
 461
 462 void ClusterServer::doSwap(void)
 463 {
 464     _clusterWindow->serverSwap(_window, _serverId);
 465 }
 466
 467 /*! return the cluster window received from the client
 468  */
 469
 470 Window *ClusterServer::getClusterWindow(void)
 471 {
 472     return _clusterWindow;
 473 }
 474
 475 /*! return the window used for rendering
 476  */
 477
 478 Window *ClusterServer::getServerWindow(void)
 479 {
 480     return _window;
 481 }
 482
 483 /*! clusterWindow changed callback. This is a callback functor.
 484     It is called for each change of a ClusterWindow.
 485  */
 486
 487 bool ClusterServer::windowChanged(FieldContainer * const fcp,
 488                                   RemoteAspect   *          )
 489 {
 490     if(_clusterWindow != NULL)
 491         return true;
 492
 493     ClusterWindow *window = dynamic_cast<ClusterWindow *>(fcp);
 494
 495     if(window->getMFServers()->size())
 496     {
 497         MFString::const_iterator sIt =
 498             window->getMFServers()->find(_serviceName);
 499
 500         if(sIt == window->getMFServers()->end())
 501         {
 502             SWARNING << "wrong window" << std::endl;
 503         }
 504         else
 505         {
 506             _clusterWindow = window;
 507
 508             if(_window != NULL)
 509             {
 510                 fprintf(stderr, "%p %" PRISize " %td\n",
 511                         static_cast<void *>(&(*_window)),
 512                         window->getMFServers()->size(),
 513                         sIt - window->getMFServers()->begin());
 514
 515                 const MFUInt32 &vIds = *(window->getMFServerIds());
 516
 517                 if(window->getMFServers()->size() <= vIds.size())
 518                 {
 519                     _window->setDrawerId(vIds[sIt -
 520                                               window->getMFServers()->begin()]);
 521                 }
 522                 else
 523                 {
 524                     _window->setDrawerId(sIt - window->getMFServers()->begin());
 525                 }
 526             }
 527         }
 528     }
 529
 530     return true;
 531 }
 532
 533 /*! Wait for incomming clients. A client can send a request for a
 534  *  special connection type or it can try to connect it it knows
 535  *  the servers address.
 536  */
 537
 538 void ClusterServer::acceptClient(void)
 539 {
 540     BinaryMessage  msg;
 541     DgramSocket    serviceSock;
 542     SocketAddress  addr;
 543     std::string    service;
 544     std::string    connectionType;
 545     UInt32         readable;
 546     bool           connected=false;
 547     std::string    address;
 548     bool           bound = false;
 549
 550     SINFO << "Waiting for request of "
 551           << _serviceName
 552           << std::endl;
 553
 554     try
 555     {
 556         if(!_requestAddress.empty())
 557         {
 558             // create connection
 559
 560             _connection = ConnectionFactory::the()->createPoint(
 561                 _connectionType);
 562
 563             if(_connection)
 564             {
 565                 // set interface
 566                 _connection->setInterface(_interface);
 567                 // bind connection
 568                 try
 569                 {
 570                     // bind to requested address
 571                     _boundAddress = _connection->bind(_requestAddress);
 572                     bound = true;
 573                 }
 574                 catch(...)
 575                 {
 576                     SINFO << "Unable to bind, use name as symbolic "
 577                           << "service name"
 578                           << std::endl;
 579                 }
 580             }
 581         }
 582
 583         serviceSock.open();
 584         serviceSock.setReusePort(true);
 585
 586         // join to multicast group
 587         if(!_serviceGroup.empty())
 588         {
 589             SocketAddress groupAddress =
 590                 SocketAddress(_serviceGroup.c_str(),
 591                               _servicePort);
 592
 593             if(groupAddress.isMulticast())
 594             {
 595                 SINFO << "wait for request on multicast:"
 596                       << _serviceGroup << std::endl;
 597
 598                 serviceSock.bind(SocketAddress(SocketAddress::ANY,
 599                                                _servicePort));
 600                 serviceSock.join(SocketAddress(groupAddress));
 601             }
 602             else
 603             {
 604                 SINFO << "wait for request by broadcast:"
 605                       << _serviceGroup << std::endl;
 606                 serviceSock.bind(SocketAddress(groupAddress));
 607             }
 608         }
 609         else
 610         {
 611             SINFO << "wait for request by broadcast" << std::endl;
 612             serviceSock.bind(SocketAddress(SocketAddress::ANY,
 613                                            _servicePort));
 614         }
 615
 616         while(!connected)
 617         {
 618             try
 619             {
 620                 if(_connection)
 621                     readable = serviceSock.waitReadable(.01);
 622                 else
 623                     readable = true;
 624
 625                 if(readable)
 626                 {
 627                     serviceSock.recvFrom(msg,addr);
 628
 629                     service        = msg.getString();
 630                     connectionType = msg.getString();
 631
 632                     SINFO << "Request for "
 633                           << service << " "
 634                           << connectionType
 635                           << std::endl;
 636
 637                     if(service == _serviceName)
 638                     {
 639                         // remove old connection if typename missmaches
 640                         if(_connection &&
 641                            _connection->getType()->getName() != connectionType)
 642                         {
 643                             delete _connection;
 644                             _connection = NULL;
 645                         }
 646
 647                         // try to create connection
 648                         if(!_connection)
 649                         {
 650                             // create connection
 651                             _connection =
 652                                 ConnectionFactory::the()->createPoint(
 653                                     connectionType);
 654
 655                             if(_connection)
 656                             {
 657                                 // set interface
 658                                 _connection->setInterface(_interface);
 659                                 // bind connection
 660                                 _boundAddress = _connection->bind(
 661                                     _requestAddress);
 662
 663                                 bound = true;
 664                             }
 665                             else
 666                             {
 667                                 SINFO << "Unknown connection type '"
 668                                       << connectionType << "'" << std::endl;
 669                             }
 670                         }
 671
 672                         if(_connection)
 673                         {
 674                             msg.clear    (             );
 675                             msg.putString(_serviceName );
 676                             msg.putString(_boundAddress);
 677                             serviceSock.sendTo(msg, addr);
 678
 679                             SINFO << "Response "
 680                                   << connectionType << ":"
 681                                   << _boundAddress
 682                                   << std::endl;
 683                         }
 684                     }
 685                 }
 686             }
 687
 688             catch(SocketConnReset &e)
 689             {
 690                 // ignore if there is a connection. This can happen, if
 691                 // a client has send a request. The server has send an
 692                 // answer meanwile the client has send a second request
 693                 // the client gets the answer to the first request and
 694                 // the server tries to send a second answer. The second
 695                 // answer can not be delivered because the client has
 696                 // closed its service port. This is a win-socket problem.
 697
 698                 SWARNING << e.what() << std::endl;
 699
 700                 // if there is no connection, then its a real problem
 701                 if(!_connection)
 702                     throw;
 703             }
 704             catch(OSG_STDEXCEPTION_NAMESPACE::exception &e)
 705             {
 706                 SWARNING << e.what() << std::endl;
 707             }
 708             try
 709             {
 710                 // try to accept
 711                 if(bound && _connection && _connection->acceptGroup(0.2) >= 0)
 712                 {
 713                     connected = true;
 714                     SINFO << "Connection accepted "
 715                           << _boundAddress
 716                           << std::endl;
 717                 }
 718             }
 719
 720             catch(OSG_STDEXCEPTION_NAMESPACE::exception &e)
 721             {
 722                 SWARNING << e.what() << std::endl;
 723             }
 724         }
 725
 726         serviceSock.close();
 727     }
 728
 729     catch(OSG_STDEXCEPTION_NAMESPACE::exception &)
 730     {
 731         throw;
 732     }
 733 }
 734