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