#include <vsnet_sockettcp.h>

Inheritance diagram for VsnetTCPSocket:

Classes
struct	Blob

struct	Header

class	SqQueues

Public Types
typedef boost::shared_ptr< Packet >	PacketPtr

Public Member Functions
	VsnetTCPSocket (int sock, const AddressIP &remote_ip, SocketSet &set)

virtual	~VsnetTCPSocket ()

virtual bool	isTcp () const

virtual int	optPayloadSize () const

virtual int	queueLen (int pri)

virtual int	sendbuf (Packet packet, const AddressIP to, int pcktflags)

virtual int	recvbuf (Packet p, AddressIP ipadr)

virtual void	dump (std::ostream &ostr) const

virtual bool	isActive ()

virtual bool	lower_selected (int datalen=-1)

virtual bool	need_test_writable ()

virtual int	get_write_fd () const

virtual int	lower_sendbuf ()

virtual void	lower_clean_sendbuf ()

Public Member Functions inherited from VsnetSocket
	VsnetSocket (int sock, const AddressIP &remote_ip, const char *socktype, SocketSet &set)

virtual	~VsnetSocket ()

bool	eq (const VsnetSocket &r) const

bool	lt (const VsnetSocket &r) const

bool	sameAddress (const VsnetSocket &r) const

const AddressIP &	getRemoteAddress () const

virtual bool	setRemoteAddress (const AddressIP &)

const AddressIP &	getLocalAddress () const

bool	setLocalAddress (const AddressIP &lip)

Public Member Functions inherited from VsnetSocketBase
	VsnetSocketBase (int fd, const char *socktype, SocketSet &set)

virtual	~VsnetSocketBase ()

bool	valid () const

int	get_fd () const

int	close_fd ()

const char *	get_socktype () const

bool	set_block ()

bool	set_nonblock ()

bool	get_nonblock () const

void	disconnect (const char *s)

virtual bool	write_on_negative ()

virtual bool	isReadyToSend (fd_set *)

void	setSet (SocketSet *set)

Protected Types
typedef std::queue< PacketPtr >	SqQueue

typedef std::queue< PacketMem >	SqQueueP

Protected Member Functions
virtual void	child_disconnect (const char *s)

void	inc_pending ()

void	dec_pending ()

void	inner_complete_a_packet (Blob *b)

void	private_nothread_conditional_write ()

Protected Attributes
Blob *	_incomplete_packet

std::queue< PacketPtr >	_cpq
	cpq = completed packet queue More...

VSMutex	_cpq_mx

int	_incomplete_header

Header	_header

bool	_connection_closed

SqQueues	_sq

SqQueueP	_sq_current

size_t	_sq_off

VSMutex	_sq_mx

int	_sq_fd

Protected Attributes inherited from VsnetSocket
AddressIP	_remote_ip

AddressIP	_local_ip

Protected Attributes inherited from VsnetSocketBase
int	_fd

SocketSet *	_set

Detailed Description

Definition at line 26 of file vsnet_sockettcp.h.

Member Typedef Documentation

typedef boost::shared_ptr< Packet > VsnetTCPSocket::PacketPtr

Definition at line 30 of file vsnet_sockettcp.h.

typedef std::queue< PacketPtr > VsnetTCPSocket::SqQueue

protected

TCP may refuse to send all bytes at once, hence we have to queue packets for sending. In addition, we use the same thread for sending and receiving, thus it would be advantageous to use asynchronous sending. For both, we need a send queue. _sq is the a of priority queues. _sq_current is the queue (with up to two entries) for sending the current packet, which consists of the packet length and the packet payload. _sq.off is the number of bytes that have already been sent from PacketMem _sq_current.front(). _sq_mx protects the queues.

Definition at line 138 of file vsnet_sockettcp.h.

typedef std::queue< PacketMem > VsnetTCPSocket::SqQueueP

protected

Definition at line 139 of file vsnet_sockettcp.h.

Constructor & Destructor Documentation

VsnetTCPSocket::VsnetTCPSocket	(	int	sock,
		const AddressIP &	remote_ip,
		SocketSet &	set
	)

Definition at line 63 of file vsnet_sockettcp.cpp.

