/*!**************************************************************************

  module      : FBM_DataVolume.cpp

  -------------------------------------------------------------------------

  responsible : TorstenS

  auhtor      : AlexanderK

  special area: FreeBlockManagement (FBM)
  description : 


  last changed: 2000-03-10  10:00
  see also    : 

  -------------------------------------------------------------------------

  copyright:    (c) 2000-2004 SAP AG



    ========== 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


*****************************************************************************/



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

#include "FreeBlockManagement/FBM_DataVolume.hpp"
#include "FreeBlockManagement/FBM_Dump.hpp"
#include "FreeBlockManagement/FBM_Exception.hpp"
#include "FreeBlockManagement/FBM_Messages.hpp"
#include "RunTime/RTE_Crash.hpp"
#include "RunTime/RTE_Message.hpp"


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

/*===========================================================================*
 *  CLASS METHOD DEFINITION                                                  *
 *===========================================================================*/

/*---------------------------------------------------------------------------*
 *  CLASS FBM_DataVolume                                              *
 *---------------------------------------------------------------------------*/

void
FBM_DataVolume::InitVolume ()
{
    SAPDBTRACE_ROUTINE_DEBUG  ("FBM_DataVolume::InitVolume", FBM_Trace, 5);

    SAPDBERR_ASSERT_STATE( 0 < m_NumBlocks );

    /* set members to their start values */
    m_LastUsedBlock                   = 0;
    m_FirstFreeBlock                  = 0;
    m_NumBlocksUsed                   = 0;
    m_NumBlocksFreeAfterSVP           = 0;
    m_NumBlocksMarkedForBackup        = 0;
    m_NumBlocksToAddUntilOptIsReached = 0;
    m_StartBlockNoToSearchFreeBlock   = m_FirstFreeBlock;
    m_VolumeIsInCompressingMode       = false;

    /* m_LastFragmentedBlock remains undefined,and gets only a value when */
    /* it is used by the data writer to compress and balance the devices  */

    /* initialize the list containing the block states */
    /* and set all block states in the list to free    */
    if(! m_BlockStateList.Create( m_NumBlocks ))
    {
        FBM_Exception errMsg( __CONTEXT__, FBM_ERR_NO_MORE_MEMORY );
        RTE_Crash( errMsg );
    }

    /* determine number of sections */
    m_NumberOfSections = (m_NumBlocks >> m_LogSectionSize);

    if ((m_NumberOfSections << m_LogSectionSize) < m_NumBlocks)
        ++m_NumberOfSections;

    /* create indizees */
    if (! m_NumBlocksFreeInSection.Resize (m_NumberOfSections, 0)            ||
            ! m_NumBlocksMarkedForBackUpInSection.Resize (m_NumberOfSections, 0) ||
            ! m_NumBlocksFreeAfterSVPInSection.Resize (m_NumberOfSections, 0))
    {
        FBM_Exception errMsg( __CONTEXT__, FBM_ERR_NO_MORE_MEMORY );
        RTE_Crash( errMsg );
    }

    /* initialize the lists of counters which contain the number of free blocks per section, the number */
    /* of blocks marked for backup per section and  number of blocks marked for backup per section      */
    for (int iSection=0; iSection<m_NumberOfSections;iSection++)
    {
        m_NumBlocksFreeInSection            [iSection] = m_SectionSize;
        m_NumBlocksMarkedForBackUpInSection [iSection] = 0;
        m_NumBlocksFreeAfterSVPInSection    [iSection] = 0;
    }

    m_NumBlocksFreeInSection[m_NumberOfSections-1] = m_NumBlocks - ((m_NumberOfSections-1) << m_LogSectionSize);

    /* check consistency of the device */
#   if COMPILEMODE_MEO00 >= QUICK_MEO00 
    if( FBM_Check.ChecksLevel( 5 ))
        Verify ();
#   endif
}

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

