GeneralMain.cpp

Go to the documentation of this file.

#define __NQ_INTERNAL_SW__ // we need internal functions in here!

#include "NQ.h"

void printStatistics( NQ_STATISTICS stats )
{
    printf("Statistics output:\n\n");
    printf("Receive Task:\n");
    printf(" SocketErrors\t\t\t%u\n", stats.RcvSocketErrors);
    printf(" SelectErrors\t\t\t%u\n", stats.RcvSelectErrors);
    printf(" NoDataRcvdErr\t\t\t%u\n", stats.RcvNoDataReceived);
    printf(" NoBuffersLeft\t\t\t%u\n", stats.RcvNoBuffersLeft);
    printf(" InvalidPeerAddress\t\t%u\n", stats.RcvInvalidPeerAddress);
    printf("\n");
    
    printf(" DroppedPacketsBadChecksum\t%u\n", stats.RcvDroppedPacketsBadChecksum);
    printf(" PacketsMissedByIndex\t\t%u\n", stats.RcvPacketsMissedByIndex);
    printf(" DroppedPacketsQueueFull\t%u\n", stats.RcvDroppedPacketsQueueFull);
    printf(" QueueErrors\t\t\t%u\n", stats.RcvQueueErrors);
    printf("\n");

    printf(" NumTelegrams\t\t\t%u\n", stats.RcvNumTelegrams);
    printf(" NumRequests\t\t\t%u\n", stats.RcvNumRequests);
    printf(" NumResponses\t\t\t%u (w_sem=%u, wo_sem=%u)\n", stats.RcvNumResponses, stats.RcvResponsesWithSemaphore, stats.RcvResponsesWithoutSemaphore);
    printf(" NumAckPackets\t\t\t%u\n", stats.RcvAckPackets );

    printf(" []NumPacketsWantAck\t\t%u\n", stats.RcvNumPacketsWantAck );

    printf("\n");
    printf("Send Task:\n");
    printf(" SocketErrors\t\t\t%u\n", stats.SntSocketErrors);
    printf(" EncodingErrors\t\t\t%u\n", stats.SntEncodingErrors);
    printf("\n");

    printf(" QueueErrors\t\t\t%u\n", stats.SntQueueErrors);
    printf(" QueueFull\t\t\t%u\n", stats.SntQueueFull);
    printf("\n");

    printf(" NumTelegrams\t\t\t%u\n", stats.SntTelegrams);
    printf(" NumRequests\t\t\t%u\n", stats.SntRequests);
    printf(" NumResponses\t\t\t%u\n", stats.SntResponses);
    printf(" NumAckPackets\t\t\t%u\n", stats.SntAckPackets);

    printf("\n");
    printf("Blocking call results:\n");
    printf(" RequestsSent\t\t\t%u\n", stats.BlockingCallRequestSent);
    printf(" ResponseTimeout\t\t%u\n", stats.BlockingCallResponseTimeout);
    printf(" ResponseReceived\t\t%u\n", stats.BlockingCallResponseReceivedOnTime);
    printf("\n");
    printf(" AckTimeout\t\t\t%u\n", stats.AckTimeout);

    printf("\n");
    UINT32 endtime = time(NULL);
    printf("Running for %u seconds\n",endtime - stats.StartTime.tv_sec);
}

