XRootD
XrdClXRootDMsgHandler.hh
Go to the documentation of this file.
1 //------------------------------------------------------------------------------
2 // Copyright (c) 2011-2014 by European Organization for Nuclear Research (CERN)
3 // Author: Lukasz Janyst <ljanyst@cern.ch>
4 //------------------------------------------------------------------------------
5 // This file is part of the XRootD software suite.
6 //
7 // XRootD is free software: you can redistribute it and/or modify
8 // it under the terms of the GNU Lesser General Public License as published by
9 // the Free Software Foundation, either version 3 of the License, or
10 // (at your option) any later version.
11 //
12 // XRootD is distributed in the hope that it will be useful,
13 // but WITHOUT ANY WARRANTY; without even the implied warranty of
14 // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
15 // GNU General Public License for more details.
16 //
17 // You should have received a copy of the GNU Lesser General Public License
18 // along with XRootD. If not, see <http://www.gnu.org/licenses/>.
19 //
20 // In applying this licence, CERN does not waive the privileges and immunities
21 // granted to it by virtue of its status as an Intergovernmental Organization
22 // or submit itself to any jurisdiction.
23 //------------------------------------------------------------------------------
24 
25 #ifndef __XRD_CL_XROOTD_MSG_HANDLER_HH__
26 #define __XRD_CL_XROOTD_MSG_HANDLER_HH__
27 
30 #include "XrdCl/XrdClDefaultEnv.hh"
31 #include "XrdCl/XrdClMessage.hh"
32 #include "XProtocol/XProtocol.hh"
33 #include "XrdCl/XrdClLog.hh"
34 #include "XrdCl/XrdClConstants.hh"
35 
41 
42 #include "XrdSys/XrdSysPthread.hh"
43 #include "XrdSys/XrdSysPageSize.hh"
45 #include "XrdSys/XrdSysPlatform.hh"
46 
48 #include "XrdOuc/XrdOucUtils.hh"
49 
50 #include <sys/uio.h>
51 #include <arpa/inet.h> // for network unmarshaling stuff
52 
53 #include <array>
54 #include <list>
55 #include <memory>
56 #include <atomic>
57 #include <memory>
58 
59 namespace XrdCl
60 {
61  class PostMaster;
62  class SIDManager;
63  class URL;
64  class LocalFileHandler;
65  class Socket;
66 
67  //----------------------------------------------------------------------------
68  // Single entry in the redirect-trace-back
69  //----------------------------------------------------------------------------
71  {
72  enum Type
73  {
77  EntryWait
78  };
79 
80  RedirectEntry( const URL &from, const URL &to, Type type ) :
81  from( from ), to( to ), type( type )
82  {
83 
84  }
85 
90 
91  std::string ToString( bool prevok = true )
92  {
93  const std::string tostr = to.GetLocation();
94  const std::string fromstr = from.GetLocation();
95 
96  if( prevok )
97  {
98  switch( type )
99  {
100  case EntryRedirect: return "Redirected from: " + fromstr + " to: "
101  + tostr;
102 
103  case EntryRedirectOnWait: return "Server responded with wait. "
104  "Falling back to virtual redirector: " + tostr;
105 
106  case EntryRetry: return "Retrying: " + tostr;
107 
108  case EntryWait: return "Waited at server request. Resending: "
109  + tostr;
110  }
111  }
112  return "Failed at: " + fromstr + ", retrying at: " + tostr;
113  }
114  };
115 
116  //----------------------------------------------------------------------------
118  //----------------------------------------------------------------------------
120  {
121  friend class HandleRspJob;
122 
123  public:
124  //------------------------------------------------------------------------
133  //------------------------------------------------------------------------
135  ResponseHandler *respHandler,
136  const URL *url,
137  std::shared_ptr<SIDManager> sidMgr,
138  LocalFileHandler *lFileHandler):
139  pRequest( msg ),
140  pResponseHandler( respHandler ),
141  pUrl( *url ),
142  pEffectiveDataServerUrl( 0 ),
143  pSidMgr( sidMgr ),
144  pLFileHandler( lFileHandler ),
145  pExpiration( 0 ),
146  pRedirectAsAnswer( false ),
147  pOksofarAsAnswer( false ),
148  pHasLoadBalancer( false ),
149  pHasSessionId( false ),
150  pChunkList( 0 ),
151  pKBuff( 0 ),
152  pRedirectCounter( 0 ),
153  pNotAuthorizedCounter( 0 ),
154 
155  pAsyncOffset( 0 ),
156  pAsyncChunkIndex( 0 ),
157 
158  pPgWrtCksumBuff( 4 ),
159  pPgWrtCurrentPageOffset( 0 ),
160  pPgWrtCurrentPageNb( 0 ),
161 
162  pOtherRawStarted( false ),
163 
164  pFollowMetalink( false ),
165 
166  pStateful( false ),
167 
168  pAggregatedWaitTime( 0 ),
169 
170  pMsgInFly( false ),
171  pSendingState( 0 ),
172 
173  pTimeoutFence( false ),
174 
175  pDirListStarted( false ),
176  pDirListWithStat( false ),
177 
178  pCV( 0 ),
179 
180  pSslErrCnt( 0 )
181  {
182  pPostMaster = DefaultEnv::GetPostMaster();
183  if( msg->GetSessionId() )
184  pHasSessionId = true;
185 
186  Log *log = DefaultEnv::GetLog();
187  log->Debug( ExDbgMsg, "[%s] MsgHandler created: %p (message: %s ).",
188  pUrl.GetHostId().c_str(), (void*)this,
189  pRequest->GetObfuscatedDescription().c_str() );
190 
191  ClientRequestHdr *hdr = (ClientRequestHdr*)pRequest->GetBuffer();
192  if( ntohs( hdr->requestid ) == kXR_pgread )
193  {
194  ClientPgReadRequest *pgrdreq = (ClientPgReadRequest*)pRequest->GetBuffer();
195  pCrc32cDigests.reserve( XrdOucPgrwUtils::csNum( ntohll( pgrdreq->offset ),
196  ntohl( pgrdreq->rlen ) ) );
197  }
198 
199  if( ntohs( hdr->requestid ) == kXR_readv )
200  pBodyReader.reset( new AsyncVectorReader( *url, *pRequest ) );
201  else if( ntohs( hdr->requestid ) == kXR_read )
202  pBodyReader.reset( new AsyncRawReader( *url, *pRequest ) );
203  else
204  pBodyReader.reset( new AsyncDiscardReader( *url, *pRequest ) );
205  }
206 
207  //------------------------------------------------------------------------
209  //------------------------------------------------------------------------
211  {
212  DumpRedirectTraceBack();
213 
214  if( !pHasSessionId )
215  delete pRequest;
216  delete pEffectiveDataServerUrl;
217 
218  pRequest = reinterpret_cast<Message*>( 0xDEADBEEF );
219  pResponseHandler = reinterpret_cast<ResponseHandler*>( 0xDEADBEEF );
220  pPostMaster = reinterpret_cast<PostMaster*>( 0xDEADBEEF );
221  pLFileHandler = reinterpret_cast<LocalFileHandler*>( 0xDEADBEEF );
222  pChunkList = reinterpret_cast<ChunkList*>( 0xDEADBEEF );
223  pEffectiveDataServerUrl = reinterpret_cast<URL*>( 0xDEADBEEF );
224 
225  Log *log = DefaultEnv::GetLog();
226  log->Debug( ExDbgMsg, "[%s] Destroying MsgHandler: %p.",
227  pUrl.GetHostId().c_str(), (void*)this );
228  }
229 
230  //------------------------------------------------------------------------
236  //------------------------------------------------------------------------
237  virtual uint16_t Examine( std::shared_ptr<Message> &msg ) override;
238 
239  //------------------------------------------------------------------------
248  //------------------------------------------------------------------------
249  virtual uint16_t InspectStatusRsp() override;
250 
251  //------------------------------------------------------------------------
255  //------------------------------------------------------------------------
256  virtual uint16_t GetSid() const override;
257 
258  //------------------------------------------------------------------------
262  //------------------------------------------------------------------------
263  virtual void Process() override;
264 
265  //------------------------------------------------------------------------
275  //------------------------------------------------------------------------
276  virtual XRootDStatus ReadMessageBody( Message *msg,
277  Socket *socket,
278  uint32_t &bytesRead ) override;
279 
280  //------------------------------------------------------------------------
285  //------------------------------------------------------------------------
286  virtual uint8_t OnStreamEvent( StreamEvent event,
287  XRootDStatus status ) override;
288 
289  //------------------------------------------------------------------------
291  //------------------------------------------------------------------------
292  virtual void OnStatusReady( const Message *message,
293  XRootDStatus status ) override;
294 
295  //------------------------------------------------------------------------
297  //------------------------------------------------------------------------
298  virtual bool IsRaw() const override;
299 
300  //------------------------------------------------------------------------
309  //------------------------------------------------------------------------
311  uint32_t &bytesWritten ) override;
312 
313  //------------------------------------------------------------------------
317  //------------------------------------------------------------------------
318  void WaitDone( time_t now );
319 
320  //------------------------------------------------------------------------
322  //------------------------------------------------------------------------
323  void SetExpiration( time_t expiration )
324  {
325  pExpiration = expiration;
326  }
327 
328  //------------------------------------------------------------------------
330  //------------------------------------------------------------------------
331  time_t GetExpiration() override
332  {
333  return pExpiration;
334  }
335 
336  //------------------------------------------------------------------------
339  //------------------------------------------------------------------------
340  void SetRedirectAsAnswer( bool redirectAsAnswer )
341  {
342  pRedirectAsAnswer = redirectAsAnswer;
343  }
344 
345  //------------------------------------------------------------------------
348  //------------------------------------------------------------------------
349  void SetOksofarAsAnswer( bool oksofarAsAnswer )
350  {
351  pOksofarAsAnswer = oksofarAsAnswer;
352  }
353 
354  //------------------------------------------------------------------------
356  //------------------------------------------------------------------------
357  const Message *GetRequest() const
358  {
359  return pRequest;
360  }
361 
362  //------------------------------------------------------------------------
364  //------------------------------------------------------------------------
365  void SetLoadBalancer( const HostInfo &loadBalancer )
366  {
367  if( !loadBalancer.url.IsValid() )
368  return;
369  pLoadBalancer = loadBalancer;
370  pHasLoadBalancer = true;
371  }
372 
373  //------------------------------------------------------------------------
375  //------------------------------------------------------------------------
376  void SetHostList( HostList *hostList )
377  {
378  pHosts.reset( hostList );
379  }
380 
381  //------------------------------------------------------------------------
383  //------------------------------------------------------------------------
384  void SetChunkList( ChunkList *chunkList )
385  {
386  pChunkList = chunkList;
387  if( pBodyReader )
388  pBodyReader->SetChunkList( chunkList );
389  if( chunkList )
390  pChunkStatus.resize( chunkList->size() );
391  else
392  pChunkStatus.clear();
393  }
394 
395  void SetCrc32cDigests( std::vector<uint32_t> && crc32cDigests )
396  {
397  pCrc32cDigests = std::move( crc32cDigests );
398  }
399 
400  //------------------------------------------------------------------------
402  //------------------------------------------------------------------------
404  {
405  pKBuff = kbuff;
406  }
407 
408  //------------------------------------------------------------------------
410  //------------------------------------------------------------------------
411  void SetRedirectCounter( uint16_t redirectCounter )
412  {
413  pRedirectCounter = redirectCounter;
414  }
415 
416  void SetFollowMetalink( bool followMetalink )
417  {
418  pFollowMetalink = followMetalink;
419  }
420 
421  void SetStateful( bool stateful )
422  {
423  pStateful = stateful;
424  }
425 
426  //------------------------------------------------------------------------
430  //------------------------------------------------------------------------
431  void PartialReceived();
432 
433  void OnReadyToSend( [[maybe_unused]] Message *msg ) override
434  {
435  pSendingState |= kSawReadySend;
436  }
437 
438  private:
439 
440  // bit flags used with pSendingState
441  static constexpr int kSendDone = 0x0001;
442  static constexpr int kSawResp = 0x0002;
443  static constexpr int kFinalResp = 0x0004;
444  static constexpr int kSawReadySend = 0x0008;
445 
446  //------------------------------------------------------------------------
448  //------------------------------------------------------------------------
449  void HandleError( XRootDStatus status );
450 
451  //------------------------------------------------------------------------
453  //------------------------------------------------------------------------
454  Status RetryAtServer( const URL &url, RedirectEntry::Type entryType );
455 
456  //------------------------------------------------------------------------
458  //------------------------------------------------------------------------
459  void HandleResponse();
460 
461  //------------------------------------------------------------------------
463  //------------------------------------------------------------------------
464  XRootDStatus *ProcessStatus();
465 
466  //------------------------------------------------------------------------
469  //------------------------------------------------------------------------
470  Status ParseResponse( AnyObject *&response );
471 
472  //------------------------------------------------------------------------
475  //------------------------------------------------------------------------
476  Status ParseXAttrResponse( char *data, size_t len, AnyObject *&response );
477 
478  //------------------------------------------------------------------------
481  //------------------------------------------------------------------------
482  Status RewriteRequestRedirect( const URL &newUrl );
483 
484  //------------------------------------------------------------------------
486  //------------------------------------------------------------------------
487  Status RewriteRequestWait();
488 
489  //------------------------------------------------------------------------
491  //------------------------------------------------------------------------
492  void UpdateTriedCGI(uint32_t errNo=0);
493 
494  //------------------------------------------------------------------------
496  //------------------------------------------------------------------------
497  void SwitchOnRefreshFlag();
498 
499  //------------------------------------------------------------------------
502  //------------------------------------------------------------------------
503  void HandleRspOrQueue();
504 
505  //------------------------------------------------------------------------
507  //------------------------------------------------------------------------
508  void HandleLocalRedirect( URL *url );
509 
510  //------------------------------------------------------------------------
515  //------------------------------------------------------------------------
516  bool IsRetriable();
517 
518  //------------------------------------------------------------------------
525  //------------------------------------------------------------------------
526  bool OmitWait( Message &request, const URL &url );
527 
528  //------------------------------------------------------------------------
534  //------------------------------------------------------------------------
535  bool RetriableErrorResponse( const Status &status );
536 
537  //------------------------------------------------------------------------
539  //------------------------------------------------------------------------
540  void DumpRedirectTraceBack();
541 
548  //------------------------------------------------------------------------
549  template<typename T>
550  Status ReadFromBuffer( char *&buffer, size_t &buflen, T& result );
551 
552  //------------------------------------------------------------------------
559  //------------------------------------------------------------------------
560  Status ReadFromBuffer( char *&buffer, size_t &buflen, std::string &result );
561 
562  //------------------------------------------------------------------------
570  //------------------------------------------------------------------------
571  Status ReadFromBuffer( char *&buffer, size_t &buflen, size_t size,
572  std::string &result );
573 
574  //------------------------------------------------------------------------
575  // Helper struct for async reading of chunks
576  //------------------------------------------------------------------------
577  struct ChunkStatus
578  {
579  ChunkStatus(): sizeError( false ), done( false ) {}
580  bool sizeError;
581  bool done;
582  };
583 
584  typedef std::list<std::unique_ptr<RedirectEntry>> RedirectTraceBack;
585 
586  static const size_t CksumSize = sizeof( uint32_t );
587  static const size_t PageWithCksum = XrdSys::PageSize + CksumSize;
588  static const size_t MaxSslErrRetry = 3;
589 
590  inline static size_t NbPgPerRsp( uint64_t offset, uint32_t dlen )
591  {
592  uint32_t pgcnt = 0;
593  uint32_t remainder = offset % XrdSys::PageSize;
594  if( remainder > 0 )
595  {
596  // account for the first unaligned page
597  ++pgcnt;
598  // the size of the 1st unaligned page
599  uint32_t _1stpg = XrdSys::PageSize - remainder;
600  if( _1stpg + CksumSize > dlen )
601  _1stpg = dlen - CksumSize;
602  dlen -= _1stpg + CksumSize;
603  }
604  pgcnt += dlen / PageWithCksum;
605  if( dlen % PageWithCksum )
606  ++ pgcnt;
607  return pgcnt;
608  }
609 
610  Message *pRequest;
611  std::shared_ptr<Message> pResponse; //< the ownership is shared with MsgReader
612  std::vector<std::shared_ptr<Message>> pPartialResps; //< the ownership is shared with MsgReader
613  ResponseHandler *pResponseHandler;
614  URL pUrl;
615  URL *pEffectiveDataServerUrl;
616  PostMaster *pPostMaster;
617  std::shared_ptr<SIDManager> pSidMgr;
618  LocalFileHandler *pLFileHandler;
619  XRootDStatus pStatus;
620  Status pLastError;
621  time_t pExpiration;
622  bool pRedirectAsAnswer;
623  bool pOksofarAsAnswer;
624  std::unique_ptr<HostList> pHosts;
625  bool pHasLoadBalancer;
626  HostInfo pLoadBalancer;
627  bool pHasSessionId;
628  std::string pRedirectUrl;
629  ChunkList *pChunkList;
630  std::vector<uint32_t> pCrc32cDigests;
631  XrdSys::KernelBuffer *pKBuff;
632  std::vector<ChunkStatus> pChunkStatus;
633  uint16_t pRedirectCounter;
634  uint16_t pNotAuthorizedCounter;
635 
636  uint32_t pAsyncOffset;
637  uint32_t pAsyncChunkIndex;
638 
639  std::unique_ptr<AsyncPageReader> pPageReader;
640  std::unique_ptr<AsyncRawReaderIntfc> pBodyReader;
641 
642  Buffer pPgWrtCksumBuff;
643  uint32_t pPgWrtCurrentPageOffset;
644  uint32_t pPgWrtCurrentPageNb;
645 
646  bool pOtherRawStarted;
647 
648  bool pFollowMetalink;
649 
650  bool pStateful;
651  int pAggregatedWaitTime;
652 
653  std::unique_ptr<RedirectEntry> pRdirEntry;
654  RedirectTraceBack pRedirectTraceBack;
655 
656  bool pMsgInFly;
657  std::atomic<int> pSendingState;
658 
659  //------------------------------------------------------------------------
660  // true if MsgHandler is both in inQueue and installed in respective
661  // Stream (this could happen if server gave oksofar response), otherwise
662  // false
663  //------------------------------------------------------------------------
664  std::atomic<bool> pTimeoutFence;
665 
666  //------------------------------------------------------------------------
667  // if we are serving chunked data to the user's handler in case of
668  // kXR_dirlist we need to memorize if the response contains stat info or
669  // not (the information is only encoded in the first chunk)
670  //------------------------------------------------------------------------
671  bool pDirListStarted;
672  bool pDirListWithStat;
673 
674  //------------------------------------------------------------------------
675  // synchronization is needed in case the MsgHandler has been configured
676  // to serve kXR_oksofar as a response to the user's handler
677  //------------------------------------------------------------------------
678  XrdSysCondVar pCV;
679 
680  //------------------------------------------------------------------------
681  // Count of consecutive `errTlsSslError` errors
682  //------------------------------------------------------------------------
683  size_t pSslErrCnt;
684  };
685 }
686 
687 #endif // __XRD_CL_XROOTD_MSG_HANDLER_HH__
kXR_unt16 requestid
Definition: XProtocol.hh:157
@ kXR_read
Definition: XProtocol.hh:125
@ kXR_readv
Definition: XProtocol.hh:137
@ kXR_pgread
Definition: XProtocol.hh:142
Object for discarding data.
Object for reading out data from the kXR_read response.
Object for reading out data from the VectorRead response.
const char * GetBuffer(uint32_t offset=0) const
Get the message buffer.
Definition: XrdClBuffer.hh:72
static Log * GetLog()
Get default log.
static PostMaster * GetPostMaster()
Get default post master.
Handle diagnostics.
Definition: XrdClLog.hh:101
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
Definition: XrdClLog.cc:282
The message representation used throughout the system.
Definition: XrdClMessage.hh:32
const std::string & GetObfuscatedDescription() const
Get the description of the message with authz parameter obfuscated.
uint64_t GetSessionId() const
Get the session ID the message is meant for.
StreamEvent
Events that may have occurred to the stream.
A hub for dispatching and receiving messages.
Handle an async response.
A network socket.
Definition: XrdClSocket.hh:43
URL representation.
Definition: XrdClURL.hh:31
std::string GetHostId() const
Get the host part of the URL (user:password@host:port)
Definition: XrdClURL.hh:99
std::string GetLocation() const
Get location (protocol://host:port/path)
Definition: XrdClURL.cc:344
bool IsValid() const
Is the url valid.
Definition: XrdClURL.cc:452
Handle/Process/Forward XRootD messages.
void SetRedirectCounter(uint16_t redirectCounter)
Set the redirect counter.
void SetFollowMetalink(bool followMetalink)
const Message * GetRequest() const
Get the request pointer.
void SetChunkList(ChunkList *chunkList)
Set the chunk list.
void SetHostList(HostList *hostList)
Set host list.
virtual uint16_t InspectStatusRsp() override
virtual void OnStatusReady(const Message *message, XRootDStatus status) override
The requested action has been performed and the status is available.
void SetCrc32cDigests(std::vector< uint32_t > &&crc32cDigests)
void SetLoadBalancer(const HostInfo &loadBalancer)
Set the load balancer.
virtual uint16_t Examine(std::shared_ptr< Message > &msg) override
void OnReadyToSend([[maybe_unused]] Message *msg) override
XRootDMsgHandler(Message *msg, ResponseHandler *respHandler, const URL *url, std::shared_ptr< SIDManager > sidMgr, LocalFileHandler *lFileHandler)
virtual void Process() override
Process the message if it was "taken" by the examine action.
virtual XRootDStatus ReadMessageBody(Message *msg, Socket *socket, uint32_t &bytesRead) override
void SetStateful(bool stateful)
void SetOksofarAsAnswer(bool oksofarAsAnswer)
time_t GetExpiration() override
Get a timestamp after which we give up.
XRootDStatus WriteMessageBody(Socket *socket, uint32_t &bytesWritten) override
void SetKernelBuffer(XrdSys::KernelBuffer *kbuff)
Set the kernel buffer.
virtual uint8_t OnStreamEvent(StreamEvent event, XRootDStatus status) override
virtual uint16_t GetSid() const override
void SetExpiration(time_t expiration)
Set a timestamp after which we give up.
virtual bool IsRaw() const override
Are we a raw writer or not?
void SetRedirectAsAnswer(bool redirectAsAnswer)
static int csNum(off_t offs, int count)
Compute the required size of a checksum vector based on offset & length.
std::vector< HostInfo > HostList
const uint64_t ExDbgMsg
std::vector< ChunkInfo > ChunkList
List of chunks.
static const int PageSize
URL url
URL of the host.
RedirectEntry(const URL &from, const URL &to, Type type)
std::string ToString(bool prevok=true)
Procedure execution status.
Definition: XrdClStatus.hh:115