/*!
  @file           Converter_Converter.cpp
  @author         TorstenS
  @author         AlexanderK
  @ingroup        Converter
  @brief          Implementation of all converter methods
*/

/*
    ========== licence begin  GPL
    Copyright (c) 2001-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

*/


/*===========================================================================*
 *  INCLUDES                                                                 *
 *===========================================================================*/

#include "hgg01.h"    // g01is_livecache
#include "hgg08.h"    // g08convert0
#include "hbd20_6.h"  // bd20GetNumberOfChangedPages
#include "hbd91.h"    // garbage collector
#include "Converter/Converter_Converter.hpp"
#include "Converter/Converter_Dump.hpp"
#include "Converter/Converter_LeafPage.hpp"
#include "Converter/Converter_Exception.hpp"
#include "Converter/Converter_MapError.hpp"
#include "Converter/Converter_Messages.hpp"
#include "Converter/Converter_PageFrameAllocator.hpp"
#include "Converter/Converter_Pager.hpp"
#include "Converter/Converter_BlockAddressIterator.hpp"
#include "FrameControl/FrameCtrl_ICommon.hpp"
#include "IOManager/IOMan_IDataInfo.hpp"
#include "Pager/Pager_IRestart.hpp"
#include "Pager/Pager_ISavepoint.hpp"
#include "Logging/Log_Types.hpp"
#include "KernelCommon/Kernel_IAdminRestartShutdown.hpp"
#include "KernelCommon/ParameterNames/KernelParam_Converter.hpp"
#include "RunTime/RTE_Message.hpp"
#include "RunTime/RTE_Crash.hpp"
#include "RunTime/Configuration/RTEConf_ParameterAccess.hpp"
#include "RunTime/Configuration/RTEConf_ParameterAccessKernelInterface.hpp"



/*===========================================================================*
 *  DEFINES                                                                  *
 *===========================================================================*/

/// Defines security space for database full handling

#define CONV_NUM_SECURITY_PAGES 10


/*===========================================================================*
 *  STATIC/INLINE FUNCTION PROTOTYPES                                        *
 *===========================================================================*/

/* -----------------------------------------------------------------------*/
/*!
    @brief  This method is used to get the configured value for a given
            integer configuration parameter from the configuration file.
    @return (SAPDB_UInt) configuration value
 */
/* -----------------------------------------------------------------------*/

static SAPDB_UInt
GetConfiguredValue( const RTEConf_Parameter::Name paramName )
{
    RTEConf_Parameter::Integer  paramValue;
    SAPDBErr_MessageList        errMsg;

    if( ! RTEConf_ParameterAccess::Instance()->GetInteger(
                paramName, paramValue, errMsg ))
    {
        RTE_Crash( errMsg );
    }
    return( ( SAPDB_UInt ) paramValue );
}

/*===========================================================================*
 *  INSTANCE METHODS OF THE INTERFACE CLASSES  OF THE CONVERTER CLASS        *
 *===========================================================================*/

Converter_IManager & Converter_IManager::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_IMigration & Converter_IMigration::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_IDiagnose & Converter_IDiagnose::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_IDump & Converter_IDump::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_IPageIO & Converter_IPageIO::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_ISavePoint & Converter_ISavePoint::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_IVerify & Converter_IVerify::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_ICommon & Converter_ICommon::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_IBackUp & Converter_IBackUp::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_IPageNoMapping & Converter_IPageNoMapping::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_IPageNoManager & Converter_IPageNoManager::Instance()
{
    return Converter_Converter::Instance();
}

/*---------------------------------------------------------------------------*/

Converter_IEvent & Converter_IEvent::Instance()
{
    return Converter_Converter::Instance();
}


/*===========================================================================*
 *  METHODS OF THE CONVERTER CLASS                                           *
 *===========================================================================*/

Converter_Converter* Converter_Converter::m_Instance = NULL;

/*---------------------------------------------------------------------------*/

Converter_Converter::Converter_Converter( SAPDBMem_IRawAllocator  &allocator )
        :
        m_Allocator( UTF8( "Converter_Manager" ), allocator),
        m_IsLiveCacheInstance( g01is_livecache()),
        m_MapBlockingSize( 8 ),
        m_VolumeNoBitCount( GetConfiguredValue( UTF8( KERNELPARAM_VOLUMENO_BIT_COUNT ))),
        m_NumRegions( GetConfiguredValue( UTF8( KERNELPARAM_CONVERTER_REGIONS ))),
        m_BaseRegionId( g08convert0 ),
        m_PageSize( FrameCtrl_ICommon::GetInstance().GetPageSize()),
        m_TempMapControl( m_Allocator, Data_PageRecoveryMode( Data_Dynamic, Data_NotRecoverable )),
        m_PermMapControl( m_Allocator, Data_PageRecoveryMode( Data_Dynamic, Data_Recoverable )),
        m_PermStaticMapControl( m_Allocator, Data_PageRecoveryMode( Data_Static, Data_Recoverable )),
        m_Index (m_Allocator),
        m_Active( false ),
        m_InMaintenanceMode( false ),
        m_DBFullWaitQueue( CONV_SUSPEND_HANDLE_DB_FULL ),
        m_FreePageNoWaitQueue( CONV_SUSPEND_FREE_PAGENO_WAIT_DURING_SVP ),
        m_ExpandConverterWaitQueue( CONV_SUSPEND_EXPAND_WAIT_DURING_SVP ),
        m_DBUsageMonitor (sp31ei_db_filling_above_limit, sp31ei_db_filling_below_limit),
        m_VolumeList(NULL)
{
    // initialize static member of the converter page defining
    // the number of bits used to address a volume

    Converter_Page::SetDevNoBitCount( m_VolumeNoBitCount );

    // the exceeding of the following limits will always be written
    // into the knldiag. all values are defined in units of 0.01 %

    m_DBUsageMonitor.EnableEventing();
    const SAPDB_Bool bOverFlowEvent          = true;
    const SAPDB_Bool bWriteAlwaysIntoKnldiag = true;

    m_DBUsageMonitor.AddEvent(8000, bOverFlowEvent, sp31ep_high, bWriteAlwaysIntoKnldiag);
    m_DBUsageMonitor.AddEvent(9000, bOverFlowEvent, sp31ep_high, bWriteAlwaysIntoKnldiag);
    m_DBUsageMonitor.AddEvent(9500, bOverFlowEvent, sp31ep_high, bWriteAlwaysIntoKnldiag);
    m_DBUsageMonitor.AddEvent(9600, bOverFlowEvent, sp31ep_high, bWriteAlwaysIntoKnldiag);
    m_DBUsageMonitor.AddEvent(9700, bOverFlowEvent, sp31ep_high, bWriteAlwaysIntoKnldiag);
    m_DBUsageMonitor.AddEvent(9800, bOverFlowEvent, sp31ep_high, bWriteAlwaysIntoKnldiag);
    m_DBUsageMonitor.AddEvent(9900, bOverFlowEvent, sp31ep_high, bWriteAlwaysIntoKnldiag);

    for (int Limit = 500; Limit < 10000; Limit+=500)
    {
        m_DBUsageMonitor.AddEvent(Limit, bOverFlowEvent,   sp31ep_low, ! bWriteAlwaysIntoKnldiag);
        m_DBUsageMonitor.AddEvent(Limit, ! bOverFlowEvent, sp31ep_low, ! bWriteAlwaysIntoKnldiag);
    }
}

/*---------------------------------------------------------------------------*/