References _sq_fd, COUT, and VsnetTCPSocket().

Referenced by VsnetTCPSocket().

                                                                                       :
     VsnetSocket( sock, remote_ip, "VsnetTCPSocket", sets )
     , _incomplete_packet( 0 )
     , _incomplete_header( 0 )
     , _connection_closed( false )
     , _sq_off( 0 )
     , _mtu_size_estimation( 1024 )
 {
     COUT<<"enter "<<__PRETTY_FUNCTION__<<endl;
     _sq_fd = sock;
 }

VsnetTCPSocket::~VsnetTCPSocket ( )

virtual

Definition at line 75 of file vsnet_sockettcp.cpp.

References _connection_closed, _incomplete_packet, VsnetSocketBase::close_fd(), COUT, VsnetSocketBase::get_fd(), and VsnetSocketBase::valid().

 {
     COUT<<"enter "<<__PRETTY_FUNCTION__<<endl;
     if (_incomplete_packet)
         delete _incomplete_packet;
     if (valid() && !_connection_closed && get_fd() >= 0)
         close_fd();
 }

Member Function Documentation

void VsnetTCPSocket::child_disconnect ( const char * s)

protectedvirtual

Reimplemented from VsnetSocketBase.

Definition at line 295 of file vsnet_sockettcp.cpp.

References VsnetSocketBase::close_fd(), COUT, and VsnetSocketBase::get_fd().

 {
     if (get_fd() > 0) {
         if (close_fd() < 0)
             COUT<<s<<" :\tWarning: disconnect error: "<<strerror( errno )<<endl;
         else
             COUT<<s<<" :\tWarning: disconnected"<<endl;
     } else {
         COUT<<s<<" :\tWarning: disconnect null socket: "<<strerror( errno )<<endl;
     }
 }

void VsnetTCPSocket::dec_pending ( )

protected

void VsnetTCPSocket::dump ( std::ostream & ostr) const

virtual

Implements VsnetSocket.

Definition at line 307 of file vsnet_sockettcp.cpp.

References VsnetSocket::_remote_ip, and VsnetSocketBase::get_fd().

 {
     ostr<<"( s="<<get_fd()<<" TCP r="<<_remote_ip<<" )";
 }

int VsnetTCPSocket::get_write_fd ( ) const

virtual

Reimplemented from VsnetSocketBase.

Definition at line 146 of file vsnet_sockettcp.cpp.

References _sq_fd.

 {
     return _sq_fd;
 }

void VsnetTCPSocket::inc_pending ( )

protected

The number of pending packets reported to the set.

void VsnetTCPSocket::inner_complete_a_packet ( Blob * b)

protected

Definition at line 457 of file vsnet_sockettcp.cpp.

References _cpq, _cpq_mx, VsnetSocketBase::_set, _sq_fd, SocketSet::add_pending(), b, VsnetTCPSocket::Blob::buf, VsnetTCPSocket::Blob::expected_len, VSMutex::lock(), VsnetTCPSocket::Blob::present_len, and VSMutex::unlock().

Referenced by lower_selected().

 {
     assert( b );
     assert( b->present_len == b->expected_len );
 
     PacketMem mem( b->buf, b->present_len, PacketMem::TakeOwnership );
     PacketPtr ptr = PacketPtr( new Packet( mem ) );
     //if (ptr->getCommand()!=CMD_POSUPDATE && ptr->getCommand()!=CMD_SNAPSHOT)
     //COUT << "Completely received a packet of type " << ptr->getCommand() << endl;
     b->buf = NULL;
     _cpq_mx.lock();
     _cpq.push( ptr );
     _cpq_mx.unlock();
     if (_set) _set->add_pending( _sq_fd );
     delete b;
 }

bool VsnetTCPSocket::isActive ( )

virtual

Implements VsnetSocketBase.

Definition at line 318 of file vsnet_sockettcp.cpp.

