XRootD
Loading...
Searching...
No Matches
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
31#include "XrdCl/XrdClMessage.hh"
33#include "XrdCl/XrdClLog.hh"
35
41
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
59namespace 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 {
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 //------------------------------------------------------------------------
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.
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.
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.
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)
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)
const Message * GetRequest() const
Get the request pointer.
void SetLoadBalancer(const HostInfo &loadBalancer)
Set the load balancer.
virtual uint16_t Examine(std::shared_ptr< Message > &msg) override
void OnReadyToSend(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 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.