CRouterInterface Class Reference

#include <routerinterface.h>

Public Member Functions
	CRouterInterface ()
	~CRouterInterface ()
int	Initialize (u16 LocalPort, char *LogsDirectory)
void	Stop ()
void	Send (StrMsg *Msg)
Private Member Functions
void	ProcessMsg (StrMsg *Elem)
void	ProcessNew (StrMsg *Elem)
void	ProcessPending (StrMsg *Elem)
void	SendNodeInfo (StrMsg *Msg)
void	SendDiskInfo (StrMsg *Msg, int Disk)
void	ProcessCancel (StrMsg *Elem)
void	SendDiskServiceTimeInfo (StrMsg *Elem, int Disk, int DiskId)
void	SendStatus (StrMsg *Msg, u16 Type, u32 RouterId, s32 Error)
void	ProcessInitialization (StrMsg *Elem, bool Restart)
	Processa a mensagem de inicializacao recebida pelo servidor de despacho.
void	SendConfirmation (StrMsg *Elem)
	Funcao para enviar, ao servidor de despacho, a confirmacao de que a inicializacao do servidor de armazenamento foi feita com sucesso.
void	ProcessStartSearchInLogs (StrMsg *Elem)
	Funcao para processar a mensagem que inicializa uma solicitacao de busca nos logs do servidor de armazenamento.
void	ProcessBlockSearchFile (StrMsg *Elem)
	Funcao para obter um dos blocos de um dos arquivos com o resultado de uma busca executada nos logs do servidor de armazenamento.
void	ProcessRemoveSearchFile (StrMsg *Elem)
	Funcao para remover um um dos arquivos com o resultado de uma busca executada nos logs do servidor de armazenamento.
Static Private Member Functions
static void *	WaitThread (void *Param)
static void *	TxThread (void *Param)
static void *	RxThread (void *Param)
Private Attributes
CvsiTCPwait	m_Connection
CvsiTCPsocket *	m_Socket
CvsiThread	m_WaitThread
CvsiThread	m_TxThread
CvsiThread	m_RxThread
CMsgQueue	m_TxQueue
int	m_Status
int	m_ConnectionActive
int	m_DisconnectFlag
CvsiCondition	m_ConnectCondition
CvsiCondition	m_DisconnectCondition
bool	m_InitializedRequestManager
	Nova variavel usada para verificar se o objeto RequestManager ja foi inicializado.
char *	m_LogsDirectory
	Nova variavel usada para armazenar o caminho em que os logs do servidor de armazenamento sao armazenados (usada ao inicializarmos o objeto de busca nos logs).

---

Detailed Description

Definition at line 36 of file routerinterface.h.

---

Constructor & Destructor Documentation

CRouterInterface::CRouterInterface

(

)

Definition at line 54 of file routerinterface.cpp.

{
    m_Status           = InterfaceInactive;
    m_Socket           = NULL;
    m_ConnectionActive = 0;
    m_DisconnectFlag   = 0;
    // Nova variavel usada para verificar se o RequestManager ja foi 
    // inicializado.
    m_InitializedRequestManager = false;
    // Inicializa a variavel que aponta para o diretorio com os logs.
    m_LogsDirectory = NULL;
}

CRouterInterface::~CRouterInterface

(

)

Definition at line 68 of file routerinterface.cpp.

{
    // Remove a variavel com o caminho dos logs.
    if( m_LogsDirectory != NULL )
        delete[] m_LogsDirectory;
    
}

---

Member Function Documentation

int CRouterInterface::Initialize	(	u16	LocalPort,
		char *	LogsDirectory
	)

Definition at line 80 of file routerinterface.cpp.

{
    if(m_Status == InterfaceActive)
    {
        return ERROR_SS_PROGRAM_ERROR;
    }

    vsiIPaddress MyAddress;

    MyAddress.Host = VSI_SOCKET_UNDEFINED_IP_ADDRESS;
    MyAddress.Port = LocalPort;

    int stat;

    stat = m_Connection.Initialize(&MyAddress,1);
    if(stat < 0)
    {
        RioErr << "CRouterInterface.Initialize - "
               << "Failed to initialize TCP socket:"
               << stat
               << " HOST: " << MyAddress.Host
               << " PORT: " << MyAddress.Port
               << endl;
        Stop();
        return ERROR_SS_TCP_INITIALIZATION;
    }

    m_Status = InterfaceActive;

    stat = m_WaitThread.Create(WaitThread,(void*) this);
    if(stat < 0)
    {
        RioErr << "Failed to create Wait Thread: "
               << stat << endl;
        return ERROR_SS_THREAD_CREATE;
    }
    
    // Inicializa a variavel que aponta para o diretorio com os logs.
    m_LogsDirectory = new char[ strlen( LogsDirectory ) + 1];
    if( m_LogsDirectory == NULL )
      return ERROR_SS_MEMORY;
      
    strcpy( m_LogsDirectory, LogsDirectory ); 

    return (0);
}

void CRouterInterface::ProcessBlockSearchFile

(

StrMsg *

Elem

)

[private]

Funcao para obter um dos blocos de um dos arquivos com o resultado de uma busca executada nos logs do servidor de armazenamento.

Parameters:

Elem

ponteiro para a estrutura onde esta armazenada a mensagem com os parametros usados pela inicializacao. Esta mensagem, do tipo MsgRSSBlockSearchFileReq, contem o identificador do arquivo, o numero do bloco, o par IP, porta UDP do cliente (para o qual devemos enviar o resultado da busca) e o identificador usado ao enviar o bloco ao cliente.

Definition at line 1156 of file routerinterface.cpp.

