CStorageDevice Class Reference

#include <storagedevice.h>

Public Member Functions
	CStorageDevice ()
	~CStorageDevice ()
int	Initialize (char *DeviceName)
void	Stop ()
void	ProcessRequest (StrRequest *Request)
int	SectorSize ()
u64	Size ()
int	Initialize (char *DeviceName, char *ip, int nDiskThreads, double EstimatedParameter, bool CollectMeasures, char *LogsDirectory)
void	GetDiskServiceTime (double averageservicetime[301])
double	GetEstimatedTime ()
void	LoadMeasures (char *filename)
Data Fields
ofstream	m_log
char	logname [MaxPathSize]
int	logopen
int	m_MaxActiveDiskThreads
pthread_mutex_t	m_Mutex
pthread_mutex_t	m_MutexThreads [300]
pthread_cond_t	m_Condition
pthread_cond_t	m_ConditionNew [300]
pthread_cond_t	m_ConditionExitThread
int	m_ActiveDiskThreads
ThreadRequestParameter	m_RequestsForThreads [300]
double	m_EstServiceTime
double	m_DevServiceTime
double	m_EstServiceTimeAccThreads [301]
double	m_DevServiceTimeAccThreads [301]
double	m_EstResponseTime
double	m_DevResponseTime
double	m_EstResponseTimeAccThreads [301]
double	m_DevResponseTimeAccThreads [301]
unsigned long long int	m_NRespTSamplesAccThreads [301]
double	m_AvgServiceTime_Model
double	m_VarServiceTime_Model
unsigned long long int	m_NSamples
double	m_AvgServiceTimeAccThreads [301]
double	m_VarServiceTimeAccThreads [301]
unsigned long long int	m_NSamplesAccThreads [301]
double	m_EstimatedParameter
bool	m_CollectMeasures
struct timeval	m_LastServiceTime
bool	m_StartingServiceInterval
Static Private Member Functions
static void *	Thread (void *Param)
static void *	BlockThread (void *Param)
Private Attributes
CvsiThread	m_Thread
CvsiDevice	m_Device
CRequestQueue	m_Queue
int	m_Status
char *	m_Name
unsigned int	m_NumberOfSaveMeasures

---

Detailed Description

Definition at line 58 of file storagedevice.h.

---

Constructor & Destructor Documentation

CStorageDevice::CStorageDevice

(

)

Definition at line 57 of file storagedevice.cpp.

{
    /* Public variables */
    logopen = 0;
    m_MaxActiveDiskThreads = 1;
    m_ActiveDiskThreads    = 0;

    m_EstServiceTime       = 0.0;
    m_DevServiceTime       = 0.0;
    memset(m_EstServiceTimeAccThreads,0,sizeof(m_EstServiceTimeAccThreads));
    memset(m_DevServiceTimeAccThreads,0,sizeof(m_DevServiceTimeAccThreads));

    m_EstResponseTime = 0.0;
    m_DevResponseTime = 0.0;
    memset(m_EstResponseTimeAccThreads,0,sizeof(m_EstResponseTimeAccThreads));
    memset(m_DevResponseTimeAccThreads,0,sizeof(m_DevResponseTimeAccThreads));
    memset(m_NRespTSamplesAccThreads,0,sizeof(m_NRespTSamplesAccThreads));

    m_AvgServiceTime_Model = 0;
    m_VarServiceTime_Model = 0;
    m_NSamples = 0;
    memset(m_AvgServiceTimeAccThreads,0,sizeof(m_AvgServiceTimeAccThreads));
    memset(m_VarServiceTimeAccThreads,0,sizeof(m_VarServiceTimeAccThreads));
    memset(m_NSamplesAccThreads,0,sizeof(m_NSamplesAccThreads));

    m_EstimatedParameter      = 0;
    m_CollectMeasures         = false;
    m_StartingServiceInterval = false;

    /* Private variables */
    m_Status = DeviceInactive;
    m_Name = NULL;
    m_NumberOfSaveMeasures = 0;

    for( int i=0; i < 300; i++ )            
    {
       m_RequestsForThreads[ i ].Request = NULL;
       m_RequestsForThreads[ i ].ActiveThreads = 0;
       m_RequestsForThreads[ i ].Available = true;
       // Inicializa o mutex e a variavel de condicao usada pela thread "i".
       pthread_mutex_init( &m_MutexThreads[ i ], NULL );
       pthread_cond_init( &m_ConditionNew[ i ], NULL );
    }

    // Inicializa os mutexes e as variaveis de condicao (principais). 
    pthread_mutex_init( &m_Mutex, NULL );
    pthread_cond_init( &m_Condition, NULL );
    pthread_cond_init( &m_ConditionExitThread, NULL );
}

CStorageDevice::~CStorageDevice

(

)

Definition at line 108 of file storagedevice.cpp.

{
    // Destroi os mutexes e as variaveis de condicao usada pelas threads.
    for( int i=0; i < 300; i++ )            
    {
       pthread_mutex_destroy( &m_MutexThreads[ i ] );
       pthread_cond_destroy( &m_ConditionNew[ i ] );
    }
    // Destroi os mutexes e as variaveis de condicao (principais). 
    pthread_mutex_destroy( &m_Mutex );
    pthread_cond_destroy( &m_Condition );
    pthread_cond_destroy( &m_ConditionExitThread );
}

---

Member Function Documentation

void * CStorageDevice::BlockThread

