SNode Class Reference

#include <DiskMgr.h>

Public Member Functions
void	SendStorageNode (EventStorageRequest *event, u16 DiskSeq)
void	GetDiskServiceTime (EventStorageRequest *event, u16 DiskSeq)
Static Public Attributes
static const int	sn_maxdisks = MAXSTORAGEDISKSARRAY
Private Member Functions
	SNode (DiskMgr *mgr, char *hostname)
	~SNode ()
int	txthread ()
int	rxthread ()
void	rxthread_nodeinfo (EventStorageReply *event)
void	rxthread_diskinfo (EventStorageReply *event)
void	rxthread_confirmation (EventStorageReply *event)
	Nova funcao para processar o recebimento da resposta a mensagem de inicializacao enviada na funcao DiskMgr::BuildNode.
void	rxthread_diskservicetimeinfo (EventStorageReply *event)
Static Private Member Functions
static void *	txthreadep (void *parm)
static void *	rxthreadep (void *parm)
Private Attributes
SNode *	sn_link
DiskMgr *	sn_mgr
char *	sn_hostname
vsiIPaddress	sn_IPaddress
CvsiTCPsocket	sn_socket
EventQueue	sn_sendqueue
pthread_t	sn_txthread
pthread_t	sn_rxthread
int	sn_termflag
int	sn_disks
int	sn_diskidorg
long long int	sn_disksize [sn_maxdisks]
bool	sn_restart
bool	sn_isenabled
bool	sn_sendevent
pthread_mutex_t	sn_mutex
pthread_cond_t	sn_waitstorage
bool	sn_readfreeblocksfile
Friends
class	DiskMgr

---

Detailed Description

Definition at line 222 of file DiskMgr.h.

---

Constructor & Destructor Documentation

SNode::SNode	(	DiskMgr *	mgr,
		char *	hostname
	)			[private]

Definition at line 1643 of file DiskMgr.cpp.

{
    int i;

    sn_mgr = mgr;
    pthread_mutex_lock( &mgr->m_mutex );
    i = strlen( hostname ) + 1;
    sn_hostname = new char[i];
    if( sn_hostname )
    {
        memcpy( sn_hostname, hostname, i );
    }

    sn_txthread = 0;
    sn_rxthread = 0;
    sn_termflag = false;

    sn_link = NULL;
    memset( (char *)&sn_IPaddress, 0, sizeof( vsiIPaddress ) );
    // sn_socket does not need to be initialized (done by it's constructorv)
    // sn_sendqueue does not need to be initialized (done by it's constructorv)
    sn_diskidorg = 0;

    SNode *nodelast = VIRTORG( mgr->m_nodelist, SNode, sn_link );

    while( nodelast->sn_link != 0 )
        nodelast = nodelast->sn_link;

    nodelast->sn_link = this;

    sn_disks = 0;
    for( i = 0; i < sn_maxdisks; i++ )
    {
        sn_disksize[i] = 0;
    }
    
    // Inicializa as novas variaveis criadas pela implementacao que permite a
    // recuperacao dos servidores de armazenamento.
    sn_restart = false;
    sn_isenabled = false;
    sn_sendevent = false;
    sn_readfreeblocksfile = false;

    pthread_mutex_unlock( &mgr->m_mutex );
    
    // Inicializa o mutex e a variavel de condicao, usados pela nova 
    // implementacao que permite a recuperacao dos storages.
    pthread_mutex_init( &sn_mutex, NULL );
    pthread_cond_init( &sn_waitstorage, NULL );
}

SNode::~SNode

(

)

[private]

Definition at line 1694 of file DiskMgr.cpp.

{
    Event *event;
    // Cancela as threads de leitura e de escrita do servidor de armazenameto
    // (storage) associado a este no.
    sn_termflag = true;
    // Fecha o socket para leitura e escrita (com isso, a thread rxthread de 
    // recepcao terminara caso esteja esperando na funcao receive, e a thread
    // txthread de transmissao nao enviara mais pacotes ao servidor de 
    // armazenamento).
    sn_socket.Close();
    // Coloca um evento especial EventTypeFinalizeThread na fila de eventos 
    // para informar a thread que ela deve terminar a sua execucao.
    if( sn_txthread != 0 )
    {
        event = EventManager.New( EventTypeFinalizeThread );
        sn_sendqueue.Put( event );
        // Desbloquea a thread txthread caso ela esteja esperando pelo reinicio do
        // servidor de armazenamento associado a esta classe.
        pthread_cond_signal( &sn_waitstorage );
        // Espera que a thread de envio termine.
        pthread_join( sn_txthread, NULL );
    }
    if( sn_rxthread != 0 ) // Espera que a thread de recebimento termine.
        pthread_join( sn_rxthread, NULL ); 
    
    // ### needs to check that there aren't any refs from a Disk to this SNode
    pthread_mutex_lock( &sn_mgr->m_mutex );
    SNode *node = VIRTORG( sn_mgr->m_nodelist, SNode, sn_link );

    while( node->sn_link != 0 )
    {
        if( node->sn_link == this )
        {
            node->sn_link = this->sn_link;
            break;
        }
        // added to fix bug --------------------------------------
        // it was not removing snode
        node = node->sn_link;
        // --------------------------------------------------------------------
    }

    pthread_mutex_unlock( &sn_mgr->m_mutex );

    delete [] sn_hostname;

    // Destroi o mutex e a variavel de condicao, usados pela nova implementacao 
    // que permite a recuperacao dos storages.
    pthread_mutex_destroy( &sn_mutex );
    pthread_cond_destroy( &sn_waitstorage );
}

