NQclass.cpp

Go to the documentation of this file.

#define __NQ_INTERNAL_SW__

#include "PlatformSW.h"
#include "NQDeclarations.h"
#include "NQLogging.h"
#include "NQPacketQueue.h"
#include "NQMessagePool.h"
#include "NQResponseWaitQueue.h"
#include "NQPeerList.h"
#include "NQclass.h"
#include <assert.h>
#include <time.h>

#define ERR(a)  \
         { errcode = a; \
        goto err;}  

//
// S = alternate socket error is provided in m_iLastSocketError
//
//      1       class was already initialized before
//      2       CNQMessage pool (CNQMessageBuffers) could not be allocated or initialized
//      3       dotted IP address could not be properly identified as valid IP address
//  S   4       could not create udp socket
//  S   5       could not bind udp socket
//      6       could not add peer to list (on InitializeQueue)
//  S   7       error setting socket to non blocking
//      8       could not create send semaphore
//      9       could not create receive semaphore
//      10      could not start send thread
//      11      could not start receive thread
//      12      unbound at the moment
//      13      thread cleanup failed
//  S   14      socket closing failed
//      15      semaphore cleanup failed    
//      16      invalid constructed message 
//      17      the send queue is already stuffed (full of elements)
//      18      internal error in send queue
//      19      unable to notify send thread/task via semaphore
//      20      Queue was simply empty, no element returned
//      21      internal Receive queue error 
//      22      timeout happened (no response received in time)
//      23      error creating temporary notification semaphore; request was not send out at all in that case
//      24      did not find the received response (semaphore was signalled, but special queue does 
//              not contain the announced response)
//      25      temporary notification semaphore could not be deleted
//      26      sendto: unknown, invalid peer


//
//
// construction: initialize variables, initialize queues with default depth
// from Declarations.h
//
//
CNetworkQueue::CNetworkQueue()
{
    m_socket = INVALID_SOCKET;
    m_iLastError = ERR_NQ_NONE;
    m_iLastSystemError = 0;
    m_bSuccessfulInitialized = false;

    m_AckPacketWaitingQueue.setQueueMaxsize(NQ_ACKQUEUE_SIZE);
    m_SendPacketQueue.setQueueMaxsize(NQ_SENDQUEUE_SIZE);
    m_ReceivePacketQueue.setQueueMaxsize(NQ_RECVQUEUE_SIZE);
}

//
//
// destruction: deinitialize instance if it was previously initialized
//
//
CNetworkQueue::~CNetworkQueue()
{
    if (m_bSuccessfulInitialized == true)
    {
        // Deinitialize if inited!
        close();
    }
}

//
//
// CNQMessage instance reference and pool handling as public API
//
//

//
CNQMessage *CNetworkQueue::createMsg()
{
    return obtainReference();
}

bool CNetworkQueue::freeMsg( CNQMessage *ref)
{
    return putbackReference( ref );
}


// internal API has different naming scheme
CNQMessage *CNetworkQueue::obtainReference()
{
    assert( m_bSuccessfulInitialized == true );

    return( bufs.acquire_reference() );
}

bool CNetworkQueue::incrementReferenceCount( CNQMessage *ref )
{
    assert( m_bSuccessfulInitialized == true );

    return( bufs.increment_referencecount(ref) );
}

// internal API has different naming scheme
bool CNetworkQueue::putbackReference( CNQMessage *ref)
{
    assert( m_bSuccessfulInitialized == true );
    
    return( bufs.putback_reference(ref) );
}

//
// public method addPeer()
//
bool CNetworkQueue::addPeer(UINT32 ip_hostorder, UINT32 port_hostorder)
{
    assert( m_bSuccessfulInitialized == true );
    
    // add if peer is not already registered
    return m_PeerList.add(ip_hostorder, port_hostorder);
}

//
// public method removePeer()
//
bool CNetworkQueue::removePeer(UINT32 ip_hostorder, UINT32 port_hostorder)
{
    assert( m_bSuccessfulInitialized == true );

    return m_PeerList.remove(ip_hostorder, port_hostorder);
}

