CAFirstMix.hpp File Reference

#include "doxygen.h"
#include "CAMix.hpp"
#include "CAMuxSocket.hpp"
#include "CAASymCipher.hpp"
#include "CAFirstMixChannelList.hpp"
#include "CAIPList.hpp"
#include "CASocketGroup.hpp"
#include "CAQueue.hpp"
#include "CAUtil.hpp"
#include "CAThread.hpp"
#include "CAThreadPool.hpp"
#include "TermsAndConditions.hpp"
#include "CALogPacketStats.hpp"
#include "CAConditionVariable.hpp"
#include "CATempIPBlockList.hpp"
#include "CASocketGroupEpoll.hpp"
#include "CAMixWithReplayDB.hpp"
#include "CAMutex.hpp"

Include dependency graph for CAFirstMix.hpp:

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

Go to the source code of this file.

Classes
class	tUINT32withLock
struct	t_FirstMixChannelToQueueList_entry
class	CAFirstMixChannelToQueueList
class	CAFirstMix

Macros
#define	TNC_SREQUEST   "TermsAndConditionsRequest"
#define	TNC_RESPONSE   "TermsAndConditionsResponse"
#define	TNC_SINTERRUPT   "TermsAndConditionsInterrupt"
#define	TNC_REQ_TRANSLATION   "Translation"
#define	TNC_RESOURCES   "Resources"
#define	TNC_RESOURCE_TEMPLATE   "Template"
#define	TNC_TEMPLATE_ROOT_ELEMENT   "TermsAndConditionsTemplate"
#define	TNC_RESOURCE_CUSTOMIZED_SECT   "CustomizedSections"
#define	TNC_RESPONSE_INVALID_REQUEST   "InvalidTermsAndConditionsRequest"
#define	TNC_CONFIRM_REQ   "TermsAndConditionsConfirm"

Typedefs
typedef struct t_FirstMixChannelToQueueList_entry	tFirstMixChannelToQueueListEntry

Functions
THREAD_RETURN	fm_loopSendToMix (void *)
	How to end this thread: 0. More...
THREAD_RETURN	fm_loopReadFromMix (void *)
THREAD_RETURN	fm_loopAcceptUsers (void *)
THREAD_RETURN	fm_loopReadFromUsers (void *)
THREAD_RETURN	fm_loopDoUserLogin (void *param)
THREAD_RETURN	fm_loopLog (void *)
THREAD_RETURN	iplist_loopDoLogCountries (void *param)

Macro Definition Documentation

TNC_CONFIRM_REQ

#define TNC_CONFIRM_REQ   "TermsAndConditionsConfirm"

Definition at line 69 of file CAFirstMix.hpp.

TNC_REQ_TRANSLATION

#define TNC_REQ_TRANSLATION   "Translation"

Definition at line 62 of file CAFirstMix.hpp.

TNC_RESOURCE_CUSTOMIZED_SECT

#define TNC_RESOURCE_CUSTOMIZED_SECT   "CustomizedSections"

Definition at line 66 of file CAFirstMix.hpp.

TNC_RESOURCE_TEMPLATE

#define TNC_RESOURCE_TEMPLATE   "Template"

Definition at line 64 of file CAFirstMix.hpp.

TNC_RESOURCES

#define TNC_RESOURCES   "Resources"

Definition at line 63 of file CAFirstMix.hpp.

TNC_RESPONSE

#define TNC_RESPONSE   "TermsAndConditionsResponse"

Definition at line 60 of file CAFirstMix.hpp.

TNC_RESPONSE_INVALID_REQUEST

#define TNC_RESPONSE_INVALID_REQUEST   "InvalidTermsAndConditionsRequest"

Definition at line 68 of file CAFirstMix.hpp.

TNC_SINTERRUPT

#define TNC_SINTERRUPT   "TermsAndConditionsInterrupt"

Definition at line 61 of file CAFirstMix.hpp.

TNC_SREQUEST

#define TNC_SREQUEST   "TermsAndConditionsRequest"

Definition at line 59 of file CAFirstMix.hpp.

TNC_TEMPLATE_ROOT_ELEMENT

#define TNC_TEMPLATE_ROOT_ELEMENT   "TermsAndConditionsTemplate"