---

Member Function Documentation

void SNode::GetDiskServiceTime	(	EventStorageRequest *	event,
		u16	DiskSeq
	)

Definition at line 2559 of file DiskMgr.cpp.

{
    if( event->Header.Type != EventTypeStorageRequest )
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "SNode::GetDiskServiceTime ERROR: "
                          << "Event has invalid type:"
                          << (int) (event->Header.Type) << endl;

        EventManager.Free((Event*)event);
        return;
    }
    u16 type = event->StorageRequest.Header.Type;
    if( type == MSG_RSS_DISKSERVICETIMEINFO_REQ )
    {
        event->StorageRequest.DiskServiceTimeInfo.Disk = DiskSeq;
    }
    else
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "SNode::GetDiskServiceTime ERROR: Storage request"
                          << " has invalid type: "
                          << event->StorageRequest.Header.Type << endl;

        EventManager.Free((Event*)event);
        return;
    }

    #ifdef RIO_DEBUG2
    if( sn_mgr->m_log.is_open() )
        sn_mgr->m_log << "SNode::GetDiskServiceTime getting service time for "
                      << "disk "
                      << event->StorageRequest.DiskServiceTimeInfo.Disk << " : "
                      << event->StorageRequest.DiskServiceTimeInfo.DiskId
                      << endl;
    #endif

    sn_sendqueue.Put((Event*) event);

    return;
}

int SNode::rxthread

(

)

[private]

Definition at line 1756 of file DiskMgr.cpp.