//
// public method create()
//
bool CNetworkQueue::create( const ONECHAR *ip_remote_peer, UINT16 portnumber_remote, UINT16 portnumber_local ) // dotted IP address of remote peer
{
    if (m_bSuccessfulInitialized == true) 
    {
        // already initialized, reinitialization is not implemented, so return false
        m_iLastError = ERR_NQ_ALREADY;
        return false; 
    }
        
    INT32 errcode;
    INT32 ret;
    struct sockaddr_in LOC;
    
    m_sendSemaphore = NULL;
    m_receiveSemaphore = NULL;
    m_hSendThread = NULL;
    m_hReceiveThread = NULL;

    if (bufs.Initialize(NQ_NUM_CNQMESSAGEBUFFERS) == false) ERR(ERR_NQ_BADPOOL); // CMessage buffers

    m_PeerList.removeAll();

    if ((ip_remote_peer != NULL))
    {
        if (strlen(ip_remote_peer)!=0)
        {
            UINT32 ip_net = platformIPAddrConvert( ip_remote_peer );

            if (ip_net == INADDR_NONE) ERR(ERR_NQ_BADIP);
            if (m_PeerList.add(ntohl(ip_net), portnumber_remote) == false) ERR(ERR_NQ_ADDPEER);
        }
    }

    m_iNextRequestId = 1;

    // bind socket to ANY/port and connect socket to anyip/anyport (filter later by recvfrom)

    LOC.sin_family      =   AF_INET;
    LOC.sin_port        =   htons( portnumber_local );
    LOC.sin_addr.s_addr =   INADDR_ANY;
  
    m_socket = platformSocketCreateInet(SOCK_DGRAM);
    if (m_socket == SOCKET_ERROR) ERR(ERR_NQ_SOCREATE);

    ret = platformSocketBind(m_socket, (const struct sockaddr *) &LOC, sizeof(LOC));
    if (ret == SOCKET_ERROR) ERR(ERR_NQ_SOBIND);

    // SW! Oct 17, 2007 - is non-blocking still necessary???
    ret = platformSocketSetNonblocking( m_socket );
    if (ret == SOCKET_ERROR) ERR(ERR_NQ_SONONBLOCK);

    // reset statistics
    memset( (void *) &m_stats, 0, sizeof(m_stats) );

    m_stats.StartTime.tv_sec = time(NULL);
    m_stats.StartTime.tv_usec = 0;

    // fire up the send and receive threads / tasks

    m_sendSemaphore = platformCreateSemaphore(NQ_SENDQUEUE_SIZE);
    if (m_sendSemaphore == NULL) ERR(ERR_NQ_SEMCREATESEND);
    
    m_receiveSemaphore = platformCreateSemaphore(NQ_RECVQUEUE_SIZE);
    if (m_receiveSemaphore == NULL) ERR(ERR_NQ_SEMCREATERECV);
    
    // start thread
    m_bShutdownSendThread = false;
    m_bSendThreadRunning = false;

    m_hSendThread = platformCreateThread(true, this, NQ_SEND_THREAD_PRIORITY);
    if (m_hSendThread == NULL) ERR(ERR_NQ_THRDCREATESEND);

    m_bShutdownReceiveThread = false;
    m_bReceiveThreadRunning = false;

    m_hReceiveThread = platformCreateThread(false, this, NQ_RECV_THREAD_PRIORITY);
    if (m_hReceiveThread == NULL) ERR(ERR_NQ_THRDCREATERECV);

    m_bSuccessfulInitialized = true;

    return true;

err:
    NQ_DBG(("NQ::create: in error stream: errcode=%d\n",errcode));
    m_iLastError = (unsigned int) errcode;

    if ((m_iLastError == ERR_NQ_SOCREATE) || 
        (m_iLastError == ERR_NQ_SOBIND) || 
        (m_iLastError == ERR_NQ_SONONBLOCK)) m_iLastSystemError = platformSocketGetLastError();
    
    if (m_iLastError > ERR_NQ_SONONBLOCK) m_iLastSystemError = platformGetLastError();

    // cleanup stuff
    if (m_iLastError > ERR_NQ_THRDCREATESEND)
    {
        // send thread was started, so shut it down
        m_bShutdownSendThread = true;
        while( m_bSendThreadRunning == true)
        {
            platformSleepMilliseconds(100);
        }

        platformDeleteThread( m_hSendThread );
    }

    if (m_iLastError > ERR_NQ_SEMCREATERECV)
        if (m_receiveSemaphore != NULL) platformDeleteSemaphore( m_receiveSemaphore );

    if (m_iLastError > ERR_NQ_SEMCREATESEND)
        if (m_sendSemaphore != NULL) platformDeleteSemaphore( m_sendSemaphore);

    if (m_iLastError > ERR_NQ_SOCREATE)
    {
        ret = platformSocketClose(m_socket);
        m_socket = INVALID_SOCKET;
    }

    if (m_iLastError > ERR_NQ_BADPOOL)
    {
        bufs.Deinitialize();
    }

        
    return false;
}


