Public Member Functions | Public Attributes | Private Member Functions | Private Attributes

fio_JourMgr Class Reference
[Classes for large file i/o, relative file, and DataBase]

#include <fio_JourMgr.h>

Collaboration diagram for fio_JourMgr:
Collaboration graph
[legend]

List of all members.

Public Member Functions

 fio_JourMgr (fio_DataBase *a_database_p)
virtual ~fio_JourMgr ()
fio_JourHandleGetJourWriteHandle (ivd_FileBufSize_t a_transSize, bool &a_newJourFile)
 When transaction end event occure than transaction is going to store its data to journal file.
fio_JourHandleGetJourReadHandle (fio_JourFile *a_jourFile_p, ivd_FilePosition_t a_position)
 When Reply only handle for read is need, but it must be released as write handle, to proper update of PDBI.
void ReleaseJourHandle (fio_JourHandle *s_jourHnd_p)
 At the end of transaction commit the jour handle is going to release.
void ReleaseWaiters ()
void Replay (fio_JourFile &jourFile, ivd_FilePosition_t position, fio_Transaction &trans)
void SetLastSyncPosition (bool a_forced)
void Synchronize ()
 all data on disk are synchronized with memory
void RemoveArhivedJourFile ()

Public Attributes

fio_PermDBinfoMgr m_permDBinfoMgr
 manage permanent DB information
ivd_FileBufSize_t m_changedDataSize
 changed data size of all transaction from last DB commit
int m_numOfTrans
 num of transactions from last DB commit
 log_CLASSID_m

Private Member Functions

void CleanJourHandleListUpTo (fio_JourHandle_p_l_t::iterator iter)
 remove journal handlers from jour hand poiter list up to specified iterator
void AddNewJourFile ()
 When current jour file has exceded minimum size a new one is created.
ivd_TransNumber_t IncTransNumber ()
 return current transaction number and increment it for future

Private Attributes

fio_JourHandle_p_l_t m_jourHandle_p_l
 list of journal handler pointers
fio_DataBasem_database_p
 pointer to DB
fio_JourFilem_currentJourFile_p
 current active file
cmn_Mutex m_jourFile_x
 exclusive lock to current journal file
ivd_TransNumber_t m_transNumber
 current available transaction number for use in journal handle
cmn_Mutex m_transNumber_x
 exclusive access to transaction number
ivd_FileID_t m_lastJourFileNumber
 number of last journal got from name
int m_maxWaitersToPDBIUpdate
 how much transaction could wait in list to its info be updatted in PDBI

Detailed Description

Definition at line 61 of file fio_JourMgr.h.


Constructor & Destructor Documentation

fio_JourMgr::fio_JourMgr ( fio_DataBase a_database_p  ) 

Definition at line 61 of file fio_JourMgr.cpp.

