62 WaitTask( XrdCl::XRootDMsgHandler *handler ): pHandler( handler )
65 o <<
"WaitTask for: 0x" << handler->
GetRequest();
69 virtual time_t Run( time_t now )
71 pHandler->WaitDone( now );
75 XrdCl::XRootDMsgHandler *pHandler;
97 virtual void Run(
void *arg )
99 pHandler->HandleResponse();
115 if( pOksofarAsAnswer )
118 while( pResponse ) pCV.Wait();
125 log->
Warning(
ExDbgMsg,
"[%s] MsgHandler is examining a response although "
126 "it already owns a response: %p (message: %s ).",
127 pUrl.GetHostId().c_str(),
this,
128 pRequest->GetObfuscatedDescription().c_str() );
132 if( msg->GetSize() < 8 )
172 pBodyReader->SetDataLength( dlen );
188 "message %s", pUrl.GetHostId().c_str(),
189 pRequest->GetObfuscatedDescription().c_str() );
229 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_oksofar response to request "
230 "%s", pUrl.GetHostId().c_str(),
231 pRequest->GetObfuscatedDescription().c_str() );
233 if( !pOksofarAsAnswer )
235 pPartialResps.emplace_back( std::move( pResponse ) );
246 pTimeoutFence.store(
true, std::memory_order_relaxed );
255 pTimeoutFence.store(
true, std::memory_order_relaxed );
264 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_status response to request "
265 "%s", pUrl.GetHostId().c_str(),
266 pRequest->GetObfuscatedDescription().c_str() );
283 pTimeoutFence.store(
true, std::memory_order_relaxed );
317 log->
Error(
XRootDMsg,
"[%s] kXR_status: invalid message size.", pUrl.GetHostId().c_str() );
338 pUrl.GetHostId().c_str() );
350 pPartialResps.push_back( std::move( pResponse ) );
363 pPageReader.reset(
new AsyncPageReader( *pChunkList, pCrc32cDigests ) );
364 pPageReader->SetRsp( rspst );
379 pResponse->GetCursor() )
409 if( pUrl.IsLocalFile() && pUrl.IsMetalink() )
418 int *qryResponse = 0;
420 qryResult.
Get( qryResponse );
421 pHosts->back().flags = *qryResponse;
delete qryResponse; qryResponse = 0;
423 qryResult.
Get( qryResponse );
424 pHosts->back().protocol = *qryResponse;
delete qryResponse;
448 pAggregatedWaitTime = 0;
457 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_ok response to request %s",
458 pUrl.GetHostId().c_str(),
459 pRequest->GetObfuscatedDescription().c_str() );
467 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_status response to request %s",
468 pUrl.GetHostId().c_str(),
469 pRequest->GetObfuscatedDescription().c_str() );
480 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_oksofar response to request %s",
481 pUrl.GetHostId().c_str(),
482 pRequest->GetObfuscatedDescription().c_str() );
493 char *errmsg =
new char[rsp->
hdr.
dlen-3]; errmsg[rsp->
hdr.
dlen-4] = 0;
494 memcpy( errmsg, rsp->
body.error.errmsg, rsp->
hdr.
dlen-4 );
495 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_error response to request %s "
496 "[%d] %s", pUrl.GetHostId().c_str(),
497 pRequest->GetObfuscatedDescription().c_str(), rsp->
body.error.errnum,
513 pUrl.GetHostId().c_str() );
519 char *urlInfoBuff =
new char[rsp->
hdr.
dlen-3];
520 urlInfoBuff[rsp->
hdr.
dlen-4] = 0;
521 memcpy( urlInfoBuff, rsp->
body.redirect.host, rsp->
hdr.
dlen-4 );
522 std::string urlInfo = urlInfoBuff;
523 delete [] urlInfoBuff;
525 "message %s: %s, port %d", pUrl.GetHostId().c_str(),
526 pRequest->GetObfuscatedDescription().c_str(), urlInfo.c_str(),
527 rsp->
body.redirect.port );
532 if( !pRedirectCounter )
535 "message %s, the last known error is: %s",
536 pUrl.GetHostId().c_str(),
537 pRequest->GetObfuscatedDescription().c_str(),
538 pLastError.ToString().c_str() );
551 uint32_t flags = pHosts->back().flags;
552 if( !pHasLoadBalancer )
561 if( ( flags &
kXR_attrMeta ) || !pLoadBalancer.url.IsValid() )
563 pLoadBalancer = pHosts->back();
564 log->
Dump(
XRootDMsg,
"[%s] Current server has been assigned "
565 "as a load-balancer for message %s",
566 pUrl.GetHostId().c_str(),
567 pRequest->GetObfuscatedDescription().c_str() );
568 HostList::iterator it;
569 for( it = pHosts->begin(); it != pHosts->end(); ++it )
570 it->loadBalancer =
false;
571 pHosts->back().loadBalancer =
true;
582 pEffectiveDataServerUrl =
new URL( pHosts->back().url );
587 std::vector<std::string> urlComponents;
591 std::ostringstream o;
593 o << urlComponents[0];
594 if( rsp->
body.redirect.port > 0 )
595 o <<
":" << rsp->
body.redirect.port <<
"/";
596 else if( rsp->
body.redirect.port < 0 )
611 std::string url( rsp->
body.redirect.host, rsp->
hdr.
dlen-4 );
612 pPostMaster->CollapseRedirect( pUrl, url );
617 std::string url( rsp->
body.redirect.host, rsp->
hdr.
dlen-4 );
619 pRedirectAsAnswer =
true;
623 URL newUrl =
URL( o.str() );
628 pUrl.GetHostId().c_str(), urlInfo.c_str() );
633 if( pUrl.GetUserName() !=
"" && newUrl.
GetUserName() ==
"" )
636 if( pUrl.GetPassword() !=
"" && newUrl.
GetPassword() ==
"" )
645 std::ostringstream ossXrd;
648 for(URL::ParamsMap::const_iterator it = urlParams.begin();
649 it != urlParams.end(); ++it )
651 if( it->first.compare( 0, 4,
"xrd." ) &&
652 it->first.compare( 0, 6,
"xrdcl." ) )
655 ossXrd << it->first <<
'=' << it->second <<
'&';
658 std::string xrdCgi = ossXrd.str();
659 pRedirectUrl = newUrl.
GetURL();
662 if( urlComponents.size() > 1 )
665 pRedirectUrl += urlComponents[1];
666 std::ostringstream o;
667 o <<
"fake://fake:111//fake?";
668 o << urlComponents[1];
670 if( urlComponents.size() == 3 )
671 o <<
'?' << urlComponents[2];
677 pRedirectUrl += xrdCgi;
680 cgiURL =
URL( o.str() );
685 std::ostringstream o;
686 o <<
"fake://fake:111//fake?";
688 cgiURL =
URL( o.str() );
690 pRedirectUrl += xrdCgi;
700 pRedirectAsAnswer =
true;
702 if( pRedirectAsAnswer )
713 Status st = RewriteRequestRedirect( newUrl );
724 if( ( pUrl.GetProtocol() ==
"roots" || pUrl.GetProtocol() ==
"xroots" ) &&
740 uint32_t waitSeconds = 0;
744 char *infoMsg =
new char[rsp->
hdr.
dlen-3];
746 memcpy( infoMsg, rsp->
body.wait.infomsg, rsp->
hdr.
dlen-4 );
747 log->
Dump(
XRootDMsg,
"[%s] Got kXR_wait response of %d seconds to "
748 "message %s: %s", pUrl.GetHostId().c_str(),
749 rsp->
body.wait.seconds, pRequest->GetObfuscatedDescription().c_str(),
752 waitSeconds = rsp->
body.wait.seconds;
756 log->
Dump(
XRootDMsg,
"[%s] Got kXR_wait response of 0 seconds to "
757 "message %s", pUrl.GetHostId().c_str(),
758 pRequest->GetObfuscatedDescription().c_str() );
761 pAggregatedWaitTime += waitSeconds;
766 if( OmitWait( *pRequest, pLoadBalancer.url ) )
770 if( pAggregatedWaitTime > maxWait )
782 Status st = RewriteRequestWait();
794 time_t resendTime = ::time(0)+waitSeconds;
796 if( resendTime < pExpiration )
798 log->
Debug(
ExDbgMsg,
"[%s] Scheduling WaitTask for MsgHandler: %p (message: %s ).",
799 pUrl.GetHostId().c_str(),
this,
800 pRequest->GetObfuscatedDescription().c_str() );
802 TaskManager *taskMgr = pPostMaster->GetTaskManager();
803 taskMgr->
RegisterTask(
new WaitTask(
this ), resendTime );
807 log->
Debug(
XRootDMsg,
"[%s] Wait time is too long, timing out %s",
808 pUrl.GetHostId().c_str(),
809 pRequest->GetObfuscatedDescription().c_str() );
826 pUrl.GetHostId().c_str() );
832 log->
Dump(
XRootDMsg,
"[%s] Got kXR_waitresp response of %d seconds to "
833 "message %s", pUrl.GetHostId().c_str(),
834 rsp->
body.waitresp.seconds,
835 pRequest->GetObfuscatedDescription().c_str() );
844 log->
Dump(
XRootDMsg,
"[%s] Got unrecognized response %d to "
845 "message %s", pUrl.GetHostId().c_str(),
846 rsp->
hdr.
status, pRequest->GetObfuscatedDescription().c_str() );
863 log->
Dump(
XRootDMsg,
"[%s] Stream event reported for msg %s",
864 pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str() );
869 if( pTimeoutFence.load( std::memory_order_relaxed ) )
872 HandleError( status );
881 uint32_t &bytesRead )
887 return pPageReader->Read( *socket, bytesRead );
889 return pBodyReader->Read( *socket, bytesRead );
906 log->
Dump(
XRootDMsg,
"[%s] Message %s has been successfully sent.",
909 log->
Debug(
ExDbgMsg,
"[%s] Moving MsgHandler: %p (message: %s ) from out-queue to in-queue.",
910 pUrl.GetHostId().c_str(),
this,
911 pRequest->GetObfuscatedDescription().c_str() );
920 log->
Error(
XRootDMsg,
"[%s] Impossible to send message %s. Trying to "
921 "recover.", pUrl.GetHostId().c_str(),
923 HandleError( status );
950 uint32_t &bytesWritten )
955 if( !pChunkList->empty() && !pCrc32cDigests.empty() )
965 int fLen = 0, lLen = 0;
971 if( pPgWrtCksumBuff.GetCursor() == 0 )
973 uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
974 memcpy( pPgWrtCksumBuff.GetBuffer(), &digest,
sizeof( uint32_t ) );
977 uint32_t btsLeft = chunk.
length - pAsyncOffset;
978 uint32_t pglen = ( pPgWrtCurrentPageNb == 0 ? fLen :
XrdSys::PageSize ) - pPgWrtCurrentPageOffset;
979 if( pglen > btsLeft ) pglen = btsLeft;
980 char* pgbuf =
static_cast<char*
>( chunk.
buffer ) + pAsyncOffset;
985 while( pPgWrtCksumBuff.GetCursor() <
sizeof( uint32_t ) )
987 uint32_t dgstlen =
sizeof( uint32_t ) - pPgWrtCksumBuff.GetCursor();
988 char* dgstbuf = pPgWrtCksumBuff.GetBufferAtCursor();
990 Status st = socket->
Send( dgstbuf, dgstlen, btswrt );
991 if( !st.
IsOK() )
return st;
992 bytesWritten += btswrt;
993 pPgWrtCksumBuff.AdvanceCursor( btswrt );
998 Status st = socket->
Send( pgbuf, pglen, btswrt );
999 if( !st.
IsOK() )
return st;
1003 bytesWritten += btswrt;
1004 pAsyncOffset += btswrt;
1010 ++pPgWrtCurrentPageNb;
1011 if( pPgWrtCurrentPageNb < nbpgs )
1014 pPgWrtCksumBuff.SetCursor( 0 );
1015 uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
1016 memcpy( pPgWrtCksumBuff.GetBuffer(), &digest,
sizeof( uint32_t ) );
1020 if( pglen > btsLeft ) pglen = btsLeft;
1022 pPgWrtCurrentPageOffset = 0;
1026 pPgWrtCurrentPageOffset += btswrt;
1030 else if( !pChunkList->empty() )
1032 size_t size = pChunkList->size();
1033 for(
size_t i = pAsyncChunkIndex ; i < size; ++i )
1035 char *buffer = (
char*)(*pChunkList)[i].buffer;
1036 uint32_t size = (*pChunkList)[i].length;
1037 size_t leftToBeWritten = size - pAsyncOffset;
1039 while( leftToBeWritten )
1042 Status st = socket->
Send( buffer + pAsyncOffset, leftToBeWritten, btswrt );
1043 bytesWritten += btswrt;
1045 pAsyncOffset += btswrt;
1046 leftToBeWritten -= btswrt;
1066 log->
Debug(
XRootDMsg,
"[%s] Channel is encrypted: cannot use kernel buffer.",
1067 pUrl.GetHostId().c_str() );
1072 pChunkList->push_back(
ChunkInfo( 0, ret, ubuff ) );
1079 while( !pKBuff->Empty() )
1083 bytesWritten += btswrt;
1087 log->
Debug(
XRootDMsg,
"[%s] Request %s payload (kernel buffer) transferred to socket.",
1088 pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str() );
1108 pTimeoutFence.store(
false, std::memory_order_relaxed );
1114 void XRootDMsgHandler::HandleResponse()
1124 log->
Debug(
ExDbgMsg,
"[%s] Calling MsgHandler: %p (message: %s ) "
1130 if( status->
IsOK() )
1132 Status st = ParseResponse( response );
1147 pRdirEntry->status = *status;
1148 pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
1154 bool finalrsp = !( pStatus.IsOK() && pStatus.code ==
suContinue );
1159 if( pSidMgr && finalrsp )
1161 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1162 if( status->
IsOK() || !pMsgInFly ||
1167 HostList *hosts = pHosts.release();
1169 pHosts.reset(
new HostList( *hosts ) );
1171 pResponseHandler->HandleResponseWithHosts( status, response, hosts );
1184 XrdSysCondVarHelper lck( pCV );
1186 pTimeoutFence.store(
false, std::memory_order_relaxed );
1197 XRootDStatus *st =
new XRootDStatus( pStatus );
1198 ServerResponse *rsp = 0;
1200 rsp = (ServerResponse *)pResponse->GetBuffer();
1202 if( !pStatus.IsOK() && rsp )
1206 st->errNo = rsp->
body.error.errnum;
1209 std::string errmsg( rsp->
body.error.errmsg, rsp->
hdr.
dlen-5 );
1211 errmsg +=
" Last seen error: " + pLastError.ToString();
1212 st->SetErrorMessage( errmsg );
1215 st->SetErrorMessage( pRedirectUrl );
1229 ServerResponse *rsp = (ServerResponse *)pResponse->GetBuffer();
1230 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1238 log->Error(
XRootDMsg,
"Internal Error: unable to process redirect" );
1243 uint32_t length = 0;
1249 if( pPartialResps.empty() )
1251 buffer = rsp->
body.buffer.data;
1260 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1262 ServerResponse *part = (ServerResponse*)pPartialResps[i]->GetBuffer();
1267 buff.Allocate( length );
1268 uint32_t offset = 0;
1269 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1271 ServerResponse *part = (ServerResponse*)pPartialResps[i]->GetBuffer();
1272 buff.Append( part->
body.buffer.data, part->
hdr.
dlen, offset );
1275 buff.Append( rsp->
body.buffer.data, rsp->
hdr.
dlen, offset );
1276 buffer = buff.GetBuffer();
1307 AnyObject *obj =
new AnyObject();
1309 char *nullBuffer =
new char[length+1];
1310 nullBuffer[length] = 0;
1311 memcpy( nullBuffer, buffer, length );
1313 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as "
1314 "LocateInfo: %s", pUrl.GetHostId().c_str(),
1315 pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
1316 LocationInfo *data =
new LocationInfo();
1318 if( data->ParseServerResponse( nullBuffer ) ==
false )
1322 delete [] nullBuffer;
1325 delete [] nullBuffer;
1337 AnyObject *obj =
new AnyObject();
1344 StatInfoVFS *data =
new StatInfoVFS();
1346 char *nullBuffer =
new char[length+1];
1347 nullBuffer[length] = 0;
1348 memcpy( nullBuffer, buffer, length );
1350 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as "
1351 "StatInfoVFS: %s", pUrl.GetHostId().c_str(),
1352 pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
1354 if( data->ParseServerResponse( nullBuffer ) ==
false )
1358 delete [] nullBuffer;
1361 delete [] nullBuffer;
1370 StatInfo *data =
new StatInfo();
1372 char *nullBuffer =
new char[length+1];
1373 nullBuffer[length] = 0;
1374 memcpy( nullBuffer, buffer, length );
1376 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as StatInfo: "
1377 "%s", pUrl.GetHostId().c_str(),
1378 pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
1380 if( data->ParseServerResponse( nullBuffer ) ==
false )
1384 delete [] nullBuffer;
1387 delete [] nullBuffer;
1400 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as ProtocolInfo",
1401 pUrl.GetHostId().c_str(),
1402 pRequest->GetObfuscatedDescription().c_str() );
1406 log->Error(
XRootDMsg,
"[%s] Got invalid redirect response.",
1407 pUrl.GetHostId().c_str() );
1411 AnyObject *obj =
new AnyObject();
1412 ProtocolInfo *data =
new ProtocolInfo( rsp->
body.protocol.pval,
1413 rsp->
body.protocol.flags );
1424 AnyObject *obj =
new AnyObject();
1425 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as "
1426 "DirectoryList", pUrl.GetHostId().c_str(),
1427 pRequest->GetObfuscatedDescription().c_str() );
1431 memcpy( path, pRequest->GetBuffer(24), req->
dirlist.
dlen );
1433 DirectoryList *data =
new DirectoryList();
1434 data->SetParentName( path );
1437 char *nullBuffer =
new char[length+1];
1438 nullBuffer[length] = 0;
1439 memcpy( nullBuffer, buffer, length );
1441 bool invalidrsp =
false;
1443 if( !pDirListStarted )
1446 pDirListStarted =
true;
1448 invalidrsp = !data->ParseServerResponse( pUrl.GetHostId(), nullBuffer );
1451 invalidrsp = !data->ParseServerResponse( pUrl.GetHostId(), nullBuffer, pDirListWithStat );
1457 delete [] nullBuffer;
1461 delete [] nullBuffer;
1472 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as OpenInfo",
1473 pUrl.GetHostId().c_str(),
1474 pRequest->GetObfuscatedDescription().c_str() );
1478 log->Error(
XRootDMsg,
"[%s] Got invalid open response.",
1479 pUrl.GetHostId().c_str() );
1483 AnyObject *obj =
new AnyObject();
1484 StatInfo *statInfo = 0;
1491 log->Dump(
XRootDMsg,
"[%s] Parsing StatInfo in response to %s",
1492 pUrl.GetHostId().c_str(),
1493 pRequest->GetObfuscatedDescription().c_str() );
1497 char *nullBuffer =
new char[rsp->
hdr.
dlen-11];
1498 nullBuffer[rsp->
hdr.
dlen-12] = 0;
1499 memcpy( nullBuffer, buffer+12, rsp->
hdr.
dlen-12 );
1501 statInfo =
new StatInfo();
1502 if( statInfo->ParseServerResponse( nullBuffer ) ==
false )
1507 delete [] nullBuffer;
1510 if( rsp->
hdr.
dlen < 12 || !statInfo )
1512 log->Error(
XRootDMsg,
"[%s] Unable to parse StatInfo in response "
1513 "to %s", pUrl.GetHostId().c_str(),
1514 pRequest->GetObfuscatedDescription().c_str() );
1520 OpenInfo *data =
new OpenInfo( (uint8_t*)buffer,
1521 pResponse->GetSessionId(),
1533 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as ChunkInfo",
1534 pUrl.GetHostId().c_str(),
1535 pRequest->GetObfuscatedDescription().c_str() );
1537 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1543 if( pPartialResps[i]->GetSize() > 8 )
1550 if( pResponse->GetSize() > 8 )
1555 return pBodyReader->GetResponse( response );
1563 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as PageInfo",
1564 pUrl.GetHostId().c_str(),
1565 pRequest->GetObfuscatedDescription().c_str() );
1570 ChunkInfo chunk = pChunkList->front();
1571 bool sizeMismatch =
false;
1572 uint32_t currentOffset = 0;
1573 char *cursor = (
char*)chunk.buffer;
1574 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1576 ServerResponseV2 *part = (ServerResponseV2*)pPartialResps[i]->GetBuffer();
1584 if( currentOffset + datalen > chunk.length )
1586 sizeMismatch =
true;
1590 currentOffset += datalen;
1594 ServerResponseV2 *rspst = (ServerResponseV2*)pResponse->GetBuffer();
1595 size_t datalen = rspst->
status.
bdy.
dlen - NbPgPerRsp( rspst->
info.pgread.offset,
1597 if( currentOffset + datalen <= chunk.length )
1598 currentOffset += datalen;
1600 sizeMismatch =
true;
1605 if( pChunkStatus.front().sizeError || sizeMismatch )
1607 log->Error(
XRootDMsg,
"[%s] Handling response to %s: user supplied "
1608 "buffer is too small for the received data.",
1609 pUrl.GetHostId().c_str(),
1610 pRequest->GetObfuscatedDescription().c_str() );
1614 AnyObject *obj =
new AnyObject();
1615 PageInfo *pgInfo =
new PageInfo( chunk.offset, currentOffset, chunk.buffer,
1616 std::move( pCrc32cDigests) );
1628 std::vector<std::tuple<uint64_t, uint32_t>> retries;
1630 ServerResponseV2 *rsp = (ServerResponseV2*)pResponse->GetBuffer();
1633 ServerResponseBody_pgWrCSE *cse = (ServerResponseBody_pgWrCSE*)pResponse->GetBuffer(
sizeof( ServerResponseV2 ) );
1635 retries.reserve( pgcnt );
1637 sizeof( ServerResponseBody_pgWrCSE ) );
1639 for(
size_t i = 0; i < pgcnt; ++i )
1642 if( i == 0 ) len = cse->
dlFirst;
1643 else if( i == pgcnt - 1 ) len = cse->
dlLast;
1644 retries.push_back( std::make_tuple( pgoffs[i], len ) );
1648 RetryInfo *info =
new RetryInfo( std::move( retries ) );
1649 AnyObject *obj =
new AnyObject();
1662 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %p as "
1663 "VectorReadInfo", pUrl.GetHostId().c_str(),
1664 pRequest->GetObfuscatedDescription().c_str() );
1666 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1672 if( pPartialResps[i]->GetSize() > 8 )
1679 if( pResponse->GetSize() > 8 )
1684 return pBodyReader->GetResponse( response );
1692 int len = rsp->hdr.dlen;
1693 char* data = rsp->body.buffer.data;
1695 return ParseXAttrResponse( data, len, response );
1706 AnyObject *obj =
new AnyObject();
1707 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as BinaryData",
1708 pUrl.GetHostId().c_str(),
1709 pRequest->GetObfuscatedDescription().c_str() );
1712 data->Allocate( length );
1713 data->Append( buffer, length );
1726 Status XRootDMsgHandler::ParseXAttrResponse(
char *data,
size_t len,
1729 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1740 if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
1744 if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
1747 std::vector<XAttrStatus> resp;
1749 for(
kXR_char i = 0; i < nattr; ++i )
1752 if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
1760 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1765 resp.push_back( XAttrStatus( name, st ) );
1772 response =
new AnyObject();
1773 response->Set(
new std::vector<XAttrStatus>( std::move( resp ) ) );
1783 if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
1787 if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
1790 std::vector<XAttr> resp;
1791 resp.reserve( nattr );
1794 for(
kXR_char i = 0; i < nattr; ++i )
1797 if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
1805 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1810 resp.push_back( XAttr( name, st ) );
1814 for(
kXR_char i = 0; i < nattr; ++i )
1817 if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
1819 vlen = ntohl( vlen );
1822 if( !( status = ReadFromBuffer( data, len, vlen,
value ) ).IsOK() )
1825 resp[i].value.swap(
value );
1832 response =
new AnyObject();
1833 response->Set(
new std::vector<XAttr>( std::move( resp ) ) );
1841 std::vector<XAttr> resp;
1846 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1850 if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
1852 vlen = ntohl( vlen );
1855 if( !( status = ReadFromBuffer( data, len, vlen,
value ) ).IsOK() )
1858 resp.push_back( XAttr( name,
value ) );
1862 response =
new AnyObject();
1863 response->Set(
new std::vector<XAttr>( std::move( resp ) ) );
1877 Status XRootDMsgHandler::RewriteRequestRedirect(
const URL &newUrl )
1885 std::string xrdCgi =
"";
1886 std::ostringstream ossXrd;
1887 for(URL::ParamsMap::const_iterator it = newCgi.begin(); it != newCgi.end(); ++it )
1889 if( it->first.compare( 0, 4,
"xrd." ) )
1891 ossXrd << it->first <<
'=' << it->second <<
'&';
1894 xrdCgi = ossXrd.str();
1904 std::string surl = newUrl.GetURL();
1905 (surl.find(
'?') == std::string::npos) ? (surl +=
'?') :
1906 ((*surl.rbegin() !=
'&') ? (surl +=
'&') : (surl +=
""));
1910 std::string surlLog = surl;
1911 if( unlikely( log->GetLevel() >= Log::ErrorMsg ) ) {
1912 surlLog = obfuscateAuth(surlLog);
1914 log->Error(
XRootDMsg,
"[%s] Failed to build redirection URL from data: %s",
1915 newUrl.GetHostId().c_str(), surl.c_str());
1932 Status XRootDMsgHandler::RewriteRequestWait()
1936 XRootDTransport::UnMarshallRequest( pRequest );
1959 XRootDTransport::SetDescription( pRequest );
1960 XRootDTransport::MarshallRequest( pRequest );
1967 void XRootDMsgHandler::HandleError( XRootDStatus status )
1975 if( pSidMgr && pMsgInFly && (
1976 status.code == errOperationExpired ||
1977 status.code == errOperationInterrupted ) )
1986 if( !noreplicas ) pLastError = status;
1988 Log *log = DefaultEnv::GetLog();
1989 log->Debug( XRootDMsg,
"[%s] Handling error while processing %s: %s.",
1990 pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str(),
1991 status.ToString().c_str() );
1997 if( status.IsFatal() && status.code == errTlsError && status.errNo == EAGAIN )
1999 if( pSslErrCnt < MaxSslErrRetry )
2001 status.status &= ~stFatal;
2017 if( status.code == errErrorResponse )
2019 if( RetriableErrorResponse( status ) )
2021 UpdateTriedCGI(status.errNo);
2023 SwitchOnRefreshFlag();
2024 HandleError( RetryAtServer( pLoadBalancer.url, RedirectEntry::EntryRetry ) );
2041 if( status.code == errOperationExpired || pRequest->GetSessionId() ||
2042 status.code == errOperationInterrupted || time(0) >= pExpiration )
2044 log->Error( XRootDMsg,
"[%s] Unable to get the response to request %s",
2045 pUrl.GetHostId().c_str(),
2046 pRequest->GetObfuscatedDescription().c_str() );
2057 if( pLoadBalancer.url.IsValid() &&
2058 pLoadBalancer.url.GetLocation() != pUrl.GetLocation() )
2061 HandleError( RetryAtServer( pLoadBalancer.url, RedirectEntry::EntryRetry ) );
2066 if( !status.IsFatal() && IsRetriable() )
2068 log->Info( XRootDMsg,
"[%s] Retrying request: %s.",
2069 pUrl.GetHostId().c_str(),
2070 pRequest->GetObfuscatedDescription().c_str() );
2073 HandleError( RetryAtServer( pUrl, RedirectEntry::EntryRetry ) );
2085 Status XRootDMsgHandler::RetryAtServer(
const URL &url, RedirectEntry::Type entryType )
2088 Log *log = DefaultEnv::GetLog();
2093 if( pRdirEntry ) pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
2094 pRdirEntry.reset(
new RedirectEntry( pUrl.GetLocation(), url.GetLocation(), entryType ) );
2096 if( pUrl.GetLocation() != url.GetLocation() )
2098 pHosts->push_back( url );
2110 pSidMgr->ReleaseSID( req->
streamid );
2117 if( !url.IsLocalFile() )
2119 pSidMgr = SIDMgrPool::Instance().GetSIDMgr( url );
2120 Status st = pSidMgr->AllocateSID( req->
streamid );
2123 log->Error( XRootDMsg,
"[%s] Impossible to send message %s.",
2124 pUrl.GetHostId().c_str(),
2125 pRequest->GetObfuscatedDescription().c_str() );
2133 if( pUrl.IsMetalink() && pFollowMetalink )
2135 log->Debug( ExDbgMsg,
"[%s] Metaling redirection for MsgHandler: %p (message: %s ).",
2136 pUrl.GetHostId().c_str(),
this,
2137 pRequest->GetObfuscatedDescription().c_str() );
2139 return pPostMaster->Redirect( pUrl, pRequest,
this );
2141 else if( pUrl.IsLocalFile() )
2143 HandleLocalRedirect( &pUrl );
2148 log->Debug( ExDbgMsg,
"[%s] Retry at server MsgHandler: %p (message: %s ).",
2149 pUrl.GetHostId().c_str(),
this,
2150 pRequest->GetObfuscatedDescription().c_str() );
2151 return pPostMaster->Send( pUrl, pRequest,
this,
true, pExpiration );
2158 void XRootDMsgHandler::UpdateTriedCGI(uint32_t errNo)
2168 if( pEffectiveDataServerUrl )
2170 tried = pEffectiveDataServerUrl->GetHostName();
2171 delete pEffectiveDataServerUrl;
2172 pEffectiveDataServerUrl = 0;
2178 tried = pUrl.GetHostName();
2183 {
if (errNo ==
kXR_NotFound) cgi[
"triedrc"] =
"enoent";
2184 else if (errNo ==
kXR_IOError) cgi[
"triedrc"] =
"ioerr";
2185 else if (errNo ==
kXR_FSError) cgi[
"triedrc"] =
"fserr";
2194 if( pLoadBalancer.url.IsValid() && (pLoadBalancer.flags &
kXR_attrMeta) )
2196 HostList::reverse_iterator it;
2197 for( it = pHosts->rbegin()+1; it != pHosts->rend(); ++it )
2199 if( it->loadBalancer )
2202 tried +=
"," + it->url.GetHostName();
2209 cgi[
"tried"] = tried;
2210 XRootDTransport::UnMarshallRequest( pRequest );
2211 MessageUtils::RewriteCGIAndPath( pRequest, cgi,
false,
"" );
2212 XRootDTransport::MarshallRequest( pRequest );
2218 void XRootDMsgHandler::SwitchOnRefreshFlag()
2220 XRootDTransport::UnMarshallRequest( pRequest );
2236 XRootDTransport::SetDescription( pRequest );
2237 XRootDTransport::MarshallRequest( pRequest );
2244 void XRootDMsgHandler::HandleRspOrQueue()
2246 JobManager *jobMgr = pPostMaster->GetJobManager();
2247 if( jobMgr->IsWorker() )
2251 Log *log = DefaultEnv::GetLog();
2252 log->Debug( ExDbgMsg,
"[%s] Passing to the thread-pool MsgHandler: %p (message: %s ).",
2253 pUrl.GetHostId().c_str(),
this,
2254 pRequest->GetObfuscatedDescription().c_str() );
2255 jobMgr->QueueJob(
new HandleRspJob(
this ), 0 );
2262 void XRootDMsgHandler::HandleLocalRedirect( URL *url )
2264 Log *log = DefaultEnv::GetLog();
2265 log->Debug( ExDbgMsg,
"[%s] Handling local redirect - MsgHandler: %p (message: %s ).",
2266 pUrl.GetHostId().c_str(),
this,
2267 pRequest->GetObfuscatedDescription().c_str() );
2269 if( !pLFileHandler )
2271 HandleError( XRootDStatus( stFatal, errNotSupported ) );
2275 AnyObject *resp = 0;
2276 pLFileHandler->SetHostList( *pHosts );
2277 XRootDStatus st = pLFileHandler->Open( url, pRequest, resp );
2284 pResponseHandler->HandleResponseWithHosts(
new XRootDStatus(),
2295 bool XRootDMsgHandler::IsRetriable()
2298 DefaultEnv::GetEnv()->GetString(
"OpenRecovery", value );
2299 if( value ==
"true" )
return true;
2310 Log *log = DefaultEnv::GetLog();
2311 log->Debug( XRootDMsg,
2312 "[%s] Not allowed to retry open request (OpenRecovery disabled): %s.",
2313 pUrl.GetHostId().c_str(),
2314 pRequest->GetObfuscatedDescription().c_str() );
2327 bool XRootDMsgHandler::OmitWait( Message &request,
const URL &url )
2330 if( !url.IsMetalink() )
2341 RedirectorRegistry ®istry = RedirectorRegistry::Instance();
2342 VirtualRedirector *redirector = registry.Get( url );
2346 if( redirector->Count( request ) > 1 )
2355 bool XRootDMsgHandler::RetriableErrorResponse(
const Status &status )
2359 if( !( pLoadBalancer.url.IsValid() &&
2360 pUrl.GetLocation() != pLoadBalancer.url.GetLocation() ) )
2371 if( !( pLoadBalancer.flags &
kXR_attrMeta ) )
return false;
2382 DefaultEnv::GetEnv()->GetInt(
"NotAuthorizedRetryLimit", limit );
2383 bool ret = pNotAuthorizedCounter < limit;
2384 ++pNotAuthorizedCounter;
2387 Log *log = DefaultEnv::GetLog();
2388 log->Error( XRootDMsg,
2389 "[%s] Reached limit of NotAuthorized retries!",
2390 pUrl.GetHostId().c_str() );
2410 void XRootDMsgHandler::DumpRedirectTraceBack()
2412 if( pRedirectTraceBack.empty() )
return;
2414 std::stringstream sstrm;
2416 sstrm <<
"Redirect trace-back:\n";
2420 auto itr = pRedirectTraceBack.begin();
2421 sstrm <<
'\t' << counter <<
". " << (*itr)->ToString() <<
'\n';
2427 for( ; itr != pRedirectTraceBack.end(); ++itr, ++prev, ++counter )
2428 sstrm <<
'\t' << counter <<
". "
2429 << (*itr)->ToString( (*prev)->status.IsOK() ) <<
'\n';
2432 DefaultEnv::GetEnv()->GetInt(
"NotAuthorizedRetryLimit", authlimit );
2434 bool warn = !pStatus.IsOK() &&
2437 ( pStatus.code ==
errAuthFailed && pNotAuthorizedCounter >= authlimit ) );
2439 Log *log = DefaultEnv::GetLog();
2441 log->Warning( XRootDMsg,
"%s", sstrm.str().c_str() );
2443 log->Debug( XRootDMsg,
"%s", sstrm.str().c_str() );
2448 template<
typename T>
2449 Status XRootDMsgHandler::ReadFromBuffer(
char *&buffer,
size_t &buflen, T& result )
2451 if(
sizeof( T ) > buflen )
return Status( stError, errDataError );
2453 memcpy(&result, buffer,
sizeof(T));
2455 buffer +=
sizeof( T );
2456 buflen -=
sizeof( T );
2464 Status XRootDMsgHandler::ReadFromBuffer(
char *&buffer,
size_t &buflen, std::string &result )
2471 if( !( status = ReadFromBuffer( buffer, buflen, c ) ).IsOK() )
2484 Status XRootDMsgHandler::ReadFromBuffer(
char *&buffer,
size_t &buflen,
2485 size_t size, std::string &result )
2489 if( size > buflen )
return Status( stError, errDataError );
2491 result.append( buffer, size );
struct ClientFattrRequest fattr
#define kXR_collapseRedir
ServerResponseStatus status
union ServerResponse::@040373375333017131300127053271011057331004327334 body
struct ClientDirlistRequest dirlist
static const int kXR_ckpXeq
struct ClientOpenRequest open
struct ServerResponseBody_Status bdy
struct ClientRequestHdr header
union ServerResponseV2::@207342300141235315373173036347114307032363217365 info
struct ClientChkPointRequest chkpoint
struct ServerResponseHeader hdr
#define kXR_PROTOCOLVERSION
struct ClientStatRequest stat
struct ClientProtocolRequest protocol
struct ClientLocateRequest locate
void Get(Type &object)
Retrieve the object being held.
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 GetInt(const std::string &key, int &value)
virtual void Run(void *arg)
The job logic.
HandleRspJob(XrdCl::XRootDMsgHandler *handler)
Interface for a job to be run by the job manager.
void Error(uint64_t topic, const char *format,...)
Report an error.
void Warning(uint64_t topic, const char *format,...)
Report a warning.
void Dump(uint64_t topic, const char *format,...)
Print a dump message.
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
static void RewriteCGIAndPath(Message *msg, const URL::ParamsMap &newCgi, bool replace, const std::string &newPath)
Append cgi to the one already present in the message.
The message representation used throughout the system.
const std::string & GetObfuscatedDescription() const
Get the description of the message with authz parameter obfuscated.
@ 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.
virtual XRootDStatus Send(const char *buffer, size_t size, int &bytesWritten)
void RegisterTask(Task *task, time_t time, bool own=true)
Interface for a task to be run by the TaskManager.
std::string GetHostId() const
Get the host part of the URL (user:password@host:port)
const std::string & GetPassword() const
Get the password.
std::map< std::string, std::string > ParamsMap
bool FromString(const std::string &url)
Parse a string and fill the URL fields.
void SetPassword(const std::string &password)
Set the password.
void SetParams(const std::string ¶ms)
Set params.
const std::string & GetUserName() const
Get the username.
std::string GetURL() const
Get the URL.
void SetProtocol(const std::string &protocol)
Set protocol.
const ParamsMap & GetParams() const
Get the URL params.
const std::string & GetProtocol() const
Get the protocol.
bool IsValid() const
Is the url valid.
void SetUserName(const std::string &userName)
Set the username.
static void splitString(Container &result, const std::string &input, const std::string &delimiter)
Split a string.
static bool CheckEC(const Message *req, const URL &url)
Check if this client can support given EC redirect.
Handle/Process/Forward XRootD messages.
virtual uint16_t Examine(std::shared_ptr< Message > &msg)
XRootDStatus WriteMessageBody(Socket *socket, uint32_t &bytesWritten)
const Message * GetRequest() const
Get the request pointer.
virtual uint8_t OnStreamEvent(StreamEvent event, XRootDStatus status)
virtual XRootDStatus ReadMessageBody(Message *msg, Socket *socket, uint32_t &bytesRead)
virtual void OnStatusReady(const Message *message, XRootDStatus status)
The requested action has been performed and the status is available.
void WaitDone(time_t now)
virtual bool IsRaw() const
Are we a raw writer or not?
virtual void Process()
Process the message if it was "taken" by the examine action.
virtual uint16_t InspectStatusRsp()
virtual uint16_t GetSid() const
const std::string & GetErrorMessage() const
Get error 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 errRedirectLimit
const int DefaultMaxMetalinkWait
const uint16_t errErrorResponse
const uint16_t errOperationExpired
const uint16_t stFatal
Fatal error, it's still an error.
const uint16_t stError
An error occurred that could potentially be retried.
const uint16_t errNotFound
std::vector< HostInfo > HostList
const uint16_t errDataError
data is corrupted
const uint16_t errInternal
Internal error.
const uint16_t stOK
Everything went OK.
const uint16_t errInvalidResponse
const uint16_t errInvalidRedirectURL
Buffer BinaryDataInfo
Binary buffer.
const uint16_t errOperationInterrupted
const uint16_t suContinue
const int DefaultNotAuthorizedRetryLimit
const uint16_t errRedirect
Response NullRef< Response >::value
const uint16_t errAuthFailed
const uint16_t errInvalidMessage
static const int PageSize
ssize_t Move(KernelBuffer &kbuff, char *&ubuff)
Describe a data chunk for vector read.
void * buffer
length of the chunk
uint32_t length
offset in the file
Procedure execution status.
uint16_t code
Error type, or additional hints on what to do.
bool IsOK() const
We're fine.
std::string ToString() const
Create a string representation.
static const uint16_t ServerFlags
returns server flags
static const uint16_t ProtocolVersion
returns the protocol version