(

void *

Param

)

[static, private]

Definition at line 487 of file storagedevice.cpp.

{
    ThreadParameter *BParam = ( ThreadParameter * ) Param;

    char *devicename        = BParam->DeviceName;
    int ThreadNumber        = BParam->ThreadNumber;
    u64 size                = BParam->DeviceSize;
    CStorageDevice* SDevice = BParam->Device;
    double ETParameter      = BParam->Device->m_EstimatedParameter;
    int activeThreads       = 0;
    int FlagStatus;
    double average, sample, sample_servicetime;
    u16 Command;
    struct timeval initialtime, finaltime, interval;

    CvsiDevice Device;
    StrRequest* Request;

    int stat = Device.OpenDeviceAgain( devicename, size );
    if( stat < 0 )
    {
        RioErr << "BlockThread: Could not open device "<< devicename << endl;
        return( (void *)ERROR_SS_THREAD_CREATE );
    }
    
    RioErr << "BLOCKTHREADID " << syscall( SYS_gettid ) << endl;

    SDevice->m_log << "Thread N " << ThreadNumber << endl;
    while( 1 )
    {
        Request = NULL;
        pthread_mutex_lock( &SDevice->m_MutexThreads[ThreadNumber] );
        //SDevice->m_log << "Thread " << ThreadNumber  << " esperando "<<endl ;
        if( SDevice->m_RequestsForThreads[ ThreadNumber ].Available == false )
        {
            Request = SDevice->m_RequestsForThreads[ ThreadNumber ].Request;
            SDevice->m_RequestsForThreads[ ThreadNumber  ].Available = true;
            activeThreads   = SDevice->m_RequestsForThreads[ ThreadNumber ].ActiveThreads;
            SDevice->m_log << "Thread " << ThreadNumber  << "(" << pthread_self() << ") entrou "<<endl ;

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

                RioErr << " ----Thread " << pthread_self()
                       << " atendendo requisicao abaixo----" << endl;
                RioErr << " ClientId: " << Request->ClientId << endl;
                RioErr << " IPaddress " << inet_ntoa( client_IP ) << endl;
                RioErr << " Port " << htons(client_PORT.sin_port) << endl;
                RioErr << " Request->Disk " << Request->Disk << endl;
            }
            #endif
        }
        else
        {

           #ifdef RIO_DEBUG2
           RioErr << " ----Esperando em m_ConditionNew[" << ThreadNumber << "]----" 
                  << endl;
           #endif

           pthread_cond_wait( &SDevice->m_ConditionNew[ThreadNumber],
                              &SDevice->m_MutexThreads[ThreadNumber] );
           if( SDevice->m_RequestsForThreads[ ThreadNumber ].Available ==
               false
             )
           {
                Request = SDevice->m_RequestsForThreads[ ThreadNumber ].Request;
                SDevice->m_RequestsForThreads[ ThreadNumber  ].Available = true;
                activeThreads   = SDevice->m_RequestsForThreads[ ThreadNumber ].ActiveThreads;
                SDevice->m_log << "Thread " << ThreadNumber  << "(" << pthread_self() << ") entrou "<<endl ;
           }
        }
        pthread_mutex_unlock(&SDevice->m_MutexThreads[ThreadNumber]);

        if( Request == NULL )
        {
            break;
        }

        Command = Request->Command;

        // Test if this I/O will require a status message to be sent to router.
        if(( Command & 0xf ) == MSG_RSS_READ )
        {
            // if command is read this is the first of two steps
            // Send status message only if bit RSS_INTERMEDIATE is set
            if (Command & RSS_INTERMEDIATE)
                FlagStatus = 1;
            else
                FlagStatus = 0;
        }
        else
        {
            // If this something else (not "read") this is the last step
            // Send status message if bit RSS_NOCOMPLETE is not set
            if( Command & RSS_NOCOMPLETE )
                FlagStatus = 0;
            else
                FlagStatus = 1;
        }

        // Reset flag bits on command
        Command = Command & 0xf;

        switch( Command )
        {
            case MSG_RSS_WRITE:
            case MSG_RSS_FLUSH:
                Request->Status = RequestStatusWrite;
                gettimeofday( &initialtime,0 );
                stat = Device.Write(Request->Pos,Request->Size,Request->Data);
                
                // Esta parte foi comentada pois ainda nao esta sendo usada pela 
                // emulacao mas futuramente pretendemos criar um script para 
                // usar esta informacao.
                //#ifdef RIO_DEBUG_EMUL
                //RioErr << "DEVICEWRITE " << ( int ) time( NULL ) << " " 
                //       << Request->Size << endl;
                //#endif       
                       
                Request->Status = RequestStatusWriteDone;
                break;
            case MSG_RSS_READ:
            case MSG_RSS_FETCH:
                Request->Status = RequestStatusRead;
                gettimeofday( &initialtime,0 );
                stat = Device.Read(Request->Pos,Request->Size,Request->Data);
                
                // Esta impressao foi comentada porque estes logs ainda nao sao 
                // necessarios. Se for necessario usar estes logs, esta linha do 
                // codigo pode ser descomentada (ja existe um script para
                // processar estes logs).
                //#ifdef RIO_DEBUG_EMUL
                //RioErr << "DEVICEREAD " << ( int ) time( NULL ) << " " 
                //       << Request->Size << endl;
                //#endif       
                       
                Request->Status = RequestStatusReadDone;
                break;
            default:
                RioErr << "Invalid Command code on Request: " << Command 
                       << endl;
                RioErr << "Unexpected error on storage device thread. "
                       << "Aborting thread " << endl;
                return 0;
        }

        //updates measures
        pthread_mutex_lock(&SDevice->m_Mutex);

        gettimeofday( &finaltime,0 );
        interval.tv_sec  = finaltime.tv_sec  - initialtime.tv_sec;
        interval.tv_usec = finaltime.tv_usec - initialtime.tv_usec;
        if( interval.tv_usec < 0)
        {
            interval.tv_sec -= 1;
            interval.tv_usec += 1000000;
        }

        sample = interval.tv_sec * 1000.0 + interval.tv_usec / 1000.0;

        //decrements number of active threads
        SDevice->m_ActiveDiskThreads--;

        if( SDevice->m_StartingServiceInterval == true )
        {
            SDevice->m_StartingServiceInterval = false;
        }
        else
        {
            interval.tv_sec  = finaltime.tv_sec  - SDevice->m_LastServiceTime.tv_sec;
            interval.tv_usec = finaltime.tv_usec - SDevice->m_LastServiceTime.tv_usec;
            if( interval.tv_usec < 0)
            {
                interval.tv_sec -= 1;
                interval.tv_usec += 1000000;
            }

            sample_servicetime = interval.tv_sec * 1000.0 + interval.tv_usec / 1000.0;

            //updates estimated service time of disk
            if( SDevice->m_EstServiceTime == 0)
            {
                SDevice->m_EstServiceTime =  sample_servicetime;
            }
            else
            {
                SDevice->m_EstServiceTime = ( 1 - ETParameter ) * SDevice->m_EstServiceTime
                                            + ETParameter * sample_servicetime;
            }

            //updates Deviation of estimated service time of disk
            SDevice->m_DevServiceTime = ( 1 - ETParameter ) * SDevice->m_DevServiceTime +
            ETParameter * fabs( sample_servicetime - SDevice->m_EstServiceTime );

            // updates estimated service time according threads
            if( SDevice->m_EstServiceTimeAccThreads[activeThreads] == 0)
            {
                SDevice->m_EstServiceTimeAccThreads[activeThreads] = sample_servicetime;
            }
            else
            {
                SDevice->m_EstServiceTimeAccThreads[activeThreads] =
                ( 1 - ETParameter ) * SDevice->m_EstServiceTimeAccThreads[activeThreads]
                                                    + ETParameter * sample_servicetime;
            }

            //updates Deviation of estimated service time according number of threads
            SDevice->m_DevServiceTimeAccThreads[activeThreads] =
                ( 1 - ETParameter ) * SDevice->m_DevServiceTimeAccThreads[activeThreads] +
            ETParameter * fabs( sample_servicetime - SDevice->m_EstServiceTimeAccThreads[activeThreads] );

            //updates average service time of disk
            average  = SDevice->m_AvgServiceTime_Model;
            SDevice->m_AvgServiceTime_Model = SDevice->m_AvgServiceTime_Model +
                (( sample_servicetime - SDevice->m_AvgServiceTime_Model )/( SDevice->m_NSamples + 1 ));

            //updates variance of average service time of disk
            if( SDevice->m_NSamples >= 1 )
            {
                SDevice->m_VarServiceTime_Model = (1 - (float) 1/SDevice->m_NSamples ) * SDevice->m_VarServiceTime_Model
                                + (SDevice->m_NSamples + 1) * pow((SDevice->m_AvgServiceTime_Model - average), 2);
            }
            SDevice->m_NSamples++;

            //updates average service time according number of threads
            average = SDevice->m_AvgServiceTimeAccThreads[activeThreads];
            SDevice->m_AvgServiceTimeAccThreads[activeThreads] = SDevice->m_AvgServiceTimeAccThreads[activeThreads] +
            (( sample_servicetime - SDevice->m_AvgServiceTimeAccThreads[activeThreads])/( SDevice->m_NSamplesAccThreads[activeThreads] + 1 ));

            //updates variance of average service time according according number of threads
            if( SDevice->m_NSamplesAccThreads[activeThreads] >= 1 )
            {
                SDevice->m_VarServiceTimeAccThreads[activeThreads] = (1 - (float) 1/SDevice->m_NSamplesAccThreads[activeThreads] ) *
                  SDevice->m_VarServiceTimeAccThreads[activeThreads]
                                + (SDevice->m_NSamplesAccThreads[activeThreads] + 1) * pow((SDevice->m_AvgServiceTimeAccThreads[activeThreads] - average), 2);
            }
            SDevice->m_NSamplesAccThreads[activeThreads]++;

            //SDevice->m_log <<"Threads " << activeThreads << "\t" 
            //               << sample_servicetime << "\t"<< sample 
            //               << " finaltime " << finaltime.tv_sec  << " " <<finaltime.tv_usec
            //               << " m_LastServiceTime " << SDevice->m_LastServiceTime.tv_sec << " " 
            //               << SDevice->m_LastServiceTime.tv_usec 
            //               << endl;
        } // fim do else do if( SDevice->m_StartingServiceInterval == true )

        SDevice->m_LastServiceTime = finaltime;

        //updates estimated service time according number of threads
        if( SDevice->m_EstResponseTimeAccThreads[activeThreads] == 0)
        {
            SDevice->m_EstResponseTimeAccThreads[activeThreads] = sample;
        }
        else
        {
            SDevice->m_EstResponseTimeAccThreads[activeThreads] =
        ( 1 - ETParameter ) * SDevice->m_EstResponseTimeAccThreads[activeThreads]
                                            + ETParameter * sample;
        }

        //updates Deviation of estimated service time according number of threads
        SDevice->m_DevResponseTimeAccThreads[activeThreads] =
        ( 1 - ETParameter ) *  SDevice->m_DevResponseTimeAccThreads[activeThreads] +
        ETParameter * fabs( sample - SDevice->m_EstResponseTimeAccThreads[activeThreads] );

        SDevice->m_NRespTSamplesAccThreads[activeThreads]++;

        //updates estimated service time of disk
        if( SDevice->m_EstResponseTime == 0)
        {
            SDevice->m_EstResponseTime =  sample;
        }
        else
        {
            SDevice->m_EstResponseTime = ( 1 - ETParameter ) * SDevice->m_EstResponseTime +
                                        ETParameter * sample;
        }

        //updates Dev of estimated service time of disk
        SDevice->m_DevResponseTime = ( 1 - ETParameter ) * SDevice->m_DevResponseTime +
            ETParameter * fabs( sample - SDevice->m_EstResponseTime );

        SDevice->m_log<< "Thread "<< ThreadNumber << "("<< pthread_self() << ") saiu " <<endl;

        #ifdef RIO_DEBUG2
        RioErr << " ----Desbloqueando a thread principal do disco----" << endl;
        #endif
        pthread_cond_broadcast(&SDevice->m_Condition);
        pthread_mutex_unlock(&SDevice->m_Mutex);

        // Prepare status message to be sent to router if required
        StrMsg* Slot = NULL;
        if(( FlagStatus ) || ( stat < 0 ))
        {
            Slot = MsgManager.New();
            if( stat < 0 )
            {
                RioErr << "Storage: IO Failed (";
                switch( stat )
                {
                    case VSI_ERROR_DEVICE_POSITION:
                        RioErr <<"VSI_ERROR_DEVICE_POSITION) ";
                        break;
                    case VSI_ERROR_DEVICE_SEEK:
                        RioErr <<"VSI_ERROR_DEVICE_SEEK) ";
                        break;
                    case VSI_ERROR_DEVICE_READ:
                        RioErr <<"VSI_ERROR_DEVICE_READ) ";
                }
                RioErr << "RouterId " << Request->RouterId << endl;
                stat = ERROR_RSS_IO_FAILED;
            }
            if(( Command == MSG_RSS_WRITE ) || ( Command == MSG_RSS_FLUSH ))
            {
                Slot->Msg.Storage.Status.Type  = MSG_RSS_WRITECOMPLETE;
            }
            else
            {
                Slot->Msg.Storage.Status.Type  = MSG_RSS_READCOMPLETE;
            }
            Slot->Msg.Storage.Status.Size      = SizeMsgRSSstatus;
            Slot->Msg.Storage.Status.RouterId  = Request->RouterId;
            Slot->Msg.Storage.Status.StorageId = Request->Id;
            Slot->Msg.Storage.Status.Error     = stat;
            Slot->Msg.Storage.Status.ActiveThreads  = activeThreads;
        }

        // I/O done. should send request to client, wait for router command
        // (i.e. do nothing) or free request.
        switch( Command )
        {
            case MSG_RSS_WRITE:
                // If write, we are done. Release request
                RequestManager.Free(Request);
                break;
            case MSG_RSS_FLUSH:
                // If flush, we are done. Release request
                RequestManager.Free(Request);
                break;
            case MSG_RSS_READ:
                // if read, do nothing now. Send data to client after status
                // message is sent to router
                break;
            case MSG_RSS_FETCH:
                // For fetch need to update Request status before sending message
                // to router to avoid race condition if the router reply arrives
                // before setting the state
                // Fetch. Keep request around
                //RioErr << "Sending status (FETCH COMPLETE) to router " << endl;
                Request->Status = RequestStatusWaitSendCommand;
                break;
            default:
                RioErr << "Invalid Command code on Request: " << Command 
                       << endl;
                RioErr << "Unexpected error on storage device thread. "
                       << "Aborting thread" << endl;
                return 0;
        }

        // Send message to router if required
        if(( FlagStatus) || (stat < 0) )
        {
            #ifdef RIO_DEBUG2
            RioErr <<"Sending status (sotoragedevice.blockthread) to router." 
                   << endl;
            #endif

            Router.Send(Slot);
        };

        // Wait to send read to client just after status is sent to
        // router. This avoids router receiving status out of order
        // (assuming network guarantees ordered delivery)
        if( Command == MSG_RSS_READ )
        {
            // Successfull Read. Send data to client
            if( stat >= 0 )
            {
                Client.Send( Request );
            }
        }
    } //fim do while (1)

    Device.Close();
    pthread_exit( NULL );
}