//
// public method close()
//
bool CNetworkQueue::close()
{
    INT32 ret;
    INT32 errcode;

    if (m_bSuccessfulInitialized == true)
    {
        m_bShutdownSendThread = true;
        m_bShutdownReceiveThread = true;

        NQ_DBG(("Waiting for send and receive threads to shut down...\n"));

        while( m_bSendThreadRunning == true)
        {
            platformSleepMilliseconds(100);
        }
    
        while( m_bReceiveThreadRunning == true)
        {
            platformSleepMilliseconds(100);
        }
        NQ_DBG(("Success. Send and receive threads are shut down.\n"));

        if (platformDeleteThread( m_hSendThread ) == false) ERR(ERR_NQ_THRDCLEANUP);
        if (platformDeleteThread( m_hReceiveThread ) == false) ERR(ERR_NQ_THRDCLEANUP);

        ret = platformSocketClose(m_socket);
        if (ret == SOCKET_ERROR) ERR(ERR_NQ_SOCLOSE);
        m_socket = INVALID_SOCKET;

        // clean up queues

        int cnt = m_SendPacketQueue.size();
        int i;

        for (i=0;i<cnt;i++)
        {
            CNQMessage *tmp;
            
            tmp = NULL;
            if (m_SendPacketQueue.pop_front(&tmp) == true)
            {
                putbackReference(tmp);
            }
        }

        cnt = m_ReceivePacketQueue.size();
        for (i=0;i<cnt;i++)
        {
            CNQMessage *tmp;
            
            tmp = NULL;
            if (m_ReceivePacketQueue.pop_front(&tmp) == true)
            {
                putbackReference(tmp);
            }
        }

        cnt = m_ReceiveResponseSemaphorePacketQueue.size();
        for (i=0;i<cnt;i++)
        {
            CNQMessage *tmp;
            
            tmp = NULL;
            if (m_ReceiveResponseSemaphorePacketQueue.pop_front(&tmp) == true)
            {
                putbackReference(tmp);
            }
        }
        // clean up queues END

        if (m_sendSemaphore != NULL) 
            if (platformDeleteSemaphore( m_sendSemaphore ) == false) ERR(ERR_NQ_SEMCLEANUP);
        if (m_receiveSemaphore != NULL) 
            if (platformDeleteSemaphore( m_receiveSemaphore ) == false) ERR(ERR_NQ_SEMCLEANUP);

        bufs.Deinitialize();
            
        m_bSuccessfulInitialized = false;

    }
    return true; // was not initialized, so deinizialization succeeded anyway

err:
    m_iLastError = (unsigned int) errcode;
    if (m_iLastError == ERR_NQ_SOCLOSE) 
        m_iLastSystemError = platformSocketGetLastError();
    else 
        m_iLastSystemError = platformGetLastError();
    
    return false;
}


//
// public method sendMsgTo()
//
/* \brief Send a message from Networkqueue
 * 
 * This method is an important part of the user API. A constructed message can be sent out
 * asynchronously to the remote peer with the help of this method.
 * \p
 * The method basically just calls SendMessageinternal(). One can find extended description there.
 * 
 *  \param      msg reference   to the CNQMessage object to send 
 *  \param      ip_hostorder    the destination peer IP address (in host byte order) 
 *  \param      port_hostorder  the destination peer port address (in host byte order)
 *  
 *  \retval true message was successfully send
 *  \retval false an error occured
 */
bool CNetworkQueue::sendMsgTo(  CNQMessage *msg, UINT32 ip_hostorder, UINT16 port_hostorder) 
{
    // make sure this packet will be no ack packet
    msg->internalPacketSpecialBits &= ~NQ_PACKET_BIT_WANT_ACK;
    
    return sendMsgToInternal( ip_hostorder, port_hostorder, msg, NQ_NO_SEMAPHORE);
}

//
// public method sendMsgTo()
//
bool CNetworkQueue::sendMsgTo( CNQMessage *msg, UINT32 ip_hostorder, UINT16 port_hostorder, const NQ_TIMEVAL &timeout)
{
    INT32 errcode;
    UINT32 msecs = timeout.tv_sec * 1000 + (timeout.tv_usec / 1000);
    bool retval;
    HANDLE ack_sem=NULL;
    bool bAckQueueRemovedTimedoutMessage;

    assert( m_bSuccessfulInitialized == true );
    assert( msg != NULL );

    // prepare:
    msg->internalPacketSpecialBits |= NQ_PACKET_BIT_WANT_ACK;

    if (msg->check_int() != 0) ERR(ERR_NQ_MSGINVALID);
    
    if (m_PeerList.isAllowed(ip_hostorder, port_hostorder) == false) ERR(ERR_NQ_INVALIDPEER);

    if (msg->isRequest() == true) 
    {
        msg->internalRequestID = m_iNextRequestId++;
    }

    msg->message_sendto_ip_host = ip_hostorder;
    msg->message_sendto_port_host = port_hostorder;

    ack_sem = platformCreateSemaphore(1);
    if (ack_sem == NULL) ERR(ERR_NQ_SEMCREATE);
    msg->notifySemaphoreOnReceiveAck = ack_sem;

    incrementReferenceCount(msg);
    if (m_AckPacketWaitingQueue.push_back(msg) == false)
    {
        putbackReference(msg);
        ERR(ERR_NQ_ACKQUEUE);
    }

    // send out stuff:
    incrementReferenceCount(msg);
    UINT32 error;
    if (m_SendPacketQueue.push_back( msg, error) == false) 
    {
        putbackReference(msg);
        if (error == ERR_PQ_FULL)
        {
            m_stats.SntQueueFull++;
            ERR(ERR_NQ_SENDQUEUEFULL);
        }
        else
        {
            NQ_DBG(("SendMessageToBlocking: Sendqueue error: %d\n",error));
            m_stats.SntQueueErrors++;
            ERR(ERR_NQ_SENDQUEUE);
        }
    }

    if (platformSignalSemaphore( m_sendSemaphore ) == false) ERR(ERR_NQ_SEMNOTIFY); // signal: there is a new packet

    if (platformWaitForSemaphore(ack_sem, msecs) == true) retval = true; else retval = false;

    // find the ack_sem attached message in the special queue and
    // delete it there
    bAckQueueRemovedTimedoutMessage = m_AckPacketWaitingQueue.find_and_remove_passed_elem( msg );
    assert(bAckQueueRemovedTimedoutMessage == true);
    putbackReference( msg );
    
    if (retval == false)
    {
        m_stats.AckTimeout++;
        ERR(ERR_NQ_TIMEOUT);
    }

    if (platformDeleteSemaphore( ack_sem ) == false) ERR(ERR_NQ_SEMDELETE);

    return retval; 

err:
    if ((errcode == ERR_NQ_SENDQUEUE) || (errcode == ERR_NQ_SENDQUEUEFULL))
    {
        if (m_AckPacketWaitingQueue.find_and_remove_passed_elem( msg ) == true)
        {
            putbackReference(msg);
        }
    }

    if ((errcode == ERR_NQ_SENDQUEUEFULL) || (errcode == ERR_NQ_SENDQUEUE) || (errcode == ERR_NQ_ACKQUEUE) || (errcode == ERR_NQ_TIMEOUT) )
    {
        if (ack_sem != NULL) platformDeleteSemaphore( ack_sem );
    }
    
    if (errcode != ERR_NQ_NONE) m_iLastError = (unsigned int) errcode;
    
    return false;
}


