/*!
 * \file    OMS_Context.cpp
 * \author  IvanS, MarkusSi, PeterG, ThomasA
 * \brief   OMS context.
 */
/*

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

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

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

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



*/

#include "Oms/OMS_Context.hpp"
#include "Oms/OMS_DumpInterface.hpp"
#include "Oms/OMS_Globals.hpp"
#include "Oms/OMS_ContainerDictionary.hpp"
#include "Oms/OMS_VersionDictionary.hpp"
#include "Oms/OMS_Session.hpp"
#include "Oms/OMS_ObjectContainer.hpp"
#include "hsp77.h"
#include "geo573.h"

#define REGISTER_ALLOCATOR true

#ifndef USE_SYSTEM_ALLOC_CO13

OMS_Context::OMS_ContextAllocator::OMS_ContextAllocator(tsp00_TaskId taskId, const OmsVersionId* pVersionId)
: SAPDBMem_RawAllocator((const SAPDB_UTF8*) "", *OMS_Globals::GetKernelInterface()->GetOmsAllocator(), 32 * 1024, 32 * 1024, 
    SAPDBMem_RawAllocator::FREE_RAW_EXTENDS, SAPDBMEM_ALLOCATOR_UNLIMITED, !REGISTER_ALLOCATOR)
    , m_heapTracer(OMS_AllocatorTracer::KERNEL_CRASH)
{
  SAPDB_UTF8 heapName[41];
  if (NULL != pVersionId) {
    sp77sprintf (REINTERPRET_CAST(char*, &heapName[0]), sizeof(heapName), "OMS Version %.22s", &pVersionId[0]);
  }
  else
  {
    sp77sprintf (REINTERPRET_CAST(char*, &heapName[0]), sizeof(heapName), "OMS default context T%03d", taskId);
  }
  SetIdentifier (&heapName[0]);
  SetTracer     (&m_heapTracer);
  SetBadAllocHandler(SAPDBMem_RawAllocator::THROW_STD_BAD_ALLOC);
  SetCheckLevel (OMS_Globals::GetKernelInterface()->GetHeapCheckLevel());
  OMS_Globals::GetKernelInterface()->RegisterAllocator(GetAllocatorInfo());
}

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

OMS_Context::OMS_ContextAllocator::~OMS_ContextAllocator()
{
  OMS_Globals::GetKernelInterface()->DeregisterAllocator(GetAllocatorInfo());
}

#endif

//externCpp
static 
tsp00_Uint4 co10_GetLastDropId() 
{
  return OMS_Globals::m_globalsInstance->m_classDictionary.GetLastDropId();
}

#define OMS_NO_ERROR   0
#define OMS_EXCEPTION  1
#define OMS_BAD_ALLOC  2

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

static
void co10_MarkUnloadable(OMS_Context* pContext, bool callFromDestructor) 
{
  OMS_Globals::m_globalsInstance->m_versionDictionary.MarkUnloadable(pContext, callFromDestructor);
}

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

OMS_Context::OMS_Context
(
 OMS_Session* session, const OmsVersionId* vid,
 const tgg01_OmsVersionContext* vctxt
 ) 
 : m_session(session)
 , m_oidDir(), m_containerDir()
 , m_consistentView()
 , m_pVersionContext(NULL)
 , m_isOpen(false)
 , m_isDropped(false)
 , m_isVersion(false)
#ifndef USE_SYSTEM_ALLOC_CO13
 , m_heap(session->GetTaskId(), vid) 
 , m_newObjCache(m_heap)
#endif
 , m_next(0)
 , m_nextUnloaded(0) 
 , m_lastDropId(0)
 , m_boundToTrans(false)
 , m_marked(false)
 , m_date()
 , m_time()
 , m_lastOpenDate()
 , m_lastOpenTime()
 , m_currLcSink(session->m_lcSink)
 , m_cntNewObjectsToFlush(0)
 , m_versionDesc(NULL)             
{
  session->m_lcSink->GetDateTime(&m_date, &m_time);
  ResetConsistentView();
  m_oidDir.Create(this);
  m_containerDir.Create(this);
  if (NULL != vid) {
    m_isVersion      = true;
    m_versionContext = *vctxt;
    SAPDB_MemCopyNoCheck (m_version, vid, sizeof(*vid));
    m_isOpen         = true;
  }
};

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

OMS_Context::~OMS_Context() 
{
#ifdef USE_SYSTEM_ALLOC_CO13
  if (m_isVersion)
  {
    for (OMS_ClassIdHash::Iter iter = m_containerDir.First(); iter; ++iter) {
      iter()->DeleteSelf(this);       
    }
    m_oidDir.Clear();
    if (m_versionDesc != NULL){  // PTS 1117690
      deallocate(m_versionDesc);
    }
  }
#endif
}

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

void OMS_Context::CleanContainerDir() {
  tsp00_Uint4 lastDropId = co10_GetLastDropId();
  if (lastDropId != m_lastDropId) {    
    m_containerDir.Clean();
    m_lastDropId = lastDropId;
  }
}

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

void OMS_Context::DeleteSelf() {
  tgg00_BasisError DBError;
  if (m_isVersion) {                /* FF 16.Nov.99  PTS 1104668 */
    m_versionContext.ovc_trans_version = m_consistentView;
    OMS_HResult hr = m_currLcSink->DropVersion((unsigned char*) &m_versionContext, &DBError);
    if (0 != DBError)
    {
      // just write a message into knldiag and ignore error
      DbpBase opMsg(m_currLcSink);
      opMsg.dbpOpError("drop version, error %d ignored", DBError);
    }
  }
  this->OMS_Context::~OMS_Context();
  OMS_Globals::m_sharedMemAllocatorInstance.deallocate(this);
}

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

void OMS_Context::Dump(OMS_DumpInterface& dumpObj) const
{
  struct ContextDumpInfo
  {
    void*                    m_this;
    OmsVersionId             m_version;
    tsp00_Int2               m_filler1;
    OMS_Session*         m_session;
    IliveCacheSink*          m_currLcSink;
    IliveCacheSink*          m_dummy; 
    tgg01_OmsVersionContext* m_pVersionContext;
    tsp00_Uint4              m_lastDropId;
    tgg91_TransNo            m_consistentView;
    bool                     m_isOpen;
    bool                     m_isDropped;
    bool                     m_isVersion;
    bool                     m_boundToTrans;
    bool                     m_marked;
    bool                     m_filler2;
  } contextDumpInfo;

  contextDumpInfo.m_this            = CONST_CAST(OMS_Context*, this);
  SAPDB_MemCopyNoCheck(&contextDumpInfo.m_version[0], &m_version[0], sizeof(m_version));
  contextDumpInfo.m_filler1         = 0;
  contextDumpInfo.m_session         = m_session;
  contextDumpInfo.m_currLcSink      = m_currLcSink;
  contextDumpInfo.m_pVersionContext = m_pVersionContext;
  contextDumpInfo.m_lastDropId      = m_lastDropId;
  contextDumpInfo.m_consistentView  = m_consistentView;
  contextDumpInfo.m_isOpen          = m_isOpen;
  contextDumpInfo.m_isDropped       = m_isDropped;
  contextDumpInfo.m_isVersion       = m_isVersion;
  contextDumpInfo.m_boundToTrans    = m_boundToTrans;
  contextDumpInfo.m_marked          = m_marked;
  contextDumpInfo.m_filler2         = false;
  dumpObj.SetDumpLabel(LABEL_OMS_CONTEXT);
  dumpObj.Dump(&contextDumpInfo, sizeof(contextDumpInfo));
  m_oidDir.Dump(dumpObj);
  m_containerDir.Dump(dumpObj);
}

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

void OMS_Context::DumpObjCache(OmsHandle& h) {
  char                  flags[4];
  flags[sizeof(flags)-1] = 0;
  OmsObjectContainerPtr curr;
  for (OMS_OidHash::OidIter iter = m_oidDir.First(); iter; ++iter) {
    curr = iter();
    if (curr->DeletedFlag()) {
      flags[0] = 'D';
    }
    else {
      flags[0] = ' ';
    }
    if (curr->StoredFlag()) {
      flags[1] = 'U';
    }
    else {
      flags[1] = ' ';
    }
    if (curr->LockedFlag()) {
      flags[2] = 'L';
    }
    else {
      flags[2] = ' ';
    }
    curr->m_pobj.omsDump(h, flags);
  }
}

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

