unix_buffer.H

Go to the documentation of this file.

#ifndef UNIX_BUFFER_
#define UNIX_BUFFER_

// some abstraction layers for using UNIX-network services in C++ (mainly TCP)
// written by Thorsten Reinecke (reinecke@thorstenreinecke.de)
// last change: 2007-06-06

#ifdef CYGWIN_COMPAT
 #warning cygwin compatibility hacks enabled
#endif

extern "C"
{
  #include <sys/socket.h>
  #include <sys/poll.h>
  #include <netinet/in.h>
  #include <netinet/tcp.h>
  #include <netdb.h>
  #include <unistd.h>

 #if defined(_REENTRANT) || defined(_THREAD_SAFE)
  // POSIX specifies, that (blocking) system calls are cancellation points,
  // but LinuxThreads may not be posix-conform to this issue,
  // (refer manpage PTHREAD_CANCEL(3))
  // so PTHREAD_TESTCANCEL is called explicitly at some important points
  #include <pthread.h>
  #define PTHREAD_TESTCANCEL pthread_testcancel(); /* explicit thread cancellation point */
 #else
  #define PTHREAD_TESTCANCEL /* this would be an explicit thread cancellation point */
 #endif 

}


// I had some problems while debugging under valgrind; this one
// hacks the problems away (but maybe not the bugs...)
//#define WEIRD_VALGRIND_HACK

#ifdef WEIRD_VALGRIND_HACK
 #if !( defined(_REENTRANT) || defined(_THREAD_SAFE) )
  // for single-threaded versions the hack needs to be disabled
  #warning "weird valgrind hack disabled"
  #undef WEIRD_VALGRIND_HACK
 #endif
#endif 

#include <cerrno>
#include <cstring>
#include <string>
#include <iostream>
#include <sstream>
#include <exception>

#ifdef WEIRD_VALGRIND_HACK
 #warning "weird valgrind hack enabled"
 #include "mutex.H"
#endif

#if defined(CYGWIN_COMPAT)
 #if defined(_REENTRANT) || defined(_THREAD_SAFE)
  #include "mutex.H"
  static CMutex cygwin_mutex;
  #define CYGW_LOCK cygwin_mutex.lock();
  #define CYGW_UNLOCK cygwin_mutex.unlock();
 #else
  #define CYGW_LOCK /* cygwin_mutex.lock(); */
  #define CYGW_UNLOCK /* cygwin_mutex.unlock(); */
 #endif
#endif   


using std::cerr;
using std::endl;
using std::flush;
using std::iostream;
using std::streambuf;


class unix_buffer_exception : public std::exception
{
 private:
  const std::string Message;

 public:
  explicit unix_buffer_exception(const std::string& Msg) throw()
   : Message(Msg)
   {
#if defined(VERBOSE) || defined(DEBUG)
     cerr << "throwing exception: " << Msg << endl;
#endif
   }
 
  virtual ~unix_buffer_exception() throw() { }
  
  virtual const char* what() const throw()
   {
     return Message.c_str();
   }  
};

inline std::string my_strerror(const int errnum)
{
  char buffer[1024];

  // please note: not the local array "buffer", but an implicitly constructed
  // std::string(buffer) is returned by this function!

#if 0
  // this is POSIX conform, but does not compile on my system:
  // POSIX -> int strerror_r(int errnum, char *buf, size_t n);
  if (strerror_r(errnum,buffer,sizeof(buffer))==0) return buffer;
  else return "";
#else
  // this is using the GNU extension and compiles on my system:
  // (see "man 3 strerror")
  char *ret = strerror_r(errnum,buffer,sizeof(buffer));
  if (ret) return ret; else return "";
#endif
}



int open_internet_port(const int port, const int qsize = 5)
{
  int sd;
  int val = 1;
  struct sockaddr_in server;
  struct linger lingval;

  if (port <= 0) return -1;

  sd = socket(AF_INET, SOCK_STREAM, 0);
  if (sd < 0) return sd;
  if (setsockopt(sd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val)) <0)
    return -1;

  // activate linger option
  lingval.l_onoff=1; // if applicable, keep socket unavailable after closing
  lingval.l_linger=10*100; // timeout = 10 seconds
  if (setsockopt(sd, SOL_SOCKET, SO_LINGER, &lingval, sizeof(lingval)) <0) return -1;

  server.sin_family = AF_INET;
  server.sin_addr.s_addr = INADDR_ANY;
  server.sin_port = htons(port);
  if (bind(sd, reinterpret_cast<struct sockaddr *>(&server), sizeof(server)) < 0) return -1;
  listen (sd, qsize);
  return sd;
}



// utils for verbose output of meta-data to streams

