StreamControl.cpp

Go to the documentation of this file.

/*
 * Copyright (C) 2009, Edmundo Albuquerque de Souza e Silva.
 *
 * This file may be distributed under the terms of the Q Public License
 * as defined by Trolltech AS of Norway and appearing in the file
 * LICENSE.QPL included in the packaging of this file.
 *
 * THIS FILE IS PROVIDED AS IS WITH NO WARRANTY OF ANY KIND, INCLUDING
 * THE WARRANTY OF DESIGN, MERCHANTABILITY AND FITNESS FOR A PARTICULAR
 * PURPOSE.  IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY SPECIAL,
 * INDIRECT OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING
 * FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF CONTRACT,
 * NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION
 * WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.
 *
 * Thanks: Jose Renato Santos
 *
 */

#include <unistd.h>

// Usado para depuracao

#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

// Unidade para podermos usar a syscall para imprimir a id de uma thread.
#include <sys/syscall.h>

#include "StreamControl.h"

///////////////////////////////////////////////////////////////////////////////
// Classe CFragmentQueue - Implementa a fila FIFO com os objetos com as      //
// informacoes dos fragmentos a serem enviados pela rede.                    //
/////////////////////////////////////////////////////////////////////////////// 

// Construtor da classe.
CFragmentQueue::CFragmentQueue()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - Construtor] Start" << endl;
    #endif
    
    // Inicializa as variaveis da classe (para uma lista vazia).
    m_Start = NULL;
    m_End = NULL;

    #ifdef RIO_DEBUG2
    // Inicializa o contador usado pela depuracao, que conta o numero de 
    // fragmentos na fila.
    FragmentQueueSize = 0;
    #endif

    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - Construtor] Finish" << endl;
    #endif
} 


// Destrutor da classe.
CFragmentQueue::~CFragmentQueue()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - Destrutor] Start" << endl;
    #endif

    // Remove a fila FIFO de objetos FragmentInfo da classe.
    DeleteFragmentQueue();

    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - Destrutor] Finish" << endl;
    #endif
}

// Funcao para remover a fila FIFO com os objetos FragmentInfo (usada pelo 
// destrutor da classe).
void CFragmentQueue::DeleteFragmentQueue()
{
    FragmentInfoList *Object, *NextObject;

    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - DeleteFragmentQueue] Start" << endl;
    #endif

    // Remove cada um dos objetos da fila FIFO da classe.
    Object = m_Start;
    while( Object != NULL )
    {
        // Salva o ponteiro para o proximo objeto FragmentInfo.
        NextObject = Object->NextObject;
        // Remove o buffer com a copia dos dados do fragmento.
        delete Object->fip->FragmentData;
        // Revome a estrutura com as informacoes do fragmento.
        delete Object->fip;
        // Remove o objeto da lista associado ao fragmento.
        delete Object;
        // Recupera o ponteiro para o proximo objeto FragmentInfo.
        Object = NextObject;
    }
    // Inicializa as variaveis da classe (para uma lista vazia).
    m_Start = NULL;
    m_End = NULL;

    #ifdef RIO_DEBUG2
    FragmentQueueSize = 0;
    #endif

    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - DeleteFragmentQueue] Finish" << endl;
    #endif
}

// Insere um objeto FragmentInfo no inicio da fila FIFO.
int CFragmentQueue::Insert( FragmentInfo *fip )
{
    FragmentInfoList *Object;

    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - Insert] Start" << endl;
    #endif

    // Tenta alocar um novo objeto da fila.
    Object = new FragmentInfoList;
    if( Object == NULL )
    {
        
        #ifdef RIO_DEBUG1
        RioErr << "[CFragmentQueue - Insert] Finish1" << endl;
        #endif

        return ERROR_STREAMCONTROL + ERROR_MEMORY;
    } 
    // Inicializa o objeto criado.
    Object->fip = fip;
    // Coloca o objeto criado no final da fila FIFO.
    Object->NextObject = NULL;
    if( m_End == NULL )
        m_Start = Object;
    else
        m_End->NextObject = Object;
    m_End = Object;

    #ifdef RIO_DEBUG2
    // Incrementa o contador usado pela depuracao (que conta o numero de 
    // fragmentos na fila), pois inserimos um fragmento no final da fila.
    FragmentQueueSize++;
    #endif

    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - Insert] Finish2" << endl;
    #endif

    return S_OK; 
}

// Remove o objeto FragmentInfo do final da fila FIFO.
FragmentInfo *CFragmentQueue::Remove()
{
    FragmentInfoList *Object;
    FragmentInfo *fip;

    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - Remove] Start" << endl;
    #endif

    // Se nao existir nenhum objeto, entao retornamos NULL.
    if( m_Start == NULL )
    {

        #ifdef RIO_DEBUG1
        RioErr << "[CFragmentQueue - Remove] Finish1" << endl;
        #endif

        return NULL;
    }
    // Obtem o ponteiro para o primeiro elemento da fila FIFO.
    Object = m_Start;
    // Atualiza a fila, removendo o objeto do inicio da fila.
    m_Start = m_Start->NextObject;
    if( m_Start == NULL )
        m_End = NULL;
    // Obtem o campo FragmentInfo do objeto (que sera retornado pela funcao).
    fip = Object->fip;
    // Remove o objeto retirado da fila.
    delete Object;

    #ifdef RIO_DEBUG2
    // Incrementa o contador usado pela depuracao (que conta o numero de 
    // fragmentos na fila), pois removemos um fragmento do final da fila.
    FragmentQueueSize--;
    #endif
    // Retorna o objeto FragmentInfo associado ao primeiro elemento da fila que 
    // acabou de ser removido.

    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - Remove] Finish2" << endl;
    #endif

    return fip;
}

// Verifica se a fifa FIFO esta vazia, e retorna true caso ela esteja vazia, e 
// false se existirem objetos FragmentInfo para serem processados.
bool CFragmentQueue::IsEmpty()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - Remove] Single" << endl;
    #endif

    // Retorna true somente se a fila estiver vazia (pois somente neste caso
    // teremos m_Start igual a NULL).
    return m_Start == NULL;
}

#ifdef RIO_DEBUG2
// Funcao usada para depuracao: imprime a lista de FragmentsInfo. O parametro 
// QueueInfoInfo contem uma string que sera impressa ao imprimirmos a lista com 
// os objetos FragmentsInfo.
void CFragmentQueue::PrintFragmentQueue( const char *QueueInfo )
{
    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - PrintFragmentQueue] Start" << endl;
    #endif

    // Esta parte esta comentada pois a impressao da lista e muito grande.
    // se for necessario imprimir a lista, descomentar esta parte do codigo
    // ate a parte nao comentada.
    //FragmentInfoList *Object;
    // Imprime a lista de objetos
    //RioErr << "- Lista com os objetos FragmentInfo (" << QueueInfo << "): "; 
    //Object = m_Start;
    //if ( Object == NULL )
    //    RioErr << " nenhum objeto armazenado." << endl;
    //else 
    //{    
    //    while( Object != NULL )
    //    {
    //        struct in_addr objectip;
    //        objectip.s_addr = Object->fip->IP;
    //        RioErr << "[data = 0x" << hex  
    //               << (unsigned int) Object->fip->FragmentData << dec 
    //               << ", size = " << Object->fip->FragmentSize 
    //               << ", IP = " << inet_ntoa( objectip ) << ", port = "
    //               << ntohs( Object->fip->Port ) << "]";
    //        Object = Object->NextObject;
    //        if( Object == NULL )
    //            RioErr << ".";
    //        else
    //            RioErr << ",";   
    //    }
    //    RioErr << FragmentQueueSize << " objeto(s) armazenado(s) na fila." 
    //           << endl;    
    //}
    // Imprime o numero de objetos na fila, usando o contador com o numero de
    // objetos na fila.
    RioErr << "- Numero de objetos FragmentInfo (" << QueueInfo << "): ";
    if ( FragmentQueueSize == 0 )
        RioErr << " nenhum objeto armazenado." << endl;
    else
        RioErr << FragmentQueueSize << " objeto(s) armazenado(s)." << endl;     
        
    #ifdef RIO_DEBUG1
    RioErr << "[CFragmentQueue - PrintFragmentQueue] Finish" << endl;
    #endif
}
#endif

///////////////////////////////////////////////////////////////////////////////
// Classe CLeakBucket - Implementa uma estrutura basica de um Leak Bucket,   //
// usada para armazenar fragmentos a serem envidaos pelos clientes.          //
/////////////////////////////////////////////////////////////////////////////// 