{
    int status;
    MsgRSSBlockSearchFileReq* Msg = &( Elem->Msg.Router.BlockSearchFile );
    #ifdef RIO_DEBUG2
    struct in_addr ip;
    ip.s_addr = Msg->IPaddr;
    RioErr << "CRouterInterface::ProcessBlockSearchFile recebida uma "
           << "requisicao para o pedido do bloco " << Msg->Block 
           << " do arquivo com a ID " << Msg->FileId << " do cliente com o IP " 
           << inet_ntoa( ip ) << " e a porta " << ntohs( Msg->Port ) 
           << ", usando a Id " << Msg->ClientId << endl;
    #endif
    status = SearchPackagesLogs.SearchResultDataRequest( Msg->IPaddr, Msg->Port,
                                                         Msg->FileId, 
                                                         Msg->Block, 
                                                         Msg->ClientId );

    #ifdef RIO_DEBUG2
    if( status != S_OK )
    {
        RioErr << "CRouterInterface::ProcessRemoveSearchFile erro " << status 
               << "(" << GetErrorDescription( status ) << ") ao remover o "
               << "arquivo com o resultado da busca." << endl;
        SendSearchLogsStatus( Elem, Msg->IPaddr, Msg->Port, Msg->ClientId, 
                              MSG_RSS_BLOCK_SEARCH_FILE_REQ, 
                              ERROR_SS_REMOVE_SEARCH_FILE );
    }                                               
    else
        SendSearchLogsStatus( Elem, Msg->IPaddr, Msg->Port, Msg->ClientId, 
                              MSG_RSS_BLOCK_SEARCH_FILE_REQ, S_OK );
    #endif       
}

void CRouterInterface::ProcessCancel

(

StrMsg *

Elem

)

[private]

Definition at line 884 of file routerinterface.cpp.

{
    MsgRSSpendRequest* Msg = & (Elem->Msg.Router.Request.Pending);

    // Compute right reply type in case an error is detected at this stage
    u16 Command = Msg->Type &0xf;
    u16 Reply;
    switch (Command)
    {
        case MSG_RSS_CANCEL:
            Reply = MSG_RSS_CANCELCOMPLETE;
            break;
        default:
            MsgManager.Free(Elem);
            RioErr << "UNEXPECTED ERROR: Processing cancel request. "
                   << "Request has unexpected type:" << Command << endl;
            return;
    }

    // Check if Request ids are valid
    // First, try to get request structure associated with request id
    StrRequest* Request = RequestManager.Get(Msg->StorageId);

    // If did not find request send error message (status) to router
    if( Request == 0 )
    {
        RioErr << "ProcessCancel:Request " << Msg->StorageId
               << " not found !!!"<<endl;
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_INVALID_REQUESTID );
        return;
    }

    // Double check if router request id match request
    if( Request->RouterId != Msg->RouterId)
    {
        RioErr << "ProcessCancel:Request " << Msg->RouterId
               << " != of " << Request->RouterId << "!!!"<<endl;
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_INVALID_REQUESTID );
        return;
    }

    // Finally, check if request status is the right one (expecting command
    // from router)
    if( ( Command == MSG_RSS_CANCEL ) &&
       ( Request->Status != RequestStatusWaitSendCommand ))
    {
        RioErr << "ProcessCancel:Request invalid !!!"<<endl;
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_INVALID_REQUESTID );
        return;
    }
    // free buffer
    #ifdef RIO_DEBUG2
    RioErr << "ProcessCancel:Request canceled ( " << Msg->StorageId << " )!!!"
           <<endl;
    #endif

    RequestManager.Free(Request);

    //send the cancel is completed
    SendStatus( Elem, Reply, Msg->RouterId, 0 );
}

void CRouterInterface::ProcessInitialization	(	StrMsg *	Elem,
		bool	Restart
	)			[private]

Processa a mensagem de inicializacao recebida pelo servidor de despacho.

Parameters:

	Elem	ponteiro para a estrutura onde esta armazenada a mensagem a ser processada.
	Restart	valor booleano que indica se esta e a primeira mensagem enviada pelo servidor de gerenciamento ou se e uma nova mensagem (indicando que o servidor de gerenciamento caiu e depois voltou).

Definition at line 1052 of file routerinterface.cpp.

{
    // Extrai o numero de blocos da mensagem
    MsgRSSInitializationReq* Msg = &( Elem->Msg.Router.Initialization );
    if( Restart )
    {
        // Neste caso, em que a mensagem nao e a primeira enviada pelo servidor
        // de geranciamento (o que provavelmente indica que o servidor caiu e
        // voltou), precisamos somente, para garantir, verificar se o tamanho
        // do bloco e o mesmo (para evitar que o disco seja incorretamente 
        // usado). Caso o tamanho do bloco nao seja o mesmo, o servidor de
        // armazenamento terminara a sua execucao.
        RioErr << "Received a duplicated initialization message. Block size = "
               <<  Msg->BlockSize << endl;
        if( Msg->BlockSize != Config.FragSize )
        {
            RioErr << "Fatal error: new block size " << Msg->BlockSize 
                   << " is different of actual block size " << Config.FragSize
                   << endl;
            exit( -1 ); // O que deveremos fazer neste caso?      
            
        }
    }
    else
    {
        // Esta e a primeira mensagem recebida. Devemos inicializar os objetos
        // RequestManager e SearchPackagesLogs.
        Config.FragSize = Msg->BlockSize;
        // Inicializa o gerenciador de requisicoes
        int stat = RequestManager.Initialize( Config.MaxReq, Config.FragSize );
        if( stat < 0 )
        {
            RioErr << "Error when initializing Request Manager: " << stat << endl;
            exit( 1 ); // O que deveremos fazer neste caso?
        }
            
        m_InitializedRequestManager = true;
              
        RioErr << "Request Manager started. Max number of requests: "
               << Config.MaxReq << " (array created). Block size: " 
               << Config.FragSize << endl;
           
        // Inicializa a classe de busca nos logs.
        stat = SearchPackagesLogs.Initialize( &SentPackagesLog, &NetManager,
                                              m_LogsDirectory, "Storage-", 
                                              Config.FragSize, &NetInterface );
        if( stat != S_OK )
        {
            RioErr << "Error when initializing SearchPackagesLogs: " << stat 
                   << endl;
            exit( 1 ); // O que deveremos fazer neste caso?
        }
    }                                             
    // Envia uma mensagem ao servidor informando que o tamanho do 
    // bloco foi recebido corretamente, para que o servidor possa
    // enviar as mensagens de informacao.
    SendConfirmation( Elem );
}

