TXSocket.h

Go to the documentation of this file.

// @(#)root/proofx:$Id$
// Author: G. Ganis Oct 2005

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

#ifndef ROOT_TXSocket
#define ROOT_TXSocket

//////////////////////////////////////////////////////////////////////////
//                                                                      //
// TXSocket                                                             //
//                                                                      //
// High level handler of connections to xproofd.                        //
//                                                                      //
//////////////////////////////////////////////////////////////////////////

#define DFLT_CONNECTMAXTRY           10

#ifndef ROOT_TSemaphore
#include "TSemaphore.h"
#endif
#ifndef ROOT_TString
#include "TString.h"
#endif
#ifndef ROOT_TList
#include "TList.h"
#endif
#ifndef ROOT_TMessage
#include "TMessage.h"
#endif
#ifndef ROOT_TUrl
#include "TUrl.h"
#endif
#ifndef ROOT_TSocket
#include "TSocket.h"
#endif
#ifndef ROOT_XrdProofConn
#include "XrdProofConn.h"
#endif
#ifndef XRC_UNSOLMSG_H
#include "XrdClient/XrdClientUnsolMsg.hh"
#endif

#include <list>
#include <mutex>

class TObjString;
class TXSockBuf;
class TXSockPipe;
class TXHandler;
class TXSocketHandler;
class XrdClientMessage;

// To transmit info to Handlers
typedef struct {
   Int_t   fInt1;
   Int_t   fInt2;
   Int_t   fInt3;
   Int_t   fInt4;
} XHandleIn_t;
typedef struct {
   Int_t   fOpt;
   const char *fMsg;
} XHandleErr_t;

class TXSocket  : public TSocket, public XrdClientAbsUnsolMsgHandler {

friend class TXProofMgr;
friend class TXProofServ;
friend class TXSlave;
friend class TXSocketHandler;
friend class TXSockPipe;
friend class TXUnixSocket;

private:
   char                fMode;          // 'e' (def) or 'i' (internal - proofsrv)
   kXR_int32           fSendOpt;       // Options for sending messages
   Short_t             fSessionID;     // proofsrv: remote ID of connected session
   TString             fUser;          // Username used for login
   TString             fHost;          // Remote host
   Int_t               fPort;          // Remote port

   Int_t               fLogLevel;      // Log level to be transmitted to servers

   TString             fBuffer;        // Container for exchanging information
   TObject            *fReference;     // Generic object reference of this socket
   TXHandler          *fHandler;       // Handler of asynchronous events (input, error)

   XrdProofConn       *fConn;          // instance of the underlying connection module

   // Asynchronous messages
   TSemaphore          fASem;          // Control access to conn async msg queue
   std::recursive_mutex fAMtx;         // To protect async msg queue
   Bool_t              fAWait;         // kTRUE if waiting at the async msg queue
   std::list<TXSockBuf *> fAQue;          // list of asynchronous messages
   Int_t               fByteLeft;      // bytes left in the first buffer
   Int_t               fByteCur;       // current position in the first buffer
   TXSockBuf          *fBufCur;        // current read buffer

   TSemaphore          fAsynProc;      // Control actions while processing async messages

   // Interrupts
   std::recursive_mutex fIMtx;         // To protect interrupt queue
   kXR_int32           fILev;          // Highest received interrupt
   Bool_t              fIForward;      // Whether the interrupt should be propagated

   // Process ID of the instatiating process (to signal interrupts)
   Int_t               fPid;

   // Whether to timeout or not
   Bool_t              fDontTimeout;   // If true wait forever for incoming messages
   Bool_t              fRDInterrupt;   // To interrupt waiting for messages

   // Version of the remote XrdProofdProtocol
   Int_t               fXrdProofdVersion;

   // Static area for input handling
   static TXSockPipe   fgPipe;         //  Pipe for input monitoring
   static TString      fgLoc;          // Location string
   static Bool_t       fgInitDone;     // Avoid initializing more than once

