DELIA-65378 : Decouple notification

Reason for change: Decouple to avoid lock issues. Test Procedure: None Risks: None Signed-off-by: Nikita Poltorapavlo <[email protected]>
rdkcentral · Jun 6, 2024 · 520161a · 520161a
1 parent 3d2b6b6
commit 520161a
Show file tree

Hide file tree

Showing 8 changed files with 186 additions and 6 deletions.
diff --git a/PersistentStore/grpc/Store2.h b/PersistentStore/grpc/Store2.h
@@ -33,6 +33,22 @@ namespace Plugin {
     namespace Grpc {
 
         class Store2 : public Exchange::IStore2 {
+        private:
+            class Job : public Core::IDispatch {
+            public:
+                Job(std::function<void()> work)
+                    : _work(work)
+                {
+                }
+                void Dispatch() override
+                {
+                    _work();
+                }
+
+            private:
+                std::function<void()> _work;
+            };
+
         private:
             Store2(const Store2&) = delete;
             Store2& operator=(const Store2&) = delete;
@@ -179,7 +195,10 @@ namespace Plugin {
                 auto status = _stub->UpdateValue(&context, request, &response);
 
                 if (status.ok()) {
-                    OnValueChanged(ns, key, value);
+                    Core::IWorkerPool::Instance().Submit(
+                        Core::ProxyType<Core::IDispatch>(Core::ProxyType<Job>::Create([&]() {
+                            OnValueChanged(ns, key, value);
+                        })));
                     result = Core::ERROR_NONE;
                 } else {
                     OnError(__FUNCTION__, status);

diff --git a/PersistentStore/grpc/l0test/Store2Test.cpp b/PersistentStore/grpc/l0test/Store2Test.cpp
@@ -4,6 +4,8 @@
 #include "../Store2.h"
 #include "SecureStorageServiceMock.h"
 #include "Server.h"
+#include "WorkerPoolImplementation.h"
+#include "WorkerPoolTest.h"
 
 using ::distp::gateway::secure_storage::v1::DeleteAllValuesRequest;
 using ::distp::gateway::secure_storage::v1::DeleteAllValuesResponse;
@@ -25,7 +27,6 @@ using ::testing::IsTrue;
 using ::testing::Le;
 using ::testing::NiceMock;
 using ::testing::Return;
-using ::testing::Test;
 using ::WPEFramework::Core::Time;
 using ::WPEFramework::Exchange::IStore2;
 using ::WPEFramework::Plugin::Grpc::Store2;
@@ -37,7 +38,7 @@ const auto kAppId = "app_id_1";
 const auto kTtl = 100;
 const auto kScope = Scope::SCOPE_ACCOUNT;
 
-class AStore2 : public Test {
+class AStore2 : public WorkerPoolTest {
 protected:
     NiceMock<SecureStorageServiceMock> service;
     Server server;

diff --git a/PersistentStore/grpc/l0test/WorkerPoolImplementation.h b/PersistentStore/grpc/l0test/WorkerPoolImplementation.h
@@ -0,0 +1,49 @@
+#pragma once
+
+#include "../../Module.h"
+
+class WorkerPoolImplementation : public WPEFramework::Core::WorkerPool, public WPEFramework::Core::ThreadPool::ICallback {
+private:
+    class Dispatcher : public WPEFramework::Core::ThreadPool::IDispatcher {
+    public:
+        Dispatcher(const Dispatcher&) = delete;
+        Dispatcher& operator=(const Dispatcher&) = delete;
+
+        Dispatcher() = default;
+        ~Dispatcher() override = default;
+
+    private:
+        void Initialize() override
+        {
+        }
+        void Deinitialize() override
+        {
+        }
+        void Dispatch(WPEFramework::Core::IDispatch* job) override
+        {
+            job->Dispatch();
+        }
+    };
+
+public:
+    WorkerPoolImplementation() = delete;
+    WorkerPoolImplementation(const WorkerPoolImplementation&) = delete;
+    WorkerPoolImplementation& operator=(const WorkerPoolImplementation&) = delete;
+
+    WorkerPoolImplementation(const uint32_t stackSize)
+        : WPEFramework::Core::WorkerPool(4 /*threadCount*/, stackSize, 32 /*queueSize*/, &_dispatch, this)
+        , _dispatch()
+    {
+        Run();
+    }
+    ~WorkerPoolImplementation() override
+    {
+        Stop();
+    }
+    void Idle() override
+    {
+    }
+
+private:
+    Dispatcher _dispatch;
+};
diff --git a/PersistentStore/grpc/l0test/WorkerPoolTest.h b/PersistentStore/grpc/l0test/WorkerPoolTest.h
@@ -0,0 +1,21 @@
+#pragma once
+
+#include "../../Module.h"
+#include <gmock/gmock.h>
+
+using ::testing::Test;
+
+class WorkerPoolTest : public Test {
+protected:
+    WPEFramework::Core::ProxyType<WorkerPoolImplementation> workerPool;
+    WorkerPoolTest()
+        : workerPool(WPEFramework::Core::ProxyType<WorkerPoolImplementation>::Create(
+              WPEFramework::Core::Thread::DefaultStackSize()))
+    {
+        WPEFramework::Core::IWorkerPool::Assign(&(*workerPool));
+    }
+    ~WorkerPoolTest() override
+    {
+        WPEFramework::Core::IWorkerPool::Assign(nullptr);
+    }
+};
diff --git a/PersistentStore/sqlite/Store2.h b/PersistentStore/sqlite/Store2.h
@@ -36,6 +36,22 @@ namespace Plugin {
                        public Exchange::IStoreCache,
                        public Exchange::IStoreInspector,
                        public Exchange::IStoreLimit {
+        private:
+            class Job : public Core::IDispatch {
+            public:
+                Job(std::function<void()> work)
+                    : _work(work)
+                {
+                }
+                void Dispatch() override
+                {
+                    _work();
+                }
+
+            private:
+                std::function<void()> _work;
+            };
+
         private:
             Store2(const Store2&) = delete;
             Store2& operator=(const Store2&) = delete;
@@ -185,7 +201,10 @@ namespace Plugin {
                 }
 
                 if (rc == SQLITE_DONE) {
-                    OnValueChanged(ns, key, value);
+                    Core::IWorkerPool::Instance().Submit(
+                        Core::ProxyType<Core::IDispatch>(Core::ProxyType<Job>::Create([&]() {
+                            OnValueChanged(ns, key, value);
+                        })));
                     result = Core::ERROR_NONE;
                 } else {
                     OnError(__FUNCTION__, rc);

diff --git a/PersistentStore/sqlite/l1test/Store2Test.cpp b/PersistentStore/sqlite/l1test/Store2Test.cpp
@@ -3,6 +3,8 @@
 
 #include "../Store2.h"
 #include "Store2NotificationMock.h"
+#include "WorkerPoolImplementation.h"
+#include "WorkerPoolTest.h"
 
 using ::testing::_;
 using ::testing::Eq;
@@ -13,7 +15,6 @@ using ::testing::IsTrue;
 using ::testing::Le;
 using ::testing::NiceMock;
 using ::testing::NotNull;
-using ::testing::Test;
 using ::WPEFramework::Exchange::IStore2;
 using ::WPEFramework::Exchange::IStoreCache;
 using ::WPEFramework::Exchange::IStoreInspector;
@@ -37,7 +38,7 @@ const auto kLimit20 = 20;
 const auto kLimit30 = 30;
 const auto kLimit40 = 40;
 
-class AStore2 : public Test {
+class AStore2 : public WorkerPoolTest {
 protected:
     WPEFramework::Core::ProxyType<Store2> store2;
     AStore2()

diff --git a/PersistentStore/sqlite/l1test/WorkerPoolImplementation.h b/PersistentStore/sqlite/l1test/WorkerPoolImplementation.h
@@ -0,0 +1,49 @@
+#pragma once
+
+#include "../../Module.h"
+
+class WorkerPoolImplementation : public WPEFramework::Core::WorkerPool, public WPEFramework::Core::ThreadPool::ICallback {
+private:
+    class Dispatcher : public WPEFramework::Core::ThreadPool::IDispatcher {
+    public:
+        Dispatcher(const Dispatcher&) = delete;
+        Dispatcher& operator=(const Dispatcher&) = delete;
+
+        Dispatcher() = default;
+        ~Dispatcher() override = default;
+
+    private:
+        void Initialize() override
+        {
+        }
+        void Deinitialize() override
+        {
+        }
+        void Dispatch(WPEFramework::Core::IDispatch* job) override
+        {
+            job->Dispatch();
+        }
+    };
+
+public:
+    WorkerPoolImplementation() = delete;
+    WorkerPoolImplementation(const WorkerPoolImplementation&) = delete;
+    WorkerPoolImplementation& operator=(const WorkerPoolImplementation&) = delete;
+
+    WorkerPoolImplementation(const uint32_t stackSize)
+        : WPEFramework::Core::WorkerPool(4 /*threadCount*/, stackSize, 32 /*queueSize*/, &_dispatch, this)
+        , _dispatch()
+    {
+        Run();
+    }
+    ~WorkerPoolImplementation() override
+    {
+        Stop();
+    }
+    void Idle() override
+    {
+    }
+
+private:
+    Dispatcher _dispatch;
+};
diff --git a/PersistentStore/sqlite/l1test/WorkerPoolTest.h b/PersistentStore/sqlite/l1test/WorkerPoolTest.h
@@ -0,0 +1,21 @@
+#pragma once
+
+#include "../../Module.h"
+#include <gmock/gmock.h>
+
+using ::testing::Test;
+
+class WorkerPoolTest : public Test {
+protected:
+    WPEFramework::Core::ProxyType<WorkerPoolImplementation> workerPool;
+    WorkerPoolTest()
+        : workerPool(WPEFramework::Core::ProxyType<WorkerPoolImplementation>::Create(
+              WPEFramework::Core::Thread::DefaultStackSize()))
+    {
+        WPEFramework::Core::IWorkerPool::Assign(&(*workerPool));
+    }
+    ~WorkerPoolTest() override
+    {
+        WPEFramework::Core::IWorkerPool::Assign(nullptr);
+    }
+};