buffer.c

用c++包装好的线程库,直接拿来使用,提高效率.

//
// Copyright (C) 2001 謗n E. Hansen
//
// Implementation of buffers, for threads library.

/*
 * There is heavy use of locking, as this buffer is espected to reside
 * between two threads, one doing the writing and the other doing the
 * reading.  To make sure, that operations such as writing and reading
 * characters or strings into the buffer, remain atomic, these routines
 * will lock access to it, during these operations.
 *
 */

#include "buffer.h"
#include "exception.h"

#include <cerrno>

#define MEMORY_PER_PAGE 4096

namespace cpp_threads {

  HeadNode::HeadNode()
  {
    Node::setHeader();
    Mutex::kind(Mutex::fast_e);
  }

  HeadNode::~HeadNode()
  {
  }

  void
  HeadNode::add(Node *elem_p)
  {
    Node *ptr = getEnd();

    if( ptr == 0 )
      ptr = this;
    ptr->AddAfter(elem_p);
  }

  BufElement::BufElement()
  {
    Pthread::debug("BufElement::BufElement()");
    _buf  = "(null)";
    _size = 0;
    _used = 0;
  }

  BufElement::BufElement(char* buf_p,int len_p)
  {
    Pthread::debug("BufElement::BufElement(%p,%d)",buf_p,len_p);
    _buf  = "(null)";
    if( len_p )
      _buf= (char *)malloc(len_p);
    _size = len_p;
    _used = len_p;
    _pos  = 0;
    if( _size > 0 )
      memcpy(_buf,buf_p,len_p);
  }
 
  BufElement::~BufElement()
  {
    Pthread::debug("BufElement::~BufElement %d bytes.",_size);
    if( _size > 0 ) {
      _size = 0;
      free(_buf);
    }
  }

  const char *
  BufElement::pos()
  {
    return &_buf[_pos];
  }

  int
  BufElement::len()
  {
    return _used - _pos;
  }

  void
  BufElement::pack_in()
  {
    if( _used < _size )
      _buf = (char *)realloc( _buf,_used );
    _size = _used;
  }

  void
  BufElement::make_room(size_t len_p)
  {
    if( _used+len_p > _size ) {
      _size += ((len_p/BUFSIZ)+1)*BUFSIZ;
      _buf = (char *)realloc( _buf,_size );
    }
  }

  IOBuffer::IOBuffer()
  {
  }

  IOBuffer::~IOBuffer()
  {
    Node *elem;

    Pthread::debug("IOBuffer::~IOBuffer");
    _head.lock();
    while( (elem = _head.getStart()) )
      delete elem;
    _head.unLock();
    Pthread::debug("-IOBuffer::~IOBuffer");
  }

  void
  IOBuffer::create_new_element(char *buf_p,int len_p)
  {
    _head.lock();
    _head.add(new BufElement(buf_p,len_p));
    _head.unLock();
  }

  bool
  IOBuffer::lock_if_not_empty()
  {
    BufElement *elem;

    _head.lock();
    while( (elem = (BufElement *)_head.getStart()) ) {
      if( elem->_pos >= elem->_used )
	delete elem;
      else
	return true;
    }
    _head.unLock();
    return false;
  }

  bool
  IOBuffer::empty()
  {
    bool rv = true;
    BufElement *elem;

    _head.lock();
    while( rv && (elem = (BufElement *)_head.getStart()) ) {
      if( elem->_pos >= elem->_used )
	delete elem;
      else
	rv = false;
    }
    _head.unLock();
    return rv;
  }

  /*
   * Get one character from the buffer, and advance the buffer
   * read position accordingly.
   *
   * @return 0 if no character available, otherwise the character.
   */
  char
  IOBuffer::getch()
  {
    char ch = '\0';

    if( lock_if_not_empty() ) {
      register BufElement *elem = (BufElement*)_head.getStart();

      ch = elem->_buf[elem->_pos++];
      _head.unLock();
    }
    return ch;
  }

  /*
   * Perform a read operation on the buffer, obtaining one character
   * without advancing the position.
   *
   * @return Character if any available, otherwise 0.
   */
  char
  IOBuffer::peekch()
  {
    char ch = '\0';

    if( lock_if_not_empty() ) {
      register BufElement *elem = (BufElement*)_head.getStart();

      ch = elem->_buf[elem->_pos];
      _head.unLock();
    }
    return ch;
  }

  /*
   * Get data from the buffer.
   *
   * This will take a whole page of data, and write onto a
   * socket.
   *
   * @param s - Socket to write to.
   */
  int
  IOBuffer::get(Socket& sock_p)
  {
    int n = 0;

    _stream_safe.lock();
    if( lock_if_not_empty() ) {
      register BufElement *elem = (BufElement*)_head.getStart();

      n = sock_p.write(elem->pos(),elem->len());
      elem->_pos = elem->_used;
      _head.unLock();
    }
    _stream_safe.unLock();
    return n;
  }

  /*
   * Get data from the buffer.
   *
   * This will take a whole page of data, and write onto a
   * string stream.
   *
   * @param s - String stream.
   */
  int
#if( GCC_VERSION >= 2096 )
  IOBuffer::get(std::stringstream& stream_p)
#else
  IOBuffer::get(std::strstream& stream_p)
#endif
  {
    int n = 0;

    _stream_safe.lock();
    if( lock_if_not_empty() ) {
      register BufElement *elem = (BufElement*)_head.getStart();

      while( elem->len() )
	stream_p << elem->_buf[elem->_pos++],n++;
      _head.unLock();
    }
    _stream_safe.unLock();
    return n;
  }

