Subversion Repositories DevTools

//---------------------------------------------------------------------------

#pragma warn -com
#include <LoggingMacros.h>
#pragma warn +com

#include <vcl.h>
#pragma hdrstop

#include "UdTransferThread.h"
#include <algorithm>
#include <dateutils.hpp>
#include <Registry.hpp>

#include "IDeviceFactory.h"
#include "IDevice.h"
#include "DeviceSimulatorThread.h"

namespace
{
    const char * REG_BASE_KEY       =   "\\Software\\ERG\\TxnTestManager\\UD Transfer";
    const char * REG_RENAME         =   "RenameSentFiles";
    const char * REG_IGNORESENT     =   "IgnoreSentFiles";
    const char * REG_MAXDEVICES     =   "MaxDeviceSimulators";
    const char * REG_DEVICEID       =   "StartingDeviceId";
    const char * REG_DEVICETYPE     =   "DeviceType";
    const char * REG_CONNECTTIMEOUT =   "DeviceConnectTimeout";

    const char * SENT_STRING        =   "sent_";
}

using namespace std;

#pragma package(smart_init)
//---------------------------------------------------------------------------

//   Important: Methods and properties of objects in VCL can only be
//   used in a method called using Synchronize, for example:
//
//      Synchronize(UpdateCaption);
//
//   where UpdateCaption could look like:
//
//      void __fastcall UdTransferThread::UpdateCaption()
//      {
//        Form1->Caption = "Updated in a thread";
//      }
//---------------------------------------------------------------------------

__fastcall UdTransferThread::UdTransferThread(const UdTransferTask & task,
                                              UdTransferManager * manager,
                                              IDeviceFactory * deviceFactory ) :
    TThread( false ),
    m_task( task ),
    m_manager( manager ),
    m_deviceFactory( deviceFactory ),
    m_lastDevice( 0 ),
    m_running( false ),
    m_error( false ),
    m_simulators( false ),
    m_renameSent( false ),
    m_ignoreSent( false ),
    m_maxDevices( 100 ),
    m_deviceId( 1 ),
    m_connectTimeout( 5 ),
    m_deviceType("000SIM")

{
    Priority = tpHigher;
}
//---------------------------------------------------------------------------

void UdTransferThread::SetName()
{
    THREADNAME_INFO info;
    info.dwType = 0x1000;
    info.szName = "UdTransferThread";
    info.dwThreadID = -1;
    info.dwFlags = 0;

    __try
    {
         RaiseException( 0x406D1388, 0, sizeof(info)/sizeof(DWORD),(DWORD*)&info );
    }
    __except (EXCEPTION_CONTINUE_EXECUTION)
    {
    }
}
//---------------------------------------------------------------------------

