// @(#)root/netx:$Name:  $:$Id: TXPhyConnection.cxx,v 1.6 2005/01/05 01:55:13 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.             *
 *************************************************************************/

//////////////////////////////////////////////////////////////////////////
//                                                                      //
// TXPhyConnection                                                      //
//                                                                      //
// Class handling physical connections to xrootd servers.               //
//                                                                      //
//////////////////////////////////////////////////////////////////////////

#include "TXPhyConnection.h"
#include "TError.h"
#include "TXDebug.h"
#include "TXMessage.h"
#include "TString.h"
#include "TEnv.h"
#include "TROOT.h"
#include "TSystem.h"
#include "TThread.h"
#include "TApplication.h"
#include "Riostream.h"


ClassImp(TXPhyConnection);

//____________________________________________________________________________
TThread::VoidFunc_t SocketReaderThread(void * arg)
{
   // This thread is the base for the async capabilities of TXPhyConnection
   // It repeatedly keeps reading from the socket, while feeding the
   // MsqQ with a stream of TXMessages containing what's happening
   // at the socket level

   TXPhyConnection *thisObj;

   if (DebugLevel() >= kHIDEBUG)
      Info("SocketReaderThread", "Reader Thread starting");

   // It should be possible to cancel the thread as soon as the
   // cancellation request is received.
   TThread::SetCancelAsynchronous();
   TThread::SetCancelOn();

   if (DebugLevel() >= kHIDEBUG)
      Info("SocketReaderThread", "Reader Thread started");

   thisObj = (TXPhyConnection *)arg;

   // Set running state ...
   thisObj->ReaderStarted();

   while (!(thisObj->ReaderThreadKilled())) {
      thisObj->BuildXMessage(kDefault, kTRUE, kTRUE);
      if (!thisObj->ReaderThreadKilled())
         thisObj->CheckAutoTerm();
   }

   if (DebugLevel() >= kHIDEBUG)
      Info("SocketReaderThread","Reader Thread exiting");
   return 0;
}

//____________________________________________________________________________
 TXPhyConnection::TXPhyConnection(TXAbsUnsolicitedMsgHandler *h)
{
   // Constructor

   // Initialization of lock mutex

   if (!(fRwMutex = new TMutex(kTRUE)))
      Error("TXPhyConnection",
            "can't create mutex for read/write: out of system resources");

   if (!(fMutex = new TMutex(kTRUE)))
      Error("TXPhyConnection",
            "can't create mutex for local locks: out of system resources");

   Touch();

   fServer = kUnknown;
   SetLogged(kNo);
   fRequestTimeout = gEnv->GetValue("XNet.RequestTimeout",
                                    DFLT_REQUESTTIMEOUT);
   UnsolicitedMsgHandler = h;

   fReaderthreadhandler = 0;
   fReaderthreadrunning = kFALSE;
   fReaderthreadkilled = kFALSE;

   fReaderCV = new TCondition();
}

//____________________________________________________________________________
 TXPhyConnection::~TXPhyConnection()
{
   // Destructor

   Disconnect();

   SafeDelete(fRwMutex);
   SafeDelete(fReaderCV);
}

//____________________________________________________________________________
 Bool_t TXPhyConnection::Connect(TString TcpAddress, Int_t TcpPort,
                                Int_t TcpWindowSize)
{
   // Connect to remote server
   R__LOCKGUARD(fMutex);

   if (DebugLevel() >= kHIDEBUG)
      Info("Connect", "Connecting to [%s:%d]", TcpAddress.Data(), TcpPort);

   fSocket = new TXSocket(TcpAddress, TcpPort, TcpWindowSize);

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

   fSocket->TryConnect();

   if (!fSocket->IsValid()) {
      Error("Connect",
            "can't open connection to xrootd/rootd on host [%s:%d]",
            TcpAddress.Data(), TcpPort);
      Disconnect();
      return kFALSE;
   }
   fSocket->SetOption(kNoDelay, 1);

   Touch();

   fTTLsec = DATA_TTL;

   if (DebugLevel() >= kHIDEBUG)
      Info("Connect", "Connected to host [%s:%d].",
	   TcpAddress.Data(), TcpPort);

   fRemoteAddress = TcpAddress;
   fRemotePort = TcpPort;
   fReaderthreadrunning = kFALSE;

   return kTRUE;
}

