* More work on asychronous network IO

* Fix of current EventQueue code to carry on compiling


git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/trunk/qpid@552001 13f79535-47bb-0310-9956-ffa450edef68

3 files changed, 208 insertions, 21 deletions

diff --git a/cpp/src/qpid/sys/posix/AsynchIO.cpp b/cpp/src/qpid/sys/posix/AsynchIO.cpp
new file mode 100644
index 0000000000..400c2080b2
--- /dev/null
+++ b/cpp/src/qpid/sys/posix/AsynchIO.cpp
@@ -0,0 +1,195 @@
+/*
+ *
+ * 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/AsynchIO.h"
+
+#include "check.h"
+
+#include <unistd.h>
+#include <fcntl.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <errno.h>
+
+#include <boost/bind.hpp>
+
+using namespace qpid::sys;
+
+namespace {
+
+/*
+ * Make file descriptor non-blocking
+ */
+void nonblocking(int fd) {
+    QPID_POSIX_CHECK(::fcntl(fd, F_SETFL, O_NONBLOCK));
+}
+
+}
+
+/*
+ * Asynch Acceptor
+ */
+
+AsynchAcceptor::AsynchAcceptor(int fd, Callback callback) :
+    acceptedCallback(callback),
+    handle(fd, boost::bind(&AsynchAcceptor::readable, this, _1), 0) {
+
+    nonblocking(fd);
+}
+
+void AsynchAcceptor::start(Poller::shared_ptr poller) {
+    handle.watch(poller);
+}
+
+/*
+ * We keep on accepting as long as there is something to accept
+ */
+void AsynchAcceptor::readable(DispatchHandle& h) {
+    int afd;
+    do {
+        errno = 0;
+        // TODO: Currently we ignore the peers address, perhaps we should
+        // log it or use it for connection acceptance.
+        afd = ::accept(h.getFD(), 0, 0);
+        if (afd >= 0) {
+            acceptedCallback(afd);
+        } else if (errno == EAGAIN) {
+            break;
+        } else {
+            QPID_POSIX_CHECK(afd);
+        }
+    } while (true);
+
+    h.rewatch();
+}
+
+/*
+ * Asynch reader/writer
+ */
+AsynchIO::AsynchIO(int fd, ReadCallback rCb, EofCallback eofCb, BuffersEmptyCallback eCb, IdleCallback iCb) :
+    readCallback(rCb),
+    eofCallback(eofCb),
+    emptyCallback(eCb),
+    idleCallback(iCb),
+    handle(fd, boost::bind(&AsynchIO::readable, this, _1), boost::bind(&AsynchIO::writeable, this, _1)) {
+
+    nonblocking(fd);
+}
+
+void AsynchIO::start(Poller::shared_ptr poller) {
+    handle.watch(poller);
+}
+
+void AsynchIO::QueueReadBuffer(const Buffer& buff) {
+    bufferQueue.push_front(buff);
+    handle.rewatchRead();
+}
+
+void AsynchIO::QueueWrite(const Buffer& buff) {
+    writeQueue.push_front(buff);
+    handle.rewatchWrite();
+}
+
+/*
+ * We keep on reading as long as we have something to read and a buffer to put
+ * it in
+ */
+void AsynchIO::readable(DispatchHandle& h) {
+    do {
+        // (Try to) get a buffer
+        if (!bufferQueue.empty()) {
+            // Read into buffer
+            Buffer buff = bufferQueue.back();
+            bufferQueue.pop_back();
+            errno = 0;
+            int rc = ::read(h.getFD(), buff.bytes, buff.byteCount);
+            if (rc == 0) {
+                eofCallback();    
+            } else if (rc > 0) {
+                readCallback(buff, rc);
+            } else {
+                // Put buffer back
+                bufferQueue.push_back(buff);
+                
+                if (errno == EAGAIN) {
+                    // We must have just put a buffer back so we know
+                    // we can do this
+                    h.rewatchRead();
+                    return;
+                } else {
+                    QPID_POSIX_CHECK(rc);
+                }
+            }
+        } else {
+            // Something to read but no buffer
+            if (emptyCallback) {
+                emptyCallback();
+            }
+            // If we still have no buffers we can't do anything more
+            if (bufferQueue.empty()) {
+                return;
+            }
+            
+        }
+    } while (true);
+}
+
+/*
+ * We carry on writing whilst we have data to write and we can write
+ */
+void AsynchIO::writeable(DispatchHandle& h) {
+    do {
+        // See if we've got something to write
+        if (!writeQueue.empty()) {
+            // Write buffer
+            Buffer buff = writeQueue.back();
+            writeQueue.pop_back();
+            errno = 0;
+            int rc = ::write(h.getFD(), buff.bytes, buff.byteCount);
+            if (rc >= 0) {
+                // Recycle the buffer
+                QueueReadBuffer(buff);
+            } else {
+                // Put buffer back
+                writeQueue.push_back(buff);
+                
+                if (errno == EAGAIN) {
+                    // We have just put a buffer back so we know
+                    // we can do this
+                    h.rewatchWrite();
+                    return;
+                } else {
+                    QPID_POSIX_CHECK(rc);
+                }
+            }
+        } else {
+            // Something to read but no buffer
+            if (idleCallback) {
+                idleCallback(h.getFD());
+            }
+            // If we still have no buffers to write we can't do anything more
+            if (writeQueue.empty()) {
+                return;
+            }
+        }
+    } while (true);
+}
+        
diff --git a/cpp/src/qpid/sys/posix/EventChannelConnection.cpp b/cpp/src/qpid/sys/posix/EventChannelConnection.cpp
index a36f096a4d..f4b6396dd1 100644
--- a/cpp/src/qpid/sys/posix/EventChannelConnection.cpp
+++ b/cpp/src/qpid/sys/posix/EventChannelConnection.cpp
@@ -62,11 +62,10 @@ EventChannelConnection::EventChannelConnection(
 }
 
 
