CALastMix.hpp File Reference

#include "CAMix.hpp"
#include "CAMuxSocket.hpp"
#include "CAASymCipher.hpp"
#include "CASocketAddrINet.hpp"
#include "CACacheLoadBalancing.hpp"
#include "CAUtil.hpp"
#include "CAQueue.hpp"
#include "CAInfoService.hpp"
#include "CALogPacketStats.hpp"
#include "CALastMixChannelList.hpp"
#include "CAMixWithReplayDB.hpp"

Include dependency graph for CALastMix.hpp:

This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Classes
class	CALastMix

Functions
THREAD_RETURN	lm_loopLog (void *)
THREAD_RETURN	lm_loopSendToMix (void *param)
	How to end this thread: 0. More...
THREAD_RETURN	lm_loopReadFromMix (void *pParam)

Function Documentation

lm_loopLog()

THREAD_RETURN lm_loopLog

(

void *

param

)

Definition at line 188 of file CALastMix.cpp.

  {
    CALastMix* pLastMix=static_cast<CALastMix*>(param);
    UINT32 countLog=0;
    UINT8 buff[256];
    while(pLastMix->m_bRunLog)
      {
        if((countLog%10)==0)
          {
            logMemoryUsage();
          }
        if(countLog==0)
          {
            CAMsg::printMsg(LOG_DEBUG,"Uploaded  Packets: %u\n",pLastMix->m_logUploadedPackets);
            CAMsg::printMsg(LOG_DEBUG,"Downloaded Packets: %u\n",pLastMix->m_logDownloadedPackets);
            print64(buff,(UINT64&)pLastMix->m_logUploadedBytes);
            CAMsg::printMsg(LOG_DEBUG,"Uploaded  Bytes  : %s\n",buff);
            print64(buff,(UINT64&)pLastMix->m_logDownloadedBytes);
            CAMsg::printMsg(LOG_DEBUG,"Downloaded Bytes  : %s\n",buff);
            countLog=30;
          }
        sSleep(30);
        countLog--;
      }
    THREAD_RETURN_SUCCESS;
  }

lm_loopReadFromMix()

THREAD_RETURN lm_loopReadFromMix

(

void *

pParam

)

ToDo: check if keep-alive is really correct here - should it not be moved upwards?

ToDo: check if keep-alive is really correct here - should it not be moved upwards?