//
// private method sendMsgInternal()
//
bool CNetworkQueue::sendMsgToInternal( UINT32 ip_hostorder, UINT16 port_hostorder, CNQMessage *msg, HANDLE recv_notify_sem)
{
    INT32 errcode;

    assert( m_bSuccessfulInitialized == true );
    assert( msg != NULL );

    if (msg->check_int() != 0) ERR(ERR_NQ_MSGINVALID);
    if (m_PeerList.isAllowed(ip_hostorder, port_hostorder) == false) ERR(ERR_NQ_INVALIDPEER);


    msg->message_sendto_ip_host = ip_hostorder;
    msg->message_sendto_port_host = port_hostorder;

    incrementReferenceCount(msg);
    UINT32 error;
    if (m_SendPacketQueue.push_back( msg, error) == false) 
    {
        putbackReference(msg);
        if (error == ERR_PQ_FULL)
        {
            m_stats.SntQueueFull++;
            ERR(ERR_NQ_SENDQUEUEFULL);
        }
        else
        {
            NQ_DBG(("SendMessageInternal(): Sendqueue error: %d\n",error));
            m_stats.SntQueueErrors++;
            ERR(ERR_NQ_SENDQUEUE);
        }
    }

    if (platformSignalSemaphore( m_sendSemaphore ) == false) ERR(ERR_NQ_SEMNOTIFY); // signal: there is a new packet
    
    return true;
err:
    m_iLastError = (unsigned int) errcode;
    
    return false;
}

//
// public method recvMsg()
//
bool CNetworkQueue::recvMsg( CNQMessage **newmsg )
{
    bool ret;
    INT32 errcode=0;
    UINT32 lasterr;

    assert( m_bSuccessfulInitialized == true );
    assert( newmsg != NULL );

    if (platformIsSemaphoreSignalled( m_receiveSemaphore ) == true)
    {
        ret = m_ReceivePacketQueue.pop_front( newmsg, lasterr );
        if (ret == false) 
        {
            if (lasterr == ERR_PQ_NONE) 
            {
                assert(26==34); // fatal: semaphore says yes, but queue is empty (huh?)
                ERR(ERR_NQ_EMPTY);
            }
            ERR(ERR_NQ_RECVQUEUE);
        }
        if ((*newmsg)->wantAckPacket() == true)
        {
            CNQMessage *ack = obtainReference();
            if (ack != NULL)
            {
                UINT32 ip_hostorder = (*newmsg)->message_came_from_ip;
                UINT16 port_hostorder = (*newmsg)->message_came_from_port;

                ack->constructAckPacket( **newmsg );

                if (sendMsgToInternal( ip_hostorder, port_hostorder, ack, NQ_NO_SEMAPHORE) == false)
                {
                    NQ_DBG(("SendMessageinternal (for ack packet) failed (1).\n"));
                }

                putbackReference(ack);
            }
            else assert( ack != NULL ); // emit assert for debugging reasons
        }

        return true;
    }
    ERR(ERR_NQ_EMPTY);

err:
    m_iLastError = (unsigned int) errcode;

    return false;
}

//
// public method recvMsg()
//
bool CNetworkQueue::recvMsg( CNQMessage **newmsg, const NQ_TIMEVAL &timeout )
{
    UINT32 timeout_msec;
    INT32 errcode=0;

    assert( m_bSuccessfulInitialized == true );
    assert( newmsg != NULL );

    timeout_msec = timeout.tv_sec * 1000 + (timeout.tv_usec / 1000);

    if (platformWaitForSemaphore( m_receiveSemaphore, timeout_msec ) == true)
    {
        UINT32 lasterr;
        bool ret = m_ReceivePacketQueue.pop_front( newmsg, lasterr );
        if (ret == false) 
        {
            if (lasterr == 0) 
            {
                assert(26==34); // fatal: semaphore says yes, but queue is empty (huh?)
                ERR(ERR_NQ_EMPTY);
            }
            ERR(ERR_NQ_RECVQUEUE);
        }

        if ((*newmsg)->wantAckPacket() == true)
        {
            CNQMessage *ack = obtainReference();
            if (ack != NULL)
            {
                UINT32 ip_hostorder = (*newmsg)->message_came_from_ip;
                UINT16 port_hostorder = (*newmsg)->message_came_from_port;

                ack->constructAckPacket( **newmsg );

                if (sendMsgToInternal(ip_hostorder, port_hostorder, ack, NQ_NO_SEMAPHORE) == false)
                {
                    NQ_DBG(("SendMessageinternal (for ack packet) failed (2).\n"));
                }

                putbackReference(ack);
            }
            else assert( ack != NULL ); // emit assert for debugging reasons
        }

        return true;
    }
    else
        m_iLastError = ERR_NQ_TIMEOUT; // timeout

    return false;

err:
    m_iLastError = (unsigned int) errcode;

    return false;
}

