RioStream Class Reference

#include <StreamManager.h>

Public Member Functions
	~RioStream ()
int	Close ()
void	Initialize (CStreamManager *)
int	MaxRequests ()
int	OpenObject (char *ObjectName, RioAccess Access, RioStreamObj **StreamObj)
int	Get_streamid ()
int	OpenObject (char *ObjectName, RioAccess Access, struct timeval RTT, int BufferSize, RioStreamObj **StreamObj)
RioTrafficType	GetType ()
uint	GetId ()
void	releaseMutexUpdateEvent ()
void	getMutexUpdateEvent ()
int	getDirection ()
void	updateNextBlockToBeFetched (PSEQBLOCKLIST Next)
void	prefetchBlock (int posAtBuffer, EventDataRequest *event, RioStreamObj *StreamObject)
int	getTokensToSend ()
void	incTokensToSend ()
void	decTokensToSend ()
void	setTokensToSend (int value)
void	decTokensToDisk ()
int	getTokensToDisk ()
void	updateTokensToDisk (struct timeval time)
u32	getMinimumBlockToSend ()
void	incMinimumBlockToSend ()
void	setMinimumBlockToSend (u32 Block)
u32	getMaximumBlockToSend ()
struct timeval	getLastArrivalTime ()
double	getRTTtoClient ()
int	getClientBufferStatus ()
int	getPlayBlock ()
int	getNumberOfBlocks ()
int	getServerBufferStatus ()
void	decServerBufferStatus ()
int	getNumberOfBuffersForEachClient ()
double	getAverageCACTime ()
double	getEstimatedDiskQueueTime (int disk)
double	getEstimatedDiskServiceTime (int disk)
double	getEstimatedDiskResponseTime (int disk)
long double	getNetworkRate ()
unsigned int	getBlockSize ()
int	getQueueSize ()
int	getClientBufferSize ()
void	cancelBlockAtBuffer (int positionAtBuffer)
bool	getPrefetchedAllBlocks ()
int	ClientCanStart ()
int	startingStream ()
void	NewBlockArrival ()
Data Fields
ofstream	m_log
ofstream	m_log_requests
Private Types
enum	StreamStatus { StreamStatusClosed = 0, StreamStatusOpened = 1 }
Private Member Functions
	RioStream ()
	RioStream (const RioStream &x)
RioStream &	operator= (const RioStream &x)
int	DataReq (u32 reqid, u16 operation, u32 ipaddr, u16 port, u32 block, u32 repbits, RioStreamObj *StreamObject)
void	RTHold (EventDataRequest *event)
void	NRTHold (EventDataRequest *event)
void	ProcessRTQueue ()
void	ProcessNRTQueue ()
void	RequestCompleted (Event *Request)
PSEQBLOCKLIST	getInfoAboutRequestList (char *filename, RioObject *object)
Private Attributes
RioStream *	Next
RioStream *	Previous
CStreamManager *	s_mgr
int	s_streamid
RioStreamDirection	s_Direction
double	s_Rate
SOCKADDR_IN	m_RemoteAddress
StreamStatus	m_Status
RioTrafficType	m_Type
int	m_nRequests
int	m_MaxRequests
int	m_UserMaxRequests
int	m_nQueue
struct timeval	m_NextRelease
struct timeval	m_ArrivalInterval
u64	m_RequestCount
u64	m_QueuedCount
CRequestQueue	m_Queue
struct timeval	m_LastArrivalTime
struct timeval	m_TokensToDiskClock
struct timeval	m_TimeToBuffer
double	m_FirstArrivalTime
double	m_SampleGenReqListTime
double	m_SampleSortListTime
double	m_SampleSimulationTime
double	m_SampleCACTime
double	m_SampleAdmissionProcessTime
double	m_SampleWaitingTime
int	m_WaitingQueueSize
int	m_TotalBlocks
PSEQBLOCKLIST	m_RequestList
uint	m_Id
PSTREAMLIST	m_PosAtClientsList
ClientBuffer *	m_Buffer
double	m_AverageRTTtoClient_msec
int	m_ClientBufferSize
int	m_ServerBufferStatus
int	m_TokensToSend
bool	m_PrefetchedAllBlocks
bool	m_FinishedMovie
int	m_TokensToDisk
bool	m_StartingToSend
u32	m_MinimumBlockToSend
pthread_mutex_t	MutexTokensToSend
pthread_mutex_t	MutexTokensToDisk
pthread_mutex_t	MutexNextBlockToBeFetched
pthread_mutex_t	MutexMinBlock
pthread_mutex_t	MutexBlock
pthread_mutex_t	MutexUpdateEvent
pthread_cond_t	ConditionBlockArrival
Friends
class	CStreamManager
class	RioStreamObj

---

Detailed Description

Definition at line 143 of file server/StreamManager.h.

---

Member Enumeration Documentation

enum RioStream::StreamStatus [private]

Enumerator:

StreamStatusClosed
StreamStatusOpened

Definition at line 237 of file server/StreamManager.h.

        {
            StreamStatusClosed  = 0, // closed or closing
            StreamStatusOpened  = 1  // Opened and active
        } StreamStatus;

---

Constructor & Destructor Documentation

RioStream::RioStream

(

)

[private]

Definition at line 961 of file server/StreamManager.cpp.

{
    s_mgr             = NULL;
    s_streamid        = 0;
    s_Direction       = RioStreamDirectionRead;
    s_Rate            = 0.0;
    m_Status          = StreamStatusClosed;
    m_Type            = RIO_TRAFFIC_INVALID;
    m_nRequests       = 0;
    m_MaxRequests     = 0;
    m_UserMaxRequests = 0;
    m_nQueue          = 0;
    memset( (void *)&m_NextRelease,     0, sizeof( m_NextRelease )     );
    memset( (void *)&m_ArrivalInterval, 0, sizeof( m_ArrivalInterval ) );
    m_RequestCount    = 0;
    m_QueuedCount     = 0;
    // m_Queue does not need to be initialized
    memset( (void *)&m_LastArrivalTime  , 0, sizeof( m_LastArrivalTime ) );
    memset( (void *)&m_TokensToDiskClock, 0, sizeof( m_TokensToDiskClock ) );
    memset( (void *)&m_TimeToBuffer     , 0, sizeof( m_TimeToBuffer ) );

    m_FirstArrivalTime           = 0.0;
    m_SampleGenReqListTime       = 0.0;
    m_SampleSortListTime         = 0.0;
    m_SampleSimulationTime       = 0.0;
    m_SampleCACTime              = 0.0;
    m_SampleAdmissionProcessTime = 0.0;
    m_SampleWaitingTime          = 0.0;
    m_WaitingQueueSize           = 0;
    m_TotalBlocks                = 0;
    m_RequestList                = NULL;
    m_Id                         = 0;
    m_PosAtClientsList           = NULL;
    m_Buffer                     = NULL;
    m_AverageRTTtoClient_msec    = 0.0;
    m_ClientBufferSize           = 0;
    m_ServerBufferStatus         = 0;
    m_TokensToSend               = 0;
    m_PrefetchedAllBlocks        = false;
    m_FinishedMovie              = false;
    m_TokensToDisk               = 0;
    m_StartingToSend             = true;
    m_MinimumBlockToSend         = 0;
    pthread_mutex_init( &MutexTokensToSend        , NULL );
    pthread_mutex_init( &MutexTokensToDisk        , NULL );
    pthread_mutex_init( &MutexNextBlockToBeFetched, NULL );
    pthread_mutex_init( &MutexMinBlock            , NULL );
    pthread_mutex_init( &MutexBlock               , NULL );
    pthread_mutex_init( &MutexUpdateEvent         , NULL );
    pthread_cond_init(  &ConditionBlockArrival    , NULL );

    Next     = NULL;
    Previous = NULL;
    // m_log          does not need to be initialized
    // m_log_requests does not need to be initialized
}

RioStream::RioStream

(

const RioStream &

x

)

[private]

RioStream::~RioStream

(

)

Definition at line 1019 of file server/StreamManager.cpp.

{
    Close();
}

---

Member Function Documentation

void RioStream::cancelBlockAtBuffer

(

int

positionAtBuffer

)

Definition at line 2496 of file server/StreamManager.cpp.

{
    getMutexUpdateEvent();

    if( m_Buffer[positionAtBuffer].event != NULL )
    {
        m_Buffer[positionAtBuffer].event->Request.Operation=RealTimeCancelBlock;
        if( m_Buffer[positionAtBuffer].event->Request.BufferId != -1 )
        {
            releaseMutexUpdateEvent();
            decServerBufferStatus();
            //if it was served by Disk
            s_mgr->m_Router->Put( (Event*)m_Buffer[positionAtBuffer].event );
            #ifdef RIO_DEBUG2
            if( m_log.is_open() ) m_log << "cancelBlockAtBuffer: block "
                << m_Buffer[positionAtBuffer].blockInfo->block
                <<" canceled!"<< endl;
            #endif
        }
        else
        {
            releaseMutexUpdateEvent();
            #ifdef RIO_DEBUG2
            if( m_log.is_open() )
                m_log << "cancelBlockAtBuffer: setting cancel flag of block "
                      << m_Buffer[positionAtBuffer].blockInfo->block << endl;
            #endif
        }
        m_Buffer[positionAtBuffer].event = NULL;
    }
    else
    {
        releaseMutexUpdateEvent();
    }
}

int RioStream::ClientCanStart

(

)

Definition at line 2569 of file server/StreamManager.cpp.

{
    struct timeval final_time;
    double time;
    // After the client sent the requests, he asks to continue
    // but only after the server side buffer is full he can start visualization
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << " Client asked 'can I start'? " << endl;
    #endif
    if( m_PosAtClientsList != NULL )
    {
        while(getServerBufferStatus()<s_mgr->GetNumberOfBuffersForEachClient())
        {
            pthread_mutex_lock( &MutexBlock );
            pthread_cond_wait( &ConditionBlockArrival, &MutexBlock );
            pthread_mutex_unlock( &MutexBlock );
        }
        final_time = Timer.CurrentTime();
        time = getInterval( m_TimeToBuffer, final_time );
        s_mgr->SetAverageBufferTime( time, this );
        m_TimeToBuffer.tv_sec  =(u64)(( time * 1000.0 ) / 1000000 );
        m_TimeToBuffer.tv_usec =(u64)((u64) ( time * 1000.0 ) % 1000000 );
        #ifdef RIO_DEBUG2
        if( m_log.is_open() ) m_log << "Time to buffer " << time << endl;
        #endif

        // after this point, the client will be controlled by tokenstodisk
        m_StartingToSend = false;
        //initializes the tokenstodisk
        m_TokensToDisk = s_mgr->GetBurstSizeOfEachClient();

        //initializes the clock to be used to accumulate tokenstodisk
        struct timeval clocknow = Timer.CurrentTime();
        m_TokensToDiskClock.tv_sec  = clocknow.tv_sec;
        m_TokensToDiskClock.tv_usec = clocknow.tv_usec;
    }
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << " Yes. You can start now. m_TokensToDisk : "
          << m_TokensToDisk << endl;
    #endif
    return S_OK;
}