bool UdTransferThread::InitialiseTask()
{
    if( DirectoryExists( m_task.sourcePath ) )
    {
        // Read generic transfer settings from the registry

        TRegistry * registry = new TRegistry;
        registry->RootKey = HKEY_CURRENT_USER;
        if( registry->OpenKey(REG_BASE_KEY, true) )
        {
            if( registry->ValueExists(REG_RENAME) )
            {
                m_renameSent = registry->ReadBool(REG_RENAME);
            }
            if( registry->ValueExists(REG_IGNORESENT) )
            {
                m_ignoreSent = registry->ReadBool(REG_IGNORESENT);
            }
            if( registry->ValueExists(REG_MAXDEVICES) )
            {
                m_maxDevices = registry->ReadInteger(REG_MAXDEVICES);
                if( m_maxDevices <= 0 ) m_maxDevices = 100;
            }
            if( registry->ValueExists(REG_DEVICEID) )
            {
                m_deviceId = registry->ReadInteger(REG_DEVICEID);
                if( m_deviceId <= 0 ) m_deviceId = 1;
            }
            if( registry->ValueExists(REG_DEVICETYPE) )
            {
                m_deviceType = registry->ReadString(REG_DEVICETYPE).SubString(1, 6);
                if( m_deviceType.Length() < 6 )
                {
                    m_deviceType.Insert( AnsiString::StringOfChar('0', 6 - m_deviceType.Length()), 1 );
                }
            }
            if( registry->ValueExists(REG_CONNECTTIMEOUT) )
            {
                m_connectTimeout = registry->ReadInteger(REG_CONNECTTIMEOUT);
                if( m_connectTimeout <= 0 ) m_connectTimeout = 5;
            }

            registry->CloseKey();
            delete registry;
            registry = NULL;
        }

        // Initialise the file list if required

        if( m_task.files.empty() )
        {
            UdFile file;        
            TSearchRec record;
            if( FindFirst( m_task.sourcePath + "\\*.devud", faReadOnly, record ) == 0 )
            {
                do
                {
                    if( (record.Attr & faDirectory) == 0 )
                    {
                        if( !(m_ignoreSent && (record.Name.Pos(SENT_STRING) == 1)) )
                        {
                            file.name = record.Name;
                            file.size = record.Size;
                            file.time = FileDateToDateTime( record.Time );

                            m_task.files.push_back( file );
                            m_stats.totalBytes += file.size;
                        }
                    }
                } while( FindNext( record ) == 0 );

                FindClose( record );
            }
        }

        // Sort the ud files by their creation time

        sort( m_task.files.begin(), m_task.files.end() );
        if( (m_task.batchSize == 0) || (m_task.batchSize > (int)m_task.files.size()) )
        {
            m_task.batchSize = m_task.files.size();
        }

        // Initialise ramp settings

        if( m_task.rampType != RampTestDisabled )
        {
            m_ramp.batchSize = m_task.batchSize;
            m_ramp.count     = 0;

            if( m_task.rampType == RampTestByRatio )
            {
                m_task.gradient = (double)m_task.files.size() / m_task.duration;
            }
        }

        m_stats.completed = false;
        m_stats.totalFiles = m_task.files.size();

        // Check if we require use of device simulators. This is the case if the destination
        // provided is not a path (no characters associated with paths).

        m_simulators = (m_task.destination.LastDelimiter("\\/:") == 0);
        if( m_simulators )
        {
            AnsiString msg;
            msg = "Connecting device simulators to " + m_task.destination + ".";
            m_manager->UpdateTask( m_task, msg );

            if( !ConnectDeviceSimulators() )
            {
                m_error = true;
                m_task.state = UdTransferFailed;
                msg = "Failed to connect device simulators to " + m_task.destination + ".";
                m_manager->UpdateTask( m_task, msg );
                return false;
            }
        }

        // Now ready to begin processing. Set the start/end time now as the previous
        // step may have consumed quite a bit of time.

        m_stats.startTime = TDateTime::CurrentDateTime();
        m_stats.endTime = m_stats.startTime + (double)m_task.duration /(24*60*60);
                
        m_task.state = UdTransferInProgress;
        m_manager->UpdateTask( m_task, "Processing started at " + m_stats.startTime + "." );

        return (m_task.batchSize > 0);
    }

    return false;
}
//---------------------------------------------------------------------------