   // List of spare buffers
   static std::mutex   fgSMtx;          // To protect spare list
   static std::list<TXSockBuf *> fgSQue; // list of spare buffers

   // Manage asynchronous message
   Int_t               PickUpReady();
   TXSockBuf          *PopUpSpare(Int_t sz);
   void                PushBackSpare();

   // Post a message into the queue for asynchronous processing
   void                PostMsg(Int_t type, const char *msg = 0);

   // Wake up all threads waiting for at the semaphore (used by TXSlave)
   void                PostSemAll();

   // Auxilliary
   Int_t               GetLowSocket() const { return (fConn ? fConn->GetLowSocket() : -1); }

   static void         SetLocation(const char *loc = ""); // Set location string

   static void         InitEnvs(); // Initialize environment variables

public:
   // Should be the same as in proofd/src/XrdProofdProtocol::Urgent
   enum EUrgentMsgType { kStopProcess = 2000 };

   TXSocket(const char *url, Char_t mode = 'M', Int_t psid = -1, Char_t ver = -1,
            const char *logbuf = 0, Int_t loglevel = -1, TXHandler *handler = 0);
   virtual ~TXSocket();

   virtual void        Close(Option_t *opt = "");
   Bool_t              Create(Bool_t attach = kFALSE);
   void                DisconnectSession(Int_t id, Option_t *opt = "");

   void                DoError(int level,
                               const char *location, const char *fmt, va_list va) const;

   virtual UnsolRespProcResult ProcessUnsolicitedMsg(XrdClientUnsolMsgSender *s,
                                                     XrdClientMessage *msg);

   virtual Int_t       GetClientID() const { return -1; }
   virtual Int_t       GetClientIDSize() const { return 1; }
   Int_t               GetLogConnID() const { return (fConn ? fConn->GetLogConnID() : -1); }
   Int_t               GetOpenError() const { return (fConn ? fConn->GetOpenError() : -1); }
   Int_t               GetServType() const { return (fConn ? fConn->GetServType() : -1); }
   Int_t               GetSessionID() const { return (fConn ? fConn->GetSessionID() : -1); }
   Int_t               GetXrdProofdVersion() const { return fXrdProofdVersion; }

   Bool_t              IsValid() const { return (fConn ? (fConn->IsValid()) : kFALSE); }
   Bool_t              IsServProofd();
   virtual void        RemoveClientID() { }
   virtual void        SetClientID(Int_t) { }
   void                SetSendOpt(ESendRecvOptions o) { fSendOpt = o; }
   void                SetSessionID(Int_t id);

   // Send interfaces
   Int_t               Send(const TMessage &mess);
   Int_t               Send(Int_t kind) { return TSocket::Send(kind); }
   Int_t               Send(Int_t status, Int_t kind)
                                        { return TSocket::Send(status, kind); }
   Int_t               Send(const char *mess, Int_t kind = kMESS_STRING)
                                        { return TSocket::Send(mess, kind); }
   Int_t               SendRaw(const void *buf, Int_t len,
                               ESendRecvOptions opt = kDontBlock);

   TObjString         *SendCoordinator(Int_t kind, const char *msg = 0, Int_t int2 = 0,
                                       Long64_t l64 = 0, Int_t int3 = 0, const char *opt = 0);

   // Recv interfaces
   Int_t               Recv(TMessage *&mess);
   Int_t               Recv(Int_t &status, Int_t &kind)
                                        { return TSocket::Recv(status, kind); }
   Int_t               Recv(char *mess, Int_t max)
                                        { return TSocket::Recv(mess, max); }
   Int_t               Recv(char *mess, Int_t max, Int_t &kind)
                                        { return TSocket::Recv(mess, max, kind); }
   Int_t               RecvRaw(void *buf, Int_t len,
                               ESendRecvOptions opt = kDefault);

   // Interrupts
   Int_t               SendInterrupt(Int_t type);
   Int_t               GetInterrupt(Bool_t &forward);

