CAFirstMixA Class Reference

#include <CAFirstMixA.hpp>

Inheritance diagram for CAFirstMixA:

Collaboration diagram for CAFirstMixA:

Public Member Functions
virtual void	shutDown ()
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 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 ()

Protected Member Functions
SINT32	loop ()
SINT32	closeConnection (fmHashTableEntry *pHashEntry)
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)

Private Member Functions
bool	sendToUsers ()
void	notifyAllUserChannels (fmHashTableEntry *pfmHashEntry, UINT16 flags)
void	finishPacket (fmHashTableEntry *pfmHashEntry)
void	checkUserConnections ()
SINT32	accountTrafficUpstream (fmHashTableEntry *pHashEntry)
SINT32	accountTrafficDownstream (fmHashTableEntry *pfmHashEntry)

Friends
THREAD_RETURN	fm_loopPacketProcessing (void *params)

Additional Inherited Members
Public Types inherited from CAMix
enum	tMixType { FIRST_MIX , MIDDLE_MIX , LAST_MIX , JAP }
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 38 of file CAFirstMixA.hpp.

Member Function Documentation

accountTrafficDownstream()

SINT32 CAFirstMixA::accountTrafficDownstream ( fmHashTableEntry *  pfmHashEntry )

private

Definition at line 969 of file CAFirstMixA.cpp.

{
  // count packet for payment
  SINT32 ret = CAAccountingInstance::handleJapPacket(pfmHashEntry, !(pfmHashEntry->bCountPacket), true);
  if (ret == CAAccountingInstance::HANDLE_PACKET_CONNECTION_OK )
  {
    // renew the timeout
    //pfmHashEntry->bRecoverTimeout = true;
    m_pChannelList->pushTimeoutEntry(pfmHashEntry);
  }
  else if (ret == CAAccountingInstance::HANDLE_PACKET_PREPARE_FOR_CLOSING_CONNECTION )
  {
    // when all control messages are sent the users connection will be closed
    //pfmHashEntry->bRecoverTimeout = false;
    m_pChannelList->setKickoutForced(pfmHashEntry, KICKOUT_FORCED);
    //m_pChannelList->pushTimeoutEntry(pfmHashEntry);
  }
  else if (ret == CAAccountingInstance::HANDLE_PACKET_CLOSE_CONNECTION )
  {
    // close users connection immediately
    CAMsg::printMsg(LOG_DEBUG, "CAFirstMixA: Closing JAP connection due to illegal payment status!\n", ret);
    closeConnection(pfmHashEntry);
    return ERR_INTERN_SOCKET_CLOSED;
  }
  //please remember that these values also may be returned even though they do not require
  //any further processing
  /*else if ( (ret == CAAccountingInstance::HANDLE_PACKET_CONNECTION_UNCHECKED) &&
        (ret == CAAccountingInstance::HANDLE_PACKET_HOLD_CONNECTION) )
  {
 
  }*/
  return E_SUCCESS;
}

References t_fmhashtableentry::bCountPacket, closeConnection(), E_SUCCESS, ERR_INTERN_SOCKET_CLOSED, CAAccountingInstance::HANDLE_PACKET_CLOSE_CONNECTION, CAAccountingInstance::HANDLE_PACKET_CONNECTION_OK, CAAccountingInstance::HANDLE_PACKET_PREPARE_FOR_CLOSING_CONNECTION, CAAccountingInstance::handleJapPacket(), KICKOUT_FORCED, CAFirstMix::m_pChannelList, CAMsg::printMsg(), CAFirstMixChannelList::pushTimeoutEntry(), and CAFirstMixChannelList::setKickoutForced().

Referenced by sendToUsers().

Here is the call graph for this function:

Here is the caller graph for this function:

accountTrafficUpstream()

SINT32 CAFirstMixA::accountTrafficUpstream ( fmHashTableEntry *  pHashEntry )

private

Definition at line 928 of file CAFirstMixA.cpp.

{
  SINT32 ret = E_SUCCESS;
 
  SINT32 handleResult = CAAccountingInstance::handleJapPacket(pHashEntry, false, false);
 
  if (CAAccountingInstance::HANDLE_PACKET_CONNECTION_OK == handleResult)
  {
    // renew the timeout
    //pHashEntry->bRecoverTimeout = true;
    m_pChannelList->pushTimeoutEntry(pHashEntry);
  }
  else if (CAAccountingInstance::HANDLE_PACKET_PREPARE_FOR_CLOSING_CONNECTION == handleResult)
  {
    // do not forward this packet
    pHashEntry->bRecoverTimeout = false;
    m_pChannelList->setKickoutForced(pHashEntry, KICKOUT_FORCED);
    CAMsg::printMsg(LOG_DEBUG, "CAFirstMixA: 1. setting bRecover timout to false for entry %x!\n", pHashEntry);
    //m_pChannelList->pushTimeoutEntry(pHashEntry);
    //don't let any upstream data messages pass for this user.
    ret = E_UNKNOWN;
  }
  else if (CAAccountingInstance::HANDLE_PACKET_CLOSE_CONNECTION == handleResult)
  {
    // kickout this user - he deserves it...
    CAMsg::printMsg(LOG_DEBUG, "CAFirstMixA: kickout upstream!\n");
    closeConnection(pHashEntry);
    ret = E_UNKNOWN;
  }
  //please remember that these values also may be returned even though they do not require
  //any further processing
  /*else if ( (ret == CAAccountingInstance::HANDLE_PACKET_CONNECTION_UNCHECKED) &&
        (ret == CAAccountingInstance::HANDLE_PACKET_HOLD_CONNECTION) )
  {
 
  }*/
  return ret;
}

References t_fmhashtableentry::bRecoverTimeout, closeConnection(), E_SUCCESS, E_UNKNOWN, CAAccountingInstance::HANDLE_PACKET_CLOSE_CONNECTION, CAAccountingInstance::HANDLE_PACKET_CONNECTION_OK, CAAccountingInstance::HANDLE_PACKET_PREPARE_FOR_CLOSING_CONNECTION, CAAccountingInstance::handleJapPacket(), KICKOUT_FORCED, CAFirstMix::m_pChannelList, CAMsg::printMsg(), CAFirstMixChannelList::pushTimeoutEntry(), and CAFirstMixChannelList::setKickoutForced().

Referenced by loop().

Here is the call graph for this function:

Here is the caller graph for this function:

checkUserConnections()

void CAFirstMixA::checkUserConnections ( )

private

Definition at line 1066 of file CAFirstMixA.cpp.

{
  // check the timeout for all connections
  fmHashTableEntry* timeoutHashEntry;
  fmHashTableEntry* firstIteratorEntry = NULL;
  bool currentEntryKickoutForced = false;
  /* this check also includes forced kickouts which have not bRecoverTimeout set. */
  while ( (timeoutHashEntry = m_pChannelList->popTimeoutEntry(true)) != NULL )
  {
    currentEntryKickoutForced = m_pChannelList->isKickoutForced(timeoutHashEntry);
    if(firstIteratorEntry == timeoutHashEntry)
    {
      m_pChannelList->pushTimeoutEntry(timeoutHashEntry, currentEntryKickoutForced);
      break;
    }
 
    if (!currentEntryKickoutForced)
    {
      //CAMsg::printMsg(LOG_ERR, "%p\n, ", timeoutHashEntry);
      if(m_pChannelList->isTimedOut(timeoutHashEntry) )
      {
        CAMsg::printMsg(LOG_DEBUG,"Client connection closed due to timeout.\n");
        closeConnection(timeoutHashEntry);
        continue;
      }
    }
    else
    {
      //A user to be kicked out: empty his downstream data queue.
      timeoutHashEntry->pQueueSend->clean();
 
      if( (timeoutHashEntry->pControlMessageQueue->getSize() == 0) ||
        (timeoutHashEntry->kickoutSendRetries <= 0) )
      {
        CAMsg::printMsg(LOG_WARNING, "Kickout immediately owner %x!\n", timeoutHashEntry);
        UINT32 authFlags = CAAccountingInstance::getAuthFlags(timeoutHashEntry);
        if (authFlags > 0)
        {
          CAMsg::printMsg(LOG_WARNING,"Client connection closed due to forced timeout! Payment auth flags: %u\n", authFlags);
        }
        else
        {
          CAMsg::printMsg(LOG_WARNING,"Client connection closed due to forced timeout!\n");
        }
        //CAAccountingInstance::setPrepaidBytesToZero(timeoutHashEntry->pAccountingInfo);
        closeConnection(timeoutHashEntry);
        continue;
      }
      else
      {
        //Note this counter initialized by calling CAFirstMixChannelList::add
        //and accessed by this thread, both do never run concurrently.
        //So we can avoid locking.
        timeoutHashEntry->kickoutSendRetries--;
        CAMsg::printMsg(LOG_INFO, "Size of control message queue for user to be kicked out: %u bytes, retries %d.\n",
            timeoutHashEntry->pControlMessageQueue->getSize(), timeoutHashEntry->kickoutSendRetries);
      }
      // Let the client obtain all his remaining control message packets
      //(which in most cases contain the error message with the kickout reason.
      CAMsg::printMsg(LOG_WARNING,"A kickout is supposed to happen. Let the user get his %u control message bytes before...\n",
          timeoutHashEntry->pControlMessageQueue->getSize());
    }
    if(firstIteratorEntry == NULL)
    {
      firstIteratorEntry = timeoutHashEntry;
    }
    m_pChannelList->pushTimeoutEntry(timeoutHashEntry, currentEntryKickoutForced);
  }
}

References CAQueue::clean(), closeConnection(), CAAccountingInstance::getAuthFlags(), CAQueue::getSize(), CAFirstMixChannelList::isKickoutForced(), CAFirstMixChannelList::isTimedOut(), t_fmhashtableentry::kickoutSendRetries, CAFirstMix::m_pChannelList, t_fmhashtableentry::pControlMessageQueue, CAFirstMixChannelList::popTimeoutEntry(), t_fmhashtableentry::pQueueSend, CAMsg::printMsg(), and CAFirstMixChannelList::pushTimeoutEntry().

Referenced by loop().

Here is the call graph for this function:

Here is the caller graph for this function:

closeConnection()

SINT32 CAFirstMixA::closeConnection ( fmHashTableEntry *  pHashEntry )

protected

Definition at line 85 of file CAFirstMixA.cpp.

{
  if (pHashEntry == NULL)
  {
    return E_UNKNOWN;
  }
 
  INIT_STACK;
  BEGIN_STACK("CAFirstMixA::closeConnection");
 
 
  fmChannelListEntry* pEntry;
  tQueueEntry* pQueueEntry = new tQueueEntry;
  MIXPACKET* pMixPacket=&pQueueEntry->packet;
 
  #ifdef LOG_TRAFFIC_PER_USER
    UINT64 current_time;
    getcurrentTimeMillis(current_time);
    CAMsg::printMsg(LOG_DEBUG,"Removing Connection wiht ID: %Lu -- login time [ms] %Lu -- logout time [ms] %Lu -- Traffic was: IN: %u  --  OUT: %u\n",pHashEntry->id,pHashEntry->timeCreated,current_time,pHashEntry->trafficIn,pHashEntry->trafficOut);
  #endif
  m_pIPList->removeIP(pHashEntry->peerIP);
 
  m_psocketgroupUsersRead->remove(*(pHashEntry->pMuxSocket));
  m_psocketgroupUsersWrite->remove(*(pHashEntry->pMuxSocket));
  pEntry = m_pChannelList->getFirstChannelForSocket(pHashEntry->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(pQueueEntry, sizeof(tQueueEntry));
    delete pEntry->pCipher;
    pEntry->pCipher = NULL;
    pEntry=m_pChannelList->getNextChannel(pEntry);
#ifdef CH_LOG_STUDY
    nrOfChOpMutex->lock();
    currentOpenedChannels--;
    nrOfChOpMutex->unlock();
#endif //CH_LOG_STUDY
  }
  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
 
  CAMuxSocket* pMuxSocket = pHashEntry->pMuxSocket;
  // Save the socket - its pointer will be deleted in this method!!! Crazy mad programming...
  m_pChannelList->remove(pHashEntry->pMuxSocket);
  delete pMuxSocket;
  pMuxSocket = NULL;
  //pHashEntry->pMuxSocket = NULL; // not needed now, but maybe in the future...
 
  delete pQueueEntry;
  pQueueEntry = NULL;
 
  FINISH_STACK("CAFirstMixA::closeConnection");
 
  return E_SUCCESS;
}

