RC9/qpid/cpp/src/qpid/client/RdmaConnector.cpp - qpid - 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 "Connector.h"

 #include "Bounds.h"
 #include "ConnectionImpl.h"
 #include "ConnectionSettings.h"
 #include "qpid/log/Statement.h"
 #include "qpid/sys/Time.h"
 #include "qpid/framing/AMQFrame.h"
 #include "qpid/sys/rdma/RdmaIO.h"
 #include "qpid/sys/Dispatcher.h"
 #include "qpid/sys/Poller.h"
 #include "qpid/Msg.h"

 #include <iostream>
 #include <boost/bind.hpp>
 #include <boost/format.hpp>
 #include <boost/lexical_cast.hpp>

 // This stuff needs to abstracted out of here to a platform specific file
 #include <netdb.h>

 namespace qpid {
 namespace client {

 using namespace qpid::sys;
 using namespace qpid::framing;
 using boost::format;
 using boost::str;

 class RdmaConnector : public Connector, private sys::Runnable
 {
     struct Buff;

     /** Batch up frames for writing to aio. */
     class Writer : public framing::FrameHandler {
         typedef Rdma::Buffer BufferBase;
         typedef std::deque<framing::AMQFrame> Frames;

         const uint16_t maxFrameSize;
         sys::Mutex lock;
         Rdma::AsynchIO* aio;
         BufferBase* buffer;
         Frames frames;
         size_t lastEof; // Position after last EOF in frames
         framing::Buffer encode;
         size_t framesEncoded;
         std::string identifier;
         Bounds* bounds;

         void writeOne();
         void newBuffer();

       public:

         Writer(uint16_t maxFrameSize, Bounds*);
         ~Writer();
         void init(std::string id, Rdma::AsynchIO*);
         void handle(framing::AMQFrame&);
         void write(Rdma::AsynchIO&);
     };

     const uint16_t maxFrameSize;
     framing::ProtocolVersion version;
     bool initiated;

     sys::Mutex pollingLock;
     bool polling;
     bool joined;

     sys::ShutdownHandler* shutdownHandler;
     framing::InputHandler* input;
     framing::InitiationHandler* initialiser;
     framing::OutputHandler* output;

     Writer writer;

     sys::Thread receiver;

     Rdma::AsynchIO* aio;
     sys::Poller::shared_ptr poller;

     ~RdmaConnector();

     void run();
     void handleClosed();
     bool closeInternal();

     // Callbacks
     void connected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams&);
     void connectionError(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, Rdma::ErrorType);
     void disconnected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&);
     void rejected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams&);

     void readbuff(Rdma::AsynchIO&, Rdma::Buffer*);
     void writebuff(Rdma::AsynchIO&);
     void writeDataBlock(const framing::AMQDataBlock& data);
     void eof(Rdma::AsynchIO&);

     std::string identifier;

     ConnectionImpl* impl;

     void connect(const std::string& host, int port);
     void close();
     void send(framing::AMQFrame& frame);

     void setInputHandler(framing::InputHandler* handler);
     void setShutdownHandler(sys::ShutdownHandler* handler);
     sys::ShutdownHandler* getShutdownHandler() const;
     framing::OutputHandler* getOutputHandler();
     const std::string& getIdentifier() const;

 public:
     RdmaConnector(framing::ProtocolVersion pVersion,
               const ConnectionSettings&,
               ConnectionImpl*);
 };

 // Static constructor which registers connector here
 namespace {
     Connector* create(framing::ProtocolVersion v, const ConnectionSettings& s, ConnectionImpl* c) {
         return new RdmaConnector(v, s, c);
     }

     struct StaticInit {
         StaticInit() {
             Connector::registerFactory("rdma", &create);
             Connector::registerFactory("ib", &create);
         };
     } init;
 }


 RdmaConnector::RdmaConnector(ProtocolVersion ver,
                      const ConnectionSettings& settings,
                      ConnectionImpl* cimpl)
     : maxFrameSize(settings.maxFrameSize),
       version(ver),
       initiated(false),
       polling(false),
       joined(true),
       shutdownHandler(0),
       writer(maxFrameSize, cimpl),
       aio(0),
       impl(cimpl)
 {
     QPID_LOG(debug, "RdmaConnector created for " << version);
 }

 RdmaConnector::~RdmaConnector() {
     close();
 }

 void RdmaConnector::connect(const std::string& host, int port){
     Mutex::ScopedLock l(pollingLock);
     assert(!polling);
     assert(joined);
     poller = Poller::shared_ptr(new Poller);

     // This stuff needs to abstracted out of here to a platform specific file
     ::addrinfo *res;
     ::addrinfo hints;
     hints.ai_flags = 0;
     hints.ai_family = AF_INET;
     hints.ai_socktype = SOCK_STREAM;
     hints.ai_protocol = 0;
     int n = ::getaddrinfo(host.c_str(), boost::lexical_cast<std::string>(port).c_str(), &hints, &res);
     if (n<0) {
         throw Exception(QPID_MSG("Cannot resolve " << host << ": " << ::gai_strerror(n)));
     }

     Rdma::Connector* c = new Rdma::Connector(
         *res->ai_addr,
         Rdma::ConnectionParams(maxFrameSize, Rdma::DEFAULT_WR_ENTRIES),
         boost::bind(&RdmaConnector::connected, this, poller, _1, _2),
         boost::bind(&RdmaConnector::connectionError, this, poller, _1, _2),
         boost::bind(&RdmaConnector::disconnected, this, poller, _1),
         boost::bind(&RdmaConnector::rejected, this, poller, _1, _2));
     c->start(poller);

     polling = true;
     joined = false;
     receiver = Thread(this);
 }

 // The following only gets run when connected
 void RdmaConnector::connected(Poller::shared_ptr poller, Rdma::Connection::intrusive_ptr& ci, const Rdma::ConnectionParams& cp) {
     Rdma::QueuePair::intrusive_ptr q = ci->getQueuePair();

     aio = new Rdma::AsynchIO(ci->getQueuePair(),
         cp.maxRecvBufferSize, cp.initialXmitCredit , Rdma::DEFAULT_WR_ENTRIES,
         boost::bind(&RdmaConnector::readbuff, this, _1, _2),
         boost::bind(&RdmaConnector::writebuff, this, _1),
         0, // write buffers full
         boost::bind(&RdmaConnector::eof, this, _1)); // data error - just close connection
     aio->start(poller);

     identifier = str(format("[%1% %2%]") % ci->getLocalName() % ci->getPeerName());
     writer.init(identifier, aio);
     ProtocolInitiation init(version);
     writeDataBlock(init);
 }

 void RdmaConnector::connectionError(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, Rdma::ErrorType) {
     QPID_LOG(trace, "Connection Error " << identifier);
     eof(*aio);
 }

 void RdmaConnector::disconnected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&) {
     eof(*aio);
 }

 void RdmaConnector::rejected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams& cp) {
     QPID_LOG(trace, "Connection Rejected " << identifier << ": " << cp.maxRecvBufferSize);
     eof(*aio);
 }

 bool RdmaConnector::closeInternal() {
     bool ret;
     {
     Mutex::ScopedLock l(pollingLock);
     ret = polling;
     if (polling) {
         polling = false;
         poller->shutdown();
     }
     if (joined || receiver.id() == Thread::current().id()) {
         return ret;
     }
     joined = true;
     }

     receiver.join();
     return ret;
 }

 void RdmaConnector::close() {
     closeInternal();
 }

 void RdmaConnector::setInputHandler(InputHandler* handler){
     input = handler;
 }

 void RdmaConnector::setShutdownHandler(ShutdownHandler* handler){
     shutdownHandler = handler;
 }

 OutputHandler* RdmaConnector::getOutputHandler(){
     return this;
 }

 sys::ShutdownHandler* RdmaConnector::getShutdownHandler() const {
     return shutdownHandler;
 }

 const std::string& RdmaConnector::getIdentifier() const {
     return identifier;
 }

 void RdmaConnector::send(AMQFrame& frame) {
     writer.handle(frame);
 }

 void RdmaConnector::handleClosed() {
     if (closeInternal() && shutdownHandler)
         shutdownHandler->shutdown();
 }

 RdmaConnector::Writer::Writer(uint16_t s, Bounds* b) :
     maxFrameSize(s),
     aio(0),
     buffer(0),
     lastEof(0),
     bounds(b)
 {
 }

 RdmaConnector::Writer::~Writer() {
     if (aio)
         aio->returnBuffer(buffer);
 }

 void RdmaConnector::Writer::init(std::string id, Rdma::AsynchIO* a) {
     Mutex::ScopedLock l(lock);
     identifier = id;
     aio = a;
     assert(aio->bufferAvailable());
     newBuffer();
 }
 void RdmaConnector::Writer::handle(framing::AMQFrame& frame) {
     Mutex::ScopedLock l(lock);
     frames.push_back(frame);
     // Don't bother to send anything unless we're at the end of a frameset (assembly in 0-10 terminology)
     if (frame.getEof()) {
         lastEof = frames.size();
         QPID_LOG(debug, "Requesting write: lastEof=" << lastEof);
         aio->notifyPendingWrite();
     }
     QPID_LOG(trace, "SENT " << identifier << ": " << frame);
 }

 void RdmaConnector::Writer::writeOne() {
     assert(buffer);
     QPID_LOG(trace, "Write buffer " << encode.getPosition()
              << " bytes " << framesEncoded << " frames ");
     framesEncoded = 0;

     buffer->dataStart = 0;
     buffer->dataCount = encode.getPosition();
     aio->queueWrite(buffer);
     newBuffer();
 }

 void RdmaConnector::Writer::newBuffer() {
     buffer = aio->getBuffer();
     encode = framing::Buffer(buffer->bytes, buffer->byteCount);
     framesEncoded = 0;
 }

 // Called in IO thread. (write idle routine)
 // This is NOT only called in response to previously calling notifyPendingWrite
 void RdmaConnector::Writer::write(Rdma::AsynchIO&) {
     Mutex::ScopedLock l(lock);
     assert(buffer);
     // If nothing to do return immediately
     if (lastEof==0)
         return;
     size_t bytesWritten = 0;
     while (aio->writable() && aio->bufferAvailable() && !frames.empty()) {
         const AMQFrame* frame = &frames.front();
         uint32_t size = frame->encodedSize();
         while (size <= encode.available()) {
             frame->encode(encode);
             frames.pop_front();
             ++framesEncoded;
             bytesWritten += size;
             if (frames.empty())
                 break;
             frame = &frames.front();
             size = frame->encodedSize();
         }
         lastEof -= framesEncoded;
         writeOne();
     }
     if (bounds) bounds->reduce(bytesWritten);
 }

 void RdmaConnector::readbuff(Rdma::AsynchIO&, Rdma::Buffer* buff) {
     framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);

     if (!initiated) {
         framing::ProtocolInitiation protocolInit;
         if (protocolInit.decode(in)) {
             //TODO: check the version is correct
             QPID_LOG(debug, "RECV " << identifier << " INIT(" << protocolInit << ")");
         }
         initiated = true;
     }
     AMQFrame frame;
     while(frame.decode(in)){
         QPID_LOG(trace, "RECV " << identifier << ": " << frame);
         input->received(frame);
     }
 }

 void RdmaConnector::writebuff(Rdma::AsynchIO& aio_) {
     writer.write(aio_);
 }

 void RdmaConnector::writeDataBlock(const AMQDataBlock& data) {
     Rdma::Buffer* buff = aio->getBuffer();
     framing::Buffer out(buff->bytes, buff->byteCount);
     data.encode(out);
     buff->dataCount = data.encodedSize();
     aio->queueWrite(buff);
 }

 void RdmaConnector::eof(Rdma::AsynchIO&) {
     handleClosed();
 }

 // TODO: astitcher 20070908 This version of the code can never time out, so the idle processing
 // will never be called
 void RdmaConnector::run(){
     // Keep the connection impl in memory until run() completes.
     //GRS: currently the ConnectionImpls destructor is where the Io thread is joined
     //boost::shared_ptr<ConnectionImpl> protect = impl->shared_from_this();
     //assert(protect);
     try {
         Dispatcher d(poller);

         //aio->start(poller);
         d.run();
         //aio->queueForDeletion();
     } catch (const std::exception& e) {
         {
         // We're no longer polling
         Mutex::ScopedLock l(pollingLock);
         polling = false;
         }
         QPID_LOG(error, e.what());
         handleClosed();
     }
 }


 }} // namespace qpid::client
	/*
	*
	* 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 "Connector.h"

	#include "Bounds.h"
	#include "ConnectionImpl.h"
	#include "ConnectionSettings.h"
	#include "qpid/log/Statement.h"
	#include "qpid/sys/Time.h"
	#include "qpid/framing/AMQFrame.h"
	#include "qpid/sys/rdma/RdmaIO.h"
	#include "qpid/sys/Dispatcher.h"
	#include "qpid/sys/Poller.h"
	#include "qpid/Msg.h"

	#include <iostream>
	#include <boost/bind.hpp>
	#include <boost/format.hpp>
	#include <boost/lexical_cast.hpp>

	// This stuff needs to abstracted out of here to a platform specific file
	#include <netdb.h>

	namespace qpid {
	namespace client {

	using namespace qpid::sys;
	using namespace qpid::framing;
	using boost::format;
	using boost::str;

	class RdmaConnector : public Connector, private sys::Runnable
	{
	struct Buff;

	/** Batch up frames for writing to aio. */
	class Writer : public framing::FrameHandler {
	typedef Rdma::Buffer BufferBase;
	typedef std::deque<framing::AMQFrame> Frames;

	const uint16_t maxFrameSize;
	sys::Mutex lock;
	Rdma::AsynchIO* aio;
	BufferBase* buffer;
	Frames frames;
	size_t lastEof; // Position after last EOF in frames
	framing::Buffer encode;
	size_t framesEncoded;
	std::string identifier;
	Bounds* bounds;

	void writeOne();
	void newBuffer();

	public:

	Writer(uint16_t maxFrameSize, Bounds*);
	~Writer();
	void init(std::string id, Rdma::AsynchIO*);
	void handle(framing::AMQFrame&);
	void write(Rdma::AsynchIO&);
	};

	const uint16_t maxFrameSize;
	framing::ProtocolVersion version;
	bool initiated;

	sys::Mutex pollingLock;
	bool polling;
	bool joined;

	sys::ShutdownHandler* shutdownHandler;
	framing::InputHandler* input;
	framing::InitiationHandler* initialiser;
	framing::OutputHandler* output;

	Writer writer;

	sys::Thread receiver;

	Rdma::AsynchIO* aio;
	sys::Poller::shared_ptr poller;

	~RdmaConnector();

	void run();
	void handleClosed();
	bool closeInternal();

	// Callbacks
	void connected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams&);
	void connectionError(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, Rdma::ErrorType);
	void disconnected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&);
	void rejected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams&);

	void readbuff(Rdma::AsynchIO&, Rdma::Buffer*);
	void writebuff(Rdma::AsynchIO&);
	void writeDataBlock(const framing::AMQDataBlock& data);
	void eof(Rdma::AsynchIO&);

	std::string identifier;

	ConnectionImpl* impl;

	void connect(const std::string& host, int port);
	void close();
	void send(framing::AMQFrame& frame);

	void setInputHandler(framing::InputHandler* handler);
	void setShutdownHandler(sys::ShutdownHandler* handler);
	sys::ShutdownHandler* getShutdownHandler() const;
	framing::OutputHandler* getOutputHandler();
	const std::string& getIdentifier() const;

	public:
	RdmaConnector(framing::ProtocolVersion pVersion,
	const ConnectionSettings&,
	ConnectionImpl*);
	};

	// Static constructor which registers connector here
	namespace {
	Connector* create(framing::ProtocolVersion v, const ConnectionSettings& s, ConnectionImpl* c) {
	return new RdmaConnector(v, s, c);
	}

	struct StaticInit {
	StaticInit() {
	Connector::registerFactory("rdma", &create);
	Connector::registerFactory("ib", &create);
	};
	} init;
	}


	RdmaConnector::RdmaConnector(ProtocolVersion ver,
	const ConnectionSettings& settings,
	ConnectionImpl* cimpl)
	: maxFrameSize(settings.maxFrameSize),
	version(ver),
	initiated(false),
	polling(false),
	joined(true),
	shutdownHandler(0),
	writer(maxFrameSize, cimpl),
	aio(0),
	impl(cimpl)
	{
	QPID_LOG(debug, "RdmaConnector created for " << version);
	}

	RdmaConnector::~RdmaConnector() {
	close();
	}

	void RdmaConnector::connect(const std::string& host, int port){
	Mutex::ScopedLock l(pollingLock);
	assert(!polling);
	assert(joined);
	poller = Poller::shared_ptr(new Poller);

	// This stuff needs to abstracted out of here to a platform specific file
	::addrinfo *res;
	::addrinfo hints;
	hints.ai_flags = 0;
	hints.ai_family = AF_INET;
	hints.ai_socktype = SOCK_STREAM;
	hints.ai_protocol = 0;
	int n = ::getaddrinfo(host.c_str(), boost::lexical_cast<std::string>(port).c_str(), &hints, &res);
	if (n<0) {
	throw Exception(QPID_MSG("Cannot resolve " << host << ": " << ::gai_strerror(n)));
	}

	Rdma::Connector* c = new Rdma::Connector(
	*res->ai_addr,
	Rdma::ConnectionParams(maxFrameSize, Rdma::DEFAULT_WR_ENTRIES),
	boost::bind(&RdmaConnector::connected, this, poller, _1, _2),
	boost::bind(&RdmaConnector::connectionError, this, poller, _1, _2),
	boost::bind(&RdmaConnector::disconnected, this, poller, _1),
	boost::bind(&RdmaConnector::rejected, this, poller, _1, _2));
	c->start(poller);

	polling = true;
	joined = false;
	receiver = Thread(this);
	}

	// The following only gets run when connected
	void RdmaConnector::connected(Poller::shared_ptr poller, Rdma::Connection::intrusive_ptr& ci, const Rdma::ConnectionParams& cp) {
	Rdma::QueuePair::intrusive_ptr q = ci->getQueuePair();

	aio = new Rdma::AsynchIO(ci->getQueuePair(),
	cp.maxRecvBufferSize, cp.initialXmitCredit , Rdma::DEFAULT_WR_ENTRIES,
	boost::bind(&RdmaConnector::readbuff, this, _1, _2),
	boost::bind(&RdmaConnector::writebuff, this, _1),
	0, // write buffers full
	boost::bind(&RdmaConnector::eof, this, _1)); // data error - just close connection
	aio->start(poller);

	identifier = str(format("[%1% %2%]") % ci->getLocalName() % ci->getPeerName());
	writer.init(identifier, aio);
	ProtocolInitiation init(version);
	writeDataBlock(init);
	}

	void RdmaConnector::connectionError(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, Rdma::ErrorType) {
	QPID_LOG(trace, "Connection Error " << identifier);
	eof(*aio);
	}

	void RdmaConnector::disconnected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&) {
	eof(*aio);
	}

	void RdmaConnector::rejected(sys::Poller::shared_ptr, Rdma::Connection::intrusive_ptr&, const Rdma::ConnectionParams& cp) {
	QPID_LOG(trace, "Connection Rejected " << identifier << ": " << cp.maxRecvBufferSize);
	eof(*aio);
	}

	bool RdmaConnector::closeInternal() {
	bool ret;
	{
	Mutex::ScopedLock l(pollingLock);
	ret = polling;
	if (polling) {
	polling = false;
	poller->shutdown();
	}
	if (joined \|\| receiver.id() == Thread::current().id()) {
	return ret;
	}
	joined = true;
	}

	receiver.join();
	return ret;
	}

	void RdmaConnector::close() {
	closeInternal();
	}

	void RdmaConnector::setInputHandler(InputHandler* handler){
	input = handler;
	}

	void RdmaConnector::setShutdownHandler(ShutdownHandler* handler){
	shutdownHandler = handler;
	}

	OutputHandler* RdmaConnector::getOutputHandler(){
	return this;
	}

	sys::ShutdownHandler* RdmaConnector::getShutdownHandler() const {
	return shutdownHandler;
	}

	const std::string& RdmaConnector::getIdentifier() const {
	return identifier;
	}

	void RdmaConnector::send(AMQFrame& frame) {
	writer.handle(frame);
	}

	void RdmaConnector::handleClosed() {
	if (closeInternal() && shutdownHandler)
	shutdownHandler->shutdown();
	}

	RdmaConnector::Writer::Writer(uint16_t s, Bounds* b) :
	maxFrameSize(s),
	aio(0),
	buffer(0),
	lastEof(0),
	bounds(b)
	{
	}

	RdmaConnector::Writer::~Writer() {
	if (aio)
	aio->returnBuffer(buffer);
	}

	void RdmaConnector::Writer::init(std::string id, Rdma::AsynchIO* a) {
	Mutex::ScopedLock l(lock);
	identifier = id;
	aio = a;
	assert(aio->bufferAvailable());
	newBuffer();
	}
	void RdmaConnector::Writer::handle(framing::AMQFrame& frame) {
	Mutex::ScopedLock l(lock);
	frames.push_back(frame);
	// Don't bother to send anything unless we're at the end of a frameset (assembly in 0-10 terminology)
	if (frame.getEof()) {
	lastEof = frames.size();
	QPID_LOG(debug, "Requesting write: lastEof=" << lastEof);
	aio->notifyPendingWrite();
	}
	QPID_LOG(trace, "SENT " << identifier << ": " << frame);
	}

	void RdmaConnector::Writer::writeOne() {
	assert(buffer);
	QPID_LOG(trace, "Write buffer " << encode.getPosition()
	<< " bytes " << framesEncoded << " frames ");
	framesEncoded = 0;

	buffer->dataStart = 0;
	buffer->dataCount = encode.getPosition();
	aio->queueWrite(buffer);
	newBuffer();
	}

	void RdmaConnector::Writer::newBuffer() {
	buffer = aio->getBuffer();
	encode = framing::Buffer(buffer->bytes, buffer->byteCount);
	framesEncoded = 0;
	}

	// Called in IO thread. (write idle routine)
	// This is NOT only called in response to previously calling notifyPendingWrite
	void RdmaConnector::Writer::write(Rdma::AsynchIO&) {
	Mutex::ScopedLock l(lock);
	assert(buffer);
	// If nothing to do return immediately
	if (lastEof==0)
	return;
	size_t bytesWritten = 0;
	while (aio->writable() && aio->bufferAvailable() && !frames.empty()) {
	const AMQFrame* frame = &frames.front();
	uint32_t size = frame->encodedSize();
	while (size <= encode.available()) {
	frame->encode(encode);
	frames.pop_front();
	++framesEncoded;
	bytesWritten += size;
	if (frames.empty())
	break;
	frame = &frames.front();
	size = frame->encodedSize();
	}
	lastEof -= framesEncoded;
	writeOne();
	}
	if (bounds) bounds->reduce(bytesWritten);
	}

	void RdmaConnector::readbuff(Rdma::AsynchIO&, Rdma::Buffer* buff) {
	framing::Buffer in(buff->bytes+buff->dataStart, buff->dataCount);

	if (!initiated) {
	framing::ProtocolInitiation protocolInit;
	if (protocolInit.decode(in)) {
	//TODO: check the version is correct
	QPID_LOG(debug, "RECV " << identifier << " INIT(" << protocolInit << ")");
	}
	initiated = true;
	}
	AMQFrame frame;
	while(frame.decode(in)){
	QPID_LOG(trace, "RECV " << identifier << ": " << frame);
	input->received(frame);
	}
	}

	void RdmaConnector::writebuff(Rdma::AsynchIO& aio_) {
	writer.write(aio_);
	}

	void RdmaConnector::writeDataBlock(const AMQDataBlock& data) {
	Rdma::Buffer* buff = aio->getBuffer();
	framing::Buffer out(buff->bytes, buff->byteCount);
	data.encode(out);
	buff->dataCount = data.encodedSize();
	aio->queueWrite(buff);
	}

	void RdmaConnector::eof(Rdma::AsynchIO&) {
	handleClosed();
	}

	// TODO: astitcher 20070908 This version of the code can never time out, so the idle processing
	// will never be called
	void RdmaConnector::run(){
	// Keep the connection impl in memory until run() completes.
	//GRS: currently the ConnectionImpls destructor is where the Io thread is joined
	//boost::shared_ptr<ConnectionImpl> protect = impl->shared_from_this();
	//assert(protect);
	try {
	Dispatcher d(poller);

	//aio->start(poller);
	d.run();
	//aio->queueForDeletion();
	} catch (const std::exception& e) {
	{
	// We're no longer polling
	Mutex::ScopedLock l(pollingLock);
	polling = false;
	}
	QPID_LOG(error, e.what());
	handleClosed();
	}
	}


	}} // namespace qpid::client