little more meat for cgc

tests/wasp_c_extensions_cgc_test.cpp

@@ -1,9 +1,23 @@
 
+#include <atomic>
+#include <condition_variable>
+#include <mutex>
+#include <thread>
+
 #include <cppunit/TestCase.h>
 #include <cppunit/extensions/HelperMacros.h>
 
 #include "wasp_c_extensions/_cgc/cgc.hpp"
 
+namespace wasp::test_case {
+    const size_t threads_num = 1024;
+    const size_t items_per_thread = 10240;
+    wasp::cgc::ConcurrentGarbageCollector* global_gc = NULL;
+
+    std::atomic<bool> start_event_flag(false);
+    std::condition_variable start_event_cv;
+    std::mutex start_event_mutex;
+}
 
 class TestWaspCGCTest:
     public CppUnit::TestFixture
@@ -30,7 +44,9 @@ class TestWaspCGCTest:
     CPPUNIT_TEST_SUITE(TestWaspCGCTest);
     CPPUNIT_TEST(test_item_destroy);
     CPPUNIT_TEST(test_item);
-    CPPUNIT_TEST(test_gc_push);
+    CPPUNIT_TEST(test_gc);
+    CPPUNIT_TEST(test_gc_push_errors);
+    CPPUNIT_TEST(test_gc_concurrent_push);
     CPPUNIT_TEST_SUITE_END();
 
     void test_item_destroy(){
@@ -50,24 +66,63 @@ class TestWaspCGCTest:
         wasp::cgc::ConcurrentGCItem::destroy(item_ptr);
     }
 
-    void test_gc_push(){
-        Item* item_ptr = NULL;
+    void test_gc(){
+        wasp::cgc::ConcurrentGarbageCollector* cgc = new wasp::cgc::ConcurrentGarbageCollector();
+        CPPUNIT_ASSERT(cgc->items() == 0);
+        delete cgc;
+    }
+
+    void test_gc_push_errors(){
+        Item* item_ptr = Item::create();
         wasp::cgc::ConcurrentGarbageCollector* cgc = new wasp::cgc::ConcurrentGarbageCollector();
 
         CPPUNIT_ASSERT_THROW(cgc->push(NULL), wasp::cgc::NullPointerException);
 
-        item_ptr = Item::create();
         item_ptr->gc_ready.store(true, std::memory_order_seq_cst);  // without it the assertion fails
         CPPUNIT_ASSERT_THROW(cgc->push(item_ptr), wasp::cgc::InvalidItemState);
+    }
+
+    void test_gc_concurrent_push(){
+        std::thread* threads[wasp::test_case::threads_num];
+
+        wasp::test_case::global_gc = new wasp::cgc::ConcurrentGarbageCollector();
+
+        for (size_t i=0; i < wasp::test_case::threads_num; i++){
+            threads[i] = new std::thread(TestWaspCGCTest::push_threaded_fn);
+        }
+
+        wasp::test_case::start_event_flag.store(true, std::memory_order_seq_cst);
+
+        {
+            std::lock_guard<std::mutex> lock(wasp::test_case::start_event_mutex);
+            wasp::test_case::start_event_cv.notify_all();
+        }
+
+        for (size_t i=0; i < wasp::test_case::threads_num; i++){
+            threads[i]->join();
+            delete threads[i];
+        }
+
+        delete wasp::test_case::global_gc;
+        wasp::test_case::global_gc = NULL;
+    }
+
+    static void push_threaded_fn(){
+        Item* items[wasp::test_case::items_per_thread];
 
-        item_ptr->gc_ready.store(false, std::memory_order_seq_cst);  // only a test may revert this flag
-        // do not do this in a real code
-        cgc->push(item_ptr);
+        while (! wasp::test_case::start_event_flag.load(std::memory_order_seq_cst)){
+            std::unique_lock<std::mutex> lock(wasp::test_case::start_event_mutex);
+            wasp::test_case::start_event_cv.wait(lock);
+        }
 
-        item_ptr->gc_ready.store(true, std::memory_order_seq_cst);  // without gc would not be
-        cgc->collect();
+        for (size_t i=0; i < wasp::test_case::items_per_thread; i++){
+            items[i] = Item::create();
+            wasp::test_case::global_gc->push(items[i]);
+        }
 
-        delete cgc;  // collection happens on destruction
+        for (size_t i=0; i < wasp::test_case::items_per_thread; i++){
+            items[i]->gc_ready.store(true, std::memory_order_seq_cst);
+        }
     }
 };
 

wasp_c_extensions/_cgc/cgc.cpp

@@ -50,12 +50,20 @@ ConcurrentGCItem::~ConcurrentGCItem()
 }
 
 ConcurrentGarbageCollector::ConcurrentGarbageCollector():
-    head(NULL)
+    head(NULL),
+    parallel_gc(0),
+    count(0)
 {}
 
 ConcurrentGarbageCollector::~ConcurrentGarbageCollector()
 {
+    assert(this->parallel_gc.load(std::memory_order_seq_cst) == 0);
     this->collect();
+    assert(this->count.load(std::memory_order_seq_cst) == 0);
+}
+
+size_t ConcurrentGarbageCollector::items(){
+    return this->count.load(std::memory_order_seq_cst);
 }
 
 void ConcurrentGarbageCollector::push(ConcurrentGCItem* item_ptr){
@@ -68,6 +76,7 @@ void ConcurrentGarbageCollector::push(ConcurrentGCItem* item_ptr){
     }
 
     while (! this->__push(item_ptr, item_ptr));
+    this->count.fetch_add(1, std::memory_order_seq_cst);
 }
 
 bool ConcurrentGarbageCollector::__push(ConcurrentGCItem* new_head_ptr, ConcurrentGCItem* new_tail_ptr)
@@ -93,16 +102,17 @@ void ConcurrentGarbageCollector::collect()
 {
     ConcurrentGCItem *head_ptr = NULL;  // pointer to a new head
     ConcurrentGCItem *tail_ptr = NULL;  // pointer to a new tail
-
     ConcurrentGCItem *next_ptr = NULL;  // pointer to a new item to check next
-
     ConcurrentGCItem *current_ptr = this->detach_head();  // current item to check
 
+    this->parallel_gc.fetch_add(1, std::memory_order_seq_cst);
+
     while(current_ptr){
         next_ptr = current_ptr->next.load(std::memory_order_seq_cst);
 
         if (current_ptr->gc_ready.load(std::memory_order_seq_cst)){
             ConcurrentGCItem::destroy(current_ptr);
+            this->count.fetch_sub(1, std::memory_order_seq_cst);
             current_ptr = next_ptr;
             continue;
         }
@@ -122,10 +132,6 @@ void ConcurrentGarbageCollector::collect()
     if (head_ptr && tail_ptr){
         while (! this->__push(head_ptr, tail_ptr));
     }
-}
 
-ConcurrentGCItem* ConcurrentGarbageCollector::pop()
-{
-    // TODO: implement
-    return NULL;
+    this->parallel_gc.fetch_sub(1, std::memory_order_seq_cst);
 }

wasp_c_extensions/_cgc/cgc.hpp

@@ -47,7 +47,8 @@ class ConcurrentGCItem{
 class ConcurrentGarbageCollector{
 
     std::atomic<ConcurrentGCItem*> head;
-    std::atomic<size_t> count;  // TODO: count and make a public method
+    std::atomic<size_t> parallel_gc;
+    std::atomic<size_t> count;
 
     bool __push(ConcurrentGCItem*, ConcurrentGCItem*);
 
@@ -57,12 +58,11 @@ class ConcurrentGarbageCollector{
         ConcurrentGarbageCollector();
         virtual ~ConcurrentGarbageCollector();
 
+        size_t items();
+
         void push(ConcurrentGCItem*);
 
         void collect();  // try to clear everything from this GC. This method should not called often
-
-        ConcurrentGCItem* pop();  // remove an item from a GC no matter gc_ready is set or not
-        // but ConcurrentGCItem::destroy is not called
 };
 
 };  // namespace wasp::cgc