/*!
    @ingroup        Restart
    @file           Rst_RedoReadTask.cpp
    @author         UweH
    @brief          defines a singleton to handle a task which reads the log
                    and creates redo files
*/
/*

    ========== licence begin  GPL
    Copyright (c) 2000-2005 SAP AG

    This program is free software; you can redistribute it and/or
    modify it under the terms of the GNU General Public License
    as published by the Free Software Foundation; either version 2
    of the License, or (at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
    along with this program; if not, write to the Free Software
    Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA  02111-1307, USA.
    ========== licence end



*/
#include "ggg92.h"   // used by hkb51.h
#include "heo52.h"   // vclock
#include "hgg01.h"   // g01restart_time
#include "hkb51.h"   // kb51GetRedoTransIndex
#include "hkb50.h"   // k50GetLogTransaction, k50RegisterTransaction
#include "hkb57.h"   // kb57RedoTransNoIncrement
#include "hkb57_1.h" // k57restartrec

#include "RunTime/Synchronisation/RTESync_Spinlock.hpp"
#include "SAPDBCommon/ErrorsAndMessages/SAPDBErr_Assertions.hpp"
#include "SAPDBCommon/Tracing/SAPDBTrace_Usage.hpp"
#include "SAPDBCommon/MemoryManagement/SAPDBMem_IRawAllocator.hpp"
#include "SAPDBCommon/MemoryManagement/SAPDBMem_NewDestroy.hpp"
#include "KernelCommon/Kernel_IAdminRestartShutdown.hpp"
#include "KernelCommon/Kernel_IAdminConfig.hpp"
#include "KernelCommon/Kernel_Migration.hpp"
#include "KernelCommon/Kernel_Version.hpp"
#include "KernelCommon/Kernel_OpMsg.hpp"
#include "Restart/Rst_RedoReadTask.hpp"
#include "Restart/Rst_IRedoTrafficControl.hpp"
#include "Restart/Rst_Exceptions.hpp"
#include "Restart/Rst_LogEntryIterator.hpp"
#include "Logging/Log_VolumeIterator.hpp"
#include "Logging/Log_SeqTapeReader.hpp"
#include "Logging/Log_Volume.hpp"
#include "Logging/Log_Savepoint.hpp"
#include "Logging/Log_AfterImage.hpp"
#include "Logging/Log_ActionPartialRollback.hpp"
#include "Logging/Log_ActionObject.hpp"
#include "Logging/Log_ActionSavepoint.hpp"
#include "Logging/Log_OpenTransFile.hpp"
#include "Logging/Log_Savepoint.hpp"

// #define TEST_DEBUG 1

/// EndOfTrans will check the IOSequence of the inserted job
#define CHECK_ASCENDING_EOT_SEQUENCE            true   // PTS 1113641 UH 2002-01-25 

/// every NUMBER_OF_ENTRIES_FOR_NEXT_WRITE entries, a message will be written in knldiag
/// also it is the checked, whether the restart_time has been expired, so that a savepoint
/// has to be written
#define NUMBER_OF_ENTRIES_FOR_NEXT_WRITE        50000

/// convenient
#define CRASH(TEXT) RTE_Crash( SAPDBErr_Exception(__CONTEXT__, SAPDBERR_ASSERT_STATE_FAILED,TEXT) );
/*!
    @class Log_AfterImageHeadSpace
    @brief This is used to implement the interface Data_ISplitSpace for reading purpose.
 */
class Log_AfterImageHeadSpace : public Data_ISplitSpace
{
public:
	/// returns the number of current part
    virtual PartNo CurrentPart () const
    {
        return 0;
    }
    /// get one part
    virtual void GetPart (PartNo       Part,
                          SAPDB_UInt  &Length,
                          SAPDB_Byte* &Value) const
    {
        Length = m_Length;
        Value  = m_Value;
    }
	/// returns lenght of complete splitspace
    virtual SAPDB_UInt Length() const
    {
        return m_TotalLength;
    }
	/// checks, whether an value is assigned or not
    virtual bool IsAssigned() const
    {
        return m_Value != 0;
    }
    /*!
    @return false, if space begin is not properly aligned
    @brief  This checks if the split space begins at an aligned position
    */
    virtual bool CheckAlignment() const
    {
         if ( ! IsAssigned() )
             return true;
             
         SAPDB_UInt8 address =
                         #ifdef BIT64
                         reinterpret_cast<SAPDB_UInt8>(m_Value);
                         #else
                         reinterpret_cast<SAPDB_UInt4>(m_Value);
                         #endif
         
         return address % LOG_PAGE_ALIGNMENT == 0;
    }