void CRouterInterface::ProcessMsg

(

StrMsg *

Elem

)

[private]

Definition at line 490 of file routerinterface.cpp.

{
    MsgRSSrouter* Msg = &(Elem->Msg.Router);

    // Ignore Flag bits in Request Type
    u16 Type = Msg->Header.Type & 0xff0f;

    #ifdef RIO_DEBUG2
    // u16 type2 = Msg->Header.Type & 0xf;
    // RioErr << "ProcessMsg (Msg->Header.Type "<< Msg->Header.Type
    //        << " " << Type << " " << type2 << ") ";
    #endif
    // ------------------------------------------------------------------------

    // Verifica se a classe RequestManager foi inicializada e, se nao foi,
    // somente aceita a mensagem , dando um aviso caso alguma outra mensagem
    // seja recebida.
    
    if( m_InitializedRequestManager )
    {
        // Neste caso, fazemos o processamento normal como antes, dando um
        // erro caso uma nova mensagem seja recebida.
        switch( Type )
        {
            case MSG_RSS_READ:
            case MSG_RSS_WRITE:
            case MSG_RSS_FETCH:
            case MSG_RSS_RECEIVE:
                ProcessNew ( Elem );
                break;
            case MSG_RSS_SEND:
            case MSG_RSS_FLUSH:
                #ifdef RIO_DEBUG2
                RioErr << "processpending" << endl;
                #endif
                // ----------------------------------------------------------------
                ProcessPending ( Elem );
                break;
            case MSG_RSS_CANCEL:
                // to free blocks that will not be used
                #ifdef RIO_DEBUG2
                RioErr << "processcancel" << endl;
                #endif
                // ----------------------------------------------------------------
                ProcessCancel ( Elem );
                break;
            // --------------------------------------------------------------------
            // Request for node info
            case MSG_RSS_NODEINFO_REQ:
                #ifdef RIO_DEBUG2
                RioErr << "sendnodeinfo" << endl;
                #endif
                // ----------------------------------------------------------------
                SendNodeInfo( Elem );
                break;
            // Request for disk info
            case MSG_RSS_DISKINFO_REQ:
                #ifdef RIO_DEBUG2
                RioErr << "sendiskinfo" << endl;
                #endif
                // ----------------------------------------------------------------
                SendDiskInfo(Elem, Msg->DiskInfo.Disk);
                break;
            case MSG_RSS_DISKSERVICETIMEINFO_REQ:
                // to send service time of disks
                #ifdef RIO_DEBUG2
                RioErr << "routerinterface - sendiskservicetimeinfo" << endl;
                #endif
                SendDiskServiceTimeInfo(Elem, Msg->DiskServiceTimeInfo.Disk,
                                        Msg->DiskServiceTimeInfo.DiskId );
                break;
            case MSG_RSS_INITIALIZATION_REQ:
                // Recebemos novamente a mensagem de inicializacao. Isso 
                // significa que o servidor de gerenciamento reiniciou a sua
                // execucao. Neste caso, deveremos chamar ProcessInitialization
                // com true, pois as inicializacoes ja foram feitas 
                // anteriormente.
                ProcessInitialization( Elem, true );
                break;   

            // Novas mensagens para processar logs no servidor de armazenamento
            // (no caso, somente os logs de trafego).
            // Mensagem para inicializar uma busca nos logs.
            case MSG_RSS_START_SEARCH_IN_LOGS_REQ:
                #ifdef RIO_DEBUG2
                RioErr << "startsearchinlogs" << endl;
                #endif
                // ----------------------------------------------------------------
                ProcessStartSearchInLogs( Elem );
                break;
            
            // Mensagem para enviar, ao cliente que fez a busca, um bloco de um
            // dos arquivos com os resultados de uma bas buscas.
            case MSG_RSS_BLOCK_SEARCH_FILE_REQ:
                #ifdef RIO_DEBUG2
                RioErr << "blocksearchfile" << endl;
                #endif
                // ----------------------------------------------------------------
                ProcessBlockSearchFile( Elem );
                break;
            
            // Mensagem para remover um dos arquivos com os resultados de uma das
            // buscas.
            case MSG_RSS_REMOVE_SEARCH_FILE_REQ:
                #ifdef RIO_DEBUG2
                RioErr << "removesearchfile" << endl;
                #endif
                // ----------------------------------------------------------------
                ProcessRemoveSearchFile( Elem );
                break;
             
            // --------------------------------------------------------------------
            // Invalid message (Ignore)
            default:
                MsgManager.Free (Elem);
                RioErr << "Received invalid message type from router: "
                       << Msg->Header.Type << " " << Type <<  endl;
        }
    }
    else
    {
        if( Type == MSG_RSS_INITIALIZATION_REQ ) {
            #ifdef RIO_DEBUG2
            RioErr << "routerinterface - initialization" << endl;
            #endif
            // Recebe e inicializa a variavel Config.FragSize com o tamanho do
            // bloco recebido do servidor. Como esta e a primeira mensagem, 
            // vamos chamar ProcessInitialization com false, para que as 
            // inicializacoes sejam feitas.
            ProcessInitialization( Elem, false );
        } 
        else
        {
            MsgManager.Free( Elem );
            RioErr << "Received a message type from router: "
                   << Msg->Header.Type << " " << Type << " after initialization"
                   << endl;
        }
    }
}