References CAQueue::add(), ASSERT, BEGIN_STACK, t_MixPacket::channel, CHANNEL_CLOSE, t_firstmixchannellist::channelOut, t_MixPacket::data, DATA_SIZE, CAFirstMix::decUsers(), E_SUCCESS, E_UNKNOWN, FINISH_STACK, t_MixPacket::flags, getcurrentTimeMillis(), CAFirstMixChannelList::getFirstChannelForSocket(), CAFirstMixChannelList::getNextChannel(), getRandom(), t_fmhashtableentry::id, INIT_STACK, CAFirstMix::m_pChannelList, CAFirstMix::m_pIPList, CAFirstMix::m_pQueueSendToMix, CAFirstMix::m_psocketgroupUsersRead, CAFirstMix::m_psocketgroupUsersWrite, t_queue_entry::packet, t_firstmixchannellist::pCipher, t_fmhashtableentry::peerIP, t_fmhashtableentry::pMuxSocket, t_fmhashtableentry::pQueueSend, CAMsg::printMsg(), t_fmhashtableentry::pSymCipher, CAFirstMixChannelList::remove(), CASocketGroupEpoll::remove(), CAIPList::removeIP(), and setZero64().

Referenced by accountTrafficDownstream(), accountTrafficUpstream(), checkUserConnections(), loop(), sendToUsers(), and shutDown().

Here is the call graph for this function:

Here is the caller graph for this function:

finishPacket()

void CAFirstMixA::finishPacket ( fmHashTableEntry *  pfmHashEntry )

private

Definition at line 1036 of file CAFirstMixA.cpp.

{
  tQueueEntry *packetToSend = &(pfmHashEntry->oQueueEntry);
  fmChannelList* cListEntry=m_pChannelList->get(packetToSend->packet.channel);
  if(cListEntry != NULL)
  {
    packetToSend->packet.channel = cListEntry->channelIn;
    cListEntry->downStreamBytes -= sizeof(tQueueEntry);
#ifdef DEBUG
    CAMsg::printMsg(LOG_DEBUG, "CAFirstMixA: channels of current packet, in: %u, out: %u, count: %u, flags: 0x%x\n",
        cListEntry->channelIn, cListEntry->channelOut, cListEntry->downStreamBytes,
        packetToSend->packet.flags);
#endif
    if(packetToSend->packet.flags == CHANNEL_CLOSE)
    {
      delete cListEntry->pCipher;
      cListEntry->pCipher = NULL;
      m_pChannelList->removeChannel(pfmHashEntry->pMuxSocket, cListEntry->channelIn);
#ifdef CH_LOG_STUDY
      nrOfChOpMutex->lock();
      currentOpenedChannels--;
      nrOfChOpMutex->unlock();
#endif //CH_LOG_STUDY
    }
  }
}

References t_MixPacket::channel, CHANNEL_CLOSE, t_firstmixchannellist::channelIn, t_firstmixchannellist::channelOut, t_MixPacket::flags, CAFirstMixChannelList::get(), CAFirstMix::m_pChannelList, t_fmhashtableentry::oQueueEntry, t_queue_entry::packet, t_firstmixchannellist::pCipher, t_fmhashtableentry::pMuxSocket, CAMsg::printMsg(), and CAFirstMixChannelList::removeChannel().

Referenced by sendToUsers().

Here is the call graph for this function:

Here is the caller graph for this function:

loop()

SINT32 CAFirstMixA::loop ( )

protectedvirtual

Todo:

check if thread is closed

Implements CAFirstMix.