void __fastcall UdTransferThread::Execute()
{
    SetName();

    m_running = true;

    try
    {
        try
        {
            bool complete = false;

            if( InitialiseTask() )
            {
                // Start transferring files, repeat if a recurring task
                do
                {
                    while( m_running && !complete && !m_error )
                    {
                        TDateTime lastBatch = TDateTime::CurrentDateTime();
                        complete = TransferNextBatch();

                        // Check if we have reached the end of the upload period

                        if( TDateTime::CurrentDateTime() >= m_stats.endTime )
                        {
                            complete = true;
                            m_running = false;
                        }
                        else if( m_running && !complete && !m_error )
                        {
                            // Sleep until the next frequency period

                            while( m_running &&
                                   SecondsBetween( TDateTime::CurrentDateTime(), lastBatch ) < m_task.frequency )
                            {
                                Sleep( 500 );
                            }

                            RampTransferBatch();
                        }

                        // Check whether the device simulator threads are still happy

                        if( m_running && !m_error && m_simulators )
                        {
                            CheckDeviceSimulatorThreads();
                        }
                    }

                } while( m_task.recurring && m_running && !m_error );
            }
            else if( !m_error )
            {
                m_error = true;
                m_task.state = UdTransferFailed;
                m_manager->UpdateTask( m_task, "Failed to initialise usage data transfer task." );
            }

            // Update task state if we are successful

            if( !m_error && complete )
            {
                m_task.state = UdTransferCompleted;
                AnsiString msg = "Successfully transferred " + IntToStr( m_stats.processedFiles ) +
                    " files (" +
                    AnsiString::FloatToStrF((long double)m_stats.processedBytes, AnsiString::sffNumber, 18, 0) +
                    " bytes).";
                m_manager->UpdateTask( m_task, msg );
            }

            // Update task statistics

            m_manager->UpdateTaskStatistics( m_task.id, m_stats );
        }
        catch( ... )
        {
            m_task.state = UdTransferFailed;
            m_manager->UpdateTask( m_task, "A fatal exception occurred during UD transfer processing." );
            m_manager->UpdateTaskStatistics( m_task.id, m_stats );
        }
    }
    __finally
    {
        m_running = false;
        m_stats.completed = true;

        // Shutdown device simulators. This is much quicker if we quickly run through
        // each and stop them first.

        vector<DeviceSimulatorThread *>::iterator device = m_devices.begin();
        for( ; device != m_devices.end(); device++ )
        {
            (*device)->Stop();
        }
        for( device = m_devices.begin(); device != m_devices.end(); device++ )
        {
            (*device)->WaitFor();
            delete (*device);
            (*device) = 0;
        }
        m_devices.clear();

        // Rename files that have been sent if required. Don't care if renames fail.

        if( m_renameSent )
        {
            int count = 0;
            AnsiString oldFile, newFile;            
            UdFileItem file = m_task.files.begin();

            while( (++count <= m_stats.processedFiles) &&
                   (file != m_task.files.end()) )
            {
                // Don't prepend SEND_STRING twice

                if( (*file).name.Pos(SENT_STRING) != 1 )
                {
                    oldFile = IncludeTrailingPathDelimiter(m_task.sourcePath) + (*file).name;
                    newFile = IncludeTrailingPathDelimiter(m_task.sourcePath) +
                        AnsiString(SENT_STRING) + (*file).name;

                    MoveFile( oldFile.c_str(), newFile.c_str() );
                }
                file++;
            }
        }
    }
}
//---------------------------------------------------------------------------

void UdTransferThread::RampTransferBatch()
{
    if( m_task.rampType != RampTestDisabled )
    {
        // Check if it is time to start ramping batch sizes

        if( !m_ramp.running )
        {
            m_ramp.running = (TDateTime::CurrentDateTime() >= m_task.rampTime);
        }

        // Calculate the new batch size for the next transfer

        if( m_ramp.running )
        {
            // Calculation is linear function, y = mx + c where:
            //  y = Next batch size
            //  m = Ramp gradient
            //  x = Ramped Upload batch count
            //  c = Batch size for upload

            m_ramp.batchSize = m_task.gradient * (++m_ramp.count) + m_task.batchSize;
        }
    }
}
//---------------------------------------------------------------------------

bool UdTransferThread::TransferNextBatch()
{
    // Find the next file to start a batch on. Account for recurring tasks.

    UdFileItem file = m_task.files.begin();
    advance( file, m_task.recurring
                        ? (m_stats.processedFiles % m_stats.totalFiles)
                        : m_stats.processedFiles );

    AnsiString sourceFile;
    int batchSize = GetBatchSize();
    int count = 0;
    int transferSize = 0;

    // Transfer files until we stop running, we reach the end of all transferrable
    // files, or the batch transfer size is reached

    while( m_running &&
          (file != m_task.files.end()) &&
          (count < batchSize) )
    {
        sourceFile = IncludeTrailingPathDelimiter(m_task.sourcePath) + (*file).name;

        if( TransferUdFile( sourceFile ) )
        {
            m_stats.processedBytes  += (*file).size;
            transferSize            += (*file).size;

            ++m_stats.processedFiles;
            ++file;
            ++count;
        }
        else
        {
            // Abort further processing

            m_error = true;
            m_task.state = UdTransferFailed;
            AnsiString msg = "Failed to transfer file " + (*file).name;
            m_manager->UpdateTask( m_task, msg );
            break;
        }

        // Roll over to the start if this is a recurring transfer

        if( m_task.recurring &&
            (file == m_task.files.end()) )
        {
            file = m_task.files.begin();
        }
    }

    if( !m_error ) m_stats.batches.push_back( transferSize );

    // Return true if there are no more files to transfer

    return (file == m_task.files.end());
}
//---------------------------------------------------------------------------