inline std::ostream& operator<< (std::ostream& ostr, const sockaddr &sa)
{
#ifdef CYGWIN_COMPAT
  // implement the obsolete stuff using gethostbyaddr
  CYGW_LOCK;
  u_int32_t ip = reinterpret_cast<const sockaddr_in& >(sa).sin_addr.s_addr;
  struct hostent *p_hostent = gethostbyaddr(reinterpret_cast<const char*>(&ip),sizeof(ip),AF_INET);
   // p_hostent is a static resource and therefore not threadsafe!
  if (p_hostent)  ostr << p_hostent->h_name;
  else ostr << "(unknown)";
  CYGW_UNLOCK;
#else
  // outputs sockaddr info to given stream
  char host[1024];
  char service[1024];
  const int flags = NI_NOFQDN;
// FIXME! sometimes hostname lookup segfaults on my system (when dynamically linked)
// this works better: const int flags = NI_NOFQDN | NI_NUMERICHOST | NI_NUMERICSERV;
// is this not threadsafe?
  const int retval = getnameinfo(&sa,sizeof(sa),host,sizeof(host),service,sizeof(service),flags);

  switch (retval)
  {
    case 0:
     ostr << host << " " << service;
     break;
    case EAI_AGAIN:
     ostr << "(The name could not be resolved at this time.)"; break;
    case EAI_BADFLAGS:
     ostr << "(The flags parameter has an invalid value.)"; break;
    case EAI_FAIL:
     ostr << "(A non-recoverable error occurred.)"; break;
    case EAI_FAMILY: 
     ostr << "(The address family was not recognized.)"; break;
    case EAI_MEMORY:
     ostr << "(Out of memory.)"; break;
    case EAI_NONAME:
     ostr << "(The  name  does  not  resolve.)"; break;
    case EAI_SYSTEM:
     ostr << "(A system error occurred.)"; break;
    default:
     ostr << "(unspecified error in getnameinfo.)"; 
   }
#endif
  return ostr;
}

inline std::string peer_info(const int socket_descriptor)
{
 // outputs meta-data for socket_descriptor
 std::ostringstream ostr;

 struct sockaddr sa;
 socklen_t sl = sizeof(sa);

 const int retval = getpeername(socket_descriptor, &sa, &sl);
 if (retval==0) ostr << sa;
 return ostr.str();
}

inline std::string socket_info(const int socket_descriptor)
{
 // outputs meta-data for socket_descriptor
 std::ostringstream ostr;

 struct sockaddr sa;
 socklen_t sl = sizeof(sa);

 const int retval = getsockname(socket_descriptor, &sa, &sl);
 if (retval==0) ostr << sa;
 return ostr.str();
}

inline std::string connection_info(const int socket_descriptor)
{
 // outputs meta-data for socket_descriptor
 std::ostringstream ostr;

 struct sockaddr sa;
 socklen_t sl = sizeof(sa);

 const int retval1 = getpeername(socket_descriptor, &sa, &sl);
 ostr << "connection from ";
 switch (retval1)
  {
    case 0       : ostr << sa; break;
    case EBADF   : ostr << "(invalid descriptor)"; break;
    case ENOTSOCK: ostr << "(file, not a socket)"; break;
    default : ostr << "(unable to perform getpeername)";
  }

 const int retval2 = getsockname(socket_descriptor, &sa, &sl);
 ostr << " to ";
 switch (retval2)
  {
    case 0       : ostr << sa; break;
    case EBADF   : ostr << "(invalid descriptor)"; break;
    case ENOTSOCK: ostr << "(file, not a socket)"; break;
    default : ostr << "(unable to perform getsockname)";
  }
 return ostr.str();
}

template<class T> inline std::string peer_info(const T& obj)
{
 // outputs connection meta-data for class object to string
 return peer_info(obj.get_descriptor());
}

template<class T> inline std::string socket_info(const T& obj)
{
 // outputs connection meta-data for class object to string
 return socket_info(obj.get_descriptor());
}

template<class T> inline std::string connection_info(const T& obj)
{
 // outputs connection meta-data for class object to string
 return connection_info(obj.get_descriptor());
}


class Cprovide_descriptor_access
{
 protected:
  virtual int _get_descriptor() const = 0;
 public:
  virtual ~Cprovide_descriptor_access() { }
};


class Cpoll_methods : public Cprovide_descriptor_access
{
public:
  virtual ~Cpoll_methods() {}

  int poll_events(short requested_events, const int timeout_ms=1) const
   {
     pollfd pfd;
     int res=0;
     do
      {
        pfd.fd=_get_descriptor(); pfd.events=requested_events; pfd.revents=0;
        res=poll(&pfd,1,timeout_ms);
        if (res==-1) if (errno!=EINTR) return -1;
      } while (res==-1);
     return pfd.revents;
   }

