source: trunk/src/testing/include/sageSync.h @ 4

/******************************************************************************
 * SAGE - Scalable Adaptive Graphics Environment
 *
 * Module: sageSync.h
 * Author : Byungil Jeong, Rajvikram Singh,
 * Description : This file includes classes to synchronize multiple groups
 *               of synchronization slaves in the SAGE processes and a class
 *               to synchronize multiple threads in a SAGE process.
 *
 * Copyright (C) 2004 Electronic Visualization Laboratory,
 * University of Illinois at Chicago
 *
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions are met:
 *
 *  * Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 *  * Redistributions in binary form must reproduce the above
 *    copyright notice, this list of conditions and the following disclaimer
 *    in the documentation and/or other materials provided with the distribution.
 *  * Neither the name of the University of Illinois at Chicago nor
 *    the names of its contributors may be used to endorse or promote
 *    products derived from this software without specific prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
 * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
 * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
 * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR
 * CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
 * EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
 * PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR
 * PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
 * LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING
 * NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
 * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 * Direct questions, comments etc about SAGE to sage_users@listserv.uic.edu or
 * http://www.evl.uic.edu/cavern/forum/
 *
 *****************************************************************************/

#ifndef _SAGESYNC_H
#define _SAGESYNC_H

#include "sageBase.h"
#include <list>
#include <map>
#include <bitset>

#define SAGE_SYNC_MSG_LEN  1280 // must be equal to SAGE_EVENT_SIZE
#define MAX_SYNC_GROUP     100
#define SYNC_MSG_BUF_LEN   64

#define SAGE_UPDATE_SETUP    1
#define SAGE_UPDATE_FOLLOW   2
#define SAGE_UPDATE_FRAME    3
#define SAGE_UPDATE_AUDIO    4

#define SAGE_CONSTANT_SYNC   1
#define SAGE_ASAP_SYNC_HARD  2
#define SAGE_ASAP_SYNC_SOFT  3

#define NORMAL_SYNC          1
#define SKIP_FRAME           2

#ifndef MSG_DONTWAIT
#define MSG_DONTWAIT    0
#endif


/**
 * class syncSlaveData.
 */
class syncSlaveData {
public:

        /**
         * a syncClient's socket file descriptor
         */
   int            clientSockFd;

   /**
    * another socket for refresh barrier - 2nd phase
        */
   int barrierClientSockFd;

   /**
    * a syncClient's IP address
        */
   sockaddr_in      clientAddr;


   /**
    * Sungwon
    * this will be filled in sageSyncServer::syncServerThread() when the object is created
    */
   int SDM; // sungwon

   /**
    * frame number.
        */
   int            frame;

   /**
    * Constructor.
    * When a syncClient calls sageSyncClinet::connectToServer(), sageSyncServer::syncServerThread() accepts it<BR>
    * then the thread creates syncSlaveData object for the client.
        */
   syncSlaveData() : frame(0), SDM(-1) {}
};

//forward declarations
class sageSyncClient;
class sageSyncServer;

#define MAX_INTERVAL_ERROR 0.1
#define SYNC_TIMEOUT       1667   /**< 1sec/60 * 10% = 1667 us */

/**
 * class syncGroup.
 */
class syncGroup {
protected:
   int noOfUpdates; /**< how many displays are updated ? */
   int slaveNum; /**< how many syncSlaves are involved in this sync group */
   pthread_mutex_t   *syncMsgLock;
   std::deque<char *> syncMsgQueue;

   int id, policy, curFrame, audioFrame, videoFrame, keyFrame, skipFrame, audioSyncCnt;

   /**
    * minimum sync signal interval<BR>
    * Even though a sync group is ready to proceed to the next frame,<BR>
    * the sync master waits until the minimum interval is reached.
    */
   double interval;

   double timeError;
   bool syncEnd;

   /**
    * when sageSyncServer::addSyncGroup() is called, the sync server object calling addSyncGroup() is assigned.
    */
   sageSyncServer *syncServer;
   pthread_t threadID;
   sageTimer timer;
   pthread_mutex_t  syncLock;
   pthread_cond_t   resumeSync; /**< condition variable */