void SimulateLocalPeer(const char *aIPAddressOtherPeer, UINT16 aPortOtherPeer, UINT16 aPort)
{
    static CNetworkQueue q;

    bool ret;
    bool running = true;
    bool send_go = false;

    NQ_TIMEVAL tmout;
    UINT32 ip_hostorder_peer;
    UINT16 port_hostorder_peer;
    const UINT32 arrlength=716;
    INT16 arr[arrlength];

    tmout.tv_usec = 10*1000;
    tmout.tv_sec = 0;

    platformInitializeConsoleInput();
    platformInitializeSocketSubsystem();
    printf("At the beginning of SimulateLocalPeer\n");

    printf("Before Queue.create()\n");
    ret = q.create(aIPAddressOtherPeer, aPortOtherPeer, aPort);
    printf("Queue.create() returned: %d\n", ret);
    if (ret != true) 
    {
        printf("Queue could not be initialized. Exiting.\n");
        platformDeinitializeConsoleInput();
        platformDeinitializeSocketSubsystem();
        return;
    }

    ip_hostorder_peer = ntohl( platformIPAddrConvert( aIPAddressOtherPeer ) );
    port_hostorder_peer = aPortOtherPeer;
    
    //UINT32 runs = 0;
    UINT32 starttime = platformGetTickCount();
    UINT32 stoptime = platformGetTickCount();
    UINT32 nextTickCount = 0;

    while( running )
    {
        if (platformkbhit())
        {
            char c = platformgetch();
            if (c == 'q' || c == 'Q') 
            {
                printf("Received 'q' uit key. Exiting.\n");
                running = false;
            }
            if (c == 's' || c == 'S')
            {
                NQ_STATISTICS stats;
                q.getStatistics( stats );
                printStatistics( stats );
            }

            if (c == 'g' || c == 'G')
            {
                if (send_go == false)
                {
                    printf("Go!\n");
                    send_go = true;
                    starttime = platformGetTickCount();
                }
                else
                {
                    printf("Stop!\n");
                    send_go = false;
                    stoptime = platformGetTickCount();
                }
            }
        }

        
        
        // telegram senden
        // telegram empfangen

        CNQMessage *tele;


      if (send_go == true)
        {
            tele = q.createMsg();
            if (tele != NULL)
            {
                NQ_TIMEVAL timeout;

                tele->setType(NQ_PT_TELEGRAM, NQ_LT_UPDATE);

                for (int t=0;t<(INT32)arrlength;t++)
                {
                    arr[t] = -((t*2)+1);
                }

                ret = tele->write(236, NQ_VARTYPE_INT16, arr, arrlength);
                if (ret == false) printf("Packet encoding failed.\n");
                
                timeout.tv_sec = 0;
                timeout.tv_usec = 900*1000; // 900 msec

                ret = q.sendMsgTo(tele, ip_hostorder_peer, port_hostorder_peer, timeout);
                //ret = q.sendMsgTo(tele, ip_hostorder_peer, port_hostorder_peer);
                if (ret == false) 
                {
                    if (q.getLastError() != ERR_NQ_SENDQUEUEFULL)   printf("Queue send error: %d\n", q.getLastError() );
                    else platformSleepMilliseconds(20);
                }


                q.freeMsg( tele );
            }

            //platformSleepMilliseconds(10);
            //printf("Queue send: ret=%d, gle=%d\n",ret,q.getLastError() );
            
        }

        if ((send_go == true) && (platformGetTickCount() >= nextTickCount) ) 
        {
            CNQMessage *syncRequest;
            NQ_TIMEVAL syncTimeout;
            static UINT32 val=0;
            UINT32 valback;

            syncRequest = q.createMsg();
            
            if (syncRequest != NULL)
            {
                syncTimeout.tv_sec = 0;
                syncTimeout.tv_usec = 999*1000; 

                syncRequest->setType(NQ_PT_REQUEST, NQ_LT_CALL); 
           
                val++;
                if (val > 300) val = 1;
                ret = syncRequest->write_UINT32(1, val); 

            
                if (ret == true)
                {
                    CNQMessage *syncResponse;
                    
                    //UINT32 start, stop;
                    //start = platformGetTickCount();

                    ret = q.sendRecvMsg(&syncResponse, syncRequest, ip_hostorder_peer, port_hostorder_peer, syncTimeout );

                    //stop = platformGetTickCount();
                    //printf("sendRecvMsg rtt=%d msec\n", stop-start);

                    if (ret == false)
                    {
                        printf("No response received (sync call) due to %d (Timeout is %d).\n", q.getLastError(), ERR_NQ_TIMEOUT);
                    }
                    else
                    {
                        UINT32 valback_expected = ((val*200)/44)%30;

                        ret = syncResponse->read_UINT32(1, valback);
                        if ((ret != true) || (valback != valback_expected)) 
                        {
                            printf("Logical error in sync call response valb=%d, valb_expected=%d, val=%d\n",valback, valback_expected, val);
                        }
                
                        q.freeMsg( syncResponse );
                    }
                }
                else printf("sync Packet encoding failed.\n");

                q.freeMsg( syncRequest );
            }
            else 
                printf("Synchronous call failed: no message buffer free!\n");

            nextTickCount = platformGetTickCount() + 1000;
        }
        


        while (q.recvMsg(&tele) == true)
        {
            //printf("MainTask: got new message\n");

            if (tele->isTelegram() == true)
            {                   
                if ((tele->getPhysicalType() == NQ_PT_TELEGRAM) &&
                    (tele->getLogicalType() == NQ_LT_UPDATE))
                {

                        memset(arr, 0, sizeof(arr));

                        UINT32 arrlength_returned=arrlength;

                        ret = tele->read(236, NQ_VARTYPE_INT16, arr, arrlength_returned);
                        if (ret == false) printf("Packet decoding failed.\n");
                            
                        if (arrlength_returned != arrlength) printf("Returned arrlength differs!\n");

                        for (int t=0;t<(INT32)arrlength;t++)
                        {
                            if (arr[t] != (-((t*4)+1))) printf("Error in array at t=%d\n",t);
                        }

                }
                else printf("Received wrong (bad) telegram. This should not happen.\n");
            }
            else    
                printf("Received wrong (bad) message. This may happen (timed out response).\n");
          
            q.freeMsg( tele );
            
        }
        
    }
    if (send_go == true) stoptime = platformGetTickCount();
    printf("Out of while loop.\n");

    ret = q.close();
    printf("Queue.close() local returned: %d\n", ret);

    NQ_STATISTICS stats1;
    q.getStatistics( stats1 );
    printStatistics( stats1 );

    printf("Data transfer was running for %u msec.\n", stoptime-starttime);

    printf("At the end. Exiting.\n");
    platformDeinitializeConsoleInput();
    platformDeinitializeSocketSubsystem();
}