References AddNewJourFile(), c_fio_JourFileExt, c_fio_maxWaitersToPDBIUpdate, dbg_NORM, fio_File::GetFileName(), fio_PermDBinfoMgr::GetLastJourFileID(), fio_PermDBinfoMgr::GetLastSyncJourFileID(), fio_PermDBinfoMgr::GetLastSyncPosition(), cmn_DirLst::GetNextName(), fio_DataBase::GetPath(), fio_PermDBinfoMgr::GetPermDBinfo(), fio_DataBase::GetTransObj(), ie_JOUR_FILE_MISSING, fio_JourFile::IncRefCount(), ivd_Error, ivd_NULLCHK_m, fio_JourFile::JourFileName(), fio_JourFile::JourFileNumber(), log_DBG_m, log_FUNC_m, log_WriteEvent(), m_currentJourFile_p, m_database_p, m_lastJourFileNumber, m_maxWaitersToPDBIUpdate, m_permDBinfoMgr, path, fio_DataBase::ReleaseTransObj(), Replay(), and s_className.

            :
            m_database_p(a_database_p),
            m_currentJourFile_p(NULL),
            m_transNumber(0),
            m_lastJourFileNumber(0),
            m_maxWaitersToPDBIUpdate(c_fio_maxWaitersToPDBIUpdate),
            m_permDBinfoMgr(false, // never read only mode
                            m_database_p->GetDataBaseID(),
                            m_database_p->GetPath(),
                            "pdbi.ldb"),
            m_changedDataSize(0),  //before DB commit
            m_numOfTrans(0)         //before DB commit
{
    log_FUNC_m(fio_JourMgr);

    cmn_Path path(m_database_p->GetPath());

    list<string> journalFiles_l;

    cmn_DirLst dirList(path, c_fio_JourFileExt);

    string name = dirList.GetNextName();
    while  (name.length() > 0) {        // get all journal files from directory
        journalFiles_l.push_back(name); // store them to list
        name = dirList.GetNextName();
    }

    if (journalFiles_l.size() > 0) {   // if already exist journal file(s)
        journalFiles_l.sort();         // do a sort 



        // only one wait transaction could be in list at Reply
        m_maxWaitersToPDBIUpdate = 1;

        ivd_FileID_t lastSyncJourID   = m_permDBinfoMgr.GetLastSyncJourFileID();
        if (lastSyncJourID == 0) {
            lastSyncJourID = 1;  // Reaply must start from first journal file
        }

        ivd_FileID_t lastJourID       = m_permDBinfoMgr.GetLastJourFileID();

        ivd_FileID_t repliedJourID = 0;

        ivd_FilePosition_t position = m_permDBinfoMgr.GetLastSyncPosition();
        log_DBG_m(dbg_NORM, "Last sync jourID 0x" << hex << lastSyncJourID << dec
                         << ",  position " << position);

        fio_Transaction *trans = m_database_p->GetTransObj();

        list<string>::iterator last = --journalFiles_l.end();
        list<string>::iterator p;

        string lastJourFileName = *last;
        if (lastJourID > fio_JourFile::JourFileNumber(*last)) {
            lastJourFileName = fio_JourFile::JourFileName(lastJourID);
        }

        ostringstream errorMessage;
        errorMessage << " DB '" << path << "' can not be properly reaplied." << endl
            << " The mentioned file and sequence of other journal files up to "
            << hex << lastJourFileName << dec
            << " are needed to reaply DB." << endl
            << " If journal files are no more available then "
            << " remove journal directory and restart the system." << endl
            << " Some transactions will be lost when removing journal directory."
            << " Use check tool to find mismatches.";

        ivd_FileID_t jourFileID = 0;
        for (p  = (journalFiles_l.begin());
             p !=  journalFiles_l.end(); 
             p++) {
            jourFileID = fio_JourFile::JourFileNumber(*p);
            log_DBG_m(dbg_NORM, "Check jour file '" << fio_JourFile::JourFileName(jourFileID) << "'.");
            if (jourFileID >= lastSyncJourID) {

                if ( repliedJourID == 0) {
                    if (jourFileID > lastSyncJourID) {
                        ostringstream sstr;
                        sstr << "Missing journal file '" << fio_JourFile::JourFileName(lastSyncJourID) << "'." << endl
                            << errorMessage.str();
                        throw ivd_Error(ie_JOUR_FILE_MISSING,
                            sstr.str(), true);
                    }
                    repliedJourID = jourFileID;
                    // write a log
                    ostringstream sstr;
                    sstr << "Reapply of DB '" << path << "' started at journal file '" << *p
                         << "' last needed journal file is '" << lastJourFileName;
                    log_WriteEvent(sstr.str());
                }
                else if (++repliedJourID != jourFileID) {
                        ostringstream sstr;
                        sstr << "Expected journal file '" << fio_JourFile::JourFileName(repliedJourID) << "' not found." << endl
                            << errorMessage.str();
                            throw ivd_Error(ie_JOUR_FILE_MISSING,
                                sstr.str(), true);
                }

                fio_JourFile *jf = new fio_JourFile(m_permDBinfoMgr.GetPermDBinfo(),
                                                    false,  // never read only mode
                                                    jourFileID,
                                                    path,
                                                   *p);
                ivd_NULLCHK_m(jf, fio_JourFile::s_className);
                if (p == last) {  // use last jour file as active
                    m_currentJourFile_p = jf;
                    m_currentJourFile_p->IncRefCount();
                }
                // reconstruct UV buffer and reply unfinished transaction 
                Replay(*jf, position, *trans);
                // fix bug 526
                position = 0; // other jour files are read from beginning

                //if (p != last) { // NOTE removing jf produce corre dump ????
                //    delete jf;
                //}
            }
        }
        if (jourFileID < lastJourID) {
            //
            //
            ostringstream sstr;
            sstr << "Expected journal file '" << fio_JourFile::JourFileName(++jourFileID) << "' not found." << endl
                << errorMessage.str();
                throw ivd_Error(ie_JOUR_FILE_MISSING,
                    sstr.str(), true);
        }

        m_database_p->ReleaseTransObj(&trans);
        // when reply is done then set trans waiters to higher value
        m_maxWaitersToPDBIUpdate = c_fio_maxWaitersToPDBIUpdate;
    }
    else {
        AddNewJourFile();
    }

    m_lastJourFileNumber =  fio_JourFile::JourFileNumber(m_currentJourFile_p->GetFileName());
}

