<commit_before>/*
// $Id$
// Fennel is a library of data storage and processing components.
// Copyright (C) 2005-2005 The Eigenbase Project
// Copyright (C) 2005-2005 Disruptive Tech
// Copyright (C) 2005-2005 LucidEra, Inc.
// Portions Copyright (C) 1999-2005 <NAME>
//
// 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 approved by The Eigenbase Project.
//
// 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
*/

#include "fennel/common/CommonPreamble.h"
#include "fennel/txn/LogicalTxnLog.h"
#include "fennel/txn/LogicalTxn.h"
#include "fennel/txn/LogicalTxnStoredStructs.h"
#include "fennel/segment/CrcSegOutputStream.h"
#include "fennel/segment/SpillOutputStream.h"
#include "fennel/segment/SegmentFactory.h"
#include "fennel/common/ByteInputStream.h"

#include <boost/bind.hpp>
#include <sstream>

FENNEL_BEGIN_CPPFILE("$Id$");

// TODO:  factor out the management of long log segments via a separate
// extensibility interface

LogicalTxnLog::LogicalTxnLog(
    SegmentAccessor const &logSegmentAccessorInit,
    PseudoUuid const &onlineUuid,
    SharedSegmentFactory pSegmentFactoryInit)
    : pSegmentFactory(pSegmentFactoryInit),
      logSegmentAccessor(logSegmentAccessorInit)
{
    // TODO: Support an option to skip CRC's for optimized non-durable logging.
    // Also support a paranoid option for recording CRC's for long logs.
    pOutputStream = CrcSegOutputStream::newCrcSegOutputStream(
        logSegmentAccessor,onlineUuid);

    // NOTE:  We only write one page at a time to the main log, and we always
    // need to wait after each page.  So request synchronous writes.
    pOutputStream->setWriteLatency(WRITE_EAGER_SYNC);
    
    nextTxnId = TxnId(1);
    pOutputStream->getSegPos(lastCheckpointMemento.logPosition);
    lastCheckpointMemento.nUncommittedTxns = 0;
    nCommittedBeforeLastCheckpoint = 0;

    groupCommitInterval = pSegmentFactory->getConfigMap().getIntParam(
        "groupCommitInterval");
}

SharedLogicalTxnLog LogicalTxnLog::newLogicalTxnLog(
    SegmentAccessor const &logSegmentAccessor,
    PseudoUuid const &onlineUuid,
    SharedSegmentFactory pSegmentFactory)
{
    return SharedLogicalTxnLog(
        new LogicalTxnLog(logSegmentAccessor,onlineUuid,pSegmentFactory));
}

LogicalTxnLog::~LogicalTxnLog()
{
    assert(uncommittedTxns.empty());
    assert(committedLongLogSegments.empty());
}

SharedLogicalTxn LogicalTxnLog::newLogicalTxn(
    SharedCacheAccessor pCacheAccessor)
{
    StrictMutexGuard mutexGuard(mutex);
    SharedLogicalTxn pTxn(
        new LogicalTxn(nextTxnId,shared_from_this(),pCacheAccessor));
    uncommittedTxns.push_back(pTxn);
    ++nextTxnId;
    return pTxn;
}

void LogicalTxnLog::removeTxn(SharedLogicalTxn pTxn)
{
    TxnListIter pFound = std::find(
        uncommittedTxns.begin(),
        uncommittedTxns.end(),
        pTxn);
    assert(pFound != uncommittedTxns.end());
    uncommittedTxns.erase(pFound);
}

