EnsemblServer/ClientServerUtils.cpp

Go to the documentation of this file.

// #######################################################################

// SSAHA : Sequence Search and Alignment by Hashing Algorithm
// Version 3.2, released 1st March 2004
// Copyright (c) Genome Research 2002

// SSAHA is free software; you can redistribute it and/or modify 
// it under the terms of version 2 of the GNU General Public Licence
// as published by the Free Software Foundation.
 
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public Licence for more details.
 
// You should have received a copy of the GNU General Public Licence
// along with this program; if not, write to the Free Software
// Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
// or see the on-line version at http://www.gnu.org/copyleft/gpl.txt

// #######################################################################

// Module Name  : ClientServerUtils
// File Name    : ClientServerUtils.cpp
// Language     : C++
// Module Author: Anthony J. Cox (ac2@sanger.ac.uk)

// Description:

// Includes:

#include <cstdlib>
#include "MatchStore.h"
#include <cerrno>
#include <sys/socket.h>
#include "ClientServerUtils.h"
#include <poll.h>

// ### Function Definitions ###

MatchInfo::MatchInfo( const Match& m ) :
  subjectNum(m.getSubjectNum()),  
  subjectStart(m.getSubjectStart()),  
  subjectEnd(m.getSubjectEnd()),  
  queryNum(m.getQueryNum()),  
  // changed so as to report reverse match positions as seen from forward 
  // direction TC 13.12.2001
  //queryStart(m.getQueryStart()),  
  //queryEnd(m.getQueryEnd()),  
  queryStart
  ( m.isQueryForward()
    ? m.getQueryStart()
    : m.getQuerySize() - m.getQueryEnd() + 1 ),
  queryEnd
  ( m.isQueryForward()
    ? m.getQueryEnd()
    : m.getQuerySize() - m.getQueryStart() + 1 ),  
  numBases(m.getNumBases()),
  isQueryForward(m.isQueryForward()) ,
  isSubjectForward(m.isSubjectForward()) 
  {}


void
err_sys(const char *fmt, ...)
{
        printf("error: %s\n", fmt);
        exit(1);
}



int Accept(int fd, struct sockaddr *sa, socklen_t *salenptr)
{
  int             n;

 again:
  if ( (n = accept(fd, sa, salenptr)) < 0) 
  {
    if (errno == ECONNABORTED)
      goto again;
    else
      throw NetworkException
      ( (string)"accept error: " + (string)strerror(errno) ); 
  }
  return(n);
} // ~Accept


void
Bind(int fd, const struct sockaddr *sa, socklen_t salen)
{
  if (bind(fd, sa, salen) < 0) throw NetworkException
      ( (string)"bind error: " + (string)strerror(errno) ); 
} // ~Bind

void
Listen(int fd, int backlog)
{
  char  *ptr;

  /*4can override 2nd argument with environment variable */
  if ( (ptr = getenv("LISTENQ")) != NULL)
    backlog = atoi(ptr);

  if (listen(fd, backlog) < 0)
    throw NetworkException
      ( (string)"listen error: " + (string)strerror(errno) ); 
} // ~Listen

Sigfunc * signal(int signo, Sigfunc *func)
{
  struct sigaction      act, oact;
  
  act.sa_handler = func;
  sigemptyset(&act.sa_mask);
  act.sa_flags = 0;
  if (signo == SIGALRM) {
#ifdef  SA_INTERRUPT
    act.sa_flags |= SA_INTERRUPT;       /* SunOS 4.x */
#endif
  } else {
#ifdef  SA_RESTART
    act.sa_flags |= SA_RESTART;         /* SVR4, 44BSD */
#endif
  }
  if (sigaction(signo, &act, &oact) < 0)
    return(SIG_ERR);
  return(oact.sa_handler);
}
/* end signal */

Sigfunc * Signal(int signo, Sigfunc *func)  /* for our signal() function */
{
  Sigfunc       *sigfunc;
  if ( (sigfunc = signal(signo, func)) == SIG_ERR) 
    throw NetworkException
      ( (string)"signal error: " + (string)strerror(errno) ); 
  return(sigfunc);
}