Converter_Converter&
Converter_Converter::CreateInstance ()
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::CreateInstance", Converter_Trace, 5 );

    if( m_Instance == NULL )
    {
        m_Instance = new (RTEMem_Allocator::Instance())
                     Converter_Converter( RTEMem_Allocator::Instance() );

        if( NULL == m_Instance )
        {
            Converter_Exception errMsg( __CONTEXT__, CONVERTER_NO_MEMORY_TO_BUILD_CONVERTER );
            RTE_Crash( errMsg );
        }
    }
    return *m_Instance;
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::Restart(
    const tsp00_TaskId              TaskId,
    const Converter_Version         &ConverterVersion,
    const IOMan_PackedBlockAddress  &packedRootBlockAddr,
    const Converter_Version         &snapShotVersion,
    const IOMan_PackedBlockAddress  &packedSnapShotRootBlockAddr,
    const Kernel_DateTime           &snapShotCreationTimeStamp,
    const SAPDB_Bool                bIncConverterVersion,
    const SAPDB_Bool                bLastSaveDataSuccessful,
    const Data_PageNo               MaxDynamicPageNo,
    const Data_PageNo               MaxStaticPageNo,
    const SAPDB_UInt                NumAllPages )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::Restart", Converter_Trace, 5 );

    SAPDBTRACE_WRITELN( Converter_Trace, 5, "ConverterVersion      : " << ConverterVersion );
    SAPDBTRACE_WRITELN( Converter_Trace, 5, "SnapShotVersion       : " << snapShotVersion );
    SAPDBTRACE_WRITELN( Converter_Trace, 7, "LastSaveDataSuccessful: " <<
                        ( bLastSaveDataSuccessful ? "yes" : "no" ));
    SAPDBTRACE_WRITELN( Converter_Trace, 7, "MaxDynamicPageNo      : " << MaxDynamicPageNo );
    SAPDBTRACE_WRITELN( Converter_Trace, 7, "MaxStaticPageNo       : " << MaxStaticPageNo );
    SAPDBTRACE_WRITELN( Converter_Trace, 7, "NumAllPages           : " << NumAllPages );


    const SAPDB_UInt numAllStaticPages = m_IsLiveCacheInstance ? NumAllPages : 0;

    if(
        ( ! m_TempMapControl.Initialize( TaskId, NumAllPages,
                                         m_BaseRegionId, m_NumRegions, m_MapBlockingSize, m_PageSize ))
        ||
        ( ! m_PermMapControl.Initialize( TaskId, SAPDB_MAX( MaxDynamicPageNo, Data_PageNo( NumAllPages )),
                                         m_BaseRegionId, m_NumRegions, m_MapBlockingSize, m_PageSize,
                                         bLastSaveDataSuccessful ))
        ||
        (! m_PermStaticMapControl.Initialize( TaskId, SAPDB_MAX( MaxStaticPageNo, Data_PageNo( numAllStaticPages )),
                                              m_BaseRegionId, m_NumRegions, m_MapBlockingSize, m_PageSize,
                                              bLastSaveDataSuccessful ))
    )
    {
        Converter_Exception errMsg( __CONTEXT__, CONVERTER_MAP_INIT_FAILED1,
                                    SAPDB_ToString( NumAllPages,      _T_d ),
                                    SAPDB_ToString( MaxDynamicPageNo, _T_d ),
                                    SAPDB_ToString( MaxStaticPageNo,  _T_d ));
        RTE_Crash( errMsg );
    }

    InitializeBase( NumAllPages );

    m_Version                   = ConverterVersion;
    m_RootBlockAddress          = packedRootBlockAddr.Unpack( m_VolumeNoBitCount );
    m_SnapShotVersion           = snapShotVersion;
    m_SnapShotCreationTimeStamp = snapShotCreationTimeStamp;
    m_LastSaveDataSuccessful    = bLastSaveDataSuccessful;

    /* the parent index is created starting with the data stored on the ConverterRootBlock */
    Converter_Pager pager( TaskId, m_SnapShotVersion );

    m_Index.Restart( m_RootBlockAddress, pager );

    /* create per device lists of all converter leaf nodes */
    VolumeList  volumeList( m_Allocator );

    m_Index.CreateInputQueues(volumeList, m_PageSize);

    m_VolumeList = &volumeList; // is used by the pager tasks

    /* read all converter leaf pages referenced by the converter tree */
    Pager_IRestart::GetInstance().ReadConverter( TaskId );

    m_VolumeList = NULL;

    volumeList.Delete(); // free occupied memory

    SAPDBERR_ASSERT_STATE( m_Index.Verify()); // PTS 1113242 TS 2001-12-21

    // Caused by the FBM handling it is necessary that
    // the current/working converter is read first.

    if( m_SnapShotVersion.IsValid()){
        RestartSnapShot( TaskId, packedSnapShotRootBlockAddr );
    }

    if( bIncConverterVersion )
        ++m_Version;
    else
        m_InMaintenanceMode = true;

    CheckDataAreaPeakFilling();

    SAPDBTRACE_WRITELN( Converter_Trace, 5, "PermPagesUsed: " << m_NumPermPagesUsed );
    SAPDBTRACE_WRITELN( Converter_Trace, 5, "UpdatedDynamicPages: " <<
                        m_PermMapControl.NumUpdatedPermPages( TaskId ));
    SAPDBTRACE_WRITELN( Converter_Trace, 5, "UpdatedStaticPages: " <<
                        m_PermStaticMapControl.NumUpdatedPermPages( TaskId ));
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::ReadPages(
    const tsp00_TaskId  taskId,
    SAPDB_UInt          pagerIndex,
    SAPDB_UInt          &totalPages,
    SAPDB_UInt          &totalIO )
{

    SAPDBTRACE_ROUTINE_DEBUG( "Converter::ReadPages", Converter_Trace, 5 );

    SAPDBERR_ASSERT_STATE (NULL != m_VolumeList);

    Converter_Pager pager( taskId, m_SnapShotVersion );

    // loop over all converter leaf pages located in one data volume. This
    // will speed up the restart time of the converter. Check if pager tasks
    // is responsible for another data volume. This could happen, if more
    // data volumes are configured than pager tasks.


    while (pagerIndex < ((*m_VolumeList).GetSize()))
    {
        Converter_Index::BlockVector::Iterator        iter    = (*m_VolumeList)[pagerIndex].Begin();
        const Converter_Index::BlockVector::Iterator  iterEnd = (*m_VolumeList)[pagerIndex].End();

        for( ; iter < iterEnd; iter++ ) // read conv pages of one data volume
        {
            SAPDB_UInt NumPermPagesUpdated        = 0;
            SAPDB_UInt NumPendingUpdatedPermPages = 0;

            Converter_LeafPage  leafPage;

            pager.Read( iter->GetBlockAddress(), leafPage );

            totalPages++;
            totalIO++;

            const Data_AddressingMode addrMode = leafPage.ReadPageRecoveryMode().GetAddressingMode();

            SAPDBERR_ASSERT_STATE (leafPage.ReadPageRecoveryMode().PageIsRecoverable());

            // write the address of the parent index page on the leaf page according the
            // iterator. This is needed for converter leaf page i/o to be able to update
            // the block address of the leaf in the corresponding converter index page.

            leafPage.WriteParentReference (iter->GetParentReference());

            SAPDB_UInt FreeEntryCount = 0;

            // mark all used blocks referenced by the current converter leaf pages as
            // occupied in the FBM. Moreover all reserved converter leaf page entries
            // are set to free.

            Converter_PageEntryIterator         pageEntryIter    = leafPage.Begin();
            const Converter_PageEntryIterator   pageEntryEndIter = leafPage.End();

            for( ; pageEntryIter != pageEntryEndIter; ++pageEntryIter )
            {
                if( ! pageEntryIter.EntryInUse())
                {
                    ++FreeEntryCount;
                    continue;
                }

                const Converter_PageEntryNo entryPos = pageEntryIter.GetPosition();

                if( pageEntryIter.GetPackedBlockAddress().IsReserved())
                {
                    leafPage.ReleaseEntry( entryPos );
                    ++FreeEntryCount;
                    continue;
                }

                m_NumPermPagesUsed.Increment();

                CheckFillingOfDataBase();

                // Update the number of changed data pages since the last save data

                const bool bEntryIsRelevantForSavePages = leafPage.SavePagesRequested( entryPos );

                if( bEntryIsRelevantForSavePages )
                    NumPermPagesUpdated+=1;

                // Do not consider pending flag if last complete backup was succesful or
                // the data page is already modified again and therefore part of the next
                // incremental backup
                if( m_LastSaveDataSuccessful || bEntryIsRelevantForSavePages )
                    continue;

                if( leafPage.SavePagesPending( entryPos ))
                    NumPendingUpdatedPermPages+=1;
            }

            Converter_BlockAddressIterator elements = leafPage.elements();

            if( SAPDB_TRUE != FBM_IManager::Instance().SetBlockStateToOccupied(taskId, elements))
            {
                Converter_Exception errMsg( __CONTEXT__, CONVERTER_BAD_ENTRY_STATE,
                                            SAPDB_ToString( leafPage.GetPageNo( elements.GetPosition()), _T_d ),
                                            (( Data_Dynamic == leafPage.ReadPageRecoveryMode().GetAddressingMode())
                                             ? "perm" : "static"));
                RTE_Crash( errMsg );
            }

            if( Data_Static == addrMode )
            {
                m_PermStaticMapControl.IntegratePage( taskId, leafPage, FreeEntryCount, NumPermPagesUpdated,
                                                      NumPendingUpdatedPermPages );
            }
            else // Data_Dynamic
            {
                m_PermMapControl.IntegratePage( taskId, leafPage, FreeEntryCount, NumPermPagesUpdated,
                                                NumPendingUpdatedPermPages );
            }
        }
        pagerIndex += g01maxdatawriter();
    }
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::ReadSnapShotPages(
    const tsp00_TaskId  taskId,
    SAPDB_UInt          pagerIndex,
    SAPDB_UInt          &totalPages,
    SAPDB_UInt          &totalIO )
{

    SAPDBTRACE_ROUTINE_DEBUG( "Converter::ReadSnapShotPages", Converter_Trace, 5 );

    SAPDBERR_ASSERT_STATE (NULL != m_VolumeList);

    // check if Pager has something to do ..
    if ((*m_VolumeList).GetSize() <= pagerIndex){
        return;
    }

    Converter_PageFrameAllocator    frameAllocator( taskId );
    Converter_SnapShotPager         pager( taskId, m_SnapShotVersion, m_Index );
    Converter_LeafPage              leafPage;

    leafPage.SetFrame( frameAllocator.New());

    // loop over all converter leaf pages located in one data volume. This
    // will speed up the restart time of the converter. Check if pager tasks
    // is responsible for another data volume. This could happen, if more
    // data volumes are configured than pager tasks.

    while (pagerIndex < ((*m_VolumeList).GetSize()))
    {
        Converter_Index::BlockVector::Iterator        iter    = (*m_VolumeList)[pagerIndex].Begin();
        const Converter_Index::BlockVector::Iterator  iterEnd = (*m_VolumeList)[pagerIndex].End();

        for( ; iter < iterEnd; iter++ ) // read conv pages of one data volume
        {
            pager.Read( iter->GetBlockAddress(), leafPage );

            totalPages++;
            totalIO++;

            const Data_PageRecoveryMode recMode  = leafPage.ReadPageRecoveryMode();
            const Data_AddressingMode   addrMode = recMode.GetAddressingMode();

            SAPDBERR_ASSERT_STATE( recMode.PageIsRecoverable());

            // mark all used blocks referenced by the current snapshot converter
            // leaf page as occupied in the FBM. Note that the corresponding FBM
            // block could be already set to occupied!

            Converter_PageEntryIterator         pageEntryIter    = leafPage.Begin();
            const Converter_PageEntryIterator   pageEntryEndIter = leafPage.End();

            for( ; pageEntryIter != pageEntryEndIter; ++pageEntryIter )
            {
                if( ! pageEntryIter.EntryInUse())
                    continue;

                if( pageEntryIter.GetPackedBlockAddress().IsReserved())
                    continue;


                const Data_PageNo        pageNo   = leafPage.GetPageNo( pageEntryIter.GetPosition());
                const IOMan_BlockAddress newBlock = pageEntryIter.GetBlockAddress(); // from snapShot
                IOMan_BlockAddress       block;

                const Converter_ReturnCode retCode = GetPositionForReading( taskId, pageNo, recMode, block );

                SAPDBERR_ASSERT_STATE( newBlock.IsValid());

                IncrementSnapShotSize();

                if(( Converter_Ok == retCode ) && ( block == newBlock ))
                {
                    SAPDBTRACE_WRITELN( Converter_Trace, 7, "Snapshot: " << ((Data_Dynamic == addrMode) ?
                                        "Permanent" : "Static") << " pageNo: " << pageNo << " on block: " <<
                                        block.GetDeviceNo() << "/" << block.GetBlockNo() << " not marked as used ");

                    continue; // given page(block) hasn't changed since snapshot creation
                }

                SAPDBTRACE_WRITELN( Converter_Trace, 7, "Snapshot: " << ((Data_Dynamic == addrMode) ?
                                    "Permanent" : "Static") << " pageNo: " << pageNo << " on block: " <<
                                    newBlock.GetDeviceNo() << "/" << newBlock.GetBlockNo() << " marked as used ");

                if( FBM_IManager::Instance().SetBlockStateToOccupied( taskId, newBlock ))
                {
                    m_NumPermPagesUsed.Increment();
                    continue;
                }

                Converter_Exception errMsg( __CONTEXT__, CONVERTER_BAD_ENTRY_STATE,
                                            SAPDB_ToString( pageNo, _T_d ),
                                            (( Data_Dynamic == addrMode) ? "perm" : "static"));
                RTE_Crash( errMsg );
            }
        }
        pagerIndex += g01maxdatawriter();
    }
    if( leafPage.IsAssigned()){
        frameAllocator.Free( leafPage.GetFrame());
    }
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::Create(
    const tsp00_TaskId taskId,
    const SAPDB_UInt   numAllPages )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::Create", Converter_Trace, 5 );

    const SAPDB_UInt numAllStaticPages = m_IsLiveCacheInstance ? numAllPages : 0;

    if(
        ( ! m_TempMapControl.Initialize( taskId, numAllPages, m_BaseRegionId,
                                         m_NumRegions, m_MapBlockingSize, m_PageSize ))
        ||
        ( ! m_PermMapControl.Initialize( taskId, numAllPages, m_BaseRegionId,
                                         m_NumRegions, m_MapBlockingSize, m_PageSize, true ))
        ||
        (! m_PermStaticMapControl.Initialize( taskId, numAllStaticPages, m_BaseRegionId,
                                              m_NumRegions, m_MapBlockingSize, m_PageSize, true ))
    )
    {
        Converter_Exception errMsg( __CONTEXT__, CONVERTER_MAP_INIT_FAILED2,
                                    SAPDB_ToString( numAllPages, _T_d ));
        RTE_Crash( errMsg );
    }

    InitializeBase( numAllPages );
    m_Version.SetToInitialVersion(); // corresponds to version 1

    m_Index.Create();
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::CreateForRestore(
    const tsp00_TaskId      taskId,
    const Converter_Version &converterVersion,
    const SAPDB_Bool        bLastSaveDataSuccessful,
    const Data_PageNo       maxDynamicPageNo,
    const Data_PageNo       maxStaticPageNo,
    const SAPDB_UInt        numAllPages )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::CreateForRestore", Converter_Trace, 5 );

    const SAPDB_UInt numAllStaticPages = m_IsLiveCacheInstance ? numAllPages : 0;

    if(
        ( ! m_TempMapControl.Initialize( taskId, numAllPages,
                                         m_BaseRegionId, m_NumRegions, m_MapBlockingSize, m_PageSize ))
        ||
        ( ! m_PermMapControl.Initialize( taskId, SAPDB_MAX( maxDynamicPageNo, Data_PageNo( numAllPages )),
                                         m_BaseRegionId, m_NumRegions, m_MapBlockingSize, m_PageSize,
                                         bLastSaveDataSuccessful ))
        ||
        (! m_PermStaticMapControl.Initialize( taskId, SAPDB_MAX( maxStaticPageNo, Data_PageNo( numAllStaticPages )),
                                              m_BaseRegionId, m_NumRegions, m_MapBlockingSize, m_PageSize,
                                              bLastSaveDataSuccessful ))
    )
    {
        Converter_Exception errMsg( __CONTEXT__, CONVERTER_MAP_INIT_FAILED1,
                                    SAPDB_ToString( numAllPages,      _T_d),
                                    SAPDB_ToString( maxDynamicPageNo, _T_d),
                                    SAPDB_ToString( maxStaticPageNo,  _T_d));
        RTE_Crash( errMsg );
    }

    InitializeBase( numAllPages );

    m_InMaintenanceMode      = true;
    m_RestoreDataActive      = true;
    m_Version                = converterVersion;
    m_LastSaveDataSuccessful = bLastSaveDataSuccessful;
    m_Index.Create();
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::Shutdown( const tsp00_TaskId taskId )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::Shutdown", Converter_Trace, 5 );

    if( ! m_Active )
        return;

    m_NumPermPagesUsed     = 0;
    m_NumTempPagesUsed     = 0;
    m_NumPermConvPages     = 0;
    m_NumTempConvPages     = 0;
    m_NumSnapShotPages     = 0;
    m_TemporaryPeakFilling = 0;
    m_PermanentPeakFilling = 0;
    m_NumAllPages          = 0;
    m_SecuritySpaceLimit   = 0;
    m_Active               = false;
    m_InMaintenanceMode    = false;
    m_SavePagesIsActiv     = false;
    m_SaveDataIsActiv      = false;
    m_RestoreDataActive    = false;
    m_RestorePagesActive   = false;
    m_ColdVerifyIsActive   = false;
    m_Version.Invalidate();
    m_SnapShotVersion.Invalidate();
    m_SnapShotCreationTimeStamp.Invalidate();

    /* resume all tasks waiting for a converter resource */
    m_DBFullWaitQueue.ResumeAllWaitingTasks( taskId );
    m_FreePageNoWaitQueue.ResumeAllWaitingTasks( taskId );

    m_TempMapControl.Shutdown( taskId );
    m_PermMapControl.Shutdown( taskId );
    m_PermStaticMapControl.Shutdown( taskId );

    m_Index.Shutdown();
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::FlushParallel (
    const tsp00_TaskId          taskId,
    Converter_Version           &ConverterVersion,
    Data_PageNo                 &MaxStaticPageNo,
    Data_PageNo                 &MaxDynamicPageNo,
    IOMan_PackedBlockAddress    &packedRootBlockAddr,
    SAPDB_UInt                  &NumPagesTotal,
    SAPDB_UInt                  &NumIOTotal)
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::FlushParallel", Converter_Trace, 5 );

    SAPDB_UInt  NumPages = 0;
    SAPDB_UInt  NumIO    = 0;

    if( ! m_Active )
    {
        Converter_Exception errMsg( __CONTEXT__, CONVERTER_INFO_SHUTDOWN );
        RTE_Message( errMsg );

        Kernel_IAdminRestartShutdown::Instance().Offline();
    }

    // MapControls dealing with permanent pageId's remembers parentReferences
    // of empty converter leaf pages between converter flushes, now it's time
    // to drop these pages from the index

    m_PermMapControl.RemoveEmptyLeaves( taskId, m_SnapShotVersion, m_Index );
    m_PermStaticMapControl.RemoveEmptyLeaves( taskId, m_SnapShotVersion, m_Index );

    Pager_ISavepoint::GetInstance().FlushConverterParallel( taskId, NumPages, NumIO );

    NumPagesTotal += NumPages;
    NumIOTotal    += NumIO;

    Converter_Pager pager ( taskId, m_SnapShotVersion );

    MaxStaticPageNo    = m_MaxUsedStaticPageNo;
    MaxDynamicPageNo   = m_MaxUsedDynamicPageNo;

    m_RootBlockAddress = m_Index.Flush( pager, NumPages, NumIO );

    NumPagesTotal += NumPages;
    NumIOTotal    += NumIO;

    packedRootBlockAddr = m_RootBlockAddress.Pack( m_VolumeNoBitCount );;
    ConverterVersion    = m_Version;

    // Converter leaf pages are set to changed if a permanent data page is
    // removed or the block address is updated. Both operations are not
    // possible during the converter flush (savepoint) and therefore the
    // number of changed converter leaves has to be zero at the end of the
    // savepoint.

    SAPDBERR_ASSERT_STATE( 0 == GetNumberOfChangedConvPages());
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::WritePages(
    const tsp00_TaskId  taskId,
    const SAPDB_UInt    iSection,
    SAPDB_UInt          &totalPages,
    SAPDB_UInt          &totalIO )
{
    totalPages = 0;
    totalIO    = 0;

    Converter_Pager pager( taskId, m_SnapShotVersion );

    m_PermMapControl.FlushSection( taskId, m_Index, pager, iSection, totalPages, totalIO );
    m_PermStaticMapControl.FlushSection( taskId, m_Index, pager, iSection, totalPages, totalIO );
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::FlushSerial(
    const tsp00_TaskId          taskId,
    Converter_Version           &ConverterVersion,
    Data_PageNo                 &MaxStaticPageNo,
    Data_PageNo                 &MaxDynamicPageNo,
    IOMan_PackedBlockAddress    &packedRootBlockAddr )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::FlushSerial", Converter_Trace, 5 );

    SAPDB_UInt  NumPages = 0;
    SAPDB_UInt  NumIO    = 0;

    if( ! m_Active )
        return( SAPDB_FALSE );

    // MapControls dealing with permanent pageId's remembers parentReferences
    // of empty converter leaf pages between converter flushes, now it's time
    // to drop these pages from the index

    m_PermMapControl.RemoveEmptyLeaves( taskId, m_SnapShotVersion, m_Index );
    m_PermStaticMapControl.RemoveEmptyLeaves( taskId, m_SnapShotVersion, m_Index );

    Converter_Pager pager ( taskId, m_SnapShotVersion );

    MaxStaticPageNo     = m_PermStaticMapControl.Flush( taskId, m_Index, pager );
    MaxDynamicPageNo    = m_PermMapControl.Flush( taskId, m_Index, pager );

    m_RootBlockAddress  = m_Index.Flush( pager, NumPages, NumIO);

    packedRootBlockAddr = m_RootBlockAddress.Pack( m_VolumeNoBitCount );;
    ConverterVersion    = m_Version;

    return( SAPDB_TRUE );
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::Expand(
    const tsp00_TaskId         TaskId,
    const IOMan_DeviceNo       &DevNo,
    const SAPDB_UInt           DevSize,
    const RTE_VolumeAccessMode VolMode,
    const SAPDB_UInt           NumPagesAdded )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::Expand", Converter_Trace, 5 );

    if( ! m_Active )
        return( SAPDB_FALSE );

    if( ! ExpandMap( TaskId, Data_PageRecoveryMode( Data_Dynamic, Data_NotRecoverable ),  NumPagesAdded ))
        return( SAPDB_FALSE );

    if( ! ExpandMap( TaskId, Data_PageRecoveryMode( Data_Dynamic, Data_Recoverable ), NumPagesAdded ))
        return( SAPDB_FALSE );

    if( m_IsLiveCacheInstance )
    {
        if( ! ExpandMap( TaskId, Data_PageRecoveryMode( Data_Static, Data_Recoverable ), NumPagesAdded ))
            return( SAPDB_FALSE );
    }

    // Add data volume to FBM before the capacity of the converter is enlarged,
    // because register device could fail. Furthermore the data base is available.

    if( ! FBM_IManager::Instance().AddVolume( TaskId, DevNo, DevSize, VolMode ))
        return( SAPDB_FALSE );

    m_NumAllPages +=NumPagesAdded;
    m_DBUsageMonitor.Rescale (m_NumAllPages);

    // PTS 1124331 TS 2003-09-23
    m_SecuritySpaceLimit = ( m_NumAllPages - SAPDB_MIN( m_NumAllPages/25, 4000));

    m_DBFullWaitQueue.ResumeAllWaitingTasks( TaskId );

    return( SAPDB_TRUE );
}

/*---------------------------------------------------------------------------*/

Converter_ReturnCode
Converter_Converter::BeginSaveData( const tsp00_TaskId  taskId )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::BeginSaveData", Converter_Trace, 5 );

    SAPDBERR_ASSERT_STATE (! BackUpIsActive());
    SAPDBERR_ASSERT_STATE (false == m_ColdVerifyIsActive);
    SAPDBERR_ASSERT_STATE (0     == m_NumStatConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumDynConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumPagesForBackUp);

    SAPDBTRACE_WRITELN( Converter_Trace, 7, "LastSaveDataSuccessful: "
                        << (m_LastSaveDataSuccessful ? "yes" : "no" ));


    m_SaveDataIsActiv = true;

    SAPDB_UInt              numPagesForBackUp = 0;
    Converter_ReturnCode    retCode;

    retCode = m_PermStaticMapControl.MarkPermPagesForSaveData( taskId,
              m_NumStatConvPagesForBackUp, numPagesForBackUp );

    if( Converter_Ok != retCode )
        return retCode;

    m_NumPagesForBackUp = numPagesForBackUp;

    retCode = m_PermMapControl.MarkPermPagesForSaveData( taskId,
              m_NumDynConvPagesForBackUp, numPagesForBackUp );

    if( Converter_Ok != retCode )
        return retCode;

    m_NumPagesForBackUp += numPagesForBackUp;

    if (g01is_archive())
    {
        m_PermStaticMapControl.ExtractArchiveConverterForSaveData( taskId );

        m_PermMapControl.ExtractArchiveConverterForSaveData( taskId );
    }

    FBM_IManager::Instance().BeginReadingBlocksMarkedForBackUp( taskId );

    return retCode;
}

/*---------------------------------------------------------------------------*/

Converter_ReturnCode
Converter_Converter::CheckConverterFBM( const tsp00_TaskId  taskId )
{
	SAPDBTRACE_ROUTINE_DEBUG( "Converter::CheckConverterFBM", Converter_Trace, 5 );

	SAPDBERR_ASSERT_STATE (! BackUpIsActive());
	SAPDBERR_ASSERT_STATE (false == m_ColdVerifyIsActive);
	SAPDBERR_ASSERT_STATE (0     == m_NumStatConvPagesForBackUp);
	SAPDBERR_ASSERT_STATE (0     == m_NumDynConvPagesForBackUp);
	SAPDBERR_ASSERT_STATE (0     == m_NumPagesForBackUp);

	SAPDB_UInt              numPagesPerm = 0;
	SAPDB_UInt              numPagesStatic = 0;
	SAPDB_UInt              numPagesTemp = 0;
	SAPDB_UInt              numReservedPerm = 0;
	SAPDB_UInt              numReservedStatic = 0;
	SAPDB_UInt              numReservedTemp = 0;
	SAPDB_UInt              numPagesConverter = 0;
	SAPDB_UInt				numPermPagesUsed = 0;
	SAPDB_UInt				numTempPagesUsed = 0;

	SAPDB_UInt              numPagesFBM = 0;

	Converter_ReturnCode    retCode;

	retCode = m_PermStaticMapControl.MarkPagesForCheckConverterFBM( taskId, numPagesStatic, numReservedStatic );

	if( Converter_Ok != retCode )
		return retCode;

	retCode = m_PermMapControl.MarkPagesForCheckConverterFBM( taskId, numPagesPerm, numReservedPerm );
	numPermPagesUsed = m_NumPermPagesUsed;

	if( Converter_Ok != retCode )
		return retCode;

	retCode = m_TempMapControl.MarkPagesForCheckConverterFBM( taskId, numPagesTemp, numReservedTemp );
	numTempPagesUsed = m_NumTempPagesUsed;

	if( Converter_Ok != retCode )
		return retCode;

	numPagesConverter = m_Index.MarkPagesForCheckConverterFBM(taskId);

	//IOMan_ReservedBlockAddress reservedAddresses;
	//FBM_IManager::Instance().SetBlockStateToBackup(taskId, reservedAddresses.GetRestartPageAddress());

	numPagesFBM = FBM_IManager::Instance().CheckConverterFBM( taskId );

	if (numPermPagesUsed != (numPagesStatic + numPagesPerm + numReservedPerm + numReservedStatic))
	{
		Converter_Exception missmatchperm( __CONTEXT__, CONVERTER_INFO_COUNTER_MISSMATCH, "Perm+Static",
								SAPDB_ToString( numPermPagesUsed, _T_d ),
								SAPDB_ToString( numPagesStatic + numPagesPerm + numReservedPerm + numReservedStatic, _T_d ));
		RTE_Message( missmatchperm );
	}
	if (m_NumTempPagesUsed != (numPagesTemp + numReservedTemp))
	{
		Converter_Exception missmatchtemp( __CONTEXT__, CONVERTER_INFO_COUNTER_MISSMATCH, "Temp",
								SAPDB_ToString( m_NumTempPagesUsed, _T_d ),
								SAPDB_ToString( numPagesTemp + numReservedTemp, _T_d ));
		RTE_Message( missmatchtemp );
	}
	if (m_NumPermConvPages != numPagesConverter)
	{
		Converter_Exception missmatchconv( __CONTEXT__,  CONVERTER_INFO_COUNTER_MISSMATCH, "Converter",
								SAPDB_ToString( m_NumPermConvPages, _T_d ),
								SAPDB_ToString( numPagesConverter, _T_d ));
		RTE_Message( missmatchconv );
	}
	if ((numPagesConverter + numPagesTemp + numPagesStatic + numPagesPerm) != numPagesFBM)
	{
		Converter_Exception missmatchFBM( __CONTEXT__, CONVERTER_INFO_COUNTER_MISSMATCH, "FBM",
								SAPDB_ToString( numPagesConverter + numPagesTemp + numPagesStatic + numPagesPerm, _T_d ),
								SAPDB_ToString( numPagesFBM, _T_d ));
		RTE_Message( missmatchFBM );
	}

	Converter_Exception stats( __CONTEXT__, CONVERTER_INFO_COUNTER_STATS,
								SAPDB_ToString(  numPagesPerm, _T_d),
								SAPDB_ToString(  numReservedPerm, _T_d),
								SAPDB_ToString(  numPagesStatic, _T_d),
								SAPDB_ToString(  numReservedStatic, _T_d),
								SAPDB_ToString(  numPagesTemp, _T_d),
								SAPDB_ToString(  numReservedTemp, _T_d),
								SAPDB_ToString(  numPagesConverter, _T_d),
								SAPDB_ToString(  numPagesFBM, _T_d));
	RTE_Message(stats);

	return retCode;
}

/*---------------------------------------------------------------------------*/

Converter_ReturnCode
Converter_Converter::BeginSavePages( const tsp00_TaskId  taskId )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::BeginSavePages", Converter_Trace, 5 );

    SAPDBERR_ASSERT_STATE (! BackUpIsActive());
    SAPDBERR_ASSERT_STATE (false == m_ColdVerifyIsActive);
    SAPDBERR_ASSERT_STATE (0     == m_NumStatConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumDynConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumPagesForBackUp);

    SAPDBTRACE_WRITELN( Converter_Trace, 7, "LastSaveDataSuccessful: "
                        << ( m_LastSaveDataSuccessful ? "yes" : "no" ));

    m_SavePagesIsActiv = true;

    SAPDB_UInt              numPagesForBackUp = 0;
    Converter_ReturnCode    retCode;

    retCode = m_PermStaticMapControl.MarkPermPagesForSavePages( taskId,
              m_NumStatConvPagesForBackUp, numPagesForBackUp, Version() );

    if( Converter_Ok != retCode )
        return retCode;

    m_NumPagesForBackUp = numPagesForBackUp;

    retCode = m_PermMapControl.MarkPermPagesForSavePages( taskId,
              m_NumDynConvPagesForBackUp, numPagesForBackUp, Version());

    if( Converter_Ok != retCode )
        return retCode;

    m_NumPagesForBackUp += numPagesForBackUp;

    FBM_IManager::Instance().BeginReadingBlocksMarkedForBackUp( taskId );

    return retCode;
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::ResetDataPageAfterBackUp(
    const tsp00_TaskId          &TaskId,
    const IOMan_BlockAddress    &BlockAddress,
    const Data_PageNo           &PageNo,
    const Data_AddressingMode   &AddressingMode )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::ResetDataPageAfterBackUp", Converter_Trace, 5 );

    SAPDB_Bool retCode = SAPDB_FALSE;

    if( m_Active )
    {
        if( Data_Dynamic == AddressingMode )
        {
            retCode = m_PermMapControl.RemovePageNoFromBackUpPageNoList( TaskId, PageNo );
        }
        else if( Data_Static == AddressingMode )
        {
            retCode = m_PermStaticMapControl.RemovePageNoFromBackUpPageNoList( TaskId, PageNo );
        }

        if( SAPDB_TRUE == retCode ){
            FBM_IManager::Instance().RestoreBlockStateMarkedForBackup( TaskId, BlockAddress );
        }
    }
    return( retCode );
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::EndSave(
    const tsp00_TaskId taskId,
    const SAPDB_Bool   bBackUpSuccessfullyFinished )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::EndSave", Converter_Trace, 5 );

    if( ! m_Active )
        return;

    m_NumDynConvPagesForBackUp  = 0;
    m_NumStatConvPagesForBackUp = 0;
    m_NumPagesForBackUp         = 0;

    if( m_SaveDataIsActiv )
    {
        m_SaveDataIsActiv = false;

        if (! bBackUpSuccessfullyFinished )
        {
            m_LastSaveDataSuccessful = false;
            m_PermMapControl.SaveDataFinishedUnSuccessfuly( taskId );
            m_PermStaticMapControl.SaveDataFinishedUnSuccessfuly( taskId );

        }
        else
        {
            m_LastSaveDataSuccessful = true;
            m_PermMapControl.SaveDataFinishedSuccessfuly( taskId );
            m_PermStaticMapControl.SaveDataFinishedSuccessfuly( taskId );
        }
    }
    else
    {
        m_SavePagesIsActiv = false;
        m_PermMapControl.RemoveBitMapPages();
        m_PermStaticMapControl.RemoveBitMapPages();
    }

    /* remove all flags and clear all lists which indicate */
    /* that a page is marked for the current back up       */
    if ( !bBackUpSuccessfullyFinished){
        FBM_IManager::Instance().RestoreAllBlockStatesMarkedForBackup( taskId );
    }
    else
    {
        SAPDBERR_ASSERT_STATE( CheckBackUp( taskId ));
    }

    m_PermMapControl.ClearBackUpPageNoList();
    m_PermStaticMapControl.ClearBackUpPageNoList();
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::RestoreConverterPage(
    const tsp00_TaskId  taskId,
    Converter_LeafPage  &leafPage )
{
    Converter_PageEntryIterator       start = leafPage.Begin();
    const Converter_PageEntryIterator end   = leafPage.End();

    const Data_AddressingMode addrMode = leafPage.ReadPageRecoveryMode().GetAddressingMode();

    while( start != end )
    {
        if (start.EntryInUse())
        {
            const Data_PageNo        pageNo       = leafPage.GetPageNo( start.GetPosition());
            const IOMan_BlockAddress blockAddress = start.GetBlockAddress();

            SetBlockAddressOfRestoredPage( taskId, blockAddress, pageNo, addrMode );
        }
        ++start;
    }
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::EndSavepoint(
    const tsp00_TaskId    taskId,
    Converter_Version     &converterVersion )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::EndSavepoint", Converter_Trace, 5 );

    ++m_Version;

    converterVersion = m_Version;

    m_PermMapControl.ReSetSavepointActiveFlag( taskId );
    m_PermStaticMapControl.ReSetSavepointActiveFlag( taskId );

    CheckDataAreaPeakFilling();
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::FreePageNosAfterSavepoint( const tsp00_TaskId taskId )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::FreePageNosAfterSavepoint", Converter_Trace, 5 );

    // release all PageNos stored in the FreePageNoList
    m_PermMapControl.FreePageNosAfterSavepoint( taskId, m_SnapShotVersion );
    m_PermStaticMapControl.FreePageNosAfterSavepoint( taskId, m_SnapShotVersion );

    // resume all tasks which were suspended during a savepoint since there
    // was no more space to insert the pageno to release into a freepagenolist
    if( ! m_FreePageNoWaitQueue.IsEmpty() )
        m_FreePageNoWaitQueue.ResumeAllWaitingTasks( taskId );

    // resume tasks which want to expand the converter and collidated with the savepoint
    if( ! m_ExpandConverterWaitQueue.IsEmpty() )
        m_ExpandConverterWaitQueue.ResumeAllWaitingTasks( taskId );

    // resume all tasks suspended because a FBM overflow occured
    if( ! m_DBFullWaitQueue.IsEmpty() )
        m_DBFullWaitQueue.ResumeAllWaitingTasks( taskId );

    // At this time all relevant FBM blocks are set to free and therefore the
    // database filling could be changed although no pages are requested or released.
    CheckFillingOfDataBase();

}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::GetPositionForWriting(
    const tsp00_TaskId          taskId,
    const Data_PageNo           &pageNo,
    const Data_PageRecoveryMode &pageRecoveryMode,
    const Converter_Version     &pageConverterVersion,
    const SAPDB_Bool            bReqSequential,
    IOMan_BlockAddress          &block )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::GetPositionForWriting", Converter_Trace, 5 );


    if( ! m_Active )
    {
        Converter_Exception errMsg( __CONTEXT__, CONVERTER_INFO_SHUTDOWN );
        RTE_Message( errMsg );

        Kernel_IAdminRestartShutdown::Instance().Offline();
    }

    const SAPDB_Bool     bIsPerm = pageRecoveryMode.PageIsRecoverable();
    Converter_ReturnCode retCode;

    if(( pageConverterVersion == m_Version ) || ( pageConverterVersion.IsValid()  && ( ! bIsPerm )))
    {
        // the currently occupied block can be reused

        if( bIsPerm )
        {
            if( pageRecoveryMode.UseDynamicPageAddressing() )
                retCode = m_PermMapControl.GetBlockAddress( taskId, pageNo, block );
            else
                retCode = m_PermStaticMapControl.GetBlockAddress( taskId, pageNo, block );
        }
        else
            retCode = m_TempMapControl.GetBlockAddress ( taskId, pageNo, block );
    }
    else // old block is not reusable
    {
        SAPDBERR_ASSERT_STATE (pageConverterVersion.IsInvalid() || (pageConverterVersion < m_Version));

        if( bIsPerm )
        {
            if( pageRecoveryMode.UseDynamicPageAddressing() )
                retCode = m_PermMapControl.GetNewPositionForWrite( taskId, pageNo,
                          pageConverterVersion, m_SnapShotVersion, bReqSequential, block );
            else
                retCode = m_PermStaticMapControl.GetNewPositionForWrite( taskId, pageNo,
                          pageConverterVersion, m_SnapShotVersion, bReqSequential, block );
        }
        else
            retCode = m_TempMapControl.GetNewPositionForWrite( taskId, pageNo, bReqSequential, block );
    }

    if( Converter_Ok == retCode )
    {
        // At this time all relevant FBM blocks are set to free and therefore the
        // database filling could be changed although no pages are requested or released.
        CheckFillingOfDataBase();
        return;
    }

    Converter_MapError mapError( __CONTEXT__ , pageRecoveryMode, pageNo, retCode );
    RTE_Crash( mapError );
}