void OMS_Context::EmptyObjCache(tsp00_Uint4 containerHandle) {
  OmsObjectContainerPtr curr;
  OMS_ClassIdEntry* pContainerInfo ;
  OMS_OidHash::OidIter iter = m_oidDir.First();

  while (iter) {
    curr = iter();
    ++iter;
    if (0xFFFFFFFF == containerHandle || curr->GetContainerHandle() == containerHandle) 
    {
      pContainerInfo = curr->GetContainerInfo(this);
      if (curr->IsNewObject())
      {
        OMS_DETAIL_TRACE(omsTrNewObj, m_session->m_lcSink, 
          "OMS_Context::EmptyObjCache : new obj " << curr->m_oid 
          << ", class: " << pContainerInfo->GetClassInfoPtr()->GetClassName());
        this->DecNewObjectsToFlush();
      }
      DeleteObjInContext(curr, pContainerInfo);
    }
  }
}

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

void OMS_Context::FlushObj(OmsObjectContainerPtr p) {
  if (p->VarObjFlag()) {
    FlushVarObject (p);
  }
  else {
    tsp00_Int2  DBError;
    OMS_ClassIdEntry* clsinfo = p->GetContainerInfo(this);
    m_session->SetDataChanged();
    OMS_HResult hr = m_currLcSink->UpdateObj (
      (unsigned char*) &m_consistentView,
      (unsigned char*) &clsinfo->GetFileId(),
      VersionContext(),
      (unsigned char*) &p->m_oid,
      (unsigned char*) &p->m_objseq, 
      clsinfo->GetPersistentSize(), 
      ((unsigned char*) &p->m_pobj) + sizeof(void*), 
      clsinfo->GetKeyDesc().GetPos(),                 // PTS 1122540
      clsinfo->GetKeyDesc().GetLen(),                 // PTS 1122540
      false,                                          // PTS 1124278
      &DBError);
    this->CheckDBError(DBError, "OMS_Context::FlushObj", p->m_oid); 
    p->RemoveStoredFlag();
    m_session->IncStore_lC();
  }
}

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

void OMS_Context::FlushDelete (OmsObjectContainerPtr pObj, bool deleteInHash) {

  tsp00_Int2     DBError;
  OMS_HResult hr;
  OMS_ClassIdEntry* clsinfo = pObj->GetContainerInfo(this);
  m_session->SetDataChanged();
  if (pObj->VarObjFlag()) {
    m_session->IncDeleteVarObjectLC ();
    bool                    isFirst = true;
    OMS_VarObjInfoNode* pInfoNode; 
    OMS_VarObjInfo*     objInfo = REINTERPRET_CAST (OMS_VarObjInfo*, &pObj->m_pobj);
    pInfoNode = &objInfo->m_vobjInfo;
    while (NULL != pInfoNode) {
      if ((!isFirst) && (!IsVersion())) {
        // ADIS 1001523, T.A. 24.09.1998
        hr = m_currLcSink->LockObj (
          (unsigned char*) &m_consistentView,
          (unsigned char*) &clsinfo->GetFileId(),
          REINTERPRET_CAST(OmsTypeOid*, &pInfoNode->m_vobjOid),
          (unsigned char*) &pInfoNode->m_vobjSeq,
          &DBError);
        this->CheckDBError(DBError, "OMS_Context::FlushDelete (VarObj)", pInfoNode->m_vobjOid);
      }
      hr = m_currLcSink->DeleteObj (
        (unsigned char*) &m_consistentView,
        (unsigned char*) &clsinfo->GetFileId(),
        VersionContext(),
        (unsigned char*) &pInfoNode->m_vobjOid,
        (unsigned char*) &pInfoNode->m_vobjSeq, 
        &DBError);
      this->CheckDBError(DBError, "OMS_Context::FlushDelete (VarObj)", pInfoNode->m_vobjOid);
      pInfoNode = pInfoNode->m_vobjNext;
      isFirst   = false;
    }
  } 
  else { // no Var Object
    m_session->IncDelete_lC();
    hr = m_currLcSink->DeleteObj (
      (unsigned char*) &m_consistentView,
      (unsigned char*) &clsinfo->GetFileId(),
      VersionContext(),
      (unsigned char*) &pObj->m_oid, 
      (unsigned char*) &pObj->m_objseq,
      &DBError);
    this->CheckDBError(DBError, "OMS_Context::FlushDelete", pObj->m_oid);
  }
  if (deleteInHash) {
    DeleteObjInContext (pObj, clsinfo);
  }
}


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

void OMS_Context::FlushLockUpdObj(OmsObjectContainerPtr p) {
  tsp00_Int2  DBError;
  OMS_ClassIdEntry* clsinfo = p->GetContainerInfo(this);
  m_session->SetDataChanged();
  OMS_HResult hr = m_currLcSink->LockUpdObj (
    (unsigned char*) &m_consistentView,
    (unsigned char*) &clsinfo->GetFileId(),
    REINTERPRET_CAST(OmsTypeOid*, &p->m_oid),
    (unsigned char*) &p->m_objseq, 
    &DBError);
  this->CheckDBError(DBError, "OMS_Context::FlushLockUpdObj", p->m_oid); 
}

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

void OMS_Context::FlushVarObject (OmsObjectContainerPtr pObj) {

  OMS_HResult hr;
  tsp00_Int2     DBError;
  tsp00_Uint4    moveLen;
  int            vObjLen;
  tgg91_PageRef  newObjSeq;
  bool           isNew    = true;
  bool           isLocked = false;
  bool           isFirst  = true;
  tsp00_Uint4    objLength;
  OMS_VarObjContainer varObjContainer;
  OMS_VarObjInfoNode*  pInfoNode; 
  OMS_VarObjInfoNode*  pLastNode;

  m_session->SetDataChanged();
  m_session->IncStoreVarObjLC();
  OMS_ClassIdEntry* clsinfo = pObj->GetContainerInfo(this);
  OMS_VarObjInfo* objInfo  = REINTERPRET_CAST (OMS_VarObjInfo*, &pObj->m_pobj);
  unsigned char* p             = objInfo->m_pvobj;
  pInfoNode                    = &objInfo->m_vobjInfo;
  objLength                    = objInfo->m_vobjSize;
  varObjContainer.m_vobjSize   = objInfo->m_vobjSize;
  do {
    isLocked = isNew;
    isNew    = false;
    moveLen = OMS_Globals::Min(VAR_OBJ_CHUNK_SIZE, objLength);
    if (moveLen >= objLength) {
      varObjContainer.m_vobjNext.setNil();
    }
    else {
      if (pInfoNode->m_vobjNext) {
        SAPDB_MemCopyNoCheck (&varObjContainer.m_vobjNext, &pInfoNode->m_vobjNext->m_vobjOid, 
          sizeof (pInfoNode->m_vobjNext->m_vobjOid));
      }
      else {
        if (IsVersion()) /* PTS 1110149 */
        {
          OmsObjectId newOid = VersionNewOid();
          SAPDB_MemCopyNoCheck(&varObjContainer.m_vobjNext, &newOid, sizeof(varObjContainer.m_vobjNext));
          newObjSeq.gg91SetNilRef();
        }
        else
        {
          hr = m_currLcSink->NewObj (
            (unsigned char*) &m_consistentView,
            (unsigned char*) &clsinfo->GetFileId(), 
            VersionContext(),
            0,
            0,
            NULL,
            REINTERPRET_CAST(OmsTypeOid*, &varObjContainer.m_vobjNext),
            (unsigned char*) &newObjSeq,
            &DBError);
          this->CheckDBError(DBError, "OMS_Context::FlushVarObject", pObj->m_oid); 
        }
        isNew                 = true;
        pInfoNode->m_vobjNext = new(*this) OMS_VarObjInfoNode (varObjContainer.m_vobjNext, newObjSeq);
      }
    }
    SAPDB_MemCopyNoCheck (&varObjContainer.vobjInfo[0], p, moveLen);
    if ((!isLocked) && (!IsVersion())) {
      hr = m_currLcSink->LockObj (
        (unsigned char*) &m_consistentView,
        (unsigned char*) &clsinfo->GetFileId(),
        REINTERPRET_CAST(OmsTypeOid*, &pInfoNode->m_vobjOid),
        (unsigned char*) &pInfoNode->m_vobjSeq, &DBError);
      this->CheckDBError(DBError, "OMS_Context::FlushVarObject", pInfoNode->m_vobjOid); 
    }
    objLength -= moveLen;
    vObjLen    = (objLength > 0) ? (int) sizeof(varObjContainer) : moveLen;
    bool contObj = (pObj->m_oid != pInfoNode->m_vobjOid);  // PTS 1124278
    hr = m_currLcSink->UpdateObj (
      (unsigned char*) &m_consistentView,
      (unsigned char*) &clsinfo->GetFileId(),
      VersionContext(),
      (unsigned char*) &pInfoNode->m_vobjOid,
      (unsigned char*) &pInfoNode->m_vobjSeq,
      vObjLen, 
      (unsigned char*) &varObjContainer, 
      clsinfo->GetKeyDesc().GetPos(),                 // PTS 1122540
      clsinfo->GetKeyDesc().GetLen(),                 // PTS 1122540
      contObj,                                        // PTS 1124278
      &DBError);
    this->CheckDBError(DBError, "OMS_Context::FlushVarObject", pInfoNode->m_vobjOid); 
    if (isFirst) {
      pObj->m_objseq = pInfoNode->m_vobjSeq;
      isFirst        = false;
    }
    p         += moveLen;
    pLastNode  = pInfoNode;
    pInfoNode  = pInfoNode->m_vobjNext;
  }
  while (objLength > 0);
  pLastNode->m_vobjNext = NULL;
  if (!IsVersion()) {
    objInfo->freeVarObj(this);
  }
  while (pInfoNode) {
    if (!IsVersion()) {
      hr = m_currLcSink->LockObj (
        (unsigned char*) &m_consistentView,
        (unsigned char*) &clsinfo->GetFileId(),
        REINTERPRET_CAST(OmsTypeOid*, &pInfoNode->m_vobjOid),
        (unsigned char*) &pInfoNode->m_vobjSeq, &DBError);
    }
    if (0 == DBError) {
      hr = m_currLcSink->DeleteObj (
        (unsigned char*) &m_consistentView,
        (unsigned char*) &clsinfo->GetFileId(),
        VersionContext(),
        (unsigned char*) &pInfoNode->m_vobjOid,
        (unsigned char*) &pInfoNode->m_vobjSeq, 
        &DBError);
    }
    this->CheckDBError(DBError, "OMS_Context::FlushVarObject", pInfoNode->m_vobjOid); 
    pLastNode = pInfoNode;
    pInfoNode = pInfoNode->m_vobjNext;
    pLastNode->deleteSelf(*this);
  }
}

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

