First draft of boost::fibers::unbuffered_channel based implementation of LLCoprocedureManager

1 files changed, 81 insertions, 0 deletions

diff --git a/indra/llmessage/tests/llcoproceduremanager_test.cpp b/indra/llmessage/tests/llcoproceduremanager_test.cpp
index e503798d88..17535abd1e 100644
--- a/indra/llmessage/tests/llcoproceduremanager_test.cpp
+++ b/indra/llmessage/tests/llcoproceduremanager_test.cpp
@@ -31,6 +31,12 @@
 
 #include "../llcoproceduremanager.h"
 
+#include <functional>
+
+#include <boost/fiber/fiber.hpp>
+#include <boost/fiber/buffered_channel.hpp>
+#include <boost/fiber/unbuffered_channel.hpp>
+
 #include "../test/lltut.h"
 
 
@@ -81,5 +87,80 @@ namespace tut
                 foo = 1;
             });
         ensure_equals("coprocedure failed to update foo", foo, 1);
+        
+        LLCoprocedureManager::instance().close("PoolName");
+    }
+
+    template<> template<>
+    void coproceduremanager_object_t::test<2>()
+    {
+        const size_t capacity = 2;
+        boost::fibers::buffered_channel<std::function<void(void)>> chan(capacity);
+
+        boost::fibers::fiber worker([&chan]() {
+            chan.value_pop()();
+        });
+
+        chan.push([]() {
+            LL_INFOS("Test") << "test 1" << LL_ENDL;
+        });
+
+        worker.join();
+    }
+
+    template<> template<>
+    void coproceduremanager_object_t::test<3>()
+    {
+        boost::fibers::unbuffered_channel<std::function<void(void)>> chan;
+
+        boost::fibers::fiber worker([&chan]() {
+            chan.value_pop()();
+        });
+
+        chan.push([]() {
+            LL_INFOS("Test") << "test 1" << LL_ENDL;
+        });
+
+        worker.join();
+    }
+
+    template<> template<>
+    void coproceduremanager_object_t::test<4>()
+    {
+        boost::fibers::buffered_channel<std::function<void(void)>> chan(4);
+
+        boost::fibers::fiber worker([&chan]() {
+            std::function<void(void)> f;
+
+            // using namespace std::chrono_literals;
+            // const auto timeout = 5s;
+            // boost::fibers::channel_op_status status;
+            while (chan.pop(f) != boost::fibers::channel_op_status::closed)
+            {
+                LL_INFOS("CoWorker") << "got coproc" << LL_ENDL;
+                f();
+            }
+            LL_INFOS("CoWorker") << "got closed" << LL_ENDL;
+        });
+
+        int counter = 0;
+
+        for (int i = 0; i < 5; ++i)
+        {
+            LL_INFOS("CoMain") << "pushing coproc " << i << LL_ENDL;
+            chan.push([&counter]() {
+                LL_INFOS("CoProc") << "in coproc" << LL_ENDL;
+                ++counter;
+            });
+        }
+
+        LL_INFOS("CoMain") << "closing channel" << LL_ENDL;
+        chan.close();
+
+        LL_INFOS("CoMain") << "joining worker" << LL_ENDL;
+        worker.join();
+
+        LL_INFOS("CoMain") << "checking count" << LL_ENDL;
+        ensure_equals("coprocedure failed to update counter", counter, 5);
     }
 }  // namespace tut