void SimulateRemotePeer(const char *aIPAddressOtherPeer, UINT16 aPortOtherPeer, UINT16 aPort)
{
    static CNetworkQueue q;

    bool ret;
    bool running = true;

    NQ_TIMEVAL tmout;
    UINT32 ip_hostorder_peer;
    UINT16 port_hostorder_peer;

    const UINT32 arrlength=716;
    INT16 arr[arrlength];

    tmout.tv_usec = 10*1000;
    tmout.tv_sec = 0;

    platformInitializeSocketSubsystem();
    platformInitializeConsoleInput();
    printf("At the beginning of SimulateRemotePeer\n");

    printf("Before Queue.create()\n");
    ret = q.create(aIPAddressOtherPeer, aPortOtherPeer, aPort);
    printf("Queue.create() returned: %d\n", ret);
    if (ret != true) 
    {
        printf("Queue could not be initialized. Exiting.\n");
        platformDeinitializeConsoleInput();
        platformDeinitializeSocketSubsystem();
        return;
    }

    ip_hostorder_peer = ntohl( platformIPAddrConvert( aIPAddressOtherPeer ) );
    port_hostorder_peer = aPortOtherPeer;
    
    //UINT32 runs = 0;
    UINT32 starttime = platformGetTickCount();

    while( running )
    {
        if (platformkbhit())
        {
            char c = platformgetch();
            if (c == 'q' || c == 'Q') 
            {
                printf("Received 'q' uit key. Exiting.\n");
                running = false;
            }
            if (c == 'b' || c == 'B') 
            {
                printf("removing remote peer %s:%d\n", aIPAddressOtherPeer, aPortOtherPeer);
                if (q.removePeer(ip_hostorder_peer, port_hostorder_peer) == false) printf("Error: Could NOT remove remote peer!\n");
            }

            if (c == 'u' || c == 'U') 
            {
                printf("registering remote peer %s:%d\n", aIPAddressOtherPeer, aPortOtherPeer);
                if (q.addPeer(ip_hostorder_peer, port_hostorder_peer) == false) printf("Error: Could NOT (re-)register remote peer!\n");
            }

            if (c == 'i' || c == 'I') 
            {
                UINT32 ip_hostorder_additionalpeer = ntohl( platformIPAddrConvert( "141.34.30.177" ) );
                UINT16 port_hostorder_additionalpeer = 31700;
                    
                printf("registering remote peer %s:%d\n", platformIPAddrConvert(ip_hostorder_additionalpeer), port_hostorder_additionalpeer);
                if (q.addPeer(ip_hostorder_additionalpeer, port_hostorder_additionalpeer) == false) printf("Error: Could NOT register additional peer!\n");
            }

            if (c == 's' || c == 'S')
            {
                NQ_STATISTICS stats;
                q.getStatistics( stats );
                printStatistics( stats );
            }


        }
        
        
        CNQMessage *tele_remote;
        if (q.recvMsg(&tele_remote, tmout) == true)
        {
            if (tele_remote->isTelegram() == true)
            {                   
                    if ((tele_remote->getPhysicalType() == NQ_PT_TELEGRAM) &&
                        (tele_remote->getLogicalType() == NQ_LT_UPDATE))
                    {

                        {
                            UINT32 arrlength_returned=arrlength;
                            
                            memset( arr, 0, sizeof(arr));

                            ret = tele_remote->read(236, NQ_VARTYPE_INT16, arr, arrlength_returned);
                            if (ret == false) printf("Packet decoding failed.\n");

                            if (arrlength_returned != arrlength) printf("Returned arrlength differs!\n");
                            
                        }

                        UINT32 ip;
                        UINT16 port;
                        bool gotIPAndPort;
                        gotIPAndPort = tele_remote->getSenderIPandPort(ip, port);

                        {
                        for (int t=0;t<(INT32)arrlength;t++)
                        {
                            if (arr[t] != (-((t*2)+1))) printf("Error in array at t=%d\n",t);
                        }
                        }

                        {
                        for (int t=0;t<(INT32)arrlength;t++)
                        {
                            arr[t] = -((t*4)+1);
                        }
                        
                        tele_remote->clear();
                        tele_remote->setType(NQ_PT_TELEGRAM, NQ_LT_UPDATE);

                        ret = tele_remote->write(236, NQ_VARTYPE_INT16, arr, arrlength);
                        if (ret == false) printf("Packet encoding failed.\n");
                        
                        }

                        if (gotIPAndPort == true)
                        {
                            // maybe: send something back
                            ret = q.sendMsgTo(tele_remote, ip, port);
                            if (ret == false) 
                            {
                                if (q.getLastError() != ERR_NQ_SENDQUEUEFULL) printf("Queue send error: %d\n", q.getLastError() );
                                else platformSleepMilliseconds(20);
                            }
                        }
                        else printf("Unable to get ip and port of remote peer.\n");
                    }
                    else printf("Received wrong (bad) telegram. This should not happen.\n");
            }
            else if (tele_remote->isRequest() == true)
            {   
                UINT32 val;
                
                // SW! no error checking!
                tele_remote->read_UINT32(1, val);
                val = ((val*200)/44)%30;
                tele_remote->write_UINT32(1, val);
                tele_remote->setType(NQ_PT_RESPONSE, tele_remote->getLogicalType());

                UINT32 ip;
                UINT16 port;
                ret = tele_remote->getSenderIPandPort(ip, port);
                if (ret == true)
                {
                    //printf("send message back to: %s:%d\n", platformIPAddrConvert(ip), port);
                    
                    ret = q.sendMsgTo(tele_remote, ip, port);
                    if (ret == false) 
                    {
                        printf("Send sync Response: Queue send error: %d\n", q.getLastError() );
                    }
                }
                else printf("Send sync Response: Could not check from where the message came.\n");

            }
            else    
                printf("Received wrong (bad) message. This should not happen.\n");


            q.freeMsg( tele_remote );
        }
        
    }
    UINT32 stoptime = platformGetTickCount();

    printf("Out of while loop.\n");

    ret = q.close();
    printf("Queue.close() remote returned: %d\n", ret);

    NQ_STATISTICS stats1;
    q.getStatistics( stats1 );
    printStatistics( stats1 );

    printf("Data transfer was running for %u msec.\n", stoptime-starttime);

    printf("At the end. Exiting.\n");
    platformDeinitializeConsoleInput();
    platformDeinitializeSocketSubsystem();
}

---