Definition at line 158 of file CAFirstMixA.cpp.

  {
#ifndef NEW_MIX_TYPE
#ifdef DELAY_USERS
    m_pChannelList->setDelayParameters( CALibProxytest::getOptions()->getDelayChannelUnlimitTraffic(),
                                      CALibProxytest::getOptions()->getDelayChannelBucketGrow(),
                                      CALibProxytest::getOptions()->getDelayChannelBucketGrowIntervall());
#endif
 
  //  CASingleSocketGroup osocketgroupMixOut;
    SINT32 countRead=0;
    //#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,Channel open timestamp (microseconds),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,Channel open timestamp (microseconds),PacketsIn (only data and open), PacketsOut (only data),ChannelDuration (open packet received)--> close packet put into send queue to next user\n");
#endif
#ifdef _DEBUG
    CAThread* pLogThread=new CAThread((UINT8*)"CAFirstMixA - LogLoop");
    pLogThread->setMainLoop(fm_loopLog);
    pLogThread->start(this);
#endif
 
#ifdef LOG_CRIME
    CAIPAddrWithNetmask* surveillanceIPs = CALibProxytest::getOptions()->getCrimeSurveillanceIPs();
    UINT32 nrOfSurveillanceIPs = CALibProxytest::getOptions()->getNrOfCrimeSurveillanceIPs();
#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:
#ifdef PAYMENT
        // while checking if there are connections to close: synch with login threads
        m_pmutexLogin->lock();
        checkUserConnections();
        m_pmutexLogin->unlock();
#endif
//First Step
//Checking for new connections
// Now in a separate 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!!
 
        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 epoll we do not need to search the whole list
            //of connected JAPs to find the ones who have sent data
            //as epoll 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
                  {
#endif
/*#ifdef DELAY_USERS
 * Don't delay upstream
                if( m_pChannelList->hasDelayBuckets(pHashEntry->delayBucketID) )
                {
#endif*/
                    countRead--;
                    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
                    #ifdef DATA_RETENTION_LOG
                      pQueueEntry->dataRetentionLogEntry.t_in=htonl(time(NULL));
                    #endif
                    if(ret<0&&ret!=E_AGAIN/*||pHashEntry->accessUntil<time()*/)
                    {
                      // remove dead connections
                      closeConnection(pHashEntry);
                    }
                    else if(ret==MIXPACKET_SIZE)                      // we've read enough data for a whole mix packet. nice!
                      {
#ifdef PAYMENT
                        if (pHashEntry->bRecoverTimeout)
                        {
                          // renew the timeout only if recovery is allowed
                          m_pChannelList->pushTimeoutEntry(pHashEntry);
                        }
#endif
                        #ifdef LOG_TRAFFIC_PER_USER
                          pHashEntry->trafficIn++;
                        #endif
                        #ifdef COUNTRY_STATS
                          m_PacketsPerCountryIN[pHashEntry->countryID].inc();
                        #endif
                        //New control channel code...!
                        if(pMixPacket->channel>0&&pMixPacket->channel<256)
                        {
                          if (pHashEntry->pControlChannelDispatcher->proccessMixPacket(pMixPacket))
                          {
                            goto NEXT_USER;
                          }
                          else
                          {
                            CAMsg::printMsg(LOG_DEBUG, "Control channel packet is invalid and could not be processed!\n");
                            closeConnection(pHashEntry);
                            goto NEXT_USER;
                          }
                        }
#ifdef ANON_DEBUG_MODE
                        bool bIsDebugPacket=false;
                        if (pMixPacket->flags&CHANNEL_DEBUG)
                          {
                          bIsDebugPacket=true;
                          UINT8 base64Payload[DATA_SIZE << 1];
                          EVP_EncodeBlock(base64Payload, pMixPacket->data, DATA_SIZE);//base64 encoding (without newline!)
                          pMixPacket->flags &= ~CHANNEL_DEBUG;
                          CAMsg::printMsg(LOG_DEBUG, "AN.ON packet debug: %s\n", base64Payload);
                          }
 
#endif
#ifdef PAYMENT
                        if(accountTrafficUpstream(pHashEntry) != E_SUCCESS) goto NEXT_USER;
#endif
                        if(pMixPacket->flags==CHANNEL_DUMMY) // just a dummy to keep the connection alife in e.g. NAT gateways
                        {
                          CAMsg::printMsg(LOG_DEBUG,"received dummy traffic\n");
                          getRandom(pMixPacket->data,DATA_SIZE);
                          #ifdef LOG_PACKET_TIMES
                            setZero64(pQueueEntry->timestamp_proccessing_start);
                          #endif
                          #ifdef LOG_TRAFFIC_PER_USER
                            pHashEntry->trafficOut++;
                          #endif
                          #ifdef COUNTRY_STATS
                            m_PacketsPerCountryOUT[pHashEntry->countryID].inc();
                          #endif
                          pHashEntry->pQueueSend->add(pQueueEntry,sizeof(tQueueEntry));
                          #ifdef HAVE_EPOLL
                            //m_psocketgroupUsersWrite->add(*pMuxSocket,pHashEntry);
                          #else
                            m_psocketgroupUsersWrite->add(*pMuxSocket);
                          #endif
                        }
                        else if(pMixPacket->flags==CHANNEL_CLOSE)     // closing one mix-channel (not the JAP<->mix connection!)
                        {
                          fmChannelListEntry* pEntry;
                          pEntry=m_pChannelList->get(pMuxSocket,pMixPacket->channel);
                          if(pEntry!=NULL)
                          {
                            pMixPacket->channel=pEntry->channelOut;
                            getRandom(pMixPacket->data,DATA_SIZE);
                            #ifdef LOG_PACKET_TIMES
                              getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end_OP);
                            #endif
                            m_pQueueSendToMix->add(pQueueEntry, sizeof(tQueueEntry));
                            /* Don't delay upstream
                            #ifdef DELAY_USERS
                            m_pChannelList->decDelayBuckets(pHashEntry->delayBucketID);
                            #endif*/
                            #ifdef LOG_CHANNEL
                              //pEntry->packetsInFromUser++;
                              getcurrentTimeMicros(current_time);
                              diff_time=diff64(current_time,pEntry->timeCreated);
                              CAMsg::printMsg(LOG_DEBUG,"1:%u,%Lu,%Lu,%u,%u,%u\n",
                                                        pEntry->channelIn,pEntry->pHead->id,pEntry->timeCreated,pEntry->packetsInFromUser,pEntry->packetsOutToUser,
                                                        diff_time);
                            #endif
                            delete pEntry->pCipher;              // forget the symetric key of this connection
                            pEntry->pCipher = NULL;
                            m_pChannelList->removeChannel(pMuxSocket,pEntry->channelIn);
 
                            #ifdef CH_LOG_STUDY
                            nrOfChOpMutex->lock();
                            currentOpenedChannels--;
                            nrOfChOpMutex->unlock();
                            #endif //CH_LOG_STUDY
                          }
                          #ifdef _DEBUG
                          else
                          {
//                            CAMsg::printMsg(LOG_DEBUG,"Invalid ID to close from Browser!\n");
                          }
                          #endif
                        }
                        else                                         // finally! a normal mix packet
                        {
                          CASymChannelCipher* 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
 
                            //check if this IP must be logged due to crime detection
#ifdef LOG_CRIME
                            crimeSurveillance(surveillanceIPs, nrOfSurveillanceIPs, pEntry->pHead->peerIP,pEntry->pHead->peerPort, pMixPacket);
#endif
#ifdef ANON_DEBUG_MODE
                            if(bIsDebugPacket)
                              {
                              pMixPacket->flags|=CHANNEL_DEBUG;
                              }
                            #endif
                            m_pQueueSendToMix->add(pQueueEntry, sizeof(tQueueEntry));
                            /* Don't delay upstream
                            #ifdef DELAY_USERS
                            m_pChannelList->decDelayBuckets(pHashEntry->delayBucketID);
                            #endif*/
                            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
 
                            //log symcrypto
                            //UINT8* tmpstr=bytes2hex(pMixPacket->data, DATA_SIZE);
                            //CAMsg::printMsg(LOG_DEBUG, "Plain Packet recevied form user: %s\n", tmpstr);
                            //delete tmpstr;
                            //tmpstr = NULL;
                            //end log symcrpyto
    
 
                            pHashEntry->pSymCipher->crypt1(pMixPacket->data,rsaBuff,FIRST_MIX_SIZE_OF_SYMMETRIC_KEYS);
                            #ifdef REPLAY_DETECTION
                            // replace time(NULL) with the real timestamp ()
                            // packet-timestamp*REPLAY_BASE + m_u64ReferenceTime
                              if(m_pReplayDB->insert(rsaBuff,time(NULL))!=E_SUCCESS)
                              {
                                CAMsg::printMsg(LOG_INFO,"Replay: Duplicate packet ignored.\n");
                                continue;
                              }
                            #endif
 
 
    
                            pCipher= CASymChannelCipherFactory::createCipher(CALibProxytest::getOptions()->getSymChannelCipherAlgorithm());
                            pCipher->setKeys(rsaBuff,FIRST_MIX_SIZE_OF_SYMMETRIC_KEYS);
                            for(int i=0;i<16;i++)
                              rsaBuff[i]=0xFF;
                            pCipher->setIV2(rsaBuff);
                            pCipher->crypt1(pMixPacket->data+FIRST_MIX_SIZE_OF_SYMMETRIC_KEYS,pMixPacket->data,DATA_SIZE-FIRST_MIX_SIZE_OF_SYMMETRIC_KEYS);
                            getRandom(pMixPacket->data+DATA_SIZE-FIRST_MIX_SIZE_OF_SYMMETRIC_KEYS,FIRST_MIX_SIZE_OF_SYMMETRIC_KEYS);
                            #if defined (LOG_CHANNEL) ||defined(DATA_RETENTION_LOG)
                              HCHANNEL tmpC=pMixPacket->channel;
                            #endif
#if defined LOG_CRIME || defined _DEBUG
                            HCHANNEL inChannel = pMixPacket->channel;
#endif
                            if(m_pChannelList->addChannel(pMuxSocket,pMixPacket->channel,pCipher,&pMixPacket->channel)!=E_SUCCESS)
                            { //todo move up ?
                              delete pCipher;
                              pCipher = NULL;
                            }
                            else
                            {
                              #ifdef CH_LOG_STUDY
                              nrOfChOpMutex->lock();
                              if(pHashEntry->channelOpenedLastIntervalTS !=
                                lastLogTime)
                              {
                                pHashEntry->channelOpenedLastIntervalTS =
                                  lastLogTime;
                                nrOfOpenedChannels++;
                              }
                              currentOpenedChannels++;
                              nrOfChOpMutex->unlock();
                              #endif //CH_LOG_STUDY
 
                              #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
                              #ifdef DATA_RETENTION_LOG
                                pQueueEntry->dataRetentionLogEntry.entity.first.channelid=htonl(pMixPacket->channel);
                                fmChannelListEntry* pTmpEntry1=m_pChannelList->get(pMuxSocket,tmpC);
                                memcpy(pQueueEntry->dataRetentionLogEntry.entity.first.ip_in,pTmpEntry1->pHead->peerIP,4);
                                pQueueEntry->dataRetentionLogEntry.entity.first.port_in=(UINT16)pTmpEntry1->pHead->peerPort;
                                pQueueEntry->dataRetentionLogEntry.entity.first.port_in=htons(pQueueEntry->dataRetentionLogEntry.entity.first.port_in);
                              #endif
 
                              //check if this IP must be logged due to crime detection
                              #ifdef LOG_CRIME
 
                                pEntry=m_pChannelList->get(pMuxSocket, inChannel);
                                if(pEntry != NULL)
                                {
                                  crimeSurveillance(surveillanceIPs, nrOfSurveillanceIPs, pEntry->pHead->peerIP,pEntry->pHead->peerPort, pMixPacket);
                                }
                              #endif
#ifdef ANON_DEBUG_MODE
                                if (bIsDebugPacket)
                                  {
                                    pMixPacket->flags |= CHANNEL_DEBUG;
                                    pEntry = m_pChannelList->get(pMuxSocket, inChannel);
                                    pEntry->bDebug = true;
                                  }
#endif
                            //log symcrypto
                            //tmpstr=bytes2hex(pMixPacket->data, DATA_SIZE);
                            //CAMsg::printMsg(LOG_DEBUG, "Plain Packet sent to next mix: %s\n", tmpstr);
                            //delete tmpstr;
                            //end log symcrpyto
 
                                m_pQueueSendToMix->add(pQueueEntry, sizeof(tQueueEntry));
                              /* Don't delay upstream
                              #ifdef DELAY_USERS
                              m_pChannelList->decDelayBuckets(pHashEntry->delayBucketID);
                              #endif*/
                              incMixedPackets();
                              #ifdef _DEBUG
                                      CAMsg::printMsg(LOG_DEBUG,"Added out channel: %u for in channel: %u\n",pMixPacket->channel,inChannel);
                              #endif
                            }
                          }
                        }
                      }
