CRouter Class Reference

#include <Router.h>

Public Member Functions
	CRouter ()
	~CRouter ()
int	Initialize (RouterConfig *Config)
int	Start ()
int	Stop ()
void	Put (Event *ep)
int	GetMaxNumberOfDisks ()
int	GetNumberOfActiveDisks ()
int	GetDiskServiceTime (EventStorageRequest *event, u16 DiskId)
void	SetDiskServiceTime (int Disk, double EstServiceTimeOfDisk, double EstServiceTime[301])
void	UpdateDiskServiceTime (double EstServiceTimeOfDisk[100], double EstServiceTime[100][301])
void	UpdateDiskResponseTime (double EstResponseTimeOfDisk[100], double EstResponseTime[100][301], double DevDiskResponseTimeOfDisk[100], double DevDiskResponseTime[100][301])
void	UpdateDiskQueueTime (double EstQueueTimeOfDisk[100], double EstQueueTime[100][301])
void	UpdateMeasures (int disk, int nThreads, struct timeval initial_time, struct timeval final_time)
long double	GetEstimatedQueueTimeOfAllDisks ()
long double	GetEstimatedDiskQueueTime (int disk)
void	SetNATMapping (NATData server_map, int stor_id, NATData stor_map)
NATData	GetNATMapping (NATData server_map, int stor_id)
	Funcao para obter um mapeamento passado como parametro identificado pelo par IP, porta do servidor de gerenciamento e o identificador do servidor de despacho.
void	RemoveNATMapping (NATData server_map, int stor_id)
	Funcao para remover um mapeamento passado como parametro identificado pelo par IP, porta do servidor de gerenciamento e o identificador do servidor de despacho.
bool	FindNATMapping (NATData server_map, int stor_id)
	Funcao para verificar se um mapeamento esta presente no mapa com os mapeamentos.
unsigned long long int	GetInvalidStorages (NATData ClientAddr)
	Deternima o vetor de bits com os servidores de armazenamento que nao podem ser usados pelo cliente com endereco passado como parametro, isto e, que estao inativos ou que foram invalidados (apos se tornarem novamente ativos).
void	GetStorageInfo (NATData ClientAddr, bool *CanRead, bool *CanWrite)
	Deternima se podemos ler ou escrever blocos, no nomento, para um dado cliente cujo endereco e passado como parametro.
int	GetNumberOfStorageNodes (unsigned int *NumberOfStorageNodes)
	Funcao para obter o numero de storage nodes.
void	RequestCompleted (RioStreamObj **StreamObjAddr, Event *event)
	Funcao para completar uma requisicao (chamando a RequestCompleted do objeto RioStreamObj passado como parametro) removendo, quando necessario, o objeto do stream e alterando o ponteiro passado como parametro para NULL.
Data Fields
NATMap	m_NAT_map
Private Member Functions
void	StorageReply (MsgRSSstorage *Msg)
void	AddDisk (int Disk)
void	SendStorage (StrDiskRequest *Request)
void	CleanUp ()
Event *	Get ()
void	routerthread ()
void	DataReq (Event *ep)
void	DataPrefetch (Event *event)
void	DataSendToClient (Event *event)
void	DataCancel (Event *event)
void	PrintQueues ()
void	StorageDown (int StorageId, bool EmptyStorageQueue)
	Funcao para definir que um servidor de armazenamento esta temporariamente indisponivel.
void	StorageUp (int StorageId)
	Funcao para definir que um servidor de armazenamento voltou a estar disponivel.
bool	CheckStorageStatus (int DiskId)
	Funcao usada para verificar se o servidor de armazenamento com uma dada ID esta ou nao parado.
Static Private Member Functions
static void *	routerthreadep (void *param)
Private Attributes
double	m_EstDiskServiceTime [100][301]
double	m_EstDiskServiceTimeOfDisk [100]
double	m_EstDiskResponseTime [100][301]
double	m_EstDiskResponseTimeOfDisk [100]
double	m_DevDiskResponseTime [100][301]
double	m_DevDiskResponseTimeOfDisk [100]
double	m_EstDiskQueueTime [100][301]
double	m_EstDiskQueueTimeOfDisk [100]
int	m_UpdatedEstimatedDiskServiceTime
bool	m_CollectMeasures
double	m_EstimatedParameter
pthread_mutex_t	m_MutexUpdated
pthread_cond_t	m_ConditionUpdated
ofstream	m_logSQT
ofstream	m_logRESP
pthread_t	m_thread
bool	m_initialized
bool	m_started
DiskMgr *	m_DiskMgr
CSystemManager *	m_SystemManager
EventQueue	m_Queue
int	m_nDisks
int	m_BlockSize
int	m_MaxPending
int	m_MaxQueueSize
CDiskRequestQueue *	m_RTqueue
CDiskRequestQueue *	m_NRTqueue
CDiskRequestQueue *	m_Pending
int *	m_nRT
int *	m_nNRT
int *	m_nPending
CDiskRequestManager	m_Request
StrDiskRequest **	m_TempDiskRequest
unsigned long long int	m_StoragesStatus
pthread_mutex_t	m_StoragesStatusMutex

---

Detailed Description

Definition at line 55 of file Router.h.

---

Constructor & Destructor Documentation

CRouter::CRouter

(

)

Definition at line 68 of file Router.cpp.

{
    m_UpdatedEstimatedDiskServiceTime = 0;
    pthread_mutex_init( &m_MutexUpdated,     NULL );
    pthread_cond_init(  &m_ConditionUpdated, NULL );
    // m_logSQT  does not need to be initialized
    // m_logRESP does not need to be initialized
    // m_Queue   does not need to be initialized
    m_thread          = 0;

    m_initialized     = false;
    m_started         = false;
    m_DiskMgr         = NULL;
    m_SystemManager   = NULL;
    m_nDisks          = 0;
    m_BlockSize       = 0;
    m_MaxPending      = 0;
    m_MaxQueueSize    = 0;

    m_RTqueue         = NULL;
    m_NRTqueue        = NULL;
    m_Pending         = NULL;
    m_nRT             = NULL;
    m_nNRT            = NULL;
    m_nPending        = NULL;

    // m_Request does not need to be initialized
    m_TempDiskRequest = NULL;

    m_EstimatedParameter = 0;
    for( int i = 0; i < 100; i++ )
    {
        for( int j = 0; j < 301; j++ )
        {
           // according number of active threads
           m_EstDiskServiceTime [ i ][ j ] = 0;
           m_EstDiskResponseTime[ i ][ j ] = 0;
           m_DevDiskResponseTime[ i ][ j ] = 0;

           // according queue size
           m_EstDiskQueueTime[ i ][ j ] = 0;
        }
        // for each disk - using all requests
        m_EstDiskQueueTimeOfDisk   [ i ] = 0;
        m_EstDiskServiceTimeOfDisk [ i ] = 0;
        m_EstDiskResponseTimeOfDisk[ i ] = 0;
        m_DevDiskResponseTimeOfDisk[ i ] = 0;
    }
    m_CollectMeasures = false;
    // ------------------------------------------------------------------------
    
    // Inicializa a variavel que define os servidores de armazenamento inativos
    // com 0 (pois inicialmente todos os servidores estarao ativos apos as
    // conexoes forem estabelecidas).
    m_StoragesStatus = 0;
    // Inicializa o mutex para garantir o acesso exclusivo a variavel 
    // m_StoragesStatus.
    pthread_mutex_init( &m_StoragesStatusMutex, NULL );
}

CRouter::~CRouter

(

)

Definition at line 129 of file Router.cpp.

{
    CleanUp();

    // Remove o mutex para garantir o acesso exclusivo a variavel 
    // m_StoragesStatus.
    pthread_mutex_destroy( &m_StoragesStatusMutex );
}

---

Member Function Documentation

void CRouter::AddDisk

(

int

Disk

)

[private]

Definition at line 1579 of file Router.cpp.

{
    RioErr << " Router: Adding disk: " << Disk << endl;
    if( ( Disk < 0 ) || ( Disk >= m_nDisks ) )
    {
        RioErr << "Router ERROR: Tried to add invalid disk " << Disk << endl;
        return;
    }

    if( m_nRT[Disk] != ( m_MaxQueueSize + 1 ) )
    {
        RioErr << "Router ERROR: Tried to add disk already active "
               << Disk << endl;
        return;
    }

    char name[1025];
    m_DiskMgr->GetDiskName( Disk, name );

    RioStorageNodeInfo sn_info;
    m_DiskMgr->GetDiskStorageNodeInfo( Disk, &sn_info );

    AddDiskEvent *it_event = new AddDiskEvent( sn_info.Hostname, name );
    m_SystemManager->PostITEvent( (MonitorEvent *) it_event );

    // Reset disk queue size (now active)
    m_nRT[Disk] = 0;
}

bool CRouter::CheckStorageStatus

(

int

DiskId

)

[private]

Funcao usada para verificar se o servidor de armazenamento com uma dada ID esta ou nao parado.

Parameters:

DiskId

identificador do disco do servidor de armazenamento.

Returns:

true se o servidor esta funcionando ou false se o servidor parou de funcionar.

Definition at line 2333 of file Router.cpp.

{
    bool Status;
    int StorageId;
    // Verifica se a DiskId e valida
    if( DiskId >= 0 )
    {
        // Descobre o identificador do servidor de armazenamento.
        StorageId = ( DiskId - 1 ) / SNode::sn_maxdisks;
        // Obtem acesso exclusivo a variavel m_StoragesStatus. 
        pthread_mutex_lock( &m_StoragesStatusMutex );
        // Verifica se o servidor esta ativo (se for o caso, o seu bit deve ser
        // igual a 0).
        Status = ( ( m_StoragesStatus & ( 1ull << StorageId ) ) == 0 );
        // Libera o acesso exclusivo a variavel m_StoragesStatus. 
        pthread_mutex_unlock( &m_StoragesStatusMutex );
    }
    else
        Status = false;
    return Status;
}

void CRouter::CleanUp

(

)

[private]

Definition at line 139 of file Router.cpp.

{
    m_initialized     = false;
    m_started         = false;

    m_DiskMgr      = 0;
    m_nDisks       = 0;
    m_BlockSize    = 0;
    m_MaxPending   = 0;
    m_MaxQueueSize = 0;

    //Pra tese: Verificar estes if's abaixo. Estao muito estranhos!!!
    //Nao deveriam ser do tipo if( x != 0 ) em vez de if(x == 0)?!?!?
    //Obs: Corrigido, pois existiam erros de memoria.
    if( m_RTqueue != 0 )
    {
        delete[] m_RTqueue;
        m_RTqueue = 0;
    }

    if( m_NRTqueue != 0 )
    {
        delete[] m_NRTqueue;
        m_NRTqueue = 0;
    }

    if( m_Pending != 0 )
    {
        delete[] m_Pending;
        m_Pending = 0;
    }

    if( m_nRT != 0 )
    {
        delete[] m_nRT;
        m_nRT = 0;
    }

    if( m_nNRT != 0 )
    {
        delete[] m_nNRT;
        m_nNRT = 0;
    }

    if( m_nPending != 0 )
    {
        delete[] m_nPending;
        m_nPending = 0;
    }

    if( m_TempDiskRequest != 0 )
    {
        delete[] m_TempDiskRequest;
        m_TempDiskRequest = 0;
    }
}

void CRouter::DataCancel

(

Event *

event

)

[private]

Definition at line 739 of file Router.cpp.

{
    DataRequest& Request = (( EventDataRequest* )( event ))->Request;

    #ifdef RIO_DEBUG2
    RioErr << "Router:DataCancel => block "<<  Request.Block
           << " BufferId " << Request.BufferId << endl;
    #endif

    if( Request.RequestCounter == -1 )
    {
        #ifdef RIO_DEBUG2
        RioErr << "Router:DataCancel => did not request this block "
               << "yet, so just free it" << endl;
        #endif

        //the request was not submit to storage yet, so just free request
        Request.Status = 0;
        RequestCompleted( &Request.streamobj, event );
    }
    else
    {
        StrDiskRequest* m_TempDiskRequest = m_Request.Get( Request.BufferId );

        if( m_TempDiskRequest )
        {
            if( ( m_TempDiskRequest->StorageId != 0 ) &&
                ( CheckStorageStatus( m_TempDiskRequest->Disk ) ) )  
            {
                if( m_CollectMeasures )
                {
                    // update queue size to get estimated queue time
                    Request.QueueSize = m_nRT[m_TempDiskRequest->Disk];
                }

                if( m_nPending[m_TempDiskRequest->Disk] < m_MaxPending )
                {
                    m_Pending[m_TempDiskRequest->Disk].Put( m_TempDiskRequest );
                    // Update disk queue sizes
                    m_nPending[m_TempDiskRequest->Disk]++;
                    m_nRT[m_TempDiskRequest->Disk]++;
                    SendStorage(m_TempDiskRequest);
                }
                // otherwise put request on disk queue
                else
                {
                    m_RTqueue[m_TempDiskRequest->Disk].Put( m_TempDiskRequest );
                    m_nRT[m_TempDiskRequest->Disk]++;
                }
                char name[1025];
                m_DiskMgr->GetDiskName( m_TempDiskRequest->Disk, name );

                RioStorageNodeInfo sn_info;
                m_DiskMgr->GetDiskStorageNodeInfo( m_TempDiskRequest->Disk, &sn_info );

                int queue_sum = m_nRT[ m_TempDiskRequest->Disk ]
                              + m_nNRT[ m_TempDiskRequest->Disk ]
                              + m_nPending[ m_TempDiskRequest->Disk ];

                UpdateDiskQueueEvent *it_event = new UpdateDiskQueueEvent(
                    sn_info.Hostname, name, queue_sum );
                m_SystemManager->PostITEvent( (MonitorEvent *)it_event );
            }
            else
            {
                RioErr << " Router:Datacancel => StorageId == 0 "
                       << "(RequestCompleted) or storage is down" << endl;
                m_Request.Free(m_TempDiskRequest);
                RequestCompleted( &Request.streamobj, event );
            }
        }
        else if ( m_TempDiskRequest->StorageId == 0 )
        {
            RioErr << "ERROR: Router:Datacancel => Could not find resquest to cancel"
                   << endl;
        } 
        else // Novo erro indicando que o servidor de armazenamento com o disco
             // esta temporariamente indisponivel.
        {
            RioErr << "ERROR: Router:Datacancel => Could not send message to "
                   << "storage to cancel the resquest" << endl;       
        }     
    }
}

