osg::PointMCastConnection Class Reference
[Network]

#include <OSGPointMCastConnection.h>

Inheritance diagram for osg::PointMCastConnection:

List of all members.

private helpers
*bool	recvNextDgram (Dgram *dgram)
void	combineAck (Dgram *dgram, SocketAddress from)
bool	recvQueue (void)
void	initialize (void)
static void	recvQueueThread (void *arg)
Public Member Functions
Constructors
*	PointMCastConnection (void)
virtual	~PointMCastConnection (void)
type info
*virtual const ConnectionType *	getType (void)
connection
*virtual Channel	connectGroup (const std::string &address, Time timeout=-1)
virtual void	disconnect (void)
virtual Channel	acceptGroup (Time timeout=-1)
synchronisation
*virtual bool	wait (Time timeout) throw (ReadError)
channel handling
*virtual Channel	selectChannel (Time timeout=-1) throw (ReadError)
Static Public Member Functions
create
*static PointConnection *	create (void)
	create conneciton
Protected Types
typedef std::vector< MemoryBlock >	BuffersT
typedef std::list< MemoryHandle >	FreeMemT
Protected Member Functions
IO Implementation
*virtual void	read (MemoryHandle mem, UInt32 size)
virtual void	readBuffer (void)
Protected Attributes
members
*DgramSocket	_mcastSocket
DgramSocket	_responseSocket
BaseThread *	_recvQueueThread
bool	_recvQueueThreadRunning
bool	_recvQueueThreadStop
UInt16	_seqNumber
SocketAddress	_mcastAddress
DgramQueue	_queue
DgramQueue	_free
Lock *	_lock
SocketAddress	_sender
SocketAddress	_ackDestination
Dgram *	_lastDgram
UInt32	_lastDgramPos
bool	_initialized
std::map< SocketAddress, UInt16 >	_combineAck
UInt16	_maxAck
Private Types
typedef PointSockConnection	Inherited
Private Member Functions
	PointMCastConnection (const PointMCastConnection &source)
PointMCastConnection &	operator= (const PointMCastConnection &source)
Static Private Attributes
static type
*static ConnectionType	_type
Classes
struct	SocketBufferHeader

---

Detailed Description

Definition at line 62 of file OSGPointMCastConnection.h.

---

Member Typedef Documentation

typedef PointSockConnection osg::PointMCastConnection::Inherited [private]

Reimplemented from osg::PointSockConnection. Definition at line 180 of file OSGPointMCastConnection.h.

typedef std::vector<MemoryBlock> osg::BinaryDataHandler::BuffersT [protected, inherited]

Definition at line 213 of file OSGBinaryDataHandler.h.

typedef std::list<MemoryHandle> osg::BinaryDataHandler::FreeMemT [protected, inherited]

Definition at line 214 of file OSGBinaryDataHandler.h.

---

Constructor & Destructor Documentation

PointMCastConnection::PointMCastConnection (  void   )

