XRootD
Loading...
Searching...
No Matches
XrdClXRootDMsgHandler.cc
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
26#include "XrdCl/XrdClLog.hh"
30#include "XrdCl/XrdClMessage.hh"
31#include "XrdCl/XrdClURL.hh"
32#include "XrdCl/XrdClUtils.hh"
39#include "XrdCl/XrdClSocket.hh"
40#include "XrdCl/XrdClTls.hh"
42
43#include "XrdOuc/XrdOucCRC.hh"
45
46#include "XrdSys/XrdSysPlatform.hh" // same as above
49#include <memory>
50#include <sstream>
51#include <numeric>
52
53namespace
54{
55 //----------------------------------------------------------------------------
56 // We need an extra task what will run the handler in the future, because
57 // tasks get deleted and we need the handler
58 //----------------------------------------------------------------------------
59 class WaitTask: public XrdCl::Task
60 {
61 public:
62 WaitTask( XrdCl::XRootDMsgHandler *handler ): pHandler( handler )
63 {
64 std::ostringstream o;
65 o << "WaitTask for: 0x" << handler->GetRequest();
66 SetName( o.str() );
67 }
68
69 virtual time_t Run( time_t now )
70 {
71 pHandler->WaitDone( now );
72 return 0;
73 }
74 private:
76 };
77}
78
79namespace XrdCl
80{
81 //----------------------------------------------------------------------------
82 // Delegate the response handling to the thread-pool
83 //----------------------------------------------------------------------------
84 class HandleRspJob: public XrdCl::Job
85 {
86 public:
87 HandleRspJob( XrdCl::XRootDMsgHandler *handler ): pHandler( handler )
88 {
89
90 }
91
92 virtual ~HandleRspJob()
93 {
94
95 }
96
97 virtual void Run( void *arg )
98 {
99 pHandler->HandleResponse();
100 delete this;
101 }
102 private:
103 XrdCl::XRootDMsgHandler *pHandler;
104 };
105
106 //----------------------------------------------------------------------------
107 // Examine an incoming message, and decide on the action to be taken
108 //----------------------------------------------------------------------------
109 uint16_t XRootDMsgHandler::Examine( std::shared_ptr<Message> &msg )
110 {
111 const int sst = pSendingState.fetch_or( kSawResp );
112
113 if( !( sst & kSendDone ) && !( sst & kSawResp ) )
114 {
115 // we must have been sent although we haven't got the OnStatusReady
116 // notification yet. Set the inflight notice.
117
118 Log *log = DefaultEnv::GetLog();
119 log->Dump( XRootDMsg, "[%s] Message %s reply received before notification "
120 "that it was sent, assuming it was sent ok.",
121 pUrl.GetHostId().c_str(),
122 pRequest->GetObfuscatedDescription().c_str() );
123
124 pMsgInFly = true;
125 }
126
127 //--------------------------------------------------------------------------
128 // if the MsgHandler is already being used to process another request
129 // (kXR_oksofar) we need to wait
130 //--------------------------------------------------------------------------
131 if( pOksofarAsAnswer )
132 {
133 XrdSysCondVarHelper lck( pCV );
134 while( pResponse ) pCV.Wait();
135 }
136 else
137 {
138 if( pResponse )
139 {
140 Log *log = DefaultEnv::GetLog();
141 log->Warning( ExDbgMsg, "[%s] MsgHandler is examining a response although "
142 "it already owns a response: %p (message: %s ).",
143 pUrl.GetHostId().c_str(), this,
144 pRequest->GetObfuscatedDescription().c_str() );
145 }
146 }
147
148 if( msg->GetSize() < 8 )
149 return Ignore;
150
151 ServerResponse *rsp = (ServerResponse *)msg->GetBuffer();
152 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
153 uint16_t status = 0;
154 uint32_t dlen = 0;
155
156 //--------------------------------------------------------------------------
157 // We only care about async responses, but those are extracted now
158 // in the SocketHandler.
159 //--------------------------------------------------------------------------
160 if( rsp->hdr.status == kXR_attn )
161 {
162 return Ignore;
163 }
164 //--------------------------------------------------------------------------
165 // We got a sync message - check if it belongs to us
166 //--------------------------------------------------------------------------
167 else
168 {
169 if( rsp->hdr.streamid[0] != req->header.streamid[0] ||
170 rsp->hdr.streamid[1] != req->header.streamid[1] )
171 return Ignore;
172
173 status = rsp->hdr.status;
174 dlen = rsp->hdr.dlen;
175 }
176
177 //--------------------------------------------------------------------------
178 // We take the ownership of the message and decide what we will do
179 // with the handler itself, the options are:
180 // 1) we want to either read in raw mode (the Raw flag) or have the message
181 // body reconstructed for us by the TransportHandler by the time
182 // Process() is called (default, no extra flag)
183 // 2) we either got a full response in which case we don't want to be
184 // notified about anything anymore (RemoveHandler) or we got a partial
185 // answer and we need to wait for more (default, no extra flag)
186 //--------------------------------------------------------------------------
187 pResponse = msg;
188 pBodyReader->SetDataLength( dlen );
189
190 Log *log = DefaultEnv::GetLog();
191 switch( status )
192 {
193 //------------------------------------------------------------------------
194 // Handle the cached cases
195 //------------------------------------------------------------------------
196 case kXR_error:
197 case kXR_redirect:
198 case kXR_wait:
199 return RemoveHandler;
200
201 case kXR_waitresp:
202 {
203 log->Dump( XRootDMsg, "[%s] Got kXR_waitresp response to "
204 "message %s", pUrl.GetHostId().c_str(),
205 pRequest->GetObfuscatedDescription().c_str() );
206
207 pResponse.reset();
208 return Ignore; // This must be handled synchronously!
209 }
210
211 //------------------------------------------------------------------------
212 // Handle the potential raw cases
213 //------------------------------------------------------------------------
214 case kXR_ok:
215 {
216 //----------------------------------------------------------------------
217 // For kXR_read we read in raw mode
218 //----------------------------------------------------------------------
219 uint16_t reqId = ntohs( req->header.requestid );
220 if( reqId == kXR_read )
221 {
222 return Raw | RemoveHandler;
223 }
224
225 //----------------------------------------------------------------------
226 // kXR_readv is the same as kXR_read
227 //----------------------------------------------------------------------
228 if( reqId == kXR_readv )
229 {
230 return Raw | RemoveHandler;
231 }
232
233 //----------------------------------------------------------------------
234 // For everything else we just take what we got
235 //----------------------------------------------------------------------
236 return RemoveHandler;
237 }
238
239 //------------------------------------------------------------------------
240 // kXR_oksofars are special, they are not full responses, so we reset
241 // the response pointer to 0 and add the message to the partial list
242 //------------------------------------------------------------------------
243 case kXR_oksofar:
244 {
245 log->Dump( XRootDMsg, "[%s] Got a kXR_oksofar response to request "
246 "%s", pUrl.GetHostId().c_str(),
247 pRequest->GetObfuscatedDescription().c_str() );
248
249 if( !pOksofarAsAnswer )
250 {
251 pPartialResps.emplace_back( std::move( pResponse ) );
252 }
253
254 //----------------------------------------------------------------------
255 // For kXR_read we either read in raw mode if the message has not
256 // been fully reconstructed already, if it has, we adjust
257 // the buffer offset to prepare for the next one
258 //----------------------------------------------------------------------
259 uint16_t reqId = ntohs( req->header.requestid );
260 if( reqId == kXR_read )
261 {
262 pTimeoutFence.store( true, std::memory_order_relaxed );
263 return Raw | ( pOksofarAsAnswer ? None : NoProcess );
264 }
265
266 //----------------------------------------------------------------------
267 // kXR_readv is similar to read, except that the payload is different
268 //----------------------------------------------------------------------
269 if( reqId == kXR_readv )
270 {
271 pTimeoutFence.store( true, std::memory_order_relaxed );
272 return Raw | ( pOksofarAsAnswer ? None : NoProcess );
273 }
274
275 return ( pOksofarAsAnswer ? None : NoProcess );
276 }
277
278 case kXR_status:
279 {
280 log->Dump( XRootDMsg, "[%s] Got a kXR_status response to request "
281 "%s", pUrl.GetHostId().c_str(),
282 pRequest->GetObfuscatedDescription().c_str() );
283
284 uint16_t reqId = ntohs( req->header.requestid );
285 if( reqId == kXR_pgwrite )
286 {
287 //--------------------------------------------------------------------
288 // In case of pgwrite by definition this wont be a partial response
289 // so we can already remove the handler from the in-queue
290 //--------------------------------------------------------------------
291 return RemoveHandler;
292 }
293
294 //----------------------------------------------------------------------
295 // Otherwise (pgread), first of all we need to read the body of the
296 // kXR_status response, we can handle the raw data (if any) only after
297 // we have the whole kXR_status body
298 //----------------------------------------------------------------------
299 pTimeoutFence.store( true, std::memory_order_relaxed );
300 return None;
301 }
302
303 //------------------------------------------------------------------------
304 // Default
305 //------------------------------------------------------------------------
306 default:
307 return RemoveHandler;
308 }
309 return RemoveHandler;
310 }
311
312 //----------------------------------------------------------------------------
313 // Reexamine the incoming message, and decide on the action to be taken
314 //----------------------------------------------------------------------------
316 {
317 if( !pResponse )
318 return 0;
319
320 Log *log = DefaultEnv::GetLog();
321 ServerResponse *rsp = (ServerResponse *)pResponse->GetBuffer();
322
323 //--------------------------------------------------------------------------
324 // Additional action is only required for kXR_status
325 //--------------------------------------------------------------------------
326 if( rsp->hdr.status != kXR_status ) return 0;
327
328 //--------------------------------------------------------------------------
329 // Ignore malformed status response
330 //--------------------------------------------------------------------------
331 if( pResponse->GetSize() < sizeof( ServerResponseStatus ) )
332 {
333 log->Error( XRootDMsg, "[%s] kXR_status: invalid message size.", pUrl.GetHostId().c_str() );
334 return Corrupted;
335 }
336
337 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
338 uint16_t reqId = ntohs( req->header.requestid );
339 //--------------------------------------------------------------------------
340 // Unmarshal the status body
341 //--------------------------------------------------------------------------
342 XRootDStatus st = XRootDTransport::UnMarshalStatusBody( *pResponse, reqId );
343
344 if( !st.IsOK() && st.code == errDataError )
345 {
346 log->Error( XRootDMsg, "[%s] %s", pUrl.GetHostId().c_str(),
347 st.GetErrorMessage().c_str() );
348 return Corrupted;
349 }
350
351 if( !st.IsOK() )
352 {
353 log->Error( XRootDMsg, "[%s] Failed to unmarshall status body.",
354 pUrl.GetHostId().c_str() );
355 pStatus = st;
356 HandleRspOrQueue();
357 return Ignore;
358 }
359
360 //--------------------------------------------------------------------------
361 // Common handling for partial results
362 //--------------------------------------------------------------------------
363 ServerResponseV2 *rspst = (ServerResponseV2*)pResponse->GetBuffer();
365 {
366 pPartialResps.push_back( std::move( pResponse ) );
367 }
368
369 //--------------------------------------------------------------------------
370 // Decide the actions that we need to take
371 //--------------------------------------------------------------------------
372 uint16_t action = 0;
373 if( reqId == kXR_pgread )
374 {
375 //----------------------------------------------------------------------
376 // The message contains only Status header and body but no raw data
377 //----------------------------------------------------------------------
378 if( !pPageReader )
379 pPageReader.reset( new AsyncPageReader( *pChunkList, pCrc32cDigests ) );
380 pPageReader->SetRsp( rspst );
381
382 action |= Raw;
383
385 action |= NoProcess;
386 else
387 action |= RemoveHandler;
388 }
389 else if( reqId == kXR_pgwrite )
390 {
391 // if data corruption has been detected on the server side we will
392 // send some additional data pointing to the pages that need to be
393 // retransmitted
394 if( size_t( sizeof( ServerResponseHeader ) + rspst->status.hdr.dlen + rspst->status.bdy.dlen ) >
395 pResponse->GetCursor() )
396 action |= More;
397 }
398
399 return action;
400 }
401
402 //----------------------------------------------------------------------------
403 // Get handler sid
404 //----------------------------------------------------------------------------
406 {
407 ClientRequest* req = (ClientRequest*) pRequest->GetBuffer();
408 return ((uint16_t)req->header.streamid[1] << 8) | (uint16_t)req->header.streamid[0];
409 }
410
411 //----------------------------------------------------------------------------
413 //----------------------------------------------------------------------------
415 {
416 Log *log = DefaultEnv::GetLog();
417
418 ServerResponse *rsp = (ServerResponse *)pResponse->GetBuffer();
419
420 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
421
422 //--------------------------------------------------------------------------
423 // If it is a local file, it can be only a metalink redirector
424 //--------------------------------------------------------------------------
425 if( pUrl.IsLocalFile() && pUrl.IsMetalink() )
426 pHosts->back().protocol = kXR_PROTOCOLVERSION;
427
428 //--------------------------------------------------------------------------
429 // We got an answer, check who we were talking to
430 //--------------------------------------------------------------------------
431 else
432 {
433 AnyObject qryResult;
434 int *qryResponse = nullptr;
435 pPostMaster->QueryTransport( pUrl, XRootDQuery::ServerFlags, qryResult );
436 qryResult.Get( qryResponse );
437 if (qryResponse) {
438 pHosts->back().flags = *qryResponse;
439 delete qryResponse;
440 qryResponse = nullptr;
441 }
442 pPostMaster->QueryTransport( pUrl, XRootDQuery::ProtocolVersion, qryResult );
443 qryResult.Get( qryResponse );
444 if (qryResponse) {
445 pHosts->back().protocol = *qryResponse;
446 delete qryResponse;
447 }
448 }
449
450 //--------------------------------------------------------------------------
451 // Process the message
452 //--------------------------------------------------------------------------
453 Status st = XRootDTransport::UnMarshallBody( pResponse.get(), req->header.requestid );
454 if( !st.IsOK() )
455 {
456 pStatus = Status( stFatal, errInvalidMessage );
457 HandleResponse();
458 return;
459 }
460
461 //--------------------------------------------------------------------------
462 // we have an response for the message so it's not in fly anymore
463 //--------------------------------------------------------------------------
464 pMsgInFly = false;
465
466 //--------------------------------------------------------------------------
467 // Reset the aggregated wait (used to omit wait response in case of Metalink
468 // redirector)
469 //--------------------------------------------------------------------------
470 if( rsp->hdr.status != kXR_wait )
471 pAggregatedWaitTime = 0;
472
473 switch( rsp->hdr.status )
474 {
475 //------------------------------------------------------------------------
476 // kXR_ok - we're done here
477 //------------------------------------------------------------------------
478 case kXR_ok:
479 {
480 log->Dump( XRootDMsg, "[%s] Got a kXR_ok response to request %s",
481 pUrl.GetHostId().c_str(),
482 pRequest->GetObfuscatedDescription().c_str() );
483 pStatus = Status();
484 HandleResponse();
485 return;
486 }
487
488 case kXR_status:
489 {
490 log->Dump( XRootDMsg, "[%s] Got a kXR_status response to request %s",
491 pUrl.GetHostId().c_str(),
492 pRequest->GetObfuscatedDescription().c_str() );
493 pStatus = Status();
494 HandleResponse();
495 return;
496 }
497
498 //------------------------------------------------------------------------
499 // kXR_ok - we're serving partial result to the user
500 //------------------------------------------------------------------------
501 case kXR_oksofar:
502 {
503 log->Dump( XRootDMsg, "[%s] Got a kXR_oksofar response to request %s",
504 pUrl.GetHostId().c_str(),
505 pRequest->GetObfuscatedDescription().c_str() );
506 pStatus = Status( stOK, suContinue );
507 HandleResponse();
508 return;
509 }
510
511 //------------------------------------------------------------------------
512 // kXR_error - we've got a problem
513 //------------------------------------------------------------------------
514 case kXR_error:
515 {
516 char *errmsg = new char[rsp->hdr.dlen-3]; errmsg[rsp->hdr.dlen-4] = 0;
517 memcpy( errmsg, rsp->body.error.errmsg, rsp->hdr.dlen-4 );
518 log->Dump( XRootDMsg, "[%s] Got a kXR_error response to request %s "
519 "[%d] %s", pUrl.GetHostId().c_str(),
520 pRequest->GetObfuscatedDescription().c_str(), rsp->body.error.errnum,
521 errmsg );
522 delete [] errmsg;
523
524 HandleError( Status(stError, errErrorResponse, rsp->body.error.errnum) );
525 return;
526 }
527
528 //------------------------------------------------------------------------
529 // kXR_redirect - they tell us to go elsewhere
530 //------------------------------------------------------------------------
531 case kXR_redirect:
532 {
533 if( rsp->hdr.dlen <= 4 )
534 {
535 log->Error( XRootDMsg, "[%s] Got invalid redirect response.",
536 pUrl.GetHostId().c_str() );
537 pStatus = Status( stError, errInvalidResponse );
538 HandleResponse();
539 return;
540 }
541
542 char *urlInfoBuff = new char[rsp->hdr.dlen-3];
543 urlInfoBuff[rsp->hdr.dlen-4] = 0;
544 memcpy( urlInfoBuff, rsp->body.redirect.host, rsp->hdr.dlen-4 );
545 std::string urlInfo = urlInfoBuff;
546 delete [] urlInfoBuff;
547 log->Dump( XRootDMsg, "[%s] Got kXR_redirect response to "
548 "message %s: %s, port %d", pUrl.GetHostId().c_str(),
549 pRequest->GetObfuscatedDescription().c_str(), urlInfo.c_str(),
550 rsp->body.redirect.port );
551
552 //----------------------------------------------------------------------
553 // Check if we can proceed
554 //----------------------------------------------------------------------
555 if( !pRedirectCounter )
556 {
557 log->Warning( XRootDMsg, "[%s] Redirect limit has been reached for "
558 "message %s, the last known error is: %s",
559 pUrl.GetHostId().c_str(),
560 pRequest->GetObfuscatedDescription().c_str(),
561 pLastError.ToString().c_str() );
562
563
564 pStatus = Status( stFatal, errRedirectLimit );
565 HandleResponse();
566 return;
567 }
568 --pRedirectCounter;
569
570 //----------------------------------------------------------------------
571 // Keep the info about this server if we still need to find a load
572 // balancer
573 //----------------------------------------------------------------------
574 uint32_t flags = pHosts->back().flags;
575 if( !pHasLoadBalancer )
576 {
577 if( flags & kXR_isManager )
578 {
579 //------------------------------------------------------------------
580 // If the current server is a meta manager then it supersedes
581 // any existing load balancer, otherwise we assign a load-balancer
582 // only if it has not been already assigned
583 //------------------------------------------------------------------
584 if( ( flags & kXR_attrMeta ) || !pLoadBalancer.url.IsValid() )
585 {
586 pLoadBalancer = pHosts->back();
587 log->Dump( XRootDMsg, "[%s] Current server has been assigned "
588 "as a load-balancer for message %s",
589 pUrl.GetHostId().c_str(),
590 pRequest->GetObfuscatedDescription().c_str() );
591 HostList::iterator it;
592 for( it = pHosts->begin(); it != pHosts->end(); ++it )
593 it->loadBalancer = false;
594 pHosts->back().loadBalancer = true;
595 }
596 }
597 }
598
599 //----------------------------------------------------------------------
600 // If the redirect comes from a data server safe the URL because
601 // in case of a failure we will use it as the effective data server URL
602 // for the tried CGI opaque info
603 //----------------------------------------------------------------------
604 if( flags & kXR_isServer )
605 pEffectiveDataServerUrl = new URL( pHosts->back().url );
606
607 //----------------------------------------------------------------------
608 // Build the URL and check it's validity
609 //----------------------------------------------------------------------
610 std::vector<std::string> urlComponents;
611 std::string newCgi;
612 Utils::splitString( urlComponents, urlInfo, "?" );
613
614 std::ostringstream o;
615
616 o << urlComponents[0];
617 if( rsp->body.redirect.port > 0 )
618 o << ":" << rsp->body.redirect.port << "/";
619 else if( rsp->body.redirect.port < 0 )
620 {
621 //--------------------------------------------------------------------
622 // check if the manager wants to enforce write recovery at himself
623 // (beware we are dealing here with negative flags)
624 //--------------------------------------------------------------------
625 if( ~uint32_t( rsp->body.redirect.port ) & kXR_recoverWrts )
626 pHosts->back().flags |= kXR_recoverWrts;
627
628 //--------------------------------------------------------------------
629 // check if the manager wants to collapse the communication channel
630 // (the redirect host is to replace the current host)
631 //--------------------------------------------------------------------
632 if( ~uint32_t( rsp->body.redirect.port ) & kXR_collapseRedir )
633 {
634 std::string url( rsp->body.redirect.host, rsp->hdr.dlen-4 );
635 pPostMaster->CollapseRedirect( pUrl, url );
636 }
637
638 if( ~uint32_t( rsp->body.redirect.port ) & kXR_ecRedir )
639 {
640 std::string url( rsp->body.redirect.host, rsp->hdr.dlen-4 );
641 if( Utils::CheckEC( pRequest, url ) )
642 pRedirectAsAnswer = true;
643 }
644 }
645
646 URL newUrl = URL( o.str() );
647 if( !newUrl.IsValid() )
648 {
650 log->Error( XRootDMsg, "[%s] Got invalid redirection URL: %s",
651 pUrl.GetHostId().c_str(), urlInfo.c_str() );
652 HandleResponse();
653 return;
654 }
655
656 if( pUrl.GetUserName() != "" && newUrl.GetUserName() == "" )
657 newUrl.SetUserName( pUrl.GetUserName() );
658
659 if( pUrl.GetPassword() != "" && newUrl.GetPassword() == "" )
660 newUrl.SetPassword( pUrl.GetPassword() );
661
662 //----------------------------------------------------------------------
663 // Forward any "xrd.*" params from the original client request also to
664 // the new redirection url
665 // Also, we need to preserve any "xrdcl.*' as they are important for
666 // our internal workflows.
667 //----------------------------------------------------------------------
668 std::ostringstream ossXrd;
669 const URL::ParamsMap &urlParams = pUrl.GetParams();
670
671 for(URL::ParamsMap::const_iterator it = urlParams.begin();
672 it != urlParams.end(); ++it )
673 {
674 if( it->first.compare( 0, 4, "xrd." ) &&
675 it->first.compare( 0, 6, "xrdcl." ) )
676 continue;
677
678 ossXrd << it->first << '=' << it->second << '&';
679 }
680
681 std::string xrdCgi = ossXrd.str();
682 pRedirectUrl = newUrl.GetURL();
683
684 URL cgiURL;
685 if( urlComponents.size() > 1 )
686 {
687 pRedirectUrl += "?";
688 pRedirectUrl += urlComponents[1];
689 std::ostringstream o;
690 o << "fake://fake:111//fake?";
691 o << urlComponents[1];
692
693 if( urlComponents.size() == 3 )
694 o << '?' << urlComponents[2];
695
696 if (!xrdCgi.empty())
697 {
698 o << '&' << xrdCgi;
699 pRedirectUrl += '&';
700 pRedirectUrl += xrdCgi;
701 }
702
703 cgiURL = URL( o.str() );
704 }
705 else {
706 if (!xrdCgi.empty())
707 {
708 std::ostringstream o;
709 o << "fake://fake:111//fake?";
710 o << xrdCgi;
711 cgiURL = URL( o.str() );
712 pRedirectUrl += '?';
713 pRedirectUrl += xrdCgi;
714 }
715 }
716
717 //----------------------------------------------------------------------
718 // Check if we need to return the URL as a response
719 //----------------------------------------------------------------------
720 if( newUrl.GetProtocol() != "root" && newUrl.GetProtocol() != "xroot" &&
721 newUrl.GetProtocol() != "roots" && newUrl.GetProtocol() != "xroots" &&
722 !newUrl.IsLocalFile() )
723 pRedirectAsAnswer = true;
724
725 if( pRedirectAsAnswer )
726 {
727 pStatus = Status( stError, errRedirect );
728 HandleResponse();
729 return;
730 }
731
732 //----------------------------------------------------------------------
733 // Rewrite the message in a way required to send it to another server
734 //----------------------------------------------------------------------
735 newUrl.SetParams( cgiURL.GetParams() );
736 Status st = RewriteRequestRedirect( newUrl );
737 if( !st.IsOK() )
738 {
739 pStatus = st;
740 HandleResponse();
741 return;
742 }
743
744 //----------------------------------------------------------------------
745 // Make sure we don't change the protocol by accident (root vs roots)
746 //----------------------------------------------------------------------
747 if( ( pUrl.GetProtocol() == "roots" || pUrl.GetProtocol() == "xroots" ) &&
748 ( newUrl.GetProtocol() == "root" || newUrl.GetProtocol() == "xroot" ) )
749 newUrl.SetProtocol( "roots" );
750
751 //----------------------------------------------------------------------
752 // Send the request to the new location
753 //----------------------------------------------------------------------
754 HandleError( RetryAtServer( newUrl, RedirectEntry::EntryRedirect ) );
755 return;
756 }
757
758 //------------------------------------------------------------------------
759 // kXR_wait - we wait, and re-issue the request later
760 //------------------------------------------------------------------------
761 case kXR_wait:
762 {
763 uint32_t waitSeconds = 0;
764
765 if( rsp->hdr.dlen >= 4 )
766 {
767 char *infoMsg = new char[rsp->hdr.dlen-3];
768 infoMsg[rsp->hdr.dlen-4] = 0;
769 memcpy( infoMsg, rsp->body.wait.infomsg, rsp->hdr.dlen-4 );
770 log->Dump( XRootDMsg, "[%s] Got kXR_wait response of %d seconds to "
771 "message %s: %s", pUrl.GetHostId().c_str(),
772 rsp->body.wait.seconds, pRequest->GetObfuscatedDescription().c_str(),
773 infoMsg );
774 delete [] infoMsg;
775 waitSeconds = rsp->body.wait.seconds;
776 }
777 else
778 {
779 log->Dump( XRootDMsg, "[%s] Got kXR_wait response of 0 seconds to "
780 "message %s", pUrl.GetHostId().c_str(),
781 pRequest->GetObfuscatedDescription().c_str() );
782 }
783
784 pAggregatedWaitTime += waitSeconds;
785
786 // We need a special case if the data node comes from metalink
787 // redirector. In this case it might make more sense to try the
788 // next entry in the Metalink than wait.
789 if( OmitWait( *pRequest, pLoadBalancer.url ) )
790 {
791 int maxWait = DefaultMaxMetalinkWait;
792 DefaultEnv::GetEnv()->GetInt( "MaxMetalinkWait", maxWait );
793 if( pAggregatedWaitTime > maxWait )
794 {
795 UpdateTriedCGI();
796 HandleError( RetryAtServer( pLoadBalancer.url, RedirectEntry::EntryRedirectOnWait ) );
797 return;
798 }
799 }
800
801 //----------------------------------------------------------------------
802 // Some messages require rewriting before they can be sent again
803 // after wait
804 //----------------------------------------------------------------------
805 Status st = RewriteRequestWait();
806 if( !st.IsOK() )
807 {
808 pStatus = st;
809 HandleResponse();
810 return;
811 }
812
813 //----------------------------------------------------------------------
814 // Register a task to resend the message in some seconds, if we still
815 // have time to do that, and report a timeout otherwise
816 //----------------------------------------------------------------------
817 time_t resendTime = ::time(0)+waitSeconds;
818
819 if( resendTime < pExpiration )
820 {
821 log->Debug( ExDbgMsg, "[%s] Scheduling WaitTask for MsgHandler: %p (message: %s ).",
822 pUrl.GetHostId().c_str(), this,
823 pRequest->GetObfuscatedDescription().c_str() );
824
825 TaskManager *taskMgr = pPostMaster->GetTaskManager();
826 taskMgr->RegisterTask( new WaitTask( this ), resendTime );
827 }
828 else
829 {
830 log->Debug( XRootDMsg, "[%s] Wait time is too long, timing out %s",
831 pUrl.GetHostId().c_str(),
832 pRequest->GetObfuscatedDescription().c_str() );
833 HandleError( Status( stError, errOperationExpired) );
834 }
835 return;
836 }
837
838 //------------------------------------------------------------------------
839 // kXR_waitresp - the response will be returned in some seconds as an
840 // unsolicited message. Currently all messages of this type are handled
841 // one step before in the XrdClStream::OnIncoming as they need to be
842 // processed synchronously.
843 //------------------------------------------------------------------------
844 case kXR_waitresp:
845 {
846 if( rsp->hdr.dlen < 4 )
847 {
848 log->Error( XRootDMsg, "[%s] Got invalid waitresp response.",
849 pUrl.GetHostId().c_str() );
850 pStatus = Status( stError, errInvalidResponse );
851 HandleResponse();
852 return;
853 }
854
855 log->Dump( XRootDMsg, "[%s] Got kXR_waitresp response of %d seconds to "
856 "message %s", pUrl.GetHostId().c_str(),
857 rsp->body.waitresp.seconds,
858 pRequest->GetObfuscatedDescription().c_str() );
859 return;
860 }
861
862 //------------------------------------------------------------------------
863 // Default - unrecognized/unsupported response, declare an error
864 //------------------------------------------------------------------------
865 default:
866 {
867 log->Dump( XRootDMsg, "[%s] Got unrecognized response %d to "
868 "message %s", pUrl.GetHostId().c_str(),
869 rsp->hdr.status, pRequest->GetObfuscatedDescription().c_str() );
870 pStatus = Status( stError, errInvalidResponse );
871 HandleResponse();
872 return;
873 }
874 }
875
876 return;
877 }
878
879 //----------------------------------------------------------------------------
880 // Handle an event other that a message arrival - may be timeout
881 //----------------------------------------------------------------------------
883 XRootDStatus status )
884 {
885 Log *log = DefaultEnv::GetLog();
886 log->Dump( XRootDMsg, "[%s] Stream event reported for msg %s",
887 pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str() );
888
889 if( event == Ready )
890 return 0;
891
892 if( pTimeoutFence.load( std::memory_order_relaxed ) )
893 return 0;
894
895 HandleError( status );
896 return RemoveHandler;
897 }
898
899 //----------------------------------------------------------------------------
900 // Read message body directly from a socket
901 //----------------------------------------------------------------------------
903 Socket *socket,
904 uint32_t &bytesRead )
905 {
906 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
907 uint16_t reqId = ntohs( req->header.requestid );
908
909 if( reqId == kXR_pgread )
910 return pPageReader->Read( *socket, bytesRead );
911
912 return pBodyReader->Read( *socket, bytesRead );
913 }
914
915 //----------------------------------------------------------------------------
916 // We're here when we requested sending something over the wire
917 // and there has been a status update on this action
918 //----------------------------------------------------------------------------
920 XRootDStatus status )
921 {
922 Log *log = DefaultEnv::GetLog();
923
924 const int sst = pSendingState.fetch_or( kSendDone );
925
926 if( sst & kFinalResp )
927 {
928 log->Dump( XRootDMsg, "[%s] Got late notification that outgoing message %s was "
929 "sent, already have final response, queuing handler callback.",
930 pUrl.GetHostId().c_str(), message->GetObfuscatedDescription().c_str() );
931 HandleRspOrQueue();
932 return;
933 }
934
935 if( sst & kSawResp )
936 {
937 log->Dump( XRootDMsg, "[%s] Got late notification that message %s has "
938 "been successfully sent.",
939 pUrl.GetHostId().c_str(), message->GetObfuscatedDescription().c_str() );
940 return;
941 }
942
943 //--------------------------------------------------------------------------
944 // We were successful, so we now need to listen for a response
945 //--------------------------------------------------------------------------
946 if( status.IsOK() )
947 {
948 log->Dump( XRootDMsg, "[%s] Message %s has been successfully sent.",
949 pUrl.GetHostId().c_str(), message->GetObfuscatedDescription().c_str() );
950
951 pMsgInFly = true;
952 return;
953 }
954
955 //--------------------------------------------------------------------------
956 // We have failed, recover if possible
957 //--------------------------------------------------------------------------
958 log->Error( XRootDMsg, "[%s] Impossible to send message %s. Trying to "
959 "recover.", pUrl.GetHostId().c_str(),
960 message->GetObfuscatedDescription().c_str() );
961 HandleError( status );
962 }
963
964 //----------------------------------------------------------------------------
965 // Are we a raw writer or not?
966 //----------------------------------------------------------------------------
968 {
969 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
970 uint16_t reqId = ntohs( req->header.requestid );
971 if( reqId == kXR_write || reqId == kXR_writev || reqId == kXR_pgwrite )
972 return true;
973 // checkpoint + execute
974 if( reqId == kXR_chkpoint && req->chkpoint.opcode == kXR_ckpXeq )
975 {
976 ClientRequest *xeq = (ClientRequest*)pRequest->GetBuffer( sizeof( ClientRequest ) );
977 reqId = ntohs( xeq->header.requestid );
978 return reqId != kXR_truncate; // only checkpointed truncate does not have raw data
979 }
980
981 return false;
982 }
983
984 //----------------------------------------------------------------------------
985 // Write the message body
986 //----------------------------------------------------------------------------
988 uint32_t &bytesWritten )
989 {
990 //--------------------------------------------------------------------------
991 // First check if it is a PgWrite
992 //--------------------------------------------------------------------------
993 if( !pChunkList->empty() && !pCrc32cDigests.empty() )
994 {
995 //------------------------------------------------------------------------
996 // PgWrite will have just one chunk
997 //------------------------------------------------------------------------
998 ChunkInfo chunk = pChunkList->front();
999 //------------------------------------------------------------------------
1000 // Calculate the size of the first and last page (in case the chunk is not
1001 // 4KB aligned)
1002 //------------------------------------------------------------------------
1003 int fLen = 0, lLen = 0;
1004 size_t nbpgs = XrdOucPgrwUtils::csNum( chunk.offset, chunk.length, fLen, lLen );
1005
1006 //------------------------------------------------------------------------
1007 // Set the crc32c buffer if not ready yet
1008 //------------------------------------------------------------------------
1009 if( pPgWrtCksumBuff.GetCursor() == 0 )
1010 {
1011 uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
1012 memcpy( pPgWrtCksumBuff.GetBuffer(), &digest, sizeof( uint32_t ) );
1013 }
1014
1015 uint32_t btsLeft = chunk.length - pAsyncOffset;
1016 uint32_t pglen = ( pPgWrtCurrentPageNb == 0 ? fLen : XrdSys::PageSize ) - pPgWrtCurrentPageOffset;
1017 if( pglen > btsLeft ) pglen = btsLeft;
1018 char* pgbuf = static_cast<char*>( chunk.buffer ) + pAsyncOffset;
1019
1020 while( btsLeft > 0 )
1021 {
1022 // first write the crc32c digest
1023 while( pPgWrtCksumBuff.GetCursor() < sizeof( uint32_t ) )
1024 {
1025 uint32_t dgstlen = sizeof( uint32_t ) - pPgWrtCksumBuff.GetCursor();
1026 char* dgstbuf = pPgWrtCksumBuff.GetBufferAtCursor();
1027 int btswrt = 0;
1028 Status st = socket->Send( dgstbuf, dgstlen, btswrt );
1029 if( !st.IsOK() ) return st;
1030 bytesWritten += btswrt;
1031 pPgWrtCksumBuff.AdvanceCursor( btswrt );
1032 if( st.code == suRetry ) return st;
1033 }
1034 // then write the raw data (one page)
1035 int btswrt = 0;
1036 Status st = socket->Send( pgbuf, pglen, btswrt );
1037 if( !st.IsOK() ) return st;
1038 pgbuf += btswrt;
1039 pglen -= btswrt;
1040 btsLeft -= btswrt;
1041 bytesWritten += btswrt;
1042 pAsyncOffset += btswrt; // update the offset to the raw data
1043 if( st.code == suRetry ) return st;
1044 // if we managed to write all the data ...
1045 if( pglen == 0 )
1046 {
1047 // move to the next page
1048 ++pPgWrtCurrentPageNb;
1049 if( pPgWrtCurrentPageNb < nbpgs )
1050 {
1051 // set the digest buffer
1052 pPgWrtCksumBuff.SetCursor( 0 );
1053 uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
1054 memcpy( pPgWrtCksumBuff.GetBuffer(), &digest, sizeof( uint32_t ) );
1055 }
1056 // set the page length
1057 pglen = XrdSys::PageSize;
1058 if( pglen > btsLeft ) pglen = btsLeft;
1059 // reset offset in the current page
1060 pPgWrtCurrentPageOffset = 0;
1061 }
1062 else
1063 // otherwise just adjust the offset in the current page
1064 pPgWrtCurrentPageOffset += btswrt;
1065
1066 }
1067 }
1068 else if( !pChunkList->empty() )
1069 {
1070 size_t size = pChunkList->size();
1071 for( size_t i = pAsyncChunkIndex ; i < size; ++i )
1072 {
1073 char *buffer = (char*)(*pChunkList)[i].buffer;
1074 uint32_t size = (*pChunkList)[i].length;
1075 size_t leftToBeWritten = size - pAsyncOffset;
1076
1077 while( leftToBeWritten )
1078 {
1079 int btswrt = 0;
1080 Status st = socket->Send( buffer + pAsyncOffset, leftToBeWritten, btswrt );
1081 bytesWritten += btswrt;
1082 if( !st.IsOK() || st.code == suRetry ) return st;
1083 pAsyncOffset += btswrt;
1084 leftToBeWritten -= btswrt;
1085 }
1086 //----------------------------------------------------------------------
1087 // Remember that we have moved to the next chunk, also clear the offset
1088 // within the buffer as we are going to move to a new one
1089 //----------------------------------------------------------------------
1090 ++pAsyncChunkIndex;
1091 pAsyncOffset = 0;
1092 }
1093 }
1094 else
1095 {
1096 Log *log = DefaultEnv::GetLog();
1097
1098 //------------------------------------------------------------------------
1099 // If the socket is encrypted we cannot use a kernel buffer, we have to
1100 // convert to user space buffer
1101 //------------------------------------------------------------------------
1102 if( socket->IsEncrypted() )
1103 {
1104 log->Debug( XRootDMsg, "[%s] Channel is encrypted: cannot use kernel buffer.",
1105 pUrl.GetHostId().c_str() );
1106
1107 char *ubuff = 0;
1108 ssize_t ret = XrdSys::Move( *pKBuff, ubuff );
1109 if( ret < 0 ) return Status( stError, errInternal );
1110 pChunkList->push_back( ChunkInfo( 0, ret, ubuff ) );
1111 return WriteMessageBody( socket, bytesWritten );
1112 }
1113
1114 //------------------------------------------------------------------------
1115 // Send the data
1116 //------------------------------------------------------------------------
1117 while( !pKBuff->Empty() )
1118 {
1119 int btswrt = 0;
1120 Status st = socket->Send( *pKBuff, btswrt );
1121 bytesWritten += btswrt;
1122 if( !st.IsOK() || st.code == suRetry ) return st;
1123 }
1124
1125 log->Debug( XRootDMsg, "[%s] Request %s payload (kernel buffer) transferred to socket.",
1126 pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str() );
1127 }
1128
1129 return Status();
1130 }
1131
1132 //----------------------------------------------------------------------------
1133 // We're here when we got a time event. We needed to re-issue the request
1134 // in some time in the future, and that moment has arrived
1135 //----------------------------------------------------------------------------
1137 {
1138 HandleError( RetryAtServer( pUrl, RedirectEntry::EntryWait ) );
1139 }
1140
1141 //----------------------------------------------------------------------------
1142 // Bookkeeping after partial response has been received.
1143 //----------------------------------------------------------------------------
1145 {
1146 pTimeoutFence.store( false, std::memory_order_relaxed ); // Take down the timeout fence
1147 }
1148
1149 //----------------------------------------------------------------------------
1150 // Unpack the message and call the response handler
1151 //----------------------------------------------------------------------------
1152 void XRootDMsgHandler::HandleResponse()
1153 {
1154 //--------------------------------------------------------------------------
1155 // Is it a final response?
1156 //--------------------------------------------------------------------------
1157 bool finalrsp = !( pStatus.IsOK() && pStatus.code == suContinue );
1158 if( finalrsp )
1159 {
1160 // Do not do final processing of the response if we haven't had
1161 // confirmation the original request was sent (via OnStatusReady).
1162 // The final processing will be triggered when we get the confirm.
1163 const int sst = pSendingState.fetch_or( kFinalResp );
1164 if( !( sst & kSendDone ) )
1165 return;
1166 }
1167
1168 //--------------------------------------------------------------------------
1169 // Process the response and notify the listener
1170 //--------------------------------------------------------------------------
1172 XRootDStatus *status = ProcessStatus();
1173 AnyObject *response = 0;
1174
1175 Log *log = DefaultEnv::GetLog();
1176 log->Debug( ExDbgMsg, "[%s] Calling MsgHandler: %p (message: %s ) "
1177 "with status: %s.",
1178 pUrl.GetHostId().c_str(), this,
1179 pRequest->GetObfuscatedDescription().c_str(),
1180 status->ToString().c_str() );
1181
1182 if( status->IsOK() )
1183 {
1184 Status st = ParseResponse( response );
1185 if( !st.IsOK() )
1186 {
1187 delete status;
1188 delete response;
1189 status = new XRootDStatus( st );
1190 response = 0;
1191 }
1192 }
1193
1194 //--------------------------------------------------------------------------
1195 // Close the redirect entry if necessary
1196 //--------------------------------------------------------------------------
1197 if( pRdirEntry )
1198 {
1199 pRdirEntry->status = *status;
1200 pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
1201 }
1202
1203 //--------------------------------------------------------------------------
1204 // Release the stream id
1205 //--------------------------------------------------------------------------
1206 if( pSidMgr && finalrsp )
1207 {
1208 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1209 if( status->IsOK() || !pMsgInFly ||
1210 !( status->code == errOperationExpired || status->code == errOperationInterrupted ) )
1211 pSidMgr->ReleaseSID( req->header.streamid );
1212 }
1213
1214 HostList *hosts = pHosts.release();
1215 if( !finalrsp )
1216 pHosts.reset( new HostList( *hosts ) );
1217
1218 pResponseHandler->HandleResponseWithHosts( status, response, hosts );
1219
1220 //--------------------------------------------------------------------------
1221 // if it is the final response there is nothing more to do ...
1222 //--------------------------------------------------------------------------
1223 if( finalrsp )
1224 delete this;
1225 //--------------------------------------------------------------------------
1226 // on the other hand if it is not the final response, we have to keep the
1227 // MsgHandler and delete the current response
1228 //--------------------------------------------------------------------------
1229 else
1230 {
1231 XrdSysCondVarHelper lck( pCV );
1232 pResponse.reset();
1233 pTimeoutFence.store( false, std::memory_order_relaxed );
1234 pCV.Broadcast();
1235 }
1236 }
1237
1238
1239 //----------------------------------------------------------------------------
1240 // Extract the status information from the stuff that we got
1241 //----------------------------------------------------------------------------
1242 XRootDStatus *XRootDMsgHandler::ProcessStatus()
1243 {
1244 XRootDStatus *st = new XRootDStatus( pStatus );
1245 ServerResponse *rsp = 0;
1246 if( pResponse )
1247 rsp = (ServerResponse *)pResponse->GetBuffer();
1248
1249 if( !pStatus.IsOK() && rsp )
1250 {
1251 if( pStatus.code == errErrorResponse )
1252 {
1253 st->errNo = rsp->body.error.errnum;
1254 // omit the last character as the string returned from the server
1255 // (acording to protocol specs) should be null-terminated
1256 std::string errmsg( rsp->body.error.errmsg, rsp->hdr.dlen-5 );
1257 if( st->errNo == kXR_noReplicas && !pLastError.IsOK() )
1258 errmsg += " Last seen error: " + pLastError.ToString();
1259 st->SetErrorMessage( errmsg );
1260 }
1261 else if( pStatus.code == errRedirect )
1262 st->SetErrorMessage( pRedirectUrl );
1263 }
1264 return st;
1265 }
1266
1267 //------------------------------------------------------------------------
1268 // Parse the response and put it in an object that could be passed to
1269 // the user
1270 //------------------------------------------------------------------------
1271 Status XRootDMsgHandler::ParseResponse( AnyObject *&response )
1272 {
1273 if( !pResponse )
1274 return Status();
1275
1276 ServerResponse *rsp = (ServerResponse *)pResponse->GetBuffer();
1277 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1278 Log *log = DefaultEnv::GetLog();
1279
1280 //--------------------------------------------------------------------------
1281 // Handle redirect as an answer
1282 //--------------------------------------------------------------------------
1283 if( rsp->hdr.status == kXR_redirect )
1284 {
1285 log->Error( XRootDMsg, "Internal Error: unable to process redirect" );
1286 return 0;
1287 }
1288
1289 Buffer buff;
1290 uint32_t length = 0;
1291 char *buffer = 0;
1292
1293 //--------------------------------------------------------------------------
1294 // We don't have any partial answers so pass what we have
1295 //--------------------------------------------------------------------------
1296 if( pPartialResps.empty() )
1297 {
1298 buffer = rsp->body.buffer.data;
1299 length = rsp->hdr.dlen;
1300 }
1301 //--------------------------------------------------------------------------
1302 // Partial answers, we need to glue them together before parsing
1303 //--------------------------------------------------------------------------
1304 else if( req->header.requestid != kXR_read &&
1305 req->header.requestid != kXR_readv )
1306 {
1307 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1308 {
1309 ServerResponse *part = (ServerResponse*)pPartialResps[i]->GetBuffer();
1310 length += part->hdr.dlen;
1311 }
1312 length += rsp->hdr.dlen;
1313
1314 buff.Allocate( length );
1315 uint32_t offset = 0;
1316 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1317 {
1318 ServerResponse *part = (ServerResponse*)pPartialResps[i]->GetBuffer();
1319 buff.Append( part->body.buffer.data, part->hdr.dlen, offset );
1320 offset += part->hdr.dlen;
1321 }
1322 buff.Append( rsp->body.buffer.data, rsp->hdr.dlen, offset );
1323 buffer = buff.GetBuffer();
1324 }
1325
1326 //--------------------------------------------------------------------------
1327 // Right, but what was the question?
1328 //--------------------------------------------------------------------------
1329 switch( req->header.requestid )
1330 {
1331 //------------------------------------------------------------------------
1332 // kXR_mv, kXR_truncate, kXR_rm, kXR_mkdir, kXR_rmdir, kXR_chmod,
1333 // kXR_ping, kXR_close, kXR_write, kXR_sync
1334 //------------------------------------------------------------------------
1335 case kXR_mv:
1336 case kXR_truncate:
1337 case kXR_rm:
1338 case kXR_mkdir:
1339 case kXR_rmdir:
1340 case kXR_chmod:
1341 case kXR_ping:
1342 case kXR_close:
1343 case kXR_write:
1344 case kXR_writev:
1345 case kXR_sync:
1346 case kXR_chkpoint:
1347 return Status();
1348
1349 //------------------------------------------------------------------------
1350 // kXR_locate
1351 //------------------------------------------------------------------------
1352 case kXR_locate:
1353 {
1354 AnyObject *obj = new AnyObject();
1355
1356 char *nullBuffer = new char[length+1];
1357 nullBuffer[length] = 0;
1358 memcpy( nullBuffer, buffer, length );
1359
1360 log->Dump( XRootDMsg, "[%s] Parsing the response to %s as "
1361 "LocateInfo: %s", pUrl.GetHostId().c_str(),
1362 pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
1363 LocationInfo *data = new LocationInfo();
1364
1365 if( data->ParseServerResponse( nullBuffer ) == false )
1366 {
1367 delete obj;
1368 delete data;
1369 delete [] nullBuffer;
1370 return Status( stError, errInvalidResponse );
1371 }
1372 delete [] nullBuffer;
1373
1374 obj->Set( data );
1375 response = obj;
1376 return Status();
1377 }
1378
1379 //------------------------------------------------------------------------
1380 // kXR_stat
1381 //------------------------------------------------------------------------
1382 case kXR_stat:
1383 {
1384 AnyObject *obj = new AnyObject();
1385
1386 //----------------------------------------------------------------------
1387 // Virtual File System stat (kXR_vfs)
1388 //----------------------------------------------------------------------
1389 if( req->stat.options & kXR_vfs )
1390 {
1391 StatInfoVFS *data = new StatInfoVFS();
1392
1393 char *nullBuffer = new char[length+1];
1394 nullBuffer[length] = 0;
1395 memcpy( nullBuffer, buffer, length );
1396
1397 log->Dump( XRootDMsg, "[%s] Parsing the response to %s as "
1398 "StatInfoVFS: %s", pUrl.GetHostId().c_str(),
1399 pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
1400
1401 if( data->ParseServerResponse( nullBuffer ) == false )
1402 {
1403 delete obj;
1404 delete data;
1405 delete [] nullBuffer;
1406 return Status( stError, errInvalidResponse );
1407 }
1408 delete [] nullBuffer;
1409
1410 obj->Set( data );
1411 }
1412 //----------------------------------------------------------------------
1413 // Normal stat
1414 //----------------------------------------------------------------------
1415 else
1416 {
1417 StatInfo *data = new StatInfo();
1418
1419 char *nullBuffer = new char[length+1];
1420 nullBuffer[length] = 0;
1421 memcpy( nullBuffer, buffer, length );
1422
1423 log->Dump( XRootDMsg, "[%s] Parsing the response to %s as StatInfo: "
1424 "%s", pUrl.GetHostId().c_str(),
1425 pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
1426
1427 if( data->ParseServerResponse( nullBuffer ) == false )
1428 {
1429 delete obj;
1430 delete data;
1431 delete [] nullBuffer;
1432 return Status( stError, errInvalidResponse );
1433 }
1434 delete [] nullBuffer;
1435 obj->Set( data );
1436 }
1437
1438 response = obj;
1439 return Status();
1440 }
1441
1442 //------------------------------------------------------------------------
1443 // kXR_protocol
1444 //------------------------------------------------------------------------
1445 case kXR_protocol:
1446 {
1447 log->Dump( XRootDMsg, "[%s] Parsing the response to %s as ProtocolInfo",
1448 pUrl.GetHostId().c_str(),
1449 pRequest->GetObfuscatedDescription().c_str() );
1450
1451 if( rsp->hdr.dlen < 8 )
1452 {
1453 log->Error( XRootDMsg, "[%s] Got invalid redirect response.",
1454 pUrl.GetHostId().c_str() );
1455 return Status( stError, errInvalidResponse );
1456 }
1457
1458 AnyObject *obj = new AnyObject();
1459 ProtocolInfo *data = new ProtocolInfo( rsp->body.protocol.pval,
1460 rsp->body.protocol.flags );
1461 obj->Set( data );
1462 response = obj;
1463 return Status();
1464 }
1465
1466 //------------------------------------------------------------------------
1467 // kXR_dirlist
1468 //------------------------------------------------------------------------
1469 case kXR_dirlist:
1470 {
1471 AnyObject *obj = new AnyObject();
1472 log->Dump( XRootDMsg, "[%s] Parsing the response to %s as "
1473 "DirectoryList", pUrl.GetHostId().c_str(),
1474 pRequest->GetObfuscatedDescription().c_str() );
1475
1476 char *path = new char[req->dirlist.dlen+1];
1477 path[req->dirlist.dlen] = 0;
1478 memcpy( path, pRequest->GetBuffer(24), req->dirlist.dlen );
1479
1480 DirectoryList *data = new DirectoryList();
1481 data->SetParentName( path );
1482 delete [] path;
1483
1484 char *nullBuffer = new char[length+1];
1485 nullBuffer[length] = 0;
1486 memcpy( nullBuffer, buffer, length );
1487
1488 bool invalidrsp = false;
1489
1490 if( !pDirListStarted )
1491 {
1492 pDirListWithStat = DirectoryList::HasStatInfo( nullBuffer );
1493 pDirListStarted = true;
1494
1495 invalidrsp = !data->ParseServerResponse( pUrl.GetHostId(), nullBuffer );
1496 }
1497 else
1498 invalidrsp = !data->ParseServerResponse( pUrl.GetHostId(), nullBuffer, pDirListWithStat );
1499
1500 if( invalidrsp )
1501 {
1502 delete data;
1503 delete obj;
1504 delete [] nullBuffer;
1505 return Status( stError, errInvalidResponse );
1506 }
1507
1508 delete [] nullBuffer;
1509 obj->Set( data );
1510 response = obj;
1511 return Status();
1512 }
1513
1514 //------------------------------------------------------------------------
1515 // kXR_open - if we got the statistics, otherwise return 0
1516 //------------------------------------------------------------------------
1517 case kXR_open:
1518 {
1519 log->Dump( XRootDMsg, "[%s] Parsing the response to %s as OpenInfo",
1520 pUrl.GetHostId().c_str(),
1521 pRequest->GetObfuscatedDescription().c_str() );
1522
1523 if( rsp->hdr.dlen < 4 )
1524 {
1525 log->Error( XRootDMsg, "[%s] Got invalid open response.",
1526 pUrl.GetHostId().c_str() );
1527 return Status( stError, errInvalidResponse );
1528 }
1529
1530 AnyObject *obj = new AnyObject();
1531 StatInfo *statInfo = 0;
1532
1533 //----------------------------------------------------------------------
1534 // Handle StatInfo if requested
1535 //----------------------------------------------------------------------
1536 if( req->open.options & kXR_retstat )
1537 {
1538 log->Dump( XRootDMsg, "[%s] Parsing StatInfo in response to %s",
1539 pUrl.GetHostId().c_str(),
1540 pRequest->GetObfuscatedDescription().c_str() );
1541
1542 if( rsp->hdr.dlen >= 12 )
1543 {
1544 char *nullBuffer = new char[rsp->hdr.dlen-11];
1545 nullBuffer[rsp->hdr.dlen-12] = 0;
1546 memcpy( nullBuffer, buffer+12, rsp->hdr.dlen-12 );
1547
1548 statInfo = new StatInfo();
1549 if( statInfo->ParseServerResponse( nullBuffer ) == false )
1550 {
1551 delete statInfo;
1552 statInfo = 0;
1553 }
1554 delete [] nullBuffer;
1555 }
1556
1557 if( rsp->hdr.dlen < 12 || !statInfo )
1558 {
1559 log->Error( XRootDMsg, "[%s] Unable to parse StatInfo in response "
1560 "to %s", pUrl.GetHostId().c_str(),
1561 pRequest->GetObfuscatedDescription().c_str() );
1562 delete obj;
1563 return Status( stError, errInvalidResponse );
1564 }
1565 }
1566
1567 OpenInfo *data = new OpenInfo( (uint8_t*)buffer,
1568 pResponse->GetSessionId(),
1569 statInfo );
1570 obj->Set( data );
1571 response = obj;
1572 return Status();
1573 }
1574
1575 //------------------------------------------------------------------------
1576 // kXR_read
1577 //------------------------------------------------------------------------
1578 case kXR_read:
1579 {
1580 log->Dump( XRootDMsg, "[%s] Parsing the response to %s as ChunkInfo",
1581 pUrl.GetHostId().c_str(),
1582 pRequest->GetObfuscatedDescription().c_str() );
1583
1584 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1585 {
1586 //--------------------------------------------------------------------
1587 // we are expecting to have only the header in the message, the raw
1588 // data have been readout into the user buffer
1589 //--------------------------------------------------------------------
1590 if( pPartialResps[i]->GetSize() > 8 )
1591 return Status( stOK, errInternal );
1592 }
1593 //----------------------------------------------------------------------
1594 // we are expecting to have only the header in the message, the raw
1595 // data have been readout into the user buffer
1596 //----------------------------------------------------------------------
1597 if( pResponse->GetSize() > 8 )
1598 return Status( stOK, errInternal );
1599 //----------------------------------------------------------------------
1600 // Get the response for the end user
1601 //----------------------------------------------------------------------
1602 return pBodyReader->GetResponse( response );
1603 }
1604
1605 //------------------------------------------------------------------------
1606 // kXR_pgread
1607 //------------------------------------------------------------------------
1608 case kXR_pgread:
1609 {
1610 log->Dump( XRootDMsg, "[%s] Parsing the response to %s as PageInfo",
1611 pUrl.GetHostId().c_str(),
1612 pRequest->GetObfuscatedDescription().c_str() );
1613
1614 //----------------------------------------------------------------------
1615 // Glue in the cached responses if necessary
1616 //----------------------------------------------------------------------
1617 ChunkInfo chunk = pChunkList->front();
1618 bool sizeMismatch = false;
1619 uint32_t currentOffset = 0;
1620 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1621 {
1622 ServerResponseV2 *part = (ServerResponseV2*)pPartialResps[i]->GetBuffer();
1623
1624 //--------------------------------------------------------------------
1625 // the actual size of the raw data without the crc32c checksums
1626 //--------------------------------------------------------------------
1627 size_t datalen = part->status.bdy.dlen - NbPgPerRsp( part->info.pgread.offset,
1628 part->status.bdy.dlen ) * CksumSize;
1629
1630 if( currentOffset + datalen > chunk.length )
1631 {
1632 sizeMismatch = true;
1633 break;
1634 }
1635
1636 currentOffset += datalen;
1637 }
1638
1639 ServerResponseV2 *rspst = (ServerResponseV2*)pResponse->GetBuffer();
1640 size_t datalen = rspst->status.bdy.dlen - NbPgPerRsp( rspst->info.pgread.offset,
1641 rspst->status.bdy.dlen ) * CksumSize;
1642 if( currentOffset + datalen <= chunk.length )
1643 currentOffset += datalen;
1644 else
1645 sizeMismatch = true;
1646
1647 //----------------------------------------------------------------------
1648 // Overflow
1649 //----------------------------------------------------------------------
1650 if( pChunkStatus.front().sizeError || sizeMismatch )
1651 {
1652 log->Error( XRootDMsg, "[%s] Handling response to %s: user supplied "
1653 "buffer is too small for the received data.",
1654 pUrl.GetHostId().c_str(),
1655 pRequest->GetObfuscatedDescription().c_str() );
1656 return Status( stError, errInvalidResponse );
1657 }
1658
1659 AnyObject *obj = new AnyObject();
1660 PageInfo *pgInfo = new PageInfo( chunk.offset, currentOffset, chunk.buffer,
1661 std::move( pCrc32cDigests) );
1662
1663 obj->Set( pgInfo );
1664 response = obj;
1665 return Status();
1666 }
1667
1668 //------------------------------------------------------------------------
1669 // kXR_pgwrite
1670 //------------------------------------------------------------------------
1671 case kXR_pgwrite:
1672 {
1673 std::vector<std::tuple<uint64_t, uint32_t>> retries;
1674
1675 ServerResponseV2 *rsp = (ServerResponseV2*)pResponse->GetBuffer();
1676 if( rsp->status.bdy.dlen > 0 )
1677 {
1678 ServerResponseBody_pgWrCSE *cse = (ServerResponseBody_pgWrCSE*)pResponse->GetBuffer( sizeof( ServerResponseV2 ) );
1679 size_t pgcnt = ( rsp->status.bdy.dlen - 8 ) / sizeof( kXR_int64 );
1680 retries.reserve( pgcnt );
1681 kXR_int64 *pgoffs = (kXR_int64*)pResponse->GetBuffer( sizeof( ServerResponseV2 ) +
1682 sizeof( ServerResponseBody_pgWrCSE ) );
1683
1684 for( size_t i = 0; i < pgcnt; ++i )
1685 {
1686 uint32_t len = XrdSys::PageSize;
1687 if( i == 0 ) len = cse->dlFirst;
1688 else if( i == pgcnt - 1 ) len = cse->dlLast;
1689 retries.push_back( std::make_tuple( pgoffs[i], len ) );
1690 }
1691 }
1692
1693 RetryInfo *info = new RetryInfo( std::move( retries ) );
1694 AnyObject *obj = new AnyObject();
1695 obj->Set( info );
1696 response = obj;
1697
1698 return Status();
1699 }
1700
1701
1702 //------------------------------------------------------------------------
1703 // kXR_readv - we need to pass the length of the buffer to the user code
1704 //------------------------------------------------------------------------
1705 case kXR_readv:
1706 {
1707 log->Dump( XRootDMsg, "[%s] Parsing the response to %p as "
1708 "VectorReadInfo", pUrl.GetHostId().c_str(),
1709 pRequest->GetObfuscatedDescription().c_str() );
1710
1711 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1712 {
1713 //--------------------------------------------------------------------
1714 // we are expecting to have only the header in the message, the raw
1715 // data have been readout into the user buffer
1716 //--------------------------------------------------------------------
1717 if( pPartialResps[i]->GetSize() > 8 )
1718 return Status( stOK, errInternal );
1719 }
1720 //----------------------------------------------------------------------
1721 // we are expecting to have only the header in the message, the raw
1722 // data have been readout into the user buffer
1723 //----------------------------------------------------------------------
1724 if( pResponse->GetSize() > 8 )
1725 return Status( stOK, errInternal );
1726 //----------------------------------------------------------------------
1727 // Get the response for the end user
1728 //----------------------------------------------------------------------
1729 return pBodyReader->GetResponse( response );
1730 }
1731
1732 //------------------------------------------------------------------------
1733 // kXR_fattr
1734 //------------------------------------------------------------------------
1735 case kXR_fattr:
1736 {
1737 int len = rsp->hdr.dlen;
1738 char* data = rsp->body.buffer.data;
1739
1740 return ParseXAttrResponse( data, len, response );
1741 }
1742
1743 //------------------------------------------------------------------------
1744 // kXR_query
1745 //------------------------------------------------------------------------
1746 case kXR_query:
1747 case kXR_set:
1748 case kXR_prepare:
1749 default:
1750 {
1751 AnyObject *obj = new AnyObject();
1752 log->Dump( XRootDMsg, "[%s] Parsing the response to %s as BinaryData",
1753 pUrl.GetHostId().c_str(),
1754 pRequest->GetObfuscatedDescription().c_str() );
1755
1756 BinaryDataInfo *data = new BinaryDataInfo();
1757 data->Allocate( length );
1758 data->Append( buffer, length );
1759 obj->Set( data );
1760 response = obj;
1761 return Status();
1762 }
1763 };
1764 return Status( stError, errInvalidMessage );
1765 }
1766
1767 //------------------------------------------------------------------------
1768 // Parse the response to kXR_fattr request and put it in an object that
1769 // could be passed to the user
1770 //------------------------------------------------------------------------
1771 Status XRootDMsgHandler::ParseXAttrResponse( char *data, size_t len,
1772 AnyObject *&response )
1773 {
1774 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1775// Log *log = DefaultEnv::GetLog(); //TODO
1776
1777 switch( req->fattr.subcode )
1778 {
1779 case kXR_fattrDel:
1780 case kXR_fattrSet:
1781 {
1782 Status status;
1783
1784 kXR_char nerrs = 0;
1785 if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
1786 return status;
1787
1788 kXR_char nattr = 0;
1789 if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
1790 return status;
1791
1792 std::vector<XAttrStatus> resp;
1793 // read the namevec
1794 for( kXR_char i = 0; i < nattr; ++i )
1795 {
1796 kXR_unt16 rc = 0;
1797 if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
1798 return status;
1799 rc = ntohs( rc );
1800
1801 // count errors
1802 if( rc ) --nerrs;
1803
1804 std::string name;
1805 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1806 return status;
1807
1808 XRootDStatus st = rc ? XRootDStatus( stError, errErrorResponse, rc ) :
1809 XRootDStatus();
1810 resp.push_back( XAttrStatus( name, st ) );
1811 }
1812
1813 // check if we read all the data and if the error count is OK
1814 if( len != 0 || nerrs != 0 ) return Status( stError, errDataError );
1815
1816 // set up the response object
1817 response = new AnyObject();
1818 response->Set( new std::vector<XAttrStatus>( std::move( resp ) ) );
1819
1820 return Status();
1821 }
1822
1823 case kXR_fattrGet:
1824 {
1825 Status status;
1826
1827 kXR_char nerrs = 0;
1828 if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
1829 return status;
1830
1831 kXR_char nattr = 0;
1832 if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
1833 return status;
1834
1835 std::vector<XAttr> resp;
1836 resp.reserve( nattr );
1837
1838 // read the name vec
1839 for( kXR_char i = 0; i < nattr; ++i )
1840 {
1841 kXR_unt16 rc = 0;
1842 if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
1843 return status;
1844 rc = ntohs( rc );
1845
1846 // count errors
1847 if( rc ) --nerrs;
1848
1849 std::string name;
1850 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1851 return status;
1852
1853 XRootDStatus st = rc ? XRootDStatus( stError, errErrorResponse, rc ) :
1854 XRootDStatus();
1855 resp.push_back( XAttr( name, st ) );
1856 }
1857
1858 // read the value vec
1859 for( kXR_char i = 0; i < nattr; ++i )
1860 {
1861 kXR_int32 vlen = 0;
1862 if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
1863 return status;
1864 vlen = ntohl( vlen );
1865
1866 std::string value;
1867 if( !( status = ReadFromBuffer( data, len, vlen, value ) ).IsOK() )
1868 return status;
1869
1870 resp[i].value.swap( value );
1871 }
1872
1873 // check if we read all the data and if the error count is OK
1874 if( len != 0 || nerrs != 0 ) return Status( stError, errDataError );
1875
1876 // set up the response object
1877 response = new AnyObject();
1878 response->Set( new std::vector<XAttr>( std::move( resp ) ) );
1879
1880 return Status();
1881 }
1882
1883 case kXR_fattrList:
1884 {
1885 Status status;
1886 std::vector<XAttr> resp;
1887
1888 while( len > 0 )
1889 {
1890 std::string name;
1891 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1892 return status;
1893
1894 kXR_int32 vlen = 0;
1895 if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
1896 return status;
1897 vlen = ntohl( vlen );
1898
1899 std::string value;
1900 if( !( status = ReadFromBuffer( data, len, vlen, value ) ).IsOK() )
1901 return status;
1902
1903 resp.push_back( XAttr( name, value ) );
1904 }
1905
1906 // set up the response object
1907 response = new AnyObject();
1908 response->Set( new std::vector<XAttr>( std::move( resp ) ) );
1909
1910 return Status();
1911 }
1912
1913 default:
1914 return Status( stError, errDataError );
1915 }
1916 }
1917
1918 //----------------------------------------------------------------------------
1919 // Perform the changes to the original request needed by the redirect
1920 // procedure - allocate new streamid, append redirection data and such
1921 //----------------------------------------------------------------------------
1922 Status XRootDMsgHandler::RewriteRequestRedirect( const URL &newUrl )
1923 {
1924 Log *log = DefaultEnv::GetLog();
1925
1926 Status st;
1927 // Append any "xrd.*" parameters present in newCgi so that any authentication
1928 // requirements are properly enforced
1929 const URL::ParamsMap &newCgi = newUrl.GetParams();
1930 std::string xrdCgi = "";
1931 std::ostringstream ossXrd;
1932 for(URL::ParamsMap::const_iterator it = newCgi.begin(); it != newCgi.end(); ++it )
1933 {
1934 if( it->first.compare( 0, 4, "xrd." ) )
1935 continue;
1936 ossXrd << it->first << '=' << it->second << '&';
1937 }
1938
1939 xrdCgi = ossXrd.str();
1940 // Redirection URL containing also any original xrd.* opaque parameters
1941 XrdCl::URL authUrl;
1942
1943 if (xrdCgi.empty())
1944 {
1945 authUrl = newUrl;
1946 }
1947 else
1948 {
1949 std::string surl = newUrl.GetURL();
1950 (surl.find('?') == std::string::npos) ? (surl += '?') :
1951 ((*surl.rbegin() != '&') ? (surl += '&') : (surl += ""));
1952 surl += xrdCgi;
1953 if (!authUrl.FromString(surl))
1954 {
1955 std::string surlLog = surl;
1956 if( unlikely( log->GetLevel() >= Log::ErrorMsg ) ) {
1957 surlLog = obfuscateAuth(surlLog);
1958 }
1959 log->Error( XRootDMsg, "[%s] Failed to build redirection URL from data: %s",
1960 newUrl.GetHostId().c_str(), surl.c_str());
1961 return Status(stError, errInvalidRedirectURL);
1962 }
1963 }
1964
1965 //--------------------------------------------------------------------------
1966 // Rewrite particular requests
1967 //--------------------------------------------------------------------------
1969 MessageUtils::RewriteCGIAndPath( pRequest, newCgi, true, newUrl.GetPath() );
1971 return Status();
1972 }
1973
1974 //----------------------------------------------------------------------------
1975 // Some requests need to be rewritten also after getting kXR_wait
1976 //----------------------------------------------------------------------------
1977 Status XRootDMsgHandler::RewriteRequestWait()
1978 {
1979 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1980
1982
1983 //------------------------------------------------------------------------
1984 // For kXR_locate and kXR_open request the kXR_refresh bit needs to be
1985 // turned off after wait
1986 //------------------------------------------------------------------------
1987 switch( req->header.requestid )
1988 {
1989 case kXR_locate:
1990 {
1991 uint16_t refresh = kXR_refresh;
1992 req->locate.options &= (~refresh);
1993 break;
1994 }
1995
1996 case kXR_open:
1997 {
1998 uint16_t refresh = kXR_refresh;
1999 req->locate.options &= (~refresh);
2000 break;
2001 }
2002 }
2003
2006 return Status();
2007 }
2008
2009 //----------------------------------------------------------------------------
2010 // Recover error
2011 //----------------------------------------------------------------------------
2012 void XRootDMsgHandler::HandleError( XRootDStatus status )
2013 {
2014 //--------------------------------------------------------------------------
2015 // If there was no error then do nothing
2016 //--------------------------------------------------------------------------
2017 if( status.IsOK() )
2018 return;
2019
2020 if( pSidMgr && pMsgInFly && (
2021 status.code == errOperationExpired ||
2022 status.code == errOperationInterrupted ) )
2023 {
2024 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
2025 pSidMgr->TimeOutSID( req->header.streamid );
2026 }
2027
2028 bool noreplicas = ( status.code == errErrorResponse &&
2029 status.errNo == kXR_noReplicas );
2030
2031 if( !noreplicas ) pLastError = status;
2032
2033 Log *log = DefaultEnv::GetLog();
2034 log->Debug( XRootDMsg, "[%s] Handling error while processing %s: %s.",
2035 pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str(),
2036 status.ToString().c_str() );
2037
2038 //--------------------------------------------------------------------------
2039 // Check if it is a fatal TLS error that has been marked as potentially
2040 // recoverable, if yes check if we can downgrade from fatal to error.
2041 //--------------------------------------------------------------------------
2042 if( status.IsFatal() && status.code == errTlsError && status.errNo == EAGAIN )
2043 {
2044 if( pSslErrCnt < MaxSslErrRetry )
2045 {
2046 status.status &= ~stFatal; // switch off fatal&error bits
2047 status.status |= stError; // switch on error bit
2048 }
2049 ++pSslErrCnt; // count number of consecutive SSL errors
2050 }
2051 else
2052 pSslErrCnt = 0;
2053
2054 //--------------------------------------------------------------------------
2055 // We have got an error message, we can recover it at the load balancer if:
2056 // 1) we haven't got it from the load balancer
2057 // 2) we have a load balancer assigned
2058 // 3) the error is either one of: kXR_FSError, kXR_IOError, kXR_ServerError,
2059 // kXR_NotFound
2060 // 4) in the case of kXR_NotFound a kXR_refresh flags needs to be set
2061 //--------------------------------------------------------------------------
2062 if( status.code == errErrorResponse )
2063 {
2064 if( RetriableErrorResponse( status ) )
2065 {
2066 UpdateTriedCGI(status.errNo);
2067 if( status.errNo == kXR_NotFound || status.errNo == kXR_Overloaded )
2068 SwitchOnRefreshFlag();
2069 HandleError( RetryAtServer( pLoadBalancer.url, RedirectEntry::EntryRetry ) );
2070 return;
2071 }
2072 else
2073 {
2074 pStatus = status;
2075 HandleRspOrQueue();
2076 return;
2077 }
2078 }
2079
2080 //--------------------------------------------------------------------------
2081 // Nothing can be done if:
2082 // 1) a user timeout has occurred
2083 // 2) has a non-zero session id
2084 // 3) if another error occurred and the validity of the message expired
2085 //--------------------------------------------------------------------------
2086 if( status.code == errOperationExpired || pRequest->GetSessionId() ||
2087 status.code == errOperationInterrupted || time(0) >= pExpiration )
2088 {
2089 log->Error( XRootDMsg, "[%s] Unable to get the response to request %s",
2090 pUrl.GetHostId().c_str(),
2091 pRequest->GetObfuscatedDescription().c_str() );
2092 pStatus = status;
2093 HandleRspOrQueue();
2094 return;
2095 }
2096
2097 //--------------------------------------------------------------------------
2098 // At this point we're left with connection errors, we recover them
2099 // at a load balancer if we have one and if not on the current server
2100 // until we get a response, an unrecoverable error or a timeout
2101 //--------------------------------------------------------------------------
2102 if( pLoadBalancer.url.IsValid() &&
2103 pLoadBalancer.url.GetLocation() != pUrl.GetLocation() )
2104 {
2105 UpdateTriedCGI( kXR_ServerError );
2106 HandleError( RetryAtServer( pLoadBalancer.url, RedirectEntry::EntryRetry ) );
2107 return;
2108 }
2109 else
2110 {
2111 if( !status.IsFatal() && IsRetriable() )
2112 {
2113 log->Info( XRootDMsg, "[%s] Retrying request: %s.",
2114 pUrl.GetHostId().c_str(),
2115 pRequest->GetObfuscatedDescription().c_str() );
2116
2117 UpdateTriedCGI( kXR_ServerError );
2118 HandleError( RetryAtServer( pUrl, RedirectEntry::EntryRetry ) );
2119 return;
2120 }
2121 pStatus = status;
2122 HandleRspOrQueue();
2123 return;
2124 }
2125 }
2126
2127 //----------------------------------------------------------------------------
2128 // Retry the message at another server
2129 //----------------------------------------------------------------------------
2130 Status XRootDMsgHandler::RetryAtServer( const URL &url, RedirectEntry::Type entryType )
2131 {
2132 pResponse.reset();
2133 Log *log = DefaultEnv::GetLog();
2134
2135 //--------------------------------------------------------------------------
2136 // Set up a redirect entry
2137 //--------------------------------------------------------------------------
2138 if( pRdirEntry ) pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
2139 pRdirEntry.reset( new RedirectEntry( pUrl.GetLocation(), url.GetLocation(), entryType ) );
2140
2141 if( pUrl.GetLocation() != url.GetLocation() )
2142 {
2143 pHosts->push_back( url );
2144
2145 //------------------------------------------------------------------------
2146 // Assign a new stream id to the message
2147 //------------------------------------------------------------------------
2148
2149 // first release the old stream id
2150 // (though it could be a redirect from a local
2151 // metalink file, in this case there's no SID)
2152 ClientRequestHdr *req = (ClientRequestHdr*)pRequest->GetBuffer();
2153 if( pSidMgr )
2154 {
2155 pSidMgr->ReleaseSID( req->streamid );
2156 pSidMgr.reset();
2157 }
2158
2159 // then get the new SIDManager
2160 // (again this could be a redirect to a local
2161 // file and in this case there is no SID)
2162 if( !url.IsLocalFile() )
2163 {
2164 pSidMgr = SIDMgrPool::Instance().GetSIDMgr( url );
2165 Status st = pSidMgr->AllocateSID( req->streamid );
2166 if( !st.IsOK() )
2167 {
2168 log->Error( XRootDMsg, "[%s] Impossible to send message %s.",
2169 pUrl.GetHostId().c_str(),
2170 pRequest->GetObfuscatedDescription().c_str() );
2171 return st;
2172 }
2173 }
2174
2175 pUrl = url;
2176 }
2177
2178 if( pUrl.IsMetalink() && pFollowMetalink )
2179 {
2180 log->Debug( ExDbgMsg, "[%s] Metaling redirection for MsgHandler: %p (message: %s ).",
2181 pUrl.GetHostId().c_str(), this,
2182 pRequest->GetObfuscatedDescription().c_str() );
2183
2184 return pPostMaster->Redirect( pUrl, pRequest, this );
2185 }
2186 else if( pUrl.IsLocalFile() )
2187 {
2188 HandleLocalRedirect( &pUrl );
2189 return Status();
2190 }
2191 else
2192 {
2193 log->Debug( ExDbgMsg, "[%s] Retry at server MsgHandler: %p (message: %s ).",
2194 pUrl.GetHostId().c_str(), this,
2195 pRequest->GetObfuscatedDescription().c_str() );
2196 return pPostMaster->Send( pUrl, pRequest, this, true, pExpiration );
2197 }
2198 }
2199
2200 //----------------------------------------------------------------------------
2201 // Update the "tried=" part of the CGI of the current message
2202 //----------------------------------------------------------------------------
2203 void XRootDMsgHandler::UpdateTriedCGI(uint32_t errNo)
2204 {
2205 URL::ParamsMap cgi;
2206 std::string tried;
2207
2208 //--------------------------------------------------------------------------
2209 // In case a data server responded with a kXR_redirect and we fail at the
2210 // node where we were redirected to, the original data server should be
2211 // included in the tried CGI opaque info (instead of the current one).
2212 //--------------------------------------------------------------------------
2213 if( pEffectiveDataServerUrl )
2214 {
2215 tried = pEffectiveDataServerUrl->GetHostName();
2216 delete pEffectiveDataServerUrl;
2217 pEffectiveDataServerUrl = 0;
2218 }
2219 //--------------------------------------------------------------------------
2220 // Otherwise use the current URL.
2221 //--------------------------------------------------------------------------
2222 else
2223 tried = pUrl.GetHostName();
2224
2225 // Report the reason for the failure to the next location
2226 //
2227 if (errNo)
2228 { if (errNo == kXR_NotFound) cgi["triedrc"] = "enoent";
2229 else if (errNo == kXR_IOError) cgi["triedrc"] = "ioerr";
2230 else if (errNo == kXR_FSError) cgi["triedrc"] = "fserr";
2231 else if (errNo == kXR_ServerError) cgi["triedrc"] = "srverr";
2232 }
2233
2234 //--------------------------------------------------------------------------
2235 // If our current load balancer is a metamanager and we failed either
2236 // at a diskserver or at an unidentified node we also exclude the last
2237 // known manager
2238 //--------------------------------------------------------------------------
2239 if( pLoadBalancer.url.IsValid() && (pLoadBalancer.flags & kXR_attrMeta) )
2240 {
2241 HostList::reverse_iterator it;
2242 for( it = pHosts->rbegin()+1; it != pHosts->rend(); ++it )
2243 {
2244 if( it->loadBalancer )
2245 break;
2246
2247 tried += "," + it->url.GetHostName();
2248
2249 if( it->flags & kXR_isManager )
2250 break;
2251 }
2252 }
2253
2254 cgi["tried"] = tried;
2256 MessageUtils::RewriteCGIAndPath( pRequest, cgi, false, "" );
2258 }
2259
2260 //----------------------------------------------------------------------------
2261 // Switch on the refresh flag for some requests
2262 //----------------------------------------------------------------------------
2263 void XRootDMsgHandler::SwitchOnRefreshFlag()
2264 {
2266 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
2267 switch( req->header.requestid )
2268 {
2269 case kXR_locate:
2270 {
2271 req->locate.options |= kXR_refresh;
2272 break;
2273 }
2274
2275 case kXR_open:
2276 {
2277 req->locate.options |= kXR_refresh;
2278 break;
2279 }
2280 }
2283 }
2284
2285 //------------------------------------------------------------------------
2286 // If the current thread is a worker thread from our thread-pool
2287 // handle the response, otherwise submit a new task to the thread-pool
2288 //------------------------------------------------------------------------
2289 void XRootDMsgHandler::HandleRspOrQueue()
2290 {
2291 //--------------------------------------------------------------------------
2292 // Is it a final response?
2293 //--------------------------------------------------------------------------
2294 bool finalrsp = !( pStatus.IsOK() && pStatus.code == suContinue );
2295 if( finalrsp )
2296 {
2297 // Do not do final processing of the response if we haven't had
2298 // confirmation the original request was sent (via OnStatusReady).
2299 // The final processing will be triggered when we get the confirm.
2300 const int sst = pSendingState.fetch_or( kFinalResp );
2301 if( !( sst & kSendDone ) )
2302 return;
2303 }
2304
2305 JobManager *jobMgr = pPostMaster->GetJobManager();
2306 if( jobMgr->IsWorker() )
2307 HandleResponse();
2308 else
2309 {
2310 Log *log = DefaultEnv::GetLog();
2311 log->Debug( ExDbgMsg, "[%s] Passing to the thread-pool MsgHandler: %p (message: %s ).",
2312 pUrl.GetHostId().c_str(), this,
2313 pRequest->GetObfuscatedDescription().c_str() );
2314 jobMgr->QueueJob( new HandleRspJob( this ), 0 );
2315 }
2316 }
2317
2318 //------------------------------------------------------------------------
2319 // Notify the FileStateHandler to retry Open() with new URL
2320 //------------------------------------------------------------------------
2321 void XRootDMsgHandler::HandleLocalRedirect( URL *url )
2322 {
2323 Log *log = DefaultEnv::GetLog();
2324 log->Debug( ExDbgMsg, "[%s] Handling local redirect - MsgHandler: %p (message: %s ).",
2325 pUrl.GetHostId().c_str(), this,
2326 pRequest->GetObfuscatedDescription().c_str() );
2327
2328 if( !pLFileHandler )
2329 {
2330 HandleError( XRootDStatus( stFatal, errNotSupported ) );
2331 return;
2332 }
2333
2334 AnyObject *resp = 0;
2335 pLFileHandler->SetHostList( *pHosts );
2336 XRootDStatus st = pLFileHandler->Open( url, pRequest, resp );
2337 if( !st.IsOK() )
2338 {
2339 HandleError( st );
2340 return;
2341 }
2342
2343 pResponseHandler->HandleResponseWithHosts( new XRootDStatus(),
2344 resp,
2345 pHosts.release() );
2346 delete this;
2347
2348 return;
2349 }
2350
2351 //------------------------------------------------------------------------
2352 // Check if it is OK to retry this request
2353 //------------------------------------------------------------------------
2354 bool XRootDMsgHandler::IsRetriable()
2355 {
2356 std::string value;
2357 DefaultEnv::GetEnv()->GetString( "OpenRecovery", value );
2358 if( value == "true" ) return true;
2359
2360 // check if it is a mutable open (open + truncate or open + create)
2361 ClientRequest *req = reinterpret_cast<ClientRequest*>( pRequest->GetBuffer() );
2362 if( req->header.requestid == htons( kXR_open ) )
2363 {
2364 bool _mutable = ( req->open.options & htons( kXR_delete ) ) ||
2365 ( req->open.options & htons( kXR_new ) );
2366
2367 if( _mutable )
2368 {
2369 Log *log = DefaultEnv::GetLog();
2370 log->Debug( XRootDMsg,
2371 "[%s] Not allowed to retry open request (OpenRecovery disabled): %s.",
2372 pUrl.GetHostId().c_str(),
2373 pRequest->GetObfuscatedDescription().c_str() );
2374 // disallow retry if it is a mutable open
2375 return false;
2376 }
2377 }
2378
2379 return true;
2380 }
2381
2382 //------------------------------------------------------------------------
2383 // Check if for given request and Metalink redirector it is OK to omit
2384 // the kXR_wait and proceed straight to the next entry in the Metalink file
2385 //------------------------------------------------------------------------
2386 bool XRootDMsgHandler::OmitWait( Message &request, const URL &url )
2387 {
2388 // we can omit kXR_wait only if we have a Metalink redirector
2389 if( !url.IsMetalink() )
2390 return false;
2391
2392 // we can omit kXR_wait only for requests that can be redirected
2393 // (kXR_read is the only stateful request that can be redirected)
2394 ClientRequest *req = reinterpret_cast<ClientRequest*>( request.GetBuffer() );
2395 if( pStateful && req->header.requestid != kXR_read )
2396 return false;
2397
2398 // we can only omit kXR_wait if the Metalink redirect has more
2399 // replicas
2400 RedirectorRegistry &registry = RedirectorRegistry::Instance();
2401 VirtualRedirector *redirector = registry.Get( url );
2402
2403 // we need more than one server as the current one is not reflected
2404 // in tried CGI
2405 if( redirector->Count( request ) > 1 )
2406 return true;
2407
2408 return false;
2409 }
2410
2411 //------------------------------------------------------------------------
2412 // Checks if the given error returned by server is retriable.
2413 //------------------------------------------------------------------------
2414 bool XRootDMsgHandler::RetriableErrorResponse( const Status &status )
2415 {
2416 // we can only retry error response if we have a valid load-balancer and
2417 // it is not our current URL
2418 if( !( pLoadBalancer.url.IsValid() &&
2419 pUrl.GetLocation() != pLoadBalancer.url.GetLocation() ) )
2420 return false;
2421
2422 // following errors are retriable at any load-balancer
2423 if( status.errNo == kXR_FSError || status.errNo == kXR_IOError ||
2424 status.errNo == kXR_ServerError || status.errNo == kXR_NotFound ||
2425 status.errNo == kXR_Overloaded || status.errNo == kXR_NoMemory )
2426 return true;
2427
2428 // check if the load-balancer is a meta-manager, if yes there are
2429 // more errors that can be recovered
2430 if( !( pLoadBalancer.flags & kXR_attrMeta ) ) return false;
2431
2432 // those errors are retriable for meta-managers
2433 if( status.errNo == kXR_Unsupported || status.errNo == kXR_FileLocked )
2434 return true;
2435
2436 // in case of not-authorized error there is an imposed upper limit
2437 // on how many times we can retry this error
2438 if( status.errNo == kXR_NotAuthorized )
2439 {
2441 DefaultEnv::GetEnv()->GetInt( "NotAuthorizedRetryLimit", limit );
2442 bool ret = pNotAuthorizedCounter < limit;
2443 ++pNotAuthorizedCounter;
2444 if( !ret )
2445 {
2446 Log *log = DefaultEnv::GetLog();
2447 log->Error( XRootDMsg,
2448 "[%s] Reached limit of NotAuthorized retries!",
2449 pUrl.GetHostId().c_str() );
2450 }
2451 return ret;
2452 }
2453
2454 // check if the load-balancer is a virtual (metalink) redirector,
2455 // if yes there are even more errors that can be recovered
2456 if( !( pLoadBalancer.flags & kXR_attrVirtRdr ) ) return false;
2457
2458 // those errors are retriable for virtual (metalink) redirectors
2459 if( status.errNo == kXR_noserver || status.errNo == kXR_ArgTooLong )
2460 return true;
2461
2462 // otherwise it is a non-retriable error
2463 return false;
2464 }
2465
2466 //------------------------------------------------------------------------
2467 // Dump the redirect-trace-back into the log file
2468 //------------------------------------------------------------------------
2469 void XRootDMsgHandler::DumpRedirectTraceBack()
2470 {
2471 if( pRedirectTraceBack.empty() ) return;
2472
2473 std::stringstream sstrm;
2474
2475 sstrm << "Redirect trace-back:\n";
2476
2477 int counter = 0;
2478
2479 auto itr = pRedirectTraceBack.begin();
2480 sstrm << '\t' << counter << ". " << (*itr)->ToString() << '\n';
2481
2482 auto prev = itr;
2483 ++itr;
2484 ++counter;
2485
2486 for( ; itr != pRedirectTraceBack.end(); ++itr, ++prev, ++counter )
2487 sstrm << '\t' << counter << ". "
2488 << (*itr)->ToString( (*prev)->status.IsOK() ) << '\n';
2489
2490 int authlimit = DefaultNotAuthorizedRetryLimit;
2491 DefaultEnv::GetEnv()->GetInt( "NotAuthorizedRetryLimit", authlimit );
2492
2493 bool warn = !pStatus.IsOK() &&
2494 ( pStatus.code == errNotFound ||
2495 pStatus.code == errRedirectLimit ||
2496 ( pStatus.code == errAuthFailed && pNotAuthorizedCounter >= authlimit ) );
2497
2498 Log *log = DefaultEnv::GetLog();
2499 if( warn )
2500 log->Warning( XRootDMsg, "%s", sstrm.str().c_str() );
2501 else
2502 log->Debug( XRootDMsg, "%s", sstrm.str().c_str() );
2503 }
2504
2505 // Read data from buffer
2506 //------------------------------------------------------------------------
2507 template<typename T>
2508 Status XRootDMsgHandler::ReadFromBuffer( char *&buffer, size_t &buflen, T& result )
2509 {
2510 if( sizeof( T ) > buflen ) return Status( stError, errDataError );
2511
2512 memcpy(&result, buffer, sizeof(T));
2513
2514 buffer += sizeof( T );
2515 buflen -= sizeof( T );
2516
2517 return Status();
2518 }
2519
2520 //------------------------------------------------------------------------
2521 // Read a string from buffer
2522 //------------------------------------------------------------------------
2523 Status XRootDMsgHandler::ReadFromBuffer( char *&buffer, size_t &buflen, std::string &result )
2524 {
2525 Status status;
2526 char c = 0;
2527
2528 while( true )
2529 {
2530 if( !( status = ReadFromBuffer( buffer, buflen, c ) ).IsOK() )
2531 return status;
2532
2533 if( c == 0 ) break;
2534 result += c;
2535 }
2536
2537 return status;
2538 }
2539
2540 //------------------------------------------------------------------------
2541 // Read a string from buffer
2542 //------------------------------------------------------------------------
2543 Status XRootDMsgHandler::ReadFromBuffer( char *&buffer, size_t &buflen,
2544 size_t size, std::string &result )
2545 {
2546 Status status;
2547
2548 if( size > buflen ) return Status( stError, errDataError );
2549
2550 result.append( buffer, size );
2551 buffer += size;
2552 buflen -= size;
2553
2554 return status;
2555 }
2556
2557}
@ kXR_NotAuthorized
@ kXR_NotFound
@ kXR_FileLocked
Definition XProtocol.hh:993
@ kXR_noReplicas
@ kXR_Unsupported
@ kXR_ServerError
@ kXR_Overloaded
@ kXR_ArgTooLong
Definition XProtocol.hh:992
@ kXR_noserver
@ kXR_IOError
Definition XProtocol.hh:997
@ kXR_FSError
Definition XProtocol.hh:995
@ kXR_NoMemory
Definition XProtocol.hh:998
#define kXR_isManager
union ServerResponse::@0 body
@ kXR_fattrDel
Definition XProtocol.hh:270
@ kXR_fattrSet
Definition XProtocol.hh:273
@ kXR_fattrList
Definition XProtocol.hh:272
@ kXR_fattrGet
Definition XProtocol.hh:271
struct ClientFattrRequest fattr
Definition XProtocol.hh:854
#define kXR_collapseRedir
ServerResponseStatus status
#define kXR_attrMeta
kXR_char streamid[2]
Definition XProtocol.hh:156
kXR_char streamid[2]
Definition XProtocol.hh:914
kXR_unt16 options
Definition XProtocol.hh:481
struct ClientDirlistRequest dirlist
Definition XProtocol.hh:852
static const int kXR_ckpXeq
Definition XProtocol.hh:216
@ kXR_delete
Definition XProtocol.hh:453
@ kXR_refresh
Definition XProtocol.hh:459
@ kXR_new
Definition XProtocol.hh:455
@ kXR_retstat
Definition XProtocol.hh:463
struct ClientOpenRequest open
Definition XProtocol.hh:860
@ kXR_waitresp
Definition XProtocol.hh:906
@ kXR_redirect
Definition XProtocol.hh:904
@ kXR_oksofar
Definition XProtocol.hh:900
@ kXR_status
Definition XProtocol.hh:907
@ kXR_ok
Definition XProtocol.hh:899
@ kXR_attn
Definition XProtocol.hh:901
@ kXR_wait
Definition XProtocol.hh:905
@ kXR_error
Definition XProtocol.hh:903
struct ServerResponseBody_Status bdy
struct ClientRequestHdr header
Definition XProtocol.hh:846
#define kXR_recoverWrts
kXR_unt16 requestid
Definition XProtocol.hh:157
@ kXR_read
Definition XProtocol.hh:125
@ kXR_open
Definition XProtocol.hh:122
@ kXR_writev
Definition XProtocol.hh:143
@ kXR_readv
Definition XProtocol.hh:137
@ kXR_mkdir
Definition XProtocol.hh:120
@ kXR_sync
Definition XProtocol.hh:128
@ kXR_chmod
Definition XProtocol.hh:114
@ kXR_dirlist
Definition XProtocol.hh:116
@ kXR_fattr
Definition XProtocol.hh:132
@ kXR_rm
Definition XProtocol.hh:126
@ kXR_query
Definition XProtocol.hh:113
@ kXR_write
Definition XProtocol.hh:131
@ kXR_set
Definition XProtocol.hh:130
@ kXR_rmdir
Definition XProtocol.hh:127
@ kXR_truncate
Definition XProtocol.hh:140
@ kXR_protocol
Definition XProtocol.hh:118
@ kXR_mv
Definition XProtocol.hh:121
@ kXR_ping
Definition XProtocol.hh:123
@ kXR_stat
Definition XProtocol.hh:129
@ kXR_pgread
Definition XProtocol.hh:142
@ kXR_chkpoint
Definition XProtocol.hh:124
@ kXR_locate
Definition XProtocol.hh:139
@ kXR_close
Definition XProtocol.hh:115
@ kXR_pgwrite
Definition XProtocol.hh:138
@ kXR_prepare
Definition XProtocol.hh:133
#define kXR_isServer
#define kXR_attrVirtRdr
struct ClientChkPointRequest chkpoint
Definition XProtocol.hh:849
struct ServerResponseHeader hdr
union ServerResponseV2::@1 info
#define kXR_PROTOCOLVERSION
Definition XProtocol.hh:70
@ kXR_vfs
Definition XProtocol.hh:763
struct ClientStatRequest stat
Definition XProtocol.hh:873
#define kXR_ecRedir
struct ClientLocateRequest locate
Definition XProtocol.hh:856
ServerResponseHeader hdr
long long kXR_int64
Definition XPtypes.hh:98
int kXR_int32
Definition XPtypes.hh:89
unsigned short kXR_unt16
Definition XPtypes.hh:67
unsigned char kXR_char
Definition XPtypes.hh:65
#define unlikely(x)
std::string obfuscateAuth(const std::string &input)
void Get(Type &object)
Retrieve the object being held.
Object for reading out data from the PgRead response.
void AdvanceCursor(uint32_t delta)
Advance the cursor.
char * GetBufferAtCursor()
Get the buffer pointer at the append cursor.
const char * GetBuffer(uint32_t offset=0) const
Get the message buffer.
void SetCursor(uint32_t cursor)
Set the cursor.
uint32_t GetCursor() const
Get append cursor.
static Log * GetLog()
Get default log.
static Env * GetEnv()
Get default client environment.
static bool HasStatInfo(const char *data)
Returns true if data contain stat info.
bool GetString(const std::string &key, std::string &value)
Definition XrdClEnv.cc:31
bool GetInt(const std::string &key, int &value)
Definition XrdClEnv.cc:89
Interface for a job to be run by the job manager.
void SetHostList(const HostList &hostList)
XRootDStatus Open(const std::string &url, uint16_t flags, uint16_t mode, ResponseHandler *handler, uint16_t timeout=0)
Handle diagnostics.
Definition XrdClLog.hh:101
@ ErrorMsg
report errors
Definition XrdClLog.hh:109
void Error(uint64_t topic, const char *format,...)
Report an error.
Definition XrdClLog.cc:231
void Warning(uint64_t topic, const char *format,...)
Report a warning.
Definition XrdClLog.cc:248
void Dump(uint64_t topic, const char *format,...)
Print a dump message.
Definition XrdClLog.cc:299
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
Definition XrdClLog.cc:282
static void RewriteCGIAndPath(Message *msg, const URL::ParamsMap &newCgi, bool replace, const std::string &newPath)
Append cgi to the one already present in the message.
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.
@ More
there are more (non-raw) data to be read
@ Ignore
Ignore the message.
StreamEvent
Events that may have occurred to the stream.
@ Ready
The stream has become connected.
void CollapseRedirect(const URL &oldurl, const URL &newURL)
Collapse channel URL - replace the URL of the channel.
XRootDStatus Send(const URL &url, Message *msg, MsgHandler *handler, bool stateful, time_t expires)
TaskManager * GetTaskManager()
Get the task manager object user by the post master.
Status Redirect(const URL &url, Message *msg, MsgHandler *handler)
Status QueryTransport(const URL &url, uint16_t query, AnyObject &result)
JobManager * GetJobManager()
Get the job manager object user by the post master.
static RedirectorRegistry & Instance()
Returns reference to the single instance.
virtual void HandleResponseWithHosts(XRootDStatus *status, AnyObject *response, HostList *hostList)
static SIDMgrPool & Instance()
std::shared_ptr< SIDManager > GetSIDMgr(const URL &url)
A network socket.
virtual XRootDStatus Send(const char *buffer, size_t size, int &bytesWritten)
void RegisterTask(Task *task, time_t time, bool own=true)
Interface for a task to be run by the TaskManager.
virtual time_t Run(time_t now)=0
void SetName(const std::string &name)
Set name of the task.
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
bool IsMetalink() const
Is it a URL to a metalink.
Definition XrdClURL.cc:465
const std::string & GetPassword() const
Get the password.
Definition XrdClURL.hh:153
std::map< std::string, std::string > ParamsMap
Definition XrdClURL.hh:33
bool FromString(const std::string &url)
Parse a string and fill the URL fields.
Definition XrdClURL.cc:62
void SetPassword(const std::string &password)
Set the password.
Definition XrdClURL.hh:161
void SetParams(const std::string &params)
Set params.
Definition XrdClURL.cc:402
const std::string & GetUserName() const
Get the username.
Definition XrdClURL.hh:135
std::string GetURL() const
Get the URL.
Definition XrdClURL.hh:86
std::string GetLocation() const
Get location (protocol://host:port/path)
Definition XrdClURL.cc:344
const std::string & GetHostName() const
Get the name of the target host.
Definition XrdClURL.hh:170
bool IsLocalFile() const
Definition XrdClURL.cc:474
void SetProtocol(const std::string &protocol)
Set protocol.
Definition XrdClURL.hh:126
const ParamsMap & GetParams() const
Get the URL params.
Definition XrdClURL.hh:244
const std::string & GetProtocol() const
Get the protocol.
Definition XrdClURL.hh:118
bool IsValid() const
Is the url valid.
Definition XrdClURL.cc:452
void SetUserName(const std::string &userName)
Set the username.
Definition XrdClURL.hh:143
static void splitString(Container &result, const std::string &input, const std::string &delimiter)
Split a string.
Definition XrdClUtils.hh:56
static bool CheckEC(const Message *req, const URL &url)
Check if this client can support given EC redirect.
Handle/Process/Forward XRootD messages.
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.
const Message * GetRequest() const
Get the request pointer.
virtual uint16_t Examine(std::shared_ptr< Message > &msg) override
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
XRootDStatus WriteMessageBody(Socket *socket, uint32_t &bytesWritten) override
virtual uint8_t OnStreamEvent(StreamEvent event, XRootDStatus status) override
virtual uint16_t GetSid() const override
virtual bool IsRaw() const override
Are we a raw writer or not?
const std::string & GetErrorMessage() const
Get error message.
static void SetDescription(Message *msg)
Get the description of a message.
static XRootDStatus UnMarshallBody(Message *msg, uint16_t reqType)
Unmarshall the body of the incoming message.
static XRootDStatus UnMarshallRequest(Message *msg)
static XRootDStatus UnMarshalStatusBody(Message &msg, uint16_t reqType)
Unmarshall the body of the status response.
static XRootDStatus MarshallRequest(Message *msg)
Marshal the outgoing message.
static int csNum(off_t offs, int count)
Compute the required size of a checksum vector based on offset & length.
const uint16_t suRetry
const uint16_t errRedirectLimit
const int DefaultMaxMetalinkWait
const uint16_t errErrorResponse
const uint16_t errTlsError
const uint16_t errOperationExpired
const uint16_t stFatal
Fatal error, it's still an error.
const uint16_t stError
An error occurred that could potentially be retried.
const uint16_t errNotFound
const uint64_t XRootDMsg
std::vector< HostInfo > HostList
const uint16_t errDataError
data is corrupted
const uint16_t errInternal
Internal error.
const uint16_t stOK
Everything went OK.
const uint64_t ExDbgMsg
const uint16_t errInvalidResponse
const uint16_t errInvalidRedirectURL
const uint16_t errNotSupported
Buffer BinaryDataInfo
Binary buffer.
const uint16_t errOperationInterrupted
const uint16_t suContinue
const int DefaultNotAuthorizedRetryLimit
const uint16_t errRedirect
const uint16_t errAuthFailed
const uint16_t errInvalidMessage
none object for initializing empty Optional
XrdSysError Log
Definition XrdConfig.cc:113
@ kXR_PartialResult
static const int PageSize
ssize_t Move(KernelBuffer &kbuff, char *&ubuff)
Describe a data chunk for vector read.
void * buffer
length of the chunk
uint32_t length
offset in the file
URL url
URL of the host.
uint32_t flags
Host type.
Procedure execution status.
uint16_t code
Error type, or additional hints on what to do.
bool IsOK() const
We're fine.
std::string ToString() const
Create a string representation.
static const uint16_t ServerFlags
returns server flags
static const uint16_t ProtocolVersion
returns the protocol version