   bool holdSync; /**< if true, then it will make checkHold() to wait on condition resumeSync */
   bool waitingInterval; /**< whether syncGroup is waiting to reach the interval */
   bool waitForKeyFrame;

public:
   /**
    * Constructor.
        */
   syncGroup() : id(0), noOfUpdates(0), curFrame(0), slaveNum(0), syncEnd(false), interval(0.0),
      policy(SAGE_ASAP_SYNC_SOFT), videoFrame(0), audioFrame(-1), keyFrame(100), holdSync(false),
      timeError(0.0), waitingInterval(false), waitForKeyFrame(false), skipFrame(0), audioSyncCnt(0)
   { syncMsgQueue.clear(); }

   int init(int startFrame, int _policy_, int groupID, int frameRate = 1, int sNum = 1); /**< syncGroup::init() */
   int setFrameRate(float frameRate);

   /**
    * TRUE, if this syncGroup reached the interval
    */
   bool checkInterval();

   /**
    * IMPORTANT<BR>
    * This function is called by sageSyncServer::manageUpdate()<BR>
    *
    * HARD_SYNC implemented<BR>
    * A syncGroup maintains the number of syncSlaves. Whenever slave sends an update, noOfUpdates increases<BR>
    * HARD_SYNC ensures every display for an app can maintain same frame number<BR>
    * by updated frame can be propagated only when noOfUpdates == number of slaves<BR>
    * i.e. all syncSlaves for an application should sends udpate
    *
    * @return 0 or NORMAL_SYNC
    */
   int processUpdate(char *data);

   int enqueSyncMsg(char *msg); /**< enqueue sync message<BR>thread safe*/
   char* dequeSyncMsg(); /**< dequeue sync message<BR>thread sage*/
   int getSyncID() { return id; }
   bool checkTimeOut();
   void blockSync(); /**< sets holdSync = true */
   void unblockSync(); /**< sets holdSync = false */

   /**
    * waits on condition resumeSync if holdSync is true
    */
   void checkHold();
   inline bool isWaiting() { return waitingInterval; }

   friend class sageSyncServer;
   friend class sageSyncClient;
};

/**
 * class sageSyncBBServer
 */
class sageSyncBBServer {
protected:
   int serverSockFd; /**< syncServer's socket */
   int maxSyncGroupID;
   int maxSlaveSockFd;
   bool syncEnd;

   /**
    * -1 old sync
    *  0 no sync
    *  1 data sync only
    *  2 two phase
    *  3 one phase
    */
   int syncLevel;

   sockaddr_in serverAddr;
   fd_set slaveFds;

   pthread_t syncThreadID, syncServerThreadID;

   std::vector<syncSlaveData> syncSlaves; /**< a list of sync slaves */

   /**
    * a list of sync slaves
    * key is SDM id
    * value is its data(mainly socket)
    */
   std::map<int,syncSlaveData> syncSlavesMap;

   sockaddr_in barrierServerAddr;
   int barrierServerSockFd;
   int barrierPort;

   fd_set barrierSlaveFds;
   int maxBarrierSlaveSockFd;

   pthread_t syncBarrierServerThreadID;
   static void* syncBarrierServerThread(void *);

   /**
    * waits syncClient's connection by calling accept() syscall<br>
    * if one connects, then it creates syncSlaveData object for it, and adds it to the vector syncSlaves
    */
   static void* syncServerThread(void*);

   /**
    * This thread is created whenever addSyncGroup() is called and under some condition.<BR>
    * 1. when the policy is SAGE_CONSTANT_SYNC<BR>
    * 2. 1 is not true and asapSyncGroupNum == 0<BR>
    * <BR>
    * if syncGroup's policy is SAGE_CONSTANT_SYNC, then this thread periodicaly sends sync(sendSync()) signal to its syncGroup.<BR>
    * Otherwise it sends sync signal(by calling manageUpdate()) only when syncClient asks ??<BR>
    * The period is defined in the syncGroup.interval
    */
   static void* mainLoopThread(void*);

   sageTimer timer;

public:
        /**
         *  Constructor. Called in sageDisplayManager::sageDisplayManager() and sail::init()
         */
   sageSyncBBServer(int syncLevel = 2);
   ~sageSyncBBServer();

   /**
    * creates syncServer socket, and starts syncServerThread
    */
   int init(int port);

   /**
    * create barrierServerSockFd
    */
   int initBarrier(int barrierPort);

   int totalRcvNum;
   int refreshInterval;
   int syncMasterPollingInterval;
   int startManagerThread(int totalRcvNum, int refreshInterval, int syncMasterPollingInterval);

   void killAllClients();  /**< closes all the open client sockets */
   int checkTimeOut();
};



