void
FBM_DataVolume::RestoreAllBlockStatesMarkedForBackup (SAPDB_Int4 &NumBlocksRestoredToFreeAfterSVP)
{
    SAPDBTRACE_ROUTINE_DEBUG  ("FBM_DataVolume::RestoreBlockStatesMarkedforBackup", FBM_Trace, 5);

    SAPDBERR_ASSERT_STATE (!m_VolumeIsInCompressingMode);
    SAPDBERR_ASSERT_STATE (m_NumBlocksMarkedForBackup > 0);

    SAPDB_Int4     BeginOfThisSection;
    SAPDB_Int4     EndOfThisSection;
    SAPDB_Int4     iSection;
    SAPDB_Int4     iBlock;
    SAPDB_Bool     bRestoredStateIsFreeAfterSVP;

    NumBlocksRestoredToFreeAfterSVP = 0;

    for (iSection=0; (iSection<m_NumberOfSections) && (m_NumBlocksMarkedForBackup>0); ++iSection)
    {
        if (m_NumBlocksMarkedForBackUpInSection[iSection]>0)
        {
            BeginOfThisSection = SectionBegin(iSection);
            EndOfThisSection   = SectionEnd  (iSection);

            for (iBlock=BeginOfThisSection; iBlock<=EndOfThisSection;  ++iBlock)
            {
                if (GetBlockState (iBlock) == BlockState_BackUp)
                {
                    /* restore the state to the one it had before the backup */
                    RestoreBlockStateMarkedForBackup (iBlock, bRestoredStateIsFreeAfterSVP);
                    if (bRestoredStateIsFreeAfterSVP)
                        ++NumBlocksRestoredToFreeAfterSVP;

                    if (m_NumBlocksMarkedForBackUpInSection[iSection] == 0) break;
                }
            }
            SAPDBERR_ASSERT_STATE (m_NumBlocksMarkedForBackUpInSection[iSection] == 0);
        }
    }
    SAPDBERR_ASSERT_STATE (m_NumBlocksMarkedForBackup == 0);

    /* reset marker of last block read for back up */
    m_ActBlockNoForBackup = FBM_UNDEFINED;

    /* check consistency of the device */
#   if COMPILEMODE_MEO00 >= QUICK_MEO00 
    if( FBM_Check.ChecksLevel( 5 ))
        Verify ();
#   endif
}

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

SAPDB_UInt
FBM_DataVolume::CheckConverterFBM(SAPDB_Int volNo)
{
	SAPDB_UInt numBlocksOK = 0;
	SAPDB_UInt numBlocksBad = 0;
	SAPDB_UInt numBlocksFreeAfterSVP = 0;
	bool dummy;
	
	for (SAPDB_Int block = 1; block < m_NumBlocks; block++)
	{
		if (BlockState_BackUp == GetBlockState (block)) // OK
		{ 
			numBlocksOK++;
			RestoreBlockStateMarkedForBackup (block, dummy);
		}
		else if  (BlockState_Occupied == GetBlockState (block))     // error
		{
			numBlocksBad++;
			FBM_BlockNoConverter blkInfo(__CONTEXT__, volNo, block, GetBlockState (block));
			RTE_Message(blkInfo);
		}
		else if  (BlockState_FreeAfterSVP == GetBlockState (block))     // skip
		{
			numBlocksFreeAfterSVP++;
		}
	}
	
	FBM_Exception counter(__CONTEXT__,
	FBM_INFO_COUNTER,
	SAPDB_ToString( m_NumBlocks, _T_d ),
	SAPDB_ToString( m_NumBlocksUsed, _T_d ),
	SAPDB_ToString( m_NumBlocksFreeAfterSVP, _T_d ),
	SAPDB_ToString( numBlocksOK + numBlocksBad + numBlocksFreeAfterSVP + 1, _T_d ),  // + 1 for info page
	SAPDB_ToString( numBlocksBad, _T_d ),
	SAPDB_ToString( numBlocksFreeAfterSVP, _T_d ));
	RTE_Message(counter);
	
	return numBlocksOK;

}
    
/*---------------------------------------------------------------------------*/