void CStorageDevice::GetDiskServiceTime

(

double

averageservicetime[301]

)

Definition at line 892 of file storagedevice.cpp.

{
    for( int aux = 0; aux < 301; aux++ )
    {
        averageservicetime[aux] = m_EstServiceTimeAccThreads[aux];
        //averageservicetime[aux] = m_AvgServiceTimeAccThreads[aux];

    }
    m_NumberOfSaveMeasures++;


    if( m_NumberOfSaveMeasures % 5 == 0 && m_CollectMeasures )
    {
        char filename[350];
        FILE *handle;
        sprintf( filename, "%s.%d", logname, m_NumberOfSaveMeasures );

        if( ( handle = fopen( filename, "w")) == NULL )
        {
            if( m_log.is_open() )
                m_log << " Could not create logtimes!" << strerror ( errno )
                      << endl;
        }
        else
        {
            fprintf( handle, "# EstimatedServiceTime of Disk %10.10f msec\n", m_EstServiceTime );
            fprintf( handle, "# DeviationServiceTime %f\n", m_DevServiceTime );
            fprintf( handle, "\n# AverageServiceTime of Disk %10.10f msec\n", m_AvgServiceTime_Model );
            fprintf( handle, "# VarianceServiceTime %f \n",m_VarServiceTime_Model  );
            fprintf( handle, "# Number of Samples %lld \n",m_NSamples );
            fprintf( handle, "\n# EstimatedResponseTime of Disk %10.10f msec\n", m_EstResponseTime );
            fprintf( handle, "# DeviationResponseTime %f\n", m_DevResponseTime );
            fprintf( handle, "\n# Measures in msec.\n");
            fprintf( handle, "#\tEstServT\tDevServT\tAvgServT\tVarServT\tSamples\t\tEstRespT\tDevRespT\tNRespTSamples\n");
            for( int aux = 1; aux < 301; aux++ )
            {
                fprintf( handle, "%3d\t%6.6f\t%6.6f\t%6.6f\t%6.6f\t%lld\t\t%6.6f\t%6.6f\t%lld\n",
                    aux,
                    m_EstServiceTimeAccThreads[aux],
                    m_DevServiceTimeAccThreads[aux],
                    m_AvgServiceTimeAccThreads[aux],
                    m_VarServiceTimeAccThreads[aux],
                    m_NSamplesAccThreads[aux],
                    m_EstResponseTimeAccThreads[aux],
                    m_DevResponseTimeAccThreads[aux],
                    m_NRespTSamplesAccThreads[aux]);
            }
            fclose( handle );
        }
    }
    return;
}

