drain on closeChannel

Signed-off-by: dorjesinpo <129227380+dorjesinpo@users.noreply.github.com>

Author: dorjesinpo

src/groups/mqb/mqbnet/mqbnet_channel.cpp

@@ -81,10 +81,11 @@ Channel::Channel(bdlbb::BlobBufferFactory* blobBufferFactory,
              d_allocators.get("ItemPool"))
 , d_buffer(1024, allocator)
 , d_doStop(false)
-, d_state(e_INITIAL)
+, d_state(e_RESET)
 , d_description(name + " - ", d_allocator_p)
 , d_name(name, d_allocator_p)
 , d_stats()
+, d_isClosing(false)
 {
     bslmt::ThreadAttributes attr = bmqsys::ThreadUtil::defaultAttributes();
     bsl::string             threadName("bmqNet-");
@@ -261,19 +262,14 @@ void Channel::resetChannel()
 
 void Channel::closeChannel()
 {
-    bsl::shared_ptr<bmqio::Channel> channel;
-    {
-        bslmt::LockGuard<bslmt::Mutex> guard(&d_mutex);  // LOCK
-        channel = d_channel_wp.lock();
-    }  // UNLOCK
+    d_isClosing.store(true);
 
-    if (BSLS_PERFORMANCEHINT_PREDICT_LIKELY(channel)) {
-        channel->close();
-        // Set the state to e_CLOSE to avoid repeated attempts to write until
-        // 'OnClose' calls 'resetChannel'.
+    // No need to 'signal' to interrupt the waiting.
+    // Waiting means there is no connection and therefore nothing to close.
 
-        d_state.testAndSwap(e_READY, e_CLOSE);
-    }
+    wakeUp();
+
+    // do not wait for all items to drain.
 }
 
 void Channel::setChannel(const bsl::weak_ptr<bmqio::Channel>& value)
@@ -747,7 +743,7 @@ void Channel::threadFn()
     bsl::string                     description;
     int                             mode = e_BLOCK;
 
-    BSLS_ASSERT(d_state == e_INITIAL || d_state == e_RESET);
+    BSLS_ASSERT(d_state == e_RESET);
 
     while (!d_doStop) {
         bmqc::MonitoredQueueState::Enum queueState;
@@ -796,17 +792,14 @@ void Channel::threadFn()
                 item.reset();
                 reset();
 
-                d_state = e_INITIAL;
                 mode    = e_BLOCK;
-            }
+                d_isClosing.store(false);
 
-            if (d_state == e_INITIAL) {
                 channel     = d_channel_wp.lock();
                 description = d_description;
 
                 if (channel) {
-                    // This is the only place for transitions:
-                    //  e_IDLE  -> e_READY
+                    // This is the only place for the transition
                     //  e_RESET -> e_READY
                     d_state = e_READY;
                 }
@@ -856,6 +849,13 @@ void Channel::threadFn()
                     // BLOCK mode and wait for the next batch of items.
                     mode = e_BLOCK;
                 }
+                // if draining, this is where it stops.
+                // Does not matter if 'flushAll' has failed; must close
+                if (d_isClosing) {
+                    channel->close();
+                    d_state.testAndSwap(e_READY, e_CLOSE);
+                    // bmqio::Channel observer will trigger 'resetChannel'
+                }
             } break;
             default: {
                 BSLS_ASSERT(false && "Unreachable by design");

src/groups/mqb/mqbnet/mqbnet_channel.h

@@ -48,7 +48,6 @@
 // Internally, 'mqbnet::Channel' starts a new thread and unconditionally
 // buffers everything it needs to write into single consumer queue.  The thread
 // reads the channel state which can be
-//  - e_INITIAL, not connected
 //  - e_RESET,   indicating a need to reset because of a connection change;
 //  - e_READY,   low-watermark
 //  - e_HWM,     high-watermark
@@ -349,17 +348,15 @@ class Channel {
   public:
     // PUBLIC TYPES
     enum EnumState {
-        /// Not connected
-        e_INITIAL = 0,
         /// Need resetting because of a connection change
-        e_RESET = 1,
+        e_RESET = 0,
         /// Between 'Channel::close; and 'resetChannel'
-        e_CLOSE = 2,
-        e_READY = 3,
+        e_CLOSE = 1,
+        e_READY = 2,
         /// LWM
-        e_LWM = 4,
+        e_LWM = 3,
         /// HWM
-        e_HWM = 5
+        e_HWM = 4
     };
 
   private:
@@ -422,6 +419,10 @@ class Channel {
     // in the internal thread.
     Stats d_stats;
 
+    /// Indicates graceful shutdown.  Drain the buffer if possible and then
+    /// close the channel.
+    bsls::AtomicBool d_isClosing;
+
   private:
     // NOT IMPLEMENTED
     Channel(const Channel&) BSLS_CPP11_DELETED;
@@ -1085,7 +1086,7 @@ Channel::enqueue(bslma::ManagedPtr<Item>& item)
 // ACCESSORS
 inline bool Channel::isAvailable() const
 {
-    return d_state != e_INITIAL && d_state != e_RESET && d_state != e_CLOSE;
+    return d_state != e_RESET && d_state != e_CLOSE && !d_isClosing;
 }
 
 inline const bsl::shared_ptr<bmqio::Channel> Channel::channel() const