// Construtor da classe. O parametro MaxCredits define o numero maximo de 
// creditos e o parametro TimeBetweenCredits define o tempo (em milisegundos) 
// entre as atualizacoes de credito. 
CLeakBucket::CLeakBucket( unsigned int MaxCredits, 
                          unsigned long long int TimeBetweenCredits )
{
    struct timeval CreateTime;

    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - Construtor] Start" << endl;
    #endif

    // Inicializa as variaveis da classe.
    SetMaxCredits( MaxCredits );
    SetTimeBetweenCredits( TimeBetweenCredits );
    m_Credits = 1; //MaxCredits;
    gettimeofday( &CreateTime, NULL );
    m_LastUpdateCredits = ( ( unsigned long long int ) CreateTime.tv_sec ) 
                          * 1000000ull + 
                          ( ( unsigned long long int ) CreateTime.tv_usec );
    #ifdef RIO_DEBUG2
    PrintLeakBucketInfo( "Construtor da classe" );
    #endif

    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - Construtor] Finish" << endl;
    #endif
} 

// Destrutor da classe
CLeakBucket::~CLeakBucket()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - Destrutor] Single" << endl;
    #endif
}

// Define o numero maximo de creditos.
void CLeakBucket::SetMaxCredits( unsigned int Value )
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - SetMaxCredits] Start" << endl;
    #endif

    // Altera o valor m_MaxCredits, com o numero maximo de creditos.
    m_MaxCredits = Value;
    // Garante que o valor de m_Credits nao e maior do que m_MaxCredits.
    if( m_Credits > m_MaxCredits )
        m_Credits = m_MaxCredits;

    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - SetMaxCredits] Finish" << endl;
    #endif
}

// Define o intervalo de tempo entre as atualizacoes (em microsegundos). 
void CLeakBucket::SetTimeBetweenCredits( unsigned long long int Value )
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - SetTimeBetweenCredits] Single" << endl;
    #endif

    // Altera o valor m_TimeBetweenCredits, com o numero maximo de creditos.
    m_TimeBetweenCredits = Value;
}

#ifdef RIO_DEBUG2
// Funcao usada para depuracao: imprime o numero de creditos, o numero maximo de 
// creditos, o tempo entre as atualizacoes e o tempo decorrido apos a ultima 
// atualizacao.
void CLeakBucket::PrintLeakBucketInfo( const char *LeakBucketInfo )
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - PrintLeakBucketInfo] Single" << endl;
    #endif
    
    RioErr << "- Informacoes sobre o objeto LeakBucket (" << LeakBucketInfo 
           << "): MaxCredits = " << m_MaxCredits << ", Credits = " << m_Credits
           << ", TimeBetweenCredits = " << m_TimeBetweenCredits
           << ", LastUpdateCredits = " << m_LastUpdateCredits << "." << endl; 
}
#endif

// Obtem o numero de creditos.
unsigned int CLeakBucket::GetCredits()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - GetCredits] Single" << endl;
    #endif

    return m_Credits;
} 

// Obtem o numero maximo de creditos.
unsigned int CLeakBucket::GetMaxCredits()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - GetMaxCredits] Single" << endl;
    #endif

    return m_MaxCredits;
} 

// Adiciona creditos ao bucket. 
void CLeakBucket::AddCredits( unsigned int Value )
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - AddCredits] Start" << endl;
    #endif

    if( m_Credits < m_MaxCredits )
    {
        // Soma Value a m_Credits.
        m_Credits = m_Credits + Value;
        // Garante que o valor de m_Credits nao e maior do que m_MaxCredits.
        if( m_Credits > m_MaxCredits )
            m_Credits = m_MaxCredits;
        #ifdef RIO_DEBUG2
        // Imprime as informacoes do LeakBucket, apos a atualizacao.
        PrintLeakBucketInfo( "Ao adicionar creditos" );
        #endif
    } 

    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - AddCredits] Finish" << endl;
    #endif
}

// Remove um credito do bucket.
void CLeakBucket::DecCredits()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - DecCredits] Start" << endl;
    #endif

    // Subtrai um credito de m_Credits  
    if( m_Credits > 0 )
    {
        m_Credits = m_Credits - 1;
        #ifdef RIO_DEBUG2
        // Imprime as informacoes do LeakBucket, apos a atualizacao.
        PrintLeakBucketInfo( "Ao decrementar creditos" );
        #endif
    }    

    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - DecCredits] Finish" << endl;
    #endif
}

// Verifica se um objeto da classe possui creditos.
bool CLeakBucket::VerifyCredits()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - VerifyCredits] Single" << endl;
    #endif

    return  m_Credits > 0;
}

// Remove um objeto da fila FIFO do bucket. O objeto somente e removido se 
// existirem creditos e, caso seja removido, o numero de creditos e decrementado 
// de acordo. A funcao retorna o objeto ou NULL caso o objeto nao possa ser 
// removido por falta de creditos ou nao exista um objeto na fila. 
FragmentInfo *CLeakBucket::Remove()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - Remove] Start" << endl;
    #endif

    if( ( !IsEmpty() ) && ( VerifyCredits() ) )
    {
        // Se existe um objeto FragmentInfo no LeakBucket e o cliente possuir
        // mais de um credito para envia-lo.
        DecCredits();

        #ifdef RIO_DEBUG1
        RioErr << "[CLeakBucket - Remove] Finish1" << endl;
        #endif

        // Remove o fragmento da fila FIFO, usando a funcao da classe pai.
        return CFragmentQueue::Remove();
    }
    else 
    {

        #ifdef RIO_DEBUG1
        RioErr << "[CLeakBucket - Remove] Finish2" << endl;
        #endif

        // Se nao existirem objetos FragmentInfo no LeakBucket ou se o cliente
        // nao possuir creditos.
        return NULL;
    }
}

// Atualiza os creditos da classe. Um credito sera adicionado apos terem passado 
// TimeBetweenCredits desde a ultima atualizacao (dada por m_LastUpdateCredits, 
// em microsegundos). Caso tenha passado um tempo maior, mais creditos serao 
// adicionados, de acordo com esta diferencaode tempo, e a variavel 
// m_LastUpdateCredits sera atualizada de acordo. 
void CLeakBucket::UpdateCredits()
{
    struct timeval tvUpdateTime;
    unsigned long long int UpdateTime, IntervalTime, Credits, RestTime;

    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - UpdateCredits] Start" << endl;
    #endif

    // Obtem o tempo em que a funcao UpdateCredits foi executada. 
    gettimeofday( &tvUpdateTime, NULL );
    // Converte o tempo para microsegundos.
    UpdateTime = ( ( unsigned long long int ) tvUpdateTime.tv_sec ) 
                 * 1000000ull + 
                 ( ( unsigned long long int ) tvUpdateTime.tv_usec );
    // Calcula o intervalo de tempo entre esta chamada e a anterior.
    IntervalTime = UpdateTime - m_LastUpdateCredits;
    // Verifica se devemos atualizar o numero de creditos.
    if( IntervalTime >= m_TimeBetweenCredits ) {
        // Como o intervalo entre as chamadas e maior ou igual ao valor de
        // m_TimeBetweenCredits, deveremos atualizar o credito de acordo com
        // o tempo passado, e ajustar a variavel m_LastUpdateCredits de
        // acordo.
        Credits = IntervalTime / m_TimeBetweenCredits;
        RestTime = IntervalTime % m_TimeBetweenCredits;
        #ifdef RIO_DEBUG2
        RioErr << "CLeakBucket::UpdateCredits Intervalo = " << IntervalTime 
               << ", m_TimeBetweenCredits = " << m_TimeBetweenCredits 
               << ", Credits = " << Credits << ", UpdateTime = " 
               << UpdateTime << ", RestTime = " << RestTime 
               << ", m_LastUpdateCredits = " << m_LastUpdateCredits << endl; 
        #endif
        AddCredits( Credits );
        m_LastUpdateCredits = UpdateTime - RestTime;
    }

    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - UpdateCredits] Finish" << endl;
    #endif
}

// Retorna o tempo (em microsegundos) que ainda falta passar para podermos fazer 
// uma atualizacao dos creditos.
unsigned long long int CLeakBucket::NextTimeToUpdateCredits()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CLeakBucket - NextTimeToUpdateCredits] Single" << endl;
    #endif

    return m_LastUpdateCredits + m_TimeBetweenCredits;
}

///////////////////////////////////////////////////////////////////////////////
// Classe CSendData - Implementa um Leak Bucket com os fragmentos a serem    //
// enviados, pelo controle de fluxo, de um servidor de armazenamento para os //
// clientes. Esta classe e derivada da classe CLeakBucket descrita           //
// anteriormente.                                                            // 
/////////////////////////////////////////////////////////////////////////////// 

