Loading...
Searching...
No Matches
XrdClInQueue.cc
Go to the documentation of this file.
//------------------------------------------------------------------------------
// Copyright (c) 2011-2012 by European Organization for Nuclear Research (CERN)
// Author: Lukasz Janyst <ljanyst@cern.ch>
//------------------------------------------------------------------------------
// XRootD is free software: you can redistribute it and/or modify
// it under the terms of the GNU Lesser General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// XRootD is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU Lesser General Public License
// along with XRootD. If not, see <http://www.gnu.org/licenses/>.
//------------------------------------------------------------------------------
#include "XProtocol/XProtocol.hh"
#include "XrdCl/XrdClInQueue.hh"
#include "XrdCl/XrdClPostMasterInterfaces.hh"
#include "XrdCl/XrdClMessage.hh"
#include "XrdCl/XrdClLog.hh"
#include "XrdCl/XrdClDefaultEnv.hh"
#include "XrdCl/XrdClConstants.hh"
#include <arpa/inet.h> // for network unmarshalling stuff
namespace XrdCl
{
//----------------------------------------------------------------------------
// Filter messages
//----------------------------------------------------------------------------
bool InQueue::DiscardMessage( Message& msg, uint16_t& sid) const
{
if( msg.GetSize() < 8 )
return true;
ServerResponse *rsp = (ServerResponse *)msg.GetBuffer();
// We only care about async responses, but those are extracted now
// in the SocketHandler
if( rsp->hdr.status == kXR_attn )
return true;
else
sid = ((uint16_t)rsp->hdr.streamid[1] << 8) | (uint16_t)rsp->hdr.streamid[0];
return false;
}
//----------------------------------------------------------------------------
// Add a listener that should be notified about incoming messages
//----------------------------------------------------------------------------
void InQueue::AddMessageHandler( MsgHandler *handler, bool &rmMsg )
{
uint16_t handlerSid = handler->GetSid();
XrdSysMutexHelper scopedLock( pMutex );
pHandlers[handlerSid] = HandlerAndExpire( handler, 0 );
}
//----------------------------------------------------------------------------
// Get a message handler interested in receiving message whose header
// is stored in msg
//----------------------------------------------------------------------------
MsgHandler *InQueue::GetHandlerForMessage( std::shared_ptr<Message> &msg,
time_t &expires,
uint16_t &action )
{
time_t exp = 0;
uint16_t act = 0;
uint16_t msgSid = 0;
MsgHandler* handler = 0;
if (DiscardMessage(*msg, msgSid))
{
return handler;
}
XrdSysMutexHelper scopedLock( pMutex );
HandlerMap::iterator it = pHandlers.find(msgSid);
if (it != pHandlers.end())
{
Log *log = DefaultEnv::GetLog();
handler = it->second.first;
act = handler->Examine( msg );
if( it->second.second == 0 ) {
it->second.second = handler->GetExpiration();
log->Debug( ExDbgMsg, "[handler: %p] Assigned expiration %lld.",
(void*)handler, (long long)it->second.second );
}
exp = it->second.second;
log->Debug( ExDbgMsg, "[msg: %p] Assigned MsgHandler: %p.",
(void*)msg.get(), (void*)handler );
if( act & MsgHandler::RemoveHandler )
{
pHandlers.erase( it );
log->Debug( ExDbgMsg, "[handler: %p] Removed MsgHandler: %p from the in-queue.",
(void*)handler, (void*)handler );
}
}
if( handler )
{
expires = exp;
action = act;
}
return handler;
}
//----------------------------------------------------------------------------
// Re-insert the handler without scanning the cached messages
//----------------------------------------------------------------------------
void InQueue::ReAddMessageHandler( MsgHandler *handler,
time_t expires )
{
uint16_t handlerSid = handler->GetSid();
XrdSysMutexHelper scopedLock( pMutex );
pHandlers[handlerSid] = HandlerAndExpire( handler, expires );
}
//----------------------------------------------------------------------------
// Remove a listener
//----------------------------------------------------------------------------
void InQueue::RemoveMessageHandler( MsgHandler *handler )
{
uint16_t handlerSid = handler->GetSid();
XrdSysMutexHelper scopedLock( pMutex );
pHandlers.erase(handlerSid);
Log *log = DefaultEnv::GetLog();
log->Debug( ExDbgMsg, "[handler: %p] Removed MsgHandler: %p from the in-queue.",
(void*)handler, (void*)handler );
}
//----------------------------------------------------------------------------
// Report an event to the handlers
//----------------------------------------------------------------------------
void InQueue::ReportStreamEvent( MsgHandler::StreamEvent event,
XRootDStatus status )
{
uint8_t action = 0;
XrdSysMutexHelper scopedLock( pMutex );
for( HandlerMap::iterator it = pHandlers.begin(); it != pHandlers.end(); )
{
action = it->second.first->OnStreamEvent( event, status );
if( action & MsgHandler::RemoveHandler )
{
auto next = it; ++next;
pHandlers.erase( it );
it = next;
}
else
++it;
}
}
//----------------------------------------------------------------------------
// Timeout handlers
//----------------------------------------------------------------------------
void InQueue::ReportTimeout( time_t now )
{
if( !now )
now = ::time(0);
XrdSysMutexHelper scopedLock( pMutex );
HandlerMap::iterator it = pHandlers.begin();
while( it != pHandlers.end() )
{
if( it->second.second && it->second.second <= now )
{
uint8_t act = it->second.first->OnStreamEvent( MsgHandler::Timeout,
Status( stError, errOperationExpired ) );
auto next = it; ++next;
if( act & MsgHandler::RemoveHandler )
pHandlers.erase( it );
it = next;
}
else
++it;
}
}
//----------------------------------------------------------------------------
// Query the handler and extract the expiration time
//----------------------------------------------------------------------------
void InQueue::AssignTimeout( MsgHandler *handler )
{
uint16_t handlerSid = handler->GetSid();
XrdSysMutexHelper scopedLock( pMutex );
HandlerMap::iterator it = pHandlers.find( handlerSid );
if( it != pHandlers.end() )
{
if( it->second.second == 0 )
{
it->second.second = handler->GetExpiration();
Log *log = DefaultEnv::GetLog();
log->Debug( ExDbgMsg, "[handler: %p] Assigned expiration %lld.",
(void*)handler, (long long)it->second.second );
}
}
}
}
XProtocol.hh
ServerResponseHeader::status
kXR_unt16 status
Definition XProtocol.hh:915
ServerResponseHeader::streamid
kXR_char streamid[2]
Definition XProtocol.hh:914
kXR_attn
@ kXR_attn
Definition XProtocol.hh:901
ServerResponse::hdr
ServerResponseHeader hdr
Definition XProtocol.hh:1288
XrdClConstants.hh
XrdClDefaultEnv.hh
XrdClInQueue.hh
XrdClLog.hh
XrdClMessage.hh
XrdClPostMasterInterfaces.hh
XrdCl::DefaultEnv::GetLog
static Log * GetLog()
Get default log.
Definition XrdClDefaultEnv.cc:493
XrdCl::InQueue::ReportTimeout
void ReportTimeout(time_t now=0)
Timeout handlers.
Definition XrdClInQueue.cc:166
XrdCl::InQueue::RemoveMessageHandler
void RemoveMessageHandler(MsgHandler *handler)
Remove a listener.
Definition XrdClInQueue.cc:129
XrdCl::InQueue::ReAddMessageHandler
void ReAddMessageHandler(MsgHandler *handler, time_t expires)
Re-insert the handler without scanning the cached messages.
Definition XrdClInQueue.cc:118
XrdCl::InQueue::ReportStreamEvent
void ReportStreamEvent(MsgHandler::StreamEvent event, XRootDStatus status)
Report an event to the handlers.
Definition XrdClInQueue.cc:143
XrdCl::InQueue::GetHandlerForMessage
MsgHandler * GetHandlerForMessage(std::shared_ptr< Message > &msg, time_t &expires, uint16_t &action)
Definition XrdClInQueue.cc:66
XrdCl::InQueue::AssignTimeout
void AssignTimeout(MsgHandler *handler)
Definition XrdClInQueue.cc:192
XrdCl::InQueue::AddMessageHandler
void AddMessageHandler(MsgHandler *handler, bool &rmMsg)
Definition XrdClInQueue.cc:54
XrdCl::Log
Handle diagnostics.
Definition XrdClLog.hh:101
XrdCl::Log::Debug
void Debug(uint64_t topic, const char *format,...)
Print a debug message.
Definition XrdClLog.cc:282
XrdCl::Message
The message representation used throughout the system.
Definition XrdClMessage.hh:32
XrdCl::MsgHandler
Message handler.
Definition XrdClPostMasterInterfaces.hh:51
XrdCl::MsgHandler::Examine
virtual uint16_t Examine(std::shared_ptr< Message > &msg)=0
XrdCl::MsgHandler::GetSid
virtual uint16_t GetSid() const =0
XrdCl::MsgHandler::RemoveHandler
@ RemoveHandler
Definition XrdClPostMasterInterfaces.hh:61
XrdCl::MsgHandler::GetExpiration
virtual time_t GetExpiration()=0
XrdCl::MsgHandler::StreamEvent
StreamEvent
Events that may have occurred to the stream.
Definition XrdClPostMasterInterfaces.hh:79
XrdCl::MsgHandler::Timeout
@ Timeout
The declared timeout has occurred.
Definition XrdClPostMasterInterfaces.hh:82
XrdCl::XRootDStatus
Request status.
Definition XrdClXRootDResponses.hh:219
XrdSysMutexHelper
Definition XrdSysPthread.hh:263
XrdCl
Definition XrdClAction.hh:34
XrdCl::errOperationExpired
const uint16_t errOperationExpired
Definition XrdClStatus.hh:90
XrdCl::stError
const uint16_t stError
An error occurred that could potentially be retried.
Definition XrdClStatus.hh:32
XrdCl::ExDbgMsg
const uint64_t ExDbgMsg
Definition XrdClConstants.hh:44
XrdCl::Status
Procedure execution status.
Definition XrdClStatus.hh:115