// PTS 1121449
void OMS_Context::FlushObjCache(bool emptyObjCache, 
                                bool adaptOidHash) 
{
  const char *msg = "OMS_Context::FlushObjCache ";

  bool               checkHash     = false;
  bool               errorOccured  = false;
  int                newObjFlushed = 0;
  int                errCnt        = 0;
  int                errorKind     = OMS_NO_ERROR;
  DbpError           dbpError(0);
  OmsObjectContainer *pFlushVarHead             = NULL;
  OmsObjectContainer *pFlushVarHeadOtherClasses = NULL;
  OmsObjectContainer *pDelVarHead               = NULL;
  OmsObjectContainer *pFreeHead                 = NULL;
  OMS_ClassIdEntry   *pClsInfoVarObj            = NULL;

  int delCnt = 0;
  // Prevent the calling of the constructor for every entry. An assignment is done in the loop.
  //OmsObjectId           pDelOid          [OMS_MASS_OPERATION_CNT];
  char                    pDelOid_u        [OMS_MASS_OPERATION_CNT * sizeof(OmsObjectId)];
  OmsObjectId            *pDelOid = reinterpret_cast<OmsObjectId*>(&pDelOid_u[0]);
  tgg91_PageRef           pDelObjVers       [OMS_MASS_OPERATION_CNT];
  tgg00_FileId           *ppDelContainerId  [OMS_MASS_OPERATION_CNT];

  int storeCnt = 0;
  // Prevent the calling of the constructor for every entry. An assignment is done in the loop.
  //OmsObjectId           pStoreOid         [OMS_MASS_OPERATION_CNT];
  char                    pStoreOid_u       [OMS_MASS_OPERATION_CNT * sizeof(OmsObjectId)];
  OmsObjectId            *pStoreOid = reinterpret_cast<OmsObjectId*>(&pStoreOid_u[0]);
  tgg91_PageRef           pStoreObjVers     [OMS_MASS_OPERATION_CNT];
  tsp00_Int4              pStoreBodySize    [OMS_MASS_OPERATION_CNT];
  unsigned char          *ppStoreObj        [OMS_MASS_OPERATION_CNT];  
  tsp00_Int4              pKeyPos           [OMS_MASS_OPERATION_CNT];
  tsp00_Int4              pKeyLen           [OMS_MASS_OPERATION_CNT];
  tgg00_FileId           *ppStoreContainerId[OMS_MASS_OPERATION_CNT];

  int lockUpdCnt = 0;
  // Prevent the calling of the constructor for every entry. An assignment is done in the loop.
  //OmsObjectId           pLockUpdOid       [OMS_MASS_OPERATION_CNT];
  char                    pLockUpdOid_u     [OMS_MASS_OPERATION_CNT * sizeof(OmsObjectId)];
  OmsObjectId            *pLockUpdOid = reinterpret_cast<OmsObjectId*>(&pLockUpdOid_u[0]);  
  tgg91_PageRef           pLockUpdObjVers   [OMS_MASS_OPERATION_CNT];
  tgg00_FileId           *ppLockUpdContainerId[OMS_MASS_OPERATION_CNT];

  tgg00_BasisError        pDBError          [OMS_MASS_OPERATION_CNT];

  OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, "Start of " << msg);
  OMS_OidHash::OidIter iter = m_oidDir.First();
  try {
    while (iter) {
      while (delCnt < OMS_MASS_OPERATION_CNT  && storeCnt < OMS_MASS_OPERATION_CNT && lockUpdCnt < OMS_MASS_OPERATION_CNT && iter){
        // Check whether oid hash structure still fulfills some invariants. 
        // Check is normally switched off
        if (checkHash) {
          m_oidDir.HashCheck();
        }

        OmsObjectContainer *curr     = iter();
        OMS_ClassIdEntry   *pClsInfo = curr->GetContainerInfoNoCheck(this);
        ++iter;

        curr->m_beforeImages = 0;  // ??
        
        bool appendToFreeList = true;
        if (curr->ReplacedFlag()){    // PTS 1125361
          // Object is replaced by a newer version and therefore nothing should be
          // done with this object besides deleting the object from the cache.
        }
        else if (curr->DeletedFlag()){
          // Object is marked as deleted 
          if (curr->IsNewObject()) {
            ++newObjFlushed;
          }
          if (curr->VarObjFlag()){
            appendToFreeList = false;

            // Collect all variable sized objects in a list for later processing
            curr->SetNext(pDelVarHead);
            pDelVarHead = curr;
            OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Object " << curr->m_oid << " append to var delete list");
          }
          else {
            // Append entry to delete-array
            pDelOid         [delCnt] = curr->m_oid;
            pDelObjVers     [delCnt] = curr->m_objseq;
            ppDelContainerId[delCnt] = &pClsInfo->GetFileId();
            ++delCnt; 
            OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Object " << curr->m_oid << " marked for mass delete");
          }
        }
        else if (curr->StoredFlag()){
          // Object is marked as to be stored
          if (curr->IsNewObject()) {
            ++newObjFlushed;
          }
          if (curr->VarObjFlag()){
            appendToFreeList = false;

            // Collect all variable sized objects in a list for later processing
            if (pClsInfoVarObj == NULL){
              pClsInfoVarObj = curr->GetContainerInfoNoCheck(this);
            }

            if (curr->GetContainerInfoNoCheck(this) == pClsInfoVarObj){
              // Object belongs to the 'container' which is currently collected in the 
              // the list pFlushVarHead.
              curr->SetNext(pFlushVarHead);
              pFlushVarHead = curr;
              OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Object " << curr->m_oid << " append to var update list1");
            }
            else {
              // Object belongs to another container and therefore this object is inserted
              // into the 'mixed' list 
              curr->SetNext(pFlushVarHeadOtherClasses);
              pFlushVarHeadOtherClasses = curr;
              OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Object " << curr->m_oid << " append to var update list2");
            }
          }
          else {
            // Append entry to store-array
            pStoreOid         [storeCnt] = curr->m_oid;
            pStoreObjVers     [storeCnt] = curr->m_objseq;
            ppStoreContainerId[storeCnt] = &pClsInfo->GetFileId();
            pStoreBodySize    [storeCnt] = pClsInfo->GetPersistentSize();
            ppStoreObj        [storeCnt] = ((unsigned char*)&curr->m_pobj) + sizeof(void*);  
            pKeyPos           [storeCnt] = pClsInfo->GetKeyDesc().GetPos();     // PTS 1122540
            pKeyLen           [storeCnt] = pClsInfo->GetKeyDesc().GetLen();     // PTS 1122540
            ++storeCnt;
            OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Object " << curr->m_oid << " marked for mass update");

            curr->RemoveStoredFlag();  // ??
          }
        }
        else if (curr->LockedFlag() && !IsVersion()){
          // Append entry to lockUpd-array
          pLockUpdOid         [lockUpdCnt] = curr->m_oid;
          pLockUpdObjVers     [lockUpdCnt] = curr->m_objseq;
          ppLockUpdContainerId[lockUpdCnt] = &pClsInfo->GetFileId();
          ++lockUpdCnt;
          OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Object " << curr->m_oid << " marked for mass lock-update");
       }

        // Append object to a list for later deletion of memory. 
        // For fixed sized objects this is only necessary if the context is not 
        // deleted as a whole. Variable sized objects are only added if they are
        // not later handled by the methods FlushVarObjMass or
        // DeleteVarObjMass. Those objects will be added in the corresponding 
        // method to the list.
        if ((emptyObjCache || curr->VarObjFlag()) && appendToFreeList) {
          curr->SetNext(pFreeHead);
          pFreeHead = curr;
          OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Object " << curr->m_oid << " appended to free list");
        }

        if (delCnt == OMS_MASS_OPERATION_CNT || (delCnt > 0 && !iter)){
          // Mass-delete from the kernel
          m_session->IncDelete_lC();
          OMS_HResult hr = m_currLcSink->DeleteObjMass(
            delCnt, &m_consistentView, ppDelContainerId, VersionContext(),
            pDelOid, pDelObjVers, errCnt, pDBError);
          OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Execute mass delete with " << delCnt << " entries");
          if (errCnt > 0){
            for (int i=0; i<delCnt; ++i){
              if (pDBError[i] != e_ok){
                OMS_Globals::Throw(DbpError(DbpError::DB_ERROR, pDBError[i], msg, pDelOid[i], __MY_FILE__, __LINE__));
              }
            }
          }
          delCnt = 0;
        }

        if (storeCnt == OMS_MASS_OPERATION_CNT || (storeCnt > 0 && !iter)){
          // Mass-store into the kernel
          m_session->IncStore_lC();
          OMS_HResult hr = m_currLcSink->UpdateObjMass(
            storeCnt, &m_consistentView, ppStoreContainerId, VersionContext(),
            pStoreOid, pStoreObjVers, pStoreBodySize, ppStoreObj, pKeyPos, pKeyLen, 
            errCnt, pDBError);
          OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Execute mass update with " << storeCnt << " entries");
          if (errCnt > 0){
            for (int i=0; i<storeCnt; ++i){
              if (pDBError[i] != e_ok){
                OMS_Globals::Throw(DbpError(DbpError::DB_ERROR, pDBError[i], msg, pStoreOid[i], __MY_FILE__, __LINE__));
              }
            }
          }
          storeCnt = 0;
        }

        if (lockUpdCnt == OMS_MASS_OPERATION_CNT || (lockUpdCnt > 0 && !iter)){
          // Mass-lock-update in the kernel
          OMS_HResult hr = m_currLcSink->LockUpdObjMass(
            lockUpdCnt, &m_consistentView, ppLockUpdContainerId, pLockUpdOid, 
            pLockUpdObjVers, errCnt, pDBError);
          OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Execute mass lock-update with " << lockUpdCnt << " entries");
          if (errCnt > 0){
            for (int i=0; i<lockUpdCnt; ++i){
              if (pDBError[i] != e_ok){
                OMS_Globals::Throw(DbpError(DbpError::DB_ERROR, pDBError[i], msg, pLockUpdOid[i], __MY_FILE__, __LINE__));
              }
            }
          }
          lockUpdCnt = 0;
        }
      }
    }

    // Flush variable sized objects
    if (pFlushVarHead)
      FlushVarObjMass(pFlushVarHead, pFlushVarHeadOtherClasses, pFreeHead);

    // Delete variable sized objects
    if (pDelVarHead)
      DeleteVarObjMass(pDelVarHead, pFreeHead);

  }
  catch (DbpError &e){
    // No call of dbpCaughtException() needed, as exception is thrown again later
    errorKind = OMS_EXCEPTION;
    dbpError  = e;
    OMS_TRACE(omsTrFlushCache, m_session->m_lcSink, " Error caught: " << e.m_errorNo << e.m_errorText);
  }
  catch (BAD_ALLOC_GEO573&){
    // No call of dbpCaughtBadAlloc() needed, as exception is thrown again later
    errorKind = OMS_BAD_ALLOC;
    OMS_TRACE(omsTrFlushCache, m_session->m_lcSink, " Bad-Alloc caught");
  }
   
  if (errorKind != OMS_NO_ERROR){
    // Append objects, which haven't been considered so far, to the list of
    // objects for which memory should be freed
    while (iter){
      OmsObjectContainer *curr = iter();
      ++iter;
      if (emptyObjCache || curr->VarObjFlag()) {
        curr->SetNext(pFreeHead);
        pFreeHead = curr;
        OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Object " << curr->m_oid << " appended to free list");
      }
    }

    // In case of an exception these lists might not be empty
    while (pFlushVarHead){
      OmsObjectContainer *next = pFlushVarHead->GetNext();
      pFlushVarHead->SetNext(pFreeHead);
      pFreeHead = pFlushVarHead;
      pFlushVarHead = next;
    }
    while (pFlushVarHeadOtherClasses){
      OmsObjectContainer *next = pFlushVarHeadOtherClasses->GetNext();
      pFlushVarHeadOtherClasses->SetNext(pFreeHead);
      pFreeHead = pFlushVarHeadOtherClasses;
      pFlushVarHeadOtherClasses = next;
    }
    while (pDelVarHead){
      OmsObjectContainer *next = pDelVarHead->GetNext();
      pDelVarHead->SetNext(pFreeHead);
      pFreeHead = pDelVarHead;
      pDelVarHead = next;
    }
  }

  // Release all memory of objects  
  while (pFreeHead){
    OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, " Free memory of object: " << pFreeHead->m_oid);
    OmsObjectContainer *pNext = pFreeHead->GetNext();
    if (pFreeHead->VarObjFlag()){
      reinterpret_cast<OMS_VarObjInfo*>(&pFreeHead->m_pobj)->freeVarObj(this);
    }
    if (emptyObjCache) {
      OMS_ClassIdEntry *pClsInfo = pFreeHead->GetContainerInfoNoCheck(this);
      pClsInfo->chainFree(*this, pFreeHead, 4);
    }
    pFreeHead = pNext;
  }

  // Clear structures
