// @(#)root/netx:$Name:  $:$Id: TXConnectionMgr.cxx,v 1.5 2005/05/08 13:49:31 rdm Exp $
// Author: Alvise Dorigo, Fabrizio Furano

/*************************************************************************
 * Copyright (C) 1995-2004, Rene Brun and Fons Rademakers.               *
 * All rights reserved.                                                  *
 *                                                                       *
 * For the licensing terms see $ROOTSYS/LICENSE.                         *
 * For the list of contributors see $ROOTSYS/README/CREDITS.             *
 *************************************************************************/

//////////////////////////////////////////////////////////////////////////
//                                                                      //
// TXConnectionMgr                                                      //
//                                                                      //
// Authors: Alvise Dorigo, Fabrizio Furano                              //
//          INFN Padova, 2003                                           //
//                                                                      //
// The Connection Manager handles socket communications for TXNetFile   //
// action: connect, disconnect, read, write. It is a static object of   //
// the TXNetFile class such that within a single application multiple   //
// TXNetFile objects share the same connection manager.                 //
// The connection manager maps multiple logical connections on a single //
// physical connection.                                                 //
// There is one and only one logical connection per client (XNTetFile   //
// object), and one and only one physical connection per server:port.   //
// Thus multiple TXNetFile objects withing a given application share    //
// the same physical TCP channel to communicate with the server.        //
// This reduces the time overhead for socket creation and reduces also  //
// the server load due to handling many sockets.                        //
//                                                                      //
//////////////////////////////////////////////////////////////////////////

#include "TEnv.h"
#include "TXConnectionMgr.h"
#include "TXNetConn.h"
#include "TXDebug.h"
#include "TXMessage.h"
#include "TError.h"

using namespace std;

extern TEnv *gEnv;

ClassImp(TXConnectionMgr);

//
// This is needed to provide a hook in TXNetConn upon successful load
// of libNetx.so
//
static TXConnectionMgr gInstance;
class TXConnectionMgrInit {
public:
   TXConnectionMgrInit() { TXNetConn::SetTXConnectionMgr(&gInstance); }
};
static TXConnectionMgrInit txconnectionmgr_init;

//_____________________________________________________________________________
extern "C" void * GarbageCollectorThread(void * arg)
{
   // Function executed in the garbage collector thread

   int i;
   TXConnectionMgr *thisObj = (TXConnectionMgr *)arg;

   TThread::SetCancelDeferred();
   TThread::SetCancelOn();

   while (!(thisObj->ThreadKilled())) {
      TThread::CancelPoint();

      if (!thisObj->ThreadKilled())
         thisObj->GarbageCollect();

      if (!thisObj->ThreadKilled()) {
         for (i = 0; i < 10; i++) {
             TThread::CancelPoint();
             if (gSystem)
                gSystem->Sleep(200);
         }
      }
   }
   return 0;
}

//____________________________________________________
 TXConnectionMgr::TXConnectionMgr()
{
   // TXConnectionMgr constructor.
   // Used only to create the static instance when loading libNetx.

   fMutex = 0;
   fThreadHandler = 0;
   fInitialized = kFALSE;
   fThreadKilled = kFALSE;
}

//____________________________________________________
 void TXConnectionMgr::Init()
{
   // Full initialization of the TXConnectionMgr instance.
   // Creates a Connection Manager object.
   // Starts the garbage collector thread, if requested

   // Do it only once
   if (fInitialized) return;

   // Initialization of lock mutex
   if (!fMutex)
      fMutex = new TMutex(kTRUE);
   if (!fMutex) {
      Info("Init", "Can't create mutex: out of system resources");
      return;
   }

   //
   // Garbage collector thread creation void *(*start_routine, void*)
   if (gEnv->GetValue("XNet.StartGarbageCollectorThread",
                      DFLT_STARTGARBAGECOLLECTORTHREAD)) {
      // The type of the thread func makes it a detached thread
      fThreadHandler =
         new TThread((TThread::VoidFunc_t) GarbageCollectorThread,this);
      if (!fThreadHandler) {
         Info("Init","Can't create garbage collector"
                     " thread: out of system resources");
         return;
      } else {
         // Start the thread
         fThreadHandler->Run();
      }
   }
   else
      if(DebugLevel() >= kHIDEBUG)
         Info("Init",
              "Explicitly requested not to start the garbage collector"
              " thread. Are you sure?");

   // Set flag
   fInitialized = kTRUE;

   return;
}