 inline bool bad() const
  {
    const int revents=poll_events(POLLIN);
    return (revents==-1) | (revents&(POLLERR|POLLHUP|POLLNVAL));
  }

 inline bool good() const
  {
    return !bad();
  }

 int readable_chars_within(const int size, const int timeout_ms=1000) const
  {
    const int revents=poll_events(POLLIN,timeout_ms);
    if (revents==-1) return -1;
    if (revents&(POLLERR|POLLHUP|POLLNVAL)) return -1;
    if (revents&POLLIN)
     {
       char *buf = new char[size];
       int count = recv(_get_descriptor(), buf, size, MSG_PEEK|MSG_NOSIGNAL);
       delete [] buf;
       return count;
     }
    return -1;
  }

 bool writable_now() const
  {
    const int revents=poll_events(POLLOUT);
    if (revents==-1) return false;
    if (revents&(POLLERR|POLLHUP|POLLNVAL)) return false;
    return (revents&POLLOUT);
  }
};



class Cpoll : public Cpoll_methods
{
private:
  int fd; // descriptor to poll

protected:
  virtual int _get_descriptor() const { return fd; }  

public:
  explicit Cpoll(const int _fd=0) : fd(_fd) {}

  template<class T> explicit Cpoll(const T &obj) : fd(obj.get_descriptor()) {}

  virtual ~Cpoll() {}

  void attach(const int _fd) { fd=_fd; }

  template<class T> void attach(const T &obj) { attach(obj.get_descriptor()); }
};



class Cnetwork_connection_methods : public Cprovide_descriptor_access
{
 public:
  virtual ~Cnetwork_connection_methods() { }

