---

Socket.cc

/*
 * Copyright 2003 Michael A. Marsh, Cornell University. All rights reserved.
 * This software is released under the modified BSD license.
 * See the file LICENSE in the top-level directory for details.
 */
//
// $Id: Socket.cc,v 1.4 2004/05/19 15:56:56 mmarsh Exp $
//
// $Log: Socket.cc,v $
// Revision 1.4  2004/05/19 15:56:56  mmarsh
// *** empty log message ***
//
// Revision 1.3  2004/01/23 14:54:57  mmarsh
// flush() is supposed to be a blocking write, so we should ignore
// EAGAIN/EWOULDBLOCK errors.
//
// Revision 1.2  2003/11/04 22:17:23  mmarsh
// General code cleanup.
//
//

#include "Socket.h"
#include "RemoteServer.h"
#include "Message.h"
#include <errno.h>
#include <unistd.h>
#include <netinet/in.h>
#include <fcntl.h>
#include <iostream>

using namespace CODEX_Quorum;

SocketBase::SocketBase( int   domain   ,
                        int   type     ,
                        int   protocol ,
                        bool  blocking ) :
   m_msgOffset(0),
   m_domain(domain),
   m_type(type),
   m_protocol(protocol),
   m_blocking(blocking),
   m_port(0),
   m_backlog(0)
{
   m_socket = ::socket(m_domain,m_type,m_protocol);
   if ( m_socket < 0 )
      throw QSESocketBaseBadSocket( __FILE__ ,
                                    __LINE__ ,
                                    m_domain , m_type , m_protocol , errno );
}

SocketBase::SocketBase( const SocketBase& aOther ) :
   m_msgOffset(0),
   m_domain(aOther.m_domain),
   m_type(aOther.m_type),
   m_protocol(aOther.m_protocol),
   m_blocking(aOther.m_blocking),
   m_port(aOther.m_port),
   m_backlog(aOther.m_backlog),
   m_socket(aOther.m_socket)
{}

SocketBase::~SocketBase()
{
   ::close( m_socket );
}

void
SocketBase::setup( int port, int backlog )
{
   protected_bind(port);
   protected_listen(backlog);
}

void
SocketBase::setup(struct sockaddr* my_addr,
                  socklen_t addrlen,
                  int backlog)
{
   protected_bind(my_addr, addrlen);
   protected_listen(backlog);
}

void
SocketBase::connect(const RemoteServer& server)
{
   if ( ::connect(m_socket, server.sockaddr(), server.addrlen()) < 0 )
      throw QSESocketBaseCannotConnect( __FILE__ ,
                                        __LINE__ ,
                                        m_socket , server , errno );
}

SocketBase*
SocketBase::accept()
{
   return protected_accept();
}

size_t
SocketBase::readFrom( void* output, size_t maxSize ) const
{
   //int bytesRead = ::recv( m_socket, output, maxSize, 0 );
   int bytesRead = ::read( m_socket, output, maxSize );
   if ( bytesRead < 0 )
   {
      throw QSESocketBaseReadFailed( __FILE__ , __LINE__ , m_socket , errno );
   }
   if ( 0 == bytesRead )
   {
      throw QSESocketBaseSocketClosed( __FILE__ , __LINE__ , m_socket , 0 );
   }
   return bytesRead;
}

size_t
SocketBase::readAll( Message& msg, size_t length ) const
{
   const size_t buffSize = 1024;
   unsigned char buff[buffSize];
   size_t readLength;
   if ( 0 == msg.length() )
   {
      readLength = 1;
   }
   else
   {
      if ( 0 == length )
      {
         readLength = buffSize;
      }
      else
      {
         readLength = ( length < buffSize ) ? length : buffSize;
      }
   }
   size_t toRead = length;
   size_t nRead = readFrom( buff, readLength ); // this may throw
   if ( 0 == nRead )
   {
      // There was a problem reading -- we're not ready yet.
      return 0;
   }
   int offset = 0;
   if ( 0 == msg.length() )
   {
      // we need to find out how long the message is -- this should be
      // the only thing read out at the moment
      unsigned char nLength = buff[0];
      if ( nLength > 4 ) // int on any platform -- > 4GB is ridiculous
      {
         throw QSESocketBaseMessageTooLong( __FILE__ , __LINE__ ,
                                            m_socket , 0 );
      }
      nLength = ( nLength < 1 ) ? 1 : nLength; // 0 == 1
      nRead += readFrom( buff+1, nLength + 1 );
      if ( nRead < nLength + 2U ) // at least one byte of actual data -- we're
                                  // willing to block at this point, since
                                  // it's critical that we read this data now
      {
         throw QSESocketBaseMessageTooShort( __FILE__ , __LINE__ ,
                                             m_socket , 0 );
      }
      toRead = 0;
      for ( int i = 1 ; i <= nLength ; ++i )
      {
         toRead = toRead*256 + buff[i];
      }
      offset = nLength + 1;
   }

   // now fill the Message with what we've just read
   msg.fill( buff+offset, nRead-offset );
   toRead -= nRead-offset;

   // read until we're done or we block
   while ( toRead > 0 )
   {
      fd_set read_set;
      FD_ZERO( &read_set );
      int s = set_fd( &read_set, SocketBase::kRead );
      struct timeval tv;
      tv.tv_sec = 0;
      tv.tv_usec = 0;
      if ( 0 > select(s+1, &read_set, 0, 0, &tv) )
      {
         throw QSESocketBaseReadFailed( __FILE__ , __LINE__ ,
                                        m_socket , errno );
      }

      if ( ! isset_fd( &read_set, SocketBase::kRead ) )
      {
         // we're blocked
         return toRead;
      }

      // get up to toRead or buffSize bytes, whichever is smaller
      nRead = readFrom( buff, (toRead<buffSize)?toRead:buffSize );
      if ( nRead > 0 )
      {
         msg.fill(buff,nRead);
      }
      // this is safe -- nRead <= toRead according to semantics of readFrom
      toRead -= nRead;
   }
   return 0;
}