-void EventChannelConnection::send(std::auto_ptr<AMQFrame> frame) {
+void EventChannelConnection::send(AMQFrame& frame) {
     {
         Monitor::ScopedLock lock(monitor);
-        assert(frame.get());
-        writeFrames.push_back(frame.release());
+        writeFrames.push_back(frame);
     }
     closeOnException(&EventChannelConnection::startWrite);
 }
@@ -119,7 +118,6 @@ void EventChannelConnection::closeOnException(MemberFnPtr f)
 // Called by endWrite and send, but only one thread writes at a time.
 // 
 void EventChannelConnection::startWrite() {
-    FrameQueue::auto_type frame;
     {
         Monitor::ScopedLock lock(monitor);
         // Stop if closed or a write event is already in progress.
@@ -130,14 +128,15 @@ void EventChannelConnection::startWrite() {
             return;
         }
         isWriting = true;
-        frame = writeFrames.pop_front();
+        AMQFrame& frame = writeFrames.front();
+        writeFrames.pop_front();
+	    // No need to lock here - only one thread can be writing at a time.
+	    out.clear();
+	    if (isTrace)
+	        cout << "Send on socket " << writeFd << ": " << frame << endl;
+	    frame.encode(out);
+	    out.flip();
     }
-    // No need to lock here - only one thread can be writing at a time.
-    out.clear();
-    if (isTrace)
-        cout << "Send on socket " << writeFd << ": " << *frame << endl;
-    frame->encode(out);
-    out.flip();
     // TODO: AMS 1/6/07 This only works because we already have the correct fd
     // in the descriptor - change not to use assigment
     writeEvent = WriteEvent(
@@ -225,11 +224,10 @@ void EventChannelConnection::endRead() {
         in.flip();
         AMQFrame frame;
         while (frame.decode(in)) {
-            // TODO aconway 2006-11-30: received should take Frame&
             if (isTrace)
                 cout << "Received on socket " << readFd
                      << ": " << frame << endl;
-            handler->received(&frame); 
+            handler->received(frame); 
         }
         in.compact();
         startRead();
diff --git a/cpp/src/qpid/sys/posix/EventChannelConnection.h b/cpp/src/qpid/sys/posix/EventChannelConnection.h
index 394df55fd9..a4ca5de517 100644
--- a/cpp/src/qpid/sys/posix/EventChannelConnection.h
+++ b/cpp/src/qpid/sys/posix/EventChannelConnection.h
@@ -50,17 +50,11 @@ class EventChannelConnection : public ConnectionOutputHandler {
         bool isTrace = false
     );
 
-    // TODO aconway 2006-11-30: SessionContext::send should take auto_ptr
-    virtual void send(qpid::framing::AMQFrame* frame) {
-        send(std::auto_ptr<qpid::framing::AMQFrame>(frame));
-    }
-            
-    virtual void send(std::auto_ptr<qpid::framing::AMQFrame> frame);
-
+    virtual void send(qpid::framing::AMQFrame& frame);
     virtual void close();
 
   private:
-    typedef boost::ptr_deque<qpid::framing::AMQFrame> FrameQueue;
+    typedef std::deque<qpid::framing::AMQFrame> FrameQueue;
     typedef void (EventChannelConnection::*MemberFnPtr)();
     struct ScopedBusy;