int RioStream::Close

(

void

)

Definition at line 1031 of file server/StreamManager.cpp.

{
    int i = 0;

    // Close any opened object on this stream
    // ###    m_ObjectManager->CloseStream(Stream);

    // just return if already closed
    if( m_Status == StreamStatusClosed )
        return S_OK;

    s_mgr->m_MutexTable->Wait();

    m_Status = StreamStatusClosed;

    if( s_mgr->m_log.is_open() )
        s_mgr->m_log << "Closing Stream ( " << m_Id << " )"
        << " Requests "      << m_nRequests
        << " Queue "         << m_nQueue
        << " Request Count " << m_RequestCount
        << " Queued Count "  << m_QueuedCount << endl;

    EventDataRequest* request;

    // empty stream queue
    while( m_nQueue > 0 )
    {
        request = m_Queue.Get();
        m_nQueue--;
        m_nRequests--;
        EventManager.Free( ( Event* )request );
    }

    if( s_mgr->m_RTHoldList.First() == this )
        s_mgr->m_RTHoldList.Remove( s_mgr->m_RTHoldList.First() );

    if( ( m_PosAtClientsList != NULL ) && ( s_mgr->m_UseServerSideBuffers ) )
    {
        // need to cancel blocks at Storage
        // for each position at buffer, cancel block
        for( i = 0 ; i < s_mgr->GetNumberOfBuffersForEachClient(); i++ )
        {
            cancelBlockAtBuffer ( i );
        }
    }
    // ------------------------------------------------------------------------

    // If there are pending requests should wait request completion
    // before finishing close
    if( m_nRequests > 0 )
    {
        if( s_mgr->m_log.is_open() )
            s_mgr->m_log << "Should wait for request completion before closing: "
                         << m_nRequests << " requests pending" << endl;
        s_mgr->m_CloseList.Put( this );
    }
    else
    {
        // If no pending request finish closing
        s_mgr->m_used--;
        // specific for real-time streams
        if( m_Type == RIO_TRAFFIC_CBR )
        {
            s_mgr->m_UsedRate -= s_Rate;
        }

        if( ( m_PosAtClientsList != NULL ) &&
            ( s_mgr->m_UseServerSideBuffers )
          )
        {
            s_mgr->RemoveClient( m_PosAtClientsList );
            if( m_Buffer )
            {
                delete m_Buffer;
                m_Buffer = NULL;
            }
        }
        #ifdef RIO_DEBUG2
        RioErr << "Closing mlog." << endl;
        m_log.close();
        #endif
        #ifdef RIO_DEBUG2
        RioErr << "Closing mlogreq." << endl;
        m_log_requests.close();
        #endif
        // --------------------------------------------------------------------

        s_mgr->m_FreeList.Put( this );
        if( s_mgr->m_log.is_open() )
            s_mgr->m_log << "Close: Stream CLOSED. Requests: " << m_RequestCount
                         << ". Queued requests: "       << m_QueuedCount
                         << ". Active streams "         << s_mgr->m_used<< endl;
    }

    s_mgr->m_MutexTable->Release();

    struct in_addr clientip;
    clientip.s_addr = m_RemoteAddress.sin_addr.s_addr;
    UpdateQueueEvent *it_event = new UpdateQueueEvent( inet_ntoa(clientip),
            htons( m_RemoteAddress.sin_port ), m_nRequests );
    s_mgr->PostITEvent( (MonitorEvent *)it_event );

    return S_OK;
}

int RioStream::DataReq	(	u32	reqid,
		u16	operation,
		u32	ipaddr,
		u16	port,
		u32	block,
		u32	repbits,
		RioStreamObj *	StreamObject
	)			[private]

Definition at line 1215 of file server/StreamManager.cpp.

