XrdClientConn.hh

Go to the documentation of this file.

#ifndef XRD_CONN_H
#define XRD_CONN_H
/******************************************************************************/
/*                                                                            */
/*                    X r d C l i e n t C o n n . h h                         */
/*                                                                            */
/* Author: Fabrizio Furano (INFN Padova, 2004)                                */
/* Adapted from TXNetFile (root.cern.ch) originally done by                   */
/*  Alvise Dorigo, Fabrizio Furano                                            */
/*          INFN Padova, 2003                                                 */
/*                                                                            */
/* This file is part of the XRootD software suite.                            */
/*                                                                            */
/* XRootD is free software: you can redistribute it and/or modify it under    */
/* the terms of the GNU Lesser General Public License as published by the     */
/* Free Software Foundation, either version 3 of the License, or (at your     */
/* option) any later version.                                                 */
/*                                                                            */
/* XRootD 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 Lesser General Public       */
/* License for more details.                                                  */
/*                                                                            */
/* You should have received a copy of the GNU Lesser General Public License   */
/* along with XRootD in a file called COPYING.LESSER (LGPL license) and file  */
/* COPYING (GPL license).  If not, see <http://www.gnu.org/licenses/>.        */
/*                                                                            */
/* The copyright holder's institutional names and contributor's names may not */
/* be used to endorse or promote products derived from this software without  */
/* specific prior written permission of the institution or contributor.       */
/******************************************************************************/

//////////////////////////////////////////////////////////////////////////
//                                                                      //
// High level handler of connections to xrootd.                         //
//                                                                      //
//////////////////////////////////////////////////////////////////////////

#include "XrdClient/XrdClientConst.hh"

#include "time.h"
#include "XrdClient/XrdClientConnMgr.hh"
#include "XrdClient/XrdClientMessage.hh"
#include "XrdClient/XrdClientUrlInfo.hh"
#include "XrdClient/XrdClientReadCache.hh"
#include "XrdOuc/XrdOucHash.hh"
#include "XrdSys/XrdSysPthread.hh"

#define ConnectionManager XrdClientConn::GetConnectionMgr()

class XrdClientAbs;
class XrdSecProtocol;

class XrdClientConn {

public:

    enum ESrvErrorHandlerRetval {
   kSEHRReturnMsgToCaller   = 0,
   kSEHRBreakLoop           = 1,
   kSEHRContinue            = 2,
   kSEHRReturnNoMsgToCaller = 3,
   kSEHRRedirLimitReached   = 4
    };
    enum EThreeStateReadHandler {
   kTSRHReturnMex     = 0,
   kTSRHReturnNullMex = 1,
   kTSRHContinue      = 2
    };

    // To keep info about an open session
    struct                     SessionIDInfo {
   char id[16];
    };

    int                        fLastDataBytesRecv;
    int                        fLastDataBytesSent;
    XErrorCode                 fOpenError;   

    XrdOucString               fRedirOpaque;        // Opaque info returned by the server when

    // redirecting. To be used in the next opens
    XrdClientConn();
    virtual ~XrdClientConn();

    inline bool                CacheWillFit(long long bytes) {
   if (!fMainReadCache)
       return FALSE;
   return fMainReadCache->WillFit(bytes);
    }

    bool                       CheckHostDomain(XrdOucString hostToCheck);
    short                      Connect(XrdClientUrlInfo Host2Conn,
                   XrdClientAbsUnsolMsgHandler *unsolhandler);
    void                       Disconnect(bool ForcePhysicalDisc);
    virtual bool               GetAccessToSrv();
    XReqErrorType              GoBackToRedirector();

    XrdOucString               GetClientHostDomain() { return fgClientHostDomain; }


    static XrdClientPhyConnection     *GetPhyConn(int LogConnID);


    // --------- Cache related stuff

    long                       GetDataFromCache(const void *buffer,
                  long long begin_offs,
                  long long end_offs,
                  bool PerfCalc,
                  XrdClientIntvList &missingblks,
                  long &outstandingblks );

    bool                       SubmitDataToCache(XrdClientMessage *xmsg,
                   long long begin_offs,
                   long long end_offs);