/*#ifdef DELAY_USERS
                  }
#endif*/
                #ifdef HAVE_EPOLL
NEXT_USER:
                  pHashEntry=(fmHashTableEntry*)m_psocketgroupUsersRead->getNextSignaledSocketData();
                #else
                  }//if is signaled
NEXT_USER:
                  pHashEntry=m_pChannelList->getNext();
                #endif
              }
              if(countRead>0)
              {
                CAMsg::printMsg(LOG_DEBUG,"CAFirstMixA::loop() - read from user --> countRead >0 after processing all connections!\n");
              }
          }
//Third step
//Sending to next mix
 
// Now in a separate Thread (see loopSendToMix())
 
//Step 4
//Step 4a Receiving from mix to queue now in a separate thread
//Step 4b Processing 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
#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, "Dequeued Downstream AN.ON packet from previous Mix debug: %s\n", base64Payload);
                pMixPacket->flags &= ~CHANNEL_DEBUG;
              }
 
#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)
                  {
                    /* a hack to solve the SSL problem:
                     * set channel of downstream packet to in channel after they are dequeued
                     * from pEntry->pQueueSend so we can retrieve the channel entry to decrement
                     * the per channel count of enqueued downstream bytes.
                     */
                    #ifndef SSL_HACK
                    pMixPacket->channel=pEntry->channelIn;
                    #endif
 
                    pEntry->pCipher->crypt2(pMixPacket->data,pMixPacket->data,DATA_SIZE);
                    //getRandom(pMixPacket->data,DATA_SIZE);
                    #ifdef LOG_PACKET_TIMES
                      getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end_OP);
                    #endif
                    pEntry->pHead->pQueueSend->add(pQueueEntry, sizeof(tQueueEntry));
                    #ifdef LOG_TRAFFIC_PER_USER
                      pEntry->pHead->trafficOut++;
                    #endif
                    #ifdef COUNTRY_STATS
                      m_PacketsPerCountryOUT[pEntry->pHead->countryID].inc();
                    #endif
                    #ifdef SSL_HACK
                      /* a hack to solve the SSL problem:
                       * per channel count of enqueued downstream bytes
                       */
                      pEntry->downStreamBytes += sizeof(tQueueEntry);
                    #endif
                    #ifdef LOG_CHANNEL
                      pEntry->packetsOutToUser++;
                      getcurrentTimeMicros(current_time);
                      diff_time=diff64(current_time,pEntry->timeCreated);
                      CAMsg::printMsg(LOG_DEBUG,"2:%u,%Lu,%Lu,%u,%u,%u\n",
                                                pEntry->channelIn,pEntry->pHead->id,pEntry->timeCreated,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
 
 
                    #ifndef SSL_HACK
                      delete pEntry->pCipher;              // forget the symetric key of this connection
                      pEntry->pCipher = NULL;
                      m_pChannelList->removeChannel(pEntry->pHead->pMuxSocket, pEntry->channelIn);
                    #ifdef CH_LOG_STUDY
                      nrOfChOpMutex->lock();
                      currentOpenedChannels--;
                      nrOfChOpMutex->unlock();
                    #endif
                    /* a hack to solve the SSL problem:
                     * remove channel after the close packet is enqueued
                     * from pEntry->pQueueSend
                     */
                    #endif
                  }
                  else
                  {
                    #ifdef DEBUG
                      CAMsg::printMsg(LOG_DEBUG, "CAFirstMixA: close channel -> client but channel does not exist.\n");
                    #endif
                  }
 
              }
            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)
                  {
                    #ifdef LOG_CRIME
                      if((pMixPacket->flags&CHANNEL_SIG_CRIME)==CHANNEL_SIG_CRIME)
                        {
                          //UINT32 id=(pMixPacket->flags>>8)&0x000000FF;
                          int log=LOG_ENCRYPTED;
                          if(!CALibProxytest::getOptions()->isEncryptedLogEnabled())
                            log=LOG_CRIT;
                          CAMsg::printMsg(log,"Detecting crime activity - next mix channel: %u -- "
                              "Incoming (User) IP:Port is: %u.%u.%u.%u:%u \n", pMixPacket->channel,
                              pEntry->pHead->peerIP[0],
                              pEntry->pHead->peerIP[1],
                              pEntry->pHead->peerIP[2],
                              pEntry->pHead->peerIP[3],
                              pEntry->pHead->peerPort);
                          continue;
                        }
                    #endif
 
                    /* a hack to solve the SSL problem:
                     * same as CHANNEL_CLOSE packets
                     */
                    #ifndef SSL_HACK
                    pMixPacket->channel=pEntry->channelIn;
                    #endif
 
                    pEntry->pCipher->crypt2(pMixPacket->data,pMixPacket->data,DATA_SIZE);
 
                    #ifdef LOG_PACKET_TIMES
                      getcurrentTimeMicros(pQueueEntry->timestamp_proccessing_end_OP);
                    #endif
#ifdef ANON_DEBUG_MODE
                      if (pEntry->bDebug)
                        { 
                        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 Dowwstream AN.ON packet debug: %s\n", base64Payload);
                        }
 
#endif
                      pEntry->pHead->pQueueSend->add(pQueueEntry, sizeof(tQueueEntry));
                    /*CAMsg::printMsg(
                        LOG_INFO,"adding data packet to queue: %x, queue size: %u bytes\n",
                        pEntry->pHead->pQueueSend, pEntry->pHead->pQueueSend->getSize());*/
                    #ifdef LOG_TRAFFIC_PER_USER
                      pEntry->pHead->trafficOut++;
                    #endif
                    #ifdef COUNTRY_STATS
                      m_PacketsPerCountryOUT[pEntry->pHead->countryID].inc();
                    #endif
                    #ifdef LOG_CHANNEL
                      pEntry->packetsOutToUser++;
                    #endif
                    #ifdef SSL_HACK
                      /* a hack to solve the SSL problem:
                       * per channel count of downstream packets in bytes
                       */
                      pEntry->downStreamBytes += sizeof(tQueueEntry);
                    #endif
 
                    #ifdef HAVE_EPOLL
                      /*int epret = m_psocketgroupUsersWrite->add(*pEntry->pHead->pMuxSocket,pEntry->pHead);
                      if(epret == E_UNKNOWN)
                      {
                        epret=errno;
                        CAMsg::printMsg(LOG_INFO,"epoll_add returns: %s (return value: %d) \n", strerror(epret), epret);
                      }*/
                      #else
                      m_psocketgroupUsersWrite->add(*pEntry->pHead->pMuxSocket);
                    #endif
 
                    incMixedPackets();
                  }
                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
                  }
              }
          }
 
        bAktiv = sendToUsers();