void CRouter::DataPrefetch

(

Event *

event

)

[private]

Definition at line 487 of file Router.cpp.

{
    NATData ClientAddr, ServerAddr;
    int storageid;
    //
    // TODO: Remover a variavel seguir quando a RioNeti e a NetMgr nao forem 
    // mais usadas.
    //
    unsigned totalStorages;

    DataRequest& Request = (( EventDataRequest* )( event ))->Request;
    StrDiskRequest* m_TempDiskRequest = m_Request.New();
    
    // Obtem o par IP, porta do cliente
    ClientAddr.nat_addr = Request.Target.IPaddress;
    ClientAddr.nat_port = Request.Target.Port;

    if( m_TempDiskRequest == 0 )
    {
        RioErr << "Router ERROR: Too many disk requests (UNEXPECTED)" << endl;
        Request.Status = ERROR_ROUTER + ERROR_UNEXPECTED;
        RequestCompleted( &Request.streamobj, event );
        return;
    }
    
    if( !CheckStorageStatus( m_TempDiskRequest->Disk ) )
    {
        RioErr << "Router ERROR (DataPrefetch): Cannot access disk: storage " 
               << "is down" << endl;
        Request.Status = ERROR_ROUTER + ERROR_SERVICE_TEMPORARY_UNAVAILABLE;
        RequestCompleted( &Request.streamobj, event );
        return;
    }
    
    // Descobre o numero de storages.
    //
    // TODO: Remover o comando seguir quando a RioNeti e a NetMgr nao forem mais
    // usadas.
    //
    m_DiskMgr->GetNumberOfStorageNodes( &totalStorages );
    
    // Descobre a id do storage.
    storageid = ( m_TempDiskRequest->Disk - 1 ) / SNode::sn_maxdisks + 1;
    // Ajusta a id para as copias em tempo nao real, caso existam pares 
    // IP, porta associados a nova implementacao de transmissao de dados.
    // (estes pares comecarao a partir da posicao do ultimo storage no vetor
    // mais 1, e o numero de pares sera o numero de storages.
    //
    // TODO: Remover todo o codigo dentro do if a seguir, incluindo o 
    // comando if, quando a RioNeti e a NetMgr nao forem mais usadas.
    //
    if( ( ( Request.Operation == NonRealTimeRead ) ||
          ( Request.Operation == NonRealTimeWrite ) ) && 
        ( FindNATMapping( ClientAddr, storageid + totalStorages + 1 ) ) )
    {    
         #ifdef RIO_DEBUG2
         RioErr << "Router:DataPrefetch usando o indice " 
                << storageid + totalStorages + 1 << " ao inves do indice "
                << storageid << "." << endl;
         #endif
         
         storageid = storageid + totalStorages + 1;
    }
    #ifdef RIO_DEBUG2
    else
        RioErr << "Router:DataPrefetch o novo indice " 
               << storageid + totalStorages + 1 << " nao foi usado. "
               << "Request.Operation = " << Request.Operation << "." << endl;
    #endif
    
    
    // Verifica se o disco deste servidor de armazenamento pode ser usado pelo
    // cliente.
    ServerAddr = GetNATMapping( ClientAddr, storageid );
    if( ( ServerAddr.nat_addr == 0 ) && ( ServerAddr.nat_port == 0 ) )
    {
        RioErr << "Router ERROR (DataPrefetch): Cannot access disk: storage " 
               << "is disabled for this client" << endl;
        Request.Status = ERROR_ROUTER + ERROR_SERVICE_TEMPORARY_UNAVAILABLE;
        RequestCompleted( &Request.streamobj, event );
        return;
    }
    

    #ifdef RIO_DEBUG2
    RioErr << "Router:DataPrefetch => block "  <<  Request.Block
           << " disk " << Request.Reps[0].disk << endl;
    #endif

    m_TempDiskRequest->Disk  = Request.Reps[0].disk;
    m_TempDiskRequest->Pos   = (((u64) Request.Reps[0].block ) * m_BlockSize );
    m_TempDiskRequest->StorageId = 0xffffffff;  // Invalid storage id
    m_TempDiskRequest->Size  = m_BlockSize;
    m_TempDiskRequest->event = ( EventDataRequest* ) event;

    if( m_CollectMeasures )
    {
        // update queue size to get estimated queue time
        Request.QueueSize = m_nRT[m_TempDiskRequest->Disk];
    }
    if( m_nPending[m_TempDiskRequest->Disk] < m_MaxPending )
    {
        m_Pending[m_TempDiskRequest->Disk].Put( m_TempDiskRequest );
        // Update disk queue sizes
        m_nPending[m_TempDiskRequest->Disk]++;
        m_nRT[m_TempDiskRequest->Disk]++;
        SendStorage(m_TempDiskRequest);
    }
    // otherwise put request on disk queue
    else
    {
        m_RTqueue[m_TempDiskRequest->Disk].Put( m_TempDiskRequest );
        m_nRT[m_TempDiskRequest->Disk]++;
    }

    char name[1025];
    m_DiskMgr->GetDiskName( m_TempDiskRequest->Disk, name );

    RioStorageNodeInfo sn_info;
    m_DiskMgr->GetDiskStorageNodeInfo( m_TempDiskRequest->Disk, &sn_info );

    int queue_sum = m_nRT[ m_TempDiskRequest->Disk ]
                  + m_nNRT[ m_TempDiskRequest->Disk ]
                  + m_nPending[ m_TempDiskRequest->Disk ];

    UpdateDiskQueueEvent *it_event = new UpdateDiskQueueEvent( sn_info.Hostname, name,
                                                            queue_sum );
    m_SystemManager->PostITEvent( (MonitorEvent *)it_event );
}

void CRouter::DataReq

(

Event *

ep

)

[private]

Definition at line 825 of file Router.cpp.