{
    int rc;
    EventStorageReply *event;
    //const int pfxsize = sizeof(event->StorageReply.Header.Type) +
    //                    sizeof(event->StorageReply.Header.Size);
    const unsigned long pfxsize = (unsigned long) &event->StorageReply.Header.Token -
                                  (unsigned long) &event->StorageReply.Header;
    bool isenabled;                    

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

    while( 1 )
    {
        if( sn_termflag )
        {
            RioErr << " DiskMgr rxthread finished. " << endl;
            pthread_exit( NULL );
        }
        
        // Obtem o valor de isenabled.
        // Obs: obter o mutex e necessario? ou podemos fazer como antes, 
        // retirando as 3 linhas de codigo a seguir e usando o if comentado?
        // Obtem o mutex para o acesso exclusivo a sn_isenabled.
        pthread_mutex_lock( &sn_mutex );
        // Obtem o valor atual de sn_isenabled.
        isenabled = sn_isenabled;
        // Libera o mutex para o acesso exclusivo a sn_isenabled.
        pthread_mutex_unlock( &sn_mutex );
        
        // Verifica se e necessario recriar a conexao com o servidor de 
        // armazenamento, ou se podemos enviar os dados.
        if( isenabled )
        {
            // A conexao esta habilitada, logo podemos receber novas mensagens.
            event = ( EventStorageReply* ) EventManager.New( 
                                                        EventTypeStorageReply );
            rc = sn_socket.Receive( ( char * ) &event->StorageReply.Header.Type, 
                                    pfxsize );

            if( sn_termflag )
            {
                EventManager.Free( ( Event* ) event );
                RioErr << " DiskMgr rxthread finished. " << endl;
                pthread_exit( NULL );
            }

            if( rc )
            {
                unsigned int storageid;

                EventManager.Free( ( Event* ) event );
                
                storageid = ( sn_diskidorg - 1 ) / SNode::sn_maxdisks;
                
                #ifdef RIO_DEBUG2
                RioErr << "SNode::rxthread receive retornou o erro " << errno 
                       << " (" << strerror( errno ) << ") para o storage na "
                       << "maquina " << sn_hostname << endl;
                #endif        

                if( sn_mgr->m_log.is_open() )
                    sn_mgr->m_log << "DiskMgr.rxthread receive1 failed." 
                                  << endl;

                RioErr << " DiskMgr rxthread error. See log file. Storage "     
                       << "server " << sn_hostname << " is down" << endl;
                       
                // Define o estado do servidor de armazenamento como 
                // temporariamente parado.
                // Obtem o mutex para o acesso exclusivo a sn_isenabled.
                pthread_mutex_lock( &sn_mutex );
                // Define o estado do servidor de armazenamento como parado.
                sn_isenabled = false;
                // Libera o mutex para o acesso exclusivo a sn_isenabled.
                pthread_mutex_unlock( &sn_mutex );
                // Gera um evento para informar ao router da queda do servidor 
                // de armazenamento
                EventStorageDown* event;
                event = ( EventStorageDown * ) EventManager.New( 
                                                         EventTypeStorageDown );

                event->StorageId = storageid;
                event->EmptyStorageQueue = true;

                #ifdef RIO_DEBUG2
                RioErr << "SNode::rxthread enviando a mensagem "
                       << "EventStorageDown ao Router do servidor de "
                       << "armazenamento com a ID " << event->StorageId << endl;
                #endif       

                sn_mgr->m_Router->Put( ( Event * ) event );
                
            } 
            else
            {
                if( event->StorageReply.Header.Size > 
                    sizeof( event->StorageReply ) )
                {
                    if( sn_mgr->m_log.is_open() )
                        sn_mgr->m_log << "DiskMgr.rxthread msg too large "
                                      << event->StorageReply.Header.Size
                                      << endl;

                    RioErr << " DiskMgr rxthread error. See log file. " << endl;
                } 
                else
                {
                    rc = sn_socket.Receive( (char *) &event->StorageReply.
                                                     Header.Token,
                                             event->StorageReply.Header.Size - 
                                             pfxsize );

                    if( sn_termflag )
                    {
                        EventManager.Free( ( Event* ) event );
                        RioErr << " DiskMgr rxthread finished. " << endl;
                        pthread_exit( NULL );
                    }
                                             
                    if( rc )
                    {
                        unsigned int storageid;

                        EventManager.Free( ( Event* ) event );

                        storageid = ( sn_diskidorg - 1 ) / SNode::sn_maxdisks;
                
                        if( sn_mgr->m_log.is_open() )
                            sn_mgr->m_log << "DiskMgr.rxthread receive2 failed." 
                                          << endl;
                                          
                        RioErr << " DiskMgr rxthread error. See log file. "     
                               << "Storage server " << sn_hostname 
                               << " is down" << endl;  

                        // Obtem o mutex para o acesso exclusivo a sn_isenabled.
                        pthread_mutex_lock( &sn_mutex );
                        // Define o estado do servidor de armazenamento como 
                        // parado.
                        sn_isenabled = false;
                        // Libera o mutex para o acesso exclusivo a 
                        // sn_isenabled.
                        pthread_mutex_unlock( &sn_mutex );

                        // Gera um evento para informar ao router da queda do 
                        // servidor de armazenamento
                        EventStorageDown* event;
                        event = ( EventStorageDown * ) EventManager.New( 
                                                         EventTypeStorageDown );

                        event->StorageId = storageid;
                        event->EmptyStorageQueue = true;                   

                        #ifdef RIO_DEBUG2 
                        RioErr << "SNode::rxthread enviando a mensagem "
                               << "EventStorageDown ao Router do servidor de "
                               << "armazenamento com a ID " << event->StorageId 
                               << endl;
                        #endif       

                        sn_mgr->m_Router->Put( ( Event * ) event ); 
                    }
                    else
                    {

                        // 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
                        //struct in_addr ip;
                        //ip.s_addr = sn_IPaddress.Host;
                        //RioErr << "RECEIVESTORAGE " << ( int ) time( NULL ) 
                        //       << " " << inet_ntoa( ip ) << " " 
                        //       << event->StorageReply.Header.Size << endl;
                        //#endif

                        if( sn_termflag )
                        {
                            EventManager.Free(( Event* )event );
                            RioErr << " DiskMgr rxthread finished. " << endl;
                            return 0;
                        }

                        switch( event->StorageReply.Header.Type )
                        {
                            case MSG_RSS_NODEINFO:
                                rxthread_nodeinfo( event );
                                break;

                            case MSG_RSS_DISKINFO:
                                rxthread_diskinfo( event );
                                break;

                            case MSG_RSS_DISKSERVICETIMEINFO:
                                rxthread_diskservicetimeinfo( event );
                                break;
                
                            // Processa a nova mensagem indicando que a 
                            // inicializacao do servidor de armazenamento foi 
                            // bem sucedida.
                            case MSG_RSS_CONFIRMATION:
                                rxthread_confirmation( event );
                                break; 
                               
                            #ifdef RIO_DEGUG2 
                            // Processa as mensagens associadas ao gerenciamento 
                            // dos logs. 
                            // Obs: atualmente estas mensagens sao somente 
                            // usadas para depurar o codigo. 
                            case MSG_RSS_SEARCHLOGS_STATUS:
                                 rxthread_searchlogsstatus( event );    
                                 break;
                            #endif    
                            // -------------------------------------------------

                           default: 
                               sn_mgr->m_Router->Put( ( Event* )event );
                               break;
                        }
                    }
                }
            }
        }
        else
        {
            // A conexao caiu ou ainda nao foi inicializada, logo precisamos 
            // restabelecer a conexao com o servidor de armazenamento.
            // Antes, porem, fechamos a conexao anterior com o servidor de 
            // armazenamento, se ela estiver aberta.
            sn_socket.Close();  
            // Tenta reiniciar a conexao com o servidor de armazenamento.
            rc =  sn_socket.Connect( &sn_IPaddress );
            if( rc )
            {
                // Se a conexao falhar, tentamos mais tarde
                RioErr << " connect failed for " << sn_hostname << ": " 
                       << GetErrorDescription( rc ) << ". Trying again!" 
                       << endl;
                // Dorme por algum tempo.
                sleep( sn_mgr->m_TimeBetweenAttempts );
            } 
            else
            {
                // Remove todos os eventos do servidor de armazenamento.
                Event *oldevent;
                do
                {
                    oldevent = ( Event * ) sn_sendqueue.Remove();
                    if( oldevent != NULL )
                        EventManager.Free( oldevent );
                } while( oldevent != NULL );
                
                RioErr << "SNode::rxthread conexao executada com sucesso para "
                       << "o servidor de armazenamento " << sn_hostname
                       << ". Enviando a mensagem MSG_RSS_INITIALIZATION_REQ "
                       << "para reiniciar o servidor de armazenamento "
                       << endl; 
                // Obtem o mutex para o acesso exclusivo a sn_isenabled.
                pthread_mutex_lock( &sn_mutex );
                // Define o estado do servidor de armazenamento como em 
                // funcionamento.
                sn_isenabled = true;
                // Define que precisamos enviar o evento EventTypeStorageUp
                // ao router
                sn_sendevent = true;
                // Envia um sinal para acordar a thread que envia dados ao
                // servidor de armazenamento.
                pthread_cond_signal( &sn_waitstorage );
                // Envia a mensagem de reinicio.
                EventStorageRequest *event;
                event = ( EventStorageRequest* ) EventManager.New( 
                                                      EventTypeStorageRequest );
                event->StorageRequest.Header.Type = MSG_RSS_INITIALIZATION_REQ;
                event->StorageRequest.Header.Size = SizeMsgRSSInitializationReq;
                event->StorageRequest.Header.Token = RSS_TOKEN_ROUTER;
                MsgRSSInitializationReq* msg = & ( event->StorageRequest.
                                                          Initialization );   
                msg->BlockSize = sn_mgr->m_BlockSize;
                sn_sendqueue.Put( ( Event* ) event );
                // Libera o mutex para o acesso exclusivo a sn_isenabled.
                pthread_mutex_unlock( &sn_mutex );

            }  
        }
    }
}