{
    // Set Arrival time with current time
    struct timeval ArrivalTime = Timer.CurrentTime();

    #ifdef RIO_DEBUG2
    if( m_log.is_open() )
    {
        struct in_addr clientip;
        double interval = getInterval( m_LastArrivalTime, ArrivalTime );
        clientip.s_addr = ipaddr;
        char *info = myInfo( false );
        m_log << endl << info
              << "Stream:DataReq => Req ip " << inet_ntoa(clientip)
              << " port "  << port   << " reqid "        << reqid
              << " block " << block  << " TokensToSend " << m_TokensToSend
              << " (m_nRequests " << m_nRequests + 1 << ")"
              << " arrival " << ArrivalTime.tv_sec  << " sec "
              << ArrivalTime.tv_usec << " usec "
              << "interval " << interval << " msec"
              << " m_MaxRequests " << m_MaxRequests << endl;
        free( info );
    }
    #endif

    // Do not accept request if Stream is closed
    if( m_Status == StreamStatusClosed )
    {
        #ifdef RIO_DEBUG2
        if( m_log.is_open() )
            m_log << "Stream:DataReq:ERROR_STREAM_NOT_OPENED!!!" << endl;
        #endif

        return ERROR_STREAMMANAGER + ERROR_STREAM_NOT_OPENED;
    }

    // Check if number of stream active requests does not exceed maximum
    if( m_nRequests >= m_MaxRequests )
    {
        #ifdef RIO_DEBUG2
        if( m_log.is_open() )
            m_log << "Stream:DataReq:Max Number of Active Requests!!! "
                  << m_nRequests << ">=" << m_MaxRequests << endl;
        #endif

        RioErr << "Stream:DataReq:Max Number of Active Requests!!! "
               << m_nRequests << ">=" << m_MaxRequests << endl;

        return ERROR_STREAMMANAGER + ERROR_MAX_STREAM_REQUESTS;
    }
    // Increment number of stream active requests
    m_nRequests++;
    m_RequestCount++;

    // Get Request slot for new request
    EventDataRequest* event;

    if( m_Type == RIO_TRAFFIC_NRT )
    {
        event = (EventDataRequest*)EventManager.New( EventTypeNRTDataRequest );
    }
    else
    {
        event = (EventDataRequest*)EventManager.New( EventTypeRTDataRequest );
    }

    if( operation == READ )
    {
        if( m_Type == RIO_TRAFFIC_NRT )
        {
            event->Request.Operation = NonRealTimeRead;
        }
        else
        {
            event->Request.Operation = RealTimeRead;
        }
    }
    else
    {
        event->Request.Operation = NonRealTimeWrite;
    }

    event->Request.streamobj        = StreamObject;
    event->Request.rioobject        = StreamObject->o_object;
    event->Request.Reqid            = reqid;
    event->Request.Target.IPaddress = ipaddr;
    event->Request.Target.Port      = port;
    event->Request.Block            = block;
    event->Request.RepBits          = repbits;
    event->Request.Status           = 0;  // ok so far
    event->Request.BufferId         = -2; // identifies normal use

    // Process request differently depending if it is real-time or not
    if( m_Type == RIO_TRAFFIC_NRT )
    {
        // Should hold request if there are others being hold or
        // if reached maximum number of pending non real-time requests
        if( s_mgr->m_HoldingNonRealTime ||
          ( s_mgr->m_ActiveRequests >= s_mgr->m_MaxActiveRequests ))
        {
            // Put request on hold until time out processing decides
            // it is time to proceess it
            NRTHold( event );
        }
        // Otherwise send request direct to router
        else
        {
            // Increment total number of requests sent to router
            s_mgr->m_ActiveRequests++;

            // Send request to router
            #ifdef RIO_DEBUG2
            if( m_log.is_open() )
                m_log << "Stream:DataReq:Enviando pedido de NRT ao router"
                      << endl;
            #endif
            s_mgr->m_Router->Put( (Event*)event );
        }
    }
    // Real-time case
    else
    {
        // Check if the object was submited to new cac or
        // if it is using server side buffers.
        if( ( m_PosAtClientsList != NULL ) &&
            ( s_mgr->m_UseServerSideBuffers )
          )
        {
            m_LastArrivalTime = ArrivalTime;

            bool    changeSequence   = false;
            bool    blockWasAtBuffer = false;
            bool    usedEvent        = false;

            incTokensToSend();                  // available buffer at client
            updateTokensToDisk( ArrivalTime );  // available requests to Disk

            #ifdef RIO_DEBUG2
            if( m_log.is_open() )
                m_log << " Looking for block at buffer ... ( tokenstodisk "
                      << getTokensToDisk() << ")"<< endl;
            #endif


            // verifies if the block is at the buffer and send
            for( int i = 0; i < s_mgr->GetNumberOfBuffersForEachClient(); i++ )
            {
                if( ( m_Buffer[i].event != NULL ) &&
                   ( (u32)m_Buffer[i].event->Request.Block == block ))
                {
                    #ifdef RIO_DEBUG2
                    if( m_log.is_open() )
                        m_log << " Found it with BufferId "
                              << m_Buffer[i].event->Request.BufferId
                              << " updates ip, port and reqid... " << endl;
                    #endif

                    blockWasAtBuffer = true;

                    // because Router could be using to update storage reply
                    getMutexUpdateEvent();

                    // updates information to send
                    m_Buffer[i].event->Request.Reqid            = reqid;
                    m_Buffer[i].event->Request.Target.IPaddress = ipaddr;
                    m_Buffer[i].event->Request.Target.Port      = port;

                    incMinimumBlockToSend();

                    if( m_Buffer[i].event->Request.BufferId == -1 )
                    {
                        #ifdef RIO_DEBUG2
                        //if it was not served by Disk, just wait
                        if( m_log.is_open() )
                            m_log << " prefetching and waiting "<< endl;
                        #endif
                        releaseMutexUpdateEvent();
                    }
                    else if( m_Buffer[i].event->Request.Status != 0 )
                    {
                        //if error at prefetch, cancel the block
                        m_Buffer[i].event->Request.Operation = 
                                                        RealTimeCancelBlock;
                        releaseMutexUpdateEvent();
                        s_mgr->m_Router->Put( (Event*)m_Buffer[i].event );
                        decTokensToSend(); //can not send block
                        decServerBufferStatus();

                        #ifdef RIO_DEBUG2
                        if( m_log.is_open() )
                            m_log << " sent to Router -> error at prefetch, "
                                  << " cancel block" << endl;
                        #endif
                    }
                    else
                    {
                        //if it was prefetched, just send the block
                        m_Buffer[i].event->Request.Operation=RealTimeSendBlock;
                        releaseMutexUpdateEvent();
                        s_mgr->m_Router->Put( (Event*)m_Buffer[i].event );
                        #ifdef RIO_DEBUG2
                        if( m_log.is_open() )
                            m_log << " sent to Router "<< endl;
                        #endif
                    }

                    m_Buffer[i].event = NULL;
                }
            }

            // updates next block to be requested
            if( block != m_PosAtClientsList->nextBlockToBeRequested->block )
            {
                //the sequence was changed, have to arrange everything
                changeSequence = true;

                #ifdef RIO_DEBUG2
                if( m_log.is_open() )
                    m_log << " sequence changed from "
                          << m_PosAtClientsList->nextBlockToBeRequested->block
                          << " to " << block << endl;
                #endif

                while( ( block >
                         m_PosAtClientsList->nextBlockToBeRequested->block ) &&
                       ( m_PosAtClientsList->nextBlockToBeRequested->next !=
                         NULL )
                     )
                    m_PosAtClientsList->nextBlockToBeRequested =
                               m_PosAtClientsList->nextBlockToBeRequested->next;

                while( ( block <
                         m_PosAtClientsList->nextBlockToBeRequested->block ) &&
                       ( m_PosAtClientsList->nextBlockToBeRequested->prev !=
                         NULL )
                     )
                    m_PosAtClientsList->nextBlockToBeRequested =
                               m_PosAtClientsList->nextBlockToBeRequested->prev;

                #ifdef RIO_DEBUG2
                if( m_log.is_open() )
                    m_log << " nbtoberequested "
                          << m_PosAtClientsList->nextBlockToBeRequested->block
                          << endl;
                #endif

                if( blockWasAtBuffer == false )
                {
                    #ifdef RIO_DEBUG2
                    if( m_log.is_open() )
                        m_log << " block was not at buffer. "<< endl;
                    #endif

                    //if block was not buffered, send the block
                    usedEvent = true;
                    decTokensToSend();
                    if( ( m_nQueue == 0 ) && ( getTokensToDisk() > 0 ) )
                    {
                        #ifdef RIO_DEBUG2
                        if( m_log.is_open() )
                            m_log << " (to Router) send block "
                                  << event->Request.Block
                                  << " bufferid "
                                  << event->Request.BufferId
                                  << " m_ActiveRequests "
                                  << s_mgr->m_ActiveRequests + 1
                                  << endl;
                        #endif

                        s_mgr->m_Router->Put( (Event*)event );
                        s_mgr->m_ActiveRequests++;
                        decTokensToDisk();
                    }
                    else
                    {
                        #ifdef RIO_DEBUG2
                        if( m_log.is_open() )
                            m_log << " (to queue) send block "
                                  << event->Request.Block
                                  << " bufferid "
                                  << event->Request.BufferId << endl;
                        #endif

                        RTHold( event );
                    }
                }
            }

            if( m_PosAtClientsList->nextBlockToBeRequested->next != NULL )
            {
                m_PosAtClientsList->nextBlockToBeRequested =
                         m_PosAtClientsList->nextBlockToBeRequested->next;
            }
            else
            {
                // Doesn't have any more requests
                #ifdef RIO_DEBUG2
                if( m_log.is_open() )
                    m_log << " last block - prefetch all true "<< endl;
                #endif

                m_FinishedMovie = true;
                m_PrefetchedAllBlocks = true;

                if( s_mgr->GetNumberOfBuffersForEachClient() != 0 )
                {
                    EventManager.Free( (Event *) event );
                }
            }

            #ifdef RIO_DEBUG2
            if( m_log.is_open() ) m_log << " updated nbtoberequested "
                  << m_PosAtClientsList->nextBlockToBeRequested->block
                  << endl;
            #endif

            if( m_PrefetchedAllBlocks == true )
            {
                m_nRequests--;
            }

            // updates next block to be fetched
            if( s_mgr->GetNumberOfBuffersForEachClient() == 0 )
            {
                updateNextBlockToBeFetched(
                                   m_PosAtClientsList->nextBlockToBeRequested);
            }
            else
            {
                if( changeSequence && ( m_FinishedMovie == false ) )
                {
                    #ifdef RIO_DEBUG2
                    if( m_log.is_open() )
                        m_log << " updating nbtobefetched ... ";
                    #endif

                    if( ( block >=
                          m_PosAtClientsList->nextBlockToBeFetched->block ) ||
                        ( ( block + s_mgr->GetNumberOfBuffersForEachClient()) <
                          ( m_PosAtClientsList->nextBlockToBeFetched->block -
                            s_mgr->GetNumberOfBuffersForEachClient()
                          )
                        )
                      )
                    {
                        //for each position at buffer, cancel block
                        for( int i=0;
                             i < s_mgr->GetNumberOfBuffersForEachClient();
                             i++
                           )
                        {
                            cancelBlockAtBuffer ( i );
                            m_nRequests++;
                        }
                        updateNextBlockToBeFetched(
                                  m_PosAtClientsList->nextBlockToBeRequested );
                    }
                    else
                    if( ( block + s_mgr->GetNumberOfBuffersForEachClient() ) <
                              m_PosAtClientsList->nextBlockToBeFetched->block )
                    {
                        updateNextBlockToBeFetched(
                             m_PosAtClientsList->nextBlockToBeRequested );
                    }
                    setMinimumBlockToSend(
                           m_PosAtClientsList->nextBlockToBeRequested->block );
                    if( m_PosAtClientsList->nextBlockToBeFetched->next == NULL )
                    {
                        m_PrefetchedAllBlocks = true;
                    }
                    else
                    {
                        m_PrefetchedAllBlocks = false;
                    }
                }
            }

            // prefetch blocks

            if( ( s_mgr->GetNumberOfBuffersForEachClient() == 0 ) &&
               ( changeSequence == false ))
            {
                decTokensToSend();
                if( m_StartingToSend == true )
                {
                    #ifdef RIO_DEBUG2
                    if( m_log.is_open() )
                        m_log << "Stream:DataReq:Sending prefetching blocks to"
                              << " router"<< endl;
                    #endif

                    s_mgr->m_Router->Put( (Event*)event );
                    s_mgr->m_ActiveRequests++;
                }
                else
                {
                    if( ( m_nQueue == 0 ) && ( getTokensToDisk() > 0 ) )
                    {
                        #ifdef RIO_DEBUG2
                        if( m_log.is_open() )
                            m_log << "Stream:DataReq:Sending prefetching blocks "
                                  << "to router 2"<< endl;
                        #endif

                        s_mgr->m_Router->Put( (Event*)event );
                        s_mgr->m_ActiveRequests++;
                        decTokensToDisk();
                    }
                    else
                    {
                        RTHold( event );
                    }
                }
            }
            int i   = 0;
            int pos = -1;
            bool prefetch;

            while(( i < s_mgr->GetNumberOfBuffersForEachClient()) &&
                  ( m_PrefetchedAllBlocks == false ))
            {
                prefetch = false;
                if( m_Buffer[i].event == NULL )
                {
                    prefetch = true;
                }
                else
                if( (( m_Buffer[i].blockInfo->block < getMinimumBlockToSend() )
                     ||( m_Buffer[i].blockInfo->block >=
                       ( m_PosAtClientsList->nextBlockToBeRequested->block
                         + s_mgr->GetNumberOfBuffersForEachClient())))
                   && changeSequence )
                {
                    cancelBlockAtBuffer( i );
                    m_nRequests++;
                    prefetch = true;
                }
                else if( ( m_Buffer[i].blockInfo->block >=
                          m_PosAtClientsList->nextBlockToBeFetched->block )
                        && ( changeSequence ))
                {
                    if( pos == -1 )
                    {
                        pos = i;
                    }
                    else
                    {
                        if( m_Buffer[pos].blockInfo->block <
                            m_Buffer[i].blockInfo->block )
                            pos = i;
                    }
                }
                if( prefetch == true )
                {
                    if( usedEvent )
                    {
                        event = (EventDataRequest*)
                                EventManager.New( EventTypeRTDataRequest );
                    }
                    else
                        usedEvent = true;

                    prefetchBlock( i, event, StreamObject );
                    updateNextBlockToBeFetched( NULL );
                }
                i++;
            }

            if( ( pos != -1 ) &&
               ( m_Buffer[pos].blockInfo->block >=
                        m_PosAtClientsList->nextBlockToBeFetched->block ))
            {
                 updateNextBlockToBeFetched( m_Buffer[pos].blockInfo->next );
            }

            if( usedEvent == false )
            {
                //have to free (it was not used)
                EventManager.Free( (Event *) event );
            }
            // ----------------------------------------------------------------
        }
        else
        {

            ///////////////////////////////////////////////////////////////////
            // Check if request satisfy rate requirement and there are no other
            // request queued

            struct timeval tolerance_time;
            tolerance_time.tv_sec =ArrivalTime.tv_sec;
            tolerance_time.tv_usec=ArrivalTime.tv_usec+RELEASE_TIME_TOLERANCE;
            if( tolerance_time.tv_usec > 1000000 )
            {
                tolerance_time.tv_sec  += 1;
                tolerance_time.tv_usec -= 1000000;
            }

            #ifdef RIO_DEBUG2
            if( m_log.is_open() ) m_log << "Stream:DataReq "
                  << " m_nQueue " << m_nQueue << endl
                  << " tolerance_time.tv_sec "
                  << ( unsigned ) tolerance_time.tv_sec
                  << " m_NextRelease.tv_sec "
                  << ( unsigned ) m_NextRelease.tv_sec
                  << endl
                  << " tolerance_time.tv_usec "
                  << ( unsigned ) tolerance_time.tv_usec
                  << " m_NextRelease.tv_usec "
                  << ( unsigned ) m_NextRelease.tv_usec
                  << endl;
            #endif

            //This part makes the control of client's requests queue
            //Now, server only sends requests to router
            if( ( m_nQueue == 0 ) &&
               (( tolerance_time.tv_sec > m_NextRelease.tv_sec ) ||
               (( tolerance_time.tv_sec == m_NextRelease.tv_sec ) &&
                ( tolerance_time.tv_usec >= m_NextRelease.tv_usec ))))
            {
                // Yes: Send request directly to Router

                // First update new release time for next request
                if( ( ArrivalTime.tv_sec > m_NextRelease.tv_sec ) ||
                  (( ArrivalTime.tv_sec == m_NextRelease.tv_sec) &&
                    (ArrivalTime.tv_usec > m_NextRelease.tv_usec)) )
                {
                    m_NextRelease.tv_sec  = ArrivalTime.tv_sec  +
                                            m_ArrivalInterval.tv_sec;
                    m_NextRelease.tv_usec = ArrivalTime.tv_usec +
                                            m_ArrivalInterval.tv_usec;
                }
                else
                {
                    m_NextRelease.tv_sec  += m_ArrivalInterval.tv_sec;
                    m_NextRelease.tv_usec += m_ArrivalInterval.tv_usec;
                }
                if( m_NextRelease.tv_usec > 1000000 )
                {
                    m_NextRelease.tv_sec  += 1;
                    m_NextRelease.tv_usec -= 1000000;
                }
            // ----------------------------------------------------------------

                // Increment total number of requests sent to router
                s_mgr->m_ActiveRequests++;

                // Send request to router
                #ifdef RIO_DEBUG2
                if( m_log.is_open() )
                    m_log << "Stream:DataReq:Enviando pedido de bloco ao router"
                          << endl;
                #endif
                s_mgr->m_Router->Put( (Event*)event );
            }
            // Rate violation: Hold request
            else
            {
                // Put request on hold until time out processing decides
                // it is time to process it
                #ifdef RIO_DEBUG2
                if( m_log.is_open() )
                    m_log << "Stream:DataReq:Putting request on hold" << endl;
                #endif

                RTHold( event );
            }

            m_LastArrivalTime = ArrivalTime;
        }
    }

    struct in_addr clientip;
    clientip.s_addr = m_RemoteAddress.sin_addr.s_addr;
    UpdateQueueEvent *it_event = new UpdateQueueEvent( inet_ntoa(clientip),
            htons( m_RemoteAddress.sin_port ), m_nRequests );
    s_mgr->PostITEvent( (MonitorEvent *)it_event );

    return S_OK;
}