// Constructor da classe. O parametro Value e o numero inicial de creditos. O 
// parametro MaxCredits define o numero maximo de creditos e o parametro 
// TimeBetweenCredits define o tempo (em milisegundos) entre as atualizacoes de 
// credito. 
CSendData::CSendData( unsigned int MaxCredits, 
                      unsigned long long int TimeBetweenCredits ) 
          :CLeakBucket( MaxCredits, TimeBetweenCredits )
{
    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - Construtor] Start" << endl;
    #endif

    // Inicializa o mutex da classe para o acesso a fila FIFO.
    pthread_mutex_init( &m_AccessFragmentQueue, NULL );
    // Inicializa o mutex da classe para o acesso as variaveis Credits e
    // MaxCredits.
    pthread_mutex_init( &m_AccessLeakBucket, NULL );
    // Cria a variavel de condicao da classe (usada para esperar por objetos).
    pthread_cond_init( &m_WaitObject, NULL );
    // Cria a variavel de condicao da classe (usada para esperar por creditos).
    pthread_cond_init( &m_WaitCredits, NULL );
    // Cria a variavel de condicao da classe (usada para esperar que os
    // creditos sejam decrementados em 1 unidade).
    pthread_cond_init( &m_WaitUseCredits, NULL );

    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - Construtor] Finish" << endl;
    #endif
}

// Destructor da classe.
CSendData::~CSendData()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - Destrutor] Start" << endl;
    #endif

    // Remove o mutex da classe para o acesso a fila FIFO.
    pthread_mutex_destroy( &m_AccessFragmentQueue );
    // Remove o mutex da classe para o acesso as variaveis Credits e
    // MaxCredits.
    pthread_mutex_destroy( &m_AccessLeakBucket );
    // Remove o mutex da classe para o acesso a fila FIFO.
    pthread_mutex_destroy( &m_AccessFragmentQueue );
    // Remove a variavel de condicao da classe (usada para esperar por
    // objetos).
    pthread_cond_destroy( &m_WaitObject );
    // Remove a variavel de condicao da classe (usada para esperar por
    // creditos).
    pthread_cond_destroy( &m_WaitCredits );
    // Remove a variavel de condicao da classe (usada para esperar que os
    // creditos sejam decrementados em 1 unidade).
    pthread_cond_destroy( &m_WaitUseCredits );
    
    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - Destrutor] Finish" << endl;
    #endif
}

// Remove um credito do bucket.
void CSendData::DecCredits()
{
    bool SendSignal;

    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - DecCredits] Start" << endl;
    #endif

    // Verifica se os creditos estavam no maximo.
    // Chama a funcao da classe pai para decrementar os creditos.
    CLeakBucket::DecCredits();
    SendSignal = ( GetCredits() == ( GetMaxCredits() - 1 ) );
    // Se o numero de creditos estava no maximo, enviamos um sinal para 
    // desbloquear, se necessario, a thread que incrementa os creditos.
    if( SendSignal )
        pthread_cond_signal( &m_WaitUseCredits );

    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - DecCredits] Finish" << endl;
    #endif
}

// Funcao para inserir um objeto FragmentInfo no final da fila SendList. A 
// insercao deste objeto ira gerar um sinal para a variavel de condicao 
// m_WaitObject caso a fila fe objetos FragmentInfo estivesse vazia antes da 
// insercao, para desbloquear a thread de envio que esta esperando por novos 
// objetos. 
int CSendData::Insert( FragmentInfo *fip )
{
    int result;
    bool SendSignal;

    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - Insert] Start" << endl;
    #endif

    // Chama o metodo da classe pai para inserir o objeto FragmentInfo.
    // Obtem acesso exclusivo a fila FIFO.
    pthread_mutex_lock( &m_AccessFragmentQueue );
    SendSignal = IsEmpty();
    result = CLeakBucket::Insert( fip );
    #ifdef RIO_DEBUG2
    // Imprime a lista obtida apos a insercao do objeto.
    CLeakBucket::PrintFragmentQueue( "Insercao em CSendData" );
    #endif
    if( ( result == S_OK ) && ( SendSignal ) )
        pthread_cond_signal( &m_WaitObject );
    // Libera o acesso exclusivo a fila FIFO.
    pthread_mutex_unlock( &m_AccessFragmentQueue );

    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - Insert] Finish" << endl;
    #endif

    return result;
}

// Funcao para remover o primeiro objeto FragmentInfo da fila da objetos a serem 
// enviados. Caso a fila esteja vazia, a funcao ira bloquear a thread de envio 
// na variavel de condicao m_WaitObject. Apos a chegada de um objeto, a funcao 
// ira verificar se existem creditos e, caso nao existam, ira bloquear a thread 
// de envio na variavel de condicao m_WaitCredits, esperando pela chegada de
// creditos. Apos isso, finalmente, a funcao ira retornar o objeto FragmentInfo.
FragmentInfo *CSendData::Remove()
{
    FragmentInfo *fip;

    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - Remove] Start" << endl;
    #endif

    // Verifica agora se existem creditos para enviar o objeto FragmentInfo. 
    // Obtem acesso exclusivo as variaveis de creditos da classe. Caso nao
    // existam creditos, espera na variavel de condicao m_WaitCredits.
    // Obtem acesso exclusivo as variaveis de creditos da classe.
    pthread_mutex_lock( &m_AccessLeakBucket );
    if( !VerifyCredits() ) 
    {
        #ifdef RIO_DEBUG2
        RioErr << "CSendData::Remove esperando por creditos." << endl;
        #endif
        pthread_cond_wait( &m_WaitCredits, &m_AccessLeakBucket );
    }
    // Decrementa o numero de creditos para garantir que nenhuma outra
    // thread use o credito obtido apos o desbloqueio.
    DecCredits();
    // Libera o acesso exclusivo as variaveis de creditos da classe.
    pthread_mutex_unlock( &m_AccessLeakBucket );
    // Obtem acesso exclusivo a fila FIFO.
    pthread_mutex_lock( &m_AccessFragmentQueue );
    // Verifica se existe um objeto na lista. Se nao existir nenhum objeto,
    // a thread sera bloqueada na variavel de condicao m_WaitObject para 
    // esperar pelo objeto.
    if( IsEmpty() ) 
    {
        #ifdef RIO_DEBUG2
        RioErr << "CSendData::Remove esperando por um objeto." << endl;
        #endif
        pthread_cond_wait( &m_WaitObject, &m_AccessFragmentQueue );
    }
    // Depois de esperarmos com sucesso (sem bloqueio, ou esperando e depois
    // sendo desbloqueado, ja existe um objeto FragmentInfo na fila da classe e 
    // temos creditos para envia-lo. Entao, basta pegar o objeto. 
    // Usamos a funcao Remove da classe pai para obter este objeto.
    fip = CFragmentQueue::Remove(); // Note que um retorno NULL indica um erro na
    #ifdef RIO_DEBUG2
    // Imprime a lista obtida apos a remocao do objeto.
    CLeakBucket::PrintFragmentQueue( "Remocao em CSendData" );
    #endif 
    // Libera o acesso exclusivo a fila FIFO.
    pthread_mutex_unlock( &m_AccessFragmentQueue );

    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - Remove] Finish" << endl;
    #endif

    return fip;
}

// Funcao para atualizar os creditos da classe. Se os creditos eram 0 antes da 
// atualizacao, gera um sinal para desbloquear a thread de envio caso esta 
// esteja bloqueada na variavel de condicao m_WaitCredits. Se os creditos 
// estiverem no maximo, a thread de atualizacao sera bloqueada ate que os 
// creditos sejam reduzidos pela funcao DecCredits.
void CSendData::UpdateCredits()
{
    bool SendSignal, Wait;

    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - UpdateCredits] Start" << endl;
    #endif

    // Obtem o tempo de espera, em que a thread de atualizacao dos creditos
    // devera ficar bloqueada (usando a funcao usleep).
    // Obtem acesso exclusivo as variaveis de creditos da classe
    pthread_mutex_lock( &m_AccessLeakBucket );
    // Verifica se precisamos bloquear esperando pelo decrescimo dos creditos.
    // Isso ocorrera se o numero de creditos for igual ao maximo.
    Wait = ( GetCredits() == GetMaxCredits() );
    if( Wait )
        pthread_cond_wait( &m_WaitUseCredits, &m_AccessLeakBucket );
    SendSignal = !VerifyCredits();
    // Atualiza os creditos da classe.
    CLeakBucket::UpdateCredits(); 
    // Envia o sinal para desbloquear a thread de envio caso nao existissem
    // creditos (a thread pode estar bloqueada esperando por threads na 
    // variavel de condicao m_WaitCredits).
    if( SendSignal )
        pthread_cond_signal( &m_WaitCredits );
    // Libera o acesso exclusivo as variaveis de creditos da classe
    pthread_mutex_unlock( &m_AccessLeakBucket );

    #ifdef RIO_DEBUG1
    RioErr << "[CSendData - UpdateCredits] Finish" << endl;
    #endif
}