void Close(int fd)
{  
  if (close(fd) == -1) 
  {
    throw NetworkException
      ( (string)"close error: " + (string)strerror(errno) ); 
  }
}


pid_t Fork(void)
{
  pid_t pid;
  if ( (pid = fork()) == -1) throw NetworkException("fork error");
  return(pid);
}

int Socket(int family, int type, int protocol)
{
  int           n;
  if ( (n = socket(family, type, protocol)) < 0) throw NetworkException("socket error");
  return(n);
}


ssize_t readn(int fd, void *vptr, size_t n)
/* Read "n" bytes from a descriptor. */
{
  size_t        nleft;
  ssize_t       nread;
  char  *ptr;
  
  ptr = (char*) vptr;
  nleft = n;
  while (nleft > 0) 
  {
    if ( (nread = read(fd, ptr, nleft)) < 0) 
    {
      if (errno == EINTR)
        nread = 0;              /* and call read() again */
      else
        return(-1);
    } else if (nread == 0)
      break;                            /* EOF */
    
    nleft -= nread;
    ptr   += nread;
  }
  return(n - nleft);            /* return >= 0 */
}
/* end readn */

ssize_t Readn(int fd, void *ptr, size_t nbytes)
{
  ssize_t               n;
  
  if ( (n = readn(fd, ptr, nbytes)) < 0)
    throw NetworkException("readn error");
  return(n);
}


#ifdef THIS_IS_A_NON_THREADSAFE_FUNCTION_SO_DONT_USE_IT
static ssize_t my_read(int fd, char *ptr)
{
  static int    read_cnt = 0;
  static char   *read_ptr;
  static char   read_buf[MAXLINE];

  cout << "@my read" << endl;
  
  if (read_cnt <= 0) 
  {
  again:
    if ( (read_cnt = read(fd, read_buf, sizeof(read_buf))) < 0) 
    {
      if (errno == EINTR)
        goto again;
      return(-1);
    } 
    else if (read_cnt == 0) return(0);
    cout << "@my read: got " << read_cnt << " bytes.\n";
    read_ptr = read_buf;
  }
  
  read_cnt--;
  *ptr = *read_ptr++;
  cout << "@my read:*ptr=" << *ptr << endl; 
  return(1);
}

ssize_t readline(int fd, void *vptr, size_t maxlen)
{
  int           n, rc;
  char  c, *ptr;
  
  ptr = (char *) vptr;
  for (n = 1; n < maxlen; n++) {
    cout << "@readline, n=" << n << endl; // %%%%
    if ( (rc = my_read(fd, &c)) == 1) {
      *ptr++ = c;
      if ((c == '\n')||(c=='\0')) // modified to stop at end of string too
        break;  /* newline is stored, like fgets() */
    } else if (rc == 0) {
      if (n == 1)
        return(0);      /* EOF, no data read */
      else
        break;          /* EOF, some data was read */
    } else
      return(-1);               /* error, errno set by read() */
  }
  
  *ptr = 0;     /* null terminate like fgets() */
  return(n);
}
/* end readline */

ssize_t Readline(int fd, void *ptr, size_t maxlen)
{
  ssize_t               n;
  if ( (n = readline(fd, ptr, maxlen)) < 0) 
    throw NetworkException("readline error");
  return(n);
}
#endif

