[libasyncio] / asyncio / async_io.cpp

/** @file
 *
 *
 * @copyright Copyright © 2007-2008 by Intra2net AG
 * @license LGPL
 * @contact info@intra2net.com
 */

//#define NOISEDEBUG

#include "async_io.hpp"

#include <list>
#include <vector>
#include <map>
#include <algorithm>
#include <utility>

#include <sys/poll.h>
#include <sys/time.h>
#include <sys/socket.h>
#include <unistd.h>
#include <errno.h>
#include <fcntl.h>

#include <boost/bind.hpp>

#include <signalfunc.hpp>
#include <timefunc.hxx>

#include <iostream>

#ifdef NOISEDEBUG
#include <iostream>
#include <iomanip>
#define DOUT(msg) std::cout << msg << std::endl
#define FODOUT(obj,msg) std::cout << typeid(*obj).name() << "[" << obj << "]:" << msg << std::endl
//#define ODOUT(msg) std::cout << typeid(*this).name() << "[" << this << "]:" << msg << std::endl
#define ODOUT(msg) std::cout << __PRETTY_FUNCTION__ << "[" << this << "]:" << msg << std::endl
#else
#define DOUT(msg) do {} while (0)
#define FODOUT(obj,msg) do {} while (0)
#define ODOUT(msg) do {} while (0)
#endif


namespace
{

using namespace AsyncIo::Utils;

/*
 * configuration:
 */


const int c_max_poll_wait= 10*60*1000; // maximal poll wait (while in backend loop): 10 min


/**
 * contains internal helper structs and functions for io handling.
 */
namespace internal_io
{

/**
 * extends struct pollfd with some convenience functions
 */
struct PollFd : public ::pollfd
{
    PollFd()
    {
        fd= 0;
        events= revents= 0;
    } // eo PollFd


    /**
     * initializes the struct with a given file descriptor and clears the event mask(s).
     * @param _fd 
     */
    PollFd(int _fd)
    {
        fd= _fd;
        events= revents= 0;
    } // eo PollFd


    /**
     * set that we want to be notified about new incoming data
     */
    void setPOLLIN()  { events |= POLLIN; }

    /**
     * set that we want to be notified if we can send (more) data.
     */
    void setPOLLOUT() { events |= POLLOUT; }

}; // eo struct PollFd


typedef std::vector<PollFd> PollVector;
typedef std::map<int,PollVector::size_type> FdPollMap;
typedef std::map<int,AsyncIo::IOImplementation*> FdIOMap;


/**
 * struct for interfacing our local structures with poll()
 */
struct PollDataCluster
{
    PollVector m_poll_vector;
    FdPollMap  m_fd_poll_map;
    FdIOMap    m_read_fd_io_map;
    FdIOMap    m_write_fd_io_map;

    void add_read_fd( int fd, AsyncIo::IOImplementation* io);
    void add_write_fd( int fd, AsyncIo::IOImplementation* io);

    pollfd* get_pollfd_ptr();
    unsigned int get_num_pollfds() const;

}; // eo struct PollDataCluster


typedef PtrList< AsyncIo::IOImplementation, true > IOList;
typedef PtrList< AsyncIo::TimerBase, true >        TimerList;

template<> int IOList::GlobalCountType::InstanceCount= 0;
template<> int TimerList::GlobalCountType::InstanceCount= 0;


/**
 * the (internal) global list of io objects (object pointers)
 */
IOList& g_io_list()
{
    static IOList _the_io_list;
    return _the_io_list;
};


/**
 * the (internal) global list of timer objects (object pointers)
 */
TimerList& g_timer_list()
{
    static TimerList _the_timer_list;
    return _the_timer_list;
}

/*
 * implementation of PollDataCluster
 */


/**
 * add a new file descriptor to the read list.
 *
 * @param fd the file descriptor.
 * @param io the io object which uses the fd for reading.
 */
void PollDataCluster::add_read_fd( int fd, AsyncIo::IOImplementation* io)
{
    FdPollMap::iterator itPollMap = m_fd_poll_map.find(fd);
    if (itPollMap != m_fd_poll_map.end())
    {
        m_poll_vector[ itPollMap->second ].setPOLLIN();
    }
    else
    {
        PollFd item(fd);
        item.setPOLLIN();
        m_fd_poll_map[fd] = m_poll_vector.size();
        m_poll_vector.push_back( item );
    }
    m_read_fd_io_map[fd]= io;
} // eo PollDataCluster::add_read_fd


/**
 * add a new file descriptor to the write list.
 *
 * @param fd the file descriptor.
 * @param io the io object which uses the fd for writing.
 */
void PollDataCluster::add_write_fd( int fd, AsyncIo::IOImplementation* io)
{
    FdPollMap::iterator itPollMap = m_fd_poll_map.find(fd);
    if (itPollMap != m_fd_poll_map.end())
    {
        m_poll_vector[ itPollMap->second ].setPOLLOUT();
    }
    else
    {
        PollFd item(fd);
        item.setPOLLOUT();
        m_fd_poll_map[fd] = m_poll_vector.size();
        m_poll_vector.push_back( item );
    }
    m_write_fd_io_map[fd]= io;
} // eo PollDataCluster::add_write_fd


/**
 * returns a pointer to a pollfd array; suitable for passing to poll()
 *
 * @return pointer to pollfd array
 */
pollfd* PollDataCluster::get_pollfd_ptr()
{
    return m_poll_vector.empty() ? NULL : &m_poll_vector.front();
} // eo get_pollfd_ptr


/**
 * returns the number of entries in the pollfd array; suitable for passing to poll()
 *
 * @return the number of entries in the pollfd array
 */
unsigned int PollDataCluster::get_num_pollfds() const
{
    return m_poll_vector.size();
} // eo get_num_pollfds



} // eo namespace internal_io


/*
 * some internal tool functions and structures
 */

struct FilterMatch {
    AsyncIo::FilterBasePtr m_filter;