void
SocketBase::writeTo( const Message& input ) const
{
   int messageSize = input.length();

   // Send the length info -- This is the most conservative way to set it,
   // with the maximum number of length bytes permitted.
   unsigned char length_info[5];
   length_info[0] = 4;
   length_info[4] = messageSize%256;
   length_info[3] = (messageSize/256)%256;
   length_info[2] = (messageSize/65536)%256;
   length_info[1] = (messageSize/16777216)%256;

   // We need to put the entire message in a single buffer so that
   // we can write a large chunk all at once.  Otherwise, there's
   // likely to be an error on the receiving end when it can't read
   // enough data.
   Message* m = new Message( length_info, 5 );
   m->fill( input );
   m_msgQueue.push( m );
}

int
SocketBase::set_fd( fd_set* fd_bitmap, SocketBase::StateType s ) const
{
   FD_SET( m_socket, fd_bitmap );
   return m_socket;
}

bool
SocketBase::isset_fd( const fd_set* fd_bitmap, SocketBase::StateType s ) const
{
   bool retVal = FD_ISSET( m_socket, fd_bitmap );
   if ( retVal && ( kWrite == s ) )
   {
      // Write as much as possible from the message queue without blocking.
      while ( retVal && !m_msgQueue.empty() )
      {
         Message* first = m_msgQueue.front();
         const unsigned char* output = first->buffer() + m_msgOffset;
         size_t maxSize = first->length() - m_msgOffset;
         int bytesWritten = internal_write( output, maxSize );
         m_msgOffset += bytesWritten;
         if ( first->length() == m_msgOffset )
         {
            // The message is written completely.  Clean up.
            delete first;
            m_msgQueue.pop();
            m_msgOffset = 0;
         }

         // Check the status of the socket again.
         fd_set write_set;
         FD_ZERO( &write_set );
         int s = set_fd( &write_set, SocketBase::kWrite );
         struct timeval tv;
         tv.tv_sec = 0;
         tv.tv_usec = 0;
         if ( 0 > select( s+1, 0, &write_set, 0, &tv ) )
         {
            throw QSESocketBaseWriteFailed( __FILE__ , __LINE__ ,
                                            m_socket , errno );
         }
         retVal = FD_ISSET( m_socket, &write_set );
      }
   }
   return retVal;
}

void
SocketBase::flush() const
{
   // This will basically spin until all of the data is consumed.
   while ( ! m_msgQueue.empty() )
   {
      Message* first = m_msgQueue.front();
      const unsigned char* output = first->buffer() + m_msgOffset;
      size_t maxSize = first->length() - m_msgOffset;
      try
      {
         int bytesWritten = internal_write( output, maxSize );
         m_msgOffset += bytesWritten;
      }
      catch ( QSESocketBaseWriteFailed& e )
      {
         // This is a blocking routine, so if we'd block, that's fine.
         if ( e.error() != EAGAIN )
         {
            throw;
         }
      }
      if ( first->length() == m_msgOffset )
      {
         // The message is written completely.  Clean up.
         delete first;
         m_msgQueue.pop();
         m_msgOffset = 0;
      }
   }
}

int
SocketBase::internal_write( const unsigned char* output, size_t maxSize ) const
{
   int bytesWritten = ::send( m_socket, output, maxSize, 0 );
   if ( bytesWritten < 0 )
   {
      throw QSESocketBaseWriteFailed( __FILE__ , __LINE__ , m_socket , errno );
   }
   return bytesWritten;
}