  inline void set_RCVBUF(const int size)
   {
     const int fd = _get_descriptor();
     int val = size; // new buffersize
     if (setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &val, sizeof(val))==-1)
       throw unix_buffer_exception(std::string("Cnetwork_connection_methods::set_RCVBUF(): "+my_strerror(errno)));
   }

  inline void set_SNDBUF(const int size)
   {
     const int fd = _get_descriptor();
     int val;
     val=size; // new buffersize
     if (setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &val, sizeof(val))==-1)
       throw unix_buffer_exception(std::string("Cnetwork_connection_methods::set_SNDBUF(): "+my_strerror(errno)));
   }

  inline int get_RCVBUF()
   {
     const int fd = _get_descriptor();
     int val = -1;
     socklen_t optlen = sizeof(val);
     if (getsockopt(fd, SOL_SOCKET, SO_RCVBUF, &val, &optlen)==-1)
      {
        perror("Cnetwork_connection_methods::get_RCVBUF()");
        return -1;
      }
     else return val;
   }

  inline int get_SNDBUF()
   {
     const int fd = _get_descriptor();
     int val = -1;
     socklen_t optlen = sizeof(val);
     if (getsockopt(fd, SOL_SOCKET, SO_SNDBUF, &val, &optlen)==-1)
      {
        perror("Cnetwork_connection_methods::get_SNDBUF()");
        return -1;
      }
     else return val;
   }

  inline void set_RCV_timeout(const int secs, const int u_secs = 0)
   {
     const int fd = _get_descriptor();
     timeval tv;
     tv.tv_sec = secs; // secs timeout
     tv.tv_usec = u_secs; // micro-secs timeout
     if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv))==-1)
       throw unix_buffer_exception(std::string("Cnetwork_connection_methods::set_RCV_timeout(): "+my_strerror(errno)));
   }

  inline void set_SND_timeout(const int secs, const int u_secs = 0)
   {
     const int fd = _get_descriptor();
     timeval tv;
     tv.tv_sec = secs; // secs timeout
     tv.tv_usec = u_secs; // micro-secs timeout
     if (setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv))==-1)
       throw unix_buffer_exception(std::string("Cnetwork_connection_methods::set_SND_timeout(): "+my_strerror(errno)));
   }

  inline int get_RCV_timeout()
   {
     const int fd = _get_descriptor();
     timeval tv;
     socklen_t optlen = sizeof(tv);
     if (getsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, &optlen)==-1)
      {
        perror("Cnetwork_connection_methods::get_RCV_timeout()");
        return -1;
      }
     else return tv.tv_sec;
   }

  inline int get_SND_timeout()
   {
     const int fd = _get_descriptor();
     timeval tv;
     socklen_t optlen = sizeof(tv);
     if (getsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, &optlen)==-1)
      {
        perror("Cnetwork_connection_methods::get_SND_timeout()");
        return -1;
      }
     else return tv.tv_sec;
   }

  inline void set_timeout(const int secs, const int u_secs = 0)
   {
     const int fd = _get_descriptor();
     timeval tv;
     tv.tv_sec = secs; // secs timeout
     tv.tv_usec = u_secs; // micro-secs timeout
     if (setsockopt(fd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv))==-1)
       throw unix_buffer_exception(std::string("Cnetwork_connection_methods::set_timeout(), RCVTIMEO: "+my_strerror(errno)));
     if (setsockopt(fd, SOL_SOCKET, SO_SNDTIMEO, &tv, sizeof(tv))==-1)
       throw unix_buffer_exception(std::string("Cnetwork_connection_methods::set_timeout(), SNDTIMEO: "+my_strerror(errno)));
   }

  inline void set_TCP_NODELAY(const bool flag = true)
   {
     const int fd = _get_descriptor();
     int optval=flag;
     if (setsockopt(fd, SOL_TCP, TCP_NODELAY, &optval, sizeof(optval))==-1)
      throw unix_buffer_exception(std::string("Cnetwork_connection_methods::set_TCP_NODELAY(): "+my_strerror(errno)));
   }

  inline bool get_TCP_NODELAY()
   {
     const int fd = _get_descriptor();
     int optval;
     socklen_t optlen = sizeof(optval);
     if (getsockopt(fd, SOL_TCP, TCP_NODELAY, &optval, &optlen)==-1)
      perror("Cnetwork_connection_methods::get_TCP_NODELAY");
     return optval;
   }

  inline void set_LINGER(const signed int seconds)
   {
     // seconds<0  : turn linger off
     // seconds>=0 : turn linger on with specified value seconds
     const int fd = _get_descriptor();
     struct linger lingval;
     if (seconds<0) { lingval.l_onoff=0; lingval.l_linger=0; }
     else { lingval.l_onoff=1; lingval.l_linger=seconds; }
     if (setsockopt(fd, SOL_SOCKET, SO_LINGER, &lingval, sizeof(lingval))==-1)
      throw unix_buffer_exception(std::string("Cnetwork_connection_methods::set_LINGER(): "+my_strerror(errno)));
   }

  inline signed int get_LINGER()
   {
     // returns linger seconds or -1 for linger off or -2 for error
     const int fd = _get_descriptor();
     struct linger lingval;
     socklen_t optlen = sizeof(lingval);
     if (getsockopt(fd, SOL_SOCKET, SO_LINGER, &lingval, &optlen)==-1)
      {
        perror("Cnetwork_connection_methods::get_LINGER()");
        return -2;
      }
     if (lingval.l_onoff==0) return -1;
     else return lingval.l_linger;
   }

  inline const std::string peer_info()
  {
   // outputs connection meta-data for class object to string
   return ::peer_info(_get_descriptor());
  }

  inline const std::string socket_info()
  {
   // outputs connection meta-data for class object to string
   return ::socket_info(_get_descriptor());
  }

  inline const std::string connection_info()
  {
   // outputs connection meta-data for class object to string
   return ::connection_info(_get_descriptor());
  }

};




class connection_waiter : public Cnetwork_connection_methods
{
 protected:
  int server_socket;
  const int client_default_timeout;

  virtual int _get_descriptor() const { return server_socket; } 
 public:

  explicit connection_waiter(const int server_port, int server_port_tries = 10, const int _client_default_timeout = 75)
   : server_socket(-1), client_default_timeout(_client_default_timeout)
   {
     while ((server_socket = open_internet_port(server_port,10))<0
            && --server_port_tries>=0) sleep(1);
     if (server_socket<0)
      {
        cerr << "connection_waiter::constructor: cannot open serverport " << server_port << endl;
        throw unix_buffer_exception("connection_waiter::constructor: cannot open serverport");
      }
   }

  ~connection_waiter()
   {
     if (shutdown(server_socket, SHUT_RDWR)==-1) perror("~connection_waiter():shutdown");
     close(server_socket);
   }

  int get_new_client_socket_descriptor() const
   {
     int sd;
     do
      {
        sd=accept(server_socket, NULL, NULL);
        PTHREAD_TESTCANCEL;
      } while (sd==0);
     return sd;
   }

  int get_client_default_timeout() const
   {
     return client_default_timeout;
   }
};


class unix_fd_buffer : public streambuf, public Cprovide_descriptor_access
{
protected:
  static const int buffersize = 4096;
  static const int pushbacksize = 10;
  int fd; // -1 will explicitly denote an invalid descriptor
  bool fd_autoclose; // whether a valid descriptor shall be closed at destructor call
  char obuffer[buffersize];
  char ibuffer[buffersize+pushbacksize];

