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

  module      : Pager_Writer.cpp

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

  responsible : Henrik

  special area:
  description : This description is currently not used, as
    documentation is only generated from header files
-
  last changed: 2002-03-07
  see also    :

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

  copyright:    (c) 2001-2004 SAP AG


    ========== 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 "Pager/Pager_Writer.hpp"
#include "Pager/Pager_Types.hpp"
#include "Pager/Pager_Controller.hpp"
#include "Pager/Pager_Exception.hpp"
#include "Pager/Pager_Messages.hpp"
#include "RunTime/RTE_Message.hpp"
#include "KernelCommon/Kernel_IAdminRestartShutdown.hpp"
#include "IOManager/IOMan_IDataIO.hpp"
#include "FreeBlockManagement/FBM_IManager.hpp"
#include "Converter/Converter_IPageIO.hpp"


#include "heo56.h"          // vsuspend
#include "hbd20_5.h"        // bd20*
#include "hgg04.h"          // gg04isStaticPage

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



/*===========================================================================*
 *  MACROS                                                                   *
 *===========================================================================*/



/*===========================================================================*
 *  LOCAL CLASSES, STRUCTURES, TYPES, UNIONS ...                             *
 *===========================================================================*/



/*===========================================================================*
 *  STATIC/INLINE FUNCTIONS (PROTOTYPES)                                     *
 *===========================================================================*/



/*===========================================================================*
 *  METHODS                                                                  *
 *===========================================================================*/


Pager_Writer::Pager_Writer(tgg00_TransContext &t)
        : m_reqFlush(RTEMem_Allocator::Instance()),
        m_trans(t),
        m_err(t.trError_gg00),
        m_pagerController(Pager_Controller::GetInstance()),
        m_blockCount(g01io_block_count()),
        m_pages(*(REINTERPRET_CAST( SAPDBMem_IRawAllocator*,                // PTS 1117889
                                    t.trAllocator_gg00)) , UTF8( "Pager:WriteDataPages" ))
{
};




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


void Pager_Writer::Initialize(){
    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::Initialize", Pager_Trace, 5 );

    const SAPDB_Int numPagerWriter = g01maxdatawriter();
    const SAPDB_Int numCaches      = g01region_cnt (rgnData_egg00);

    m_pages.Initialize(m_blockCount);

    m_reqFlush.Initialize(numCaches,false);

    CleanupActiveCache();

    m_myIndex = m_pagerController.RegisterNextPagerWriter(m_trans.trTaskId_gg00, this);

    PagerWriterIterator current = Begin();
    const PagerWriterIterator& iterEnd = End();

    Pager_Exception StartTaskID(__CONTEXT__, PAGER_WRITER_START_TASKID,
                                SAPDB_ToString(m_trans.trTaskId_gg00, _T_d));
    RTE_Message (StartTaskID);
    Pager_Exception FirstCache(__CONTEXT__, PAGER_WRITER_FIRST_CACHE,
                               SAPDB_ToString(current.value(), _T_d));
    RTE_Message (FirstCache);

    current++;

    while (current != iterEnd) {
        Pager_Exception NextCache(__CONTEXT__, PAGER_WRITER_NEXT_CACHE,
                                  SAPDB_ToString(current.value(), _T_d));
        RTE_Message (NextCache);

        current++;
    }
    Run();
}

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

void Pager_Writer::CleanupActiveCache(){
    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::CleanupActiveCache", Pager_Trace, 5 );

    const SAPDB_Int numCaches = g01region_cnt (rgnData_egg00);

    m_reqFlush.Initialize(numCaches,false);
}

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

void Pager_Writer::ReqFlushForDataCache(SAPDB_Int regOffset){

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::ReqFlushForDataCache", Pager_Trace, 5 );

    m_reqFlush[regOffset] = true;
}

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

tsp00_TaskId Pager_Writer::GetTaskId() const{

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::GetTaskId", Pager_Trace, 5 );

    return m_trans.trTaskId_gg00;
}

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