Definition at line 187 of file CALastMix.cpp.

  {
    #ifdef ENABLE_GPERFTOOLS_CPU_PROFILER
      ProfilerRegisterThread();
    #endif
    CALastMix* pLastMix = static_cast<CALastMix*>(pParam);
    CAMuxSocket* pMuxSocket=pLastMix->m_pMuxIn;
    CAQueue* pQueue=pLastMix->m_pQueueReadFromMix;
    tQueueEntry* pQueueEntry=new tQueueEntry;
    MIXPACKET* pMixPacket=&pQueueEntry->packet;
    CASingleSocketGroup* pSocketGroup=new CASingleSocketGroup(false);
    pSocketGroup->add(*pMuxSocket);
    #ifdef USE_POOL
      CAPool* pPool=new CAPool(MIX_POOL_SIZE);
    #endif
    UINT64 keepaliveNow,keepaliveLast;
    UINT32 u32KeepAliveRecvInterval=pLastMix->m_u32KeepAliveRecvInterval;
    getcurrentTimeMillis(keepaliveLast);
    while(!pLastMix->m_bRestart)
      {
        if(pQueue->getSize()>MAX_READ_FROM_PREV_MIX_QUEUE_SIZE)
          {
#ifdef DEBUG
            CAMsg::printMsg(LOG_DEBUG,"CAFirstMix::Queue is full!\n");
#endif
            msSleep(200);
            getcurrentTimeMillis(keepaliveLast);
            continue;
          }
        //check if the connection is broken because we did not receive a Keep_alive-Message
        getcurrentTimeMillis(keepaliveNow);
        UINT32 keepaliveDiff=diff64(keepaliveNow,keepaliveLast);
        if(keepaliveDiff>u32KeepAliveRecvInterval)
          {
            CAMsg::printMsg(LOG_ERR,"CALastMix::loopReadFromMix() -- restart because of KeepAlive-Traffic Timeout!\n");
            pLastMix->m_bRestart=true;
            MONITORING_FIRE_NET_EVENT(ev_net_prevConnectionClosed);
            break;
          }
        SINT32 ret=pSocketGroup->select(MIX_POOL_TIMEOUT);
        if(ret < 0)
          {
            if (ret == E_TIMEDOUT)
              {
                #ifdef USE_POOL
                  pMixPacket->flags=CHANNEL_DUMMY;
                  pMixPacket->channel=DUMMY_CHANNEL;
                  getRandom(pMixPacket->data,DATA_SIZE);
                  #ifdef LOG_PACKET_TIMES
                    setZero64(pQueueEntry->timestamp_proccessing_start);
                  #endif
                #else
                  continue;
                #endif
              }
            else
              {
                /* another error occured (happens sometimes while debugging because
                 * of interruption, if a breakpoint is reached -> poll() returns
                 * errorcode EINTR)
                 * Note: Any Error on select() does not mean, that the underliny connections have some error state, because
                 * in this case select() returns the socket and than this socket returns the error
                 */
                continue;
              }
          }
        else if(ret>0)
        {
          ret=pMuxSocket->receive(pMixPacket); //receives a whole MixPacket
          #ifdef LOG_PACKET_TIMES
            getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_start);
          #endif
          #ifdef DATA_RETENTION_LOG
            pQueueEntry->dataRetentionLogEntry.t_in=htonl(time(NULL));
          #endif
          if(ret!=MIXPACKET_SIZE)
          {//something goes wrong...
            CAMsg::printMsg(LOG_ERR,"CALastMix::lm_loopReadFromMix - received returned: %i\n",ret);
            pLastMix->m_bRestart=true;
            MONITORING_FIRE_NET_EVENT(ev_net_prevConnectionClosed);
            break;
          }
        }
    #ifdef USE_POOL
      #ifdef LOG_PACKET_TIMES
        getcurrentTimeMicros(pQueueEntry->pool_timestamp_in);
      #endif
        pPool->pool((tPoolEntry*) pQueueEntry);
      #ifdef LOG_PACKET_TIMES
        getcurrentTimeMicros(pQueueEntry->pool_timestamp_out);
      #endif
    #endif
        pQueue->add(pQueueEntry,sizeof(tQueueEntry));
        getcurrentTimeMillis(keepaliveLast);
      }
    delete pQueueEntry;
    pQueueEntry = NULL;
    delete pSocketGroup;
    pSocketGroup = NULL;
    #ifdef USE_POOL
      delete pPool;
      pPool = NULL;
    #endif
    THREAD_RETURN_SUCCESS;
  }

lm_loopSendToMix()

THREAD_RETURN lm_loopSendToMix

(

void *

param

)

How to end this thread: 0.

set m_bRestart=true;

1. Close connection to next mix
2. put a byte in the Mix-Output-Queue