//
// public method sendRecvMsg()
//
bool CNetworkQueue::sendRecvMsg(CNQMessage **response, CNQMessage *request, UINT32 ip_hostorder, UINT16 port_hostorder, const NQ_TIMEVAL &timeout)
{
    bool retval = false;

    INT32 errcode;
    UINT32 msecs = timeout.tv_sec * 1000 + (timeout.tv_usec / 1000);
    CNQMessage *l_response;
    bool pushToResponseWaitQueue;

    assert( m_bSuccessfulInitialized == true );
    assert( request != NULL );
    assert( response != NULL );

    if (request->isRequest() == false) return false;

    request->internalRequestID = m_iNextRequestId++;

    HANDLE recv_sem = platformCreateSemaphore(1);
    if (recv_sem == NULL) ERR(ERR_NQ_SEMCREATE);
    
    // add wait for response listener
    NQ_WAITFORRESPONSE resp;
    resp.semaphore_to_signal = recv_sem;
    resp.requestid = request->internalRequestID;

    pushToResponseWaitQueue = m_ResponseWaitQueue.push_back( resp );
    if ( pushToResponseWaitQueue == false) ERR(ERR_NQ_RESPWAITQUEUE);


    if (sendMsgToInternal(ip_hostorder, port_hostorder, request, recv_sem) == false) ERR(0); // pass old error code

    m_stats.BlockingCallRequestSent++;

    if (platformWaitForSemaphore(recv_sem, msecs) == true)
    {
        // extract response from queue
        bool success = m_ReceiveResponseSemaphorePacketQueue.find_and_remove_elem(request->internalRequestID, response);
        if (success == true) 
        {
            m_stats.BlockingCallResponseReceivedOnTime++;
            retval = true;
        }
        else ERR(ERR_NQ_RESPTROUBLE);
    }
    else 
    {
        // kill the response wait block out of the response wait queue
        bool killed = m_ResponseWaitQueue.remove_elem( request->internalRequestID );
        if (killed == false) 
        {
            // if it could not be removed, that means we just miss the response, but it was processed in the 
            // meantime, so silently drop it from RRSemaphorePacketQueue

            bool success = m_ReceiveResponseSemaphorePacketQueue.find_and_remove_elem(request->internalRequestID, &l_response);
            
            {
            if (success == true) NQ_DBG(("Response came just a bit too late!\n")); 
            if (success == false) NQ_DBG(("Could not remove.\n"));
            }

            //if (success == true) {exampleNQLoggingSetParameters( NQ_DBG_TYP, __FILE__, __LINE__); exampleNQLoggingHandler("Response came just a bit too late!\n");} else {exampleNQLoggingSetParameters( NQ_DBG_TYP, __FILE__, __LINE__); exampleNQLoggingHandler("Could not remove!\n");}


            success=true; // make compiler happy;
            
        }
        m_stats.BlockingCallResponseTimeout++;
        ERR(ERR_NQ_TIMEOUT);

    }

    if (platformDeleteSemaphore( recv_sem ) == false) ERR(ERR_NQ_SEMDELETE);

    return retval; 

err:
    if (errcode == ERR_NQ_NONE)
    {
        m_ResponseWaitQueue.remove_elem( request->internalRequestID );
    }
    
    if ((errcode == ERR_NQ_TIMEOUT) || (errcode == ERR_NQ_RESPTROUBLE) || (errcode == ERR_NQ_RESPWAITQUEUE) || (errcode == ERR_NQ_NONE) )
    {
        if (recv_sem != NULL) platformDeleteSemaphore( recv_sem );
    }
    
    if (errcode != ERR_NQ_NONE) m_iLastError = (unsigned int) errcode;
    
    return false;
}



bool CNetworkQueue::getStatistics( NQ_STATISTICS &stats )
{
    stats = m_stats;
    return true;
}


//
// The current set of error codes are listed at the top of this cpp file.
//
UINT32 CNetworkQueue::getLastError()
{
    return m_iLastError;
}