void Pager_Writer::Run() {

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::Run", Pager_Trace, 5 );

    const SAPDB_Bool isArchive = g01is_archive();
    SAPDB_Bool found;

    while(true) {
        if ((e_ok != m_err ) &&
                (e_shutdown != m_err)) {        // PTS 1107037 2000-06-26
            Kernel_IAdminRestartShutdown::Instance().Offline(m_err, true);
            break;
        }

        // scan for preDisplace Requests ...
        PagerWriterIterator iter = Begin();
        const PagerWriterIterator& iterEnd = End();
        do {
            found = m_reqFlush[iter.value()];
            if  (!found){
                iter++;
            }
        } while ((iter != iterEnd) && !found);

        if ((! found)  ||
                (m_pagerController.IsPreparePagerWriter()) ||
                (e_shutdown == m_err)) {          // PTS 1107037 TS 2000-06-26
            m_err  = e_ok; // PTS 1107037 TS 2000-06-26 *)

            m_pagerController.SuspendPagerWriter(m_myIndex, m_trans.trTaskId_gg00);  // sleep

        }
        switch(m_pagerController.GetPagerOrder()) {
        case poFlushDataCacheBeforePrepare :
            if (isArchive)
            {
                FlushDataCacheBeforePrepareSequential();
            }
            else
            {
                FlushDataCacheBeforePrepare();
            }
            break;
        case poFlushDataCacheAfterPrepare :
            if (isArchive)
            {
                FlushDataCacheAfterPrepareSequential();
            }
            else
            {
                FlushDataCacheAfterPrepare();
            }
            break;
        case poReadConverterCache:
            ReadConverter( SAPDB_FALSE );
            break;
        case poReadSnapShotConverter:
            ReadConverter( SAPDB_TRUE );
            break;
        case poFlushConverterCache:
            FlushConverter();
            break;
        default:
            if (isArchive)
            {
                PreDisplaceIOSequential ();
            }
            else
            {
                PreDisplaceIO();
            }
            break;
        }
    }
}

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

void Pager_Writer::PreDisplaceIO() {

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::PreDisplaceIO", Pager_Trace, 5 );

    SAPDB_Bool              found;
    SAPDB_Bool1             dummy;
    SAPDB_Bool1             useIOChain;
    SAPDB_Int2              curPageCnt;
    SAPDB_Int               pageCnt;
    tbd00_NodePtrList       nodePtrList;
    tbd2_data_cb_flush_list cbptr_list;

    IOMan_BlockAddress startBlock;
    IOMan_ClusterAddress cluster;

    found      = false;
    useIOChain = SAPDB_FALSE;


    PagerWriterIterator iter = Begin();
    const PagerWriterIterator& iterEnd = End();

    do {
        found = m_reqFlush[iter.value()];
        if  (!found){
            iter++;
        }
    } while ((iter != iterEnd) && !found);


    while ((e_ok == m_err) &&
            (iter != iterEnd) &&
            !m_pagerController.IsPreparePagerWriter()){

        pageCnt = m_blockCount;

        const SAPDB_Bool bReqSequential = SAPDB_FALSE;

        cluster = FBM_IManager::Instance().GetMultFreeBlocks(
                      m_trans.trTaskId_gg00, pageCnt, bReqSequential, m_err);

        startBlock = cluster.GetBlockAddress(0);
        pageCnt    = cluster.GetBlockCount();

        if (e_ok == m_err){
            curPageCnt = pageCnt;
            bd20PreDisplaceIO (m_trans.trTaskId_gg00, iter.value(), m_err, useIOChain,
                               curPageCnt, nodePtrList, cbptr_list, dummy);
        }

        if (e_ok == m_err) {
            if (curPageCnt > 0) {
                cluster.SetCluster(startBlock, curPageCnt);
                WriteDataPages (m_trans, nodePtrList, 0, cluster);
                if (e_ok == m_err){
                    bd20ResetIOState(m_trans.trTaskId_gg00, iter.value(), curPageCnt, cbptr_list);
                    pageCnt  -= curPageCnt;
                    startBlock.SetBlockNo(startBlock.GetBlockNo() + curPageCnt);
                    cluster.SetCluster(startBlock, pageCnt);
                }
            }
            if  ((pageCnt > 0) && (e_ok == m_err)) {
                useIOChain = SAPDB_FALSE;

                m_reqFlush[iter.value()] = false;

                do {
                    found = m_reqFlush[iter.value()];
                    if  (!found){
                        iter++;
                    }
                } while ((iter != iterEnd) && !found);

                FBM_IManager::Instance().SetClusterStateToFree(m_trans.trTaskId_gg00, cluster);

            }
        }
    }
}

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