void CRouterInterface::ProcessNew

(

StrMsg *

Elem

)

[private]

Definition at line 633 of file routerinterface.cpp.

{
    // First check if message parameters are valid
    // Reply with Status error if any parameter is not valid
    MsgRSSnewRequest* Msg = &(Elem->Msg.Router.Request.New);

    // Compute right reply type in case an error is detected at this stage
    u16 Command = Msg->Type &0xf;
    u16 Reply;
    switch (Command)
    {
        case MSG_RSS_WRITE:
            Reply = MSG_RSS_WRITECOMPLETE;
            break;
        case MSG_RSS_RECEIVE:
            Reply = MSG_RSS_RECEIVECOMPLETE;
            break;
        case MSG_RSS_READ:
            Reply = MSG_RSS_SENDCOMPLETE;
            break;
        case MSG_RSS_FETCH:
            Reply = MSG_RSS_READCOMPLETE;
            break;
        default:
            MsgManager.Free(Elem);
            #ifdef RIO_DEBUG2
            RioErr << "UNEXPECTED ERROR: Processing new request. "
                   << "Request has unexpected type:" << Command << endl;
            #endif
            return;
    }

    // Check disk number
    unsigned int disk = Msg->Disk;
    if(disk >= Config.nDisks)
    {
        #ifdef RIO_DEBUG2
        RioErr << "ProcessNew:Request invalid disk!!!"<<endl;
        #endif
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_INVALID_DISK );
        return;
    }

    // Check Data position on disk
    if( ( Msg->Pos + Msg->DataSize) > Config.Disk[disk].Size )
    {
        #ifdef RIO_DEBUG2
        RioErr << "ProcessNew:Request invalid position!!!"<<endl;
        #endif
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_INVALID_POSITION );
        return;
    }

    // Check Data size
    if( ( Msg->DataSize > (Config.MaxReqSize * Config.FragSize )) ||
       ( Msg->DataSize == 0 ))
    {
        #ifdef RIO_DEBUG2
        RioErr << "ProcessNew:Request invalid data size!!!"<<endl;
        #endif
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_INVALID_DATASIZE );
        return;
    }

    StrRequest* Request = RequestManager.New();

    if( Request == 0 )
    {
        #ifdef RIO_DEBUG2
        RioErr << "ProcessNew:Request not available!!!"<<endl;
        #endif
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_MAX_REQUEST );
        return;
    }


    Request->Command    = Msg->Type;
    Request->Status     = RequestStatusAccepted;
    Request->ClientId   = Msg->ClientId;
    Request->RouterId   = Msg->RouterId;
    Request->IPaddress  = Msg->IPaddr;
    Request->Port       = Msg->Port;
    Request->Disk       = Msg->Disk;
    Request->Pos        = Msg->Pos;
    Request->Size       = Msg->DataSize;
    /*not send ack*/
    Request->StreamTraffic = Msg->StreamTraffic;
    // Inicializa o novo campo que define a taxa de transferencia de um video.
    Request->VideoRate = Msg->VideoRate;

    #ifdef RIO_DEBUG2
    struct in_addr client_IP;
    client_IP.s_addr = Request->IPaddress;
    struct sockaddr_in client_PORT;
    client_PORT.sin_port = Request->Port;

    RioErr     << " ------------------------------- "               << endl;
    RioErr     << " ClientId: "       << Request->ClientId          << endl;
    RioErr     << " IPaddress "       << inet_ntoa( client_IP )     << endl;
    RioErr     << " Port "            << htons(client_PORT.sin_port)<< endl;
    RioErr     << " Request->Disk "   << Request->Disk              << endl;
    #endif

    // Don't have to send any error message if reached this point.
    // Free Message Slot
    MsgManager.Free(Elem);

    switch (Command)
    {
        case MSG_RSS_READ:
        case MSG_RSS_FETCH:
            #ifdef RIO_DEBUG2
            RioErr << " RouterInterface: Device : ProcessRequest " << endl;
            #endif
            Device[Request->Disk].ProcessRequest( Request );
            break;
        case MSG_RSS_WRITE:
        case MSG_RSS_RECEIVE:
            #ifdef RIO_DEBUG2
            RioErr << " RouterInterface: sending to client " << endl;
            #endif
            Client.Receive(Request);
            break;
        default:
            RioErr << "UNEXPECTED ERROR: Router interface processing a "
                   << "new request and found invalid request type: "
                   << Command << endl;
    }
}

void CRouterInterface::ProcessPending

(

StrMsg *

Elem

)

[private]

Definition at line 777 of file routerinterface.cpp.