   // Urgent message
   void                SendUrgent(Int_t type, Int_t int1, Int_t int2);

   // Interrupt the low level socket
   inline void         SetInterrupt(Bool_t i = kTRUE) {
                                        std::lock_guard<std::recursive_mutex> lock(fAMtx);
                                        fRDInterrupt = i;
                                        if (i && fConn) fConn->SetInterrupt();
                                        if (i && fAWait) fASem.Post(); }
   inline Bool_t       IsInterrupt()  { std::lock_guard<std::recursive_mutex> lock(fAMtx);
                                        return fRDInterrupt; }
   // Set / Check async msg queue waiting status
   inline void         SetAWait(Bool_t w = kTRUE) {
                                        std::lock_guard<std::recursive_mutex> lock(fAMtx);
                                        fAWait = w; }
   inline Bool_t       IsAWait()  { std::lock_guard<std::recursive_mutex> lock(fAMtx);
                                        return fAWait; }
   // Flush the asynchronous queue
   Int_t               Flush();

   // Ping the counterpart
   Bool_t              Ping(const char *ord = 0);

   // Request remote touch of the admin file associated with this connection
   void                RemoteTouch();
   // Propagate a Ctrl-C
   void                CtrlC();

   // Standard options cannot be set
   Int_t               SetOption(ESockOptions, Int_t) { return 0; }

   // Disable / Enable read timeout
   void                DisableTimeout() { fDontTimeout = kTRUE; }
   void                EnableTimeout() { fDontTimeout = kFALSE; }

   // Try reconnection after error
   virtual Int_t       Reconnect();

   ClassDef(TXSocket, 0) //A high level connection class for PROOF
};


//
// The following structure is used to store buffers received asynchronously
//
class TXSockBuf {
public:
   Int_t   fSiz;
   Int_t   fLen;
   Char_t *fBuf;
   Bool_t  fOwn;
   Int_t   fCid;

   TXSockBuf(Char_t *bp=0, Int_t sz=0, Bool_t own=1);
  ~TXSockBuf();

   void Resize(Int_t sz);

   static Long64_t BuffMem();
   static Long64_t GetMemMax();
   static void     SetMemMax(Long64_t memmax);

private:
   Char_t *fMem;
   static Long64_t fgBuffMem; // Total allocated memory
   static Long64_t fgMemMax;  // Max allocated memory allowed
};

//
// The following class describes internal pipes
//
class TXSockPipe {
public:

   TXSockPipe(const char *loc = "");
   virtual ~TXSockPipe();

   Bool_t       IsValid() const { return ((fPipe[0] >= 0 && fPipe[1] >= 0) ? kTRUE : kFALSE); }

   TXSocket    *GetLastReady();

   Int_t        GetRead() const { return fPipe[0]; }
   Int_t        Post(TSocket *s);  // Notify socket ready via global pipe
   Int_t        Clean(TSocket *s); // Clean previous pipe notification
   Int_t        Flush(TSocket *s); // Remove any instance of 's' from the pipe
   void         DumpReadySock();

   void         SetLoc(const char *loc = "") { fLoc = loc; }

private:
   std::recursive_mutex fMutex;  // Protect access to the sockets-ready list
   Int_t        fPipe[2];        // Pipe for input monitoring
   TString      fLoc;            // Location string
   TList        fReadySock;      // List of sockets ready to be read
};

//
// Guard for a semaphore
//
class TXSemaphoreGuard {
public:

   TXSemaphoreGuard(TSemaphore *sem) : fSem(sem), fValid(kTRUE) { if (!fSem || fSem->TryWait()) fValid = kFALSE; }
   virtual ~TXSemaphoreGuard() { if (fValid && fSem) fSem->Post(); }

   Bool_t       IsValid() const { return fValid; }

private:
   TSemaphore  *fSem;
   Bool_t       fValid;
};

#endif