References _connection_closed, _cpq, _cpq_mx, VsnetSocketBase::_set, _sq_fd, VSMutex::lock(), private_nothread_conditional_write(), SocketSet::rem_pending(), and VSMutex::unlock().

 {
     /* True is the correct answer when the connection is closed:
      * the app must call recvbuf once after this to receive 0
      * and start close processing.
      * We could return packets from the queue first, but that may
      * trigger an answer packet from the application, and give
      * us trouble because of the closed socket.
      */
     if (_connection_closed)
         return true;
     bool retval = false;
 
     _cpq_mx.lock();
     if (_cpq.empty() == false)
         retval = true;
     else if (_set)
         _set->rem_pending( _sq_fd );
     _cpq_mx.unlock();
 
     private_nothread_conditional_write();
 
     return retval;
 }

virtual bool VsnetTCPSocket::isTcp ( ) const

inlinevirtual

Implements VsnetSocket.

Definition at line 36 of file vsnet_sockettcp.h.

     {
         return true;
     }

void VsnetTCPSocket::lower_clean_sendbuf ( )

virtual

Called when we noticed that the primary file descriptor is closed but data remains in the send queue.

Reimplemented from VsnetSocketBase.

Definition at line 251 of file vsnet_sockettcp.cpp.

References _sq, _sq_current, _sq_mx, _sq_off, COUT, VsnetTCPSocket::SqQueues::empty(), VSMutex::lock(), VsnetTCPSocket::SqQueues::pop(), and VSMutex::unlock().

 {
     _sq_mx.lock();
     while ( !_sq.empty() ) {
         _sq.pop();
 #ifdef VSNET_DEBUG
         COUT<<"forgetting a packet in _sq"<<endl;
 #endif
     }
     while ( !_sq_current.empty() ) {
         _sq_current.pop();
 #ifdef VSNET_DEBUG
         COUT<<"forgetting a segment in _sq_current"<<endl;
 #endif
         _sq_off = 0;
     }
     _sq_mx.unlock();
 }

bool VsnetTCPSocket::lower_selected ( int datalen = -1)

virtual

Implements VsnetSocketBase.

Definition at line 343 of file vsnet_sockettcp.cpp.