void RioStream::decServerBufferStatus

(

)

Definition at line 2720 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexBlock );
    m_ServerBufferStatus--;
    pthread_mutex_unlock( &MutexBlock );
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "BufferStatus decremented : " << m_ServerBufferStatus << endl;
    #endif
}

void RioStream::decTokensToDisk

(

)

Definition at line 2644 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexTokensToDisk );
    if( m_TokensToDisk > 0 )
    {
        m_TokensToDisk--;
    }
    pthread_mutex_unlock( &MutexTokensToDisk );
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "TokensToDisk decremented : " << m_TokensToDisk << endl;
    #endif
}

void RioStream::decTokensToSend

(

)

Definition at line 2821 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexTokensToSend );
    m_TokensToSend--;
    pthread_mutex_unlock( &MutexTokensToSend );
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "TokensToSend decremented : " << m_TokensToSend << endl;
    #endif
}

int RioStream::Get_streamid

(

)

Definition at line 1208 of file server/StreamManager.cpp.

{
    return s_streamid;
}

double RioStream::getAverageCACTime

(

)

Definition at line 2737 of file server/StreamManager.cpp.

{
    return s_mgr->GetAverageCACTime();
}

unsigned int RioStream::getBlockSize

(

)

Definition at line 2758 of file server/StreamManager.cpp.

{
    return s_mgr->GetBlockSize();
}

int RioStream::getClientBufferSize

(

)

Definition at line 2699 of file server/StreamManager.cpp.

{
    return m_ClientBufferSize;
}

int RioStream::getClientBufferStatus

(

)

Definition at line 2669 of file server/StreamManager.cpp.

{
    if( m_StartingToSend )
    {
        return 0;
    }
    else
    {
        return m_ClientBufferSize - m_TokensToSend;
    }
}

int RioStream::getDirection

(

)

Definition at line 1136 of file server/StreamManager.cpp.

{
    return s_Direction;
}

double RioStream::getEstimatedDiskQueueTime

(

int

disk

)

Definition at line 2753 of file server/StreamManager.cpp.

{
    return s_mgr->GetEstimatedDiskQueueTime( disk );
}

double RioStream::getEstimatedDiskResponseTime

(

int

disk

)

Definition at line 2748 of file server/StreamManager.cpp.

{
    return s_mgr->GetEstimatedDiskResponseTime( disk );
}

double RioStream::getEstimatedDiskServiceTime

(

int

disk

)

Definition at line 2743 of file server/StreamManager.cpp.

{
    return s_mgr->GetEstimatedDiskServiceTime( disk );
}

uint RioStream::GetId

(

)

Definition at line 1202 of file server/StreamManager.cpp.

{
    return m_Id;
}

PSEQBLOCKLIST RioStream::getInfoAboutRequestList	(	char *	filename,
		RioObject *	object
	)			[private]

Definition at line 2889 of file server/StreamManager.cpp.

{
    FILE *file;
    PSEQBLOCKLIST curblock, prevblock, init_stream;
    struct timeval initial_time, final_time;

    float       duration = 0;
    uint        block    = 0;
    int         rc       = 0;
    int         numreps;
    long double time = 0; // event time

    RioDiskBlock Reps[MaxReplications];

    initial_time = Timer.CurrentTime();

    m_FirstArrivalTime = initial_time.tv_sec  * 1000.0 +
                         initial_time.tv_usec / 1000.0;

    uint SBufferSize  = getNumberOfBuffersForEachClient();
    uint CBufferSize  = m_ClientBufferSize;

    uint buffersize   = CBufferSize + SBufferSize;

    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log <<"---------------------------------------------------------------- "<<endl;
    if( m_log.is_open() ) m_log <<"                      getInfoAboutRequestList                    "<<endl;
    if( m_log.is_open() ) m_log <<"---------------------------------------------------------------- "<<endl;
    if( m_log.is_open() ) m_log << "Getting disk service time info ... ";
    #endif

    pthread_mutex_lock( &s_mgr->MutexUpdateMeasures );
    s_mgr->GetDiskServiceTime();
    s_mgr->UpdateDiskQueueAndServiceTime();
    pthread_mutex_unlock( &s_mgr->MutexUpdateMeasures );

    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "OK." << endl;
    #endif

    double msecRunCAC  = getAverageCACTime();
    double NetworkRate = getNetworkRate();

    unsigned int BlockSize = getBlockSize();

    double timeToTransmit_msec = ( BlockSize/NetworkRate ) * 1000.0;

    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "Client ID   " << m_Id << endl;
    if( m_log.is_open() ) m_log << "SBufferSize " << SBufferSize << endl;
    if( m_log.is_open() ) m_log << "CBufferSize " << CBufferSize << endl;
    if( m_log.is_open() ) m_log << "RTT (msec)  " << m_AverageRTTtoClient_msec << endl;
    if( m_log.is_open() ) m_log << "msecRunCAC  " << msecRunCAC << endl;
    if( m_log.is_open() ) m_log << "NetworkRate " << NetworkRate << "B/sec" << endl;
    if( m_log.is_open() ) m_log << "BlockSize   " << BlockSize << "Bytes" << endl;
    if( m_log.is_open() ) m_log << "timeToTransmit_msec " << timeToTransmit_msec << " msec" << endl;
    #endif

    // Open the file with times of the requests
    if( (file = fopen( filename, "r" )) == NULL )
    {
        if( m_log.is_open() )
            m_log<<"getInfoAboutRequestList: "<<filename<<": File not found."<<endl;
        return NULL;
    }

    // Start to construct request list
    init_stream = prevblock = NULL;

    float *req_time   = (float *) malloc( buffersize * sizeof(float) );
    float *disk_block = (float *) malloc( buffersize * sizeof(float) );
    // get number of disks
    int disks = s_mgr->m_MaxNumberOfDisks;
    //time to service next request
    long double *disk_time = (long double*) malloc( disks * sizeof(long double) );
    for( int i = 0 ; i < disks ; i++ )
        disk_time[ i ] = msecRunCAC;

    memset( req_time, 0, buffersize*sizeof(float) );
    memset( disk_block, 0, buffersize*sizeof(float) );
    memset( disk_time, 0, disks*sizeof(long double) );

    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "getInfoAboutRequestList: Starting reading ... " << endl;
    #endif

    // Read next request until the end of file
    while( !feof( file ) )
    {
        if( fscanf( file, "%f\n", &duration ) == 1 )
        {
            duration = duration * 1000.0; // duration in msec

            curblock = ( PSEQBLOCKLIST )malloc( sizeof( SEQBLOCKLIST ) );
            if( curblock != NULL )
            {
                curblock->id       = m_Id; //to change
                curblock->block    = block;
                curblock->duration = duration;

                rc = object->MapBlock( (RioBlock) block, 0, &numreps, Reps);
                if( rc )
                {
                    RioErr << "getInfoAboutRequestList: Error at MapBlock"
                           << endl;
                }
                else
                {
                    curblock->disk          = Reps[0].disk;
                    curblock->physicalblock = Reps[0].block;
                }

                curblock->next     = NULL;
                curblock->prev     = NULL;

                if( block >=  buffersize )
                {
                    curblock->csystime = req_time[block % buffersize] + time;
                    time += req_time[ block % buffersize ];
                }
                else
                {
                    if( block < SBufferSize )
                    {
                        // When the requests to full server side buffer will be:
                        // shift msec to run cac
                        curblock->csystime = msecRunCAC;
                    }
                    else
                    {
                        // When the requests to full client buffer will be:
                        // shift msec to run CAC + 1 RTT
                        curblock->csystime = msecRunCAC +
                                             m_AverageRTTtoClient_msec;
                    }

                    if( disk_time[ curblock->disk ] < curblock->csystime )
                        disk_time[ curblock->disk ] = curblock->csystime;

                    disk_time[ curblock->disk ] +=
                            getEstimatedDiskServiceTime( curblock->disk );

                    disk_block[ block ] = disk_time[ curblock->disk ];

                    #ifdef RIO_DEBUG2
                    if( m_log.is_open() ) m_log << " id "     <<  curblock->id
                          << " block "  <<  curblock->block
                          << " disk "   << curblock->disk
                          << " servicetime "
                          << getEstimatedDiskServiceTime( curblock->disk )
                          << " disk_time "
                          << disk_time[ curblock->disk]  << endl;
                    #endif

                    if( (block + 1) == buffersize )
                    {
                        double maxdisktime = disk_time[ 0 ];
                        for( int i = 1; i < disks; i++ )
                        {
                            if( disk_time [ i ] > maxdisktime )
                                maxdisktime = disk_time[ i ];
                        }
                        double timetotx = msecRunCAC +
                                          m_AverageRTTtoClient_msec;
                        for( uint i = 0; i < CBufferSize; i++ )
                        {
                            if( disk_block [ i ] > timetotx )
                                timetotx = disk_block[ i ];
                            timetotx += timeToTransmit_msec;
                        }

                        //time is set to when the requests for Totalbuffers
                        //blocks will be:
                        //max between until tx all BC blocks and read BS + BC blocks
                        double maxtime;
                        if( timetotx > maxdisktime )
                            maxtime = timetotx;
                        else
                            maxtime = maxdisktime;
                        time = maxtime + m_AverageRTTtoClient_msec;

                        #ifdef RIO_DEBUG2
                        if( m_log.is_open() )
                        {
                            m_log << "maxdisktime " << maxdisktime << endl;
                            m_log << "timetotx " << timetotx << endl;
                            m_log << "max + RTT = time => " << time << endl;
                        }
                        #endif
                    }
                }
                req_time[ block % buffersize ] = duration;
            }
            else 
            {
                RioErr << "getInfoAboutRequestList: Could not allocate memory. "
                       << endl;
    
                free( req_time );
                free( disk_block );
                free( disk_time );

                return NULL;
            }
            if( prevblock ) {
                curblock->prev = prevblock;
                prevblock = prevblock->next = curblock;
            }
            else
                prevblock = init_stream = curblock;
            duration = 0;
        }
        else
        {
            if( !feof( file ) )
            {
                RioErr << "getInfoAboutRequestList: Could not read from file."
                       << endl;

                free( req_time );
                free( disk_block );
                free( disk_time );

                return NULL;
            }
        }
        block++;
    }

    m_TotalBlocks = block;

    fclose( file );

    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "getInfoAboutRequestList: Closing file." << endl;
    #endif

    final_time = Timer.CurrentTime();
    time = getInterval( initial_time, final_time );
    m_SampleGenReqListTime = time;
    s_mgr->SetAverageGenReqListTime( time , this );

    #ifdef RIO_DEBUG2
    if( m_log.is_open() )
    {
        m_log<<"---------------------------------------------------------------- "<<endl;
        m_log<<"getInfoAboutRequestList Time = " << time << " msec "              <<endl;
        m_log<<"---------------------------------------------------------------- "<<endl;
    }
    #endif

    free( req_time );
    free( disk_block );
    free( disk_time );

    return init_stream;
}

