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();
111 const int sst = pSendingState.fetch_or( kSawResp );
113 if( !( sst & kSendDone ) && !( sst & kSawResp ) )
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() );
129 if( pOksofarAsAnswer )
132 while( pResponse ) pCV.Wait();
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() );
146 if( msg->GetSize() < 8 )
186 pBodyReader->SetDataLength( dlen );
202 "message %s", pUrl.GetHostId().c_str(),
203 pRequest->GetObfuscatedDescription().c_str() );
243 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_oksofar response to request "
244 "%s", pUrl.GetHostId().c_str(),
245 pRequest->GetObfuscatedDescription().c_str() );
247 if( !pOksofarAsAnswer )
249 pPartialResps.emplace_back( std::move( pResponse ) );
260 pTimeoutFence.store(
true, std::memory_order_relaxed );
269 pTimeoutFence.store(
true, std::memory_order_relaxed );
278 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_status response to request "
279 "%s", pUrl.GetHostId().c_str(),
280 pRequest->GetObfuscatedDescription().c_str() );
297 pTimeoutFence.store(
true, std::memory_order_relaxed );
331 log->
Error(
XRootDMsg,
"[%s] kXR_status: invalid message size.", pUrl.GetHostId().c_str() );
352 pUrl.GetHostId().c_str() );
364 pPartialResps.push_back( std::move( pResponse ) );
377 pPageReader.reset(
new AsyncPageReader( *pChunkList, pCrc32cDigests ) );
378 pPageReader->SetRsp( rspst );
393 pResponse->GetCursor() )
423 if( pUrl.IsLocalFile() && pUrl.IsMetalink() )
432 int *qryResponse =
nullptr;
434 qryResult.
Get( qryResponse );
436 pHosts->back().flags = *qryResponse;
438 qryResponse =
nullptr;
441 qryResult.
Get( qryResponse );
443 pHosts->back().protocol = *qryResponse;
462 pSendingState.fetch_or( kInFlyDone );
469 pAggregatedWaitTime = 0;
478 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_ok response to request %s",
479 pUrl.GetHostId().c_str(),
480 pRequest->GetObfuscatedDescription().c_str() );
488 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_status response to request %s",
489 pUrl.GetHostId().c_str(),
490 pRequest->GetObfuscatedDescription().c_str() );
501 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_oksofar response to request %s",
502 pUrl.GetHostId().c_str(),
503 pRequest->GetObfuscatedDescription().c_str() );
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,
534 pUrl.GetHostId().c_str() );
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;
546 "message %s: %s, port %d", pUrl.GetHostId().c_str(),
547 pRequest->GetObfuscatedDescription().c_str(), urlInfo.c_str(),
548 rsp->
body.redirect.port );
553 if( !pRedirectCounter )
556 "message %s, the last known error is: %s",
557 pUrl.GetHostId().c_str(),
558 pRequest->GetObfuscatedDescription().c_str(),
559 pLastError.ToString().c_str() );
572 uint32_t flags = pHosts->back().flags;
573 if( !pHasLoadBalancer )
582 if( ( flags &
kXR_attrMeta ) || !pLoadBalancer.url.IsValid() )
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;
603 pEffectiveDataServerUrl =
new URL( pHosts->back().url );
608 std::vector<std::string> urlComponents;
612 std::ostringstream o;
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 )
632 std::string url( rsp->
body.redirect.host, rsp->
hdr.
dlen-4 );
633 pPostMaster->CollapseRedirect( pUrl, url );
638 std::string url( rsp->
body.redirect.host, rsp->
hdr.
dlen-4 );
640 pRedirectAsAnswer =
true;
644 URL newUrl =
URL( o.str() );
649 pUrl.GetHostId().c_str(), urlInfo.c_str() );
654 if( pUrl.GetUserName() !=
"" && newUrl.
GetUserName() ==
"" )
657 if( pUrl.GetPassword() !=
"" && newUrl.
GetPassword() ==
"" )
666 std::ostringstream ossXrd;
669 for(URL::ParamsMap::const_iterator it = urlParams.begin();
670 it != urlParams.end(); ++it )
672 if( it->first.compare( 0, 4,
"xrd." ) &&
673 it->first.compare( 0, 6,
"xrdcl." ) )
676 ossXrd << it->first <<
'=' << it->second <<
'&';
679 std::string xrdCgi = ossXrd.str();
680 pRedirectUrl = newUrl.
GetURL();
683 if( urlComponents.size() > 1 )
686 pRedirectUrl += urlComponents[1];
687 std::ostringstream o;
688 o <<
"fake://fake:111//fake?";
689 o << urlComponents[1];
691 if( urlComponents.size() == 3 )
692 o <<
'?' << urlComponents[2];
698 pRedirectUrl += xrdCgi;
701 cgiURL =
URL( o.str() );
706 std::ostringstream o;
707 o <<
"fake://fake:111//fake?";
709 cgiURL =
URL( o.str() );
711 pRedirectUrl += xrdCgi;
721 pRedirectAsAnswer =
true;
723 if( pRedirectAsAnswer )
734 std::string prevPath;
735 Status st = RewriteRequestRedirect( newUrl, prevPath );
754 if( ( pUrl.GetProtocol() ==
"roots" || pUrl.GetProtocol() ==
"xroots" ) &&
770 uint32_t waitSeconds = 0;
774 char *infoMsg =
new char[rsp->
hdr.
dlen-3];
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(),
782 waitSeconds = rsp->
body.wait.seconds;
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() );
791 pAggregatedWaitTime += waitSeconds;
796 if( OmitWait( *pRequest, pLoadBalancer.url ) )
800 if( pAggregatedWaitTime > maxWait )
812 Status st = RewriteRequestWait();
824 time_t resendTime = ::time(0)+waitSeconds;
826 if( resendTime < pExpiration )
828 log->
Debug(
ExDbgMsg,
"[%s] Scheduling WaitTask for MsgHandler: %p (message: %s ).",
829 pUrl.GetHostId().c_str(), (
void*)
this,
830 pRequest->GetObfuscatedDescription().c_str() );
832 TaskManager *taskMgr = pPostMaster->GetTaskManager();
833 taskMgr->
RegisterTask(
new WaitTask(
this ), resendTime );
837 log->
Debug(
XRootDMsg,
"[%s] Wait time is too long, timing out %s",
838 pUrl.GetHostId().c_str(),
839 pRequest->GetObfuscatedDescription().c_str() );
856 pUrl.GetHostId().c_str() );
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() );
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() );
893 log->
Dump(
XRootDMsg,
"[%s] Stream event reported for msg %s",
894 pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str() );
899 if( pTimeoutFence.load( std::memory_order_relaxed ) )
902 HandleError( status );
911 uint32_t &bytesRead )
917 return pPageReader->Read( *socket, bytesRead );
919 return pBodyReader->Read( *socket, bytesRead );
935 log->
Dump(
XRootDMsg,
"[%s] Got notification that outgoing message %s "
936 "was sent successfully.", pUrl.GetHostId().c_str(),
943 const int sst = pSendingState.fetch_or( kSendDone );
946 if( status.
IsOK() && ( sst & kSendDone ) )
return;
950 if( !status.
IsOK() && ( ( sst & kFinalResp ) || ( sst & kSawResp ) ) )
952 log->
Error(
XRootDMsg,
"[%s] Unexpected error for message %s. Trying to "
953 "recover.", pUrl.GetHostId().c_str(),
955 HandleError( status );
959 if( sst & kFinalResp )
967 if( sst & kRetryAtSrv )
971 HandleError( RetryAtServer( pRetryAtUrl, pRetryAtEntryType ) );
996 log->
Error(
XRootDMsg,
"[%s] Impossible to send message %s. Trying to "
997 "recover.", pUrl.GetHostId().c_str(),
999 HandleError( status );
1026 uint32_t &bytesWritten )
1031 if( !pChunkList->empty() && !pCrc32cDigests.empty() )
1041 int fLen = 0, lLen = 0;
1047 if( pPgWrtCksumBuff.GetCursor() == 0 )
1049 uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
1050 memcpy( pPgWrtCksumBuff.GetBuffer(), &digest,
sizeof( uint32_t ) );
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;
1058 while( btsLeft > 0 )
1061 while( pPgWrtCksumBuff.GetCursor() <
sizeof( uint32_t ) )
1063 uint32_t dgstlen =
sizeof( uint32_t ) - pPgWrtCksumBuff.GetCursor();
1064 char* dgstbuf = pPgWrtCksumBuff.GetBufferAtCursor();
1066 Status st = socket->
Send( dgstbuf, dgstlen, btswrt );
1067 if( !st.
IsOK() )
return st;
1068 bytesWritten += btswrt;
1069 pPgWrtCksumBuff.AdvanceCursor( btswrt );
1074 Status st = socket->
Send( pgbuf, pglen, btswrt );
1075 if( !st.
IsOK() )
return st;
1079 bytesWritten += btswrt;
1080 pAsyncOffset += btswrt;
1086 ++pPgWrtCurrentPageNb;
1087 if( pPgWrtCurrentPageNb < nbpgs )
1090 pPgWrtCksumBuff.SetCursor( 0 );
1091 uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
1092 memcpy( pPgWrtCksumBuff.GetBuffer(), &digest,
sizeof( uint32_t ) );
1096 if( pglen > btsLeft ) pglen = btsLeft;
1098 pPgWrtCurrentPageOffset = 0;
1102 pPgWrtCurrentPageOffset += btswrt;
1106 else if( !pChunkList->empty() )
1108 size_t size = pChunkList->size();
1109 for(
size_t i = pAsyncChunkIndex ; i < size; ++i )
1111 char *buffer = (
char*)(*pChunkList)[i].buffer;
1112 uint32_t size = (*pChunkList)[i].length;
1113 size_t leftToBeWritten = size - pAsyncOffset;
1115 while( leftToBeWritten )
1118 Status st = socket->
Send( buffer + pAsyncOffset, leftToBeWritten, btswrt );
1119 bytesWritten += btswrt;
1121 pAsyncOffset += btswrt;
1122 leftToBeWritten -= btswrt;
1142 log->
Debug(
XRootDMsg,
"[%s] Channel is encrypted: cannot use kernel buffer.",
1143 pUrl.GetHostId().c_str() );
1148 pChunkList->push_back(
ChunkInfo( 0, ret, ubuff ) );
1155 while( !pKBuff->Empty() )
1159 bytesWritten += btswrt;
1163 log->
Debug(
XRootDMsg,
"[%s] Request %s payload (kernel buffer) transferred to socket.",
1164 pUrl.GetHostId().c_str(), pRequest->GetObfuscatedDescription().c_str() );
1184 pTimeoutFence.store(
false, std::memory_order_relaxed );
1190 void XRootDMsgHandler::HandleResponse()
1201 const int sst = pSendingState.fetch_or( kFinalResp );
1202 if( ( sst & kSawReadySend ) && !( sst & kSendDone ) )
1210 XRootDStatus *status = ProcessStatus();
1211 AnyObject *response = 0;
1214 log->Debug(
ExDbgMsg,
"[%s] Calling MsgHandler: %p (message: %s ) "
1218 status->ToString().c_str() );
1220 if( status->IsOK() )
1222 Status st = ParseResponse( response );
1227 status =
new XRootDStatus( st );
1237 pRdirEntry->status = *status;
1238 pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
1244 if( pSidMgr && finalrsp )
1246 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1247 if( status->IsOK() || !IsInFly() ||
1252 HostList *hosts = pHosts.release();
1254 pHosts.reset(
new HostList( *hosts ) );
1256 pResponseHandler->HandleResponseWithHosts( status, response, hosts );
1269 XrdSysCondVarHelper lck( pCV );
1271 pTimeoutFence.store(
false, std::memory_order_relaxed );
1282 XRootDStatus *st =
new XRootDStatus( pStatus );
1283 ServerResponse *rsp = 0;
1285 rsp = (ServerResponse *)pResponse->GetBuffer();
1287 if( !pStatus.IsOK() && rsp )
1291 st->errNo = rsp->
body.error.errnum;
1294 std::string errmsg( rsp->
body.error.errmsg, rsp->
hdr.
dlen-5 );
1296 errmsg +=
" Last seen error: " + pLastError.ToString();
1297 st->SetErrorMessage( errmsg );
1300 st->SetErrorMessage( pRedirectUrl );
1314 ServerResponse *rsp = (ServerResponse *)pResponse->GetBuffer();
1315 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1323 log->Error(
XRootDMsg,
"Internal Error: unable to process redirect" );
1328 uint32_t length = 0;
1334 if( pPartialResps.empty() )
1336 buffer = rsp->
body.buffer.data;
1345 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1347 ServerResponse *part = (ServerResponse*)pPartialResps[i]->GetBuffer();
1352 buff.Allocate( length );
1353 uint32_t offset = 0;
1354 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1356 ServerResponse *part = (ServerResponse*)pPartialResps[i]->GetBuffer();
1357 buff.Append( part->
body.buffer.data, part->
hdr.
dlen, offset );
1360 buff.Append( rsp->
body.buffer.data, rsp->
hdr.
dlen, offset );
1361 buffer = buff.GetBuffer();
1392 AnyObject *obj =
new AnyObject();
1394 char *nullBuffer =
new char[length+1];
1395 nullBuffer[length] = 0;
1396 memcpy( nullBuffer, buffer, length );
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();
1403 if( data->ParseServerResponse( nullBuffer ) ==
false )
1407 delete [] nullBuffer;
1410 delete [] nullBuffer;
1422 AnyObject *obj =
new AnyObject();
1429 StatInfoVFS *data =
new StatInfoVFS();
1431 char *nullBuffer =
new char[length+1];
1432 nullBuffer[length] = 0;
1433 memcpy( nullBuffer, buffer, length );
1435 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as "
1436 "StatInfoVFS: %s", pUrl.GetHostId().c_str(),
1437 pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
1439 if( data->ParseServerResponse( nullBuffer ) ==
false )
1443 delete [] nullBuffer;
1446 delete [] nullBuffer;
1455 StatInfo *data =
new StatInfo();
1457 char *nullBuffer =
new char[length+1];
1458 nullBuffer[length] = 0;
1459 memcpy( nullBuffer, buffer, length );
1461 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as StatInfo: "
1462 "%s", pUrl.GetHostId().c_str(),
1463 pRequest->GetObfuscatedDescription().c_str(), nullBuffer );
1465 if( data->ParseServerResponse( nullBuffer ) ==
false )
1469 delete [] nullBuffer;
1472 delete [] nullBuffer;
1485 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as ProtocolInfo",
1486 pUrl.GetHostId().c_str(),
1487 pRequest->GetObfuscatedDescription().c_str() );
1491 log->Error(
XRootDMsg,
"[%s] Got invalid redirect response.",
1492 pUrl.GetHostId().c_str() );
1496 AnyObject *obj =
new AnyObject();
1497 ProtocolInfo *data =
new ProtocolInfo( rsp->
body.protocol.pval,
1498 rsp->
body.protocol.flags );
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() );
1516 memcpy( path, pRequest->GetBuffer(24), req->
dirlist.
dlen );
1518 DirectoryList *data =
new DirectoryList();
1519 data->SetParentName( path );
1522 char *nullBuffer =
new char[length+1];
1523 nullBuffer[length] = 0;
1524 memcpy( nullBuffer, buffer, length );
1526 bool invalidrsp =
false;
1528 if( !pDirListStarted )
1531 pDirListStarted =
true;
1533 invalidrsp = !data->ParseServerResponse( pUrl.GetHostId(), nullBuffer );
1536 invalidrsp = !data->ParseServerResponse( pUrl.GetHostId(), nullBuffer, pDirListWithStat );
1542 delete [] nullBuffer;
1546 delete [] nullBuffer;
1557 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as OpenInfo",
1558 pUrl.GetHostId().c_str(),
1559 pRequest->GetObfuscatedDescription().c_str() );
1563 log->Error(
XRootDMsg,
"[%s] Got invalid open response.",
1564 pUrl.GetHostId().c_str() );
1568 AnyObject *obj =
new AnyObject();
1569 StatInfo *statInfo = 0;
1576 log->Dump(
XRootDMsg,
"[%s] Parsing StatInfo in response to %s",
1577 pUrl.GetHostId().c_str(),
1578 pRequest->GetObfuscatedDescription().c_str() );
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 );
1586 statInfo =
new StatInfo();
1587 if( statInfo->ParseServerResponse( nullBuffer ) ==
false )
1592 delete [] nullBuffer;
1595 if( rsp->
hdr.
dlen < 12 || !statInfo )
1597 log->Error(
XRootDMsg,
"[%s] Unable to parse StatInfo in response "
1598 "to %s", pUrl.GetHostId().c_str(),
1599 pRequest->GetObfuscatedDescription().c_str() );
1605 OpenInfo *data =
new OpenInfo( (uint8_t*)buffer,
1606 pResponse->GetSessionId(),
1618 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as ChunkInfo",
1619 pUrl.GetHostId().c_str(),
1620 pRequest->GetObfuscatedDescription().c_str() );
1622 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1628 if( pPartialResps[i]->GetSize() > 8 )
1635 if( pResponse->GetSize() > 8 )
1640 return pBodyReader->GetResponse( response );
1648 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as PageInfo",
1649 pUrl.GetHostId().c_str(),
1650 pRequest->GetObfuscatedDescription().c_str() );
1655 ChunkInfo chunk = pChunkList->front();
1656 bool sizeMismatch =
false;
1657 uint32_t currentOffset = 0;
1658 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1660 ServerResponseV2 *part = (ServerResponseV2*)pPartialResps[i]->GetBuffer();
1668 if( currentOffset + datalen > chunk.length )
1670 sizeMismatch =
true;
1674 currentOffset += datalen;
1677 ServerResponseV2 *rspst = (ServerResponseV2*)pResponse->GetBuffer();
1678 size_t datalen = rspst->
status.
bdy.
dlen - NbPgPerRsp( rspst->
info.pgread.offset,
1680 if( currentOffset + datalen <= chunk.length )
1681 currentOffset += datalen;
1683 sizeMismatch =
true;
1688 if( pChunkStatus.front().sizeError || sizeMismatch )
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() );
1697 AnyObject *obj =
new AnyObject();
1698 PageInfo *pgInfo =
new PageInfo( chunk.offset, currentOffset, chunk.buffer,
1699 std::move( pCrc32cDigests) );
1711 std::vector<std::tuple<uint64_t, uint32_t>> retries;
1713 ServerResponseV2 *rsp = (ServerResponseV2*)pResponse->GetBuffer();
1716 ServerResponseBody_pgWrCSE *cse = (ServerResponseBody_pgWrCSE*)pResponse->GetBuffer(
sizeof( ServerResponseV2 ) );
1718 retries.reserve( pgcnt );
1720 sizeof( ServerResponseBody_pgWrCSE ) );
1722 for(
size_t i = 0; i < pgcnt; ++i )
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 ) );
1731 RetryInfo *info =
new RetryInfo( std::move( retries ) );
1732 AnyObject *obj =
new AnyObject();
1745 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as "
1746 "VectorReadInfo", pUrl.GetHostId().c_str(),
1747 pRequest->GetObfuscatedDescription().c_str() );
1749 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1755 if( pPartialResps[i]->GetSize() > 8 )
1762 if( pResponse->GetSize() > 8 )
1767 return pBodyReader->GetResponse( response );
1775 int len = rsp->hdr.dlen;
1776 char* data = rsp->body.buffer.data;
1778 return ParseXAttrResponse( data, len, response );
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() );
1795 data->Allocate( length );
1796 data->Append( buffer, length );
1809 Status XRootDMsgHandler::ParseXAttrResponse(
char *data,
size_t len,
1812 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
1823 if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
1827 if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
1830 std::vector<XAttrStatus> resp;
1832 for(
kXR_char i = 0; i < nattr; ++i )
1835 if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
1843 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1848 resp.push_back( XAttrStatus( name, st ) );
1855 response =
new AnyObject();
1856 response->Set(
new std::vector<XAttrStatus>( std::move( resp ) ) );
1866 if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
1870 if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
1873 std::vector<XAttr> resp;
1874 resp.reserve( nattr );
1877 for(
kXR_char i = 0; i < nattr; ++i )
1880 if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
1888 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1893 resp.push_back( XAttr( name, st ) );
1897 for(
kXR_char i = 0; i < nattr; ++i )
1900 if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
1902 vlen = ntohl( vlen );
1905 if( !( status = ReadFromBuffer( data, len, vlen, value ) ).IsOK() )
1908 resp[i].value.swap( value );
1915 response =
new AnyObject();
1916 response->Set(
new std::vector<XAttr>( std::move( resp ) ) );
1924 std::vector<XAttr> resp;
1929 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1933 if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
1935 vlen = ntohl( vlen );
1938 if( !( status = ReadFromBuffer( data, len, vlen, value ) ).IsOK() )
1941 resp.push_back( XAttr( name, value ) );
1945 response =
new AnyObject();
1946 response->Set(
new std::vector<XAttr>( std::move( resp ) ) );
1960 Status XRootDMsgHandler::RewriteRequestRedirect(
const URL &newUrl, std::string &opath )
1965 if ( !newUrl.IsValid() )
1967 std::string surlLog = newUrl.GetURL();
1971 log->Error(
XRootDMsg,
"[%s] Failed to build redirection URL from data: %s",
1972 newUrl.GetHostId().c_str(), surlLog.c_str());
1989 Status XRootDMsgHandler::RewriteRequestWait()
1991 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
2024 void XRootDMsgHandler::HandleError(
XRootDStatus status )
2032 if( pSidMgr && IsInFly() && (
2036 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
2043 if( !noreplicas ) pLastError = status;
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() );
2054 if( status.IsFatal() && status.code ==
errTlsError && status.errNo == EAGAIN )
2056 if( pSslErrCnt < MaxSslErrRetry )
2058 status.status &= ~stFatal;
2076 if( RetriableErrorResponse( status ) )
2078 UpdateTriedCGI(status.errNo);
2080 SwitchOnRefreshFlag();
2101 log->Error(
XRootDMsg,
"[%s] Unable to get the response to request %s",
2102 pUrl.GetHostId().c_str(),
2103 pRequest->GetObfuscatedDescription().c_str() );
2114 if( pLoadBalancer.url.IsValid() &&
2115 pLoadBalancer.url.GetLocation() != pUrl.GetLocation() )
2123 if( !status.IsFatal() && IsRetriable() )
2125 log->Info(
XRootDMsg,
"[%s] Retrying request: %s.",
2126 pUrl.GetHostId().c_str(),
2127 pRequest->GetObfuscatedDescription().c_str() );
2144 if( &pRetryAtUrl != &url ) pRetryAtUrl = url;
2145 pRetryAtEntryType = entryType;
2146 const int sst = pSendingState.fetch_or( kRetryAtSrv );
2152 if( ( sst & kSawReadySend ) && !( sst & kSendDone ) )
return Status();
2153 pSendingState &= ~kRetryAtSrv;
2161 if( pRdirEntry ) pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
2162 pRdirEntry.reset(
new RedirectEntry( pUrl.GetLocation(), url.GetLocation(), entryType ) );
2164 if( pUrl.GetLocation() != url.GetLocation() )
2166 pHosts->push_back( url );
2173 if( pUrl.GetPath() != url.GetPath() )
2188 ClientRequestHdr *req = (ClientRequestHdr*)pRequest->GetBuffer();
2191 pSidMgr->ReleaseSID( req->
streamid );
2198 if( !url.IsLocalFile() )
2201 Status st = pSidMgr->AllocateSID( req->
streamid );
2204 log->Error(
XRootDMsg,
"[%s] Impossible to send message %s.",
2205 pUrl.GetHostId().c_str(),
2206 pRequest->GetObfuscatedDescription().c_str() );
2214 if( pUrl.IsMetalink() && pFollowMetalink )
2216 log->Debug(
ExDbgMsg,
"[%s] Metaling redirection for MsgHandler: %p (message: %s ).",
2217 pUrl.GetHostId().c_str(), (
void*)
this,
2218 pRequest->GetObfuscatedDescription().c_str() );
2220 return pPostMaster->Redirect( pUrl, pRequest,
this );
2222 else if( pUrl.IsLocalFile() )
2224 HandleLocalRedirect( &pUrl );
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 );
2239 void XRootDMsgHandler::UpdateTriedCGI(uint32_t errNo)
2243 HostList::reverse_iterator itst = pHosts->rbegin();
2250 if( pEffectiveDataServerUrl )
2252 for( ; itst != pHosts->rend(); ++itst )
2254 if( itst->url.GetURL() == pEffectiveDataServerUrl->GetURL() )
2257 tried = pEffectiveDataServerUrl->GetHostName();
2258 delete pEffectiveDataServerUrl;
2259 pEffectiveDataServerUrl = 0;
2265 else if ( !pUrl.IsLocalFile() )
2266 tried = pUrl.GetHostName();
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";
2282 if( pLoadBalancer.url.IsValid() && (pLoadBalancer.flags &
kXR_attrMeta) )
2284 HostList::reverse_iterator it;
2285 if( itst == pHosts->rend() )
2286 itst = pHosts->rbegin();
2287 for( it = itst+1; it != pHosts->rend(); ++it )
2289 if( it->loadBalancer )
2292 tried += ( tried.length() ?
"," :
"" ) + it->url.GetHostName();
2299 if( !tried.length() )
2302 cgi[
"tried"] = tried;
2311 void XRootDMsgHandler::SwitchOnRefreshFlag()
2314 ClientRequest *req = (ClientRequest *)pRequest->GetBuffer();
2337 void XRootDMsgHandler::HandleRspOrQueue()
2342 bool finalrsp = !( pStatus.IsOK() && pStatus.code ==
suContinue );
2348 const int sst = pSendingState.fetch_or( kFinalResp );
2349 if( ( sst & kSawReadySend ) && !( sst & kSendDone ) )
2353 JobManager *jobMgr = pPostMaster->GetJobManager();
2354 if( jobMgr->IsWorker() )
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() );
2369 void XRootDMsgHandler::HandleLocalRedirect(
URL *url )
2372 log->Debug(
ExDbgMsg,
"[%s] Handling local redirect - MsgHandler: %p (message: %s ).",
2373 pUrl.GetHostId().c_str(), (
void*)
this,
2374 pRequest->GetObfuscatedDescription().c_str() );
2376 if( !pLFileHandler )
2382 AnyObject *resp = 0;
2383 pLFileHandler->SetHostList( *pHosts );
2384 XRootDStatus st = pLFileHandler->Open( url, pRequest, resp );
2391 pResponseHandler->HandleResponseWithHosts(
new XRootDStatus(),
2402 bool XRootDMsgHandler::IsRetriable()
2406 if( value ==
"true" )
return true;
2409 ClientRequest *req =
reinterpret_cast<ClientRequest*
>( pRequest->GetBuffer() );
2419 "[%s] Not allowed to retry open request (OpenRecovery disabled): %s.",
2420 pUrl.GetHostId().c_str(),
2421 pRequest->GetObfuscatedDescription().c_str() );
2434 bool XRootDMsgHandler::OmitWait(
Message &request,
const URL &url )
2437 if( !url.IsMetalink() )
2442 ClientRequest *req =
reinterpret_cast<ClientRequest*
>( request.GetBuffer() );
2449 VirtualRedirector *redirector = registry.Get( url );
2453 if( redirector->Count( request ) > 1 )
2462 bool XRootDMsgHandler::RetriableErrorResponse(
const Status &status )
2466 if( !( pLoadBalancer.url.IsValid() &&
2467 pUrl.GetLocation() != pLoadBalancer.url.GetLocation() ) )
2478 if( !( pLoadBalancer.flags &
kXR_attrMeta ) )
return false;
2490 bool ret = pNotAuthorizedCounter < limit;
2491 ++pNotAuthorizedCounter;
2496 "[%s] Reached limit of NotAuthorized retries!",
2497 pUrl.GetHostId().c_str() );
2517 void XRootDMsgHandler::DumpRedirectTraceBack()
2519 if( pRedirectTraceBack.empty() )
return;
2521 std::stringstream sstrm;
2523 sstrm <<
"Redirect trace-back:\n";
2527 auto itr = pRedirectTraceBack.begin();
2528 sstrm <<
'\t' << counter <<
". " << (*itr)->ToString() <<
'\n';
2534 for( ; itr != pRedirectTraceBack.end(); ++itr, ++prev, ++counter )
2535 sstrm <<
'\t' << counter <<
". "
2536 << (*itr)->ToString( (*prev)->status.IsOK() ) <<
'\n';
2541 bool warn = !pStatus.IsOK() &&
2544 ( pStatus.code ==
errAuthFailed && pNotAuthorizedCounter >= authlimit ) );
2548 log->Warning(
XRootDMsg,
"%s", sstrm.str().c_str() );
2550 log->Debug(
XRootDMsg,
"%s", sstrm.str().c_str() );
2555 template<
typename T>
2556 Status XRootDMsgHandler::ReadFromBuffer(
char *&buffer,
size_t &buflen, T& result )
2560 memcpy(&result, buffer,
sizeof(T));
2562 buffer +=
sizeof( T );
2563 buflen -=
sizeof( T );
2571 Status XRootDMsgHandler::ReadFromBuffer(
char *&buffer,
size_t &buflen, std::string &result )
2578 if( !( status = ReadFromBuffer( buffer, buflen, c ) ).IsOK() )
2591 Status XRootDMsgHandler::ReadFromBuffer(
char *&buffer,
size_t &buflen,
2592 size_t size, std::string &result )
2598 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
std::string obfuscateAuth(const std::string &input)
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 GetString(const std::string &key, std::string &value)
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, std::string *opathp=nullptr)
Append cgi to the one already present in the message.
The message representation used throughout the system.
const std::string & GetObfuscatedDescription() const
Get the description of the message with authz parameter obfuscated.
@ More
there are more (non-raw) data to be read
@ Ignore
Ignore the message.
StreamEvent
Events that may have occurred to the stream.
@ Ready
The stream has become connected.
static RedirectorRegistry & Instance()
Returns reference to the single instance.
static SIDMgrPool & Instance()
std::shared_ptr< SIDManager > GetSIDMgr(const URL &url)
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.
const std::string & GetPath() const
Get the path.
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
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 SetPath(const std::string &path)
Set the path.
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 InspectStatusRsp() override
friend class HandleRspJob
virtual void OnStatusReady(const Message *message, XRootDStatus status) override
The requested action has been performed and the status is available.
const Message * GetRequest() const
Get the request pointer.
virtual uint16_t Examine(std::shared_ptr< Message > &msg) override
void WaitDone(time_t now)
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 errRedirectLimit
const int DefaultMaxMetalinkWait
const uint16_t errErrorResponse
const uint16_t errTlsError
const uint16_t errOperationExpired
const uint16_t stFatal
Fatal error, it's still an error.
const uint16_t stError
An error occurred that could potentially be retried.
const uint16_t errNotFound
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
const uint16_t errNotSupported
const uint16_t errLocalError
Buffer BinaryDataInfo
Binary buffer.
const uint16_t errOperationInterrupted
const uint16_t suContinue
const int DefaultNotAuthorizedRetryLimit
const uint16_t errRedirect
const uint16_t errAuthFailed
const uint16_t errInvalidMessage
static const int PageSize
ssize_t Move(KernelBuffer &kbuff, char *&ubuff)
Describe a data chunk for vector read.
void * buffer
length of the chunk
uint32_t length
offset in the file
Procedure execution status.
uint16_t code
Error type, or additional hints on what to do.
bool IsOK() const
We're fine.
static const uint16_t ServerFlags
returns server flags
static const uint16_t ProtocolVersion
returns the protocol version