//_____________________________________________________________________________
 TXConnectionMgr::~TXConnectionMgr()
{
   // Deletes mutex locks, stops garbage collector thread.

   UInt_t i=0;

   {
      R__LOCKGUARD(fMutex);
      for (i = 0; i < fLogVec.size(); i++)
	 if (fLogVec[i])
            Disconnect(i, kFALSE);
      for (i = 0; i < fLogVec.size(); i++)
	 if (fLogVec[i])
            delete fLogVec[i];
      for (i = 0; i < fPhyVec.size(); i++)
	 if (fPhyVec[i])
            delete fPhyVec[i];
   }

   if (fThreadHandler) {
      fThreadKilled = kTRUE;
      fThreadHandler->Kill();
   }
}

//_____________________________________________________________________________
 void TXConnectionMgr::GarbageCollect()
{
   // Get rid of unused physical connections. 'Unused' means not used for a
   // TTL time from any logical one. The TTL depends on the kind of remote
   // server. For a load balancer the TTL is very high, while for a data server
   // is quite small.

   // Mutual exclusion on the vectors and other vars
   {
      R__LOCKGUARD(fMutex);

      if (DebugLevel() >= kDUMPDEBUG)
         Info("GarbageCollect",
              "%d: collecting gargage (%d phys conn, %d log conn)",
              TThread::SelfId(),fPhyVec.size(),fLogVec.size());

      // We cycle all the physical connections
      for (unsigned short int i = 0; i < fPhyVec.size(); i++) {

	 // If a single physical connection has no linked logical connections,
	 // then we kill it if its TTL has expired
	 if ( fPhyVec[i] && (GetPhyConnectionRefCount(fPhyVec[i]) <= 0) &&
	      fPhyVec[i]->ExpiredTTL() ) {

	    if (DebugLevel() >= kDUMPDEBUG)
	       Info("GarbageCollect", "Purging physical connection %d", i);
	    // Wait until the physical connection is unlocked (it may be
            // in use by slow processes)

            if (fPhyVec[i])
	       delete(fPhyVec[i]);
            RemovePhyConn(fPhyVec[i]);

	    if (DebugLevel() >= kHIDEBUG)
	       Info("GarbageCollect", "Purged physical connection %d", i);
	 }
      }
   }
}

//_____________________________________________________________________________
short int TXConnectionMgr::Connect(TString RemoteAddress,
                                   Int_t TcpPort, Int_t TcpWindowSize)
{
   // Connects to the remote server:
   //  - Looks for an existing physical connection already bound to
   //    RemoteAddress:TcpPort;
   //  - If needed, creates a TCP channel to RemoteAddress:TcpPort
   //    (this is a physical connection);
   //  - Creates a logical connection and binds it to the previous
   //    created physical connection;
   //  - Returns the logical connection ID. Every client will use this
   //    ID to deal with the server.

   TXLogConnection *logconn;
   TXPhyConnection *phyconn;
   short int  newid;
   Bool_t phyfound;

   // First we get a new logical connection object
   if (DebugLevel() >= kHIDEBUG)
      Info("Connect", "Creating a logical connection...");

   logconn = new TXLogConnection();
   if (!logconn) {
      Error("Connect","Fatal ERROR *** Object creation with new failed !"
                      " Probable system resources exhausted.");
      gSystem->Abort();
   }

   if(DebugLevel() >= kDUMPDEBUG)
      Info("Connect", "Getting lock...");

   {
      R__LOCKGUARD(fMutex);
      // If we already have a physical connection to that host:port,
      // then we use that
      phyfound = kFALSE;
      if (DebugLevel() >= kHIDEBUG)
	 Info("Connect","Looking for an available physical"
                        " connection for address [%s:%d]",
	                RemoteAddress.Data(), TcpPort);

      for (unsigned short int i=0; i < fPhyVec.size(); i++) {
	 if (fPhyVec[i] && fPhyVec[i]->IsValid() &&
	     fPhyVec[i]->IsPort(TcpPort) &&
             fPhyVec[i]->IsAddress(RemoteAddress)) {
	    // We link that physical connection to the new logical connection
	    fPhyVec[i]->Touch();
	    logconn->SetPhyConnection( fPhyVec[i] );
	    phyfound = kTRUE;
	    break;
	 }
      }
   }

   if (!phyfound) {

      if (DebugLevel() >= kHIDEBUG)
         Info("Connect",
              "Physical connection not found. Creating a new one...");

      // If not already present, then we must build a new physical connection,
      // and try to connect it
      // While we are trying to connect, the mutex must be unlocked
      // Note that at this point logconn is a pure local instance, so it
      // does not need to be protected by mutex
      phyconn = new TXPhyConnection(this);

      if (!phyconn) {
         Error("Connect","Fatal ERROR *** Object creation with new failed !"
                         " Probable system resources exhausted.");
         gSystem->Abort();
      }
      if (phyconn &&
          phyconn->Connect(RemoteAddress, TcpPort, TcpWindowSize)) {

         logconn->SetPhyConnection(phyconn);

         if (DebugLevel() >= kHIDEBUG)
            Info("Connect", "New physical connection to server [%s:%d]"
                            " succesfully created.",
                 RemoteAddress.Data(), TcpPort);
      } else
         return -1;
   }

   // Now, we are connected to the host desired.
   // The physical connection can be old or newly created
   {
      R__LOCKGUARD(fMutex);
      // Then, if needed, we push the physical connection into its vector
      if (!phyfound)
	 fPhyVec.push_back(phyconn);

      // Then we push the logical connection into its vector
      fLogVec.push_back(logconn);

      // Its ID is its position inside the vector, we must return it later
      newid = fLogVec.size()-1;

      // Now some debug log
      if (DebugLevel() >= kHIDEBUG) {
	 Int_t logCnt = 0, phyCnt = 0;

	 for (unsigned short int i=0; i < fPhyVec.size(); i++)
	    if (fPhyVec[i])
	       phyCnt++;
	 for (unsigned short int i=0; i < fLogVec.size(); i++)
	    if (fLogVec[i])
	       logCnt++;

	 Info("Connect",
	      "LogConn: size:%d, count:%d - PhyConn: size:%d, count:%d",
	      fLogVec.size(), logCnt, phyCnt, fPhyVec.size());
      }

   }


   return newid;
}