Here is the call graph for this function:

fio_JourMgr::~fio_JourMgr (  )  [virtual]

Definition at line 205 of file fio_JourMgr.cpp.

References fio_JourFile::DecRefCount(), log_FUNC_m, m_currentJourFile_p, and NULL.

                          {
    log_FUNC_m(~fio_JourMgr);
// fix bug 507
// all other jour file are deletted when
// last jourHandle of particular file is deletted too
    if (m_currentJourFile_p != NULL) {
        m_currentJourFile_p->DecRefCount();
    }
// end
}

Here is the call graph for this function:


Member Function Documentation

void fio_JourMgr::AddNewJourFile (  )  [private]

When current jour file has exceded minimum size a new one is created.

Definition at line 377 of file fio_JourMgr.cpp.

References fio_JourFile::DecRefCount(), fio_DataBase::GetPath(), fio_PermDBinfoMgr::GetPermDBinfo(), fio_JourFile::IncRefCount(), ivd_NULLCHK_m, log_FUNC_m, m_currentJourFile_p, m_database_p, m_lastJourFileNumber, m_permDBinfoMgr, NULL, and s_className.

Referenced by fio_JourMgr(), and GetJourWriteHandle().

Here is the call graph for this function:

Here is the caller graph for this function:

void fio_JourMgr::CleanJourHandleListUpTo ( fio_JourHandle_p_l_t::iterator  iter  )  [private]

remove journal handlers from jour hand poiter list up to specified iterator

Definition at line 325 of file fio_JourMgr.cpp.

References m_jourHandle_p_l.

Referenced by ReleaseWaiters().

                                                                           {
//    log_FUNC_m(CleanJourHandleListUpTo);

    fio_JourHandle_p_l_t::iterator p;
    for (p  = m_jourHandle_p_l.begin();
         p != iter;
         p++) {
        delete (*p);
    }
    m_jourHandle_p_l.erase(m_jourHandle_p_l.begin(), iter);
}

Here is the caller graph for this function:

fio_JourHandle * fio_JourMgr::GetJourReadHandle ( fio_JourFile a_jourFile_p,
ivd_FilePosition_t  a_position 
)

When Reply only handle for read is need, but it must be released as write handle, to proper update of PDBI.

Definition at line 238 of file fio_JourMgr.cpp.

References fio_JourFile::GetJourReadHandle(), cmn_Mutex::Lock(), m_jourFile_x, m_jourHandle_p_l, and cmn_Mutex::Unlock().

Referenced by Replay().

                                                {
//    log_FUNC_m(GetJourReadHandle);
    m_jourFile_x.Lock();
    fio_JourHandle *jourHandle = a_jourFile_p->GetJourReadHandle(a_position);
    m_jourHandle_p_l.push_back(jourHandle);
    m_jourFile_x.Unlock(); // guarantee sequence of transaction number

    return jourHandle;
}