{
    int  i, j, rc;

    // Get Request from RequestEle
    DataRequest& Request = (( EventDataRequest* )( event ))->Request;

    #ifdef RIO_DEBUG2
    struct in_addr clientip;
    clientip.s_addr = Request.Target.IPaddress;
    RioErr << "No DataReq do Router: "
           << "IP " << inet_ntoa(clientip)
           << " port "  << ntohs( Request.Target.Port )
           << " reqid " << Request.Reqid
           << " block " << Request.Block
           << endl;
    #endif

    Request.streamobj->Stream()->getMutexUpdateEvent();

    if( Request.Operation == RealTimePrefetchBlock )
    {
        DataPrefetch( event );
        Request.streamobj->Stream()->releaseMutexUpdateEvent();
    }
    else if( Request.Operation == RealTimeCancelBlock )
    {
        DataCancel( event );
        Request.streamobj->Stream()->releaseMutexUpdateEvent();
    }
    else if( Request.Operation == RealTimeSendBlock )
    {
        DataSendToClient( event );
        Request.streamobj->Stream()->releaseMutexUpdateEvent();
    }
    // ------------------------------------------------------------------------
    else
    {
        Request.streamobj->Stream()->releaseMutexUpdateEvent();
        ///////////////////////////////////////
        //RioErr << "Router:DataReq => RealTimeRead or NonRealTime[Read|Write]"
        //       << endl;

        // ### before request was queued to Router should have
        // increased pending count in streamobj so that won't be deleted
        // while I/O running?

        // -- map request to absolute blocks
        // ### this could be a lot cleaner!!
        //  shouldn't know about internals of RioDiskBlock...

        rc = Request.rioobject->MapBlock( Request.Block, Request.RepBits,
                                          &Request.RepNum, Request.Reps );

        if( rc )
        {
            // failed in mapping logical block to physical block
            //  must send error status to client

            RioErr << "Router ERROR: Failed mapping logical block." << endl;

            Request.Status = ERROR_ROUTER + ERROR_INVALID_BLOCK;
            RequestCompleted( &Request.streamobj, event );
            return;
        }

        // Compute number of disk requests for this data request
        int nDiskRequests;
        if( Request.Operation == NonRealTimeWrite )
        {
            // For write, need to send disk request to all replicas
            nDiskRequests = Request.RepNum;
        }
        else if( ( Request.Operation == RealTimeRead ) ||
                ( Request.Operation == NonRealTimeRead ))
        {
            // For read send disk request to just one replica
            nDiskRequests = 1;
        }
        else
        {
            RioErr << "Router ERROR: Got data request with invalid operation: "
                   << (int) Request.Operation << endl;
            Request.Status = ERROR_ROUTER + ERROR_UNEXPECTED;
            RequestCompleted( &Request.streamobj, event );
            return;
        }

        if( nDiskRequests > MaxReplications )
        {
            RioErr << "Router ERROR: Got data request with too many replicas: "
                   << nDiskRequests << endl;
            Request.Status = ERROR_ROUTER + ERROR_UNEXPECTED;
            RequestCompleted( &Request.streamobj, event );
            return;
        }
       
        // allocate disk requests
        for( i = 0; i < nDiskRequests; i++ )
        {
            m_TempDiskRequest[i] = m_Request.New();

            // This should not happen if number of data requests does
            // not exceed maximum.
            if( m_TempDiskRequest[i] == 0 )
            {
                // If there is no available disk request, free previous
                // allocated disk requests and send error status to client
                for( j = 0; j < i; j++ )
                {
                    m_Request.Free( m_TempDiskRequest[j] );
                }
                RioErr << "Router ERROR: Too many disk requests (UNEXPECTED)"
                       << endl;
                Request.Status = ERROR_ROUTER + ERROR_UNEXPECTED;
                RequestCompleted( &Request.streamobj, event );
                return;
            } 
        }

        // Fill disk request fields: Disk and Pos
        if( ( Request.Operation == RealTimeRead ) ||
            ( Request.Operation == NonRealTimeRead ))
        {
            // find shortest disk queue
            int Min  = m_MaxQueueSize + 1;
            int Rep  = -1;
            int nrep = Request.RepNum;
            int dk;
            int nQueue;
            int storageid;
            // TODO: Esta variavel devera ser removida quando a NetMgr e a 
            // RioNeti forem definitivamente removidas do codigo.
            unsigned int totalStorages;
            // Obtem o endereco do servidor de despacho usado como base do 
            // mapeamento
            NATData input( Request.Target.IPaddress, Request.Target.Port );
            // TODO: Inicializa totalStorages com o numero de storages. Este
            // codigo devera ser removido quando a NetMgr e a RioNeti forem 
            // definitivamente removidas do codigo.
            m_DiskMgr->GetNumberOfStorageNodes( &totalStorages );
            
            for( i = 0; i < nrep ; i++ )
            {
                // Garante que somente os discos do servidor de armazenamento
                // ativo sejam utilizados.
                dk = Request.Reps[i].disk;
                if( CheckStorageStatus( dk ) )
                {
                    // Descobre a ID do storage a partir do disco acessado.
                    storageid = ( dk - 1 ) / ( SNode::sn_maxdisks ) + 1;
                    // Se o cliente estiver atrás de NAT, o par (IP, porta) que 
                    // o servidor informar ao storage são diferentes, variam 
                    // para cada storage.
                    // Para descobrir qual par deve ser usado, deve-se consultar 
                    // o heap m_NAT_map, que sera usado como o par IP, porta a 
                    // ser enviado ao servidor de armazenamento
                    // Obs: Note que agora sempre deveremos ter o par IP, porta 
                    // do cliente no mapa, porque agora sempre enviamos o 
                    // mapeamento para o servidor, independentemente de o 
                    // cliente estar ou nao atras de NAT.
                    //
                    // TODO: Precisamos verificar agora se estamos usando a nova
                    // ou a implementacao antiga. Todo o codigo dentro do if e 
                    // do seu else devem ser removidos quando a NetMgr e a 
                    // RioNeti forem definitivamente retiradas do codigo.
                    if( ( Request.Operation == NonRealTimeRead )  && 
                        ( FindNATMapping( input, storageid + 
                                          totalStorages + 1 ) ) )
                    {
                        #ifdef RIO_DEBUG2
                        RioErr << "CRouter::DataReq usando o indice " 
                               << storageid + totalStorages + 1 << " ao inves "
                               << "do indice " << storageid << "." << endl;
                        #endif
                
                        storageid = storageid + totalStorages + 1;
                    }
                    #ifdef RIO_DEBUG2
                    else
                        RioErr << "Router::DataReq o novo indice " 
                               << storageid + totalStorages + 1 << " nao foi "
                               << "usado. Request.Operation = " 
                               << Request.Operation << "." << endl;
                    #endif
                    
                    NATData result = m_NAT_map.getElement( input, storageid );
                    if( ( result.nat_addr != -1 ) || ( result.nat_port != 0 ) )
                    { 
                        if( Request.Operation == RealTimeRead )
                            nQueue = m_nRT[dk];
                        else
                            nQueue = m_nNRT[dk] + m_nRT[dk];
                        if( nQueue < Min )
                        {
                            Min = nQueue;
                            Rep = i;
                            //ClientAddr = result;
                        }
                    } 
                    #ifdef RIO_DEBUG2
                    else
                        RioErr << "Router WARNING: storage with ID " 
                               << ( ( dk - 1 ) / SNode::sn_maxdisks ) 
                               << " cannot be used" << endl; 
                    #endif           
                }
                #ifdef RIO_DEBUG2
                else
                    RioErr << "Router WARNING: storage with ID " 
                           << ( ( dk - 1 ) / SNode::sn_maxdisks ) 
                           << " is down" << endl;
                #endif            
            }

            if( Rep == -1 )
            {
                // no disk queue found -- fail request
                RioErr << "Router ERROR: Unavailable disk on read block " 
                       << Request.Block << " with reqid " << Request.Reqid 
                       << endl;
                m_Request.Free( m_TempDiskRequest[0] );
                Request.Status = ERROR_ROUTER + ERROR_INVALID_DISK;
                RequestCompleted( &Request.streamobj, event );
                return;
            }

           m_TempDiskRequest[0]->Disk = Request.Reps[Rep].disk;
           m_TempDiskRequest[0]->Pos  =
           ( ( (u64) Request.Reps[Rep].block ) * m_BlockSize );
        }
        // Write case
        else
        {
            int nQueue;
            int dk;
            int storageid;
            // TODO: Esta variavel devera ser removida quando a NetMgr e a 
            // RioNeti forem definitivamente removidas do codigo.
            unsigned int totalStorages;
            // Obtem o endereco do servidor de despacho usado como base do 
            // mapeamento
            NATData input( Request.Target.IPaddress, Request.Target.Port );
            // TODO: Inicializa totalStorages com o numero de storages. Este
            // codigo devera ser removido quando a NetMgr e a RioNeti forem 
            // definitivamente removidas do codigo.
            m_DiskMgr->GetNumberOfStorageNodes( &totalStorages );

            for( i = 0; i < nDiskRequests ; i++ )
            {
                dk = Request.Reps[i].disk;
                if( Request.Operation == NonRealTimeWrite )
                    nQueue = m_nNRT[dk] + m_nRT[dk];
                else
                    nQueue = m_nRT[dk];

                // Descobre a ID do storage a partir do disco acessado.
                storageid = ( dk - 1 ) / ( SNode::sn_maxdisks ) + 1;
                // Se o cliente estiver atrás de NAT, o par (IP, porta) 
                // que o servidor informar ao storage são diferentes, 
                // variam para cada storage.
                // Para descobrir qual par deve ser usado, deve-se 
                // consultar o heap m_NAT_map, que sera usado como o par 
                // IP, porta a ser enviado ao servidor de armazenamento.
                // Obs: Note que agora sempre deveremos ter o par IP, 
                // porta do cliente no mapa, porque agora sempre 
                // enviamos o mapeamento para o servidor, 
                // independentemente de o cliente estar ou nao atras de 
                // NAT.
                //
                // TODO: Precisamos verificar agora se estamos usando a nova
                // ou a implementacao antiga. Todo o codigo dentro do if e 
                // do seu else devem ser removidos quando a NetMgr e a 
                // RioNeti forem definitivamente retiradas do codigo.
                if( ( Request.Operation == NonRealTimeWrite )  &&  
                    ( FindNATMapping( input, storageid + totalStorages + 1 ) ) )
                {
                    #ifdef RIO_DEBUG2
                    RioErr << "CRouter::DataReq usando o indice " 
                           << storageid + totalStorages + 1 << " ao inves do "
                           << "indice " << storageid << "." << endl;
                    #endif
              
                    storageid = storageid + totalStorages + 1;
                }
                #ifdef RIO_DEBUG2
                else
                    RioErr << "Router::DataReq o novo indice " 
                           << storageid + totalStorages + 1 << " nao foi "
                           << "usado. Request.Operation = " 
                           << Request.Operation << "." << endl;
                #endif
                
                NATData result = m_NAT_map.getElement( input, storageid );

                if( ( nQueue > m_MaxQueueSize ) || 
                    ( !CheckStorageStatus( dk ) ) || 
                    ( ( result.nat_addr == -1 ) && ( result.nat_port == 0 ) ) )
                {
                    RioErr << "Router ERROR: Unavailable disk " << dk 
                           << " on write block " << Request.Block 
                           << " with reqid " << Request.Reqid << endl;
                    RioErr << "nQueue " << nQueue << " m_MaxQueueSize "
                           << m_MaxQueueSize << endl;
                    for( j = 0; j < nDiskRequests; j++ )
                    {
                        m_Request.Free( m_TempDiskRequest[j] );
                    }
                    Request.Status = ERROR_ROUTER + ERROR_INVALID_DISK;
                    RequestCompleted( &Request.streamobj, event );
                    return;
                }
                // Se o disco e valido, inicializamos a sua requisicao.
                m_TempDiskRequest[i]->Disk = dk;
                m_TempDiskRequest[i]->Pos  =
                                (((u64) Request.Reps[i].block ) * m_BlockSize );
            }
        }

        Request.RequestCounter = nDiskRequests;

        // Informa que a funcao RequestCompleted deve ser chamada.
        Request.RequestCancelled = false;        

        //Filling Disk request fields
        int dk;
        // Fill other fields of disk requests and include it of disk queue
        for( i = 0; i < nDiskRequests ; i++ )
        {
            
            #ifdef RIO_DEBUG2
            RioErr << "CRouter::DataReq colocando a requicao para a "
                   << "replicacao " << i+1 << " na fila do disco " 
                   << m_TempDiskRequest[ i ]->Disk << endl;
            #endif
                   
            m_TempDiskRequest[i]->StorageId = 0xffffffff;// Invalid storage id
            m_TempDiskRequest[i]->Size      = m_BlockSize;
            m_TempDiskRequest[i]->event     = ( EventDataRequest* ) event;
            dk = m_TempDiskRequest[i]->Disk;

            // Send request to storage node if there are not many pending
            // requests
            if( m_nPending[dk] < m_MaxPending )
            {
                m_Pending[dk].Put( m_TempDiskRequest[i] );
                // update disk queue sizes
                m_nPending[dk]++;
                if( ( Request.Operation == NonRealTimeWrite ) ||
                    ( Request.Operation == NonRealTimeRead ))
                {
                    m_nNRT[dk]++;
                }
                else
                {
                    if( m_CollectMeasures )
                    {
                        // update queue size to get estimated queue time
                        Request.QueueSize = m_nRT[dk];
                    }
                    // ----------------------------------------------------
                    m_nRT[dk]++;
                }
                
                SendStorage(m_TempDiskRequest[i]);
            }
            // otherwise put request on disk queue
            else // else do if( m_nPending[dk] < m_MaxPending )
            {
                // insert disk request at real time or non real time queue
                if( ( Request.Operation == NonRealTimeWrite ) ||
                    ( Request.Operation == NonRealTimeRead ))
                {
                    m_NRTqueue[dk].Put( m_TempDiskRequest[i]);
                    m_nNRT[dk]++;
                }
                else
                {
                    m_RTqueue[dk].Put( m_TempDiskRequest[i]);
                    if( m_CollectMeasures )
                    {
                        // update queue size to get estimated queue time
                        Request.QueueSize = m_nRT[dk];
                    }
                    // --------------------------------------------------------
                    m_nRT[dk]++;
                    // --------------------------------------------------------
                }
            } // fim do else do if( m_nPending[dk] < m_MaxPending )

            char name[1025];
            m_DiskMgr->GetDiskName( dk, name );

            RioStorageNodeInfo sn_info;
            m_DiskMgr->GetDiskStorageNodeInfo( dk, &sn_info );

            int queue_sum = m_nRT[ dk ]
                            + m_nNRT[ dk ]
                            + m_nPending[ dk ];

            UpdateDiskQueueEvent *it_event = new UpdateDiskQueueEvent(
                                            sn_info.Hostname, name, queue_sum );
            m_SystemManager->PostITEvent( (MonitorEvent *)it_event );
        } // fim do for( i = 0; i < nDiskRequests ; i++ )
    } // fim do else
}

void CRouter::DataSendToClient

(

Event *

event

)

[private]

Definition at line 620 of file Router.cpp.

{
    NATData ClientAddr, ServerAddr;
    bool StorageUp, StorageEnabled;
    int storageid;
    //
    // TODO: Remover a variavel seguir quando a RioNeti e a NetMgr nao forem 
    // mais usadas.
    //
    unsigned int totalStorages;

    DataRequest& Request = (( EventDataRequest* )( event ))->Request;
    StrDiskRequest* m_TempDiskRequest = m_Request.Get( Request.BufferId );
    
    StorageUp = CheckStorageStatus( m_TempDiskRequest->Disk );

    // Obtem o par IP, porta do cliente
    ClientAddr.nat_addr = Request.Target.IPaddress;
    ClientAddr.nat_port = Request.Target.Port;

    // Descobre o numero de storages.
    //
    // TODO: Remover o comando seguir quando a RioNeti e a NetMgr nao forem mais
    // usadas.
    //
    m_DiskMgr->GetNumberOfStorageNodes( &totalStorages );
    
    // Descobre a id do storage.
    storageid = ( m_TempDiskRequest->Disk - 1 ) / SNode::sn_maxdisks + 1;
    // Ajusta a id para as copias em tempo nao real, caso existam pares 
    // IP, porta associados a nova implementacao de transmissao de dados.
    // (estes pares comecarao a partir da posicao do ultimo storage no vetor
    // mais 1, e o numero de pares sera o numero de storages.
    //
    // TODO: Remover a variavel seguir quando a RioNeti e a NetMgr nao forem 
    // mais usadas.
    //
    if( ( ( Request.Operation == NonRealTimeRead ) ||
          ( Request.Operation == NonRealTimeWrite ) ) && 
        ( FindNATMapping( ClientAddr, storageid + totalStorages + 1 ) ) )
    {
         #ifdef RIO_DEBUG2
         RioErr << "Router:DataSendToClient usando o indice " 
                << storageid + totalStorages + 1 << " ao inves do indice "
                << storageid << "." << endl;
         #endif
                
         storageid = storageid + totalStorages + 1;
    }
    #ifdef RIO_DEBUG2
    else
        RioErr << "Router:DataSendToClient o novo indice " 
               << storageid + totalStorages + 1 << " nao foi usado. "
               << "Request.Operation = " << Request.Operation << "." << endl;
    #endif
    
    // Verifica se o disco deste servidor de armazenamento pode ser usado pelo
    // cliente.
    ServerAddr = GetNATMapping( ClientAddr, storageid );
    StorageEnabled = ( ( ServerAddr.nat_addr != 0 ) || 
                       ( ServerAddr.nat_port != 0 ) );

    if( ( m_TempDiskRequest ) && ( StorageUp ) && ( StorageEnabled ) )   
    {

        #ifdef RIO_DEBUG2
        RioErr << "Router:DataSendToClient => block "<<  Request.Block
               << " bufferid " << Request.BufferId
               << " disk " << m_TempDiskRequest->Disk << endl;
        #endif

        //update bufferstatus
        Request.streamobj->Stream()->decServerBufferStatus();
        //update tokenstosend
        Request.streamobj->Stream()->decTokensToSend();

        m_Pending[m_TempDiskRequest->Disk].Put( m_TempDiskRequest );

        if( m_CollectMeasures )
        {
            // don't use this sample
            Request.QueueSize = -1;
        }
        m_nPending[m_TempDiskRequest->Disk]++;
        SendStorage(m_TempDiskRequest);
        
        char name[1025];
        m_DiskMgr->GetDiskName( m_TempDiskRequest->Disk, name );

        RioStorageNodeInfo sn_info;
        m_DiskMgr->GetDiskStorageNodeInfo( m_TempDiskRequest->Disk, &sn_info );

        int queue_sum = m_nRT[ m_TempDiskRequest->Disk ]
                      + m_nNRT[ m_TempDiskRequest->Disk ]
                      + m_nPending[ m_TempDiskRequest->Disk ];

        UpdateDiskQueueEvent *it_event = new UpdateDiskQueueEvent( sn_info.Hostname, name,
                                                                queue_sum );
        m_SystemManager->PostITEvent( (MonitorEvent *) it_event);
    }
    else if( m_TempDiskRequest == 0 )
    {
        RioErr << "ERROR: Router:DataSendToClient => Could not find request to send"
               << endl;
    }
    else if( StorageEnabled ) 
    {
        RioErr << "Router ERROR (DataSendToClient): Cannot access disk: " 
               << "storage is down" << endl;
    }
    else  
    {
        RioErr << "Router ERROR (DataSendToClient): Cannot access disk: " 
               << "storage is disabled for this client" << endl;
    }

}