SAPDB_Int4
FBM_DataVolume::GetMultFreeBlocks  (const SAPDB_Int4  NumBlocksWanted,
                                    SAPDB_Int4       &NumBlocksSupplied)
{
    SAPDBTRACE_ROUTINE_DEBUG  ("FBM_DataVolume::GetMultFreeBlocks", FBM_Trace, 5);

    /* this routine is called to get  a cluster of free blocks.  */
    /* before the routine is called it is already obvious that   */
    /* there must be at least one free block on this device      */

    SAPDB_Bool bFirstFreeBlockWasUsed = false;
    SAPDB_Bool bFreePairFound;
    SAPDB_Int4 FreePairBlockNo;
    SAPDB_Int4 LastSuppliedFreeBlock;
    SAPDB_Int4 FreeBlockNo;
    SAPDB_Int4 CurrSection;

    /* find free BlockNo */
    FindFreeBlock (m_StartBlockNoToSearchFreeBlock, FreeBlockNo, CurrSection,  bFirstFreeBlockWasUsed);

    /* check if this free block belongs to a cluster of free blocks. if not the next cluster is searched */
    if (((FreeBlockNo + 1) < m_NumBlocks) && (GetBlockState (FreeBlockNo + 1) == BlockState_Free ))
        bFreePairFound = true;
    else
    {
        FindPairOfFreeBlocks (FreeBlockNo, bFreePairFound, FreePairBlockNo, m_StartBlockNoToSearchFreeBlock);
        if (bFreePairFound)
        {
            bFirstFreeBlockWasUsed &= (FreeBlockNo == FreePairBlockNo);
            FreeBlockNo = FreePairBlockNo;
            CurrSection   = (FreeBlockNo >> m_LogSectionSize);
            SAPDBERR_ASSERT_STATE ((GetBlockState(FreeBlockNo)     == BlockState_Free) &&
                                   (GetBlockState(FreeBlockNo + 1) == BlockState_Free));
        }
    }

    /* change state of free block to occupied */
    m_BlockStateList.SetBlockState (FreeBlockNo,BlockState_Occupied);

    /* reserve all following blocks which are free too until NumBlocksWanted */
    /* free blocks are found or the first block which is not free occurs     */
    for (LastSuppliedFreeBlock=FreeBlockNo, NumBlocksSupplied = 1 ;
            ((NumBlocksSupplied<NumBlocksWanted)  &&
             (LastSuppliedFreeBlock + 1) < m_NumBlocks) &&
            (GetBlockState (LastSuppliedFreeBlock + 1) == BlockState_Free);
            NumBlocksSupplied++)
    {
        LastSuppliedFreeBlock++;
        m_BlockStateList.SetBlockState (LastSuppliedFreeBlock, BlockState_Occupied);
    }

    if (bFreePairFound)
        m_StartBlockNoToSearchFreeBlock = LastSuppliedFreeBlock;

    /* write found block no into vtrace */
    SAPDBTRACE_WRITELN( FBM_Trace, 5, "FreeBlockNo: " << FreeBlockNo
                        << " #WantedFreeB: "  << NumBlocksWanted
                        << " #FoundFreeB: "  << NumBlocksSupplied);

    /* update counter of used blocks */
    m_NumBlocksUsed += NumBlocksSupplied;

    /* update counter of free blocks in the current section */
    const SAPDB_Int4 CurrSectionEnd = SectionEnd(CurrSection);
    if (LastSuppliedFreeBlock <= CurrSectionEnd)
        m_NumBlocksFreeInSection[CurrSection]-=NumBlocksSupplied;
    else
    {
        m_NumBlocksFreeInSection[CurrSection]  -=(CurrSectionEnd          - FreeBlockNo + 1);
        m_NumBlocksFreeInSection[CurrSection+1]-=(LastSuppliedFreeBlock - CurrSectionEnd);
    }

    /* update number of last used block */
    if (LastSuppliedFreeBlock > m_LastUsedBlock)
        m_LastUsedBlock = LastSuppliedFreeBlock;

    /* update hint where to start to find the next free block */
    if (bFirstFreeBlockWasUsed || (m_FirstFreeBlock == FreeBlockNo))
        m_FirstFreeBlock = LastSuppliedFreeBlock + 1;

    /* update the counter which controlls the distribution of blocks onto devices */
    m_NumBlocksToAddUntilOptIsReached -=NumBlocksSupplied;

    /* check consistency of the device */
#   if COMPILEMODE_MEO00 >= QUICK_MEO00 
    if( FBM_Check.ChecksLevel( 5 ))
        Verify ();
#   endif

    return ( FreeBlockNo ) ;
}

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

SAPDB_Int4
FBM_DataVolume::GetFreeBlock  ()
{
    SAPDBTRACE_ROUTINE_DEBUG  ("FBM_DataVolume::GetFreeBlock", FBM_Trace, 5);

    /* this routine is called to get the adress of a free block. */
    /* before the routine is called it is already obvious that   */
    /* there must be at least one free block on this device      */

    SAPDB_Bool bFirstFreeBlockWasUsed = false;
    SAPDB_Int4 FreeBlockNo;
    SAPDB_Int4 CurrSection;

    /* find free BlockNo */
    FindFreeBlock (m_StartBlockNoToSearchFreeBlock, FreeBlockNo, CurrSection,  bFirstFreeBlockWasUsed);

    /* change state of free block to occupy */
    m_BlockStateList.SetBlockState (FreeBlockNo,BlockState_Occupied);

    /* write found block no into vtrace */
    SAPDBTRACE_WRITELN( FBM_Trace, 5, "FreeSBlockNo: " << FreeBlockNo);

    /* update counter of used blocks and free blocks in the current section */
    ++m_NumBlocksUsed;
    --m_NumBlocksFreeInSection[CurrSection];

    /* update number of last used block */
    if (FreeBlockNo > m_LastUsedBlock)
        m_LastUsedBlock = FreeBlockNo;

    /* update hint where to start to find the next and the first free block */
    m_StartBlockNoToSearchFreeBlock = FreeBlockNo + 1;
    if (bFirstFreeBlockWasUsed)
        m_FirstFreeBlock = FreeBlockNo + 1;

    /* update the counter which controlls the distribution of blocks onto devices */
    --m_NumBlocksToAddUntilOptIsReached;

    /* check consistency of the device */
#   if COMPILEMODE_MEO00 >= QUICK_MEO00 
    if( FBM_Check.ChecksLevel( 5 ))
        Verify ();
#   endif

    return ( FreeBlockNo ) ;
}

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