Definition at line 65 of file CAFirstMix.hpp.

Typedef Documentation

tFirstMixChannelToQueueListEntry

typedef struct t_FirstMixChannelToQueueList_entry tFirstMixChannelToQueueListEntry

Function Documentation

fm_loopAcceptUsers()

THREAD_RETURN fm_loopAcceptUsers

(

void *

param

)

Definition at line 365 of file CAFirstMix.cpp.

  {
    INIT_STACK;
    BEGIN_STACK("CAFirstMix::fm_loopAcceptUsers");
    CAMsg::printMsg(LOG_DEBUG, "Start acceptusers\n");
 
    CAFirstMix* pFirstMix=(CAFirstMix*)param;
    CASocket** socketsIn=pFirstMix->m_arrSocketsIn;
    CAIPList* pIPList=pFirstMix->m_pIPList;
    CATempIPBlockList* pIPBlockList = pFirstMix->m_pIPBlockList;
    CAThreadPool* pthreadsLogin=pFirstMix->m_pthreadsLogin;
    UINT32 nSocketsIn=pFirstMix->m_nSocketsIn;
#ifdef __BUILD_AS_SHADOW_PLUGIN__
    CASocketGroupEpoll* psocketgroupAccept=new CASocketGroupEpoll(false);
#else
    CASocketGroup* psocketgroupAccept = new CASocketGroup(false);
#endif
    CAMuxSocket* pNewMuxSocket;
    UINT8* peerIP=new UINT8[4];
    UINT32 i=0;
    SINT32 countRead;
    SINT32 ret;
    SINT32 retPeerIP = E_SUCCESS;
 
    pFirstMix->m_newConnections = 0;
 
    // kick out users that already have connected
    for(i=0;i<nSocketsIn;i++)
    {
        while (socketsIn[i]->close() != E_SUCCESS)
        {
          sSleep(1);
        }
        delete socketsIn[i];
    }
    CAMsg::printMsg(LOG_DEBUG, "Creating incoming sockets\n");
 
    if (CALibProxytest::getOptions()->createSockets(false,pFirstMix-> m_arrSocketsIn, pFirstMix->m_nSocketsIn) != E_SUCCESS)
    {
      goto END_THREAD;
    }
    CAMsg::printMsg(LOG_DEBUG, "Add sockets to select group\n");
    for(i=0;i<nSocketsIn;i++)
    {
      psocketgroupAccept->add(*socketsIn[i]);
    }
#ifdef REPLAY_DETECTION //before we can start to accept users we have to ensure that we received the replay timestamps form the over mixes
    CAMsg::printMsg(LOG_DEBUG,"Waiting for Replay Timestamp from next mixes\n");
    i=0;
    while(!pFirstMix->m_bRestart && i < pFirstMix->m_u32MixCount-1)
    {
      if(pFirstMix->m_arMixParameters[i].m_u32ReplayOffset==0)//not set yet
        {
          msSleep(100);//wait a little bit and try again
          continue;
        }
      i++;
    }
    CAMsg::printMsg(LOG_DEBUG,"All Replay Timestamp received\n");
#endif
    CAMsg::printMsg(LOG_DEBUG, "Start accept users inner loop\n");
 
    while(!pFirstMix->m_bRestart)
      {
        if (pIPBlockList->count()>40)
        {
          CAMsg::printMsg(LOG_DEBUG,"UserAcceptLoop: login timeout list counts %d. We have %d users, %d open sockets and %d new connections. Restarting server sockets...\n",pIPBlockList->count(), pFirstMix->getNrOfUsers() ,CASocket::countOpenSockets(), pFirstMix->m_newConnections);
          for(i=0;i<nSocketsIn;i++)
          {
            psocketgroupAccept->remove(*socketsIn[i]);
            while (socketsIn[i]->close() != E_SUCCESS)
            {
              sSleep(1);
            }
          }
  
          if (CALibProxytest::getOptions()->createSockets(false,pFirstMix-> m_arrSocketsIn, pFirstMix->m_nSocketsIn) != E_SUCCESS)
          {
            // could not listen
            goto END_THREAD;
          }
          for(i=0;i<nSocketsIn;i++)
          {
            psocketgroupAccept->add(*socketsIn[i]);
          }
          sSleep(1);
        }
#ifdef  __BUILD_AS_SHADOW_PLUGIN__
        CAMsg::printMsg(LOG_DEBUG, "Before acceptusers->select()\n");
        countRead=psocketgroupAccept->select();
        CAMsg::printMsg(LOG_DEBUG, "after acceptusers->select()\n");
#else
        countRead = psocketgroupAccept->select(10000);
#endif
        if(countRead<0)
          { //check for Error - are we restarting ?
            if(pFirstMix->m_bRestart ||countRead!=E_TIMEDOUT)
              goto END_THREAD;
          }
        i=0;
#ifdef _DEBUG
        CAMsg::printMsg(LOG_DEBUG,"UserAcceptLoop: countRead=%i\n",countRead);
#endif
        while(countRead>0&&i<nSocketsIn)
        {
          if(psocketgroupAccept->isSignaled(*socketsIn[i]))
          {
            countRead--;
            #ifdef _DEBUG
              CAMsg::printMsg(LOG_DEBUG,"New direct Connection from Client!\n");
            #endif
            #ifdef SYM_CHANNEL_CIPHER_CTR
              pNewMuxSocket=new CAMuxSocket(CTR);
            #elif defined NO_ENCRYPTION
              pNewMuxSocket = new CAMuxSocket(NULL_CIPHER);
#else
              pNewMuxSocket=new CAMuxSocket(OFB);
            #endif
            ret=socketsIn[i]->accept(*(pNewMuxSocket->getCASocket()));
            pFirstMix->incNewConnections();
 
            if(ret!=E_SUCCESS)
            {
              // may return E_SOCKETCLOSED or E_SOCKET_LIMIT
              CAMsg::printMsg(LOG_ERR,"Accept Error %u - direct Connection from Client!\n",GET_NET_ERROR);
            }
            else if( (CALibProxytest::getOptions()->getMaxNrOfUsers() > 0 &&
                pFirstMix->getNrOfUsers() >= CALibProxytest::getOptions()->getMaxNrOfUsers())
                && (isAllowedToPassRestrictions(pNewMuxSocket->getCASocket()) != E_SUCCESS)
 
                )
            {
              CAMsg::printMsg(LOG_DEBUG,"CAFirstMix User control: Too many users (Maximum:%d)! Rejecting user...\n", pFirstMix->getNrOfUsers(),CALibProxytest::getOptions()->getMaxNrOfUsers());
              ret = E_UNKNOWN;
            }
            else if ((pFirstMix->m_newConnections > CAFirstMix::MAX_CONCURRENT_NEW_CONNECTIONS)
                && (isAllowedToPassRestrictions(pNewMuxSocket->getCASocket()) != E_SUCCESS)
                )
 
            {
              /* This should protect the mix from flooding attacks
               * No more than MAX_CONCURRENT_NEW_CONNECTIONS are allowed.
               */
#ifdef _DEBUG
              CAMsg::printMsg(LOG_DEBUG,"CAFirstMix Flooding protection: Too many concurrent new connections (Maximum:%d)! Rejecting user...\n", CAFirstMix::MAX_CONCURRENT_NEW_CONNECTIONS);
#endif
              ret = E_UNKNOWN;
            }
//#ifndef PAYMENT
            else if ((ret = pNewMuxSocket->getCASocket()->getPeerIP(peerIP)) != E_SUCCESS ||
                (retPeerIP = pIPList->insertIP(peerIP)) < 0) 
                // || (pIPBlockList->checkIP(peerIP) == E_UNKNOWN && isAllowedToPassRestrictions(pNewMuxSocket->getCASocket()) != E_SUCCESS))
            {
              if (ret != E_SUCCESS)
              {
                CAMsg::printMsg(LOG_DEBUG,"Could not insert IP address as IP could not be retrieved! We have %d login threads currently running.\n", pthreadsLogin->countRequests()); 
  
              }
              else if (retPeerIP < 0)
              {
                CAMsg::printMsg(LOG_DEBUG,"CAFirstMix Flooding protection: Could not insert IP address! We have %d login threads currently running.\n", pthreadsLogin->countRequests()); 
                pIPBlockList->insertIP(peerIP);
              }
              else if (pIPBlockList->checkIP(peerIP) == E_UNKNOWN)
              {
                CAMsg::printMsg(LOG_DEBUG, "Client IP address %u.%u.x.x is temporarily blocked! User login denied. We have %d open sockets and %d new connections.\n", peerIP[0],peerIP[1], CASocket::countOpenSockets(), pFirstMix->m_newConnections);
                pIPList->removeIP(peerIP);
              } 
              ret = E_UNKNOWN;
            }
//#endif
            else
            {
              t_UserLoginData* d=new t_UserLoginData;
              d->pNewUser=pNewMuxSocket;
              d->pMix=pFirstMix;
              memcpy(d->peerIP,peerIP,4);
#ifdef DEBUG
              CAMsg::printMsg(LOG_DEBUG,"%d concurrent client connections.\n", pFirstMix->m_newConnections);
#endif
              if(pthreadsLogin->addRequest(fm_loopDoUserLogin,d)!=E_SUCCESS)
              {
                CAMsg::printMsg(LOG_ERR,"Could not add an login request to the login thread pool!\n");
                ret=E_UNKNOWN;
              }
            }
 
            if (ret != E_SUCCESS)
            {
              delete pNewMuxSocket;
              pNewMuxSocket = NULL;
              pFirstMix->decNewConnections();
              if(ret==E_SOCKETCLOSED&&pFirstMix->m_bRestart) //Hm, should we restart ??
              {
                goto END_THREAD;
              }
              else //if(ret==E_SOCKET_LIMIT) // Hm no free sockets - wait some time to hope to get a free one...
              {
                msSleep(400);
              }
            }
          }
          i++;
        }
      }
END_THREAD:
    FINISH_STACK("CAFirstMix::fm_loopAcceptUsers");
 
    delete[] peerIP;
    peerIP = NULL;
    delete psocketgroupAccept;
    psocketgroupAccept = NULL;
 
    CAMsg::printMsg(LOG_DEBUG,"Exiting Thread AcceptUser\n");
    THREAD_RETURN_SUCCESS;
  }