///////////////////////////////////////////////////////////////////////////////
// Classe CFPool - Implementa a classe que armaeza a lista com os fragmentos //
// recebidos e que ainda nao foram processados pelo controle de fluxo.       // 
/////////////////////////////////////////////////////////////////////////////// 

// Construtor da classe.
CFPool::CFPool()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - Construtor] Start" << endl;
    #endif

    // Cria a variavel de condicao da classe.
    pthread_cond_init( &m_WaitObject, NULL );
    // Cria o mutex usado para garantir o acesso exclusivo a fila FIFO.
    pthread_mutex_init( &m_AccessFragmentQueue, NULL );

    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - Construtor] Finish" << endl;
    #endif
}

// Destrutor da classe.
CFPool::~CFPool()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - Destrutor] Start" << endl;
    #endif

    // Remove a variavel de condicao da classe.
    pthread_cond_destroy( &m_WaitObject );
    // remove o mutex usado para garantir o acesso exclusivo a fila FIFO.
    pthread_mutex_destroy( &m_AccessFragmentQueue );

    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - Destrutor] Finish" << endl;
    #endif
}

// Funcao para inserir um objeto FragmentInfo no final da fila de objetos 
// FragmentInfo. Se a fila estiver vazia, a funcao ira gerar um sinal para a 
// variavel de condicao WaitObject.
int CFPool::Insert( FragmentInfo *fip )
{
    bool SendSignal;
    int result;

    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - Insert] Start" << endl;
    #endif

    // Usa a funcao Insert da classe pai para inserir o objeto FragmentInfo na
    // fila FIFO.
    // Obtem acesso exclusivo a fila FIFO.
    pthread_mutex_lock( &m_AccessFragmentQueue );
    SendSignal = IsEmpty();
    result = CFragmentQueue::Insert( fip );
    #ifdef RIO_DEBUG2
    // Imprime a lista obtida apos a insercao do objeto.
    CFragmentQueue::PrintFragmentQueue( "Insercao em CFPool" );
    #endif
    // Se nao ocorreu nenhum erro na insercao, devemos gerar um sinal
    // informando que um objeto foi inserido, se o pool estava vazio antes
    // da insercao (obs: devemos usar S_OK ao inves de S_OK?) 
    if( ( result == S_OK ) && ( SendSignal ) ) 
        // Envia um sinal para a variavel de condicao, para desbloquear, se
        // necessario, uma thread que esteja esperando pela condicao, que e a
        // insercao de um objeto FragmentInfo no pool.
        pthread_cond_signal( &m_WaitObject );
    // Libera o acesso exclusivo a fila FIFO.
    pthread_mutex_unlock( &m_AccessFragmentQueue );

    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - Insert] Finish" << endl;
    #endif

    return result;
}

// Funcao para esperar que o pool de objetos FragmentInfo nao esteja vazio. Se o 
// pool estiver vazio, a thread que chamou WaitObject sera bloqueada em uma 
// variavel de condicao ate que um objeto seja Inserido e um sinal seja enviado 
// para desbloquear a thread.
void CFPool::WaitObject()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - WaitObject] Start" << endl;
    #endif

    // Usa a funcao Remove da classe pai para tentar remover um objeto
    // FragmentInfo da fila FIFO.
    // Obtem acesso exclusivo a fila FIFO.
    pthread_mutex_lock( &m_AccessFragmentQueue );
    if( IsEmpty() ) 
    {
        // Espera na variavel de condicao pela chegada de um objeto 
        // FragmentInfo. A thread que chamou Insert sera bloqueada ate que a 
        // funcao Insert seja chamada e gere um sinal para a variavel de 
        // condicao, o que ira em consequencia desbloquear a thread.
        #ifdef RIO_DEBUG2
        RioErr << "CFPool::WaitObject esperando por um objeto." << endl;
        #endif
        pthread_cond_wait( &m_WaitObject, &m_AccessFragmentQueue );
    }
    // Libera o acesso exclusivo a fila FIFO.
    pthread_mutex_unlock( &m_AccessFragmentQueue );

    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - WaitObject] Finish" << endl;
    #endif
}

// Funcao para remover o objeto FragmentInfo do inicio da fila de objetos 
// FragmentInfo. Se nao existir nenhum objeto, a funcao ira retornar NULL.
FragmentInfo *CFPool::Remove()
{
    FragmentInfo *fip;

    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - Remove] Start" << endl;
    #endif

    // Usa a funcao Remove da classe pai para tentar remover um objeto
    // FragmentInfo da fila FIFO.
    // Obtem acesso exclusivo a fila FIFO.
    pthread_mutex_lock( &m_AccessFragmentQueue );
    fip = CFragmentQueue::Remove();
    #ifdef RIO_DEBUG2
    // Imprime a lista obtida apos a remocao do objeto.
    CFragmentQueue::PrintFragmentQueue( "Remocao em CFPool" );
    #endif
    // Libera o acesso exclusivo a fila FIFO.
    pthread_mutex_unlock( &m_AccessFragmentQueue );

    #ifdef RIO_DEBUG1
    RioErr << "[CFPool - Remove] Finish" << endl;
    #endif

    return fip;
}

///////////////////////////////////////////////////////////////////////////////
// Classe CClientInfo - Implementa a classe com a lista de fragmentos de um  //
// cliente que ainda nao escalonados para serem  enviados, pelo controle de  //
// fluxo, de um servidor de armazenamento para este cliente. Esta classe     // 
// tambem e derivada da classe CLeakBucket descrita anteriormente.           //                                                 // 
/////////////////////////////////////////////////////////////////////////////// 

// Constructor da classe. O parametro Value e o numero inicial de creditos. O 
// parametro MaxCredits define o numero maximo de creditos e o parametro 
// TimeBetweenCredits define o tempo (em milisegundos) entre as atualizacoes de 
// credito. O parametro ip define o endereco IP do cliente. Finalmente, o 
// parametro port define a porta do cliente.
CClientInfo::CClientInfo( unsigned int MaxCredits, 
                          unsigned long long int TimeBetweenCredits,
                          int ip, int port ) 
          :CLeakBucket( MaxCredits, TimeBetweenCredits )
{
    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - Construtor] Start" << endl;
    #endif

    // Inicializa as variaveis da classe.
    IP = ip; // IP do cliente (na ordem da rede).
    Port = port; // Porta do cliente (na ordem da rede).
    IdleCount = MAXIDLECOUNT; // Inicializa o contador de inatividade.
    // Cria o mutex para garantir o acesso exclusivo a fila do LeakBucket.
    pthread_mutex_init( &m_AccessFragmentQueue, NULL );

    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - Construtor] Finish" << endl;
    #endif
}

// Destructor da classe.
CClientInfo::~CClientInfo()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - Destrutor] Start" << endl;
    #endif

    // Remove o mutex para garantir o acesso exclusivo a fila do LeakBucket.
    pthread_mutex_destroy( &m_AccessFragmentQueue );

    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - Destrutor] Finish" << endl;
    #endif
}

// Funcao para inserir um objeto FragmentInfo no final da fila do cliente. O 
// contador de inatividade do cliente sera alterado para o valor maximo, dado 
// pela constante MAXIDLECOUNT, pois o cliente deixou de estar inativo.  
int CClientInfo::Insert( FragmentInfo *fip )
{
    int result;

    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - Insert] Start" << endl;
    #endif

    // Obtem o acesso ao mutex para garantir o acesso exclusivo a fila do 
    // LeakBucket.
    pthread_mutex_lock( &m_AccessFragmentQueue );
    // Inicializa o contador de inatividade com o valor maximo.
    IdleCount = MAXIDLECOUNT;
    // Chama a funcao da classe pai para inserir o fragmento na fila.
    result = CLeakBucket::Insert( fip );
    // Libera o mutex para garantir o acesso exclusivo a fila do LeakBucket.
    pthread_mutex_unlock( &m_AccessFragmentQueue );

    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - Insert] Finish" << endl;
    #endif

    return result;    
}

// Funcao para remover o primeiro objeto FragmentInfo da fila da objetos a serem 
// enviados. 
FragmentInfo *CClientInfo::Remove()
{
    FragmentInfo *fip;

    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - Remove] Start" << endl;
    #endif

    // Obtem o acesso ao mutex para garantir o acesso exclusivo a fila do 
    // LeakBucket.
    pthread_mutex_lock( &m_AccessFragmentQueue );
    // Chama a funcao da classe pai para remover o fragmento na fila.
    fip = CLeakBucket::Remove();
    // Libera o mutex para garantir o acesso exclusivo a fila do LeakBucket.
    pthread_mutex_unlock( &m_AccessFragmentQueue );

    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - Remove] Finish" << endl;
    #endif

    return fip;    
}