bool CNetworkQueue::getLastErrorString( ONECHAR *buffer, UINT32 buflength)
{

    ONECHAR os_additional_error[300];
    static ONECHAR last_error_buffer[200];
    static const ONECHAR classname[]="CNetworkQueue";

    static const ONECHAR clserrors[ERR_NQ_NUMERRORS][72]=
    {
        "No error.",                                                                //0 0 nix
        "The class was already initialized before (at InitializeQueue).",           //1
        "Error creating CNQMessage pool (CNQMessageBuffers) at InitializeQueue..",      //2
        "The dotted IP address is invalid at InitializeQueue.",                     //3
        "Could not create udp socket at InitializeQueue().",                        //4     S
        "Could not bind udp socket at InitializeQueue().",                          //5     S
        "Could not add peer to list (on InitializeQueue).",                         //6     
        "Error putting socket into nonblocking mode at InitializeQueue().",         //7     S
        "Could not create Send semaphore at InitializeQueue().",                    //8
        "Could not create Receive semaphore at InitializeQueue().",                 //9
        "Could not start send thread/task at InitializeQueue().",                   //10
        "Could not start receive thread/task at InitializeQueue().",                //11
        "Could not cleanup tasks/threads at DeinitializeQueue().",                  //12
        "Could not close socket at DeinitializeQueue().",                           //13    S
        "Could not cleanup semaphores at DeinitializeQueue().",                     //14
        "The packet to be sent is not valid (SendMessage()).",                      //15 
        "The sent queue is filled up at the moment (SendMessage()).",               //16
        "Pushing the packet to send queue failed (SendMessage()).",                 //17
        "Semaphore could not be signalled (SendMessage()).",                        //18
        "Queue was empty (isNewMessageAvailable()).",                               //19
        "Error getting message out of nonempty queue (isNewMessageAvailable()).",   //20
        "Timeout happened (isNewMessageAvailable or SendRequestAndWaitForResp.)",   //21
        "Error creating temp. semaphore (Request/Response blocking).",              //22
        "Through semaphore announced response was not found in dedicated queue!",   //23
        "Temporary semaphore could not be deleted (Request/Response blocking).",    //24
        "Invalid remote peer: (ip, port) is not contained in peer list.",           //25
        "Internal error in 'ack' queue.",                                           //26
        "Internal error in 'response wait' queue."                                  //27

    };
    
    last_error_buffer[0] = 0;
    os_additional_error[0] = 0;

    assert( buffer != NULL );
    assert( m_iLastError < ERR_NQ_NUMERRORS);

    if ((m_iLastError == ERR_NQ_SOCREATE) ||
        (m_iLastError == ERR_NQ_SOBIND) ||
        (m_iLastError == ERR_NQ_SONONBLOCK) ||
        (m_iLastError == ERR_NQ_SOCLOSE))
    {
            sprintf((char *) last_error_buffer, "%s Error: %s\n(sys error was: %d -> \"%s\")", 
                classname, clserrors[m_iLastError-ERR_NQ_BASE], m_iLastSystemError, 
                platformGetOSErrorString(m_iLastSystemError, os_additional_error, 
                sizeof(os_additional_error)) );
    }
    else
    {
        if (m_iLastError >= ERR_NQ_BASE && m_iLastError < ERR_NQ_NUMERRORS)
        {
            // no OS error might have happened
            sprintf((char *) last_error_buffer, "%s Error: %s", classname, clserrors[m_iLastError-ERR_NQ_BASE] );
        }
        else
        {
            // no error or unknown error!
            sprintf((char *) last_error_buffer, "%s Error: %s", classname, "Unknown internal error." );
        }
    }
    
    UINT32 total_length = strlen((char *)last_error_buffer)+1;
    if (total_length > buflength) return false;
    memcpy(buffer, last_error_buffer, total_length);

    return true;
} 

