semaphore: allow wait() and signal() after broken()

This patch makes semaphore more robust by allowing wait(), try_wait()
and signal() calls after it was broken(). The reason for this is to
simplify the users of semaphore so that they do not need any additional
synchronization between calls to broken() and any other semaphore
function.

After semaphore is broken all calls to wait() will return a failed
future, signal() will be ignored and try_wait() will return false.
Moreoever, the semaphore will appear to have no available units and no
waiters.

Signed-off-by: Paweł Dziepak <pdziepak@scylladb.com>
Message-Id: <1458221941-32369-1-git-send-email-pdziepak@scylladb.com>

core/semaphore.hh

@@ -69,6 +69,7 @@ public:
 class semaphore {
 private:
     size_t _count;
+    std::exception_ptr _ex;
     struct entry {
         promise<> pr;
         size_t nr;
@@ -111,6 +112,9 @@ public:
             _count -= nr;
             return make_ready_future<>();
         }
+        if (_ex) {
+            return make_exception_future(_ex);
+        }
         promise<> pr;
         auto fut = pr.get_future();
         _wait_list.push_back(entry(std::move(pr), nr));
@@ -155,6 +159,9 @@ public:
     ///
     /// \param nr Number of units to deposit (default 1).
     void signal(size_t nr = 1) {
+        if (_ex) {
+            return;
+        }
         _count += nr;
         while (!_wait_list.empty() && _wait_list.front().nr <= _count) {
             auto& x = _wait_list.front();
@@ -195,17 +202,11 @@ public:
     /// Signal to waiters that an error occurred.  \ref wait() will see
     /// an exceptional future<> containing a \ref broken_semaphore exception.
     /// The future is made available immediately.
-    ///
-    /// This may only be used once per semaphore; after using it the
-    /// semaphore is in an indeterminate state and should not be waited on.
     void broken() { broken(std::make_exception_ptr(broken_semaphore())); }
 
     /// Signal to waiters that an error occurred.  \ref wait() will see
     /// an exceptional future<> containing the provided exception parameter.
     /// The future is made available immediately.
-    ///
-    /// This may only be used once per semaphore; after using it the
-    /// semaphore is in an indeterminate state and should not be waited on.
     template <typename Exception>
     void broken(const Exception& ex) {
         broken(std::make_exception_ptr(ex));
@@ -214,15 +215,14 @@ public:
     /// Signal to waiters that an error occurred.  \ref wait() will see
     /// an exceptional future<> containing the provided exception parameter.
     /// The future is made available immediately.
-    ///
-    /// This may only be used once per semaphore; after using it the
-    /// semaphore is in an indeterminate state and should not be waited on.
     void broken(std::exception_ptr ex);
 };
 
 inline
 void
 semaphore::broken(std::exception_ptr xp) {
+    _ex = xp;
+    _count = 0;
     while (!_wait_list.empty()) {
         auto& x = _wait_list.front();
         x.pr.set_exception(xp);

tests/semaphore_test.cc

@@ -106,7 +106,7 @@ SEASTAR_TEST_CASE(test_semaphore_mix_1) {
 SEASTAR_TEST_CASE(test_broken_semaphore) {
     auto sem = make_lw_shared<semaphore>(0);
     struct oops {};
-    auto ret = sem->wait().then_wrapped([sem] (future<> f) {
+    auto check_result = [sem] (future<> f) {
         try {
             f.get();
             BOOST_FAIL("expecting exception");
@@ -118,9 +118,12 @@ SEASTAR_TEST_CASE(test_broken_semaphore) {
         }
         BOOST_FAIL("unreachable");
         return make_ready_future<>();
-    });
+    };
+    auto ret = sem->wait().then_wrapped(check_result);
     sem->broken(oops());
-    return ret;
+    return sem->wait().then_wrapped(check_result).then([ret = std::move(ret)] () mutable {
+        return std::move(ret);
+    });
 }
 
 SEASTAR_TEST_CASE(test_shared_mutex_exclusive) {