Author: kpvdr Date: 2010-12-10 15:52:48 -0500 (Fri, 10 Dec 2010) New Revision: 4436 Added: store/trunk/cpp/lib/jrnl2/Journal.cpp store/trunk/cpp/lib/jrnl2/Journal.hpp store/trunk/cpp/lib/jrnl2/JournalState.cpp store/trunk/cpp/lib/jrnl2/JournalState.hpp store/trunk/cpp/lib/jrnl2/ScopedLock.hpp Removed: store/trunk/cpp/lib/jrnl2/jrnl.cpp store/trunk/cpp/lib/jrnl2/jrnl.hpp store/trunk/cpp/lib/jrnl2/jrnl_state.cpp store/trunk/cpp/lib/jrnl2/jrnl_state.hpp store/trunk/cpp/lib/jrnl2/slock.cpp store/trunk/cpp/lib/jrnl2/slock.hpp store/trunk/cpp/lib/jrnl2/smutex.cpp store/trunk/cpp/lib/jrnl2/smutex.hpp store/trunk/cpp/lib/jrnl2/txn_ctxt.hpp Modified: store/trunk/cpp/lib/jrnl2/DataToken.cpp store/trunk/cpp/lib/jrnl2/DataToken.hpp store/trunk/cpp/lib/jrnl2/JournalErrors.hpp store/trunk/cpp/lib/jrnl2/JournalException.hpp store/trunk/cpp/lib/jrnl2/JournalParameters.cpp store/trunk/cpp/lib/jrnl2/JournalParameters.hpp store/trunk/cpp/lib/jrnl2/Makefile.am store/trunk/cpp/perf/JournalInstance.cpp store/trunk/cpp/perf/JournalInstance.hpp store/trunk/cpp/perf/JournalPerformanceTest.cpp store/trunk/cpp/perf/m Log: Additional code tidy-up and refactoring in lib/jrnl2 Modified: store/trunk/cpp/lib/jrnl2/DataToken.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/DataToken.cpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/DataToken.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -40,7 +40,7 @@ recordId_t AtomicRecordIdCounter::_recordId; // static - smutex AtomicRecordIdCounter::_recordIdMutex; + ScopedMutex AtomicRecordIdCounter::_recordIdMutex; DataToken::DataToken() : _dataTokenState(), Modified: store/trunk/cpp/lib/jrnl2/DataToken.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/DataToken.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/DataToken.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -33,11 +33,10 @@ #define mrg_journal2_DataToken_hpp #include <string> -#include <sys/types.h> // u_int16_t, u_int32_t +#include <sys/types.h> // u_int64_t #include "DataTokenState.hpp" -#include "slock.hpp" -#include "smutex.hpp" +#include "ScopedLock.hpp" namespace mrg { @@ -49,9 +48,13 @@ class AtomicRecordIdCounter { static recordId_t _recordId; - static smutex _recordIdMutex; + static ScopedMutex _recordIdMutex; public: - inline static recordId_t s_getNextRecordId() { slock l(_recordIdMutex); return ++_recordId; } + inline static recordId_t s_getNextRecordId() + { // --- START OF CRITICAL SECTION --- + ScopedLock l(_recordIdMutex); + return ++_recordId; + } // --- START OF CRITICAL SECTION --- }; class DataToken Copied: store/trunk/cpp/lib/jrnl2/Journal.cpp (from rev 4435, store/trunk/cpp/lib/jrnl2/jrnl.cpp) =================================================================== --- store/trunk/cpp/lib/jrnl2/Journal.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/Journal.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -0,0 +1,188 @@ +/** + * \file Journal.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library 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 + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "Journal.hpp" + +// --- temp code --- +#include <iostream> +// --- end temp code --- + +namespace mrg +{ +namespace journal2 +{ + + std::string + g_ioResAsString(const ioRes /*res*/) + { + // TODO - provide implementation + return ".[g_ioResAsString]."; + } + + // static + u_int32_t Journal::_s_listSizeThreshold = 50; + + Journal::Journal(const std::string& jrnlId, + const std::string& jrnlDir, + const std::string& baseFileName) : + _jrnlId(jrnlId), + _jrnlDir(jrnlDir), + _baseFileName(baseFileName), + _jrnlParamsPtr(0), + _aioCallbackPtr(0) + // --- temp code --- + , _callBackDataTokenListSwitch(false) + // --- end temp code --- + {} + + void + Journal::initialize(const JournalParameters* jpPtr, AioCallback* const aiocbPtr) + { + _jrnlParamsPtr = jpPtr; + _aioCallbackPtr = aiocbPtr; + } + + ioRes + Journal::enqueue(const void* const /*msgPtr*/, const std::size_t /*msgSize*/, DataToken* const dtokPtr) + { + dtokPtr->getDataTokenState().setOpStateToEnqueue(); + // --- temp code --- + bool flushFlag; + { // --- START OF CRITICAL SECTION --- + ScopedLock l(_writeDataTokenListLock); + _writeDataTokenList.push_back(dtokPtr); + flushFlag = _writeDataTokenList.size() >= _s_listSizeThreshold; + if (flushFlag) flushNoLock(false); + } // --- END OF CRITICAL SECTION --- + if (flushFlag) processCompletedAioWriteEvents(0); + // --- end temp code --- + return 0; + } + + ioRes + Journal::dequeue(DataToken* const dtokPtr) + { + dtokPtr->getDataTokenState().setOpStateToDequeue(); + dtokPtr->setDequeueRecordId(dtokPtr->getRecordId()); + // --- temp code --- + bool flushFlag; + { // --- START OF CRITICAL SECTION --- + ScopedLock l(_writeDataTokenListLock); + _writeDataTokenList.push_back(dtokPtr); + flushFlag = _writeDataTokenList.size() >= _s_listSizeThreshold; + if (flushFlag) flushNoLock(false); + } // --- END OF CRITICAL SECTION --- + if (flushFlag) processCompletedAioWriteEvents(0); + // --- end temp code --- + return 0; + } + + ioRes + Journal::commit() + { + // TODO + return 0; + } + + ioRes + Journal::abort() + { + // TODO + return 0; + } + + u_int32_t + Journal::getWriteAioEventsRemaining() const + { + while (true) { // --- START OF CRITICAL SECTION --- + ScopedTryLock l1(_callBackDataTokenListLock); + ScopedTryLock l2(_writeDataTokenListLock); + if (l1.isLocked() && l2.isLocked()) { + return _callBackDataTokenList[0].size() + _callBackDataTokenList[1].size(); + } else { + //::usleep(10); + } + }; // --- END OF CRITICAL SECTION --- + } + + void + Journal::flush(const bool blockTillAioCompleteFlag) + { + // --- temp code --- + // --- START OF CRITICAL SECTION --- + ScopedTryLock l(_writeDataTokenListLock); + if (l.isLocked()) { + flushNoLock(blockTillAioCompleteFlag); + } + // --- END OF CRITICAL SECTION --- + // --- end temp code --- + } + + // protected + void + Journal::flushNoLock(const bool /*blockTillAioCompleteFlag*/) + { + // --- temp code --- + int i = _callBackDataTokenListSwitch ? 1 : 0; + while (_writeDataTokenList.size()) { + _callBackDataTokenList[i].push_back(_writeDataTokenList.back()); + _writeDataTokenList.pop_back(); + } + // --- end temp code --- + } + + void + Journal::processCompletedAioWriteEvents(timespec* const /*timeout*/) + { + // --- temp code --- + // --- START OF CRITICAL SECTION 1 --- + ScopedTryLock l1(_callBackDataTokenListLock); + if (l1.isLocked()) { + int i = _callBackDataTokenListSwitch ? 0 : 1; + if (_callBackDataTokenList[i].size() && _aioCallbackPtr) { + // NOTE: Callback made with _callBackDataTokenListLock isLocked + _aioCallbackPtr->writeAioCompleteCallback(_callBackDataTokenList[i]); + } + _callBackDataTokenList[i].clear(); + + // take both locks before allowing switch to change + { // --- START OF CRITICAL SECTION 2 --- + ScopedLock l2(_writeDataTokenListLock); + _callBackDataTokenListSwitch = !_callBackDataTokenListSwitch; + } // --- END OF CRITICAL SECTION 2 --- + } + // --- END OF CRITICAL SECTION 1 --- + // --- end temp code --- + } + +} // namespace journal2 +} // namespace mrg + Copied: store/trunk/cpp/lib/jrnl2/Journal.hpp (from rev 4435, store/trunk/cpp/lib/jrnl2/jrnl.hpp) =================================================================== --- store/trunk/cpp/lib/jrnl2/Journal.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/Journal.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -0,0 +1,112 @@ +/** + * \file Journal.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library 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 + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_Journal_hpp +#define mrg_journal2_Journal_hpp + +#include <string> +#include <sys/types.h> // u_int64_t, u_int32_t, etc. +#include <time.h> // timespec + +#include "AioCallback.hpp" +#include "DataToken.hpp" +#include "JournalState.hpp" +#include "JournalParameters.hpp" + +// --- temp code --- +#include "ScopedLock.hpp" // ScopedMutex +#include <vector> +// --- end temp code --- + +namespace mrg +{ +namespace journal2 +{ + + // TODO - decide if this is the right place to expose these codes and flags + typedef u_int64_t ioRes; // TODO - this needs to be expressed as flags + const ioRes RHM_IORES_ENQCAPTHRESH = 0x1; + const ioRes RHM_IORES_BUSY = 0x2; + std::string g_ioResAsString(const ioRes /*res*/); + + class Journal + { + protected: + std::string _jrnlId; + std::string _jrnlDir; + std::string _baseFileName; + JournalState _jrnlState; + const JournalParameters* _jrnlParamsPtr; + AioCallback* _aioCallbackPtr; + + // --- temp code --- + static u_int32_t _s_listSizeThreshold; + std::vector<DataToken*> _writeDataTokenList; + std::vector<DataToken*> _callBackDataTokenList[2]; + bool _callBackDataTokenListSwitch; + ScopedMutex _writeDataTokenListLock; + ScopedMutex _callBackDataTokenListLock; + // --- end temp code --- + + void flushNoLock(const bool blockTillAioCompleteFlag); + + public: + Journal(const std::string& jrnlId, const std::string& jrnlDir, const std::string& baseFileName); + + // get functions + inline std::string getJournalId() { return _jrnlId; } + inline std::string getJournalDir() { return _jrnlDir; } + inline std::string getBaseFileName() { return _baseFileName; } + inline const JournalState& getJournalState() { return _jrnlState; } + inline const JournalParameters* getJournalParameters() const { return _jrnlParamsPtr; } + + // msg ops + void initialize(const JournalParameters* jpPtr, AioCallback* const aiocbPtr); + ioRes enqueue(const void* const msgPtr, const std::size_t msgSize, DataToken* const dtokPtr); + ioRes dequeue(DataToken* const dtokPtr); + ioRes commit(); + ioRes abort(); + + // aio ops and status + // --- temp code --- + u_int32_t getWriteAioEventsRemaining() const; + // --- end of temp code --- + void flush(const bool blockTillAioCompleteFlag); + void processCompletedAioWriteEvents(timespec* const timeout); + }; + + +} // namespace journal2 +} // namespace mrg + + +#endif // mrg_journal2_Journal_hpp + Modified: store/trunk/cpp/lib/jrnl2/JournalErrors.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/JournalErrors.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/JournalErrors.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -33,7 +33,7 @@ #define mrg_journal2_JournalErrors_hpp #include <map> -#include <sys/types.h> // u_int16_t, u_int32_t +#include <sys/types.h> // u_int32_t namespace mrg { Modified: store/trunk/cpp/lib/jrnl2/JournalException.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/JournalException.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/JournalException.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -32,10 +32,10 @@ #ifndef mrg_journal2_JournalException_hpp #define mrg_journal2_JournalException_hpp -#include <cstring> +#include <cstring> // std::strerror #include <sstream> #include <string> -#include <sys/types.h> // u_int16_t, u_int32_t +#include <sys/types.h> // u_int32_t #include "JournalErrors.hpp" Modified: store/trunk/cpp/lib/jrnl2/JournalParameters.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/JournalParameters.cpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/JournalParameters.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -38,92 +38,91 @@ namespace journal2 { -// static declarations -std::string JournalParameters::_s_defaultJrnlDir = "/tmp/store"; -std::string JournalParameters::_s_defaultJrnlBaseFileName = "JournalData"; -u_int16_t JournalParameters::_s_defaultNumJrnlFiles = 8; -u_int32_t JournalParameters::_s_defaultJrnlFileSize_sblks = 3072; -bool JournalParameters::_s_defaultAutoExpand = false; -u_int16_t JournalParameters::_s_defaultAutoExpandMaxJrnlFiles = 0; -u_int16_t JournalParameters::_s_defaultWriteBuffNumPgs = 32; -u_int32_t JournalParameters::_s_defaultWriteBuffPgSize_sblks = 128; + // static declarations + std::string JournalParameters::_s_defaultJrnlDir = "/tmp/store"; + std::string JournalParameters::_s_defaultJrnlBaseFileName = "JournalData"; + u_int16_t JournalParameters::_s_defaultNumJrnlFiles = 8; + u_int32_t JournalParameters::_s_defaultJrnlFileSize_sblks = 3072; + bool JournalParameters::_s_defaultAutoExpand = false; + u_int16_t JournalParameters::_s_defaultAutoExpandMaxJrnlFiles = 0; + u_int16_t JournalParameters::_s_defaultWriteBuffNumPgs = 32; + u_int32_t JournalParameters::_s_defaultWriteBuffPgSize_sblks = 128; -JournalParameters::JournalParameters() : - _jrnlDir(_s_defaultJrnlDir), - _jrnlBaseFileName(_s_defaultJrnlBaseFileName), - _numJrnlFiles(_s_defaultNumJrnlFiles), - _jrnlFileSize_sblks(_s_defaultJrnlFileSize_sblks), - _autoExpand(_s_defaultAutoExpand), - _autoExpandMaxJrnlFiles(_s_defaultAutoExpandMaxJrnlFiles), - _writeBuffNumPgs(_s_defaultWriteBuffNumPgs), - _writeBuffPgSize_sblks(_s_defaultWriteBuffPgSize_sblks) -{} + JournalParameters::JournalParameters() : + _jrnlDir(_s_defaultJrnlDir), + _jrnlBaseFileName(_s_defaultJrnlBaseFileName), + _numJrnlFiles(_s_defaultNumJrnlFiles), + _jrnlFileSize_sblks(_s_defaultJrnlFileSize_sblks), + _autoExpand(_s_defaultAutoExpand), + _autoExpandMaxJrnlFiles(_s_defaultAutoExpandMaxJrnlFiles), + _writeBuffNumPgs(_s_defaultWriteBuffNumPgs), + _writeBuffPgSize_sblks(_s_defaultWriteBuffPgSize_sblks) + {} -JournalParameters::JournalParameters(const std::string& jrnlDir, - const std::string& jrnlBaseFileName, - const u_int16_t numJrnlFiles, - const bool autoExpand, - const u_int16_t autoExpandMaxJrnlFiles, - const u_int32_t jrnlFileSize_sblks, - const u_int16_t writeBuffNumPgs, - const u_int32_t writeBuffPgSize_sblks) : - _jrnlDir(jrnlDir), - _jrnlBaseFileName(jrnlBaseFileName), - _numJrnlFiles(numJrnlFiles), - _jrnlFileSize_sblks(jrnlFileSize_sblks), - _autoExpand(autoExpand), - _autoExpandMaxJrnlFiles(autoExpandMaxJrnlFiles), - _writeBuffNumPgs(writeBuffNumPgs), - _writeBuffPgSize_sblks(writeBuffPgSize_sblks) -{} + JournalParameters::JournalParameters(const std::string& jrnlDir, + const std::string& jrnlBaseFileName, + const u_int16_t numJrnlFiles, + const bool autoExpand, + const u_int16_t autoExpandMaxJrnlFiles, + const u_int32_t jrnlFileSize_sblks, + const u_int16_t writeBuffNumPgs, + const u_int32_t writeBuffPgSize_sblks) : + _jrnlDir(jrnlDir), + _jrnlBaseFileName(jrnlBaseFileName), + _numJrnlFiles(numJrnlFiles), + _jrnlFileSize_sblks(jrnlFileSize_sblks), + _autoExpand(autoExpand), + _autoExpandMaxJrnlFiles(autoExpandMaxJrnlFiles), + _writeBuffNumPgs(writeBuffNumPgs), + _writeBuffPgSize_sblks(writeBuffPgSize_sblks) + {} -JournalParameters::JournalParameters(const JournalParameters& sp) : - _jrnlDir(sp._jrnlDir), - _jrnlBaseFileName(sp._jrnlBaseFileName), - _numJrnlFiles(sp._numJrnlFiles), - _jrnlFileSize_sblks(sp._jrnlFileSize_sblks), - _autoExpand(sp._autoExpand), - _autoExpandMaxJrnlFiles(sp._autoExpandMaxJrnlFiles), - _writeBuffNumPgs(sp._writeBuffNumPgs), - _writeBuffPgSize_sblks(sp._writeBuffPgSize_sblks) + JournalParameters::JournalParameters(const JournalParameters& sp) : + _jrnlDir(sp._jrnlDir), + _jrnlBaseFileName(sp._jrnlBaseFileName), + _numJrnlFiles(sp._numJrnlFiles), + _jrnlFileSize_sblks(sp._jrnlFileSize_sblks), + _autoExpand(sp._autoExpand), + _autoExpandMaxJrnlFiles(sp._autoExpandMaxJrnlFiles), + _writeBuffNumPgs(sp._writeBuffNumPgs), + _writeBuffPgSize_sblks(sp._writeBuffPgSize_sblks) + {} -{} + void + JournalParameters::toStream(std::ostream& os) const + { + os << "Store Parameters:" << std::endl; + os << " jrnlDir = \"" << _jrnlDir << "\"" << std::endl; + os << " jrnlBaseFileName = \"" << _jrnlBaseFileName << "\"" << std::endl; + os << " numJrnlFiles = " << _numJrnlFiles << std::endl; + os << " jrnlFileSize_sblks = " << _jrnlFileSize_sblks << std::endl; + os << " autoExpand = " << _autoExpand << std::endl; + os << " autoExpandMaxJrnlFiles = " << _autoExpandMaxJrnlFiles << std::endl; + os << " writeBuffNumPgs = " << _writeBuffNumPgs << std::endl; + os << " writeBuffPgSize_sblks = " << _writeBuffPgSize_sblks << std::endl; + } -void -JournalParameters::toStream(std::ostream& os) const -{ - os << "Store Parameters:" << std::endl; - os << " jrnlDir = \"" << _jrnlDir << "\"" << std::endl; - os << " jrnlBaseFileName = \"" << _jrnlBaseFileName << "\"" << std::endl; - os << " numJrnlFiles = " << _numJrnlFiles << std::endl; - os << " jrnlFileSize_sblks = " << _jrnlFileSize_sblks << std::endl; - os << " autoExpand = " << _autoExpand << std::endl; - os << " autoExpandMaxJrnlFiles = " << _autoExpandMaxJrnlFiles << std::endl; - os << " writeBuffNumPgs = " << _writeBuffNumPgs << std::endl; - os << " writeBuffPgSize_sblks = " << _writeBuffPgSize_sblks << std::endl; -} + std::string + JournalParameters::toString() const + { + std::ostringstream oss; + toStream(oss); + return oss.str(); + } -std::string -JournalParameters::toString() const -{ - std::ostringstream oss; - toStream(oss); - return oss.str(); -} + std::ostream& + operator<<(std::ostream& os, const JournalParameters& jp) + { + jp.toStream(os); + return os; + } -std::ostream& -operator<<(std::ostream& os, const JournalParameters& jp) -{ - jp.toStream(os); - return os; -} + std::ostream& + operator<<(std::ostream& os, const JournalParameters* jpPtr) + { + jpPtr->toStream(os); + return os; + } -std::ostream& -operator<<(std::ostream& os, const JournalParameters* jpPtr) -{ - jpPtr->toStream(os); - return os; -} - } // namespace journal2 } // namespace mrg Modified: store/trunk/cpp/lib/jrnl2/JournalParameters.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/JournalParameters.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/JournalParameters.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -34,7 +34,7 @@ #include <iostream> #include <string> -#include <sys/types.h> +#include <sys/types.h> // u_int16_t, u_int32_t namespace mrg { Copied: store/trunk/cpp/lib/jrnl2/JournalState.cpp (from rev 4435, store/trunk/cpp/lib/jrnl2/jrnl_state.cpp) =================================================================== --- store/trunk/cpp/lib/jrnl2/JournalState.cpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/JournalState.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -0,0 +1,171 @@ +/** + * \file JournalState.cpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library 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 + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#include "JournalState.hpp" + +#include "JournalException.hpp" + +namespace mrg +{ +namespace journal2 +{ + + JournalState::JournalState() : + _journalState(JS_NONE) + {} + + JournalState::JournalState(const JournalState& s) : + _journalState(s._journalState) + {} + + JournalState::JournalState(const journalState_t s) : + _journalState(s) + {} + + const journalState_t& + JournalState::get() const + { + return _journalState; + } + + void + JournalState::set(const journalState_t s) + { + _journalState = s; + } + + void + JournalState::reset() + { + _journalState = JS_NONE; + } + + void + JournalState::setStateInitializing() + { + if (_journalState != JS_NONE) + THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, + s_getStateAsString(JS_INITIALIZING), + s_getStateAsString(_journalState), + "JournalState", + "setStateInitializing"); + _journalState = JS_INITIALIZING; + } + + void + JournalState::setStateRecoverPhase1() + { + if (_journalState != JS_NONE) + THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, + s_getStateAsString(JS_RECOVERING_PHASE_1), + s_getStateAsString(_journalState), + "JournalState", + "setStateRecoverPhase1"); + _journalState = JS_RECOVERING_PHASE_1; + } + + void + JournalState::setStateRecoverPhase2() + { + if (_journalState != JS_RECOVERING_PHASE_1) + THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, + s_getStateAsString(JS_RECOVERING_PHASE_2), + s_getStateAsString(_journalState), + "JournalState", + "setStateRecoverPhase2"); + _journalState = JS_RECOVERING_PHASE_2; + } + + void + JournalState::setStateRunning() + { + if (_journalState != JS_INITIALIZING ||_journalState != JS_RECOVERING_PHASE_2 ) + THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, + s_getStateAsString(JS_RUNNING), + s_getStateAsString(_journalState), + "JournalState", + "setStateRunning"); + _journalState = JS_RUNNING; + } + + void + JournalState::setStateStopping() + { + if (_journalState != JS_RUNNING) + THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, + s_getStateAsString(JS_STOPPING), + s_getStateAsString(_journalState), + "JournalState", + "setStateStopping"); + _journalState = JS_STOPPING; + } + + void + JournalState::setStateStopped() + { + if (_journalState != JS_STOPPING) + THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, + s_getStateAsString(JS_STOPPED), + s_getStateAsString(_journalState), + "JournalState", + "setStateStopped"); + _journalState = JS_STOPPED; + } + + //static + std::string + JournalState::s_getStateAsString(journalState_t s) + { + switch (s) { + case JS_NONE: + return "JS_NONE"; + case JS_RECOVERING_PHASE_1: + return "JS_RECOVERING_PHASE_1"; + case JS_RECOVERING_PHASE_2: + return "JS_RECOVERING_PHASE_2"; + case JS_INITIALIZING: + return "JS_INITIALIZING"; + case JS_RUNNING: + return "JS_RUNNING"; + case JS_STOPPING: + return "JS_STOPPING"; + case JS_STOPPED: + return "JS_STOPPED"; + default: + std::ostringstream oss; + oss << "<unknown state (" << "s" << ")>"; + return oss.str(); + } + } + +} // namespace journal2 +} // namespace mrg + Copied: store/trunk/cpp/lib/jrnl2/JournalState.hpp (from rev 4435, store/trunk/cpp/lib/jrnl2/jrnl_state.hpp) =================================================================== --- store/trunk/cpp/lib/jrnl2/JournalState.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/JournalState.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -0,0 +1,103 @@ +/** + * \file JournalState.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library 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 + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_JournalState_hpp +#define mrg_journal2_JournalState_hpp + +#include <string> + +namespace mrg +{ +namespace journal2 +{ + + /** + * Journal state machine + * + * JS_NONE * + * / \ + * / v + * | o JS_RECOVERING_PHASE_1 + * v | + * JS_INITIALIZING o v + * | o JS_RECOVERING_PHASE_2 + * \ / + * v v + * JS_RUNNING o + * | + * v + * JS_STOPPING o + * | + * v + * JS_STOPPED * + */ + typedef enum + { + JS_NONE = 0, + JS_RECOVERING_PHASE_1, + JS_RECOVERING_PHASE_2, + JS_INITIALIZING, + JS_RUNNING, + JS_STOPPING, + JS_STOPPED + } journalState_t; + + class JournalState + { + protected: + journalState_t _journalState; + public: + JournalState(); + JournalState(const JournalState& s); + JournalState(const journalState_t s); + + // Raw state get/set functions + const journalState_t& get() const; + void set(const journalState_t s); + + // State change functions + void reset(); + void setStateInitializing(); + void setStateRecoverPhase1(); + void setStateRecoverPhase2(); + void setStateRunning(); + void setStateStopping(); + void setStateStopped(); + + static std::string s_getStateAsString(journalState_t s); + }; + +} // namespace journal2 +} // namespace mrg + + +#endif // mrg_journal2_JournalState_hpp + Modified: store/trunk/cpp/lib/jrnl2/Makefile.am =================================================================== --- store/trunk/cpp/lib/jrnl2/Makefile.am 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/Makefile.am 2010-12-10 20:52:48 UTC (rev 4436) @@ -23,27 +23,23 @@ lib_LTLIBRARIES = libasyncjrnl2.la -libasyncjrnl2_la_SOURCES = \ - DataToken.cpp \ - DataTokenState.cpp \ - JournalDirectory.cpp \ - JournalErrors.cpp \ - JournalException.cpp \ - jrnl.cpp \ - jrnl_state.cpp \ - slock.cpp \ - smutex.cpp \ - JournalParameters.cpp \ - AioCallback.hpp \ - dtok.hpp \ - DataTokenState.hpp \ - JournalDirectory.hpp \ - JournalErrors.hpp \ - JournalException.hpp \ - jrnl.hpp \ - jrnl_state.hpp \ - slock.hpp \ - smutex.hpp \ - JournalParameters.hpp \ - txn_ctxt.hpp +libasyncjrnl2_la_SOURCES = \ + DataToken.cpp \ + DataTokenState.cpp \ + Journal.cpp \ + JournalDirectory.cpp \ + JournalErrors.cpp \ + JournalException.cpp \ + JournalParameters.cpp \ + JournalState.cpp \ + AioCallback.hpp \ + DataToken.hpp \ + DataTokenState.hpp \ + Journal.hpp \ + JournalDirectory.hpp \ + JournalErrors.hpp \ + JournalException.hpp \ + JournalParameters.hpp \ + JournalState.hpp \ + ScopedLock.hpp Copied: store/trunk/cpp/lib/jrnl2/ScopedLock.hpp (from rev 4435, store/trunk/cpp/lib/jrnl2/slock.hpp) =================================================================== --- store/trunk/cpp/lib/jrnl2/ScopedLock.hpp (rev 0) +++ store/trunk/cpp/lib/jrnl2/ScopedLock.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -0,0 +1,110 @@ +/** + * \file ScopedLock.hpp + * + * Qpid asynchronous store plugin library + * + * This file contains async journal code (v.2). + * + * \author Kim van der Riet + * + * Copyright (c) 2010 Red Hat, Inc. + * + * This file is part of the Qpid async store library msgstore.so. + * + * This library is free software; you can redistribute it and/or + * modify it under the terms of the GNU Lesser General Public + * License as published by the Free Software Foundation; either + * version 2.1 of the License, or (at your option) any later version. + * + * This library 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 + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this library; if not, write to the Free Software + * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 + * USA + * + * The GNU Lesser General Public License is available in the file COPYING. + */ + +#ifndef mrg_journal2_ScopedLock_hpp +#define mrg_journal2_ScopedLock_hpp + +#include <cerrno> // EBUSY +#include <pthread.h> + +#include "JournalException.hpp" + +namespace mrg +{ +namespace journal2 +{ + + // Ultra-simple scoped mutex class that allows a posix mutex to be initialized and destroyed with error checks + class ScopedMutex + { + protected: + mutable pthread_mutex_t _m; + public: + inline ScopedMutex() + { + PTHREAD_CHK(::pthread_mutex_init(&_m, 0), "::pthread_mutex_init", "smutex", "smutex"); + } + inline ~ScopedMutex() + { + PTHREAD_CHK(::pthread_mutex_destroy(&_m), "::pthread_mutex_destroy", "smutex", "~smutex"); + } + inline pthread_mutex_t* get() const { return &_m; } + }; + + // Scoped-mutex (sm) container, superclass for scoped lock classes + class ScopedMutexContainer + { + protected: + const ScopedMutex& _sm; + public: + ScopedMutexContainer(const ScopedMutex& sm) : _sm(sm) {} + virtual ~ScopedMutexContainer() {} + inline const ScopedMutex& get() const { return _sm; } + }; + + // Ultra-simple scoped lock class, auto-releases mutex when it goes out-of-scope + class ScopedLock : public ScopedMutexContainer + { + public: + inline ScopedLock(const ScopedMutex& sm) : ScopedMutexContainer(sm) + { + PTHREAD_CHK(::pthread_mutex_lock(_sm.get()), "::pthread_mutex_lock", "slock", "slock"); + } + virtual inline ~ScopedLock() + { + PTHREAD_CHK(::pthread_mutex_unlock(_sm.get()), "::pthread_mutex_unlock", "slock", "~slock"); + } + }; + + // Ultra-simple scoped try-lock class, auto-releases mutex when it goes out-of-scope + class ScopedTryLock : public ScopedMutexContainer + { + protected: + bool _lockedFlag; + public: + inline ScopedTryLock(const ScopedMutex& sm) : ScopedMutexContainer(sm), _lockedFlag(false) + { + int ret = ::pthread_mutex_trylock(_sm.get()); + _lockedFlag = (ret == 0); // check if lock obtained + if (!_lockedFlag && ret != EBUSY) PTHREAD_CHK(ret, "::pthread_mutex_trylock", "stlock", "stlock"); + } + virtual inline ~ScopedTryLock() + { + if (_lockedFlag) + PTHREAD_CHK(::pthread_mutex_unlock(_sm.get()), "::pthread_mutex_unlock", "stlock", "~stlock"); + } + inline bool isLocked() const { return _lockedFlag; } + }; + +} // namespace journal2 +} // namespace mrg + +#endif // ifndef mrg_journal2_ScopedLock_hpp Deleted: store/trunk/cpp/lib/jrnl2/jrnl.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jrnl.cpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/jrnl.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -1,184 +0,0 @@ -/** - * \file jrnl.cpp - * - * Qpid asynchronous store plugin library - * - * This file contains async journal code (v.2). - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#include "jrnl.hpp" - -// --- temp code --- -#include <iostream> -#include <vector> -// --- end temp code --- - -namespace mrg -{ -namespace journal2 -{ - -std::string -iores_str(const iores /*res*/) -{ - // TODO - provide implementation - return ".[iores_str]."; -} - -// static -u_int32_t jrnl::list_thresh = 50; - -jrnl::jrnl(const std::string& jid, - const std::string& jdir, - const std::string& base_filename) : - _jid(jid), - _jdir(jdir), - _base_filename(base_filename), - _store_params_ptr(0), - _cbp(0) - // --- temp code --- - , cb_dtok_list_switch(false) - // --- end temp code --- -{} - -void -jrnl::initialize(const JournalParameters* sp, AioCallback* const cbp) -{ - _store_params_ptr = sp; - _cbp = cbp; -} - -iores -jrnl::enqueue(const void* const /*msg_ptr*/, const std::size_t /*msg_size*/, DataToken* const dtokp) -{ - dtokp->getDataTokenState().setOpStateToEnqueue(); - // --- temp code --- - bool flush_flag; - { - slock l(wr_dtok_list_lock); - wr_dtok_list.push_back(dtokp); - flush_flag = wr_dtok_list.size() >= list_thresh; - if (flush_flag) flush_nl(false); - } - if (flush_flag) get_wr_events(0); - // --- end temp code --- - return 0; -} - -iores -jrnl::dequeue(DataToken* const dtokp) -{ - dtokp->getDataTokenState().setOpStateToDequeue(); - dtokp->setDequeueRecordId(dtokp->getRecordId()); - // --- temp code --- - bool flush_flag; - { - slock l(wr_dtok_list_lock); - wr_dtok_list.push_back(dtokp); - flush_flag = wr_dtok_list.size() >= list_thresh; - if (flush_flag) flush_nl(false); - } - if (flush_flag) get_wr_events(0); - // --- end temp code --- - return 0; -} - -iores -jrnl::commit() -{ - // TODO - return 0; -} - -iores -jrnl::abort() -{ - // TODO - return 0; -} - -u_int32_t -jrnl::get_wr_aio_evt_rem() const -{ - while (true) { - stlock l1(cb_dtok_list_lock); - stlock l2(wr_dtok_list_lock); - if (l1.locked() && l2.locked()) { - return cb_dtok_list[0].size() + cb_dtok_list[1].size(); - } else { - //::usleep(10); - } - }; -} - -void -jrnl::flush(const bool block_till_aio_cmpl) -{ - // --- temp code --- - stlock l(wr_dtok_list_lock); - if (l.locked()) { - flush_nl(block_till_aio_cmpl); - } - // --- end temp code --- -} - -// protected -void -jrnl::flush_nl(const bool /*block_till_aio_cmpl*/) -{ - // --- temp code --- - int i = cb_dtok_list_switch ? 1 : 0; - while (wr_dtok_list.size()) { - cb_dtok_list[i].push_back(wr_dtok_list.back()); - wr_dtok_list.pop_back(); - } - // --- end temp code --- -} - -void -jrnl::get_wr_events(timespec* const /*timeout*/) -{ - // --- temp code --- - stlock l1(cb_dtok_list_lock); - if (l1.locked()) { - int i = cb_dtok_list_switch ? 0 : 1; - if (cb_dtok_list[i].size() && _cbp) { - _cbp->writeAioCompleteCallback(cb_dtok_list[i]); - } - cb_dtok_list[i].clear(); - - // take both locks before allowing switch to change - { - slock l2(wr_dtok_list_lock); - cb_dtok_list_switch = !cb_dtok_list_switch; - } - } - // --- end temp code --- -} - -} // namespace journal2 -} // namespace mrg - Deleted: store/trunk/cpp/lib/jrnl2/jrnl.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jrnl.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/jrnl.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -1,112 +0,0 @@ -/** - * \file jrnl.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains async journal code (v.2). - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_journal2_jnrl_hpp -#define mrg_journal2_jnrl_hpp - -#include <string> -#include <sys/types.h> // u_int64_t, u_int32_t, etc. -#include <time.h> // timespec - -#include "AioCallback.hpp" -#include "DataToken.hpp" -#include "jrnl_state.hpp" -#include "JournalParameters.hpp" - -// --- temp code --- -#include "slock.hpp" -#include "smutex.hpp" -#include <vector> -// --- end temp code --- - -namespace mrg -{ -namespace journal2 -{ - - // TODO - decide if this is the right place to expose these codes and flags - typedef u_int64_t iores; // TODO - this needs to be expressed as flags - const iores RHM_IORES_ENQCAPTHRESH = 0x1; - const iores RHM_IORES_BUSY = 0x2; - std::string iores_str(const iores /*res*/); - - class jrnl - { - protected: - std::string _jid; - std::string _jdir; - std::string _base_filename; - jrnl_state _jrnl_state; - const JournalParameters* _store_params_ptr; - AioCallback* _cbp; - - // --- temp code --- - static u_int32_t list_thresh; - std::vector<DataToken*> wr_dtok_list; - std::vector<DataToken*> cb_dtok_list[2]; - bool cb_dtok_list_switch; - smutex wr_dtok_list_lock; - smutex cb_dtok_list_lock; - // --- end temp code --- - - void flush_nl(const bool block_till_aio_cmpl); - public: - jrnl(const std::string& jid, const std::string& jdir, const std::string& base_filename); - - // get functions - inline std::string get_id() { return _jid; } - inline std::string get_dir() { return _jdir; } - inline std::string get_base_filename() { return _base_filename; } - inline const jrnl_state& get_jrnl_state() { return _jrnl_state; } - inline const JournalParameters* get_store_params() const { return _store_params_ptr; } - - // msg ops - void initialize(const JournalParameters* sp, AioCallback* const cbp); - iores enqueue(const void* const msg_ptr, const std::size_t msg_size, DataToken* const dtokp); - iores dequeue(DataToken* const dtokp); - iores commit(); - iores abort(); - - // aio ops and status - // --- temp code --- - u_int32_t get_wr_aio_evt_rem() const; - // --- end of temp code --- - void flush(const bool block_till_aio_cmpl); - void get_wr_events(timespec* const timeout); - }; - - -} // namespace journal2 -} // namespace mrg - - -#endif // mrg_journal2_jnrl_hpp - Deleted: store/trunk/cpp/lib/jrnl2/jrnl_state.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jrnl_state.cpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/jrnl_state.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -1,180 +0,0 @@ -/** - * \file jrnl_state.cpp - * - * Qpid asynchronous store plugin library - * - * This file contains async journal code (v.2). - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#include "jrnl_state.hpp" - -#include "JournalException.hpp" - -namespace mrg -{ -namespace journal2 -{ - -jrnl_state::jrnl_state() : _jrnl_state(JS_NONE) -{} - -jrnl_state::jrnl_state(const jrnl_state& s) : _jrnl_state(s._jrnl_state) -{} - -jrnl_state::jrnl_state(const jrnl_state_t s) : _jrnl_state(s) -{} - -const jrnl_state_t& -jrnl_state::get() const -{ - return _jrnl_state; -} - -void -jrnl_state::set(const jrnl_state_t s) -{ - _jrnl_state = s; -} - -void -jrnl_state::reset() -{ - _jrnl_state = JS_NONE; -} - -void -jrnl_state::initialize() -{ - if (_jrnl_state != JS_NONE) - THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, - get_state_str(JS_INITIALIZING), - get_state_str(_jrnl_state), - "jrnl_state", - "initialize"); - _jrnl_state = JS_INITIALIZING; -} - -void -jrnl_state::initialize_compl() -{ - if (_jrnl_state != JS_INITIALIZING) - THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, - get_state_str(JS_RUNNING), - get_state_str(_jrnl_state), - "jrnl_state", - "initialize_compl"); - _jrnl_state = JS_RUNNING; -} - -void -jrnl_state::recover_phase_1() -{ - if (_jrnl_state != JS_NONE) - THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, - get_state_str(JS_RECOVERING_PHASE_1), - get_state_str(_jrnl_state), - "jrnl_state", - "recover_phase_1"); - _jrnl_state = JS_RECOVERING_PHASE_1; -} - -void -jrnl_state::recover_phase_2() -{ - if (_jrnl_state != JS_RECOVERING_PHASE_1) - THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, - get_state_str(JS_RECOVERING_PHASE_2), - get_state_str(_jrnl_state), - "jrnl_state", - "recover_phase_2"); - _jrnl_state = JS_RECOVERING_PHASE_2; -} - -void -jrnl_state::recover_compl() -{ - if (_jrnl_state != JS_RECOVERING_PHASE_2) - THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, - get_state_str(JS_RUNNING), - get_state_str(_jrnl_state), - "jrnl_state", - "recover_compl"); - _jrnl_state = JS_RUNNING; -} - -void -jrnl_state::stop() -{ - if (_jrnl_state != JS_RUNNING) - THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, - get_state_str(JS_STOPPING), - get_state_str(_jrnl_state), - "jrnl_state", - "stop"); - _jrnl_state = JS_STOPPING; -} - -void -jrnl_state::stop_compl() -{ - if (_jrnl_state != JS_STOPPING) - THROW_STATE_EXCEPTION(JournalErrors::JERR_BADJRNLSTATE, - get_state_str(JS_STOPPED), - get_state_str(_jrnl_state), - "jrnl_state", - "stop_compl"); - _jrnl_state = JS_STOPPED; -} - -//static -std::string -jrnl_state::get_state_str(jrnl_state_t s) -{ - switch (s) { - case JS_NONE: - return "JS_NONE"; - case JS_RECOVERING_PHASE_1: - return "JS_RECOVERING_PHASE_1"; - case JS_RECOVERING_PHASE_2: - return "JS_RECOVERING_PHASE_2"; - case JS_INITIALIZING: - return "JS_INITIALIZING"; - case JS_RUNNING: - return "JS_RUNNING"; - case JS_STOPPING: - return "JS_STOPPING"; - case JS_STOPPED: - return "JS_STOPPED"; - default: - std::ostringstream oss; - oss << "<unknown state (" << "s" << ")>"; - return oss.str(); - } -} - -} // namespace journal2 -} // namespace mrg - Deleted: store/trunk/cpp/lib/jrnl2/jrnl_state.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/jrnl_state.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/jrnl_state.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -1,104 +0,0 @@ -/** - * \file jrnl_state.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains async journal code (v.2). - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_journal2_jrnl_state_hpp -#define mrg_journal2_jrnl_state_hpp - -#include <string> - -namespace mrg -{ -namespace journal2 -{ - - /** - * Journal state machine - * - * JS_NONE * - * / \ - * / v - * | o JS_RECOVERING_PHASE_1 - * v | - * JS_INITIALIZING o v - * | o JS_RECOVERING_PHASE_2 - * \ / - * v v - * JS_RUNNING o - * | - * v - * JS_STOPPING o - * | - * v - * JS_STOPPED * - */ - typedef enum - { - JS_NONE = 0, - JS_RECOVERING_PHASE_1, - JS_RECOVERING_PHASE_2, - JS_INITIALIZING, - JS_RUNNING, - JS_STOPPING, - JS_STOPPED - } jrnl_state_t; - - class jrnl_state - { - protected: - jrnl_state_t _jrnl_state; - public: - jrnl_state(); - jrnl_state(const jrnl_state& s); - jrnl_state(const jrnl_state_t s); - - // Raw state get/set functions - const jrnl_state_t& get() const; - void set(const jrnl_state_t s); - - // State change functions - void reset(); - void initialize(); - void initialize_compl(); - void recover_phase_1(); - void recover_phase_2(); - void recover_compl(); - void stop(); - void stop_compl(); - - static std::string get_state_str(jrnl_state_t s); - }; - -} // namespace journal2 -} // namespace mrg - - -#endif // mrg_journal2_jrnl_state_hpp - Deleted: store/trunk/cpp/lib/jrnl2/slock.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/slock.cpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/slock.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -1,32 +0,0 @@ -/** - * \file slock.cpp - * - * Qpid asynchronous store plugin library - * - * This file contains async journal code (v.2). - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#include "slock.hpp" Deleted: store/trunk/cpp/lib/jrnl2/slock.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/slock.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/slock.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -1,94 +0,0 @@ -/** - * \file slock.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains async journal code (v.2). - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_journal2_slock_hpp -#define mrg_journal2_slock_hpp - -#include <cerrno> // EBUSY -#include <pthread.h> - -#include "JournalException.hpp" -#include "smutex.hpp" - -namespace mrg -{ -namespace journal2 -{ - - // Scoped-mutex (sm) container, superclass for scoped lock classes - class sm_cntnr - { - protected: - const smutex& _sm; - public: - sm_cntnr(const smutex& sm) : _sm(sm) {} - virtual ~sm_cntnr() {} - inline const smutex& get() const { return _sm; } - }; - - // Ultra-simple scoped lock class, auto-releases mutex when it goes out-of-scope - class slock : public sm_cntnr - { - public: - inline slock(const smutex& sm) : sm_cntnr(sm) - { - PTHREAD_CHK(::pthread_mutex_lock(_sm.get()), "::pthread_mutex_lock", "slock", "slock"); - } - virtual inline ~slock() - { - PTHREAD_CHK(::pthread_mutex_unlock(_sm.get()), "::pthread_mutex_unlock", "slock", "~slock"); - } - }; - - // Ultra-simple scoped try-lock class, auto-releases mutex when it goes out-of-scope - class stlock : public sm_cntnr - { - protected: - bool _locked; - public: - inline stlock(const smutex& sm) : sm_cntnr(sm), _locked(false) - { - int ret = ::pthread_mutex_trylock(_sm.get()); - _locked = (ret == 0); // check if lock obtained - if (!_locked && ret != EBUSY) PTHREAD_CHK(ret, "::pthread_mutex_trylock", "stlock", "stlock"); - } - virtual inline ~stlock() - { - if (_locked) - PTHREAD_CHK(::pthread_mutex_unlock(_sm.get()), "::pthread_mutex_unlock", "stlock", "~stlock"); - } - inline bool locked() const { return _locked; } - }; - -} // namespace journal2 -} // namespace mrg - -#endif // ifndef mrg_journal2_slock_hpp Deleted: store/trunk/cpp/lib/jrnl2/smutex.cpp =================================================================== --- store/trunk/cpp/lib/jrnl2/smutex.cpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/smutex.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -1,32 +0,0 @@ -/** - * \file smutex.cpp - * - * Qpid asynchronous store plugin library - * - * This file contains async journal code (v.2). - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#include "smutex.hpp" Deleted: store/trunk/cpp/lib/jrnl2/smutex.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/smutex.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/smutex.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -1,64 +0,0 @@ -/** - * \file smutex.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains async journal code (v.2). - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_journal2_smutex_hpp -#define mrg_journal2_smutex_hpp - -#include <pthread.h> - -#include "JournalException.hpp" - -namespace mrg -{ -namespace journal2 -{ - - // Ultra-simple scoped mutex class that allows a posix mutex to be initialized and destroyed with error checks - class smutex - { - protected: - mutable pthread_mutex_t _m; - public: - inline smutex() - { - PTHREAD_CHK(::pthread_mutex_init(&_m, 0), "::pthread_mutex_init", "smutex", "smutex"); - } - inline ~smutex() - { - PTHREAD_CHK(::pthread_mutex_destroy(&_m), "::pthread_mutex_destroy", "smutex", "~smutex"); - } - inline pthread_mutex_t* get() const { return &_m; } - }; - -} // namespace journal2 -} // namespace mrg - -#endif // ifndef mrg_journal2_smutex_hpp Deleted: store/trunk/cpp/lib/jrnl2/txn_ctxt.hpp =================================================================== --- store/trunk/cpp/lib/jrnl2/txn_ctxt.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/lib/jrnl2/txn_ctxt.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -1,54 +0,0 @@ -/** - * \file txn_ctxt.hpp - * - * Qpid asynchronous store plugin library - * - * This file contains async journal code (v.2). - * - * \author Kim van der Riet - * - * Copyright (c) 2010 Red Hat, Inc. - * - * This file is part of the Qpid async store library msgstore.so. - * - * This library is free software; you can redistribute it and/or - * modify it under the terms of the GNU Lesser General Public - * License as published by the Free Software Foundation; either - * version 2.1 of the License, or (at your option) any later version. - * - * This library 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 - * Lesser General Public License for more details. - * - * You should have received a copy of the GNU Lesser General Public - * License along with this library; if not, write to the Free Software - * Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 - * USA - * - * The GNU Lesser General Public License is available in the file COPYING. - */ - -#ifndef mrg_test_txn_ctxt_hpp -#define mrg_test_txn_ctxt_hpp - -#include <string> - -namespace mrg -{ -namespace journal2 -{ - - class txn_ctxt - { - public: - virtual ~txn_ctxt() {} - virtual std::string& tid() = 0; - virtual bool is_dist_txn() = 0; - }; - -} // namespace journal2 -} // namespace mrg - - -#endif // mrg_test_txn_ctxt_hpp Modified: store/trunk/cpp/perf/JournalInstance.cpp =================================================================== --- store/trunk/cpp/perf/JournalInstance.cpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/perf/JournalInstance.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -42,7 +42,7 @@ const uint32_t msgSize, const char* msgData, #ifdef JOURNAL2 - mrg::journal2::jrnl* const jrnlPtr) : + mrg::journal2::Journal* const jrnlPtr) : #else mrg::journal::jcntl* const jrnlPtr) : #endif @@ -70,7 +70,7 @@ while (i < _numMsgs) { #ifdef JOURNAL2 mrg::journal2::DataToken* dtokPtr = new mrg::journal2::DataToken(); - mrg::journal2::iores jrnlIoRes = _jrnlPtr->enqueue(_msgData, _msgSize, dtokPtr); + mrg::journal2::ioRes jrnlIoRes = _jrnlPtr->enqueue(_msgData, _msgSize, dtokPtr); #else mrg::journal::data_tok* dtokPtr = new mrg::journal::data_tok(); mrg::journal::iores jrnlIoRes = _jrnlPtr->enqueue_data_record(_msgData, _msgSize, _msgSize, dtokPtr); @@ -108,7 +108,7 @@ default: delete dtokPtr; #ifdef JOURNAL2 - std::cerr << "enqueue_data_record FAILED with " << mrg::journal2::iores_str(jrnlIoRes) << std::endl; + std::cerr << "enqueue_data_record FAILED with " << mrg::journal2::g_ioResAsString(jrnlIoRes) << std::endl; #else std::cerr << "enqueue_data_record FAILED with " << mrg::journal::iores_str(jrnlIoRes) << std::endl; #endif @@ -132,24 +132,28 @@ mrg::journal::data_tok* dtokPtr = 0; #endif while (!dtokPtr) { - bool getEventsFlag; // thread local + bool aioEventsFlag; // thread local { // --- START OF CRITICAL SECTION --- std::lock_guard<std::mutex> l(_unprocCallbackListMutex); - getEventsFlag = _unprocCallbackList.size() == 0; - if (!getEventsFlag) { + aioEventsFlag = _unprocCallbackList.size() == 0; + if (!aioEventsFlag) { dtokPtr = _unprocCallbackList.front(); _unprocCallbackList.pop(); } } // --- END OF CRITICAL SECTION --- - if (getEventsFlag) { + if (aioEventsFlag) { +#ifdef JOURNAL2 + _jrnlPtr->processCompletedAioWriteEvents(0); +#else _jrnlPtr->get_wr_events(0); +#endif ::usleep(1); } } bool done = false; while (!done) { #ifdef JOURNAL2 - mrg::journal2::iores jrnlIoRes = _jrnlPtr->dequeue(dtokPtr); + mrg::journal2::ioRes jrnlIoRes = _jrnlPtr->dequeue(dtokPtr); #else mrg::journal::iores jrnlIoRes = _jrnlPtr->dequeue_data_record(dtokPtr); #endif @@ -171,7 +175,7 @@ break; default: #ifdef JOURNAL2 - std::cerr << "dequeue_data_record FAILED with " << mrg::journal2::iores_str(jrnlIoRes) << ": " + std::cerr << "dequeue_data_record FAILED with " << mrg::journal2::g_ioResAsString(jrnlIoRes) << ": " << dtokPtr->statusStr() << std::endl; #else std::cerr << "dequeue_data_record FAILED with " << mrg::journal::iores_str(jrnlIoRes) << ": " @@ -181,7 +185,11 @@ done = true; } } +#ifdef JOURNAL2 + _jrnlPtr->processCompletedAioWriteEvents(0); +#else _jrnlPtr->get_wr_events(0); +#endif } _jrnlPtr->flush(true); } Modified: store/trunk/cpp/perf/JournalInstance.hpp =================================================================== --- store/trunk/cpp/perf/JournalInstance.hpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/perf/JournalInstance.hpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -39,7 +39,7 @@ #ifdef JOURNAL2 #include "jrnl2/AioCallback.hpp" #include "jrnl2/DataToken.hpp" -#include "jrnl2/jrnl.hpp" +#include "jrnl2/Journal.hpp" #else #include "jrnl/aio_callback.hpp" #include "jrnl/data_tok.hpp" @@ -68,7 +68,7 @@ const uint32_t _msgSize; ///< Size of each message (in bytes) const char* _msgData; ///< Pointer to message content to be used for each message. #ifdef JOURNAL2 - mrg::journal2::jrnl* const _jrnlPtr; ///< Journal instance pointer + mrg::journal2::Journal* const _jrnlPtr; ///< Journal instance pointer std::queue<mrg::journal2::DataToken*> _unprocCallbackList; ///< Queue of unprocessed callbacks to be dequeued #else mrg::journal::jcntl* const _jrnlPtr; ///< Journal instance pointer @@ -114,7 +114,7 @@ JournalInstance(const uint32_t numMsgs, const uint32_t msgSize, const char* msgData, - mrg::journal2::jrnl* const jrnlPtr); + mrg::journal2::Journal* const jrnlPtr); #else JournalInstance(const uint32_t numMsgs, const uint32_t msgSize, Modified: store/trunk/cpp/perf/JournalPerformanceTest.cpp =================================================================== --- store/trunk/cpp/perf/JournalPerformanceTest.cpp 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/perf/JournalPerformanceTest.cpp 2010-12-10 20:52:48 UTC (rev 4436) @@ -81,7 +81,7 @@ mrg::journal2::JournalDirectory::s_delete(_jrnlParams._jrnlDir); } mrg::journal2::JournalDirectory::s_create(_jrnlParams._jrnlDir); - mrg::journal2::jrnl* jp; + mrg::journal2::Journal* jp; #else if (mrg::journal::jdir::exists(_jrnlParams._jrnlDir)) { mrg::journal::jdir::delete_dir(_jrnlParams._jrnlDir); @@ -96,7 +96,7 @@ std::ostringstream jdir; jdir << _jrnlParams._jrnlDir << "/" << jname.str(); #ifdef JOURNAL2 - jp = new mrg::journal2::jrnl(jname.str(), jdir.str(), _jrnlParams._jrnlBaseFileName); + jp = new mrg::journal2::Journal(jname.str(), jdir.str(), _jrnlParams._jrnlBaseFileName); #else jp = new mrg::journal::jcntl(jname.str(), jdir.str(), _jrnlParams._jrnlBaseFileName); #endif Modified: store/trunk/cpp/perf/m =================================================================== --- store/trunk/cpp/perf/m 2010-12-09 14:33:48 UTC (rev 4435) +++ store/trunk/cpp/perf/m 2010-12-10 20:52:48 UTC (rev 4436) @@ -5,7 +5,7 @@ # The variable JOURNAL2, if defined, will link with the new journal2 namespace journal. Otherwise the old journal # namespace will be used. -JOURNAL2=1 +#JOURNAL2=1 # Optimization options #OPT="-O0 -ggdb" @@ -58,16 +58,14 @@ DEFINES=-DJOURNAL2 -JRNL_FILES="../lib/jrnl2/DataToken.cpp \ +JRNL_FILES="../lib/jrnl2/DataTokenState.cpp \ + ../lib/jrnl2/DataToken.cpp \ + ../lib/jrnl2/Journal.cpp \ ../lib/jrnl2/JournalDirectory.cpp \ - ../lib/jrnl2/JournalException.cpp \ - ../lib/jrnl2/jrnl_state.cpp \ - ../lib/jrnl2/smutex.cpp \ - ../lib/jrnl2/DataTokenState.cpp \ ../lib/jrnl2/JournalErrors.cpp \ - ../lib/jrnl2/jrnl.cpp \ - ../lib/jrnl2/slock.cpp \ - ../lib/jrnl2/JournalParameters.cpp" + ../lib/jrnl2/JournalException.cpp \ + ../lib/jrnl2/JournalParameters.cpp \ + ../lib/jrnl2/JournalState.cpp" WARN="${WARN_COMMON} ${WARN_OTHER} -Wshadow -Wunsafe-loop-optimizations" fi