struct timeval RioStream::getLastArrivalTime

(

)

[read]

Definition at line 2664 of file server/StreamManager.cpp.

{
    return m_LastArrivalTime;
}

u32 RioStream::getMaximumBlockToSend

(

)

Definition at line 2844 of file server/StreamManager.cpp.

{
    if( m_FinishedMovie == true )
    {
        return m_PosAtClientsList->nextBlockToBeRequested->block + 1;
    }
    else
    {
        return m_PosAtClientsList->nextBlockToBeRequested->block;
    }
}

u32 RioStream::getMinimumBlockToSend

(

)

Definition at line 2856 of file server/StreamManager.cpp.

{
    return m_MinimumBlockToSend;
}

void RioStream::getMutexUpdateEvent

(

)

Definition at line 2395 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexUpdateEvent );
}

long double RioStream::getNetworkRate

(

)

Definition at line 2764 of file server/StreamManager.cpp.

{
    return s_mgr->GetNetworkRate();
}

int RioStream::getNumberOfBlocks

(

)

Definition at line 2694 of file server/StreamManager.cpp.

{
    return m_TotalBlocks;
}

int RioStream::getNumberOfBuffersForEachClient

(

)

Definition at line 2731 of file server/StreamManager.cpp.

{
    return s_mgr->GetNumberOfBuffersForEachClient();
}

int RioStream::getPlayBlock

(

)

Definition at line 2681 of file server/StreamManager.cpp.

{
    if( m_StartingToSend )
    {
        return 0;
    }
    else
    {
        return(m_PosAtClientsList->nextBlockToBeRequested->block -
               getClientBufferSize());
    }
}

bool RioStream::getPrefetchedAllBlocks

(

)

Definition at line 2882 of file server/StreamManager.cpp.

{
    return m_PrefetchedAllBlocks;
}

int RioStream::getQueueSize

(

)

Definition at line 2770 of file server/StreamManager.cpp.

{
    return m_nQueue;
}

double RioStream::getRTTtoClient

(

)

Definition at line 2776 of file server/StreamManager.cpp.

{
    return m_AverageRTTtoClient_msec;
}

int RioStream::getServerBufferStatus

(

)

Definition at line 2705 of file server/StreamManager.cpp.

{
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << " m_ServerBufferStatus " << m_ServerBufferStatus
          << " m_TokensToSend " << m_TokensToSend
          << endl;
    #endif

    if( m_TokensToSend > 0 )
        return ( m_ServerBufferStatus - m_TokensToSend );
    else
        return m_ServerBufferStatus;
}

int RioStream::getTokensToDisk

(

)

Definition at line 2658 of file server/StreamManager.cpp.

{
    return m_TokensToDisk;
}

int RioStream::getTokensToSend

(

)

Definition at line 2804 of file server/StreamManager.cpp.

{
    return m_TokensToSend;
}

RioTrafficType RioStream::GetType

(

)

Definition at line 1196 of file server/StreamManager.cpp.

{
    return m_Type;
}

void RioStream::incMinimumBlockToSend

(

)

Definition at line 2861 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexMinBlock );
    m_MinimumBlockToSend++;
    pthread_mutex_unlock( &MutexMinBlock );
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "MinimumBlockToSend incremented : " << m_MinimumBlockToSend
          << endl;
    #endif
}

void RioStream::incTokensToSend

(

)

Definition at line 2810 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexTokensToSend );
    m_TokensToSend++;
    pthread_mutex_unlock( &MutexTokensToSend );
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "TokensToSend incremented : " << m_TokensToSend << endl;
    #endif
}

void RioStream::Initialize

(

CStreamManager *

mgr

)

Definition at line 1025 of file server/StreamManager.cpp.

{
    s_mgr = mgr;
}

int RioStream::MaxRequests

(

)

Definition at line 1184 of file server/StreamManager.cpp.

{
    if( m_Status != StreamStatusOpened )
    {
        return ERROR_STREAMMANAGER + ERROR_STREAM_NOT_OPENED;
    }

    return m_UserMaxRequests;
}

void RioStream::NewBlockArrival

(

)

Definition at line 2791 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexBlock );
    m_ServerBufferStatus++;
    pthread_cond_broadcast(&ConditionBlockArrival);
    pthread_mutex_unlock( &MutexBlock );
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "BufferStatus incremented : " << m_ServerBufferStatus << endl;
    #endif
}

void RioStream::NRTHold

(

EventDataRequest *

event

)

[private]

Definition at line 1819 of file server/StreamManager.cpp.

{
    // If this is the first hold request for this stream
    // put stream on list of stream which are currently holding requests
    if( m_nQueue == 0 )
    {
        s_mgr->m_NRTHoldList.Put( this );
    }

    m_Queue.Put( event );
    m_nQueue++;
    m_QueuedCount++;
}

int RioStream::OpenObject	(	char *	ObjectName,
		RioAccess	Access,
		struct timeval	RTT,
		int	BufferSize,
		RioStreamObj **	StreamObj
	)

Definition at line 2157 of file server/StreamManager.cpp.

