XRootD
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"
27 #include "XrdCl/XrdClDefaultEnv.hh"
28 #include "XrdCl/XrdClConstants.hh"
30 #include "XrdCl/XrdClMessage.hh"
31 #include "XrdCl/XrdClURL.hh"
32 #include "XrdCl/XrdClUtils.hh"
34 #include "XrdCl/XrdClJobManager.hh"
35 #include "XrdCl/XrdClSIDManager.hh"
39 #include "XrdCl/XrdClSocket.hh"
40 #include "XrdCl/XrdClTls.hh"
41 #include "XrdCl/XrdClOptimizers.hh"
42 
43 #include "XrdOuc/XrdOucCRC.hh"
45 
46 #include "XrdSys/XrdSysPlatform.hh" // same as above
47 #include "XrdSys/XrdSysAtomics.hh"
48 #include "XrdSys/XrdSysPthread.hh"
49 #include <memory>
50 #include <sstream>
51 #include <numeric>
52 
53 namespace
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 
79 namespace XrdCl
80 {
81  //----------------------------------------------------------------------------
82  // Delegate the response handling to the thread-pool
83  //----------------------------------------------------------------------------
84  class HandleRspJob: public XrdCl::Job
85  {
86  public:
87  HandleRspJob( XrdCl::XRootDMsgHandler *handler ): pHandler( handler )
88  {
89 
90  }
91 
92  virtual ~HandleRspJob()
93  {
94 
95  }
96 
97  virtual void Run( void *arg )
98  {
99  pHandler->HandleResponse();
100  delete this;
101  }
102  private:
103  XrdCl::XRootDMsgHandler *pHandler;
104  };
105 
106  //----------------------------------------------------------------------------
107  // Examine an incoming message, and decide on the action to be taken
108  //----------------------------------------------------------------------------
109  uint16_t XRootDMsgHandler::Examine( std::shared_ptr<Message> &msg )
110  {
111  const int sst = pSendingState.fetch_or( kSawResp );
112 
113  if( !( sst & kSendDone ) && !( sst & kSawResp ) )
114  {
115  // we must have been sent although we haven't got the OnStatusReady
116  // notification yet. Set the inflight notice.
117 
118  Log *log = DefaultEnv::GetLog();
119  log->Dump( XRootDMsg, "[%s] Message %s reply received before notification "
120  "that it was sent, assuming it was sent ok.",
121  pUrl.GetHostId().c_str(),
122  pRequest->GetObfuscatedDescription().c_str() );
123  }
124 
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  //----------------------------------------------------------------------------
403  uint16_t XRootDMsgHandler::GetSid() const
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  {
647  pStatus = Status( stError, errInvalidRedirectURL );
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 
2016  XRootDTransport::SetDescription( pRequest );
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  }
2329  XRootDTransport::SetDescription( pRequest );
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  {
2488  int limit = DefaultNotAuthorizedRetryLimit;
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
Definition: XProtocol.hh:1042
@ kXR_NotFound
Definition: XProtocol.hh:1043
@ kXR_FileLocked
Definition: XProtocol.hh:1035
@ kXR_noReplicas
Definition: XProtocol.hh:1061
@ kXR_Unsupported
Definition: XProtocol.hh:1045
@ kXR_ServerError
Definition: XProtocol.hh:1044
@ kXR_Overloaded
Definition: XProtocol.hh:1056
@ kXR_ArgTooLong
Definition: XProtocol.hh:1034
@ kXR_noserver
Definition: XProtocol.hh:1046
@ kXR_IOError
Definition: XProtocol.hh:1039
@ kXR_FSError
Definition: XProtocol.hh:1037
@ kXR_NoMemory
Definition: XProtocol.hh:1040
#define kXR_isManager
Definition: XProtocol.hh:1198
union ServerResponse::@0 body
@ 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
Definition: XProtocol.hh:1209
ServerResponseStatus status
Definition: XProtocol.hh:1352
#define kXR_attrMeta
Definition: XProtocol.hh:1201
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
Definition: XProtocol.hh:1304
struct ClientRequestHdr header
Definition: XProtocol.hh:887
#define kXR_recoverWrts
Definition: XProtocol.hh:1208
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
Definition: XProtocol.hh:1199
#define kXR_attrVirtRdr
Definition: XProtocol.hh:1204
struct ClientChkPointRequest chkpoint
Definition: XProtocol.hh:890
struct ServerResponseHeader hdr
Definition: XProtocol.hh:1303
union ServerResponseV2::@1 info
#define kXR_PROTOCOLVERSION
Definition: XProtocol.hh:70
@ kXR_vfs
Definition: XProtocol.hh:799
struct ClientStatRequest stat
Definition: XProtocol.hh:915
kXR_char options
Definition: XProtocol.hh:809
#define kXR_ecRedir
Definition: XProtocol.hh:1210
struct ClientLocateRequest locate
Definition: XProtocol.hh:898
ServerResponseHeader hdr
Definition: XProtocol.hh:1330
long long kXR_int64
Definition: XPtypes.hh:98
int kXR_int32
Definition: XPtypes.hh:89
unsigned short kXR_unt16
Definition: XPtypes.hh:67
unsigned char kXR_char
Definition: XPtypes.hh:65
#define unlikely(x)
std::string obfuscateAuth(const std::string &input)
void Get(Type &object)
Retrieve the object being held.
Object for reading out data from the PgRead response.
void AdvanceCursor(uint32_t delta)
Advance the cursor.
Definition: XrdClBuffer.hh:156
const char * GetBuffer(uint32_t offset=0) const
Get the message buffer.
Definition: XrdClBuffer.hh:72
void SetCursor(uint32_t cursor)
Set the cursor.
Definition: XrdClBuffer.hh:148
uint32_t GetCursor() const
Get append cursor.
Definition: XrdClBuffer.hh:140
char * GetBufferAtCursor()
Get the buffer pointer at the append cursor.
Definition: XrdClBuffer.hh:189
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.
void SetHostList(const HostList &hostList)
XRootDStatus Open(const std::string &url, uint16_t flags, uint16_t mode, ResponseHandler *handler, time_t timeout=0)
Handle diagnostics.
Definition: XrdClLog.hh:101
@ ErrorMsg
report errors
Definition: XrdClLog.hh:109
void Error(uint64_t topic, const char *format,...)
Report an error.
Definition: XrdClLog.cc:231
void Warning(uint64_t topic, const char *format,...)
Report a warning.
Definition: XrdClLog.cc:248
void Dump(uint64_t topic, const char *format,...)
Print a dump message.
Definition: XrdClLog.cc:299
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
Definition: XrdClLog.cc:282
static void RewriteCGIAndPath(Message *msg, const URL::ParamsMap &newCgi, bool replace, const std::string &newPath, std::string *opathp=nullptr)
Append cgi to the one already present in the message.
The message representation used throughout the system.
Definition: XrdClMessage.hh:32
const std::string & GetObfuscatedDescription() const
Get the description of the message with authz parameter obfuscated.
uint64_t GetSessionId() const
Get the session ID the message is meant for.
@ More
there are more (non-raw) data to be read
@ Ignore
Ignore the message.
StreamEvent
Events that may have occurred to the stream.
@ Ready
The stream has become connected.
void CollapseRedirect(const URL &oldurl, const URL &newURL)
Collapse channel URL - replace the URL of the channel.
XRootDStatus Send(const URL &url, Message *msg, MsgHandler *handler, bool stateful, time_t expires)
TaskManager * GetTaskManager()
Get the task manager object user by the post master.
Status Redirect(const URL &url, Message *msg, MsgHandler *handler)
Status QueryTransport(const URL &url, uint16_t query, AnyObject &result)
JobManager * GetJobManager()
Get the job manager object user by the post master.
static RedirectorRegistry & Instance()
Returns reference to the single instance.
virtual void HandleResponseWithHosts(XRootDStatus *status, AnyObject *response, HostList *hostList)
static SIDMgrPool & Instance()
std::shared_ptr< SIDManager > GetSIDMgr(const URL &url)
A network socket.
Definition: XrdClSocket.hh:43
virtual XRootDStatus Send(const char *buffer, size_t size, int &bytesWritten)
Definition: XrdClSocket.cc:461
bool IsEncrypted()
Definition: XrdClSocket.cc:867
void RegisterTask(Task *task, time_t time, bool own=true)
Interface for a task to be run by the TaskManager.
virtual time_t Run(time_t now)=0
void SetName(const std::string &name)
Set name of the task.
URL representation.
Definition: XrdClURL.hh:31
std::string GetHostId() const
Get the host part of the URL (user:password@host:port)
Definition: XrdClURL.hh:99
bool IsMetalink() const
Is it a URL to a metalink.
Definition: XrdClURL.cc:465
const std::string & GetHostName() const
Get the name of the target host.
Definition: XrdClURL.hh:170
std::map< std::string, std::string > ParamsMap
Definition: XrdClURL.hh:33
void SetPassword(const std::string &password)
Set the password.
Definition: XrdClURL.hh:161
const std::string & GetProtocol() const
Get the protocol.
Definition: XrdClURL.hh:118
void SetParams(const std::string &params)
Set params.
Definition: XrdClURL.cc:402
std::string GetURL() const
Get the URL.
Definition: XrdClURL.hh:86
std::string GetLocation() const
Get location (protocol://host:port/path)
Definition: XrdClURL.cc:344
const std::string & GetUserName() const
Get the username.
Definition: XrdClURL.hh:135
void SetPath(const std::string &path)
Set the path.
Definition: XrdClURL.hh:225
const std::string & GetPassword() const
Get the password.
Definition: XrdClURL.hh:153
bool IsLocalFile() const
Definition: XrdClURL.cc:474
const ParamsMap & GetParams() const
Get the URL params.
Definition: XrdClURL.hh:244
void SetProtocol(const std::string &protocol)
Set protocol.
Definition: XrdClURL.hh:126
const std::string & GetPath() const
Get the path.
Definition: XrdClURL.hh:217
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.
Definition: XrdClUtils.cc:703
Handle/Process/Forward XRootD messages.
const Message * GetRequest() const
Get the request pointer.
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.
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
Definition: XrdClStatus.hh:40
const uint16_t errRedirectLimit
Definition: XrdClStatus.hh:102
const int DefaultMaxMetalinkWait
const uint16_t errErrorResponse
Definition: XrdClStatus.hh:105
const uint16_t errTlsError
Definition: XrdClStatus.hh:80
const uint16_t errOperationExpired
Definition: XrdClStatus.hh:90
const uint16_t stFatal
Fatal error, it's still an error.
Definition: XrdClStatus.hh:33
const uint16_t stError
An error occurred that could potentially be retried.
Definition: XrdClStatus.hh:32
const uint16_t errNotFound
Definition: XrdClStatus.hh:100
const uint64_t XRootDMsg
std::vector< HostInfo > HostList
const uint16_t errDataError
data is corrupted
Definition: XrdClStatus.hh:63
const uint16_t errInternal
Internal error.
Definition: XrdClStatus.hh:56
const uint16_t stOK
Everything went OK.
Definition: XrdClStatus.hh:31
const uint64_t ExDbgMsg
const uint16_t errInvalidResponse
Definition: XrdClStatus.hh:99
const uint16_t errInvalidRedirectURL
Definition: XrdClStatus.hh:98
const uint16_t errNotSupported
Definition: XrdClStatus.hh:62
const uint16_t errLocalError
Definition: XrdClStatus.hh:107
Buffer BinaryDataInfo
Binary buffer.
const uint16_t errOperationInterrupted
Definition: XrdClStatus.hh:91
const uint16_t suContinue
Definition: XrdClStatus.hh:39
const int DefaultNotAuthorizedRetryLimit
const uint16_t errRedirect
Definition: XrdClStatus.hh:106
const uint16_t errAuthFailed
Definition: XrdClStatus.hh:88
const uint16_t errInvalidMessage
Definition: XrdClStatus.hh:85
none object for initializing empty Optional
XrdSysError Log
Definition: XrdConfig.cc:113
@ kXR_PartialResult
Definition: XProtocol.hh:1293
static const int PageSize
ssize_t Move(KernelBuffer &kbuff, char *&ubuff)
Describe a data chunk for vector read.
void * buffer
length of the chunk
uint32_t length
offset in the file
URL url
URL of the host.
uint32_t flags
Host type.
Procedure execution status.
Definition: XrdClStatus.hh:115
uint16_t code
Error type, or additional hints on what to do.
Definition: XrdClStatus.hh:147
bool IsOK() const
We're fine.
Definition: XrdClStatus.hh:124
std::string ToString() const
Create a string representation.
Definition: XrdClStatus.cc:97
static const uint16_t ServerFlags
returns server flags
static const uint16_t ProtocolVersion
returns the protocol version