{
    MsgRSSpendRequest* Msg = & (Elem->Msg.Router.Request.Pending);

    // Compute right reply type in case an error is detected at this stage
    u16 Command = Msg->Type &0xf;
    u16 Reply;
    switch (Command)
    {
        case MSG_RSS_FLUSH:
            Reply = MSG_RSS_WRITECOMPLETE;
            break;
        case MSG_RSS_SEND:
            Reply = MSG_RSS_SENDCOMPLETE;
            break;
        default:
            MsgManager.Free(Elem);
            RioErr << "UNEXPECTED ERROR: Processing pending request. "
                   << "Request has unexpected type:" << Command << endl;
            return;
    }

    // Check if Request ids are valid

    // First, try to get request structure associated with request id
    StrRequest* Request = RequestManager.Get(Msg->StorageId);

    // If did not find request send error message (status) to router
    if( Request == 0 )
    {
        RioErr << "ProcessPending:Request " << Msg->StorageId
               << " not found !!!"<<endl;
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_INVALID_REQUESTID );
        return;
    }

    // Double check if router request id match request
    if( Request->RouterId != Msg->RouterId)
    {
        RioErr << "ProcessPending:Request " << Msg->RouterId
               << " != of " << Request->RouterId << "!!!"<<endl;
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_INVALID_REQUESTID );
        return;
    }

    // Finally, check if request status is the right one (expecting command
    // from router)
    if(   (( Command == MSG_RSS_FLUSH ) &&
           ( Request->Status != RequestStatusWaitFlushCommand ))
       || (( Command == MSG_RSS_SEND  ) &&
           ( Request->Status != RequestStatusWaitSendCommand )) )
    {
        RioErr << "ProcessPending:Request invalid !!!"<<endl;
        SendStatus( Elem,
                    Reply,
                    Msg->RouterId,
                    ERROR_RSS_INVALID_REQUESTID );
        return;
    }


    // Update request command to new operation
    Request->Command     =  Msg->Type;

    // Update ip, port, reqid
    // It is necessary because using buffers at the Router (Server side),
    // the Storage reads the block before the request is sent by the client
    // so it has to update the ip, port and reqid that will be used
    // to send the block to the client
    Request->ClientId   = Msg->ClientId;
    Request->IPaddress  = Msg->IPaddr;
    Request->Port       = Msg->Port;
    /*not send ack*/
    Request->StreamTraffic = Msg->StreamTraffic;
    // Inicializa o novo campo que define a taxa de transferencia de um video.
    Request->VideoRate = Msg->VideoRate;
    // ------------------------------------------------------------------------

    // Don't have to send any error message if reached this point.
    // Free Message Slot
    MsgManager.Free(Elem);

    // Send request to right place: client (send), or device (flush)
    switch (Command)
    {
        case MSG_RSS_FLUSH:
            Device[Request->Disk].ProcessRequest(Request);
            break;
        case MSG_RSS_SEND:
            RioErr << "YJYJYJYJYJYJYJYJY" <<endl;
            Client.Send(Request);
            break;
        default:
            RioErr << "UNEXPECTED ERROR: Router interface processing a "
                   << "pending request "
                   << "and found invalid request type "<< Command << endl;
    }

}

void CRouterInterface::ProcessRemoveSearchFile

(

StrMsg *

Elem

)

[private]

Funcao para remover um um dos arquivos com o resultado de uma busca executada nos logs do servidor de armazenamento.

Parameters:

Elem

ponteiro para a estrutura onde esta armazenada a mensagem com os parametros usados pela inicializacao. Esta mensagem, do tipo MsgRSSRemoveSearchFileReq, contem o identificador do arquivo e o par IP, porta UDP do cliente.

Definition at line 1190 of file routerinterface.cpp.

{
    int status;
    MsgRSSRemoveSearchFileReq* Msg = &( Elem->Msg.Router.RemoveSearchFile );
    #ifdef RIO_DEBUG2
    RioErr << "CRouterInterface::ProcessRemoveSearchFile recebida uma "
           << "requisicao para remover o arquivo com a ID " << Msg->FileId 
           << endl;
    #endif
    status = SearchPackagesLogs.RemoveSearchResultFile( Msg->FileId );
    #ifdef RIO_DEBUG2
    if( status != S_OK )
    {
        RioErr << "CRouterInterface::ProcessRemoveSearchFile erro " << status 
               << "(" << GetErrorDescription( status ) << ") ao remover o "
               << "arquivo com o resultado da busca." << endl;
        SendSearchLogsStatus( Elem, 0, 0, 0, MSG_RSS_REMOVE_SEARCH_FILE_REQ,
                              ERROR_SS_REMOVE_SEARCH_FILE );
    }                                               
    else
        SendSearchLogsStatus( Elem, 0, 0, 0, MSG_RSS_REMOVE_SEARCH_FILE_REQ, 
                              S_OK );
    #endif       
}

void CRouterInterface::ProcessStartSearchInLogs

(

StrMsg *

Elem

)

[private]

Funcao para processar a mensagem que inicializa uma solicitacao de busca nos logs do servidor de armazenamento.

Parameters:

Elem

ponteiro para a estrutura onde esta armazenada a mensagem com os parametros usados pela inicializacao. Esta mensagem, do tipo MsgRSSStartSearchInLogsReq, contem o tipo da busca, o tempo inicial, o tempo final, o par IP, porta UDP do cliente (para o qual devemos enviar o resultado da busca) e o identificador usado ao enviar a resposta da busca.

Definition at line 1123 of file routerinterface.cpp.

{
    int status;
    MsgRSSStartSearchInLogsReq* Msg = &( Elem->Msg.Router.StartSearchInLogs );
    #ifdef RIO_DEBUG2
    struct in_addr ip;
    ip.s_addr = Msg->IPaddr;
    RioErr << "CRouterInterface::ProcessStartSearchInLogs recebida uma "
           << "requisicao de busca do tipo " << Msg->SearchType 
           << ", do tempo " << Msg->StartTime << " ate o tempo " 
           << Msg->EndTime << " para o cliente com o IP " << inet_ntoa( ip )
           << " e a porta " << ntohs( Msg->Port ) << ", usando a Id " 
           << Msg->ClientId << endl;
    #endif
    status = SearchPackagesLogs.SearchLogsRequest( Msg->StartTime, Msg->EndTime,
                                                   Msg->IPaddr, Msg->Port,
                                                   Msg->ClientId );
    #ifdef RIO_DEBUG2
    if( status != S_OK )
    {
        RioErr << "CRouterInterface::ProcessStartSearchInLogs erro " << status 
               << "(" << GetErrorDescription( status ) << ") ao processar a "
               << " solicitacao de busca." << endl;
        SendSearchLogsStatus( Elem, Msg->IPaddr, Msg->Port, Msg->ClientId,
                              MSG_RSS_START_SEARCH_IN_LOGS_REQ,
                              ERROR_SS_SEARCH_IN_LOGS );
    }                                               
    else
        SendSearchLogsStatus( Elem, Msg->IPaddr, Msg->Port, Msg->ClientId,
                              MSG_RSS_START_SEARCH_IN_LOGS_REQ, S_OK );
    #endif       
}