{
    int rc;

    // Check if object access matches stream direction
    if( (s_Direction == RioStreamDirectionRead )&&( Access & RIO_WRITE_ACCESS) )
    {
        return ERROR_STREAMMANAGER + ERROR_INVALID_ACCESS_TYPE;
    }
    if( (s_Direction == RioStreamDirectionWrite )&&( Access & RIO_READ_ACCESS) )
    {
        return ERROR_STREAMMANAGER + ERROR_INVALID_ACCESS_TYPE;
    }

    if( m_Status != StreamStatusOpened )
    {
        return ERROR_STREAMMANAGER + ERROR_STREAM_NOT_OPENED;
    }
    
    RioStreamObj *op = new RioStreamObj( s_mgr, this );
    if( op == 0 )
    {
        return ERROR_STREAMMANAGER + ERROR_MEMORY;
    }

    op->o_object = 0;

    rc = s_mgr->m_ObjectManager->Open( ObjectName, Access, &op->o_object );

    if( rc == 0 )
    {
        *StreamObj = op;

        // if it is VBR stream, if it is using buffer, initialize it
        if( (GetType() == RIO_TRAFFIC_VBR )&&( s_mgr->m_UseServerSideBuffers ) )
        {
            struct timeval initial_time, final_time;
            double time;

            initial_time = Timer.CurrentTime();

            m_ClientBufferSize = BufferSize;
            m_AverageRTTtoClient_msec = RTT.tv_sec * 1000.0 +
                                        RTT.tv_usec / 1000.0;

            // get file name (with information to admission control)
            char infoFileName[255]="", mmobjectname[255]="";
            strcpy( mmobjectname, ObjectName );
            char *pos = strrchr( mmobjectname, '.' );
            strcpy( pos+1, "cactimes" );
            strcpy( infoFileName, s_mgr->m_FileRoot );
            strcat( infoFileName, "/" ) ;
            strcat( infoFileName, mmobjectname ) ;

            m_FirstArrivalTime = initial_time.tv_sec  * 1000.0 +
                                 initial_time.tv_usec / 1000.0;

            #ifdef RIO_DEBUG2
            if( m_log.is_open() )
                m_log << "OpenObject: Video : " << ObjectName
                      << endl << "OpenObject: CAC info: " << infoFileName
                      << endl << "OpenObject: FirstArrivalTime "
                      << m_FirstArrivalTime << endl;
            #endif

            m_WaitingQueueSize = s_mgr->UpdateNumberOfWaitingClients( 1 );

            pthread_mutex_lock( &s_mgr->MutexRunCAC );

            if( ( m_RequestList = getInfoAboutRequestList( infoFileName,
                                                      op->o_object )) != NULL )
            {
                if( s_mgr->m_UseNewCAC )
                {

                    m_PosAtClientsList = s_mgr->CanAdmit( m_RequestList,
                                                     this,
                                                     m_AverageRTTtoClient_msec,
                                                     m_ClientBufferSize );
                    final_time = Timer.CurrentTime();
                    time = getInterval( initial_time, final_time );
                    m_SampleCACTime = time;
                    int index;

                    if( m_PosAtClientsList != NULL )
                    {
                        index = 0;
                    }
                    else
                    {
                        index = 1;
                    }

                    m_SampleAdmissionProcessTime =  m_SampleGenReqListTime+
                                                    m_SampleSortListTime+
                                                    m_SampleSimulationTime;

                    m_SampleWaitingTime =  m_SampleCACTime - 
                                           m_SampleAdmissionProcessTime;

                    s_mgr->SetAverageWaitingTime( m_SampleWaitingTime,
                                                  this, index,
                                                  m_WaitingQueueSize );
                    s_mgr->SetAverageAdmissionProcessTime(
                                            m_SampleAdmissionProcessTime,
                                            this, index );

                    s_mgr->SetAverageCACTime( time, this, index );

                    if( s_mgr->m_log.is_open() )
                    {
                        s_mgr->m_log << "Time to generate list: "
                                     << m_SampleGenReqListTime << " msec."
                                     << endl;
                        s_mgr->m_log << "Time to sort list: "
                                     << m_SampleSortListTime << " msec."
                                     << endl;
                        s_mgr->m_log << "Time to simulate list: "
                                     << m_SampleSimulationTime << " msec."
                                     << endl;
                        s_mgr->m_log << "Time to admission: "
                                     << m_SampleCACTime << " msec."
                                     << endl;
                        s_mgr->m_log << "Waiting time: "
                                     <<  m_SampleWaitingTime << " msec."
                                     << endl;
                        s_mgr->m_log << "Waiting queue size: "
                                     <<  m_WaitingQueueSize << endl;
                    }

                    #ifdef RIO_DEBUG2
                    if( m_log.is_open() )
                        m_log << "Time to admission: " << time << " msec."
                              << endl;
                    #endif
                    if( m_PosAtClientsList != NULL )
                    {
                        if( s_mgr->m_log.is_open() )
                            s_mgr->m_log << "OpenObject: New client admitted. "
                                         << "(RTT " << m_AverageRTTtoClient_msec
                                         << " msec)" << endl;

                        #ifdef RIO_DEBUG2
                        if( m_log.is_open() )
                            m_log << "OpenObject: New client admitted." << endl;
                        #endif
                    }
                    else
                    {
                        if( s_mgr->m_log.is_open() )
                            s_mgr->m_log << "OpenObject: Could not admit new "
                                         << "client" << endl;

                        #ifdef RIO_DEBUG2
                        if( m_log.is_open() )
                            m_log << "OpenObject: Could not admit new client "
                                  << endl;
                        #endif
                        if( s_mgr->m_log.is_open() )
                            s_mgr->m_log << "Deleting StreamObj " << op << endl;

                        delete op;
                        return -1;
                    }
                }
                else
                {
                    m_PosAtClientsList = s_mgr->InsertNewClient( m_RequestList,
                                                                 this );
                    s_mgr->UpdateNumberOfWaitingClients(-1);
                    pthread_mutex_unlock( &s_mgr->MutexRunCAC );
                }

                // initialize StartingToSend
                m_StartingToSend = true;

                setMinimumBlockToSend(
                        m_PosAtClientsList->nextBlockToBeRequested->block );
                m_FinishedMovie = false;

                // now it creates the server side buffer equals to
                // s_mgr->GetNumberOfBuffersForEachClient()
                if( s_mgr->GetNumberOfBuffersForEachClient() > 0 )
                {
                    m_Buffer = new ClientBuffer[
                                    s_mgr->GetNumberOfBuffersForEachClient()];
                    if( m_Buffer == 0 )
                    {
                        RioErr << "Could not allocate buffer."<< endl;
                        return -1;
                    }
                    else
                    {
                        for( int i = 0;
                             i < s_mgr->GetNumberOfBuffersForEachClient();
                             i++
                           )
                        {
                            prefetchBlock( i,
                              (EventDataRequest*)EventManager.New( 
                                                       EventTypeRTDataRequest ),
                                                       op);
                            m_nRequests++;
                            updateNextBlockToBeFetched( NULL );
                        }
                    }
                }
                m_TimeToBuffer = Timer.CurrentTime();
                //After this point, send Ok to client
            }
            else
            {
                RioErr << "Could not find " << infoFileName
                       << ", allocate memory or read from file."
                       << "Not using new control admission and buffers."
                       << endl;

                pthread_mutex_unlock( &s_mgr->MutexRunCAC );
                return S_OK;
            }
        }
        return S_OK;
    }

    if( s_mgr->m_log.is_open() )
        s_mgr->m_log << "Deleting StreamObj " << op << endl;

    delete op;
    return rc;
}

int RioStream::OpenObject	(	char *	ObjectName,
		RioAccess	Access,
		RioStreamObj **	StreamObj
	)

Definition at line 1142 of file server/StreamManager.cpp.

{
    int rc;

    // Check if object access matches stream direction
    if( ( s_Direction == RioStreamDirectionRead ) &&
       ( Access & RIO_WRITE_ACCESS ))
    {
        return ERROR_STREAMMANAGER + ERROR_INVALID_ACCESS_TYPE;
    }
    if( ( s_Direction == RioStreamDirectionWrite ) &&
       ( Access & RIO_READ_ACCESS ))
    {
        return ERROR_STREAMMANAGER + ERROR_INVALID_ACCESS_TYPE;
    }

    if( m_Status != StreamStatusOpened )
    {
        return ERROR_STREAMMANAGER + ERROR_STREAM_NOT_OPENED;
    }

    RioStreamObj *op = new RioStreamObj( s_mgr, this );
    if( op == 0 )
    {
        return ERROR_STREAMMANAGER + ERROR_MEMORY;
    }

    op->o_object = 0;

    rc = s_mgr->m_ObjectManager->Open( ObjectName, Access, &op->o_object );

    if( rc == 0 )
    {
        *StreamObj = op;
        return S_OK;
    }
    delete op;
    return rc;
}

RioStream& RioStream::operator=

(

const RioStream &

x

)

[private]

void RioStream::prefetchBlock	(	int	posAtBuffer,
		EventDataRequest *	event,
		RioStreamObj *	StreamObject
	)

Definition at line 2408 of file server/StreamManager.cpp.

{

    m_Buffer[posAtBuffer].blockInfo = m_PosAtClientsList->nextBlockToBeFetched;

    #ifdef RIO_DEBUG2
    struct timeval time = Timer.CurrentTime();
    long double time_msec = time.tv_sec * 1000.0 + time.tv_usec/1000.0 ;
    if( m_log_requests.is_open() )
    {
        m_log_requests << m_Buffer[posAtBuffer].blockInfo->block << "\t\t"
                       << m_Buffer[posAtBuffer].blockInfo->csystime << "\t\t"
                       << time_msec - m_FirstArrivalTime << "\t\t"
                       << ( time_msec - m_FirstArrivalTime ) -
                          m_Buffer[posAtBuffer].blockInfo->csystime
                       << endl;
    }
    #endif

    m_Buffer[posAtBuffer].event = event;
    m_Buffer[posAtBuffer].event->Request.BufferId         = -1;
    // to be used to confirm if the request was submit to
    // storage before cancel it
    m_Buffer[posAtBuffer].event->Request.RequestCounter   = -1;
    m_Buffer[posAtBuffer].event->Request.Reqid            = 0;
    m_Buffer[posAtBuffer].event->Request.Target.IPaddress = 0;
    m_Buffer[posAtBuffer].event->Request.Target.Port      = 0;
    m_Buffer[posAtBuffer].event->Request.Block            =
                                 m_Buffer[posAtBuffer].blockInfo->block;
    m_Buffer[posAtBuffer].event->Request.Reps[0].block    =
                                 m_Buffer[posAtBuffer].blockInfo->physicalblock;
    m_Buffer[posAtBuffer].event->Request.Reps[0].disk     =
                                 m_Buffer[posAtBuffer].blockInfo->disk;
    m_Buffer[posAtBuffer].event->Request.RepBits   = 0;
    m_Buffer[posAtBuffer].event->Request.Status    = 0;
    m_Buffer[posAtBuffer].event->Request.Operation = RealTimePrefetchBlock;
    m_Buffer[posAtBuffer].event->Request.streamobj = StreamObject;
    m_Buffer[posAtBuffer].event->Request.rioobject = StreamObject->o_object;

    StreamObject->m_PendingRequests++;

    if( m_StartingToSend == true )
    {
        #ifdef RIO_DEBUG2
        if( m_log.is_open() )
            m_log << "prefetchBlock:starting prefetching block "
                  << m_Buffer[posAtBuffer].event->Request.Block
                  << " (m_nRequests " << m_nRequests << ")"
                  << " m_ActiveRequests "
                  << s_mgr->m_ActiveRequests + 1
                  << endl;
        #endif
        s_mgr->m_Router->Put( (Event*)m_Buffer[posAtBuffer].event );
        s_mgr->m_ActiveRequests++;
    }
    else
    {
        if( ( m_nQueue == 0 ) && ( getTokensToDisk() > 0 ) )
        {
            #ifdef RIO_DEBUG2
            if( m_log.is_open() )
                m_log << "prefetchBlock:prefetching block (Router) "
                      << m_Buffer[posAtBuffer].event->Request.Block
                      << " (m_nRequests " << m_nRequests << ")"
                      << " m_ActiveRequests "
                      << s_mgr->m_ActiveRequests + 1
                      << endl;
            #endif
            s_mgr->m_Router->Put( (Event*)m_Buffer[posAtBuffer].event );
            s_mgr->m_ActiveRequests++;
            decTokensToDisk();
        }
        else
        {
            #ifdef RIO_DEBUG2
            if( m_log.is_open() )
                m_log << "prefetchBlock:prefetching block (RTQueue) "
                      << m_Buffer[posAtBuffer].event->Request.Block
                      << " (m_nRequests " << m_nRequests << ")"
                      << endl;
            #endif
            RTHold( m_Buffer[posAtBuffer].event );
        }
    }
}