void
FBM_DataVolume::SetAllBlocksMarkedAsFreeAfterSVPToFree ()
{
    SAPDBTRACE_ROUTINE_DEBUG  ("FBM_DataVolume::SetAllBlocksMarkedAsFreeAfterSVPToFree", FBM_Trace, 5);

    SAPDB_Int4     BeginOfThisSection;
    SAPDB_Int4     EndOfThisSection;
    SAPDB_Int4     iSection;
    SAPDB_Int4     iBlock;

    /* go through all blocks starting with the largest block */
    for (iSection=m_LastUsedBlock>>m_LogSectionSize; (iSection>=0) && (m_NumBlocksFreeAfterSVP>0); --iSection)
    {
        if (m_NumBlocksFreeAfterSVPInSection[iSection] > 0)
        {
            BeginOfThisSection = SectionBegin(iSection);
            EndOfThisSection   = SectionEnd  (iSection);

            for (iBlock=EndOfThisSection; iBlock>=BeginOfThisSection ;--iBlock)
            {
                if (m_BlockStateList.GetBlockState (iBlock) == BlockState_FreeAfterSVP)
                {
                    m_BlockStateList.SetBlockState (iBlock,BlockState_Free);
                    ++m_NumBlocksFreeInSection[iSection];
                    --m_NumBlocksUsed;
                    --m_NumBlocksFreeAfterSVP;
                    --m_NumBlocksFreeAfterSVPInSection[iSection];
                    if (m_NumBlocksFreeAfterSVPInSection[iSection] == 0)
                    {
                        /* since this blockno must be the smalllest which became free    */
                        /* only this number has to be checked to be the first free block */
                        if (iBlock < m_FirstFreeBlock) m_FirstFreeBlock = iBlock;
                        break;
                    }
                }
            }
            SAPDBERR_ASSERT_STATE (m_NumBlocksFreeAfterSVPInSection[iSection] == 0);
        }
    }
    SAPDBERR_ASSERT_STATE (m_NumBlocksFreeAfterSVP == 0);

    /* find the last used block */
    if (m_NumBlocksUsed == 0)
        m_LastUsedBlock = 0;
    else
    {

        /* first search first last used section */
        for (iSection = (m_LastUsedBlock >> m_LogSectionSize); iSection >= 0;--iSection)
        {
            if (m_NumBlocksFreeInSection [iSection] < (EndOfThisSection-BeginOfThisSection+1))
            {
                BeginOfThisSection = SectionBegin(iSection);
                EndOfThisSection   = SectionEnd  (iSection);
                break;
            }
        }

        /* find last used block in last used section */
        for (iBlock=SAPDB_MIN(EndOfThisSection,m_LastUsedBlock);iBlock>=BeginOfThisSection;--iBlock)
        {
            if (m_BlockStateList.GetBlockState (iBlock) != BlockState_Free)
            {
                m_LastUsedBlock = iBlock;
                break;
            }
        }
    }

    /* check that no blocks remain in the state free after save point */
    SAPDBERR_ASSERT_STATE (m_NumBlocksFreeAfterSVP == 0);

    /* check consistency of the device */
#   if COMPILEMODE_MEO00 >= QUICK_MEO00 
    if( FBM_Check.ChecksLevel( 5 ))
        Verify ();
#   endif
}

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


void
FBM_DataVolume::GetNextBlocksForBackUp (const SAPDB_Int4   MaxNumBlocksWanted,
                                        SAPDB_Int4        &SuppliedNumBlocks,
                                        SAPDB_Int4        &BlockNo)