fm_loopDoUserLogin()

THREAD_RETURN fm_loopDoUserLogin

(

void *

param

)

Definition at line 367 of file CAFirstMix.cpp.

  {
    INIT_STACK;
    BEGIN_STACK("CAFirstMix::fm_loopDoUserLogin");
#ifdef COUNTRY_STATS
    my_thread_init();
#endif
 
    t_UserLoginData* d=(t_UserLoginData*)param;
    d->pMix->doUserLogin(d->pNewUser,d->peerIP);
 
    SAVE_STACK("CAFirstMix::fm_loopDoUserLogin", "after user login");
    d->pMix->decNewConnections();
    delete d;
    d = NULL;
 
#ifdef COUNTRY_STATS
    my_thread_end();
#endif
    FINISH_STACK("CAFirstMix::fm_loopDoUserLogin");
 
    THREAD_RETURN_SUCCESS;
  }

fm_loopLog()

THREAD_RETURN fm_loopLog

(

void *

param

)

Definition at line 551 of file CAFirstMix.cpp.

  {
    CAFirstMix* pFirstMix=(CAFirstMix*)param;
    pFirstMix->m_bRunLog=true;
    UINT32 countLog=0;
    while(pFirstMix->m_bRunLog)
      {
        if(countLog==0)
          {
            logMemoryUsage();
            countLog=10;
          }
        sSleep(30);
        countLog--;
      }
    THREAD_RETURN_SUCCESS;
  }

