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(), 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(), 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 = 0;
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 
445  //------------------------------------------------------------------------
447  //------------------------------------------------------------------------
448  void HandleError( XRootDStatus status );
449 
450  //------------------------------------------------------------------------
452  //------------------------------------------------------------------------
453  Status RetryAtServer( const URL &url, RedirectEntry::Type entryType );
454 
455  //------------------------------------------------------------------------
457  //------------------------------------------------------------------------
458  void HandleResponse();
459 
460  //------------------------------------------------------------------------
462  //------------------------------------------------------------------------
463  XRootDStatus *ProcessStatus();
464 
465  //------------------------------------------------------------------------
468  //------------------------------------------------------------------------
469  Status ParseResponse( AnyObject *&response );
470 
471  //------------------------------------------------------------------------
474  //------------------------------------------------------------------------
475  Status ParseXAttrResponse( char *data, size_t len, AnyObject *&response );
476 
477  //------------------------------------------------------------------------
480  //------------------------------------------------------------------------
481  Status RewriteRequestRedirect( const URL &newUrl );
482 
483  //------------------------------------------------------------------------
485  //------------------------------------------------------------------------
486  Status RewriteRequestWait();
487 
488  //------------------------------------------------------------------------
490  //------------------------------------------------------------------------
491  void UpdateTriedCGI(uint32_t errNo=0);
492 
493  //------------------------------------------------------------------------
495  //------------------------------------------------------------------------
496  void SwitchOnRefreshFlag();
497 
498  //------------------------------------------------------------------------
501  //------------------------------------------------------------------------
502  void HandleRspOrQueue();
503 
504  //------------------------------------------------------------------------
506  //------------------------------------------------------------------------
507  void HandleLocalRedirect( URL *url );
508 
509  //------------------------------------------------------------------------
514  //------------------------------------------------------------------------
515  bool IsRetriable();
516 
517  //------------------------------------------------------------------------
524  //------------------------------------------------------------------------
525  bool OmitWait( Message &request, const URL &url );
526 
527  //------------------------------------------------------------------------
533  //------------------------------------------------------------------------
534  bool RetriableErrorResponse( const Status &status );
535 
536  //------------------------------------------------------------------------
538  //------------------------------------------------------------------------
539  void DumpRedirectTraceBack();
540 
547  //------------------------------------------------------------------------
548  template<typename T>
549  Status ReadFromBuffer( char *&buffer, size_t &buflen, T& result );
550 
551  //------------------------------------------------------------------------
558  //------------------------------------------------------------------------
559  Status ReadFromBuffer( char *&buffer, size_t &buflen, std::string &result );
560 
561  //------------------------------------------------------------------------
569  //------------------------------------------------------------------------
570  Status ReadFromBuffer( char *&buffer, size_t &buflen, size_t size,
571  std::string &result );
572 
573  //------------------------------------------------------------------------
574  // Helper struct for async reading of chunks
575  //------------------------------------------------------------------------
576  struct ChunkStatus
577  {
578  ChunkStatus(): sizeError( false ), done( false ) {}
579  bool sizeError;
580  bool done;
581  };
582 
583  typedef std::list<std::unique_ptr<RedirectEntry>> RedirectTraceBack;
584 
585  static const size_t CksumSize = sizeof( uint32_t );
586  static const size_t PageWithCksum = XrdSys::PageSize + CksumSize;
587  static const size_t MaxSslErrRetry = 3;
588 
589  inline static size_t NbPgPerRsp( uint64_t offset, uint32_t dlen )
590  {
591  uint32_t pgcnt = 0;
592  uint32_t remainder = offset % XrdSys::PageSize;
593  if( remainder > 0 )
594  {
595  // account for the first unaligned page
596  ++pgcnt;
597  // the size of the 1st unaligned page
598  uint32_t _1stpg = XrdSys::PageSize - remainder;
599  if( _1stpg + CksumSize > dlen )
600  _1stpg = dlen - CksumSize;
601  dlen -= _1stpg + CksumSize;
602  }
603  pgcnt += dlen / PageWithCksum;
604  if( dlen % PageWithCksum )
605  ++ pgcnt;
606  return pgcnt;
607  }
608 
609  Message *pRequest;
610  std::shared_ptr<Message> pResponse; //< the ownership is shared with MsgReader
611  std::vector<std::shared_ptr<Message>> pPartialResps; //< the ownership is shared with MsgReader
612  ResponseHandler *pResponseHandler;
613  URL pUrl;
614  URL *pEffectiveDataServerUrl;
615  PostMaster *pPostMaster;
616  std::shared_ptr<SIDManager> pSidMgr;
617  LocalFileHandler *pLFileHandler;
618  XRootDStatus pStatus;
619  Status pLastError;
620  time_t pExpiration;
621  bool pRedirectAsAnswer;
622  bool pOksofarAsAnswer;
623  std::unique_ptr<HostList> pHosts;
624  bool pHasLoadBalancer;
625  HostInfo pLoadBalancer;
626  bool pHasSessionId;
627  std::string pRedirectUrl;
628  ChunkList *pChunkList;
629  std::vector<uint32_t> pCrc32cDigests;
630  XrdSys::KernelBuffer *pKBuff;
631  std::vector<ChunkStatus> pChunkStatus;
632  uint16_t pRedirectCounter;
633  uint16_t pNotAuthorizedCounter;
634 
635  uint32_t pAsyncOffset;
636  uint32_t pAsyncChunkIndex;
637 
638  std::unique_ptr<AsyncPageReader> pPageReader;
639  std::unique_ptr<AsyncRawReaderIntfc> pBodyReader;
640 
641  Buffer pPgWrtCksumBuff;
642  uint32_t pPgWrtCurrentPageOffset;
643  uint32_t pPgWrtCurrentPageNb;
644 
645  bool pOtherRawStarted;
646 
647  bool pFollowMetalink;
648 
649  bool pStateful;
650  int pAggregatedWaitTime;
651 
652  std::unique_ptr<RedirectEntry> pRdirEntry;
653  RedirectTraceBack pRedirectTraceBack;
654 
655  bool pMsgInFly;
656  std::atomic<int> pSendingState;
657 
658  //------------------------------------------------------------------------
659  // true if MsgHandler is both in inQueue and installed in respective
660  // Stream (this could happen if server gave oksofar response), otherwise
661  // false
662  //------------------------------------------------------------------------
663  std::atomic<bool> pTimeoutFence;
664 
665  //------------------------------------------------------------------------
666  // if we are serving chunked data to the user's handler in case of
667  // kXR_dirlist we need to memorize if the response contains stat info or
668  // not (the information is only encoded in the first chunk)
669  //------------------------------------------------------------------------
670  bool pDirListStarted;
671  bool pDirListWithStat;
672 
673  //------------------------------------------------------------------------
674  // synchronization is needed in case the MsgHandler has been configured
675  // to serve kXR_oksofar as a response to the user's handler
676  //------------------------------------------------------------------------
677  XrdSysCondVar pCV;
678 
679  //------------------------------------------------------------------------
680  // Count of consecutive `errTlsSslError` errors
681  //------------------------------------------------------------------------
682  size_t pSslErrCnt;
683  };
684 }
685 
686 #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