void Pager_Writer::PreDisplaceIOSequential() {

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::PreDisplaceIOSequential", Pager_Trace, 5 );

    SAPDB_Bool              found;
    SAPDB_Bool1             isSequential;
    SAPDB_Bool1             useIOChain;
    IOMan_BlockCount        oPageCnt;
    SAPDB_Int               pagesLeftToWrite;
    SAPDB_Int2              iPageCnt;
    tbd00_NodePtrList       nodePtrList;
    tbd2_data_cb_flush_list cbptr_list;

    IOMan_ClusterAddress cluster;

    found      = false;
    useIOChain = SAPDB_FALSE;


    PagerWriterIterator iter = Begin();
    const PagerWriterIterator& iterEnd = End();

    do {
        found = m_reqFlush[iter.value()];
        if  (!found){
            iter++;
        }
    } while ((iter != iterEnd) && !found);


    while ((e_ok == m_err) &&
            (iter != iterEnd) &&
            !m_pagerController.IsPreparePagerWriter())
    {

        iPageCnt = m_blockCount;
        bd20PreDisplaceIO (m_trans.trTaskId_gg00, iter.value(), m_err, useIOChain,
                           iPageCnt, nodePtrList, cbptr_list, isSequential);

        pagesLeftToWrite = iPageCnt;

        while ((e_ok == m_err) && (pagesLeftToWrite > 0) )
        {
            cluster = FBM_IManager::Instance().GetMultFreeBlocks(
                          m_trans.trTaskId_gg00, pagesLeftToWrite, isSequential,  m_err);
            if (e_ok != m_err)
            {
                break;
            }
            oPageCnt   = cluster.GetBlockCount();

            WriteDataPages (m_trans, nodePtrList, iPageCnt-pagesLeftToWrite, cluster);

            pagesLeftToWrite -= oPageCnt;
        }

        if (e_ok == m_err){
            bd20ResetIOState(m_trans.trTaskId_gg00, iter.value(), iPageCnt, cbptr_list);
        }

        if  ((0 == iPageCnt) && (e_ok == m_err)) {  // finished with datacache segment
            useIOChain = SAPDB_FALSE;

            m_reqFlush[iter.value()] = false;

            do {
                found = m_reqFlush[iter.value()];
                if  (!found){
                    iter++;
                }
            } while ((iter != iterEnd) && !found);
        }
    }
}

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