References _connection_closed, _header, _incomplete_header, _incomplete_packet, VsnetSocketBase::_set, _sq_fd, SocketSet::add_pending(), b, VsnetTCPSocket::Blob::base(), VsnetTCPSocket::Blob::buf, VsnetSocketBase::close_fd(), COUT, PacketMem::dump(), VsnetTCPSocket::Blob::expected_len, VsnetSocketBase::get_fd(), VsnetSocketBase::get_nonblock(), VsnetTCPSocket::Header::h_len(), i, inner_complete_a_packet(), VsnetTCPSocket::Blob::missing(), VsnetTCPSocket::Blob::present_len, VsnetOSS::recv(), vsnetEConnAborted(), vsnetEConnReset(), vsnetEWouldBlock(), vsnetGetLastError(), and vsnetLastError().

 {
     if (_connection_closed) {
         COUT<<"Connection already closed"<<endl;
         return true;         /* Pretty sure that recv will return 0.  */
     }
     bool endless    = true;
     bool successful = false;
     if (get_nonblock() == false && datalen == -1) endless = false;
     do {
         if ( (_incomplete_header > 0)
             || (_incomplete_header == 0 && _incomplete_packet == 0) ) {
             assert( _incomplete_packet == 0 );             //we expect a len, can not have data yet
             assert( (unsigned int) _incomplete_header < sizeof (Header) );              //len is coded in sizeof(Header) bytes
             unsigned int len = sizeof (Header)-_incomplete_header;
             char *b = (char*) &_header;
             int ret = VsnetOSS::recv( get_fd(), &b[_incomplete_header], len, 0 );
             if (ret <= 0) {
                 if ( ret == 0 || vsnetEConnAborted() || vsnetEConnReset() ) {
                     COUT<<"Connection closed in header"<<endl;
                     _connection_closed = true;
                     close_fd();
                     if (_set) _set->add_pending( _sq_fd );
                     return true;
                 } else if (vsnetEWouldBlock() == false) {
                     COUT<<"recv returned "<<ret
                     <<" errno "<<vsnetGetLastError()
                     <<" = "<<vsnetLastError()
                     <<endl;
                     perror( "receiving TCP packetlength bytes" );
                     _connection_closed = true;
                     close_fd();
                     return true;
                 } else {
                     //COUT << "Received EWOULDBLOCK." << (get_nonblock()?"true":"false") << endl;
                 }
                 return successful;
             }
             assert( (unsigned int)ret <= len );
             if (ret > 0) _incomplete_header += ret;
             if (datalen != -1) datalen -= ret;
             if ( _incomplete_header == sizeof (Header) ) {
                 _incomplete_header = 0;
                 len = _header.h_len();
                 _incomplete_packet = new Blob( len );
             }
         }
         if (_incomplete_packet != 0) {
             int len = _incomplete_packet->missing();
             if (datalen > 0)
                 len = datalen < len ? datalen : len;
             int ret = VsnetOSS::recv( get_fd(), _incomplete_packet->base(), len, 0 );
             assert( ret <= len );
             if (ret <= 0) {
                 if (ret == 0) {
                     COUT<<"Connection closed in data"<<endl;
                     _connection_closed = true;
                     close_fd();
                     if (_set) _set->add_pending( _sq_fd );
                     return true;
                 } else if ( vsnetEConnAborted() ) {
                     perror( "receiving TCP packet" );
                     if (get_fd() == -1) {
                         perror( "receiving dead TCP packet" );
                     } else {
                         COUT<<"Connection closed in error"<<endl;
                         _connection_closed = true;
                         close_fd();
                         if (_set) _set->add_pending( _sq_fd );
                     }
                     return true;
                 } else if ( vsnetEWouldBlock() ) {
                     static int i = 0;
                     if (i++%128 == 0)
                         COUT<<"Received EWOULDBLOCK in data."<<(get_nonblock() ? "true" : "false")<<endl;
                 }
                 return successful;
             } else {
                 if (datalen != -1) {
                     datalen -= ret;
                     if (datalen <= 0)
                         endless = false;
                 }
                 _incomplete_packet->present_len += ret;
                 if (ret == len) {
                     assert( _incomplete_packet->expected_len
                             == _incomplete_packet->present_len );
 #ifdef VSNET_DEBUG
                     //
                     //DEBUG block - remove soon
                     //
                     {
                         Blob     *b = _incomplete_packet;
                         COUT<<"received buffer with len "<<b->present_len<<": "<<endl;
                         PacketMem m( b->buf, b->present_len, PacketMem::LeaveOwnership );
 #if 0
                         m.dump( cout, 3 );
 #endif
                     }
 #endif
 
                     inner_complete_a_packet( _incomplete_packet );
                     _incomplete_packet = 0;
                     successful = true;
                     endless    = false;                     //If we're done, let's stop while we're ahead.
 
                     //either endless is false, or we exit with EWOULDBLOCK
                 }
             }
         }
     } while (endless);     //exit only for EWOULDBLOCK or closed socket
     return successful;
 }

int VsnetTCPSocket::lower_sendbuf ( )

virtual

Reimplemented from VsnetSocketBase.

Definition at line 164 of file vsnet_sockettcp.cpp.