bool CRouter::FindNATMapping	(	NATData	server_map,
		int	stor_id
	)

Funcao para verificar se um mapeamento esta presente no mapa com os mapeamentos.

TODO: Remover a funcao quando a RioNeti e a NetMgr nao forem mais usadas.

Parameters:

	server_map	estrutura com o par IP, porta do servidor
	stor_id	identificador do servidor de armazenamento.

Returns:

true se o mapeamento esta presente ou false caso o mapeamento nao esteja presente.

Definition at line 2142 of file Router.cpp.

{
    return m_NAT_map.findElement( server_map, stor_id );
}

Event* CRouter::Get

(

void

)

[inline, private]

Definition at line 146 of file Router.h.

{ return m_Queue.Get(); }

int CRouter::GetDiskServiceTime	(	EventStorageRequest *	event,
		u16	DiskId
	)

Definition at line 1983 of file Router.cpp.

{
    int result;
    result = m_DiskMgr->GetDiskServiceTime( event, DiskId );
    if( result == -1 )
    {
        m_EstDiskServiceTimeOfDisk[ DiskId ] = -1;
    }
    return result;
}

long double CRouter::GetEstimatedDiskQueueTime

(

int

disk

)

Definition at line 1945 of file Router.cpp.

{
    return m_EstDiskQueueTimeOfDisk[disk];
}

long double CRouter::GetEstimatedQueueTimeOfAllDisks

(

)

unsigned long long int CRouter::GetInvalidStorages

(

NATData

ClientAddr

)

Deternima o vetor de bits com os servidores de armazenamento que nao podem ser usados pelo cliente com endereco passado como parametro, isto e, que estao inativos ou que foram invalidados (apos se tornarem novamente ativos).

Parameters:

ClientAddr

endereco UDP do cliente. Se os campos da estrutura NATData forem ambos 0, somente iremos considerar se os servidores estao ou nao ativos. Em caso contrario, tambem vamos contar os servidores que nao podem ser usados pelo cliente.

Returns:

vetor com as informacoes dos storages inativos. Cada bit esta associado a um servidor de armazenamento e, quando igual a 1, indica que o servidor associado ao bit nao pode ser usado (e 0 quando pode ser usado).

Definition at line 2358 of file Router.cpp.

{
    unsigned long long int InvalidStorages;
    unsigned int TotalStorages;
    NATData StorageAddr;

    // Determina o numero de servidores de armazenamento ativos.
    m_DiskMgr->GetNumberOfStorageNodes( &TotalStorages );
    
    // Obtem acesso exclusivo a variavel m_StoragesStatus. 
    pthread_mutex_lock( &m_StoragesStatusMutex );
    // Inicialmente os servidores que nao podem ser usados sao os que estao
    // inativos.
    InvalidStorages = m_StoragesStatus;
    // Libera o acesso exclusivo a variavel m_StoragesStatus. 
    pthread_mutex_unlock( &m_StoragesStatusMutex );
    // Adiciona os servidores de armazenamento que se tornaram ativos depois de
    // o cliente ter criado a conexao e que, portanto, nao podem ser usados.
    if( ( ClientAddr.nat_addr != 0 ) && ( ClientAddr.nat_port != 0 ) )
    {
        for( unsigned int Storage = 0; Storage < TotalStorages; Storage++ )
        {
            // Devemos adicionar mais 1 ao chamar GetNATMapping porque, no mapa
            // de mapeamentos do NAT, os servidores de armazenamento estao a 
            // partir da posicao 1 (a posicao 0 e usada pelo servidor de 
            // gerenciamento).
            StorageAddr = GetNATMapping( ClientAddr, Storage + 1 );
            if( ( StorageAddr.nat_addr == -1 ) && 
                ( StorageAddr.nat_port == 0 ) )
            {
                
                #ifdef RIO_DEBUG2
                RioErr << "CRouter::GetInvalidStorages o servidor de "
                       << "armazenamento com a ID " << Storage << " nao pode "
                       << "ser usado pelo cliente pois seu mapeamento foi "
                       << "invalidado." << endl;
                #endif
                       
                InvalidStorages = InvalidStorages | ( 1ull << Storage );
            } 
            
            // Verifica se o outro endereco para o storage para a nova classe e
            // valido (se existir). Este codigo devera ser removido quando a 
            // classe NetMgr for definitivamente retirada do codigo.
            //
            // TODO: Remover todo o codigo dentro do if a seguir, incluindo o 
            // comando if, quando a RioNeti e a NetMgr nao forem mais usadas.
            //
            if( FindNATMapping( ClientAddr, Storage + TotalStorages + 1 ) )
            {
                StorageAddr = GetNATMapping( ClientAddr, Storage + 
                                             TotalStorages + 1 );
                if( ( StorageAddr.nat_addr == -1 ) && 
                    ( StorageAddr.nat_port == 0 ) )
                {
                
                    #ifdef RIO_DEBUG2
                    RioErr << "CRouter::GetInvalidStorages o servidor de "
                           << "armazenamento com a ID " << Storage << " nao "
                           << "pode ser usado pelo cliente pois seu mapeamento "
                           << " foi invalidado (na nova implementacao)." 
                           << endl;
                    #endif
                       
                    InvalidStorages = InvalidStorages | ( 1ull << Storage );
                }
            }
            #ifdef RIO_DEBUG2
            else
            RioErr << "CRouter::GetInvalidStorages a ID "  
                   << Storage + TotalStorages + 1 << " nao existe no mapa e "
                   << "nao foi considerada!" << endl;
            #endif
        }
    } 
    #ifdef RIO_DEBUG2
    // Imprime o vetor com os servidores que nao podem ser usados.
    RioErr << "CRouter::GetInvalidStorages Vetor com a identificacao dos "
           << "servidores de armazenamento que nao podem ser usados: ";
    for( int p = MAXNUMSTORAGES - 1; p >= 0 ; p-- )
    {
        unsigned int q = ( InvalidStorages >> p ) & 1ull; 
        RioErr << q;
    }
    RioErr << endl;
    #endif
           
    return InvalidStorages;
}

int CRouter::GetMaxNumberOfDisks

(

)

Definition at line 1967 of file Router.cpp.

{
    return m_DiskMgr->GetMaxNumberOfDisks();
}

NATData CRouter::GetNATMapping	(	NATData	server_map,
		int	stor_id
	)

Funcao para obter um mapeamento passado como parametro identificado pelo par IP, porta do servidor de gerenciamento e o identificador do servidor de despacho.

Parameters:

	server_map	estrutura com o par IP, porta do servidor
	stor_id	identificador do servidor de armazenamento.

Definition at line 2126 of file Router.cpp.

{
    return m_NAT_map.getElement( server_map, stor_id );
}

int CRouter::GetNumberOfActiveDisks

(

)

Definition at line 1974 of file Router.cpp.

{
    unsigned int numd;
    m_DiskMgr->GetNumberOfActiveDisks( &numd);
    return (int) numd;
}

int CRouter::GetNumberOfStorageNodes

(

unsigned int *

NumberOfStorageNodes

)

Funcao para obter o numero de storage nodes.

Parameters:

NumberOfStorageNodes

ponteiro para um inteiro nao sinalizado que armazenara o numero de storages.

Returns:

S_OK se nenhum erro ocorreu ao obter o numero de storages, ou um valor diferente de S_OK em caso contrario.

Definition at line 2493 of file Router.cpp.

{
    return m_DiskMgr->GetNumberOfStorageNodes( NumberOfStorageNodes );
}

void CRouter::GetStorageInfo	(	NATData	ClientAddr,
		bool *	CanRead,
		bool *	CanWrite
	)

Deternima se podemos ler ou escrever blocos, no nomento, para um dado cliente cujo endereco e passado como parametro.

Parameters:

	ClientAddr	endereco UDP do cliente. Se os campos da estrutura NATData forem ambos 0, somente iremos considerar se os servidores estao ou nao ativos. Em caso contrario, tambem vamos contar os servidores que nao podem ser usados pelo cliente.
	CanRead	ponteiro para um valor booleano que informa se uma leitura de um bloco pode ser feita. Consideraremos que uma leitura nao podera ser feita se nao for possivel garantir que um bloco pode ser lido, isto e, se o numero de servidores de armazenamento inativos for maior ou igual ao numero de replicacoes.
	CanWrite	ponteiro para um valor booleano que informa se uma escrita de um bloco pode ser feita. Consideraremos que uma escrita nao podera ser feita se nao for possivel garantir um disco em um servidor de armazenamento diferente para cada replicacao do bloco, isto e, se o numero de servidores de armazenamento ativos for menor do que o numero de replicacoes.

Definition at line 2450 of file Router.cpp.

{
    unsigned long long int InvalidStorages;
    unsigned int TotalStorages;
    int InactiveStorages, ActiveStorages;
    int NumberOfReplications;

    // Determina o numero de servidores de armazenamento ativos.
    m_DiskMgr->GetNumberOfStorageNodes( &TotalStorages );
    // Determina o numero de replicacoes.
    NumberOfReplications = m_DiskMgr->GetNumberOfReplications();
    // Determina quais servidores de armazenamento nao podem ser usados pelo
    // cliente
    InvalidStorages = GetInvalidStorages( ClientAddr );     
    // Conta o numero de servidores de armazenamento nao ativos.
    InactiveStorages = 0;
    for( unsigned int i = 0; i < MAXNUMSTORAGES; i++ )
        if( ( InvalidStorages & ( 1ull << i ) ) != 0 )
            ( InactiveStorages )++;
    // Define o numero de servidores ativos (igual ao numero total de servidores
    // menos os servidores inativos).        
    ActiveStorages = TotalStorages - InactiveStorages;
    // A leitura somente podera ser feita, com garantia de sucesso, se o numero
    // de servidores inativos for menor do que o numero de replicacoes.
    *CanRead = ( InactiveStorages < NumberOfReplications );
    // A escrita somente podera ser feita se o numero de servidores ativos for
    // maior ou igual ao numero de replicacoes pois, em caso contrario, nao 
    // existirao servidores suficientes para salvarmos todas as replicacoes de
    // um bloco.
    *CanWrite = ( ActiveStorages >= NumberOfReplications );
    #ifdef RIO_DEBUG2         
    RioErr << "RioSession::GetStorageInfo estado dos servidores de "
           << "armazenamento ao receber um pedido de dados : ativos = " 
           << ActiveStorages << ", inativos = " << InactiveStorages 
           << ", totais = " << TotalStorages << ", replicacoes = "
           << NumberOfReplications << ", CanRead = " 
           << ( *CanRead ? "true": "false" ) << ", CanWrite = " 
           << ( *CanWrite ? "true": "false" ) << endl;
    #endif        
}

int CRouter::Initialize

(

RouterConfig *

Config

)

Definition at line 199 of file Router.cpp.