// Funcao para atualizar os creditos da classe. Se os creditos da classe forem 
// atualizados e a fila FIFO de fragmentos estiver vazia, entao o contador de 
// inatividade sera decrementado de uma unidade, se ainda for maior do que 0.
void CClientInfo::UpdateCredits()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - UpdateCredits] Start" << endl;
    #endif

    // Obtem o acesso ao mutex para garantir o acesso exclusivo a fila do 
    // LeakBucket.
    pthread_mutex_lock( &m_AccessFragmentQueue );
    // Verifica se a fila FIFO esta vazia, pois neste caso deveremos decrementar
    // o contador de inatividade em uma unidade (se ele nao for zero).
    if( ( IsEmpty() ) && ( IdleCount > 0 ) )
        IdleCount--;
    // Libera o mutex para garantir o acesso exclusivo a fila do LeakBucket.
    pthread_mutex_unlock( &m_AccessFragmentQueue );
    // Chama a funcao da classe pai para atualizar os creditos do cliente.
    CLeakBucket::UpdateCredits();    

    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - UpdateCredits] Finish" << endl;
    #endif
}

// Funcao usada para verificar se o cliente deve ser removido, isto e, se o 
// contador de inatividade for igual a zero.
bool CClientInfo::MustRemoveClient()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - MustRemoveClient] Single" << endl;
    #endif

    return ( IdleCount == 0 ); // Se o contador for zero, entao o cliente 
                               // devera ser removido. Em caso contrario, ainda
                               // nao devera ser removido.
}

// Retorna o endereco IP do cliente.
int CClientInfo::GetIPAddress()
{
    return IP; // Retorna o IP do cliente.
}

// Retorna a porta do cliente.
int CClientInfo::GetPort()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CClientInfo - GetPort] Single" << endl;
    #endif

    return Port; // Retorna a porta do cliente.
}

///////////////////////////////////////////////////////////////////////////////
// Classe CClientHeap - Implementa um heap com os clientes atualmente        // 
// conectados, em tempo real, ao servidor. Cada no do heap esta associado a  //
// um cliente, e aponta para o objeto da classe CClientInfo associado ao     //
// objeto com a lista de fragmentos deste cliente. O heap sera ordenado pelo //
// tempo de atualizacao dos creditos dos clientes e, com isso, o elemento no //
// topo do heap e aquele com o tempo de atualizacao dos creditos mais proximo//
// do tempo atual. Alem disso, devido ao modo como o heap e ordenado, ele    //
// pode ser visto como uma lista de eventos, com cada evento sendo a         //
// atualizacao dos creditos do heap seguida pelo envio de um numero de       //
// fragmentos nao maior do que o numero de creditos deste cliente apos a     //
// atualizacao dos creditos.                                                 //
/////////////////////////////////////////////////////////////////////////////// 

// Construtor da classe. O parametro e o numero maximo de creditos para o envio 
// de fragmentos pelo cliente.
CClientHeap::CClientHeap( unsigned int MaxClientsCredits )
{
    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - Construtor] Start" << endl;
    #endif

    // Inicializa as variaveis da classe
    // ClientsHeap: a fila de prioridades nao precisa ser inicializada.
    // ClientsMap: o mapa dos clientes nao precisa ser inicializado.
    m_MaxClientsCredits = MaxClientsCredits;
    // Inicializa a variavel m_LastClient.
    m_LastClient = NULL;
    // Cria o mutex usado para garantir o acesso exclusivo ao mapa e a 
    // variavel m_LastClient.
    pthread_mutex_init( &m_AccessClients, NULL );
    // Cria o mutex usado para garantir o acesso exclusivo ao heap.
    pthread_mutex_init( &m_AccessClientsHeap, NULL );
    // Cria a variavel de condicao da classe (usada para esperar por clientes).
    pthread_cond_init( &m_WaitClients, NULL );

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - Construtor] Finish" << endl;
    #endif
}

// Destrutor da classe.
CClientHeap::~CClientHeap()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - Destrutor] Start" << endl;
    #endif

    // Remove o mutex usado para garantir o acesso exclusivo ao mapa e a 
    // variavel m_LastClient.
    pthread_mutex_destroy( &m_AccessClients );
    // Remove o mutex da classe para o acesso ao heap.
    pthread_mutex_destroy( &m_AccessClientsHeap );
    // Cria a variavel de condicao da classe (usada para esperar por clientes).
    pthread_cond_destroy( &m_WaitClients );

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - Destrutor] Finish" << endl;
    #endif
}

// Funcao para buscar um cliente no heap, usando o seu ip e a sua porta (os 
// bytes dos inteiros devem estar na ordem de bytes da rede (network byte 
// order). Se o cliente nao estiver no heap, sera retornado NULL e, em caso 
// contrario, sera retornado a estrutura do heap para este cliente.
CClientInfo *CClientHeap::FindClient( int ip, int port )
{
    unsigned long long int ClientMapId;
    // Define um iterador para uma posicao qualquer do mapa, e um iterador para 
    // o final do mapa.   
    TClientsMap::iterator ClientIt, EndIt;

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - FindClient] Start" << endl;
    #endif

    #ifdef RIO_DEBUG2
    struct in_addr clientaddr;
    clientaddr.s_addr = ip;
    RioErr << "CClientList::FindClient procurando pelo cliente com o IP " 
           << inet_ntoa( clientaddr ) << " e a porta " 
           << ntohs( port ) << endl;
    #endif       
    // Determina o ID (com 64 bits) do cliente (os 32 bits superiores sao a
    // porta do cliente, e os 32 bits inferiores sao o IP do cliente).
    ClientMapId = ( unsigned int ) port;
    ClientMapId = ( ClientMapId << 32 ) | ( ( unsigned int ) ip );
    // Verifica se o cliente com a ID computada anteriormente esta no mapa.
    ClientIt = m_ClientsMap.find( ClientMapId );
    EndIt = m_ClientsMap.end();
    if( ClientIt != EndIt ) 
    {
        #ifdef RIO_DEBUG2
        RioErr << "CClientList::FindClient cliente com a ID " << ClientMapId
               << " achado no mapa." << endl;
        #endif

        #ifdef RIO_DEBUG1
        RioErr << "[CClientHeap - FindClient] Finish1" << endl;
        #endif

        // Se o cliente estiver no mapa, entao retornamos um ponteiro para uma 
        // estrutura CHeapElement com as informacoes do cliente (este ponteiro 
        // e dado por it->second. O valor id->first e igual ao valor de ID).
        return ClientIt->second;
    }    
    else
    {
        #ifdef RIO_DEBUG2
        RioErr << "CClientList::FindClient cliente com a ID " << ClientMapId
               << " nao achado no mapa." << endl;
        #endif       

        #ifdef RIO_DEBUG1
        RioErr << "[CClientHeap - FindClient] Finish2" << endl;
        #endif

        // Se o cliente nao estiver no mapa, entao retornamos NULL.
        return NULL;
    }
}

// Funcao para criar uma nova lista para um cliente identiticado pelos
// paremetros ip e port, e cujo tempo da atualizacao entre os creditos e dado 
// em TimeBetweenCredits. 
CClientInfo *CClientHeap::CreateClient( int ip, int port, 
                                        unsigned long long int TimeBetweenCredits )
{
    unsigned long long int ClientMapId;
    CClientInfo *Client;

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - CreateClient] Start" << endl;
    #endif

    #ifdef RIO_DEBUG2
    struct in_addr clientaddr;
    clientaddr.s_addr = ip;
    RioErr << "CClientList::CreateClient criando uma lista para o cliente com " 
           << "o IP " << inet_ntoa( clientaddr ) << " e a porta " 
           << ntohs( port ) << endl;
    #endif        
    // Tentamos criar uma lista para o cliente identificado por IP e port. 
    Client = new CClientInfo( m_MaxClientsCredits, TimeBetweenCredits, ip, 
                              port );
    if( Client == NULL )
    {

        #ifdef RIO_DEBUG1
        RioErr << "[CClientHeap - CreateClient] Finish1" << endl;
        #endif

        return NULL;
    }                          
    // Adiciona o cliente no heap.
    // Obtem acesso exclusivo ao heap.
    pthread_mutex_lock( &m_AccessClientsHeap );
    // Verifica se o heap esta vazio, pois neste caso, precisamos gerar um 
    // sinal para desbloquear a thread m_ClientsThread que processa os 
    // clientes do heap
    if( m_ClientsHeap.empty() )
        pthread_cond_signal( &m_WaitClients );
    m_ClientsHeap.push( Client );
    // Libera o acesso exclusivo ao heap.
    pthread_mutex_unlock( &m_AccessClientsHeap );
    // Determina o ID (com 64 bits) do cliente (os 32 bits superiores sao a
    // porta do cliente, e os 32 bits inferiores sao o IP do cliente).
    ClientMapId = ( unsigned int ) port;
    ClientMapId = ( ClientMapId << 32 ) | ( ( unsigned int ) ip );       
    // Devemos agora inserir o cliente no mapa.
    m_ClientsMap.insert( pair< unsigned long long int, CClientInfo * >
                         ( ClientMapId, Client ) );
    #ifdef RIO_DEBUG2
    RioErr << "CClientList::CreateClient inserindo no mapa o cliente com o IP " 
           << inet_ntoa( clientaddr ) << ", a porta " << ntohs( port ) 
           << ", e a ID " << ClientMapId << endl;
    #endif

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - CreateClient] Finish2" << endl;
    #endif

    return Client;
}