//____________________________________________________________________________
 void TXPhyConnection::StartReader()
{
   // Start reader thread

   bool running;
   {
      R__LOCKGUARD(fMutex);
      running = fReaderthreadrunning;
   }

   // Parametric asynchronous stuff.
   // If we are going Sync, then nothing has to be done,
   // otherwise the reader thread must be started
   if ( !running &&
         gEnv->GetValue("XNet.GoAsynchronous", DFLT_GOASYNC) ) {

      if (DebugLevel() >= kHIDEBUG)
         Info("StartReader", "Starting reader thread...");

      // Now we launch  the reader thread
      fReaderthreadhandler =
         new TThread((TThread::VoidFunc_t) SocketReaderThread, this);

      if (!fReaderthreadhandler)
         Info("StartReader",
              "Can't create reader thread: out of system resources");
      else {
         // Start the thread
         fReaderthreadhandler->Run();
         // Make sure that it is really running
         do {
            {
               R__LOCKGUARD(fMutex);
               running = fReaderthreadrunning;
            }
            if (!running) {
               if (DebugLevel() >= kHIDEBUG)
                  Info("StartReader","Waiting a little bit ...");
               fReaderCV->TimedWaitRelative(100);
            }
         } while (!running);
      }
   }
}

//____________________________________________________________________________
 void TXPhyConnection::ReaderStarted()
{
   // Called inside SocketReaderThread to flag the running status
   // of the thread

   R__LOCKGUARD(fMutex);
   fReaderthreadrunning = kTRUE;
   fReaderCV->Signal();
}

//____________________________________________________________________________
 Bool_t TXPhyConnection::ReConnect(TString TcpAddress, Int_t TcpPort,
                                  Int_t TcpWindowSize)
{
   // Re-connection attempt

   Disconnect();
   return Connect(TcpAddress, TcpPort, TcpWindowSize);
}

//____________________________________________________________________________
 TSocket *TXPhyConnection::SaveSocket()
{
   // Return copy of the TSocket part of the existing socket
   // Used to save an open connection to rootd daemons

   TSocket *opensock = 0;
   if (fSocket) {

      if (fReaderthreadrunning) {
         fReaderthreadkilled = kTRUE;
         fReaderthreadhandler->Kill();
      }

      // Extract TSocket part of TXSocket
      opensock = fSocket->ExtractSocket();

      // Signal deactivation
      fTTLsec = 0;
   }
   return opensock;
}

//____________________________________________________________________________
 void TXPhyConnection::Disconnect()
{
   // Terminate connection

   if (fReaderthreadrunning) {
      fReaderthreadkilled = kTRUE;
      fReaderthreadhandler->Kill();
   }

   fSocket = 0;
}

//____________________________________________________________________________
 void TXPhyConnection::Touch()
{
   // Set last-use-time to present time

   R__LOCKGUARD(fMutex);

   time_t t = time(0);
   if (DebugLevel() >= kDUMPDEBUG)
      Info("Touch", "Setting 'fLastUseTimestamp' to current time: %d",t);

   fLastUseTimestamp = t;
}

//____________________________________________________________________________
 Int_t TXPhyConnection::ReadRaw(void *buf, Int_t len, ESendRecvOptions opt)
{
   // Receive 'len' bytes from the connected server and store them in 'buf'.
   // Return number of bytes received.

   Int_t res;

   Touch();

   if (IsValid()) {

      if (DebugLevel() >= kDUMPDEBUG)
         Info("ReadRaw", "Reading from socket: %d [%s:%d]",
               fSocket->GetDescriptor(), fRemoteAddress.Data(), fRemotePort);

      res = fSocket->RecvRaw(buf, len, opt);

      if ((res <= 0) && (res != TXSOCK_ERR_TIMEOUT) && fReaderthreadkilled &&
          (DebugLevel() >= kHIDEBUG)) {
         Info("ReadRaw", "Reader thread has been killed");
         return res;
      }

      if ((res <= 0) && (res != TXSOCK_ERR_TIMEOUT) &&
          (DebugLevel() >= kHIDEBUG) && (gSystem->GetErrno()) )
         Info("ReadRaw", "Read error [%s:%d]. Errno %d:'%s'.",
               fRemoteAddress.Data(), fRemotePort, gSystem->GetErrno(),
               gSystem->GetError() );

      // If a socket error comes, then we disconnect (and destroy the fSocket)
      // but we have not to disconnect in the case of a timeout
      if (((res <= 0) && (res != TXSOCK_ERR_TIMEOUT)) ||
          (!fSocket->IsValid())) {

         if (DebugLevel() >= kHIDEBUG)
            Info("ReadRaw",
                 "Socket reported a disconnection (server[%s:%d]). Closing.",
                 fRemoteAddress.Data(), fRemotePort);
         Disconnect();
      }

      Touch();

      return res;
   }
   else {
      // Socket already destroyed or disconnected
      if (DebugLevel() >= kDUMPDEBUG)
         Info("ReadRaw", "Socket is disconnected (server [%s:%d])",
              fRemoteAddress.Data(), fRemotePort);
      return TXSOCK_ERR;
   }
}