    /// default constructor
    Log_AfterImageHeadSpace ()
    : m_Value       (NULL),
      m_Length      (0),
      m_TotalLength (0)
    {}
    /*!
        @brief  assigns a new space
        @param  iter the logentry, which will be assigned to the Afterimage
     */
    void Assign (const Rst_LogEntryIterator& iter)
    {
        m_Value       = iter.GetValuePtr();
        m_Length      = iter.Length();
        m_TotalLength = iter.RestLength();
    }
private:
    /// pointer to the value of the current entry
    SAPDB_Byte *m_Value;
    /// length of the current entry
    SAPDB_UInt m_Length;
    /// total length of the current entry
    SAPDB_UInt m_TotalLength;
};
/* --------------------------------------------------------------------------- */
void Rst_RedoReadTask::GetReadPosition (Log_EntryInfo &entryInfo) const
{
    SAPDBTRACE_METHOD_DEBUG ("Rst_RedoReadTask::GetReadPosition", LogTrans_Trace, 5);
    m_logIter.GetReadPosition(entryInfo);
}
/* --------------------------------------------------------------------------- */
void Rst_RedoReadTask::ReadLogAndCreateRedoFiles (tgg00_TransContext &readerTrans)
{
    SAPDBTRACE_METHOD_DEBUG ("Rst_RedoReadTask::ReadLogAndCreateRedoFiles", LogTrans_Trace, 5);

    SAPDBTRACE_WRITELN (LogTrans_Trace, 6, "trError: " << readerTrans.trError_gg00);

    if (e_ok != readerTrans.trError_gg00) return;

    SAPDBMem_IRawAllocator  &allocator      = *(reinterpret_cast <SAPDBMem_IRawAllocator*>
                                                (readerTrans.trAllocator_gg00));
    Rst_IRedoTrafficControl &coordinator    = Rst_IRedoTrafficControl::Instance();
    Log_Volume          &log            = Log_Volume::Instance();

    ReCreateOpenTransactions (readerTrans, allocator);
    if ( readerTrans.trError_gg00 != e_ok )
    {
        SAPDBTRACE_WRITELN (LogTrans_Trace, 6, "trError: " << readerTrans.trError_gg00);
        CRASH("ReCreateOpenTransactions failed");
    }
    SAPDBTRACE_WRITELN (LogTrans_Trace, 6, "open trans are re-created");

    m_logIter.Initialize(readerTrans.trTaskId_gg00,
                         &readerTrans.trRteCommPtr_gg00->to_cancel ) ;
    InternReadLogAndCreateRedoFile( readerTrans,
                                    coordinator,
                                    m_logIter,
                                    allocator);

    if ( m_redoIsAborted )
    {
        RTE_Message( Restart_Exception(__CONTEXT__, RST_REDO_ABORTED) );
        coordinator.SetEndOfLogRead (readerTrans.trTaskId_gg00);
        return;
    }

    CloseOpenTransactions (readerTrans, m_logIter.GetLastProcessedIOSeq() );
    coordinator.SetEndOfLogRead (readerTrans.trTaskId_gg00);

    SAPDBTRACE_WRITELN (LogTrans_Trace, 6, "TrError: " << readerTrans.trError_gg00);
}
/* --------------------------------------------------------------------------- */
void Rst_RedoReadTask::InternReadLogAndCreateRedoFile(tgg00_TransContext      &readerTrans,
                                    Rst_IRedoTrafficControl &coordinator,
                                    Rst_LogEntryIterator    &iter,
                                    SAPDBMem_IRawAllocator  &allocator)
{
    SAPDBTRACE_METHOD_DEBUG ("Rst_RedoReadTask::InternReadLogAndCreateRedoFile", LogTrans_Trace, 5);

    Log_AfterImageHeadSpace  entryHeadSpace;
    Log_AfterImage           afterImage;
    bool                     isOK            = true;
    bool                     checkForSavepointEntry
                             = Log_Savepoint::GetLastSavepointEntryType() != Log_NoOp; // for migration
    bool                     nextEntryMustBeFinishOpenTrans = false; // PTS 1115065 UH 2002-04-10
    bool                     endOfTransAlreadyProcessed = false; // PTS 1123461 UH 2003-08-28
    tgg00_TransIndex         currTransIndex;
    Log_Transaction         *pTransaction    = NULL;
    SAPDB_UInt               readEntryCount  = 0;
    SAPDB_Int                timeInSec;
    SAPDB_Int                dummyTimeInMSec;
    SAPDB_Int                lastSavepointTriggerTime;
    Log_IOSequenceNo         lastIOSequence;

    vclock (&lastSavepointTriggerTime, &dummyTimeInMSec);

    while ( iter.GetState() == Log_VolumeIterator::Iter_ok )
    {
        iter.CheckSuspendForStandby(readerTrans.trTaskId_gg00);
        entryHeadSpace.Assign           (iter);
        Data_SplitSpaceReader reader    (entryHeadSpace);
        afterImage.ReadPersistentFormat (reader, isOK);

        if ( m_redoIsAborted )
        {
            RTE_Message( Restart_Exception(__CONTEXT__, RST_REDO_ABORTED) );
            coordinator.SetEndOfLogRead (readerTrans.trTaskId_gg00);
            return;
        }

        if ( ! isOK )
        {
            iter.WriteToTrace ("LogEntryIterator");
            CRASH("cannot read afterimage");
        }

        if ( checkForSavepointEntry )
        {
            checkForSavepointEntry = false;
            if ( afterImage.GetActionType() != Log_Savepoint::GetLastSavepointEntryType() )
            {
                Kernel_VTrace() << "LastSavepointEntryType: "
                                << Log_GetActionTypeString(Log_Savepoint::GetLastSavepointEntryType());
                Kernel_VTrace() << "readlog: T" << afterImage.GetTransNo().gg90GetInt4()
                                <<" (" << afterImage.GetSequence() << ") '"
                                << Log_GetActionTypeString(afterImage.GetActionType()) << "' ["
                                << iter.IOSeqNo() << "]\n";
                CRASH("redo begin not found (look into vtrace)");
            }
        }
        
        if ( LogTrans_Trace.TracesLevel(7) )
            Kernel_VTrace() << "readlog: T" << afterImage.GetTransNo().gg90GetInt4()\
                            <<" (" << afterImage.GetSequence() << ") '" \
                            << Log_GetActionTypeString(afterImage.GetActionType()) << "' ["
                            << iter.IOSeqNo() << "]\n";

        // PTS 1115065 UH 2002-04-10 begin
        if ( nextEntryMustBeFinishOpenTrans )
        {
            if( afterImage.GetActionType() != Log_FinishOpenTrans )
            {
                // the previous log entry was truncated
                // this can only happen if the kernel crashed and restarted and now the restart begins
                // at an earlier time. (e.g. restore data)
                Kernel_VTrace() << "readlog: T" << afterImage.GetTransNo().gg90GetInt4()
                                <<" (" << afterImage.GetSequence() << ") '"
                                << Log_GetActionTypeString(afterImage.GetActionType()) << "' ["
                                << iter.IOSeqNo() << "]\n";
                iter.WriteToTrace ("LogEntryIterator");
                CRASH("previous logentry was corrupt");
            }
            nextEntryMustBeFinishOpenTrans = false;
        }
        // PTS 1115065 UH 2002-04-10 end

        lastIOSequence = iter.IOSeqNo();

        if ( ! afterImage.GetTransNo().gg90IsNil() )
        {
            // PTS 1135302 UH 2005-04-29 Moved the skip after the checks above
            // PTS 1134103 UH 2005-03-07 skip already processed transactions
            if ( m_TransactionsToSkip.Includes(afterImage.GetTransNo()) )
            {
                #ifdef SAPDB_FAST
                if ( LogTrans_Trace.TracesLevel(6) )
                #endif
                    iter.Message ( "entry skipped, because", readEntryCount,
                                   afterImage.GetTransNo().gg90GetInt4(),
                                   afterImage.GetSequence() );
                ++iter;
                continue;
            }
            
            pTransaction = GetOrCreateRedoTransaction ( readerTrans,
                                                        allocator,
                                                        afterImage.GetTransNo(),
                                                        currTransIndex );
        }
        else
        {
            // PTS 1133929 mb 2005-18-02 (savepoints entries)
            pTransaction = NULL;
        }
        

        // The read task builds then redo files with a correct transid.
        readerTrans.trWriteTransId_gg00 = afterImage.GetTransNo();
        readerTrans.trIndex_gg00        = currTransIndex;

        // save current read entry position for redo-savepoint
        iter.SetCurrentEntry (afterImage.GetActionType());
        ++readEntryCount;
        
        // write progress messages
        if ( LogTrans_Trace.TracesLevel(9)
             ||
             (readEntryCount % NUMBER_OF_ENTRIES_FOR_NEXT_WRITE == 0) )
             iter.Message ( "redo-read", readEntryCount,
                                     afterImage.GetTransNo().gg90GetInt4(),
                                     afterImage.GetSequence() );

        if ( readEntryCount % NUMBER_OF_ENTRIES_FOR_NEXT_WRITE == 0 )
        {
            vclock (&timeInSec, &dummyTimeInMSec);
            if ( lastSavepointTriggerTime + g01restart_time <= timeInSec )
            {
                lastSavepointTriggerTime = timeInSec;
                Log_SavepointManager.StartSavepoint (readerTrans, Log_SVPReasonDistance);
            }
        }
        
        endOfTransAlreadyProcessed = false; // PTS 1123461 UH 2003-08-28
        
        if ( pTransaction != 0
             &&
             pTransaction->GetTransType() != Log_Transaction::redoRead )
        {
			// PTS 1123461 UH 2003-08-28 new if-clause
            // The context is already given to the redo tasks.
            // Do not insert it again to the redo traffic control.
            // But check the correct EndOfTrans handling.
            if ( (pTransaction->GetTransState() == Log_Transaction::committed
                  &&
                  afterImage.GetActionType() != Log_Commit)
                  /* PTS 1132673 UH 2004-11-25
                     if trans is marked as rollbacked the next entry may be of any kind
                     but it must be the last entry in the log !!!
                 ||
                 (pTransaction->GetTransState() == Log_Transaction::rollbacked
                  &&
                  afterImage.GetActionType() != Log_Rollback)
                 */
                 ||
                 (pTransaction->GetTransState() == Log_Transaction::opened) )
            {
                Kernel_VTrace() << "Found different end of transaction";
                iter.WriteToTrace ("LogEntryIterator");
                afterImage.WriteToTrace ("CurrentAction");
                pTransaction->WriteToTrace ("CurrentTrans");
                CRASH("Found different end of transaction");
            }
            endOfTransAlreadyProcessed = true;
        }
        
        switch ( afterImage.GetActionType() )
        {
        case Log_Commit:
        {
            SAPDBERR_ASSERT_STATE( iter.RestLength() == iter.Length() );
            Log_ActionEndOfTransaction commitAction;
        
            if (iter.RestLength() == (commitAction.GetPersistentLength() + Log_AfterImage::GetPersistentLength()))
            // PTS 1124045 mb 2003-09-08 commit-logentry with timestamp for point-in-time recovery
            {
                commitAction.ReadPersistentFormat (reader, allocator, isOK);
                if ( ! isOK )
                    CRASH("read Log_ActionEndOfTransaction");
            }
            
            if ( LogTrans_Trace.TracesLevel(7) )
            {
                commitAction.WriteToTrace ("EndOfRedoTrans");
            }
            
            if ( ! iter.UntilDateIsReached(commitAction.GetCommitDate(),commitAction.GetCommitTime()))
            {
                if ( ! endOfTransAlreadyProcessed ) // PTS 1123461 UH 2003-08-28
                {
                    EndOfTransaction ( readerTrans.trTaskId_gg00,
                                       allocator,
                                       pTransaction,
                                       afterImage.GetActionType() == Log_Commit
                                       ? Log_Transaction::committed
                                       : Log_Transaction::rollbacked,
                                       iter.IOSeqNo(),
                                       currTransIndex,
                                       CHECK_ASCENDING_EOT_SEQUENCE ); // PTS 1113641 UH 2002-01-25
                }
            }
            break;
        }
        case Log_Rollback:
            SAPDBERR_ASSERT_STATE( iter.RestLength() == iter.Length() );

            if ( ! endOfTransAlreadyProcessed ) // PTS 1123461 UH 2003-08-28
            {
                EndOfTransaction ( readerTrans.trTaskId_gg00,
                                   allocator,
                                   pTransaction,
                                   afterImage.GetActionType() == Log_Commit
                                   ? Log_Transaction::committed
                                   : Log_Transaction::rollbacked,
                                   iter.IOSeqNo(),
                                   currTransIndex,
                                   CHECK_ASCENDING_EOT_SEQUENCE ); // PTS 1113641 UH 2002-01-25
            }
            break;
        case Log_FinishOpenTrans:
        {
            SAPDBERR_ASSERT_STATE( iter.RestLength() == iter.Length() );

            // PTS 1121011 mb 2003-03-11 extract kernel-version from savepoint
            Log_ActionSavepoint SVPAction;
            SVPAction.ReadPersistentFormat (reader, allocator, isOK);
            if (SVPAction.HasKernelVersionString())
            {
                if ( ! Kernel_IAdminConfig::Instance().MigrationHandler().
                            CheckCompatibility( Kernel_Migration::log,
                                                Kernel_Version(SVPAction.GetKernelVersionString()) ) )
                {
                    iter.WriteToTrace ("LogEntryIterator");
                    CRASH("restartsavepoint with too new Softwareversion");
                }
            }
            CloseOpenTransactions (readerTrans, lastIOSequence);
            break;
        }
        case Log_PartialRollback:
        {
            SAPDBERR_ASSERT_STATE( iter.RestLength() == iter.Length() );
            Log_ActionPartialRollback Action;
            Action.ReadPersistentFormat (reader, allocator, isOK);
            if ( ! isOK )
                CRASH("read PartialRollback failed");
            if ( Action.GetRedoStopSequence() // PTS 1111 UH 2000-08-10
                 ==
                 afterImage.GetSequence() )
            {
                // this should be prevented by Log_Transaction,
                // but for more safety it is checked here again
                SAPDBTRACE_WRITELN (LogTrans_Trace, 7, "truncate: stopseq: "
                                                       << Action.GetRedoStopSequence()
                                                       << ", nothing to do.");
                break;
            }
            SAPDBTRACE_WRITELN (LogTrans_Trace, 7, "truncate: inclusive stopseqence: "
                                                   << Action.GetRedoStopSequence());
            // PTS 1113230 UH 2001-12-20 begin
            pTransaction->RedoPartialRollback ( Action.GetUndoStopSequence(),
                                                Action.GetRedoStopSequence() );
            // PTS 1113230 UH 2001-12-20 end
            break;
        }
        case Log_SavepointEntry:
            SAPDBERR_ASSERT_STATE( iter.RestLength() == iter.Length() );
             // nothing to do
            break;
        default:
            nextEntryMustBeFinishOpenTrans = // PTS 1115065 UH 2002-04-10
                ! CopyEntry (*pTransaction, iter, afterImage, readerTrans);
            if ( nextEntryMustBeFinishOpenTrans )
            {
                if ( LogTrans_Trace.TracesLevel(6) )
                {
                    Kernel_VTrace() << "nextEntryMustBeFinishOpenTrans:\n";
                    iter.WriteToTrace ("read iterator");
                    if ( iter.GetState() == Log_VolumeIterator::Iter_ok )
                        Kernel_VTrace() << "readlog: T" << afterImage.GetTransNo().gg90GetInt4()
                                        <<" (" << afterImage.GetSequence() << ") '"
                                        << Log_GetActionTypeString(afterImage.GetActionType()) << "' ["
                                        << iter.IOSeqNo() << "]\n";
                }
            }
            break;
        }//switch

        if ( Log_VolumeIterator::Iter_ok == iter.GetState() )
        {
            // PTS 1115065 UH 2002-04-10 begin
            // if current entry was truncated and the current queue io sequence is 0
            // then the log writing was beginning here with a restart and the first
            // log entry must be the FinishOpenTrans entry.
            // So the iterator must not be incremented because it is pointing to the beginning.
            if ( ! nextEntryMustBeFinishOpenTrans
                 ||
                 iter.QueueSeqNo() != MIN_IOSEQUENCE )
                ++iter;
            // PTS 1115065 UH 2002-04-10 end
        }
    } //endwhile
    iter.Message ( "last-redo-read", readEntryCount,
                             afterImage.GetTransNo().gg90GetInt4(),
                             afterImage.GetSequence() );
}
/* --------------------------------------------------------------------------- */
void Rst_RedoReadTask::CloseOpenTransactions (tgg00_TransContext &redoReadTrans,
                                              Log_IOSequenceNo    lastIOSequence)
{
    SAPDBTRACE_METHOD_DEBUG ("Rst_RedoReadTask::CloseOpenTransactions", LogTrans_Trace, 5);

    tgg00_TransIndex        openTransIndex       = cgg_nil_transindex;
    Log_Transaction        *pTransaction         = NULL;
    SAPDBMem_IRawAllocator &allocator            =
        *(reinterpret_cast <SAPDBMem_IRawAllocator*>(redoReadTrans.trAllocator_gg00));

    if ( lastIOSequence.IsInvalid() )
    {
        // PTS 1124684 UH 2003-10-16
        lastIOSequence = Rst_IRedoTrafficControl::Instance().
            GetLastIOSequence(redoReadTrans.trTaskId_gg00);
        if ( lastIOSequence.IsInvalid() )
            lastIOSequence  = 0;
    }
    
    kb51GetFirstOpenRedoTrans (redoReadTrans.trTaskId_gg00, openTransIndex);

    while ( openTransIndex != cgg_nil_transindex )
    {
        pTransaction = REINTERPRET_CAST(Log_Transaction*,k50GetLogTransaction (openTransIndex));
        if ( NULL == pTransaction )
            CRASH("k50GetLogTransaction() returns NULL");

        if ( pTransaction->GetRedoEotSequence().IsValid() )
            CRASH("eot sequence is valid");

        EndOfTransaction ( redoReadTrans.trTaskId_gg00,
                           allocator,
                           pTransaction,
                           Log_Transaction::rollbacked,
                           lastIOSequence,
                           openTransIndex,
                           ! CHECK_ASCENDING_EOT_SEQUENCE ); // PTS 1113641 UH 2002-01-25

        // force search at the beginning of the transactions list
        // until there exists no more open transaction
        openTransIndex = cgg_nil_transindex;
        kb51GetFirstOpenRedoTrans (redoReadTrans.trTaskId_gg00, openTransIndex);
    }//while
    // PTS 1129746 UH 2004-06-18 added wait for all transactions which end at this time
    Rst_IRedoTrafficControl::Instance().WaitForAllRedoTasks(redoReadTrans.trTaskId_gg00);
}
/* --------------------------------------------------------------------------- */
void Rst_RedoReadTask::ReCreateOpenTransactions (tgg00_TransContext     &redoReaderTrans,
                                                 SAPDBMem_IRawAllocator &allocator)
{
    SAPDBTRACE_METHOD_DEBUG ("Rst_RedoReadTask::ReCreateOpenTransactions", LogTrans_Trace, 5);

    tkb00_SaveptParam &saveptParam = k57restartrec->rstLastSavept_kb00();

    Log_OpenTransFile file ( redoReaderTrans, saveptParam.svpOpenTransRoot_kb00 );

    if ( ! file.IsCreated() )
        return;
    
    m_OldestKnownEOTSequence = file.GetOldestKnownIOSequence(); // PTS 1124684 UH 2003-10-16
    if ( LogTrans_Trace.TracesLevel(6) )
        Kernel_VTrace() << "ReCreateOpenTransactions: opentransfile.GetOldestKnownIOSequence() "
                        << m_OldestKnownEOTSequence << NewLine ;
    if (m_OldestKnownEOTSequence.IsInvalid()) 
    // for migration or if no transactions had been open during the savepoint
    {
        m_OldestKnownEOTSequence = saveptParam.svpIOsequence_kb00; // PTS 1127009 UH 2004-01-09, PTS 1127704
        if ( LogTrans_Trace.TracesLevel(6) )
            Kernel_VTrace() << "ReCreateOpenTransactions: m_OldestKnownEOTSequence was nil now set to "
                            << m_OldestKnownEOTSequence << NewLine;
    }
    // PTS 1114727 mb 2003-11-04 corrected settting of OldestKnownEOTSequence 
    // PTS 1127009 UH 2004-01-09 update of oldestknown for Log_Volume removed

    Log_OpenTransFile::Iterator &iter = file.GetFirstOpenTrans(saveptParam.svpId_kb00);

    tgg00_TransIndex currentTransIndex = cgg_nil_transindex;
    tgg00_BasisError trError           = e_ok;

    // PTS 1134103 UH 2005-03-07
    if ( ! m_TransactionsToSkip.Init ( allocator, Rst_IRedoTrafficControl::Instance().GetMaximumNumberOfTransactions() ) )
    {
        RTE_Crash ( Restart_Exception(__CONTEXT__, SAPDBERR_ASSERT_STATE_FAILED, "m_TransactionsToSkip.Init() failed") );
    }

    const Kernel_Migration::ChangeInfo &repeatedRedoFlag = Kernel_IAdminConfig::Instance().MigrationHandler().SearchForChangeInfo ("RepeatedRedoFlagInOpenTransFile");
    const bool lastSavepointDoneDuringRedo               = repeatedRedoFlag.IsEnabled() ? false : file.CreatedDuringRedo();
          bool dontReadLogOfTransAgain;
          
    if ( lastSavepointDoneDuringRedo )
        Kernel_OpInfo (csp3_n_restart) << "Previous restart was interrupted.";

    while ( iter.IsValid() )
    {
        Log_Transaction* newtrans = new (allocator) Log_Transaction ();

        if ( newtrans == NULL )
            CRASH("allocate failed");

        newtrans->ReadPersistentFormat( *iter, file.GetSpaceSize() );

        if ( newtrans->GetRedoTransNo().gg90IsNil() )
        {
            // the savepoint may collect such trans entries
            destroy (newtrans, allocator);
            ++iter;
            continue;
        }
        
        dontReadLogOfTransAgain = ! lastSavepointDoneDuringRedo && newtrans->GetTransState() == Log_Transaction::committed;
        
        if ( dontReadLogOfTransAgain )
        {
            // This trans is definitly complety done so ignore every log entry which is found in the future
            if ( ! m_TransactionsToSkip.Add ( newtrans->GetRedoTransNo() ) ) // PTS 1134103 UH 2005-03-07
                CRASH("m_TransactionsToSkip.Add() failed");

            // The transaction is prepared to be completly removed with all its files
            if ( ! newtrans->InitializeForRedo (redoReaderTrans) )
                CRASH("InitializeForRedo for delete failed");

            // delete the log transaction context
            redoReaderTrans.trWriteTransId_gg00 = newtrans->GetRedoTransNo();
            newtrans->DeleteAfterRedo();
            redoReaderTrans.trWriteTransId_gg00.gg90SetNil();

            destroy (newtrans, allocator);

            // skip to next open trans entry
            ++iter;
            continue;
        }
        
        if ( ! newtrans->InitializeForRedoCopy (redoReaderTrans,
                                                newtrans->GetRedoTransNo(),
                                                allocator) )
            CRASH("InitializeForRedoCopy failed");

        newtrans->UpdateUndoRedoInfoForRestart(m_OldestKnownEOTSequence);
        
        kb51GetRedoTransIndex (redoReaderTrans.trTaskId_gg00,
                               newtrans->GetRedoTransNo(),
                               currentTransIndex,
                               trError);

        if ( trError != e_ok )
            CRASH("kb51GetRedoTransIndex failed");
        k50RegisterTransaction (currentTransIndex, newtrans);

        if ( LogTrans_Trace.TracesLevel(6) )
            newtrans->WriteToTrace("OpenTrans");

        kb57RedoTransNoIncrement ( redoReaderTrans.trTaskId_gg00, // PTS 1127838 UH 2004-02-20 added
                                   newtrans->GetRedoTransNo() );

        if ( dontReadLogOfTransAgain // PTS 1138299 UH 2005-10-12
             ||
             (newtrans->GetTransState() != Log_Transaction::opened
              &&
              newtrans->GetRedoEotSequence().IsValid()) )
        {
            EndOfTransaction ( redoReaderTrans.trTaskId_gg00,
                               allocator,
                               newtrans,
                               newtrans->GetTransState(),
                               newtrans->GetRedoEotSequence(),
                               currentTransIndex,
                               ! CHECK_ASCENDING_EOT_SEQUENCE ); // PTS 1113641 UH 2002-01-25
        }

        ++iter;
    }
	
    if ( LogTrans_Trace.TracesLevel(6) )
        Rst_IRedoTrafficControl::Instance().
            WriteToTrace("End of Rst_RedoReadTask::ReCreateOpenTransactions()");

    // PTS 1134103 UH 2005-03-07
    if ( ! Rst_IRedoTrafficControl::Instance().GetTransactions(redoReaderTrans.trTaskId_gg00,
			                                       m_TransactionsToSkip) )
        CRASH("GetTransactions failed");

    Rst_IRedoTrafficControl::Instance().EnableRedoProcessing(redoReaderTrans.trTaskId_gg00);
}
/* --------------------------------------------------------------------------- */
Log_Transaction* Rst_RedoReadTask::GetOrCreateRedoTransaction
    ( tgg00_TransContext     &readerTrans,
      SAPDBMem_IRawAllocator &allocator,
      tgg91_TransNo           transno,
      tgg00_TransIndex       &currTransIndex )
{
    SAPDBTRACE_METHOD_DEBUG ("Rst_RedoReadTask::GetOrCreateRedoTransaction", LogTrans_Trace, 5);

    tgg00_BasisError &trError        = readerTrans.trError_gg00;
    tsp00_TaskId     &readerTaskId   = readerTrans.trTaskId_gg00;
    Log_Transaction  *pTransaction   = NULL;

    currTransIndex = cgg_nil_transindex;

    kb51GetRedoTransIndex ( readerTaskId, transno, currTransIndex, trError );

    if ( trError != e_ok || currTransIndex == cgg_nil_transindex )
    {
        SAPDBTRACE_WRITELN (LogTrans_Trace, 6, "trError: " << trError);
        CRASH("kb51GetRedoTransIndex failed");
    }

    pTransaction = reinterpret_cast<Log_Transaction*>
                   (k50GetLogTransaction (currTransIndex));

    SAPDBTRACE_WRITELN (LogTrans_Trace, 6, "TransNo: " << transno.gg90GetInt4() << \
                                      " => TransIndex: " << currTransIndex << \
                                      (NULL==pTransaction?" NEW":""));

    if ( NULL == pTransaction )
    {
        pTransaction = new (allocator) Log_Transaction ();
        if ( pTransaction == NULL )
            CRASH("allocate failed");

        k50RegisterTransaction (currTransIndex, pTransaction);

        bool isOK = pTransaction->InitializeForRedoCopy
                                  ( readerTrans, transno, allocator );
        if ( ! isOK )
            CRASH("InitializeForRedoCopy failed");

        kb57RedoTransNoIncrement (readerTaskId, transno);
    }

    return pTransaction;
}
/* --------------------------------------------------------------------------- */
bool Rst_RedoReadTask::CopyEntry (Log_Transaction        &transaction, // PTS 1115065 UH 2002-04-10 return bool
                                  Rst_LogEntryIterator   &iter,
                                  Log_AfterImage         &afterimage,
                                  tgg00_TransContext     &readerTrans)
{
    SAPDBTRACE_METHOD_DEBUG ("Rst_RedoReadTask::CopyEntry", LogTrans_Trace, 5);
    SAPDBTRACE_WRITELN (LogTrans_Trace, 7, "CopyEntry: length:" << iter.RestLength() <<
                                           ", minimal: " << GetMinimalLengthToCopyEntry (afterimage.GetActionType()));

    // PTS 1127732 UH 2004-02-11 added SkipCurrentEntry 
    // If SkipCurrentEntry is true then nothing is to do - this is a redo of a redo.
    // This entry was already copied into redo file.
    // PTS 1115065 UH 2002-04-10
    const bool SkipCurrentEntry = transaction.GetLastCopiedRedoEntrySequence().IsValid()
                                  &&
                                  afterimage.GetSequence() <= transaction.GetLastCopiedRedoEntrySequence();
    // PTS 1131107 UH 2004-08-24 added handling for Log_MultipleActions
    const bool isMultipleAction = Log_MultipleActions == afterimage.GetActionType();
    Log_AfterImageHeadSpace dummySplitSpace;
    Data_SplitSpaceWriter   Writer =
                                SkipCurrentEntry          // PTS 1127732 UH 2004-02-11
                                ? Data_SplitSpaceWriter(dummySplitSpace)   // PTS 1127732 UH 2004-02-11 use dummy writer
                                : isMultipleAction
                                  ? transaction.GetWriterForLocalRedoEntries
                                          (iter.GetValuePtr() + Log_AfterImage::GetHeaderSize(),
                                           iter.RestLength() - Log_AfterImage::GetHeaderSize())
                                  : transaction.ReserveSpaceForRedoEntryCopy
                                            ( iter.RestLength(),
                                              GetMinimalLengthToCopyEntry (afterimage.GetActionType()),
                                              afterimage.GetSequence(),
                                              iter.IOSeqNo() );


    Data_SplitSpaceWriter::Result writeResult      = Data_SplitSpaceWriter::ok;
    bool                          loopagain        = false;
    bool                          entryInterrupted = false; // PTS 1125914 UH 2003-12-03

    do
    {
        if ( loopagain
             &&
             iter.GetState() == Log_VolumeIterator::Iter_ok )
        {
            ++iter;
            // PTS 1125914 UH 2003-12-03 next if clause
            if ( iter.GetState() == Log_VolumeIterator::Iter_ok
				 &&
				 iter.QueueSeqNo() == MIN_IOSEQUENCE )
            {
                // The current entry was interrupted.
                entryInterrupted = true;
            }
        }
        
        if ( entryInterrupted
             ||
             iter.GetState() != Log_VolumeIterator::Iter_ok )
        {
            loopagain = false;
            if ( LogTrans_Trace.TracesLevel(6) )
			{
                Kernel_VTrace() << "INFO: *** LOG IS TRUNCATED ***";
            	iter.WriteToTrace("iterator");
				Writer.WriteToTrace("writer");
			}
            writeResult = Data_SplitSpaceWriter::ok;
            if ( ! SkipCurrentEntry // PTS 1127732 UH 2004-02-11
                 &&
                 ! isMultipleAction )
                transaction.ReleaseSpaceAfterRedoEntryCopy(false); // release resources unchanged
            break;
        }

        if ( SkipCurrentEntry ) // PTS 1127732 UH 2004-02-11
            loopagain = iter.RestLength() > iter.Length();
        else
        {
            if ( ! loopagain && isMultipleAction )
            {
                const SAPDB_UInt firstTimeSkipLength = Log_AfterImage::GetHeaderSize()
                                                       +
                                                       Log_ActionMultiple::GetHeaderSize();
                if ( firstTimeSkipLength < iter.Length() )
                {
                    SAPDBTRACE_WRITELN (LogTrans_Trace, 7, "CopyEntry: write: length = " << iter.Length() << " - " << firstTimeSkipLength);
                    Writer.Write ( iter.GetValuePtr() + firstTimeSkipLength,
                                   iter.Length()      - firstTimeSkipLength, writeResult );
                }
                else
                {
                    SAPDBTRACE_WRITELN (LogTrans_Trace, 7, "CopyEntry: write: NO WRITE firstTimeSkipLength:" \
                                                            << firstTimeSkipLength << " < iter.Length():" \
                                                            << iter.Length());
                    writeResult = Data_SplitSpaceWriter::moreSpaceAvailable;
                }
            }
            else   
            {
                SAPDBTRACE_WRITELN (LogTrans_Trace, 7, "CopyEntry: write: length = " << iter.Length());
                Writer.Write (iter.GetValuePtr(), iter.Length(), writeResult);
            }

            if ( writeResult != Data_SplitSpaceWriter::moreSpaceAvailable
                 &&
                 writeResult != Data_SplitSpaceWriter::ok )
            {
                Kernel_VTrace() << "writeResult: " << writeResult;
                iter.WriteToTrace("iterator");
                Writer.WriteToTrace("writer");
                CRASH("Writer.Write failed");
            }
            loopagain = writeResult == Data_SplitSpaceWriter::moreSpaceAvailable
                        &&
                        iter.RestLength() > iter.Length();
        }
        
    }
    while ( loopagain );
    
    if ( Data_SplitSpaceWriter::ok == writeResult
         &&
         ! entryInterrupted )
    {
        if ( ! SkipCurrentEntry ) // PTS 1127732 UH 2004-02-11
        {
            if ( ! isMultipleAction )
                transaction.ReleaseSpaceAfterRedoEntryCopy(true);
            else
            {
                if ( iter.GetState() != Log_VolumeIterator::Iter_ok )
                {
                    transaction.ResetLocalRedoEntries();
                    return false;
                }
                if (! transaction.RestoreLocalRedoEntries (iter.IOSeqNo()))
                {
                    Writer.WriteToTrace("writer");
                    iter.WriteToTrace ("read iterator");
                    RTE_Crash(Log_Exception(__CONTEXT__,SAPDBERR_ASSERT_STATE_FAILED,"transaction.RestoreLocalRedoEntries"));
                }
            }
        }
        return true;
    }
    // PTS 1115065 UH 2002-04-10 no abort and no change
    if ( LogTrans_Trace.TracesLevel(6) )
    {
        Kernel_VTrace() << "INFO: *** LOGENTRY TRUNCATED ***, writeResult: " << writeResult;
        iter.WriteToTrace("iterator");
    }
    if ( ! entryInterrupted
         &&
         ! SkipCurrentEntry // PTS 1127732 UH 2004-02-11
         &&
         ! isMultipleAction ) 
        transaction.ReleaseSpaceAfterRedoEntryCopy(false);
    return false;
}
/* --------------------------------------------------------------------------- */
void Rst_RedoReadTask::EndOfTransaction
        ( tsp00_TaskId                 redoReaderTaskid,
          SAPDBMem_IRawAllocator      &allocator,
          Log_Transaction*            &pTransaction,
          Log_Transaction::TransState  state,
          Log_IOSequenceNo             eotSequence,
          tgg00_TransIndex             transindex,
          bool                         checkAscendingSequence ) // PTS 1113641 UH 2002-01-25
{
    SAPDBTRACE_METHOD_DEBUG ("Rst_RedoReadTask::EndOfTransaction", LogTrans_Trace, 5);

    pTransaction->SetEOTInRedo ( state, eotSequence );

    if ( LogTrans_Trace.TracesLevel(7) )
        pTransaction->WriteToTrace ("EndOfRedoTrans");

    // PTS 1114946 UH 2002-03-21 begin
    bool             transIsAlreadyInRedoList = false;
    Log_Transaction* pPublicTrans = reinterpret_cast<Log_Transaction*>(k50GetLogTransaction(transindex));

    if ( NULL == pPublicTrans )
        CRASH("no public Log_Transaction");

    if ( pPublicTrans != pTransaction )
    {
        // there is another log transcontext registered in the public transaction entry in locklist
        // Check if this correct.
        if ( pTransaction->GetTransState()      != pPublicTrans->GetTransState()
             ||
             pTransaction->GetRedoEotSequence() != pPublicTrans->GetRedoEotSequence() )
        {
            pTransaction->WriteToTrace ("pTransaction");
            pPublicTrans->WriteToTrace ("pPublicTrans");
            CRASH("transaction end found twice");
        }
        transIsAlreadyInRedoList = true;
    }
    else
    {
        // 2004-02-18 UH added
        // This can happen if trans was open in last redo savepoint and last entry read was the commit.
        // So ReCreate has already inserted the transaction into the redo list and after that
        // the redo reader finds the commit.
        // In this case the transaction need not to be inserted.
        transIsAlreadyInRedoList =
            pPublicTrans->GetTransState() == Log_Transaction::illegalTransType;
    }
    // PTS 1114946 UH 2002-03-21 end
    
    bool           isOK    = true;
    Log_ActionType eottype = state ==  Log_Transaction::committed
                                       ? Log_Commit
                                       : Log_Rollback; // <- default

    if ( (eottype == Log_Commit
          &&
          ! pTransaction->EntriesToRedoExist())
         ||
         (eottype == Log_Rollback
          &&
          ! pTransaction->WasOpenInLastSavepoint()) )
    {
        tgg00_TransState dummyState;
        dummyState.clear();
        pTransaction->Delete();
        k51remove_locks ( redoReaderTaskid, transindex, dummyState );
        k51del_transentry_from_locklist ( redoReaderTaskid, transindex );
    }
    else
    {
        pTransaction->DeleteAfterRedoCopy();

        if ( ! transIsAlreadyInRedoList ) // PTS 1114946 UH 2002-03-21
            Rst_IRedoTrafficControl::Instance().
                InsertJob ( redoReaderTaskid,
                            transindex, eotSequence, eottype,
                            m_OldestKnownEOTSequence, // PTS 1124684 UH 2003-10-16
                            checkAscendingSequence,   // PTS 1113641 UH 2002-01-25
                            *pTransaction,
                            isOK );
        if ( ! isOK )
        {
            Rst_IRedoTrafficControl::Instance().WriteToTrace("Rst_RedoReadTask::EndOfTransaction: InsertJob failed");
            CRASH("InsertJob failed");
        }
    }

    allocator.Deallocate(pTransaction);
    pTransaction = NULL;
}
/* --------------------------------------------------------------------------- */
SAPDB_UInt Rst_RedoReadTask::GetMinimalLengthToCopyEntry (Log_ActionType type)
{
    SAPDB_UInt length = Log_AfterImage::GetPersistentLength();
    switch (type)
    {
        case Log_NewObject:
        case Log_InsertObject:
        case Log_UpdateObject:
        case Log_DeleteObject:
            length += Log_ActionObject::GetMinimalLengthToCopyEntry();
            break;
        case Log_PartialRollback:
            length += Log_ActionPartialRollback::GetMinimalLengthToCopyEntry();
            break;
        case Log_SavepointEntry:
        case Log_FinishOpenTrans:
            length += Log_ActionSavepoint::GetMinimalLengthToCopyEntry();
            break;
        case Log_MultipleActions:
            length += Log_ActionMultiple::GetMinimalLengthToCopyEntry();
            break;
        default:
            break;
    }
    return length;
}