References _sq, _sq_current, _sq_fd, _sq_mx, _sq_off, COUT, VsnetTCPSocket::SqQueues::empty(), fprintf, h, PacketMem::len(), VSMutex::lock(), VsnetTCPSocket::SqQueues::pop(), and VSMutex::unlock().

 {
     _sq_mx.lock();
     if ( _sq_current.empty() ) {
         if ( _sq.empty() ) {
 #ifdef VSNET_DEBUG
             COUT<<"both queues are empty"<<endl;
 #endif
             _sq_mx.unlock();
             return 0;
         } else {
             PacketPtr m = _sq.pop();
             int len     = m->getSendBufferLength();
             Header    h( len );
             _sq_current.push( PacketMem( &h, sizeof (Header) ) );
             _sq_current.push( PacketMem( m->getSendBuffer(), len ) );
             _sq_off = 0;
         }
     }
     PacketMem    packet( _sq_current.front() );
 
     int numsent;
     unsigned int len = packet.len();
     assert( len > _sq_off );
     len    -= _sq_off;
     const char  *buf = packet.getConstBuf();
     numsent = ::send( _sq_fd, &buf[_sq_off], len, 0 );
     if (numsent < 0) {
         switch (errno)
         {
 #if defined (_WIN32) && !defined (__CYGWIN__)
         case WSAECONNRESET:         //other side closed socket
         case WSAECONNABORTED:         //other side closed socket
         case WSAEWOULDBLOCK:
 #else
         case ECONNRESET:         //other side closed socket
         case ECONNABORTED:         //other side closed socket
         case EWOULDBLOCK:
 #endif
         case EINTR:
             _sq_mx.unlock();
             return 0;
 
         case EFAULT:
             _sq_mx.unlock();
             COUT<<"EFAULT"<<endl
                 <<"   *** An invalid user space address was specified for a parameter."<<endl
                 <<"   *** fd        : "<<_sq_fd<<endl
                 <<"   *** buf       : "<<std::hex<<(long) buf<<std::dec<<endl
                 <<"   *** sq offset : "<<_sq_off<<endl
                 <<"   *** packet len: "<<packet.len()<<endl;
             return 0;
 
         default:
             fprintf( stderr, "Failed sending TCP data of %d len to socket %d\n", len, _sq_fd );
             perror( "\tsending TCP data" );
             _sq_mx.unlock();
             return -1;
         }
     } else if (numsent == 0) {
         //other side closed socket - what to do now?
         _sq_mx.unlock();
         return numsent;
     } else if ( (unsigned int) numsent == len ) {
         _sq_off = 0;
         _sq_current.pop();
         _sq_mx.unlock();
 #ifdef VSNET_DEBUG
         //
         //DEBUG block - remove soon
         //
         {
             COUT<<"sent buffer with len "<<packet.len()<<": "<<endl;
 #if 0
             packet.dump( cout, 0 );
 #endif
         }
 #endif
         return numsent;
     } else {
         assert( (unsigned int) numsent < len );         //should be impossible
         _sq_off += numsent;
         _sq_mx.unlock();
         return numsent;
     }
 }

bool VsnetTCPSocket::need_test_writable ( )

virtual

Reimplemented from VsnetSocketBase.

Definition at line 131 of file vsnet_sockettcp.cpp.

References _sq, _sq_current, _sq_mx, COUT, e, VsnetTCPSocket::SqQueues::empty(), VSMutex::lock(), and VSMutex::unlock().

 {
     _sq_mx.lock();
     bool e = ( _sq.empty() && _sq_current.empty() );
 #ifdef VSNET_DEBUG
     if (!e)
         COUT<<"_sq "
             <<(_sq.empty() ? "empty" : "not empty")
             <<", _sq_current "
             <<(_sq_current.empty() ? "empty" : "not empty")<<endl;
 #endif
     _sq_mx.unlock();
     return !e;
 }

int VsnetTCPSocket::optPayloadSize ( ) const

virtual

Implements VsnetSocket.

Definition at line 151 of file vsnet_sockettcp.cpp.

 {
     return _mtu_size_estimation;
 }

void VsnetTCPSocket::private_nothread_conditional_write ( )

protected

Definition at line 123 of file vsnet_sockettcp.cpp.

References _connection_closed, VsnetSocketBase::_set, and SocketSet::waste_time().

Referenced by isActive().

 {
 #ifdef USE_NO_THREAD
     if (_connection_closed) return;
     if (_set) _set->waste_time( 0, 0 );       //waste zero time, but check sockets
 #endif
 }

int VsnetTCPSocket::queueLen ( int pri)

virtual

Implements VsnetSocket.

Definition at line 156 of file vsnet_sockettcp.cpp.

References _sq, _sq_mx, VsnetTCPSocket::SqQueues::getLength(), VSMutex::lock(), and VSMutex::unlock().

 {
     _sq_mx.lock();
     int retval = _sq.getLength( pri );
     _sq_mx.unlock();
     return retval;
 }

int VsnetTCPSocket::recvbuf	(	Packet *	p,
		AddressIP *	from
	)

virtual

This function copies or moves data into the given PacketMem variable. It is preferred over the other recvbuf function because it may reduce the number of copy operations by at least one.

Implements VsnetSocket.

Definition at line 270 of file vsnet_sockettcp.cpp.

