You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by gs...@apache.org on 2013/04/29 20:04:18 UTC
svn commit: r1477236 - in /qpid/trunk/qpid/cpp/src: ./ qpid/broker/
qpid/broker/amqp_0_10/ qpid/sys/ qpid/sys/posix/ qpid/sys/windows/ tests/
Author: gsim
Date: Mon Apr 29 18:04:17 2013
New Revision: 1477236
URL: http://svn.apache.org/r1477236
Log:
QPID-4339: simple paged queue implementation
Added:
qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.h
qpid/trunk/qpid/cpp/src/qpid/sys/MemoryMappedFile.h
qpid/trunk/qpid/cpp/src/qpid/sys/posix/MemoryMappedFile.cpp
qpid/trunk/qpid/cpp/src/qpid/sys/windows/MemoryMappedFile.cpp
qpid/trunk/qpid/cpp/src/tests/run_paged_queue_tests (with props)
Modified:
qpid/trunk/qpid/cpp/src/CMakeLists.txt
qpid/trunk/qpid/cpp/src/Makefile.am
qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.h
qpid/trunk/qpid/cpp/src/qpid/broker/QueueCursor.h
qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h
qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessage.h
qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessageImpl.h
qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp
qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.h
qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt
qpid/trunk/qpid/cpp/src/tests/Makefile.am
qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp
Modified: qpid/trunk/qpid/cpp/src/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/CMakeLists.txt?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/CMakeLists.txt (original)
+++ qpid/trunk/qpid/cpp/src/CMakeLists.txt Mon Apr 29 18:04:17 2013
@@ -748,6 +748,7 @@ if (CMAKE_SYSTEM_NAME STREQUAL Windows)
qpid/sys/windows/IocpPoller.cpp
qpid/sys/windows/IOHandle.cpp
qpid/sys/windows/LockFile.cpp
+ qpid/sys/windows/MemoryMappedFile.cpp
qpid/sys/windows/PipeHandle.cpp
qpid/sys/windows/PollableCondition.cpp
qpid/sys/windows/Shlib.cpp
@@ -849,6 +850,7 @@ else (CMAKE_SYSTEM_NAME STREQUAL Windows
qpid/sys/posix/FileSysDir.cpp
qpid/sys/posix/IOHandle.cpp
qpid/sys/posix/LockFile.cpp
+ qpid/sys/posix/MemoryMappedFile.cpp
qpid/sys/posix/Mutex.cpp
qpid/sys/posix/PipeHandle.cpp
qpid/sys/posix/PollableCondition.cpp
@@ -1196,6 +1198,7 @@ set (qpidbroker_SOURCES
qpid/broker/MessageStoreModule.cpp
qpid/broker/NameGenerator.cpp
qpid/broker/NullMessageStore.cpp
+ qpid/broker/PagedQueue.cpp
qpid/broker/QueueBindings.cpp
qpid/broker/QueuedMessage.cpp
qpid/broker/QueueCursor.cpp
Modified: qpid/trunk/qpid/cpp/src/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/Makefile.am?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/Makefile.am Mon Apr 29 18:04:17 2013
@@ -39,6 +39,7 @@ windows_dist = \
qpid/sys/windows/IOHandle.cpp \
qpid/sys/windows/IoHandlePrivate.h \
qpid/sys/windows/LockFile.cpp \
+ qpid/sys/windows/MemoryMappedFile.cpp \
qpid/sys/windows/mingw32_compat.h \
qpid/sys/windows/PollableCondition.cpp \
qpid/sys/windows/PipeHandle.cpp \
@@ -167,6 +168,7 @@ libqpidcommon_la_SOURCES += \
qpid/sys/posix/AsynchIO.cpp \
qpid/sys/posix/FileSysDir.cpp \
qpid/sys/posix/LockFile.cpp \
+ qpid/sys/posix/MemoryMappedFile.cpp \
qpid/sys/posix/Time.cpp \
qpid/sys/posix/Thread.cpp \
qpid/sys/posix/Shlib.cpp \
@@ -492,6 +494,7 @@ libqpidcommon_la_SOURCES += \
qpid/sys/Fork.h \
qpid/sys/LockFile.h \
qpid/sys/LockPtr.h \
+ qpid/sys/MemoryMappedFile.h \
qpid/sys/MemStat.h \
qpid/sys/OutputControl.h \
qpid/sys/OutputTask.h \
@@ -654,6 +657,8 @@ libqpidbroker_la_SOURCES = \
qpid/broker/MessageStore.h \
qpid/broker/MessageStoreModule.cpp \
qpid/broker/MessageStoreModule.h \
+ qpid/broker/PagedQueue.h \
+ qpid/broker/PagedQueue.cpp \
qpid/broker/PriorityQueue.h \
qpid/broker/PriorityQueue.cpp \
qpid/broker/Protocol.h \
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Message.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Message.h?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Message.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Message.h Mon Apr 29 18:04:17 2013
@@ -145,9 +145,6 @@ public:
void annotationsChanged();
};
-QPID_BROKER_EXTERN void encode(const Message&, std::string&);
-QPID_BROKER_EXTERN void decode(const std::string&, Message&);
-
}}
Added: qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.cpp?rev=1477236&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.cpp Mon Apr 29 18:04:17 2013
@@ -0,0 +1,425 @@
+/*
+ *
+ * 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/broker/PagedQueue.h"
+#include "qpid/broker/Protocol.h"
+#include "qpid/broker/QueueCursor.h"
+#include "qpid/broker/Message.h"
+#include "qpid/log/Statement.h"
+#include <string.h>
+
+namespace qpid {
+namespace broker {
+namespace {
+using qpid::sys::MemoryMappedFile;
+
+size_t encodedSize(const Message& msg)
+{
+ return msg.getPersistentContext()->encodedSize() + 4/*content-size*/ + 4/*sequence-number*/;
+}
+
+size_t encode(const Message& msg, char* data, size_t size)
+{
+ uint32_t encoded = msg.getPersistentContext()->encodedSize();
+ uint32_t required = encoded + 4/*content-size*/ + 4/*sequence-number*/;
+ if (required > size) return 0;
+ qpid::framing::Buffer buffer(data, required);
+ buffer.putLong(encoded);
+ buffer.putLong(msg.getSequence());
+ msg.getPersistentContext()->encode(buffer);
+ assert(buffer.getPosition() == required);
+ return required;
+}
+
+size_t decode(ProtocolRegistry& protocols, Message& msg, const char* data, size_t size)
+{
+ qpid::framing::Buffer metadata(const_cast<char*>(data), size);
+ uint32_t encoded = metadata.getLong();
+ uint32_t sequence = metadata.getLong();
+ assert(metadata.available() >= encoded);
+ qpid::framing::Buffer buffer(const_cast<char*>(data) + metadata.getPosition(), encoded);
+ msg = protocols.decode(buffer);
+ assert(buffer.getPosition() == encoded);
+ msg.setSequence(qpid::framing::SequenceNumber(sequence));
+ return encoded + metadata.getPosition();
+}
+
+}
+
+PagedQueue::PagedQueue(const std::string& name, const std::string& directory, uint m, uint factor, ProtocolRegistry& p)
+ : pageSize(file.getPageSize()*factor), maxLoaded(m), protocols(p), offset(0)
+{
+ path = file.open(name, directory);
+ QPID_LOG(debug, "PagedQueue[" << path << "]");
+}
+
+size_t PagedQueue::size()
+{
+ size_t total(0);
+ for (Used::const_iterator i = used.begin(); i != used.end(); ++i) {
+ total += i->second.available();
+ }
+ return total;
+}
+
+bool PagedQueue::deleted(const QueueCursor& cursor)
+{
+ if (cursor.valid) {
+ Used::iterator page = findPage(cursor.position, false);
+ if (page == used.end()) {
+ return false;
+ }
+ page->second.deleted(cursor.position);
+ if (page->second.empty()) {
+ //move page to free list
+ --loaded;
+ page->second.clear(file);
+ free.push_back(page->second);
+ used.erase(page);
+ }
+ return true;
+ } else {
+ return false;
+ }
+}
+
+void PagedQueue::publish(const Message& added)
+{
+ if (encodedSize(added) > pageSize) {
+ throw qpid::Exception(QPID_MSG("Message is larger than page size for queue backed by " << path));
+ }
+ Used::reverse_iterator i = used.rbegin();
+ if (i != used.rend()) {
+ if (!i->second.isLoaded()) load(i->second);
+ if (i->second.add(added)) return;
+ }
+ //used is empty or last page is full, need to add a new page
+ if (!newPage(added.getSequence()).add(added)) {
+ throw qpid::Exception(QPID_MSG("Could not add message to paged queue backed by " << path));
+ }
+}
+
+Message* PagedQueue::next(QueueCursor& cursor)
+{
+ Used::iterator i = used.begin();
+ if (cursor.valid) {
+ qpid::framing::SequenceNumber position(cursor.position);
+ ++position;
+ i = findPage(position, true);
+ if (i == used.end() && !used.empty() && used.begin()->first > position) i = used.begin();
+ }
+ while (i != used.end()) {
+ Message* m = i->second.next(version, cursor);
+ QPID_LOG(debug, "PagedQueue::next(" << cursor.valid << ":" << cursor.position << "): " << m);
+ if (m) return m;
+ ++i;
+ }
+ QPID_LOG(debug, "PagedQueue::next(" << cursor.valid << ":" << cursor.position << ") returning 0 ");
+ return 0;
+}
+
+Message* PagedQueue::release(const QueueCursor& cursor)
+{
+ if (cursor.valid) {
+ Used::iterator i = findPage(cursor.position, true);
+ if (i == used.end()) return 0;
+ return i->second.release(cursor.position);
+ } else {
+ return 0;
+ }
+}
+
+Message* PagedQueue::find(const framing::SequenceNumber& position, QueueCursor* cursor)
+{
+ Used::iterator i = findPage(position, true);
+ if (i != used.end()) {
+ Message* m = i->second.find(position);
+ if (cursor) {
+ cursor->setPosition(version, m ? m->getSequence() : position);
+ }
+ return m;
+ } else {
+ return 0;
+ }
+}
+
+void PagedQueue::foreach(Functor)
+{
+ //TODO:
+}
+
+Message* PagedQueue::find(const QueueCursor& cursor)
+{
+ if (cursor.valid) return find(cursor.position, 0);
+ else return 0;
+}
+
+PagedQueue::Page::Page(size_t s, size_t o) : size(s), offset(o), region(0), used(0)
+{
+ QPID_LOG(debug, "Created Page[" << offset << "], size=" << size);
+}
+
+void PagedQueue::Page::deleted(qpid::framing::SequenceNumber s)
+{
+ if (isLoaded()) {
+ Message* message = find(s);
+ assert(message);//could this ever legitimately be 0?
+ message->setState(DELETED);
+ }
+ contents.remove(s);
+ acquired.remove(s);
+}
+
+Message* PagedQueue::Page::release(qpid::framing::SequenceNumber s)
+{
+ Message* m = find(s);
+ if (m) {
+ m->setState(AVAILABLE);
+ }
+ acquired.remove(s);
+ return m;
+}
+
+bool PagedQueue::Page::add(const Message& message)
+{
+ assert(region);
+ assert (size >= used);
+ size_t encoded = encode(message, region + used, size - used);
+ QPID_LOG(debug, "Calling Page[" << offset << "]::add() used=" << used << ", size=" << size << ", encoded=" << encoded << ")");
+ if (encoded) {
+ used += encoded;
+ messages.push_back(message);
+ messages.back().setState(AVAILABLE);
+ contents.add(message.getSequence());
+ return true;
+ } else {
+ return false;
+ }
+}
+
+bool PagedQueue::Page::empty() const
+{
+ return contents.empty();
+}
+
+bool PagedQueue::Page::isLoaded() const
+{
+ return region;
+}
+
+Message* PagedQueue::Page::next(uint32_t version, QueueCursor& cursor)
+{
+ if (messages.empty()) return 0;
+
+ qpid::framing::SequenceNumber position;
+ if (cursor.valid) {
+ position = cursor.position + 1;
+ if (position < messages.front().getSequence()) {
+ position = messages.front().getSequence();
+ cursor.setPosition(position, version);
+ }
+ } else {
+ position = messages.front().getSequence();
+ cursor.setPosition(position, version);
+ }
+
+ Message* m;
+ do {
+ m = find(position);
+ if (m) cursor.setPosition(position, version);
+ ++position;
+ } while (m != 0 && !cursor.check(*m));
+ return m;
+ //if it is the first in the page, increment the hint count of the page
+ //if it is the last in the page, decrement the hint count of the page
+}
+
+/**
+ * Called before adding to the free list
+ */
+void PagedQueue::Page::clear(MemoryMappedFile& file)
+{
+ if (region) file.unmap(region, size);
+ region = 0;
+ used = 0;
+ contents.clear();
+ messages.clear();
+}
+
+size_t PagedQueue::Page::available() const
+{
+ return contents.size() - acquired.size();
+}
+
+Message* PagedQueue::Page::find(qpid::framing::SequenceNumber position)
+{
+ if (messages.size()) {
+ assert(position >= messages.front().getSequence());
+
+ size_t index = position - messages.front().getSequence();
+ if (index < messages.size()) return &(messages[index]);
+ else return 0;
+ } else {
+ //page is empty, is this an error?
+ QPID_LOG(warning, "Could not find message at " << position << "; empty page.");
+ return 0;
+ }
+
+ //if it is the first in the page, increment the hint count of the page
+ //if it is the last in the page, decrement the hint count of the page
+}
+
+void PagedQueue::Page::load(MemoryMappedFile& file, ProtocolRegistry& protocols)
+{
+ QPID_LOG(debug, "Page[" << offset << "]::load" << " used=" << used << ", size=" << size);
+ assert(region == 0);
+ region = file.map(offset, size);
+ assert(region != 0);
+ bool haveData = used > 0;
+ used = 4;//first 4 bytes are the count
+ if (haveData) {
+ uint32_t count = *(reinterpret_cast<uint32_t*>(region));
+ //decode messages into Page::messages
+ for (size_t i = 0; i < count; ++i) {
+ Message message;
+ used += decode(protocols, message, region + used, size - used);
+ if (!contents.contains(message.getSequence())) {
+ message.setState(DELETED);
+ QPID_LOG(debug, "Setting state to deleted for message loaded at " << message.getSequence());
+ } else if (acquired.contains(message.getSequence())) {
+ message.setState(ACQUIRED);
+ } else {
+ message.setState(AVAILABLE);
+ }
+ messages.push_back(message);
+ }
+ if (messages.size()) {
+ QPID_LOG(debug, "Page[" << offset << "]::load " << messages.size() << " messages loaded from "
+ << messages.front().getSequence() << " to " << messages.back().getSequence());
+ } else {
+ QPID_LOG(debug, "Page[" << offset << "]::load no messages loaded");
+ }
+ }//else there is nothing we need to explicitly load, just needed to map region
+}
+
+void PagedQueue::Page::unload(MemoryMappedFile& file)
+{
+ if (messages.size()) {
+ QPID_LOG(debug, "Page[" << offset << "]::unload " << messages.size() << " messages to unload from "
+ << messages.front().getSequence() << " to " << messages.back().getSequence());
+ } else {
+ QPID_LOG(debug, "Page[" << offset << "]::unload no messages to unload");
+ }
+ for (std::deque<Message>::iterator i = messages.begin(); i != messages.end(); ++i) {
+ if (i->getState() == ACQUIRED) acquired.add(i->getSequence());
+ }
+ uint32_t count = messages.size();
+ ::memcpy(region, &count, sizeof(count));
+ file.flush(region, size);
+ file.unmap(region, size);
+ //remove messages from memory
+ messages.clear();
+ region = 0;
+}
+
+void PagedQueue::load(Page& page)
+{
+ //if needed, release another page
+ if (loaded == maxLoaded) {
+ //which page to select?
+ Used::reverse_iterator i = used.rbegin();
+ while (i != used.rend() && !i->second.isLoaded()) {
+ ++i;
+ }
+ assert(i != used.rend());
+ unload(i->second);
+ }
+ page.load(file, protocols);
+ ++loaded;
+ QPID_LOG(debug, "PagedQueue[" << path << "] loaded page, " << loaded << " pages now loaded");
+}
+
+void PagedQueue::unload(Page& page)
+{
+ page.unload(file);
+ --loaded;
+ QPID_LOG(debug, "PagedQueue[" << path << "] unloaded page, " << loaded << " pages now loaded");
+}
+
+
+PagedQueue::Page& PagedQueue::newPage(qpid::framing::SequenceNumber id)
+{
+ if (loaded == maxLoaded) {
+ //need to release a page from memory to make way for a new one
+
+ //choose last one?
+ Used::reverse_iterator i = used.rbegin();
+ while (!i->second.isLoaded() && i != used.rend()) {
+ ++i;
+ }
+ assert(i != used.rend());
+ unload(i->second);
+ }
+ if (free.empty()) {
+ //need to extend file and add some pages to the free list
+ addPages(4/*arbitrary number, should this be config item?*/);
+ }
+ std::pair<Used::iterator, bool> result = used.insert(Used::value_type(id, free.front()));
+ QPID_LOG(debug, "Added page for sequence starting from " << id);
+ assert(result.second);
+ free.pop_front();
+ load(result.first->second);
+ return result.first->second;
+}
+
+void PagedQueue::addPages(size_t count)
+{
+ for (size_t i = 0; i < count; ++i) {
+ free.push_back(Page(pageSize, offset));
+ offset += pageSize;
+ file.expand(offset);
+ }
+ QPID_LOG(debug, "Added " << count << " pages to free list; now have " << used.size() << " used, and " << free.size() << " free");
+}
+
+PagedQueue::Used::iterator PagedQueue::findPage(const QueueCursor& cursor)
+{
+ Used::iterator i = used.begin();
+ if (cursor.valid) {
+ i = findPage(cursor.position, true);
+ } else if (i != used.end() && !i->second.isLoaded()) {
+ load(i->second);
+ }
+ return i;
+}
+
+PagedQueue::Used::iterator PagedQueue::findPage(qpid::framing::SequenceNumber n, bool loadIfRequired)
+{
+ Used::iterator i = used.end();
+ for (Used::iterator j = used.begin(); j != used.end() && j->first <= n; ++j) {
+ i = j;
+ }
+ if (loadIfRequired && i != used.end() && !i->second.isLoaded()) {
+ load(i->second);
+ }
+ return i;
+}
+
+}} // namespace qpid::broker
Added: qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.h?rev=1477236&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/PagedQueue.h Mon Apr 29 18:04:17 2013
@@ -0,0 +1,98 @@
+#ifndef QPID_BROKER_PAGEDQUEUE_H
+#define QPID_BROKER_PAGEDQUEUE_H
+
+/*
+ *
+ * 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/broker/Messages.h"
+#include "qpid/framing/SequenceSet.h"
+#include "qpid/sys/MemoryMappedFile.h"
+#include <deque>
+#include <list>
+#include <map>
+
+namespace qpid {
+namespace broker {
+class ProtocolRegistry;
+/**
+ *
+ */
+class PagedQueue : public Messages {
+ public:
+ PagedQueue(const std::string& name, const std::string& directory, uint maxLoaded, uint pageFactor, ProtocolRegistry& protocols);
+ size_t size();
+ bool deleted(const QueueCursor&);
+ void publish(const Message& added);
+ Message* next(QueueCursor& cursor);
+ Message* release(const QueueCursor& cursor);
+ Message* find(const framing::SequenceNumber&, QueueCursor*);
+ Message* find(const QueueCursor&);
+ void foreach(Functor);
+ private:
+ class Page {
+ public:
+ Page(size_t size, size_t offset);
+ void read();//decode messages from memory mapped file
+ void write();//encode messages into memory mapped file
+ bool isLoaded() const;
+ bool empty() const;
+ void deleted(qpid::framing::SequenceNumber);
+ Message* release(qpid::framing::SequenceNumber);
+ bool add(const Message&);
+ Message* next(uint32_t version, QueueCursor&);
+ Message* find(qpid::framing::SequenceNumber);
+ void load(qpid::sys::MemoryMappedFile&,ProtocolRegistry&);
+ void unload(qpid::sys::MemoryMappedFile&);
+ void clear(qpid::sys::MemoryMappedFile&);
+ size_t available() const;
+ private:
+ size_t size;
+ size_t offset;
+
+ char* region;//0 implies not mapped
+ qpid::framing::SequenceSet contents;
+ qpid::framing::SequenceSet acquired;
+ std::deque<Message> messages;//decoded representation
+ size_t used;//amount of data used to encode current set of messages held
+ };
+
+ qpid::sys::MemoryMappedFile file;
+ std::string path;
+ const size_t pageSize;
+ const uint maxLoaded;
+ ProtocolRegistry& protocols;
+ size_t offset;
+ typedef std::map<qpid::framing::SequenceNumber, Page> Used;
+ Used used;
+ std::list<Page> free;
+ uint loaded;
+ uint32_t version;
+
+ void addPages(size_t count);
+ Page& newPage(qpid::framing::SequenceNumber);
+ Used::iterator findPage(const QueueCursor& cursor);
+ Used::iterator findPage(qpid::framing::SequenceNumber n, bool loadIfRequired);
+ void load(Page&);
+ void unload(Page&);
+ bool deleted(qpid::framing::SequenceNumber);
+};
+}} // namespace qpid::broker
+
+#endif /*!QPID_BROKER_PAGEDQUEUE_H*/
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.cpp?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.cpp Mon Apr 29 18:04:17 2013
@@ -20,6 +20,7 @@
*/
#include "Protocol.h"
#include "qpid/sys/ConnectionCodec.h"
+#include "qpid/broker/RecoverableMessageImpl.h"
#include "qpid/broker/amqp_0_10/MessageTransfer.h"
#include "qpid/log/Statement.h"
@@ -47,13 +48,25 @@ boost::intrusive_ptr<const qpid::broker:
}
boost::shared_ptr<RecoverableMessage> ProtocolRegistry::recover(qpid::framing::Buffer& b)
{
- boost::shared_ptr<RecoverableMessage> msg;
- for (Protocols::const_iterator i = protocols.begin(); !msg && i != protocols.end(); ++i) {
- msg = i->second->recover(b);
+ uint32_t position = b.getPosition();
+ for (Protocols::const_iterator i = protocols.begin(); i != protocols.end(); ++i) {
+ boost::shared_ptr<RecoverableMessage> msg = i->second->recover(b);
+ if (msg) return msg;
+ else b.setPosition(position);
}
+ boost::intrusive_ptr<qpid::broker::amqp_0_10::MessageTransfer> transfer(new qpid::broker::amqp_0_10::MessageTransfer());
+ transfer->decodeHeader(b);
+ boost::shared_ptr<RecoverableMessage> msg(new RecoverableMessageImpl(Message(transfer, transfer)));
return msg;
}
+Message ProtocolRegistry::decode(qpid::framing::Buffer& buffer)
+{
+ boost::shared_ptr<RecoverableMessage> r = recover(buffer);
+ r->decodeContent(buffer);
+ return r->getMessage();
+}
+
ProtocolRegistry::~ProtocolRegistry()
{
for (Protocols::const_iterator i = protocols.begin(); i != protocols.end(); ++i) {
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.h?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/Protocol.h Mon Apr 29 18:04:17 2013
@@ -68,6 +68,7 @@ class ProtocolRegistry : public Protocol
qpid::sys::ConnectionCodec* create(const qpid::framing::ProtocolVersion&, qpid::sys::OutputControl&, const std::string&, const qpid::sys::SecuritySettings&);
boost::intrusive_ptr<const qpid::broker::amqp_0_10::MessageTransfer> translate(const Message&);
boost::shared_ptr<RecoverableMessage> recover(qpid::framing::Buffer&);
+ Message decode(qpid::framing::Buffer&);
~ProtocolRegistry();
void add(const std::string&, Protocol*);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueCursor.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueCursor.h?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueCursor.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueCursor.h Mon Apr 29 18:04:17 2013
@@ -64,6 +64,7 @@ class QueueCursor
friend class MessageDeque;
friend class MessageMap;
friend class PriorityQueue;
+ friend class PagedQueue;
template <typename T> friend class IndexedDeque;
};
}} // namespace qpid::broker
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueFactory.cpp Mon Apr 29 18:04:17 2013
@@ -30,17 +30,18 @@
#include "qpid/broker/Fairshare.h"
#include "qpid/broker/MessageDeque.h"
#include "qpid/broker/MessageMap.h"
+#include "qpid/broker/PagedQueue.h"
#include "qpid/broker/PriorityQueue.h"
#include "qpid/broker/QueueFlowLimit.h"
#include "qpid/broker/ThresholdAlerts.h"
#include "qpid/broker/FifoDistributor.h"
+#include "qpid/log/Statement.h"
#include <map>
#include <memory>
namespace qpid {
namespace broker {
-
QueueFactory::QueueFactory() : broker(0), store(0), parent(0) {}
boost::shared_ptr<Queue> QueueFactory::create(const std::string& name, const QueueSettings& settings)
@@ -66,6 +67,17 @@ boost::shared_ptr<Queue> QueueFactory::c
} else {
queue->messages = std::auto_ptr<Messages>(new PriorityQueue(settings.priorities));
}
+ } else if (settings.paging) {
+ if (!broker) {
+ QPID_LOG(warning, "Cannot create paged queue without broker context");
+ } else if (!qpid::sys::MemoryMappedFile::isSupported()) {
+ QPID_LOG(warning, "Cannot create paged queue; memory mapped file support not available on this platform");
+ } else {
+ queue->messages = std::auto_ptr<Messages>(new PagedQueue(name, broker->getOptions().dataDir,
+ settings.maxPages ? settings.maxPages : 4,
+ settings.pageFactor ? settings.pageFactor : 1,
+ broker->getProtocolRegistry()));
+ }
} else if (settings.lvqKey.empty()) {//LVQ already handled above
queue->messages = std::auto_ptr<Messages>(new MessageDeque());
}
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.cpp Mon Apr 29 18:04:17 2013
@@ -54,6 +54,9 @@ const std::string ALERT_SIZE_DOWN("qpid.
const std::string PRIORITIES("qpid.priorities");
const std::string FAIRSHARE("qpid.fairshare");
const std::string FAIRSHARE_ALIAS("x-qpid-fairshare");
+const std::string PAGING("qpid.paging");
+const std::string MAX_PAGES("qpid.max_pages_loaded");
+const std::string PAGE_FACTOR("qpid.page_factor");
const std::string LVQ_LEGACY("qpid.last_value_queue");
const std::string LVQ_LEGACY_KEY("qpid.LVQ_key");
@@ -87,6 +90,9 @@ QueueSettings::QueueSettings(bool d, boo
shareGroups(false),
addTimestamp(false),
dropMessagesAtLimit(false),
+ paging(false),
+ maxPages(0),
+ pageFactor(0),
noLocal(false),
isBrowseOnly(false),
autoDeleteDelay(0),
@@ -187,6 +193,15 @@ bool QueueSettings::handle(const std::st
} else if (key == MAX_FILE_SIZE && value.asUint64() > 0) {
maxFileSize = value.asUint64();
return false; // 'handle' here and also pass to store
+ } else if (key == PAGING) {
+ paging = value;
+ return true;
+ } else if (key == MAX_PAGES) {
+ maxPages = value;
+ return true;
+ } else if (key == PAGE_FACTOR) {
+ pageFactor = value;
+ return true;
} else {
return false;
}
@@ -218,6 +233,25 @@ void QueueSettings::validate() const
throw qpid::framing::InvalidArgumentException(QPID_MSG("Only shared groups are supported at present; " << MessageGroupManager::qpidSharedGroup
<< " is required if " << MessageGroupManager::qpidMessageGroupKey << " is set"));
}
+
+ if (paging) {
+ if(lvqKey.size()) {
+ throw qpid::framing::InvalidArgumentException(QPID_MSG("Cannot specify " << LVQ_KEY << " and " << PAGING << " for the same queue"));
+ }
+ if(priorities) {
+ throw qpid::framing::InvalidArgumentException(QPID_MSG("Cannot specify " << PRIORITIES << " and " << PAGING << " for the same queue"));
+ }
+ if(groupKey.size()) {
+ throw qpid::framing::InvalidArgumentException(QPID_MSG("Cannot specify " << MessageGroupManager::qpidMessageGroupKey << " and " << PAGING << " for the same queue"));
+ }
+ } else {
+ if (maxPages) {
+ throw qpid::framing::InvalidArgumentException(QPID_MSG("Can only specify " << MAX_PAGES << " if " << PAGING << " is set"));
+ }
+ if (pageFactor) {
+ throw qpid::framing::InvalidArgumentException(QPID_MSG("Can only specify " << PAGE_FACTOR << " if " << PAGING << " is set"));
+ }
+ }
}
void QueueSettings::populate(const std::map<std::string, qpid::types::Variant>& inputs, std::map<std::string, qpid::types::Variant>& unused)
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/QueueSettings.h Mon Apr 29 18:04:17 2013
@@ -59,6 +59,11 @@ struct QueueSettings
QueueDepth maxDepth;
bool dropMessagesAtLimit;//aka ring queue policy
+ //PagedQueue:
+ bool paging;
+ uint maxPages;
+ uint pageFactor;
+
bool noLocal;
bool isBrowseOnly;
std::string traceId;
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessage.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessage.h?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessage.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessage.h Mon Apr 29 18:04:17 2013
@@ -30,7 +30,7 @@
namespace qpid {
namespace broker {
class ExpiryPolicy;
-
+class Message;
/**
* The interface through which messages are reloaded on recovery.
*/
@@ -53,6 +53,7 @@ public:
* of length as necessary)
*/
virtual void decodeContent(framing::Buffer& buffer) = 0;
+ virtual Message getMessage() = 0;
virtual ~RecoverableMessage() {};
};
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessageImpl.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessageImpl.h?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessageImpl.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/RecoverableMessageImpl.h Mon Apr 29 18:04:17 2013
@@ -22,11 +22,11 @@
*
*/
#include "RecoverableMessage.h"
+#include "Message.h"
namespace qpid {
namespace broker {
class DtxBuffer;
-class Message;
class Queue;
class RecoverableMessageImpl : public RecoverableMessage
@@ -43,6 +43,7 @@ public:
void recover(boost::shared_ptr<Queue> queue);
void enqueue(boost::shared_ptr<DtxBuffer> buffer, boost::shared_ptr<Queue> queue);
void dequeue(boost::shared_ptr<DtxBuffer> buffer, boost::shared_ptr<Queue> queue);
+ Message getMessage();
};
}} // namespace qpid::broker
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/RecoveryManagerImpl.cpp Mon Apr 29 18:04:17 2013
@@ -118,15 +118,8 @@ RecoverableQueue::shared_ptr RecoveryMan
RecoverableMessage::shared_ptr RecoveryManagerImpl::recoverMessage(framing::Buffer& buffer)
{
- framing::Buffer sniffer(buffer.getPointer(), buffer.available());
- RecoverableMessage::shared_ptr m = protocols.recover(sniffer);
- if (m) {
- return m;
- } else {
- boost::intrusive_ptr<qpid::broker::amqp_0_10::MessageTransfer> transfer(new qpid::broker::amqp_0_10::MessageTransfer());
- transfer->decodeHeader(buffer);
- return RecoverableMessage::shared_ptr(new RecoverableMessageImpl(Message(transfer, transfer)));
- }
+ RecoverableMessage::shared_ptr m = protocols.recover(buffer);
+ return m;
}
RecoverableTransaction::shared_ptr RecoveryManagerImpl::recoverTransaction(const std::string& xid,
@@ -191,6 +184,11 @@ void RecoverableMessageImpl::computeExpi
msg.computeExpiration(ep);
}
+Message RecoverableMessageImpl::getMessage()
+{
+ return msg;
+}
+
void RecoverableQueueImpl::recover(RecoverableMessage::shared_ptr msg)
{
dynamic_pointer_cast<RecoverableMessageImpl>(msg)->recover(queue);
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.cpp Mon Apr 29 18:04:17 2013
@@ -228,11 +228,16 @@ void MessageTransfer::decodeHeader(frami
}
void MessageTransfer::decodeContent(framing::Buffer& buffer)
{
- if (buffer.available()) {
+ decodeContent(buffer, buffer.available());
+}
+
+void MessageTransfer::decodeContent(framing::Buffer& buffer, size_t size)
+{
+ if (size) {
//get the data as a string and set that as the content
//body on a frame then add that frame to the frameset
AMQFrame frame((AMQContentBody()));
- frame.castBody<AMQContentBody>()->decode(buffer, buffer.available());
+ frame.castBody<AMQContentBody>()->decode(buffer, size);
frame.setFirstSegment(false);
frames.append(frame);
} else {
@@ -373,25 +378,4 @@ boost::intrusive_ptr<PersistableMessage>
}
return clone;
}
-}
-// qpid::broker namespace, TODO: move these elsewhere!
-void encode(const Message& in, std::string& out)
-{
- const amqp_0_10::MessageTransfer& transfer = amqp_0_10::MessageTransfer::get(in);
- uint32_t size = transfer.encodedSize();
- std::vector<char> data(size);
- qpid::framing::Buffer buffer(&(data[0]), size);
- transfer.encode(buffer);
- buffer.reset();
- buffer.getRawData(out, size);
-}
-void decode(const std::string& in, Message& out)
-{
- boost::intrusive_ptr<amqp_0_10::MessageTransfer> transfer(new amqp_0_10::MessageTransfer);
- qpid::framing::Buffer buffer(const_cast<char*>(in.data()), in.size());
- transfer->decodeHeader(buffer);
- transfer->decodeContent(buffer);
- out = Message(transfer, transfer);
-}
-
-}} // namespace qpid::broker::amqp_0_10
+}}} // namespace qpid::broker::amqp_0_10
Modified: qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.h?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.h (original)
+++ qpid/trunk/qpid/cpp/src/qpid/broker/amqp_0_10/MessageTransfer.h Mon Apr 29 18:04:17 2013
@@ -93,6 +93,7 @@ class MessageTransfer : public qpid::bro
void decodeHeader(framing::Buffer& buffer);
void decodeContent(framing::Buffer& buffer);
+ void decodeContent(framing::Buffer& buffer, size_t size);
void encode(framing::Buffer& buffer) const;
uint32_t encodedSize() const;
Added: qpid/trunk/qpid/cpp/src/qpid/sys/MemoryMappedFile.h
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/MemoryMappedFile.h?rev=1477236&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/MemoryMappedFile.h (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/MemoryMappedFile.h Mon Apr 29 18:04:17 2013
@@ -0,0 +1,74 @@
+#ifndef QPID_SYS_MEMORYMAPPEDFILE_H
+#define QPID_SYS_MEMORYMAPPEDFILE_H
+
+/*
+ *
+ * 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 <memory>
+#include <string>
+
+namespace qpid {
+namespace sys {
+
+class MemoryMappedFilePrivate;
+/**
+ * Abstraction of memory mapping functionality
+ */
+class MemoryMappedFile {
+ public:
+ MemoryMappedFile();
+ /**
+ * Opens a file that can be mapped by region into memory
+ */
+ std::string open(const std::string& name, const std::string& directory);
+ /**
+ * Returns the page size
+ */
+ size_t getPageSize();
+ /**
+ * Load a portion of the file into memory
+ */
+ char* map(size_t offset, size_t size);
+ /**
+ * Evict a portion of the file from memory
+ */
+ void unmap(char* region, size_t size);
+ /**
+ * Flush any changes to a previously mapped region of the file
+ * back to disk
+ */
+ void flush(char* region, size_t size);
+ /**
+ * Expand the capacity of the file
+ */
+ void expand(size_t offset);
+ /**
+ * Returns true if memory mapping is supported, false otherwise
+ */
+ static bool isSupported();
+ private:
+ std::auto_ptr<MemoryMappedFilePrivate> state;
+
+ MemoryMappedFile(const MemoryMappedFile&);
+ MemoryMappedFile& operator=(const MemoryMappedFile&);
+};
+}} // namespace qpid::sys
+
+#endif /*!QPID_SYS_MEMORYMAPPEDFILE_H*/
Added: qpid/trunk/qpid/cpp/src/qpid/sys/posix/MemoryMappedFile.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/posix/MemoryMappedFile.cpp?rev=1477236&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/posix/MemoryMappedFile.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/posix/MemoryMappedFile.cpp Mon Apr 29 18:04:17 2013
@@ -0,0 +1,105 @@
+/*
+ *
+ * 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/sys/MemoryMappedFile.h"
+#include "qpid/Exception.h"
+#include "qpid/Msg.h"
+#include <sys/mman.h>
+#include <unistd.h>
+#include <fcntl.h>
+
+namespace qpid {
+namespace sys {
+namespace {
+const std::string PATH_SEPARATOR("/");
+const std::string ESCAPE("%");
+const std::string VALID("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_-.");
+std::string getFileName(const std::string& name, const std::string& dir)
+{
+ std::stringstream filename;
+ if (dir.size()) filename << dir << PATH_SEPARATOR;
+ size_t start = 0;
+ while (true) {
+ size_t i = name.find_first_not_of(VALID, start);
+ if (i == std::string::npos) {
+ filename << name.substr(start);
+ return filename.str();
+ } else {
+ if (i > start) filename << name.substr(start, i-start);
+ filename << ESCAPE << (int) name.at(i);
+ start = i+1;
+ }
+ }
+
+}
+}
+
+class MemoryMappedFilePrivate
+{
+ friend class MemoryMappedFile;
+ const int fd;
+ MemoryMappedFilePrivate(int fd_) : fd(fd_) {}
+};
+MemoryMappedFile::MemoryMappedFile() {}
+
+std::string MemoryMappedFile::open(const std::string& name, const std::string& directory)
+{
+ std::string path = getFileName(name, directory);
+
+ int flags = O_CREAT | O_TRUNC | O_RDWR;
+ int fd = ::open(path.c_str(), flags, S_IRWXU);
+ if (fd == -1) throw qpid::Exception(QPID_MSG("Failed to open memory mapped file " << path << ": " << qpid::sys::strError(errno) << " [flags=" << flags << "]"));
+ state = std::auto_ptr<MemoryMappedFilePrivate>(new MemoryMappedFilePrivate(fd));
+ return path;
+}
+size_t MemoryMappedFile::getPageSize()
+{
+ return ::sysconf(_SC_PAGE_SIZE);
+}
+char* MemoryMappedFile::map(size_t offset, size_t size)
+{
+ int protection = PROT_READ | PROT_WRITE;
+ char* region = (char*) ::mmap(0, size, protection, MAP_SHARED, state->fd, offset);
+ if (region == MAP_FAILED) {
+ throw qpid::Exception(QPID_MSG("Failed to map page into memory: " << qpid::sys::strError(errno)));
+ }
+ return region;
+
+}
+void MemoryMappedFile::unmap(char* region, size_t size)
+{
+ ::munmap(region, size);
+}
+void MemoryMappedFile::flush(char* region, size_t size)
+{
+ ::msync(region, size, MS_ASYNC);
+}
+void MemoryMappedFile::expand(size_t offset)
+{
+ if ((::lseek(state->fd, offset - 1, SEEK_SET) == -1) || (::write(state->fd, "", 1) == -1)) {
+ throw qpid::Exception(QPID_MSG("Failed to expand paged queue file: " << qpid::sys::strError(errno)));
+ }
+}
+bool MemoryMappedFile::isSupported()
+{
+ return true;
+}
+
+}} // namespace qpid::sys
Added: qpid/trunk/qpid/cpp/src/qpid/sys/windows/MemoryMappedFile.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/qpid/sys/windows/MemoryMappedFile.cpp?rev=1477236&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/qpid/sys/windows/MemoryMappedFile.cpp (added)
+++ qpid/trunk/qpid/cpp/src/qpid/sys/windows/MemoryMappedFile.cpp Mon Apr 29 18:04:17 2013
@@ -0,0 +1,55 @@
+/*
+ *
+ * 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/sys/MemoryMappedFile.h"
+
+namespace qpid {
+namespace sys {
+class MemoryMappedFilePrivate {};
+
+MemoryMappedFile::MemoryMappedFile() {}
+
+std::string MemoryMappedFile::open(const std::string& /*name*/, const std::string& /*directory*/)
+{
+ return std::string();
+}
+size_t MemoryMappedFile::getPageSize()
+{
+ return 0;
+}
+char* MemoryMappedFile::map(size_t /*offset*/, size_t /*size*/)
+{
+ return 0;
+}
+void MemoryMappedFile::unmap(char* /*region*/, size_t /*size*/)
+{
+}
+void MemoryMappedFile::flush(char* /*region*/, size_t /*size*/)
+{
+}
+void MemoryMappedFile::expand(size_t /*offset*/)
+{
+}
+bool MemoryMappedFile::isSupported()
+{
+ return false;
+}
+
+}} // namespace qpid::sys
Modified: qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt (original)
+++ qpid/trunk/qpid/cpp/src/tests/CMakeLists.txt Mon Apr 29 18:04:17 2013
@@ -319,6 +319,10 @@ add_test (qpid-client-test ${test_wrap}
add_test (quick_perftest ${test_wrap} ${qpid-perftest_LOCATION} --summary --count 100)
add_test (quick_topictest ${test_wrap} ${CMAKE_CURRENT_SOURCE_DIR}/quick_topictest${test_script_suffix})
add_test (quick_txtest ${test_wrap} ${qpid-txtest_LOCATION} --queues 4 --tx-count 10 --quiet)
+if (NOT CMAKE_SYSTEM_NAME STREQUAL Windows)
+ # paged queue not yet implemented for windows
+ add_test (paged_queue_tests ${shell} ${CMAKE_CURRENT_SOURCE_DIR}/run_paged_queue_tests${test_script_suffix})
+endif (NOT CMAKE_SYSTEM_NAME STREQUAL Windows)
if (PYTHON_EXECUTABLE)
add_test (run_header_test ${shell} ${CMAKE_CURRENT_SOURCE_DIR}/run_header_test${test_script_suffix})
add_test (python_tests ${test_wrap} ${CMAKE_CURRENT_SOURCE_DIR}/python_tests${test_script_suffix})
Modified: qpid/trunk/qpid/cpp/src/tests/Makefile.am
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/Makefile.am?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/Makefile.am (original)
+++ qpid/trunk/qpid/cpp/src/tests/Makefile.am Mon Apr 29 18:04:17 2013
@@ -325,6 +325,7 @@ EXTRA_DIST += \
run_perftest \
ring_queue_test \
run_ring_queue_test \
+ run_paged_queue_tests \
dynamic_log_level_test \
dynamic_log_hires_timestamp \
qpid-ctrl \
@@ -362,7 +363,7 @@ CLEANFILES+=valgrind.out *.log *.vglog*
# Not run under valgrind, too slow
LONG_TESTS+=start_broker \
- fanout_perftest shared_perftest multiq_perftest topic_perftest run_ring_queue_test \
+ fanout_perftest shared_perftest multiq_perftest topic_perftest run_ring_queue_test run_paged_queue_tests\
run_msg_group_tests_soak \
stop_broker \
run_long_federation_sys_tests
Modified: qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp?rev=1477236&r1=1477235&r2=1477236&view=diff
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp (original)
+++ qpid/trunk/qpid/cpp/src/tests/MessageTest.cpp Mon Apr 29 18:04:17 2013
@@ -19,6 +19,7 @@
*
*/
#include "qpid/broker/Message.h"
+#include "qpid/broker/Protocol.h"
#include "qpid/framing/AMQP_HighestVersion.h"
#include "qpid/framing/AMQFrame.h"
#include "qpid/framing/MessageTransferBody.h"
@@ -56,10 +57,12 @@ QPID_AUTO_TEST_CASE(testEncodeDecode)
properties["abc"] = "xyz";
Message msg = MessageUtils::createMessage(properties, data);
- std::string buffer;
- encode(msg, buffer);
- msg = Message();
- decode(buffer, msg);
+ std::vector<char> bytes(msg.getPersistentContext()->encodedSize());
+ qpid::framing::Buffer buffer(&bytes[0], bytes.size());
+ msg.getPersistentContext()->encode(buffer);
+ buffer.reset();
+ ProtocolRegistry registry;
+ msg = registry.decode(buffer);
BOOST_CHECK_EQUAL(routingKey, msg.getRoutingKey());
BOOST_CHECK_EQUAL((uint64_t) data.size(), msg.getContentSize());
Added: qpid/trunk/qpid/cpp/src/tests/run_paged_queue_tests
URL: http://svn.apache.org/viewvc/qpid/trunk/qpid/cpp/src/tests/run_paged_queue_tests?rev=1477236&view=auto
==============================================================================
--- qpid/trunk/qpid/cpp/src/tests/run_paged_queue_tests (added)
+++ qpid/trunk/qpid/cpp/src/tests/run_paged_queue_tests Mon Apr 29 18:04:17 2013
@@ -0,0 +1,32 @@
+#!/bin/bash
+
+#
+# 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.
+#
+
+#setup path to find qpid-config and sender/receiver test progs
+source ./test_env.sh
+
+export PATH=$PWD:$srcdir:$PYTHON_COMMANDS:$PATH
+
+#set port to connect to via env var
+test -s qpidd.port && QPID_PORT=`cat qpidd.port`
+export QPID_PORT
+
+qpid-cpp-benchmark --broker "localhost:$QPID_PORT" --create-option "node:{x-declare:{arguments:{'qpid.paging':True,'qpid.max_size':0,'qpid.max_count':0,'qpid.flow_stop_size':0,'qpid.flow_resume_size':0,'qpid.flow_stop_count':0,'qpid.flow_resume_count':0}}}"
+qpid-cpp-benchmark --broker "localhost:$QPID_PORT" --create-option "node:{x-declare:{arguments:{'qpid.paging':True,'qpid.max_size':0,'qpid.max_count':0,'qpid.flow_stop_size':0,'qpid.flow_resume_size':0,'qpid.flow_stop_count':0,'qpid.flow_resume_count':0}}}" --fill-drain
\ No newline at end of file
Propchange: qpid/trunk/qpid/cpp/src/tests/run_paged_queue_tests
------------------------------------------------------------------------------
svn:executable = *
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org