void LogicalTxnLog::commitTxn(SharedLogicalTxn pTxn)
{
    LogicalTxnEventMemento memento;
    memento.event = LogicalTxnEventMemento::EVENT_COMMIT;
    memento.txnId = pTxn->txnId;
    memento.cbActionLast = pTxn->svpt.cbActionPrev;
    memento.nParticipants = pTxn->participants.size();
    SharedSegment pSegment = pTxn->pOutputStream->getSegment();
    if (pSegment) {
        assert(pTxn->pOutputStream.unique());
        pTxn->pOutputStream->hardPageBreak();
        pTxn->pOutputStream->getSegOutputStream()->getSegPos(
            memento.logPosition);
        pTxn->pOutputStream.reset();
        pSegment->checkpoint(CHECKPOINT_FLUSH_AND_UNMAP);
        committedLongLogSegments.push_back(pSegment);
    } else {
        CompoundId::setPageId(memento.logPosition.segByteId,NULL_PAGE_ID);
        CompoundId::setByteOffset(
            memento.logPosition.segByteId,
            pTxn->svpt.cbLogged);
        memento.logPosition.cbOffset = pTxn->svpt.cbLogged;
    }
    memento.longLog = pSegment ? true : false;
    StrictMutexGuard mutexGuard(mutex);
    pOutputStream->writeValue(memento);
    if (!pSegment) {
        SharedByteInputStream pInputStream =
            pTxn->pOutputStream->getInputStream();
        uint cbActual;
        PConstBuffer pBuffer = pInputStream->getReadPointer(1,&cbActual);
        pOutputStream->writeBytes(pBuffer,cbActual);
    }

    // REVIEW jvs 27-Dec-2005:  when groupCommitInterval is 0, maybe
    // we should bypass some of the overhead here?
    
    boost::xtime groupCommitExpiration;
    convertTimeout(groupCommitInterval,groupCommitExpiration);
    SegStreamPosition logPos;
    pOutputStream->getSegPos(logPos);
    PageId startPageId = CompoundId::getPageId(logPos.segByteId);
    for (;;) {
        bool timeout = !condition.timed_wait(mutexGuard,groupCommitExpiration);

        pOutputStream->getSegPos(logPos);
        PageId lastPageId = CompoundId::getPageId(logPos.segByteId);
        if (lastPageId != startPageId) {
            // someone else has flushed for us
            break;
        }

        if (timeout) {
            // timeout:  we're in charge of flushing
            
            // NOTE:  Since we're using synchronous writes, there's no need to
            // checkpoint (assuming the underlying device has been correctly<commit_msg>FENNEL:  for unit tests, revert default group commit interval to 30ms (workaround for problem in BTreeTxnTest exposed by elimination of group commit delay); optimize no-delay case<commit_after>/*
// $Id$
// Fennel is a library of data storage and processing components.
// Copyright (C) 2005-2005 The Eigenbase Project
// Copyright (C) 2005-2005 Disruptive Tech
// Copyright (C) 2005-2005 LucidEra, Inc.
// Portions Copyright (C) 1999-2005 <NAME>
//
// 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 approved by The Eigenbase Project.
//
// 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
*/

#include "fennel/common/CommonPreamble.h"
#include "fennel/txn/LogicalTxnLog.h"
#include "fennel/txn/LogicalTxn.h"
#include "fennel/txn/LogicalTxnStoredStructs.h"
#include "fennel/segment/CrcSegOutputStream.h"
#include "fennel/segment/SpillOutputStream.h"
#include "fennel/segment/SegmentFactory.h"
#include "fennel/common/ByteInputStream.h"

#include <boost/bind.hpp>
#include <sstream>

FENNEL_BEGIN_CPPFILE("$Id$");

// TODO:  factor out the management of long log segments via a separate
// extensibility interface

LogicalTxnLog::LogicalTxnLog(
    SegmentAccessor const &logSegmentAccessorInit,
    PseudoUuid const &onlineUuid,
    SharedSegmentFactory pSegmentFactoryInit)
    : pSegmentFactory(pSegmentFactoryInit),
      logSegmentAccessor(logSegmentAccessorInit)
{
    // TODO: Support an option to skip CRC's for optimized non-durable logging.
    // Also support a paranoid option for recording CRC's for long logs.
    pOutputStream = CrcSegOutputStream::newCrcSegOutputStream(
        logSegmentAccessor,onlineUuid);

    // NOTE:  We only write one page at a time to the main log, and we always
    // need to wait after each page.  So request synchronous writes.
    pOutputStream->setWriteLatency(WRITE_EAGER_SYNC);
    
    nextTxnId = TxnId(1);
    pOutputStream->getSegPos(lastCheckpointMemento.logPosition);
    lastCheckpointMemento.nUncommittedTxns = 0;
    nCommittedBeforeLastCheckpoint = 0;

    groupCommitInterval = pSegmentFactory->getConfigMap().getIntParam(
        "groupCommitInterval", 30);
}

SharedLogicalTxnLog LogicalTxnLog::newLogicalTxnLog(
    SegmentAccessor const &logSegmentAccessor,
    PseudoUuid const &onlineUuid,
    SharedSegmentFactory pSegmentFactory)
{
    return SharedLogicalTxnLog(
        new LogicalTxnLog(logSegmentAccessor,onlineUuid,pSegmentFactory));
}

LogicalTxnLog::~LogicalTxnLog()
{
    assert(uncommittedTxns.empty());
    assert(committedLongLogSegments.empty());
}

SharedLogicalTxn LogicalTxnLog::newLogicalTxn(
    SharedCacheAccessor pCacheAccessor)
{
    StrictMutexGuard mutexGuard(mutex);
    SharedLogicalTxn pTxn(
        new LogicalTxn(nextTxnId,shared_from_this(),pCacheAccessor));
    uncommittedTxns.push_back(pTxn);
    ++nextTxnId;
    return pTxn;
}

void LogicalTxnLog::removeTxn(SharedLogicalTxn pTxn)
{
    TxnListIter pFound = std::find(
        uncommittedTxns.begin(),
        uncommittedTxns.end(),
        pTxn);
    assert(pFound != uncommittedTxns.end());
    uncommittedTxns.erase(pFound);
}

void LogicalTxnLog::commitTxn(SharedLogicalTxn pTxn)
{
    LogicalTxnEventMemento memento;
    memento.event = LogicalTxnEventMemento::EVENT_COMMIT;
    memento.txnId = pTxn->txnId;
    memento.cbActionLast = pTxn->svpt.cbActionPrev;
    memento.nParticipants = pTxn->participants.size();
    SharedSegment pSegment = pTxn->pOutputStream->getSegment();
    if (pSegment) {
        assert(pTxn->pOutputStream.unique());
        pTxn->pOutputStream->hardPageBreak();
        pTxn->pOutputStream->getSegOutputStream()->getSegPos(
            memento.logPosition);
        pTxn->pOutputStream.reset();
        pSegment->checkpoint(CHECKPOINT_FLUSH_AND_UNMAP);
        committedLongLogSegments.push_back(pSegment);
    } else {
        CompoundId::setPageId(memento.logPosition.segByteId,NULL_PAGE_ID);
        CompoundId::setByteOffset(
            memento.logPosition.segByteId,
            pTxn->svpt.cbLogged);
        memento.logPosition.cbOffset = pTxn->svpt.cbLogged;
    }
    memento.longLog = pSegment ? true : false;
    StrictMutexGuard mutexGuard(mutex);
    pOutputStream->writeValue(memento);
    if (!pSegment) {
        SharedByteInputStream pInputStream =
            pTxn->pOutputStream->getInputStream();
        uint cbActual;
        PConstBuffer pBuffer = pInputStream->getReadPointer(1,&cbActual);
        pOutputStream->writeBytes(pBuffer,cbActual);
    }

    boost::xtime groupCommitExpiration;
    if (groupCommitInterval) {
        convertTimeout(groupCommitInterval,groupCommitExpiration);
    }
    SegStreamPosition logPos;
    pOutputStream->getSegPos(logPos);
    PageId startPageId = CompoundId::getPageId(logPos.segByteId);
    for (;;) {
        bool timeout = true;
        if (groupCommitInterval) {
            timeout = !condition.timed_wait(mutexGuard,groupCommitExpiration);

            pOutputStream->getSegPos(logPos);
            PageId lastPageId = CompoundId::getPageId(logPos.segByteId);
            if (lastPageId != startPageId) {
                // someone else has flushed for us
                break;
            }
        }

        if (timeout) {
            // timeout:  we're in charge of flushing
            
            // NOTE:  Since we're using synchronous writes, there's no need to
            // checkpoint (assuming the underlying device has been correctly
{'fraction_non_alphanumeric': 0.09374208259437547, 'fraction_numerical': 0.009543112912760747, 'mean_word_length': 3.640282131661442, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7030786', 'n_tokens_mistral': 3733, 'n_tokens_neox': 3571, 'n_words': 857}