#ifdef USE_SYSTEM_ALLOC_CO13
  ClearObjCache();
#endif
  m_oidDir.SetEmpty(adaptOidHash);
  m_newObjCache.SetEmpty();

  if (errorKind == OMS_EXCEPTION) {
    m_cntNewObjectsToFlush = 0;
    throw dbpError;
  }
  else if (errorKind == OMS_BAD_ALLOC){
    m_cntNewObjectsToFlush = 0;
    throw BAD_ALLOC_GEO573();
  }
  else if ((!IsVersion()) && (newObjFlushed != m_cntNewObjectsToFlush)){
    // Check whether number of new flushed objects match the expected number,
    // which is stored as variable in the context 
    OMS_TRACE(omsTrFlushCache | omsTrNewObj, m_session->m_lcSink, 
      "OMS_Context::FlushObjCache : flushed new objects: "
      << newObjFlushed << ", expected to flush: " << m_cntNewObjectsToFlush);
    DbpBase b(m_currLcSink);
    b.dbpOpError ("Flushed new objects           : %d", newObjFlushed);
    b.dbpOpError ("expected new objects to flush : %d", m_cntNewObjectsToFlush);

    m_cntNewObjectsToFlush = 0;
    OMS_Globals::Throw(DbpError(DbpError::DB_ERROR, e_missing_object_flush, __MY_FILE__, __LINE__));
  }
}

