#include <CAFirstMixB.hpp>

Inheritance diagram for CAFirstMixB:

Collaboration diagram for CAFirstMixB:

Protected Member Functions
SINT32	loop ()

Protected Member Functions inherited from CAFirstMix
bool	isShuttingDown ()

SINT32	init ()

SINT32	clean ()

virtual SINT32	initOnce ()

virtual SINT32	processKeyExchange ()

SINT32	initMixParameters (DOMElement *elemMixes)
	Initialises the MixParameters info for each mix form the <Mixes> element received from the second mix. More...

SINT32	incUsers (LP_fmHashTableEntry pHashEntry)

SINT32	decUsers (LP_fmHashTableEntry pHashEntry)

SINT32	incMixedPackets ()

SINT32	doUserLogin (CAMuxSocket *pNewUSer, UINT8 perrIP[4])

SINT32	reconfigure ()

SINT32	deleteCountryStats ()

Protected Member Functions inherited from CAMix
SINT32	checkCompatibility (DOMNode a_parent, const char a_mixPosition)

SINT32	appendCompatibilityInfo (DOMNode *a_parent)

SINT32	addMixInfo (DOMNode *a_element, bool a_bForceFirstNode)

virtual SINT32	initMixCascadeInfo (DOMElement *elemMixes)
	This will initialize the XML Cascade Info struct XMLFirstMixToInfoService that is sent to the InfoService in `CAInfoService::sendCascadeHelo()` More...

SINT32	signXML (DOMNode *a_element)

Additional Inherited Members
Public Types inherited from CAMix
enum	tMixType { FIRST_MIX , MIDDLE_MIX , LAST_MIX , JAP }

Public Member Functions inherited from CAFirstMix
	CAFirstMix ()

virtual	~CAFirstMix ()

tMixType	getType () const

bool	forceKickout (fmHashTableEntry pHashTableEntry, const XERCES_CPP_NAMESPACE::DOMDocument pErrDoc=NULL)

CAMutex *	getLoginMutex ()

SINT32	connectToNextMix (CASocketAddr *a_pAddrNext)

SINT32	getMixedPackets (UINT64 &ppackets)

UINT32	getNrOfUsers ()

SINT32	getLevel (SINT32 puser, SINT32 prisk, SINT32 *ptraffic)

TermsAndConditions *	getTermsAndConditions (const UINT8 *opSki)

DOMNode *	getTermsAndConditionsTemplate (UINT8 *templateRefID)

SINT32	getMixCount ()

tMixParameters *	getMixParameters ()
	Returns the ordered list of the mix parameters from the first mix to the last mix. More...

SINT32	setMixParameters (const tMixParameters &params)
	Sets the parameters for the mix specified in the params.m_strMixID field. More...

SINT32	handleKeyInfoExtensions (DOMElement *root)

SINT32	handleTermsAndConditionsExtension (DOMElement *extensionRoot)

Public Member Functions inherited from CAMixWithReplayDB
	CAMixWithReplayDB ()

CADatabase *	getReplayDB () const

Public Member Functions inherited from CAMix
	CAMix ()

virtual	~CAMix ()

SINT32	start ()

virtual void	shutDown ()

virtual bool	isShutDown ()

SINT32	getMixCascadeInfo (XERCES_CPP_NAMESPACE::DOMDocument *&docMixCascadeInfo)
	Returns the Mix-Cascade info which should be send to the InfoService. More...

bool	acceptsReconfiguration ()

CAControlChannelDispatcher *	getDownstreamControlChannelDispatcher () const

CAControlChannelDispatcher *	getUpstreamControlChannelDispatcher () const

UINT32	getLastConnectionTime ()

bool	isConnected ()

Public Attributes inherited from CAFirstMix
UINT64	m_u64LastTimestampReceived

Public Attributes inherited from CAMixWithReplayDB
UINT64	m_u64ReferenceTime

Static Public Attributes inherited from CAMix
static const UINT32	TIMEOUT_MIX_CONNECTION_ESTABLISHEMENT = 60000

Protected Attributes inherited from CAFirstMix
CAIPList *	m_pIPList

CATempIPBlockList *	m_pIPBlockList