/**
 * class sageSyncServer. It is created when syncMaster == true
 */
class sageSyncServer {
protected:
   int serverSockFd; /**< syncServer's socket */
   int asapSyncGroupNum, maxSyncGroupID;
   int maxSlaveSockFd;
   bool syncEnd;

   sockaddr_in serverAddr;
   fd_set slaveFds;

   pthread_t   syncThreadID, syncServerThreadID;
   pthread_t   groupThreadID;

   std::vector<syncSlaveData> syncSlaves; /**< a list of sync slaves */

   /**
    * key - sdm id
        */
   std::map<int,syncSlaveData> syncSlavesMap;


   //syncGroup *syncGroupArray[MAX_SYNC_GROUP]; /**< an array of sync groups */
   std::vector<syncGroup *> syncGroupArray; /**< an array of sync groups */

   /**
    * waits syncClient's connection by calling accept() syscall<br>
    * if one connects, then it creates syncSlaveData object for it, and adds it to the vector syncSlaves
    */
   static void* syncServerThread(void*);

   /**
    * This thread is created whenever addSyncGroup() is called and under some condition.<BR>
    * 1. when the policy is SAGE_CONSTANT_SYNC<BR>
    * 2. 1 is not true and asapSyncGroupNum == 0<BR>
    * <BR>
    * if syncGroup's policy is SAGE_CONSTANT_SYNC, then this thread periodicaly sends sync(sendSync()) signal to its syncGroup.<BR>
    * Otherwise it sends sync signal(by calling manageUpdate()) only when syncClient asks ??<BR>
    * The period is defined in the syncGroup.interval
    */
   static void* managerThread(void*);
   static void* managerConstantThread(void*);

   /**
    * It seems that each syncGroup has corresponding managerThread.<BR>
    *
    * This function waits this syncServer object's syncSlaves' update from socket connection<BR>
    * then, extracts and identifies groupID from a slave's message<BR>
    * then, calls corresponding syncGroup's syncGroup::processUpdate()<BR>
    * If syncGroup's processUpdate() returns NORMAL_SYNC<BR>
    * then, syncGroup::checkInterval(), syncGroup::checkHold(), syndSync() could be called.
    *
    * respond with sync signal ( sendSync() ) when a syncClient sends sageSyncClient::sendSlaveUpdate().
    */
   int manageUpdate();

   /**
    * sends sync message (group's current frame: grp->curFrame) to the syncSlaves.<BR>
    * sync message includes grp->id, grp->curFrame, dataLen, cmd<BR>
    * if grp->syncGroup::dequeSyncMsg() returns data then sends the data as well
    */
   int sendSync(syncGroup *grp, int cmd = NORMAL_SYNC);
   sageTimer timer;

        syncGroup* findSyncGroup(int id, int& index);

public:
        /**
         *  Constructor. Called in sageDisplayManager::sageDisplayManager() and sail::init()
         */
   sageSyncServer();
   ~sageSyncServer();

   /**
    * creates syncServer socket, and starts syncServerThread
    */
   int init(int port);

   /**
    * Adds syncGroup into syncGroupArray and creates corresponding managerThread() if group's policy is SAGE_CONSTANT_SYNC<BR>
    * or policy is not SAGE_CONSTANT_SYNC and asapSyncGroupNum == 0<BR>
    * triggered by sageDisplayManager::initStreams() and sail
    */
   int addSyncGroup(syncGroup *grp);

   int removeSyncGroup(int id);

   void killAllClients();  /**< closes all the open client sockets */
   int checkTimeOut();
}; // End of sageSyncServer

class sageCircBufSingle;

/**
 * class syncMsgStruct
 */
class syncMsgStruct {
public:
   int frameID;
   char *data;

   syncMsgStruct() : frameID(0), data(NULL) {}
   syncMsgStruct(int size) : frameID(0) { data = new char[size]; }
   ~syncMsgStruct() { if (data) delete [] data; }
};

/**
 * class sageSyncClient
 */
class sageSyncClient {
private:
        int syncLevel;

   int clientSockFd;

   int barrierClientSockFd;
   int refreshBarrierDeltaT;

   int maxGroupID;
   //sageCircBufSingle *syncMsgBuf[MAX_SYNC_GROUP];
   std::vector<sageCircBufSingle *> syncMsgBuf;
   bool syncEnd;
   pthread_t syncThreadID;