// Funcao para remover do heap o cliente cujo ponteiro e passado como parametro. 
// A funcao retorna true se a remocao do cliente foi feita com sucesso, e false 
// se ela nao foi feita com sucesso (isto e, se a thread que insere fragmentos 
// nos clientes estiver inserindo um fragmento neste cliente).
bool CClientHeap::RemoveClient( CClientInfo *Client )
{
    unsigned long long int ClientMapId;
    bool result;

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - RemoveClient] Start" << endl;
    #endif

    // Note que o cliente ja estara fora do heap (pois sempre removemos o 
    // cliente do heap e o recolocamos apos atualizar o tempo de atualizacao
    // entre os creditos).        
    // Obtem acesso exclusivo ao mapa e a variavel m_LastClient.
    pthread_mutex_lock( &m_AccessClients );
    if( ( Client->MustRemoveClient() ) && ( m_LastClient != Client ) )
    {
        #ifdef RIO_DEBUG2
        struct in_addr clientaddr;
        clientaddr.s_addr = Client->GetIPAddress();
        RioErr << "CClientList::RemoveClient removendo a lista do cliente com o "
               << "IP " << inet_ntoa( clientaddr ) << " e a porta " 
               << ntohs( Client->GetPort() ) << endl;
        #endif
        // Determina o ID (com 64 bits) do cliente (os 32 bits superiores sao a
        // porta do cliente, e os 32 bits inferiores sao o IP do cliente).
        ClientMapId = ( unsigned int ) Client->GetPort();
        ClientMapId = ( ClientMapId << 32 ) | 
                      ( ( unsigned int ) Client->GetIPAddress() );       
        // Remove o cliente do mapa.
        m_ClientsMap.erase( ClientMapId );
        // Corrige o ponteiro de  busca caso ele aponte para este cliente.
        if( m_LastClient == Client )
            m_LastClient = NULL;
        #ifdef RIO_DEBUG2
        RioErr << "CClientList::RemoveClient removendo do mapa o cliente com " 
               << "o IP " << inet_ntoa( clientaddr ) << ", a porta " 
               << ntohs( Client->GetPort() ) << ", e a ID " << ClientMapId  
               << endl;
        #endif
        // Deleta o objeto associado ao cliente.    
        delete Client;
        result = true;
    }
    else
        result = false;
    // Libera o acesso exclusivo ao mapa e a variavel m_LastClient.
    pthread_mutex_unlock( &m_AccessClients );

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - RemoveClient] Finish" << endl;
    #endif

    return result;
}

// Funcao para inserir um object FragmentInfo na lista do cliente. Se nao 
// existir uma lista de fragmentos do cliente, ela sera criada pela funcao.
int CClientHeap::Insert( FragmentInfo *fip )
{
    int result;
    CClientInfo *Client;

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - Insert] Start" << endl;
    #endif

    // Obtem acesso exclusivo ao mapa e a variavel m_LastClient.
    pthread_mutex_lock( &m_AccessClients );
    // Determina se precisamos buscar um novo cliente (isso sera visto pelo par 
    // IP, porta do cliente dado em m_LastClient. Se o valor de m_LastClient for 
    // NULL, e se o IP ou a porta forem diferentes do cliente refenreciado por
    // m_LastClient, faremos uma nova busca pelo cliente e, se nao acharmos este
    // cliente, criaremos uma nova lista para ele.
    if( ( m_LastClient == NULL ) || ( m_LastClient->GetIPAddress() != fip->IP ) ||
        ( m_LastClient->GetPort() != fip->Port ) )
    {
        // Tenta obter a lista do cliente. Se necessario, cria uma nova lista 
        // para o cliente. Tentaremos buscar o cliente ate conseguirmos obter 
        // um ponteiro.
        Client = FindClient( fip->IP, fip->Port );
        if( Client == NULL ) 
        {
            // Precisamos criar uma nova lista para o cliente, pois ele nao
            // foi achado no mapa dos clientes.
            Client = CreateClient( fip->IP, fip->Port, 
                                   fip->TimeBetweenCredits );
            if( Client == NULL )
            {

                #ifdef RIO_DEBUG1
                RioErr << "[CClientHeap - Insert] Finish1" << endl;
                #endif

                return ERROR_STREAMCONTROL + ERROR_MEMORY;
            }
        }
        m_LastClient = Client;
    } 
    else
        Client = m_LastClient; // Como o ultimo fragmento inserido foi o do
                               // cliente para o qual devemos enviar fip, 
                               // entao simplesmente usados o ponteiro ao
                               // inves de busca-lo no mapa. 
    // Libera o acesso exclusivo ao mapa e a variavel m_LastClient.
    pthread_mutex_unlock( &m_AccessClients );
    // Agora adicionamos o novo objeto FragmentInfo na fila FIFO de objetos 
    // do cliente referenciado por Client (ou seja, no final da fila).
    result = Client->Insert( fip );
    if( result == S_OK ) 
    {
        // Agora imprimimos a lista do cliente, com usando seu par IP, porta
        // para identifica-lo. 
        #ifdef RIO_DEBUG2
        struct in_addr clientaddr;
        clientaddr.s_addr = Client->GetIPAddress();
        char StrAux[ 256 ];
        sprintf( StrAux, "Insercao em CClientList [IP=%s,Port=%u]",
                 inet_ntoa( clientaddr ), ntohs( Client->GetPort() ) );                
        Client->PrintFragmentQueue( StrAux );
        #endif 
        
    }

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - Insert] Finish2" << endl;
    #endif

    return result;  
}