/*---------------------------------------------------------------------------*/

Converter_ReturnCode
Converter_Converter::GetPositionForReading(
    const tsp00_TaskId          taskId,
    const Data_PageNo           pageNo,
    const Data_PageRecoveryMode &pageRecoveryMode,
    IOMan_BlockAddress          &block )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::GetPositionForReading", Converter_Trace, 5 );

    if( ! m_Active )
    {
        Converter_Exception errMsg( __CONTEXT__, CONVERTER_INFO_SHUTDOWN );
        RTE_Message( errMsg );

        Kernel_IAdminRestartShutdown::Instance().Offline();
    }

    if( pageRecoveryMode.PageIsRecoverable() )
    {
        if( pageRecoveryMode.UseDynamicPageAddressing() )
            return( m_PermMapControl.GetBlockAddress( taskId, pageNo, block ));
        else
            return( m_PermStaticMapControl.GetBlockAddress( taskId, pageNo, block ));
    }
    else
        return( m_TempMapControl.GetBlockAddress ( taskId, pageNo, block ));
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::SetPositionForWriting(
    const tsp00_TaskId          taskId,
    const Converter_Version     &pageConverterVersion,
    const Data_PageNo           &pageNo,
    const Data_PageRecoveryMode &pageRecoveryMode,
    const IOMan_BlockAddress    &block )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::SetPositionForWriting", Converter_Trace, 5 );

    if( ! m_Active )
    {
        Converter_Exception errMsg( __CONTEXT__, CONVERTER_INFO_SHUTDOWN );
        RTE_Message( errMsg );

        Kernel_IAdminRestartShutdown::Instance().Offline();
    }

    if( pageRecoveryMode.PageIsRecoverable() )
    {
        if( pageRecoveryMode.UseDynamicPageAddressing() )
            m_PermMapControl.SetPositionForWriting( taskId, m_Version, pageConverterVersion,
                                                    m_SnapShotVersion, pageNo, block );
        else
            m_PermStaticMapControl.SetPositionForWriting( taskId, m_Version, pageConverterVersion,
                    m_SnapShotVersion, pageNo, block );
    }
    else
        m_TempMapControl.SetPositionForWriting ( taskId, pageConverterVersion, pageNo, block );

    // At this time all relevant FBM blocks are set to free and therefore the
    // database filling could be changed although no pages are requested or released
    CheckFillingOfDataBase();
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::RequestSpecificiedNewStaticPageNo(
    const tsp00_TaskId  taskId,
    const Data_PageNo   pageNo )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::RequestSpecificiedNewStaticPageNo", Converter_Trace, 5 );

    // PTS 1116298 UH 2002-06-21
    // Added the wait interface and the loop because RequestSpecifiedNewPageNo()
    // has to wait for a savepoint.
    // It may happen that there exists a pending Free for the requested pageno.

    SAPDB_Bool                          bWaitUntilSavepointFinished = SAPDB_FALSE;
    Converter_WaitQueue::WaitContext    waitContext;

    do
    {
        if( ! m_Active )
        {
            Converter_Exception errMsg( __CONTEXT__, CONVERTER_INFO_SHUTDOWN );
            RTE_Message( errMsg );

            Kernel_IAdminRestartShutdown::Instance().Offline();
        }

        if( ! m_PermStaticMapControl.RequestSpecifiedNewPageNo( taskId, pageNo,
                waitContext, m_FreePageNoWaitQueue, bWaitUntilSavepointFinished ))
        {
            // means that the page no is in use
            SAPDBTRACE_WRITELN( Converter_Trace, 5, "PageNo: " <<  pageNo << "not available" );
            return( SAPDB_FALSE );
        }

        if( bWaitUntilSavepointFinished )
            m_FreePageNoWaitQueue.Wait( taskId );
    }
    while( bWaitUntilSavepointFinished );

    m_NumPermPagesUsed.Increment();
    CheckFillingOfDataBase ();

    SAPDBTRACE_WRITELN( Converter_Trace, 7, "PermPagesUsed: " << m_NumPermPagesUsed );
    return( SAPDB_TRUE );
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::FreeTempPageNo(
    const tsp00_TaskId  taskId,
    const Data_PageNo   pageNo )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::FreeTempPageNo", Converter_Trace, 5 );

    SAPDBTRACE_WRITELN( Converter_Trace, 5, "Free temp PageNo: " << pageNo );

    if( ! m_Active )
        return( SAPDB_FALSE );

    m_TempMapControl.FreePageNo( taskId, pageNo );

    m_NumTempPagesUsed.Decrement();
    SAPDBTRACE_WRITELN( Converter_Trace, 7, "TempPagesUsed: " << m_NumTempPagesUsed );

    CheckFillingOfDataBase();

    if( ! m_DBFullWaitQueue.IsEmpty() ){
        m_DBFullWaitQueue.ResumeAllWaitingTasks( taskId );
    }
    return( SAPDB_TRUE );
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::HandleDBFull (
    tgg00_TransContext  &Trans,
    SAPDB_UInt          NumDataPagesRequested)
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::HandleDBFull", Converter_Trace, 6 );

    tgg00_BasisError    &trError = Trans.trError_gg00;

    NumDataPagesRequested += CONV_NUM_SECURITY_PAGES; // add some pages for security reasons

    while (m_Active)
    {
        const SAPDB_Int4 numChangedPages  = NumChangedPages();
        //const SAPDB_Bool pageNosExhausted = GetNumberOfUsedPages() + NumDataPagesRequested >= GetNumberOfAllPages();
        // PTS 1124331 TS 2003-09-23
        const SAPDB_Bool pageNosExhausted = ! IsSpaceAvailable( NumDataPagesRequested );
        const SAPDB_Bool blocksExhausted  = ! FBM_IManager::Instance().IsSpaceAvailable( numChangedPages + NumDataPagesRequested );

        if( ! pageNosExhausted && ! blocksExhausted )
            return; // comman way to leave this method

        if( blocksExhausted && SavepointIsPossible() && FBM_IManager::Instance().FreeAfterSVPExist() )
            StartSavepoint( Trans, numChangedPages ); // Do not check necessity of SVP

        if(( m_IsLiveCacheInstance ) && ( ! SavepointNeeded() ))
            bd91StartOMSGarbageCollection( Trans, false );

        if( ! TransactionIsSuspendable( Trans ))
            return; // the task is qualified to proceed with his job

        const SAPDB_Bool bAddVolumeIsPossible = IOMan_IDataInfo::GetInstance().AddDataVolumePossible();
        const SAPDB_Bool bIsRollback          = Trans.trState_gg00.includes( tsRollback_egg00 );

        if
        (
            (
                (
                    (SAPDB_UInt(Trans.trTempCount_gg00) > m_NumTempPagesUsed/20) &&
                    (m_NumTempPagesUsed                 > m_NumAllPages/100    ) &&
                    (SAPDB_UInt(Trans.trTempCount_gg00) > 100                  )
                )
                ||
                ( ! bAddVolumeIsPossible )
            )
            &&
            ( ! bIsRollback )
        )
        {
            trError = e_no_more_temp_space; // force rollback but do not suspend task

            if (! bAddVolumeIsPossible)
            {
                Converter_Exception errMsg (__CONTEXT__, CONVERTER_NO_ADD_DATA_DEVICE,
                                            SAPDB_ToString( Trans.trTaskId_gg00, _T_d ));
                RTE_Message( errMsg );
            }
            else
            {
                Converter_Exception errMsg(__CONTEXT__,CONVERTER_FORCE_ROLLBACK_DUE_TO_DB_FULL,
                                           SAPDB_ToString( Trans.trTaskId_gg00, _T_d ),
                                           SAPDB_ToString(m_NumAllPages, _T_d),
                                           SAPDB_ToString(m_NumTempPagesUsed, _T_d),
                                           SAPDB_ToString(Trans.trTempCount_gg00, _T_d));
                RTE_Message( errMsg );
            }
            return;
        }

        if( blocksExhausted )
        {
            Converter_Exception errMsg( __CONTEXT__,CONVERTER_SUSPEND_TASK_DUE_TO_FBM_FULL,
                                        SAPDB_ToString( Trans.trTaskId_gg00, _T_d ),
                                        SAPDB_ToString( numChangedPages + NumDataPagesRequested, _T_d ),
                                        SAPDB_ToString( FBM_IManager::Instance().GetNumberOfUsedBlocks(), _T_d ),
                                        SAPDB_ToString( FBM_IManager::Instance().GetNumberOfFreeBlocks(), _T_d ));
            RTE_Message( errMsg );

            SAPDBTrace_Stream   traceStream( &Converter_Trace.GetTracePrinter() );

            traceStream << "*** " << errMsg.DateTime( SAPDB_TRUE ) << " "
            << errMsg.Type( SAPDB_TRUE ) << " "
            << errMsg.ID() << " "
            << errMsg.Component() << " " << NewLine
            << "*** " << errMsg.Message() << NewLine;
        }
        else // pageNosExhausted
        {
            Converter_Exception errMsg( __CONTEXT__,CONVERTER_SUSPEND_TASK_DUE_TO_DB_FULL,
                                        SAPDB_ToString( Trans.trTaskId_gg00, _T_d ),
                                        SAPDB_ToString( GetNumberOfUsedPages(), _T_d ),
                                        SAPDB_ToString( NumDataPagesRequested, _T_d ),
                                        SAPDB_ToString( GetNumberOfAllPages(), _T_d ));
            RTE_Message( errMsg );

            SAPDBTrace_Stream   traceStream( &Converter_Trace.GetTracePrinter() );

            traceStream << "*** " << errMsg.DateTime( SAPDB_TRUE ) << " "
            << errMsg.Type( SAPDB_TRUE ) << " "
            << errMsg.ID() << " "
            << errMsg.Component() << " " << NewLine
            << "*** " << errMsg.Message() << NewLine;
        }

        m_DBFullWaitQueue.InsertTaskAndWait( Trans.trTaskId_gg00 );

        if(( Trans.trRteCommPtr_gg00->to_cancel ) && ( ! bIsRollback ))
        {
            Converter_Exception errMsg( __CONTEXT__,CONVERTER_INFO_CANCELLED_DURING_DB_FULL,
                                        SAPDB_ToString( Trans.trTaskId_gg00, _T_d ));
            RTE_Message( errMsg );

            SAPDBTrace_Stream   traceStream( &Converter_Trace.GetTracePrinter() );

            traceStream << "*** " << errMsg.DateTime( SAPDB_TRUE ) << " "
            << errMsg.Type( SAPDB_TRUE ) << " "
            << errMsg.ID() << " "
            << errMsg.Component() << " " << NewLine
            << "*** " << errMsg.Message() << NewLine;

            trError  = e_cancelled;
            return;
        }
    }
    if( ! m_Active )
    {
        Converter_Exception errMsg( __CONTEXT__, CONVERTER_INFO_SHUTDOWN );
        RTE_Message( errMsg );

        Kernel_IAdminRestartShutdown::Instance().Offline();
    }

    return;
}