CAQueue *	m_pQueueSendToMix

CAQueue *	m_pQueueReadFromMix

volatile UINT32	m_nUser

UINT32	m_nSocketsIn

volatile bool	m_bRestart

CASocket **	m_arrSocketsIn

UINT32	m_u32MixCount

tMixParameters *	m_arMixParameters

CAFirstMixChannelList *	m_pChannelList

CASocketGroupEpoll *	m_psocketgroupUsersRead

CASocketGroupEpoll *	m_psocketgroupUsersWrite

CAMuxSocket *	m_pMuxOut

UINT8 *	m_xmlKeyInfoBuff

UINT16	m_xmlKeyInfoSize

XERCES_CPP_NAMESPACE::DOMDocument *	m_docMixCascadeInfo

UINT64	m_nMixedPackets

CAASymCipher *	m_pRSA

CAMutex *	m_pmutexUser

CAMutex *	m_pmutexMixedPackets

CAMutex *	m_pmutexLoginThreads

CAThread *	m_pthreadAcceptUsers

CAThreadPool *	m_pthreadsLogin

CAThread *	m_pthreadSendToMix

CAThread *	m_pthreadReadFromMix

UINT32	m_nrOfTermsAndConditionsDefs

TermsAndConditions **	m_tnCDefs

UINT32	m_nrOfTermsAndConditionsTemplates

DOMNode **	m_tcTemplates

XERCES_CPP_NAMESPACE::DOMDocument *	m_templatesOwner

const XMLCh *	TNC_REQUEST

const XMLCh *	TNC_CONFIRM

const XMLCh *	TNC_INTERRUPT

tUINT32withLock *	m_PacketsPerCountryIN

tUINT32withLock *	m_PacketsPerCountryOUT

bool	m_bIsShuttingDown

volatile bool	m_bRunLog

CAMutex *	m_pmutexLogin

Protected Attributes inherited from CAMixWithReplayDB
CADatabase *	m_pReplayDB

CAReplayCtrlChannelMsgProc *	m_pReplayMsgProc

Protected Attributes inherited from CAMix
volatile bool	m_bLoop

bool	m_bReconfiguring

volatile bool	m_bShutDown

CAMultiSignature *	m_pMultiSignature

CAInfoService *	m_pInfoService

UINT32	m_u32KeepAliveRecvInterval

UINT32	m_u32KeepAliveSendInterval

bool	m_acceptReconfiguration

volatile bool	m_bConnected

volatile UINT32	m_lLastConnectionTime

XERCES_CPP_NAMESPACE::DOMDocument *	m_docMixCascadeInfo

CAControlChannelDispatcher *	m_pMuxOutControlChannelDispatcher

CAControlChannelDispatcher *	m_pMuxInControlChannelDispatcher

Detailed Description

Definition at line 34 of file CAFirstMixB.hpp.

Member Function Documentation

◆ loop()

SINT32 CAFirstMixB::loop ( )

protectedvirtual

Implements CAFirstMix.

