semaphore: disallow move after used

Referring to scylladb#1323 and discussion in scylladb#1477.

Avi Kivity <[email protected]> asserted that:
> We have a choice:
> declare that most things can be movable and live with the complication and slowdown
> declare that control structures (like semaphores) have stable addresses (as they typically do anyway)
> There's really no reason to move a sempahore.

Gleb Natapov <[email protected]> proposed:
> add an assertion in debug mode if a semaphore is moved after the first use. It can be a simple bool that is set to true on the first wait().

Since we still need to be able to move the semaphore
after it created this patch adds a `used_flag` class
that asserts that it isn't used when moved or move-assigned.

This patch reverts commit 3061b2f
but keeps the unit test introduced by it.

Close scylladb#1323

Signed-off-by: Benny Halevy <[email protected]>

include/seastar/core/semaphore.hh

@@ -188,6 +188,35 @@ private:
         }
     };
     internal::abortable_fifo<entry, expiry_handler> _wait_list;
+
+#ifdef SEASTAR_SEMAPHORE_DEBUG
+    struct used_flag {
+        // set to true from the wait path
+        // prevents the semaphore from being moved or move-reassigned when _used
+        bool _used = false;
+
+        used_flag() = default;
+        used_flag(used_flag&& o) noexcept {
+            assert(!_used && "semaphore cannot be moved after it has been used");
+        }
+        used_flag& operator=(used_flag&& o) noexcept {
+            if (this != &o) {
+                assert(!_used && !o._used && "semaphore cannot be moved after it has been used");
+            }
+            return *this;
+        }
+        void use() noexcept {
+            _used = true;
+        }
+    };
+#else
+    struct used_flag {
+        void use() noexcept {}
+    };
+#endif
+
+    [[no_unique_address]] used_flag _used;
+
     bool has_available_units(size_t nr) const noexcept {
         return _count >= 0 && (static_cast<size_t>(_count) >= nr);
     }
@@ -244,6 +273,7 @@ public:
     ///         \ref semaphore_timed_out exception.  If the semaphore was
     ///         \ref broken(), may contain a \ref broken_semaphore exception.
     future<> wait(time_point timeout, size_t nr = 1) noexcept {
+        _used.use();
         if (may_proceed(nr)) {
             _count -= nr;
             return make_ready_future<>();
@@ -279,6 +309,7 @@ public:
     ///         \ref semaphore_aborted exception.  If the semaphore was
     ///         \ref broken(), may contain a \ref broken_semaphore exception.
     future<> wait(abort_source& as, size_t nr = 1) noexcept {
+        _used.use();
         if (may_proceed(nr)) {
             _count -= nr;
             return make_ready_future<>();
@@ -343,6 +374,7 @@ public:
     ///
     /// \param nr Amount of units to consume (default 1).
     void consume(size_t nr = 1) noexcept {
+        _used.use();
         if (_ex) {
             return;
         }
@@ -360,6 +392,7 @@ public:
     /// \param nr number of units to reduce the counter by (default 1).
     /// \return `true` if the counter had sufficient units, and was decremented.
     bool try_wait(size_t nr = 1) noexcept {
+        _used.use();
         if (may_proceed(nr)) {
             _count -= nr;
             return true;