{
    SAPDBTRACE_ROUTINE_DEBUG  ("FBM_DataVolume::GetNextBlocksForBackUp", FBM_Trace, 5);

    /* check that the device ist not in the compression mode */
    SAPDBERR_ASSERT_STATE (!m_VolumeIsInCompressingMode);

    /* initialize output parameter */
    SuppliedNumBlocks = 0;

    if (m_ActBlockNoForBackup == FBM_UNDEFINED)
    {
        BlockNo = FBM_UNDEFINED;
        return;
    }

    /* search next block marked for back up starting from the current backup block no */
    SAPDB_Int4 CurrSection    = m_ActBlockNoForBackup>>m_LogSectionSize;
    SAPDB_Int4 CurrSectionEnd = SectionEnd(CurrSection);
    SAPDB_Bool bBlockFound    = false;

    /* go through current section and find the first block which is marked as back up */
    while (m_ActBlockNoForBackup <= CurrSectionEnd)
    {
        if (GetBlockState (m_ActBlockNoForBackup) == BlockState_BackUp)
        {
            /* check if the found block is really marked for backup and not for compression */
            SAPDBERR_ASSERT_STATE ((GetBlockStateAfterBackup (m_ActBlockNoForBackup) == BlockState_Occupied) ||
                                   (GetBlockStateAfterBackup (m_ActBlockNoForBackup) == BlockState_FreeAfterSVP));
            bBlockFound = true;
            break;
        }
        ++m_ActBlockNoForBackup;
    }

    /* goto next section which contains blocks marked for backup */
    if (!bBlockFound)
    {
        SAPDB_Bool bSectionFound  = false;

        while ((++CurrSection)<m_NumberOfSections)
        {
            if (m_NumBlocksMarkedForBackUpInSection[CurrSection] > 0)
            {
                bSectionFound = true;
                break;
            }
        }

        if (!bSectionFound)
            m_ActBlockNoForBackup = FBM_UNDEFINED;
        else
        {
            /* go through section and search first block marked for backup */
            m_ActBlockNoForBackup = SectionBegin(CurrSection);
            CurrSectionEnd        = SectionEnd(CurrSection);
            while (m_ActBlockNoForBackup <= CurrSectionEnd)
            {
                if (GetBlockState (m_ActBlockNoForBackup) == BlockState_BackUp)
                {
                    bBlockFound = true;
                    break;
                }
                ++m_ActBlockNoForBackup;
            }

            /* the found section must contain blocks marked for back up */
            SAPDBERR_ASSERT_STATE (bBlockFound);
        }
    }

    /* set return value where the next set of neighbouring blocks marked for back up starts */
    BlockNo = m_ActBlockNoForBackup;

    /* check if there is really a next block which is          */
    /* marked for back up and not yet fetched by an datawriter */
    if (bBlockFound)
    {
        /* check out if the following blocks are also in the state BlockState_BackUp */
        SuppliedNumBlocks = 1;
        ++m_ActBlockNoForBackup;

        while ((SuppliedNumBlocks      <  MaxNumBlocksWanted) &&
                (m_ActBlockNoForBackup  <= m_LastUsedBlock)    &&
                (GetBlockState (m_ActBlockNoForBackup) == BlockState_BackUp))
        {

            /* check if the found block is really marked for backup and not for compression */
            SAPDBERR_ASSERT_STATE ((GetBlockStateAfterBackup (m_ActBlockNoForBackup) == BlockState_Occupied) ||
                                   (GetBlockStateAfterBackup (m_ActBlockNoForBackup) == BlockState_FreeAfterSVP));
            ++ SuppliedNumBlocks;
            ++ m_ActBlockNoForBackup;
        }

        /* set marker to display that there are no more blocks for backup */
        if (m_ActBlockNoForBackup > m_LastUsedBlock)
            m_ActBlockNoForBackup = FBM_UNDEFINED;
    }

    /* check consistency of the device */
#   if COMPILEMODE_MEO00 >= QUICK_MEO00 
    if( FBM_Check.ChecksLevel( 5 ))
        Verify ();
#   endif
}

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

void
FBM_DataVolume::GetDeviceStatistics (SAPDB_Int4 &NumBlocks,
                                     SAPDB_Int4 &NumBlocksFree,
                                     SAPDB_Int4 &NumBlockFreeAfterSVP) const
{
    NumBlocks            = m_NumBlocks;
    NumBlocksFree        = m_NumBlocks - m_NumBlocksUsed;
    NumBlockFreeAfterSVP = m_NumBlocksFreeAfterSVP;
}

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