void * CRouterInterface::RxThread

(

void *

Param

)

[static, private]

Definition at line 334 of file routerinterface.cpp.

{
    int res;

    StrMsg* Elem;
    MsgRSSrouter* Msg;

    // Get this class pointer
    CRouterInterface* Interface = (CRouterInterface*) Param;

    // Get members of this class
    CvsiTCPsocket* &m_Socket = Interface->m_Socket;
    int            &m_ConnectionActive    = Interface->m_ConnectionActive;
    int            &m_DisconnectFlag      = Interface->m_DisconnectFlag;
    CvsiCondition  &m_ConnectCondition    = Interface->m_ConnectCondition;
    CvsiCondition  &m_DisconnectCondition = Interface->m_DisconnectCondition;

    int& m_Status = Interface->m_Status;

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

    m_ConnectCondition.Lock();
    while( m_ConnectionActive == 0)
    {
        m_ConnectCondition.Wait();
    }
    m_DisconnectFlag = 0;
    m_ConnectCondition.Unlock();

    while(1)
    {
        // Get free message slot to store next message
        Elem = MsgManager.New();
        Msg  = &(Elem->Msg.Router);

        // Get only first 4 bytes of messages to check message size
        res = m_Socket->Receive ( (char*)Msg, 4 );

        // Exit thread if interface is not active anymore
        if(m_Status != InterfaceActive)
        {
            // Free message buffer
            MsgManager.Free(Elem);
            RioErr << "Exiting Rx thread ..." << endl;
            return (void*) 0;
        }

        // Wait for connection change if previous connection was shut down
        if( m_ConnectionActive == 0 )
        {
            // Free message buffer
            MsgManager.Free(Elem);

            // Signal end of previous connection
            m_DisconnectCondition.Lock();
            m_DisconnectFlag = 1;
            m_DisconnectCondition.Signal();
            m_DisconnectCondition.Unlock();

            m_ConnectCondition.Lock();
            while( m_ConnectionActive == 0)
            {
                m_ConnectCondition.Wait();
            }
            m_DisconnectFlag = 0;
            m_ConnectCondition.Unlock();
        }
        else
        {
            if( res < 0 )
            {
                // Free message buffer
                MsgManager.Free(Elem);
                RioErr << "Unexpected error on TCP receive(): " << res << endl;

                // Signal end of previous connection
                m_DisconnectCondition.Lock();
                m_ConnectionActive = 0;
                m_DisconnectFlag = 1;
                m_DisconnectCondition.Signal();
                m_DisconnectCondition.Unlock();
            }
            else
            {
                #ifdef RIO_DEBUG2
                // Imprime uma informacao sobre a mensagem
                RioErr << "CRouterInterface::RxThread informacao sobre a "
                       << "mensagem com  o tipo " << Msg->Header.Type 
                       << " e com o tamanho de " << Msg->Header.Size << endl;
                #endif        
                // Get message size
                u16 Size = Msg->Header.Size;

                // Discard messages with invalid sizes
                if( (Size > SizeMsgRSS) || (Size < 4) )
                {
                    int n,nr;
                    RioErr << "Received message with invalid size: Type="
                           << Msg->Header.Type << " Size:" << Msg->Header.Size
                           << endl;
                    nr = 4;
                    while( nr < Size )
                    {
                        n = Size - nr;
                        if( n > SizeMsgRSS )
                            n = SizeMsgRSS;
                        res = m_Socket->Receive( (char*)Msg,n );

                        // Check for error on socket receive
                        if(res < 0)
                        {
                            RioErr << "Unexpected Error on TCP "
                                   << "connection Receive()"
                                   << endl;
                            nr = Size;
                        }
                        else
                        {
                            nr += n;
                        }
                    }
                    MsgManager.Free(Elem);
                }
                // Process (possibly) valid message
                else
                {
                    // Get remaining bytes of message
                    if(Size > 4)
                    {
                        m_Socket->Receive (((char*)Msg)+4,Size-4);
                    }
                    
                    #ifdef RIO_DEBUG2
                    ///RioErr << "routerinterface - processando msg recebida do router " << endl;
                    #endif
                    
                    // Esta impressao foi comentada porque estes logs ainda nao 
                    // sao necessarios. Se for necessario usar estes logs, esta 
                    // linha do codigo pode ser descomentada (ja existe um 
                    //script para processar estes logs).
                    //#ifdef RIO_DEBUG_EMUL
                    //RioErr << "RECEIVESERVER " << ( int ) time( NULL ) << " " 
                    //       << Size << endl;
                    //#endif
                    
                    Interface->ProcessMsg(Elem);
                }
            }
        }
    }
}

void CRouterInterface::Send

(

StrMsg *

Msg

)

Definition at line 156 of file routerinterface.cpp.

{
    m_TxQueue.Put(Elem);
}

void CRouterInterface::SendConfirmation

(

StrMsg *

Elem

)

[private]

Funcao para enviar, ao servidor de despacho, a confirmacao de que a inicializacao do servidor de armazenamento foi feita com sucesso.

Parameters:

Elem

ponteiro para a estrutura onde esta armazenada a mensagem a ser enviada.

Definition at line 1113 of file routerinterface.cpp.

{
    MsgRSSConfirmation* Msg = &( Elem->Msg.Storage.Confirmation );

    Msg->Type       = MSG_RSS_CONFIRMATION;
    Msg->Size       = SizeMsgRSSConfirmation;

    Send( Elem );
}