Definition at line 41 of file CAFirstMixB.cpp.

 {
 #ifdef NEW_MIX_TYPE
   /* should only be compiled if new NEW_MIX_TYPE is defined */
 #ifdef DELAY_USERS
   m_pChannelList->setDelayParameters(pglobalOptions->getDelayChannelUnlimitTraffic(),
     pglobalOptions->getDelayChannelBucketGrow(),
     pglobalOptions->getDelayChannelBucketGrowIntervall());
 #endif    
  
   //  CASingleSocketGroup osocketgroupMixOut;
   SINT32 countRead;
   //#ifdef LOG_PACKET_TIMES
   //  tPoolEntry* pPoolEntry=new tPoolEntry;
   //  MIXPACKET* pMixPacket=&pPoolEntry->mixpacket;
   //#else
   tQueueEntry* pQueueEntry = new tQueueEntry;
   MIXPACKET* pMixPacket = &pQueueEntry->packet;
   //#endif  
   m_nUser = 0;
   SINT32 ret;
   //osocketgroupMixOut.add(*m_pMuxOut);
  
   UINT8* tmpBuff = new UINT8[sizeof(tQueueEntry)];
   CAMsg::printMsg(LOG_DEBUG, "Starting Message Loop... \n");
   bool bAktiv;
   UINT8 rsaBuff[RSA_SIZE];
 #ifdef LOG_TRAFFIC_PER_USER
   UINT64 current_time;
   UINT32 diff_time;
   CAMsg::printMsg(LOG_DEBUG, "Channel log formats:\n");
   CAMsg::printMsg(LOG_DEBUG, "1. Close received from user (times in micros) - 1:Channel-ID,Connection-ID,PacketsIn (only data and open),PacketsOut (only data),ChannelDuration (open packet received --> close packet put into send queue to next mix)\n");
   CAMsg::printMsg(LOG_DEBUG, "2. Channel close from Mix(times in micros)- 2.:Channel-ID,Connection-ID,PacketsIn (only data and open), PacketsOut (only data),ChannelDuration (open packet received)--> close packet put into send queue to next user\n");
 #endif
   //    CAThread threadReadFromUsers;
   //    threadReadFromUsers.setMainLoop(loopReadFromUsers);
   //    threadReadFromUsers.start(this);
  
   while (!m_bRestart)                                                            /* the main mix loop as long as there are things that are not handled by threads. */
   {
     bAktiv = false;
     //LOOP_START:
  
     //First Step
     //Checking for new connections    
     // Now in a separat Thread.... 
  
     // Second Step 
     // Checking for data from users
     // Now in a separate Thread (see loopReadFromUsers())
     //Only proccess user data, if queue to next mix is not to long!!
 #define MAX_NEXT_MIX_QUEUE_SIZE 10000000 //10 MByte
     if (m_pQueueSendToMix->getSize() < MAX_NEXT_MIX_QUEUE_SIZE)
     {
       countRead = m_psocketgroupUsersRead->select(/*false,*/0);        // how many JAP<->mix connections have received data from their coresponding JAP
       if (countRead > 0)
         bAktiv = true;
 #ifdef HAVE_EPOLL
       //if we have epool we do not need to search the whole list
       //of connected JAPs to find the ones who have sent data
       //as epool will return ONLY these connections.
       fmHashTableEntry* pHashEntry = (fmHashTableEntry*)m_psocketgroupUsersRead->getFirstSignaledSocketData();
       while (pHashEntry != NULL)
       {
         CAMuxSocket* pMuxSocket = pHashEntry->pMuxSocket;
 #else
       //if we do not have epoll we have to go to the whole
       //list of open connections to find the ones which
       //actually have sent some data
       fmHashTableEntry* pHashEntry = m_pChannelList->getFirst();
       while (pHashEntry != NULL&&countRead > 0)                      // iterate through all connections as long as there is at least one active left
       {
         CAMuxSocket* pMuxSocket = pHashEntry->pMuxSocket;
         if (m_psocketgroupUsersRead->isSignaled(*pMuxSocket))  // if this one seems to have data
         {
           countRead--;
 #endif
           ret = pMuxSocket->receive(pMixPacket, 0);
 #if defined LOG_PACKET_TIMES||defined(LOG_CHANNEL)
           getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_start);
           set64(pQueueEntry->timestamp_proccessing_start_OP, pQueueEntry->timestamp_proccessing_start);
 #endif  
           if (ret == SOCKET_ERROR/*||pHashEntry->accessUntil<time()*/)
           {
             // remove dead connections
 #ifdef LOG_TRAFFIC_PER_USER
             getcurrentTimeMillis(current_time);
             diff_time = diff64(current_time, pHashEntry->timeCreated);
             m_pIPList->removeIP(pHashEntry->peerIP, diff_time, pHashEntry->trafficIn, pHashEntry->trafficOut);
 #else
             m_pIPList->removeIP(pHashEntry->peerIP);
 #endif
             m_psocketgroupUsersRead->remove(pMuxSocket->getSocket());
             m_psocketgroupUsersWrite->remove(pMuxSocket->getSocket());
             ASSERT(pHashEntry->pQueueSend != NULL, "Send queue is NULL");
             delete pHashEntry->pQueueSend;
             pHashEntry->pQueueSend = NULL;
             delete pHashEntry->pSymCipher;
             pHashEntry->pSymCipher = NULL;
 #ifdef COUNTRY_STATS
             decUsers(pHashEntry);
 #else
             decUsers();
 #endif
             /* remove the client part (MuxSocket) but keep the
              * outgoing channels until a CHANNEL-CLOSE from the
              * last mix is received
              */
             m_pChannelList->removeClientPart(pMuxSocket);
             delete pMuxSocket;
             pMuxSocket = NULL;
           }
           else if (ret == MIXPACKET_SIZE)                       // we've read enough data for a whole mix packet. nice!
           {
 #ifdef LOG_TRAFFIC_PER_USER
             pHashEntry->trafficIn++;
 #endif
 #ifdef COUNTRY_STATS
             m_PacketsPerCountryIN[pHashEntry->countryID].inc();
 #endif  
             //New control channel code...!
             SINT32 ret = 0;
             if (pMixPacket->channel > 0 && pMixPacket->channel < 256)
             {
               if (pHashEntry->pControlChannelDispatcher->proccessMixPacket(pMixPacket))
               {
                 goto NEXT_USER;
               }
               else
               {
                 CAMsg::printMsg(LOG_DEBUG, "Packet is invalid and could not be processed!\n");
                 ret = 3;
               }
             }
 #ifdef PAYMENT
             // payment code added by Bastian Voigt
             if (ret == 0)
             {
               ret = CAAccountingInstance::handleJapPacket(pHashEntry);
             }
             if (ret == 2)
             {
               goto NEXT_USER;
             }
 #endif                                                  
             if (ret == 3)
             {
               // this jap is evil! terminate connection and add IP to blacklist
               CAMsg::printMsg(LOG_DEBUG, "CAFirstMixB: Detected evil Jap.. closing connection! Removing IP..\n", ret);
               fmChannelListEntry* pEntry;
               pEntry = m_pChannelList->getFirstChannelForSocket(pMuxSocket);
               while (pEntry != NULL)
               {
                 getRandom(pMixPacket->data, DATA_SIZE);
                 pMixPacket->flags = CHANNEL_CLOSE;
                 pMixPacket->channel = pEntry->channelOut;
 #ifdef LOG_PACKET_TIMES
                 setZero64(pQueueEntry->timestamp_proccessing_start);
 #endif
                 m_pQueueSendToMix->add(pMixPacket, sizeof(tQueueEntry));
                 delete pEntry->pCipher;
                 pEntry->pCipher = NULL;
                 pEntry = m_pChannelList->getNextChannel(pEntry);
               }
               m_pIPList->removeIP(pHashEntry->peerIP);
               m_psocketgroupUsersRead->remove(pMuxSocket->getSocket());
               m_psocketgroupUsersWrite->remove(pMuxSocket->getSocket());
               delete pHashEntry->pQueueSend;
               pHashEntry->pQueueSend = NULL;
               delete pHashEntry->pSymCipher;
               pHashEntry->pSymCipher = NULL;
               m_pChannelList->remove(pMuxSocket);
               delete pMuxSocket;
               pMuxSocket = NULL;
               decUsers();
               goto NEXT_USER;
             }
  
             if (pMixPacket->flags == CHANNEL_DUMMY)          // just a dummy to keep the connection alife in e.g. NAT gateways 
             {
               getRandom(pMixPacket->data, DATA_SIZE);
 #ifdef LOG_PACKET_TIMES
               setZero64(pQueueEntry->timestamp_proccessing_start);
 #endif
               pHashEntry->pQueueSend->add(pMixPacket, sizeof(tQueueEntry));
 #ifdef HAVE_EPOLL
               m_psocketgroupUsersWrite->add(*pMuxSocket, pHashEntry);
 #else
               m_psocketgroupUsersWrite->add(*pMuxSocket);
 #endif
             }
             else                                         // finally! a normal mix packet
             {
               CASymCipher* pCipher = NULL;
               fmChannelListEntry* pEntry;
               pEntry = m_pChannelList->get(pMuxSocket, pMixPacket->channel);
               if (pEntry != NULL&&pMixPacket->flags == CHANNEL_DATA)
               {
                 pMixPacket->channel = pEntry->channelOut;
                 pCipher = pEntry->pCipher;
                 pCipher->crypt1(pMixPacket->data, pMixPacket->data, DATA_SIZE);
                 // queue the packet for sending to the next mix.
 #ifdef LOG_PACKET_TIMES
                 getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end_OP);
 #endif
                 m_pQueueSendToMix->add(pMixPacket, sizeof(tQueueEntry));
                 incMixedPackets();
 #ifdef LOG_CHANNEL
                 pEntry->packetsInFromUser++;
 #endif
               }
               else if (pEntry == NULL&&pMixPacket->flags == CHANNEL_OPEN)  // open a new mix channel
               { // stefan: muesste das nicht vor die behandlung von CHANNEL_DATA? oder gilt OPEN => !DATA ? 
                 //es gilt: open -> data
                 pHashEntry->pSymCipher->crypt1(pMixPacket->data, rsaBuff, KEY_SIZE);
                 pCipher = new CASymCipher();
                 pCipher->setKey(rsaBuff);
                 for (int i = 0; i < 16; i++)
                   rsaBuff[i] = 0xFF;
                 pCipher->setIV2(rsaBuff);
                 pCipher->crypt1(pMixPacket->data + KEY_SIZE, pMixPacket->data, DATA_SIZE - KEY_SIZE);
                 getRandom(pMixPacket->data + DATA_SIZE - KEY_SIZE, KEY_SIZE);
 #ifdef LOG_CHANNEL
                 HCHANNEL tmpC = pMixPacket->channel;
 #endif
                 if (m_pChannelList->addChannel(pMuxSocket, pMixPacket->channel, pCipher, &pMixPacket->channel) != E_SUCCESS)
                 { //todo move up ?
                   delete pCipher;
                   pCipher = NULL;
                 }
                 else
                 {
 #ifdef LOG_PACKET_TIMES
                   getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end_OP);
 #endif
 #ifdef LOG_CHANNEL
                   fmChannelListEntry* pTmpEntry = m_pChannelList->get(pMuxSocket, tmpC);
                   pTmpEntry->packetsInFromUser++;
                   set64(pTmpEntry->timeCreated, pQueueEntry->timestamp_proccessing_start);
 #endif
                   m_pQueueSendToMix->add(pMixPacket, sizeof(tQueueEntry));
                   incMixedPackets();
 #ifdef _DEBUG
                   //                                      CAMsg::printMsg(LOG_DEBUG,"Added out channel: %u\n",pMixPacket->channel);
 #endif
                 }
               }
             }
           }
 #ifdef HAVE_EPOLL
           NEXT_USER :
                 pHashEntry = (fmHashTableEntry*)m_psocketgroupUsersRead->getNextSignaledSocketData();
 #else
         }//if is signaled
       NEXT_USER:
         pHashEntry = m_pChannelList->getNext();
 #endif
       }
       }
     //Third step
     //Sending to next mix
  
     // Now in a separate Thread (see loopSendToMix())
  
     //Step 4
     //Stepa 4a Receiving form Mix to Queue now in separat Thread
     //Step 4b Proccesing MixPackets received from Mix
     //todo check for error!!!
     countRead = m_nUser + 1;
     while (countRead > 0 && m_pQueueReadFromMix->getSize() >= sizeof(tQueueEntry))
     {
       bAktiv = true;
       countRead--;
       ret = sizeof(tQueueEntry);
       m_pQueueReadFromMix->get((UINT8*)pQueueEntry, (UINT32*)&ret);
 #ifdef LOG_PACKET_TIMES
       getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_start_OP);
 #endif
       if (pMixPacket->flags == CHANNEL_CLOSE) //close event
       {
 #if defined(_DEBUG) && !defined(__MIX_TEST)
         //                  CAMsg::printMsg(LOG_DEBUG,"Closing Channel: %u ...\n",pMixPacket->channel);
 #endif
         fmChannelList* pEntry = m_pChannelList->get(pMixPacket->channel);
         if (pEntry != NULL)
         {
           if (pEntry->pHead != NULL) {
             pMixPacket->channel = pEntry->channelIn;
             getRandom(pMixPacket->data, DATA_SIZE);
 #ifdef LOG_PACKET_TIMES
             getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end_OP);
 #endif
             pEntry->pHead->pQueueSend->add(pMixPacket, sizeof(tQueueEntry));
 #ifdef LOG_TRAFFIC_PER_USER
             pEntry->pHead->trafficOut++;
 #endif
 #ifdef COUNTRY_STATS
             m_PacketsPerCountryOUT[pEntry->pHead->countryID].inc();
 #endif  
 #ifdef LOG_CHANNEL  
             //pEntry->packetsOutToUser++;
             getcurrentTimeMicros(current_time);
             diff_time = diff64(current_time, pEntry->timeCreated);
             CAMsg::printMsg(LOG_DEBUG, "2:%u,%Lu,%u,%u,%u\n",
               pEntry->channelIn, pEntry->pHead->id, pEntry->packetsInFromUser, pEntry->packetsOutToUser,
               diff_time);
 #endif
  
 #ifdef HAVE_EPOLL
             m_psocketgroupUsersWrite->add(*pEntry->pHead->pMuxSocket, pEntry->pHead);
 #else
             m_psocketgroupUsersWrite->add(*pEntry->pHead->pMuxSocket);
 #endif                    
             delete pEntry->pCipher;
             pEntry->pCipher = NULL;
             m_pChannelList->removeChannel(pEntry->pHead->pMuxSocket, pEntry->channelIn);
           }
           else {
             /* the client has already closed the connection but we
              * have waited for the CHANNEL-CLOSE packet from the
              * last mix -> now we have it
              */
 #ifdef LOG_PACKET_TIMES
             getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end_OP);
 #endif
             delete pEntry->pCipher;
             pEntry->pCipher = NULL;
             m_pChannelList->removeVacantOutChannel(pEntry);
             pEntry = NULL;
           }
         }
       }
       else
       {//flag !=close
 #if defined(_DEBUG) && !defined(__MIX_TEST)
 //                  CAMsg::printMsg(LOG_DEBUG,"Sending Data to Browser!\n");
 #endif
         fmChannelList* pEntry = m_pChannelList->get(pMixPacket->channel);
         if (pEntry != NULL) {
           if (pEntry->pHead != NULL) {
 #ifdef LOG_CRIME
             if ((pMixPacket->flags&CHANNEL_SIG_CRIME) == CHANNEL_SIG_CRIME)
             {
               UINT32 id = (pMixPacket->flags >> 8) & 0x000000FF;
               int log = LOG_ENCRYPTED;
               if (!pglobalOptions->isEncryptedLogEnabled())
                 log = LOG_CRIT;
               CAMsg::printMsg(log, "Detecting crime activity - ID: %u -- In-IP is: %u.%u.%u.%u \n", id, pEntry->pHead->peerIP[0], pEntry->pHead->peerIP[1], pEntry->pHead->peerIP[2], pEntry->pHead->peerIP[3]);
               continue;
             }
 #endif
             pMixPacket->channel = pEntry->channelIn;
             pEntry->pCipher->crypt2(pMixPacket->data, pMixPacket->data, DATA_SIZE);
  
 #ifdef LOG_PACKET_TIMES
             getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end_OP);
 #endif
             pEntry->pHead->pQueueSend->add(pMixPacket, sizeof(tQueueEntry));
 #ifdef LOG_TRAFFIC_PER_USER
             pEntry->pHead->trafficOut++;
 #endif
 #ifdef COUNTRY_STATS
             m_PacketsPerCountryOUT[pEntry->pHead->countryID].inc();
 #endif  
 #ifdef LOG_CGANNEL  
             pEntry->packetsOutToUser++;
 #endif
 #ifdef HAVE_EPOLL
             m_psocketgroupUsersWrite->add(*pEntry->pHead->pMuxSocket, pEntry->pHead);
 #else
             m_psocketgroupUsersWrite->add(*pEntry->pHead->pMuxSocket);
 #endif    
             incMixedPackets();
           }
           else {
             /* connection to client is already closed -> we wait for
              * CLOSE-CHANNEL from last mix (but this is no CLOSE-
              * CHANNEL packet -> do only some compatibility things and
              * ignore the packet)
              */
 #ifdef LOG_CRIME
              /* we don't have the user-information any more (but also
               * the user cannot receive any data); nevertheless write
               * a log-message to show what happened
               */
             if ((pMixPacket->flags&CHANNEL_SIG_CRIME) == CHANNEL_SIG_CRIME) {
               UINT32 id = (pMixPacket->flags >> 8) & 0x000000FF;
               int log = LOG_ENCRYPTED;
               if (!pglobalOptions->isEncryptedLogEnabled()) {
                 log = LOG_CRIT;
               }
               CAMsg::printMsg(log, "Detecting crime activity - ID: %u -- In-IP is: not available (user has already closed connection)\n", id);
               continue;
             }
 #endif
 #ifdef LOG_PACKET_TIMES
             getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end_OP);
 #endif
           }
         }
         else
         {
 #ifdef _DEBUG
           if (pMixPacket->flags != CHANNEL_DUMMY)
           {
             /*                          CAMsg::printMsg(LOG_DEBUG,"Error Sending Data to Browser -- "
                             "Channel-Id %u not valid!\n",pMixPacket->channel
                           );*/
 #ifdef LOG_CHANNEL
             CAMsg::printMsg(LOG_INFO, "Packet late arrive for channel: %u\n", pMixPacket->channel);
 #endif
           }
 #endif
         }
       }
     }
  
     //Step 5 
     //Writing to users...
     countRead = m_psocketgroupUsersWrite->select(/*true,*/0);
 #ifdef HAVE_EPOLL    
     fmHashTableEntry* pfmHashEntry = (fmHashTableEntry*)m_psocketgroupUsersWrite->getFirstSignaledSocketData();
     while (pfmHashEntry != NULL)
     {
 #else
     fmHashTableEntry* pfmHashEntry = m_pChannelList->getFirst();
     while (countRead > 0 && pfmHashEntry != NULL)
     {
       if (m_psocketgroupUsersWrite->isSignaled(*pfmHashEntry->pMuxSocket))
       {
         countRead--;
 #endif
 #ifdef DELAY_USERS
         if (pfmHashEntry->delayBucket > 0)
         {
 #endif
           if (pfmHashEntry->pQueueSend->getSize() > 0)
           {
             bAktiv = true;
             UINT32 len = sizeof(tQueueEntry);
             if (pfmHashEntry->uAlreadySendPacketSize == -1)
             {
               pfmHashEntry->pQueueSend->get((UINT8*)&pfmHashEntry->oQueueEntry, &len);
 #ifdef PAYMENT
               //do not count control channel packets!
               if (pfmHashEntry->oQueueEntry.packet.channel > 0 && pfmHashEntry->oQueueEntry.packet.channel < 256)
                 pfmHashEntry->bCountPacket = false;
               else
                 pfmHashEntry->bCountPacket = true;
 #endif
               pfmHashEntry->pMuxSocket->prepareForSend(&(pfmHashEntry->oQueueEntry.packet));
               pfmHashEntry->uAlreadySendPacketSize = 0;
             }
             len = MIXPACKET_SIZE - pfmHashEntry->uAlreadySendPacketSize;
             ret = pfmHashEntry->pMuxSocket->getCASocket()->send(((UINT8*)&(pfmHashEntry->oQueueEntry)) + pfmHashEntry->uAlreadySendPacketSize, len);
             if (ret > 0)
             {
               pfmHashEntry->uAlreadySendPacketSize += ret;
               if (pfmHashEntry->uAlreadySendPacketSize == MIXPACKET_SIZE)
               {
 #ifdef PAYMENT
                 if (pfmHashEntry->bCountPacket)
                 {
                   // count packet for payment
                   if (CAAccountingInstance::handleJapPacket(pfmHashEntry) == 2)
                   {
                     goto NEXT_USER_WRITING;
                   }
                 }
 #endif
 #ifdef DELAY_USERS
                 pfmHashEntry->delayBucket--;
 #endif
                 pfmHashEntry->uAlreadySendPacketSize = -1;
 #ifdef LOG_PACKET_TIMES
                 if (!isZero64(pfmHashEntry->oQueueEntry.timestamp_proccessing_start))
                 {
                   getcurrentTimeMicros(pfmHashEntry->oQueueEntry.timestamp_proccessing_end);
                   m_pLogPacketStats->addToTimeingStats(pfmHashEntry->oQueueEntry, CHANNEL_DATA, false);
                 }
 #endif
               }
             }
           }
 #ifdef DELAY_USERS
         }
 #endif
         //todo error handling
 #ifdef HAVE_EPOLL
         NEXT_USER_WRITING :
                   pfmHashEntry = (fmHashTableEntry*)m_psocketgroupUsersWrite->getNextSignaledSocketData();
 #else
       }//if is socket signaled
     NEXT_USER_WRITING:
       pfmHashEntry = m_pChannelList->getNext();
 #endif
     }
     if (!bAktiv)
       msSleep(100);
     }
   //ERR:
   //@todo move cleanup to clean() !
   CAMsg::printMsg(LOG_CRIT, "Seams that we are restarting now!!\n");
   m_bRestart = true;
   m_pMuxOut->close();
   for (UINT32 i = 0; i < m_nSocketsIn; i++)
     m_arrSocketsIn[i].close();
   //writng some bytes to the queue...
   UINT8 b[sizeof(tQueueEntry) + 1];
   m_pQueueSendToMix->add(b, sizeof(tQueueEntry) + 1);
   //#if !defined(_DEBUG) && !defined(NO_LOOPACCEPTUSER)
   CAMsg::printMsg(LOG_CRIT, "Wait for LoopAcceptUsers!\n");
   m_pthreadAcceptUsers->join();
   //#endif
   CAMsg::printMsg(LOG_CRIT, "Wait for LoopSendToMix!\n");
   m_pthreadSendToMix->join(); //will not join if queue is empty (and so wating)!!!
   CAMsg::printMsg(LOG_CRIT, "Wait for LoopReadFromMix!\n");
   m_pthreadReadFromMix->join();
 #ifdef LOG_PACKET_TIMES
   CAMsg::printMsg(LOG_CRIT, "Wait for LoopLogPacketStats to terminate!\n");
   m_pLogPacketStats->stop();
 #endif  
   //waits until all login threads terminates....
   // we have to be sure that the Accept thread was alread stoped!
   m_pthreadsLogin->destroy(true);
   CAMsg::printMsg(LOG_CRIT, "Before deleting CAFirstMixChannelList()!\n");
   CAMsg::printMsg(LOG_CRIT, "Memeory usage before: %u\n", getMemoryUsage());
   fmHashTableEntry* pHashEntry = m_pChannelList->getFirst();
   while (pHashEntry != NULL)
   {
     CAMuxSocket * pMuxSocket = pHashEntry->pMuxSocket;
     delete pHashEntry->pQueueSend;
     pHashEntry->pQueueSend = NULL;
     delete pHashEntry->pSymCipher;
     pHashEntry->pSymCipher = NULL;
  
     fmChannelListEntry* pEntry = m_pChannelList->getFirstChannelForSocket(pHashEntry->pMuxSocket);
     while (pEntry != NULL)
     {
       delete pEntry->pCipher;
       pEntry->pCipher = NULL;
  
       pEntry = m_pChannelList->getNextChannel(pEntry);
     }
     m_pChannelList->remove(pHashEntry->pMuxSocket);
     pMuxSocket->close();
     delete pMuxSocket;
     pMuxSocket = NULL;
     pHashEntry = m_pChannelList->getNext();
   }
   /* clean all vacant out-channels (the connection from the client was
    * closed but we've waited for the CHANNEL-CLOSE from the last mix)
    */
   m_pChannelList->cleanVacantOutChannels();
   CAMsg::printMsg(LOG_CRIT, "Memory usage after: %u\n", getMemoryUsage());
   delete pQueueEntry;
   pQueueEntry = NULL;
   delete[]tmpBuff;
   tmpBuff = NULL;
   CAMsg::printMsg(LOG_CRIT, "Main Loop exited!!\n");
 #endif // NEW_MIX_TYPE
   return E_UNKNOWN;
 }

Here is the call graph for this function:

The documentation for this class was generated from the following files:

Protected Member Functions

Additional Inherited Members

Detailed Description

Member Function Documentation

◆ loop()