void SNode::rxthread_confirmation

(

EventStorageReply *

event

)

[private]

Nova funcao para processar o recebimento da resposta a mensagem de inicializacao enviada na funcao DiskMgr::BuildNode.

Parameters:

event

ponteiro para evento a ser processado, com a mensagem recebida.

Definition at line 2048 of file DiskMgr.cpp.

{
    EventStorageRequest *TxEvent;
    bool isenabled;

    if( ( event->StorageReply.NodeInfo.Size != SizeMsgRSSConfirmation )
        || ( event->StorageReply.NodeInfo.Token != RSS_TOKEN_STORAGE ))
    {
        RioErr << "event->StorageReply.NodeInfo.Size = " 
               << event->StorageReply.NodeInfo.Size << endl;
        RioErr << "SizeMsgRSSConfirmation = " << SizeMsgRSSConfirmation 
               << endl;
        RioErr << "event->StorageReply.NodeInfo.Token = " << hex 
               << event->StorageReply.NodeInfo.Token << dec << endl;
        RioErr << "RSS_TOKEN_STORAGE = " << hex << RSS_TOKEN_STORAGE << dec 
               << endl;
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "DiskMgr.rxthread invalid confirmation msg "
                          << "ignored for " << sn_hostname  << endl;

        EventManager.Free(( Event* ) event );
        return;
    }

    EventManager.Free(( Event* ) event );

    // Obs: obter o mutex e necessario? ou podemos fazer como antes, 
    // retirando as 3 linhas de codigo a seguir e usando o if comentado?
    // Obtem o mutex para o acesso exclusivo a sn_isenabled.
    pthread_mutex_lock( &sn_mutex );
    // Define o estado do servidor de armazenamento como parado.
    isenabled = sn_isenabled;
    // Libera o mutex para o acesso exclusivo a sn_isenabled.
    pthread_mutex_unlock( &sn_mutex );

    // Envia uma mensagem ao roteador informando que o servidor de armazenamento
    // inicializou com sucesso. A mensagem somente devera ser enviada nesta 
    // funcao se o servidor estiver reinicializando, isto e, o servidor 
    // estava ativo antes de inicializarmos o servidor de gerenciamento. Em caso
    // contrario (quando o servidor de armazenamento inicializar depois do 
    // servidor de gerenciamento), esta mensagem somente devera ser enviada apos
    // recebermos e inicializarmos todos os discos deste servidor. 
    if( ( sn_sendevent ) && ( sn_restart ) )
    {
        unsigned int storageid;
                 
        storageid = ( sn_diskidorg - 1 ) / SNode::sn_maxdisks;
                
        // O servidor de armazenamento reiniciou com sucesso. Precisamos entao
        // enviar o evento, ao Router, informando que este servidor esta 
        // novamente ativo.   
        EventStorageUp* event;
        // Depois de restabelecida a conexao com sucesso, enviamos um 
        // outro evento indicando que o servidor de armazenamento voltou 
        // a ficar disponivel.
        event = ( EventStorageUp * ) EventManager.New( EventTypeStorageUp );
        event->StorageId = storageid;

        #ifdef RIO_DEBUG2
        RioErr << "SNode::rxthread_confirmation enviando a mensagem "
               << "EventStorageUp ao Router do servidor de armazenamento com "
               << "a ID " << event->StorageId << endl;
        #endif       

        sn_mgr->m_Router->Put( ( Event * ) event );

        // Somente para evitar o envio de um outro evento.
        sn_sendevent = false;
    }
    
    if( !sn_restart )
    {
        // Se for a primeira vez que o servidor de armazenamento iniciou, envia 
        // a mensagem originalmente enviada pela funcao DiskMgr::BuildNode:
       
        // send node info request to this node to start things off...
        TxEvent = ( EventStorageRequest* ) EventManager.New( 
                                                      EventTypeStorageRequest );
        TxEvent->StorageRequest.Header.Type  = MSG_RSS_NODEINFO_REQ;
        TxEvent->StorageRequest.Header.Size  = SizeMsgRSSnodeInfoReq;
        TxEvent->StorageRequest.Header.Token = RSS_TOKEN_ROUTER;

        sn_sendqueue.Put( (Event*) TxEvent );
        
        // Se esta funcao for chamada novamente, entao estaremos tentado 
        // reiniciar a conexao com o servidor de armazenamento.
        sn_restart = true;
    }  
}