#ifndef FAST_PROCESSING
        if(!bAktiv)
          msSleep(1);
#endif
      }
//ERR:
    CAMsg::printMsg(LOG_CRIT,"Seems that we are restarting now!!\n");
    m_bRunLog=false;
    clean();
    delete pQueueEntry;
    pQueueEntry = NULL;
    delete []tmpBuff;
    tmpBuff = NULL;
#ifdef _DEBUG
    pLogThread->join();
    delete pLogThread;
    pLogThread = NULL;
#endif
    CAMsg::printMsg(LOG_CRIT,"Main Loop exited!!\n");
#endif//NEW_MIX_TYPE
    return E_UNKNOWN;
  }

References accountTrafficUpstream(), CASocketGroupEpoll::add(), CAQueue::add(), CAFirstMixChannelList::addChannel(), t_fmhashtableentry::bRecoverTimeout, t_MixPacket::channel, CHANNEL_CLOSE, CHANNEL_DATA, CHANNEL_DEBUG, CHANNEL_DUMMY, CHANNEL_OPEN, CHANNEL_SIG_CRIME, t_firstmixchannellist::channelIn, t_firstmixchannellist::channelOut, checkUserConnections(), CAFirstMix::clean(), closeConnection(), t_fmhashtableentry::countryID, CASymChannelCipherFactory::createCipher(), CASymChannelCipher::crypt1(), CASymChannelCipher::crypt2(), t_MixPacket::data, DATA_SIZE, diff64(), E_SUCCESS, E_UNKNOWN, FIRST_MIX_SIZE_OF_SYMMETRIC_KEYS, t_MixPacket::flags, CAFirstMix::fm_loopLog, CAFirstMixChannelList::get(), CAQueue::get(), getcurrentTimeMicros(), CAFirstMixChannelList::getFirst(), CASocketGroupEpoll::getFirstSignaledSocketData(), CAFirstMixChannelList::getNext(), CASocketGroupEpoll::getNextSignaledSocketData(), CALibProxytest::getOptions(), getRandom(), CAQueue::getSize(), t_fmhashtableentry::id, tUINT32withLock::inc(), CAFirstMix::incMixedPackets(), CADatabase::insert(), CASocketGroupEpoll::isSignaled(), CAThread::join(), CALockAble::lock(), CAMutex::lock(), LOG_ENCRYPTED, CAFirstMix::m_bRestart, CAFirstMix::m_bRunLog, CAFirstMix::m_nUser, CAFirstMix::m_PacketsPerCountryIN, CAFirstMix::m_PacketsPerCountryOUT, CAFirstMix::m_pChannelList, CAFirstMix::m_pmutexLogin, CAFirstMix::m_pQueueReadFromMix, CAFirstMix::m_pQueueSendToMix, CAMixWithReplayDB::m_pReplayDB, CAFirstMix::m_psocketgroupUsersRead, CAFirstMix::m_psocketgroupUsersWrite, MAX_NEXT_MIX_QUEUE_SIZE, MIXPACKET_SIZE, msSleep(), t_queue_entry::packet, t_firstmixchannellist::pCipher, t_fmhashtableentry::pControlChannelDispatcher, t_fmhashtableentry::peerIP, t_firstmixchannellist::pHead, t_fmhashtableentry::pMuxSocket, t_fmhashtableentry::pQueueSend, CAMsg::printMsg(), CAControlChannelDispatcher::proccessMixPacket(), t_fmhashtableentry::pSymCipher, CAFirstMixChannelList::pushTimeoutEntry(), CAMuxSocket::receive(), CAFirstMixChannelList::removeChannel(), RSA_SIZE, CASocketGroupEpoll::select(), sendToUsers(), set64(), CAFirstMixChannelList::setDelayParameters(), CASymChannelCipher::setIV2(), CASymChannelCipher::setKeys(), CAThread::setMainLoop(), setZero64(), CAThread::start(), and CAMutex::unlock().