  virtual int _get_descriptor() const { return fd; } 

  virtual int empty_buffer()
   {
     if (fd<0) // invalid file descriptor?
      {
        cerr << "unix_fd_buffer::empty buffer(): invalid descriptor! fd=" << fd << endl;
        throw unix_buffer_exception("unix_fd_buffer::empty buffer(): invalid descriptor!");
        return EOF;
      }
     const int total_to_write = pptr()-pbase();
     if (!total_to_write) return 0;
     if (total_to_write<0 || total_to_write>buffersize)
      {
        throw unix_buffer_exception("unix_fd_buffer::empty_buffer() sanity check failed!");
      }
     int total_written = 0;
     signed int retries = 2; // maximum of retries in case of temporary errors
     do
     {
       PTHREAD_TESTCANCEL;
       const int count = total_to_write-total_written;
       const int written=write(fd, obuffer, count);
       PTHREAD_TESTCANCEL;
       if (written!=count)
        {
          cerr << "unix_fd_buffer::empty_buffer(): "
               << written << " bytes of " << count << " sent." << endl;
          if (written<0)
           {
             --retries;
             const int err=errno;
             perror("unix_fd_buffer::empty_buffer()");
             if (written==-1 && err == EAGAIN && retries>=0) continue; // this may fix some problems...
             if (written==-1 && err == EINTR && retries>=0) continue; // this may fix some problems...
             throw unix_buffer_exception(std::string("unix_fd_buffer::empty_buffer() send failed: ")+my_strerror(err));
             return EOF;
           }
        }
       pbump(-written);
       total_written+=written;
     } while (total_written<total_to_write);
     return total_written;
   }

  virtual int overflow (int c) 
   {
     if (c != EOF) { *pptr() = c; pbump(1); }
     if (empty_buffer() == EOF) return EOF;
     return c;
   }

  virtual int sync ()
   {
     return (empty_buffer() == EOF) ? -1 : 0;
   }

  virtual int underflow ()
   {
     if (gptr() < egptr()) return *gptr();
     int pbcount = gptr() - eback();
     if (pbcount > pushbacksize) pbcount = pushbacksize;
     if (pbcount > 0) memmove (ibuffer+(pushbacksize-pbcount), gptr()-pbcount, pbcount);
     int count;
     signed int retries = 2; // maximum of retries in case of temporary errors
     do
     {
       PTHREAD_TESTCANCEL;
       count = read(fd, ibuffer+pushbacksize, buffersize);
       PTHREAD_TESTCANCEL;
       if (count <= 0) // any unexpected behaviour?
        {
         if (count==0) return EOF;
         --retries;
         const int err=errno;
         perror("unix_fd_buffer::underflow()");
         if (count==-1 && err == EAGAIN && retries>=0) continue; // this should fix some problems...
         if (count==-1 && err == EINTR && retries>=0) continue; // this should fix some problems...
         throw unix_buffer_exception(std::string("unix_fd_buffer::underflow() recv failed: ")+my_strerror(err));
         return EOF;
        }
     } while (count<=0);
     //cerr << "---> count: " << count << endl;
     setg(ibuffer+(pushbacksize-pbcount), ibuffer+pushbacksize,
          ibuffer+pushbacksize+count);
     return *gptr();
   }
  
public:

  inline int get_descriptor() const { return fd; }


  void attach(const int _fd, const bool autoclose = true)
  {
    if (fd>0)
     {
       throw unix_buffer_exception("unix_fd_buffer::open(const int _fd): descriptor is already in use!");
     }
    if (_fd<0)
     {
       throw unix_buffer_exception("unix_fd_buffer::open(const int _fd): descriptor is invalid!");
     }
    fd = _fd; fd_autoclose=autoclose;
    setp(obuffer, obuffer+(buffersize-1));
    setg(ibuffer+pushbacksize, ibuffer+pushbacksize, ibuffer+pushbacksize);
  }

  virtual int detach()
  {
    if (fd<0) return -1; // no valid descriptor attached
    sync();
    if (fd_autoclose)
     {
       if (shutdown(fd, SHUT_RDWR)==-1) perror("~unix_fd_buffer():shutdown");
       close(fd);
       fd=-1; // fd now invalid
     }
    const int h=fd;
    fd=-1; // no descriptor attached to this object now
    return h;
  }

  unix_fd_buffer(void) // default constructor
   : fd(-1), // -1 denotes no descriptor defined yet
     fd_autoclose(true)
  {
    setp(obuffer, obuffer+(buffersize-1));
    setg(ibuffer+pushbacksize, ibuffer+pushbacksize, ibuffer+pushbacksize);
  }