// Pega os proximos fragmentos do cliente que esta no topo do heap. O vetor
// Fragments (do STL) contera os fragmentos. Note que o tamanho do vetor dara o 
// numero de fragmentos a serem enviados. A funcao ira dormir, se necessario, 
// para esperar pelo tempo em que os creditos do cliente devem ser atualizados. 
// O cliente somente sera reinserido no heap (com um novo tempo de atualizacao 
// do cretido), se possuir fragmentos ou se nao possuir fragmentos e o contador 
// de inatividade for maior do que 0. No primeiro caso, como o cliente possui 
// fragmentos, o contador de inatividade sera alterado para o valor maximo 
// MAXREMOVECOUNT. A variavel NextFragment e um ponteiro para um valor booleano 
// que sera true se devemos continuar a pegar fragmentos do heap, e false se ja 
// esta na hora de pegarmos mais framentos do pool e copia-los para o heap.
void CClientHeap::GetFragments( TFragments &Fragments )
{
    unsigned long long int CurrentTime, UpdateTime;
    bool InsertClient;
    FragmentInfo *fip;
    CClientInfo *Client;
    struct timeval tvCurrentTime;

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - GetFragments] Start" << endl;
    #endif

    // Obtem a cabeca do heap.
    // Obtem acesso exclusivo ao heap.
    pthread_mutex_lock( &m_AccessClientsHeap );
    // Verifica se existem elementos no heap. Caso o heap esteja vazio, espera
    // pela chegada de novos clientes.
    if( m_ClientsHeap.empty() )
        pthread_cond_wait( &m_WaitClients, &m_AccessClientsHeap );
    // Agora que temos pelo menos um  cliente (pois o heap nao esta vazio), 
    // entao vamos processar o proximo cliente, que esta no topo do heap, 
    // associado ao evento que devera ser primeiramente executado.
    Client = m_ClientsHeap.top();
    // Remove a cabeca do heap.
    m_ClientsHeap.pop();
    // Libera o acesso exclusivo ao heap.
    pthread_mutex_unlock( &m_AccessClientsHeap );
    // Obtem o tempo de atualizacao do credito.
    UpdateTime = Client->NextTimeToUpdateCredits(); 
    // Determina o tempo atual.
    gettimeofday( &tvCurrentTime, NULL );
    // Converte o tempo atual (computado anteriormente) para microsegundos.
    CurrentTime = ( ( unsigned long long int ) tvCurrentTime.tv_sec ) 
                  * 1000000ull + 
                  ( ( unsigned long long int ) tvCurrentTime.tv_usec );
    // Verifica se precisamos dormir por alguns microsegundos. Isso ocorrera 
    // se o tempo de atualizacao x for maior do que o tempo atual a. Se o 
    // tempo de atualizacao x for maior, entao deverenis dormir por a - x 
    // microsegundos.
    if( UpdateTime > CurrentTime )
        usleep( UpdateTime - CurrentTime );
    // Agora, como ja passou o tempo de atualizacao do cliente (pelo menos, 
    // isso deveria ocorrer), atualizamos os creditos do cliente.
    Client->UpdateCredits();
    // Inicializa a variavel que diz se devemos ou nao inserir novamente o 
    // cliente no heap.
    InsertClient = true;
    // Verificamos se temos fragmentos para o cliente. Note que o cliente 
    // deveria possuir creditos, pois dormimos ate o tempo da proxima 
    // atualizacao do credito do cliente.
    Fragments.clear();
    while ( ( fip = Client->Remove() ) != NULL )
    {
        // Insere os fragmentos no vetor enquanto o cliente possuir creditos 
        // ou fragmentos para serem enviados.
        Fragments.insert( Fragments.end(), fip );
    }
    if( Fragments.size() > 0 ) 
    {
        #ifdef RIO_DEBUG2
        struct in_addr clientaddr;
        char StrAux[ 256 ];
        unsigned int f;
        for( f = 0; f < Fragments.size(); f++ ) 
        {
            clientaddr.s_addr = Fragments[ f ]->IP;
            sprintf( StrAux, 
                     "enviando fragmento %u para o cliente [IP=%s,Port=%u]", 
                     f+1, inet_ntoa( clientaddr ), 
                     ntohs( Fragments[ f ]->Port ) );          
            RioErr << "CClientHeap::GetNextFragment " << StrAux << endl;
        }
        #endif
    }    
    else
    {
       // O cliente esta inativo por muito tempo e precisa ser removido do
       // mapa. Como o cliente ja foi removido do heap, bastara remove-lo do 
       // mapa e deletar a estrutura Client. Isso sera feito pela funcao
       // RemoveClient da classe, que retornara true se o cliente pode ser
       // removido. O cliente nao podera ser removido se outra thread
       // estiver tentando inserir fragmentos na fila dele. Note que o valor
       // de InsertClient e a negacao do retorno da funcao porque somente
       // deveremos inserir o cliente no heap se ele nao pode ser removido.
       InsertClient = !RemoveClient( Client );  
    }
    // Obtem acesso exclusivo ao heap.
    pthread_mutex_lock( &m_AccessClientsHeap );
    // Se InsertClient for true, deveremos reinserir o cliente no heap.
    if( InsertClient ) 
        m_ClientsHeap.push( Client );
    // Libera o acesso exclusivo ao heap.
    pthread_mutex_unlock( &m_AccessClientsHeap );

    #ifdef RIO_DEBUG1
    RioErr << "[CClientHeap - GetFragments] Finish" << endl;
    #endif
}

///////////////////////////////////////////////////////////////////////////////
// Classe CStreamControl - Classe principal do controle de fluxo, responsavel//
// por processar os fragmentos deste que eles sao colocados no controle de   //
// fluxo pela objeto da classe RioNeti (usando a funcao Put), ate eles serem //
// enviados aos clientes (com a funcao nbSend do objeto da classe RioNeti).  //
/////////////////////////////////////////////////////////////////////////////// 

// Construtor da classe.
CStreamControl::CStreamControl()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - Construtor] Start" << endl;
    #endif

    // Inicializa as variaveis da classe.
    m_SendData = NULL;
    m_FPool = NULL;
    m_ClientHeap = NULL;
    m_SendThread = 0;
    m_PoolThread = 0;
    m_ClientsThread = 0;
    m_UpdateThread = 0;
    m_RioNeti = NULL;
    m_Started = false;

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - Construtor] Finish" << endl;
    #endif
}

// Destrutor da classe.
CStreamControl::~CStreamControl()
{
    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - Destrutor] Start" << endl;
    #endif

    Stop();

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - Destrutor] Finish" << endl;
    #endif
}

// Funcao que implementa a thread que processa os fragmentos do pool.
void *CStreamControl::PoolThread( void *Param )
{
    CStreamControl *psc;
    FragmentInfo *fip;
    int result;

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - PoolThread] Start" << endl;
    #endif

    psc = (CStreamControl *) Param;

    RioErr << "CStreamControl::PoolThread thread iniciada." << endl;

    RioErr << "POOLTHREADID " << syscall( SYS_gettid ) << endl;
   
    for( ; ; )
    
    {
        // Se o pool estiver vazio, entao espera na variavel de condicao 
        // da classe CFPool pela chegada de Fragmentos.
        psc->m_FPool->WaitObject();
        // Remove todos os objetos FragmentInfo do pool m_FPool.
        while( ( fip = psc->m_FPool->Remove() ) != NULL ) 
        {
            // Insere o objeto na lista do cliente correto, usando a
            // funcao Insert da classe CClientList. 
            while ( ( result = psc->m_ClientHeap->Insert( fip ) ) 
                      != S_OK ) 
            {
                RioErr << "StreamControl::PoolThread erro de alocacao "
                       << "de memoria ao inserir um fragmento na lista "
                       << "de um cliente da classe CClientHeap. Tentando " 
                       << "novamente." << endl;
            }
        } 
    }

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - PoolThread] Finish" << endl;
    #endif

    return NULL;
}

// Funcao que implementa a thread que processa os fragmentos das listas dos 
// clientes.
void *CStreamControl::ClientsThread( void *Param )
{
    CStreamControl *psc;
    int result;
    unsigned int f; 
    TFragments Fragments;

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - ClientsThread] Start" << endl;
    #endif

    psc = (CStreamControl *) Param;
    
    RioErr << "CStreamControl::ClientsThread thread iniciada." << endl;

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

    for( ; ; )
    {
        // Agora vamos obter o primeiro fragmento a ser copiado para a fila
        // FIFO de envio dos fragmentos. Se a funcao retornar NULL, entao
        // o cliente que estava no topo do heap (o cliente para o qual
        // atualizamos os creditos), nao possui fragmentos a serem enviados
        // (note que o cliente deveria possuir creditos, pois dormimos, no
        // procedimento GetFragments chamado a seguir, ate que esteja na 
        // hora de atualizar os creditos do cliente associado ao topo do
        // heap.
        psc->m_ClientHeap->GetFragments( Fragments );
        for( f = 0; f < Fragments.size(); f++ ) 
        {
            // Devemos inserir o objeto FragmentInfo no final da fila FIFO de
            // envio.
            while ( ( result = psc->m_SendData->Insert( Fragments[ f ] ) ) 
                             != S_OK ) 
            {
                RioErr << "StreamControl::ClientsThread erro de alocacao "
                       << "de memoria ao criar um objeto da classe "
                       << "CSendData. Tentando novamente." << endl;
            }
        }
    }

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - ClientsThread] Finish" << endl;
    #endif

    return NULL;
}

// Funcão que implementa a thread que envia os objetos Fragment usando a 
// CallBack.
void *CStreamControl::SendDataThread( void *Param )
{
    CStreamControl *psc;
    FragmentInfo *fip;
    int result;

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - SendDataThread] Start" << endl;
    #endif

    psc = (CStreamControl *) Param;
    
    RioErr << "CStreamControl::SendDataThread thread iniciada." << endl;

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

    for( ; ; )
    {
        // Tenta obter um objeto FragmentInfo. Se a fila FIFO da classe CSendData
        // estiver vazia, a thread sera bloqueada ate que um objeto seja
        // colocado na fila. A thread tambem sera bloqueada se nao existirem
        // creditos.
        fip = psc->m_SendData->Remove();
        if( fip != NULL ) 
        {  
            // Envia os dados do fragmento associado ao objeto FragmentInfo.
            result = psc->m_RioNeti->nbSend( NULL, fip->FragmentData, 
                                             fip->FragmentSize, fip->IP, 
                                             fip->Port);
            // Remove os dados do fragmento (eles sao uma copia) e a estrutura 
            // com estes dados
            #ifdef RIO_DEBUG2
            psc->TotalSendFragments++;
            RioErr << "CStreamControl::SendDataThread enviado o fragmento "
                   << "numero " << psc->TotalSendFragments << endl;
            #endif       
            delete fip->FragmentData;
            delete fip;
        } 
        else
           RioErr << "CStreamControl::SendDataThread erro - fip = NULL" 
                  << endl;                               
    }

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - SendDataThread] Finish" << endl;
    #endif

    return NULL;
}