SocketBase*
SocketBase::protected_accept()
{
   SocketBase* newSocket = clone();
   // We don't know what accept(2) is going to do, memory-wise, so
   // we need to use malloc(3) instead of new.
   struct sockaddr* addr = 0;
   socklen_t* paddrlen = 0;
   socklen_t addrlen = 0;
   switch( m_domain )
   {
      case PF_INET :
         addr = (struct sockaddr*) malloc(sizeof(struct sockaddr_in));
         addrlen = sizeof(struct sockaddr_in);
         paddrlen = &addrlen;
         break;
   }
   int newSocketFD = ::accept( m_socket, addr, paddrlen );
   if ( newSocketFD < 0 )
   {
      free( addr );
      throw QSESocketBaseAcceptFailed( __FILE__ , __LINE__ , m_socket, errno );
   }
   newSocket->setSocket( newSocketFD );
   newSocket->setBacklog( 0 );
   switch( m_domain )
   {
      case PF_INET :
         struct sockaddr_in* addr_in = (struct sockaddr_in*) addr;
         newSocket->setPort(addr_in->sin_port);
         break;
   }
   free(addr);
   if ( ! m_blocking )
   { // only using the POSIX variety
      int flags;
      if ( -1 == (flags = fcntl(newSocket->socket(), F_GETFL, 0)) )
         flags = 0;
      fcntl(newSocket->socket(), F_SETFL, flags | O_NONBLOCK);
   }
   newSocket->finish_accept();
   return newSocket;
}

void
SocketBase::finish_accept()
{
}

SocketBase*
SocketBase::clone()
{
   SocketBase* copy = new SocketBase( *this );
   return copy;
}

void
SocketBase::protected_bind(int port)
{
   m_port = port;
   struct sockaddr_in* my_addr_in = new struct sockaddr_in;
   my_addr_in->sin_family         = AF_INET;
   my_addr_in->sin_port           = htons(m_port);
   my_addr_in->sin_addr.s_addr    = htonl(INADDR_ANY);

   struct sockaddr* my_addr = (struct sockaddr*)my_addr_in;
   socklen_t addrlen = sizeof(*my_addr_in);

   protected_bind(my_addr, addrlen);
   delete my_addr_in;
}

void
SocketBase::protected_bind(struct sockaddr* my_addr, socklen_t addrlen)
{
   int on = 1;
   ::setsockopt( m_socket, SOL_SOCKET, SO_REUSEADDR, (void*)&on, sizeof(on) );
   if ( ::bind(m_socket, my_addr, addrlen) < 0 )
      throw QSESocketBaseCannotBind( __FILE__ , __LINE__ , m_socket , errno );
}

void
SocketBase::protected_listen(int backlog)
{
   m_backlog = backlog;
   if ( ::listen(m_socket, m_backlog) < 0 )
      throw QSESocketBaseListenFailed( __FILE__ , __LINE__ ,
                                       m_socket , m_backlog , errno );
}

void
SocketBase::setSocket( int socketFD )
{
   m_socket = socketFD;
}

void
SocketBase::setBacklog( int backlog )
{
   m_backlog = backlog;
}

void
SocketBase::setPort( int port )
{
   m_port = port;
}

//------ Exceptions ------//

void
QSESocketBase::derivedMsg() const
{
   cerr << "SocketBase -- ";
   errMsg();
   cerr << "\n   errno = " << m_error;
}

void
QSESocketBaseBadSocket::errMsg() const
{
   cerr << "call to socket( "
        << m_domain   << " , "
        << m_type     << " , "
        << m_protocol << " ) failed";
}

void
QSESocketBaseCannotBind::errMsg() const
{
   cerr << "call to bind failed for socket " << m_socket;
}

void
QSESocketBaseListenFailed::errMsg() const
{
   cerr << "call to listen failed for socket " << m_socket
        << ", backlog " << m_backlog;
}

void
QSESocketBaseAcceptFailed::errMsg() const
{
   cerr << "call to accept failed for socket " << m_socket;
}

void
QSESocketBaseCannotConnect::errMsg() const
{
   cerr << "cannot connect:\n"
        << "   socket = " << m_socket << "\n"
        << "   server = " << m_server.name() << ":" << m_server.port();
}

void
QSESocketBaseSocketClosed::errMsg() const
{
   cerr << "socket " << m_socket << " is closed";
}

void
QSESocketBaseReadFailed::errMsg() const
{
   cerr << "read failed for socket " << m_socket;
}

void
QSESocketBaseWriteFailed::errMsg() const
{
   cerr << "write failed for socket " << m_socket;
}

void
QSESocketBaseMessageTooLong::errMsg() const
{
   cerr << "message too long on socket " << m_socket;
}

void
QSESocketBaseMessageTooShort::errMsg() const
{
   cerr << "message too short on socket " << m_socket;
}

---