References _connection_closed, _cpq, _cpq_mx, VsnetSocket::_remote_ip, VsnetSocketBase::_set, _sq_fd, VsnetSocketBase::close_fd(), Packet::copyfrom(), COUT, VSMutex::lock(), SocketSet::rem_pending(), and VSMutex::unlock().

 {
     _cpq_mx.lock();
     if (_cpq.empty() == false) {
         PacketPtr ptr = _cpq.front();
         _cpq.pop();
         _cpq_mx.unlock();
 
         AddressIP dummy;
         if (ipadr) *ipadr = _remote_ip;
         p->copyfrom( *ptr.get() );
         return ptr->getDataLength()+ptr->getHeaderLength();
     } else if (_connection_closed) {
         if (_set) _set->rem_pending( _sq_fd );
         _cpq_mx.unlock();
         close_fd();
         COUT<<__PRETTY_FUNCTION__<<" connection is closed"<<endl;
         return 0;
     } else {
         if (_set) _set->rem_pending( _sq_fd );
         _cpq_mx.unlock();
         return -1;
     }
 }

int VsnetTCPSocket::sendbuf	(	Packet *	packet,
		const AddressIP *	to,
		int	pcktflags
	)

virtual

Use a priority queue instead of a standard queue.

Implements VsnetSocket.

Definition at line 104 of file vsnet_sockettcp.cpp.

References VsnetSocketBase::_set, _sq, _sq_mx, e, VsnetTCPSocket::SqQueues::empty(), Packet::getSendBufferLength(), HIPRI, VSMutex::lock(), LOPRI, VsnetTCPSocket::SqQueues::push(), VSMutex::unlock(), and SocketSet::wakeup().

 {
     int idx = 1;
     if (pcktflags&LOPRI) idx = 0;
     if (pcktflags&HIPRI) idx = 2;
     /* Add a timestamp here -- drop packets when they get too old.
      */
 
     PacketPtr enq = PacketPtr( new Packet( *packet ) );
 
     _sq_mx.lock();
     bool e = _sq.empty();
 
     _sq.push( idx, enq );
     _sq_mx.unlock();
     if (e && _set) _set->wakeup();
     return packet->getSendBufferLength();
 }

Member Data Documentation

bool VsnetTCPSocket::_connection_closed

protected

Closed connections are noticed in isActive but evaluated by the application after recvbuf. So, we remember the situation here until the application notices it.

Definition at line 122 of file vsnet_sockettcp.h.

Referenced by isActive(), lower_selected(), private_nothread_conditional_write(), recvbuf(), and ~VsnetTCPSocket().

std::queue< PacketPtr > VsnetTCPSocket::_cpq

protected

cpq = completed packet queue

Definition at line 106 of file vsnet_sockettcp.h.

Referenced by inner_complete_a_packet(), isActive(), and recvbuf().

VSMutex VsnetTCPSocket::_cpq_mx

protected

Definition at line 107 of file vsnet_sockettcp.h.

Referenced by inner_complete_a_packet(), isActive(), and recvbuf().

Header VsnetTCPSocket::_header

protected

Definition at line 116 of file vsnet_sockettcp.h.

Referenced by lower_selected().

int VsnetTCPSocket::_incomplete_header

protected

We send sizeof(Header) bytes as a packet length indicator. Unfortunately, even these sizeof(Header) bytes may be split by TCP. These two variables are needed for collecting the sizeof(Header) bytes. Note: for the obvious reason that this happens rarely, the collection code can not be considered tested.

Definition at line 115 of file vsnet_sockettcp.h.

Referenced by lower_selected().

Blob* VsnetTCPSocket::_incomplete_packet

protected

if we have received part of a TCP packet but not the complete packet, the expected length and received number of bytes are stored in _incomplete_packet. If several packets have been received at once, but the application processes them one at a time, the received, unprocessed packets are stored in the list.

Definition at line 103 of file vsnet_sockettcp.h.

Referenced by lower_selected(), and ~VsnetTCPSocket().

SqQueues VsnetTCPSocket::_sq

protected

Definition at line 160 of file vsnet_sockettcp.h.

Referenced by lower_clean_sendbuf(), lower_sendbuf(), need_test_writable(), queueLen(), and sendbuf().