void Pager_Writer::FlushDataCacheBeforePrepare () {  // non sequential

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::FlushDataCacheBeforePrepare", Pager_Trace, 5 );

    SAPDB_Bool1             dummy;
    SAPDB_Bool1             isSequential;
    IOMan_BlockCount        pageCnt;
    SAPDB_Int2              curPageCnt;
    SAPDB_Int               totalIO;
    SAPDB_Int               totalPages;
    SAPDB_Int               totalIOElems;
    tbd00_NodePtrList       nodePtrList;
    tbd2_data_cb_flush_list cbptrList;

    PagerWriterIterator current = Begin();
    const PagerWriterIterator& iterEnd = End();

    IOMan_BlockAddress  startBlock;
    IOMan_ClusterAddress cluster;

    m_err      = e_ok;
    totalPages = 0;
    totalIO    = 0;

    do {  // all assigned cache segments
        totalIOElems = bd20IOElemCount (m_trans.trTaskId_gg00, current.value() );
        do {
            pageCnt = m_blockCount;

            const SAPDB_Bool bReqSequential = SAPDB_FALSE;

            cluster = FBM_IManager::Instance().GetMultFreeBlocks(
                          m_trans.trTaskId_gg00, pageCnt, bReqSequential,  m_err);
            startBlock = cluster.GetBlockAddress(0);
            pageCnt    = cluster.GetBlockCount();

            while ((e_ok == m_err) && (pageCnt > 0) && (totalIOElems > 0)){
                curPageCnt = pageCnt;
                bd20GetPageForIO (m_trans.trTaskId_gg00, current.value(), m_err,
                                  curPageCnt, nodePtrList, cbptrList, dummy, isSequential);
                if  (0 == curPageCnt){
                    totalIOElems = 0;
                }
                else {
                    totalIOElems -= curPageCnt;
                }

                if ((e_ok == m_err) && (curPageCnt >0))
                {
                    totalIO++;
                    totalPages += curPageCnt;

                    cluster.SetCluster(startBlock, curPageCnt);
                    WriteDataPages (m_trans, nodePtrList, 0, cluster);

                    if (e_ok == m_err){
                        bd20ResetIOState (m_trans.trTaskId_gg00, current.value(), curPageCnt, cbptrList);
                        pageCnt   -= curPageCnt;
                        startBlock.SetBlockNo(startBlock.GetBlockNo() + curPageCnt);
                        cluster.SetCluster(startBlock, pageCnt);
                    }
                }
            }
        } while ((e_ok == m_err) && (totalIOElems > 0 ));

        FBM_IManager::Instance().SetClusterStateToFree(m_trans.trTaskId_gg00, cluster);

        current++;
    } while ((e_ok == m_err) && (current != iterEnd));

    m_pagerController.AddDataIOTotal(totalIO);
    m_pagerController.AddDataPagesTotal(totalPages);
}

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

void Pager_Writer::FlushDataCacheBeforePrepareSequential () {

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::FlushDataCacheBeforePrepareSequential", Pager_Trace, 5 );

    SAPDB_Bool1             dummy;
    SAPDB_Bool1             isSequential;
    IOMan_BlockCount        oPageCnt;
    SAPDB_Int2              iPageCnt;
    SAPDB_Int               pagesLeftToWrite;
    SAPDB_Int               totalIO;
    SAPDB_Int               totalPages;
    SAPDB_Int               totalIOElems;
    tbd00_NodePtrList       nodePtrList;
    tbd2_data_cb_flush_list cbptrList;

    PagerWriterIterator current = Begin();
    const PagerWriterIterator& iterEnd = End();

    IOMan_ClusterAddress cluster;

    m_err      = e_ok;
    totalPages = 0;
    totalIO    = 0;

    do {  // all assigned cache segments
        totalIOElems = bd20IOElemCount (m_trans.trTaskId_gg00, current.value() );
        do {
            iPageCnt = m_blockCount;

            bd20GetPageForIO (m_trans.trTaskId_gg00, current.value(), m_err,
                              iPageCnt, nodePtrList, cbptrList, dummy, isSequential);
            if  (0 == iPageCnt){
                break;
            }

            pagesLeftToWrite = iPageCnt;

            while ((e_ok == m_err) && (pagesLeftToWrite > 0) ){
                cluster = FBM_IManager::Instance().GetMultFreeBlocks(
                              m_trans.trTaskId_gg00, pagesLeftToWrite, isSequential,  m_err);
                if (e_ok != m_err)
                {
                    break;
                }
                oPageCnt   = cluster.GetBlockCount();

                WriteDataPages (m_trans, nodePtrList, iPageCnt-pagesLeftToWrite, cluster);

                pagesLeftToWrite -= oPageCnt;

                totalIO++;
                totalPages += oPageCnt;
                totalIOElems -= oPageCnt;
            }
            if (e_ok == m_err){
                bd20ResetIOState (m_trans.trTaskId_gg00, current.value(), iPageCnt, cbptrList);
            }
        } while ((e_ok == m_err) && (totalIOElems > 0 ));

        current++;
    } while ((e_ok == m_err) && (current != iterEnd));

    m_pagerController.AddDataIOTotal(totalIO);
    m_pagerController.AddDataPagesTotal(totalPages);
}

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