Definition at line 186 of file CALastMix.cpp.

  {
    #ifdef ENABLE_GPERFTOOLS_CPU_PROFILER
      ProfilerRegisterThread();
    #endif
 
    CALastMix* pLastMix = static_cast<CALastMix*>(param);
    CAQueue* pQueue=pLastMix->m_pQueueSendToMix;
    CAMuxSocket* pMuxSocket=pLastMix->m_pMuxIn;
    SINT32 ret;
    UINT32 len;
    UINT32 u32KeepAliveSendInterval=pLastMix->m_u32KeepAliveSendInterval;
#ifndef USE_POOL
    tQueueEntry* pQueueEntry=new tQueueEntry;
    MIXPACKET* pMixPacket=&pQueueEntry->packet;
 
    while(!pLastMix->m_bRestart)
      {
        len=sizeof(tQueueEntry);
        ret=pQueue->getOrWait((UINT8*)pQueueEntry,&len,u32KeepAliveSendInterval);
        if(ret==E_TIMEDOUT)
          {
            pMixPacket->flags=CHANNEL_DUMMY;
            pMixPacket->channel=DUMMY_CHANNEL;
            getRandom(pMixPacket->data,DATA_SIZE);
          }
        else if(ret!=E_SUCCESS||len!=sizeof(tQueueEntry))
          {
            CAMsg::printMsg(LOG_ERR,"CALastMix::lm_loopSendToMix - Error in dequeueing MixPaket\n");
            CAMsg::printMsg(LOG_ERR,"ret=%i len=%i\n",ret,len);
            MONITORING_FIRE_NET_EVENT(ev_net_prevConnectionClosed);
            break;
          }
#ifdef ANON_DEBUG_MODE
        if (pMixPacket->flags&CHANNEL_DEBUG)
          {
          UINT8 base64Payload[DATA_SIZE << 1];
          EVP_EncodeBlock(base64Payload, pMixPacket->data, DATA_SIZE);//base64 encoding (without newline!)
          CAMsg::printMsg(LOG_DEBUG, "Send Downstream AN.ON packet to previous Mix debug: %s\n", base64Payload);
          }
#endif
        if (pMuxSocket->send(pMixPacket) != MIXPACKET_SIZE)
          {
            CAMsg::printMsg(LOG_ERR,"CALastMix::lm_loopSendToMix - Error in sending MixPaket\n");
            MONITORING_FIRE_NET_EVENT(ev_net_prevConnectionClosed);
            break;
          }
#ifdef LOG_PACKET_TIMES
        if(!isZero64(pQueueEntry->timestamp_proccessing_start))
          {
            getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end);
            pLastMix->m_pLogPacketStats->addToTimeingStats(*pQueueEntry,CHANNEL_DATA,false);
          }
#endif
      }
    delete pQueueEntry;
    pQueueEntry = NULL;
#else
    CAPool* pPool=new CAPool(MIX_POOL_SIZE);
    tPoolEntry* pPoolEntry=new tPoolEntry;
    MIXPACKET* pMixPacket=&pPoolEntry->packet;
    while(!pLastMix->m_bRestart)
      {
        len=sizeof(tQueueEntry);
        SINT32 ret=pQueue->getOrWait((UINT8*)pPoolEntry,&len,MIX_POOL_TIMEOUT);
        if(ret==E_TIMEDOUT)
          {
            pMixPacket->flags=CHANNEL_DUMMY;
            pMixPacket->channel=DUMMY_CHANNEL;
            getRandom(pMixPacket->data,DATA_SIZE);
            #ifdef LOG_PACKET_TIMES
              setZero64(pPoolEntry->timestamp_proccessing_start);
            #endif
          }
        else if(ret!=E_SUCCESS||len!=sizeof(tQueueEntry))
          {
            break;
          }
        #ifdef LOG_PACKET_TIMES
          getcurrentTimeMicros(pPoolEntry->pool_timestamp_in);
        #endif
        pPool->pool(pPoolEntry);
        #ifdef LOG_PACKET_TIMES
          getcurrentTimeMicros(pPoolEntry->pool_timestamp_out);
        #endif
        if(pMuxSocket->send(pMixPacket)!=MIXPACKET_SIZE)
          break;
#ifdef LOG_PACKET_TIMES
        if(!isZero64(pPoolEntry->timestamp_proccessing_start))
          {
            getcurrentTimeMicros(pPoolEntry->timestamp_proccessing_end);
            pLastMix->m_pLogPacketStats->addToTimeingStats(*pPoolEntry,CHANNEL_DATA,false);
          }
#endif
      }
    delete pPoolEntry;
    pPoolEntry = NULL;
    delete pPool;
    pPool = NULL;
#endif
    CAMsg::printMsg(LOG_DEBUG,"Exiting Thread SendToMix\n");
    THREAD_RETURN_SUCCESS;
  }