XrdClXRootDMsgHandler.hh

Go to the documentation of this file.

//------------------------------------------------------------------------------
// Copyright (c) 2011-2014 by European Organization for Nuclear Research (CERN)
// Author: Lukasz Janyst <ljanyst@cern.ch>
//------------------------------------------------------------------------------
// 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 General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD.  If not, see <http://www.gnu.org/licenses/>.
//
// In applying this licence, CERN does not waive the privileges and immunities
// granted to it by virtue of its status as an Intergovernmental Organization
// or submit itself to any jurisdiction.
//------------------------------------------------------------------------------
 
#ifndef __XRD_CL_XROOTD_MSG_HANDLER_HH__
#define __XRD_CL_XROOTD_MSG_HANDLER_HH__
 
#include "XrdCl/XrdClPostMasterInterfaces.hh"
#include "XrdCl/XrdClXRootDResponses.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClMessage.hh"
#include "XProtocol/XProtocol.hh"
#include "XrdCl/XrdClLog.hh"
#include "XrdCl/XrdClConstants.hh"
 
#include "XrdCl/XrdClAsyncPageReader.hh"
#include "XrdCl/XrdClAsyncVectorReader.hh"
#include "XrdCl/XrdClAsyncRawReader.hh"
#include "XrdCl/XrdClAsyncDiscardReader.hh"
#include "XrdClAsyncRawReaderIntfc.hh"
 
#include "XrdSys/XrdSysPthread.hh"
#include "XrdSys/XrdSysPageSize.hh"
#include "XrdSys/XrdSysKernelBuffer.hh"
#include "XrdSys/XrdSysPlatform.hh"
 
#include "XrdOuc/XrdOucPgrwUtils.hh"
 
#include <sys/uio.h>
#include <arpa/inet.h> // for network unmarshaling stuff
 
#include <array>
#include <list>
#include <memory>
#include <atomic>
#include <memory>
 
namespace XrdCl
{
  class PostMaster;
  class SIDManager;
  class URL;
  class LocalFileHandler;
  class Socket;
 
  //----------------------------------------------------------------------------
  // Single entry in the redirect-trace-back
  //----------------------------------------------------------------------------
  struct RedirectEntry
  {
      enum Type
      {
        EntryRedirect,
        EntryRedirectOnWait,
        EntryRetry,
        EntryWait
      };
 
      RedirectEntry( const URL &from, const URL &to, Type type ) :
        from( from ), to( to ), type( type )
      {
 
      }
 
      URL          from;
      URL          to;
      Type         type;
      XRootDStatus status;
 
      std::string ToString( bool prevok = true )
      {
        const std::string tostr   = to.GetLocation();
        const std::string fromstr = from.GetLocation();
 
        if( prevok )
        {
          switch( type )
          {
            case EntryRedirect: return "Redirected from: " + fromstr + " to: "
                + tostr;
 
            case EntryRedirectOnWait: return "Server responded with wait. "
                "Falling back to virtual redirector: " + tostr;
 
            case EntryRetry: return "Retrying: " + tostr;
 
            case EntryWait: return "Waited at server request. Resending: "
                + tostr;
          }
        }
        return "Failed at: " + fromstr + ", retrying at: " + tostr;
      }
  };
 
  //----------------------------------------------------------------------------
  //----------------------------------------------------------------------------
  class XRootDMsgHandler: public MsgHandler
  {
      friend class HandleRspJob;
 
    public:
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      XRootDMsgHandler( Message                     *msg,
                        ResponseHandler             *respHandler,
                        const URL                   *url,
                        std::shared_ptr<SIDManager>  sidMgr,
                        LocalFileHandler            *lFileHandler):
        pRequest( msg ),
        pResponseHandler( respHandler ),
        pUrl( *url ),
        pEffectiveDataServerUrl( 0 ),
        pSidMgr( sidMgr ),
        pLFileHandler( lFileHandler ),
        pExpiration( 0 ),
        pRedirectAsAnswer( false ),
        pOksofarAsAnswer( false ),
        pHasLoadBalancer( false ),
        pHasSessionId( false ),
        pChunkList( 0 ),
        pKBuff( 0 ),
        pRedirectCounter( 0 ),
        pNotAuthorizedCounter( 0 ),
 
        pAsyncOffset( 0 ),
        pAsyncChunkIndex( 0 ),
 
        pPgWrtCksumBuff( 4 ),
        pPgWrtCurrentPageOffset( 0 ),
        pPgWrtCurrentPageNb( 0 ),
 
        pOtherRawStarted( false ),
 
        pFollowMetalink( false ),
 
        pStateful( false ),
 
        pAggregatedWaitTime( 0 ),
 
        pMsgInFly( false ),
 
        pTimeoutFence( false ),
 
        pDirListStarted( false ),
        pDirListWithStat( false ),
 
        pCV( 0 ),
 