/*----------------------------------------------------------------------*/
// PTS 1121449
void OMS_Context::DeleteVarObjMass(OmsObjectContainer  *&pHead,
                                   OmsObjectContainer  *&pFreeList)
{
  const char *msg = "OMS_Context::DeleteVarObjMass ";

  int delCnt = 0;
  // Prevent the calling of the constructor for every entry. An assignment is done in the loop.
  //OmsObjectId           pDelOid           [OMS_MASS_OPERATION_CNT];
  char                    pDelOid_u         [OMS_MASS_OPERATION_CNT * sizeof(OmsObjectId)];
  OmsObjectId            *pDelOid = reinterpret_cast<OmsObjectId*>(&pDelOid_u[0]);
  tgg91_PageRef           pDelObjVers       [OMS_MASS_OPERATION_CNT];
  tgg00_FileId           *ppDelContainerId  [OMS_MASS_OPERATION_CNT];

  int lockCnt = 0;
  // Prevent the calling of the constructor for every entry. An assignment is done in the loop.
  //OmsObjectId           pLockOid          [OMS_MASS_OPERATION_CNT];
  char                    pLockOid_u         [OMS_MASS_OPERATION_CNT * sizeof(OmsObjectId)];
  OmsObjectId            *pLockOid = reinterpret_cast<OmsObjectId*>(&pLockOid_u[0]);
  tgg91_PageRef           pLockObjVers      [OMS_MASS_OPERATION_CNT];
  tgg00_FileId           *ppLockContainerId [OMS_MASS_OPERATION_CNT];
  
  tgg00_BasisError        pDBError          [OMS_MASS_OPERATION_CNT];

  int                 errCnt    = 0;
  OMS_VarObjInfo     *objInfo   = NULL;
  OMS_VarObjInfoNode *pInfoNode = NULL;
  OMS_ClassIdEntry   *pClsInfo  = NULL;

  OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "Start DELETE of VAR-Objects");

  while (pHead) {
    while (pHead && delCnt < OMS_MASS_OPERATION_CNT){

      if (pInfoNode == NULL){
        // Consider first/next object in list
        objInfo   = REINTERPRET_CAST (OMS_VarObjInfo*, &pHead->m_pobj);
        pInfoNode = &objInfo->m_vobjInfo; 
        pClsInfo  = pHead->GetContainerInfoNoCheck(this);
      }

      while (NULL != pInfoNode && delCnt < OMS_MASS_OPERATION_CNT) {
        // Get a lock for this object from the kernel if needed
        if (!IsVersion()) {
          // Remember entries for locking
          pLockOid         [lockCnt] = pInfoNode->m_vobjOid;
          pLockObjVers     [lockCnt] = pInfoNode->m_vobjSeq;
          ppLockContainerId[lockCnt] = &pClsInfo->GetFileId();
          ++lockCnt;
          OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "  Infonode: " << pInfoNode->m_vobjOid << "  locked");
       }

        // Append entry to delete-array
        pDelOid         [delCnt] = pInfoNode->m_vobjOid;
        pDelObjVers     [delCnt] = pInfoNode->m_vobjSeq;
        ppDelContainerId[delCnt] = &pClsInfo->GetFileId();
        ++delCnt; 
        OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "  Infonode: " << pInfoNode->m_vobjOid << "  deleted");
        
        pInfoNode = pInfoNode->m_vobjNext;
      }

      if (pInfoNode == NULL){
        // Previous object is completely flushed, therefore consider next object
        OmsObjectContainer *pNext = pHead->GetNext();
        
        // Append object to list for later freeing of memory
        pHead->SetNext(pFreeList);
        pFreeList = pHead;

        pHead = pNext;
      }
    }

    if (delCnt == OMS_MASS_OPERATION_CNT || (delCnt > 0 && pHead == NULL)){
      m_session->IncDeleteVarObjectLC ();

      // Mass lock
      tsp00_Int8 timeout = 0;
      OMS_HResult hr = m_currLcSink->LockObjMass(
        lockCnt, &m_consistentView, ppLockContainerId, pLockOid, 
        pLockObjVers, timeout, errCnt, pDBError);
      LockResult(lockCnt, pDBError, NULL, pLockOid, msg);
      lockCnt = 0;

      // Mass deletion
      hr = m_currLcSink->DeleteObjMass(
        delCnt, &m_consistentView, ppDelContainerId, VersionContext(),
        pDelOid, pDelObjVers, errCnt, pDBError);
      if (errCnt > 0){
        for (int i=0; i<delCnt; ++i){
          if (pDBError[i] != e_ok){
            OMS_Globals::Throw(DbpError(DbpError::DB_ERROR, pDBError[i], msg, pDelOid[i], __MY_FILE__, __LINE__));
          }
        }
      }
      delCnt = 0;
    }
  }

  OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "End DELETE of VAR-Objects");
}

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