void
FBM_DataVolume::Verify ()
{
    SAPDBTRACE_ROUTINE_DEBUG  ("FBM_DataVolume::Verify", FBM_Trace, 5);

    SAPDBERR_ASSERT_STATE (m_NumBlocks == m_BlockStateList.Capacity());

    SAPDB_Int4 NumBlocksFree        = 0;
    SAPDB_Int4 NumBlockFreeAfterSVP = 0;

    SAPDB_Int4 FirstFreeBlock                = m_NumBlocks;
    SAPDB_Int4 LastUsedBlock                 = 0;
    SAPDB_Int4 NumBlocksMarkedForBackUp      = 0;
    SAPDB_Int4 NumBlocksMarkedForCompression = 0;

    /* note: this can be written a better way when exceptions are allowed */
    FBM_CounterList  NumBlocksFreeInSection( m_NumBlocksFreeInSection.GetAllocator() );
    FBM_CounterList  NumBlocksFreeAfterSVPInSection( m_NumBlocksFreeAfterSVPInSection.GetAllocator() );
    FBM_CounterList  NumBlocksMarkedForBackUpInSection( m_NumBlocksMarkedForBackUpInSection.GetAllocator() );

    if (!NumBlocksFreeInSection.Resize(m_NumberOfSections, 0)         ||
            !NumBlocksFreeAfterSVPInSection.Resize(m_NumberOfSections, 0) ||
            !NumBlocksMarkedForBackUpInSection.Resize(m_NumberOfSections, 0))
    {
        FBM_Exception Error(__CONTEXT__, FBM_NO_MEMORY_FOR_VERIFICATION, SAPDB_ToString (m_NumberOfSections, _T_d));
        RTE_Message (Error);
        return;
    }


    /* go through all entries in the blocklist and count the occurence of each state */
    for (SAPDB_Int4 iBlockNo=0; iBlockNo < m_NumBlocks; ++iBlockNo)
    {
        FBM_BlockState BlockStateAfterBackUp = GetBlockStateAfterBackup(iBlockNo);

        switch (GetBlockState(iBlockNo))
        {
        case BlockState_Free:
            ++NumBlocksFree;
            ++NumBlocksFreeInSection [iBlockNo >> m_LogSectionSize];
            if (FirstFreeBlock > iBlockNo)
                FirstFreeBlock = iBlockNo;
            SAPDBERR_ASSERT_STATE (BlockState_Free == BlockStateAfterBackUp);
            break;
        case BlockState_Occupied:
            if (LastUsedBlock < iBlockNo)
                LastUsedBlock = iBlockNo;
            SAPDBERR_ASSERT_STATE (BlockState_Free == BlockStateAfterBackUp);
            break;
        case BlockState_FreeAfterSVP:
            ++NumBlockFreeAfterSVP;
            ++NumBlocksFreeAfterSVPInSection [iBlockNo >> m_LogSectionSize];
            if (LastUsedBlock < iBlockNo)
                LastUsedBlock = iBlockNo;
            SAPDBERR_ASSERT_STATE (BlockState_Free == BlockStateAfterBackUp);
            break;
        case BlockState_BackUp:
            ++ NumBlocksMarkedForBackUp;
            ++ NumBlocksMarkedForBackUpInSection [iBlockNo >> m_LogSectionSize];
            if (m_VolumeIsInCompressingMode)
            {
                //blocks are marked for compression
                SAPDBERR_ASSERT_STATE ((BlockState_BackUp  == BlockStateAfterBackUp) ||
                                       (BlockState_Free    == BlockStateAfterBackUp));
            }
            else
            {
                //blocks are marked for back up
                SAPDBERR_ASSERT_STATE ((BlockState_Occupied     == BlockStateAfterBackUp) ||
                                       (BlockState_FreeAfterSVP == BlockStateAfterBackUp));
            }

            if (LastUsedBlock < iBlockNo)
                LastUsedBlock = iBlockNo;
            if ((BlockState_BackUp == BlockStateAfterBackUp) ||
                    (BlockState_Free   == BlockStateAfterBackUp  ))
                ++NumBlocksMarkedForCompression;
            break;
        default:
            /* write error message into knldiag and vtrace */
            FBM_IllegalState IllegalState(__CONTEXT__, iBlockNo, GetBlockState (iBlockNo), BlockStateAfterBackUp);
            RTE_Crash(IllegalState);
        }
    }

    /* check counters and pointers */
    SAPDBERR_ASSERT_STATE (NumBlocksMarkedForBackUp == m_NumBlocksMarkedForBackup);
    SAPDBERR_ASSERT_STATE (m_NumBlocksUsed == (m_NumBlocks - NumBlocksFree));
    SAPDBERR_ASSERT_STATE (m_NumBlocksFreeAfterSVP == NumBlockFreeAfterSVP);
    SAPDBERR_ASSERT_STATE (FirstFreeBlock >= m_FirstFreeBlock);
    SAPDBERR_ASSERT_STATE (LastUsedBlock <= m_LastUsedBlock);

    /* check that either all blocks are marked for backup or for compression */
    SAPDBERR_ASSERT_STATE ((NumBlocksMarkedForCompression == NumBlocksMarkedForBackUp) ||
                           (NumBlocksMarkedForCompression == 0));

    /* check that the device is not used for a backup and gets compressed at the same time */
    SAPDBERR_ASSERT_STATE (m_VolumeIsInCompressingMode ==  (NumBlocksMarkedForCompression>0));
    SAPDBERR_ASSERT_STATE (m_VolumeIsInCompressingMode  || (m_LastFragmentedBlock == FBM_UNDEFINED));
    SAPDBERR_ASSERT_STATE (!m_VolumeIsInCompressingMode || (m_ActBlockNoForBackup == FBM_UNDEFINED));

    /* check content of the counters of free blocks in each section */
    for (int iSection=0; iSection<m_NumberOfSections;iSection++)
    {
        SAPDBERR_ASSERT_STATE (NumBlocksFreeInSection[iSection]            == m_NumBlocksFreeInSection[iSection]);
        SAPDBERR_ASSERT_STATE (NumBlocksFreeAfterSVPInSection[iSection]    == m_NumBlocksFreeAfterSVPInSection[iSection]);
        SAPDBERR_ASSERT_STATE (NumBlocksMarkedForBackUpInSection[iSection] == m_NumBlocksMarkedForBackUpInSection[iSection]);
    }
}

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