  explicit unix_fd_buffer(const int _fd)
   : fd(_fd), fd_autoclose(true)
  {
    setp(obuffer, obuffer+(buffersize-1));
    setg(ibuffer+pushbacksize, ibuffer+pushbacksize, ibuffer+pushbacksize);
  }


  virtual ~unix_fd_buffer()
   {
     if (fd>=0)
      {
        // valid descriptor -> free resource
        sync();
        if (fd_autoclose) close(fd);
        fd=-1;
      }
   }

};



class unix_buffer : public unix_fd_buffer, public Cnetwork_connection_methods
{
protected:
  virtual int _get_descriptor() const { return fd; } 

  virtual int empty_buffer()
   {
     if (fd<0) // invalid file descriptor?
      {
        cerr << "unix_buffer::empty buffer(): invalid descriptor! fd=" << fd << endl;
        throw unix_buffer_exception("unix_buffer::empty buffer(): invalid descriptor!");
        return EOF;
      }
     const int total_to_write = pptr()-pbase();
     if (!total_to_write) return 0;
     if (total_to_write<0 || total_to_write>buffersize)
      {
        throw unix_buffer_exception("unix_buffer::empty_buffer() sanity check failed!");
      }
     int total_written = 0;
     signed int retries = 2; // maximum of retries in case of temporary errors
     do
     {
       PTHREAD_TESTCANCEL;
       const int count = total_to_write-total_written;
       const int written=send(fd, obuffer, count, MSG_NOSIGNAL);
       PTHREAD_TESTCANCEL;
       if (written!=count)
        {
          cerr << "unix_buffer::empty_buffer(): "
               << written << " bytes of " << count << " sent." << endl;
          if (written<0)
           {
             --retries;
             const int err=errno;
             perror("unix_buffer::empty_buffer()");
             if (written==-1 && err == EAGAIN && retries>=0) continue; // this may fix some problems...
             if (written==-1 && err == EINTR && retries>=0) continue; // this may fix some problems...
             throw unix_buffer_exception(std::string("unix_buffer::empty_buffer() send failed: ")+my_strerror(err));
             return EOF;
           }
        }
       pbump(-written);
       total_written+=written;
     } while (total_written<total_to_write);
     return total_written;
   }

  virtual int underflow ()
   {
     if (gptr() < egptr()) return *gptr();
     int pbcount = gptr() - eback();
     if (pbcount > pushbacksize) pbcount = pushbacksize;
     if (pbcount > 0) memmove (ibuffer+(pushbacksize-pbcount), gptr()-pbcount, pbcount);
     int count;
     signed int retries = 2; // maximum of retries in case of temporary errors
     do
     {
       PTHREAD_TESTCANCEL;
       count = recv(fd, ibuffer+pushbacksize, buffersize, MSG_NOSIGNAL);
       PTHREAD_TESTCANCEL;
       if (count <= 0) // any unexpected behaviour?
        {
         if (count==0) return EOF;
         --retries;
         const int err=errno;
         perror("unix_buffer::underflow()");
         if (count==-1 && err == EAGAIN && retries>=0) continue; // this should fix some problems...
         if (count==-1 && err == EINTR && retries>=0) continue; // this should fix some problems...
         throw unix_buffer_exception(std::string("unix_buffer::underflow() recv failed: ")+my_strerror(err));
         return EOF;
        }
     } while (count<=0);
     //cerr << "---> count: " << count << endl;
     setg(ibuffer+(pushbacksize-pbcount), ibuffer+pushbacksize,
          ibuffer+pushbacksize+count);
     return *gptr();
   }
  
public:

  virtual int detach()
  {
    if (fd<0) return -1; // no valid descriptor attached
    sync();
    if (fd_autoclose)
     {
       if (shutdown(fd, SHUT_RDWR)==-1) perror("~unix_buffer():shutdown");
       close(fd);
       fd=-1; // fd now invalid
     }
    const int h=fd;
    fd=-1; // no descriptor attached to this object now
    return h;
  }

  unix_buffer(void) // Default constructor
   : unix_fd_buffer() // use defaults from unix_fd_buffer
  {
    set_timeout(75); // default: 75 seconds
  }

  explicit unix_buffer(const int _fd, const int secs_timeout = 75)
   : unix_fd_buffer(_fd)
  {
    set_timeout(secs_timeout);
  }

  explicit unix_buffer(const connection_waiter& cw)
   : unix_fd_buffer(cw.get_new_client_socket_descriptor())
  {
    set_timeout(cw.get_client_default_timeout());
  }