//_____________________________________________________________________________
 void TXConnectionMgr::Disconnect(short int LogConnectionID,
                                 Bool_t ForcePhysicalDisc)
{
   // Deletes a logical connection.
   // Also deletes the related physical one if ForcePhysicalDisc=TRUE.

   if (DebugLevel() >= kDUMPDEBUG)
      Info("Disconnect", "Getting lock...");

   {
      R__LOCKGUARD(fMutex);
      if ((UInt_t(LogConnectionID) >= fLogVec.size()) ||
          (!fLogVec[LogConnectionID])) {
         Error("Disconnect", "Destroying nonexistent logconnid %d.",
                LogConnectionID);
	 return;
      }

      if (ForcePhysicalDisc) {
	 // We disconnect the phyconn, but it will be removed by the
         // garbagecollector as soon as possible. Note that here we
         // cannot destroy the phyconn, since there can be other logconns
         // pointing to it the phyconn will be removed when there are no
	 // more logconns pointing to it
	 fLogVec[LogConnectionID]->GetPhyConnection()->SetTTL(0);
	 fLogVec[LogConnectionID]->GetPhyConnection()->Disconnect();
      }

      fLogVec[LogConnectionID]->GetPhyConnection()->Touch();
      SafeDelete(fLogVec[LogConnectionID]);
      RemoveLogConn(fLogVec[LogConnectionID]);
      fLogVec[LogConnectionID] = 0;

      if (DebugLevel() >= kDUMPDEBUG)
	 Info("Disconnect", "Unlocking...");
   }
}

//_____________________________________________________________________________
 Int_t TXConnectionMgr::ReadRaw(short int LogConnectionID, void *buffer,
                               Int_t BufferLength, ESendRecvOptions opt)
{
   // Read BufferLength bytes from the logical connection LogConnectionID

   TXLogConnection *logconn;

   logconn = GetConnection(LogConnectionID);
   if (logconn) {
      if (DebugLevel() >= kDUMPDEBUG)
         Info("ReadRaw", "Reading from logical connection %d",
              LogConnectionID);

      return logconn->ReadRaw(buffer, BufferLength, opt);
   }
   else {
      Info("ReadRaw", "There's not a logical connection with id=%d",
           LogConnectionID);

      return(-1);
   }
}