const SAPDB_Int2
FBM_DataVolume::m_LogSectionSize = 8;

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

const SAPDB_Int2
FBM_DataVolume::m_SectionSize    = 1 << (FBM_DataVolume::m_LogSectionSize);

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

void
FBM_DataVolume::FindPairOfFreeBlocks (SAPDB_Int4  StartSearchBlockNo,
                                      SAPDB_Bool &bFreePairFound,
                                      SAPDB_Int4 &FreePairBlockNo,
                                      SAPDB_Int4 &StartBlockNoForNextSearch) const
{
    SAPDB_Int4 SearchBlockNo           = StartSearchBlockNo;
    const SAPDB_Int4 OptLastUsedBlock  = GetOptLastUsedBlock();
    const SAPDB_Int4 MaxBlocksToScan   = 256;

    for (SAPDB_Int4 BlockCount=0; BlockCount < MaxBlocksToScan; ++BlockCount)
    {
        if (GetBlockState(SearchBlockNo) != BlockState_Free || ((SearchBlockNo + 1) >= m_NumBlocks))
            ++SearchBlockNo;
        else
        {
            if (GetBlockState(SearchBlockNo + 1) != BlockState_Free)
                SearchBlockNo += 2;
            else
            {
                bFreePairFound            = true;
                FreePairBlockNo           = SearchBlockNo;
                StartBlockNoForNextSearch = SearchBlockNo;
                return;
            }
        }

        if (SearchBlockNo > OptLastUsedBlock)
            SearchBlockNo = m_FirstFreeBlock;
    }

    bFreePairFound            = false;
    StartBlockNoForNextSearch = SearchBlockNo;
}

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

void
FBM_DataVolume::FindFreeBlock (SAPDB_Int4  StartSearchBlockNo,
                               SAPDB_Int4 &FreeBlockNo,
                               SAPDB_Int4 &CurrSection,
                               SAPDB_Bool &bFirstFreeBlockWasUsed)
{
    SAPDBERR_ASSERT_STATE ((m_NumBlocks - m_NumBlocksUsed) > 0);

    bool             bFreeBlockFound   = false;

    const SAPDB_Int4 OptLastUsedBlock  = GetOptLastUsedBlock();

    /* get starting position and section to search the first free block */
    SAPDB_Int4 SearchBlockNo = StartSearchBlockNo + 1;
    if (SearchBlockNo > OptLastUsedBlock)
    {
        SearchBlockNo = m_FirstFreeBlock;
        bFirstFreeBlockWasUsed = true;
    }

    CurrSection = (SearchBlockNo >> m_LogSectionSize);

    /* check if there is a free block in the current section behind the start position for search */
    if (m_NumBlocksFreeInSection[CurrSection] > 0)
    {
        const SAPDB_Int4 StopBlockNoInSection = SectionEnd(CurrSection);
        while(SearchBlockNo<=StopBlockNoInSection)
        {
            if (GetBlockState (SearchBlockNo) == BlockState_Free)
            {
            bFreeBlockFound = true;
            break;
            }
            ++SearchBlockNo;
        }
    }

    /* if no free block was found goto next section with free blocks */
    if (!bFreeBlockFound)
    {
        SAPDB_Int4 OptLastUsedSection = (OptLastUsedBlock >> m_LogSectionSize);

        /* find the next section with a free block */
        do
        {
            if (CurrSection < OptLastUsedSection)
                ++CurrSection;
            else
            {
                SAPDBERR_ASSERT_STATE (bFirstFreeBlockWasUsed == false);

                CurrSection = (m_FirstFreeBlock >> m_LogSectionSize);
                bFirstFreeBlockWasUsed = true;
            }
        }
        while (m_NumBlocksFreeInSection[CurrSection] == 0);

        /* find the free block within the section which was found to contain free blocks */
        const SAPDB_Int4 StopBlockNoInSection  = SectionEnd(CurrSection);
        const SAPDB_Int4 StartBlockNoInSection = SectionBegin(CurrSection);
        for (SearchBlockNo=StartBlockNoInSection; SearchBlockNo<=StopBlockNoInSection; ++SearchBlockNo)
        {
            if (GetBlockState (SearchBlockNo) == BlockState_Free) break;
        }
    }
    FreeBlockNo = SearchBlockNo;

    /* check that a free block was found */
    SAPDBERR_ASSERT_STATE (m_BlockStateList.GetBlockState(FreeBlockNo) == BlockState_Free);
}

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