Here is the call graph for this function:

notifyAllUserChannels()

void CAFirstMixA::notifyAllUserChannels ( fmHashTableEntry *  pfmHashEntry, UINT16  flags  )

private

Definition at line 1004 of file CAFirstMixA.cpp.

{
  if(pfmHashEntry == NULL) 
    return;
  fmChannelListEntry* pEntry = m_pChannelList->getFirstChannelForSocket(pfmHashEntry->pMuxSocket);
  tQueueEntry* pQueueEntry=new tQueueEntry;
  MIXPACKET *notifyPacket = &(pQueueEntry->packet);
  memset(notifyPacket, 0, MIXPACKET_SIZE);
 
  notifyPacket->flags = flags;
  while(pEntry != NULL)
  {
    if(pEntry->bIsSuspended)
    {
      notifyPacket->channel = pEntry->channelOut;
      getRandom(notifyPacket->data,DATA_SIZE);
#ifdef _DEBUG
      CAMsg::printMsg(LOG_INFO,"Sent flags %u for channel: %u\n", flags, notifyPacket->channel);
#endif
      m_pQueueSendToMix->add(pQueueEntry, sizeof(tQueueEntry));
      pEntry->bIsSuspended = false;
    }
    pEntry=m_pChannelList->getNextChannel(pEntry);
  }
  pfmHashEntry->cSuspend=0;
  delete pQueueEntry;
}

References CAQueue::add(), t_firstmixchannellist::bIsSuspended, t_MixPacket::channel, t_firstmixchannellist::channelOut, t_fmhashtableentry::cSuspend, t_MixPacket::data, DATA_SIZE, flags, t_MixPacket::flags, CAFirstMixChannelList::getFirstChannelForSocket(), CAFirstMixChannelList::getNextChannel(), getRandom(), CAFirstMix::m_pChannelList, CAFirstMix::m_pQueueSendToMix, MIXPACKET_SIZE, t_queue_entry::packet, t_fmhashtableentry::pMuxSocket, and CAMsg::printMsg().

Here is the call graph for this function:

sendToUsers()

bool CAFirstMixA::sendToUsers ( )

private

Definition at line 773 of file CAFirstMixA.cpp.

{
#ifndef HAVE_EPOLL
  SINT32 countRead = 
#endif
  m_psocketgroupUsersWrite->select(/*true,*/0);
  tQueueEntry *packetToSend = NULL;
  SINT32 packetSize = sizeof(tQueueEntry);
  CAQueue *controlMessageUserQueue = NULL;
  CAQueue *dataMessageUserQueue = NULL;
 
  CAQueue *processedQueue = NULL; /* one the above queues that should be used for processing*/
  UINT32 extractSize = 0;
  bool bAktiv = false;
  UINT32 iSocketErrors = 0;
 
/* Cyclic polling: gets all open sockets that will not block when invoking send()
 * but will only send at most one packet. After that control is returned to loop()
 */
#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
      /* loop turn init */
      extractSize = 0;
      processedQueue = NULL;
      packetToSend = &(pfmHashEntry->oQueueEntry);
      controlMessageUserQueue = pfmHashEntry->pControlMessageQueue;
      dataMessageUserQueue = pfmHashEntry->pQueueSend;
 
      //Control messages have a higher priority.
      if(controlMessageUserQueue->getSize() > 0)
      {
        processedQueue = controlMessageUserQueue;
        pfmHashEntry->bCountPacket = false;
      }
      else if( (dataMessageUserQueue->getSize() > 0)
#ifdef DELAY_USERS
          && m_pChannelList->hasDelayBuckets(pfmHashEntry->delayBucketID)
#endif
      )
      {
        processedQueue = dataMessageUserQueue;
        pfmHashEntry->bCountPacket = true;
      }
 
      if(processedQueue != NULL)
      {
        extractSize = packetSize;
        bAktiv=true;
 
        if(pfmHashEntry->uAlreadySendPacketSize == -1)
        {
          processedQueue->get((UINT8*) packetToSend, &extractSize);
 
          /* Hack for SSL BUG */
#ifdef SSL_HACK
          finishPacket(pfmHashEntry);
#endif //SSL_HACK
#ifdef ANON_DEBUG_MODE
          if (packetToSend->packet.flags&CHANNEL_DEBUG)
            {
            UINT8 base64Payload[DATA_SIZE << 1];
            EVP_EncodeBlock(base64Payload, packetToSend->packet.data, DATA_SIZE);//base64 encoding (without newline!)
            CAMsg::printMsg(LOG_DEBUG, "Send Downstream AN.ON packet to user debug: %s\n", base64Payload);
            packetToSend->packet.flags &= ~CHANNEL_DEBUG;
            }
 
#endif
          pfmHashEntry->pMuxSocket->prepareForSend(&(packetToSend->packet));
          pfmHashEntry->uAlreadySendPacketSize = 0;
        }
      }
 
      if( (extractSize > 0) || (pfmHashEntry->uAlreadySendPacketSize > 0) )
      {
        SINT32 len =  MIXPACKET_SIZE - pfmHashEntry->uAlreadySendPacketSize;
        UINT8* packetToSendOffset = ((UINT8*)&(packetToSend->packet)) + pfmHashEntry->uAlreadySendPacketSize;
        CASocket* clientSocket = pfmHashEntry->pMuxSocket->getCASocket();
 
        SINT32 ret = clientSocket->send(packetToSendOffset, len);
 
        if(ret > 0)
        {
#ifdef PAYMENT
          SINT32 accounting = E_SUCCESS;
#endif
          pfmHashEntry->uAlreadySendPacketSize += ret;
 
          if(pfmHashEntry->uAlreadySendPacketSize == MIXPACKET_SIZE)
          {
            #ifdef DELAY_USERS
            if(processedQueue != controlMessageUserQueue)
            {
              m_pChannelList->decDelayBuckets(pfmHashEntry->delayBucketID);
            }
            #endif
 
            #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
            pfmHashEntry->uAlreadySendPacketSize=-1;
#ifdef PAYMENT
            /* count this packet for accounting */
            accounting = accountTrafficDownstream(pfmHashEntry);
#endif
          }
 
        }
        else if(ret<0&&ret!=E_AGAIN)
        {
          iSocketErrors++;
          // if (iSocketErrors == 1) // show debug message only at the first error; otherwise, the log may get huge
          {
            SOCKET sock=clientSocket->getSocket();
            CAMsg::printMsg(LOG_DEBUG,"CAFirstMixA::sendtoUser() - send error %d on socket %d. Reason: %s (%i)\n", ret, sock, GET_NET_ERROR_STR(GET_NET_ERROR), GET_NET_ERROR);
          }
          // kick the user out - these only happens in extreme situations...
          closeConnection(pfmHashEntry);
        }
        //TODO error handling
      }
 
#ifdef HAVE_EPOLL
    pfmHashEntry=(fmHashTableEntry*)m_psocketgroupUsersWrite->getNextSignaledSocketData();
  }