/// Flushes a list of var objects 
/*!
** This can flushed a list of objects with variable size to the kernel.
** The method can be split roughly into 4 phases:
** In phase ONE for each object it is checked how many pages are needed to store the 
** object and how many pages are already assigned to the object. If the object owns more
** pages than it needs, than these pages are appended to a freelist. Objects which needs
** more pages tries to get them from the freelist and if there are not enough, then these
** objects are collected in a separate list for later processing.
** In phase TWO new pages are allocated per mass operation from the kernel for all those
** objects which still need pages. After the pages have been allocated, these pages are
** assigned to the objects which needs additional pages.
** If there had been more pages assigned to the object than are needed to store them,
** then these superfluous pages are freed per mass-operation in phase THREE.
** This means that there will never be phase TWO and phase THREE active in the same call;
** furthermore if the sizes don't change, then none of them will be needed.
** Finally in phase FOUR the objects are updated per mass-operations in the kernel.
** To support the reuse of pages (in phase ONE and TWO) it is necessary, that all objects
** considered in this phase belong to the same schema, class, and container. Therefore
** the objects to be flushed, are divided into two list: one which contains only objects
** belonging to the same schema, class and container and the other list, which contains
** all the rest. The algorithm (all 4 phases) is executed solely for the first list and
** and the end it is checked whether there are other objects left and if so, this list
** is split again into one with only one class and the rest.
** This means, that if there are a very large number of different schema, class, container
** combinations, then the algorithm will have a quadratic running time and should be
** changed.
**
** \param pHeadOneClass Pointer to the head of a list of the objects which should be flushed.
**    All objects in this list must belong to the same schema and container number.
** \param pHeadOtherClasses If there are objects which belong to other classes than the
**    objects of the list pHeadOneClass, then these objects are listed in this list.
** \param pFreeList All processed objects are linked to this list. So this list can be 
**    used to free the memory which is allocated by the corresponding objects.
**
** \assumption All objects in the list, headed by pHead, belong to the same
** schema, class, and container
**
** \attention After normal execution the lists pHeadOneClass and pHeadOtherClasses should
** be empty. But in case of an exception, this might not be true, and therefore they are 
** passed as reference.
**
** \since PTS 1121449
*/
void OMS_Context::FlushVarObjMass (OmsObjectContainer  *&pHeadOneClass, 
                                   OmsObjectContainer  *&pHeadOtherClasses,
                                   OmsObjectContainer  *&pFreeList) 
{
  const char *msg = "OMS_Context::FlushVarObjMass ";

  int cnt = 0;
  // Prevent the calling of the constructor for every entry. An assignment is done in the loop.
  //OmsObjectId           pOid           [OMS_MASS_OPERATION_CNT];
  char                    pOid_u         [OMS_MASS_OPERATION_CNT * sizeof(OmsObjectId)];
  OmsObjectId            *pOid = reinterpret_cast<OmsObjectId*>(&pOid_u[0]);
  tgg91_PageRef           pObjVers       [OMS_MASS_OPERATION_CNT];
  tgg00_FileId           *ppContainerId  [OMS_MASS_OPERATION_CNT];
  tsp00_Int4              pObjSize       [OMS_MASS_OPERATION_CNT];
  tsp00_Int4              pBodySize      [OMS_MASS_OPERATION_CNT];
  unsigned char          *ppObj          [OMS_MASS_OPERATION_CNT];
  // Prevent the calling of the constructor for every entry. An assignment is done in the loop.
  //OmsObjectId           pNextOid       [OMS_MASS_OPERATION_CNT];
  char                    pNextOid_u     [OMS_MASS_OPERATION_CNT * sizeof(OmsObjectId)];
  OmsObjectId            *pNextOid = reinterpret_cast<OmsObjectId*>(&pNextOid_u[0]);
  short                   pDBError       [OMS_MASS_OPERATION_CNT];
  bool                    pContObj       [OMS_MASS_OPERATION_CNT];

  int lockCnt = 0;
  // Prevent the calling of the constructor for every entry. An assignment is done in the loop.
  //OmsObjectId           pLockOid           [OMS_MASS_OPERATION_CNT];
  char                    pLockOid_u         [OMS_MASS_OPERATION_CNT * sizeof(OmsObjectId)];
  OmsObjectId            *pLockOid = reinterpret_cast<OmsObjectId*>(&pLockOid_u[0]);
  tgg91_PageRef           pLockObjVers       [OMS_MASS_OPERATION_CNT];
  tgg00_FileId           *ppLockContainerId  [OMS_MASS_OPERATION_CNT];

  OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "Start FLUSH of VAR-Objects");
  while (pHeadOneClass){
    int                 errCnt              = 0;
    int                 availInfoNodeCnt    = 0;          
    int                 neededInfoNodeCnt   = 0;
    int                 remainingSize       = 0;
    unsigned char      *pCurrAddr           = NULL;
    OmsObjectContainer *pNext               = NULL;
    OmsObjectContainer *pPrev               = NULL;
    OmsObjectContainer *pObjContainer       = NULL;
    OmsObjectContainer *pUnfinishedObjs     = NULL;
    OMS_VarObjInfo     *pObjInfo            = NULL;
    OMS_VarObjInfoNode *pFirstInfoNode      = NULL;
    OMS_VarObjInfoNode *pAvailInfoNodes     = NULL;

    // All Objects share the same ContainerInfo. See assumption!
    OMS_ClassIdEntry   *pClsInfo = pHeadOneClass->GetContainerInfoNoCheck(this);
    OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "Schema: " << pClsInfo->GetSchema()
      << " Class: " << pClsInfo->GetClassInfoPtr()->GetClassName()
      << " Container: " << pClsInfo->GetContainerNo());

    // ********* PHASE I **********
    OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "Start of PHASE I");

    pObjContainer = pHeadOneClass;
    while (pObjContainer){
      pNext = pObjContainer->GetNext();

      pObjInfo       = REINTERPRET_CAST (OMS_VarObjInfo*, &pObjContainer->m_pobj);
      pFirstInfoNode = &pObjInfo->m_vobjInfo;

      // Compute number of pages needed to store the object.
      // pagesNeeded = floor (pObjInfo->m_vobjSize / VAR_OBJ_CHUNK_SIZE);
      int pagesNeeded = (pObjInfo->m_vobjSize + VAR_OBJ_CHUNK_SIZE - 1) / VAR_OBJ_CHUNK_SIZE;
      // Determine number of pages which are currently assigned to the object
      int pagesCurr = 0;
      if (pFirstInfoNode->m_nodeCnt > 0){
        // Normally the number of infonodes is stored in this variable
        pagesCurr = pFirstInfoNode->m_nodeCnt;
      }
      else {
        // In the very unlikely case, that the size of a single object is larger then 0.5 GB
        // then the variable is too small to store the number and therefore the number is 
        // determined by running over the chain
        OMS_VarObjInfoNode *p = pFirstInfoNode;
        pagesCurr = 1;
        while (p->m_vobjNext){
          p = p->m_vobjNext;
          ++pagesCurr;
        }
      }
      OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink,
          "  Object: " << pObjContainer->m_oid
          << " pages existing: " << pagesCurr  << " pages needed: " << pagesNeeded); 

      if (pagesNeeded < pagesCurr) {
        // The objects owns more infonodes as it needs to store the current data.

        // Position the pointer on the last infonode which is needed
        OMS_VarObjInfoNode *pLastNeededInfoNode = pFirstInfoNode;
        for(int i=0; i<pagesNeeded-1; ++i) {
          pLastNeededInfoNode  = pLastNeededInfoNode->m_vobjNext;
        }

        // Append superfluous infonodes to a separate list for possible later reuse
        OMS_VarObjInfoNode *pLastInfoNode = pLastNeededInfoNode;
        while (pLastInfoNode->m_vobjNext){
          pLastInfoNode = pLastInfoNode->m_vobjNext;
          ++availInfoNodeCnt;
          OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "    Infonode: " << pLastInfoNode->m_vobjOid << "  appended to freelist ");
        }
        // pLastNeededInfoNode is still pointing to the last element which is needed and
        // pLastInfoNode is now pointing to the last infonode for the current object. 
        // Append already existing list behind this object and initialize head 
        // with new starting point of the list.
        pLastInfoNode->m_vobjNext   = pAvailInfoNodes;
        pAvailInfoNodes             = pLastNeededInfoNode->m_vobjNext;

        // Terminate the list of infonodes for the current object
        pLastNeededInfoNode->m_vobjNext = NULL;
      }
      else if (pagesNeeded > pagesCurr) {
        // The object does not own enough infonodes to store the current data

        if (pagesNeeded-pagesCurr > availInfoNodeCnt){
          // There are not enough infonodes in the available list for the current
          // object. Therefore store the object in a separate list for later processing.
          if (pPrev == NULL)
            pHeadOneClass             = pNext;
          else
            pPrev->SetNext(pNext);

          pObjContainer->SetNext(pUnfinishedObjs);
          pUnfinishedObjs             = pObjContainer;
          neededInfoNodeCnt          += (pagesNeeded-pagesCurr);

          OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "    pages in freelist " << availInfoNodeCnt << " => Objects is marked for later processing");
        }
        else {
          // Position the pointer at the end of the chain
          OMS_VarObjInfoNode *pLastInfoNode = pFirstInfoNode;
          while (pLastInfoNode->m_vobjNext){
            pLastInfoNode = pLastInfoNode->m_vobjNext;
          }

          // Take infonodes from the reuse list
          for (int i=0; i<pagesNeeded-pagesCurr; ++i) {
            OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "    Infonode: " << pAvailInfoNodes->m_vobjOid << "  appended to object ");

            pLastInfoNode->m_vobjNext = pAvailInfoNodes;
            pAvailInfoNodes           = pAvailInfoNodes->m_vobjNext;
            pLastInfoNode             = pLastInfoNode->m_vobjNext;
            pLastInfoNode->m_vobjNext = NULL;
            --availInfoNodeCnt;
          }
        }
      }

      if (!(pNext == NULL && pUnfinishedObjs != NULL)){
        // Initialization for next iteration
        if (pObjContainer != pUnfinishedObjs){
          // Remember previous entry in the original list. Necessary for deletion of entries out
          // of the single linked list. This entry only changes if the last object was not appended
          // to the list of unfinished objects
          pPrev = pObjContainer;
        }
        pObjContainer = pNext;
      }
      else {
        // ********* PHASE II **********
        OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "Start of PHASE II");

        // All objects have been considered, but there are still some objects which does
        // not own enough infonodes for their current object size. Either take infonodes
        // from the availableInfoNode list or if there are not enough allocate some from 
        // the kernel.

        OMS_VarObjInfoNode *pInfoNode = NULL;
        tgg91_PageRef       ObjVers;
        while (neededInfoNodeCnt > availInfoNodeCnt){
          // Create new infonodes, with oid ...
          if (IsVersion()){
            // ... from the version
            while (neededInfoNodeCnt > availInfoNodeCnt){
              OmsObjectId   newOid = VersionNewOid();
              OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "  Infonode: " << newOid << "  allocated in version");
              ObjVers.gg91SetNilRef();
              pInfoNode             = new(*this) OMS_VarObjInfoNode(newOid, ObjVers);
              ++availInfoNodeCnt;

              pInfoNode->m_vobjNext = pAvailInfoNodes;
              pAvailInfoNodes       = pInfoNode;
            }
            break; // Workaround LinuxIA64 compiler bug (endless loop). PTS 1144800
          }
          else {
            // ... from the kernel
            cnt = 0;
            while (cnt < OMS_MASS_OPERATION_CNT && neededInfoNodeCnt > availInfoNodeCnt){
              ppContainerId[cnt] = &pClsInfo->GetFileId();
              ++cnt;
              ++availInfoNodeCnt;
            }
            
            // Mass-New from the kernel
            m_currLcSink->NewObjMass(cnt, &m_consistentView, ppContainerId, NULL, NULL, NULL, NULL,
              pOid, pObjVers, errCnt, pDBError);

            // Error handling
            if (errCnt > 0){
              for (int j=0; j<cnt; ++j){
                if (pDBError[j] != e_ok){
                  OMS_Globals::Throw(DbpError(DbpError::DB_ERROR, pDBError[j], msg, pOid[j], __MY_FILE__, __LINE__));
                }
              }
            }

            // Create new infonodes and append then to the list of available infonodes
            for (int j=0; j<cnt; ++j){
              pInfoNode             = new(*this) OMS_VarObjInfoNode(pOid[j], pObjVers[j]);
              pInfoNode->m_newFlag  = true; 
              pInfoNode->m_vobjNext = pAvailInfoNodes;
              pAvailInfoNodes       = pInfoNode;
              OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "  Infonode: " << pOid[j] << "  allocated in kernel");
            }
          }
        }

        // Now append list with unfinished objects to original list and continue processing
        // with these objects. So these objects are considered twice.
        if (pObjContainer == pUnfinishedObjs){
          if (pPrev != NULL)
            pPrev->SetNext(pUnfinishedObjs);
          else
            pHeadOneClass     = pUnfinishedObjs;
        }
        else {
          pObjContainer->SetNext(pUnfinishedObjs);
          pPrev                     = pObjContainer;
        }
        pObjContainer   = pUnfinishedObjs;
        pUnfinishedObjs = NULL;
      } 
    }


    // ********* PHASE III **********
    OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "Start of PHASE III");

    cnt     = 0;
    lockCnt = 0;
    // Remove infonodes which are not needed anymore
    while (pAvailInfoNodes){
      while (cnt < OMS_MASS_OPERATION_CNT && pAvailInfoNodes){
        if (!IsVersion()) {
          // Remember entries for locking
          pLockOid         [lockCnt] = pAvailInfoNodes->m_vobjOid;
          pLockObjVers     [lockCnt] = pAvailInfoNodes->m_vobjSeq;
          ppLockContainerId[lockCnt] = &pClsInfo->GetFileId();
          ++lockCnt;
          OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "  Infonode: " << pAvailInfoNodes->m_vobjOid << "  locked");
        }

        // Append entry to delete-array
        pOid         [cnt] = pAvailInfoNodes->m_vobjOid;
        pObjVers     [cnt] = pAvailInfoNodes->m_vobjSeq;
        ppContainerId[cnt] = &pClsInfo->GetFileId();
        ++cnt; 
        OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "  Infonode: " << pAvailInfoNodes->m_vobjOid << "  deleted");

        OMS_VarObjInfoNode *p = pAvailInfoNodes;
        pAvailInfoNodes = pAvailInfoNodes->m_vobjNext;
        p->deleteSelf(*this);
      }

      if (cnt == OMS_MASS_OPERATION_CNT || (cnt > 0 && pAvailInfoNodes == NULL)){
        // Mass lock
        tsp00_Int8 timeout = 0;
        OMS_HResult hr = m_currLcSink->LockObjMass(
          lockCnt, &m_consistentView, ppLockContainerId, pLockOid, 
          pLockObjVers, timeout, errCnt, pDBError);
        LockResult(lockCnt, pDBError, NULL, pLockOid, msg);
        lockCnt = 0;

        // Mass deletion
        hr = m_currLcSink->DeleteObjMass(
          cnt, &m_consistentView, ppContainerId, VersionContext(),
          pOid, pObjVers, errCnt, pDBError);
        if (errCnt > 0){
          for (int i=0; i<cnt; ++i){
            if (pDBError[i] != e_ok){
              OMS_Globals::Throw(DbpError(DbpError::DB_ERROR, pDBError[i], msg, pOid[i], __MY_FILE__, __LINE__));
            }
          }
        }
        cnt = 0;
      }
    }


    // ********* PHASE IV **********
    OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "Start of PHASE IV");

    OMS_VarObjInfoNode *pInfoNode = NULL;
    pObjContainer = pHeadOneClass;
    cnt           = 0;
    lockCnt       = 0;
    while (pObjContainer) {
      while (pObjContainer && cnt < OMS_MASS_OPERATION_CNT){

        if (pInfoNode == NULL){
          // Consider first/next object in list
          pObjInfo      = REINTERPRET_CAST (OMS_VarObjInfo*, &pObjContainer->m_pobj);
          pInfoNode     = &pObjInfo->m_vobjInfo; 
          pCurrAddr     = pObjInfo->m_pvobj;
          remainingSize = pObjInfo->m_vobjSize;
        }

        while (NULL != pInfoNode && cnt < OMS_MASS_OPERATION_CNT) {
          if (!IsVersion() && (pObjContainer->m_oid != pInfoNode->m_vobjOid || !pObjContainer->LockedFlag())) {
            if (!pInfoNode->m_newFlag){            
              // Entries must be locked if no version is active and if either the current infonode is
              // a following node (which is not newly created) or if the base node is not locked.
              pLockOid         [lockCnt] = pInfoNode->m_vobjOid;
              pLockObjVers     [lockCnt] = pInfoNode->m_vobjSeq;
              ppLockContainerId[lockCnt] = &pClsInfo->GetFileId();
              ++lockCnt;
              OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "  Infonode: " << pInfoNode->m_vobjOid << "  locked");
            }
          }

          // Insert entry into array
          pOid         [cnt] = pInfoNode->m_vobjOid;
          pObjVers     [cnt] = pInfoNode->m_vobjSeq;
          ppContainerId[cnt] = &pClsInfo->GetFileId();
          pObjSize     [cnt] = pObjInfo->m_vobjSize;
          pBodySize    [cnt] = (remainingSize > VAR_OBJ_CHUNK_SIZE ? VAR_OBJ_CHUNK_SIZE : remainingSize);
          ppObj        [cnt] = pCurrAddr;
          pContObj     [cnt] = (pObjContainer->m_oid != pInfoNode->m_vobjOid);  // PTS 1124278

          if (pInfoNode->m_vobjNext){
            pNextOid   [cnt] = pInfoNode->m_vobjNext->m_vobjOid;
          }
          else {
            pNextOid   [cnt].setNil();
          }
          OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "  Infonode: " << pInfoNode->m_vobjOid << "  updated");

          pCurrAddr      += pBodySize[cnt]; 
          remainingSize  -= pBodySize[cnt]; 
          ++cnt; 
          pInfoNode = pInfoNode->m_vobjNext;
        }

        if (pInfoNode == NULL){
          // Previous object is completely flushed, therefore consider next object
          pNext = pObjContainer->GetNext();

          // Append object to list for later freeing of memory
          pObjContainer->SetNext(pFreeList);
          pFreeList     = pObjContainer;

          pObjContainer = pNext;  
          pHeadOneClass = pObjContainer;
        }
      }

      if (cnt == OMS_MASS_OPERATION_CNT || (cnt > 0 && pObjContainer == NULL)){
        int errCnt = 0;

        // Mass lock
        tsp00_Int8 timeout = 0;
        OMS_HResult hr = m_currLcSink->LockObjMass(
          lockCnt, &m_consistentView, ppLockContainerId, pLockOid, 
          pLockObjVers, timeout, errCnt, pDBError);
        LockResult(lockCnt, pDBError, NULL, pLockOid, msg);
        lockCnt = 0;

        // Mass update
        hr = m_currLcSink->UpdateVarObjMass(
          cnt, &m_consistentView, ppContainerId, VersionContext(), pOid, pObjVers, 
          pObjSize, pBodySize, ppObj, pNextOid, pContObj, errCnt, pDBError);
        if (errCnt > 0){
          for (int i=0; i<cnt; ++i){
            if (pDBError[i] != e_ok){
              OMS_Globals::Throw(DbpError(DbpError::DB_ERROR, pDBError[i], msg, pOid[i], __MY_FILE__, __LINE__));
            }
          }
        }
        cnt = 0;
      }
    }

    pHeadOneClass = NULL;
    if (pHeadOtherClasses){
      // Take all objects which belong to the same class (as the first object) from the 
      // list pHeadOtherClasses and insert them into the list pHeadOneClass for next 
      // iteration of algorithm. All objects which do not belong to this class remain
      // in the old list. 
      // Assumption: It is assumed that (almost) all objects, which are stored in the local
      //   cache during a single transaction, belong to the same container. If there are many
      //   containers for var-objects this algorithm must be changed because of its
      //   quatratic running time in such a case.
      OMS_ClassIdEntry   *pClsInfo = pHeadOtherClasses->GetContainerInfoNoCheck(this);

      OmsObjectContainer *next = NULL;
      OmsObjectContainer *prev = NULL;
      OmsObjectContainer *curr = pHeadOtherClasses;
      while (curr){
        next = curr->GetNext();
        if (curr->GetContainerInfoNoCheck(this) == pClsInfo){
          if (prev)
            prev->SetNext(next);
          else 
            pHeadOtherClasses = curr->GetNext();
          curr->SetNext(pHeadOneClass); 
          pHeadOneClass = curr;
        }
        else {
          prev = curr;
        }
        curr = next;
      }
    }
  }

  OMS_DETAIL_TRACE(omsTrFlushCache | omsTrVarObject, m_session->m_lcSink, "End FLUSH of VAR-Objects");
}