fm_loopReadFromMix()

THREAD_RETURN fm_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 364 of file CAFirstMix.cpp.

  {
    INIT_STACK;
    BEGIN_STACK("CAFirstMix::fm_loopReadFromMix");
 
    CAFirstMix* pFirstMix=(CAFirstMix*)pParam;
    CAMuxSocket* pMuxSocket=pFirstMix->m_pMuxOut;
    CAQueue* pQueue=pFirstMix->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=pFirstMix->m_u32KeepAliveRecvInterval;
    getcurrentTimeMillis(keepaliveLast);
    CAControlChannelDispatcher* pControlChannelDispatcher=pFirstMix->m_pMuxOutControlChannelDispatcher;
    while (!pFirstMix->m_bRestart)
      {
      if (pQueue->getSize() > MAX_READ_FROM_NEXT_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 received a Keep_alive-Message
      getcurrentTimeMillis(keepaliveNow);
      UINT32 keepaliveDiff = diff64(keepaliveNow, keepaliveLast);
      if (keepaliveDiff > u32KeepAliveRecvInterval)
        {
        CAMsg::printMsg(LOG_DEBUG, "CAFirstMix::loopReadFromMix() -- restart because of KeepAlive-Traffic Timeout!\n");
        pFirstMix->m_bRestart = true;
        MONITORING_FIRE_NET_EVENT(ev_net_nextConnectionClosed);
        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
          */
#ifdef DEBUG
          CAMsg::printMsg(LOG_DEBUG, "CAFirstMix::loopReadFromMix() - socket select error: %i\n",ret);
#endif
          continue;
          }
        }
      else if (ret > 0)
        {
        ret = pMuxSocket->receive(pMixPacket);
#ifdef LOG_PACKET_TIMES
        getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_start);
#endif
        if (ret != MIXPACKET_SIZE)
          {
          pFirstMix->m_bRestart = true;
          CAMsg::printMsg(LOG_ERR, "CAFirstMix::lm_loopReadFromMix - received returned: %i -- restarting!\n", ret);
          MONITORING_FIRE_NET_EVENT(ev_net_nextConnectionClosed);
          break;
        }
        if (pMixPacket->channel > 0 && pMixPacket->channel < 256)
          {
#ifdef DEBUG
          CAMsg::printMsg(LOG_DEBUG, "CAFirstMix - sent a packet from the next mix to the ControlChanelDispatcher... \n");
#endif
          pControlChannelDispatcher->proccessMixPacket(pMixPacket);
          getcurrentTimeMillis(keepaliveLast);
          continue;
          }
      }
#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
#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, "Received Downstream AN.ON packet from previous Mix debug: %s\n", base64Payload);
          }
 