double CStorageDevice::GetEstimatedTime

(

)

Definition at line 946 of file storagedevice.cpp.

{
    return m_EstServiceTime;
}

int CStorageDevice::Initialize	(	char *	DeviceName,
		char *	ip,
		int	nDiskThreads,
		double	EstimatedParameter,
		bool	CollectMeasures,
		char *	LogsDirectory
	)

Definition at line 160 of file storagedevice.cpp.

{
    char device_name[200] = "" , log[ MaxPathSize ] = ""; // 200

    // changes '/' by '_'
    int i = 0;
    while( DeviceName[i] != '\0' )
    {
        if( DeviceName[i] == '/' )
            device_name[i] = '_';
        else
            device_name[i] = DeviceName[i];
        i++;
        if( i == 190 ) break;
    }

    RioErr << "[StorageDevice] CollectMeasures: \t" << CollectMeasures
           << endl;
    if( CollectMeasures )
    {
        // Alteracao para armazenar os logs em um diretorio diferente daquele com
        // o codigo do storage server.
        if( LogsDirectory == NULL )
            // Arquivo armazenado no diretorio com o codigo do storage
            // server
            sprintf( log, "RIOdisk.%s.%s.log", ip, device_name );
        else if( LogsDirectory[ strlen( LogsDirectory ) - 1 ] != '/' ) 
            // Devemos colocar a "/" entre o nome do arquivo e do diretorio.
            sprintf( log, "%s/RIOdisk.%s.%s.log", LogsDirectory, ip, device_name );
        else
            // Nao devemos colocar a "/" entre o nome do arquivo e do diretorio.
            sprintf( log, "%sRIOdisk.%s.%s.log", LogsDirectory, ip, device_name );
        logopen = 1;
    }

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

    m_Status = DeviceActive;

    int stat;

    stat = m_Device.Open( DeviceName );
    if(stat < 0)
    {
        Stop();
        return ERROR_SS_INVALID_DEVICE;
    }

    // Save device name for message trace purposes
    m_Name = DeviceName;
    m_MaxActiveDiskThreads = nDiskThreads;

    if( CollectMeasures )
    {
        m_log.open( log );
        m_log << "Device Name : " << m_Name <<endl;
        m_log << "Max Active Threads : " <<  m_MaxActiveDiskThreads  <<endl;
    }

    // Alteracao para armazenar os logs em um diretorio diferente daquele com
    // o codigo do storage server.
    if( LogsDirectory == NULL )
        // Arquivo armazenado no diretorio com o codigo do storage server
        sprintf( logname, "RIOdisk.%s.%s.log.times", ip, device_name );
    else if( LogsDirectory[ strlen( LogsDirectory ) - 1 ] != '/' ) 
        // Devemos colocar a "/" entre o nome do arquivo e do diretorio.
        sprintf( logname, "%s/RIOdisk.%s.%s.log.times", LogsDirectory, ip, 
                 device_name );
    else
        // Nao devemos colocar a "/" entre o nome do arquivo e do diretorio.
        sprintf( logname, "%sRIOdisk.%s.%s.log.times", LogsDirectory, ip, 
                 device_name );

    stat = m_Thread.Create(Thread,(void*) this);
    if( stat < 0 )
    {
        Stop();
        return ERROR_SS_THREAD_CREATE;
    }

    m_CollectMeasures    = CollectMeasures;
    m_EstimatedParameter = EstimatedParameter;

    if( CollectMeasures )
    {
        m_log << "CollectMeasures \t"        << m_CollectMeasures    <<endl;
        m_log << "EstimatedTimeParameter \t" << m_EstimatedParameter <<endl;
    }

    for( int i = 0; i < m_MaxActiveDiskThreads; i++ )
    {
        pthread_t thread_id;
        pthread_attr_t attr;
        ThreadParameter *Param = new ThreadParameter;
        Param->DeviceName     = m_Name;
        Param->DeviceSize     = m_Device.Size();
        Param->Device         = this;
        Param->ThreadNumber   = i;

        pthread_attr_init( &attr );
        pthread_attr_setdetachstate( &attr, PTHREAD_CREATE_DETACHED );
        pthread_attr_setstacksize( &attr, 2*PTHREAD_STACK_MIN );
        if((pthread_create( &thread_id, &attr, &BlockThread,(void*) Param )) != 0 )
        {
            RioErr << "Storage: Could not start BlockThread " << i << "("
                   << strerror(errno) << ")!"<<endl;

            return ERROR_SS_THREAD_CREATE;
        }                                
    }
    LoadMeasures( logname );
    return 0;
}