//____________________________________________________________________________
 TXMessage *TXPhyConnection::ReadXMessage(Int_t streamid)
{
   // Gets a full loaded TXMessage from this phyconn.
   // May be a pure msg pick from a queue

   Touch();
   return fMsgQ.GetMsg(streamid, fRequestTimeout );

}

//____________________________________________________________________________
 TXMessage *TXPhyConnection::BuildXMessage(ESendRecvOptions opt,
                                          Bool_t IgnoreTimeouts, Bool_t Enqueue)
{
   // Builds an TXMessage, and makes it read its header/data from the socket
   // Also put automatically the msg into the queue

   TXMessage *m;

   m = new TXMessage();
   if (!m) {
      Error("BuildXMessage", "Fatal ERROR *** Object creation with new failed !"
                             " Probable system resources exhausted.");
      gSystem->Abort();
   }
   m->ReadRaw(this, opt);

   if (fReaderthreadkilled) {
      if (m) delete m;
      return (TXMessage *)0;
   }

   if (m->IsAttn()) {

      // Here we insert the PhyConn-level support for unsolicited responses
      // Some of them will be propagated in some way to the upper levels
      //  TLogConn, TConnMgr, TXNetConn
      HandleUnsolicited(m);

      // The purpose of this message ends here
      delete m;
      m = 0;
   }
   else
      if (Enqueue) {
         // If we have to ignore the socket timeouts, then we have not to
         // feed the queue with them. In this case, the newly created TXMessage
         // has to be freed.
         bool waserror;
         if (IgnoreTimeouts) {

            if (m->GetStatusCode() != TXMessage::kXMSC_timeout) {
               waserror = m->IsError();
               fMsgQ.PutMsg(m);
               if (waserror)
                  for (int kk=0; kk < 10; kk++)
                      fMsgQ.PutMsg(0);
            } else {
               delete m;
               m = 0;
            }
         } else
            fMsgQ.PutMsg(m);
      }

   return m;
}

//____________________________________________________________________________
 void TXPhyConnection::CheckAutoTerm()
{
   // Check if auto-termination is needed

   Bool_t doexit = kFALSE;
   {
      R__LOCKGUARD(fMutex);

      // Parametric asynchronous stuff
      // If we are going async, we might be willing to term ourself
      if (!IsValid() && gEnv->GetValue("XNet.GoAsynchronous", DFLT_GOASYNC)) {
         if (TThread::SelfId() == fReaderthreadhandler->GetId()) {
            // Notify termination, if requested
            if (DebugLevel() >= kHIDEBUG)
               Info("CheckAutoTerm", "self-Cancelling reader thread.");
            // Reset thread handlers (real termination will be done
            // at ::Exit() )
            fReaderthreadhandler = 0;
            fReaderthreadrunning = kFALSE;
            // Destroy the socket
            delete fSocket;
            fSocket = 0;
            // We are going to exit
            doexit = kTRUE;
         }
      }
   }

   // Now exit, if requested
   if (doexit) {
      UnlockChannel();
      TThread::Exit(0);
   }
}

//____________________________________________________________________________
 void TXPhyConnection::HandleUnsolicited(TXMessage *m)
{
   // Local processing of unsolicited responses is done here

   Bool_t ProcessingToGo = kTRUE;
   struct ServerResponseBody_Attn *attnbody;

   Touch();

   // Local processing of the unsolicited TXMessage
   attnbody = (struct ServerResponseBody_Attn *)m->GetData();
   if (attnbody) {

      switch (attnbody->actnum) {
      case kXR_asyncms:
         // A message arrived from the server. Let's print it.
         Info("HandleUnsolicited",
              "Message from server at socket %d[%s:%d]: '%s'.",
              fSocket->GetDescriptor(), fRemoteAddress.Data(),
              fRemotePort, attnbody->parms);
         ProcessingToGo = kFALSE;
         break;
      }
   }

   // Now we propagate the message to the interested object, if any
   // It could be some sort of upper layer of the architecture
   if (ProcessingToGo)
      SendUnsolicitedMsg(this, m);
}