#endif
        pQueue->add(pQueueEntry, sizeof(tQueueEntry));
        getcurrentTimeMillis(keepaliveLast); 
        }
    delete pQueueEntry;
    pQueueEntry = NULL;
    delete pSocketGroup;
    pSocketGroup = NULL;
    #ifdef USE_POOL
      delete pPool;
      pPool = NULL;
    #endif
 
    FINISH_STACK("CAFirstMix::fm_loopReadFromMix");
    THREAD_RETURN_SUCCESS;
  }

fm_loopReadFromUsers()

THREAD_RETURN fm_loopReadFromUsers

(

void *

)

fm_loopSendToMix()

THREAD_RETURN fm_loopSendToMix

(

void *

param

)

How to end this thread: 0.

set bRestart=true;

1. Close connection to next mix
2. put some bytes (len>MIX_PACKET_SIZE) in the Mix-Output-Queue

Definition at line 363 of file CAFirstMix.cpp.

  {
    INIT_STACK;
    BEGIN_STACK("CAFirstMix::fm_loopSendToMix");
 
    CAFirstMix* pFirstMix=(CAFirstMix*)param;
    CAQueue* pQueue=((CAFirstMix*)param)->m_pQueueSendToMix;
    CAMuxSocket* pMuxSocket=pFirstMix->m_pMuxOut;
 
    UINT32 len;
    SINT32 ret;
 
/*#ifdef DATA_RETENTION_LOG
    t_dataretentionLogEntry* pDataRetentionLogEntry=new t_dataretentionLogEntry;
#endif
*/
#ifndef USE_POOL
    tQueueEntry* pQueueEntry=new tQueueEntry;
    MIXPACKET* pMixPacket=&pQueueEntry->packet;
    UINT32 u32KeepAliveSendInterval=pFirstMix->m_u32KeepAliveSendInterval;
    while(!pFirstMix->m_bRestart)
      {
        len=sizeof(tQueueEntry);
        ret=pQueue->getOrWait((UINT8*)pQueueEntry,&len,u32KeepAliveSendInterval);
        if(ret==E_TIMEDOUT)
          {//send a dummy as keep-alvie-traffic
            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,"CAFirstMix::lm_loopSendToMix - Error in dequeueing MixPaket\n");
            CAMsg::printMsg(LOG_ERR,"ret=%i len=%i\n",ret,len);
            break;
          }
 
        if(pMuxSocket->send(pMixPacket)!=MIXPACKET_SIZE)
          {
            CAMsg::printMsg(LOG_ERR,"CAFirstMix::lm_loopSendToMix - Error in sending MixPaket\n");
            break;
          }
#if defined (LOG_PACKET_TIMES)
        if(!isZero64(pQueueEntry->timestamp_proccessing_start))
          {
            getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end);
            pFirstMix->m_pLogPacketStats->addToTimeingStats(*pQueueEntry,pMixPacket->flags,true);
          }