/*----------------------------------------------------------------------*/
// PTS 1122280
void OMS_Context::FlushVersionObjCache() 
{
  const char *msg = "OMS_Context::FlushVersionObjCache ";

  bool checkHash = false;

  OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, "Start of " << msg);
  try {
    OMS_OidHash::OidIter iter = m_oidDir.First();
    while (iter) {
      // Check whether oid hash structure still fulfills some invariants. 
      // Check is normally switched off
      if (checkHash) {
        m_oidDir.HashCheck();
      }

      OmsObjectContainer *curr = iter();
      ++iter;

      if (curr->DeletedFlag() && IsVersionOid(curr->m_oid)) {
        // Object is created in a version and marked as deleted
        OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, "  Delete object " << curr->m_oid);

        OMS_ClassIdEntry* pInfo = curr->GetContainerInfoNoCheck(this);

        // Delete entry in the oid-directory. Do not adapt key-structure as this is done
        // in the method VersionDelKey
        m_oidDir.HashDelete(curr->m_oid, false);

        if (pInfo->IsKeyedObject()){
          // Check is necessary as there might be several objects with the same key,
          // but only one with the delete-flag = false. If there are several entries
          // with the same key, then the corresponding entry in the key-tree is pointing 
          // to the non-deleted object and therefore it must be checked whether the
          // current object is equal with the object the tree-entry is pointing to. 
          // ASSUMPTION: The number of objects with the delete flag is quite small 
          //   compared with the overall number. Therefore the running time of O(2*m*log(n))
          //   (where m is the number of deleted objects and n is the number of objects
          //   in the version) is acceptable. If it is expected, that the number of deleted
          //   objects is about the number of objects, then it is better to remember the
          //   non-deleted objects, drop the whole tree and reinsert the remembered objects
          //   again. Unfortunately it is not possible to iterate over the tree and delete
          //   all objects which are marked as deleted (running time O(n)), because the 
          //   iterator will get confused when updates are executed during iteration. 
          if (curr == pInfo->VersionFindKey(pInfo->GetKeyPtr(curr))) {
            // Delete entry out of the tree with version objects
            pInfo->VersionDelKey(curr, this);
          }
        }

        if (!pInfo->IsVarObject()){
          // Delete entry out of the list of version objects
          m_newObjCache.removeObject(curr, this);
        }

        // Append object frame to free-list
        pInfo->chainFree(*this, curr, 5);
      }
    }
  }
  catch (DbpError &e){
    OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, msg << "Error occured: " << e.m_errorNo);
    OMS_Globals::Throw(e);
  }
  OMS_DETAIL_TRACE(omsTrFlushCache, m_session->m_lcSink, "End of " << msg);
}

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