//
//
// internal; platform-independent implementation of the send task or thread
//
// pseudocode:
//  get a new message out of the send queue, encode a physical packet,
//  send it out, adjust statistics and move/adjust data (e.g. response that might come)
//
void CNetworkQueue::SendProcessTask()
{
    UINT8 initial_buffer[NQ_MAX_PACKET_SIZE];
    CNQMessage *m_temp;

    NQ_DBG(("SendTask: Start!\n"));

    m_bSendThreadRunning = true;
    while( 1 == 1)
    {
        // send out all stuff
        // afterwards: receive what can be received and put it into queue

        bool ret = platformWaitForSemaphore(m_sendSemaphore, NQ_SEND_SEMAPHORE_TIMEOUT_MS); // 500msec

        if (ret == true)
        {
            // here we can get an entry out of the queue and sent it to the remote end
            
            bool ret_get = m_SendPacketQueue.pop_front(&m_temp);
            if (ret_get == false) 
            {
                m_stats.SntQueueErrors++;
                continue;
            }
            
            INT32 message_size = 0;

            // fill packet index
            bool success = m_PeerList.getSendPacketIndex(m_temp->message_sendto_ip_host,m_temp->message_sendto_port_host, m_temp->internalPacketIndex);
            assert( success == true);
            success = true; // make compiler happy

            
            message_size = NQ_MAX_PACKET_SIZE;
            bool ret_enc = m_temp->ExportMessageToPacketBuffer(initial_buffer, message_size);
                
            // check for boundaries
            if (ret_enc == false) 
            {
                NQ_DBG(("SendTask: Encoding failed.\n"));
                m_stats.SntEncodingErrors++;
                putbackReference(m_temp);
                continue;
            }

            INT32 still_to_write = message_size;
            UINT8 *bufptr = initial_buffer;
            bool abort=false;

            // if packet contains

            // wrap inside: must not break until send out anything

            INT32 accumsockettimeouts = 0;

            while (abort == false)
            {
                struct timeval tmout;
                fd_set fdwset;

                FD_ZERO( &fdwset );
                FD_SET( m_socket, &fdwset);

                tmout.tv_sec = NQ_SEND_SELECT_TIMEOUT_MS / 1000;
                tmout.tv_usec = (NQ_SEND_SELECT_TIMEOUT_MS % 1000)*1000; 
    
                INT32 ret_select = platformSocketSelect(platformMaxFdSets, NULL, &fdwset, NULL, &tmout);
                if (ret_select != 1)
                {
                    accumsockettimeouts++;
                    if (accumsockettimeouts > 10)
                    {
                        abort = true;
                        m_stats.SntSocketErrors++;
                    }
                }


                INT32 ret_send = platformSocketSendToFunction(m_socket, bufptr, still_to_write,m_temp->message_sendto_ip_host,m_temp->message_sendto_port_host);
                
                if (ret_send >= 0)
                {
                    still_to_write -= ret_send;
                    bufptr += ret_send;

                    if (still_to_write <= 0) abort = true; 
                }
                else
                {
                    if (ret_send == SOCKET_ERROR)
                    {
                        if (platformSocketGetLastError() != EAGAIN) 
                        {
                            abort = true;
                            m_stats.SntSocketErrors++;
                        }
                    }
                }
            }

            if (still_to_write != 0)
            {
                /* SW! Oct 11, 2007 -> removed; now in SendMessageToBlocking
                // packet was NOT successfully send out
                // so clear it from the ack packet queue if 
                if (m_temp->wantAckPacket() == true)
                {
                    if (m_AckPacketWaitingQueue.find_and_remove_passed_elem( m_temp ) == false)
                    {
                        NQ_DBG(("SendTask: Huh, could not find original message in ack queue to remove!\n"));
                    }
                    else putbackReference( m_temp );
                }
                */

            }
            else
            {
                m_PeerList.incSendPacketIndex(m_temp->message_sendto_ip_host,m_temp->message_sendto_port_host); 
                
                if (m_temp->isAckPacket() == true) m_stats.SntAckPackets++;
                // success sending, increment field
                else if (m_temp->isTelegram() == true) m_stats.SntTelegrams++;
                else if (m_temp->isRequest() == true) m_stats.SntRequests++;
                else m_stats.SntResponses++;

            }
            putbackReference(m_temp);
        }

        if (m_bShutdownSendThread == true) break;

    }
    m_bSendThreadRunning = false;
    NQ_DBG(("SendTask: At the very end of task.\n"));
}


//
// internal; platform-independent implementation of the receive task or thread
//
// pseudocode:
//  wait on the socket for a new received packet, read in the contents, decode into a CNQMessage,
//  adjust statistics and move/adjust data (e.g. if packet was an awaited response) and put the
//  decoded message into the appropriate receive queue and notify appropriate semaphore
//