   /**
    * keeps trying to receive message from syncServer by calling readSyncMsg()<BR>
    * started by addSyncGroup()
        */
   static void* syncClientThread(void *args);

   /**
    * receive messages from sageSyncServer::sendSync()<BR>
    * receives message(syncgroupID, frameNum, dataLen) from syncServer by calling sage::recv() and reads sync data.<BR>
    * This function is continuously called by syncClientThread().
        */
   int readSyncMsg();

        sageCircBufSingle* findSyncGroup(int id, int& index);

public:
   /**
        * creates socket(clientSockFd) and sets TCP_NODELAY, SO_OOBINLINE socket option
        */
   sageSyncClient(int syncLevel = -1);

   /**
    * shuts down the socket, and calls pthread_join
        */
   ~sageSyncClient();

   /**
    * connects to a syncServer
        *
        * @param syncServerIP a char *
        * @param port an integer
        * @return -1 on error, 0 otherwise
        */
   int connectToServer(const char *serverIP, int port, int SDMnum=-1);

   int connectToBarrierServer(const char *serverIP, int port, int SDMnum=-1);


   /**
    * creates sageCircBufSingle object, assigns it into syncMsgBuf array<BR>
    * starts syncClientThread() if no other syncgroup was added.<BR>
    * So, syncClientThread() is started when the first syncgroup is added.
        *
        * @param id an integer, sync group id
        * @return -1 on error, 0 otherwise
        */
   int addSyncGroup(int id);


   /**
    * removes element(id) from syncMsgBuf array
        *
        * @param id an integer
        * @return -1 on error, 0 otherwise
        */
   int removeSyncGroup(int id);


   /**
    * sends update msg to the syncServer.<BR>
    * sageSyncServer::manageUpdate() will respond to this.<BR>
    *
    * Triggered by below events and message<BR>
    * EVENT_READ_BLOCK (pixelDownloader::fetchSageBlocks()) <BR>
    * EVENT_SYNC_MESSAGE (sageDisplayManager::processSync() -> pixelDownloader::fetchSageBlocks()) <BR>
    * RCV_UPDATE_DISPLAY (sageDisplayManager::updateDisplay() -> pixelDownloader::fetchSageBlocks()) <BR>
    * <BR>
    * in the sageAudioManager::initStream()<BR>
    * in the sageAudioCircBuf::updateReadIndex() <- (sageAudioReceiver::readData(), sageAudioStreamer::streamLoop())<BR>
    *
        *
        * @param frame an integer frame number
        * @param id an integer, default 0
        * @param rcvNum an integer, default 0
        * @param type an integer, default SAGE_UPDATE_FOLLOW
        * @return -1 on error, 0 otherwise
        */
   int sendSlaveUpdate(int frame, int id = 0, int rcvNum = 0, int type = SAGE_UPDATE_FOLLOW, int nodeID=-1);

   /**
    * when a PDL received new frame (END_FRAME flag) it reports to the syncMaster before doing swapMontage()
        */
   int sendSlaveUpdateToBBS(int frame, int id = 0, int rcvNum = 0, int SDMnum = -1, int delayCompenLatency=0);

   /** THE FINAL CASE , sageSync_theFinal.cpp */
   int sendRefreshBarrier(int nodeID);
   int recvRefreshBarrier(bool nonblock=false); // block or nonblock
   /** end THE FINAL CASE */

   /**
    * receives sync message without additional data
        *
        * @param msg a char *
        * @return -1 on error, 0 otherwise
        */
   int waitForSync(char* msg, int len = -1);

   /**
    * It uses MSG_PEEK to find out the length of sync message
    *
    * @param void
    * @return -1 on error, message size otherwise
    */
   int waitForSyncPeek();

   /**
    * receives sync message with group ID and additional data
        *
        * @param id an integer
        * @return syncMsg, NULL on error
        */
   syncMsgStruct* waitForSync(int id);

   /**
    * receives sync message with additional data
        *
        * @param data a char *
        * @return -1 on error, 0 otherwise
        */
   int waitForSyncData(char* &data);


   /**
    * ch
        */
   int checkSync(char* &msg);
}; // End of tvSyncClient


/**
 * For synchronization among master and slave threads
 */
class sageThreadSync {
private:
   pthread_mutex_t   *slaveLock;
   pthread_mutex_t   *masterLock;
   int slaveNum;

public:
   sageThreadSync(int sNum = 1);
   ~sageThreadSync();

   int synchronize(int rank = 0);
};

#endif