ssize_t writen(int fd, const void *vptr, size_t n) 
/* Write "n" bytes to a descriptor. */
{
  size_t                nleft;
  ssize_t               nwritten;
  const char    *ptr;

  //  pollfd pollDetails;
  //  pollDetails.fd=fd;
  //  pollDetails.events=POLLIN|POLLRDNORM|POLLRDBAND|POLLPRI|POLLOUT|POLLWRNORM|POLLWRBAND;
  //  int pollOut(0);


  ptr = (char *) vptr;
  nleft = n;
  while (nleft > 0) {

    //    do { pollOut=poll(&pollDetails,1,10); } while (pollOut==0);
    //  cout << "POLL Writen: " <<   pollOut
    //   << ((pollDetails.revents|POLLIN!=0)?'1':'0')
    //    << ((pollDetails.revents|POLLRDNORM!=0)?'1':'0')
    //    << ((pollDetails.revents|POLLRDBAND!=0)?'1':'0')
    //   << ((pollDetails.revents|POLLPRI!=0)?'1':'0')
    //   << ((pollDetails.revents|POLLOUT!=0)?'1':'0')
    //   << ((pollDetails.revents|POLLWRNORM!=0)?'1':'0')
    //   << ((pollDetails.revents|POLLWRBAND!=0)?'1':'0')
    //   << ((pollDetails.revents|POLLERR!=0)?'1':'0')
    //   << ((pollDetails.revents|POLLHUP!=0)?'1':'0')
    //   << ((pollDetails.revents|POLLNVAL!=0)?'1':'0') << endl;


    if ( (nwritten = write(fd, ptr, nleft)) <= 0) {
      if (errno == EINTR)
        nwritten = 0;           /* and call write() again */
      else
        return(-1);                     /* error */
    }
    
    nleft -= nwritten;
    ptr   += nwritten;
  }
  return(n);
}
/* end writen */

void Writen(int fd, void *ptr, size_t nbytes)
{


  if (writen(fd, ptr, nbytes) != nbytes) 
  {
    //    cout << "Error no :"<< errno << endl;
    throw BrokenSocketException();
    //    else throw NetworkException("writen error");
  }
}


char* Fgets(char *ptr, int n, FILE *stream)
{
  char  *rptr;
  if ( (rptr = fgets(ptr, n, stream)) == NULL && ferror(stream))
    throw NetworkException("fgets error");
  return (rptr);
}

void Fputs(const char *ptr, FILE *stream)
{
  if (fputs(ptr, stream) == EOF) throw NetworkException("fputs error");
}

void Connect(int fd, const struct sockaddr *sa, socklen_t salen)
{
  if (connect(fd, sa, salen) < 0) throw NetworkException("connect error");
}



// SocketInterface member functions

void SocketInterface::sendSequence( const WordSequence& seq )
{
    SequenceHeader sinfo;
    sinfo.size = seq.end() - seq.begin();
    sinfo.basesInLast = seq.getNumBasesInLast();

    sendStruct(&sinfo);
    Writen(portNum_, (void*)&(*seq.begin()), sinfo.size*sizeof(Word));

    //    bytesSent_+=sinfo.size*sizeof(Word)+sizeof(sinfo);
    //   cout << "@sent seq " << sinfo.size*sizeof(Word)+sizeof(sinfo) 
    //   << " bytes, " << bytesSent_ << "total.\n";

}


void SocketInterface::receiveSequence( WordSequence& seq )
{
  SequenceHeader sinfo;
  receiveStruct(&sinfo);
  //  cout << "receiveSeq: " << sinfo.size << " " << sinfo.basesInLast << endl;
  seq.resize(sinfo.size);

  copy((char*)&(*seq.begin()), sinfo.size*sizeof(Word));
  seq.setNumBasesInLast(sinfo.basesInLast);

  bytesRead_+=sinfo.size*sizeof(Word)+sizeof(sinfo);
  //  cout << "@read seq " << sinfo.size*sizeof(Word)+sizeof(sinfo) 
  //       << " bytes, " << bytesRead_ << "total.\n";
}

void SocketInterface::sendString( const string& s )
{
  Writen(portNum_, (void*)(s.c_str()), s.size()+1);
  bytesSent_+=s.size()+1;
  //  cout << "@send string " << s.size()+1 << " bytes, " << bytesSent_
  //     << "total.\n";
  // +1 cos we need to send the \0 at the end of the string
}

void SocketInterface::sendChars( const char* pChar, int numChars )
{
  static const char c('\0');
  Writen(portNum_, (void*)pChar, numChars);
  Writen(portNum_, (void*)&c, 1);
  bytesSent_+=numChars+1;
}


void SocketInterface::receiveString( string& s )
{
  s="";
  char c;
  while(1)
  {
    receiveStruct(&c);
    if((c=='\0')||(c=='\n')) return;
    s+=c;
  } 
}