    bool                       SubmitRawDataToCache(const void *buffer,
                   long long begin_offs,
                   long long end_offs);

    void                       SubmitPlaceholderToCache(long long begin_offs,
                     long long end_offs) {
   if (fMainReadCache)
       fMainReadCache->PutPlaceholder(begin_offs, end_offs);
    }

  
    void                       RemoveAllDataFromCache(bool keepwriteblocks=true) {
        if (fMainReadCache)
            fMainReadCache->RemoveItems(keepwriteblocks);
    }

    void                       RemoveDataFromCache(long long begin_offs,
                                                   long long end_offs, bool remove_overlapped = false) {
        if (fMainReadCache)
     fMainReadCache->RemoveItems(begin_offs, end_offs, remove_overlapped);
    }

    void                       RemovePlaceholdersFromCache() {
        if (fMainReadCache)
            fMainReadCache->RemovePlaceholders();
    }

    void                       PrintCache() {
        if (fMainReadCache)
            fMainReadCache->PrintCache();
    }


  bool                       GetCacheInfo(
                 // The actual cache size
                 int &size,

                 // The number of bytes submitted since the beginning
                 long long &bytessubmitted,

                 // The number of bytes found in the cache (estimate)
                 long long &byteshit,

                 // The number of reads which did not find their data
                                          // (estimate)
                 long long &misscount,

                 // miss/totalreads ratio (estimate)
                 float &missrate,

                 // number of read requests towards the cache
                 long long &readreqcnt,

                 // ratio between bytes found / bytes submitted
                 float &bytesusefulness
                 ) {
      if (!fMainReadCache) return false;

      fMainReadCache->GetInfo(size,
               bytessubmitted,
               byteshit,
               misscount,
               missrate,
               readreqcnt,
               bytesusefulness);
      return true;
    }  
                 

    void                       SetCacheSize(int CacheSize) {
        if (!fMainReadCache && CacheSize)
     fMainReadCache = new XrdClientReadCache();

        if (fMainReadCache)
      fMainReadCache->SetSize(CacheSize);
    }

    void                       SetCacheRmPolicy(int RmPolicy) {
        if (fMainReadCache)
      fMainReadCache->SetBlkRemovalPolicy(RmPolicy);
    }

    void                       UnPinCacheBlk(long long begin_offs, long long end_offs) {
        fMainReadCache->UnPinCacheBlk(begin_offs, end_offs);
   // Also use this to signal the possibility to proceed for a hard checkpoint
   fWriteWaitAck->Broadcast();
    }


    // -------------------


    int                        GetLogConnID() const { return fLogConnID; }

    ERemoteServerType          GetServerType() const { return fServerType; }

    kXR_unt16                  GetStreamID() const { return fPrimaryStreamid; }

    inline XrdClientUrlInfo    *GetLBSUrl() { return fLBSUrl; }
    inline XrdClientUrlInfo    *GetMetaUrl() { return fMetaUrl; }
    inline XrdClientUrlInfo    GetCurrentUrl() { return fUrl; }
    inline XrdClientUrlInfo    GetRedirUrl() { return fREQUrl; }

    XErrorCode                 GetOpenError() const { return fOpenError; }
    virtual XReqErrorType      GoToAnotherServer(XrdClientUrlInfo &newdest);
    virtual XReqErrorType      GoToMetaManager();
    bool                       IsConnected() const { return fConnected; }
    bool                       IsPhyConnConnected();

    struct ServerResponseHeader
                               LastServerResp;

    struct ServerResponseBody_Error
                               LastServerError;

    void                       ClearLastServerError() {
                                   memset(&LastServerError, 0, sizeof(LastServerError));
                                   LastServerError.errnum = kXR_noErrorYet;
                               }

    UnsolRespProcResult        ProcessAsynResp(XrdClientMessage *unsolmsg);

    virtual bool               SendGenCommand(ClientRequest *req, 
                     const void *reqMoreData,       
                     void **answMoreDataAllocated,
                     void *answMoreData, bool HasToAlloc,
                     char *CmdName, int substreamid = 0);

    int                        GetOpenSockFD() const { return fOpenSockFD; }