{
    int i;

    if( m_initialized )
    {
        RioErr << "Router.Initialize: already initialized" << endl;
        return ERROR_ROUTER + ERROR_INITIALIZED;
    }

    // Update number of disks
    m_nDisks = Config->nDisks;

    // Update block size
    m_BlockSize = Config->BlockSize;

    // Update maximum number of pending requests per disk
    m_MaxPending = Config->MaxPending;

    // Update flag to collect measures
    m_CollectMeasures = Config->CollectMeasures;
    // Update Parameter to collect measures
    m_EstimatedParameter = Config->EstimatedTimeParameter;
    m_UpdatedEstimatedDiskServiceTime = 0;

    #ifdef __QUEUE_RESP_LOG
    if( m_CollectMeasures )
    {
        // open files to save measures
        // Variavel usada para compor os nomes dos arquivos com os logs.
        char LogFileName[ MaxPathSize ];
        // Compoe o nome do arquivo com os logs.
        strcpy( LogFileName, Config->LogsDirectory );
        strcat( LogFileName, LOGRESP );
        m_logRESP.open( LogFileName );
        strcpy( LogFileName, Config->LogsDirectory );
        strcat( LogFileName, LOGSQT );
        m_logSQT.open ( LogFileName );
        m_logRESP << "#Measures in msec" << endl;
        m_logSQT  << "#Measures in msec" << endl;
    }
    #endif
    // ------------------------------------------------------------------------

    m_DiskMgr       = Config->DiskManager;
    m_SystemManager = Config->SystemManager;

    // allocate disk queues. 3 queues: real-time, non real-time and
    // pending requests (i.e. submited to storage node )

    m_RTqueue = new CDiskRequestQueue[m_nDisks];
    if( m_RTqueue == 0 )
    {
        RioErr << "Router.Initialize: Out of memory" << endl;
        CleanUp();
        return ERROR_ROUTER + ERROR_MEMORY;
    }

    m_NRTqueue = new CDiskRequestQueue[m_nDisks];
    if( m_NRTqueue == 0 )
    {
        RioErr << "Router.Initialize: Out of memory" << endl;
        CleanUp();
        return ERROR_ROUTER + ERROR_MEMORY;
    }

    m_Pending = new CDiskRequestQueue[m_nDisks];
    if( m_Pending == 0 )
    {
        RioErr << "Router.Initialize: Out of memory" << endl;
        CleanUp();
        return ERROR_ROUTER + ERROR_MEMORY;
    }
    m_nRT = new int [m_nDisks];
    if( m_nRT == 0 )
    {
        RioErr << "Router.Initialize: Out of memory" << endl;
        CleanUp();
        return ERROR_ROUTER + ERROR_MEMORY;
    }

    m_nNRT = new int [m_nDisks];
    if( m_nNRT == 0 )
    {
        RioErr << "Router.Initialize: Out of memory" << endl;
        CleanUp();
        return ERROR_ROUTER + ERROR_MEMORY;
    }

    m_nPending = new int [m_nDisks];
    if( m_nPending == 0 )
    {
        RioErr << "Router.Initialize: Out of memory" << endl;
        CleanUp();
        return ERROR_ROUTER + ERROR_MEMORY;
    }

    m_MaxQueueSize = ( Config->MaxDataRequests * MaxReplications );

    for( i = 0; i < m_nDisks; i++ )
    {
        // Set size of real time queues to large value
        // This indicates that the disk is not available
        m_nRT[i] = m_MaxQueueSize + 1;
        m_nNRT[i] = 0;
        m_nPending[i] = 0;
    }

    m_TempDiskRequest = new StrDiskRequest* [MaxReplications];
    if( m_TempDiskRequest == 0 )
    {
        RioErr << "Router.Initialize: Out of memory" << endl;
        CleanUp();
        return ERROR_ROUTER + ERROR_MEMORY;
    }

    // number of requests must be >= maxbuffer * maxclients so set
    // maxrequests to max value.
    unsigned int maxrequests = 0xffff;

    #ifdef RIO_DEBUG2
    RioErr << "Router: Max number of Requests " << maxrequests << endl;
    #endif
    // ------------------------------------------------------------------------

    int hResult;
    hResult = m_Request.Initialize( maxrequests );
    if( hResult )
    {
        RioErr << "Could not initialize m_Requests at Router"
               << ". (" << maxrequests << ")" << endl;
        return hResult;
    }

    m_initialized = true;

    return S_OK;
}

void CRouter::PrintQueues

(

)

[private]

Definition at line 1951 of file Router.cpp.

{
    int i;
    RioErr << "Queues:" << endl;
    for( i = 0; i < m_nDisks; i++ )
    {
        RioErr << "  Disk " << i << ": "
               << m_nRT[i] << " "
               << m_nNRT[i] << " "
               << m_nPending[i] << endl;

    }
}

void CRouter::Put

(

Event *

ep

)

Definition at line 2092 of file Router.cpp.

{
    EventDataRequest* event;
    event = ( EventDataRequest *) ep;

    if( ( ( ep->Type==EventTypeRTDataRequest  ) ||
          ( ep->Type==EventTypeNRTDataRequest ) 
        ) && ( m_CollectMeasures )
      )
    {
        gettimeofday( &event->Request.ArrivalTime, 0 );
    }

    m_Queue.Put(ep);
}

void CRouter::RemoveNATMapping	(	NATData	server_map,
		int	stor_id
	)

Funcao para remover um mapeamento passado como parametro identificado pelo par IP, porta do servidor de gerenciamento e o identificador do servidor de despacho.

Parameters:

	server_map	estrutura com o par IP, porta do servidor
	stor_id	identificador do servidor de armazenamento.

Returns:

mapeamento para o servidor de armazenamenro.

Definition at line 2118 of file Router.cpp.

{
    m_NAT_map.removeElement( server_map, stor_id );
}

void CRouter::RequestCompleted	(	RioStreamObj **	StreamObjAddr,
		Event *	event
	)

Funcao para completar uma requisicao (chamando a RequestCompleted do objeto RioStreamObj passado como parametro) removendo, quando necessario, o objeto do stream e alterando o ponteiro passado como parametro para NULL.

Obs: Se o objeto passado for NULL, indicando que o objeto foi incorretamente removido, um erro sera impresso no log do servidor informando deste erro.

Parameters:

	StreamObjAddr	ponteiro com o endereco com o ponteiro para o objeto para o qual deveremos chamar a funcao RequestCompleted (isso e necessario porque, se removermos o objeto, o ponteiro para ele devera ser alterado para NULL).
	event	ponteiro do evento a ser passado a funcao RequestCompleted.

Definition at line 2501 of file Router.cpp.

{
    // Variavel booleana para armazenar a informacao de se o objeto de stream
    // deve ou nao ser removido nesta funcao.
    bool RemoveStreamObj;

    if( ( *StreamObjAddr ) != NULL )
    {
        ( *StreamObjAddr )->RequestCompleted( event, &RemoveStreamObj );
        if( RemoveStreamObj )
        {
             // Neste caso, tratamos da ultima requisicao recebida e o stream ja
             // foi fechado, mas nao foi removido por causa das requisicoes que
             // ainda estavam pendentes quando ele foi fechado. Entao, devemos
             // remover o stream e alterar o ponteiro passado como parametro
             // para NULL, para podermos detectar eventuais erros futuros. Note
             // que o ponteiro nao deveria ser usado em outras partes do Router
             // porque, nestas partes, o stream ainda deveria estar aberto.
             //RioErr << "CRouter::RequestCompleted deletando o objeto "
             //        << "RioStreamObj cujo ponteiro e igual a 0x " << hex
             //    << ( unsigned long ) *StreamObjAddr << dec << "." << endl;
             delete *StreamObjAddr;
             *StreamObjAddr = NULL;
        }
    }
    else
    {
         RioErr << "[CRouter] RequestCompleted nao foi possivel executar a "
                << "funcao porque foi passado um ponteiro NULO!" << endl;
    }
}

void CRouter::routerthread

(

)

[private]

Definition at line 398 of file Router.cpp.

{
    Event* event;

    RioErr << "ROUTERTHREADID " << syscall( SYS_gettid ) << endl;

    while( 1 )
    {
        event = Get();

        #ifdef RIO_DEBUG2    
        RioErr << "CRouter::routerthread processando o evento com o tipo "
               << event->Type << endl;
        #endif       

        switch( event->Type )
        {
            case EventTypeRTDataRequest:
            case EventTypeNRTDataRequest:
                DataReq( event );
                break;

            case EventTypeStorageReply:
                // -- free specific request for that disk
                StorageReply(& (((EventStorageReply*)(event))->StorageReply ));
                EventManager.Free(event);
                break;

            case EventTypeAddDisk:
                AddDisk((( EventAddDisk* )( event ))->Disk );
                EventManager.Free(event);
                break;
                
            // Novos eventos usados para tratar do caso em que um servidor de
            // armazenamento cai e depois volta a estar disponivel.
            // Evento para tratar a queda de um servidor de armazenamento.
            case EventTypeStorageDown:
            
                #ifdef RIO_DEBUG2
                RioErr << "CRouter::routerthread recebida a mensagem "
                       << "EventTypeStorageDown da classe DiskMgr informando "
                       << "que o servidor de armazanamento com a ID "
                       << ( ( EventStorageDown* ) ( event ) )->StorageId 
                       << " parou de funcionar" << endl;
                #endif       
                
                StorageDown( ( ( EventStorageDown* ) ( event ) )->StorageId,
                             ( ( EventStorageDown* ) ( event ) )->
                                EmptyStorageQueue );
                EventManager.Free( event );
                break;
            // Evento para tratar a volta de um servidor de armazenamento.
            case EventTypeStorageUp:
            
                #ifdef RIO_DEBUG2
                RioErr << "CRouter::routerthread recebida a mensagem "
                       << "EventTypeStorageUp da classe DiskMgr informando "
                       << "que o servidor de armazanamento com a ID "
                       << ( ( EventStorageDown* ) ( event ) )->StorageId 
                       << " voltou a funcionar" << endl;
                #endif       
                       
                StorageUp( ( ( EventStorageUp* )( event ) )->StorageId );
                EventManager.Free( event );
                break;
                
            // Novo evento usado para parar a thread do Router.
            case EventTypeFinalizeThread:
                #ifdef RIO_DEBUG2
                RioErr << "CRouter::routerthread recebida a mensagem "
                       << "EventTypeFinalizeThread informando que a thread "
                       << "deve ser finalizada" << endl;
                #endif
                EventManager.Free( event );
                //pthread_exit( NULL );
                return;       
                break;    

            default:
                RioErr << "Router.Routerthread unknown element type "
                       << (int *) event->Type << endl;
                EventManager.Free(event);
        }
    }
}

void * CRouter::routerthreadep

(

void *

param

)

[static, private]

Definition at line 391 of file Router.cpp.

{
    ( (CRouter *) parm)->routerthread();
    return 0;
}

void CRouter::SendStorage

(

StrDiskRequest *

Request

)

[private]

Definition at line 1609 of file Router.cpp.