#endif
#ifdef DATA_RETENTION_LOG
        if((pQueueEntry->packet.flags&CHANNEL_OPEN)!=0)
          {
            pQueueEntry->dataRetentionLogEntry.t_out=htonl(time(NULL));
            pFirstMix->m_pDataRetentionLog->log(&(pQueueEntry->dataRetentionLogEntry));
          }
#endif
      }
    delete pQueueEntry;
    pQueueEntry = NULL;
#else
    CAPool* pPool=new CAPool(MIX_POOL_SIZE);
    tPoolEntry* pPoolEntry=new tPoolEntry;
    MIXPACKET* pMixPacket=&pPoolEntry->packet;
    while(!pFirstMix->m_bRestart)
      {
        len=sizeof(tQueueEntry);
        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);
              pFirstMix->m_pLogPacketStats->addToTimeingStats(*pPoolEntry,pMixPacket->flags,true);
            }
        #endif
      }
    delete pPoolEntry;
    pPoolEntry = NULL;
    delete pPool;
    pPool = NULL;
#endif
 
/*#ifdef DATA_RETENTION_LOG
    delete pDataRetentionLogEntry;
#endif
*/
    FINISH_STACK("CAFirstMix::fm_loopSendToMix");
 
    CAMsg::printMsg(LOG_DEBUG,"Exiting Thread SendToMix\n");
    THREAD_RETURN_SUCCESS;
  }

iplist_loopDoLogCountries()

THREAD_RETURN iplist_loopDoLogCountries

(

void *

param

)

Definition at line 541 of file CAFirstMix.cpp.

  {
    mysql_thread_init();
    CAMsg::printMsg(LOG_DEBUG,"Starting iplist_loopDoLogCountries\n");
    CAFirstMix* pFirstMix=(CAFirstMix*)param;
    UINT32 s=0;
    UINT8 buff[255];
    memset(buff,0,255);
    CALibProxytest::getOptions()->getCascadeName(buff,255);
    mysqlEscapeTableName(buff);
    while(pFirstMix->m_bRunLogCountries)
      {
        if(s==LOG_COUNTRIES_INTERVALL)
          {
            UINT8 aktDate[255];
            time_t aktTime=time(NULL);
            strftime((char*)aktDate,255,"%Y%m%d%H%M%S",gmtime(&aktTime));
            char query[1024];
            sprintf(query,"INSERT into `stats_%s` (date,id,count,packets_in,packets_out) VALUES (\"%s\",\"%%u\",\"%%u\",\"%%u\",\"%%u\")",buff,aktDate);
            pFirstMix->m_pmutexUser->lock();
            for(UINT32 i=0;i<NR_OF_COUNTRIES+1;i++)
              {
                if(pFirstMix->m_CountryStats[i]>0)
                  {
                    char aktQuery[1024];
                    sprintf(aktQuery,query,i,pFirstMix->m_CountryStats[i],pFirstMix->m_PacketsPerCountryIN[i].getAndzero(),pFirstMix->m_PacketsPerCountryOUT[i].getAndzero());
                    SINT32 ret=mysql_query(pFirstMix->m_mysqlCon,aktQuery);
                    if(ret!=0)
                    {
                      CAMsg::printMsg(LOG_INFO,"CountryStats DB - failed to update CountryStats DB with new values - error %i\n",ret);
                    }
                  }
              }
            pFirstMix->m_pmutexUser->unlock();
            s=0;
          }
        sSleep(10);
        s++;
      }
    CAMsg::printMsg(LOG_DEBUG,"Exiting iplist_loopDoLogCountries\n");
    mysql_thread_end();
    THREAD_RETURN_SUCCESS;
  }

References CAMsg::printMsg().

Here is the call graph for this function: