65 o <<
"WaitTask for: 0x" << handler->
GetRequest();
69 virtual time_t
Run( time_t now )
71 pHandler->WaitDone( now );
92 virtual ~HandleRspJob()
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.",
131 if( pOksofarAsAnswer )
134 while( pResponse ) pCV.
Wait();
141 log->
Warning(
ExDbgMsg,
"[%s] MsgHandler is examining a response although "
142 "it already owns a response: %p (message: %s ).",
148 if( msg->GetSize() < 8 )
188 pBodyReader->SetDataLength( dlen );
245 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_oksofar response to request "
249 if( !pOksofarAsAnswer )
251 pPartialResps.emplace_back( std::move( pResponse ) );
262 pTimeoutFence.store(
true, std::memory_order_relaxed );
271 pTimeoutFence.store(
true, std::memory_order_relaxed );
280 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_status response to request "
299 pTimeoutFence.store(
true, std::memory_order_relaxed );
366 pPartialResps.push_back( std::move( pResponse ) );
379 pPageReader.reset(
new AsyncPageReader( *pChunkList, pCrc32cDigests ) );
380 pPageReader->SetRsp( rspst );
395 pResponse->GetCursor() )
434 int *qryResponse =
nullptr;
436 qryResult.
Get( qryResponse );
438 pHosts->back().flags = *qryResponse;
440 qryResponse =
nullptr;
443 qryResult.
Get( qryResponse );
445 pHosts->back().protocol = *qryResponse;
471 pAggregatedWaitTime = 0;
480 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_ok response to request %s",
490 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_status response to request %s",
503 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_oksofar response to request %s",
516 char *errmsg =
new char[rsp->
hdr.
dlen-3]; errmsg[rsp->
hdr.
dlen-4] = 0;
517 memcpy( errmsg, rsp->
body.error.errmsg, rsp->
hdr.
dlen-4 );
518 log->
Dump(
XRootDMsg,
"[%s] Got a kXR_error response to request %s "
542 char *urlInfoBuff =
new char[rsp->
hdr.
dlen-3];
543 urlInfoBuff[rsp->
hdr.
dlen-4] = 0;
544 memcpy( urlInfoBuff, rsp->
body.redirect.host, rsp->
hdr.
dlen-4 );
545 std::string urlInfo = urlInfoBuff;
546 delete [] urlInfoBuff;
548 "message %s: %s, port %d", pUrl.
GetHostId().c_str(),
550 rsp->
body.redirect.port );
555 if( !pRedirectCounter )
558 "message %s, the last known error is: %s",
574 uint32_t flags = pHosts->back().flags;
575 if( !pHasLoadBalancer )
586 pLoadBalancer = pHosts->back();
587 log->
Dump(
XRootDMsg,
"[%s] Current server has been assigned "
588 "as a load-balancer for message %s",
591 HostList::iterator it;
592 for( it = pHosts->begin(); it != pHosts->end(); ++it )
593 it->loadBalancer =
false;
594 pHosts->back().loadBalancer =
true;
605 pEffectiveDataServerUrl =
new URL( pHosts->back().url );
610 std::vector<std::string> urlComponents;
614 std::ostringstream o;
616 o << urlComponents[0];
617 if( rsp->
body.redirect.port > 0 )
618 o <<
":" << rsp->
body.redirect.port <<
"/";
619 else if( rsp->
body.redirect.port < 0 )
634 std::string url( rsp->
body.redirect.host, rsp->
hdr.
dlen-4 );
640 std::string url( rsp->
body.redirect.host, rsp->
hdr.
dlen-4 );
642 pRedirectAsAnswer =
true;
646 URL newUrl =
URL( o.str() );
651 pUrl.
GetHostId().c_str(), urlInfo.c_str() );
668 std::ostringstream ossXrd;
671 for(URL::ParamsMap::const_iterator it = urlParams.begin();
672 it != urlParams.end(); ++it )
674 if( it->first.compare( 0, 4,
"xrd." ) &&
675 it->first.compare( 0, 6,
"xrdcl." ) )
678 ossXrd << it->first <<
'=' << it->second <<
'&';
681 std::string xrdCgi = ossXrd.str();
682 pRedirectUrl = newUrl.
GetURL();
685 if( urlComponents.size() > 1 )
688 pRedirectUrl += urlComponents[1];
689 std::ostringstream o;
690 o <<
"fake://fake:111//fake?";
691 o << urlComponents[1];
693 if( urlComponents.size() == 3 )
694 o <<
'?' << urlComponents[2];
700 pRedirectUrl += xrdCgi;
703 cgiURL =
URL( o.str() );
708 std::ostringstream o;
709 o <<
"fake://fake:111//fake?";
711 cgiURL =
URL( o.str() );
713 pRedirectUrl += xrdCgi;
723 pRedirectAsAnswer =
true;
725 if( pRedirectAsAnswer )
736 Status st = RewriteRequestRedirect( newUrl );
763 uint32_t waitSeconds = 0;
767 char *infoMsg =
new char[rsp->
hdr.
dlen-3];
769 memcpy( infoMsg, rsp->
body.wait.infomsg, rsp->
hdr.
dlen-4 );
770 log->
Dump(
XRootDMsg,
"[%s] Got kXR_wait response of %d seconds to "
771 "message %s: %s", pUrl.
GetHostId().c_str(),
775 waitSeconds = rsp->
body.wait.seconds;
779 log->
Dump(
XRootDMsg,
"[%s] Got kXR_wait response of 0 seconds to "
784 pAggregatedWaitTime += waitSeconds;
789 if( OmitWait( *pRequest, pLoadBalancer.
url ) )
793 if( pAggregatedWaitTime > maxWait )
805 Status st = RewriteRequestWait();
817 time_t resendTime = ::time(0)+waitSeconds;
819 if( resendTime < pExpiration )
821 log->
Debug(
ExDbgMsg,
"[%s] Scheduling WaitTask for MsgHandler: %p (message: %s ).",
826 taskMgr->
RegisterTask(
new WaitTask(
this ), resendTime );
830 log->
Debug(
XRootDMsg,
"[%s] Wait time is too long, timing out %s",
855 log->
Dump(
XRootDMsg,
"[%s] Got kXR_waitresp response of %d seconds to "
857 rsp->
body.waitresp.seconds,
867 log->
Dump(
XRootDMsg,
"[%s] Got unrecognized response %d to "
886 log->
Dump(
XRootDMsg,
"[%s] Stream event reported for msg %s",
892 if( pTimeoutFence.load( std::memory_order_relaxed ) )
895 HandleError( status );
904 uint32_t &bytesRead )
910 return pPageReader->Read( *socket, bytesRead );
912 return pBodyReader->Read( *socket, bytesRead );
926 const int sst = pSendingState.fetch_or( kSendDone );
931 if( !status.
IsOK() && ( ( sst & kFinalResp ) || ( sst & kSawResp ) ) )
933 log->
Error(
XRootDMsg,
"[%s] Unexpected error for message %s. Trying to "
936 HandleError( status );
940 if( sst & kFinalResp )
942 log->
Dump(
XRootDMsg,
"[%s] Got late notification that outgoing message %s was "
943 "sent, already have final response, queuing handler callback.",
951 log->
Dump(
XRootDMsg,
"[%s] Got late notification that message %s has "
952 "been successfully sent.",
962 log->
Dump(
XRootDMsg,
"[%s] Message %s has been successfully sent.",
972 log->
Error(
XRootDMsg,
"[%s] Impossible to send message %s. Trying to "
975 HandleError( status );
1002 uint32_t &bytesWritten )
1007 if( !pChunkList->empty() && !pCrc32cDigests.empty() )
1017 int fLen = 0, lLen = 0;
1025 uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
1026 memcpy( pPgWrtCksumBuff.
GetBuffer(), &digest,
sizeof( uint32_t ) );
1029 uint32_t btsLeft = chunk.
length - pAsyncOffset;
1030 uint32_t pglen = ( pPgWrtCurrentPageNb == 0 ? fLen :
XrdSys::PageSize ) - pPgWrtCurrentPageOffset;
1031 if( pglen > btsLeft ) pglen = btsLeft;
1032 char* pgbuf =
static_cast<char*
>( chunk.
buffer ) + pAsyncOffset;
1034 while( btsLeft > 0 )
1037 while( pPgWrtCksumBuff.
GetCursor() <
sizeof( uint32_t ) )
1039 uint32_t dgstlen =
sizeof( uint32_t ) - pPgWrtCksumBuff.
GetCursor();
1042 Status st = socket->
Send( dgstbuf, dgstlen, btswrt );
1043 if( !st.
IsOK() )
return st;
1044 bytesWritten += btswrt;
1050 Status st = socket->
Send( pgbuf, pglen, btswrt );
1051 if( !st.
IsOK() )
return st;
1055 bytesWritten += btswrt;
1056 pAsyncOffset += btswrt;
1062 ++pPgWrtCurrentPageNb;
1063 if( pPgWrtCurrentPageNb < nbpgs )
1067 uint32_t digest = htonl( pCrc32cDigests[pPgWrtCurrentPageNb] );
1068 memcpy( pPgWrtCksumBuff.
GetBuffer(), &digest,
sizeof( uint32_t ) );
1072 if( pglen > btsLeft ) pglen = btsLeft;
1074 pPgWrtCurrentPageOffset = 0;
1078 pPgWrtCurrentPageOffset += btswrt;
1082 else if( !pChunkList->empty() )
1084 size_t size = pChunkList->size();
1085 for(
size_t i = pAsyncChunkIndex ; i < size; ++i )
1087 char *buffer = (
char*)(*pChunkList)[i].buffer;
1088 uint32_t size = (*pChunkList)[i].length;
1089 size_t leftToBeWritten = size - pAsyncOffset;
1091 while( leftToBeWritten )
1094 Status st = socket->
Send( buffer + pAsyncOffset, leftToBeWritten, btswrt );
1095 bytesWritten += btswrt;
1097 pAsyncOffset += btswrt;
1098 leftToBeWritten -= btswrt;
1118 log->
Debug(
XRootDMsg,
"[%s] Channel is encrypted: cannot use kernel buffer.",
1124 pChunkList->push_back(
ChunkInfo( 0, ret, ubuff ) );
1131 while( !pKBuff->
Empty() )
1135 bytesWritten += btswrt;
1139 log->
Debug(
XRootDMsg,
"[%s] Request %s payload (kernel buffer) transferred to socket.",
1160 pTimeoutFence.store(
false, std::memory_order_relaxed );
1166 void XRootDMsgHandler::HandleResponse()
1177 const int sst = pSendingState.fetch_or( kFinalResp );
1178 if( !( sst & kSendDone ) )
1186 XRootDStatus *status = ProcessStatus();
1187 AnyObject *response = 0;
1190 log->Debug(
ExDbgMsg,
"[%s] Calling MsgHandler: %p (message: %s ) "
1194 status->ToString().c_str() );
1196 if( status->IsOK() )
1198 Status st = ParseResponse( response );
1203 status =
new XRootDStatus( st );
1213 pRdirEntry->status = *status;
1214 pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
1220 if( pSidMgr && finalrsp )
1223 if( status->IsOK() || !pMsgInFly ||
1228 HostList *hosts = pHosts.release();
1230 pHosts.reset(
new HostList( *hosts ) );
1247 pTimeoutFence.store(
false, std::memory_order_relaxed );
1256 XRootDStatus *XRootDMsgHandler::ProcessStatus()
1258 XRootDStatus *st =
new XRootDStatus( pStatus );
1263 if( !pStatus.
IsOK() && rsp )
1267 st->errNo = rsp->
body.error.errnum;
1270 std::string errmsg( rsp->
body.error.errmsg, rsp->
hdr.
dlen-5 );
1272 errmsg +=
" Last seen error: " + pLastError.
ToString();
1273 st->SetErrorMessage( errmsg );
1276 st->SetErrorMessage( pRedirectUrl );
1285 Status XRootDMsgHandler::ParseResponse( AnyObject *&response )
1299 log->Error(
XRootDMsg,
"Internal Error: unable to process redirect" );
1304 uint32_t length = 0;
1310 if( pPartialResps.empty() )
1312 buffer = rsp->
body.buffer.data;
1321 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1328 buff.Allocate( length );
1329 uint32_t offset = 0;
1330 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1333 buff.Append( part->
body.buffer.data, part->
hdr.
dlen, offset );
1336 buff.Append( rsp->
body.buffer.data, rsp->
hdr.
dlen, offset );
1337 buffer = buff.GetBuffer();
1368 AnyObject *obj =
new AnyObject();
1370 char *nullBuffer =
new char[length+1];
1371 nullBuffer[length] = 0;
1372 memcpy( nullBuffer, buffer, length );
1374 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as "
1375 "LocateInfo: %s", pUrl.
GetHostId().c_str(),
1377 LocationInfo *data =
new LocationInfo();
1379 if( data->ParseServerResponse( nullBuffer ) ==
false )
1383 delete [] nullBuffer;
1386 delete [] nullBuffer;
1398 AnyObject *obj =
new AnyObject();
1405 StatInfoVFS *data =
new StatInfoVFS();
1407 char *nullBuffer =
new char[length+1];
1408 nullBuffer[length] = 0;
1409 memcpy( nullBuffer, buffer, length );
1411 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as "
1412 "StatInfoVFS: %s", pUrl.
GetHostId().c_str(),
1415 if( data->ParseServerResponse( nullBuffer ) ==
false )
1419 delete [] nullBuffer;
1422 delete [] nullBuffer;
1431 StatInfo *data =
new StatInfo();
1433 char *nullBuffer =
new char[length+1];
1434 nullBuffer[length] = 0;
1435 memcpy( nullBuffer, buffer, length );
1437 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as StatInfo: "
1441 if( data->ParseServerResponse( nullBuffer ) ==
false )
1445 delete [] nullBuffer;
1448 delete [] nullBuffer;
1461 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as ProtocolInfo",
1467 log->Error(
XRootDMsg,
"[%s] Got invalid redirect response.",
1472 AnyObject *obj =
new AnyObject();
1473 ProtocolInfo *data =
new ProtocolInfo( rsp->
body.protocol.pval,
1474 rsp->
body.protocol.flags );
1485 AnyObject *obj =
new AnyObject();
1486 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as "
1487 "DirectoryList", pUrl.
GetHostId().c_str(),
1494 DirectoryList *data =
new DirectoryList();
1495 data->SetParentName( path );
1498 char *nullBuffer =
new char[length+1];
1499 nullBuffer[length] = 0;
1500 memcpy( nullBuffer, buffer, length );
1502 bool invalidrsp =
false;
1504 if( !pDirListStarted )
1507 pDirListStarted =
true;
1509 invalidrsp = !data->ParseServerResponse( pUrl.
GetHostId(), nullBuffer );
1512 invalidrsp = !data->ParseServerResponse( pUrl.
GetHostId(), nullBuffer, pDirListWithStat );
1518 delete [] nullBuffer;
1522 delete [] nullBuffer;
1533 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as OpenInfo",
1539 log->Error(
XRootDMsg,
"[%s] Got invalid open response.",
1544 AnyObject *obj =
new AnyObject();
1545 StatInfo *statInfo = 0;
1552 log->Dump(
XRootDMsg,
"[%s] Parsing StatInfo in response to %s",
1558 char *nullBuffer =
new char[rsp->
hdr.
dlen-11];
1559 nullBuffer[rsp->
hdr.
dlen-12] = 0;
1560 memcpy( nullBuffer, buffer+12, rsp->
hdr.
dlen-12 );
1562 statInfo =
new StatInfo();
1563 if( statInfo->ParseServerResponse( nullBuffer ) ==
false )
1568 delete [] nullBuffer;
1571 if( rsp->
hdr.
dlen < 12 || !statInfo )
1573 log->Error(
XRootDMsg,
"[%s] Unable to parse StatInfo in response "
1581 OpenInfo *data =
new OpenInfo( (uint8_t*)buffer,
1582 pResponse->GetSessionId(),
1594 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as ChunkInfo",
1598 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1604 if( pPartialResps[i]->GetSize() > 8 )
1611 if( pResponse->GetSize() > 8 )
1616 return pBodyReader->GetResponse( response );
1624 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as PageInfo",
1631 ChunkInfo chunk = pChunkList->front();
1632 bool sizeMismatch =
false;
1633 uint32_t currentOffset = 0;
1634 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1644 if( currentOffset + datalen > chunk.length )
1646 sizeMismatch =
true;
1650 currentOffset += datalen;
1654 size_t datalen = rspst->
status.
bdy.
dlen - NbPgPerRsp( rspst->
info.pgread.offset,
1656 if( currentOffset + datalen <= chunk.length )
1657 currentOffset += datalen;
1659 sizeMismatch =
true;
1664 if( pChunkStatus.front().sizeError || sizeMismatch )
1666 log->Error(
XRootDMsg,
"[%s] Handling response to %s: user supplied "
1667 "buffer is too small for the received data.",
1673 AnyObject *obj =
new AnyObject();
1674 PageInfo *pgInfo =
new PageInfo( chunk.offset, currentOffset, chunk.buffer,
1675 std::move( pCrc32cDigests) );
1687 std::vector<std::tuple<uint64_t, uint32_t>> retries;
1694 retries.reserve( pgcnt );
1698 for(
size_t i = 0; i < pgcnt; ++i )
1701 if( i == 0 ) len = cse->
dlFirst;
1702 else if( i == pgcnt - 1 ) len = cse->
dlLast;
1703 retries.push_back( std::make_tuple( pgoffs[i], len ) );
1707 RetryInfo *info =
new RetryInfo( std::move( retries ) );
1708 AnyObject *obj =
new AnyObject();
1721 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %p as "
1722 "VectorReadInfo", pUrl.
GetHostId().c_str(),
1725 for( uint32_t i = 0; i < pPartialResps.size(); ++i )
1731 if( pPartialResps[i]->GetSize() > 8 )
1738 if( pResponse->GetSize() > 8 )
1743 return pBodyReader->GetResponse( response );
1751 int len = rsp->hdr.dlen;
1752 char* data = rsp->body.buffer.data;
1754 return ParseXAttrResponse( data, len, response );
1765 AnyObject *obj =
new AnyObject();
1766 log->Dump(
XRootDMsg,
"[%s] Parsing the response to %s as BinaryData",
1771 data->Allocate( length );
1772 data->Append( buffer, length );
1785 Status XRootDMsgHandler::ParseXAttrResponse(
char *data,
size_t len,
1786 AnyObject *&response )
1799 if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
1803 if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
1806 std::vector<XAttrStatus> resp;
1808 for(
kXR_char i = 0; i < nattr; ++i )
1811 if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
1819 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1824 resp.push_back( XAttrStatus( name, st ) );
1831 response =
new AnyObject();
1832 response->Set(
new std::vector<XAttrStatus>( std::move( resp ) ) );
1842 if( !( status = ReadFromBuffer( data, len, nerrs ) ).IsOK() )
1846 if( !( status = ReadFromBuffer( data, len, nattr ) ).IsOK() )
1849 std::vector<XAttr> resp;
1850 resp.reserve( nattr );
1853 for(
kXR_char i = 0; i < nattr; ++i )
1856 if( !( status = ReadFromBuffer( data, len, rc ) ).IsOK() )
1864 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1869 resp.push_back( XAttr( name, st ) );
1873 for(
kXR_char i = 0; i < nattr; ++i )
1876 if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
1878 vlen = ntohl( vlen );
1881 if( !( status = ReadFromBuffer( data, len, vlen, value ) ).IsOK() )
1884 resp[i].value.swap( value );
1891 response =
new AnyObject();
1892 response->Set(
new std::vector<XAttr>( std::move( resp ) ) );
1900 std::vector<XAttr> resp;
1905 if( !( status = ReadFromBuffer( data, len, name ) ).IsOK() )
1909 if( !( status = ReadFromBuffer( data, len, vlen ) ).IsOK() )
1911 vlen = ntohl( vlen );
1914 if( !( status = ReadFromBuffer( data, len, vlen, value ) ).IsOK() )
1917 resp.push_back( XAttr( name, value ) );
1921 response =
new AnyObject();
1922 response->Set(
new std::vector<XAttr>( std::move( resp ) ) );
1936 Status XRootDMsgHandler::RewriteRequestRedirect(
const URL &newUrl )
1944 std::string xrdCgi =
"";
1945 std::ostringstream ossXrd;
1946 for(URL::ParamsMap::const_iterator it = newCgi.begin(); it != newCgi.end(); ++it )
1948 if( it->first.compare( 0, 4,
"xrd." ) )
1950 ossXrd << it->first <<
'=' << it->second <<
'&';
1953 xrdCgi = ossXrd.str();
1963 std::string surl = newUrl.
GetURL();
1964 (surl.find(
'?') == std::string::npos) ? (surl +=
'?') :
1965 ((*surl.rbegin() !=
'&') ? (surl +=
'&') : (surl +=
""));
1969 std::string surlLog = surl;
1973 log->Error(
XRootDMsg,
"[%s] Failed to build redirection URL from data: %s",
1974 newUrl.GetHostId().c_str(), surl.c_str());
1991 Status XRootDMsgHandler::RewriteRequestWait()
2026 void XRootDMsgHandler::HandleError( XRootDStatus status )
2034 if( pSidMgr && pMsgInFly && (
2045 if( !noreplicas ) pLastError = status;
2048 log->Debug(
XRootDMsg,
"[%s] Handling error while processing %s: %s.",
2050 status.ToString().c_str() );
2056 if( status.IsFatal() && status.code ==
errTlsError && status.errNo == EAGAIN )
2058 if( pSslErrCnt < MaxSslErrRetry )
2060 status.status &= ~stFatal;
2078 if( RetriableErrorResponse( status ) )
2080 UpdateTriedCGI(status.errNo);
2082 SwitchOnRefreshFlag();
2103 log->Error(
XRootDMsg,
"[%s] Unable to get the response to request %s",
2125 if( !status.IsFatal() && IsRetriable() )
2127 log->Info(
XRootDMsg,
"[%s] Retrying request: %s.",
2159 if( pRdirEntry ) pRedirectTraceBack.push_back( std::move( pRdirEntry ) );
2160 pRdirEntry.reset(
new RedirectEntry( pUrl.
GetLocation(), url.GetLocation(), entryType ) );
2164 pHosts->push_back( url );
2176 pSidMgr->ReleaseSID( req->
streamid );
2183 if( !url.IsLocalFile() )
2186 Status st = pSidMgr->AllocateSID( req->
streamid );
2189 log->Error(
XRootDMsg,
"[%s] Impossible to send message %s.",
2201 log->Debug(
ExDbgMsg,
"[%s] Metaling redirection for MsgHandler: %p (message: %s ).",
2205 return pPostMaster->
Redirect( pUrl, pRequest,
this );
2209 HandleLocalRedirect( &pUrl );
2214 log->Debug(
ExDbgMsg,
"[%s] Retry at server MsgHandler: %p (message: %s ).",
2217 return pPostMaster->
Send( pUrl, pRequest,
this,
true, pExpiration );
2224 void XRootDMsgHandler::UpdateTriedCGI(uint32_t errNo)
2234 if( pEffectiveDataServerUrl )
2237 delete pEffectiveDataServerUrl;
2238 pEffectiveDataServerUrl = 0;
2249 {
if (errNo ==
kXR_NotFound) cgi[
"triedrc"] =
"enoent";
2250 else if (errNo ==
kXR_IOError) cgi[
"triedrc"] =
"ioerr";
2251 else if (errNo ==
kXR_FSError) cgi[
"triedrc"] =
"fserr";
2262 HostList::reverse_iterator it;
2263 for( it = pHosts->rbegin()+1; it != pHosts->rend(); ++it )
2265 if( it->loadBalancer )
2268 tried +=
"," + it->url.GetHostName();
2275 cgi[
"tried"] = tried;
2284 void XRootDMsgHandler::SwitchOnRefreshFlag()
2310 void XRootDMsgHandler::HandleRspOrQueue()
2321 const int sst = pSendingState.fetch_or( kFinalResp );
2322 if( !( sst & kSendDone ) )
2327 if( jobMgr->IsWorker() )
2332 log->Debug(
ExDbgMsg,
"[%s] Passing to the thread-pool MsgHandler: %p (message: %s ).",
2342 void XRootDMsgHandler::HandleLocalRedirect( URL *url )
2345 log->Debug(
ExDbgMsg,
"[%s] Handling local redirect - MsgHandler: %p (message: %s ).",
2349 if( !pLFileHandler )
2355 AnyObject *resp = 0;
2357 XRootDStatus st = pLFileHandler->
Open( url, pRequest, resp );
2375 bool XRootDMsgHandler::IsRetriable()
2379 if( value ==
"true" )
return true;
2392 "[%s] Not allowed to retry open request (OpenRecovery disabled): %s.",
2407 bool XRootDMsgHandler::OmitWait( Message &request,
const URL &url )
2410 if( !url.IsMetalink() )
2422 VirtualRedirector *redirector = registry.Get( url );
2426 if( redirector->Count( request ) > 1 )
2435 bool XRootDMsgHandler::RetriableErrorResponse(
const Status &status )
2463 bool ret = pNotAuthorizedCounter < limit;
2464 ++pNotAuthorizedCounter;
2469 "[%s] Reached limit of NotAuthorized retries!",
2490 void XRootDMsgHandler::DumpRedirectTraceBack()
2492 if( pRedirectTraceBack.empty() )
return;
2494 std::stringstream sstrm;
2496 sstrm <<
"Redirect trace-back:\n";
2500 auto itr = pRedirectTraceBack.begin();
2501 sstrm <<
'\t' << counter <<
". " << (*itr)->ToString() <<
'\n';
2507 for( ; itr != pRedirectTraceBack.end(); ++itr, ++prev, ++counter )
2508 sstrm <<
'\t' << counter <<
". "
2509 << (*itr)->ToString( (*prev)->status.IsOK() ) <<
'\n';
2514 bool warn = !pStatus.
IsOK() &&
2521 log->Warning(
XRootDMsg,
"%s", sstrm.str().c_str() );
2523 log->Debug(
XRootDMsg,
"%s", sstrm.str().c_str() );
2528 template<
typename T>
2529 Status XRootDMsgHandler::ReadFromBuffer(
char *&buffer,
size_t &buflen, T& result )
2533 memcpy(&result, buffer,
sizeof(T));
2535 buffer +=
sizeof( T );
2536 buflen -=
sizeof( T );
2544 Status XRootDMsgHandler::ReadFromBuffer(
char *&buffer,
size_t &buflen, std::string &result )
2551 if( !( status = ReadFromBuffer( buffer, buflen, c ) ).IsOK() )
2564 Status XRootDMsgHandler::ReadFromBuffer(
char *&buffer,
size_t &buflen,
2565 size_t size, std::string &result )
2571 result.append( buffer, size );
union ServerResponse::@0 body
struct ClientFattrRequest fattr
#define kXR_collapseRedir
ServerResponseStatus status
struct ClientDirlistRequest dirlist
static const int kXR_ckpXeq
struct ClientOpenRequest open
struct ServerResponseBody_Status bdy
struct ClientRequestHdr header
struct ClientChkPointRequest chkpoint
struct ServerResponseHeader hdr
union ServerResponseV2::@1 info
#define kXR_PROTOCOLVERSION
struct ClientStatRequest stat
struct ClientLocateRequest locate
std::string obfuscateAuth(const std::string &input)
void Get(Type &object)
Retrieve the object being held.
void AdvanceCursor(uint32_t delta)
Advance the cursor.
char * GetBufferAtCursor()
Get the buffer pointer at the append cursor.
const char * GetBuffer(uint32_t offset=0) const
Get the message buffer.
void SetCursor(uint32_t cursor)
Set the cursor.
uint32_t GetCursor() const
Get append cursor.
static Log * GetLog()
Get default log.
static Env * GetEnv()
Get default client environment.
static bool HasStatInfo(const char *data)
Returns true if data contain stat info.
bool GetString(const std::string &key, std::string &value)
bool GetInt(const std::string &key, int &value)
Interface for a job to be run by the job manager.
void SetHostList(const HostList &hostList)
XRootDStatus Open(const std::string &url, uint16_t flags, uint16_t mode, ResponseHandler *handler, uint16_t timeout=0)
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.
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)
virtual XRootDStatus Send(const char *buffer, size_t size, int &bytesWritten)
void RegisterTask(Task *task, time_t time, bool own=true)
Interface for a task to be run by the TaskManager.
virtual time_t Run(time_t now)=0
void SetName(const std::string &name)
Set name of the task.
std::string GetHostId() const
Get the host part of the URL (user:password@host:port)
bool IsMetalink() const
Is it a URL to a metalink.
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.
std::string GetLocation() const
Get location (protocol://host:port/path)
const std::string & GetHostName() const
Get the name of the target host.
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
Buffer BinaryDataInfo
Binary buffer.
const uint16_t errOperationInterrupted
const uint16_t suContinue
const int DefaultNotAuthorizedRetryLimit
const uint16_t errRedirect
const uint16_t errAuthFailed
const uint16_t errInvalidMessage
none object for initializing empty Optional
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