  /*
   * Put data into the buffer, the data to fill the buffer with
   * originates from a socket.
   *
   * @param s - Socket to read data from.
   */
  int
  IOBuffer::put(Socket& sock_p)
  {
    char *bptr;
    int n = 0;

    bptr = (char *)malloc( MEMORY_PER_PAGE );
    if( !bptr )
      exception::fatal( ENOMEM );
    n = sock_p.read( bptr,MEMORY_PER_PAGE );
    if( n > 0 )
      create_new_element( (char *)realloc(bptr,n),n );
    else
      delete bptr;
    return n;
  }

  /*
   * Fill the buffer with data, the data is to be taken
   * from a string.
   *
   * @param s String to insert into the buffer.
   */
  void
  IOBuffer::put(const std::string& str_p)
  {
    create_new_element( strdup(str_p.c_str()),str_p.length() );
  }

  /*
   * Put a single character into the buffer.
   *
   * @param c Character to insert.
   */
  void
  IOBuffer::put(char ch_p)
  {
    BufElement *elem;

    _head.lock();
    elem = (BufElement*)_head.getEnd();
    if( elem == 0 || elem->_used == elem->_size )
      _head.add( (elem = new BufElement( 0,0 )) );
    elem->make_room(1);
    elem->_buf[elem->_used++] = ch_p;
    if( ch_p == '\0' || ch_p == '\n' )
      elem->pack_in();
    _head.unLock();
  }

  /*
   * Obtain the next character in the buffer.
   *
   */
  char
  IOBuffer::getchar()
  {
    char ch;

    // Make sure we don't steal a character from
    // the middle of a stream.
    _stream_safe.lock();
    ch = getch();
    _stream_safe.unLock();
    return ch;
  }

  /*
   * Get a string of data from the buffer.  The string is assumed
   * to end with a terminating 0 character.  If reading the buffer
   * clears enough space, it is return to the system.
   *
   * @return String of data from the buffer.
   */
  std::string
  IOBuffer::gets()
  {
    char ch;
    std::string s;

    _stream_safe.lock();
    ch = getch();
    while( !isspace(ch) && ch != '\0' ) {
      s += ch;
      ch = getch();
    }
    _stream_safe.unLock();
    return s;
  }

  /*
   * Get a string of data from the buffer.  The string is either
   * terminated with an end of line character, or an end of string
   * character.  If reading the buffer clears enough space, it is
   * returned to the system.
   *
   * @return String of data from the buffer.
   */
  std::string
  IOBuffer::getline(char delim_p)
  {
    char ch;
    std::string s;

    _stream_safe.lock();
    ch = getch();
    while( ch != delim_p && ch != '\0' ) {
      s += ch;
      ch = getch();
    }
    _stream_safe.unLock();
    return s;
  }

  /*
   * Obtain the number of bytes, contained in the buffer.
   *
   * @return Number of bytes in buffer.
   */
  uint
  IOBuffer::count()
  {
    size_t total = 0;
    Node *elem;

    _head.lock();
    for( elem=_head.getStart();elem;elem=elem->getNext() )
      total += ((BufElement*)elem)->_used;
    _head.unLock();
    return total;
  }

  IOBuffer&
  IOBuffer::operator <<(Socket& sock_p)
  {
    put(sock_p);
    return *this;
  }

  IOBuffer&
  IOBuffer::operator >>(Socket& sock_p)
  {
    std::string s;

    s = gets();
    sock_p.write( s.c_str(),s.length() );
    return *this;
  }

  IOBuffer&
  IOBuffer::operator >>(std::string& str_p)
  {

    str_p = gets();
    return *this;
  }

#if( GCC_VERSION >= 2096 )
  IOBuffer&
  IOBuffer::operator >>(std::stringstream& sstr_p)
#else
  IOBuffer&
  IOBuffer::operator >>(std::strstream& sstr_p)
#endif
  {
    sstr_p << gets();
    return *this;
  }

  IOBuffer&
  IOBuffer::operator <<(std::string& str_p)
  {
    put(str_p);
    return *this;
  }

#if( GCC_VERSION >= 2096 )
  IOBuffer&
  IOBuffer::operator <<(std::stringstream& sstr_p)
#else
  IOBuffer&
  IOBuffer::operator <<(std::strstream& sstr_p)
#endif
  {
    std::string s;

    sstr_p >> s;
    put(s);
    return *this;
  }

  IOBuffer&
  IOBuffer::operator >>(int& val_p)
  {
    bool ok = true;
    int val = 0;
    char ch;
    std::string s;

    _stream_safe.lock();
    while( ok ) {
      ch = peekch();
      ok = isdigit(ch);
      if( ok ) {
	ch = getch();
	val = (val * 10) + (ch - '0');
      }
    }
    _stream_safe.unLock();
    return *this;
  }

  IOBuffer&
  IOBuffer::operator <<(int& val_p)
  {
#if( GCC_VERSION>=2096 )
    std::stringstream s;
#else
    std::strstream s;
#endif

    s << val_p << std::ends;
    put( s.str() );
    return *this;
  }

  IOBuffer&
  IOBuffer::operator <<(double& val_p)
  {
#if( GCC_VERSION>=2096 )
    std::stringstream s;
#else
    std::strstream s;
#endif

    s << val_p << std::ends;
    put( s.str() );
    return *this;
  }

  IOBuffer&
  IOBuffer::operator >>(double& val_p)
  {
    char ch;
    bool ok = true;
    char *endp;
    std::string s;

    _stream_safe.lock();
    while( ok ) {
      ch = peekch();
      ok = isdigit(ch) || ispunct(ch);
      if( ok )
	s += getch();
    }
    val_p = strtod( s.c_str(),&endp );
    _stream_safe.unLock();
    return *this;
  }

}; // namespace