void SNode::rxthread_diskinfo

(

EventStorageReply *

event

)

[private]

Definition at line 2225 of file DiskMgr.cpp.

{
    int diskid;
    Disk *dp;
    bool sendevent;

    #ifdef RIO_DEBUG1
    RioErr << "### [SNode - rxthread_diskinfo] Start" << endl;
    #endif

    if( ( event->StorageReply.DiskInfo.Size != SizeMsgRSSdiskInfo )
        || ( event->StorageReply.DiskInfo.Token != RSS_TOKEN_STORAGE ))
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "DiskMgr.rxthread invalid diskinfo msg "
                          << "ignored for " << sn_hostname  << endl;

        EventManager.Free((Event*) event);
        return;
    }

    if( sn_mgr->m_log.is_open() )
        sn_mgr->m_log << " DiskMgr.rxthread_diskinfo: node " << sn_hostname
                      << " Disk "     << event->StorageReply.DiskInfo.Disk
                      << " DiskSize " << event->StorageReply.DiskInfo.DiskSize
                      << " DiskName " << event->StorageReply.DiskInfo.DiskName;
                  // ----------------------------------------------------------

    char diskname[256] = "", diskdevice[256] = "";
    strcpy( diskname, event->StorageReply.DiskInfo.DiskName );
    for( int aux = 0 ; diskname[aux] != '\0'; aux++ )
        if( diskname[aux] == '/' )
            diskname[aux] = '.';
    sprintf( diskdevice, "%s/%s.%s", sn_mgr->m_MetaRoot, sn_hostname, diskname);

    if( sn_mgr->m_log.is_open() )
        sn_mgr->m_log << " DiskBitsName " << diskdevice << endl;
    // ------------------------------------------------------------------------

    if( event->StorageReply.DiskInfo.Disk > sn_maxdisks )
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "DiskMgr.rxthread Disk > maxdisks "
                          << "ignored for " << sn_hostname << endl;
        EventManager.Free((Event*) event);
        return;
    }

    sn_disksize[event->StorageReply.DiskInfo.Disk] =
                                        event->StorageReply.DiskInfo.DiskSize;

    diskid = sn_diskidorg + event->StorageReply.DiskInfo.Disk;

    if( sn_mgr->m_disks[diskid] )
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "Diskmgr.rxthread duplicate disk info for disk "
                          << diskid << " from " << sn_hostname
                          << " ignored " << endl;

        EventManager.Free((Event*) event);
        return;
    }

    // ### should locking be here or in Disk ctor/dtor??
    pthread_mutex_lock( &sn_mgr->m_mutex );
    dp = new Disk( sn_mgr, this );
    dp->d_diskid = diskid;
    dp->d_diskseq = event->StorageReply.DiskInfo.Disk;
    dp->d_ntot = sn_disksize[dp->d_diskseq] / sn_mgr->m_BlockSize;
    // ### use log file instead of cerr (?) - need to pass *ostream?

    // the diskbits file name will be --------------------
    // storagenode.devicename instead of DiskBits1 and so on
    // It will avoid errors if the device order is modified at disk.cfg
    dp->d_BitMap.Initialize( diskdevice, dp->d_ntot, &cerr );
    
    //added to support df command ----------------------------------
    strcpy( dp->d_DiskName, event->StorageReply.DiskInfo.DiskName );
    // ------------------------------------------------------------------------

    // ### should delay allowing alloc until load or format has been called
    //     for this d_BitMap?
    dp->d_status = Disk::STAT_READ | Disk::STAT_WRITE | Disk::STAT_ALLOC;
    sn_mgr->m_disks[diskid] = dp;
    
    // Verifica se todos os discos ja foram inicializados. Existe um modo melhor
    // do que percorrer todas as entradas para este servidor de armazenamento?
    sendevent = true;
    for( int d = 0; d < sn_disks; d++ )
        sendevent = sendevent && ( sn_mgr->m_disks[ d + sn_diskidorg ] != 0 );     
    
    pthread_mutex_unlock( &sn_mgr->m_mutex );

    // Why this was not here before ????????
    EventManager.Free(( Event* ) event );

     // Send add disk message to Router
    EventAddDisk* eventdisk;
    eventdisk = ( EventAddDisk* ) EventManager.New( EventTypeAddDisk );
    eventdisk->Disk = diskid;
    sn_mgr->m_Router->Put(( Event* )eventdisk );

    // Envia uma mensagem ao roteador informando que o servidor de armazenamento
    // inicializou com sucesso. Isso somente sera feito se este servidor 
    // inicializar apos o servidor de gerenciamento (isso e indicado pela 
    // variavel booleana sn_sendevent, que indica a necessidade de enviarmos
    // um evento ao Router). A mensagem somente devera ser enviada apos todas
    // as mensagens de todos os discos forem recebidas com sucesso. Note que,
    // no caso de o servidor esta inicializando pela primeira vez, apos o 
    // servidor de gerenciamento, precisamos carregar todos os bitmaps dos
    // discos.
    if( ( sn_sendevent ) && ( sendevent ) ) 
    {
        unsigned int storageid;
                 
        storageid = ( sn_diskidorg - 1 ) / SNode::sn_maxdisks;

        // Carrega os bitmaps dos discos deste servidor de armazenamento. Note
        // que quando o servidor de armazenamento iniciar antes do servidor de
        // gerenciamento, isso sera feito na funcao Start da classe DiskMgr.                
        sn_mgr->LoadDisksBitMaps( storageid );
        
        // Libera todos os blocos associados a este servidor de armazenamento
        // que acabou de inicializar, caso existam blocos a serem liberados.
        if( sn_readfreeblocksfile )
        {
            sn_mgr->FreeSalvedBlocks( sn_hostname );
            sn_readfreeblocksfile = false;
        }
        
        // Contabiliza os discos que passaram a ficar ativos.
        sn_mgr->m_NumberOfActiveDisks = sn_mgr->m_NumberOfActiveDisks + 
                                        sn_disks;

        // O servidor de armazenamento reiniciou com sucesso. Precisamos entao
        // enviar o evento, ao Router, informando que este servidor esta 
        // novamente ativo.   
        EventStorageUp* event;
        // Depois de restabelecida a conexao com sucesso, enviamos um 
        // outro evento indicando que o servidor de armazenamento voltou 
        // a ficar disponivel.
        event = ( EventStorageUp * ) EventManager.New( EventTypeStorageUp );
        event->StorageId = storageid;

        #ifdef RIO_DEBUG2
        RioErr << "SNode::rxthread_confirmation enviando a mensagem "
               << "EventStorageUp ao Router do servidor de armazenamento com "
               << "a ID " << event->StorageId << endl;
        #endif       

        sn_mgr->m_Router->Put( ( Event * ) event );
        
        // Somente para evitar o envio de um outro evento.
        sn_sendevent = false;
    }

    #ifdef RIO_DEBUG1
    RioErr << "### [SNode - rxthread_diskinfo] End" << endl;
    #endif
}