    void                       SetClientHostDomain(const char *src) { fgClientHostDomain = src; }
    void                       SetConnected(bool conn) { fConnected = conn; }

    void                       SetOpenError(XErrorCode err) { fOpenError = err; }

    // Gets a parallel stream id to use to set the return path for a re
    int                        GetParallelStreamToUse(int reqsperstream);
    int                        GetParallelStreamCount();     // Returns the total number of connected streams

    void                       SetRedirHandler(XrdClientAbs *rh) { fRedirHandler = rh; }

    void                       SetRequestedDestHost(char *newh, kXR_int32 port) {
   fREQUrl = fUrl;
   fREQUrl.Host = newh;
   fREQUrl.Port = port;
   fREQUrl.SetAddrFromHost();
    }

    // Puts this instance in pause state for wsec seconds.
    // A value <= 0 revokes immediately the pause state
    void                       SetREQPauseState(kXR_int32 wsec) {
   // Lock mutex
   fREQWait->Lock();

   if (wsec > 0)
       fREQWaitTimeLimit = time(0) + wsec;
   else {
       fREQWaitTimeLimit = 0;
       fREQWait->Broadcast();
   }

   // UnLock mutex
   fREQWait->UnLock();
    }

    // Puts this instance in connect-pause state for wsec seconds.
    // Any future connection attempt will not happen before wsec
    //  and the first one will be towards the given host
    void                       SetREQDelayedConnectState(kXR_int32 wsec) {
   // Lock mutex
   fREQConnectWait->Lock();

   if (wsec > 0)
       fREQConnectWaitTimeLimit = time(0) + wsec;
   else {
       fREQConnectWaitTimeLimit = 0;
       fREQConnectWait->Broadcast();
   }

   // UnLock mutex
   fREQConnectWait->UnLock();
    }

    void                       SetSID(kXR_char *sid);
    inline void                SetUrl(XrdClientUrlInfo thisUrl) { fUrl = thisUrl; }

    // Sends the request to the server, through logconn with ID LogConnID
    // The request is sent with a streamid 'child' of the current one, then marked as pending
    // Its answer will be caught asynchronously
    XReqErrorType              WriteToServer_Async(ClientRequest *req, 
                     const void* reqMoreData,
                     int substreamid = 0);

    static XrdClientConnectionMgr *GetConnectionMgr()
    { return fgConnectionMgr;} //Instance of the conn manager

    static void DelSessionIDRepo()         {fSessionIDRMutex.Lock();
                                            fSessionIDRepo.Purge();
                                            fSessionIDRMutex.UnLock();
                                           }

    void GetSessionID(SessionIDInfo &sess) {sess = mySessionID;}

    long                       GetServerProtocol() { return fServerProto; }

    short                      GetMaxRedirCnt() const { return fMaxGlobalRedirCnt; }
    void                       SetMaxRedirCnt(short mx) {fMaxGlobalRedirCnt = mx; }
    short                      GetRedirCnt() const { return fGlobalRedirCnt; }

    bool                       DoWriteSoftCheckPoint();
    bool                       DoWriteHardCheckPoint();
    void                       UnPinCacheBlk();


    // To give a max number of seconds for an operation to complete, no matter what happens inside
    // e.g. redirections, sleeps, failed connection attempts etc.
    void                       SetOpTimeLimit(int delta_secs);
    bool                       IsOpTimeLimitElapsed(time_t timenow);


protected:
    void                       SetLogConnID(int cid) { fLogConnID = cid; }
    void                       SetStreamID(kXR_unt16 sid) { fPrimaryStreamid = sid; }



    // The handler which first tried to connect somewhere
    XrdClientAbsUnsolMsgHandler *fUnsolMsgHandler;

    XrdClientUrlInfo           fUrl;                // The current URL
    XrdClientUrlInfo           *fLBSUrl;            // Needed to save the load balancer url
    XrdClientUrlInfo           fREQUrl;             // For explicitly requested redirs

    short                      fGlobalRedirCnt;    // Number of redirections

private:

    static XrdOucString        fgClientHostDomain; // Save the client's domain name
    bool                       fConnected;
    bool                       fGettingAccessToSrv; // To avoid recursion in desperate situations
    time_t                     fGlobalRedirLastUpdateTimestamp; // Timestamp of last redirection

    int                        fLogConnID;        // Logical connection ID used
    kXR_unt16                  fPrimaryStreamid;  // Streamid used for normal communication
    // NB it's a copy of the one contained in
    // the logconn

    short                      fMaxGlobalRedirCnt;
    XrdClientReadCache         *fMainReadCache;

    // The time limit for a transaction
    time_t                     fOpTimeLimit;

    XrdClientAbs               *fRedirHandler;      // Pointer to a class inheriting from
    // XrdClientAbs providing methods
    // to handle a redir at higher level

    XrdOucString               fRedirInternalToken; // Token returned by the server when
    // redirecting. To be used in the next logins

    XrdSysCondVar              *fREQWaitResp;           // For explicitly requested delayed async responses
    ServerResponseBody_Attn_asynresp *
                               fREQWaitRespData;        // For explicitly requested delayed async responses

    time_t                     fREQWaitTimeLimit;   // For explicitly requested pause state
    XrdSysCondVar              *fREQWait;           // For explicitly requested pause state
    time_t                     fREQConnectWaitTimeLimit;   // For explicitly requested delayed reconnect
    XrdSysCondVar              *fREQConnectWait;           // For explicitly requested delayed reconnect

    long                       fServerProto;        // The server protocol
    ERemoteServerType          fServerType;         // Server type as returned by doHandShake() 
    SessionIDInfo              mySessionID;         // Login session ID


    static XrdSysMutex         fSessionIDRMutex;    // Mutex for the Repo
    static XrdOucHash<SessionIDInfo>
    fSessionIDRepo;      // The repository of session IDs, shared.
    // Association between
    // <hostname>:<port>.<user> and a SessionIDInfo struct

    int                        fOpenSockFD;         // Descriptor of the underlying socket
    static XrdClientConnectionMgr *fgConnectionMgr; //Instance of the Connection Manager

    XrdSysCondVar              *fWriteWaitAck;
    XrdClientVector<ClientRequest> fWriteReqsToRetry; // To store the write reqs to retry in case of a disconnection

    bool                       CheckErrorStatus(XrdClientMessage *, short &, char *);
    void                       CheckPort(int &port);
    void                       CheckREQPauseState();
    void                       CheckREQConnectWaitState();
    bool                       CheckResp(struct ServerResponseHeader *resp, const char *method);
    XrdClientMessage           *ClientServerCmd(ClientRequest *req,
                  const void *reqMoreData,
                  void **answMoreDataAllocated,
                  void *answMoreData,
                  bool HasToAlloc,
                  int substreamid = 0);
    XrdSecProtocol            *DoAuthentication(char *plist, int plsiz);

    ERemoteServerType          DoHandShake(short log);

    bool                       DoLogin();
    bool                       DomainMatcher(XrdOucString dom, XrdOucString domlist);

    XrdOucString               GetDomainToMatch(XrdOucString hostname);

    ESrvErrorHandlerRetval     HandleServerError(XReqErrorType &, XrdClientMessage *,
                   ClientRequest *);
    bool                       MatchStreamid(struct ServerResponseHeader *ServerResponse);

    // Sends a close request, without waiting for an answer
    // useful (?) to be sent just before closing a badly working stream
    bool                       PanicClose();

    XrdOucString               ParseDomainFromHostname(XrdOucString hostname);

    XrdClientMessage           *ReadPartialAnswer(XReqErrorType &, size_t &, 
                    ClientRequest *, bool, void**,
                    EThreeStateReadHandler &);

//  void                       ClearSessionID();

    XReqErrorType              WriteToServer(ClientRequest *req, 
                    const void* reqMoreData,
                    short LogConnID,
                    int substreamid = 0);

    bool                       WaitResp(int secsmax);

    XrdClientUrlInfo           *fMetaUrl;           // Meta manager url
    bool                        fLBSIsMeta;         // Is current redirector a meta manager?

public:
    XrdOucString               fRedirCGI;           // Same as fRedirOpaque but persistent

};
#endif