int CStorageDevice::Initialize

(

char *

DeviceName

)

Definition at line 124 of file storagedevice.cpp.

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

    m_Status = DeviceActive;

    int stat;
    stat = m_Device.Open( DeviceName );
    if(stat < 0)
    {
        Stop();
        return ERROR_SS_INVALID_DEVICE;
    }

    stat = m_Thread.Create(Thread,(void*) this);
    if( stat < 0 )
    {
        Stop();
        return ERROR_SS_THREAD_CREATE;
    }

    // Save device name for message trace purposes
    m_Name = DeviceName;

    return 0;
}

void CStorageDevice::LoadMeasures

(

char *

filename

)

Definition at line 951 of file storagedevice.cpp.

{
        FILE *handle;
        int i;

        if( ( handle = fopen( filename, "r")) == NULL )
        {
            if( m_log.is_open() )
                m_log << " Could not read logtimes(" << filename << ")!"
                      << strerror ( errno )<< endl;
        }
        else
        {
            if( fscanf( handle, "# EstimatedServiceTime of Disk %lf msec\n", 
                &m_EstServiceTime ) != 1 )
            {
                RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                       << "ler o arquivo " << filename << endl;
            }
            if( fscanf( handle, "# DeviationServiceTime %lf\n", 
                &m_DevServiceTime ) != 1 )
            {
                RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                       << "ler o arquivo " << filename << endl;
            }
            if( fscanf( handle, "\n# AverageServiceTime of Disk %lf msec\n", 
                &m_AvgServiceTime_Model ) != 1 )
            {
                RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                       << "ler o arquivo " << filename << endl;
            }
            if( fscanf( handle, "# VarianceServiceTime %lf \n",
                &m_VarServiceTime_Model  ) != 1 )
            {
                RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                       << "ler o arquivo " << filename << endl;
            }
            if( fscanf( handle, "# Number of Samples %lld \n",
                &m_NSamples ) != 1 )
            {
                RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                       << "ler o arquivo " << filename << endl;
            }
            if( fscanf( handle, "\n# EstimatedResponseTime of Disk %lf msec\n", 
                &m_EstResponseTime ) != 1 )
            {
                RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                       << "ler o arquivo " << filename << endl;
            }
            if( fscanf( handle, "# DeviationResponseTime %lf\n", 
                &m_DevResponseTime ) != 1 )
            {
                RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                       << "ler o arquivo " << filename << endl;
            }
            if( fscanf( handle, "\n# Measures in msec.\n") != 0 )
            {
                RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                       << "ler o arquivo " << filename << endl;
            }
            if( fscanf( handle, "#\tEstServT\tDevServT\tAvgServT\tVarServT\tSamples\t\tEstRespT\tDevRespT\tNRespTSamples\n") 
                != 0 )
            {
                RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                       << "ler o arquivo " << filename << endl;
            }
            for( int aux = 1; (aux < 301) && (!feof( handle )); aux++ )
            {
                if( fscanf( handle, "%3d\t%lf\t%lf\t%lf\t%lf\t%lld\t\t%lf\t%lf\t%lld\n",
                    &i,
                    &m_EstServiceTimeAccThreads[aux],
                    &m_DevServiceTimeAccThreads[aux],
                    &m_AvgServiceTimeAccThreads[aux],
                    &m_VarServiceTimeAccThreads[aux],
                    &m_NSamplesAccThreads[aux],
                    &m_EstResponseTimeAccThreads[aux],
                    &m_DevResponseTimeAccThreads[aux],
                    &m_NRespTSamplesAccThreads[aux] ) != 9 )
                {
                    RioErr << "CStorageDevice::LoadMeasures: aviso - erro ao "
                           << "ler o arquivo " << filename << endl;
                }
                        
            }
            fclose( handle );
        }

    return;
}

