Gave the pipe the ability to disable dequeue operations.

--HG--
extra : convert_revision : svn%3Afdd8eb12-d10e-0410-9acb-85c331704f74/trunk%404268

dlib/pipe/pipe_kernel_1.h

@@ -97,6 +97,15 @@ namespace dlib
         void enable_enqueue (
         );
 
+        bool is_dequeue_enabled (
+        ) const;
+
+        void disable_dequeue (
+        );
+
+        void enable_dequeue (
+        );
+
         bool is_enabled (
         ) const;
 
@@ -144,6 +153,7 @@ namespace dlib
         mutable unsigned long enqueue_waiters;
         mutable unsigned long unblock_sig_waiters;
         bool enqueue_enabled;
+        bool dequeue_enabled;
 
         // restricted functions
         pipe(const pipe&);        // copy constructor
@@ -176,7 +186,8 @@ namespace dlib
         dequeue_waiters(0),
         enqueue_waiters(0),
         unblock_sig_waiters(0),
-        enqueue_enabled(true)
+        enqueue_enabled(true),
+        dequeue_enabled(true)
     {
     }
 
@@ -236,7 +247,7 @@ namespace dlib
         // this function is sort of like a call to enqueue so treat it like that
         ++enqueue_waiters;
 
-        while (pipe_size > 0 && enabled )
+        while (pipe_size > 0 && enabled && dequeue_enabled )
             enqueue_sig.wait();
 
         // let the destructor know we are ending if it is blocked waiting
@@ -424,11 +435,11 @@ namespace dlib
         }
 
         // wait until there is something in the pipe or we are disabled 
