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(), (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 //------------------------------------------------------------------------
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.
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.
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.
Handle XRootD stream IDs.
A network socket.
URL representation.
Definition XrdClURL.hh:31
bool IsValid() const
Is the url valid.
Definition XrdClURL.cc:452
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.