 unix_buffer(const std::string &host, const int port)
  : unix_fd_buffer()
  {

#ifdef CYGWIN_COMPAT
    // cygwin compatibility mode

    struct sockaddr_in server;
    struct hostent *hp;
    int ret;
    int err_number;
    fd = socket(AF_INET, SOCK_STREAM, 0);
    if (fd < 0)
     {
       throw unix_buffer_exception("unix_buffer::constructor: Can't open socket");
     }
    server.sin_family = AF_INET;

    CYGW_LOCK;
    hp = gethostbyname(host.c_str());
    if (hp == NULL)
     {
       CYGW_UNLOCK;
       throw unix_buffer_exception("unix_buffer::constructor: Unknown host: "+host);
     }
    memcpy((char*)&server.sin_addr, (char*)hp->h_addr, hp->h_length);
    CYGW_UNLOCK;

    server.sin_port = htons(port);
    {
      int retries_on_EAGAIN = 3;
      while ((ret = connect(fd, reinterpret_cast<struct sockaddr *>(&server), sizeof(server))) < 0)
       {
        switch (err_number=errno)
         {
           case ETIMEDOUT:
            cerr << "." << flush; sleep(10); break;
           case ENETUNREACH:
            cerr << "%" << flush; sleep(10); break;
           case EAGAIN:
            if (retries_on_EAGAIN--)
             {
               cerr << "Can't connect socket (" << my_strerror(err_number) << ")" << endl;
               if (err_number==EAGAIN)
                {
                  cerr << "Trying it again in 10 secs..." << endl;
                  sleep(10);
                }
               break;
             }
           default:
            throw unix_buffer_exception(std::string("unix_buffer::constructor: Can't connect socket: ")+my_strerror(err_number));
         }
       }
    }

#else
    // POSIX mode

    // refer "man getaddrinfo" and "man socket" 
    struct addrinfo hints; // our wishes are placed here
    memset(&hints,0,sizeof(hints)); // must be zeroed for default options
    hints.ai_family=PF_INET; // we want IPv4 as protocol
    hints.ai_socktype=SOCK_STREAM; // and we need a stream, not datagram!
    struct addrinfo *addrinfo_res = NULL; // here the result will be delivered
#ifdef WEIRD_VALGRIND_HACK
    static CMutex serial;
    serial.lock();
    // normally (by specification) getaddrinfo is thread-safe!
    // however: under valgrind, multithreaded sessions will hang exactly at this
    // point; probably there are other errors as well; but how to track them, when
    // every session hangs here (without any useful information)?
    // -> with locked access: no error occurs!
    const int retval = getaddrinfo(host.c_str(),NULL,&hints,&addrinfo_res);
    serial.unlock();
#else
    const int retval = getaddrinfo(host.c_str(),NULL,&hints,&addrinfo_res);
#endif
    if ( retval || addrinfo_res==NULL ) // any error?
     {
       cerr << "can't reach " << "\"" <<  host << "\"" << endl;
       cerr << "Error given by getaddrinfo: " << endl;
       cerr << gai_strerror(retval) << endl;
       throw unix_buffer_exception("unix_buffer::constructor: Can't reach "+host+": "+gai_strerror(retval));
     }
    if (addrinfo_res->ai_socktype!=SOCK_STREAM) // we got a "stream"-protocol?
     {
       throw unix_buffer_exception("unix_buffer::constructor: provided protocol doesn't support SOCK_STREAM");
     }
    switch (addrinfo_res->ai_family)
     {
      case PF_INET:
       //cerr << "PF_INET: using IPv4!" << endl;
       fd = socket(AF_INET, SOCK_STREAM, 0);
       reinterpret_cast<sockaddr_in*>(addrinfo_res->ai_addr)->sin_port=htons(port);
       break;
      case PF_INET6:
       cerr << "PF_INET6: try using IPv6 (untested feature!)" << endl;
       fd = socket(AF_INET6, SOCK_STREAM, 0);
       reinterpret_cast<sockaddr_in6*>(addrinfo_res->ai_addr)->sin6_port=htons(port);
       break;
      default:
       throw unix_buffer_exception("unix_buffer::constructor: too bad! ai_family isn't supported by unix_buffer!");
     }

    if (fd < 0) // is the descriptor valid?
     {
       throw unix_buffer_exception("unix_buffer::constructor: Can't open socket");
     }

    {
      int retries_on_EAGAIN = 3;
      while ( connect(fd, addrinfo_res->ai_addr, addrinfo_res->ai_addrlen) <0 )
       {
        PTHREAD_TESTCANCEL;
        switch (int err_number=errno)
         {
           case ETIMEDOUT:
            cerr << "." << flush; sleep(10); break;
           case ENETUNREACH:
            cerr << "%" << flush; sleep(10); break;
           case EAGAIN:
            if (retries_on_EAGAIN--)
             {
               cerr << "Can't connect socket (" << my_strerror(err_number) << ")" << endl;
               if (err_number==EAGAIN)
                {
                  cerr << "Trying it again in 10 secs..." << endl;
                  sleep(10);
                }
               break;
             }
           default:
            throw unix_buffer_exception(std::string("unix_buffer::constructor: Can't connect socket: ")+my_strerror(err_number));
         }
       }
    }
    freeaddrinfo(addrinfo_res);
#endif

    setp(obuffer, obuffer+(buffersize-1));
    setg(ibuffer+pushbacksize, ibuffer+pushbacksize, ibuffer+pushbacksize);
  }