-        while (pipe_size == 0 && enabled &&
+        while (pipe_size == 0 && enabled && dequeue_enabled &&
                !(pipe_max_size == 0 && first == 0 && last == 0) )
             dequeue_sig.wait();
 
-        if (enabled == false)
+        if (enabled == false || dequeue_enabled == false)
         {
             --dequeue_waiters;
             // let the destructor know we are unblocking
@@ -581,7 +592,7 @@ namespace dlib
 
         bool timed_out = false;
         // wait until there is something in the pipe or we are disabled or we timeout.
-        while (pipe_size == 0 && enabled && 
+        while (pipe_size == 0 && enabled && dequeue_enabled &&
                !(pipe_max_size == 0 && first == 0 && last == 0) )
         {
             if (timeout == 0 || dequeue_sig.wait_or_timeout(timeout) == false)
@@ -591,7 +602,7 @@ namespace dlib
             }
         }
 
-        if (enabled == false || timed_out)
+        if (enabled == false || timed_out || dequeue_enabled == false)
         {
             --dequeue_waiters;
             // let the destructor know we are unblocking
@@ -636,7 +647,7 @@ namespace dlib
         auto_mutex M(m);
         ++unblock_sig_waiters;
 
-        while ( (dequeue_waiters < num || pipe_size != 0) && enabled)
+        while ( (dequeue_waiters < num || pipe_size != 0) && enabled && dequeue_enabled)
             unblock_sig.wait();
 
         // let the destructor know we are ending if it is blocked waiting
@@ -687,6 +698,47 @@ namespace dlib
         enqueue_enabled = true;
     }
 
+// ----------------------------------------------------------------------------------------
+
+    template <
+        typename T
+        >
+    bool pipe<T>::
+    is_dequeue_enabled (
+    ) const
+    {
+        auto_mutex M(m);
+        return dequeue_enabled;
+    }
+
+// ----------------------------------------------------------------------------------------
+
+    template <
+        typename T
+        >
+    void pipe<T>::
+    disable_dequeue (
+    )
+    {
+        auto_mutex M(m);
+        dequeue_enabled = false;
+        dequeue_sig.broadcast();
+    }
+
+
+// ----------------------------------------------------------------------------------------
+
+    template <
+        typename T
+        >
+    void pipe<T>::
+    enable_dequeue (
+    )
+    {
+        auto_mutex M(m);
+        dequeue_enabled = true;
+    }
+
 // ----------------------------------------------------------------------------------------
 
 }

dlib/pipe/pipe_kernel_abstract.h

@@ -22,6 +22,7 @@ namespace dlib
                 size() == 0
                 is_enabled() == true
                 is_enqueue_enabled() == true
+                is_dequeue_enabled() == true
 
             WHAT THIS OBJECT REPRESENTS
                 This is a first in first out queue with a fixed maximum size containing 
@@ -64,6 +65,26 @@ namespace dlib
                 - #is_enabled() == true
         !*/
 
+        void disable (
+        );
+        /*!
+            ensures
+                - #is_enabled() == false
+                - causes all current and future calls to enqueue(), dequeue(),
+                  enqueue_or_timeout() and dequeue_or_timeout() to not block but 
+                  to return false immediately until enable() is called.
+                - causes all current and future calls to wait_until_empty() and
+                  wait_for_num_blocked_dequeues() to not block but return
+                  immediately until enable() is called.
+        !*/
+
+        bool is_enabled (
+        ) const;
+        /*!
+            ensures
+                - returns true if this pipe is currently enabled, false otherwise.
+        !*/
+
         void empty (
         );
         /*!
@@ -78,6 +99,7 @@ namespace dlib
                 - blocks until one of the following is the case:
                     - size() == 0  
                     - is_enabled() == false
+                    - is_dequeue_enabled() == false
         !*/
 
         void wait_for_num_blocked_dequeues (
@@ -90,6 +112,7 @@ namespace dlib
                       to dequeue() and dequeue_or_timeout() is greater than 
                       or equal to num.
                     - is_enabled() == false
+                    - is_dequeue_enabled() == false
         !*/
 
         bool is_enqueue_enabled (
@@ -122,24 +145,34 @@ namespace dlib
                 - #is_enqueue_enabled() == true
         !*/
 
-        void disable (
+        bool is_dequeue_enabled (
+        ) const;
+        /*!
+            ensures
+                - returns true if the dequeue() and dequeue_or_timeout() functions are 
+                  currently enabled, returns false otherwise.  (note that the higher 
+                  level is_enabled() function can overrule this one.  So if 
+                  is_enabled() == false then dequeue functions are still disabled even
+                  if is_dequeue_enabled() returns true.  But if is_dequeue_enabled() == false 
+                  then dequeue functions are always disabled no matter the state of 
+                  is_enabled())
+        !*/
+
+        void disable_dequeue (
         );
         /*!
             ensures
-                - #is_enabled() == false
-                - causes all current and future calls to enqueue(), dequeue(),
-                  enqueue_or_timeout() and dequeue_or_timeout() to not block but 
-                  to return false immediately until enable() is called.
-                - causes all current and future calls to wait_until_empty() and
-                  wait_for_num_blocked_dequeues() to not block but return
-                  immediately until enable() is called.
+                - #is_dequeue_enabled() == false 
+                - causes all current and future calls to dequeue() and
+                  dequeue_or_timeout() to not block but to return false 
+                  immediately until enable_dequeue() is called.
         !*/
 
-        bool is_enabled (
-        ) const;
+        void enable_dequeue (
+        );
         /*!
             ensures
-                - returns true if this pipe is currently enabled, false otherwise.
+                - #is_dequeue_enabled() == true
         !*/
 
         unsigned long max_size (
@@ -227,6 +260,7 @@ namespace dlib
                         - max_size() == 0 and another thread is trying to enqueue an item 
                           onto this pipe and we can receive our item directly from that thread.  
                         - someone calls disable()
+                        - someone calls disable_dequeue()
                 - else
                     - this call does not block.
                 - if (this call to dequeue() returns true) then
@@ -250,6 +284,7 @@ namespace dlib
                         - max_size() == 0 and another thread is trying to enqueue an item onto this 
                           pipe and we can receive our item directly from that thread.  
                         - someone calls disable() 
+                        - someone calls disable_dequeue()
                         - timeout milliseconds passes
                 - else
                     - this call does not block.

dlib/test/pipe.cpp

@@ -191,6 +191,10 @@ namespace
         DLIB_TEST(test_1.size() == 0);
         DLIB_TEST(test2_1.size() == 0);
 
+        DLIB_TEST(test.is_enqueue_enabled() == true);
+        DLIB_TEST(test.is_dequeue_enabled() == true);
+        DLIB_TEST(test.is_enabled() == true);
+
         test.empty();
         test2.empty();
         DLIB_TEST(test.size() == 0);
@@ -515,6 +519,67 @@ namespace
         DLIB_TEST(test_1.size() == 0);
 
         DLIB_TEST(found_error == false);
+
+
+
+
+        {
+            test.enable();
+            test.enable_enqueue();
+            test.empty();
+            DLIB_TEST(test.size() == 0);
+            DLIB_TEST(test.is_enabled() == true);
+            DLIB_TEST(test.is_enqueue_enabled() == true);
+            DLIB_TEST(test.is_dequeue_enabled() == true);
+            test.disable_dequeue();
+            dlog << LINFO << "Make sure disable_dequeue() works right...";
+            DLIB_TEST(test.is_dequeue_enabled() == false);
+            DLIB_TEST(test.dequeue(a) == false);
+            test.wait_until_empty();
+            a = 4;
+            test.enqueue(a);
+            test.wait_until_empty();
+            test.wait_for_num_blocked_dequeues(4);
+            DLIB_TEST(test.size() == 1);
+            DLIB_TEST(test.dequeue(a) == false);
+            DLIB_TEST(test.dequeue_or_timeout(a,10000) == false);
+            DLIB_TEST(test.size() == 1);
+            a = 0;
+            test.enable_dequeue();
+            DLIB_TEST(test.is_dequeue_enabled() == true);
+            DLIB_TEST(test.dequeue(a) == true);
+            DLIB_TEST(a == 4);
+            test_1.wait_until_empty();
+        }
+        {
+            test_1.enable();
+            test_1.enable_enqueue();
+            test_1.empty();
+            DLIB_TEST(test_1.size() == 0);
+            DLIB_TEST(test_1.is_enabled() == true);
+            DLIB_TEST(test_1.is_enqueue_enabled() == true);
+            DLIB_TEST(test_1.is_dequeue_enabled() == true);
+            test_1.disable_dequeue();
+            dlog << LINFO << "Make sure disable_dequeue() works right...";
+            DLIB_TEST(test_1.is_dequeue_enabled() == false);
+            DLIB_TEST(test_1.dequeue(a) == false);
+            a = 4;
+            test_1.wait_for_num_blocked_dequeues(4);
+            test_1.wait_for_num_blocked_dequeues(0);
+            test_1.enqueue(a);
+            test_1.wait_until_empty();
+            DLIB_TEST(test_1.size() == 1);
+            DLIB_TEST(test_1.dequeue(a) == false);
+            DLIB_TEST(test_1.dequeue_or_timeout(a,10000) == false);
+            DLIB_TEST(test_1.size() == 1);
+            a = 0;
+            test_1.enable_dequeue();
+            DLIB_TEST(test_1.is_dequeue_enabled() == true);
+            DLIB_TEST(test_1.dequeue(a) == true);
+            DLIB_TEST(a == 4);
+            test_1.wait_until_empty();
+        }
+
     }