// Funcao que implementa a thread que atualiza os creditos da classe CSendData.
void *CStreamControl::UpdateThread( void *Param )
{
    CStreamControl *psc;

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - UpdateThread] Start" << endl;
    #endif

    psc = (CStreamControl *) Param;
    
    RioErr << "CStreamControl::UpdateThread thread iniciada." << endl;

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

    for( ; ; )
    {
        psc->m_SendData->UpdateCredits();
    }

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - UpdateThread] Finish" << endl;
    #endif

    return NULL;
}

// Funcao para inicializar a classe. Os parametros da funcao sao:
//     rn : ponteiro para a classe RioNeti. 
//     MaxClientsCredits : Numero maximo de creditos para o envio de fragmentos 
//                         para um cliente.
//     MaxNetworkCredits : Numero maximo de creditos para o envio de fragmentos 
//                         para a rede (para todos os clientes).
//     TimeBetweenCredits: Tempo entre duas atualizacoes dos creditos da classe 
//                         CSendData.
int CStreamControl::Start( RioNeti *rn, unsigned int MaxClientsCredits,
                           unsigned int MaxNetworkCredits, 
                           unsigned long long int TimeBetweenCredits )
{
    // Atributos para as threads criadas pela classe.
    pthread_attr_t attribSendThread;
    pthread_attr_t attribPoolThread;
    pthread_attr_t attribClientsThread;
    pthread_attr_t attribUpdateThread;

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

    // Verifica se a classe ja foi inicializada.
    if( m_Started )
    {

        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Start] Finish1" << endl;
        #endif
        
        return ERROR_STREAMCONTROL + ERROR_STARTED; 
    }
    // Cria as classes usadas pela classe StreamControl.
    // Cria a classe com a fila FIFO com os objetos FragmentInfo a serem enviados
    // pela thread m_SendThread.
    // (corrigir os parametros 0, 0 para os valores corretos do numero 
    // maximo de creditos e do tempo da atualizacao entre os creditos.
    m_SendData = new CSendData( MaxNetworkCredits, TimeBetweenCredits );
    if( m_SendData == NULL )  
    {
        
        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Start] Finish2" << endl;
        #endif

        return ERROR_STREAMCONTROL + ERROR_MEMORY; 
    }
    // Cria o pool de objetos FragmentInfo que armazenara os objetos colocados
    // pela funcao Put e retirados pela thread m_ProcessThread.
    m_FPool = new CFPool;
    if( m_FPool == NULL ) 
    {
        
        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Start] Finish3" << endl;
        #endif

        delete m_SendData;
        return ERROR_STREAMCONTROL + ERROR_MEMORY; 
    }
    // Cria a lista com com os objetos FragmentInfo que devem ser enviados para
    // os clientes.
    m_ClientHeap = new CClientHeap( MaxClientsCredits );
    if ( m_ClientHeap == NULL )
    {
        
        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Start] Finish4" << endl;
        #endif

        delete m_SendData;
        delete m_FPool;
        return ERROR_STREAMCONTROL + ERROR_MEMORY; 
    }
    // Inicializa os atributos das threads usadas pela classe.
    pthread_attr_init( &attribSendThread );
    pthread_attr_init( &attribPoolThread );
    pthread_attr_init( &attribClientsThread );
    pthread_attr_init( &attribUpdateThread );
    pthread_attr_setstacksize( &attribSendThread, 2*PTHREAD_STACK_MIN );
    pthread_attr_setstacksize( &attribPoolThread,  2*PTHREAD_STACK_MIN );
    pthread_attr_setstacksize( &attribClientsThread,  2*PTHREAD_STACK_MIN );
    pthread_attr_setstacksize( &attribUpdateThread,  2*PTHREAD_STACK_MIN );
    // Cria as threads usadas pela classe.
    if ( pthread_create( &m_PoolThread, &attribPoolThread, PoolThread, this )   
                        != 0 )
    {
        
        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Start] Finish5" << endl;
        #endif

        delete m_SendData;
        delete m_FPool;
        delete m_ClientHeap;
        return ERROR_STREAMCONTROL + ERROR_CREATE_THREAD; 
    }
    if ( pthread_create( &m_ClientsThread, &attribClientsThread, ClientsThread, 
                         this ) != 0 )
    {
        
        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Start] Finish6" << endl;
        #endif

        delete m_SendData;
        delete m_FPool;
        delete m_ClientHeap;
        pthread_cancel( m_PoolThread );
        return ERROR_STREAMCONTROL + ERROR_CREATE_THREAD; 
    }
    if ( pthread_create( &m_SendThread, &attribSendThread, SendDataThread, 
                         this ) != 0 )
    {
        
        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Start] Finish7" << endl;
        #endif

        delete m_SendData;
        delete m_FPool;
        delete m_ClientHeap;
        pthread_cancel( m_PoolThread );
        pthread_cancel( m_ClientsThread );
        return ERROR_STREAMCONTROL + ERROR_CREATE_THREAD; 
    }
    if ( pthread_create( &m_UpdateThread, &attribUpdateThread, UpdateThread, 
                         this ) != 0 )
    {
        
        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Start] Finish8" << endl;
        #endif

        delete m_SendData;
        delete m_FPool;
        delete m_ClientHeap;
        pthread_cancel( m_PoolThread );
        pthread_cancel( m_ClientsThread );
        pthread_cancel( m_SendThread );
        return ERROR_STREAMCONTROL + ERROR_CREATE_THREAD; // Esta correto?
    }
    // Inicializa as variaveis internas.
    m_RioNeti = rn;
    m_Started = true;

    #ifdef RIO_DEBUG2
    // Acumula o numero de fragmentos enviados (usado para depuracao).
    TotalSendFragments = 0;
    #endif

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - Start] Finish9" << endl;
    #endif

    return S_OK; 
}

// Funcao para parar o processamento da classe iniciado por Start.
void CStreamControl::Stop() 
{
    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - Stop] Start" << endl;
    #endif

    if( m_SendThread != 0 )
        pthread_cancel( m_SendThread );    
    if( m_PoolThread != 0 )
        pthread_cancel( m_PoolThread );    
    if( m_ClientsThread != 0 )
        pthread_cancel( m_ClientsThread );    
    if( m_UpdateThread != 0 )
        pthread_cancel( m_UpdateThread );    
    if( m_SendData != NULL )
        delete m_SendData;
    if( m_FPool != NULL )
        delete m_FPool;
    if( m_ClientHeap != NULL )
        delete m_ClientHeap;
    // Inicializa as variaveis da classe.
    m_SendData = NULL;
    m_FPool = NULL;
    m_ClientHeap = NULL;
    m_SendThread = 0;
    m_PoolThread = 0;
    m_ClientsThread = 0;
    m_UpdateThread = 0;
    m_RioNeti = NULL;
    m_Started = false;

    #ifdef RIO_DEBUG2
    TotalSendFragments = 0;
    #endif

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - Stop] Finish" << endl;
    #endif
}

// Funcao para inserir um novo objeto Fragment para ser futuramente processado 
// na classe e colocado no LeackBucket associado ao cliente ao qual este objeto 
// se refenrecia.
int CStreamControl::Put( NetBuf *nbp )
{
    FragmentInfo *fip;

    #ifdef RIO_DEBUG1
    RioErr << "[CStreamControl - Put] Start" << endl;
    #endif

    // Adiciona um elemento Fragment no pool m_FPool para ser processado pela
    // thread ProcessThread.
    // Verifica se a classe foi inicializada, e insere um elemento no pool
    // somente se a classe foi inicializada.
    if( m_Started )
    {
        // Cria uma novo objeto FragmentInfo e copia os dados do fragmento
        // nele.
        fip = new FragmentInfo;
        if( fip == NULL )
        {

            #ifdef RIO_DEBUG1
            RioErr << "[CStreamControl - Put] Finish1" << endl;
            #endif

            return ERROR_STREAMCONTROL + ERROR_MEMORY;
        }
        // Obtem os dados do objeto NetBuf
        // Obtem uma copia do fragmento que deveria ter sido enviado.
        if( nbp->GetFragmentInfo( &fip->FragmentData, &fip->FragmentSize ) != 
            S_OK )
        {

            #ifdef RIO_DEBUG1
            RioErr << "[CStreamControl - Put] Finish2" << endl;
            #endif

            delete fip;
            return ERROR_STREAMCONTROL + ERROR_MEMORY;
        }    
        // Obtem o IP e a porta do cliente.
        nbp->GetIPAndPort( &fip->IP, &fip->Port );
        // Obtem o tempo da atualizacao entre os creditos.
        fip->TimeBetweenCredits = nbp->GetTimeBetweenCredits();

        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Put] Finish3" << endl;
        #endif

        return m_FPool->Insert( fip );
    }
    else
    { 

        #ifdef RIO_DEBUG1
        RioErr << "[CStreamControl - Put] Finish4" << endl;
        #endif

        return ERROR_STREAMCONTROL + ERROR_NOT_STARTED;
    }
}

---