void CRouterInterface::SendDiskInfo	(	StrMsg *	Msg,
		int	Disk
	)			[private]

Definition at line 990 of file routerinterface.cpp.

{
    MsgRSSdiskInfo* Msg = &(Elem->Msg.Storage.DiskInfo);
    if( Disk >= ((int)Config.nDisks ))
    {
        RioErr << "Unexpected ERROR on CRouterInterface::SendDiskInfo : "
               << "Invalid Disk: " << Disk << endl;
        MsgManager.Free (Elem);
    }

    Msg->Type       = MSG_RSS_DISKINFO;
    Msg->Size       = SizeMsgRSSdiskInfo;
    Msg->Disk       = Disk;
    Msg->DiskSize   = Config.Disk[Disk].Size;
    Msg->SectorSize = Config.Disk[Disk].SectorSize;
    //support df command ----------------------------------
    strcpy( Msg->DiskName, Config.Disk[Disk].Name );
    // ------------------------------------------------------------------------
    Send (Elem);
}

void CRouterInterface::SendDiskServiceTimeInfo	(	StrMsg *	Elem,
		int	Disk,
		int	DiskId
	)			[private]

Definition at line 1013 of file routerinterface.cpp.

{
    MsgRSSdiskServiceTimeInfo* Msg = &(Elem->Msg.Storage.DiskServiceTimeInfo);
    double etaccthreads[301];
    double estimatedtime;

    if( Disk >= ((int)Config.nDisks ))
    {
        RioErr << "Unexpected ERROR on CRouterInterface::"
               << "SendDiskServiceTimeInfo."
               << "Invalid Disk: " << Disk << endl;
        MsgManager.Free (Elem);
        return;
    }

    memset(etaccthreads,0,sizeof(etaccthreads));
    Msg->Type           = MSG_RSS_DISKSERVICETIMEINFO;
    Msg->Size           = SizeMsgRSSdiskServiceTimeInfo;
    Msg->Disk           = Disk;
    Msg->DiskId         = DiskId;
    Device[ Disk ].GetDiskServiceTime( etaccthreads );
    estimatedtime = Device[ Disk ].GetEstimatedTime();
    Msg->EstimatedDiskServTime  = estimatedtime;
    //Start - GCC 3
    //Msg->EstimatedServTimeAccT  = etaccthreads;
    memcpy( Msg->EstimatedServTimeAccT,etaccthreads,sizeof(etaccthreads));
    //End - GCC 3

    #ifdef RIO_DEBUG2
    RioErr << "Sending disk service time info of " << Disk
           << " DiskId " << DiskId  << endl;
    RioErr << "EstimatedTime " << Msg->EstimatedDiskServTime << " msec"<< endl;
    #endif
    Send (Elem);
}

void CRouterInterface::SendNodeInfo

(

StrMsg *

Msg

)

[private]

Definition at line 973 of file routerinterface.cpp.

{
    MsgRSSnodeInfo* Msg = &(Elem->Msg.Storage.NodeInfo);

    Msg->Type       = MSG_RSS_NODEINFO;
    Msg->Size       = SizeMsgRSSnodeInfo;
    Msg->nDisks     = Config.nDisks;
    Msg->FragSize   = Config.FragSize;
    Msg->BufferSize = Config.BufferSize;
    Msg->MaxReqSize = Config.MaxReqSize;
    Msg->MaxReq     = Config.MaxReq;

    Send (Elem);
}

void CRouterInterface::SendStatus	(	StrMsg *	Msg,
		u16	Type,
		u32	RouterId,
		s32	Error
	)			[private]

Definition at line 957 of file routerinterface.cpp.

{
    MsgRSSstatus* Msg   = &(Elem->Msg.Storage.Status);
    Msg->Type           = Type;
    Msg->Size           = SizeMsgRSSstatus;
    Msg->RouterId       = RouterId;
    Msg->StorageId      = 0;
    Msg->Error          = Error;
    Send(Elem);
}

void CRouterInterface::Stop

(

void

)

Definition at line 129 of file routerinterface.cpp.

{
    // change status to cause threads to exit
    m_Status = InterfaceInactive;

    // close sockets to cause Wait and Rx Threads to return from socket call
    m_Connection.Close();

    m_DisconnectCondition.Lock();
    if(m_ConnectionActive)
    {
        m_ConnectionActive = 0;
        m_Socket->Close();
    }
    m_DisconnectCondition.Unlock();

    // ### need to Send fake message to force tx thread to
    //     return from waiting on message queue;

    m_TxThread.Join(0);
    m_RxThread.Join(0);
}

void * CRouterInterface::TxThread

(

void *

Param

)

[static, private]

Definition at line 270 of file routerinterface.cpp.

{
    int stat;

    // Pointer to current message to be sent
    StrMsg* Msg;

    // Get this class pointer
    CRouterInterface* Interface = (CRouterInterface*) Param;

    // Get members of this class
    CMsgQueue &m_TxQueue = Interface->m_TxQueue;
    int &m_Status = Interface->m_Status;
    CvsiTCPsocket* &m_Socket = Interface->m_Socket;

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

    // Loop forever
    while(1)
    {
        // Wait and get next message
        Msg = m_TxQueue.Get();

        // Exit thread if interface is not active anymore
        // Stop() called
        if(m_Status != InterfaceActive)
        {
            RioErr << "TxThread finished." << endl;
            return ( (void*) 0 );
        }
        
        #ifdef RIO_DEBUG2
        // Imprime uma informacao sobre a mensagem
        RioErr << "CRouterInterface::TxThread informacao sobre a mensagem com "
               << " o tipo " << Msg->Msg.Header.Type << " e com o tamanho de "
               << Msg->Msg.Header.Size << endl;
        #endif        

        // Set Token
        Msg->Msg.Header.Token = RSS_TOKEN_STORAGE;

        // Send message to router
        stat = m_Socket->Send((char*)& (Msg->Msg), Msg->Msg.Header.Size);
        if(stat < 0)
        {
            RioErr << "Unexpected Error on TCP socket Send()" << endl;
        }

        // Esta impressao foi comentada porque estes logs ainda nao sao 
        // necessarios. Se for necessario usar estes logs, esta linha do codigo 
        // pode ser descomentada (ja existe um script para processar estes 
        // logs).
        //#ifdef RIO_DEBUG_EMUL
        //RioErr << "SENDSERVER " << ( int ) time( NULL ) << " "
        //       << Msg->Msg.Header.Size << endl;
        //#endif

        MsgManager.Free (Msg);
    }

}