void SNode::rxthread_diskservicetimeinfo

(

EventStorageReply *

event

)

[private]

Definition at line 2389 of file DiskMgr.cpp.

{
    if( ( event->StorageReply.DiskServiceTimeInfo.Size !=
         SizeMsgRSSdiskServiceTimeInfo ) ||
       ( event->StorageReply.DiskServiceTimeInfo.Token != RSS_TOKEN_STORAGE ))
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "DiskMgr.rxthread invalid diskServiceTimeinfo msg "
                          << "ignored for " << sn_hostname << endl;
        EventManager.Free((Event*) event);
        return;
    }

    if( event->StorageReply.DiskServiceTimeInfo.Disk > sn_maxdisks )
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "DiskMgr.rxthread Disk > maxdisks "
                          << "ignored for " << sn_hostname
                          << endl;

        EventManager.Free((Event*) event);
        return;
    }

    #ifdef RIO_DEBUG2
    if( sn_mgr->m_log.is_open() )
        sn_mgr->m_log << "DiskMgr.rxthread received servicetimeinfo of disk "
                      << event->StorageReply.DiskServiceTimeInfo.Disk << ":"
                      << event->StorageReply.DiskServiceTimeInfo.DiskId
                      << endl;
    #endif


    sn_mgr->m_Router->SetDiskServiceTime(
                event->StorageReply.DiskServiceTimeInfo.DiskId,
                event->StorageReply.DiskServiceTimeInfo.EstimatedDiskServTime,
                event->StorageReply.DiskServiceTimeInfo.EstimatedServTimeAccT);
    EventManager.Free((Event*)event);
}