void
FBM_DataVolume::RestoreBlockStateMarkedForBackup (SAPDB_Int4  BlockNo,
        SAPDB_Bool &bRestoredStateIsFreeAfterSVP)
{
    /* check if the state to be restored is correct */
    SAPDBERR_ASSERT_STATE ((GetBlockState             (BlockNo) == BlockState_BackUp)       &&
                           ((GetBlockStateAfterBackup (BlockNo) == BlockState_Occupied) ||
                            (GetBlockStateAfterBackup  (BlockNo) == BlockState_FreeAfterSVP)));

    /* update counter of blocks marked for back up */
    --m_NumBlocksMarkedForBackup;
    --m_NumBlocksMarkedForBackUpInSection[BlockNo>>m_LogSectionSize];

    /* restore the state to the one it had before the backup. If  */
    /* the restored state (=return value of the restore function) */
    /* is FreeAfterSVP the appropriate counter is updated         */
    if (m_BlockStateList.RestoreBlockStateMarkedForBackup (BlockNo) != BlockState_FreeAfterSVP)
        bRestoredStateIsFreeAfterSVP = false;
    else
    {
        ++m_NumBlocksFreeAfterSVP;
        ++m_NumBlocksFreeAfterSVPInSection[BlockNo>>m_LogSectionSize];
        bRestoredStateIsFreeAfterSVP = true;
    }

#   if COMPILEMODE_MEO00 >= QUICK_MEO00 
    if( FBM_Check.ChecksLevel( 5 ))
        Verify ();
#   endif
}

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

void
FBM_DataVolume::SetBlockStateMarkedForCompressionToFreeAfterSVP (SAPDB_Int4 BlockNo)
{

    /* check if the FBM is in the compression mode where this state can only exist */
    SAPDBERR_ASSERT_STATE (m_VolumeIsInCompressingMode);

    /* check if the transition is correct */
    SAPDBERR_ASSERT_STATE ((GetBlockStateAfterBackup (BlockNo) == BlockState_BackUp) &&
                           (GetBlockState            (BlockNo) == BlockState_BackUp));

    /* set new block state */
    m_BlockStateList.SetBlockStateAfterBackup (BlockNo, BlockState_Free);
    m_BlockStateList.SetBlockState            (BlockNo, BlockState_FreeAfterSVP);

    /* update counter of blocks marked for back up */
    --m_NumBlocksMarkedForBackup;
    --m_NumBlocksMarkedForBackUpInSection[BlockNo>>m_LogSectionSize];

    /* remove compression mode if there are no more blocks to compress */
    m_VolumeIsInCompressingMode = (m_NumBlocksMarkedForBackup != 0);

    /* update counter of blocks marked as free after savepoint */
    ++m_NumBlocksFreeAfterSVP;
    ++m_NumBlocksFreeAfterSVPInSection[BlockNo>>m_LogSectionSize];

#   if COMPILEMODE_MEO00 >= QUICK_MEO00 
    if( FBM_Check.ChecksLevel( 5 ))
        Verify ();
#   endif
}

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

void
FBM_DataVolume::Dump(
    Kernel_Dump  &dump,
    const SAPDB_Int4 devNo) const
{
    struct FBMDataDevice   fbmDev;

    fbmDev.dmpDevNo                           = devNo;
    fbmDev.dmpLastUsedBlock                   = m_LastUsedBlock;
    fbmDev.dmpFirstFreeBlock                  = m_FirstFreeBlock;
    fbmDev.dmpLastFragmentedBlock             = m_LastFragmentedBlock;
    fbmDev.dmpFirstFragmentedBlock            = m_FirstFragmentedBlock;
    fbmDev.dmpNumBlocks                       = m_NumBlocks;
    fbmDev.dmpNumBlocksUsed                   = m_NumBlocksUsed;
    fbmDev.dmpNumBlocksMarkedForBackup        = m_NumBlocksMarkedForBackup;
    fbmDev.dmpNumBlocksFreeAfterSVP           = m_NumBlocksFreeAfterSVP;
    fbmDev.dmpNumBlocksToAddUntilOptIsReached = m_NumBlocksToAddUntilOptIsReached;
    fbmDev.dmpNumberOfSections                = m_NumberOfSections;
    fbmDev.dmpActBlockNoForBackup             = m_ActBlockNoForBackup;
    fbmDev.dmpStartBlockNoToSearchFreeBlock   = m_StartBlockNoToSearchFreeBlock;
    fbmDev.dmpDeviceIsInCompressingMode       = m_VolumeIsInCompressingMode;
    fbmDev.dmpFiller1                         = false;
    fbmDev.dmpFiller2                         = 0;

    dump.InsertEntry( Kernel_Dump::DmpFBMDataDevice,
                      Kernel_DumpPage::Entry( &fbmDev, sizeof( fbmDev )));
}
