[DO NOT BUILD] save my work

condition_variable.hpp

@@ -4,6 +4,7 @@
 #include <condition_variable>
 #include <mutex>
 #include <atomic>
+#include <numeric>
 
 #include <rlib/class_decorator.hpp>
 
@@ -11,31 +12,47 @@ namespace rlib {
     class easy_condition_variable : rlib::noncopyable {
     public:
         easy_condition_variable()
-            : stdcv(), stdcv_mutex(), notify_count(0) {}
+            : stdcv(), stdcv_mutex(), notify_count(0), notify_all_mark(false) {}
         
         void notify_one() {
+            ++notify_count;
             stdcv.notify_one();
         }
         void notify_all() {
+            notify_all_mark = true;
             stdcv.notify_all();
         }
 
+        const std::function<bool()> determine_wait_cond = [this]{
+                if(notify_all_mark)
+                    return true;
+                while(true) {
+                    auto val = notify_count.load();
+                    if(val <= 0)
+                        return false;
+                    if(notify_count.compare_exchange_strong(val, val-1))
+                        return true;
+                }
+            };
         void wait() {
+            notify_all_mark = false;
             std::unique_lock<std::mutex> lk(stdcv_mutex);
-            stdcv.wait(lk, [this]{return notify_count.load() > 0;});
+            stdcv.wait(lk, determine_wait_cond);
             lk.unlock();
         }
         template< class Rep, class Period >
         bool wait_for(const std::chrono::duration<Rep, Period>& rel_time) {
+            notify_all_mark = false;
             std::unique_lock<std::mutex> lk(stdcv_mutex);
-            bool ret = stdcv.wait_for(lk, rel_time, [this]{return notify_count.load() > 0;});
+            bool ret = stdcv.wait_for(lk, rel_time, determine_wait_cond);
             lk.unlock();
             return ret;
         }
         template< class Clock, class Duration >
         bool wait_until(const std::chrono::time_point<Clock, Duration>& timeout_time) {
+            notify_all_mark = false;
             std::unique_lock<std::mutex> lk(stdcv_mutex);
-            bool ret = stdcv.wait_until(lk, timeout_time, [this]{return notify_count.load() > 0;});
+            bool ret = stdcv.wait_until(lk, timeout_time, determine_wait_cond);
             lk.unlock();
             return ret;
         }
@@ -43,6 +60,7 @@ namespace rlib {
         std::condition_variable stdcv;
         std::mutex stdcv_mutex;
         std::atomic<int> notify_count;
+        std::atomic<bool> notify_all_mark;
     };
 } // namespace rlib
 

pool.hpp

@@ -22,48 +22,28 @@ using size_t = unsigned long;
 
 namespace rlib {
     struct object_pool_policy_fixed {
-        object_pool_policy_fixed(size_t size) : size(size), used(new std::atomic<size_t>(0)) {}
-        bool borrow_should_alloc(/*const size_t free_objects, const size_t existing_objects*/) { // TODO: add size argument?
-            // object_pool::borrow calls this funcion.
-            // If it returns true, object pool will alloc a new object.
-            // If it returns false, object pool will not alloc a new object.
-            // If it returns false, THIS FUNCTION MUST HAVE NO SIDE EFFECT!
-            // If there's no free object, object pool::borrow blocks.
-            while(true) {
-                auto used_old = used->load(std::memory_order_acquire);
-                if(used_old >= size) {
-                    return false;
-                }
-                auto used_new = used_old + 1;
-                if(used->compare_exchange_strong(used_old, used_new, std::memory_order_acq_rel))
-                    break; // success
-            }
-            return true;
-        }
-        bool release_should_free() {
-            if(used->load() == 0)
-                throw std::runtime_error("POLICY detected error: Release object of zero-sized object pool.");
-            (*used)--;
-            return false;
-        }
-
-        size_t objects_should_alloc(const size_t inuse_objects, const size_t avail_objects) {
+        object_pool_policy_fixed(size_t size) : size(size) {}
+        size_t objects_should_alloc(const size_t inuse_objects, const size_t avail_objects) const {
             return avail_objects < size ? size - avail_objects : 0;
         }
     private:
         const size_t size;
-        std::unique_ptr<std::atomic<size_t> > used;
     };
-    struct object_pool_policy_dynamic_never_free {
-        object_pool_policy_dynamic_never_free() : fixed(std::numeric_limits<size_t>::max()) {}
-        bool borrow_should_alloc() {
-            return fixed.borrow_should_alloc();
-        }
-        bool release_should_free() {
-            return fixed.release_should_free();
+    struct object_pool_policy_watermarks {
+        // If free/all < alloc_threshold_rate, then alloc more objects.
+        object_pool_policy_watermarks(float alloc_threshold_rate = 0.8, size_t min_objects = 8)
+            : _alloc_threshold_rate(1.0/alloc_threshold_rate), min_objects(min_objects)  {}
+        size_t objects_should_alloc(const size_t inuse_objects, const size_t avail_objects) const {
+            if(avail_objects < min_objects)
+                return min_objects - avail_objects;
+            size_t threshold = inuse_objects * _alloc_threshold_rate;
+            if(threshold > avail_objects)
+                return threshold - avail_objects;
+            return 0;
         }
     private:
-        object_pool_policy_fixed fixed;
+        const float _alloc_threshold_rate;
+        const size_t min_objects;
     };
     struct object_pool_policy_dynamic_smart {
 
@@ -146,7 +126,7 @@ namespace rlib {
         std::tuple<_bound_construct_args_t ...> _bound_args;
 
         size_t inuse_objects = 0;
-        policy_t policy;
+        const policy_t policy;
         std::condition_variable borrow_cv; // use buffer_mutex on notifying alloc event. 
         volatile bool new_obj_ready = false;
         void notify_new_object_allocated(size_t how_many) {

test/Makefile

@@ -22,7 +22,7 @@
 MODULES=string meta trait stdio sio scope_guard
 
 XTRA_FLAGS ?=
-CXXFLAGS=-I. -I../.. $(XTRA_FLAGS)
+CXXFLAGS=-I. -I../.. $(XTRA_FLAGS) -pthread
 STD ?= 14
 FLAGS11=-std=c++11 rlib/libr.a
 FLAGS14=-std=c++14 rlib/libr.a
@@ -40,19 +40,12 @@ endif
 
 POSTFIX=$(STD)_$(CXX)
 
-all: string common pool
+all: string.test common.test pool.test threading.test
 
-common:
-	$(CXX) $(CXXFLAGS) src/common.cc $(CXXFLAGS) -o src/common_$(POSTFIX).out
-	src/common_$(POSTFIX).out
-
-string:
-	$(CXX) $(CXXFLAGS) src/string.cc $(CXXFLAGS) -o src/string_$(POSTFIX).out
-	src/string_$(POSTFIX).out
-
-pool:
-	$(CXX) $(CXXFLAGS) src/pool.cc $(CXXFLAGS) -o src/pool_$(POSTFIX).out
-	src/pool_$(POSTFIX).out
+%.test: src/%.cc
+	@echo Testing $< ...
+	$(CXX) $< $(CXXFLAGS) -o src/$@_$(POSTFIX).out
+	src/$@_$(POSTFIX).out
 
 clean:
 	rm -f src/*.out

test/src/pool.cc

@@ -59,13 +59,13 @@ TEST_CASE("fixed object pool") {
             objs.pop_front();
         }
     }
-    REQUIRE(fixed_pool.size() == 1);
+    REQUIRE(fixed_pool.size() == pool_size);
 }
 
 TEST_CASE("infinite dynamic object pool") {
     const auto arg1 = string("fuck you 2");
-    rlib::object_pool<rlib::object_pool_policy_dynamic_never_free, pooled_obj2_t, decltype(arg1)>
-        inf_pool(rlib::object_pool_policy_dynamic_never_free(), arg1);
+    rlib::object_pool<rlib::object_pool_policy_watermarks, pooled_obj2_t, decltype(arg1)>
+        inf_pool(rlib::object_pool_policy_watermarks(), arg1);
     
     auto res = inf_pool.try_borrow_one();
     REQUIRE(res != nullptr);
@@ -90,8 +90,6 @@ TEST_CASE("infinite dynamic object pool") {
         }
         // "Leak" one object.
     }
-
-    REQUIRE(inf_pool.size() == 1+test_rounds);
 }
 
 TEST_CASE("fixed object pool parallel test") {

test/src/threading.cc

@@ -3,4 +3,43 @@
 
 #include <rlib/condition_variable.hpp>
 
+#include <thread>
+#include <iostream>
+
+rlib::easy_condition_variable rcv, rcv_done;
+volatile int globalTestVar = 0;
+void t1() {
+    rcv.wait();
+    globalTestVar *= 3;
+    rcv_done.notify_all();
+}
+void t2() {
+    rcv.wait();
+    globalTestVar *= 2;
+    rcv_done.notify_all();
+}
+void t3() {
+    globalTestVar += 3;
+    rcv.notify_one();
+    rcv_done.wait();
+}
+
+
+TEST_CASE("condition_variable_1") {
+    const auto tests = 512;
+    std::cout << "Running tests... It may be slow." << std::endl;
+    for(auto cter = 0; cter < 1024; ++cter) {
+        globalTestVar = 0;
+        auto th1 = std::thread(t1);
+        auto th2 = std::thread(t2);
+        auto th3 = std::thread(t3);
+        th3.join();
+        REQUIRE((globalTestVar == 9 || globalTestVar == 6));
+        rcv.notify_all();
+        th1.join();
+        th2.join();
+        REQUIRE(globalTestVar == 18);
+    }
+}
+