void RioStream::ProcessNRTQueue

(

)

[private]

Definition at line 1930 of file server/StreamManager.cpp.

{
    EventDataRequest* request;

    // Process just one queued request at time
    // (round robin across all non real-time streams)
    if( m_nQueue > 0 ) // Just in case: Not really necessary
    {
        // retrieve request from queue
        request = m_Queue.Get();
        m_nQueue--;

        // Increment total number of requests sent to router
        s_mgr->m_ActiveRequests++;

        // Send request to router
        s_mgr->m_Router->Put( (Event*)request );
    }
    // If no requests left on queue, remove stream from list of streams
    // with hold requests
    if( m_nQueue == 0 )
        s_mgr->m_NRTHoldList.Remove( this );
}

void RioStream::ProcessRTQueue

(

)

[private]

Definition at line 1834 of file server/StreamManager.cpp.

{

    EventDataRequest* request;

    struct timeval t = Timer.CurrentTime();
    t.tv_usec += RELEASE_TIME_TOLERANCE;
    if( t.tv_usec > 1000000 )
    {
        t.tv_sec  += 1;
        t.tv_usec -= 1000000;
    }

    if( m_PosAtClientsList != NULL )
    {
        struct timeval clocknow = Timer.CurrentTime();

        // updates number of tokensToDisk
        updateTokensToDisk( clocknow );

        // Process maximum queued requests without violating stream rate
        while(( m_nQueue > 0 ) && ( getTokensToDisk() > 0 ))
        {
            #ifdef RIO_DEBUG2
            if( m_log.is_open() )
                m_log << "Stream:DataReq 1 - ProcessRTQueue "
                      << " Recuperando da fila."
                      << endl;
            #endif
            // retrieve request from queue
            request = m_Queue.Get();
            m_nQueue--;

            // Increment total number of requests sent to router
            s_mgr->m_ActiveRequests++;

            // Send request to router
            #ifdef RIO_DEBUG2
            if( m_log.is_open() ) m_log << "Stream:DataReq - ProcessRTQueue "
                  << " Enviando pedidos recuperados da fila para o router111 "
                  << endl;
            #endif
            s_mgr->m_Router->Put( (Event*)request );
            decTokensToDisk();
        }
    }
    else
    {
        // Process maximum queued requests without violating stream rate
        while( ( m_nQueue > 0 ) &&
               ( ( t.tv_sec > m_NextRelease.tv_sec) ||
                 ( ( t.tv_sec == m_NextRelease.tv_sec  ) &&
                   ( t.tv_usec >= m_NextRelease.tv_usec)
                 )
               )
             )
        {
            // retrieve request from queue
            #ifdef RIO_DEBUG2
            if( m_log.is_open() )
                m_log << "Stream:DataReq 2 - ProcessRTQueue "
                      << " Recuperando da fila."
                      << endl;
            #endif
            request = m_Queue.Get();
            m_nQueue--;

            // update new release time for next request
            m_NextRelease.tv_sec  += m_ArrivalInterval.tv_sec;
            m_NextRelease.tv_usec += m_ArrivalInterval.tv_usec;
            if( m_NextRelease.tv_usec > 1000000 )
            {
                m_NextRelease.tv_sec  += 1;
                m_NextRelease.tv_usec -= 1000000;
            }

            // Increment total number of requests sent to router
            s_mgr->m_ActiveRequests++;

            // Send request to router
            #ifdef RIO_DEBUG2
            if( m_log.is_open() )
                m_log << "Stream:DataReq - ProcessRTQueue "
                      << " Enviando pedidos recuperados da fila para o router "
                      << endl;
            #endif
            s_mgr->m_Router->Put( (Event*)request );
        }
    }
    // If no requests left on queue, remove stream from list of streams
    // with hold requests
    if( m_nQueue == 0 )
        s_mgr->m_RTHoldList.Remove( this );
}

void RioStream::releaseMutexUpdateEvent

(

)

Definition at line 2400 of file server/StreamManager.cpp.

{
    pthread_mutex_unlock( &MutexUpdateEvent );
}

void RioStream::RequestCompleted

(

Event *

Request

)

[private]

Definition at line 1955 of file server/StreamManager.cpp.

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

    // Save request info for sending reply to client after request is free
    DataRequestOperation Operation = Req->Operation;
    CommunicationAddress Target    = Req->Target;
    int Reqid                     = Req->Reqid;
    RioResult Status               = Req->Status;
    NATData result;
    // TODO: Remover depois, quando a RioNeti e a NetMgr nao forem mais usadas.
    unsigned int StorageNumber;

    // TODO: Remover depois, quando a RioNeti e a NetMgr nao forem mais usadas
    // (obtem o numero de storages).
    s_mgr->m_Router->GetNumberOfStorageNodes( &StorageNumber );

    // Free request
    EventManager.Free( Request );

    m_nRequests--;
    s_mgr->m_ActiveRequests--;

    #ifdef RIO_DEBUG2
    struct in_addr client_address;
    client_address.s_addr = Target.IPaddress;
    RioErr << "[RioStream] RequestCompleted Target: "
           << "IPAddr = " << inet_ntoa(client_address)
           << ", Port = " << Target.Port << endl;
    #endif

    // Check if stream was closed
    if( m_Status == StreamStatusClosed )
    {
        if( m_nRequests == 0 )
        {
            s_mgr->m_CloseList.Remove( this );
            s_mgr->m_used--;
            // specific for real-time streams
            if( m_Type == RIO_TRAFFIC_CBR )
            {
                s_mgr->m_UsedRate -= s_Rate;
            }
            if( (m_PosAtClientsList != NULL) &&
                (s_mgr->m_UseServerSideBuffers)
              )
            {
                s_mgr->RemoveClient( m_PosAtClientsList );
                if( m_Buffer )
                {
                    delete m_Buffer;
                    m_Buffer = NULL;
                }
            }
            #ifdef RIO_DEBUG2
            RioErr << "Closing mlog"    << endl;
            m_log.close();
            #endif
            #ifdef RIO_DEBUG2
            RioErr << "Closing mlogreq" << endl;
            m_log_requests.close();
            #endif
            s_mgr->m_FreeList.Put( this );
            if( s_mgr->m_log.is_open() )
                s_mgr->m_log << "RequestCompleted: Stream CLOSED. "
                             << " Requests: " << m_RequestCount
                             << ". Queued requests: " << m_QueuedCount << endl;
        }
    }
    else // else do if( m_Status == StreamStatusClosed )
    {
        
        #ifdef RIO_DEBUG2
        RioErr << "RioStream::RequestCompleted Status = " << Status 
               << ", Operation = " << Operation << ", ReqID = " << Reqid 
               << endl;
        #endif       
               
        // Send reply to client if required
        if( ( Operation == NonRealTimeRead ) ||
            ( Operation == RealTimeRead    ) ||
            ( Operation == RealTimeCancelBlock )
          )
        {
            // For read, just send result to client if there was an error
            if( Status != 0 )
            {
                #ifdef RIO_DEBUG2
                if( m_log.is_open() )
                    m_log << "RequestCompleted SendResult Error " << endl;
                #endif
                RioErr << "RequestCompleted SendResult Error " << endl;

                NATData input( Target.IPaddress, Target.Port );
                
                // Verifica se existe o mapeamento do servidor para a nova 
                // implementacao.
                // TODO: Remover este if quando a RioNeti e a NetMgr deixarem de
                // ser usadas.
                if( s_mgr->m_Router->m_NAT_map.findElement( input, 
                    StorageNumber + 1 ) )
                    result = s_mgr->m_Router->m_NAT_map.getElement( input, 
                                                            StorageNumber + 1 );
                else                    
                    result = s_mgr->m_Router->m_NAT_map.getElement( input, 0 );
                if( result.nat_addr != 0 )
                {    
                    // A nova chamada a funcao NetInterface.FindIPAndPort e para 
                    // verificarmos se devemos usar a funcao do objeto da nova 
                    // classe ou a funcao do objeto da classe antiga.
                    if( s_mgr->m_NetInterface->FindIPAndPort( result.nat_addr, 
                                                              result.nat_port ) 
                      )
                        s_mgr->m_NetInterface->SendResult( result.nat_addr, 
                                                           result.nat_port,
                                                           Reqid, Status );
                    else
                        s_mgr->m_NetMgr->SendResult( result.nat_addr, 
                                                     result.nat_port,
                                                     Reqid, Status );
                }
                else
                {                
                    // A nova chamada a funcao NetInterface.FindIPAndPort e para 
                    // verificarmos se devemos usar a funcao do objeto da nova 
                    // classe ou a funcao do objeto da classe antiga.
                    if( s_mgr->m_NetInterface->FindIPAndPort( Target.IPaddress, 
                                                              Target.Port ) )
                        s_mgr->m_NetInterface->SendResult( Target.IPaddress, 
                                                           Target.Port,
                                                           Reqid, Status );
                    else                                         
                        s_mgr->m_NetMgr->SendResult( Target.IPaddress, 
                                                     Target.Port,
                                                     Reqid, Status );
                }
            }
        }
        // write case
        else if( Operation == NonRealTimeWrite )
        {
            NATData input( Target.IPaddress, Target.Port );
            // Verifica se existe o mapeamento do servidor para a nova 
            // implementacao.
            // TODO: Remover este if quando a RioNeti e a NetMgr deixarem de ser
            // usadas.
            if( s_mgr->m_Router->m_NAT_map.findElement( input, 
                                                           StorageNumber + 1 ) )
                result = s_mgr->m_Router->m_NAT_map.getElement( input, 
                                                            StorageNumber + 1 );
            else
                result = s_mgr->m_Router->m_NAT_map.getElement( input, 0 );
            if( result.nat_addr != 0 )
            {
                Target.IPaddress = result.nat_addr;
                Target.Port = result.nat_port;
            }                
            if( result.nat_addr != 0 )
            {
                // A nova chamada a funcao NetInterface.FindIPAndPort e para 
                // verificarmos se devemos usar a funcao do objeto da nova 
                // classe ou a funcao do objeto da classe antiga.
                if( s_mgr->m_NetInterface->FindIPAndPort( result.nat_addr, 
                                                          result.nat_port ) )
                    s_mgr->m_NetInterface->SendResult( result.nat_addr, 
                                                       result.nat_port,
                                                       Reqid, Status );
                else                                        
                    s_mgr->m_NetMgr->SendResult( result.nat_addr, 
                                                 result.nat_port,
                                                 Reqid, Status );
            }
            else
            {          
                if( s_mgr->m_NetInterface->FindIPAndPort( Target.IPaddress, 
                                                          Target.Port ) )
                    s_mgr->m_NetInterface->SendResult( Target.IPaddress, 
                                                       Target.Port, Reqid, 
                                                       Status );
                else                                              
                    s_mgr->m_NetMgr->SendResult( Target.IPaddress, Target.Port,
                                                 Reqid, Status );
            }
        }
    } // fim do else do if( m_Status == StreamStatusClosed )

    struct in_addr clientip;
    clientip.s_addr = m_RemoteAddress.sin_addr.s_addr;
    UpdateQueueEvent *it_event = new UpdateQueueEvent( inet_ntoa(clientip),
                            htons( m_RemoteAddress.sin_port ), m_nRequests );
    s_mgr->PostITEvent( (MonitorEvent *)it_event );
}