bool UdTransferThread::TransferUdFile( const AnsiString & filename )
{
    bool transferred = true;

    if( m_simulators )
    {
        // Get the next device simulator to send the UD file

        vector<DeviceSimulatorThread *>::iterator device = m_devices.begin();
        advance( device, ++m_lastDevice );
        if( device == m_devices.end() )
        {
            m_lastDevice = 0;
            device = m_devices.begin();
        }

        // Can't check the state of the transfer as the transfer is handled
        // in another thread. After a batch is sent we check with all the device
        // simulator threads for potential transfer problems

        if( (*device) != NULL )
        {
            (*device)->UploadUDFile( filename );
        }
    }
    else
    {
        // Simply copy the file to the destination directory

        AnsiString destFile = IncludeTrailingPathDelimiter(m_task.destination) +
            ExtractFileName(filename);
        transferred = (CopyFile( filename.c_str(), destFile.c_str(), false ) != 0);
    }

    return transferred;
}
//---------------------------------------------------------------------------

int  UdTransferThread::GetBatchSize()
{
    return m_ramp.running ? m_ramp.batchSize : m_task.batchSize;
}
//---------------------------------------------------------------------------

void __fastcall UdTransferThread::GetTransferStatistics( UdTransferStatistics & stats )
{
    stats.completed      = m_stats.completed;
    stats.startTime      = m_stats.startTime;
    stats.endTime        = m_stats.endTime;
    stats.processedFiles = m_stats.processedFiles;
    stats.totalFiles     = m_stats.totalFiles;
    stats.processedBytes = m_stats.processedBytes;
    stats.totalBytes     = m_stats.totalBytes;
    stats.batches        = m_stats.batches;
}
//---------------------------------------------------------------------------

bool UdTransferThread::ConnectDeviceSimulators()
{
    bool connected = true;

    int number = m_maxDevices;
    if( m_task.batchSize < number )
    {
        number = m_task.batchSize;
    }

    // Create and start simulator threads

    for( int i = 0; i < number; i++ )
    {
        unsigned long deviceId = m_deviceId + i;
        if( deviceId <= 0 ) deviceId = 1;

        AnsiString installationId;
        installationId.sprintf( "??????????-???-%s-%08d", m_deviceType, deviceId );

        IDevice * device = &m_deviceFactory->createMASSDevice(
            m_deviceType.c_str(), deviceId, installationId.c_str() );
        if( device != NULL )
        {
            // Simulator thread object now owns the device
            
            DeviceSimulatorThread * thread =
                new DeviceSimulatorThread( device, m_deviceFactory, m_task.destination );
            if( thread != NULL )
            {
                m_devices.push_back( thread );

                // Wait for the first connection to confirm that the destination host is
                // running a site computer, and that the device configuration is correct.
                // Subsequent disconnections (if any) will be handled by the device thread.

                int count = m_connectTimeout * 2;

                while( (count >= 0) && !thread->IsConnected() )
                {
                    count--;
                    Sleep( 500 );
                }
                if( count == 0 )
                {
                    connected = false;
                    break;
                }

                // Need a slight delay to confirm that the device has connected all channels
                // (CMD/UD/CD) -- if not, we may overlap channels between devices and this can
                // take a long time to sort itself out (lots of device disconnections).

                Sleep( 500 );
            }
        }
        if( !m_running ) break;
    }

    return connected;
}
//---------------------------------------------------------------------------

void UdTransferThread::CheckDeviceSimulatorThreads()
{
    AnsiString error = "";

    // We are expecting all threads to be running here, so a terminal error
    // has occurred if this is not the case

    vector<DeviceSimulatorThread *>::iterator device = m_devices.begin();
    for( ; device != m_devices.end(); device++ )
    {
        if( ((*device) != NULL) &&
            !(*device)->IsRunning() )
        {
            error = (*device)->GetErrorMessage();
            if( !error.IsEmpty() )
            {
                break;
            }
        }
    }

    // Abort further processing

    if( !error.IsEmpty() )
    {
        m_error = true;
        m_task.state = UdTransferFailed;
        m_manager->UpdateTask( m_task, error );
    }
}