Here is the call graph for this function:

Here is the caller graph for this function:

fio_JourHandle * fio_JourMgr::GetJourWriteHandle ( ivd_FileBufSize_t  a_transSize,
bool &  a_newJourFile 
)

When transaction end event occure than transaction is going to store its data to journal file.

To do that needs journal handle. The jour handle has method to store and read data to jour file. To get a journal handle you need size of transaciton.

Definition at line 218 of file fio_JourMgr.cpp.

References AddNewJourFile(), c_fio_MaxJourSize, fio_JourFile::GetFreeTransPosition(), fio_JourFile::GetJourWriteHandle(), IncTransNumber(), cmn_Mutex::Lock(), m_currentJourFile_p, m_jourFile_x, m_jourHandle_p_l, fio_JourHandle::SetTransNumber(), and cmn_Mutex::Unlock().

Referenced by fio_Transaction::EndTransaction().

                                                                     {
//    log_FUNC_m(GetJourWriteHandle);
    m_jourFile_x.Lock();
    if (m_currentJourFile_p->GetFreeTransPosition() > c_fio_MaxJourSize) {
        a_newJourFile = true;
        AddNewJourFile();
    }
    fio_JourHandle *jourHandle = m_currentJourFile_p->GetJourWriteHandle(a_transSize);

    // update journal handle with proper transaction number
    jourHandle->SetTransNumber(IncTransNumber());
    m_jourHandle_p_l.push_back(jourHandle);
    m_jourFile_x.Unlock(); // guarantee sequence of transaction number

    return jourHandle;
}

Here is the call graph for this function:

Here is the caller graph for this function:

ivd_TransNumber_t fio_JourMgr::IncTransNumber (  )  [private]

return current transaction number and increment it for future

Definition at line 396 of file fio_JourMgr.cpp.

References cmn_Mutex::Lock(), log_FUNC_m, m_transNumber, m_transNumber_x, and cmn_Mutex::Unlock().

Referenced by GetJourWriteHandle().

Here is the call graph for this function:

Here is the caller graph for this function:

void fio_JourMgr::ReleaseJourHandle ( fio_JourHandle s_jourHnd_p  ) 

At the end of transaction commit the jour handle is going to release.

this even is used to update transaction list of finished transaction and update permanent DB info.

See also:
fio_PermDBinfoMgr

Definition at line 252 of file fio_JourMgr.cpp.

References fio_JourHandle::Release(), and ReleaseWaiters().

Referenced by fio_Transaction::EndTransaction(), and Replay().

                                                               {
//    log_FUNC_m(ReleaseJourHandle);
//    log_DBG_m(dbg_NORM, "ReleaseJourHandle");
    s_jourHnd_p->Release();
    ReleaseWaiters();
}

Here is the call graph for this function:

Here is the caller graph for this function:

void fio_JourMgr::ReleaseWaiters (  ) 

Definition at line 261 of file fio_JourMgr.cpp.

References CleanJourHandleListUpTo(), fio_DataBase::Commit(), dbg_NORM, fio_JourHandle::GetFileID(), fio_JourHandle::GetLimitPosition(), fio_JourHandle::GetTransNumber(), cmn_Mutex::Lock(), log_DBG_m, log_FUNC_m, m_database_p, m_jourFile_x, m_jourHandle_p_l, m_maxWaitersToPDBIUpdate, m_permDBinfoMgr, fio_PermDBinfoMgr::SetLastTransaction(), and cmn_Mutex::Unlock().

