src/qpid/broker/PagedQueue.cpp - qpid-cpp - Git at Google

 /*
  *
  * 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 "qpid/framing/reply_exceptions.h"
 #include "qpid/sys/Time.h"
 #include <string.h>

 namespace qpid {
 namespace broker {
 namespace {
 using qpid::sys::AbsTime;
 using qpid::sys::Duration;
 using qpid::sys::ZERO;
 using qpid::sys::FAR_FUTURE;
 using qpid::sys::MemoryMappedFile;
 const uint32_t OVERHEAD(4/*content-size*/ + 4/*sequence-number*/ + 8/*persistence-id*/ + 8/*expiration*/);

 size_t encodedSize(const Message& msg)
 {
     return msg.getPersistentContext()->encodedSize() + OVERHEAD;
 }

 size_t encode(const Message& msg, char* data, size_t size)
 {
     uint32_t encoded = msg.getPersistentContext()->encodedSize();
     uint32_t required = encoded + OVERHEAD;
     if (required > size) return 0;
     qpid::framing::Buffer buffer(data, required);
     buffer.putLong(encoded);
     buffer.putLong(msg.getSequence());
     buffer.putLongLong(msg.getPersistentContext()->getPersistenceId());
     sys::AbsTime expiration = msg.getExpiration();
     int64_t t(0);
     if (expiration < FAR_FUTURE) {
         // Just need an integer that will round trip
         t = Duration(ZERO, expiration);
     }
     buffer.putLongLong(t);
     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();
     uint64_t persistenceId = metadata.getLongLong();
     int64_t t = metadata.getLongLong();
     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));
     msg.getPersistentContext()->setPersistenceId(persistenceId);
     if (t) {
         sys::AbsTime expiration(ZERO, t);
         msg.getSharedState().setExpiration(expiration);
     }
     return encoded + metadata.getPosition();
 }

 }

 PagedQueue::PagedQueue(const std::string& name_, const std::string& directory, uint m, uint factor, ProtocolRegistry& p)
     : name(name_), pageSize(file.getPageSize()*factor), maxLoaded(m), protocols(p), offset(0), loaded(0), version(0)
 {
     if (directory.empty()) {
         throw qpid::Exception(QPID_MSG("Cannot create paged queue: No paged queue directory specified"));
     }
     file.open(name, directory);
     QPID_LOG(debug, "PagedQueue[" << name << "]");
 }

 PagedQueue::~PagedQueue()
 {
     file.close();
 }

 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::check(const Message& added)
 {
     if (encodedSize(added) > pageSize) {
         QPID_LOG(error, "Message is larger than page size for queue " << name);
         throw qpid::framing::PreconditionFailedException(QPID_MSG("Message is larger than page size for queue " << name));
     }
 }

 void PagedQueue::publish(const Message& added)
 {
     check(added);
     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)) {
         QPID_LOG(error, "Could not add message to paged queue " << name);
         throw qpid::Exception(QPID_MSG("Could not add message to paged queue " << name));
     }
 }

 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()) {
         if (!i->second.isLoaded()) load(i->second);
         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) {
         qpid::framing::Buffer buffer(region, sizeof(uint32_t));
         uint32_t count = buffer.getLong();
         //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();
     qpid::framing::Buffer buffer(region, sizeof(uint32_t));
     buffer.putLong(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[" << name << "] loaded page, " << loaded << " pages now loaded");
 }

 void PagedQueue::unload(Page& page)
 {
     page.unload(file);
     --loaded;
     QPID_LOG(debug, "PagedQueue[" << name << "] 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
	/*
	*
	* 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 "qpid/framing/reply_exceptions.h"
	#include "qpid/sys/Time.h"
	#include <string.h>

	namespace qpid {
	namespace broker {
	namespace {
	using qpid::sys::AbsTime;
	using qpid::sys::Duration;
	using qpid::sys::ZERO;
	using qpid::sys::FAR_FUTURE;
	using qpid::sys::MemoryMappedFile;
	const uint32_t OVERHEAD(4/content-size/ + 4/sequence-number/ + 8/persistence-id/ + 8/expiration/);

	size_t encodedSize(const Message& msg)
	{
	return msg.getPersistentContext()->encodedSize() + OVERHEAD;
	}

	size_t encode(const Message& msg, char* data, size_t size)
	{
	uint32_t encoded = msg.getPersistentContext()->encodedSize();
	uint32_t required = encoded + OVERHEAD;
	if (required > size) return 0;
	qpid::framing::Buffer buffer(data, required);
	buffer.putLong(encoded);
	buffer.putLong(msg.getSequence());
	buffer.putLongLong(msg.getPersistentContext()->getPersistenceId());
	sys::AbsTime expiration = msg.getExpiration();
	int64_t t(0);
	if (expiration < FAR_FUTURE) {
	// Just need an integer that will round trip
	t = Duration(ZERO, expiration);
	}
	buffer.putLongLong(t);
	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();
	uint64_t persistenceId = metadata.getLongLong();
	int64_t t = metadata.getLongLong();
	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));
	msg.getPersistentContext()->setPersistenceId(persistenceId);
	if (t) {
	sys::AbsTime expiration(ZERO, t);
	msg.getSharedState().setExpiration(expiration);
	}
	return encoded + metadata.getPosition();
	}

	}

	PagedQueue::PagedQueue(const std::string& name_, const std::string& directory, uint m, uint factor, ProtocolRegistry& p)
	: name(name_), pageSize(file.getPageSize()*factor), maxLoaded(m), protocols(p), offset(0), loaded(0), version(0)
	{
	if (directory.empty()) {
	throw qpid::Exception(QPID_MSG("Cannot create paged queue: No paged queue directory specified"));
	}
	file.open(name, directory);
	QPID_LOG(debug, "PagedQueue[" << name << "]");
	}

	PagedQueue::~PagedQueue()
	{
	file.close();
	}

	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::check(const Message& added)
	{
	if (encodedSize(added) > pageSize) {
	QPID_LOG(error, "Message is larger than page size for queue " << name);
	throw qpid::framing::PreconditionFailedException(QPID_MSG("Message is larger than page size for queue " << name));
	}
	}

	void PagedQueue::publish(const Message& added)
	{
	check(added);
	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)) {
	QPID_LOG(error, "Could not add message to paged queue " << name);
	throw qpid::Exception(QPID_MSG("Could not add message to paged queue " << name));
	}
	}

	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()) {
	if (!i->second.isLoaded()) load(i->second);
	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) {
	qpid::framing::Buffer buffer(region, sizeof(uint32_t));
	uint32_t count = buffer.getLong();
	//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();
	qpid::framing::Buffer buffer(region, sizeof(uint32_t));
	buffer.putLong(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[" << name << "] loaded page, " << loaded << " pages now loaded");
	}

	void PagedQueue::unload(Page& page)
	{
	page.unload(file);
	--loaded;
	QPID_LOG(debug, "PagedQueue[" << name << "] 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