void SocketInterface::receiveChars( vector<char>& v )
{
  v.clear();
  char c;
  while(1)
  {
    receiveStruct(&c);
    if((c=='\0')||(c=='\n')) return;
    v.push_back(c);
  } 
}


//string SocketInterface::receiveString( void )
//{
//  int a(Readline(portNum_, buffer_, MAXLINE));
//  bytesRead_+=a;
//  cout << "@receive string " << a << " bytes, " << bytesRead_
//       << "total.\n";
  // == 0)
  //     throw NetworkException("Data stream from server ended prematurely!");
  //  if (Readline(portNum_, buffer_, MAXLINE) == 0)
  //     throw NetworkException("Data stream from server ended prematurely!");
//  return string(buffer_);
//}

void SocketInterface::checkSocketEmpty( void )
{
  
  //  Linux and other Unixes doesn't have MSG_NONBLOCK as an argument to recv. Workaround by DJC.
  

  //  Is the socket blocking at the moment?
  bool blocking;
  
  int flags=fcntl(portNum_,F_GETFL,0);
  int newflags=flags;

  if (flags==-1) 
    throw NetworkException("Problem reading status (fcntl) of socket");
  

  // Make it non-blocking
  newflags|=O_NONBLOCK;
  fcntl(portNum_,F_SETFL,newflags);

  // See if we can read data
  int n(recv( portNum_, buffer_, MAXLINE, MSG_PEEK));
 
  // Put the socket back to how it was before (blocking or non-blocking)
  fcntl(portNum_,F_SETFL,flags); 

  //perror("error status = ");
  if (errno!=EWOULDBLOCK) 
    {
      //    cout << "There are " << n << " bytes waiting to be read.\n";
      if (n>0) throw NetworkException("Unexpected data at socket!");
    }

}

void SocketInterface::copy( char* target, int numBytes)
{
  //  cout << "copy - requested " << numBytes << ", got " << size() << endl;
  if (size()<numBytes) getAtLeast(numBytes-size());

  for( int i(0) ; i < numBytes ; ++ i )
    { *target++ = front(); pop_front(); }

  //  cout << "final size " << size() << endl;
}

void SocketInterface::getAtLeast( int numBytes )
{

  //  cout << "  getAtLeast " << numBytes << " bytes\n";

  int numRead(0);
  int totalNumRead(0); 
  int numTries(0);
  fd_set fds;
  FD_ZERO(&fds);
  FD_SET(portNum_, &fds);
  
  while(1)
  {
    // bit of a cheat using numRead twice here ...
    numRead=select(portNum_+1, &fds, NULL, NULL, &timeOut_ );
    if (numRead==0)
    {
      throw NetworkException
        ("Timed out getting data from socket: " + (string)strerror(errno) ); 
    } // ~if
    else if (numRead<0)
    {
      throw NetworkException
        ("Select error getting data from socket: " + (string)strerror(errno) ); 
    } // ~else if

    //    numRead=read(portNum_,(void*)&buffer_[0],MaxChunkSize);
    numRead=recv(portNum_,(void*)&buffer_[0],MaxChunkSize,0);
    if (numRead==0)
    {
      if (++numTries==100)
      throw NetworkException
        ("Reading zero bytes from socket, probable peer disconnect: " 
         + (string)strerror(errno) ); 
    }    
    if ( numRead < 0 )
    {
      if (errno==EINTR) continue;
      else 
      {
        //      throw NetworkException("getAtLeast error");
        throw NetworkException
        ( (string)"getAtLeast error: " + (string)strerror(errno) ); 
      } // ~else
    } // ~if
    //    cout << " got " << numRead << " so far...\n";
    insert(end(),&buffer_[0],&buffer_[numRead]);
    totalNumRead +=numRead;
    if (totalNumRead>=numBytes) break;

  } // ~while

  //  cout << "  total got " << totalNumRead << endl;

}




// Name:
// Arguments:
// TYPE  NAME  IN/OUT COMMENT
// Returns: TYPE COMMENT

// End of file ClientServerUtils.cpp

---