void CStorageDevice::ProcessRequest

(

StrRequest *

Request

)

Definition at line 301 of file storagedevice.cpp.

{
    #ifdef RIO_DEBUG2
    struct in_addr client_IP;
    struct sockaddr_in client_PORT;
    if( Request != NULL )
    {
        client_IP.s_addr = Request->IPaddress;
        client_PORT.sin_port = Request->Port;
        RioErr << " ----A requisicao abaixo sera colocada da fila do disco----"
               << endl;
        RioErr << " ClientId: " << Request->ClientId
               << endl;
        RioErr << " IPaddress " << inet_ntoa( client_IP )
               << endl;
        RioErr << " Port " << htons(client_PORT.sin_port)
               << endl;
        RioErr << " Request->Disk " << Request->Disk << endl;
    }
    else
        RioErr << " ----A requisicao sera nula colocada da fila do disco----" 
               << endl;
    #endif

    m_Queue.Put(Request);

    #ifdef RIO_DEBUG2
    if( Request != NULL )
    {
        client_IP.s_addr = Request->IPaddress;
        client_PORT.sin_port = Request->Port;
        RioErr << " ----A requisicao abaixo foi colocada com sucesso na fila "
               << "do disco----" << endl;
        RioErr << " ClientId: " << Request->ClientId
               << endl;
        RioErr << " IPaddress " << inet_ntoa( client_IP )
               << endl;
        RioErr << " Port " << htons(client_PORT.sin_port)
               << endl;
        RioErr << " Request->Disk " << Request->Disk << endl;
    }
    else
        RioErr << " ----A requisicao nula foi colocada da fila do disco----" 
               << endl;
    #endif
}