//_____________________________________________________________________________
 TXMessage *TXConnectionMgr::ReadMsg(short int LogConnectionID,
                                    ESendRecvOptions opt)
{
   TXLogConnection *logconn;
   TXMessage *mex;

   logconn = GetConnection(LogConnectionID);
   if (logconn) {
      //    if (DebugLevel() >= kDUMPDEBUG)
      //      Info("ReadMsg", "Reading from logical connection %d",
      // 	   LogConnectionID);
   }

   // Parametric asynchronous stuff.
   // If we are going Sync, then we must build the message here,
   // otherwise the messages come directly from the queue
   if ( !gEnv->GetValue("XNet.GoAsynchronous", DFLT_GOASYNC) ) {

      // We get a new message directly from the socket.
      // The message gets inserted inside the phyconn queue
      // This line of code will be moved to a reader thread
      // inside TXPhyConnection
      // Timeouts must not be ignored here, indeed they are an error
      // because we are waiting for a message that must come quickly
      mex = logconn->GetPhyConnection()->BuildXMessage(opt, kFALSE, kFALSE);

   }
   else {
      // Now we get the message from the queue, with the timeouts needed
      mex = logconn->GetPhyConnection()->ReadXMessage(LogConnectionID);
   }

   // Return the message unmarshalled to ClientServerCmd
   return mex;
}

//_____________________________________________________________________________
 Int_t TXConnectionMgr::WriteRaw(short int LogConnectionID, const void *buffer,
                                Int_t BufferLength, ESendRecvOptions opt)
{
   // Write BufferLength bytes into the logical connection LogConnectionID

   TXLogConnection *logconn;
   logconn = GetConnection(LogConnectionID);
   if (logconn) {
      if (DebugLevel() >= kDUMPDEBUG)
         Info("WriteRaw", "Writing %d bytes to logical connection %d.",
              BufferLength, LogConnectionID);

      return logconn->WriteRaw(buffer, BufferLength, opt);
   }
   else {
      Info("WriteRaw", "There's not a logical connection with id=%d",
           LogConnectionID);

      return(-1);
   }
}

//_____________________________________________________________________________
 TXLogConnection *TXConnectionMgr::GetConnection(short int LogConnectionID)
{
   // Return a logical connection object that has LogConnectionID as its ID.

   TXLogConnection *res;
   {
      R__LOCKGUARD(fMutex);
      res = fLogVec[LogConnectionID];
   }

   return res;
}

//_____________________________________________________________________________
short int TXConnectionMgr::GetPhyConnectionRefCount(TXPhyConnection *PhyConn)
{
   // Return the number of logical connections bound to the physical
   // one 'PhyConn'
   int cnt = 0;
   {
      R__LOCKGUARD(fMutex);
      for (unsigned short int i = 0; i < fLogVec.size(); i++)
	 if (fLogVec[i] &&
            (fLogVec[i]->GetPhyConnection() == PhyConn))
            cnt++;
   }
   return cnt;
}

//_____________________________________________________________________________
 Bool_t TXConnectionMgr::ProcessUnsolicitedMsg(TXUnsolicitedMsgSender *sender,
                                              TXMessage *unsolmsg)
{
   // We are here if an unsolicited response comes from a physical connection
   // The response comes in the form of an TXMessage *, that must NOT be
   // destroyed after processing. It is destroyed by the first sender.
   // Remember that we are in a separate thread, since unsolicited responses
   // are asynchronous by nature.

   Info("Write", "Processing unsolicited response");

   // Local processing ....

   // Now we propagate the message to the interested objects.
   // In our architecture, the interested objects are the objects which
   // self-registered in the logical connections belonging to the Phyconn
   // which threw the event
   // So we throw the evt towards each logical connection
   {
      R__LOCKGUARD(fMutex);
      for (unsigned short int i = 0; i < fLogVec.size(); i++)
	 if (fLogVec[i] &&
            (fLogVec[i]->GetPhyConnection() == sender)) {
	     fLogVec[i]->ProcessUnsolicitedMsg(sender, unsolmsg);
	 }
   }

   return kTRUE;
}

//_____________________________________________________________________________
 void TXConnectionMgr::RemoveLogConn(TXLogConnection *logc)
{
   // Remove logc from the list of logical connections

   {
      R__LOCKGUARD(fMutex);

      vector<TXLogConnection *>::iterator i;
      for (i = fLogVec.begin(); i != fLogVec.end(); ++i) {
         if (logc == *i) {
            fLogVec.erase(i);
            break;
         }
      }
   }
}

//_____________________________________________________________________________
 void TXConnectionMgr::RemovePhyConn(TXPhyConnection *phyc)
{
   // Remove phyc from the list of physical connections

   {
      R__LOCKGUARD(fMutex);

      vector<TXPhyConnection *>::iterator i;
      for (i = fPhyVec.begin(); i != fPhyVec.end(); ++i) {
         if (phyc == *i) {
            fPhyVec.erase(i);
            break;
         }
      }
   }
}