    FilterMatch(AsyncIo::FilterBasePtr filter)
    : m_filter(filter)
    {}

    bool operator () (const AsyncIo::FilterBasePtr& item)
    {
        return item && item == m_filter;
    }

}; // eo struct FilterMatch



} // eo anonymous namespace




namespace AsyncIo
{


/*
 * implementation of TimerBase
 */

/**
 * constructor. Adds the object to the internal timer list.
 */
TimerBase::TimerBase()
: m_active(false)
, m_marked(false)
{
    internal_io::g_timer_list().add_item(this);
} // eo TimerBase::TimerBase


/**
 * destructor. Removes the object from the internal timer list.
 */
TimerBase::~TimerBase()
{
    ODOUT("enter");
    if (internal_io::TimerList::Instances())
    {
        ODOUT("remove from list");
        internal_io::g_timer_list().remove_item(this);
    }
} // eo TimerBase::~TimerBase


/**
 * @brief returns the point in time when the time is executed in real time.
 * @return the point in time when the timer is to be executed.
 */
MilliTime TimerBase::getRealWhenTime() const
{
    MilliTime mono_time;
    MilliTime real_time;
    get_current_monotonic_time(mono_time);
    get_current_real_time(real_time);
    MilliTime result= m_when - mono_time + real_time;
    return result;
} // eo TimerBase::getRealWhenTime() const


/**
 * sets the time when the event should be executed.
 * @param sec the seconds part of the point in time.
 * @param msec  the milliseconds part of the point in time.
 */
void TimerBase::setWhenTime(long sec, long msec)
{
    m_when.set(sec,msec);
    m_marked= false;
} // eo TimerBase::setWhenTime


/**
 * sets the time when the event should be executed.
 * @param mt the point in time.
 */
void TimerBase::setWhenTime(const MilliTime& mt)
{
    m_when= mt;
    m_marked= false;
} // eo TimerBase::setWhenTime


/**
 * sets the time delta measured from current time when the event should be executed.
 * @param sec the seconds of the time delta
 * @param msec the milli seconds of the time delta
 */
void TimerBase::setDeltaWhenTime(long sec, long msec)
{
    setDeltaWhenTime( MilliTime(sec,msec) );
} // eo TimerBase::setWhenTime



/**
 * sets the time delta measured from current time when the event should be executed.
 * @param mt the time delta
 */
void TimerBase::setDeltaWhenTime(const MilliTime& mt)
{
    get_current_monotonic_time(m_when);
    m_when+= mt;
    m_marked= false;
} // eo TimerBase::setWhenTime


/**
 * set the active state of the timer event.
 * @param active determines if the object should be active (default: yes).
 */
void TimerBase::activate(bool active)
{
    m_active = active;
    if (!active)
    {
        // clear the mark if we are not active.
        m_marked= false;
    }
} // eo TimerBase::activate


/** @fn void TimerBase::deactivate()
 *  deactivates the event by clearing the active state.
 */


/**
 * called when the timer event occured.
 */
void TimerBase::execute()
{
} // eo TimerBase::execute


/*
 * implementation of FilterBase class
 */


FilterBase::FilterBase()
: m_io(NULL)
{
} // eo FilterBase::FilterBase()


/**
 * injects incoming data.
 * @param data  the new data
 */
void FilterBase::injectIncomingData(const std::string& data)
{
    if (m_io)
    {
        FilterBasePtr ptr= get_ptr_as< FilterBase >();
        if (ptr)
        {
            m_io->injectIncomingData(ptr,data);
        }
    }
} // FilterBase::injectIncomingData(const std::string&)


/**
 * injects outgoing data.
 * @param data the new data
 */
void FilterBase::injectOutgoingData(const std::string& data)
{
    if (m_io)
    {
        FilterBasePtr ptr= get_ptr_as< FilterBase >();
        if (ptr)
        {
            m_io->injectOutgoingData(ptr,data);
        }
    }
} // eo FilterBase::injectOutgoingData(const std::string&)



/**
 * called when EOF detected on incoming channel (or incoming channel closed)
 */
void FilterBase::endOfIncomingData()
{
} // eo FilterBase::endOfIncomingData()

/**
 * called when the filter should reset.
 * This is used when a new channel is opened or when the filter is taken out of a filter chain.
 */
void FilterBase::reset()
{
} // eo FilterBase::reset()


/*
 * implementation of IOImplementation class
 */


/**
 * constructor for the base io class.
 *
 * Also adds the object to internal list of io objects (which is used by the backend).
 *
 * @param read_fd the file descriptor which should be used for reading (default -1 for no value)
 * @param write_fd  the file descriptor which should be used for writing (default -1 for no value)
 */
IOImplementation::IOImplementation(int read_fd, int write_fd)
: m_read_fd(-1)
, m_write_fd(-1)
, m_eof(false)
, m_not_writable(false)
, m_input_buffer()
, m_output_buffer()
, m_marked_for_reading(false)
, m_marked_for_writing(false)
{
    internal_io::g_io_list().add_item(this);
    if (read_fd >= 0)
    {
        setReadFd( read_fd );
    }
    if (write_fd >= 0)
    {
        setWriteFd( write_fd );
    }
} // eo IOImplementation::IOImplementation


/**
 * destructor of the base io class.
 *
 * Removes the object from the interal list of io objects.
 */
IOImplementation::~IOImplementation()
{
    close();
    if (internal_io::IOList::Instances())
    {
        internal_io::g_io_list().remove_item(this);
    }
    // now clear the filters:
    while (! m_filter_chain.empty() )
    {
        FilterChain::iterator it = m_filter_chain.begin();
        (*it)->reset();
        (*it)->m_io= NULL;
        //TODO: signal the filter that it is removed ?!
        m_filter_chain.erase(it);
    }
} // eo IOImplementation::~IOImplementation


/**
 * adds another filter to the filter chain.
 * @param filter pointer to the new filter.
 */
void IOImplementation::addFilter
(
    FilterBasePtr filter
)
{
    if (!filter)
    {
        return; // nothing to do
    }
    if (filter->m_io)
    {
        filter->m_io->removeFilter(filter);
    }
    m_filter_chain.push_back( filter );
} // eo IOImplementation::addFilter


/**
 * removes a filter from the filter chain.
 * @param filter the pointer to the filter which is removed.
 * @note if the filter is removed the class gives away the ownership; i.,e. the caller is responsible for
 * deleting the filter if it was dynamically allocated.
 */
void IOImplementation::removeFilter
(
    FilterBasePtr filter
)
{
    FilterChain::iterator it =
        std::find_if( m_filter_chain.begin(), m_filter_chain.end(), FilterMatch(filter) );
    if (it != m_filter_chain.end())
    {
        filter->reset();
        filter->m_io= NULL;
        //TODO: signal the filter that it is removed ?!
        m_filter_chain.erase(it);
    }
} // eo IOImplementation::removeFilter


/**
 * closes the file descriptors (/ the connection).
 *
 * @param direction the direction which should be closed (default: @a Direction::both for all).
 */
void IOImplementation::close(Direction direction)
{
    bool had_read_fd= (m_read_fd >= 0);
    m_errno= 0;
    if (direction == Direction::unspecified) direction= Direction::both;
    if (direction != Direction::both && m_read_fd==m_write_fd && m_read_fd>=0)
    { // special case: half closed (socket) connections...
        // NOTE: for file descriptors m_errno will set to ENOTSOCK, but since we "forget" the desired part
        // (read_fd or write_fd) this class works as desired.
        switch(direction)
        {
            case Direction::in:
                {
                    int res= ::shutdown(m_read_fd, SHUT_RD);
                    if (res<0) 
                    {
                        m_errno= errno;
                    }
                    m_read_fd= -1;
                    if (!m_eof)
                    {
                        for(FilterChain::iterator it= m_filter_chain.begin();
                            it != m_filter_chain.end();
                            ++it)
                        {
                            (*it)->endOfIncomingData();
                        }
                    }
                }
                return;

            case Direction::out:
                {
                    int res= ::shutdown(m_write_fd, SHUT_WR);
                    if (res<0)
                    {
                        m_errno= errno;
                    }
                    m_write_fd= -1;
                    m_output_buffer.clear();
                }
                return;
        }
    }
    if (m_write_fd >= 0 && (direction & Direction::out) )
    {
        int res1 = ::close(m_write_fd);
        if (m_write_fd == m_read_fd)
        {
            m_read_fd= -1;
        }
        m_write_fd= -1;
        m_output_buffer.clear();
        if (res1<0) m_errno= errno;
    }
    if (m_read_fd >=0 && (direction & Direction::in) )
    {
        int res1 = ::close(m_read_fd);
        m_read_fd= -1;
        if (res1<0) m_errno= errno;
    }
    if (had_read_fd && !m_eof && (m_read_fd<0))
    {
        for(FilterChain::iterator it= m_filter_chain.begin();
            it != m_filter_chain.end();
            ++it)
        {
            (*it)->endOfIncomingData();
        }
    }
} // eo IOImplementation::close


/**
 * determines if the io class wants to read data.
 * Default implementation checks only for a valid file descriptor value.
 *
 * @return @a true if the objects wants to read data
 */
bool IOImplementation::wantRead()
{
    return (m_read_fd >= 0) && ! m_eof;
} // eo IOImplementation::wantRead


/**
 * determines if the io class wants to write data.
 * Default implementation checks for a valid file descriptor value and if the object
 * cannot write data immediately.
 *
 * @return @a true if the objects wants to write data
 */
bool IOImplementation::wantWrite()
{
    return (m_write_fd >= 0) && ! m_marked_for_writing && ! m_not_writable;
} // eo IOImplementation::wantWrite


/**
 * delivers if opened.
 * The default returns @a true if at least one file descriptor (read or write) is valid.
 * @return @a true if opened.
 */
bool IOImplementation::opened() const
{
    return (m_read_fd>=0) || (m_write_fd>=0);
} // eo IOImplementation::opened() const


/**
 * returns if the read side detected an end of file (EOF).
 * @return @a true if end of file was detected on read file descriptor (or read file descriptor isn't valid).
 */
bool IOImplementation::eof() const
{
    return (m_read_fd < 0) || m_eof;
} // eo IOImplementatio::eof() const


/**
 * @brief returns of the write side didn't detect that it cannot write.
 * @return @a true if we can write.
 */
bool IOImplementation::writable() const
{
    return (m_write_fd >=0 ) and not m_not_writable;
} // eo IOImplementation::writable() const


/**
 * returns if the output buffer is empty.
 * @return 
 */
bool IOImplementation::empty() const
{
    return m_output_buffer.empty();
} // eo IOImplementation::empty

/**
 * puts data into the output buffer and sends it immediately if possible,
 *
 * The data is passed through the filter chain before it's stored in the output buffer
 * (i.e. the output buffer contains data as it should be send directly to the descriptor).
 * @param _data the data which should be send.
 */
void IOImplementation::lowSend(const std::string& _data)
{
    std::string data(_data);

    for(FilterChain::reverse_iterator it_filter= m_filter_chain.rbegin();
        it_filter!= m_filter_chain.rend();
        ++it_filter)
    {
        data= (*it_filter)->filterOutgoingData(data);
    }
    m_output_buffer+= data;

    // if we can send immediately, do it:
    if (! m_output_buffer.empty() && m_marked_for_writing)
    {
        doWrite();
    }
} // eo IOImplementation::lowSend


/**
 * called by the backend when there is data to read for this object.
 *
 * Reads the data from the connection (read file descriptor) and passes the data through the filter chain.
 * The final data is appended to the input buffer and the signal @a m_signal_read() is called.
 *
 * If EOF is detected (i,e, no data was received) then the signal @a m_signal_eof() is called.
 *
 * @note overload this method only when You know what You are doing!
 * (overloading is necessary when handling server sockets.)
 */
void IOImplementation::doRead()
{
    // static read buffer; should be ok as long as we don't use threads
    static char buffer[8*1024]; // 8 KiB

    m_errno = 0;
    if (m_read_fd<0 || !m_marked_for_reading)
    {
        ODOUT("exit0; read_fd="<<m_read_fd << " mark=" << m_marked_for_reading);
        return;
    }

    // reset the mark:
    m_marked_for_reading = false;

    // now read the data:
    ssize_t count;
    count = ::read(m_read_fd, buffer, sizeof(buffer));

    ODOUT("::read -> " << count);

    // interpret what we got:
    if (count < 0) // error
    {
        m_errno = errno;
        int fd= m_read_fd;

        switch(m_errno)
        {
            case EINVAL:
            case EBADF:
            case ECONNRESET:
            case ENETRESET:
                if (fd == m_read_fd)
                {
                    close( (m_read_fd == m_write_fd) ? Direction::both : Direction::in );
                }
                break;
        }
    }
    else if (count==0) // EOF
    {
        // remember the read fd:
        int fd = m_read_fd;
        // remember the EOF:
        m_eof= true;
        // signal EOF
        m_signal_eof();
        // if the fd is still the same: close it.
        if (fd == m_read_fd)
        {
            close( Direction::in );
        }
    }
    else // we have valid data
    {
        std::string data(buffer,count);
        ODOUT(" got \"" << data << "\"");
        for(FilterChain::iterator it_filter= m_filter_chain.begin();
            it_filter != m_filter_chain.end();
            ++it_filter)
        {
            data= (*it_filter)->filterIncomingData(data);
        }
        m_input_buffer+= data;
        m_signal_read();
    }
} // eo IOImplementation::doRead


/**
 * interface for filter classes to inject data into the filter chain (emulating new incoming data).
 * @param from_filter the filter which injects the new data.
 * @param _data  the new data.
 */
void IOImplementation::injectIncomingData(FilterBasePtr from_filter, const std::string& _data)
{
    FilterChain::iterator it_filter =
        std::find_if( m_filter_chain.begin(), m_filter_chain.end(), FilterMatch(from_filter) );
    if (it_filter == m_filter_chain.end())
    {
        // dont accept data inject from a unknown filter
        return;
    }
    // well: pass the data through the remaining filters:
    // NOTE: processing is (nearly) the same as in IOImplementation::doRead()
    std::string data(_data);
    for(++it_filter;
        it_filter != m_filter_chain.end();
        ++it_filter)
    {
        data= (*it_filter)->filterIncomingData(data);
    }
    m_input_buffer+= data;
    m_signal_read();
} // eo IOImplementation::injectIncomingData(FilterBase*,const std::string&)


/**
 * interface for filter classes to inject data into the filter chain (emulating new outgoing data).
 * @param from_filter the filter which injects the new data.
 * @param _data  the new data.
 */
void IOImplementation::injectOutgoingData(FilterBasePtr from_filter, const std::string& _data)
{
    FilterChain::reverse_iterator it_filter =
        std::find_if( m_filter_chain.rbegin(), m_filter_chain.rend(), FilterMatch(from_filter) );
    if (it_filter == m_filter_chain.rend())
    {
        // dont accept data inject from a unknown filter
        return;
    }
    // well: pass the data through the remaining filters:
    // NOTE: processing is (nearly) the same as in IOImplementation::lowSend()
    std::string data(_data);
    for(++it_filter;
        it_filter!= m_filter_chain.rend();
        ++it_filter)
    {
        data= (*it_filter)->filterOutgoingData(data);
    }
    m_output_buffer+= data;

    // if we can send immediately, do it:
    if (! m_output_buffer.empty() && m_marked_for_writing)
    {
        doWrite();
    }
} // eo IOImplementation::injectOutgoingData(FilterBase*,const std::string&)


/**
 * set the read file descriptor.
 * Although a derived class can also set the read fd directly; this method should be used
 * for this task since it updates some flags on the fd for async operation.
 * @param fd the new read file descriptor.
 */
void IOImplementation::setReadFd(int fd)
{
    // test if we already have a valid descriptor (and may have to close it):
    if (m_read_fd >=0 )
    {
        if (m_read_fd == fd)
        {
            // fd was already right; consider it to be ok.
            return;
        }
        close(Direction::in);
    }
    // reset our errno:
    m_errno= 0;
    // if the new descriptor looks valid, set some flags:
    if (fd >= 0)
    {
        long flags= ::fcntl(fd, F_GETFL);
        if (flags != -1)
        {
            // set the flags for non blocking, async operation
            flags |= O_NONBLOCK|O_ASYNC;
            ::fcntl(fd,F_SETFL, flags);
        }
        else if ( errno == EBADF )
        {
            // well, we seemed to be fed with an invalid descriptor...:
            m_errno = errno;
            fd= -1;
        }
    }
    if (fd >= 0) // if still valid:
    {
        // set the close-on-exec flag
        ::fcntl(fd,F_SETFD, FD_CLOEXEC);
    }
    m_read_fd= fd;
    m_marked_for_reading= false;
    m_eof= false;
} // eo IOImplementation::setReadFd(int)



/**
 * set the write file descriptor.
 * Although a derived class can also set the write fd directly; this method should be used
 * for this task since it updates some flags on the fd for async operation.
 * @param fd the new write file descriptor.
 */
void IOImplementation::setWriteFd(int fd)
{
    if (m_write_fd >=0 )
    {
        if (m_write_fd == fd)
        {
            // fd was already right; consider it to be ok.
            return;
        }
        close(Direction::out);
    }
    // reset our errno:
    m_errno= 0;
    // if the new descriptor looks valid, set some flags:
    if (fd >= 0)
    {
        long flags= ::fcntl(fd, F_GETFL);
        if (flags != -1)
        {
            // set the flags for non blocking, async operation
            flags |= O_NONBLOCK|O_ASYNC;
            ::fcntl(fd,F_SETFL, flags);
        }
        else if (errno == EBADF)
        {
            // well, we seemed to be fed with an invalid descriptor...:
            m_errno = errno;
            fd= -1;
        }
    }
    if (fd >= 0) // if still valid:
    {
        // set the close-on-exec flag
        ::fcntl(fd,F_SETFD, FD_CLOEXEC);
    }
    m_write_fd = fd;
    m_marked_for_writing= false;
    m_not_writable= false;
} // eo IOImplementation::setWriteFd(int)



/**
 * called by the backend when this object can write data.
 *
 * If some data was sended, the signal @a m_signal_write is called.
 *
 * @internal tries to write all buffered data to output; if this succeeds,
 * the connection is assumed to be still able to accept more data.
 * (i.e. the internal write mark is kept!)
 *
  * @note overload this method only when You know what You are doing!
*/
void IOImplementation::doWrite()
{
    m_errno = 0;
    if ( m_write_fd<0 || !m_marked_for_writing || m_output_buffer.empty())
    {
        return;
    }

    ODOUT("doWrite, d=\"" << m_output_buffer << "\"");

    //reset mark:
    m_marked_for_writing= false;

    // now write the data
    ssize_t count= ::write( m_write_fd, m_output_buffer.data(), m_output_buffer.size());

    ODOUT("::write -> " << count);

    if (count < 0) // error
    {
        m_errno= errno;
        int fd= m_write_fd;

        switch(m_errno)
        {
            case EPIPE:
                m_not_writable= true;
                // emit a signal
                m_signal_not_writable();
                // fall through
            case EINVAL:
            case EBADF:
            case ECONNRESET:
            case ENETRESET:
                if (fd == m_write_fd)
                {
                    close( (m_write_fd == m_read_fd) ? Direction::both : Direction::out );
                }
                break;
        }
    }
    else
    {
        m_output_buffer.erase(0, count);
        if (m_output_buffer.empty())
        {
            // special case: if we were able to send all the data, we keep the write mark:
            m_marked_for_writing= true;
        }
    }
    if (count > 0)
    {
        m_signal_write();
    }
} // eo IOImplementation::doWrite


/*
 * implementation of SimpleIO
 */


SimpleIO::SimpleIO(int read_fd, int write_fd)
: inherited(read_fd, write_fd)
{
    m_signal_read.connect(boost::bind(&SimpleIO::slotReceived,this));
} // eo SimpleIO::SimpleIO()


SimpleIO::~SimpleIO()
{
} // eo SimpleIO::~SimpleIO()


/**
 * sends a string.
 * @param data the string.
 */
void SimpleIO::sendString(const std::string& data)
{
    lowSend(data);
} // eo SimpleIO::sendString(const std::string&)


/**
 * emits the signal signalReceived with the received data.
 * This slot is connected to IOImplementation::m_signal_read.
 */
void SimpleIO::slotReceived()
{
    std::string data;
    data.swap(m_input_buffer);
    signal_received_string(data);
} // eo SimpleIO::slotReceived()



/*
 * implementation of SimpleIO2
 */


SimpleIO2::SimpleIO2(int read_fd, int write_fd)
: inherited(read_fd, write_fd)
{
    m_signal_read.connect(boost::bind(&SimpleIO2::slotReceived,this));
} // eo SimpleIO2::SimpleIO2()


SimpleIO2::~SimpleIO2()
{
} // eo SimpleIO2::~SimpleIO2()


/**
 * sends a string.
 * @param data the string.
 */
void SimpleIO2::sendString(const std::string& data)
{
    lowSend(data);
} // eo SimpleIO2::sendString(const std::string&)


/**
 * emits the signal signalReceived with the received data.
 * This slot is connected to IOImplementation::m_signal_read.
 */
void SimpleIO2::slotReceived()
{
    std::string data;
    data.swap(m_input_buffer);
    signal_received_string(data);
} // eo SimpleIO2::slotReceived()



/*
 * implementation of class Backend (singleton)
 */

Backend* Backend::g_backend= NULL;

int Backend::m_count_active_steps=0;


Backend::Backend()
: m_count_active_loops(0)
, m_count_stop_requests(0)
{
    SystemTools::ignore_signal( SystemTools::Signal::PIPE );
} // eo Backend::Backend


Backend::~Backend()
{
    SystemTools::restore_signal_handler( SystemTools::Signal::PIPE );
} // eo Backend::~Backend()

/**
 * delivers pointer to the current backend, instantiating a new backend if there was no current one.
 *
 * This should be the only way to access the backend which should be a singleton.
 *
 * @return the pointer to the current backend.
 */
Backend* Backend::getBackend()
{
    if (!g_backend)
    {
        g_backend = new Backend();
    }
    return g_backend;
} // eo Backend::getBackend




/**
 * performs one backend cycle.
 *
 * Collects all file descriptors from the active io objects which should be selected for reading and/or writing.
 * Also determines the timer events which become due and adjusts the timeout.
 * Constructs the necessary structures and calls poll().
 * Finally interprets the results from poll() (i.e. performs the reading/writing/timer events)
 *
 * @param timeout maximal wait value in milliseconds; negative value waits until at least one event occured.
 * @return @a true if there was at least one active object; otherwise @a false
 *
 * @note this method is a little beast.
 *
 * @internal 
 * The cycle is divided into four steps: collecting; poll; mark and execute.
 * The "mark" step is necessary to avoid some bad side effects when method calls in the execution stage
 * are calling @a Backup::doOneStep or open their own local backend loop.
 *
 * @todo handle some more error cases.
 * @todo provide a plugin interface for external handler.
 * (currently inclusion of external handler is possible by (ab)using timer classes)
 */
bool Backend::doOneStep(int timeout)
{
    ODOUT( "timeout=" << timeout );
    internal_io::PollDataCluster poll_data;
    bool had_active_object = false;

    ++m_count_active_steps;

    try {

        FdSetType local_read_fds;
        FdSetType local_write_fds;

        // step 1 ; collect

        { // step 1.1: collect fds for read/write operations
            for(internal_io::IOList::iterator itIOList = internal_io::g_io_list().begin();
                itIOList != internal_io::g_io_list().end();
                ++itIOList)
            {
                if (! *itIOList) continue; // skip NULL entries
                int  read_fd  = (*itIOList)->m_read_fd;
                int  write_fd = (*itIOList)->m_write_fd;
                bool want_read  = (read_fd >= 0) and (*itIOList)->wantRead();
                bool want_write = (write_fd >= 0) and (*itIOList)->wantWrite();
                if (read_fd >= 0 )
                {
                    local_read_fds.insert( read_fd );
                }
                if (write_fd >= 0)
                {
                    local_write_fds.insert( write_fd );
                }
                if (!want_read && !want_write) continue;
                if (want_read)
                {
                    FODOUT( (*itIOList), "wants to read  (fd=" << read_fd << ")");
                    poll_data.add_read_fd(read_fd, *itIOList);
                }
                if (want_write)
                {
                    FODOUT( (*itIOList), "wants to write  (fd=" << write_fd << ")");
                    poll_data.add_write_fd(write_fd, *itIOList);
                }
                had_active_object= true;
            }
        }

        { // step 1.2: collect timer events
            MilliTime current_time;
            MilliTime min_event_time;

            get_current_monotonic_time(current_time);
            bool min_event_time_set;

            if (timeout >= 0)
            {
                min_event_time = current_time + MilliTime(0,timeout);
                min_event_time_set= true;
            }
            else
            {
                min_event_time = current_time + MilliTime(86400,0);
                min_event_time_set= false;
            }
            // TODO

            for(internal_io::TimerList::iterator it_timer= internal_io::g_timer_list().begin();
                it_timer != internal_io::g_timer_list().end() 
                && (!had_active_object || !min_event_time_set || current_time < min_event_time);
                ++ it_timer)
            {
                if (! *it_timer) continue; // skip NULL entries
                if (! (*it_timer)->m_active) continue; // skip if not enabled
                if ( !min_event_time_set || (*it_timer)->m_when < min_event_time)
                {
                    min_event_time = (*it_timer)->m_when;
                    min_event_time_set= true;
                }
                had_active_object= true;
            }

            if (min_event_time_set)
            { // we have at a minimal event time, so (re)compute the timeout value:
                MilliTime delta= (min_event_time - current_time);
                long long delta_ms = std::min( delta.get_milliseconds(), 21600000LL); // max 6h
                if (delta_ms <= 0L)
                {
                    timeout= 0L;
                }
                else
                {
                    timeout= delta_ms + (delta_ms<5 ? 1 : 3);
                }
            }
        }

        // step 2 : poll
        ODOUT(" poll timeout is " << timeout);
        {
            MilliTime current_time;
            get_current_monotonic_time(current_time);
            ODOUT("  current time is sec="<<current_time.mt_sec << ", msec=" << current_time.mt_msec);
        }
        int poll_result= ::poll(poll_data.get_pollfd_ptr(), poll_data.get_num_pollfds(), timeout);

        ODOUT("poll -> " << poll_result);
        {
            MilliTime current_time;
            get_current_monotonic_time(current_time);
            ODOUT("  current time is sec="<<current_time.mt_sec << ", msec=" << current_time.mt_msec);
        }

        if (poll_result < 0)
        {
            //TODO poll error handling (signals ?!)
        }

        // step 3 : mark

        // step 3.1: mark io objects (if necessary)
        if (poll_result > 0)
        {
            for(internal_io::PollVector::iterator itPollItem = poll_data.m_poll_vector.begin();
                itPollItem != poll_data.m_poll_vector.end();
                ++itPollItem)
            {
                ODOUT("  fd=" << itPollItem->fd << ", events=" << itPollItem->events << ", revents=" << itPollItem->revents);
                if ( 0 == (itPollItem->revents))
                { // preliminary continuation if nothing is to handle for this item(/fd)...
                    continue;
                }
                if ( 0!= (itPollItem->revents & (POLLIN|POLLHUP)))
                {
                    IOImplementation *io= poll_data.m_read_fd_io_map[ itPollItem->fd ];
                    if (io && io->m_read_fd==itPollItem->fd)
                    {
                        FODOUT(io,"marked for reading");
                        io->m_marked_for_reading= true;
                    }
                }
                if ( 0!= (itPollItem->revents & POLLOUT))
                {
                    IOImplementation *io= poll_data.m_write_fd_io_map[ itPollItem->fd ];
                    if (io && io->m_write_fd==itPollItem->fd)
                    {
                        io->m_marked_for_writing= true;
                    }
                }
                if ( 0!= (itPollItem->revents & POLLERR))
                {
                    IOImplementation *io= poll_data.m_write_fd_io_map[ itPollItem->fd ];
                    if (0!= (itPollItem->events & POLLOUT))
                    {
                        if (io && io->m_write_fd==itPollItem->fd)
                        {
                            io->m_marked_for_writing= false;
                            //io->close( Direction::out );
                        }
                    }
                }
                // TODO error handling (POLLERR, POLLHUP, POLLNVAL)
            }
        }

        //Step 3.2: mark timer objects
        {
            MilliTime current_time;

            get_current_monotonic_time(current_time);
            ODOUT("  current time is sec="<<current_time.mt_sec << ", msec=" << current_time.mt_msec);

            for(internal_io::TimerList::iterator it_timer= internal_io::g_timer_list().begin();
                it_timer != internal_io::g_timer_list().end();
                ++ it_timer)
            {
                ODOUT("  check timer " << *it_timer);
                if (! *it_timer) continue; // skip NULL entries
                if (! (*it_timer)->m_active) continue; // skip if not enabled
                if ( (*it_timer)->m_when <= current_time)
                {
                    ODOUT("   ==> MARK");
                    (*it_timer)->m_marked = true;
                }
            }
        }


        // step 4 : execute

        // step 4.1: execute io
        ODOUT("execute stage");
        for(internal_io::IOList::iterator it_io = internal_io::g_io_list().begin();
            it_io != internal_io::g_io_list().end();
            ++ it_io)
        {
            ODOUT("  check obj " << *it_io);
            if (NULL == *it_io) continue;
            if ((*it_io)->m_marked_for_writing)
            {
                FODOUT((*it_io),"exec doWrite");
                (*it_io)->doWrite();
                if ((*it_io) == NULL) continue; // skip remaining if we lost the object
                if ((*it_io)->m_errno)
                {
                    continue;
                }
            }
            if ((*it_io)->m_marked_for_reading)
            {
                FODOUT((*it_io),"exec doRead");
                (*it_io)->doRead();
                if ((*it_io) == NULL) continue; // skip remaining if we lost the object
                if ((*it_io)->m_errno)
                {
                    continue;
                }
            }
        }

        // step 4.2: execute timer events
        {
            for(internal_io::TimerList::iterator it_timer= internal_io::g_timer_list().begin();
                it_timer != internal_io::g_timer_list().end();
                ++ it_timer)
            {
                if (! *it_timer) continue; // skip NULL entries
                if (! (*it_timer)->m_active) continue; // skip if not enabled
                if (! (*it_timer)->m_marked) continue; // skip if not marked

                // reset the mark and deactivate object now since the execute() method might activate it again
                (*it_timer)->m_marked= false;
                (*it_timer)->m_active= false;

                // now execute the event:
                (*it_timer)->execute();
            }
        }

    } // eo try
    catch(...)
    {
        // clean up our counter
        --m_count_active_steps;
        // and forward the exception
        throw;
    }

    if ( 0 == --m_count_active_steps)
    {
        internal_io::g_io_list().clean_list();
        internal_io::g_timer_list().clean_list();
    }

    return had_active_object;
} // eo Backend::doOneStep


/**
 * enters a backend loop.
 *
 * Calls @a Backend::doOneStep within a loop until @a Backend::stop was called or there are no more
 * active objects (io objects or timer objects).
 */
void Backend::run()
{
    ++m_count_active_loops;
    do
    {
        try
        {
            if (!doOneStep(c_max_poll_wait))
            {
                // stop if there are no more active objects.
                stop();
            }
        }
        catch(...)
        {
            // clean up our counter
            --m_count_active_loops;
            // and forward the exception
            throw;
        }
    } 
    while (0 == m_count_stop_requests);
    --m_count_active_loops;
    --m_count_stop_requests;
} // eo Backend::run


/**
 * @brief stops the latest loop currently run by Backend::run().
 * @see Backend::run()
 */
void Backend::stop()
{
    if (m_count_active_loops)
    {
        ++m_count_stop_requests;
    }
} // eo Backend::stop()



} // eo namespace AsyncIo