void SNode::rxthread_nodeinfo

(

EventStorageReply *

event

)

[private]

Definition at line 2168 of file DiskMgr.cpp.

{
    EventStorageRequest *TxEvent;
    int i;

    if( ( event->StorageReply.NodeInfo.Size != SizeMsgRSSnodeInfo )
        || ( event->StorageReply.NodeInfo.Token != RSS_TOKEN_STORAGE ))
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "DiskMgr.rxthread invalid nodeinfo msg "
                          << "ignored for " << sn_hostname  << endl;

        EventManager.Free(( Event* ) event );
        return;
    }

    if( sn_disks != 0 )
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "DiskMgr.rxthread duplicate nodeinfo "
                          << "ignored for " << sn_hostname  << endl;

        EventManager.Free(( Event* ) event );
        return;
    }

    i = event->StorageReply.NodeInfo.nDisks;

    if( i > sn_maxdisks )
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "DiskMgr.rxthread nDisks " << i
                          << " > sn_maxdisks "  << sn_maxdisks
                          << " ignored for "    << sn_hostname << endl;

        EventManager.Free((Event*) event);
        return;
    }

    sn_disks = i;
    EventManager.Free(( Event* ) event );

    // send disk info request for each disk
    for( i = 0; i < sn_disks; i++ )
    {
        TxEvent = ( EventStorageRequest* )
                    EventManager.New(EventTypeStorageRequest);
        TxEvent->StorageRequest.DiskInfo.Type   = MSG_RSS_DISKINFO_REQ;
        TxEvent->StorageRequest.DiskInfo.Size   = SizeMsgRSSdiskInfoReq;
        TxEvent->StorageRequest.DiskInfo.Token  = RSS_TOKEN_ROUTER;
        TxEvent->StorageRequest.DiskInfo.Disk   = i;
        sn_sendqueue.Put(( Event* )TxEvent );
    }
}

void * SNode::rxthreadep

(

void *

parm

)

[static, private]

Definition at line 1747 of file DiskMgr.cpp.

{
    ((SNode *)parm)->rxthread();
    return 0;
}

void SNode::SendStorageNode	(	EventStorageRequest *	event,
		u16	DiskSeq
	)

Definition at line 2521 of file DiskMgr.cpp.

{
    if( event->Header.Type != EventTypeStorageRequest )
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "SNode::SendStorageNode ERROR: Event has "
                          << "invalid type:" << (int) (event->Header.Type)
                          << endl;

        EventManager.Free((Event*)event);
        return;
    }
    u16 type = event->StorageRequest.Header.Type & 0xf;

    if( ( type == MSG_RSS_READ  ) || ( type == MSG_RSS_WRITE   ) ||
       ( type == MSG_RSS_FETCH ) || ( type == MSG_RSS_RECEIVE ))
    {
        event->StorageRequest.Request.New.Disk = DiskSeq;
    }
    //bug: "if" changed - "or" by "and"
    else if( ( type != MSG_RSS_SEND   ) &&
             ( type != MSG_RSS_FLUSH  ) &&
             ( type != MSG_RSS_CANCEL ) )
    {
        if( sn_mgr->m_log.is_open() )
            sn_mgr->m_log << "SNode::SendStorageNode ERROR: Storage request"
                          << " has invalid type: "
                          << event->StorageRequest.Header.Type << endl;

        EventManager.Free((Event*)event);
        return;
    }
    sn_sendqueue.Put((Event*) event);

    return;
}

int SNode::txthread

(

)

[private]

Definition at line 2439 of file DiskMgr.cpp.