#else
    }//if is socket signaled
    pfmHashEntry=m_pChannelList->getNext();
  }
#endif
  if (iSocketErrors > 1)
  {
    CAMsg::printMsg(LOG_ERR, "CAFirstMixA::sendtoUser() - %d send errors on a socket occured!\n", iSocketErrors);
  }
  
  return bAktiv;
}

References accountTrafficDownstream(), t_fmhashtableentry::bCountPacket, CHANNEL_DATA, CHANNEL_DEBUG, closeConnection(), t_MixPacket::data, DATA_SIZE, CAFirstMixChannelList::decDelayBuckets(), DELAY_USERS, t_fmhashtableentry::delayBucketID, E_AGAIN, E_SUCCESS, finishPacket(), t_MixPacket::flags, CAQueue::get(), GET_NET_ERROR, GET_NET_ERROR_STR, CAMuxSocket::getCASocket(), getcurrentTimeMicros(), CAFirstMixChannelList::getFirst(), CASocketGroupEpoll::getFirstSignaledSocketData(), CAFirstMixChannelList::getNext(), CASocketGroupEpoll::getNextSignaledSocketData(), CAQueue::getSize(), CAFirstMixChannelList::hasDelayBuckets(), CASocketGroupEpoll::isSignaled(), isZero64(), len, CAFirstMix::m_pChannelList, CAFirstMix::m_psocketgroupUsersWrite, MIXPACKET_SIZE, t_fmhashtableentry::oQueueEntry, t_queue_entry::packet, t_fmhashtableentry::pControlMessageQueue, t_fmhashtableentry::pMuxSocket, t_fmhashtableentry::pQueueSend, CAMuxSocket::prepareForSend(), CAMsg::printMsg(), CASocketGroupEpoll::select(), CASocket::send(), SOCKET, and t_fmhashtableentry::uAlreadySendPacketSize.

Referenced by loop().

Here is the call graph for this function:

Here is the caller graph for this function:

shutDown()

void CAFirstMixA::shutDown ( )

virtual

Reimplemented from CAMix.

Definition at line 44 of file CAFirstMixA.cpp.

{
  m_bIsShuttingDown = true;
  m_bLoop=false;
#ifdef PAYMENT
  UINT32 connectionsClosed = 0;
  fmHashTableEntry* timeoutHashEntry;
 
 
  /* make sure no reconnect is possible when shutting down */
  if(m_pthreadAcceptUsers!=NULL)
  {
    CAMsg::printMsg(LOG_CRIT,"Wait for LoopAcceptUsers!\n");
    m_bRestart=true;
    m_pthreadAcceptUsers->join();
    delete m_pthreadAcceptUsers;
  }
  m_pthreadAcceptUsers=NULL;
 
  if(m_pInfoService != NULL)
  {
    CAMsg::printMsg(LOG_DEBUG,"Shutting down infoservice.\n");
    m_pInfoService->stop();
  }
  if(m_pChannelList!=NULL) // may happen if mixes did not yet connect to each other
  {
    while ((timeoutHashEntry = m_pChannelList->popTimeoutEntry(true)) != NULL)
    {
      CAMsg::printMsg(LOG_DEBUG,"Shutting down, closing client connection.\n");
      connectionsClosed++;
      closeConnection(timeoutHashEntry);
    }
    CAMsg::printMsg(LOG_DEBUG,"Closed %i client connections.\n", connectionsClosed);
  }
#endif
  m_bRestart = true;
  m_bIsShuttingDown = false;
}

References closeConnection(), CAThread::join(), CAFirstMix::m_bIsShuttingDown, CAMix::m_bLoop, CAFirstMix::m_bRestart, CAFirstMix::m_pChannelList, CAMix::m_pInfoService, CAFirstMix::m_pthreadAcceptUsers, CAFirstMixChannelList::popTimeoutEntry(), CAMsg::printMsg(), and CAInfoService::stop().

Here is the call graph for this function:

Friends And Related Function Documentation

fm_loopPacketProcessing

THREAD_RETURN fm_loopPacketProcessing ( void *  params )

friend

---

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

• CAFirstMixA.hpp
• CAFirstMixA.cpp