3:28 p.m.
Author: kpvdr
Date: 2008-07-30 16:28:39 -0400 (Wed, 30 Jul 2008)
New Revision: 2238
Modified:
store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp
store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h
Log:
Backports of trunk r.2236,2237: Improvements on fix for BZ457239, also tightening up on
exception handling around transactions in BdbMessageStore. Also added txn to one addtional
call omitted on previouis fixes.
Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp
===================================================================
--- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp 2008-07-30 20:09:46 UTC (rev 2237)
+++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.cpp 2008-07-30 20:28:39 UTC (rev 2238)
@@ -249,7 +249,7 @@
} catch (const DbException& e) {
txn.abort();
THROW_STORE_EXCEPTION_2("Error opening databases", e);
- } catch (const std::exception& e) {
+ } catch (...) {
txn.abort();
throw;
}
@@ -359,7 +359,6 @@
THROW_STORE_EXCEPTION(std::string("Queue ") + queue.getName() + ":
create() failed: " + e.what());
}
try {
- qpid::sys::Mutex::ScopedLock sl(bdbQueueLock);
if (!create(queueDb, queueIdSequence, queue)) {
THROW_STORE_EXCEPTION("Queue already exists: " + queue.getName());
}
@@ -435,7 +434,16 @@
Dbt key(&id, sizeof(id));
BufferValue value (p);
- int status = db.put(0, &key, &value, DB_NOOVERWRITE | DB_AUTO_COMMIT);
+ int status;
+ TxnCtxt txn;
+ txn.begin(env, true);
+ try {
+ status = db.put(txn.get(), &key, &value, DB_NOOVERWRITE);
+ txn.commit();
+ } catch (...) {
+ txn.abort();
+ throw;
+ }
if (status == DB_KEYEXIST) {
return false;
} else {
@@ -461,8 +469,13 @@
BindingDbt value(e, q, k, a);
TxnCtxt txn;
txn.begin(env, true);
- put(bindingDb, txn.get(), key, value);
- txn.commit();
+ try {
+ put(bindingDb, txn.get(), key, value);
+ txn.commit();
+ } catch (...) {
+ txn.abort();
+ throw;
+ }
}
void BdbMessageStore::unbind(const PersistableExchange& e,
@@ -498,10 +511,12 @@
recoverGeneral(txn, registry);
txn.commit();
-
} catch (const DbException& e) {
txn.abort();
THROW_STORE_EXCEPTION_2("Error on recovery", e);
+ } catch (...) {
+ txn.abort();
+ throw;
}
//recover transactions:
@@ -785,7 +800,7 @@
value.buffer.record();
TxnCtxt txn;
- txn.begin(env);
+ txn.begin(env, true);
try {
if (messageDb.get(txn.get(), &key, &value, 0) == DB_NOTFOUND) {
txn.abort();
@@ -804,6 +819,9 @@
} catch (const DbException& e) {
txn.abort();
THROW_STORE_EXCEPTION("Unexpected BDB error in
BdbMessageStore::getExternMessage(): " + string(e.what()));
+ } catch (...) {
+ txn.abort();
+ throw;
}
return ret;
}
@@ -922,7 +940,7 @@
store(NULL, &txn, key, msg, true);
msg->setPersistenceId(messageId);
txn.commit();
- } catch (const std::exception& e) {
+ } catch (...) {
txn.abort();
throw;
}
@@ -943,6 +961,9 @@
} catch (const DbException& e) {
txn.abort();
THROW_STORE_EXCEPTION_2("Error destroying message", e);
+ } catch (...) {
+ txn.abort();
+ throw;
}
}
}
@@ -950,7 +971,17 @@
u_int64_t BdbMessageStore::getRecordSize(Db& db,
Dbt& key)
{
- return getRecordSize(0, db, key);
+ u_int64_t ret = 0;
+ TxnCtxt txn;
+ txn.begin(env, true);
+ try {
+ ret = getRecordSize(txn.get(), db, key);
+ txn.commit();
+ } catch (...) {
+ txn.abort();
+ throw;
+ }
+ return ret;
}
u_int64_t BdbMessageStore::getRecordSize(DbTxn* txn,
@@ -977,6 +1008,8 @@
checkInit();
u_int64_t messageId (msg->getPersistenceId());
if (messageId != 0) {
+ TxnCtxt txn;
+ txn.begin(env, true);
try {
Dbt key (&messageId, sizeof(messageId));
u_int64_t offset = getRecordSize(messageDb, key);
@@ -992,9 +1025,14 @@
value.set_flags(DB_DBT_USERMEM | DB_DBT_PARTIAL);
value.set_doff(offset);
value.set_dlen(size);
- messageDb.put(0, &key, &value, DB_AUTO_COMMIT);
+ messageDb.put(txn.get(), &key, &value, 0);
+ txn.commit();
} catch (const DbException& e) {
+ txn.abort();
THROW_STORE_EXCEPTION_2("Error appending content", e);
+ } catch (...) {
+ txn.abort();
+ throw;
}
} else {
THROW_STORE_EXCEPTION("Cannot append content. Message not known to
store!");
@@ -1035,16 +1073,18 @@
value.set_dlen(length);
int status = messageDb.get(txn.get(), &key, &value, 0);
if (status == DB_NOTFOUND) {
- txn.abort();
delete [] buffer;
THROW_STORE_EXCEPTION("Cannot load content. Message not known to
store!");
- } else {
- txn.commit();
- data.assign(buffer, value.get_size());
- delete [] buffer;
}
+ data.assign(buffer, value.get_size());
+ delete [] buffer;
+ txn.commit();
} catch (const DbException& e) {
+ txn.abort();
THROW_STORE_EXCEPTION_2("Error loading content", e);
+ } catch (...) {
+ txn.abort();
+ throw;
}
} else {
THROW_STORE_EXCEPTION("Cannot load content. Message not known to
store!");
@@ -1378,26 +1418,32 @@
TxnCtxt txn;
txn.begin(env, true);
try {
- Cursor bindings;
- bindings.open(bindingDb, txn.get());
+ {
+ Cursor bindings;
+ bindings.open(bindingDb, txn.get());
- IdDbt key;
- Dbt value;
- while (bindings.next(key, value)) {
- Buffer buffer(reinterpret_cast<char*>(value.get_data()),
value.get_size());
- if (buffer.available() < 8) {
- THROW_STORE_EXCEPTION("Not enough data for binding");
+ IdDbt key;
+ Dbt value;
+ while (bindings.next(key, value)) {
+ Buffer buffer(reinterpret_cast<char*>(value.get_data()),
value.get_size());
+ if (buffer.available() < 8) {
+ THROW_STORE_EXCEPTION("Not enough data for binding");
+ }
+ uint64_t queueId = buffer.getLongLong();
+ if (queue.getPersistenceId() == queueId) {
+ bindings->del(0);
+ QPID_LOG(debug, "Deleting binding for " <<
queue.getName() << " " << key.id << "->" <<
queueId);
+ }
}
- uint64_t queueId = buffer.getLongLong();
- if (queue.getPersistenceId() == queueId) {
- bindings->del(0);
- QPID_LOG(debug, "Deleting binding for " <<
queue.getName() << " " << key.id << "->" <<
queueId);
- }
}
+ txn.commit();
} catch (const std::exception& e) {
+ txn.abort();
THROW_STORE_EXCEPTION_2("Error deleting bindings", e.what());
+ } catch (...) {
+ txn.abort();
+ throw;
}
- txn.commit();
QPID_LOG(debug, "Deleted all bindings for " << queue.getName()
<< ":" << queue.getPersistenceId());
}
@@ -1408,33 +1454,39 @@
TxnCtxt txn;
txn.begin(env, true);
try {
- Cursor bindings;
- bindings.open(bindingDb, txn.get());
+ {
+ Cursor bindings;
+ bindings.open(bindingDb, txn.get());
- IdDbt key(exchange.getPersistenceId());
- Dbt value;
+ IdDbt key(exchange.getPersistenceId());
+ Dbt value;
- for (int status = bindings->get(&key, &value, DB_SET); status == 0;
status = bindings->get(&key, &value, DB_NEXT_DUP)) {
- Buffer buffer(reinterpret_cast<char*>(value.get_data()),
value.get_size());
- if (buffer.available() < 8) {
- THROW_STORE_EXCEPTION("Not enough data for binding");
- }
- uint64_t queueId = buffer.getLongLong();
- if (queue.getPersistenceId() == queueId) {
- std::string q;
- std::string k;
- buffer.getShortString(q);
- buffer.getShortString(k);
- if (bkey == k) {
- bindings->del(0);
- QPID_LOG(debug, "Deleting binding for " <<
queue.getName() << " " << key.id << "->" <<
queueId);
+ for (int status = bindings->get(&key, &value, DB_SET); status ==
0; status = bindings->get(&key, &value, DB_NEXT_DUP)) {
+ Buffer buffer(reinterpret_cast<char*>(value.get_data()),
value.get_size());
+ if (buffer.available() < 8) {
+ THROW_STORE_EXCEPTION("Not enough data for binding");
}
+ uint64_t queueId = buffer.getLongLong();
+ if (queue.getPersistenceId() == queueId) {
+ std::string q;
+ std::string k;
+ buffer.getShortString(q);
+ buffer.getShortString(k);
+ if (bkey == k) {
+ bindings->del(0);
+ QPID_LOG(debug, "Deleting binding for " <<
queue.getName() << " " << key.id << "->" <<
queueId);
+ }
+ }
}
}
+ txn.commit();
} catch (const std::exception& e) {
+ txn.abort();
THROW_STORE_EXCEPTION_2("Error deleting bindings", e.what());
+ } catch (...) {
+ txn.abort();
+ throw;
}
- txn.commit();
}
string BdbMessageStore::getJrnlBaseDir()
Modified: store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h
===================================================================
--- store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h 2008-07-30 20:09:46 UTC (rev 2237)
+++ store/branches/mrg-1.0/cpp/lib/BdbMessageStore.h 2008-07-30 20:28:39 UTC (rev 2238)
@@ -115,7 +115,6 @@
static qpid::sys::Duration defJournalFlushTimeout;
qpid::management::Store::shared_ptr mgmtObject;
qpid::sys::Mutex jrnlCreateLock;
- qpid::sys::Mutex bdbQueueLock;
// Parameter validation and calculation
static u_int16_t chkJrnlNumFilesParam(const u_int16_t param,