/*---------------------------------------------------------------------------*/

Converter_ReturnCode
Converter_Converter::BeginCheckData( const tsp00_TaskId    taskId )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::BeginCheckDataWithUpdate", Converter_Trace, 5 );

    SAPDBERR_ASSERT_STATE (! BackUpIsActive());
    SAPDBERR_ASSERT_STATE (false == m_ColdVerifyIsActive);
    SAPDBERR_ASSERT_STATE (0     == m_NumStatConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumDynConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumPagesForBackUp);

    m_InMaintenanceMode  = true;
    m_ColdVerifyIsActive = true;

    Converter_ReturnCode retCode;

    retCode = m_PermMapControl.InitializeBackUpPageNoList( taskId );

    if( Converter_Ok == retCode ){
        retCode = m_PermStaticMapControl.InitializeBackUpPageNoList( taskId );
    }

    return retCode;
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::EndCheckData(
    const tsp00_TaskId  taskId,
    Data_PageCount      &NumStaticPermDataPages,
    Data_PageCount      &NumDynamicPermDataPages,
    Data_PageCount      &NumPermConvPages,
    Data_PageCount      &NumReleaseBlocks )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::EndCheckDataWithUpdate", Converter_Trace, 5 );

    SAPDBERR_ASSERT_STATE (! BackUpIsActive());
    SAPDBERR_ASSERT_STATE (true == m_ColdVerifyIsActive);
    SAPDBERR_ASSERT_STATE (0     == m_NumStatConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumDynConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumPagesForBackUp);

    SAPDB_UInt NumPermStaticConvPages   = 0;
    SAPDB_UInt NumPermDynamicConvPages  = 0;
    SAPDB_UInt NumStaticTempDataPages   = 0;
    SAPDB_UInt NumDynamictempDataPages  = 0;
    SAPDB_UInt numStaticReleasedBlocks  = 0;
    SAPDB_UInt numDynamicReleasedBlocks = 0;

    m_PermStaticMapControl.ReleaseUnusedPageNos( taskId, m_SnapShotVersion,
            NumStaticPermDataPages, NumPermStaticConvPages, numStaticReleasedBlocks );

    m_PermMapControl.ReleaseUnusedPageNos(       taskId, m_SnapShotVersion,
            NumDynamicPermDataPages, NumPermDynamicConvPages, numDynamicReleasedBlocks );

    NumPermConvPages = NumPermStaticConvPages + NumPermDynamicConvPages;
    NumReleaseBlocks = numStaticReleasedBlocks + numDynamicReleasedBlocks;

    m_NumPermPagesUsed = NumStaticPermDataPages + NumDynamicPermDataPages;

    m_PermStaticMapControl.ClearBackUpPageNoList();
    m_PermMapControl.ClearBackUpPageNoList();

    m_ColdVerifyIsActive = false;
    m_InMaintenanceMode  = false;
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::EndFailedCheckData ( const tsp00_TaskId  taskId )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::EndFailedCheckDataWithUpdate", Converter_Trace, 5 );

    SAPDBERR_ASSERT_STATE (! BackUpIsActive());
    SAPDBERR_ASSERT_STATE (true == m_ColdVerifyIsActive);
    SAPDBERR_ASSERT_STATE (0     == m_NumStatConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumDynConvPagesForBackUp);
    SAPDBERR_ASSERT_STATE (0     == m_NumPagesForBackUp);

    m_PermStaticMapControl.EndFailedColdVerify( taskId );
    m_PermMapControl.EndFailedColdVerify( taskId );

    m_ColdVerifyIsActive = false;
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::GetDbSpaceStatistics(
    const tsp00_TaskId  taskId,
    tgg00_DbSpaceInfo   &dbSpaceInfo )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::GetDbSpaceStatistics", Converter_Trace, 5 );

    dbSpaceInfo.db_total_pages    = IOMan_IDataInfo::GetInstance().TotalDataPages();
    dbSpaceInfo.db_perm_pages     = IOMan_IDataInfo::GetInstance().TotalDataPages();
    dbSpaceInfo.db_real_perm_used = m_NumPermPagesUsed + m_NumPermConvPages;
    dbSpaceInfo.db_perm_percent   = (tsp00_Int2)((dbSpaceInfo.db_real_perm_used * 100.0)/dbSpaceInfo.db_total_pages);
    dbSpaceInfo.db_temp_pages     = m_NumTempPagesUsed;
    dbSpaceInfo.db_temp_percent   = (tsp00_Int2)((m_NumTempPagesUsed*100.0)/dbSpaceInfo.db_total_pages);
    dbSpaceInfo.db_used_pages     = GetNumberOfUsedPages();
	dbSpaceInfo.db_unused_pages   = dbSpaceInfo.db_total_pages - ( dbSpaceInfo.db_used_pages + FBM_IManager::Instance().GetNumberOfFreeAfterSVPBlocks());
    dbSpaceInfo.db_updated_pages  = UpdatedPagesSinceLastSaveData( taskId );
    dbSpaceInfo.db_max_perm_used  = GetPeakForPermPages();
    dbSpaceInfo.db_used_blocks    = FBM_IManager::Instance().GetNumberOfUsedBlocks();
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::IsDBFull( const tsp00_TaskId taskId ) const
{
    if( ! m_DBFullWaitQueue.IsEmpty())
        return( SAPDB_TRUE ); // task is suspended because of db full

    const SAPDB_UInt numDataPagesRequested = CONV_NUM_SECURITY_PAGES * 2;// see HandledbFull

    // PTS 1124331 TS 2003-09-23
    if( ! IsSpaceAvailable( numDataPagesRequested ))
        return( SAPDB_TRUE ); // pageNos exhausted

    //if( GetNumberOfUsedPages() + numDataPagesRequested >= GetNumberOfAllPages())
    //    return( SAPDB_TRUE ); // pageNos exhausted

    if( ! FBM_IManager::Instance().IsSpaceAvailable( NumChangedPages() + numDataPagesRequested ))
        return( SAPDB_TRUE ); // blocks in FBM exhausted

    return( SAPDB_FALSE );
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::SecuritySpaceReached() const
{
    // PTS 1124331 TS 2003-09-23

    if( ! IsSpaceAvailable( CONV_NUM_SECURITY_PAGES ))
        return( SAPDB_TRUE );

    if( ! FBM_IManager::Instance().IsSpaceAvailable( NumChangedPages()))
        return( SAPDB_TRUE );

    return( SAPDB_FALSE );
}

/*---------------------------------------------------------------------------*/

Converter_GarbageCollectionReturnCode
Converter_Converter::GarbageCollectionNeeded() const // PTS 1116157 FF 2002-06-14
{
    if( ! SavepointIsPossible() )
        return( Converter_GCNotWanted );

    if( SavepointNeeded() ) // Savepoint will come immediately started by TimeOut Task
        return( Converter_GCNotWanted );

    // data base is filled up to 90 percent inclusive the changed pages of the data cache

    if( FBM_IManager::Instance().GarbageCollectionNeeded( NumChangedPages() ))
        return( Converter_GCNeeded );
    else
        return( Converter_GCNotNeeded );
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::StartSavepoint( tgg00_TransContext &trans )
{
    if( ! SavepointNeeded() )
        return;

    StartSavepoint( trans, NumChangedPages() );
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::SavepointNeeded() const
{
    if( ! SavepointIsPossible() )
        return( false );

    if( FBM_IManager::Instance().SavepointNeeded() ) // more blocks in state free after SVP than free
        return( true );

    if( NumChangedPages() > (( FBM_IManager::Instance().GetNumberOfFreeBlocks()) / 3 ))
        return( true );

    return( false );
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::Dump(
    const tsp00_TaskId  TaskId,
    Kernel_Dump         &Dump,
    const SAPDB_Bool    bEnterRegion )
{
    struct ConverterManager   CM;

    CM.dmpVersion                         = m_Version;
    CM.dmpNumPermPagesUsed                = m_NumPermPagesUsed;
    CM.dmpNumTempPagesUsed                = m_NumTempPagesUsed;
    CM.dmpNumAllPages                     = m_NumAllPages;
    CM.dmpDeltaSnapShotPages              = m_NumSnapShotPages;
    CM.dmpNumPagesForBackup               = m_NumPagesForBackUp;
    CM.dmpNumStatConvPagesForBackup       = m_NumStatConvPagesForBackUp;
    CM.dmpNumDynConvPagesForBackup        = m_NumDynConvPagesForBackUp;
    CM.dmpNumPermConvPages                = m_NumPermConvPages;
    CM.dmpNumTempConvPages                = m_NumTempConvPages;
    CM.dmpMapBlockingSize                 = m_MapBlockingSize;
    CM.dmpConverterIsActive               = m_Active;
    CM.dmpSavePagesIsActive               = m_SavePagesIsActiv;
    CM.dmpSaveDataIsActive                = m_SaveDataIsActiv;
    CM.dmpRestoreDataIsActive             = m_RestoreDataActive;
    CM.dmpRestorePagesIsActive            = m_RestorePagesActive;
    CM.dmpColdVerifyIsActive              = m_ColdVerifyIsActive;
    CM.dmpLastSaveDataSuccessful          = m_LastSaveDataSuccessful;

    if( m_RootBlockAddress.IsValid() )
    {
        CM.dmpRootDevNo   = m_RootBlockAddress.GetDeviceNo();
        CM.dmpRootBlockNo = m_RootBlockAddress.GetBlockNo();
    }
    else
    {
        CM.dmpRootDevNo   = IOMan_DeviceNo();
        CM.dmpRootBlockNo = IOMan_BlockNo();
    }

    Dump.InsertEntry( Kernel_Dump::DmpConverterManager, Kernel_DumpPage::Entry( &CM, sizeof( CM )));

    m_Index.Dump( Dump );

    m_TempMapControl.Dump( TaskId, Dump, bEnterRegion );
    m_PermMapControl.Dump( TaskId, Dump, bEnterRegion );
    m_PermStaticMapControl.Dump( TaskId, Dump, bEnterRegion );
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::SetBlockAddressForMigratedPage (
    const tsp00_TaskId          taskId,
    const IOMan_BlockAddress    &block,
    const Data_PageNo           &pageNo,
    const SAPDB_Bool            bSetToOccupied )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::SetBlockAddressForMigratedPage", Converter_Trace, 5 );

    SAPDBERR_ASSERT_STATE( m_RestoreDataActive );

    // At this time the corresponding converter page entry is invalid (initialized),
    // because the converter is build up the current migration step.

    m_PermMapControl.SetBlockAddressOfRestoredPage( taskId, pageNo, m_SnapShotVersion,
            block, m_RestoreDataActive);

    m_NumPermPagesUsed.Increment();

    CheckFillingOfDataBase();

    if( SAPDB_TRUE != bSetToOccupied )
        return;

    // After updating the converter page entry, the block is set to occupied within
    // the FBM. Note that the given block must not specify one of the reserved blocks
    // in the data volumes.

    if( SAPDB_TRUE != FBM_IManager::Instance().SetBlockStateToOccupied( taskId, block ))
    {
        Converter_Exception errMsg( __CONTEXT__, CONVERTER_BAD_ENTRY_STATE,
                                    SAPDB_ToString( pageNo, _T_d ), "perm");
        RTE_Crash( errMsg );
    }
}

/*---------------------------------------------------------------------------*/

IOMan_BlockAddress
Converter_Converter::DetermineAndSetBlockAddressForMigratedPage(
    const tsp00_TaskId  taskId,
    const Data_PageNo   &pageNo )
{
    const SAPDB_Bool         bSetToOccupied = SAPDB_FALSE;
    const SAPDB_Bool         bReqSequential = SAPDB_FALSE;
    const IOMan_BlockAddress newBlock       = FBM_IManager::Instance().GetFreeBlock( taskId , bReqSequential );

    // Note that the newBlock is set to occupied by the GetFreeBlock
    // method and therefore not set again to occupupied by the subsequent call.

    SetBlockAddressForMigratedPage( taskId, newBlock, pageNo, bSetToOccupied );

    return( newBlock );
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::DeleteBlockAddress(
    const tsp00_TaskId          taskId,
    const Data_PageNo           &pageNo,
    const Data_PageRecoveryMode &recMode )
{
    SAPDBERR_ASSERT_STATE( m_Active );

    IOMan_BlockAddress          block;
    Converter_ReturnCode        retCode;
    const Data_AddressingMode   addrMode = recMode.GetAddressingMode();

    if( ! recMode.PageIsRecoverable())
        return( SAPDB_TRUE );

    if( Data_Dynamic == addrMode ){
        retCode = m_PermMapControl.GetBlockAddress( taskId, pageNo, block );
    }
    else if( Data_Static == addrMode ){
        retCode = m_PermStaticMapControl.GetBlockAddress( taskId, pageNo, block );
    }
    else{
        retCode = Converter_NoConverterEntry;
    }

    if( Converter_Ok != retCode )
        return( SAPDB_FALSE );

    Converter_Version  pageConverterVersion; // dummy version, not used in this case

    pageConverterVersion.SetToInitialVersion(); // set to a defined value, which will not influence the execution

    if( Data_Dynamic == addrMode )
        return( FreePermDynamicPageNo( taskId, pageNo, pageConverterVersion ));
    else
        return( FreePermStaticPageNo( taskId, pageNo, pageConverterVersion ));
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::GetBlockAddress(
    const tsp00_TaskId          taskId,
    const Data_PageNo           &pageNo,
    const Data_PageRecoveryMode &recMode,
    IOMan_BlockAddress          &block )
{
    SAPDBERR_ASSERT_STATE( m_Active );

    Converter_ReturnCode        retCode;
    const Data_AddressingMode   addrMode = recMode.GetAddressingMode();

    if( Data_Dynamic == addrMode )
    {
        if( recMode.PageIsRecoverable())
            retCode = m_PermMapControl.GetBlockAddress( taskId, pageNo, block );
        else
            retCode = m_TempMapControl.GetBlockAddress( taskId, pageNo, block );
    }
    else if( Data_Static == addrMode ){
        retCode = m_PermStaticMapControl.GetBlockAddress( taskId, pageNo, block );
    }
    else{
        retCode = Converter_NoConverterEntry;
    }

    if( Converter_Ok == retCode )
        return( SAPDB_TRUE );

    block.Invalidate();

    return( SAPDB_FALSE );
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::SetNewBlockAddress(
    const tsp00_TaskId          taskId,
    const Data_PageNo           &pageNo,
    const Data_PageRecoveryMode &recMode,
    const IOMan_BlockAddress    &blockAddr )
{
    SAPDBERR_ASSERT_STATE( m_Active );

    if( ! recMode.PageIsRecoverable())
        return( SAPDB_TRUE );

    if( recMode.UseStaticPageAddressing() )
        m_PermStaticMapControl.SetBlockAddressOfRestoredPage( taskId, pageNo,
                m_SnapShotVersion, blockAddr, m_RestoreDataActive );
    else
        m_PermMapControl.SetBlockAddressOfRestoredPage( taskId, pageNo,
                m_SnapShotVersion, blockAddr, m_RestoreDataActive );

    return( SAPDB_TRUE );
}

//----------------------------------------------------------------------------
//  private methods
//----------------------------------------------------------------------------

void
Converter_Converter::InitializeBase( const SAPDB_UInt NumAllPages )
{
    m_SnapShotVersion.Invalidate();
    m_SnapShotCreationTimeStamp.Invalidate();

    m_NumPermPagesUsed                 = 0;
    m_NumTempPagesUsed                 = 0;
    m_NumPagesForBackUp                = 0;
    m_NumStatConvPagesForBackUp        = 0;
    m_NumDynConvPagesForBackUp         = 0;
    m_NumPermConvPages                 = 0;
    m_NumTempConvPages                 = 0;
    m_NumSnapShotPages                 = 0;
    m_TemporaryPeakFilling             = 0;
    m_PermanentPeakFilling             = 0;
    m_MaxUsedDynamicPageNo             = 0;
    m_MaxUsedStaticPageNo              = 0;
    m_NumAllPages                      = NumAllPages;
    // PTS 1124331 TS 2003-09-23
    m_SecuritySpaceLimit               = ( m_NumAllPages - SAPDB_MIN( m_NumAllPages/25, 4000));
    m_Active                           = true;
    m_InMaintenanceMode                = false;
    m_SavePagesIsActiv                 = false;
    m_SaveDataIsActiv                  = false;
    m_RestoreDataActive                = false;
    m_RestorePagesActive               = false;
    m_ColdVerifyIsActive               = false;
    m_LastSaveDataSuccessful           = true;
    m_DBUsageMonitor.Rescale (m_NumAllPages);
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::ExpandMap(
    const tsp00_TaskId           taskId,
    const Data_PageRecoveryMode  &recoveryMode,
    const SAPDB_UInt             numPagesAdded )
{
    Converter_WaitQueue::WaitContext    waitContext;
    Converter_ReturnCode                retCode;
    const SAPDB_UInt                    newAllPages = m_NumAllPages + numPagesAdded;

    do
    {
        if( recoveryMode.PageIsRecoverable() )
        {
            if( recoveryMode.UseDynamicPageAddressing() ){
                retCode = m_PermMapControl.Expand( taskId, newAllPages,
                                                   waitContext, m_ExpandConverterWaitQueue );
            }
            else{
                retCode = m_PermStaticMapControl.Expand( taskId, newAllPages,
                          waitContext, m_ExpandConverterWaitQueue );
            }
        }
        else{
            retCode = m_TempMapControl.Expand( taskId, newAllPages );
        }


        switch( retCode )
        {
        case Converter_Ok:
            return( SAPDB_TRUE );
        case Converter_SavePointActive:
            m_ExpandConverterWaitQueue.Wait( taskId );
            break;
        default:
            {
                Converter_Exception ExpandFailed(__CONTEXT__,CONVERTER_EXPAND_FAILED,
                                                 SAPDB_ToString( m_NumAllPages, _T_d) ,
                                                 SAPDB_ToString( newAllPages, _T_d ));
                RTE_Message( ExpandFailed );
            }
            return( SAPDB_FALSE );
        }
    }
    while( m_Active );

    return( SAPDB_FALSE );
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::FreePermPageNo(
    const tsp00_TaskId          taskId,
    Converter_PermMapControl    &mapControl,
    const Data_PageNo           pageNo,
    const Converter_Version     &pageConverterVersion )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::FreePermPageNo", Converter_Trace, 5 );

    SAPDBTRACE_WRITELN( Converter_Trace, 5, "Free perm PageNo: " << pageNo );

    if( ! m_Active ){
        return( SAPDB_FALSE );
    }

    SAPDB_Bool  bWaitUntilSavepointFinished  = false;
    SAPDB_Bool  bOldBlockIsUsableImmediately = false;
    SAPDB_Bool  bDoDecrementPermUseCount     = true;

    do
    {
        Converter_WaitQueue::WaitContext    waitContext;

        mapControl.FreePageNo( taskId, pageConverterVersion, m_SnapShotVersion,
                               pageNo, waitContext, m_FreePageNoWaitQueue,
                               bWaitUntilSavepointFinished, bOldBlockIsUsableImmediately,
                               bDoDecrementPermUseCount );

        if( bWaitUntilSavepointFinished )
        {
            m_FreePageNoWaitQueue.Wait( taskId );

            if( ! m_Active ){
                return( SAPDB_FALSE );
            }
        }
    }
    while( bWaitUntilSavepointFinished );

    if( bDoDecrementPermUseCount )
        m_NumPermPagesUsed.Decrement();

    SAPDBTRACE_WRITELN( Converter_Trace, 7, "PermPagesUsed: " << m_NumPermPagesUsed );

    CheckFillingOfDataBase();

    if(( bOldBlockIsUsableImmediately ) && ( ! m_DBFullWaitQueue.IsEmpty() )){
        m_DBFullWaitQueue.ResumeAllWaitingTasks( taskId );
    }
    return( SAPDB_TRUE );
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::StartSavepoint(
    tgg00_TransContext &trans,
    const SAPDB_Int4    numChangedPages )
{
    Converter_Exception errMsg( __CONTEXT__, CONVERTER_SAVEPOINT_NEEDED,
                                SAPDB_ToString( numChangedPages, _T_d ),
                                SAPDB_ToString( FBM_IManager::Instance().GetNumberOfUsedBlocks(), _T_d ),
                                SAPDB_ToString( FBM_IManager::Instance().GetNumberOfFreeBlocks(), _T_d ));
    RTE_Message( errMsg );

    Log_SavepointManager.StartSavepoint( trans, Log_SVPReasonConverter );
}

/*---------------------------------------------------------------------------*/

void
Converter_Converter::RestartSnapShot(
    const tsp00_TaskId              taskId,
    const IOMan_PackedBlockAddress  &packedSnapShotRootBlock )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::RestartSnapShot", Converter_Trace, 6 );

    SAPDBERR_ASSERT_STATE( NULL == m_VolumeList );

    const SAPDB_Bool            bReadSnapShot     = SAPDB_TRUE;
    const IOMan_BlockAddress    snapShotRootBlock = packedSnapShotRootBlock.Unpack( m_VolumeNoBitCount );
    Converter_SnapShotPager     pager( taskId, m_SnapShotVersion, m_Index );
    Converter_Index             snapShotIndex( m_Allocator);
    VolumeList                  volumeList( m_Allocator );

    SAPDBERR_ASSERT_STATE( snapShotRootBlock.IsValid());

    snapShotIndex.Restart( snapShotRootBlock, pager );

    SAPDBERR_ASSERT_STATE( snapShotIndex.Verify());

    snapShotIndex.CreateInputQueues( volumeList, m_PageSize );

    m_VolumeList = &volumeList; // is used by the pager tasks

    Pager_IRestart::GetInstance().ReadConverter( taskId, bReadSnapShot );

    m_VolumeList = NULL;

    volumeList.Delete(); // free occupied memory

    snapShotIndex.Shutdown();
}

/*---------------------------------------------------------------------------*/

SAPDB_Int4
Converter_Converter::NumChangedPages() const
{
    return( bd20GetNumberOfChangedPages() + GetNumberOfChangedConvPages());
}

/*---------------------------------------------------------------------------*/

SAPDB_Bool
Converter_Converter::IsLeafPageMember(
    const tsp00_TaskId        taskId,
    const Converter_LeafPage  &page )
{
    SAPDBTRACE_ROUTINE_DEBUG( "Converter::IsLeafPageMember", Converter_Trace, 7 );


    const Data_PageRecoveryMode recMode    = page.ReadPageRecoveryMode();
    const Data_PageNo           basePageNo = page.ReadBasePageNo();
    const IOMan_BlockAddress    block      = page.ReadBlockAddress();

    if( recMode.UseDynamicPageAddressing())
        return( m_PermMapControl.IsMember( taskId, basePageNo, block ));

    if( recMode.UseStaticPageAddressing())
        return(m_PermStaticMapControl.IsMember( taskId, basePageNo, block ));

    return( SAPDB_FALSE );
}

/*===========================================================================*
 *  END OF CODE                                                              *
 *===========================================================================*/