Referenced by ReleaseJourHandle(), and Synchronize().

                                 {
//    log_DBG_m(dbg_NORM, "ReleaseWaiters");
    m_jourFile_x.Lock();

    // iterate through journal handle pointer list
    // and check if all from begin until current are released
    // if so, then update PDBI in case this is the last in list
    // or there are enought journal handle delayed (it's time to flush)
    // otherwise is just set as released.
    fio_JourHandle_p_l_t::iterator p;
    int relesedWaiters = 0;
    for (p  = m_jourHandle_p_l.begin();
         p != m_jourHandle_p_l.end();
         p++) {
        if ((*p)->GetReleased()) {
             relesedWaiters++;
             continue; // count countinuous realsed journal handles
        }
        break; // break as soon found one not released
    }
//    log_DBG_m(dbg_NORM, "ReleasedWaiters " << relesedWaiters);

    // when enought journal handleas are released then update PDBI
    if (relesedWaiters >= m_maxWaitersToPDBIUpdate) {
        log_FUNC_m(ReleaseWaiters);
        log_DBG_m(dbg_NORM, "Time to commit DB and release waiting journal handles JH."
                            " Released JH" << relesedWaiters
                            << "  All JH " << m_jourHandle_p_l.size());
        m_database_p->Commit();
        p--; // get previous iterator that is last released
        fio_JourHandle* lastJourHnd_p = *p;
        m_permDBinfoMgr.SetLastTransaction(lastJourHnd_p->GetTransNumber(),
                                           lastJourHnd_p->GetFileID(),
                                           lastJourHnd_p->GetLimitPosition());
        p++;  // remove all until next
        CleanJourHandleListUpTo(p);
    }
    m_jourFile_x.Unlock();
}

Here is the call graph for this function:

Here is the caller graph for this function:

void fio_JourMgr::RemoveArhivedJourFile (  ) 

Definition at line 408 of file fio_JourMgr.cpp.

References c_fio_JourFileExt, cmn_File::DeleteFile(), fio_PermDBinfoMgr::GetArhivedSyncJourFileID(), cmn_DirLst::GetNextName(), fio_DataBase::GetPath(), fio_JourFile::JourFileNumber(), m_database_p, m_permDBinfoMgr, and path.

Referenced by fio_DataBase::Continue().

                                        {
    // get last sink possition
    ivd_FileID_t syncJourID = m_permDBinfoMgr.GetArhivedSyncJourFileID();

    if (syncJourID == 0) {
        return;
    }
    
    // remove all jour files older than last arhived sync
    cmn_Path path = m_database_p->GetPath();
    cmn_DirLst dirList(path, c_fio_JourFileExt);

    string name = dirList.GetNextName();
    while  (name.length() > 0) {        // get all journal files from directory
        ivd_FileID_t jourID = fio_JourFile::JourFileNumber(name);
        if (jourID < syncJourID) {
            cmn_File::DeleteFile(path + name);
        }
        name = dirList.GetNextName();
    }
}

Here is the call graph for this function:

Here is the caller graph for this function:

void fio_JourMgr::Replay ( fio_JourFile jourFile,
ivd_FilePosition_t  position,
fio_Transaction trans 
)

Definition at line 339 of file fio_JourMgr.cpp.

References dbg_DETAIL, fio_JourHandle::GetCurrentPosition(), fio_File::GetFileName(), fio_JourFile::GetFreeTransPosition(), GetJourReadHandle(), fio_PermDBinfoMgr::GetLastTransNumber(), fio_JourHandle::GetTransNumber(), ie_FATAL_ERROR, log_DBG_m, log_FUNC_m, m_permDBinfoMgr, m_transNumber, fio_Transaction::ReadTransactionFromLog(), ReleaseJourHandle(), and fio_JourHandle::SetLastDBTransNum().