void * CRouterInterface::WaitThread

(

void *

Param

)

[static, private]

Definition at line 163 of file routerinterface.cpp.

{
    // Get this class pointer
    CRouterInterface* Interface   = (CRouterInterface*) Param;
    CvsiThread        &m_TxThread = Interface->m_TxThread;
    CvsiThread        &m_RxThread = Interface->m_RxThread;
    int               &m_Status   = Interface->m_Status;
    int               &m_ConnectionActive = Interface->m_ConnectionActive;
    int               &m_DisconnectFlag   = Interface->m_DisconnectFlag;
    CvsiTCPwait       &m_Connection       = Interface->m_Connection;
    CvsiTCPsocket*    &m_Socket   = Interface->m_Socket;
    CvsiCondition     &m_ConnectCondition = Interface->m_ConnectCondition;
    CvsiCondition     &m_DisconnectCondition = Interface->m_DisconnectCondition;

    CvsiTCPsocket*    sock;

    int stat;
    int errorcount = 0;

    m_ConnectionActive = 0;
    m_DisconnectFlag   = 1;

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

    stat = m_TxThread.Create(TxThread,Param);
    if(stat < 0)
    {
        RioErr << "ERROR on Wait thread: Failed to create Tx thread: "
               << stat << endl;
        RioErr << "Aborting thread ..." << endl;
        Interface->Stop();
        return (void*) 0;
    }

    stat = m_RxThread.Create(RxThread,(void*) Param);
    if(stat < 0)
    {
        RioErr << "ERROR on Wait thread: Failed to create Rx thread: " << stat
               << endl;
        RioErr << "Aborting thread ..." << endl;
        Interface->Stop();
        return (void*) 0;
    }

    while(1)
    {
        sock = new CvsiTCPsocket;

        // Wait for router connection
        stat = m_Connection.Accept(sock,NULL);

        RioErr << "Connection.Accept(): " << stat << endl;

        // Exit thread if interface is not active anymore
        if(m_Status != InterfaceActive)
        {
            Interface->Stop();
            RioErr << "Exiting Wait thread ..." << endl;
            return (void*) 0;
        }

        // check for errors
        if( stat < 0 )
        {
            errorcount++;
            RioErr << "Unexpected Error on TCP connection Accept()" << endl;
            if( errorcount > 4 )
            {
                RioErr << "Too many TCP connection errors: Aborting wait Thread"
                       << endl;
                Interface->Stop();
                return (void *)0;
            }
        }
        else
        {
            // reset error count
            // (this time connection was accepted successfully)
            errorcount = 0;

            // Wait for previous connection to shut down
            m_DisconnectCondition.Lock();

            if( m_ConnectionActive )
            {
                m_ConnectionActive = 0;
                m_Socket->Close();
            }
            while(m_DisconnectFlag == 0)
            {
                m_DisconnectCondition.Wait();
            }
            m_Socket = sock;
            m_DisconnectCondition.Unlock();

            // Signal Rx thread that connection is active
            m_ConnectCondition.Lock();
            m_ConnectionActive = 1;
            m_ConnectCondition.Signal();
            m_ConnectCondition.Unlock();
        }
    }
    return 0;
}

---

Field Documentation

CvsiCondition CRouterInterface::m_ConnectCondition [private]

Definition at line 142 of file routerinterface.h.

CvsiTCPwait CRouterInterface::m_Connection [private]

Definition at line 133 of file routerinterface.h.

int CRouterInterface::m_ConnectionActive [private]

Definition at line 140 of file routerinterface.h.

CvsiCondition CRouterInterface::m_DisconnectCondition [private]

Definition at line 143 of file routerinterface.h.

int CRouterInterface::m_DisconnectFlag [private]

Definition at line 141 of file routerinterface.h.

bool CRouterInterface::m_InitializedRequestManager [private]

Nova variavel usada para verificar se o objeto RequestManager ja foi inicializado.

Definition at line 146 of file routerinterface.h.

char* CRouterInterface::m_LogsDirectory [private]

Nova variavel usada para armazenar o caminho em que os logs do servidor de armazenamento sao armazenados (usada ao inicializarmos o objeto de busca nos logs).

Definition at line 150 of file routerinterface.h.

CvsiThread CRouterInterface::m_RxThread [private]

Definition at line 137 of file routerinterface.h.

CvsiTCPsocket* CRouterInterface::m_Socket [private]

Definition at line 134 of file routerinterface.h.

int CRouterInterface::m_Status [private]

Definition at line 139 of file routerinterface.h.

CMsgQueue CRouterInterface::m_TxQueue [private]

Definition at line 138 of file routerinterface.h.

CvsiThread CRouterInterface::m_TxThread [private]

Definition at line 136 of file routerinterface.h.

CvsiThread CRouterInterface::m_WaitThread [private]

Definition at line 135 of file routerinterface.h.

---

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

• routerinterface.h
• routerinterface.cpp