void RioStream::RTHold

(

EventDataRequest *

event

)

[private]

Definition at line 1800 of file server/StreamManager.cpp.

{
    // If this is the first hold request for this stream
    // put stream on list of stream which are currently holding requests
    if( m_nQueue == 0 )
    {
        #ifdef RIO_DEBUG2
        if( m_log.is_open() )
            m_log << "Stream:DataReq:colocando first na fila " << endl;
        #endif
        s_mgr->m_RTHoldList.Put( this );
    }

    m_Queue.Put( event );
    m_nQueue++;
    m_QueuedCount++;
}

void RioStream::setMinimumBlockToSend

(

u32

Block

)

Definition at line 2871 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexMinBlock );
    m_MinimumBlockToSend = Block;
    pthread_mutex_unlock( &MutexMinBlock );
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "MinimumBlockToSend set : " << m_MinimumBlockToSend << endl;
    #endif
}

void RioStream::setTokensToSend

(

int

value

)

Definition at line 2831 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexTokensToSend );
    m_TokensToSend = value;
    pthread_mutex_unlock( &MutexTokensToSend );
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "TokensToSend set : " << m_TokensToSend << endl;
    #endif
}

int RioStream::startingStream

(

)

Definition at line 2784 of file server/StreamManager.cpp.

{
    return m_StartingToSend;
}

void RioStream::updateNextBlockToBeFetched

(

PSEQBLOCKLIST

Next

)

Definition at line 2535 of file server/StreamManager.cpp.

{
    if( Next == NULL )
    {
        pthread_mutex_lock( &MutexNextBlockToBeFetched );
        if( m_PosAtClientsList->nextBlockToBeFetched->next == NULL )
        {
            #ifdef RIO_DEBUG2
            if( m_log.is_open() ) m_log << "updateNextBlockToBeFetched: prefetchAll true"<< endl;
            #endif
            m_PrefetchedAllBlocks = true;
        }
        else
        {
            m_PosAtClientsList->nextBlockToBeFetched =
                           m_PosAtClientsList->nextBlockToBeFetched->next;
        }
        pthread_mutex_unlock( &MutexNextBlockToBeFetched );
    }
    else
    {
        pthread_mutex_lock( &MutexNextBlockToBeFetched );
        m_PosAtClientsList->nextBlockToBeFetched = Next;
        pthread_mutex_unlock( &MutexNextBlockToBeFetched );
    }
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "updateNextBlockToBeFetched: "
          << m_PosAtClientsList->nextBlockToBeFetched->block<< endl;
    #endif
}

void RioStream::updateTokensToDisk

(

struct timeval

time

)

Definition at line 2615 of file server/StreamManager.cpp.

{
    pthread_mutex_lock( &MutexTokensToDisk );
    while( ( (m_TokensToDiskClock.tv_sec < time.tv_sec ) ||
             ( ( m_TokensToDiskClock.tv_sec == time.tv_sec) &&
               (m_TokensToDiskClock.tv_usec < time.tv_usec )
             )
           )
            && (m_StartingToSend == false ))
    {
        m_TokensToDiskClock.tv_sec += m_ArrivalInterval.tv_sec;
        m_TokensToDiskClock.tv_usec += m_ArrivalInterval.tv_usec;
        if( m_TokensToDiskClock.tv_usec > 1000000 )
        {
            m_TokensToDiskClock.tv_sec  += 1;
            m_TokensToDiskClock.tv_usec -= 1000000;
        }

        if( m_TokensToDisk < s_mgr->GetBurstSizeOfEachClient() )
        {
            m_TokensToDisk++;
        }
    }
    pthread_mutex_unlock( &MutexTokensToDisk );
    #ifdef RIO_DEBUG2
    if( m_log.is_open() ) m_log << "TokensToDisk updated : " << m_TokensToDisk << endl;
    #endif
}

---

Friends And Related Function Documentation

friend class CStreamManager [friend]

Definition at line 300 of file server/StreamManager.h.

friend class RioStreamObj [friend]

Definition at line 301 of file server/StreamManager.h.

---

Field Documentation

pthread_cond_t RioStream::ConditionBlockArrival [private]

Definition at line 297 of file server/StreamManager.h.

struct timeval RioStream::m_ArrivalInterval [private]

Definition at line 257 of file server/StreamManager.h.

double RioStream::m_AverageRTTtoClient_msec [private]

Definition at line 282 of file server/StreamManager.h.

ClientBuffer* RioStream::m_Buffer [private]

Definition at line 281 of file server/StreamManager.h.

int RioStream::m_ClientBufferSize [private]

Definition at line 283 of file server/StreamManager.h.

bool RioStream::m_FinishedMovie [private]

Definition at line 287 of file server/StreamManager.h.

double RioStream::m_FirstArrivalTime [private]

Definition at line 268 of file server/StreamManager.h.

uint RioStream::m_Id [private]

Definition at line 279 of file server/StreamManager.h.

struct timeval RioStream::m_LastArrivalTime [private]

Definition at line 264 of file server/StreamManager.h.

ofstream RioStream::m_log

Definition at line 233 of file server/StreamManager.h.

ofstream RioStream::m_log_requests

Definition at line 233 of file server/StreamManager.h.

int RioStream::m_MaxRequests [private]

Definition at line 253 of file server/StreamManager.h.

u32 RioStream::m_MinimumBlockToSend [private]

Definition at line 290 of file server/StreamManager.h.

struct timeval RioStream::m_NextRelease [private]

Definition at line 256 of file server/StreamManager.h.

int RioStream::m_nQueue [private]

Definition at line 255 of file server/StreamManager.h.

int RioStream::m_nRequests [private]

Definition at line 252 of file server/StreamManager.h.

PSTREAMLIST RioStream::m_PosAtClientsList [private]

Definition at line 280 of file server/StreamManager.h.

bool RioStream::m_PrefetchedAllBlocks [private]

Definition at line 286 of file server/StreamManager.h.

CRequestQueue RioStream::m_Queue [private]

Definition at line 261 of file server/StreamManager.h.

u64 RioStream::m_QueuedCount [private]

Definition at line 260 of file server/StreamManager.h.

SOCKADDR_IN RioStream::m_RemoteAddress [private]

Definition at line 249 of file server/StreamManager.h.

u64 RioStream::m_RequestCount [private]

Definition at line 259 of file server/StreamManager.h.

PSEQBLOCKLIST RioStream::m_RequestList [private]

Definition at line 278 of file server/StreamManager.h.

double RioStream::m_SampleAdmissionProcessTime [private]

Definition at line 273 of file server/StreamManager.h.

double RioStream::m_SampleCACTime [private]

Definition at line 272 of file server/StreamManager.h.

double RioStream::m_SampleGenReqListTime [private]

Definition at line 269 of file server/StreamManager.h.

double RioStream::m_SampleSimulationTime [private]

Definition at line 271 of file server/StreamManager.h.

double RioStream::m_SampleSortListTime [private]

Definition at line 270 of file server/StreamManager.h.

double RioStream::m_SampleWaitingTime [private]

Definition at line 274 of file server/StreamManager.h.

int RioStream::m_ServerBufferStatus [private]

Definition at line 284 of file server/StreamManager.h.

bool RioStream::m_StartingToSend [private]

Definition at line 289 of file server/StreamManager.h.

StreamStatus RioStream::m_Status [private]

Definition at line 250 of file server/StreamManager.h.

struct timeval RioStream::m_TimeToBuffer [private]

Definition at line 266 of file server/StreamManager.h.

int RioStream::m_TokensToDisk [private]

Definition at line 288 of file server/StreamManager.h.

struct timeval RioStream::m_TokensToDiskClock [private]

Definition at line 265 of file server/StreamManager.h.

int RioStream::m_TokensToSend [private]

Definition at line 285 of file server/StreamManager.h.

int RioStream::m_TotalBlocks [private]

Definition at line 276 of file server/StreamManager.h.

RioTrafficType RioStream::m_Type [private]

Definition at line 251 of file server/StreamManager.h.

int RioStream::m_UserMaxRequests [private]

Definition at line 254 of file server/StreamManager.h.

int RioStream::m_WaitingQueueSize [private]

Definition at line 275 of file server/StreamManager.h.

pthread_mutex_t RioStream::MutexBlock [private]

Definition at line 295 of file server/StreamManager.h.

pthread_mutex_t RioStream::MutexMinBlock [private]

Definition at line 294 of file server/StreamManager.h.

pthread_mutex_t RioStream::MutexNextBlockToBeFetched [private]

Definition at line 293 of file server/StreamManager.h.

pthread_mutex_t RioStream::MutexTokensToDisk [private]

Definition at line 292 of file server/StreamManager.h.

pthread_mutex_t RioStream::MutexTokensToSend [private]

Definition at line 291 of file server/StreamManager.h.

pthread_mutex_t RioStream::MutexUpdateEvent [private]

Definition at line 296 of file server/StreamManager.h.

RioStream* RioStream::Next [private]

Definition at line 147 of file server/StreamManager.h.

RioStream* RioStream::Previous [private]

Definition at line 148 of file server/StreamManager.h.

RioStreamDirection RioStream::s_Direction [private]

Definition at line 247 of file server/StreamManager.h.

CStreamManager* RioStream::s_mgr [private]

Definition at line 243 of file server/StreamManager.h.

double RioStream::s_Rate [private]

Definition at line 248 of file server/StreamManager.h.

int RioStream::s_streamid [private]

Definition at line 245 of file server/StreamManager.h.

---

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

• server/StreamManager.h
• server/StreamManager.cpp