void CNetworkQueue::ReceiveProcessTask()
{
    UINT8 initial_buffer[NQ_MAX_PACKET_SIZE];
    CNQMessage *m_temp;
    UINT32 error;
    UINT32 ipfrom;
    UINT16 portfrom;

    NQ_DBG(("ReceiveTask: Start!\n"));
  
    m_bReceiveThreadRunning = true;
    while( 1 == 1)
    {
        struct timeval tmout;
        fd_set fdrset;

        FD_ZERO( &fdrset );
        FD_SET( m_socket, &fdrset);

        tmout.tv_sec = NQ_RECV_SELECT_TIMEOUT_MS / 1000;
        tmout.tv_usec = (NQ_RECV_SELECT_TIMEOUT_MS % 1000)*1000; 
    
        INT32 ret_select = platformSocketSelect(platformMaxFdSets, &fdrset, NULL, NULL, &tmout);
        if (m_bShutdownReceiveThread == true) break;
        
        if (ret_select == 1)
        {
            INT32 ret_recv = platformSocketReceiveFromFunction(m_socket, initial_buffer, NQ_MAX_PACKET_SIZE, &ipfrom, &portfrom);
            if (ret_recv != SOCKET_ERROR)
            {
                if (ret_recv == 0)  // connection gracefully closed (huh?)
                {                   
                    // anyhow, with 0 bytes we can not do anything
                    m_stats.RcvNoDataReceived++;
                    continue; 
                }

                if (m_PeerList.isAllowed(ipfrom, portfrom) == false)
                {
                    m_stats.RcvInvalidPeerAddress++;
                    continue;
                }

                m_temp = obtainReference();
                if (m_temp == NULL)
                {
                    m_stats.RcvNoBuffersLeft++;
                    continue;
                }

                bool valid = m_temp->ImportMessageFromPacketBuffer(initial_buffer, ret_recv, ipfrom, portfrom );
                if (valid == false)
                {
                    m_stats.RcvDroppedPacketsBadChecksum++;
                    putbackReference(m_temp);
                    continue;
                }

                // wenn keine Gegenstelle (count(m_PeerList)==0) definiert ist, wird der packet index NICHT abgeprüft
                // getpacketindex gibt immer false und setpacketindex immer true zurück

                {
                    UINT16 last_packet_index;
                    static UINT16 ret_recv_lpi;
                    
                    if (m_PeerList.getReceivePacketIndex( ipfrom, portfrom, last_packet_index) == false)
                    {
                        bool success = m_PeerList.setReceivePacketIndex( ipfrom, portfrom, m_temp->internalPacketIndex);
                        assert( success == true );
                        success=true; // make compiler happy
                        ret_recv_lpi = ret_recv;

                    }
                    else
                    {
                        INT32 lpi = (INT32) last_packet_index;
                        INT32 cpi = (INT32) m_temp->internalPacketIndex;
                        INT32 tmp;

                        tmp = ((cpi-lpi)&0xffff)-1;
                        assert( tmp >= 0 );

                        if (tmp != 0) 
                        {
                            m_stats.RcvPacketsMissedByIndex += tmp;
                            NQ_DBG(("ReceiveTask: Missed %d (for big numbers: at least) packet(s) cpi=%d(size=%d); lpi=%d(size=%d) remotepeer=%s:%d\n", tmp, cpi, ret_recv, lpi, ret_recv_lpi, (char*)platformIPAddrConvert(ipfrom),portfrom));
                        }
                        ret_recv_lpi = ret_recv;


                        bool success = m_PeerList.setReceivePacketIndex( ipfrom, portfrom, m_temp->internalPacketIndex);
                        assert( success == true );
                        success=true; // make compiler happy
                    }
                }

                if (m_temp->isAckPacket() == true)
                {
                    if (m_AckPacketWaitingQueue.find_elem_ack_and_signal( *m_temp ) == false)
                    {
                        NQ_DBG(("ReceiveTask: No waiting packet for ack found!"));
                    }

                    m_stats.RcvAckPackets++;
                    // throw away ack packet; continue;
                    putbackReference( m_temp );
                    continue;
                }

                incrementReferenceCount(m_temp);

                if (m_temp->wantAckPacket())
                {
                    m_stats.RcvNumPacketsWantAck++;
                }

                // maybe signal, if proper
                if (m_temp->isResponse() == true)
                {
                    // aha, it is a response, so traverse the response wait queue
                    NQ_WAITFORRESPONSE temp_wait;

                    m_stats.RcvNumResponses++;
          
                    bool found_evt = m_ResponseWaitQueue.find_and_remove_elem(m_temp->internalRequestID, temp_wait);
                    if (found_evt == true)
                    {
                        if (m_ReceiveResponseSemaphorePacketQueue.push_back( m_temp, error ) == true)
                        {
                            m_stats.RcvResponsesWithSemaphore++;
                            platformSignalSemaphore(temp_wait.semaphore_to_signal);
                        }
                        else
                        {
                            putbackReference(m_temp);
                            if (error == 2)
                                m_stats.RcvDroppedPacketsQueueFull++;
                            else
                                m_stats.RcvQueueErrors++;
                        }
                    }
                    else 
                    {
                        if (m_ReceivePacketQueue.push_back( m_temp, error ) == true) // push response without any semaphore
                        {
                            m_stats.RcvResponsesWithoutSemaphore++;
                        
                            platformSignalSemaphore( m_receiveSemaphore ); // notify usual receive functions
                        }
                        else
                        {
                            putbackReference(m_temp);
                            if (error == 2)
                                m_stats.RcvDroppedPacketsQueueFull++;
                            else
                                m_stats.RcvQueueErrors++;
                        }
                    }

                }
                else
                { // for requests and telegrams:

                    if (m_ReceivePacketQueue.push_back( m_temp, error ) == true)
                    {
                        if (m_temp->isTelegram() == true) m_stats.RcvNumTelegrams++;
                        else if (m_temp->isRequest() == true) m_stats.RcvNumRequests++;
                        else assert(33==34); // BAD: here: either telegram or request must be!
                        
                        platformSignalSemaphore( m_receiveSemaphore );
                    }
                    else
                    {
                        putbackReference(m_temp);
                        if (error == 2)
                            m_stats.RcvDroppedPacketsQueueFull++;
                        else
                            m_stats.RcvQueueErrors++;
                    }

                }
                
                putbackReference(m_temp);

            }
            else 
            {
                if (platformSocketGetLastError() != EAGAIN) m_stats.RcvSocketErrors++;
                else NQ_DBG(("ReceiveTask: Eagain.\n"));
            }
        }
        else if (ret_select == SOCKET_ERROR) m_stats.RcvSelectErrors++;
        

    }
    m_bReceiveThreadRunning = false;
    NQ_DBG(("ReceiveTask: At the very end of task.\n"));
}


/* debug code, was removed Oct 17 2007 SW
    SW Apr 17, 2008; changed from print f to NQ_DBG
void CNetworkQueue::PrintQueueSizes()
{
    NQ_DBG(("=========================================\n"));
    NQ_DBG(("= Queue Sizes\n"));
    NQ_DBG(("=========================================\n"));
    NQ_DBG(("m_AckPacketWaitingQueue.size() = %d\n", m_AckPacketWaitingQueue.size()));
    NQ_DBG(("m_ResponseWaitQueue.size() = %d\n", m_ResponseWaitQueue.size()));
    NQ_DBG(("m_ReceiveResponseSemaphorePacketQueue.size() = %d\n", m_ReceiveResponseSemaphorePacketQueue.size()));
    NQ_DBG(("m_ReceivePacketQueue.size() = %d\n", m_ReceivePacketQueue.size()));
    NQ_DBG(("m_SendPacketQueue.size() = %d\n", m_SendPacketQueue.size()));
    NQ_DBG(("=========================================\n"));

}
*/

---