int CStorageDevice::SectorSize

(

)

Definition at line 880 of file storagedevice.cpp.

{
    return m_Device.SectorSize();
}

u64 CStorageDevice::Size

(

)

Definition at line 886 of file storagedevice.cpp.

{
    return m_Device.Size();
}

void CStorageDevice::Stop

(

void

)

Definition at line 284 of file storagedevice.cpp.

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

    // Make sure thread return from Get() on request queue.
    m_Queue.Signal();

    m_Thread.Join(0);

    if( logopen == 1 ) m_log.close();
}

void * CStorageDevice::Thread

(

void *

Param

)

[static, private]

Definition at line 351 of file storagedevice.cpp.

{
    int i;

    // Get this class pointer
    CStorageDevice* Device = (CStorageDevice*) Param;

    // Get members of this class
    CRequestQueue &m_Queue = Device->m_Queue;
    int &m_Status          = Device->m_Status;

    if( Device->m_log.is_open() )
    {
        ofstream &m_log        = Device->m_log;

        m_log << "Device Thread running ..." << endl;
    }
    
    RioErr << "DEVICETHREADID " << syscall( SYS_gettid ) << endl;

    // Loop forever
    while( 1 )
    {
        pthread_mutex_lock(&Device->m_Mutex);
        if( Device->m_ActiveDiskThreads < Device->m_MaxActiveDiskThreads )
        {

            pthread_mutex_unlock(&Device->m_Mutex);

            #ifdef RIO_DEBUG2
            // Variavel booleana usada para indicar que o pedido foi atribuido
            // a alguma thread de bloco.
            bool isServed = false;
            #endif

            // Wait and get next request

            #ifdef RIO_DEBUG2
            RioErr << " ----Tentando obter e esperando, se necessario, uma "
                   << "requisicao da fila----" << endl;
            #endif
            StrRequest* Request = m_Queue.Get();

            #ifdef RIO_DEBUG2
            struct in_addr client_IP;
            struct sockaddr_in client_PORT;
            if( Request != NULL )
            {
                client_IP.s_addr = Request->IPaddress;
                client_PORT.sin_port = Request->Port;
                RioErr << " ----A requisicao abaixo foi removida da fila----"
                       << endl;
                RioErr << " ClientId: " << Request->ClientId
                       << endl;
                RioErr << " IPaddress " << inet_ntoa( client_IP )
                       << endl;
                RioErr << " Port " << htons(client_PORT.sin_port)
                       << endl;
                RioErr << " Request->Disk " << Request->Disk << endl;
            }
            else
                RioErr << " ----A requisicao nula foi removida da fila----"
                       << endl;
            #endif

            if(( m_Status != DeviceActive ) || ( Request == 0 ))
            {
                if(m_Status != DeviceActive)
                    RioErr << "Storage: loop: Interface is not active!" << endl;
                else
                    RioErr << "Storage: loop: Request = 0!" << endl;
                return( 0 );
            }

            pthread_mutex_lock(&Device->m_Mutex);
            //increments number of active threads
            Device->m_ActiveDiskThreads++;
            if(  Device->m_ActiveDiskThreads == 1 )
                Device->m_StartingServiceInterval = true;

            //Device->m_log << "vai pegar mutex "<<endl ;
            pthread_mutex_unlock( &Device->m_Mutex );

            for( i = 0; i < Device->m_MaxActiveDiskThreads; i++ )
            {
                if( Device->m_RequestsForThreads[ i ].Available == true )
                {
                    pthread_mutex_lock( &Device->m_MutexThreads[i] );
                    Device->m_RequestsForThreads[ i ].Request = Request;
                    Device->m_RequestsForThreads[ i ].Available = false;
                    Device->m_RequestsForThreads[ i ].ActiveThreads =
                        Device->m_ActiveDiskThreads;

                    #ifdef RIO_DEBUG2
                    RioErr << " ----Liberando thread no storagedevice para "
                           << " a requisicao abaixo----" << endl;
                    RioErr << " ClientId: " << Request->ClientId
                           << endl;
                    RioErr << " IPaddress " << inet_ntoa( client_IP )
                           << endl;
                    RioErr << " Port " << htons(client_PORT.sin_port)
                           << endl;
                    RioErr << " Request->Disk " << Request->Disk << endl;

                    isServed = true;
                    #endif

                    pthread_cond_broadcast(&Device->m_ConditionNew[i]);
                    pthread_mutex_unlock( &Device->m_MutexThreads[i] );
                    break;
                }
            }
            #ifdef RIO_DEBUG2
            if( !isServed )
            {
                RioErr << " ----Aviso: a requisicao abaixo foi ignorada, pois "
                       << "incorretamente achamos que existiam de disco "
                       << "disponiveis! threads ativas = "
                       << Device->m_ActiveDiskThreads << " e threads totais = "
                       << Device->m_MaxActiveDiskThreads << " ----" << endl;
                RioErr << " ClientId: " << Request->ClientId << endl;
                RioErr << " IPaddress " << inet_ntoa( client_IP ) << endl;
                RioErr << " Port " << htons( client_PORT.sin_port ) << endl;
                RioErr << " Request->Disk " << Request->Disk << endl;
            }
            #endif
        }
        else // else do if( Device->m_ActiveDiskThreads < Device->m_MaxActiv...
        {
            RioErr << "Storage: Waiting Available Thread!" << endl;
            pthread_cond_wait( &Device->m_Condition,&Device->m_Mutex );
            pthread_mutex_unlock( &Device->m_Mutex );
        }
    } // fim do while( 1 )*/
}

