9:05 a.m.
Author: cctrieloff
Date: 2007-11-07 10:05:52 -0500 (Wed, 07 Nov 2007)
New Revision: 1258
Modified:
store/trunk/cpp/lib/BdbMessageStore.cpp
store/trunk/cpp/lib/BdbMessageStore.h
Log:
- staging support
- no staged enqueue, content released, then loaded not yet supported on async
mode
Modified: store/trunk/cpp/lib/BdbMessageStore.cpp
===================================================================
--- store/trunk/cpp/lib/BdbMessageStore.cpp 2007-11-06 21:52:47 UTC (rev 1257)
+++ store/trunk/cpp/lib/BdbMessageStore.cpp 2007-11-07 15:05:52 UTC (rev 1258)
@@ -492,16 +492,22 @@
{
case rhm::journal::RHM_IORES_SUCCESS:{
msg_count++;
- char* data = (char*)dbuff;
- unsigned headerSize = Buffer(data, preambleLength).getLong();
- Buffer headerBuff(data+ preambleLength, headerSize); /// do we want
read size or header size ????
-
- RecoverableMessage::shared_ptr msg = recovery.recoverMessage(headerBuff);
+ RecoverableMessage::shared_ptr msg;
+ char* data = (char*)dbuff;
+
+ unsigned headerSize;
+ if (externalFlag){
+ msg = getExternMessage(recovery, dtokp.rid(), headerSize); // large message
external to jrnl
+ } else {
+ headerSize = Buffer(data, preambleLength).getLong();
+ Buffer headerBuff(data+ preambleLength, headerSize); /// do we
want read size or header size ????
+ msg = recovery.recoverMessage(headerBuff);
+ }
msg->setPersistenceId(dtokp.rid());
u_int32_t contentOffset = headerSize + preambleLength;
u_int64_t contentSize = readSize - contentOffset;
- if (msg->loadContent(contentSize)) {
+ if (msg->loadContent(contentSize) && !externalFlag) {
//now read the content
Buffer contentBuff(data + contentOffset, contentSize);
msg->decodeContent(contentBuff);
@@ -545,6 +551,29 @@
}
+RecoverableMessage::shared_ptr
BdbMessageStore::getExternMessage(qpid::broker::RecoveryManager& recovery,
+ uint64_t messageId, unsigned& headerSize)
+{
+ Dbt key (&messageId, sizeof(messageId));
+ size_t preamble_length = sizeof(u_int32_t)/*header size*/;
+
+ BufferValue value(preamble_length, 0);
+ value.buffer.record();
+ if (messageDb.get(0, &key, &value, 0) == DB_NOTFOUND) {
+ THROW_STORE_EXCEPTION("Cannot load content. Message not known to
store!");
+ }
+ //read header only to begin with
+ headerSize = value.buffer.getLong();
+
+ BufferValue header(headerSize, preamble_length);
+ if (messageDb.get(0, &key, &header, 0) == DB_NOTFOUND) {
+ THROW_STORE_EXCEPTION("Cannot load content. Message not known to
store!");
+ }
+
+ return recovery.recoverMessage(header.buffer);
+}
+
+
// bdb version
void BdbMessageStore::recoverMessages(TxnCtxt& txn, RecoveryManager& recovery,
queue_index& index,
txn_list& locked, message_index& prepared)
@@ -700,13 +729,13 @@
txn.begin(env);
u_int64_t messageId (msg.getPersistenceId());
- if (messageId == 0) {
+ if (messageId == 0 || !msg.isContentReleased()) {
try {
Dbt key (&messageId, sizeof(messageId));
messageId = messageIdSequence.next();
store(NULL, &txn, key, msg, true);
msg.setPersistenceId(messageId);
- txn.commit();
+ txn.commit();
} catch (std::exception& e) {
txn.abort();
throw e;
@@ -813,8 +842,6 @@
JournalImpl* jc = static_cast<JournalImpl*>(queue.getExternalQueueStore());
if (jc){
jc->flush();
-// ::usleep(10000); /////////////// hack ----------- FIX!!
-// jc->get_wr_events();
}
}catch ( journal::jexception& e) {
THROW_STORE_EXCEPTION("Flush failed: " + e.to_string() );
@@ -854,6 +881,7 @@
if (usingJrnl()){
// add queue* to the txn map..
if (ctxt) txn->addXidRecord(queue.getExternalQueueStore());
+ if (msg.isContentReleased()) put(mappingDb, txn->get(), key, value); // TODO -
remove once jrnl is used for transient policy see **
}else{
msg.enqueueComplete(); // set enqueued for ack
put(mappingDb, txn->get(), key, value);
@@ -879,10 +907,14 @@
{
u_int32_t headerSize = message.encodedHeaderSize();
u_int64_t size = message.encodedSize() + sizeof(u_int32_t);
- char* buff = static_cast<char*>(::alloca(size)); // long + headers + content
- Buffer buffer(buff,size);
- buffer.putLong(headerSize);
- message.encode(buffer);
+ char* buff= 0;
+ if (!message.isContentReleased() )
+ {
+ buff = static_cast<char*>(::alloca(size)); // long + headers + content
+ Buffer buffer(buff,size);
+ buffer.putLong(headerSize);
+ message.encode(buffer);
+ }
try {
@@ -898,9 +930,17 @@
JournalImpl* jc =
static_cast<JournalImpl*>(queue->getExternalQueueStore());
rhm::journal::iores eres;
if (txn->getXid().empty()){
- eres = jc->enqueue_data_record(buff, size, size, dtokp.get(), false);
+ if (message.isContentReleased()){
+ eres = jc->enqueue_extern_data_record(0, dtokp.get(), false);
+ }else {
+ eres = jc->enqueue_data_record(buff, size, size, dtokp.get(), false);
+ }
}else {
- eres = jc->enqueue_txn_data_record(buff, size, size, dtokp.get(),
txn->getXid(), false);
+ if (message.isContentReleased()){
+ eres = jc->enqueue_extern_txn_data_record(0, dtokp.get(), txn->getXid(),
false);
+ } else {
+ eres = jc->enqueue_txn_data_record(buff, size, size, dtokp.get(),
txn->getXid(), false);
+ }
}
switch (eres)
{
@@ -963,6 +1003,13 @@
if (ctxt) txn->addXidRecord(queue.getExternalQueueStore());
async_dequeue(ctxt, msg, queue);
// added here as we are not doing it async on call back
+ if (msg.isContentReleased()) // TODO remove this code once jrnl is used for transient
policy see **
+ {
+ Dbt key (&messageId, sizeof(messageId));
+ Dbt value (&queueId, sizeof(queueId));
+ dequeue(txn->get(), key, value);
+ }
+
msg.dequeueComplete();
if ( msg.isDequeueComplete() ) // clear id after last dequeue
msg.setPersistenceId(0);
Modified: store/trunk/cpp/lib/BdbMessageStore.h
===================================================================
--- store/trunk/cpp/lib/BdbMessageStore.h 2007-11-06 21:52:47 UTC (rev 1257)
+++ store/trunk/cpp/lib/BdbMessageStore.h 2007-11-07 15:05:52 UTC (rev 1258)
@@ -88,6 +88,8 @@
void recoverMessages(TxnCtxt& txn, qpid::broker::RecoveryManager&
recovery,
qpid::broker::RecoverableQueue::shared_ptr& queue,
txn_list& locked, message_index& prepared);
+ qpid::broker::RecoverableMessage::shared_ptr
getExternMessage(qpid::broker::RecoveryManager& recovery,
+ uint64_t mId, unsigned& headerSize);
void recoverExchanges(TxnCtxt& txn, qpid::broker::RecoveryManager&
recovery, exchange_index& index);
void recoverBindings(TxnCtxt& txn, exchange_index& exchanges,
queue_index& queues);
int enqueueMessage(TxnCtxt& txn, IdDbt& msgId,
qpid::broker::RecoverableMessage::shared_ptr& msg,