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 2013/12/03 22:46:39 UTC
svn commit: r1547601 - in /qpid/trunk/qpid/cpp/src: ./ qpid/linearstore/
qpid/linearstore/journal/
Author: kpvdr
Date: Tue Dec 3 21:46:39 2013
New Revision: 1547601
URL: http://svn.apache.org/r1547601
Log:
QPID-5358: Checksum not implemented in record tail, not checked during read. Implemented the Alder-32 algorithm to check every record from the start of the header to the start of the record tail. Upon recovery, the recovered record checksum is compared to the record tail checksum. This should detect the condition where a multi-page record may have its header and tail written, but one or more of its inbetween pages may not be fully written.
Added:
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.cpp
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.h
Modified:
qpid/trunk/qpid/cpp/src/linearstore.cmake
qpid/trunk/qpid/cpp/src/qpid/linearstore/ISSUES
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.cpp
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.h
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.cpp
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.h
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/jrec.h
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.cpp
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.h
qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/wmgr.cpp
Modified: qpid/trunk/qpid/cpp/src/linearstore.cmake
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/linearstore.cmake?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/linearstore.cmake (original)
+++ qpid/trunk/qpid/cpp/src/linearstore.cmake Tue Dec 3 21:46:39 2013
@@ -93,6 +93,7 @@ if (BUILD_LINEARSTORE)
# Journal source files
set (linear_jrnl_SOURCES
+ qpid/linearstore/journal/Checksum.cpp
qpid/linearstore/journal/data_tok.cpp
qpid/linearstore/journal/deq_rec.cpp
qpid/linearstore/journal/EmptyFilePool.cpp
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/ISSUES
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/ISSUES?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/ISSUES (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/ISSUES Tue Dec 3 21:46:39 2013
@@ -6,10 +6,10 @@ Store:
1. (SOLVED) Overwrite identity: When recovering a previously used file, if the write boundary coincides with old record
start, no way of discriminating old from new at boundary (used to use OWI).
-2. QPID-5357: Recycling files while in use not working, however, files are recovered to EFP during recovery. Must solve
+2. (SOLVED) QPID-5357: Recycling files while in use not working, however, files are recovered to EFP during recovery. Must solve
#1 first.
-3. QPID-5358: Checksum not implemented in record tail, not checked during read.
+3. (SOLVED) QPID-5358: Checksum not implemented in record tail, not checked during read.
4. QPID-5359: Rework qpid management parameters and controls (QMF).
@@ -27,9 +27,15 @@ Store:
* Store analysis and status
* Recovery/reading of message content
+Current bugs and performance issues:
+------------------------------------
+1. RH Bugzilla 1035843 - Slow performance for producers
+2. RH Bugzilla 1036071 - Crash when deleting queue
+3. RH Bugzilla 1035802 - Broker won't recover with durable queue
+4. RH Bugzilla 1036026 - Unable to create durable queue - framing error
+
Code tidy-up
------------
-
* Remove old comments
* Use c++ cast templates instead of (xxx)y
* Member names: xxx_
Added: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.cpp?rev=1547601&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.cpp Tue Dec 3 21:46:39 2013
@@ -0,0 +1,43 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#include "qpid/linearstore/journal/Checksum.h"
+
+namespace qpid {
+namespace linearstore {
+namespace journal {
+
+Checksum::Checksum() : a(1UL), b(0UL), MOD_ADLER(65521UL) {}
+
+Checksum::~Checksum() {}
+
+void Checksum::addData(const unsigned char* data, const std::size_t len) {
+ for (uint32_t i = 0; i < len; i++) {
+ a = (a + data[i]) % MOD_ADLER;
+ b = (a + b) % MOD_ADLER;
+ }
+}
+
+uint32_t Checksum::getChecksum() {
+ return (b << 16) | a;
+}
+
+}}}
Added: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.h?rev=1547601&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/Checksum.h Tue Dec 3 21:46:39 2013
@@ -0,0 +1,54 @@
+/*
+ *
+ * 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.
+ *
+ */
+
+#ifndef QPID_LINEARSTORE_JOURNAL_CHECKSUM_H_
+#define QPID_LINEARSTORE_JOURNAL_CHECKSUM_H_
+
+#include <cstddef>
+#include <stdint.h>
+
+namespace qpid {
+namespace linearstore {
+namespace journal {
+
+/*
+ * This checksum routine uses the Adler-32 algorithm as described in
+ * http://en.wikipedia.org/wiki/Adler-32. It is structured so that the
+ * data for which the checksum must be calculated can be added in several
+ * stages through the addData() function, and when complete, the checksum
+ * is obtained through a call to getChecksum().
+ */
+class Checksum
+{
+private:
+ uint32_t a;
+ uint32_t b;
+ const uint32_t MOD_ADLER;
+public:
+ Checksum();
+ virtual ~Checksum();
+ void addData(const unsigned char* data, const std::size_t len);
+ uint32_t getChecksum();
+};
+
+}}}
+
+#endif // QPID_LINEARSTORE_JOURNAL_CHECKSUM_H_
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/RecoveryManager.cpp Tue Dec 3 21:46:39 2013
@@ -24,6 +24,7 @@
#include <algorithm>
#include <cstdlib>
#include <iomanip>
+#include "qpid/linearstore/journal/Checksum.h"
#include "qpid/linearstore/journal/data_tok.h"
#include "qpid/linearstore/journal/deq_rec.h"
#include "qpid/linearstore/journal/EmptyFilePool.h"
@@ -201,6 +202,32 @@ bool RecoveryManager::readNextRemainingR
}
readJournalData((char*)*dataPtrPtr, dataSize);
+ // Check enqueue record checksum
+ Checksum checksum;
+ checksum.addData((unsigned char*)&enqueueHeader, sizeof(::enq_hdr_t));
+ if (xidSize > 0) {
+ checksum.addData((unsigned char*)*xidPtrPtr, xidSize);
+ }
+ if (dataSize > 0) {
+ checksum.addData((unsigned char*)*dataPtrPtr, dataSize);
+ }
+ ::rec_tail_t enqueueTail;
+ inFileStream_.read((char*)&enqueueTail, sizeof(::rec_tail_t));
+ uint32_t cs = checksum.getChecksum();
+//std::cout << std::hex << "### rid=0x" << enqueueHeader._rhdr._rid << " rtcs=0x" << enqueueTail._checksum << " cs=0x" << cs << std::dec << std::endl; // DEBUG
+ int res = ::rec_tail_check(&enqueueTail, &enqueueHeader._rhdr, cs);
+ if (res != 0) {
+ std::stringstream oss;
+ switch (res) {
+ case 1: oss << std::hex << "Magic: expected 0x" << ~enqueueHeader._rhdr._magic << "; found 0x" << enqueueTail._xmagic; break;
+ case 2: oss << std::hex << "Serial: expected 0x" << enqueueHeader._rhdr._serial << "; found 0x" << enqueueTail._serial; break;
+ case 3: oss << std::hex << "Record Id: expected 0x" << enqueueHeader._rhdr._rid << "; found 0x" << enqueueTail._rid; break;
+ case 4: oss << std::hex << "Checksum: expected 0x" << cs << "; found 0x" << enqueueTail._checksum; break;
+ default: oss << "Unknown error " << res;
+ }
+ throw jexception(jerrno::JERR_JREC_BADRECTAIL, oss.str(), "enq_rec", "decode"); // TODO: Don't throw exception, log info
+ }
+
// Set data token
dtokp->set_wstate(data_tok::ENQ);
dtokp->set_rid(enqueueHeader._rhdr._rid);
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.cpp?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.cpp Tue Dec 3 21:46:39 2013
@@ -23,6 +23,7 @@
#include <cassert>
#include <cstring>
+#include "qpid/linearstore/journal/Checksum.h"
#include "qpid/linearstore/journal/jexception.h"
namespace qpid {
@@ -55,10 +56,11 @@ deq_rec::reset(const uint64_t serial, co
_buff = 0;
_deq_tail._serial = serial;
_deq_tail._rid = rid;
+ _deq_tail._checksum = 0UL;
}
uint32_t
-deq_rec::encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks)
+deq_rec::encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum)
{
assert(wptr != 0);
assert(max_size_dblks > 0);
@@ -68,6 +70,7 @@ deq_rec::encode(void* wptr, uint32_t rec
std::size_t rec_offs = rec_offs_dblks * QLS_DBLK_SIZE_BYTES;
std::size_t rem = max_size_dblks * QLS_DBLK_SIZE_BYTES;
std::size_t wr_cnt = 0;
+
if (rec_offs_dblks) // Continuation of split dequeue record (over 2 or more pages)
{
if (size_dblks(rec_size()) - rec_offs_dblks > max_size_dblks) // Further split required
@@ -84,8 +87,10 @@ deq_rec::encode(void* wptr, uint32_t rec
rem -= wsize;
}
rec_offs -= _deq_hdr._xidsize - wsize2;
+ checksum.addData((unsigned char*)wptr, wr_cnt);
if (rem)
{
+ _deq_tail._checksum = checksum.getChecksum();
wsize = sizeof(_deq_tail) > rec_offs ? sizeof(_deq_tail) - rec_offs : 0;
wsize2 = wsize;
if (wsize)
@@ -109,8 +114,10 @@ deq_rec::encode(void* wptr, uint32_t rec
{
std::memcpy(wptr, (const char*)_xidp + rec_offs, wsize);
wr_cnt += wsize;
+ checksum.addData((unsigned char*)wptr, wr_cnt);
}
rec_offs -= _deq_hdr._xidsize - wsize;
+ _deq_tail._checksum = checksum.getChecksum();
wsize = sizeof(_deq_tail) > rec_offs ? sizeof(_deq_tail) - rec_offs : 0;
if (wsize)
{
@@ -142,8 +149,10 @@ deq_rec::encode(void* wptr, uint32_t rec
wr_cnt += wsize;
rem -= wsize;
}
+ checksum.addData((unsigned char*)wptr, wr_cnt);
if (rem)
{
+ _deq_tail._checksum = checksum.getChecksum();
wsize = rem >= sizeof(_deq_tail) ? sizeof(_deq_tail) : rem;
std::memcpy((char*)wptr + wr_cnt, (void*)&_deq_tail, wsize);
wr_cnt += wsize;
@@ -157,6 +166,8 @@ deq_rec::encode(void* wptr, uint32_t rec
{
std::memcpy((char*)wptr + wr_cnt, _xidp, _deq_hdr._xidsize);
wr_cnt += _deq_hdr._xidsize;
+ checksum.addData((unsigned char*)wptr, wr_cnt);
+ _deq_tail._checksum = checksum.getChecksum();
std::memcpy((char*)wptr + wr_cnt, (void*)&_deq_tail, sizeof(_deq_tail));
wr_cnt += sizeof(_deq_tail);
}
@@ -172,14 +183,12 @@ deq_rec::encode(void* wptr, uint32_t rec
bool
deq_rec::decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs)
{
- uint32_t checksum = 0UL; // TODO: Add checksum math
if (rec_offs == 0)
{
- //_deq_hdr.hdr_copy(h);
::rec_hdr_copy(&_deq_hdr._rhdr, &h);
ifsp->read((char*)&_deq_hdr._deq_rid, sizeof(_deq_hdr._deq_rid));
ifsp->read((char*)&_deq_hdr._xidsize, sizeof(_deq_hdr._xidsize));
- rec_offs = sizeof(_deq_hdr);
+ rec_offs = sizeof(::deq_hdr_t);
// Read header, allocate (if req'd) for xid
if (_deq_hdr._xidsize)
{
@@ -223,14 +232,18 @@ deq_rec::decode(::rec_hdr_t& h, std::ifs
ifsp->ignore(rec_size_dblks() * QLS_DBLK_SIZE_BYTES - rec_size());
assert(!ifsp->fail() && !ifsp->bad());
if (_deq_hdr._xidsize) {
- int res = ::rec_tail_check(&_deq_tail, &_deq_hdr._rhdr, checksum);
+ Checksum checksum;
+ checksum.addData((unsigned char*)&_deq_hdr, sizeof(_deq_hdr));
+ checksum.addData((unsigned char*)_buff, _deq_hdr._xidsize);
+ uint32_t cs = checksum.getChecksum();
+ int res = ::rec_tail_check(&_deq_tail, &_deq_hdr._rhdr, cs);
if (res != 0) {
std::stringstream oss;
switch (res) {
case 1: oss << std::hex << "Magic: expected 0x" << ~_deq_hdr._rhdr._magic << "; found 0x" << _deq_tail._xmagic; break;
case 2: oss << std::hex << "Serial: expected 0x" << _deq_hdr._rhdr._serial << "; found 0x" << _deq_tail._serial; break;
case 3: oss << std::hex << "Record Id: expected 0x" << _deq_hdr._rhdr._rid << "; found 0x" << _deq_tail._rid; break;
- case 4: oss << std::hex << "Checksum: expected 0x" << checksum << "; found 0x" << _deq_tail._checksum; break;
+ case 4: oss << std::hex << "Checksum: expected 0x" << cs << "; found 0x" << _deq_tail._checksum; break;
default: oss << "Unknown error " << res;
}
throw jexception(jerrno::JERR_JREC_BADRECTAIL, oss.str(), "deq_rec", "decode"); // TODO: Don't throw exception, log info
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.h?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/deq_rec.h Tue Dec 3 21:46:39 2013
@@ -48,7 +48,7 @@ public:
void reset(const uint64_t serial, const uint64_t rid, const uint64_t drid, const void* const xidp,
const std::size_t xidlen, const bool txn_coml_commit);
- uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks);
+ uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum);
bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs);
inline bool is_txn_coml_commit() const { return ::is_txn_coml_commit(&_deq_hdr); }
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.cpp?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.cpp Tue Dec 3 21:46:39 2013
@@ -23,6 +23,7 @@
#include <cassert>
#include <cstring>
+#include "qpid/linearstore/journal/Checksum.h"
#include "qpid/linearstore/journal/jexception.h"
namespace qpid {
@@ -62,7 +63,7 @@ enq_rec::reset(const uint64_t serial, co
}
uint32_t
-enq_rec::encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks)
+enq_rec::encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum)
{
assert(wptr != 0);
assert(max_size_dblks > 0);
@@ -102,8 +103,10 @@ enq_rec::encode(void* wptr, uint32_t rec
}
rec_offs -= _enq_hdr._dsize - wsize2;
}
+ checksum.addData((unsigned char*)wptr, wr_cnt);
if (rem)
{
+ _enq_tail._checksum = checksum.getChecksum();
wsize = sizeof(_enq_tail) > rec_offs ? sizeof(_enq_tail) - rec_offs : 0;
wsize2 = wsize;
if (wsize)
@@ -122,21 +125,25 @@ enq_rec::encode(void* wptr, uint32_t rec
else // No further split required
{
rec_offs -= sizeof(_enq_hdr);
- std::size_t wsize = _enq_hdr._xidsize > rec_offs ? _enq_hdr._xidsize - rec_offs : 0;
- if (wsize)
+ std::size_t xid_wsize = _enq_hdr._xidsize > rec_offs ? _enq_hdr._xidsize - rec_offs : 0;
+ if (xid_wsize)
{
- std::memcpy(wptr, (const char*)_xidp + rec_offs, wsize);
- wr_cnt += wsize;
+ std::memcpy(wptr, (const char*)_xidp + rec_offs, xid_wsize);
+ wr_cnt += xid_wsize;
}
- rec_offs -= _enq_hdr._xidsize - wsize;
- wsize = _enq_hdr._dsize > rec_offs ? _enq_hdr._dsize - rec_offs : 0;
- if (wsize && !::is_enq_external(&_enq_hdr))
+ rec_offs -= _enq_hdr._xidsize - xid_wsize;
+ std::size_t data_wsize = _enq_hdr._dsize > rec_offs ? _enq_hdr._dsize - rec_offs : 0;
+ if (data_wsize && !::is_enq_external(&_enq_hdr))
{
- std::memcpy((char*)wptr + wr_cnt, (const char*)_data + rec_offs, wsize);
- wr_cnt += wsize;
+ std::memcpy((char*)wptr + wr_cnt, (const char*)_data + rec_offs, data_wsize);
+ wr_cnt += data_wsize;
+ }
+ rec_offs -= _enq_hdr._dsize - data_wsize;
+ if (xid_wsize || data_wsize) {
+ checksum.addData((unsigned char*)wptr, wr_cnt);
}
- rec_offs -= _enq_hdr._dsize - wsize;
- wsize = sizeof(_enq_tail) > rec_offs ? sizeof(_enq_tail) - rec_offs : 0;
+ _enq_tail._checksum = checksum.getChecksum();
+ std::size_t wsize = sizeof(_enq_tail) > rec_offs ? sizeof(_enq_tail) - rec_offs : 0;
if (wsize)
{
std::memcpy((char*)wptr + wr_cnt, (char*)&_enq_tail + rec_offs, wsize);
@@ -174,8 +181,10 @@ enq_rec::encode(void* wptr, uint32_t rec
wr_cnt += wsize;
rem -= wsize;
}
+ checksum.addData((unsigned char*)wptr, wr_cnt);
if (rem)
{
+ _enq_tail._checksum = checksum.getChecksum();
wsize = rem >= sizeof(_enq_tail) ? sizeof(_enq_tail) : rem;
std::memcpy((char*)wptr + wr_cnt, (void*)&_enq_tail, wsize);
wr_cnt += wsize;
@@ -195,6 +204,8 @@ enq_rec::encode(void* wptr, uint32_t rec
std::memcpy((char*)wptr + wr_cnt, _data, _enq_hdr._dsize);
wr_cnt += _enq_hdr._dsize;
}
+ checksum.addData((unsigned char*)wptr, wr_cnt);
+ _enq_tail._checksum = checksum.getChecksum();
std::memcpy((char*)wptr + wr_cnt, (void*)&_enq_tail, sizeof(_enq_tail));
wr_cnt += sizeof(_enq_tail);
#ifdef QLS_CLEAN
@@ -209,15 +220,13 @@ enq_rec::encode(void* wptr, uint32_t rec
bool
enq_rec::decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs)
{
- uint32_t checksum = 0UL; // TODO: Add checksum math
if (rec_offs == 0)
{
// Read header, allocate (if req'd) for xid
- //_enq_hdr.hdr_copy(h);
::rec_hdr_copy(&_enq_hdr._rhdr, &h);
ifsp->read((char*)&_enq_hdr._xidsize, sizeof(_enq_hdr._xidsize));
ifsp->read((char*)&_enq_hdr._dsize, sizeof(_enq_hdr._dsize));
- rec_offs = sizeof(_enq_hdr);
+ rec_offs = sizeof(::enq_hdr_t);
if (_enq_hdr._xidsize > 0)
{
_buff = std::malloc(_enq_hdr._xidsize);
@@ -280,18 +289,6 @@ enq_rec::decode(::rec_hdr_t& h, std::ifs
}
ifsp->ignore(rec_size_dblks() * QLS_DBLK_SIZE_BYTES - rec_size());
assert(!ifsp->fail() && !ifsp->bad());
- int res = ::rec_tail_check(&_enq_tail, &_enq_hdr._rhdr, checksum);
- if (res != 0) {
- std::stringstream oss;
- switch (res) {
- case 1: oss << std::hex << "Magic: expected 0x" << ~_enq_hdr._rhdr._magic << "; found 0x" << _enq_tail._xmagic; break;
- case 2: oss << std::hex << "Serial: expected 0x" << _enq_hdr._rhdr._serial << "; found 0x" << _enq_tail._serial; break;
- case 3: oss << std::hex << "Record Id: expected 0x" << _enq_hdr._rhdr._rid << "; found 0x" << _enq_tail._rid; break;
- case 4: oss << std::hex << "Checksum: expected 0x" << checksum << "; found 0x" << _enq_tail._checksum; break;
- default: oss << "Unknown error " << res;
- }
- throw jexception(jerrno::JERR_JREC_BADRECTAIL, oss.str(), "enq_rec", "decode"); // TODO: Don't throw exception, log info
- }
return true;
}
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.h?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/enq_rec.h Tue Dec 3 21:46:39 2013
@@ -49,7 +49,7 @@ public:
void reset(const uint64_t serial, const uint64_t rid, const void* const dbuf, const std::size_t dlen,
const void* const xidp, const std::size_t xidlen, const bool transient, const bool external);
- uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks);
+ uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum);
bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs);
std::size_t get_xid(void** const xidpp);
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/jrec.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/jrec.h?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/jrec.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/jrec.h Tue Dec 3 21:46:39 2013
@@ -32,6 +32,8 @@ namespace qpid {
namespace linearstore {
namespace journal {
+class Checksum;
+
/**
* \class jrec
* \brief Abstract class for all file jrecords, both data and log. This class establishes
@@ -95,7 +97,7 @@ public:
* \param max_size_dblks Maximum number of data-blocks to write to pointer wptr.
* \returns Number of data-blocks encoded.
*/
- virtual uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks) = 0;
+ virtual uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum) = 0;
virtual bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs) = 0;
virtual std::string& str(std::string& str) const = 0;
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.cpp?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.cpp Tue Dec 3 21:46:39 2013
@@ -23,6 +23,7 @@
#include <cassert>
#include <cstring>
+#include "qpid/linearstore/journal/Checksum.h"
#include "qpid/linearstore/journal/jexception.h"
namespace qpid {
@@ -55,10 +56,11 @@ txn_rec::reset(const bool commitFlag, co
_txn_tail._xmagic = ~_txn_hdr._rhdr._magic;
_txn_tail._serial = serial;
_txn_tail._rid = rid;
+ _txn_tail._checksum = 0UL;
}
uint32_t
-txn_rec::encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks)
+txn_rec::encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum)
{
assert(wptr != 0);
assert(max_size_dblks > 0);
@@ -83,8 +85,10 @@ txn_rec::encode(void* wptr, uint32_t rec
rem -= wsize;
}
rec_offs -= _txn_hdr._xidsize - wsize2;
+ checksum.addData((unsigned char*)wptr, wr_cnt);
if (rem)
{
+ _txn_tail._checksum = checksum.getChecksum();
wsize = sizeof(_txn_tail) > rec_offs ? sizeof(_txn_tail) - rec_offs : 0;
wsize2 = wsize;
if (wsize)
@@ -108,8 +112,10 @@ txn_rec::encode(void* wptr, uint32_t rec
{
std::memcpy(wptr, (const char*)_xidp + rec_offs, wsize);
wr_cnt += wsize;
+ checksum.addData((unsigned char*)wptr, wr_cnt);
}
rec_offs -= _txn_hdr._xidsize - wsize;
+ _txn_tail._checksum = checksum.getChecksum();
wsize = sizeof(_txn_tail) > rec_offs ? sizeof(_txn_tail) - rec_offs : 0;
if (wsize)
{
@@ -141,8 +147,10 @@ txn_rec::encode(void* wptr, uint32_t rec
wr_cnt += wsize;
rem -= wsize;
}
+ checksum.addData((unsigned char*)wptr, wr_cnt);
if (rem)
{
+ _txn_tail._checksum = checksum.getChecksum();
wsize = rem >= sizeof(_txn_tail) ? sizeof(_txn_tail) : rem;
std::memcpy((char*)wptr + wr_cnt, (void*)&_txn_tail, wsize);
wr_cnt += wsize;
@@ -154,6 +162,8 @@ txn_rec::encode(void* wptr, uint32_t rec
{
std::memcpy((char*)wptr + wr_cnt, _xidp, _txn_hdr._xidsize);
wr_cnt += _txn_hdr._xidsize;
+ checksum.addData((unsigned char*)wptr, wr_cnt);
+ _txn_tail._checksum = checksum.getChecksum();
std::memcpy((char*)wptr + wr_cnt, (void*)&_txn_tail, sizeof(_txn_tail));
wr_cnt += sizeof(_txn_tail);
#ifdef QLS_CLEAN
@@ -168,14 +178,12 @@ txn_rec::encode(void* wptr, uint32_t rec
bool
txn_rec::decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs)
{
- uint32_t checksum = 0UL; // TODO: Add checksum math
if (rec_offs == 0)
{
// Read header, allocate for xid
- //_txn_hdr.hdr_copy(h);
::rec_hdr_copy(&_txn_hdr._rhdr, &h);
ifsp->read((char*)&_txn_hdr._xidsize, sizeof(_txn_hdr._xidsize));
- rec_offs = sizeof(txn_hdr_t);
+ rec_offs = sizeof(::txn_hdr_t);
_buff = std::malloc(_txn_hdr._xidsize);
MALLOC_CHK(_buff, "_buff", "txn_rec", "rcv_decode");
}
@@ -216,14 +224,19 @@ txn_rec::decode(::rec_hdr_t& h, std::ifs
throw jexception(jerrno::JERR_JREC_BADRECTAIL); // TODO: complete exception detail
}
assert(!ifsp->fail() && !ifsp->bad());
- int res = ::rec_tail_check(&_txn_tail, &_txn_hdr._rhdr, checksum);
+ assert(_txn_hdr._xidsize > 0);
+ Checksum checksum;
+ checksum.addData((unsigned char*)&_txn_hdr, sizeof(_txn_hdr));
+ checksum.addData((unsigned char*)_buff, _txn_hdr._xidsize);
+ uint32_t cs = checksum.getChecksum();
+ int res = ::rec_tail_check(&_txn_tail, &_txn_hdr._rhdr, cs);
if (res != 0) {
std::stringstream oss;
switch (res) {
case 1: oss << std::hex << "Magic: expected 0x" << ~_txn_hdr._rhdr._magic << "; found 0x" << _txn_tail._xmagic; break;
case 2: oss << std::hex << "Serial: expected 0x" << _txn_hdr._rhdr._serial << "; found 0x" << _txn_tail._serial; break;
case 3: oss << std::hex << "Record Id: expected 0x" << _txn_hdr._rhdr._rid << "; found 0x" << _txn_tail._rid; break;
- case 4: oss << std::hex << "Checksum: expected 0x" << checksum << "; found 0x" << _txn_tail._checksum; break;
+ case 4: oss << std::hex << "Checksum: expected 0x" << cs << "; found 0x" << _txn_tail._checksum; break;
default: oss << "Unknown error " << res;
}
throw jexception(jerrno::JERR_JREC_BADRECTAIL, oss.str(), "txn_rec", "decode"); // TODO: Don't throw exception, log info
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.h?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/txn_rec.h Tue Dec 3 21:46:39 2013
@@ -48,7 +48,7 @@ public:
void reset(const bool commitFlag, const uint64_t serial, const uint64_t rid, const void* const xidp,
const std::size_t xidlen);
- uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks);
+ uint32_t encode(void* wptr, uint32_t rec_offs_dblks, uint32_t max_size_dblks, Checksum& checksum);
bool decode(::rec_hdr_t& h, std::ifstream* ifsp, std::size_t& rec_offs);
std::size_t get_xid(void** const xidpp);
Modified: qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/wmgr.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/wmgr.cpp?rev=1547601&r1=1547600&r2=1547601&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/wmgr.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/linearstore/journal/wmgr.cpp Tue Dec 3 21:46:39 2013
@@ -23,6 +23,7 @@
#include <cassert>
#include "qpid/linearstore/journal/aio_callback.h"
+#include "qpid/linearstore/journal/Checksum.h"
#include "qpid/linearstore/journal/data_tok.h"
#include "qpid/linearstore/journal/jcntl.h"
#include "qpid/linearstore/journal/JournalFile.h"
@@ -158,6 +159,7 @@ wmgr::enqueue(const void* const data_buf
}
//std::cout << "---+++ wmgr::enqueue() ENQ rid=0x" << std::hex << rid << " po=0x" << _pg_offset_dblks << " cs=0x" << (_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) << " " << std::dec << std::flush; // DEBUG
bool done = false;
+ Checksum checksum;
while (!done)
{
//std::cout << "*" << std::flush; // DEBUG
@@ -165,7 +167,7 @@ wmgr::enqueue(const void* const data_buf
void* wptr = (void*)((char*)_page_ptr_arr[_pg_index] + _pg_offset_dblks * QLS_DBLK_SIZE_BYTES);
uint32_t data_offs_dblks = dtokp->dblocks_written();
uint32_t ret = _enq_rec.encode(wptr, data_offs_dblks,
- (_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) - _pg_offset_dblks);
+ (_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) - _pg_offset_dblks, checksum);
// Remember fid which contains the record header in case record is split over several files
if (data_offs_dblks == 0) {
@@ -278,6 +280,7 @@ wmgr::dequeue(data_tok* dtokp,
}
//std::cout << "---+++ wmgr::dequeue() DEQ rid=0x" << std::hex << rid << " drid=0x" << dequeue_rid << " " << std::dec << std::flush; // DEBUG
bool done = false;
+ Checksum checksum;
while (!done)
{
//std::cout << "*" << std::flush; // DEBUG
@@ -285,7 +288,7 @@ wmgr::dequeue(data_tok* dtokp,
void* wptr = (void*)((char*)_page_ptr_arr[_pg_index] + _pg_offset_dblks * QLS_DBLK_SIZE_BYTES);
uint32_t data_offs_dblks = dtokp->dblocks_written();
uint32_t ret = _deq_rec.encode(wptr, data_offs_dblks,
- (_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) - _pg_offset_dblks);
+ (_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) - _pg_offset_dblks, checksum);
// Remember fid which contains the record header in case record is split over several files
if (data_offs_dblks == 0) {
@@ -396,13 +399,14 @@ wmgr::abort(data_tok* dtokp,
_abort_busy = true;
}
bool done = false;
+ Checksum checksum;
while (!done)
{
assert(_pg_offset_dblks < _cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS);
void* wptr = (void*)((char*)_page_ptr_arr[_pg_index] + _pg_offset_dblks * QLS_DBLK_SIZE_BYTES);
uint32_t data_offs_dblks = dtokp->dblocks_written();
uint32_t ret = _txn_rec.encode(wptr, data_offs_dblks,
- (_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) - _pg_offset_dblks);
+ (_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) - _pg_offset_dblks, checksum);
// Remember fid which contains the record header in case record is split over several files
if (data_offs_dblks == 0)
@@ -494,13 +498,14 @@ wmgr::commit(data_tok* dtokp,
_commit_busy = true;
}
bool done = false;
+ Checksum checksum;
while (!done)
{
assert(_pg_offset_dblks < _cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS);
void* wptr = (void*)((char*)_page_ptr_arr[_pg_index] + _pg_offset_dblks * QLS_DBLK_SIZE_BYTES);
uint32_t data_offs_dblks = dtokp->dblocks_written();
uint32_t ret = _txn_rec.encode(wptr, data_offs_dblks,
- (_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) - _pg_offset_dblks);
+ (_cache_pgsize_sblks * QLS_SBLK_SIZE_DBLKS) - _pg_offset_dblks, checksum);
// Remember fid which contains the record header in case record is split over several files
if (data_offs_dblks == 0)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org