void Pager_Writer::FlushDataCacheAfterPrepare() {

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::FlushDataCacheAfterPrepare", Pager_Trace, 5 );

    SAPDB_Bool1             svpCompleted;
    SAPDB_Bool1             isSequential;
    SAPDB_Bool              stopLoop;
    IOMan_BlockCount        pageCnt;
    SAPDB_Int2              curPageCnt;
    SAPDB_Int               loopCnt;
    SAPDB_Int               totalIO;
    SAPDB_Int               totalPages;
    tbd00_NodePtrList       nodePtrList;
    tbd2_data_cb_flush_list cbptrList;

    PagerWriterIterator current = Begin();
    const PagerWriterIterator& iterEnd = End();

    IOMan_BlockAddress  startBlock;
    IOMan_ClusterAddress cluster;

    m_err      = e_ok;
    totalPages = 0;
    totalIO    = 0;

    do {
        svpCompleted = SAPDB_FALSE;
        loopCnt      = 0;
        while ((e_ok == m_err) && !svpCompleted){
            pageCnt = m_blockCount;

            const SAPDB_Bool bReqSequential = SAPDB_FALSE;

            IOMan_ClusterAddress cluster = FBM_IManager::Instance().GetMultFreeBlocks(
                                               m_trans.trTaskId_gg00, pageCnt, bReqSequential, m_err);
            startBlock = cluster.GetBlockAddress(0);
            pageCnt   = cluster.GetBlockCount();

            stopLoop = false;
            while ((e_ok == m_err) && (pageCnt > 0) && (!stopLoop)){
                curPageCnt = pageCnt;
                bd20GetPageForIO (m_trans.trTaskId_gg00, current.value(), m_err, curPageCnt,
                                  nodePtrList, cbptrList, svpCompleted, isSequential);
                if (e_ok == m_err){
                    if (curPageCnt > 0){
                        totalIO++;
                        totalPages += curPageCnt;

                        cluster.SetCluster(startBlock, curPageCnt);
                        WriteDataPages (m_trans, nodePtrList, 0, cluster);

                        if (e_ok == m_err){
                            bd20ResetIOState (m_trans.trTaskId_gg00, current.value(), curPageCnt, cbptrList);
                            pageCnt -= curPageCnt;
                            startBlock.SetBlockNo(startBlock.GetBlockNo() + curPageCnt);
                            cluster.SetCluster(startBlock, pageCnt);
                        }
                    }
                    else {
                        stopLoop = true;
                    }
                }
            }

            FBM_IManager::Instance().SetClusterStateToFree(m_trans.trTaskId_gg00, cluster);

            if ((e_ok == m_err) && (!svpCompleted) && stopLoop) {
                if (loopCnt >= 5){
                    vsleep (m_trans.trTaskId_gg00, 1);
                    loopCnt = 0;
                }
                else {
                    loopCnt++;
                    vsleep (m_trans.trTaskId_gg00, 0);
                }
            }
        }
        current++;
    }
    while ((e_ok == m_err) && (current != iterEnd));

    m_pagerController.AddDataIOTotal(totalIO);
    m_pagerController.AddDataPagesTotal(totalPages);
}

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

