#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:

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

Function Documentation

THREAD_RETURN lm_loopLog ( void * )

Definition at line 567 of file CALastMix.cpp.

References logMemoryUsage(), CALastMix::m_bRunLog, CALastMix::m_logDownloadedBytes, CALastMix::m_logDownloadedPackets, CALastMix::m_logUploadedBytes, CALastMix::m_logUploadedPackets, print64(), CAMsg::printMsg(), sSleep(), and THREAD_RETURN_SUCCESS.

  {
    CALastMix* pLastMix=(CALastMix*)param;
    pLastMix->m_bRunLog=true;
    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;
  }

Here is the call graph for this function:

THREAD_RETURN lm_loopReadFromMix ( void * pParam )

Definition at line 696 of file CALastMix.cpp.

References CASocketGroup::add(), CAQueue::add(), t_MixPacket::channel, CHANNEL_DUMMY, t_MixPacket::data, DATA_SIZE, diff64(), DUMMY_CHANNEL, E_TIMEDOUT, ev_net_prevConnectionClosed, t_MixPacket::flags, getcurrentTimeMicros(), getcurrentTimeMillis(), getRandom(), CAQueue::getSize(), CALastMix::m_bRestart, CALastMix::m_pMuxIn, CALastMix::m_pQueueReadFromMix, CAMix::m_u32KeepAliveRecvInterval, MAX_READ_FROM_PREV_MIX_QUEUE_SIZE, MIX_POOL_TIMEOUT, MIXPACKET_SIZE, MONITORING_FIRE_NET_EVENT, msSleep(), t_queue_entry::packet, CAPool::pool(), CAMsg::printMsg(), CAMuxSocket::receive(), CASocketGroup::select(), setZero64(), and THREAD_RETURN_SUCCESS.

  {
    CALastMix* pLastMix=(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;
  }

Here is the call graph for this function:

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 600 of file CALastMix.cpp.

References t_MixPacket::channel, CHANNEL_DATA, CHANNEL_DUMMY, t_MixPacket::data, DATA_SIZE, DUMMY_CHANNEL, E_SUCCESS, E_TIMEDOUT, ev_net_prevConnectionClosed, t_MixPacket::flags, getcurrentTimeMicros(), CAQueue::getOrWait(), getRandom(), isZero64(), len, CALastMix::m_bRestart, CALastMix::m_pMuxIn, CALastMix::m_pQueueSendToMix, CAMix::m_u32KeepAliveSendInterval, MIX_POOL_TIMEOUT, MIXPACKET_SIZE, MONITORING_FIRE_NET_EVENT, t_queue_entry::packet, CAPool::pool(), CAMsg::printMsg(), CAMuxSocket::send(), setZero64(), and THREAD_RETURN_SUCCESS.

  {
    CALastMix* pLastMix=(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;
          }
        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;
  }

Here is the call graph for this function:

Classes

Functions

Function Documentation