  virtual ~unix_buffer()
   {
     if (fd>=0)
      {
        // valid descriptor -> close the socket
        // but be careful: the socket may be already dead, so we nee to catch exceptions!
        try { sync(); }
        catch (unix_buffer_exception &e)
         {
           cerr << "unix_buffer::destructor caught exception: " << e.what() << endl;
         }
        if (fd_autoclose)
         {
           if (shutdown(fd, SHUT_RDWR)==-1) perror("~unix_buffer():shutdown");
           close(fd);
         }
        fd=-1; // very important! tell the ancestor not to destruct fd twice!
      }
   }

};




class unix_io_stream : public iostream, public Cnetwork_connection_methods
{
protected:
  unix_buffer buf;
  virtual int _get_descriptor() const { return buf.get_descriptor(); } 

public:

  inline int get_descriptor() const { return buf.get_descriptor(); }


  unix_io_stream(const std::string &host, const int port) : iostream(&buf), buf(host, port)
   {
   }


  explicit unix_io_stream(const connection_waiter& cw)
   : iostream(&buf), buf(cw)
   {
   }


  explicit unix_io_stream(const int _fd, const int secs_timeout = 75) : iostream(&buf), buf(_fd, secs_timeout)
   {
   }


  virtual ~unix_io_stream()
   {
   }
};


class fd_iostream : public std::iostream, public Cprovide_descriptor_access
 {
   protected:
    unix_fd_buffer buf;
    virtual int _get_descriptor() const { return buf.get_descriptor(); } 
   public:

    inline int get_descriptor() const { return buf.get_descriptor(); }

    explicit fd_iostream(const int _fd) : std::iostream(&buf), buf(_fd) { }
    virtual ~fd_iostream() { }
 };

class sd_iostream : public fd_iostream, public Cnetwork_connection_methods
 {
   protected:
    virtual int _get_descriptor() const { return buf.get_descriptor(); }
   public:
    explicit sd_iostream(const int _fd) : fd_iostream(_fd) { }
    virtual ~sd_iostream() { }
 };


class socket_piper
{
 private:
  int socketdes[2];

 public:

  socket_piper()
  {
    socketdes[0]=socketdes[1]=-1;
    int retval=socketpair(AF_UNIX, SOCK_STREAM, 0, socketdes); // create a pipe (refer manpage SOCKETPAIR(2))
    if (retval) throw unix_buffer_exception(std::string("socket_piper::constructor: "+my_strerror(errno)));
  }

  ~socket_piper()
  {
    // delete undetached descriptors
    if (socketdes[0]>=0) close(socketdes[0]);
    if (socketdes[1]>=0) close(socketdes[1]);
  }

  std::istream* detach_InPipe(const int bufsize=-1)
  {
    if (socketdes[0]<0) throw unix_buffer_exception("socket_piper::detach_InPipe: no descriptor available!"); 
    sd_iostream *pInPipe = new sd_iostream(socketdes[0]);
    //std::cout << "InPipe bufsize (prev): " << pInPipe->get_RCVBUF() << endl;
    if (bufsize>=0) pInPipe->set_RCVBUF(bufsize);
    //std::cout << "InPipe bufsize (post): " << pInPipe->get_RCVBUF() << endl;
    socketdes[0]=-1; // detach
    return pInPipe;
  }

  std::ostream* detach_OutPipe(const int bufsize=-1)
  {
    if (socketdes[1]<0) throw unix_buffer_exception("socket_piper::detach_OutPipe: no descriptor available!"); 
    sd_iostream *pOutPipe = new sd_iostream(socketdes[1]);
    //std::cout << "OutPipe bufsize (prev): " << pOutPipe->get_SNDBUF() << endl;
    if (bufsize>=0) pOutPipe->set_SNDBUF(bufsize);
    //std::cout << "OutPipe bufsize (post): " << pOutPipe->get_SNDBUF() << endl;
    socketdes[1]=-1; // detach
    return pOutPipe;
  }  
};


#endif

---