You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kp...@apache.org on 2012/05/09 16:19:33 UTC
svn commit: r1336220 [2/6] - in /qpid/branches/asyncstore/cpp/src: ./
qpid/asyncStore/ qpid/asyncStore/jrnl2/ qpid/broker/ tests/
tests/storePerfTools/ tests/storePerfTools/asyncPerf/
tests/storePerfTools/common/ tests/storePerfTools/jrnlPerf/
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AsyncJournal.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AsyncJournal.cpp?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AsyncJournal.cpp (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AsyncJournal.cpp Wed May 9 14:19:28 2012
@@ -0,0 +1,231 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file AsyncJournal.cpp
+ */
+
+#include "AsyncJournal.h"
+
+#include "AioCallback.h"
+#include "DataToken.h"
+
+// --- temp code ---
+#include <fstream>
+#include <iostream>
+// --- end temp code ---
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+std::string
+g_ioResAsString(const jrnlOpRes /*res*/)
+{
+ /// \todo TODO - provide implementation
+ return ".[g_ioResAsString].";
+}
+
+AsyncJournal::AsyncJournal(const std::string& jrnlId,
+ const std::string& jrnlDir,
+ const std::string& baseFileName) :
+ m_jrnlId(jrnlId),
+ m_jrnlDir(jrnlDir),
+ m_baseFileName(baseFileName),
+ m_jrnlParamsPtr(0),
+ m_aioCallbackPtr(0)
+{
+ m_writeDataTokens.reserve(s_listSizeThreshold);
+ m_callbackDataTokens.reserve(s_listSizeThreshold);
+}
+
+std::string
+AsyncJournal::getId() const
+{
+ return m_jrnlId;
+}
+
+JournalDirectory
+AsyncJournal::getJournalDir() const
+{
+ return m_jrnlDir;
+}
+
+std::string
+AsyncJournal::getJournalDirName() const
+{
+ return m_jrnlDir.getFqName();
+}
+
+std::string
+AsyncJournal::getBaseFileName() const
+{
+ return m_baseFileName;
+}
+
+const JournalRunState&
+AsyncJournal::getState() const
+{
+ return m_jrnlState;
+}
+
+const JournalParameters*
+AsyncJournal::getParameters() const
+{
+ return m_jrnlParamsPtr;
+}
+
+void
+AsyncJournal::initialize(const JournalParameters* jpPtr,
+ AioCallback* const aiocbPtr)
+{
+ m_jrnlParamsPtr = jpPtr;
+ m_aioCallbackPtr = aiocbPtr;
+ // --- temp code ---
+ m_jrnlDir.create();
+ /// --- end temp code ---
+}
+
+jrnlOpRes
+AsyncJournal::enqueue(DataToken* dtokPtr,
+ const void* /*dataPtr*/,
+ const std::size_t /*dataLen*/,
+ const void* /*tidPtr*/,
+ const std::size_t /*tidLen*/,
+ const bool /*transientFlag*/)
+{
+ dtokPtr->getDataOpState().enqueue();
+ // --- temp code ---
+ { // --- START OF CRITICAL SECTION ---
+ ScopedLock l(m_writeDataTokensLock);
+ m_writeDataTokens.push_back(dtokPtr);
+ if (m_writeDataTokens.size() >= s_listSizeThreshold) {
+ flushNoLock();
+ }
+ } // --- END OF CRITICAL SECTION ---
+ //processCompletedAioWriteEvents(0);
+ // --- end temp code ---
+ return 0;
+}
+
+jrnlOpRes
+AsyncJournal::dequeue(DataToken* const dtokPtr,
+ const void* /*tidPtr*/,
+ const std::size_t /*tidLen*/)
+{
+ dtokPtr->getDataOpState().dequeue();
+ dtokPtr->setDequeueRecordId(dtokPtr->getRecordId());
+ // --- temp code ---
+ { // --- START OF CRITICAL SECTION ---
+ ScopedLock l(m_writeDataTokensLock);
+ m_writeDataTokens.push_back(dtokPtr);
+ if (m_writeDataTokens.size() >= s_listSizeThreshold) {
+ flushNoLock();
+ }
+ } // --- END OF CRITICAL SECTION ---
+ //processCompletedAioWriteEvents(0);
+ // --- end temp code ---
+ return 0;
+}
+
+jrnlOpRes
+AsyncJournal::commit()
+{
+ /// \todo TODO - provide implementation
+ return 0;
+}
+
+jrnlOpRes
+AsyncJournal::abort()
+{
+ /// \todo TODO - provide implementation
+ return 0;
+}
+
+jrnlOpRes
+AsyncJournal::flush()
+{
+ // --- temp code ---
+ // --- START OF CRITICAL SECTION ---
+ ScopedTryLock l(m_writeDataTokensLock);
+ if (l.isLocked()) {
+ return flushNoLock();
+ }
+ return 0;
+ // --- END OF CRITICAL SECTION ---
+ // --- end temp code ---
+}
+
+// protected
+jrnlOpRes
+AsyncJournal::flushNoLock()
+{
+ // --- temp code ---
+ // Normally the page would be written to disk using libaio here (still to do).
+ uint32_t cnt = 0UL;
+ ScopedLock l(m_callbackDataTokensLock);
+ while (!m_writeDataTokens.empty()) {
+ m_callbackDataTokens.push_back(m_writeDataTokens.back());
+ m_writeDataTokens.pop_back();
+ ++cnt;
+ }
+ return 0;
+ // --- end temp code ---
+}
+
+jrnlOpRes
+AsyncJournal::sync(const double timeout)
+{
+ // --- temp code ---
+ // --- START OF CRITICAL SECTION ---
+ ScopedTryLock l(m_writeDataTokensLock);
+ if (l.isLocked()) {
+ return syncNoLock(timeout);
+ }
+ return 0;
+ // --- END OF CRITICAL SECTION ---
+ // --- end temp code ---
+}
+
+// protected
+jrnlOpRes
+AsyncJournal::syncNoLock(const double /*timeout*/)
+{
+ // --- temp code ---
+ if (m_callbackDataTokens.size()) {
+ processCompletedAioWriteEvents();
+ }
+ return 0;
+ // --- end temp code ---
+}
+
+void
+AsyncJournal::processCompletedAioWriteEvents(const double /*timeout*/)
+{
+ // --- temp code ---
+ // --- START OF CRITICAL SECTION 1 ---
+ ScopedLock l1(m_callbackDataTokensLock);
+ m_aioCallbackPtr->writeAioCompleteCallback(m_callbackDataTokens);
+ // --- END OF CRITICAL SECTION 1 ---
+ // --- end temp code ---
+}
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AsyncJournal.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AsyncJournal.h?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AsyncJournal.h (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AsyncJournal.h Wed May 9 14:19:28 2012
@@ -0,0 +1,258 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file AsyncJournal.h
+ */
+
+#ifndef qpid_asyncStore_jrnl2_AsyncJournal_h_
+#define qpid_asyncStore_jrnl2_AsyncJournal_h_
+
+#include "JournalDirectory.h"
+#include "JournalRunState.h"
+#include "ScopedLock.h"
+
+#include <string>
+#include <stdint.h> // uint64_t, uint32_t, etc.
+
+// --- temp code ---
+#include <vector>
+// --- end temp code ---
+
+namespace qpid { ///< Namespace for top-level qpid domain
+namespace asyncStore { ///< Namespace for AsyncStore code
+namespace jrnl2 { ///< Namespace for AsyncStore journal v.2 code.
+
+class AioCallback;
+class DataToken;
+class JournalParameters;
+
+/**
+ * \brief Type to return results from journal operations.
+ * \todo TODO - decide if this is the right place to expose these codes and flags. Also express ioRes as flags.
+ */
+typedef uint64_t jrnlOpRes;
+
+const jrnlOpRes RHM_IORES_ENQCAPTHRESH = 0x1; ///< Error flag indicating an enqueue capacity threshold was reached.
+const jrnlOpRes RHM_IORES_BUSY = 0x2; ///< Error flag indicating that a call could not be completed because the Store is busy.
+
+/**
+ * \brief Global function to convert a return code into a textual representation.
+ */
+std::string g_ioResAsString(const jrnlOpRes /*res*/);
+
+/**
+ * \brief Top level of a single journal instance, which is usually deployed on a per-queue basis.
+ *
+ * Each journal has its own set of journal files and when recovered will result in restored data (messages) being
+ * sent to the queue to which this instance is connected.
+ */
+class AsyncJournal
+{
+public:
+ /**
+ * \brief Constructor, which creates a single instance of a journal.
+ *
+ * \param jrnlId Journal identifier (JID), typically the name of the queue with which this journal
+ * is associated.
+ * \param jrnlDir Absolute path to the directory in which the journal will be placed. If the path does not exist,
+ * it will be created.
+ * \param baseFileName The base name of all files in the journal directory.
+ */
+ AsyncJournal(const std::string& jrnlId,
+ const std::string& jrnlDir,
+ const std::string& baseFileName);
+
+ // Get functions
+
+ /**
+ * \brief Get the journal identifier (JID).
+ *
+ * \returns Journal identifier (JID) of this journal instance.
+ */
+ std::string getId() const;
+
+ /**
+ * \brief Get the URI or directory where this journal is deployed (writes its files).
+ *
+ * \returns JournalDirectory instance controlling where this journal instance is deployed.
+ */
+ JournalDirectory getJournalDir() const;
+
+ /**
+ * \brief Get the URI or directory where this journal is deployed (writes its files).
+ *
+ * \returns URI or directory (as a std::string) where this journal instance is deployed.
+ */
+ std::string getJournalDirName() const;
+
+ /**
+ * \brief Get the base file name used for all journal and metadata files associated with this journal instance.
+ *
+ * \returns String containing the base file name used for all files associated with this journal instance.
+ */
+ std::string getBaseFileName() const;
+
+ /**
+ * \brief Get the journal state object, which can be queried for the journal state.
+ *
+ * \returns \c const reference to the JournalSate instance associated with this journal instance.
+ */
+ const JournalRunState& getState() const;
+
+ /**
+ * \brief Return the Journal parameter object which contains the journal options and settings. It may be
+ * queried directly to obtain the options and settings.
+ *
+ * \returns Pointer to the JournalParameters instance associated with this journal instance.
+ */
+ const JournalParameters* getParameters() const;
+
+ // Data ops
+
+ /**
+ * \brief Initialize the journal and its files, making it ready for use.
+ *
+ * \param jpPtr Pointer to an instance of JournalParameters containing the journal options and settings.
+ * \param aiocbPtr Pointer to an instance of AioCallback, whcih sets the broker AIO callback functions.
+ *
+ * \todo TODO: Make this call async for large/slow ops
+ */
+ void initialize(const JournalParameters* jpPtr,
+ AioCallback* const aiocbPtr);
+
+ /**
+ * \brief Enqueue data of size dataLen and pointed to by dataPtr. If transactional, then tidPtr points to
+ * the transaction id and tidLen indicates its size. The DataToken instance must be kept in the service of this
+ * data record until it has been fully dequeued.
+ *
+ * \param dtokPtr Pointer to the DataToken object of a previouisly enqueued data record.
+ * \param dataPtr Pointer to data to be stored (enqueued). If \b NULL, when parameter \a dataLen > 0, then this
+ * is assumed to be an externally stored data record.
+ * \param dataLen Length of the data pointed to
+ * \param tidPtr Pointer to the transaction ID for this operation. If \b NULL together with \a tidLen, then
+ * the operation is non-transactional.
+ * \param tidLen Size of the transaction ID pointed to in parameter \a tidPtr. Must be 0 when \a tidPtr is
+ * \b NULL to make non-transactional.
+ * \param transientFlag If \b true, sets a flag indicating that on recover, this record is to be ignored and
+ * treated as if transient (rather than durable).
+ *
+ * \returns Return code for this operation. A zero value (0x0) indicates success, a non-zero value indicates
+ * an error has occurred or an issue is present.
+ */
+ jrnlOpRes enqueue(DataToken* dtokPtr,
+ const void* dataPtr, // if null and dataLen > 0, extern assumed
+ const std::size_t dataLen,
+ const void* tidPtr, // if null and tidLen == 0, non transactional
+ const std::size_t tidLen,
+ const bool transientFlag);
+
+ /**
+ * \brief Dequeue the data record previously enqueued using the DataToken object dtokPtr.
+ *
+ * \param dtokPtr Pointer to the DataToken object of a previouisly enqueued data record.
+ * \param tidPtr Pointer to the transaction ID for this operation. If \b NULL together with \a tidLen, then
+ * the operation is non-transactional.
+ * \param tidLen Size of the transaction ID pointed to in parameter \a tidPtr. Must be 0 when \a tidPtr is
+ * \b NULL to make non-transactional.
+ * \returns Return code for this operation. A zero value (0x0) indicates success, a non-zero value indicates
+ * an error has occurred or an issue is present.
+ */
+ jrnlOpRes dequeue(DataToken* const dtokPtr,
+ const void* tidPtr, // if null and tidLen == 0, non transactional
+ const std::size_t tidLen);
+
+ /**
+ * \brief Commit the transaction Id (XID) used for previoius enqueue(s) and/or dequeue(s).
+ *
+ * \return Return code for this operation. A zero value (0x0) indicates success, a non-zero value indicates
+ * an error has occurred or an issue is present.
+ *
+ * \todo TODO: Create and add an XID type as a parameter to this call.
+ */
+ jrnlOpRes commit();
+
+ /**
+ * \brief Abort (roll back) the transaction Id (XID) used for previoius enqueue(s) and/or dequeue(s).
+ *
+ * \return Return code for this operation. A zero value (0x0) indicates success, a non-zero value indicates
+ * an error has occurred or an issue is present.
+ *
+ * \todo TODO: Create and add an XID type as a parameter to this call.
+ */
+ jrnlOpRes abort();
+
+ // AIO ops and status
+
+ /**
+ * \brief Flush all unwritten buffered records for this journal.
+ */
+ jrnlOpRes flush();
+
+ /**
+ * \brief Wait until all AIOs outstanding at the last flush() have returned for this journal.
+ *
+ * It is assumed that flush() will have been previously called. This call by definition blocks until \a timeout
+ * has elapsed or all AIO operations outstanding at the last flush() call have returned. A zero timeout implies
+ * an indefinite block.
+ */
+ jrnlOpRes sync(const double timeout = 0.0);
+
+ /**
+ * \brief Search for and process completed AIO write events.
+ *
+ * \param timeout Maximum time to wait for completion, otherwise return without performing any work.
+ *
+ * \todo TODO: This may become obsolete if epoll is used instead.
+ * \todo TODO: Should this return the number of completed AIO events?
+ */
+ void processCompletedAioWriteEvents(const double timeout = 0.0);
+
+protected:
+ std::string m_jrnlId; ///< Identifier for this journal instance (JID), typically queue name.
+ JournalDirectory m_jrnlDir; ///< Directory in which this journal is deployed.
+ std::string m_baseFileName; ///< Base file name used for all journal files belonging to this instance.
+ JournalRunState m_jrnlState; ///< Journal state manager, controls the state of this journal.
+ const JournalParameters* m_jrnlParamsPtr; ///< Journal options and parameters associated with this journal.
+ AioCallback* m_aioCallbackPtr; ///< Pointers to the broker's callback functions for AIO completion callbacks.
+
+ // --- temp code ---
+ static const uint32_t s_listSizeThreshold = 250; ///< [TEMP CODE] Number of data records at which a flush will occur.
+ std::vector<DataToken*> m_writeDataTokens; ///< [TEMP CODE] List of data tokens held before a flush.
+ std::vector<DataToken*> m_callbackDataTokens; ///< [TEMP CODE] List of data tokens ready for callbacks.
+ ScopedMutex m_writeDataTokensLock; ///< [TEMP CODE] Lock to protect the write token list.
+ ScopedMutex m_callbackDataTokensLock; ///< [TEMP CODE] Lock to protect the callback token list.
+ // --- end temp code ---
+
+ /**
+ * \brief Internal-use flush call which operates without taking a lock.
+ */
+ jrnlOpRes flushNoLock();
+
+ /**
+ * \brief Internal-use sync call which operates without taking a lock.
+ */
+ jrnlOpRes syncNoLock(const double timeout);
+
+};
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+#endif // qpid_asyncStore_jrnl2_AsyncJournal_h_
+
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AtomicCounter.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AtomicCounter.h?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AtomicCounter.h (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/AtomicCounter.h Wed May 9 14:19:28 2012
@@ -0,0 +1,77 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file AtomicCounter.h
+ */
+
+#ifndef qpid_asyncStore_jrnl2_AtomicCounter_h_
+#define qpid_asyncStore_jrnl2_AtomicCounter_h_
+
+#include "ScopedLock.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+/**
+ * \brief Template integral atomic counter class which provides a next() operation to increment then return the
+ * updated count of integral type T. This operation is performed under a thread lock.
+ */
+template <class T> class AtomicCounter
+{
+public:
+ /**
+ * \brief Constructor with an option to set an inital value for the counter.
+ */
+ AtomicCounter(T initialValue = T(0)) :
+ m_cnt(initialValue)
+ {}
+
+ /**
+ * \brief Destructor
+ */
+ virtual ~AtomicCounter()
+ {}
+
+ /**
+ * \brief Increment, then return new value of the internal counter. This is thread-safe and takes a lock
+ * in order to perform the increment.
+ *
+ * \note This is a non-zero counter. By default, the counter starts with the value 0, and consequently the
+ * first call to next() will return 1. Upon overflow, the counter will be incremented twice so as to avoid
+ * returning the value 0.
+ */
+ virtual T next()
+ {
+ // --- START OF CRITICAL SECTION ---
+ ScopedLock l(m_mutex);
+ while (!++m_cnt) ; // Cannot return 0x0 if m_cnt should overflow
+ return m_cnt;
+ } // --- END OF CRITICAL SECTION ---
+
+protected:
+ T m_cnt; ///< Internal count value
+ ScopedMutex m_mutex; ///< Internal lock used to increment the counter
+
+};
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+#endif // qpid_asyncStore_jrnl2_AtomicCounter_h_
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/Configuration.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/Configuration.h?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/Configuration.h (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/Configuration.h Wed May 9 14:19:28 2012
@@ -0,0 +1,106 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file Configuration.h
+ */
+
+#ifndef qpid_asyncStore_jrnl2_Configuration_h_
+#define qpid_asyncStore_jrnl2_Configuration_h_
+
+#include <stdint.h> // uint8_t
+
+#if defined(__i386__) /* little endian, 32 bits */
+ #define JRNL_LITTLE_ENDIAN
+// #define JRNL_32_BIT
+#elif defined(__PPC__) || defined(__s390__) /* big endian, 32 bits */
+ #define JRNL_BIG_ENDIAN
+// #define JRNL_32_BIT
+#elif defined(__ia64__) || defined(__x86_64__) || defined(__alpha__) /* little endian, 64 bits */
+ #define JRNL_LITTLE_ENDIAN
+// #define JRNL_64_BIT
+#elif defined(__powerpc64__) || defined(__s390x__) /* big endian, 64 bits */
+ #define JRNL_BIG_ENDIAN
+// #define JRNL_64_BIT
+#else
+ #error Unable to determine endianness
+#endif
+
+
+/**
+* <b>Rule:</b> Data block size (JRNL_DBLK_SIZE) MUST be a power of 2 such that
+* <pre>
+* JRNL_DBLK_SIZE * JRNL_SBLK_SIZE == n * 512 (n = 1,2,3...)
+* </pre>
+* (The disk softblock size is 512 for Linux kernels >= 2.6)
+*/
+//#define JRNL_DBLK_SIZE 128 ///< Data block size in bytes (CANNOT BE LESS THAN 32!)
+//#define JRNL_SBLK_SIZE 4 ///< Disk softblock size in multiples of JRNL_DBLK_SIZE
+//#define JRNL_MIN_FILE_SIZE 128 ///< Min. jrnl file size in sblks (excl. FileHeader)
+//#define JRNL_MAX_FILE_SIZE 4194304 ///< Max. jrnl file size in sblks (excl. FileHeader)
+//#define JRNL_MIN_NUM_FILES 4 ///< Min. number of journal files
+//#define JRNL_MAX_NUM_FILES 64 ///< Max. number of journal files
+//#define JRNL_ENQ_THRESHOLD 80 ///< Percent full when enqueue connection will be closed
+//
+//#define JRNL_RMGR_PAGE_SIZE 128 ///< Journal page size in softblocks
+//#define JRNL_RMGR_PAGES 16 ///< Number of pages to use in wmgr
+//
+//#define JRNL_WMGR_DEF_PAGE_SIZE 64 ///< Journal write page size in softblocks (default)
+//#define JRNL_WMGR_DEF_PAGES 32 ///< Number of pages to use in wmgr (default)
+//
+//#define JRNL_WMGR_MAXDTOKPP 1024 ///< Max. dtoks (data blocks) per page in wmgr
+//#define JRNL_WMGR_MAXWAITUS 100 ///< Max. wait time (us) before submitting AIO
+//
+//#define JRNL_INFO_EXTENSION "jinf" ///< Extension for journal info files
+//#define JRNL_DATA_EXTENSION "jdat" ///< Extension for journal data files
+//#define RHM_JDAT_TXA_MAGIC 0x614d4852 ///< ("RHMa" in little endian) Magic for dtx abort hdrs
+//#define RHM_JDAT_TXC_MAGIC 0x634d4852 ///< ("RHMc" in little endian) Magic for dtx commit hdrs
+//#define RHM_JDAT_DEQ_MAGIC 0x644d4852 ///< ("RHMd" in little endian) Magic for deq rec hdrs
+//#define RHM_JDAT_ENQ_MAGIC 0x654d4852 ///< ("RHMe" in little endian) Magic for enq rec hdrs
+//#define RHM_JDAT_FILE_MAGIC 0x664d4852 ///< ("RHMf" in little endian) Magic for file hdrs
+//#define RHM_JDAT_EMPTY_MAGIC 0x784d4852 ///< ("RHMx" in little endian) Magic for empty dblk
+//#define RHM_JDAT_VERSION 0x01 ///< Version (of file layout)
+//#define RHM_CLEAN_CHAR 0xff ///< Char used to clear empty space on disk
+
+#define RHM_LENDIAN_FLAG 0x0 ///< Value of little endian flag on disk
+#define RHM_BENDIAN_FLAG 0x1 ///< Value of big endian flag on disk
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+/**
+ * \brief Class aggregating the journal and store hard-coded configuration values.
+ */
+class Configuration
+{
+public:
+ static const uint8_t s_endianValue =
+#if defined(JRNL_LITTLE_ENDIAN)
+ RHM_LENDIAN_FLAG;
+#elif defined(JRNL_BIG_ENDIAN)
+ RHM_BENDIAN_FLAG;
+#else
+# error Unknown endianness
+#endif
+};
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+#endif // qpid_asyncStore_jrnl2_Configuration_h_
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataOpState.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataOpState.cpp?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataOpState.cpp (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataOpState.cpp Wed May 9 14:19:28 2012
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file DataOpState.cpp
+ */
+
+#include "DataOpState.h"
+
+#include "JournalError.h"
+
+#include <sstream> // std::ostringstream
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+DataOpState::DataOpState() :
+ State<opState_t>()
+{}
+
+DataOpState::DataOpState(const opState_t s) :
+ State<opState_t>(s)
+{}
+
+DataOpState::DataOpState(const DataOpState& s) :
+ State<opState_t>(s)
+{}
+
+DataOpState::~DataOpState()
+{}
+
+void
+DataOpState::enqueue()
+{
+ set(OP_ENQUEUE);
+}
+
+void
+DataOpState::dequeue()
+{
+ set(OP_DEQUEUE);
+}
+
+const char*
+DataOpState::getAsStr() const
+{
+ return s_toStr(m_state);
+}
+
+// static
+const char*
+DataOpState::s_toStr(const opState_t s)
+{
+ switch (s) {
+ case OP_NONE:
+ return "OP_NONE";
+ case OP_ENQUEUE:
+ return "OP_ENQUEUE";
+ case OP_DEQUEUE:
+ return "OP_DEQUEUE";
+ default:
+ std::ostringstream oss;
+ oss << "<unknown state (" << "s" << ")>";
+ return oss.str().c_str();
+ }
+}
+
+// protected
+void
+DataOpState::set(const opState_t s)
+{
+ // State transition logic: set stateError to true if an invalid transition is attempted
+ bool stateTransitionError = false;
+ switch(m_state) {
+ case OP_NONE:
+ if (s != OP_ENQUEUE) stateTransitionError = true;
+ break;
+ case OP_ENQUEUE:
+ if (s != OP_DEQUEUE) stateTransitionError = true;
+ break;
+ case OP_DEQUEUE: // Cannot move out of this state except with reset()
+ stateTransitionError = true;
+ break;
+ }
+ if (stateTransitionError) {
+ THROW_STATE_EXCEPTION(JournalError::JERR_MSGOPSTATE, s_toStr(s), s_toStr(m_state), "DataOpState", "set");
+ }
+ m_state = s;
+}
+
+}}} // namespace qpid::asyncStore::jrnl2
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataOpState.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataOpState.h?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataOpState.h (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataOpState.h Wed May 9 14:19:28 2012
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file DataOpState.h
+ */
+
+#ifndef qpid_asyncStore_jrnl2_DataOpState_h_
+#define qpid_asyncStore_jrnl2_DataOpState_h_
+
+#include "State.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+/**
+ * \brief Enumeration of valid data record enqueue/dequeue states.
+ */
+typedef enum {
+ OP_NONE = 0, ///< Data record has not been processed by async store
+ OP_ENQUEUE, ///< Data record has been enqueued
+ OP_DEQUEUE ///< Data record has been dequeued
+} opState_t;
+
+/**
+ * \brief State machine to control the data record operational state (ie none -> enqueued -> dequeued).
+ *
+ * The following state diagram shows valid data record operational state transitions:
+ * \dot
+ * digraph DataOpState {
+ * node [fontname=Helvetica, fontsize=8];
+ * OP_NONE [URL="\ref OP_NONE"]
+ * OP_ENQUEUE [URL="\ref OP_ENQUEUE"]
+ * OP_DEQUEUE [URL="\ref OP_DEQUEUE"]
+ *
+ * edge [fontname=Helvetica, fontsize=8]
+ * OP_NONE->OP_ENQUEUE [label=" enqueue()" URL="\ref qpid::jrnl2::DataOpState::enqueue()"]
+ * OP_ENQUEUE->OP_DEQUEUE [label=" dequeue()" URL="\ref qpid::jrnl2::DataOpState::enqueue()"]
+ * }
+ * \enddot
+ */
+class DataOpState: public State<opState_t>
+{
+public:
+ /**
+ * \brief Default constructor, setting internal state to OP_NONE.
+ */
+ DataOpState();
+
+ /**
+ * \brief Constructor allowing the internal state to be preset to any valid value of opState_t.
+ *
+ * \param s State value to which the internal state of this new instance should be initialized.
+ */
+ DataOpState(const opState_t s);
+
+ /**
+ * \brief Copy constructor.
+ *
+ * \param s Instance from which this new instance state should be copied.
+ */
+ DataOpState(const DataOpState& s);
+
+ /**
+ * \brief Virtual destructor
+ */
+ virtual ~DataOpState();
+
+ /**
+ * \brief Changes the data record operational state from OP_NONE to OP_ENQUEUE.
+ *
+ * \throws JournalException with JERR_ILLEGALDTOKOPSTATECHANGE if current state makes it illegal to change
+ * to OP_ENQUEUE according to the state machine semantics.
+ */
+ void enqueue();
+
+ /**
+ * \brief Changes the data record operational state from OP_ENQUEUE to OP_DEQUEUE.
+ *
+ * \throws JournalException with JERR_ILLEGALDTOKOPSTATECHANGE if current state makes it illegal to change
+ * to OP_DEQUEUE according to the state machine semantics.
+ */
+ void dequeue();
+
+ /**
+ * \brief Get the current state in a string form.
+ *
+ * \returns String representation of internal state m_state as returned by static function s_toStr().
+ */
+ const char* getAsStr() const;
+
+ /**
+ * \brief Static helper function to convert a numeric opState_t type to a string representation.
+ */
+ static const char* s_toStr(const opState_t s);
+
+protected:
+ /**
+ * \brief Set (or change) the value of m_state. This function implements the state machine checks for
+ * legal state change transitions, and throw an exception if an illegal state transition is requested.
+ *
+ * \param s State to which this machine should be changed.
+ * \throws JournalException with JERR_ILLEGALDTOKOPSTATECHANGE if current state makes it illegal to change
+ * to the requested state \a s.
+ */
+ void set(const opState_t s);
+
+};
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+#endif // qpid_asyncStore_jrnl2_DataOpState_h_
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataToken.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataToken.cpp?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataToken.cpp (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataToken.cpp Wed May 9 14:19:28 2012
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file DataToken.cpp
+ */
+
+#include "DataToken.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+DataToken::DataToken() :
+ m_dataOpState(),
+ m_recordId(s_recordIdCounter.next()),
+ m_externalRecordIdFlag(false)
+{}
+
+DataToken::DataToken(const recordId_t rid) :
+ m_dataOpState(),
+ m_recordId(rid),
+ m_externalRecordIdFlag(true)
+{}
+
+DataToken::~DataToken() {}
+
+const DataOpState&
+DataToken::getDataOpState() const
+{
+ return m_dataOpState;
+}
+
+DataOpState&
+DataToken::getDataOpState()
+{
+ return m_dataOpState;
+}
+
+const DataWrComplState&
+DataToken::getDataWrComplState() const
+{
+ return m_dataWrComplState;
+}
+
+DataWrComplState&
+DataToken::getDataWrComplState()
+{
+ return m_dataWrComplState;
+}
+
+bool
+DataToken::isTransient() const
+{
+ return m_transientFlag;
+}
+
+bool
+DataToken::isExternal() const
+{
+ return m_externalFlag;
+}
+
+const std::string&
+DataToken::getExternalLocation() const
+{
+ return m_externalLocation;
+}
+
+recordId_t
+DataToken::getRecordId() const
+{
+ return m_recordId;
+}
+
+bool
+DataToken::isRecordIdExternal() const
+{
+ return m_externalRecordIdFlag;
+}
+
+recordId_t
+DataToken::getDequeueRecordId() const
+{
+ return m_dequeueRecordId;
+}
+
+void
+DataToken::setRecordId(const recordId_t rid)
+{
+ m_recordId = rid;
+ m_externalRecordIdFlag = true;
+}
+
+void
+DataToken::setDequeueRecordId(const recordId_t drid)
+{
+ m_dequeueRecordId = drid;
+}
+
+void
+DataToken::toStream(std::ostream& os) const
+{
+ /// \todo TODO: Implementation required
+ os << "status string";
+}
+
+// private static
+RecordIdCounter_t DataToken::s_recordIdCounter;
+
+}}} // namespace qpid::asyncStore::jrnl2
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataToken.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataToken.h?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataToken.h (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataToken.h Wed May 9 14:19:28 2012
@@ -0,0 +1,176 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file DataToken.h
+ */
+
+#ifndef qpid_jrnl2_asyncStore_DataToken_h_
+#define qpid_jrnl2_asyncStore_DataToken_h_
+
+#include "DataOpState.h"
+#include "DataWrComplState.h"
+#include "RecordIdCounter.h"
+
+#include <string>
+#include <stdint.h> // uint64_t
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+/**
+ * \brief Contains the state and metadata for each record handled by the async store.
+ *
+ * The DataToken keeps the state of the data record through the AIO write and recovery process. It may also
+ * be used to set certain properties of the data record.
+ */
+class DataToken : public Streamable
+{
+public:
+ /**
+ * \brief Default constructor, which internally assigns a unique record Id (RID).
+ */
+ DataToken();
+
+ /**
+ * \brief Constructor which accepts an external record id (RID). No check is made of the uniqueness of this RID.
+ *
+ * \param rid The externally assigned record Id, which must be unique across this journal instance within the
+ * window presented by the journal files.
+ */
+ DataToken(const recordId_t rid);
+
+ /**
+ * \brief Virtual destructor
+ */
+ virtual ~DataToken();
+
+ // Get functions
+
+ /**
+ * \brief Get the const DataOpState state manager object for this data token.
+ *
+ * \return \c const state of the data record associated with this DataToken instance.
+ */
+ const DataOpState& getDataOpState() const;
+
+ /**
+ * \brief Get the DataOpState state manager object for this data token.
+ *
+ * \return Non -\c const (modifiable) state of the data record associated with this DataToken instance.
+ */
+ DataOpState& getDataOpState();
+
+ /**
+ * \brief Get the const DataWrComplState state manager object for this data token.
+ *
+ * \return \c const state of the data record associated with this DataToken instance.
+ */
+ const DataWrComplState& getDataWrComplState() const;
+
+ /**
+ * \brief Get the DataWrComplState state manager object for this data token.
+ *
+ * \return Non -\c const (modifiable) state of the data record associated with this DataToken instance.
+ */
+ DataWrComplState& getDataWrComplState();
+
+ /**
+ * \brief Check if this data is marked as transient (will not be recovered).
+ *
+ * \return \c true if the data is transient, \c false otherwise.
+ */
+ bool isTransient() const;
+
+ /**
+ * \brief Check if this data is marked as external (i.e. has its content stored outside the store journal).
+ *
+ * \return \c true if the data is external (has its content stored outside the journal), \c false otherwise.
+ */
+ bool isExternal() const;
+
+ /**
+ * \brief Return the external storage location of the data if it is marked as external. This may be a path
+ * or URL.
+ *
+ * \return Location or URI pointing to the location of the storage for this data record if it is marked external.
+ */
+ const std::string& getExternalLocation() const;
+
+ /**
+ * \brief Return the Record Id (RID) for this data record.
+ *
+ * \return Record Id (RID) for this data record.
+ */
+ recordId_t getRecordId() const;
+
+ /**
+ * \brief Checks if the Record Id assigned to this data record was supplied externally.
+ *
+ * \return \c true if the data record had its Record Id (RID) assigned externally from the store, \c false
+ * otherwise.
+ */
+ bool isRecordIdExternal() const;
+
+ /**
+ * \brief Returns the recordId of the dequeuing record (if dequeued)
+ *
+ * \return The Record Id of the dequeue record which dequeues this record, or 0x0 if it has not been dequeued
+ * (or cannot be dequeued).
+ */
+ recordId_t getDequeueRecordId() const;
+
+ // Set functions
+
+ /**
+ * \brief Set the record Id (RID) for this record. No check is made of the uniqueness of this RID.
+ */
+ void setRecordId(const recordId_t rid);
+
+ /**
+ * \brief Set the record Id (RID) of the record which dequeues this record.
+ */
+ void setDequeueRecordId(const recordId_t drid);
+
+ // Debug aid(s)
+
+ /**
+ * \brief Stream a formatted summary of the state of this DataToken object.
+ */
+ virtual void toStream(std::ostream& os) const;
+
+protected:
+ DataOpState m_dataOpState; ///< Data operational state (none, enqueued, dequeued)
+ DataWrComplState m_dataWrComplState; ///< Data write completion state (none, part, complete)
+ bool m_transientFlag; ///< True if the data record is transient (eg for Flow-To-Disk)
+ bool m_externalFlag; ///< True if the data record is stored externally from the store
+ std::string m_externalLocation; ///< Location of data record when stored externally
+ recordId_t m_recordId; ///< Record Id which is unique to this store instance
+ bool m_externalRecordIdFlag; ///< True if the record id was set through this token
+ recordId_t m_dequeueRecordId; ///< Record Id of the dequeue record for this data record
+
+private:
+ static RecordIdCounter_t s_recordIdCounter; ///< Static instance keeps record Ids unique across system
+
+};
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+#endif // qpid_jrnl2_asyncStore_DataToken_h_
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataWrComplState.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataWrComplState.cpp?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataWrComplState.cpp (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataWrComplState.cpp Wed May 9 14:19:28 2012
@@ -0,0 +1,109 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file DataWrComplState.cpp
+ */
+
+#include "DataWrComplState.h"
+
+#include "JournalError.h"
+
+#include <sstream> // std::ostringstream
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+DataWrComplState::DataWrComplState() :
+ State<wrComplState_t>()
+{}
+
+DataWrComplState::DataWrComplState(const wrComplState_t s) :
+ State<wrComplState_t>(s)
+{}
+
+DataWrComplState::DataWrComplState(const DataWrComplState& s) :
+ State<wrComplState_t>(s)
+{}
+
+DataWrComplState::~DataWrComplState()
+{}
+
+void
+DataWrComplState::complete()
+{
+ set(WR_COMPLETE);
+}
+
+void
+DataWrComplState::partComplete()
+{
+ set(WR_PART);
+}
+
+const char*
+DataWrComplState::getAsStr() const
+{
+ return s_toStr(m_state);
+}
+
+// static
+const char*
+DataWrComplState::s_toStr(const wrComplState_t s)
+{
+ switch (s) {
+ case WR_NONE:
+ return "WR_NONE";
+ case WR_PART:
+ return "WR_PART";
+ case WR_COMPLETE:
+ return "WR_COMPLETE";
+ default:
+ std::ostringstream oss;
+ oss << "<unknown state (" << "s" << ")>";
+ return oss.str().c_str();
+ }
+}
+
+// protected
+void
+DataWrComplState::set(const wrComplState_t s)
+{
+ // State transition logic: set stateError to true if an invalid transition is attempted
+ bool stateTransitionError = false;
+ switch(m_state) {
+ case WR_NONE:
+ if (s != WR_PART && s != WR_COMPLETE) stateTransitionError = true;
+ break;
+ case WR_PART:
+ if (s != WR_COMPLETE) stateTransitionError = true;
+ break;
+ case WR_COMPLETE: // Cannot move out of this state except with reset()
+ stateTransitionError = true;
+ break;
+ }
+ if (stateTransitionError) {
+ THROW_STATE_EXCEPTION(JournalError::JERR_MSGWRCMPLSTATE, s_toStr(s), s_toStr(m_state), "DataWrComplState",
+ "set");
+ }
+ m_state = s;
+}
+
+}}} // namespace qpid::asyncStore::jrnl2
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataWrComplState.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataWrComplState.h?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataWrComplState.h (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DataWrComplState.h Wed May 9 14:19:28 2012
@@ -0,0 +1,130 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file DataWrComplState.h
+ */
+
+#ifndef qpid_asyncStore_jrnl2_DataWrComplState_h_
+#define qpid_asyncStore_jrnl2_DataWrComplState_h_
+
+#include "State.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+/**
+ * \brief Enumeration of valid data record write completion states.
+ */
+typedef enum {
+ WR_NONE = 0, ///< Data record has not been written into async store buffer
+ WR_PART, ///< Data record has been partly written to store buffer
+ WR_COMPLETE ///< Data record write to store buffer is complete
+} wrComplState_t;
+
+/**
+ * \brief State machine for the data record write completion state - ie whether the data record has been written
+ * to the store buffer.
+ *
+ * The following state diagram shows valid write buffer states and transitions:
+ * \dot
+ * digraph wrComplState_t {
+ * node [fontname=Helvetica, fontsize=8];
+ * WR_NONE [URL="\ref WR_NONE"]
+ * WR_PART [URL="\ref WR_PART"]
+ * WR_COMPLETE [URL="\ref WR_COMPLETE"]
+ *
+ * edge [fontname=Helvetica, fontsize=8]
+ * WR_NONE->WR_PART [label=" partComplete()" URL="\ref qpid::jrnl2::DataWrComplState::partComplete()"]
+ * WR_NONE->WR_COMPLETE [label=" complete()" URL="\ref qpid::jrnl2::DataWrComplState::complete()"]
+ * WR_PART->WR_COMPLETE [label=" complete()" URL="\ref qpid::jrnl2::DataWrComplState::complete()"]
+ * }
+ * \enddot
+ */
+class DataWrComplState: public State<wrComplState_t> {
+public:
+ /**
+ * \brief Default constructor, setting internal state to WR_NONE.
+ */
+ DataWrComplState();
+
+ /**
+ * \brief Constructor allowing the internal state to be preset to any valid value of wrComplState_t.
+ *
+ * \param s State value to which the internal state of this new instance should be initialized.
+ */
+ DataWrComplState(const wrComplState_t s);
+
+ /**
+ * \brief Copy constructor.
+ *
+ * \param s Instance from which this new instance state should be copied.
+ */
+ DataWrComplState(const DataWrComplState& s);
+
+ /**
+ * \brief Virtual destructor
+ */
+ virtual ~DataWrComplState();
+
+ /**
+ * \brief Changes the data record operational state from WR_NONE or WR_PART to WR_COMPLETE.
+ *
+ * \throws JournalException with JERR_ILLEGALDTOKOPSTATECHANGE if current state makes it illegal to change
+ * to WR_COMPLETE according to the state machine semantics.
+ */
+ void complete();
+
+ /**
+ * \brief Changes the data record operational state from WR_NONE to WR_PART.
+ *
+ * \throws JournalException with JERR_ILLEGALDTOKOPSTATECHANGE if current state makes it illegal to change
+ * to WR_PART according to the state machine semantics.
+ */
+ void partComplete();
+
+ /**
+ * \brief Get the current state in a string form.
+ *
+ * \returns String representation of internal state m_state as returned by static function s_toStr().
+ */
+ const char* getAsStr() const;
+
+ /**
+ * \brief Static helper function to convert a numeric wrComplState_t type to a string representation.
+ */
+ static const char* s_toStr(const wrComplState_t s);
+
+protected:
+ /**
+ * \brief Set (or change) the value of m_state. This function implements the state machine checks for
+ * legal state change transitions, and throw an exception if an illegal state transition is requested.
+ *
+ * \param s State to which this machine should be changed.
+ * \throws JournalException with JERR_ILLEGALDTOKOPSTATECHANGE if current state makes it illegal to change
+ * to the requested state \a s.
+ */
+ void set(const wrComplState_t s);
+
+};
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+#endif // qpid_asyncStore_jrnl2_DataWrComplState_h_
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DequeueHeader.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DequeueHeader.cpp?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DequeueHeader.cpp (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DequeueHeader.cpp Wed May 9 14:19:28 2012
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file DequeueHeader.cpp
+ */
+
+#include "DequeueHeader.h"
+
+#include "RecordTail.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+DequeueHeader::DequeueHeader() :
+ RecordHeader(),
+ m_dequeuedRecordId(0),
+ m_xidSize(0)
+{}
+
+DequeueHeader::DequeueHeader(const uint32_t magic,
+ const uint8_t version,
+ const uint64_t recordId,
+ const uint64_t dequeuedRecordId,
+ const uint64_t xidSize,
+ const bool overwriteIndicator,
+ const bool tplCommitOnTxnComplFlag) :
+ RecordHeader(magic, version, recordId, overwriteIndicator),
+ m_dequeuedRecordId(dequeuedRecordId),
+ m_xidSize(xidSize)
+{
+ setTplCommitOnTxnComplFlag(tplCommitOnTxnComplFlag);
+}
+
+DequeueHeader::DequeueHeader(const DequeueHeader& dh) :
+ RecordHeader(dh),
+ m_dequeuedRecordId(dh.m_dequeuedRecordId),
+ m_xidSize(dh.m_xidSize)
+{}
+
+DequeueHeader::~DequeueHeader()
+{}
+
+void
+DequeueHeader::copy(const DequeueHeader& dh)
+{
+ RecordHeader::copy(dh);
+ m_dequeuedRecordId = dh.m_dequeuedRecordId;
+ m_xidSize = dh.m_xidSize;
+}
+
+void
+DequeueHeader::reset()
+{
+ RecordHeader::reset();
+ m_dequeuedRecordId = 0;
+ m_xidSize = 0;
+}
+
+bool
+DequeueHeader::getTplCommitOnTxnComplFlag() const
+{
+ return m_flags & DEQ_HDR_TPL_COMMIT_ON_TXN_COMPL_MASK;
+}
+
+void
+DequeueHeader::setTplCommitOnTxnComplFlag(const bool commitOnTxnCompl)
+{
+ m_flags = commitOnTxnCompl ?
+ m_flags | DEQ_HDR_TPL_COMMIT_ON_TXN_COMPL_MASK :
+ m_flags & (~DEQ_HDR_TPL_COMMIT_ON_TXN_COMPL_MASK);
+}
+
+//static
+uint64_t
+DequeueHeader::getHeaderSize()
+{
+ return static_cast<uint64_t>(sizeof(DequeueHeader));
+}
+
+uint64_t
+DequeueHeader::getBodySize() const
+{
+ return m_xidSize;
+}
+
+uint64_t
+DequeueHeader::getRecordSize() const
+{
+ return getHeaderSize() + (getBodySize() > 0LL ?
+ getBodySize() + RecordTail::getSize() :
+ 0);
+}
+
+}}} // namespace qpid::asyncStore::jrnl2
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DequeueHeader.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DequeueHeader.h?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DequeueHeader.h (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/DequeueHeader.h Wed May 9 14:19:28 2012
@@ -0,0 +1,230 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file DequeueHeader.h
+ *
+ * List of journal record structs:
+ * struct DequeueHeader <-- This file
+ * struct EnqueueHeader
+ * struct EventHeader
+ * struct FileHeader
+ * struct RecordHeader
+ * struct RecordTail
+ * struct TransactionHeader
+ *
+ * Overview of journal record structs:
+ *
+ * <pre>
+ * +------------+ +--------------+
+ * | RecordTail | | RecordHeader |
+ * +------------+ | (abstract) |
+ * +--------------+
+ * ^
+ * |
+ * +----------------+---------+-------+-------------------+
+ * | | | |
+ * +------------+ +-------------+ +---------------+ +-------------------+
+ * | FileHeader | | EventHeader | | DequeueHeader | | TransactionHeader |
+ * +------------+ +-------------+ +---------------+ +-------------------+
+ * ^
+ * |
+ * +---------------+
+ * | EnqueueHeader |
+ * +---------------+
+ * </pre>
+ */
+
+#ifndef qpid_asyncStore_jrnl2_DequeueHeader_h_
+#define qpid_asyncStore_jrnl2_DequeueHeader_h_
+
+#include "RecordHeader.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+#pragma pack(1)
+
+/**
+ * \brief Struct for dequeue record header. This record marks a previous enqueue as
+ * logically deleted.
+ *
+ * The dequeue record conceptually marks a previous enqueue record as dequeued,
+ * or logically deleted. This is achieved by recording the record ID (rid) of
+ * the enqueue record being deleted in the m_dequeueRecordId field. Note that this
+ * rid is distinct from the rid assigned to the dequeue record itself in m_recordId.
+ *
+ * Dequeue records do not carry data. However, if it is transactional (and thus has
+ * a transaction ID (xid), then this record will be terminated by a RecordTail struct.
+ * If, on the other hand, this record is non-transactional, then the rec_tail
+ * is absent.
+ *
+ * Note that this record had its own rid distinct from the rid of the record it is dequeuing.
+ * The rid field below is the rid of the dequeue record itself; the deq-rid field is the rid of a
+ * previous enqueue record being dequeued by this record.
+ *
+ * Record layout in binary format (32 bytes):
+ * <pre>
+ * 0x0 0x7
+ * +-----+-----+-----+-----+-----+-----+-----+-----+ -+
+ * 0x00 | m_magic | v | e | m_flags | |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+ | struct RecordHeader
+ * 0x08 | m_recordId | |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+ -+
+ * 0x10 | m_dequeuedRecordId |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+
+ * 0x18 | m_xidSize |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+
+ * </pre>
+ * <table>
+ * <tr>
+ * <td>v</td>
+ * <td>file version [ <code>_version</code> ] (If the format or encoding of
+ * this file changes, then this number should be incremented)</td>
+ * </tr>
+ * <tr>
+ * <td>e</td>
+ * <td>endian flag [ <code>_bigEndianFlag</code> ], <b>false</b> (0x00) for
+ * little endian, <b>true</b> (0x01) for big endian</td>
+ * </tr>
+ * </table>
+ * <pre>
+ * Non-transactional:
+ * +---------+
+ * | dequeue |
+ * | header |
+ * +---------+
+ * <-- 32 --->
+ *
+ * Transactional:
+ * +---------+------------------------+--------+
+ * | dequeue | XID | record |
+ * | header | | tail |
+ * +---------+------------------------+--------+
+ * <-- 32 ---><----- m_xidSize ------><-- 16 -->
+ * </pre>
+ */
+class DequeueHeader : public RecordHeader
+{
+public:
+ uint64_t m_dequeuedRecordId; ///< Record ID of dequeued record
+ uint64_t m_xidSize; ///< XID size
+
+ /**
+ * \brief Mask for the record header flags field m_flags which is used in dequeue records in the Transaction
+ * Prepared List (TPL) to indicate that a closed transaction resulted in a commit (if the flag is set)
+ * or an abort (if the flag is not set).
+ */
+ static const uint16_t DEQ_HDR_TPL_COMMIT_ON_TXN_COMPL_MASK = 0x10;
+
+ /**
+ * \brief Default constructor, which sets all values to 0.
+ */
+ DequeueHeader();
+
+ /**
+ * \brief Convenience constructor which initializes values during construction.
+ *
+ * \param magic The magic for this record
+ * \param version Version of this record
+ * \param recordId Record identifier for this record
+ * \param dequeuedRecordId Record identifier of the record being dequeued by this record
+ * \param xidSize Size of the transaction (or distributed transaction) ID for this record
+ * \param overwriteIndicator Flag indicating the present value of the overwrite indicator when writing this
+ * record
+ * \param tplCommitOnTxnComplFlag
+ */
+ DequeueHeader(const uint32_t magic,
+ const uint8_t version,
+ const uint64_t recordId,
+ const uint64_t dequeuedRecordId,
+ const uint64_t xidSize,
+ const bool overwriteIndicator,
+ const bool tplCommitOnTxnComplFlag = false);
+
+ /**
+ * \brief Copy constructor
+ *
+ * \param dh Instance to be copied
+ */
+ DequeueHeader(const DequeueHeader& dh);
+
+ /**
+ * \brief Virtual destructor
+ */
+ virtual ~DequeueHeader();
+
+ /**
+ * \brief Convenience copy method.
+ */
+ void copy(const DequeueHeader& dh);
+
+ /**
+ * \brief Reset this record to default values (mostly 0)
+ */
+ void reset();
+
+ /**
+ * \brief Return the value of the tplCommitOnTxnComplFlag for this record. This flag is used only within the
+ * TPL, and if set, indicates that the transaction was closed using a commit. If not set, the transaction was
+ * closed using an abort. This is used during recovery of the transactions in the store.
+ *
+ * \returns \b true if the tplCommitOnTxnComplFlag flag for this record is set, \b false otherwise.
+ */
+ bool getTplCommitOnTxnComplFlag() const;
+
+ /**
+ * \brief Set the value of the tplCommitOnTxnComplFlag for this record. This is only used in the TPL, and is
+ * ignored elsewhere.
+ *
+ * \param commitOnTxnCompl The value to be set in the tplCommitOnTxnComplFlag. If \b true, the transaction was
+ * closed with a commit; if \b false, with an abort.
+ */
+ void setTplCommitOnTxnComplFlag(const bool commitOnTxnCompl);
+
+ /**
+ * \brief Return the header size of this record in bytes.
+ *
+ * \returns Size of record header in bytes.
+ */
+ static uint64_t getHeaderSize();
+
+ /**
+ * \brief Return the body (xid and data) size of this record in bytes.
+ *
+ * \returns Size of record body in bytes.
+ */
+ uint64_t getBodySize() const;
+
+ /**
+ * \brief Return total size of this record in bytes, being in the case of the dequeue record the size of the
+ * header, the size of the body (xid only) and the size of the tail.
+ *
+ * \returns Total size of record in bytes.
+ */
+ uint64_t getRecordSize() const;
+
+};
+
+#pragma pack()
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+#endif // qpid_asyncStore_jrnl2_DequeueHeader_h_
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EnqueueHeader.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EnqueueHeader.cpp?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EnqueueHeader.cpp (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EnqueueHeader.cpp Wed May 9 14:19:28 2012
@@ -0,0 +1,90 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file EnqueueHeader.cpp
+ */
+
+#include "EnqueueHeader.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+EnqueueHeader::EnqueueHeader() :
+ EventHeader()
+{}
+
+EnqueueHeader::EnqueueHeader(const uint32_t magic,
+ const uint8_t version,
+ const uint64_t recordId,
+ const uint64_t xidSize,
+ const uint64_t dataSize,
+ const bool overwriteIndicator,
+ const bool transient,
+ const bool external) :
+ EventHeader(magic, version, recordId, xidSize, dataSize, overwriteIndicator)
+{
+ setTransientFlag(transient);
+ setExternalFlag(external);
+}
+
+EnqueueHeader::EnqueueHeader(const EnqueueHeader& eh) :
+ EventHeader(eh)
+{}
+
+EnqueueHeader::~EnqueueHeader()
+{}
+
+bool
+EnqueueHeader::getTransientFlag() const
+{
+ return m_flags & ENQ_HDR_TRANSIENT_MASK;
+}
+
+void
+EnqueueHeader::setTransientFlag(const bool transient)
+{
+ m_flags = transient ?
+ m_flags | ENQ_HDR_TRANSIENT_MASK :
+ m_flags & (~ENQ_HDR_TRANSIENT_MASK);
+}
+
+bool
+EnqueueHeader::getExternalFlag() const
+{
+ return m_flags & ENQ_HDR_EXTERNAL_MASK;
+}
+
+void
+EnqueueHeader::setExternalFlag(const bool external)
+{
+ m_flags = external ?
+ m_flags | ENQ_HDR_EXTERNAL_MASK :
+ m_flags & (~ENQ_HDR_EXTERNAL_MASK);
+}
+
+//static
+uint64_t
+EnqueueHeader::getHeaderSize()
+{
+ return sizeof(EnqueueHeader);
+}
+
+}}} // namespace qpid::asyncStore::jrnl2
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EnqueueHeader.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EnqueueHeader.h?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EnqueueHeader.h (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EnqueueHeader.h Wed May 9 14:19:28 2012
@@ -0,0 +1,219 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file EnqueueHeader.h
+ *
+ * List of journal record structs:
+ * struct DequeueHeader
+ * struct EnqueueHeader <-- This file
+ * struct EventHeader
+ * struct FileHeader
+ * struct RecordHeader
+ * struct RecordTail
+ * struct TransactionHeader
+ *
+ * Overview of journal record structs:
+ *
+ * <pre>
+ * +------------+ +--------------+
+ * | RecordTail | | RecordHeader |
+ * +------------+ | (abstract) |
+ * +--------------+
+ * ^
+ * |
+ * +----------------+---------+-------+-------------------+
+ * | | | |
+ * +------------+ +-------------+ +---------------+ +-------------------+
+ * | FileHeader | | EventHeader | | DequeueHeader | | TransactionHeader |
+ * +------------+ +-------------+ +---------------+ +-------------------+
+ * ^
+ * |
+ * +---------------+
+ * | EnqueueHeader |
+ * +---------------+
+ * </pre>
+ */
+
+#ifndef qpid_asyncStore_jrnl2_EnqueueHeader_h_
+#define qpid_asyncStore_jrnl2_EnqueueHeader_h_
+
+#include "EventHeader.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+#pragma pack(1)
+
+/**
+ * \brief Struct for enqueue record header. This record stores message data to the
+ * journal.
+ *
+ * Enqueue records record the content of messages for possible later recovery, and
+ * are so-called because they correspond with the event of enqueuing the record on
+ * a queue.
+ *
+ * In addition to the fields inherited from RecordHeader, this struct includes both the
+ * transaction id (xid) and data blob sizes.
+ *
+ * This header precedes the enqueued message data in journal files, and unless there is
+ * no xid and no data (both with 0 length), is followed by a RecordTail.
+ *
+ * Record layout in binary format (32 bytes):
+ * <pre>
+ * 0x0 0x7
+ * +-----+-----+-----+-----+-----+-----+-----+-----+ -+
+ * 0x00 | m_magic | v | e | m_flags | |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+ | struct RecordHeader
+ * 0x08 | m_recordId | |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+ -+
+ * 0x10 | m_xidSize |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+
+ * 0x18 | m_dataSize |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+
+ * </pre>
+ * <table>
+ * <tr>
+ * <td>v</td>
+ * <td>file version [ <code>_version</code> ] (If the format or encoding of
+ * this file changes, then this number should be incremented)</td>
+ * </tr>
+ * <tr>
+ * <td>e</td>
+ * <td>endian flag [ <code>_bigEndianFlag</code> ], <b>false</b> (0x00) for
+ * little endian, <b>true</b> (0x01) for big endian</td>
+ * </tr>
+ * </table>
+ * <pre>
+ * No content (ie m_xidSize==0 AND m_dataSize==0):
+ * +---------+
+ * | enqueue |
+ * | header |
+ * +---------+
+ * <-- 32 --->
+ *
+ * With content:
+ * +---------+----------------+---------------------------+--------+
+ * | enqueue | XID | Data | record |
+ * | header | | | tail |
+ * +---------+----------------+---------------------------+--------+
+ * <-- 32 ---><-- m_xidSize --><------ m_dataSize -------><-- 16 -->
+ * </pre>
+ */
+class EnqueueHeader : public EventHeader
+{
+public:
+ /**
+ * \brief Mask for the record header flags field m_flags which is used to indicate that a record is transient
+ * (if the flag is set) or durable (if the flag is not set).
+ */
+ static const uint16_t ENQ_HDR_TRANSIENT_MASK = 0x10;
+
+ /**
+ * \brief Mask for the record header flags field m_flags which is used to indicate that a record is using
+ * external storage for its data content (if the flag is set). If the flag is not set, then the data content
+ * is stored in the journal itself.
+ */
+ static const uint16_t ENQ_HDR_EXTERNAL_MASK = 0x20;
+
+ /**
+ * \brief Default constructor, which sets all values to 0.
+ */
+ EnqueueHeader();
+
+ /**
+ * \brief Convenience constructor which initializes values during construction.
+ *
+ * \param magic The magic for this record
+ * \param version Version of this record
+ * \param recordId Record identifier for this record
+ * \param xidSize Size of the transaction (or distributed transaction) ID for this record
+ * \param dataSize Size of the opaque data block for this record
+ * \param overwriteIndicator Flag indicating the present value of the overwrite indicator when writing this
+ * record
+ * \param transient Flag indicating that this record is transient (ie to be discarded on recovery)
+ * \param external Flag indicating that this record's data is stored externally to the journal, the data portion
+ * of the record identifies the storage location.
+ */
+ EnqueueHeader(const uint32_t magic,
+ const uint8_t version,
+ const uint64_t recordId,
+ const uint64_t xidSize,
+ const uint64_t dataSize,
+ const bool overwriteIndicator,
+ const bool transient = false,
+ const bool external = false);
+
+ /**
+ * \brief Copy constructor
+ *
+ * \param eh Instance to be copied
+ */
+ EnqueueHeader(const EnqueueHeader& eh);
+
+ /**
+ * \brief Virtual destructor
+ */
+ virtual ~EnqueueHeader();
+
+ /**
+ * \brief Return the value of the Transient flag for this record. If set, this record is ignored during
+ * recovery.
+ *
+ * \returns true if the Transient flag for this record is set, false otherwise.
+ */
+ bool getTransientFlag() const;
+
+ /**
+ * \brief Set the value of the Transient flag for this record.
+ *
+ * \param transient The value to be set in the transient flag.
+ */
+ void setTransientFlag(const bool transient = true);
+
+ /**
+ * \brief Return the value of the External flag for this record. If set, this record data is not within the
+ * journal but external to it. The data part of this record contains the location of the stored data.
+ *
+ * \returns true if the Transient flag for this record is set, false otherwise.
+ */
+ bool getExternalFlag() const;
+
+ /**
+ * \brief Set the value of the External flag for this record.
+ *
+ * \param external The value to be set in the External flag.
+ */
+ void setExternalFlag(const bool external = true);
+
+ /**
+ * \brief Return the header size of this record in bytes.
+ *
+ * \returns Size of record header in bytes.
+ */
+ static uint64_t getHeaderSize();
+
+};
+
+#pragma pack()
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+#endif // qpid_asyncStore_jrnl2_EnqueueHeader_h_
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EventHeader.cpp
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EventHeader.cpp?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EventHeader.cpp (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EventHeader.cpp Wed May 9 14:19:28 2012
@@ -0,0 +1,95 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file EventHeader.cpp
+ */
+
+#include "EventHeader.h"
+
+#include "RecordTail.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+EventHeader::EventHeader() :
+ RecordHeader(),
+ m_xidSize(0),
+ m_dataSize(0)
+{}
+
+EventHeader::EventHeader(const uint32_t magic,
+ const uint8_t version,
+ const uint64_t recordId,
+ const uint64_t xidSize,
+ const uint64_t dataSize,
+ const bool overwriteIndicator) :
+ RecordHeader(magic, version, recordId, overwriteIndicator),
+ m_xidSize(xidSize),
+ m_dataSize(dataSize)
+{}
+
+EventHeader::EventHeader(const EventHeader& eh) :
+ RecordHeader(eh),
+ m_xidSize(eh.m_xidSize),
+ m_dataSize(eh.m_dataSize)
+{}
+
+EventHeader::~EventHeader()
+{}
+
+void
+EventHeader::copy(const EventHeader& e)
+{
+ RecordHeader::copy(e);
+ m_xidSize = e.m_xidSize;
+ m_dataSize = e.m_dataSize;
+}
+
+void
+EventHeader::reset()
+{
+ RecordHeader::reset();
+ m_xidSize = 0;
+ m_dataSize = 0;
+}
+
+//static
+uint64_t
+EventHeader::getHeaderSize()
+{
+ return sizeof(EventHeader);
+}
+
+uint64_t
+EventHeader::getBodySize() const
+{
+ return m_xidSize + m_dataSize;
+}
+
+uint64_t
+EventHeader::getRecordSize() const
+{
+ return getHeaderSize() + (getBodySize() ?
+ getBodySize() + RecordTail::getSize() :
+ 0);
+}
+
+}}} // namespace qpid::asyncStore::jrnl2
Added: qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EventHeader.h
URL: http://svn.apache.org/viewvc/qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EventHeader.h?rev=1336220&view=auto
==============================================================================
--- qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EventHeader.h (added)
+++ qpid/branches/asyncstore/cpp/src/qpid/asyncStore/jrnl2/EventHeader.h Wed May 9 14:19:28 2012
@@ -0,0 +1,201 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+/**
+ * \file EventHeader.h
+ *
+ * List of journal record structs:
+ * struct DequeueHeader
+ * struct EnqueueHeader
+ * struct EventHeader <-- This file
+ * struct FileHeader
+ * struct RecordHeader
+ * struct RecordTail
+ * struct TransactionHeader
+ *
+ * Overview of journal record structs:
+ *
+ * <pre>
+ * +------------+ +--------------+
+ * | RecordTail | | RecordHeader |
+ * +------------+ | (abstract) |
+ * +--------------+
+ * ^
+ * |
+ * +----------------+---------+-------+-------------------+
+ * | | | |
+ * +------------+ +-------------+ +---------------+ +-------------------+
+ * | FileHeader | | EventHeader | | DequeueHeader | | TransactionHeader |
+ * +------------+ +-------------+ +---------------+ +-------------------+
+ * ^
+ * |
+ * +---------------+
+ * | EnqueueHeader |
+ * +---------------+
+ * </pre>
+ */
+
+#ifndef qpid_asyncStore_jrnl2_EventHeader_h_
+#define qpid_asyncStore_jrnl2_EventHeader_h_
+
+#include "RecordHeader.h"
+
+namespace qpid {
+namespace asyncStore {
+namespace jrnl2 {
+
+#pragma pack(1)
+
+/**
+ * \brief Struct for an event record header. Event records can be used to record
+ * system events in the store.
+ *
+ * The EventHeader record type may be used to store events into the journal which do
+ * not constitute data content but changes of state in the broker. These can be
+ * recovered and used to set appropriate state in the broker.
+ *
+ * This record is almost identical to EnqueueRecord, but without the flags. It
+ * precedes the xid and stored event data in journal files, and unless there is
+ * no xid and no data (both with 0 length), is followed by a RecordTail.
+ *
+ * \todo TODO: I am uncertain at this time whether it is necessary to set an XID on an event
+ * record, but in case, I have left this feature in. In any event, there is only a
+ * 1 byte size penalty in the header size for doing so.
+ *
+ * Record layout in binary format (32 bytes):
+ * <pre>
+ * 0x0 0x7
+ * +-----+-----+-----+-----+-----+-----+-----+-----+ -+
+ * 0x00 | m_magic | v | e | m_flags | |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+ | struct RecordHeader
+ * 0x08 | m_recordId | |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+ -+
+ * 0x10 | m_xidSize |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+
+ * 0x18 | m_dataSize |
+ * +-----+-----+-----+-----+-----+-----+-----+-----+
+ * </pre>
+ * <table>
+ * <tr>
+ * <td>v</td>
+ * <td>file version [ <code>_version</code> ] (If the format or encoding of
+ * this file changes, then this number should be incremented)</td>
+ * </tr>
+ * <tr>
+ * <td>e</td>
+ * <td>endian flag [ <code>_bigEndianFlag</code> ], <b>false</b> (0x00) for
+ * little endian, <b>true</b> (0x01) for big endian</td>
+ * </tr>
+ * </table>
+ * <pre>
+ * No content (ie m_xidSize==0 AND m_dataSize==0):
+ * +---------+
+ * | event |
+ * | header |
+ * +---------+
+ * <-- 32 --->
+ *
+ * With content:
+ * +---------+----------------+---------------------------+--------+
+ * | event | XID | Event data | record |
+ * | header | | | tail |
+ * +---------+----------------+---------------------------+--------+
+ * <-- 32 ---><-- m_xidSize --><------ m_dataSize -------><-- 16 -->
+ * </pre>
+ */
+class EventHeader : public RecordHeader
+{
+public:
+ uint64_t m_xidSize; ///< XID size
+ uint64_t m_dataSize; ///< Record data size
+
+ /**
+ * \brief Default constructor, which sets all values to 0.
+ */
+ EventHeader();
+
+ /**
+ * \brief Convenience constructor which initializes values during construction.
+ *
+ * \param magic The magic for this record
+ * \param version Version of this record
+ * \param recordId Record identifier for this record
+ * \param xidSize Size of the transaction (or distributed transaction) ID for this record
+ * \param dataSize Size of the opaque data block for this record
+ * \param overwriteIndicator Flag indicating the present value of the overwrite indicator when writing this
+ * record
+ */
+ EventHeader(const uint32_t magic,
+ const uint8_t version,
+ const uint64_t recordId,
+ const uint64_t xidSize,
+ const uint64_t dataSize,
+ const bool overwriteIndicator);
+
+ /**
+ * \brief Copy constructor
+ *
+ * \param eh Instance to be copied
+ */
+ EventHeader(const EventHeader& eh);
+
+ /**
+ * \brief Virtual destructor
+ */
+ virtual ~EventHeader();
+
+ /**
+ * \brief Convenience copy method.
+ */
+ virtual void copy(const EventHeader& eh);
+
+ /**
+ * \brief Reset this record to default values (mostly 0)
+ */
+ virtual void reset();
+
+ /**
+ * \brief Return the header size of this record in bytes.
+ *
+ * \returns Size of record header in bytes.
+ */
+ static uint64_t getHeaderSize();
+
+ /**
+ * \brief Return the body (data) size of this record in bytes.
+ *
+ * \returns Size of record body in bytes.
+ */
+ virtual uint64_t getBodySize() const;
+
+ /**
+ * \brief Return total size of this record in bytes, being in the case of the enqueue record the size of the
+ * header, the size of the body (xid and data) and the size of the tail.
+ *
+ * \returns Total size of record in bytes.
+ */
+ virtual uint64_t getRecordSize() const;
+
+};
+
+#pragma pack()
+
+}}} // namespace qpid::asyncStore::jrnl2
+
+#endif // qpid_asyncStore_jrnl2_EventHeader_h_
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org