{
    EventStorageRequest *event = NULL;
    Event *auxevent;
    int rc;
    
    RioErr << "TXTHREADID " << syscall( SYS_gettid ) << endl;

    while( true )
    {
        //if( sn_termflag ) return 0;
        
        // Obtem o mutex para o acesso exclusivo a sn_isenabled.
        pthread_mutex_lock( &sn_mutex );
        // Verifica se precisamos esperar pela conexao.
        if( !sn_isenabled ) // Neste caso, esperamos ate que a conexao seja
                            // restabelecida.
            pthread_cond_wait( &sn_waitstorage, &sn_mutex );
        // Libera o mutex para o acesso exclusivo a sn_isenabled.
        pthread_mutex_unlock( &sn_mutex );
        
        // O Get abaixo retorna um Event
        //event = (EventStorageRequest*) sn_sendqueue.Get();
        auxevent = sn_sendqueue.Get();
        // Verifica se e um evento gerado para terminar a thread.
        if( auxevent->Type == EventTypeFinalizeThread )
        {
            EventManager.Free( (Event*) auxevent );
            RioErr << " DiskMgr txthread finished. " << endl;
            // Como recebemos um evento para terminar a thread, devemos entao
            // termina-la.
            //pthread_exit( NULL );
            return 0;
        } 

        event = ( EventStorageRequest* ) auxevent;
        #ifdef RIO_DEBUG2
        DataRequest& Request = (( EventDataRequest* )( event ))->Request;
        struct in_addr clientip;
        
        clientip.s_addr = Request.Target.IPaddress;
        RioErr << "Enviando ao storage agora!: "
               << "IP " << inet_ntoa(clientip)
               << " port "  << Request.Target.Port
               << " reqid " << Request.Reqid
               << " block " << Request.Block
               << endl;
        #endif

        /*if( sn_termflag )
        {
            EventManager.Free((Event*)event);
            return 0;
        }*/

        rc = sn_socket.Send( (char *) &(event->StorageRequest.Header.Type),
                             event->StorageRequest.Header.Size );
        if( rc )
        {
            if( sn_mgr->m_log.is_open() )
                sn_mgr->m_log << "SNode.txthread send error " << rc << endl;
             RioErr << " DiskMgr txthread error. See log file. " << 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
        //struct in_addr ip;
        //ip.s_addr = sn_IPaddress.Host;
        //RioErr << "SENDSTORAGE " << ( int ) time( NULL ) << " " 
        //       << inet_ntoa( ip ) << " " << event->StorageRequest.Header.Size 
        //       << endl;
        //#endif

        EventManager.Free((Event*)event);
    } // fim do while( true );
}

void * SNode::txthreadep

(

void *

parm

)

[static, private]

Definition at line 2431 of file DiskMgr.cpp.

{
    (( SNode * )parm )->txthread();
    return 0;
}

---

Friends And Related Function Documentation

friend class DiskMgr [friend]

Definition at line 294 of file DiskMgr.h.

---

Field Documentation

int SNode::sn_diskidorg [private]

Definition at line 269 of file DiskMgr.h.

int SNode::sn_disks [private]

Definition at line 268 of file DiskMgr.h.

long long int SNode::sn_disksize[sn_maxdisks] [private]

Definition at line 271 of file DiskMgr.h.

char* SNode::sn_hostname [private]

Definition at line 261 of file DiskMgr.h.

vsiIPaddress SNode::sn_IPaddress [private]

Definition at line 262 of file DiskMgr.h.

bool SNode::sn_isenabled [private]

Definition at line 279 of file DiskMgr.h.

SNode* SNode::sn_link [private]

Definition at line 259 of file DiskMgr.h.

const int SNode::sn_maxdisks = MAXSTORAGEDISKSARRAY [static]

Definition at line 225 of file DiskMgr.h.

DiskMgr* SNode::sn_mgr [private]

Definition at line 260 of file DiskMgr.h.

pthread_mutex_t SNode::sn_mutex [private]

Definition at line 284 of file DiskMgr.h.

bool SNode::sn_readfreeblocksfile [private]

Definition at line 292 of file DiskMgr.h.

bool SNode::sn_restart [private]

Definition at line 275 of file DiskMgr.h.

pthread_t SNode::sn_rxthread [private]

Definition at line 266 of file DiskMgr.h.

bool SNode::sn_sendevent [private]

Definition at line 282 of file DiskMgr.h.

EventQueue SNode::sn_sendqueue [private]

Definition at line 264 of file DiskMgr.h.

CvsiTCPsocket SNode::sn_socket [private]

Definition at line 263 of file DiskMgr.h.

int SNode::sn_termflag [private]

Definition at line 267 of file DiskMgr.h.

pthread_t SNode::sn_txthread [private]

Definition at line 265 of file DiskMgr.h.

pthread_cond_t SNode::sn_waitstorage [private]

Definition at line 287 of file DiskMgr.h.

---

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

• DiskMgr.h
• DiskMgr.cpp