{
    EventStorageRequest* event = NULL;

    #ifdef RIO_DEBUG2
    int bufferid = ((EventDataRequest*)(Request->event))->Request.BufferId;
    RioErr << "Router:SendStorage => RouterId " << Request->Id
           << " BufferId " << bufferid
           << " Disk " << Request->Disk
           << " pending requests " <<  m_nPending[Request->Disk] << endl;

    if( m_nPending[Request->Disk] > m_MaxPending )
    {
        RioErr << "Router:SendStorage => "
               << " Disk " << Request->Disk
               << " pending requests " <<  m_nPending[Request->Disk]
               << " RT " <<m_nRT[Request->Disk] << endl;
    }
    #endif

    if( ( m_CollectMeasures ) && ( Request->event->Request.QueueSize != -1 ) )
    {
        struct timeval now;
        gettimeofday( &now, 0 );
        double sample_msec = getInterval( Request->event->Request.ArrivalTime,
                                          now );
        Request->event->Request.ArrivalTime = now;

        // updates m_EstimatedQueueTime of the disk
        int disk = Request->event->Request.Reps[0].disk;
        int queuesize = Request->event->Request.QueueSize;

        if( queuesize <  m_MaxPending + 1 )
            queuesize = 0;
        else if( queuesize > ( m_MaxPending + 300 ))
            queuesize = 300;
        else
            queuesize = queuesize - m_MaxPending;

        if( m_EstDiskQueueTime[ disk ][ queuesize ] == 0 )
        {
            m_EstDiskQueueTime[ disk ][ queuesize ] = sample_msec;
        }
        else
        {
            m_EstDiskQueueTime[ disk ][ queuesize ] =
                                                 (( 1 - m_EstimatedParameter) *
                                     m_EstDiskQueueTime[ disk ][ queuesize ]) +
                                            m_EstimatedParameter * sample_msec;
        }
        if( m_EstDiskQueueTimeOfDisk[ disk ] == 0 )
        {
            m_EstDiskQueueTimeOfDisk[ disk ] = sample_msec;
        }
        else
        {
            m_EstDiskQueueTimeOfDisk[ disk ] =    ( 1 - m_EstimatedParameter) *
                                             m_EstDiskQueueTimeOfDisk[ disk ] +
                                            m_EstimatedParameter * sample_msec;
        }

        #ifdef __QUEUE_RESP_LOG
        m_logSQT << " disk "   << disk
                 << " sample " << sample_msec
                 << " est "    << m_EstDiskQueueTimeOfDisk[ disk ] << endl;
        #endif
    }

    if( Request->event->Request.Operation == RealTimePrefetchBlock )
    {
        MsgRSSnewRequest* msg;

        Request->event->Request.RequestCounter = 1;

        event = (EventStorageRequest*) EventManager.New(
                                                    EventTypeStorageRequest );
        msg = &( event->StorageRequest.Request.New );
        msg->Type = MSG_RSS_FETCH | RSS_INTERMEDIATE;
        msg->Size = SizeMsgRSSnewRequest;
        msg->Token = RSS_TOKEN_STORAGE;
        // Doesn't know yet ClientId
        msg->ClientId = ((EventDataRequest*)(Request->event))->Request.Reqid;
        msg->RouterId = Request->Id;
        // Doesn't know yet IP
        msg->IPaddr =
          ((EventDataRequest*)(Request->event))->Request.Target.IPaddress;
        // Doesn't know yet Port
        msg->Port = ((EventDataRequest*)(Request->event))->Request.Target.Port;
        // Se o cliente estiver atrás de NAT, o par (IP, porta) que o servidor
        // deve informar ao storage são diferentes, variam para cada storage
        // Para descobrir qual par deve ser usado, deve-se consultar o heap
        // m_NAT_map. Caso a resposta seja o par default (0, 0), o cliente não
        // está atrás de NAT e o par atual deve ser mantido; caso contrário,
        // utiliza-se o resultado da consulta
        // Aparentemente este trecho de código não é necessário, visto que o IP
        // e porta não são conhecidos neste momento. Se preciso, descomentar
        // o código abaixo e testar.
        //NATData input( msg->IPaddr, msg->Port );
        //NATData result = m_NAT_map.getElement( input,
        //                           (Request->Disk-1)/(SNode::sn_maxdisks)+1 );
        //if( result.nat_addr != 0 )
        //{
        //    msg->IPaddr = result.nat_addr;
        //    msg->Port = result.nat_port;
        //}
        msg->Disk = Request->Disk;
        msg->Pos = Request->Pos;
        msg->DataSize = Request->Size;

        #ifdef RIO_DEBUG2
        RioErr << " prefetch RouterId "<< Request->Id
               << " ClientId "<< msg->ClientId
               << " msg-Type (F+Inter)" << msg->Type << endl;
        #endif
    }
    else if( Request->event->Request.Operation == RealTimeCancelBlock )
    {
        if( Request->event->Request.BufferId == -1 )
        {
            #ifdef RIO_DEBUG2
            RioErr << "request block was not submit to storage yet, "
                   << " so just free request" << endl;  
            #endif

            m_nRT[Request->Disk]--;
            m_nPending[Request->Disk]--;

            //the request was not submit to storage yet, so just free request
            Request->event->Request.Status = 0;
            RequestCompleted( &Request->event->Request.streamobj,
                              ( Event* ) ( Request->event ) );
            char name[1025];
            m_DiskMgr->GetDiskName( Request->Disk, name );

            RioStorageNodeInfo sn_info;
            m_DiskMgr->GetDiskStorageNodeInfo( Request->Disk, &sn_info );

            int queue_sum = m_nRT[ Request->Disk ]
                          + m_nNRT[ Request->Disk ]
                          + m_nPending[ Request->Disk ];

            UpdateDiskQueueEvent *it_event = new UpdateDiskQueueEvent(
                sn_info.Hostname, name, queue_sum );
            m_SystemManager->PostITEvent( (MonitorEvent *) it_event );
            return;
        }
        else
        {
            MsgRSSpendRequest* msg;
            event = (EventStorageRequest*) EventManager.New(
                                                    EventTypeStorageRequest );
            msg = &( event->StorageRequest.Request.Pending );
            msg->Type = MSG_RSS_CANCEL;
            msg->Size = SizeMsgRSSpendRequest;
            msg->Token = RSS_TOKEN_STORAGE;
            //Update ClientId
            msg->ClientId = ((EventDataRequest*)(
                                            Request->event))->Request.Reqid;
            //Update StorageId
            msg->StorageId = Request->StorageId;
            msg->RouterId = Request->Id;
            //Update IP
            msg->IPaddr = ((EventDataRequest*)(
                                    Request->event))->Request.Target.IPaddress;
            //Update Port
            msg->Port = ((EventDataRequest*)(
                                    Request->event))->Request.Target.Port;
            // Se o cliente estiver atrás de NAT, o par (IP, porta) que o 
            // servidor deve informar ao storage são diferentes, variam para 
            // cada storage
            // Para descobrir qual par deve ser usado, deve-se consultar o heap
            // m_NAT_map. Caso a resposta seja o par default (0, 0), o cliente 
            // não está atrás de NAT e o par atual deve ser mantido; caso 
            // contrário, utiliza-se o resultado da consulta
            NATData input( msg->IPaddr, msg->Port );
            NATData result = m_NAT_map.getElement( input,
                                   (Request->Disk-1)/(SNode::sn_maxdisks)+1 );
            if( result.nat_addr != 0 )
            {
                msg->IPaddr = result.nat_addr;
                msg->Port = result.nat_port;
            }

            #ifdef RIO_DEBUG2
            RioErr << " cancel RouterId " << Request->Id
                   << " ClientId "        << msg->ClientId
                   << " StorageId "       << msg->StorageId
                   << " msg-Type C "      << msg->Type << endl;
            #endif
        }
    }
    else if( Request->event->Request.Operation == RealTimeSendBlock )
    {
        MsgRSSpendRequest* msg;
        event = (EventStorageRequest*) EventManager.New(
                                                    EventTypeStorageRequest );
        msg = &( event->StorageRequest.Request.Pending );
        msg->Type = MSG_RSS_SEND;
        msg->Size = SizeMsgRSSpendRequest;
        msg->Token = RSS_TOKEN_STORAGE;
        //Update ClientId
        msg->ClientId = ((EventDataRequest*)(Request->event))->Request.Reqid;
        //Update StorageId
        msg->StorageId = Request->StorageId;
        msg->RouterId = Request->Id;
        //Update IP
        msg->IPaddr = ((EventDataRequest*)(
                                    Request->event))->Request.Target.IPaddress;
        //Update Port
        msg->Port = ((EventDataRequest*)(Request->event))->Request.Target.Port;
        // Se o cliente estiver atrás de NAT, o par (IP, porta) que o servidor
        // deve informar ao storage são diferentes, variam para cada storage
        // Para descobrir qual par deve ser usado, deve-se consultar o heap
        // m_NAT_map. Caso a resposta seja o par default (0, 0), o cliente não
        // está atrás de NAT e o par atual deve ser mantido; caso contrário,
        // utiliza-se o resultado da consulta
        NATData input( msg->IPaddr, msg->Port );
        NATData result = m_NAT_map.getElement( input,
                                   (Request->Disk-1)/(SNode::sn_maxdisks)+1 );
        if( result.nat_addr != 0 )
        {
            msg->IPaddr = result.nat_addr;
            msg->Port = result.nat_port;
        }

        /*not send ack*/
        msg->StreamTraffic = ((Event*)(Request->event))->Type;
        
        // Adiciona a mensagem a taxa de transferencia do Video.
        ((EventDataRequest*)(Request->event))->Request.rioobject->GetVideoRate( &msg->VideoRate );

        #ifdef RIO_DEBUG2
        RioErr << " send RouterId " << Request->Id
               << " ClientId "      << msg->ClientId
               << " StorageId "     << msg->StorageId
               << " msg-Type S "    << msg->Type << endl;
        #endif
    }
    // ------------------------------------------------------------------------
    else
    {
        MsgRSSnewRequest* msg;
        // Armazena o identificador do storage.
        int storageid;
        // TODO: Esta variavel devera ser removida quando a NetMgr e a 
        // RioNeti forem definitivamente removidas do codigo.
        unsigned int totalStorages;
        
        event = (EventStorageRequest*)EventManager.New(EventTypeStorageRequest);
        msg = &( event->StorageRequest.Request.New );
        if( (  Request->event->Request.Operation == RealTimeRead ) ||
           (  Request->event->Request.Operation == NonRealTimeRead))
            msg->Type = MSG_RSS_READ;
        else
            msg->Type = MSG_RSS_WRITE;

        msg->Size     = SizeMsgRSSnewRequest;
        msg->Token    = RSS_TOKEN_STORAGE;
        msg->ClientId = ((EventDataRequest*)(Request->event))->Request.Reqid;
        msg->RouterId = Request->Id;
        msg->IPaddr   = ((EventDataRequest*)(
                                    Request->event))->Request.Target.IPaddress;
        msg->Port     = ((EventDataRequest*)(Request->event))->Request.Target.Port;
        
        // Descobre o identificador do storage.
        storageid = ( Request->Disk - 1 )/( SNode::sn_maxdisks ) + 1; 

        // TODO: Inicializa totalStorages com o numero de storages. Este codigo
        // devera ser removido quando a NetMgr e a RioNeti forem 
        // definitivamente removidas do codigo.
        m_DiskMgr->GetNumberOfStorageNodes( &totalStorages );
        
        // Obtem o endereco do servidor de despacho usado como base do 
        // mapeamento.
        NATData input( msg->IPaddr, msg->Port );

        // Se o cliente estiver atrás de NAT, o par (IP, porta) que o servidor
        // deve informar ao storage são diferentes, variam para cada storage
        // Para descobrir qual par deve ser usado, deve-se consultar o heap
        // m_NAT_map. Caso a resposta seja o par default (0, 0), o cliente não
        // está atrás de NAT e o par atual deve ser mantido; caso contrário,
        // utiliza-se o resultado da consulta
        //
        // TODO: Precisamos verificar agora se estamos usando a nova ou a 
        // implementacao antiga. Todo o codigo dentro do if e do seu else devem 
        // ser removidos quando a NetMgr e a RioNeti forem definitivamente 
        // retiradas do codigo.
        if( ( ( Request->event->Request.Operation == NonRealTimeRead ) ||
              ( Request->event->Request.Operation == NonRealTimeWrite ) ) && 
              ( FindNATMapping( input, storageid + totalStorages + 1 ) ) )
        {
            #ifdef RIO_DEBUG2
            RioErr << "CRouter::SendStorage usando o indice "  
                   << storageid + totalStorages + 1 << " ao inves do indice " 
                   << storageid << "." << endl;
            #endif
              
            storageid = storageid + totalStorages + 1;
        }
        #ifdef RIO_DEBUG2
        else
            RioErr << "Router::SendStorage o novo indice " 
                   << storageid + totalStorages + 1 << " nao foi usado. "
                   << "Request->event->Request.Operation = " 
                   << Request->event->Request.Operation << "." << endl;
        #endif
        
        // Obtem o mapeamento a ser usado pelo storage.
        NATData result = m_NAT_map.getElement( input, storageid );
        if( result.nat_addr != 0 )
        {
            msg->IPaddr = result.nat_addr;
            msg->Port = result.nat_port;
        }
        msg->Disk     = Request->Disk;
        msg->Pos      = Request->Pos;
        msg->DataSize = Request->Size;

        /*not send ack*/
        msg->StreamTraffic = ((Event*)(Request->event))->Type;

        // Adiciona a mensagem a taxa de transferencia do Video.
        ((EventDataRequest*)(Request->event))->Request.rioobject->GetVideoRate( &msg->VideoRate );

        #ifdef RIO_DEBUG2
        RioErr << " read and send RouterId "<< Request->Id
               << " ClientId "<< msg->ClientId
               << " msg-Type RS|RW " << msg->Type << endl;
        #endif
    }

    m_DiskMgr->SendStorageNode( event, u16 ( Request->Disk ));
    return;
}

void CRouter::SetDiskServiceTime	(	int	Disk,
		double	EstServiceTimeOfDisk,
		double	EstServiceTime[301]
	)

Definition at line 1996 of file Router.cpp.

{
    for(int j = 0; j < 301; j++)
    {
        m_EstDiskServiceTime[ Disk ][ j ] = EstServiceTime[ j ];
    }

    m_EstDiskServiceTimeOfDisk[ Disk ] = EstServiceTimeOfDisk;
    pthread_mutex_lock(&m_MutexUpdated);
    m_UpdatedEstimatedDiskServiceTime++;
    pthread_cond_broadcast(&m_ConditionUpdated);
    pthread_mutex_unlock(&m_MutexUpdated);

    return;
}

void CRouter::SetNATMapping	(	NATData	server_map,
		int	stor_id,
		NATData	stor_map
	)

Definition at line 2111 of file Router.cpp.

{
    m_NAT_map.insertElement( server_map, stor_id, stor_map );
}

int CRouter::Start

(

void

)

Definition at line 338 of file Router.cpp.

{
    pthread_attr_t attrib;

    if( !m_initialized )
    {
        RioErr << "Router.Start: not initiallized" << endl;
        return ERROR_ROUTER + ERROR_NOT_INITIALIZED;
    }
    if( m_started )
    {
        RioErr << "Router.Start: already started" << endl;
        return ERROR_ROUTER + ERROR_STARTED;
    }

    pthread_attr_init( &attrib );
    pthread_attr_setstacksize( &attrib, 3*PTHREAD_STACK_MIN );

    if( pthread_create( &m_thread, &attrib, &routerthreadep, (void *)this ) )
    {
        RioErr << "Router.Start pthread_create failed "
               << strerror(errno) << endl;
        return (ERROR_ROUTER + ERROR_CREATE_THREAD);
    }

    m_started = true;
    return S_OK;
}

int CRouter::Stop

(

void

)

Definition at line 367 of file Router.cpp.

{
    Event *event;
    
    if( !m_started )
    {
        RioErr << "Router.Stop: not started" << endl;
        return( ERROR_ROUTER + ERROR_NOT_STARTED );
    }

    // Coloca o evento do termino da thread na fila de eventos.
    event = EventManager.New( EventTypeFinalizeThread );
    Put( event );                                                  

    // ### int term should have timer event which will wake router
    //     soon so don't need to do anything other than set flag...

    pthread_join( m_thread, NULL );

    m_started = false;
    return S_OK;
}

void CRouter::StorageDown	(	int	StorageId,
		bool	EmptyStorageQueue
	)			[private]

Funcao para definir que um servidor de armazenamento esta temporariamente indisponivel.

