Ryzom

// NeL - MMORPG Framework <http://dev.ryzom.com/projects/nel/>

// Copyright (C) 2010  Winch Gate Property Limited

//

// This program is free software: you can redistribute it and/or modify

// it under the terms of the GNU Affero General Public License as

// published by the Free Software Foundation, either version 3 of the

// License, or (at your option) any later version.

//

// This program is distributed in the hope that it will be useful,

// but WITHOUT ANY WARRANTY; without even the implied warranty of

// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

// GNU Affero General Public License for more details.

//

// You should have received a copy of the GNU Affero General Public License

// along with this program.  If not, see <http://www.gnu.org/licenses/>.

#include "stdnet.h"

#include <limits>

#include "nel/misc/entity_id.h" // for createMessage()

#include "nel/misc/variable.h"

#include "nel/misc/thread.h"

#include "nel/misc/mutex.h"

#include "nel/net/unified_network.h"

#include "nel/net/module_common.h"

#include "nel/net/naming_client.h"

#ifdef NL_OS_UNIX

#include <sched.h>

#endif

using namespace std;

using namespace NLMISC;

namespace NLNET {

static size_t ThreadCreator = 0;

static const uint64 AppIdDeadConnection = 0xDEAD;

uint32 TotalCallbackCalled = 0;

uint32 TimeInCallback =0;

#ifdef NL_OS_UNIX

/// Yield method (Unix only)

CVariable<uint32> UseYieldMethod("nel", "UseYieldMethod", "0=select 1=usleep 2=nanosleep 3=sched_yield 4=none", 0, 0, true );

#endif

/// Reduce sending lag

CVariable<bool> FlushSendsBeforeSleep("nel", "FlushSendsBeforeSleep", "If true, send buffers will be flushed before sleep, not in next update", true, 0, true );

/// Network congestion monitoring

CVariable<uint> L5TotalBytesInLowLevelSendQueues("nel", "L5TotalBytesInLowLevelSendQueues", "Number of bytes pending in send queues (postponed by non-blocking send()) for network congestion monitoring. N/A if FlushSendsBeforeSleep disabled)", 0, 0, true );

/// Receiving size limit

CVariablePtr<uint32> DefaultMaxExpectedBlockSize("nel", "DefaultMaxExpectedBlockSize", "If receiving more than this value in bytes, the connection will be dropped", &CBufNetBase::DefaultMaxExpectedBlockSize, true );

/// Sending size limit

CVariablePtr<uint32> DefaultMaxSentBlockSize("nel", "DefaultMaxSentBlockSize", "If sending more than this value in bytes, the program may be stopped", &CBufNetBase::DefaultMaxSentBlockSize, true );

#define AUTOCHECK_DISPLAY nlwarning

//#define AUTOCHECK_DISPLAY CUnifiedNetwork::getInstance()->displayInternalTables (), nlerror

const TServiceId        TServiceId::InvalidId = TServiceId(std::numeric_limits<uint16>::max());

//

// Callbacks from NAMING SERVICE

//

// when a service registers

void        uNetRegistrationBroadcast(const string &name, TServiceId sid, const vector<CInetAddress> &addr)

{

        nldebug ("HNETL5: + naming %s-%u '%s'", name.c_str(), sid.get(), vectorCInetAddressToString(addr).c_str ());

        CUnifiedNetwork *uni= CUnifiedNetwork::getInstance();

        if (uni->_SId == sid)

        {

                // it's me! don't add me!!!

                return;

        }

        // add the unified connection

        if(sid.get() >= uni->_IdCnx.size ())

                uni->_IdCnx.resize (sid.get()+1);

        if (uni->_IdCnx[sid.get()].State == CUnifiedNetwork::CUnifiedConnection::NotUsed)

        {

                uni->_IdCnx[sid.get()] = CUnifiedNetwork::CUnifiedConnection(name, sid, false);

                uni->_UsedConnection.push_back (sid);

        }

        if (!uni->_IdCnx[sid.get()].ExtAddress.empty ())

                AUTOCHECK_DISPLAY ("HNETL5: %s-%u already inserted in the table with '%s'", name.c_str(), sid.get(), vectorCInetAddressToString (uni->_IdCnx[sid.get()].ExtAddress).c_str ());

        // set the list of external addresses

        nlassert (!addr.empty());

        uni->_IdCnx[sid.get()].ExtAddress = addr;

        // associate nid with ext address

        uni->_IdCnx[sid.get()].setupNetworkAssociation (uni->_NetworkAssociations, uni->_DefaultNetwork);

}

// when a service unregisters

void        uNetUnregistrationBroadcast(const string &name, TServiceId sid, const vector<CInetAddress> &addr)

{

        nldebug ("HNETL5: - naming %s-%u '%s'", name.c_str(), sid.get(), vectorCInetAddressToString (addr).c_str ());

        // get the service connection

        CUnifiedNetwork *uni = CUnifiedNetwork::getInstance();

        CUnifiedNetwork::CUnifiedConnection *uc = uni->getUnifiedConnection (sid);

        if (uc == 0) return;        // should never happen, the getUnifiedConnection() will generate a AUTOCHECK_DISPLAY

        // call the user callback

        uni->callServiceDownCallback(uc->ServiceName, uc->ServiceId);

        if(!uc->Connections.empty ())

        {

                // set all connection to dead, now, all messages received on this socket will be ignored and closed

                for (uint i = 0; i < uc->Connections.size (); ++i)

                {

                        if (uc->Connections[i].valid())

                                uc->Connections[i].setAppId (AppIdDeadConnection);

                }

                //

                // It's the first connection that added the _NamedCnx so if there s no connection, no need to

                // remove entry in _NamedCnx

                //

                uni->removeNamedCnx (uc->ServiceName, uc->ServiceId);

        }

        // remove the _UsedConnection

        bool found = false;

        for (vector<TServiceId>::iterator it = uni->_UsedConnection.begin (); it != uni->_UsedConnection.end(); it++)

        {

                if (*it == uc->ServiceId)

                {

                        found = true;

                        uni->_UsedConnection.erase (it);

                        break;

                }

        }

        if (!found)

                AUTOCHECK_DISPLAY ("HNETL5: can't find the sid %hu in the _UsedConnection", uc->ServiceId.get());

        // reset the unified connection

        uc->reset ();

}

//

// Callbacks from connection/disconnection services

//

void        uncbConnection(TSockId from, void * /* arg */)

{

        nlinfo ("HNETL5: + connec '%s'", from->asString().c_str());

        from->setAppId (AppIdDeadConnection);

}

void        uncbDisconnection(TSockId from, void * /* arg */)

{

        if(from->appId () == AppIdDeadConnection)

        {

                nlinfo ("HNETL5: - connec '%s'", from->asString().c_str());

        }

        else

        {

                CUnifiedNetwork        *uni = CUnifiedNetwork::getInstance();

                TServiceId                sid(uint16(from->appId()));

                CUnifiedNetwork::CUnifiedConnection *uc = uni->getUnifiedConnection (sid);

                if (uc == 0)

                {

                        nlinfo ("HNETL5: - connec '%s' sid %hu", from->asString().c_str(), sid.get());

                }

                else

                {

                        nlinfo ("HNETL5: - connec '%s' %s-%hu", from->asString().c_str(), uc->ServiceName.c_str (), sid.get());

                        if (uc->IsExternal)

                        {

                                if (!uc->AutoRetry)

                                {

                                        // If it s a external service with no auto retry, remove the connection

                                        // call the user callback

                                        uni->callServiceDownCallback(uc->ServiceName, uc->ServiceId);

                                        uni->removeNamedCnx (uc->ServiceName, uc->ServiceId);

                                        uni->_ConnectionToReset.push_back(uc->ServiceId);

                                }

                                else

                                {

                                        // call the user callback

                                        uni->callServiceDownCallback(uc->ServiceName, uc->ServiceId);

                                }

                        }

                        else

                        {

                                // reset the connection

                                uint i;

                                for (i = 0; i < uc->Connections.size (); i++)

                                {

                                        if (uc->Connections[i].valid() && uc->Connections[i].CbNetBase->getSockId(uc->Connections[i].HostId) == from)

                                        {

                                                if (uc->Connections[i].IsServerConnection)

                                                {

                                                        // we have to remove the stuffs now because HostId will not be accessible later

                                                        uc->Connections[i].reset();

                                                }

                                                else

                                                {

                                                        // if it s a client, we can't delete now because the callback client is currently in use

                                                        // only disconnect

                                                        if(uc->Connections[i].CbNetBase->connected ())

                                                        {

                                                                uc->Connections[i].CbNetBase->disconnect (uc->Connections[i].HostId);

                                                        }

                                                }

                                                break;

                                        }

                                }

                                if (i == uc->Connections.size ())

                                {

                                        AUTOCHECK_DISPLAY ("HNETL5: received a disconnection from a service but the connection is not in my list!");

                                }

                        }

                }

                from->setAppId (AppIdDeadConnection);

        }

}

//

// Callback from identification services

//

void        uncbServiceIdentification(CMessage &msgin, TSockId from, CCallbackNetBase &netbase)

{

        string                inSName;

        TServiceId        inSid;

        if (from->appId () != AppIdDeadConnection)

                AUTOCHECK_DISPLAY ("HNETL5: received a connect ident from an unknown connection 0x%"NL_I64"X", from->appId ());

        // recover the service name and id

        msgin.serial(inSName);

        msgin.serial(inSid);

        uint8 pos;

        msgin.serial (pos);

        bool isExternal;

        msgin.serial (isExternal);

        nlinfo ("HNETL5: + connect ident '%s' %s-%hu pos %hu ext %d", from->asString().c_str(), inSName.c_str(), inSid.get(), (uint16)pos, (uint8)isExternal);

        if(isExternal)

        {

                nlassert (pos == 0);

        }

        if (inSid.get() == 0)

        {

                if (isExternal)

                {

                        inSid = CUnifiedNetwork::getInstance ()->_ExtSId;

                        CUnifiedNetwork::getInstance ()->_ExtSId.set(CUnifiedNetwork::getInstance ()->_ExtSId.get()+1);

                        // the following was an nlwarning but this is in fact desired behavior and not an error so we have changed to nlinfo

                        nlinfo ("HNETL5: Received a connection from a service with a SId 0, we give him the SId %d", inSid.get());

                }

                else

                {

                        nlwarning ("HNETL5: Received a connection from a service with a SId 0 and wasn't external, disconnecting it");

                        netbase.disconnect();

                        return;

                }

        }

        from->setAppId(inSid.get());

        // add a new connection to the list

        CUnifiedNetwork                *uni= CUnifiedNetwork::getInstance();

        if(inSid.get() >= uni->_IdCnx.size ())

        {

                uni->_IdCnx.resize (inSid.get()+1);

        }

        switch(uni->_IdCnx[inSid.get()].State)

        {

        case CUnifiedNetwork::CUnifiedConnection::NotUsed:                // add the new unified connection

                uni->_IdCnx[inSid.get()] = CUnifiedNetwork::CUnifiedConnection(inSName, inSid, isExternal);

                uni->_UsedConnection.push_back (inSid);

                break;

        default:

                break;

        }

        if (uni->_IdCnx[inSid.get()].IsExternal != isExternal)

        {

                AUTOCHECK_DISPLAY ("HNETL5: Receive a connection that is not totally external %d %d", uni->_IdCnx[inSid.get()].IsExternal, isExternal);

                return;

        }

        bool FirstConnection = (uni->_IdCnx[inSid.get()].Connections.size () == 0);

        // add the connection to the already inserted unified connection

        if (pos >= uni->_IdCnx[inSid.get()].Connections.size ())

        {

                uni->_IdCnx[inSid.get()].Connections.resize(pos+1);

        }

        uni->_IdCnx[inSid.get()].Connections[pos] = CUnifiedNetwork::CUnifiedConnection::TConnection(&netbase, from);

        // If the connection is external, we'll never receive the ExtAddress by the naming service, so add it manually

        if (isExternal)

        {

                uni->_IdCnx[inSid.get()].ExtAddress.push_back (netbase.hostAddress (from));

                uni->_IdCnx[inSid.get()].setupNetworkAssociation (uni->_NetworkAssociations, uni->_DefaultNetwork);

        }

        // send the callback to the user with the first connection

        if (FirstConnection)

        {

                // insert the name in the map to be able to send message with the name

                uni->addNamedCnx (inSName, inSid);

                uni->callServiceUpCallback (inSName, inSid);

        }

}

// the callbacks wrapper

void        uncbMsgProcessing(CMessage &msgin, TSockId from, CCallbackNetBase &/* netbase */)

{

        if (from->appId() == AppIdDeadConnection)

        {

                AUTOCHECK_DISPLAY ("HNETL5: Receive a message from a dead connection");

                return;

        }

        CUnifiedNetwork                                                                        *uni = CUnifiedNetwork::getInstance();

        TServiceId                                                                                sid(uint16(from->appId()));

        CUnifiedNetwork::TMsgMappedCallback::iterator        itcb;

        itcb = uni->_Callbacks.find(msgin.getName());

        if (itcb == uni->_Callbacks.end())

        {

                // the callback doesn't exist

                nlwarning ("HNETL5: Can't find callback '%s' called by service %hu", msgin.getName().c_str(), sid.get());

        }

        else

        {

                CUnifiedNetwork::CUnifiedConnection *uc = uni->getUnifiedConnection (sid);

                if (uc == 0)

                {

                        nlwarning ("HNETL5: Received a message from a service %hu that is not ready (bad appid? 0x%"NL_I64"X)", sid.get(), from->appId ());

                        return;

                }

                if((*itcb).second == 0)

                {

                        nlwarning ("HNETL5: Received message %s from a service %hu but the associated callback is NULL", msgin.getName ().c_str(), sid.get());

                        return;

                }

                {

                        static map<string, CHTimer> timers;

                        map<string, CHTimer>::iterator it;

                        {

                                H_AUTO(L5UCHTimerOverhead);

                                string callbackName = "USRCB_" + msgin.getName();

                                it = timers.find(callbackName);

                                if(it == timers.end())

                                {

                                        it = timers.insert(make_pair(callbackName, CHTimer(NULL))).first;

                                        (*it).second.setName((*it).first.c_str());

                                }

                        }

                        {

                                H_AUTO(L5UserCallback);

                                TTime before = CTime::getLocalTime();

                                (*it).second.before();

//                                const std::string &cbName = itcb->first;

                                (*itcb).second (msgin, uc->ServiceName, sid);

                                (*it).second.after();

                                TTime after = CTime::getLocalTime();

                                // sum the time used to do callback

                                TimeInCallback += uint32((after-before));

                        }

                }

                uc->TotalCallbackCalled++;

                TotalCallbackCalled++;

        }

}

TCallbackItem        unServerCbArray[] =

{

        { "UN_SIDENT", uncbServiceIdentification }

};

//

// Alive check thread

//

class CAliveCheck : public NLMISC::IRunnable

{

public:

        CAliveCheck() : ExitRequired(false)        { }

        virtual void        run();

        virtual                        ~CAliveCheck()        { }

        volatile bool        ExitRequired;

        static CAliveCheck*        Thread;

        struct CCheckAddress

        {

                CCheckAddress() : ConnectionId(0xDEAD), NeedCheck(false), AddressValid(false) { }

                CInetAddress        Address;

                std::string                ServiceName;

                TServiceId                ServiceId;

                uint                        ConnectionId;

                uint                        ConnectionIndex;

                volatile bool        NeedCheck;

                volatile bool        AddressValid;

        };

        CCheckAddress                CheckList[128];

        void                        checkService(CInetAddress address, uint connectionId, uint connectionIndex, const std::string &service, TServiceId id);

};

CAliveCheck*        CAliveCheck::Thread = NULL;

IThread*                AliveThread = NULL;

void        CAliveCheck::run()

{

        // setup thread

        Thread = this;

        TTime        t = CTime::getLocalTime();

        while (!ExitRequired)

        {

                if (CTime::getLocalTime() - t < 10000)

                {

                        nlSleep(100);

                        continue;

                }

                uint        i;

                for (i=0; i<sizeof(CheckList)/sizeof(CheckList[0]); ++i)

                {

                        if (CheckList[i].NeedCheck && !CheckList[i].AddressValid)

                        {

                                try

                                {

                                        CCallbackClient        cbc;

                                        cbc.connect(CheckList[i].Address);

                                        // success (no exception)

                                        CheckList[i].AddressValid = true;

                                        cbc.disconnect();

                                }

                                catch (ESocketConnectionFailed &e)

                                {

#if FINAL_VERSION

                                        nlinfo ("HNETL5: can't connect to %s-%hu now (%s)", CheckList[i].ServiceName.c_str(), CheckList[i].ServiceId.get(), e.what ());

#else

                                        nlwarning ("HNETL5: can't connect to %s-%hu now (%s)", CheckList[i].ServiceName.c_str(), CheckList[i].ServiceId.get(), e.what ());

#endif

                                }

                        }

                }

                t = CTime::getLocalTime();

        }

        ExitRequired = false;

}

void        CAliveCheck::checkService(CInetAddress address, uint connectionId, uint connectionIndex, const std::string &service, TServiceId id)

{

        uint        i;

        for (i=0; i<sizeof(CheckList)/sizeof(CheckList[0]); ++i)

        {

                if (CheckList[i].NeedCheck)

                        continue;

                CheckList[i].Address = address;

                CheckList[i].ConnectionId = connectionId;

                CheckList[i].ConnectionIndex = connectionIndex;

                CheckList[i].ServiceName = service;

                CheckList[i].ServiceId = id;

                CheckList[i].AddressValid = false;

                CheckList[i].NeedCheck = true;

                return;

        }

}

//

//

//

bool        CUnifiedNetwork::init(const CInetAddress *addr, CCallbackNetBase::TRecordingState rec,

                                                          const string &shortName, uint16 port, TServiceId &sid)

{

        // the commands can now be invoked

        registerCommandsHandler();

        if (_Initialised)

        {

                AUTOCHECK_DISPLAY ("HNETL5: Unified network layer already initialized");

                return true;

        }

        ThreadCreator = NLMISC::getThreadId();

        vector<CInetAddress> laddr = CInetAddress::localAddresses();

        _RecordingState = rec;

        _Name = shortName;

        _SId = sid;

        if (addr != 0)

                _NamingServiceAddr = *addr;

        // if the address isn't 0, uses the naming service

        if (_NamingServiceAddr.isValid ())

        {

                // connect the callback to know when a new service comes in or goes down

                CNamingClient::setRegistrationBroadcastCallback(uNetRegistrationBroadcast);

                CNamingClient::setUnregistrationBroadcastCallback(uNetUnregistrationBroadcast);

                // connect to the naming service (may generate a ESocketConnectionFailed exception)

                CNamingClient::connect(_NamingServiceAddr, _RecordingState, laddr);

                if (port == 0)

                        port = CNamingClient::queryServicePort ();

        }

#ifdef NL_OS_UNIX

        /// Init the main pipe to select() on data available

        if ( ::pipe( _MainDataAvailablePipe ) != 0 )

                nlwarning( "Unable to create main D.A. pipe" );

        //nldebug( "Pipe: created" );

#endif

        // setup the server callback only if server port != 0, otherwise there's no server callback

        _ServerPort = port;

        if(_ServerPort != 0)

        {

                nlassert (_CbServer == 0);

                _CbServer = new CCallbackServer( CCallbackNetBase::Off, "", true, false ); // don't init one pipe per connection

#ifdef NL_OS_UNIX

                _CbServer->setExternalPipeForDataAvailable( _MainDataAvailablePipe ); // the main pipe is shared for all connections

                //nldebug( "Pipe: set (server %p)", _CbServer );

#endif

                bool retry = false;

                do

                {

                        retry = false;

                        try

                        {

                                _CbServer->init(port);

                        }

                        catch (ESocket &)

                        {

                                nlwarning("Failed to init the listen socket on port %u, is the service already running ?", port);

                                // wait a little before retrying

                                nlSleep(5000);

                                retry = true;

                        }

                } while(retry);

                _CbServer->addCallbackArray(unServerCbArray, 1);                                // the service ident callback

                _CbServer->setDefaultCallback(uncbMsgProcessing);                                // the default callback wrapper

                _CbServer->setConnectionCallback(uncbConnection, NULL);

                _CbServer->setDisconnectionCallback(uncbDisconnection, NULL);

        }

        else

        {

                nlinfo ("HNETL5: ServerPort is 0 so I don't create a CCallbackServer");

        }

        if (CNamingClient::connected())

        {

                // register the service

                for (uint i = 0; i < laddr.size(); i++)

                        laddr[i].setPort(_ServerPort);

                if (_SId.get() == 0)

                {

                        if ( ! CNamingClient::registerService(_Name, laddr, _SId) )

                        {

                                nlinfo ("HNETL5: Registration denied");

                                return false;

                        }

                }

                else

                {

                        CNamingClient::registerServiceWithSId(_Name, laddr, _SId);

                }

                sid = _SId;

                nlinfo ("HNETL5: Server '%s' added, registered and listen to port %hu", _Name.c_str (), _ServerPort);

        }

        AliveThread = IThread::create(new CAliveCheck(), 1024*4);

        AliveThread->start();

        _Initialised = true;

        return true;

}

void        CUnifiedNetwork::connect()

{

        nlassertex(_Initialised == true, ("Try to CUnifiedNetwork::connect() whereas it is not initialised yet"));

        if (ThreadCreator != NLMISC::getThreadId()) nlwarning ("HNETL5: Multithread access but this class is not thread safe thread creator = %u thread used = %u", ThreadCreator, NLMISC::getThreadId());

        if (CNamingClient::connected())

        {

                // get the services list

                const list<CNamingClient::CServiceEntry>        &services = CNamingClient::getRegisteredServices();

                // connects to the registered services

                list<CNamingClient::CServiceEntry>::const_iterator        its;

                // don't connect to itself

                for (its = services.begin(); its != services.end(); ++its)

                {

                        if (_SId != (*its).SId)

                        {

                                // add service with name, address, ident, not external, service id, and not autoretry (obsolete)

                                // we put the last true because the name callback should already inserted it by uNetRegistrationBroadcast()

                                addService((*its).Name, (*its).Addr, true, false, (*its).SId, false, true);

                        }

                        else

                        {

                                // don't process services received after mine because they'll connect to me

                                break;

                        }

                }

        }

}

void        CUnifiedNetwork::release(bool mustFlushSendQueues, const std::vector<std::string>& namesOfOnlyServiceToFlushSending)

{

        if (!_Initialised)

                return;

        // the commands can't be invoked

        unregisterCommandsHandler();

        // terminate the auto reconnection thread

        if (CAliveCheck::Thread)

        {

                CAliveCheck::Thread->ExitRequired = true;

                AliveThread->wait();

                delete CAliveCheck::Thread;

                delete AliveThread;

        }

        if (ThreadCreator != NLMISC::getThreadId()) nlwarning ("HNETL5: Multithread access but this class is not thread safe thread creator = %u thread used = %u", ThreadCreator, NLMISC::getThreadId());

        // Ensure all outgoing data are sent before disconnecting, if requested

        if ( mustFlushSendQueues )

        {

                nlinfo( "HNETL5: Flushing sending queues..." );

                float totalBytes = 0;

                uint bytesRemaining, i=0;

                while ( (bytesRemaining = tryFlushAllQueues( namesOfOnlyServiceToFlushSending )) != 0 )

                {

                        if ( i == 0 )

                                totalBytes = (float)bytesRemaining;

                        if ( i % 20 == 0 )

                        {

                                nldebug( "%.1f%% of %.3f KB flushed so far", // display without HNETL5 to bypass filters!

                                        ((float)(bytesRemaining-totalBytes))/totalBytes, totalBytes / 1024.0f );

                        }

                        ++i;

                        nlSleep( 100 );

                }

                nldebug( "HNETL5: Flush done in %u steps", i+1 );

        }

        // disconnect all clients

        if(_CbServer)

        {

                _CbServer->disconnect(InvalidSockId);

                delete _CbServer;

                _CbServer = 0;

        }

        // disconnect all connections to servers

        for (uint i = 0; i<_IdCnx.size(); ++i)

        {

                if (_IdCnx[i].State != CUnifiedNetwork::CUnifiedConnection::NotUsed)

                {

                        for(uint j = 0 ; j < _IdCnx[i].Connections.size (); j++)

                        {

                                if (_IdCnx[i].Connections[j].valid() && !_IdCnx[i].Connections[j].IsServerConnection)

                                {

                                        if (_IdCnx[i].Connections[j].CbNetBase->connected ())

                                                _IdCnx[i].Connections[j].CbNetBase->disconnect();

                                        delete _IdCnx[i].Connections[j].CbNetBase;

                                }

                        }

                        _IdCnx[i].Connections.clear ();

                }

        }

        // clear all other data

        _IdCnx.clear();

        _UsedConnection.clear ();

        _NamedCnx.clear();

        _UpCallbacks.clear();

        _DownCallbacks.clear();

        _Callbacks.clear();

        // disconnect the connection with the naming service

        if (CNamingClient::connected ())

                CNamingClient::disconnect ();

#ifdef NL_OS_UNIX

        ::close( _MainDataAvailablePipe[PipeRead] );

        ::close( _MainDataAvailablePipe[PipeWrite] );

#endif

}

void        CUnifiedNetwork::addService(const string &name, const CInetAddress &addr, bool sendId, bool external, TServiceId sid, bool autoRetry, bool shouldBeAlreayInserted)

{

        vector <CInetAddress> addrs;

        addrs.push_back (addr);

        addService (name, addrs, sendId, external, sid, autoRetry, shouldBeAlreayInserted);

}

void        CUnifiedNetwork::addService(const string &name, const vector<CInetAddress> &addr, bool sendId, bool external, TServiceId sid, bool autoRetry, bool shouldBeAlreayInserted)

{

        nlassertex(_Initialised == true, ("Try to CUnifiedNetwork::addService() whereas it is not initialised yet"));

        if (ThreadCreator != NLMISC::getThreadId()) nlwarning ("HNETL5: Multithread access but this class is not thread safe thread creator = %u thread used = %u", ThreadCreator, NLMISC::getThreadId());

        if (external)

        {

                sid = _ExtSId;

                _ExtSId.set(_ExtSId.get()+1);

        }

        nlinfo("HNETL5: addService %s-%hu '%s'", name.c_str(), sid.get(), vectorCInetAddressToString(addr).c_str());

        if (external && addr.size () != 1)

        {

                AUTOCHECK_DISPLAY ("HNETL5: Can't add external service with more than one connection");

        }

        // add the entry in the unified connection table

        if (sid.get() >= _IdCnx.size())

                _IdCnx.resize(sid.get()+1);

        CUnifiedConnection        *uc = &_IdCnx[sid.get()];

        // at this point it s possible that the service already added in the _IdCnx by the uNetRegistrationBroadcast()

        if (shouldBeAlreayInserted && _IdCnx[sid.get()].State == CUnifiedNetwork::CUnifiedConnection::NotUsed)  AUTOCHECK_DISPLAY ("HNETL5: the unified connection should already set by the naming reg broadcast and is not (%hu)", sid.get());

        if (!shouldBeAlreayInserted && _IdCnx[sid.get()].State == CUnifiedNetwork::CUnifiedConnection::Ready)  AUTOCHECK_DISPLAY ("HNETL5: the unified connection should not already set but is (%hu)", sid.get());

        if (_IdCnx[sid.get()].State == CUnifiedNetwork::CUnifiedConnection::NotUsed)

        {

                *uc = CUnifiedConnection(name, sid, external);

                _UsedConnection.push_back (sid);

        }

        else

        {

                // If the entry already set, check that all is correct

                if (name != uc->ServiceName) AUTOCHECK_DISPLAY ("HNETL5: name are different in addService %s %s", name.c_str (), uc->ServiceName.c_str ());

                if (sid != uc->ServiceId) AUTOCHECK_DISPLAY ("HNETL5: sid are different in addService %hu %hu", sid.get(), uc->ServiceId.get());

                if (addr != uc->ExtAddress) AUTOCHECK_DISPLAY ("HNETL5: external addr are different in addService '%s' '%s'", vectorCInetAddressToString(addr).c_str(), vectorCInetAddressToString(uc->ExtAddress).c_str ());

        }

        uc->AutoRetry = autoRetry;

        uc->SendId = sendId;

        uc->ExtAddress = addr;

        nlassert (!addr.empty());

        // associate nid with ext address

        uc->setupNetworkAssociation (_NetworkAssociations, _DefaultNetwork);

        // connect to all connection

        bool        connectSuccess;

        if (uc->Connections.size () < addr.size ())

        {

                uc->Connections.resize (addr.size ());

        }

        vector<CInetAddress> laddr = CInetAddress::localAddresses();

        for (uint i = 0; i < addr.size(); i++)

        {

                // first we have to look if we have a network that can established the connection

                uint j = 0;

                // it s 127.0.0.1, it s ok

                if (!addr[i].is127001 ())

                {

                        for (j = 0; j < laddr.size (); j++)

                        {

                                if (laddr[j].internalNetAddress () == addr[i].internalNetAddress ())

                                {

                                        // it's ok, we can try

                                        break;

                                }

                        }

                        // If we don't found a valid network, we'll try with the first one.

                        // It's happen, for example, when you try to connect to a service that is not in the network but use IP translation

                        if (j == laddr.size ())

                        {

                                nlwarning ("HNETL5: I can't access '%s' because I haven't a net card on this network, we'll use the first network", addr[i].asString ().c_str ());

                                j = 0;

                        }

                }

                // Create a new connection with the service, setup callback and connect

                CCallbackClient        *cbc = new CCallbackClient( CCallbackNetBase::Off, "", true, false ); // don't init one pipe per connection

#ifdef NL_OS_UNIX

                cbc->setExternalPipeForDataAvailable( _MainDataAvailablePipe ); // the main pipe is shared for all connections

                //nldebug( "Pipe: set (client %p)", cbc );

#endif

                cbc->setDisconnectionCallback(uncbDisconnection, NULL);

                cbc->setDefaultCallback(uncbMsgProcessing);

                cbc->getSockId()->setAppId(sid.get());

                try

                {

                        cbc->connect(addr[i]);

                        connectSuccess = true;

                }

                catch (ESocketConnectionFailed &e)

                {

                        nlwarning ("HNETL5: can't connect to %s (sid %u) now (%s) '%s'", name.c_str(), sid.get(), e.what (), addr[i].asString ().c_str());

                        connectSuccess = false;

                }

                if (!connectSuccess && !autoRetry)

                {

                        nlwarning ("HNETL5: Can't add service because no retry and can't connect");

                        delete cbc;

                }

                else

                {

                        uc->Connections[i] = CUnifiedNetwork::CUnifiedConnection::TConnection(cbc);

                }

                if (connectSuccess && sendId)

                {

                        // send identification to the service

                        CMessage        msg("UN_SIDENT");

                        msg.serial(_Name);

                        TServiceId                ssid = _SId;

                        if (uc->IsExternal)

                        {

                                // in the case that the service is external, we can't send our sid because the external service can

                                // have other connection with the same sid (for example, LS can have 2 WS with same sid => sid = 0 and leave

                                // the other side to find a good number

                                ssid.set(0);

                        }

                        msg.serial(ssid);        // serializes a 16 bits service id

                        uint8 pos = uint8(j);

                        msg.serial(pos);        // send the position in the connection table

                        msg.serial (uc->IsExternal);

                        cbc->send (msg);

                }

        }

        if (addr.size () != uc->Connections.size())

        {

                nlwarning ("HNETL5: Can't connect to all connections to the service %d/%d", addr.size (), uc->Connections.size());

        }

        bool cntok = false;

        for (uint j = 0; j < uc->Connections.size(); j++)

        {

                if (uc->Connections[j].CbNetBase != NULL)

                {

                        if (uc->Connections[j].CbNetBase->connected ())

                        {

                                cntok = true;

                                break;

                        }

                }

        }

        if (cntok)

        {

                // add the name only if at least one connection is ok

                addNamedCnx (name, sid);

                callServiceUpCallback (name, sid); // global callback ("*") will be called even for external service

        }

        nldebug ("HNETL5: addService was successful");

}

//

//

//

#define        TIME_BLOCK(tick, instr) \

{ \

        TTicks        _time_block_before = CTime::getPerformanceTime(); \

        instr ; \

        TTicks        _time_block_after = CTime::getPerformanceTime(); \

        tick += (_time_block_after - _before); \

}

/*

 *

 */

void        CUnifiedNetwork::update(TTime timeout)

{

        H_AUTO(CUnifiedNetworkUpdate);

        H_BEFORE(UNMisc1);

        nlassertex(_Initialised == true, ("Try to CUnifiedNetwork::update() whereas it is not initialised yet"));

        if (ThreadCreator != NLMISC::getThreadId()) nlwarning ("HNETL5: Multithread access but this class is not thread safe thread creator = %u thread used = %u", ThreadCreator, NLMISC::getThreadId());

        bool        enableRetry;        // true every 5 seconds to reconnect if necessary

        // Compute the real timeout based on the next update timeout

        TTime t0 = CTime::getLocalTime ();

        if (timeout > 0)

        {

                if (_NextUpdateTime == 0)

                {

                        _NextUpdateTime = t0 + timeout;

                }

                else

                {

                        TTime err = t0 - _NextUpdateTime;

                        _NextUpdateTime += timeout;

                        // if we are too late, resync to the next value

                        while (err > timeout)

                        {

                                err -= timeout;

                                _NextUpdateTime += timeout;

                        }

                        timeout -= err;

                        if (timeout < 0) timeout = 0;

                }

        }

        TTime remainingTime = timeout;

        // check if we need to retry to connect to the client

        enableRetry = (t0-_LastRetry > 5000);

        if (enableRetry)

                _LastRetry = t0;

        H_AFTER(UNMisc1);

        H_BEFORE(UNNamingCheck);

        // Try to reconnect to the naming service if connection lost

        if (_NamingServiceAddr.isValid ())

        {

                if (CNamingClient::connected ())

                {

                        CNamingClient::update ();

                }

                else if (enableRetry)

                {

                        H_AUTO(L5NSReconnect);

                        try

                        {

                                vector<CInetAddress> laddr = CInetAddress::localAddresses();

                                CNamingClient::connect (_NamingServiceAddr, _RecordingState, laddr);

                                // re-register the service

                                for (uint i = 0; i < laddr.size(); i++)

                                        laddr[i].setPort(_ServerPort);

                                CNamingClient::resendRegisteration (_Name, laddr, _SId);

                        }

                        catch (ESocketConnectionFailed &)

                        {

                                nlwarning ("HNETL5: Could not connect to the Naming Service (%s). Retrying in a few seconds...", _NamingServiceAddr.asString().c_str());

                        }

                }

        }

        H_AFTER(UNNamingCheck);

        H_BEFORE(UNUpdateCnx);

        for(;;)

        {

                H_AUTO(L5OneLoopUpdate);

                if (CAliveCheck::Thread)

                {

                        uint        i;

                        for (i=0; i<sizeof(CAliveCheck::Thread->CheckList)/sizeof(CAliveCheck::Thread->CheckList[0]); ++i)

                        {

                                CAliveCheck::CCheckAddress        &address = CAliveCheck::Thread->CheckList[i];

                                if (address.NeedCheck && address.AddressValid)

                                {

                                        CUnifiedConnection &uc = _IdCnx[address.ConnectionId];

                                        if (uc.ServiceName == address.ServiceName &&

                                                uc.ServiceId == address.ServiceId &&

                                                uc.ValidRequested)

                                        {

                                                uc.ValidRequested = false;

                                                uc.ValidExternal = true;

                                        }

                                        address.NeedCheck = false;

                                        address.AddressValid = false;

                                }

                        }

                }

                // update all server connections

                if (_CbServer)

                {

                        _CbServer->update2((sint32)remainingTime, 0);

                }

                // update all client connections

                for (uint k = 0; k<_UsedConnection.size(); ++k)

                {

                        H_AUTO(UNBrowseConnections);

                        nlassert (_IdCnx[_UsedConnection[k].get()].State == CUnifiedNetwork::CUnifiedConnection::Ready);

                        for (uint j = 0; j < _IdCnx[_UsedConnection[k].get()].Connections.size (); j++)

                        {

                                // WARNING : don't take a reference in the outside loop because

                                //                        _IdCnx can be resized by execution of a callback

                                CUnifiedConnection &uc = _IdCnx[_UsedConnection[k].get()];

                                nlassert(_IdCnx[_UsedConnection[k].get()].Connections.size() > j);

                                CUnifiedConnection::TConnection &conn = _IdCnx[_UsedConnection[k].get()].Connections[j];

                                H_AUTO(UNBrowseSubConnections);

                                if (!conn.valid())

                                        continue;

                                if (conn.IsServerConnection)

                                        continue;

                                nlassert(uc.Connections.size() > j);

                                if (conn.CbNetBase->connected ())

                                {

                                nlassert(uc.Connections.size() > j);

                                        conn.CbNetBase->update2((sint32)remainingTime, 0);

                                }

                                else if (enableRetry && uc.AutoRetry)

                                {

                                        if (uc.ValidExternal)

                                        {

                                                uc.ValidExternal = false;

                                                uc.ValidRequested = false;

                                                autoReconnect( uc, j );

                                        }

                                        else if (!uc.ValidRequested && CAliveCheck::Thread)

                                        {

                                                uc.ValidRequested = true;

                                                CAliveCheck::Thread->checkService(uc.ExtAddress[j], _UsedConnection[k].get(), j, uc.ServiceName, uc.ServiceId);

                                        }

                                }

                        }

                }

                // reset closed client connection

                for (uint i=0; i<_ConnectionToReset.size(); ++i)

                {

                        // remove the _UsedConnection

                        bool found = false;

                        for (vector<TServiceId>::iterator it = _UsedConnection.begin (); it != _UsedConnection.end(); it++)

                        {

                                if (*it == _IdCnx[_ConnectionToReset[i].get()].ServiceId)

                                {

                                        found = true;

                                        _UsedConnection.erase (it);

                                        break;

                                }

                        }

                        if (!found)

                                AUTOCHECK_DISPLAY ("HNETL5: can't find the sid %hu in the _UsedConnection", _IdCnx[_ConnectionToReset[i].get()].ServiceId.get());

                        _IdCnx[_ConnectionToReset[i].get()].reset();

                }

                _ConnectionToReset.clear();

                enableRetry = false;

                if ( FlushSendsBeforeSleep.get() )

                {

                        // Flush all connections

                        L5TotalBytesInLowLevelSendQueues = tryFlushAllQueues();

                }

                //      t0 --------- previousTime -------------------------- t0 + timeout

                //                                    prevRemainingTime

                //

                //      t0 -------------- currentTime ---------------------- t0 + timeout

                //                                        remainingTime

                TTime prevRemainingTime = remainingTime;

                TTime currentTime = CTime::getLocalTime();

                remainingTime = t0 + timeout - currentTime;

                // If it's the end (or if the Unix system time was changed forwards), don't sleep (currentTime > t0 + timeout)

                if ( remainingTime <= 0 )

                        break;

                // If the Unix system time was changed backwards, don't wait more than requested and don't provide an erroneous time to the sleep function that would fail (currentTime < previousTime)

                if ( remainingTime > prevRemainingTime )

                {

                        // Restart at previousTime

                        nldebug( "Backward time sync detected (at least -%"NL_I64"d ms)", remainingTime - prevRemainingTime );

                        remainingTime = prevRemainingTime;

                        t0 = currentTime - (timeout - remainingTime);

                }

#ifdef NL_OS_UNIX

                // Sleep until the time expires or we receive a message

                H_BEFORE(L5UpdateSleep);

                switch ( UseYieldMethod.get() )

                {

                case 0: sleepUntilDataAvailable( remainingTime ); break; // accurate sleep with select()

                case 1: ::usleep(1000); break; // 20 ms

                case 2: nlSleep(1); break; // 10 ms (by nanosleep, but 20 ms measured on kernel 2.4.20)

                case 3:        ::sched_yield(); break; // makes all slow (at least on kernel 2.4.20) !

                default: break; // don't sleep at all, makes all slow!

                }

                H_AFTER(L5UpdateSleep);

#else

                // Enable windows multithreading before rescanning all connections

                H_TIME(L5UpdateSleep, nlSleep(1);); // 0 (yield) would be too harmful to other applications

#endif

        }

        H_AFTER(UNUpdateCnx);

        H_TIME(UNAutoCheck, autoCheck(););

}

/*

 * Auto-reconnect

 */

void CUnifiedNetwork::autoReconnect( CUnifiedConnection &uc, uint connectionIndex )

{

        H_AUTO(L5AutoReconnect);

        try

        {

                CCallbackClient *cbc = (CCallbackClient *)uc.Connections[connectionIndex].CbNetBase;

                cbc->connect(uc.ExtAddress[connectionIndex]);

                uc.Connections[connectionIndex].CbNetBase->getSockId()->setAppId(uc.ServiceId.get());

                nlinfo ("HNETL5: reconnection to %s-%hu success", uc.ServiceName.c_str(), uc.ServiceId.get());

                // add the name only if at least one connection is ok

                if (!haveNamedCnx (uc.ServiceName, uc.ServiceId))

                        addNamedCnx (uc.ServiceName, uc.ServiceId);

                // resend the identification is necessary

                if (uc.SendId)

                {

                        // send identification to the service

                        CMessage        msg("UN_SIDENT");

                        msg.serial(_Name);

                        TServiceId                ssid = _SId;

                        if (uc.IsExternal)

                        {

                                // in the case that the service is external, we can't send our sid because the external service can

                                // have other connection with the same sid (for example, LS can have 2 WS with same sid => sid = 0 and leave

                                // the other side to find a good number

                                ssid.set(0);

                        }

                        msg.serial(ssid);        // serializes a 16 bits service id

                        uint8 pos = uint8(connectionIndex);

                        msg.serial(pos);        // send the position in the connection table

                        msg.serial (uc.IsExternal);

                        uc.Connections[connectionIndex].CbNetBase->send (msg, uc.Connections[connectionIndex].HostId);

                }

                // call the user callback

                callServiceUpCallback (uc.ServiceName, uc.ServiceId);

        }

        catch (ESocketConnectionFailed &e)

        {

#if FINAL_VERSION

                nlinfo ("HNETL5: can't connect to %s-%hu now (%s)", uc.ServiceName.c_str(), uc.ServiceId.get(), e.what ());

#else

                nlwarning ("HNETL5: can't connect to %s-%hu now (%s)", uc.ServiceName.c_str(), uc.ServiceId.get(), e.what ());

#endif

        }

}

#ifdef NL_OS_UNIX

/*

 *

 */

void CUnifiedNetwork::sleepUntilDataAvailable( TTime msecMax )

{

        // Prevent looping infinitely if an erroneous time was provided

        if ( msecMax > 999 ) // limit not told in Linux man but here: http://docs.hp.com/en/B9106-90009/select.2.html

                msecMax = 999;

        // Prepare for select()

        fd_set readers;

        FD_ZERO( &readers );

        FD_SET( _MainDataAvailablePipe[PipeRead], &readers );

        SOCKET descmax = _MainDataAvailablePipe[PipeRead] + 1;

        // Select

        timeval tv;

        tv.tv_sec = 0;

        tv.tv_usec = msecMax * 1000;

        //nldebug( "Select %u ms", (uint)msecMax );

        //TTime before = CTime::getLocalTime();

        int res = ::select( descmax+1, &readers, NULL, NULL, &tv );

        if ( res == -1 )

                nlwarning( "HNETL5: Select failed in sleepUntilDataAvailable");

        //nldebug( "Slept %u ms", (uint)(CTime::getLocalTime()-before) );

}

#endif

bool CUnifiedNetwork::isConnectionConnected(TServiceId sid) const

{

        // a Connected connection is a connection that is Ready but that is not yet connected (serverUp will be called latter via L5).

        return sid.get() < _IdCnx.size()

                && _IdCnx[sid.get()].State == CUnifiedConnection::Ready

                && !_IdCnx[sid.get()].Connections.empty();

}

//

//

//

uint8 CUnifiedNetwork::findConnectionId (TServiceId sid, uint8 nid)

{

        if (_IdCnx[sid.get()].Connections.size () == 0)

        {

                nlwarning ("HNETL5: Can't send message to %s because no connection are available", _IdCnx[sid.get()].ServiceName.c_str ());

                return 0xFF;

        }

        // by default, connection id will be the default one

        uint8 connectionId = _IdCnx[sid.get()].DefaultNetwork;

        if (nid == 0xFF)

        {

                // it s often happen because they didn't set a good network configuration, so it s in debug to disable it easily

                //nldebug ("HNETL5: nid %hu, will use the default connection %hu", (uint16)nid, (uint16)connectionId);

        }

        else if (nid >= _IdCnx[sid.get()].NetworkConnectionAssociations.size())

        {

                nlwarning ("HNETL5: No net association for nid %hu, use the default connection %hu", (uint16)nid, (uint16)connectionId);

        }

        else

        {

                if (_IdCnx[sid.get()].NetworkConnectionAssociations[nid] >= _IdCnx[sid.get()].Connections.size ())

                {

                        nlwarning ("HNETL5: Can't send message to %s because nid %d point on a bad connection (%d and only have %d cnx), use default connection", _IdCnx[sid.get()].ServiceName.c_str (), nid, connectionId, _IdCnx[sid.get()].Connections.size ());

                }

                else

                {

                        connectionId = _IdCnx[sid.get()].NetworkConnectionAssociations[nid];

                }

        }

        if (connectionId >= _IdCnx[sid.get()].Connections.size() || !_IdCnx[sid.get()].Connections[connectionId].valid() || !_IdCnx[sid.get()].Connections[connectionId].CbNetBase->connected())

        {

                // there's a problem with the selected connectionID, so try to find a valid one

                nlwarning ("HNETL5: Can't find selected connection id %hu to send message to %s because connection is not valid or connected, find a valid connection id", (uint16)connectionId, _IdCnx[sid.get()].ServiceName.c_str ());

                for (connectionId = 0; connectionId < _IdCnx[sid.get()].Connections.size(); connectionId++)

                {

                        if (_IdCnx[sid.get()].Connections[connectionId].valid() && _IdCnx[sid.get()].Connections[connectionId].CbNetBase->connected())

                        {

                                // we found one at last, use this one

                                //nldebug ("HNETL5: Ok, we found a valid connectionid, use %hu",  (uint16)connectionId);

                                break;

                        }

                }

                if (connectionId == _IdCnx[sid.get()].Connections.size())

                {

                        nlwarning ("HNETL5: Can't send message to %s because default connection is not exist, valid or connected", _IdCnx[sid.get()].ServiceName.c_str ());

                        return 0xFF;

                }

        }

        return connectionId;

}

//

//

//

uint        CUnifiedNetwork::send(const string &serviceName, const CMessage &msgout, bool warnIfNotFound, uint8 nid)

{

        nlassertex(_Initialised == true, ("Try to CUnifiedNetwork::send(const string&, const CMessage&) whereas it is not initialised yet"));

        if (ThreadCreator != NLMISC::getThreadId()) nlwarning ("HNETL5: Multithread access but this class is not thread safe thread creator = %u thread used = %u", ThreadCreator, NLMISC::getThreadId());

        TNameMappedConnection::const_iterator                                                                it;

        pair<TNameMappedConnection::const_iterator,TNameMappedConnection::const_iterator>        range;

        range = _NamedCnx.equal_range(serviceName);

        uint found = 0;

        if (range.first != _NamedCnx.end())

        {

                for (it=range.first; it!=range.second; ++it)

                {

                        TServiceId        sid = it->second;

                        if (sid.get() >= _IdCnx.size () || _IdCnx[sid.get()].State != CUnifiedNetwork::CUnifiedConnection::Ready)

                        {

                                // It often happen when the service is down (connection broke and the naming not already say that it s down)

                                // In this case, just warn

                                nlwarning ("HNETL5: Can't send %s to the service '%s' because it was in the _NamedCnx but not in _IdCnx (means that the service is down)", msgout.getName().c_str(), serviceName.c_str ());

                                return false;

                        }

                        ++found;

                        uint8 connectionId = findConnectionId (sid, nid);

                        if (connectionId == 0xff)        // failed

                        {

                                nlwarning ("HNETL5: Can't send %s message to %hu because no connection available", msgout.getName().c_str(), sid.get());

                                continue;

                        }

                        _IdCnx[sid.get()].Connections[connectionId].CbNetBase->send (msgout, _IdCnx[sid.get()].Connections[connectionId].HostId);

                }

        }

        if (!found && warnIfNotFound)

                nlwarning ("HNETL5: can't find service %s to send message %s", serviceName.c_str(), msgout.getName().c_str());

        return found;

}

bool        CUnifiedNetwork::send(TServiceId sid, const CMessage &msgout, uint8 nid)

{

        nlassertex(_Initialised == true, ("Try to CUnifiedNetwork::send(TServiceId, const CMessage&) whereas it is not initialised yet"));