Constructor Definition at line 67 of file OSGPointMCastConnection.cpp. References osg::PointSockConnection::_acceptSocket, _free, _lock, osg::Lock::get(), osg::StreamSocket::open(), OSG_DGRAM_QUEUE_LEN, osg::DgramQueue::put(), and osg::Socket::setReusePort(). Referenced by create(). : Inherited(), _lastDgram(NULL), _initialized(false) { char lockName[256]; sprintf(lockName,"PointMCastConnection%p",this); // create locks _lock = Lock::get(lockName); // fill dgramqueue for(UInt32 dI = 0 ; dI < OSG_DGRAM_QUEUE_LEN ; ++dI) _free.put(new Dgram()); _acceptSocket.open(); _acceptSocket.setReusePort(true); /* _socketWriteBuffer.resize(131071); // reserve first bytes for buffer size writeBufAdd(&_socketWriteBuffer[sizeof(SocketBufferHeader)], _socketWriteBuffer.size()-sizeof(SocketBufferHeader)); */ }

PointMCastConnection::~PointMCastConnection (  void   )  [virtual]

Destructor Definition at line 95 of file OSGPointMCastConnection.cpp. References osg::PointSockConnection::_acceptSocket, _free, _lock, _mcastSocket, _queue, _recvQueueThread, _recvQueueThreadStop, osg::Lock::aquire(), osg::StreamSocket::close(), osg::DgramSocket::close(), osg::DgramQueue::empty(), osg::DgramQueue::get(), osg::BaseThread::join(), and osg::Lock::release(). { // indicate thread stop _recvQueueThreadStop = true; // wait for stop BaseThread::join(_recvQueueThread); // close socket _mcastSocket.close(); // free queues _lock->aquire(); while(!_free.empty()) delete _free.get(_lock); while(!_queue.empty()) delete _queue.get(_lock); _lock->release(); // close socket _acceptSocket.close(); }

osg::PointMCastConnection::PointMCastConnection (  const PointMCastConnection &  source  )  [private]

---

Member Function Documentation

const ConnectionType * PointMCastConnection::getType (  void   )  [virtual]

get connection type Reimplemented from osg::PointSockConnection. Definition at line 116 of file OSGPointMCastConnection.cpp. References _type. { return &_type; }

Connection::Channel PointMCastConnection::connectGroup (  const std::string &  address, Time  timeout = -1 )  [virtual]

connect to the given group. If timeout is reached, -1 is returned Reimplemented from osg::PointSockConnection. Definition at line 127 of file OSGPointMCastConnection.cpp. References osg::Connection::Channel, and osg::PointSockConnection::connectGroup(). { Channel channel = Inherited::connectGroup(address,timeout); return channel; }

void PointMCastConnection::disconnect (  void   )  [virtual]

disconnect the given channel Reimplemented from osg::PointSockConnection. Definition at line 137 of file OSGPointMCastConnection.cpp. References osg::PointSockConnection::_socket, and osg::StreamSocket::close(). { _socket.close(); }

Connection::Channel PointMCastConnection::acceptGroup (  Time  timeout = -1  )  [virtual]

accept an icomming grop connection. If timeout is reached, -1 is returned. If timeout is -1 then wait without timeout Reimplemented from osg::PointSockConnection. Definition at line 145 of file OSGPointMCastConnection.cpp. References osg::PointSockConnection::acceptGroup(), and osg::Connection::Channel. { Channel channel = Inherited::acceptGroup(timeout); return channel; }

bool PointMCastConnection::wait (  Time  timeout  )  throw (ReadError) [virtual]

wait for signal Reimplemented from osg::PointSockConnection. Definition at line 202 of file OSGPointMCastConnection.cpp. References FFATAL. { UInt32 tag; if(_pointToPoint) return Inherited::wait(timeout); try { if(selectChannel(timeout) < 0) return false; getValue(tag); if(tag != 314156) { FFATAL(("Stream out of sync in PointMCastConnection\n")); throw ReadError("Stream out of sync"); } } catch(SocketError &e) { throw ReadError(e.what()); } return true; }

Connection::Channel PointMCastConnection::selectChannel (  Time  timeout = -1  )  throw (ReadError) [virtual]

select the next channel for reading. If timeout is not -1 then -1 is returned if timeout is reached Reimplemented from osg::PointSockConnection. Definition at line 157 of file OSGPointMCastConnection.cpp. { if(_pointToPoint) return Inherited::selectChannel(timeout); try { if(!_initialized) initialize(); // todo if(isReadBufferEmpty() && !_lastDgram && _queue.empty()) { if(timeout == -1) { // wait for a dgram _lock->aquire(); _queue.wait(_lock); _lock->release(); return 0; } if(timeout == 0) return -1; while(_queue.empty() && timeout > 0) { _mcastSocket.waitReadable(.1); timeout-=.1; } if(_queue.empty()) return -1; } } catch(SocketException &e) { throw ReadError(e.what()); } return 0; }

PointConnection * PointMCastConnection::create (  void   )  [static]

Reimplemented from osg::PointSockConnection. Definition at line 231 of file OSGPointMCastConnection.cpp. References PointMCastConnection(). { return new PointMCastConnection(); }

void PointMCastConnection::read (  MemoryHandle  mem, UInt32  size )  [protected, virtual]

Read data into given memory Read data form the current read socket. The read socket is that socket, that was selectet in selectChannel. Reimplemented from osg::PointSockConnection. Definition at line 246 of file OSGPointMCastConnection.cpp. References _free, _initialized, _lastDgram, _lastDgramPos, _lock, osg::PointConnection::_pointToPoint, _queue, osg::Lock::aquire(), osg::DgramQueue::get(), osg::Dgram::getData(), osg::Dgram::getSize(), initialize(), osg::osgMin(), osg::DgramQueue::put(), osg::PointSockConnection::read(), and osg::Lock::release(). { if(_pointToPoint) { Inherited::read(mem,size); return; } Dgram *dgram = NULL; char *buffer = (char*)mem; UInt32 len; UInt32 dgramPos; if(!_initialized) initialize(); while(size) { if(_lastDgram) { dgramPos = _lastDgramPos; dgram = _lastDgram; } else { // get next dgram _lock->aquire(); dgram = _queue.get(_lock); _lock->release(); dgramPos = 0; if(dgram->getSize() == 0) throw ReadError("Channel closed\n"); } // copy to buffer len = osgMin(size,dgram->getSize()-dgramPos); memcpy(buffer,dgram->getData()+dgramPos,len); buffer += len; size -= len; dgramPos += len; if(dgramPos == dgram->getSize()) { // put to free queue _lock->aquire(); _free.put(dgram); _lock->release(); _lastDgram = NULL; } else { _lastDgram = dgram; _lastDgramPos = dgramPos; } } }

void PointMCastConnection::readBuffer (  void   )  [protected, virtual]

Read next data block The stream connection uses only BinaryDataHandler buffer. If more then one buffer is present, then this methode must be changed! Reimplemented from osg::PointSockConnection. Definition at line 307 of file OSGPointMCastConnection.cpp. References _free, _initialized, _lastDgram, _lastDgramPos, _lock, osg::PointConnection::_pointToPoint, _queue, osg::Lock::aquire(), osg::DgramQueue::empty(), osg::DgramQueue::get(), osg::Dgram::getData(), osg::Dgram::getSize(), initialize(), osg::osgMin(), osg::DgramQueue::put(), osg::BinaryDataHandler::readBufBegin(), osg::PointSockConnection::readBuffer(), and osg::Lock::release(). { if(_pointToPoint) { Inherited::readBuffer(); return; } static int sumSize=0; Dgram *dgram = NULL; UInt32 size = readBufBegin()->getSize(); MemoryHandle buffer = readBufBegin()->getMem(); UInt32 len; UInt32 dgramPos; if(!_initialized) initialize(); do { if(_lastDgram) { dgramPos = _lastDgramPos; dgram = _lastDgram; } else { // get next dgram _lock->aquire(); dgram = _queue.get(_lock); _lock->release(); dgramPos = 0; if(dgram->getSize() == 0) throw ReadError("Channel closed"); } // copy to buffer len = osgMin(size,dgram->getSize()-dgramPos); memcpy(buffer,dgram->getData()+dgramPos,len); buffer += len; size -= len; dgramPos += len; if(dgramPos == dgram->getSize()) { // put to free queue _lock->aquire(); _free.put(dgram); _lock->release(); _lastDgram = NULL; } else { _lastDgram = dgram; _lastDgramPos = dgramPos; } } while(size && !_queue.empty()); // set data size readBufBegin()->setDataSize(readBufBegin()->getSize()-size); sumSize += readBufBegin()->getDataSize(); }

bool PointMCastConnection::recvNextDgram (  Dgram *  dgram  )  [private]

read next dgram from mcast or private socket Definition at line 373 of file OSGPointMCastConnection.cpp. References _ackDestination, _combineAck, _maxAck, _mcastSocket, _responseSocket, _sender, combineAck(), osg::Dgram::getBuffer(), osg::Dgram::getBufferCapacity(), osg::Dgram::getBufferSize(), osg::Dgram::getId(), osg::SocketSelection::isSetRead(), osg::DgramSocket::recvFrom(), osg::SocketSelection::select(), osg::DgramSocket::sendTo(), osg::Dgram::setBufferSize(), osg::Dgram::setId(), osg::SocketSelection::setRead(), osg::Dgram::setResponseAck(), and osg::Dgram::setResponseSize(). Referenced by recvQueue(). { SocketSelection selection; SocketAddress from; UInt32 length; selection.setRead(_mcastSocket); selection.setRead(_responseSocket); if(!selection.select(0.5)) return false; if(selection.isSetRead(_responseSocket)) { length = _responseSocket.recvFrom(dgram->getBuffer(), dgram->getBufferCapacity(), from); dgram->setBufferSize(length); #if 0 // ???? // from sender if(from == _sender && !_combineAck.empty()) { exit(0); if(_maxAck == dgram->getId()) { // do we have all acks ? dgram->setId(_maxAck); dgram->setResponseSize(); dgram->setResponseAck(true); #ifdef TEST_LOST_DGRAM_RATE if(drand48()>TEST_LOST_DGRAM_RATE) #endif _responseSocket.sendTo( dgram->getBuffer(), dgram->getBufferSize(), _ackDestination); return false; } else { return true; } } #endif combineAck(dgram,from); } if(selection.isSetRead(_mcastSocket)) { length = _mcastSocket.recvFrom(dgram->getBuffer(), dgram->getBufferCapacity(), from); dgram->setBufferSize(length); // ignore packages from wrong destination if(from != _sender) return false; else return true; } else { return false; } }

void PointMCastConnection::combineAck (  Dgram *  dgram, SocketAddress  from )  [private]

combine several acks to 1 ack stream Definition at line 440 of file OSGPointMCastConnection.cpp. References _ackDestination, _combineAck, _maxAck, _responseSocket, _seqNumber, FFATAL, osg::Dgram::getBuffer(), osg::Dgram::getBufferSize(), osg::Dgram::getId(), osg::Dgram::less(), osg::DgramSocket::sendTo(), osg::Dgram::setId(), osg::Dgram::setResponseAck(), and osg::Dgram::setResponseSize(). Referenced by recvNextDgram(), and recvQueue(). { UInt16 maxAck; if(dgram) { // do we expect acks from different source if(_combineAck.count(from)==0) { FFATAL(("no ack from other expected\n")); return; } // ack retransmission if( Dgram::less(dgram->getId(),_combineAck[from] ) ) { // printf("Ack restranmisson\n"); return; } _combineAck[from] = dgram->getId(); } maxAck = _seqNumber-1; for(std::map<SocketAddress,UInt16>::iterator aI = _combineAck.begin() ; aI != _combineAck.end() ; ++aI) { if( Dgram::less(aI->second,maxAck) ) maxAck = aI->second; } // when _max ack is now greate if( Dgram::less(_maxAck,maxAck)) { Dgram response; _maxAck = maxAck; response.setResponseSize(); response.setId(_maxAck); response.setResponseAck(true); _responseSocket.sendTo( response.getBuffer(), response.getBufferSize(), _ackDestination); } }

void PointMCastConnection::recvQueueThread (  void *  arg  )  [static, private]

read next dgram from mcast or private socket Definition at line 633 of file OSGPointMCastConnection.cpp. References _recvQueueThreadRunning, recvQueue(), and SFATAL. Referenced by initialize(). { PointMCastConnection *the=(PointMCastConnection*)arg; try { the->recvQueue(); } catch(SocketException &e) { SFATAL << "Error in dgram reader thread:" << e.what() << std::endl; } the->_recvQueueThreadRunning = false; }

bool PointMCastConnection::recvQueue (  void   )  [private]

recv queue Definition at line 489 of file OSGPointMCastConnection.cpp. References _ackDestination, _combineAck, _free, _lock, _maxAck, _mcastSocket, _queue, _recvQueueThreadStop, _responseSocket, _sender, _seqNumber, osg::PointSockConnection::_socket, osg::Lock::aquire(), combineAck(), FLOG, osg::DgramQueue::get(), osg::Dgram::getBuffer(), osg::Dgram::getBufferSize(), osg::Dgram::getId(), osg::Dgram::getResponseAck(), osg::Dgram::getSize(), osg::getSystemTime(), osg::Dgram::less(), osg::DgramQueue::put(), osg::Socket::recv(), recvNextDgram(), osg::Lock::release(), osg::DgramSocket::sendTo(), osg::Dgram::setId(), osg::Dgram::setResponseAck(), osg::Dgram::setResponseSize(), osg::Dgram::setSize(), and osg::Socket::waitReadable(). Referenced by recvQueueThread(). { SocketAddress from; Dgram *dgram; Dgram response; UInt32 readCount=0; UInt32 length; bool missing=false; Time ignoreT=getSystemTime(); UInt16 id; #ifdef TEST_LOST_DGRAM_RATE srand48((long int)(10000*getSystemTime())); #endif for(;;) { // get free dgram _lock->aquire(); dgram =_free.get(_lock); _lock->release(); do { // ignore for a while if(missing) ignoreT = getSystemTime(); do { while(!recvNextDgram(dgram)) { if(_recvQueueThreadStop) return true; try { while(_socket.waitReadable(0)) { char buffer; if(_socket.recv(&buffer,1) <= 0) { // put EOT to the queue dgram->setSize(0); _lock->aquire(); _queue.put(dgram); _lock->release(); FLOG(("Connection lost\n")); return false; } } } catch(SocketException &e) { // put EOT to the queue dgram->setSize(0); _lock->aquire(); _queue.put(dgram); _lock->release(); FLOG(("Connection lost\n")); return false; } } id = dgram->getId(); } while( missing && id != _seqNumber && (getSystemTime() - ignoreT) < 0.01); missing = false; response.setId(id); // ack request ? if(dgram->getSize() == 0) { if( !Dgram::less(id,_seqNumber ) ) { missing = true; response.setId(_seqNumber); } // printf("ack request %d %d\n",id,missing); } else { // printf("%d got %d\n",id,_seqNumber); if( dgram->getId() == _seqNumber) { // got expected dgram. // put to queue _lock->aquire(); _queue.put(dgram); _lock->release(); } else { // ignore if unneccesary retransmission if( Dgram::less(id,_seqNumber ) ) { continue; } else { missing = true; response.setId(_seqNumber); } } } // printf("Responde %d\n",response.getId()); // prepare response response.setResponseAck(!missing); response.setResponseSize(); // send response if nak or no data in the queue if(!response.getResponseAck() || !_mcastSocket.waitReadable(0)) { #ifdef TEST_LOST_DGRAM_RATE if(drand48()>TEST_LOST_DGRAM_RATE) #endif if(response.getResponseAck()) { // send response if no ack combination // or ack request if(_combineAck.empty() || id == _maxAck) _responseSocket.sendTo(response.getBuffer(), response.getBufferSize(), _ackDestination); else combineAck(NULL,_sender); } else { // send nak to sender _responseSocket.sendTo(response.getBuffer(), response.getBufferSize(), _sender); } } } while(id != _seqNumber || missing); _seqNumber++; } // return true; }

void PointMCastConnection::initialize (  void   )  [private]

initialize connection Definition at line 649 of file OSGPointMCastConnection.cpp. References _ackDestination, _combineAck, _initialized, _maxAck, _mcastSocket, _recvQueueThread, _recvQueueThreadRunning, _recvQueueThreadStop, _responseSocket, _sender, _seqNumber, osg::PointSockConnection::_socket, osg::SocketAddress::ANY, osg::Socket::bind(), osg::Connection::Channel, osg::BinaryMessage::clear(), osg::BaseThread::get(), osg::Socket::getAddress(), osg::Connection::getInterface(), osg::SocketAddress::getPort(), osg::BinaryMessage::getString(), osg::BinaryMessage::getUInt32(), osg::DgramSocket::join(), osg::DgramSocket::open(), osg::BinaryMessage::putString(), osg::BinaryMessage::putUInt32(), osg::Socket::recv(), recvQueueThread(), osg::BaseThread::runFunction(), osg::Socket::send(), osg::DgramSocket::setMCastInterface(), osg::Socket::setReadBufferSize(), and osg::Socket::setReusePort(). Referenced by read(), and readBuffer(). { std::string group; Channel channel; BinaryMessage message; char hostname[256]; std::string fromHost; UInt32 fromPort; UInt32 combineCount; std::string host; UInt32 port; char threadName[256]; sprintf(threadName,"PointMCastConnection%p",this); // get info about the group _socket.recv(message); // group and port group = message.getString(); port = message.getUInt32(); // get seq number _seqNumber = message.getUInt32(); _maxAck = _seqNumber - 1; // server port fromHost = message.getString(); fromPort = message.getUInt32(); _sender = SocketAddress(fromHost.c_str(),fromPort); // prepare socket to receive mcast packages _mcastSocket.open(); _mcastSocket.setReusePort(true); _mcastSocket.setReadBufferSize(524288); _mcastSocket.bind(SocketAddress(SocketAddress::ANY,port)); try { _mcastSocket.join(SocketAddress(group.c_str())); } catch(...) {} // set multicast interface if(!getInterface().empty()) { _mcastSocket.setMCastInterface(SocketAddress(getInterface().c_str())); } _responseSocket.open(); _responseSocket.bind(SocketAddress(SocketAddress::ANY,0)); // tell the group from wich port requests are comming hostname[255] = '\0'; gethostname(hostname,255); message.clear(); message.putString(hostname); message.putUInt32(_responseSocket.getAddress().getPort()); _socket.send(message); // get ack destination info _socket.recv(message); message.getUInt32(combineCount); // printf("%d\n",combineCount); while(combineCount--) { host=message.getString(); port=message.getUInt32(); _combineAck[SocketAddress(host.c_str(),port)]=_seqNumber-1; } host=message.getString(); port=message.getUInt32(); _ackDestination = SocketAddress(host.c_str(),port); // start reader thread _recvQueueThread=BaseThread::get(threadName); _recvQueueThreadRunning = true; _recvQueueThreadStop = false; _recvQueueThread->runFunction( recvQueueThread, (void *) (this) ); _initialized = true; // printf("init end\n"); }

PointMCastConnection& osg::PointMCastConnection::operator= (  const PointMCastConnection &  source  )  [private]

Connection::Channel PointSockConnection::connectPoint (  const std::string &  address,