        pSslErrCnt( 0 )
      {
        pPostMaster = DefaultEnv::GetPostMaster();
        if( msg->GetSessionId() )
          pHasSessionId = true;
 
        Log *log = DefaultEnv::GetLog();
        log->Debug( ExDbgMsg, "[%s] MsgHandler created: 0x%x (message: %s ).",
                    pUrl.GetHostId().c_str(), this,
                    pRequest->GetDescription().c_str() );
 
        ClientRequestHdr *hdr = (ClientRequestHdr*)pRequest->GetBuffer();
        if( ntohs( hdr->requestid ) == kXR_pgread )
        {
          ClientPgReadRequest *pgrdreq = (ClientPgReadRequest*)pRequest->GetBuffer();
          pCrc32cDigests.reserve( XrdOucPgrwUtils::csNum( ntohll( pgrdreq->offset ),
                                                         ntohl( pgrdreq->rlen ) ) );
        }
 
        if( ntohs( hdr->requestid ) == kXR_readv )
          pBodyReader.reset( new AsyncVectorReader( *url, *pRequest ) );
        else if( ntohs( hdr->requestid ) == kXR_read )
          pBodyReader.reset( new AsyncRawReader( *url, *pRequest ) );
        else
          pBodyReader.reset( new AsyncDiscardReader( *url, *pRequest ) );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      ~XRootDMsgHandler()
      {
        DumpRedirectTraceBack();
 
        if( !pHasSessionId )
          delete pRequest;
        delete pEffectiveDataServerUrl;
 
        pRequest                = reinterpret_cast<Message*>( 0xDEADBEEF );
        pResponseHandler        = reinterpret_cast<ResponseHandler*>( 0xDEADBEEF );
        pPostMaster             = reinterpret_cast<PostMaster*>( 0xDEADBEEF );
        pLFileHandler           = reinterpret_cast<LocalFileHandler*>( 0xDEADBEEF );
        pChunkList              = reinterpret_cast<ChunkList*>( 0xDEADBEEF );
        pEffectiveDataServerUrl = reinterpret_cast<URL*>( 0xDEADBEEF );
 
        Log *log = DefaultEnv::GetLog();
        log->Debug( ExDbgMsg, "[%s] Destroying MsgHandler: 0x%x.",
                    pUrl.GetHostId().c_str(), this );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual uint16_t Examine( std::shared_ptr<Message> &msg  );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual uint16_t InspectStatusRsp();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual uint16_t GetSid() const;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual void Process();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual XRootDStatus ReadMessageBody( Message  *msg,
                                            Socket   *socket,
                                            uint32_t &bytesRead );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual uint8_t OnStreamEvent( StreamEvent  event,
                                     XRootDStatus status );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual void OnStatusReady( const Message *message,
                                  XRootDStatus   status );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      virtual bool IsRaw() const;
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      XRootDStatus WriteMessageBody( Socket   *socket,
                                     uint32_t &bytesWritten );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void WaitDone( time_t now );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void SetExpiration( time_t expiration )
      {
        pExpiration = expiration;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      time_t GetExpiration()
      {
        return pExpiration;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void SetRedirectAsAnswer( bool redirectAsAnswer )
      {
        pRedirectAsAnswer = redirectAsAnswer;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void SetOksofarAsAnswer( bool oksofarAsAnswer )
      {
        pOksofarAsAnswer = oksofarAsAnswer;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      const Message *GetRequest() const
      {
        return pRequest;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void SetLoadBalancer( const HostInfo &loadBalancer )
      {
        if( !loadBalancer.url.IsValid() )
          return;
        pLoadBalancer    = loadBalancer;
        pHasLoadBalancer = true;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void SetHostList( HostList *hostList )
      {
        pHosts.reset( hostList );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void SetChunkList( ChunkList *chunkList )
      {
        pChunkList = chunkList;
        if( pBodyReader )
          pBodyReader->SetChunkList( chunkList );
        if( chunkList )
          pChunkStatus.resize( chunkList->size() );
        else
          pChunkStatus.clear();
      }
 
      void SetCrc32cDigests( std::vector<uint32_t> && crc32cDigests )
      {
        pCrc32cDigests = std::move( crc32cDigests );
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void SetKernelBuffer( XrdSys::KernelBuffer *kbuff )
      {
        pKBuff = kbuff;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void SetRedirectCounter( uint16_t redirectCounter )
      {
        pRedirectCounter = redirectCounter;
      }
 
      void SetFollowMetalink( bool followMetalink )
      {
        pFollowMetalink = followMetalink;
      }
 
      void SetStateful( bool stateful )
      {
        pStateful = stateful;
      }
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void PartialReceived();
 
    private:
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void HandleError( XRootDStatus status );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Status RetryAtServer( const URL &url, RedirectEntry::Type entryType );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void HandleResponse();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      XRootDStatus *ProcessStatus();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Status ParseResponse( AnyObject *&response );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Status ParseXAttrResponse( char *data, size_t len, AnyObject *&response );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Status RewriteRequestRedirect( const URL &newUrl );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Status RewriteRequestWait();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void UpdateTriedCGI(uint32_t errNo=0);
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void SwitchOnRefreshFlag();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void HandleRspOrQueue();
 
      //------------------------------------------------------------------------ 
      //------------------------------------------------------------------------
      void HandleLocalRedirect( URL *url );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool IsRetriable();
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool OmitWait( Message &request, const URL &url );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      bool RetriableErrorResponse( const Status &status );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      void DumpRedirectTraceBack();
      
      //------------------------------------------------------------------------
      template<typename T>
      Status ReadFromBuffer( char *&buffer, size_t &buflen, T& result );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Status ReadFromBuffer( char *&buffer, size_t &buflen, std::string &result );
 
      //------------------------------------------------------------------------
      //------------------------------------------------------------------------
      Status ReadFromBuffer( char *&buffer, size_t &buflen, size_t size,
                             std::string &result );
 
      //------------------------------------------------------------------------
      // Helper struct for async reading of chunks
      //------------------------------------------------------------------------
      struct ChunkStatus
      {
        ChunkStatus(): sizeError( false ), done( false ) {}
        bool sizeError;
        bool done;
      };
 
      typedef std::list<std::unique_ptr<RedirectEntry>> RedirectTraceBack;
 
      static const size_t             CksumSize      = sizeof( uint32_t );
      static const size_t             PageWithCksum  = XrdSys::PageSize + CksumSize;
      static const size_t             MaxSslErrRetry = 3;
 
      inline static size_t NbPgPerRsp( uint64_t offset, uint32_t dlen )
      {
        uint32_t pgcnt = 0;
        uint32_t remainder = offset % XrdSys::PageSize;
        if( remainder > 0 )
        {
          // account for the first unaligned page
          ++pgcnt;
          // the size of the 1st unaligned page
          uint32_t _1stpg = XrdSys::PageSize - remainder;
          if( _1stpg + CksumSize > dlen )
            _1stpg = dlen - CksumSize;
          dlen -= _1stpg + CksumSize;
        }
        pgcnt += dlen / PageWithCksum;
        if( dlen % PageWithCksum )
          ++ pgcnt;
        return pgcnt;
      }
 
      Message                               *pRequest;
      std::shared_ptr<Message>               pResponse; //< the ownership is shared with MsgReader
      std::vector<std::shared_ptr<Message>>  pPartialResps; //< the ownership is shared with MsgReader
      ResponseHandler                       *pResponseHandler;
      URL                                    pUrl;
      URL                                   *pEffectiveDataServerUrl;
      PostMaster                            *pPostMaster;
      std::shared_ptr<SIDManager>            pSidMgr;
      LocalFileHandler                      *pLFileHandler;
      XRootDStatus                           pStatus;
      Status                                 pLastError;
      time_t                                 pExpiration;
      bool                                   pRedirectAsAnswer;
      bool                                   pOksofarAsAnswer;
      std::unique_ptr<HostList>              pHosts;
      bool                                   pHasLoadBalancer;
      HostInfo                               pLoadBalancer;
      bool                                   pHasSessionId;
      std::string                            pRedirectUrl;
      ChunkList                             *pChunkList;
      std::vector<uint32_t>                  pCrc32cDigests;
      XrdSys::KernelBuffer                  *pKBuff;
      std::vector<ChunkStatus>               pChunkStatus;
      uint16_t                               pRedirectCounter;
      uint16_t                               pNotAuthorizedCounter;
 
      uint32_t                               pAsyncOffset;
      uint32_t                               pAsyncChunkIndex;
 
      std::unique_ptr<AsyncPageReader>       pPageReader;
      std::unique_ptr<AsyncRawReaderIntfc>   pBodyReader;
 
      Buffer                                 pPgWrtCksumBuff;
      uint32_t                               pPgWrtCurrentPageOffset;
      uint32_t                               pPgWrtCurrentPageNb;
 
      bool                                   pOtherRawStarted;
 
      bool                                   pFollowMetalink;
 
      bool                                   pStateful;
      int                                    pAggregatedWaitTime;
 
      std::unique_ptr<RedirectEntry>         pRdirEntry;
      RedirectTraceBack                      pRedirectTraceBack;
 
      bool                                   pMsgInFly;
 
      //------------------------------------------------------------------------
      // true if MsgHandler is both in inQueue and installed in respective
      // Stream (this could happen if server gave oksofar response), otherwise
      // false
      //------------------------------------------------------------------------
      std::atomic<bool>                      pTimeoutFence;
 
      //------------------------------------------------------------------------
      // if we are serving chunked data to the user's handler in case of
      // kXR_dirlist we need to memorize if the response contains stat info or
      // not (the information is only encoded in the first chunk)
      //------------------------------------------------------------------------
      bool                                   pDirListStarted;
      bool                                   pDirListWithStat;
 
      //------------------------------------------------------------------------
      // synchronization is needed in case the MsgHandler has been configured
      // to serve kXR_oksofar as a response to the user's handler
      //------------------------------------------------------------------------
      XrdSysCondVar                          pCV;
 
      //------------------------------------------------------------------------
      // Count of consecutive `errTlsSslError` errors
      //------------------------------------------------------------------------
      size_t                                 pSslErrCnt;
  };
}
 
#endif // __XRD_CL_XROOTD_MSG_HANDLER_HH__