void Pager_Writer::FlushDataCacheAfterPrepareSequential() {

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::FlushDataCacheAfterPrepareSequential", Pager_Trace, 5 );

    SAPDB_Bool1             svpCompleted;
    SAPDB_Bool1             isSequential;
    SAPDB_Bool              stopLoop;
    IOMan_BlockCount        oPageCnt;
    SAPDB_Int2              iPageCnt;
    SAPDB_Int               loopCnt;
    SAPDB_Int               totalIO;
    SAPDB_Int               totalPages;
    SAPDB_Int               pagesLeftToWrite;
    tbd00_NodePtrList       nodePtrList;
    tbd2_data_cb_flush_list cbptrList;

    PagerWriterIterator current = Begin();
    const PagerWriterIterator& iterEnd = End();

    IOMan_ClusterAddress cluster;

    m_err      = e_ok;
    totalPages = 0;
    totalIO    = 0;

    do {
        svpCompleted = SAPDB_FALSE;
        loopCnt      = 0;
        stopLoop = false;

        while ((e_ok == m_err) && !svpCompleted){

            iPageCnt = m_blockCount;
            bd20GetPageForIO (m_trans.trTaskId_gg00, current.value(), m_err, iPageCnt,
                              nodePtrList, cbptrList, svpCompleted, isSequential);

            if (0 == iPageCnt)
            {
                stopLoop = true;
            }

            pagesLeftToWrite = iPageCnt;

            while ((e_ok == m_err) && (pagesLeftToWrite > 0) )
            {
                cluster = FBM_IManager::Instance().GetMultFreeBlocks(
                              m_trans.trTaskId_gg00, pagesLeftToWrite, isSequential,  m_err);
                if (e_ok != m_err)
                {
                    break;
                }

                oPageCnt   = cluster.GetBlockCount();

                WriteDataPages (m_trans, nodePtrList, iPageCnt-pagesLeftToWrite, cluster);

                totalIO++;
                totalPages += oPageCnt;
                pagesLeftToWrite -= oPageCnt;
            }
            if (e_ok == m_err){
                bd20ResetIOState (m_trans.trTaskId_gg00, current.value(), iPageCnt, cbptrList);
            }

            if ((e_ok == m_err) && (!svpCompleted) && stopLoop)
            {
                if (loopCnt >= 5){
                    vsleep (m_trans.trTaskId_gg00, 1);
                    loopCnt = 0;
                }
                else {
                    loopCnt++;
                    vsleep (m_trans.trTaskId_gg00, 0);
                }
            }
        }
        current++;
    }
    while ((e_ok == m_err) && (current != iterEnd));

    m_pagerController.AddDataIOTotal(totalIO);
    m_pagerController.AddDataPagesTotal(totalPages);
}

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

void Pager_Writer::WriteDataPages(tgg00_TransContext  &Trans,
                                  tbd00_NodePtrList   &pNodes,
                                  SAPDB_Int            Offset,
                                  IOMan_ClusterAddress Cluster) {

    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::WriteDataPages", Pager_Trace, 5 );

    const IOMan_BlockCount BlockCount = Cluster.GetBlockCount();

    m_pages.Clear();

    Data_BasePage Page;
    for(SAPDB_Int Index = Offset ; Index < (BlockCount + Offset); ++Index )
    {
        Page.Assign( pNodes[ Index ], !gg04IsStaticPage( pNodes[ Index ]->nd_pt2() ),
                     pNodes[ Index ]->nd_pmode().includes( pmTemp_egg00 ));

        m_pages.Add( Page );
    }
    IOMan_IDataIO::GetInstance().WriteDataPages( Trans.trTaskId_gg00, m_pages, Cluster );

}

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

void Pager_Writer::FlushConverter()
{
    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::FlushConverter", Pager_Trace, 5 );

    const SAPDB_Int numPagerWriter = g01maxdatawriter();
    const SAPDB_Int numConverter   = g01region_cnt (rgnConvert_egg00);
	SAPDB_Int   section = m_myIndex;

	while(section < numConverter)
	{
		SAPDB_UInt  totalIO    = 0;
		SAPDB_UInt  totalPages = 0;

		Converter_IPageIO::Instance().WritePages( m_trans.trTaskId_gg00, section, totalPages, totalIO );
		section += numPagerWriter;

		m_pagerController.AddConverterIOTotal( totalIO );
		m_pagerController.AddConverterPagesTotal( totalPages );
	}
}

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

void Pager_Writer::ReadConverter( const SAPDB_Bool bReadSnapShotConverter )
{
    SAPDBTRACE_ROUTINE_DEBUG("Pager_Writer::ReadConverter", Pager_Trace, 5 );

    SAPDB_UInt  totalIO    = 0;
    SAPDB_UInt  totalPages = 0;

    if( bReadSnapShotConverter )
        Converter_IPageIO::Instance().ReadSnapShotPages( m_trans.trTaskId_gg00,
                m_myIndex, totalPages, totalIO );
    else
        Converter_IPageIO::Instance().ReadPages( m_trans.trTaskId_gg00,
                m_myIndex, totalPages, totalIO );

    m_pagerController.AddConverterIOTotal( totalIO );
    m_pagerController.AddConverterPagesTotal( totalPages );
}

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