Parameters:

	ServerId	identificador do servidor de armazenamento.
	EmptyStorageQueue	true se a fila do servidor de armazenamento deve ser esvaziada e false em caso contrario. Obs: a fila nao deve ser esvaziada se a mensagem e enviada devido ao servidor nao estiver habilitado quando o servidor de gerenciamento iniciar a sua execucao.

Definition at line 2157 of file Router.cpp.

{
    StrDiskRequest *DiskRequest;
    RioStorageNodeInfo NodeInfo;
    RioStorageNodeInfo sn_info;
    char name[1025];
    int FirstDiskId;
    bool EmptyQueues;
    bool IsPending;
    // Obtem acesso exclusivo a variavel m_StoragesStatus. 
    pthread_mutex_lock( &m_StoragesStatusMutex );
    // Define o servidor StorageId como inativo.
    m_StoragesStatus = m_StoragesStatus | ( 1ull << StorageId );
    // Libera o acesso exclusivo a variavel m_StoragesStatus. 
    pthread_mutex_unlock( &m_StoragesStatusMutex );
    if( EmptyStorageQueue )
    {
        // Determina o numero de discos do servidor de armazenamento.
        m_DiskMgr->GetStorageNodeInfo( StorageId, &NodeInfo );
        // Calcula o identificador do primeiro disco do servidor de 
        // armazenamento. Ele sera igual ao Id do servidor multiplado pelo 
        // numero maximo de discos em cada servidor de armazanemento, dado por 
        // SNode::sn_maxdisks. Devemos somar mais 1 porque o identificador 0 e 
        // reservado.
        FirstDiskId = StorageId * SNode::sn_maxdisks + 1;
        // Remove as solicitacoes das filas de todos os discos do servidor de
        // armazenamento, e gera os codigos de erro esperados.
        for( int dk = FirstDiskId; dk < FirstDiskId + NodeInfo.NumberOfDisks; 
             dk++ )
        {
            EmptyQueues = true;
            // Remove as socilitacoes de todas as filas deste disco, gerando um
            // erro para o cliente
            while( EmptyQueues )
            {
                // Primeiramente tentamos remover todas as solicitacoes de disco 
                // da fila m_Pending, depois da fila m_RTqueue, e finalmente as 
                // da fila m_NRTqueue.
                if( m_nPending[ dk ] > 0 )
                {
                    DiskRequest = m_Pending[ dk ].Get();
                    
                    #ifdef RIO_DEBUG2
                    RioErr << "CRouter::StorageDown removendo a requisicao "
                           << DiskRequest << " de m_Pending. m_Pending[ " << dk 
                           << " ] = " << m_nPending[ dk ] << endl;
                    #endif       
                           
                    m_nPending[ dk ]--;
                    // Verifica qual fila deve ser decrementada (pelo que notei, 
                    // sempre que algo e colocado em m_Pending, ou m_nRT[ dk ]
                    // e incrementado, ou m_nNRT[ dk ] e incrementado.
                    DataRequest& Request = DiskRequest->event->Request;
                    if( ( Request.Operation == NonRealTimeWrite ) ||
                        ( Request.Operation == NonRealTimeRead ) )
                        m_nNRT[dk]--;
                    else    
                        m_nRT[dk]--;
                    IsPending = true;
                } 
                else if( m_nRT[ dk ] > 0 )
                {
                    DiskRequest = m_RTqueue[ dk ].Get();
                    
                    #ifdef RIO_DEBUG2
                    RioErr << "CRouter::StorageDown removendo a requisicao "
                           << DiskRequest << " de m_RTqueue. m_nRT[ " << dk 
                           << " ] = " << m_nRT[ dk ] << endl;
                    #endif       
                           
                    m_nRT[ dk ]--;
                    IsPending = false;
                }
                else if( m_nNRT[ dk ] > 0 )
                {
                    DiskRequest = m_NRTqueue[ dk ].Get();
                    
                    #ifdef RIO_DEBUG2
                    RioErr << "CRouter::StorageDown removendo a requisicao "
                           << DiskRequest << " de m_NRTqueue. m_nNRT[ " << dk 
                           << " ] = " << m_nNRT[ dk ] << endl;
                    #endif       
                           
                    m_nNRT[ dk ]--;
                    IsPending = false;
                }
                else 
                   EmptyQueues = false; 
                // Se ainda tivermos uma solicitacao de disco, retornamos o erro 
                // ao cliente (informando que o servidor de armazanamento do 
                // disco se tornou indisponivel).
                if( ( EmptyQueues ) && ( DiskRequest != NULL ) )
                {
                    // Obtem a solicitacao do cliente associada a solicitacao do 
                    // disco, a partir do evento armazenado em DiskRequest.
                    DataRequest& Request = DiskRequest->event->Request;
                    if( ( IsPending ) || 
                        ( ( Request.Operation != RealTimeRead ) &&
                          ( Request.Operation != RealTimePrefetchBlock ) ) )
                    {
                        // Cancela a requisicao (ou o pedido).
                        
                        #ifdef RIO_DEBUG2
                        RioErr << "CRouter::StorageDown removendo uma "
                               << "solicitacao  do disco " << dk << " do "
                               << "servidor de armazenamento " << StorageId 
                               << endl;
                        #endif       
                        // Define o codigo de erro para a requisicao. No caso da 
                        // operacao ser um cancelamento e BufferId for igual a 
                        // 1, devemos enviar uma resposta positiva (sem erros) 
                        // ao cliente. Em caso contrario, a resposta deve ser o 
                        // erro indicando que o servidor deixou de funcionar.
                        if( ( Request.Operation == RealTimeCancelBlock ) && 
                            ( Request.BufferId == -1 ) )
                            Request.Status = 0;
                        else
                            Request.Status = ERROR_ROUTER + 
                                        ERROR_SERVICE_TEMPORARY_UNAVAILABLE;
                        RequestCompleted( &Request.streamobj, ( Event * )
                                          DiskRequest->event );

                        // Isso e para evitar o envio de termino duplicado.
                        Request.RequestCancelled = true;        
                    } 
                    else
                    {
                        // Reenvia o evento para o Router.
                        #ifdef RIO_DEBUG2
                        RioErr << "CRouter::StorageDown Reenviando o evento "
                               << "com a requisicao para o Router" << endl;
                        #endif
                               
                        Put( ( Event *) DiskRequest->event );
                        
                    }
                    // Libera a requisicao do disco.
                    m_Request.Free( DiskRequest );
                } 
                else if( ( EmptyQueues ) && ( DiskRequest == NULL ) )
                    RioErr << "CRouter::StorageDown aviso DiskRequest == NULL"
                           << endl;
            }
        
            // Gera um evento para a classe de monitoramento, informando que as 
            // filas do disco ficaram vazias.
            m_DiskMgr->GetDiskName( dk, name );
            m_DiskMgr->GetDiskStorageNodeInfo( dk, &sn_info );
            UpdateDiskQueueEvent *it_event = new UpdateDiskQueueEvent( 
                                                 sn_info.Hostname, name, 0 );
            m_SystemManager->PostITEvent( ( MonitorEvent * ) it_event );
        }
    }
}

void CRouter::StorageReply

(

MsgRSSstorage *

Msg

)

[private]

Definition at line 1293 of file Router.cpp.

{
    StrDiskRequest*    Request;
    EventDataRequest*  event;
    DataRequest*       DataRequest;

    StrDiskRequest* NewRequest;
    int Disk;
    u16 Type;
    u32 Id;
    s32 Error;
    int ActiveThreads;
    // ------------------------------------------------------------------------

    switch( Message->Header.Type )
    {
        case MSG_RSS_READCOMPLETE:
        case MSG_RSS_WRITECOMPLETE:
        case MSG_RSS_RECEIVECOMPLETE:
        case MSG_RSS_SENDCOMPLETE:
        case MSG_RSS_READY:
        case MSG_RSS_CANCELCOMPLETE:
        {
            Type  = Message->Status.Type;
            Id    = Message->Status.RouterId;
            Error = Message->Status.Error;
            ActiveThreads = Message->Status.ActiveThreads;
            // ----------------------------------------------------------------

            Request = m_Request.Get( Id );
            if( Request == 0 )
            {
                RioErr << "Router Error: Message from storage node"
                        " has invalid Router Id " << Id << endl;
                return;
            }

            // Get data request event associated with disk request
            event = ( EventDataRequest* ) Request->event;
            DataRequest = & ( event->Request );

            // Status contains the first error detected
            // If more errors are detected they are ignored
            if( DataRequest->Status == 0 )
            {
                DataRequest->Status = Error;
            }

            #ifdef RIO_DEBUG2
            struct in_addr clientip;
            clientip.s_addr = DataRequest->Target.IPaddress;
            RioErr << "Router:StorageReply => RouterId " << Id
                   << " StorageId "<< Message->Status.StorageId
                   << " Status " << DataRequest->Status
                   << " client IP " << inet_ntoa(clientip)
                   << " client Port " << ntohs( DataRequest->Target.Port )
                   << " block " << DataRequest->Block
                   << " BufferId " << DataRequest->BufferId
                   << " Msg->Error " << Error
                   << " RequestCounter " << DataRequest->RequestCounter
                   << endl;
            #endif

            Disk = Request->Disk;
            
            // Verifica se o disco perterce a um servidor de armazenamento que
            // acabou de parar de funcionar. 
            if( !CheckStorageStatus( Disk ) )
            {
                RioErr << "Router Error: Message from storage node that is "
                          "down " << Id << endl;
                // Decrementa o contador de solicitacoes, removendo a 
                // solicitacao do servidor de armazenamento que caiu.
                DataRequest->RequestCounter--;       
                return;
                        
            }

            if( ( Type == MSG_RSS_READCOMPLETE ) && ( m_CollectMeasures ) )
            {
                struct timeval now;
                gettimeofday( &now, 0 );
                UpdateMeasures( Disk, ActiveThreads,
                                Request->event->Request.ArrivalTime, now );
            }
            // ----------------------------------------------------------------

            // Remove request from pending queue
            m_Pending[Disk].Remove(Request);

            // Check if there are queued requests not sent to storage node
            // and send one of them
            if( (( m_nRT[Disk] + m_nNRT[Disk] ) > m_nPending[Disk] ) &&
                ( m_nPending[Disk] < m_MaxPending ))
            {
                // Try to send real time request first
                NewRequest = m_RTqueue[Disk].Get();
                // If no RT request available, send a non RT one
                // There should be at least one (if condition above)
                if( NewRequest == 0 )
                {
                    RioErr <<" Could not get next request from RT queue"
                           << endl;
                    NewRequest = m_NRTqueue[Disk].Get();
                }
                m_Pending[Disk].Put( NewRequest );
                SendStorage( NewRequest );
            }
            // If no new disk request sent decrement number of pending
            // requests
            else
            {
                m_nPending[Disk]--;
            }

            // Decrement size of queue
            if( ( DataRequest->Operation == NonRealTimeRead ) ||
                ( DataRequest->Operation == NonRealTimeWrite)  )
            {
                m_nNRT[Disk]--;
            }
            else
            {
                if( DataRequest->Operation != RealTimeSendBlock )
                {
                      m_nRT[Disk]--;
                }
            }

            char name[1025];
            m_DiskMgr->GetDiskName( Disk, name );

            RioStorageNodeInfo sn_info;
            m_DiskMgr->GetDiskStorageNodeInfo( Disk, &sn_info );

            int queue_sum = m_nRT[ Disk ]
                          + m_nNRT[ Disk ]
                          + m_nPending[ Disk ];

            UpdateDiskQueueEvent *it_event = new UpdateDiskQueueEvent(
                sn_info.Hostname, name, queue_sum );
            m_SystemManager->PostITEvent( (MonitorEvent *)it_event );

            // Check if this is the last message storage node will send for
            // this disk request ( any message with an error code is the last)
            // ( success I/O is ended with
            // SENDCOMPLETE(read) or WRITECOMPLETE(write) )
            if( ( Type == MSG_RSS_SENDCOMPLETE )  ||
                ( Type == MSG_RSS_WRITECOMPLETE ) ||
                ( Type == MSG_RSS_CANCELCOMPLETE )||
                ( Error != 0) )
            {

                // if error != 0 and bufferid == -1  so prefetch failed
                if( ( DataRequest->BufferId == -1 ) &&
                    ( Type == MSG_RSS_READCOMPLETE ) )
                {
                    RioErr << "error at prefetch!! Error: "<< Error << endl;

                    DataRequest->streamobj->Stream()->getMutexUpdateEvent();
                    // updates StorageId and BufferId
                    Request->StorageId = Message->Status.StorageId;
                    DataRequest->BufferId = Id;
                    #ifdef RIO_DEBUG2
                    RioErr << "Router:StorageReply => new BufferId "  << Id
                           << " and StorageId " << Request->StorageId << endl;
                    #endif
                    // verifies if the block was canceled
                    if( DataRequest->Operation == RealTimeCancelBlock )
                    {
                        #ifdef RIO_DEBUG2
                        RioErr << "Block was canceled." << endl;
                        #endif
                        DataCancel( (Event *) event );
                        DataRequest->streamobj->Stream()->releaseMutexUpdateEvent();
                    }
                    else
                    {
                        // tells a new block arrived
                        DataRequest->streamobj->Stream()->NewBlockArrival();
                        // verifies if the client requested the block
                        if( ( DataRequest->Target.IPaddress != 0 ) &&
                           ( DataRequest->Target.Port != 0 ))
                        {
                            #ifdef RIO_DEBUG2
                            RioErr << "Block was requested by client (cancel it)."
                                   << endl;
                            #endif
                            DataRequest->Operation = RealTimeCancelBlock;
                            DataCancel( (Event *) event );

                            //update tokenstosend
                            DataRequest->streamobj->Stream()->decTokensToSend();
                            //update buffer status
                            DataRequest->streamobj->Stream()->decServerBufferStatus();
                        }
                        DataRequest->streamobj->Stream()->releaseMutexUpdateEvent();
                    }
                }
                // ------------------------------------------------------------
                else
                {
                    m_Request.Free(Request);
                    
                    #ifdef RIO_DEBUG2
                    RioErr << "CRouter::StorageReply requisicao " 
                           << DataRequest->RequestCounter << " recebida!" 
                           << endl;
                    #endif       

                    DataRequest->RequestCounter--;
                    // Check if we are done (all disk requests for this data
                    // request completed)
                    // Obs: RequestCompleted somente deve ser chamada se a 
                    // requisicao nao foi cancelada (porque um dos discos 
                    // associados a ela pertence a um servidor de armazenamento
                    // que ficou indisponivel) pois neste caso a funcao ja
                    // foi chamada.
                    if( ( DataRequest->RequestCounter == 0 ) &&
                        ( !DataRequest->RequestCancelled ) )
                    {
                        
                        #ifdef RIO_DEBUG2
                        RioErr << "CRouter::StorageReply operation " 
                               << DataRequest->Operation << " completed "
                               << "(DataRequest->RequestCounter == 0)" << endl;
                        #endif
                                
                        RequestCompleted( &DataRequest->streamobj,
                                          ( Event* ) event );
                    }
                }
            }
            // Not last message from storage node for this disk request
            else
            {
                if( DataRequest->BufferId == -1 )
                {
                    DataRequest->streamobj->Stream()->getMutexUpdateEvent();
                    // updates StorageId and BufferId
                    Request->StorageId = Message->Status.StorageId;
                    DataRequest->BufferId = Id;
                    #ifdef RIO_DEBUG2
                    RioErr << "Router:StorageReply => new BufferId "  << Id
                           << " and StorageId " << Request->StorageId << endl;
                    #endif

                    // verifies if the block was canceled
                    if( DataRequest->Operation == RealTimeCancelBlock )
                    {
                        #ifdef RIO_DEBUG2
                        RioErr << "Block was canceled." << endl;
                        #endif

                        DataCancel( (Event *) event );
                        DataRequest->streamobj->Stream()->releaseMutexUpdateEvent();
                    }
                    else
                    {
                        // tells a new block arrived
                        DataRequest->streamobj->Stream()->NewBlockArrival();

                        // verifies if the client requested the block
                        if( ( DataRequest->Target.IPaddress != 0 )&&
                           ( DataRequest->Target.Port != 0))
                        {
                            #ifdef RIO_DEBUG2
                            RioErr << "Block was requested by client." << endl;
                            #endif
                            DataRequest->Operation = RealTimeSendBlock;
                            DataSendToClient( (Event *) event );
                        }
                        DataRequest->streamobj->Stream()->releaseMutexUpdateEvent();
                    }
                }
                // ------------------------------------------------------------
            }
            break;
        }
        default:
            RioErr << "Router ERROR: Received invalid message from storage node."
                   << endl;
    }
}

