EventChannel.cpp: Fixed error handling problems causing EBADF exceptions.

git-svn-id: https://svn.apache.org/repos/asf/incubator/qpid/branches/event-queue-2006-12-20@489225 13f79535-47bb-0310-9956-ffa450edef68

2 files changed, 26 insertions, 38 deletions

diff --git a/cpp/lib/common/sys/posix/EventChannel.cpp b/cpp/lib/common/sys/posix/EventChannel.cpp
index 37e3c2257a..26437484f8 100644
--- a/cpp/lib/common/sys/posix/EventChannel.cpp
+++ b/cpp/lib/common/sys/posix/EventChannel.cpp
@@ -211,9 +211,11 @@ Event* EventChannel::Queue::wake(uint32_t epollFlags) {
         assert(!queue.empty());
         Event* e = queue.front();
         assert(e);
-        // TODO aconway 2006-12-20: Can/should we move event completion
-        // out into dispatch() so it doesn't happen in Descriptor locks?
-        e->complete(descriptor);
+        if (!e->getException()) {
+            // TODO aconway 2006-12-20: Can/should we move event completion
+            // out into dispatch() so it doesn't happen in Descriptor locks?
+            e->complete(descriptor);
+        }
         queue.pop_front();
         return e;
     }
@@ -236,9 +238,9 @@ void EventChannel::Queue::shutdown() {
 
 void EventChannel::Descriptor::activate(int epollFd_, int myFd_) {
     Mutex::ScopedLock l(lock);
-    assert(myFd < 0 || (myFd == myFd_)); // Can't change fd.
-    if (epollFd < 0) {          // Means we're not polling.
-        epollFd = epollFd_;
+    assert(isShutdown() || (myFd == myFd_)); // Can't change fd while running.
+    if (isShutdown()) {
+        epollFd = epollFd_;     // We're back in business.
         myFd = myFd_;
         epollCtl(EPOLL_CTL_ADD, 0);
     }
@@ -252,14 +254,15 @@ void EventChannel::Descriptor::shutdown() {
 void EventChannel::Descriptor::shutdownUnsafe() {
     // Caller holds lock.
     ::close(myFd);
-    epollFd = -1;               // Indicate we are not polling.
+    epollFd = -1;               // Mark myself as shutdown.
     inQueue.shutdown();
     outQueue.shutdown();
-    epollCtl(EPOLL_CTL_DEL, 0);
 }
 
 void EventChannel::Descriptor::update() {
     // Caller holds lock.
+    if (isShutdown())             // Nothing to do
+        return;
     uint32_t events =  EPOLLONESHOT | EPOLLERR | EPOLLHUP;
     inQueue.setBit(events);
     outQueue.setBit(events);
@@ -268,6 +271,7 @@ void EventChannel::Descriptor::update() {
     
 void EventChannel::Descriptor::epollCtl(int op, uint32_t events) {
     // Caller holds lock
+    assert(!isShutdown());
     CleanStruct<epoll_event> ee;
     ee.data.ptr = this;
     ee.events = events;
@@ -279,13 +283,16 @@ void EventChannel::Descriptor::epollCtl(int op, uint32_t events) {
 
 EventPair EventChannel::Descriptor::wake(uint32_t epollEvents) {
     Mutex::ScopedLock l(lock);
-    // If we have an error:
+    // On error, shut down the Descriptor and both queues.
     if (epollEvents & (EPOLLERR | EPOLLHUP)) {
         shutdownUnsafe();
-        // Complete both sides on error so the event can fail and
-        // mark itself with an exception.
-        epollEvents |= EPOLLIN | EPOLLOUT;
+        // TODO aconway 2006-12-20: This error handling models means
+        // that any error reported by epoll will result in a shutdown
+        // exception on the events. Can we get more accurate error
+        // reporting somehow?
     }
+    // Check the queues even if shutdown - we want to drain the
+    // events that have been marked with exceptions.
     EventPair ready(inQueue.wake(epollEvents), outQueue.wake(epollEvents));
     update();
     return ready;
diff --git a/cpp/tests/EventChannelTest.cpp b/cpp/tests/EventChannelTest.cpp
index bba0a9cd9b..46eb7d231b 100644
--- a/cpp/tests/EventChannelTest.cpp
+++ b/cpp/tests/EventChannelTest.cpp
@@ -30,7 +30,7 @@
 #include <netinet/in.h>
 #include <netdb.h>
 #include <iostream>
-
+using namespace std;
 using namespace qpid::sys;
 
 
@@ -116,39 +116,19 @@ class EventChannelTest : public CppUnit::TestCase
         CPPUNIT_ASSERT_EQUAL(std::string(hello, size/2),
                              std::string(readBuf, size/2));
     }
-    
 
-    void testFailedRead() 
-    {
+    void testFailedRead() {
         ReadEvent re(pipe[0], readBuf, size);
         ec->post(re);
-
-        // EOF before all data read.
+        // Close the write end while reading, causes a HUP.
         ::close(pipe[1]);
         CPPUNIT_ASSERT(isNextEvent(re));
         CPPUNIT_ASSERT(re.getException());
-        try {
-            re.throwIfException();
-            CPPUNIT_FAIL("Expected QpidError.");
-        }
-        catch (const qpid::QpidError&) { }
 
-        
-        //  Try to read from closed file descriptor.
+        // Try to read from closed fd. Fails in post() and throws.
         try {
             ec->post(re);
-            CPPUNIT_ASSERT(isNextEvent(re));
-            re.throwIfException();
-            CPPUNIT_FAIL("Expected an exception.");
-        }
-        catch (const qpid::QpidError&) {}
-        
-        //  Bad file descriptor. Note in this case we fail
-        //  in post and throw immediately.
-        try {
-            ReadEvent bad(-1, readBuf, size);
-            ec->post(bad);
-            CPPUNIT_FAIL("Expected QpidError.");
+            CPPUNIT_FAIL("Expected exception");
         }
         catch (const qpid::QpidError&) {}
     }
@@ -164,8 +144,9 @@ class EventChannelTest : public CppUnit::TestCase
 
     void testFailedWrite() {
         WriteEvent wr(pipe[1], hello, size);
-        ::close(pipe[0]);
         ec->post(wr);
+        // Close the read end while writing.
+        ::close(pipe[0]);
         CPPUNIT_ASSERT(isNextEvent(wr));
         CPPUNIT_ASSERT(wr.getException());
     }