//____________________________________________________________________________
 Int_t TXPhyConnection::WriteRaw(const void *buf, Int_t len, ESendRecvOptions opt)
{
   // Send 'len' bytes located at 'buf' to the connected server.
   // Return number of bytes sent.

   Int_t res;

   Touch();

   if (IsValid()) {

      if (DebugLevel() >= kDUMPDEBUG)
         Info("WriteRaw", "Writing to socket %d[%s:%d]",
              fSocket->GetDescriptor(), fRemoteAddress.Data(), fRemotePort);

      res = fSocket->SendRaw(buf, len, opt);

      if ((res <= 0)  && (res != TXSOCK_ERR_TIMEOUT) &&
          (DebugLevel() >= kHIDEBUG) && (gSystem->GetErrno()))
         Info("WriteRaw", "Write error [%s:%d]. Errno: %d:'%s'.",
              fRemoteAddress.Data(), fRemotePort, gSystem->GetErrno(),
              gSystem->GetError() );

      // If a socket error comes, then we disconnect (and destroy the fSocket)
      if ((res < 0) || (!fSocket->IsValid())) {
         if (DebugLevel() >= kHIDEBUG)
            Info("WriteRaw",
                 "Socket reported a disconnection (server[%s:%d]). Closing.",
                 fRemoteAddress.Data(), fRemotePort);
         Disconnect();
      }

      Touch();
      return( res );
   }
   else {
      // Socket already destroyed or disconnected
      if (DebugLevel() >= kDUMPDEBUG)
         Info("WriteRaw", "Socket is disconnected (server [%s:%d])",
              fRemoteAddress.Data(), fRemotePort);
      return TXSOCK_ERR;
   }
}

//____________________________________________________________________________
 UInt_t TXPhyConnection::GetBytesSent()
{
   // Return number of bytes sent

   if (IsValid())
      return fSocket->GetBytesSent();
   else {
      // Socket already destroyed or disconnected
      if (DebugLevel() >= kDUMPDEBUG)
         Info("GetBytesSent",
              "Socket is disconnected (server [%s:%d])",
              fRemoteAddress.Data(), fRemotePort);
      return 0;
   }
}

//____________________________________________________________________________
 UInt_t TXPhyConnection::GetBytesRecv()
{
   // Return number of bytes received

   if (IsValid())
      return fSocket->GetBytesRecv();
   else {
      // Socket already destroyed or disconnected
      if(DebugLevel() >= kDUMPDEBUG)
         Info("GetBytesRecv",
              "Socket is disconnected (server [%s:%d])",
              fRemoteAddress.Data(), fRemotePort);
      return 0;
   }
}

//____________________________________________________________________________
 UInt_t TXPhyConnection::GetSocketBytesSent()
{
   // Return number of bytes sent

   if (IsValid())
      return fSocket->GetSocketBytesSent();
   else {
      // Socket already destroyed or disconnected
      if(DebugLevel() >= kDUMPDEBUG)
         Info("GetSocketBytesSent",
              "Socket is disconnected (server [%s:%s])",
              fRemoteAddress.Data(), fRemotePort);
      return 0;
   }
}

//____________________________________________________________________________
 UInt_t TXPhyConnection::GetSocketBytesRecv()
{
   // Return number of bytes received

   if (IsValid())
      return fSocket->GetSocketBytesRecv();
   else {
      // Socket already destroyed or disconnected
      if(DebugLevel() >= kDUMPDEBUG)
         Info("GetSocketBytesRecv",
              "Socket is disconnected (server [%s:%s])",
              fRemoteAddress.Data(), fRemotePort);
      return 0;
   }
}

//____________________________________________________________________________
 Bool_t TXPhyConnection::ExpiredTTL()
{
   // Check expiration time
   return( (time(0) - fLastUseTimestamp) > fTTLsec ? kTRUE : kFALSE);
}

//____________________________________________________________________________
 void TXPhyConnection::LockChannel()
{
   // Lock
   fRwMutex->Lock();
}

//____________________________________________________________________________
 void TXPhyConnection::UnlockChannel()
{
   // Unlock
   fRwMutex->UnLock();
}