void OMS_Context::MarkNotBoundToTrans(bool callFromDestructor) { 
  m_boundToTrans = false;
  if (!m_isOpen) {
    co10_MarkUnloadable(this, callFromDestructor);
    this->ResetLcSink();
  }
}

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

void  OMS_Context::OpenVersion(OMS_Session* session) {
  AssignLcSink(session->m_lcSink);
  m_currLcSink->GetDateTime(&m_lastOpenDate, &m_lastOpenTime);
  m_session    = session;
  m_isOpen     = true;
}

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

void OMS_Context::ReduceHeapUsage() /* PTS 1110149 */
{
  this->UnLoad();
  m_containerDir.UnregisterAll();
}

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

void OMS_Context::ResetLcSink() {
  m_currLcSink = NULL;
}

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

/* PTS 1115134 */

void  OMS_Context::ResetVersion(OMS_Session& session) 
{
  if (m_isVersion)
  {
    this->AssignLcSink(session.m_lcSink);
    this->EmptyObjCache();
    for (OMS_ClassIdHash::Iter iter = m_containerDir.First(); iter; ++iter)
    {
      iter()->VersionDelIndex(false, this);
    }
    this->AssignLcSink(NULL);
    tgg00_BasisError DBError;
    session.m_lcSink->ResetVersion((unsigned char*) &m_versionContext, DBError);
    this->CheckDBError(DBError, "OMS_Context::ResetVersion", OmsObjectId());
  }
}

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

void OMS_Context::UnLoad() 
{
  const bool emptyObjCache = true;
  pasbool*     pToCancel;
  tsp00_TaskId taskId;
  OMS_Session *oldSession = m_session;

  // reassign pointer to session, may not be valid, if called from bad alloc handling
  OMS_Globals::GetCurrentLcSink()->GetDefaultContext(REINTERPRET_CAST(void**, &this->m_session), &pToCancel, taskId);
  AssignLcSink(m_session->m_lcSink);  // Reassign sink pointer, too  PTS 1133345
  
  m_pVersionContext = &m_versionContext;
  FlushObjCache(emptyObjCache, false);  // No hash-adaptation PTS 1118855 
  m_containerDir.ClearFreeLists(1);
  for (OMS_ClassIdHash::Iter iter = m_containerDir.First(); iter; ++iter) {
    iter()->VersionDelIndex(false, this);
  }
  OMS_Globals::GetKernelInterface()->IncOmsVersionUnloadCounter();

  // Reset local pointers if they have been set by this routine  PTS 1133345
  if (oldSession != m_session){
    m_session = NULL;
    AssignLcSink(NULL);
  }
}

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

void  OMS_Context::VersionClearObjCache() {

  if (m_session->IsDataChanged()) {
    // already flushed something, cannot clear cache before opening version
    this->CheckDBError(e_transaction_end_required, "OMS_Context::VersionClearObjCache", OmsObjectId()); 
  }
  OmsObjectContainerPtr curr;

  for (OMS_OidHash::OidIter iter = m_oidDir.First(); iter; ++iter) {
    curr = iter();
    if ((curr->StoredFlag()) || (curr->DeletedFlag())) {
      this->CheckDBError(e_transaction_end_required, "OMS_Context::VersionClearObjCache", OmsObjectId()); 
    }
    if (curr->VarObjFlag()) {
      REINTERPRET_CAST(OMS_VarObjInfo*, &curr->m_pobj)->freeVarObj(this);
    }
    else {
      if (curr->LockedFlag()) {
        this->FlushObj(curr);
      }
    }
  }  
  m_oidDir.SetEmpty();
  m_newObjCache.SetEmpty();
}

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

void OMS_Context::NewConsistentView() {
  tsp00_Int2 DBError;

  tgg91_TransNo consistentView = m_consistentView;
  OMS_HResult hr = m_currLcSink->NewConsistentView ((unsigned char*) &m_consistentView, &DBError);
  if (0 != DBError) {
    m_consistentView = consistentView;
    m_session->ThrowDBError (DBError, "omsNewConsistentView", OmsObjectId(), __MY_FILE__, __LINE__);
    return;
  }
  m_session->ChangedConsistentView();
}

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

OmsObjectId OMS_Context::VersionNewOid()
{
  OmsObjectId oid;
  oid.setPno(OMS_VERSION_OBJ_PAGE_NO);
  IncVersionKey();
  tgg92_OmsTypeOid* pOid = REINTERPRET_CAST(tgg92_OmsTypeOid*, &oid);
  pOid->generation = CurrentVersionKey() / (256 * 256);  
  pOid->pagePos    = CurrentVersionKey() % (256 * 256);
  return oid;
}

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

/*! Try to aquire locks for all objects in the local oms-cache, for which
**  there exists before-images, but which are not locked. 
**  This might happen, as it is possible to get an 'forupd'-pointer without
**  getting a lock (see PTS 1128108).
*/
void OMS_Context::LockObjWithBeforeImage()   
{
  const char* msg = "OMS_Context::LockObjWithBeforeImage";

  int                  cnt        = 0;
  int                  errCnt     = 0;
  OmsObjectContainer  *pObj       = NULL;

  char                    pOid_u         [OMS_MASS_OPERATION_CNT * sizeof(OmsObjectId)];
  OmsObjectId            *pOid = reinterpret_cast<OmsObjectId*>(&pOid_u[0]);
  tgg91_PageRef           pObjVers       [OMS_MASS_OPERATION_CNT];
  tgg00_FileId           *ppContainerId  [OMS_MASS_OPERATION_CNT];
  char                    pContainerId_u [OMS_MASS_OPERATION_CNT * sizeof(OMS_UnknownContainerId)];
  OMS_UnknownContainerId *pContainerId = reinterpret_cast<OMS_UnknownContainerId*>(&pContainerId_u[0]);
  tgg00_BasisError        pDBError       [OMS_MASS_OPERATION_CNT];
  OmsObjectContainer     *ppObj          [OMS_MASS_OPERATION_CNT];

  OMS_OidHash::OidIter iter       = m_oidDir.First();
  while (iter) {
    while (iter && cnt < OMS_MASS_OPERATION_CNT) {
      pObj = iter();

      if (pObj->m_beforeImages != 0 && !pObj->LockedFlag()){
        pOid         [cnt] =  pObj->m_oid;
        pObjVers     [cnt] =  pObj->m_objseq;
        ppContainerId[cnt] = &pObj->m_containerInfo->GetFileId();
        ppObj        [cnt] =  pObj;
        ++cnt;
      }
      ++iter;
    }

    if (cnt > 0) {
      // Mass lock
      tsp00_Int8 remainingTime = 0;
      OMS_HResult hr = m_session->m_lcSink->LockObjMass(
        cnt, &m_consistentView, ppContainerId, pOid, 
        pObjVers, remainingTime, errCnt, pDBError);

      for (int i=0; i<cnt; ++i){
        if (pDBError[i] != e_ok) {
          m_session->ThrowDBError(pDBError[i], msg, pOid[i], __MY_FILE__, __LINE__);
        }
        else {
          // Mark object as locked in the liboms
          ppObj[i]->MarkLocked();
        }
      }
      cnt = 0;
    }
  }
}

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

void OMS_Context::ReleaseAllUnchanged()   // PTS 1128262
{
  OMS_OidHash::OidIter iter = m_oidDir.First();
  while (iter) {
    OmsObjectContainer *pObj = iter();

    ++iter;
    m_session->ReleaseObj(pObj, /*unchangedOnly=*/true);
  }
}


