3:13 p.m.
Author: kpvdr
Date: 2010-12-01 09:13:40 -0500 (Wed, 01 Dec 2010)
New Revision: 4418
Modified:
store/trunk/cpp/lib/MessageStoreImpl.cpp
store/trunk/cpp/lib/MessageStoreImpl.h
Log:
Fix for BZ656385 "Data store can become corrupt with small store file size and large
wcache-page-size". This is an undetected illegal combination of parameters, allowing
the write cache to be larger than the journal file. Fixed by adding detection for this
condition.
Modified: store/trunk/cpp/lib/MessageStoreImpl.cpp
===================================================================
--- store/trunk/cpp/lib/MessageStoreImpl.cpp 2010-11-29 21:04:56 UTC (rev 4417)
+++ store/trunk/cpp/lib/MessageStoreImpl.cpp 2010-12-01 14:13:40 UTC (rev 4418)
@@ -96,7 +96,7 @@
return p;
}
-u_int32_t MessageStoreImpl::chkJrnlFileSizeParam(const u_int32_t param, const std::string
paramName)
+u_int32_t MessageStoreImpl::chkJrnlFileSizeParam(const u_int32_t param, const std::string
paramName, const u_int32_t wCachePgSizeSblks)
{
u_int32_t p = param;
u_int32_t min = JRNL_MIN_FILE_SIZE / JRNL_RMGR_PAGE_SIZE;
@@ -108,10 +108,15 @@
p = max;
QPID_LOG(warning, "parameter " << paramName << "
(" << param << ") is above allowable maximum (" << max
<< "); changing this parameter to maximum value.");
}
+ if (wCachePgSizeSblks > p * JRNL_RMGR_PAGE_SIZE) {
+ std::ostringstream oss;
+ oss << "Cannot create store with file size less than write page cache
size. [file size = " << p << " (" << (p *
JRNL_RMGR_PAGE_SIZE / 2) << " kB); write page cache = " <<
(wCachePgSizeSblks / 2) << " kB]";
+ THROW_STORE_EXCEPTION(oss.str());
+ }
return p;
}
-u_int32_t MessageStoreImpl::chkJrnlWrPageCacheSize(const u_int32_t param, const
std::string paramName)
+u_int32_t MessageStoreImpl::chkJrnlWrPageCacheSize(const u_int32_t param, const
std::string paramName, const u_int16_t jrnlFsizePgs)
{
u_int32_t p = param;
switch (p)
@@ -124,6 +129,10 @@
case 32:
case 64:
case 128:
+ if (jrnlFsizePgs == 1) {
+ p = 64;
+ QPID_LOG(warning, "parameter " << paramName << "
(" << param << ") cannot set a page size greater than the journal
file size; changing this parameter to the journal file size (" << p <<
")");
+ }
break;
default:
if (p == 0) {
@@ -245,10 +254,10 @@
const StoreOptions* opts = static_cast<const StoreOptions*>(options);
u_int16_t numJrnlFiles = chkJrnlNumFilesParam(opts->numJrnlFiles,
"num-jfiles");
u_int32_t jrnlFsizePgs = chkJrnlFileSizeParam(opts->jrnlFsizePgs,
"jfile-size-pgs");
- u_int32_t jrnlWrCachePageSizeKib = chkJrnlWrPageCacheSize(opts->wCachePageSizeKib,
"wcache-page-size");
+ u_int32_t jrnlWrCachePageSizeKib = chkJrnlWrPageCacheSize(opts->wCachePageSizeKib,
"wcache-page-size", jrnlFsizePgs);
u_int16_t tplNumJrnlFiles = chkJrnlNumFilesParam(opts->tplNumJrnlFiles,
"tpl-num-jfiles");
u_int32_t tplJrnlFSizePgs = chkJrnlFileSizeParam(opts->tplJrnlFsizePgs,
"tpl-jfile-size-pgs");
- u_int32_t tplJrnlWrCachePageSizeKib =
chkJrnlWrPageCacheSize(opts->tplWCachePageSizeKib, "tpl-wcache-page-size");
+ u_int32_t tplJrnlWrCachePageSizeKib =
chkJrnlWrPageCacheSize(opts->tplWCachePageSizeKib, "tpl-wcache-page-size",
tplJrnlFSizePgs);
bool autoJrnlExpand;
u_int16_t autoJrnlExpandMaxFiles;
chkJrnlAutoExpandOptions(opts, autoJrnlExpand, autoJrnlExpandMaxFiles,
"auto-expand-max-jfiles", numJrnlFiles, "num-jfiles");
@@ -496,7 +505,7 @@
value = args.get("qpid.file_size");
if (value.get() != 0 && !value->empty() &&
value->convertsTo<int>())
- localFileSizeSblks = chkJrnlFileSizeParam((u_int32_t) value->get<int>(),
"qpid.file_size") * JRNL_RMGR_PAGE_SIZE;
+ localFileSizeSblks = chkJrnlFileSizeParam((u_int32_t) value->get<int>(),
"qpid.file_size", wCachePgSizeSblks) * JRNL_RMGR_PAGE_SIZE;
if (queue.getName().size() == 0)
{
Modified: store/trunk/cpp/lib/MessageStoreImpl.h
===================================================================
--- store/trunk/cpp/lib/MessageStoreImpl.h 2010-11-29 21:04:56 UTC (rev 4417)
+++ store/trunk/cpp/lib/MessageStoreImpl.h 2010-12-01 14:13:40 UTC (rev 4418)
@@ -161,9 +161,11 @@
static u_int16_t chkJrnlNumFilesParam(const u_int16_t param,
const std::string paramName);
static u_int32_t chkJrnlFileSizeParam(const u_int32_t param,
- const std::string paramName);
+ const std::string paramName,
+ const u_int32_t wCachePgSizeSblks = 0);
static u_int32_t chkJrnlWrPageCacheSize(const u_int32_t param,
- const std::string paramName);
+ const std::string paramName,
+ const u_int16_t jrnlFsizePgs);
static u_int16_t getJrnlWrNumPages(const u_int32_t wrPageSizeKib);
void chkJrnlAutoExpandOptions(const MessageStoreImpl::StoreOptions* opts,
bool& autoJrnlExpand,