---

Field Documentation

char CStorageDevice::logname[MaxPathSize]

Definition at line 87 of file storagedevice.h.

int CStorageDevice::logopen

Definition at line 88 of file storagedevice.h.

int CStorageDevice::m_ActiveDiskThreads

Definition at line 94 of file storagedevice.h.

double CStorageDevice::m_AvgServiceTime_Model

Definition at line 108 of file storagedevice.h.

double CStorageDevice::m_AvgServiceTimeAccThreads[301]

Definition at line 111 of file storagedevice.h.

bool CStorageDevice::m_CollectMeasures

Definition at line 116 of file storagedevice.h.

pthread_cond_t CStorageDevice::m_Condition

Definition at line 91 of file storagedevice.h.

pthread_cond_t CStorageDevice::m_ConditionExitThread

Definition at line 93 of file storagedevice.h.

pthread_cond_t CStorageDevice::m_ConditionNew[300]

Definition at line 92 of file storagedevice.h.

CvsiDevice CStorageDevice::m_Device [private]

Definition at line 124 of file storagedevice.h.

double CStorageDevice::m_DevResponseTime

Definition at line 103 of file storagedevice.h.

double CStorageDevice::m_DevResponseTimeAccThreads[301]

Definition at line 105 of file storagedevice.h.

double CStorageDevice::m_DevServiceTime

Definition at line 98 of file storagedevice.h.

double CStorageDevice::m_DevServiceTimeAccThreads[301]

Definition at line 100 of file storagedevice.h.

double CStorageDevice::m_EstimatedParameter

Definition at line 115 of file storagedevice.h.

double CStorageDevice::m_EstResponseTime

Definition at line 102 of file storagedevice.h.

double CStorageDevice::m_EstResponseTimeAccThreads[301]

Definition at line 104 of file storagedevice.h.

double CStorageDevice::m_EstServiceTime

Definition at line 97 of file storagedevice.h.

double CStorageDevice::m_EstServiceTimeAccThreads[301]

Definition at line 99 of file storagedevice.h.

struct timeval CStorageDevice::m_LastServiceTime

Definition at line 117 of file storagedevice.h.

ofstream CStorageDevice::m_log

Definition at line 86 of file storagedevice.h.

int CStorageDevice::m_MaxActiveDiskThreads

Definition at line 89 of file storagedevice.h.

pthread_mutex_t CStorageDevice::m_Mutex

Definition at line 90 of file storagedevice.h.

pthread_mutex_t CStorageDevice::m_MutexThreads[300]

Definition at line 90 of file storagedevice.h.

char* CStorageDevice::m_Name [private]

Definition at line 127 of file storagedevice.h.

unsigned long long int CStorageDevice::m_NRespTSamplesAccThreads[301]

Definition at line 106 of file storagedevice.h.

unsigned long long int CStorageDevice::m_NSamples

Definition at line 110 of file storagedevice.h.

unsigned long long int CStorageDevice::m_NSamplesAccThreads[301]

Definition at line 113 of file storagedevice.h.

unsigned int CStorageDevice::m_NumberOfSaveMeasures [private]

Definition at line 128 of file storagedevice.h.

CRequestQueue CStorageDevice::m_Queue [private]

Definition at line 125 of file storagedevice.h.

ThreadRequestParameter CStorageDevice::m_RequestsForThreads[300]

Definition at line 95 of file storagedevice.h.

bool CStorageDevice::m_StartingServiceInterval

Definition at line 118 of file storagedevice.h.

int CStorageDevice::m_Status [private]

Definition at line 126 of file storagedevice.h.

CvsiThread CStorageDevice::m_Thread [private]

Definition at line 123 of file storagedevice.h.

double CStorageDevice::m_VarServiceTime_Model

Definition at line 109 of file storagedevice.h.

double CStorageDevice::m_VarServiceTimeAccThreads[301]

Definition at line 112 of file storagedevice.h.

---

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

• storagedevice.h
• storagedevice.cpp