disposable_semaphore for one-shot usage

Author: lihuiba

thread/test/test.cpp

@@ -32,6 +32,7 @@ limitations under the License.
 #include "../../test/ci-tools.h"
 #include "../future.h"
 
+#include <photon/common/lockfree_queue.h>
 
 using namespace std;
 using namespace photon;
@@ -698,6 +699,40 @@ TEST(Semaphore, heavy) {
     wait_for_completion();
 }
 
+using Ring = photon::common::RingChannel<
+    LockfreeSPSCRingQueue<Delegate<void>, 16>>;
+
+void test_disposable_semaphore(Ring* ring, uint64_t count, uint64_t slat, uint64_t oplat) {
+    while (count--) {
+        disposable_semaphore sem;
+        auto op = [oplat, &sem]() {
+            if (oplat) ::usleep(oplat); // operation latency
+            sem.signal();               // completion signal
+        };
+        ring->send(op);
+        if (slat) ::usleep(slat);       // submission latency
+        int ret = sem.wait(1000*1000);  // wait for completion
+        EXPECT_EQ(0, ret);
+        if (ret) return;
+    }
+}
+
+TEST(disposable_semaphore, basic) {
+    Ring ring;
+    std::thread worker([&]() {
+        vcpu_init();
+        while(auto cb = ring.recv(0, 0)) cb();
+        vcpu_fini();
+    });
+
+    test_disposable_semaphore(&ring, 10000, 0, 100);
+    test_disposable_semaphore(&ring, 10000, 100, 0);
+    test_disposable_semaphore(&ring, 10000, 00, 00);
+
+    ring.send({});
+    worker.join();
+}
+
 TEST(Sleep, sleep_only_thread) {    //Sleep_sleep_only_thread_Test::TestBody
     // If current thread is only thread and sleeping
     // it should be able to avoid crash during long time sleep

thread/thread.cpp

@@ -1899,6 +1899,10 @@ R"(
         }
         return 0;
     }
+    void disposable_semaphore::defer(void* arg) {
+        auto _this = (disposable_semaphore*)arg;
+        _this->signal();
+    }
     bool is_master_event_engine_default() {
         return CURRENT->get_vcpu()->is_master_event_engine_default();
     }

thread/thread.h

@@ -461,6 +461,35 @@ namespace photon
         void try_resume(uint64_t count);
     };
 
+    // one-shot semaphore
+    class disposable_semaphore {
+    public:
+        int wait(Timeout timeout) {
+            uint64_t th = 0, current = (uint64_t)CURRENT;
+            assert(current && !(current & 1));
+            if (!_waiter.compare_exchange_strong(th, current)) {
+                assert(th == 0x1);
+                return 0;
+            }
+            int ret = thread_usleep_defer(timeout, &defer, this);
+            if (likely(ret < 0)) { assert(errno == -1); return 0; }
+            errno = ETIMEDOUT;
+            return -1;
+        }
+        void signal() {
+            auto th = _waiter.fetch_or(0x1);
+            if (!th) { } // first signal (async op completion) before the CAS, AKA _waiter.compare_exchange_strong()
+            else if (!(th&1)) { } // first signal after the CAS, either async op completion or usleep defer
+            else { assert((th>>1) && (th&1)); // second signal()
+                thread_interrupt((thread*)(th&~1), -1);
+            }
+        }
+
+    protected:
+        std::atomic<uint64_t> _waiter{0};
+        static void defer(void* arg);
+    };
+
     // to be different to timer flags
     // mark flag should be larger than 999, and not touch lower bits
     // here we selected