void CRouter::StorageUp

(

int

StorageId

)

[private]

Funcao para definir que um servidor de armazenamento voltou a estar disponivel.

Parameters:

ServerId

identificador do servidor de armazenamento.

Definition at line 2314 of file Router.cpp.

{
    // Obtem acesso exclusivo a variavel m_StoragesStatus. 
    pthread_mutex_lock( &m_StoragesStatusMutex );
    // Invalida todos os pares IP, porta dos clientes conectados associados a
    // este servidor de armazenamento. Precisamos somar 1 ao identificador 
    // porque o identificador 0 e usado pelo mapeamento do servidor de 
    // gerenciamento.
    m_SystemManager->InvalidateStorageNatMappings( StorageId + 1 );
    // Define o servidor StorageId como ativo.
    m_StoragesStatus = m_StoragesStatus & ( ~( 1ull << StorageId ) );
    // Libera o acesso exclusivo a variavel m_StoragesStatus. 
    pthread_mutex_unlock( &m_StoragesStatusMutex );
}

void CRouter::UpdateDiskQueueTime	(	double	EstQueueTimeOfDisk[100],
		double	EstQueueTime[100][301]
	)

Definition at line 2072 of file Router.cpp.

{
    int num = GetMaxNumberOfDisks();

    for( int i = 1; i < num; i++ )
    {
        for(int j = 0; j < 301; j++)
        {
            EstimatedQueueTime[i][j] = m_EstDiskQueueTime[i][j];
        }
        EstimatedQueueTimeOfDisk[i] = m_EstDiskQueueTimeOfDisk[i];
    }

    return;
}

void CRouter::UpdateDiskResponseTime	(	double	EstResponseTimeOfDisk[100],
		double	EstResponseTime[100][301],
		double	DevDiskResponseTimeOfDisk[100],
		double	DevDiskResponseTime[100][301]
	)

Definition at line 2044 of file Router.cpp.

{
    int num = GetMaxNumberOfDisks();

    for( int i = 1; i < num; i++ )
    {
        for(int j = 0; j < 301; j++)
        {
            if( m_EstDiskServiceTime[i][j] > m_EstDiskResponseTime[i][j] )
                // server was initialized now
                m_EstDiskResponseTime[i][j] = m_EstDiskServiceTime[i][j];

            EstimatedResponseTime[i][j] = m_EstDiskResponseTime[i][j];
            DevResponseTime[i][j] = m_DevDiskResponseTime[i][j];
        }
        EstimatedResponseTimeOfDisk[i] = m_EstDiskResponseTimeOfDisk[i];
        DevResponseTimeOfDisk[i] = m_DevDiskResponseTimeOfDisk[i];
    }

    return;
}

void CRouter::UpdateDiskServiceTime	(	double	EstServiceTimeOfDisk[100],
		double	EstServiceTime[100][301]
	)

Definition at line 2016 of file Router.cpp.

{
    int num  = GetMaxNumberOfDisks();
    int numd = GetNumberOfActiveDisks();

    //RioErr << "Router:UpdateDiskServiceTime " << numd << " of " << num <<endl;

    while( m_UpdatedEstimatedDiskServiceTime < numd )
    {
        pthread_mutex_lock( &m_MutexUpdated );
        pthread_cond_wait( &m_ConditionUpdated, &m_MutexUpdated );
        pthread_mutex_unlock( &m_MutexUpdated );
    }

    for( int i = 1; i < num; i++ )
    {
        for(int j = 0; j < 301; j++)
        {
            EstimatedServiceTime[i][j] = m_EstDiskServiceTime[i][j];
        }
        EstimatedServiceTimeOfDisk[i] = m_EstDiskServiceTimeOfDisk[i];
    }

    m_UpdatedEstimatedDiskServiceTime = 0;
    return;
}

void CRouter::UpdateMeasures	(	int	disk,
		int	nThreads,
		struct timeval	initial_time,
		struct timeval	final_time
	)

Definition at line 1233 of file Router.cpp.

{
    double sample_msec = getInterval( initial_time, final_time );

    if( nThreads > m_MaxPending )
    {
        nThreads = m_MaxPending;
    }
    else if( nThreads < 1  )
    {
        RioErr << " ActiveThreads " << nThreads << endl;
        nThreads = 1;
    }

    if( m_EstDiskResponseTime[ disk ][ nThreads ] == 0 )
    {
        m_EstDiskResponseTime[ disk ][ nThreads ] = sample_msec;
    }
    else
    {
        m_EstDiskResponseTime[ disk ][ nThreads ] =
                                                  (( 1 - m_EstimatedParameter) *
                                   m_EstDiskResponseTime[ disk ][ nThreads ] ) +
                                           m_EstimatedParameter * sample_msec;
    }

    //updates Dev of estimated response time of disk
    m_DevDiskResponseTime[ disk ][ nThreads ] =  ( 1 - m_EstimatedParameter ) *
                                    m_DevDiskResponseTime[ disk ][ nThreads ] +
                                                         m_EstimatedParameter *
               fabs( sample_msec - m_EstDiskResponseTime[ disk ][ nThreads ] );

    if( m_EstDiskResponseTimeOfDisk[ disk ] == 0 )
    {
        m_EstDiskResponseTimeOfDisk[ disk ] = sample_msec;
    }
    else
    {
        m_EstDiskResponseTimeOfDisk[ disk ] =     ( 1 - m_EstimatedParameter) *
                                          m_EstDiskResponseTimeOfDisk[ disk ] +
                                          m_EstimatedParameter * sample_msec;
    }

    m_DevDiskResponseTimeOfDisk[ disk ] =        ( 1 - m_EstimatedParameter ) *
                                          m_DevDiskResponseTimeOfDisk[ disk ] +
                                                         m_EstimatedParameter *
                      fabs( sample_msec - m_EstDiskResponseTimeOfDisk[ disk ]);

    #ifdef __QUEUE_RESP_LOG
    m_logRESP << " disk "   << disk
              << " sample " << sample_msec
              << " est "    << m_EstDiskResponseTimeOfDisk[ disk ]
              << endl;
    #endif
}

---

Field Documentation

int CRouter::m_BlockSize [private]

Definition at line 100 of file Router.h.

bool CRouter::m_CollectMeasures [private]

Definition at line 78 of file Router.h.

pthread_cond_t CRouter::m_ConditionUpdated [private]

Definition at line 82 of file Router.h.

double CRouter::m_DevDiskResponseTime[100][301] [private]

Definition at line 68 of file Router.h.

double CRouter::m_DevDiskResponseTimeOfDisk[100] [private]

Definition at line 69 of file Router.h.

DiskMgr* CRouter::m_DiskMgr [private]

Definition at line 90 of file Router.h.

double CRouter::m_EstDiskQueueTime[100][301] [private]

Definition at line 74 of file Router.h.

double CRouter::m_EstDiskQueueTimeOfDisk[100] [private]

Definition at line 75 of file Router.h.

double CRouter::m_EstDiskResponseTime[100][301] [private]

Definition at line 66 of file Router.h.

double CRouter::m_EstDiskResponseTimeOfDisk[100] [private]

Definition at line 67 of file Router.h.

double CRouter::m_EstDiskServiceTime[100][301] [private]

Definition at line 61 of file Router.h.

double CRouter::m_EstDiskServiceTimeOfDisk[100] [private]

Definition at line 62 of file Router.h.

double CRouter::m_EstimatedParameter [private]

Definition at line 79 of file Router.h.

bool CRouter::m_initialized [private]

Definition at line 87 of file Router.h.

ofstream CRouter::m_logRESP [private]

Definition at line 84 of file Router.h.

ofstream CRouter::m_logSQT [private]

Definition at line 84 of file Router.h.

int CRouter::m_MaxPending [private]

Definition at line 103 of file Router.h.

int CRouter::m_MaxQueueSize [private]

Definition at line 108 of file Router.h.

pthread_mutex_t CRouter::m_MutexUpdated [private]

Definition at line 81 of file Router.h.

NATMap CRouter::m_NAT_map

Definition at line 194 of file Router.h.

int CRouter::m_nDisks [private]

Definition at line 98 of file Router.h.

int* CRouter::m_nNRT [private]

Definition at line 115 of file Router.h.

int* CRouter::m_nPending [private]

Definition at line 116 of file Router.h.

int* CRouter::m_nRT [private]

Definition at line 114 of file Router.h.

CDiskRequestQueue* CRouter::m_NRTqueue [private]

Definition at line 111 of file Router.h.

CDiskRequestQueue* CRouter::m_Pending [private]

Definition at line 112 of file Router.h.

EventQueue CRouter::m_Queue [private]

Definition at line 94 of file Router.h.

CDiskRequestManager CRouter::m_Request [private]

Definition at line 119 of file Router.h.

CDiskRequestQueue* CRouter::m_RTqueue [private]

Definition at line 110 of file Router.h.

bool CRouter::m_started [private]

Definition at line 88 of file Router.h.

unsigned long long int CRouter::m_StoragesStatus [private]

Definition at line 127 of file Router.h.

pthread_mutex_t CRouter::m_StoragesStatusMutex [private]

Definition at line 133 of file Router.h.

CSystemManager* CRouter::m_SystemManager [private]

Definition at line 91 of file Router.h.

StrDiskRequest** CRouter::m_TempDiskRequest [private]

Definition at line 122 of file Router.h.

pthread_t CRouter::m_thread [private]

Definition at line 86 of file Router.h.

int CRouter::m_UpdatedEstimatedDiskServiceTime [private]

Definition at line 77 of file Router.h.

---

The documentation for this class was generated from the following files:

• Router.h
• Router.cpp