Referenced by fio_JourMgr().

                                                                                                    {
    log_FUNC_m(Replay);

    fio_JourHandle *jourHnd = GetJourReadHandle(&jourFile, position);
    // update DB from this transacton further
    jourHnd->SetLastDBTransNum(m_permDBinfoMgr.GetLastTransNumber());
    
    log_DBG_m(dbg_DETAIL, "Reply journal file " << jourFile.GetFileName() 
                       << " position " << position 
                       << " last trans in DB " << m_permDBinfoMgr.GetLastTransNumber());

    ivd_FilePosition_t fileSize = jourFile.GetFreeTransPosition();
    if (fileSize < position) {
        ostringstream sstr;
        sstr << " Journal file is shorter than expected. Journal file size=" << fileSize
             << ", but expected size=" << position << " " << endl
             << *jourHnd;
//        log_DBG_m(dbg_DETAIL, sstr.str());
        throw ivd_InternalError(ie_FATAL_ERROR, sstr.str());
    }
    
    while (jourHnd->GetCurrentPosition() < fileSize) {
        // use jourhandle to proper file position
        // update DB from transaction number
        // transaction with trans numb < a_transNum is used for UV list build
        trans.ReadTransactionFromLog(jourHnd);
    }

    ivd_TransNumber_t    transNum = jourHnd->GetTransNumber();
    if (transNum > 0) {
        m_transNumber = transNum;
    }
    log_DBG_m(dbg_DETAIL, "Last transaction number = " << m_transNumber);
    ReleaseJourHandle(jourHnd);
}

Here is the call graph for this function:

Here is the caller graph for this function:

void fio_JourMgr::SetLastSyncPosition ( bool  a_forced  ) 

Definition at line 314 of file fio_JourMgr.cpp.

References fio_File::GetFileID(), fio_JourFile::GetFreeTransPosition(), log_FUNC_m, m_currentJourFile_p, m_permDBinfoMgr, fio_PermDBinfoMgr::SetLastSyncPosition(), and Synchronize().

Referenced by fio_DataBase::Synchronize().

Here is the call graph for this function:

Here is the caller graph for this function:

void fio_JourMgr::Synchronize (  ) 

all data on disk are synchronized with memory

Definition at line 304 of file fio_JourMgr.cpp.

References c_fio_maxWaitersToPDBIUpdate, log_FUNC_m, m_maxWaitersToPDBIUpdate, and ReleaseWaiters().

Referenced by SetLastSyncPosition().

Here is the call graph for this function:

Here is the caller graph for this function:


Member Data Documentation

Definition at line 105 of file fio_JourMgr.h.

changed data size of all transaction from last DB commit

Definition at line 99 of file fio_JourMgr.h.

current active file

Definition at line 76 of file fio_JourMgr.h.

Referenced by AddNewJourFile(), fio_JourMgr(), GetJourWriteHandle(), SetLastSyncPosition(), and ~fio_JourMgr().

pointer to DB

Definition at line 73 of file fio_JourMgr.h.

Referenced by AddNewJourFile(), fio_JourMgr(), ReleaseWaiters(), and RemoveArhivedJourFile().

exclusive lock to current journal file

Definition at line 79 of file fio_JourMgr.h.

Referenced by GetJourReadHandle(), GetJourWriteHandle(), and ReleaseWaiters().

list of journal handler pointers

Definition at line 70 of file fio_JourMgr.h.

Referenced by CleanJourHandleListUpTo(), GetJourReadHandle(), GetJourWriteHandle(), and ReleaseWaiters().

number of last journal got from name

Definition at line 89 of file fio_JourMgr.h.

Referenced by AddNewJourFile(), and fio_JourMgr().

how much transaction could wait in list to its info be updatted in PDBI

Definition at line 92 of file fio_JourMgr.h.

Referenced by fio_JourMgr(), ReleaseWaiters(), and Synchronize().

num of transactions from last DB commit

Definition at line 102 of file fio_JourMgr.h.

manage permanent DB information

Definition at line 96 of file fio_JourMgr.h.

Referenced by AddNewJourFile(), fio_JourMgr(), ReleaseWaiters(), RemoveArhivedJourFile(), Replay(), and SetLastSyncPosition().

current available transaction number for use in journal handle

Definition at line 82 of file fio_JourMgr.h.

Referenced by IncTransNumber(), and Replay().

exclusive access to transaction number

Definition at line 85 of file fio_JourMgr.h.

Referenced by IncTransNumber().


The documentation for this class was generated from the following files: