Merge pull request #5716 from npoltorapavlo/DELIA-66118-main

Delia 66118 main
rdkcentral · Sep 25, 2024 · 4064bca · 4064bca
2 parents 22b1fff + 938d8a5
commit 4064bca
Show file tree

Hide file tree

Showing 16 changed files with 373 additions and 67 deletions.
diff --git a/.github/workflows/L2-CloudStore.yml b/.github/workflows/L2-CloudStore.yml
@@ -37,4 +37,4 @@ jobs:
 # Usage:
 # PATH=${PWD}/install/bin:${PATH} LD_LIBRARY_PATH=${PWD}/install/lib:${LD_LIBRARY_PATH} valgrind --tool=memcheck --log-file=valgrind_log --leak-check=yes --show-reachable=yes --track-fds=yes --fair-sched=try Thunder -f -c ${PWD}/install/etc/Thunder/config.json
 # (to stop press q & enter)
-# curl -d '{"jsonrpc":"2.0","id":0,"method":"org.rdk.CloudStore.setValue","params":{"namespace":"test","key":"key1","value":"1","ttl":100}}' http://localhost:55555/jsonrpc
+# curl -d '{"jsonrpc":"2.0","id":0,"method":"org.rdk.CloudStore.setValue","params":{"scope":"account","namespace":"test","key":"key1","value":"1","ttl":100}}' http://localhost:55555/jsonrpc
diff --git a/CloudStore/CHANGELOG.md b/CloudStore/CHANGELOG.md
@@ -16,6 +16,10 @@ All notable changes to this RDK Service will be documented in this file.
 
 * For more details, refer to [versioning](https://github.com/rdkcentral/rdkservices#versioning) section under Main README.
 
+## [1.0.1] - 2024-09-17
+### Fixed
+- Decouple notification, add timeouts
+
 ## [1.0.0] - 2024-07-15
 ### Added
 - Add CHANGELOG

diff --git a/CloudStore/CloudStore.cpp b/CloudStore/CloudStore.cpp
@@ -24,7 +24,7 @@
 
 #define API_VERSION_NUMBER_MAJOR 1
 #define API_VERSION_NUMBER_MINOR 0
-#define API_VERSION_NUMBER_PATCH 0
+#define API_VERSION_NUMBER_PATCH 1
 
 namespace WPEFramework {
 

diff --git a/CloudStore/Module.h b/CloudStore/Module.h
@@ -32,6 +32,14 @@
 #define URI_ENV "CLOUDSTORE_URI"
 #define IARM_INIT_NAME "Thunder_Plugins"
 #define URI_RFC "Device.DeviceInfo.X_RDKCENTRAL-COM_RFC.CloudStore.Uri"
+#define PARTNER_ID_FILENAME "/opt/www/authService/partnerId3.dat"
+#define ACCOUNT_ID_FILENAME "/opt/www/authService/said.dat"
+#define DEVICE_ID_FILENAME "/opt/www/authService/xdeviceid.dat"
+#define SECURITY_AGENT_FILENAME "/tmp/SecurityAgent/token"
+#define IARM_TIMEOUT 1000
+#define COM_RPC_TIMEOUT 1000
+#define JSON_RPC_TIMEOUT 2000
+#define GRPC_TIMEOUT 3000
 
 #undef EXTERNAL
 #define EXTERNAL
diff --git a/CloudStore/grpc/Store2.h b/CloudStore/grpc/Store2.h
diff --git a/CloudStore/grpc/l0test/Store2Test.cpp b/CloudStore/grpc/l0test/Store2Test.cpp
@@ -4,6 +4,7 @@
 #include "../Store2.h"
 #include "SecureStorageServiceMock.h"
 #include "Server.h"
+#include "WorkerPoolImplementation.h"
 
 using ::distp::gateway::secure_storage::v1::DeleteAllValuesRequest;
 using ::distp::gateway::secure_storage::v1::DeleteAllValuesResponse;
@@ -39,13 +40,21 @@ const auto kScope = Scope::SCOPE_ACCOUNT;
 
 class AStore2 : public Test {
 protected:
+    WPEFramework::Core::ProxyType<WorkerPoolImplementation> workerPool;
     NiceMock<SecureStorageServiceMock> service;
     Server server;
     WPEFramework::Core::ProxyType<IStore2> store2;
     AStore2()
-        : server(kUri, &service)
+        : workerPool(WPEFramework::Core::ProxyType<WorkerPoolImplementation>::Create(
+              WPEFramework::Core::Thread::DefaultStackSize()))
+        , server(kUri, &service)
         , store2(WPEFramework::Core::ProxyType<Store2>::Create(kUri))
     {
+        WPEFramework::Core::IWorkerPool::Assign(&(*workerPool));
+    }
+    ~AStore2() override
+    {
+        WPEFramework::Core::IWorkerPool::Assign(nullptr);
     }
 };
 

diff --git a/CloudStore/grpc/l0test/WorkerPoolImplementation.h b/CloudStore/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/CloudStore/l0test/ServiceMock.h b/CloudStore/l0test/ServiceMock.h
@@ -53,6 +53,7 @@ class ServiceMock : public WPEFramework::PluginHost::IShell,
     MOCK_METHOD(void, Unregister, (const IShell::ICOMLink::INotification*), (override));
     MOCK_METHOD(WPEFramework::RPC::IRemoteConnection*, RemoteConnection, (const uint32_t), (override));
     MOCK_METHOD(void*, Instantiate, (const WPEFramework::RPC::Object&, const uint32_t, uint32_t&), (override));
+    MOCK_METHOD(WPEFramework::RPC::IStringIterator*, GetLibrarySearchPaths, (const string&), (const, override));
     BEGIN_INTERFACE_MAP(ServiceMock)
     INTERFACE_ENTRY(IShell)
     INTERFACE_ENTRY(IShell::ICOMLink)

diff --git a/PersistentStore/CHANGELOG.md b/PersistentStore/CHANGELOG.md
@@ -16,6 +16,10 @@ All notable changes to this RDK Service will be documented in this file.
 
 * For more details, refer to [versioning](https://github.com/rdkcentral/rdkservices#versioning) section under Main README.
 
+## [2.0.1] - 2024-09-17
+### Fixed
+- Decouple notification, add timeouts
+
 ## [2.0.0] - 2024-08-06
 ### Removed
 - Move secure store into a separate plugin

diff --git a/PersistentStore/Module.h b/PersistentStore/Module.h
@@ -34,6 +34,8 @@
 #define MAXVALUE_ENV "PERSISTENTSTORE_MAXVALUE"
 #define LIMIT_ENV "PERSISTENTSTORE_LIMIT"
 #define IARM_INIT_NAME "Thunder_Plugins"
+#define IARM_TIMEOUT 1000
+#define SQLITE_TIMEOUT 1000
 
 #undef EXTERNAL
 #define EXTERNAL
diff --git a/PersistentStore/PersistentStore.cpp b/PersistentStore/PersistentStore.cpp
@@ -22,7 +22,7 @@
 
 #define API_VERSION_NUMBER_MAJOR 2
 #define API_VERSION_NUMBER_MINOR 0
-#define API_VERSION_NUMBER_PATCH 0
+#define API_VERSION_NUMBER_PATCH 1
 
 namespace WPEFramework {
 

diff --git a/PersistentStore/l0test/ServiceMock.h b/PersistentStore/l0test/ServiceMock.h
@@ -53,6 +53,7 @@ class ServiceMock : public WPEFramework::PluginHost::IShell,
     MOCK_METHOD(void, Unregister, (const IShell::ICOMLink::INotification*), (override));
     MOCK_METHOD(WPEFramework::RPC::IRemoteConnection*, RemoteConnection, (const uint32_t), (override));
     MOCK_METHOD(void*, Instantiate, (const WPEFramework::RPC::Object&, const uint32_t, uint32_t&), (override));
+    MOCK_METHOD(WPEFramework::RPC::IStringIterator*, GetLibrarySearchPaths, (const string&), (const, override));
     BEGIN_INTERFACE_MAP(ServiceMock)
     INTERFACE_ENTRY(IShell)
     INTERFACE_ENTRY(IShell::ICOMLink)

diff --git a/PersistentStore/l1test/ServiceMock.h b/PersistentStore/l1test/ServiceMock.h
@@ -53,6 +53,7 @@ class ServiceMock : public WPEFramework::PluginHost::IShell,
     MOCK_METHOD(void, Unregister, (const IShell::ICOMLink::INotification*), (override));
     MOCK_METHOD(WPEFramework::RPC::IRemoteConnection*, RemoteConnection, (const uint32_t), (override));
     MOCK_METHOD(void*, Instantiate, (const WPEFramework::RPC::Object&, const uint32_t, uint32_t&), (override));
+    MOCK_METHOD(WPEFramework::RPC::IStringIterator*, GetLibrarySearchPaths, (const string&), (const, override));
     BEGIN_INTERFACE_MAP(ServiceMock)
     INTERFACE_ENTRY(IShell)
     INTERFACE_ENTRY(IShell::ICOMLink)

diff --git a/PersistentStore/sqlite/Store2.h b/PersistentStore/sqlite/Store2.h
@@ -36,6 +36,35 @@ namespace Plugin {
                        public Exchange::IStoreCache,
                        public Exchange::IStoreInspector,
                        public Exchange::IStoreLimit {
+        private:
+            class Job : public Core::IDispatch {
+            public:
+                Job(Store2* parent, const IStore2::ScopeType scope, const string& ns, const string& key, const string& value)
+                    : _parent(parent)
+                    , _scope(scope)
+                    , _ns(ns)
+                    , _key(key)
+                    , _value(value)
+                {
+                    _parent->AddRef();
+                }
+                ~Job() override
+                {
+                    _parent->Release();
+                }
+                void Dispatch() override
+                {
+                    _parent->OnValueChanged(_scope, _ns, _key, _value);
+                }
+
+            private:
+                Store2* _parent;
+                const IStore2::ScopeType _scope;
+                const string _ns;
+                const string _key;
+                const string _value;
+            };
+
         private:
             Store2(const Store2&) = delete;
             Store2& operator=(const Store2&) = delete;
@@ -75,22 +104,59 @@ namespace Plugin {
                 if (rc != SQLITE_OK) {
                     OnError(__FUNCTION__, rc);
                 }
+                rc = sqlite3_busy_timeout(_data, SQLITE_TIMEOUT); // Timeout
+                if (rc != SQLITE_OK) {
+                    OnError(__FUNCTION__, rc);
+                }
                 const std::vector<string> statements = {
                     "pragma foreign_keys = on;",
-                    "pragma busy_timeout = 1000000;",
-                    "create table if not exists namespace (id integer primary key,name text unique);",
-                    "create table if not exists item (ns integer,key text,value text,foreign key(ns) references namespace(id) on delete cascade on update no action,unique(ns,key) on conflict replace);",
-                    "create table if not exists limits (n integer,size integer,foreign key(n) references namespace(id) on delete cascade on update no action,unique(n) on conflict replace);",
+                    "create table if not exists namespace"
+                    " (id integer primary key,name text unique);",
+                    "create table if not exists item"
+                    " (ns integer,key text,value text,"
+                    "foreign key(ns) references namespace(id) on delete cascade on update no action,"
+                    "unique(ns,key) on conflict replace);",
+                    "create table if not exists limits"
+                    " (n integer,size integer,"
+                    "foreign key(n) references namespace(id) on delete cascade on update no action,"
+                    "unique(n) on conflict replace);",
                     "alter table item add column ttl integer;",
-                    "create temporary trigger if not exists ns_empty insert on namespace begin select case when length(new.name) = 0 then raise (fail, 'empty') end; end;",
-                    "create temporary trigger if not exists key_empty insert on item begin select case when length(new.key) = 0 then raise (fail, 'empty') end; end;",
-                    "create temporary trigger if not exists ns_maxvalue insert on namespace begin select case when length(new.name) > " + std::to_string(_maxValue) + " then raise (fail, 'max value') end; end;",
-                    "create temporary trigger if not exists key_maxvalue insert on item begin select case when length(new.key) > " + std::to_string(_maxValue) + " then raise (fail, 'max value') end; end;",
-                    "create temporary trigger if not exists value_maxvalue insert on item begin select case when length(new.value) > " + std::to_string(_maxValue) + " then raise (fail, 'max value') end; end;",
-                    "create temporary trigger if not exists ns_maxsize insert on namespace begin select case when (select sum(s) from (select sum(length(key)+length(value)) s from item union all select sum(length(name)) s from namespace union all select length(new.name) s)) > " + std::to_string(_maxSize) + " then raise (fail, 'max size') end; end;",
-                    "create temporary trigger if not exists item_maxsize insert on item begin select case when (select sum(s) from (select sum(length(key)+length(value)) s from item union all select sum(length(name)) s from namespace union all select length(new.key)+length(new.value) s)) > " + std::to_string(_maxSize) + " then raise (fail, 'max size') end; end;",
-                    "create temporary trigger if not exists item_limit_default insert on item begin select case when (select length(new.key)+length(new.value)+sum(length(key)+length(value)) from item where ns = new.ns) > " + std::to_string(_limit) + " then raise (fail, 'limit') end; end;",
-                    "create temporary trigger if not exists item_limit insert on item begin select case when (select size-length(new.key)-length(new.value)-sum(length(key)+length(value)) from limits inner join item on limits.n = item.ns where n = new.ns) < 0 then raise (fail, 'limit') end; end;"
+                    "create temporary trigger if not exists ns_empty insert on namespace"
+                    " begin select case when length(new.name) = 0"
+                    " then raise (fail, 'empty') end; end;",
+                    "create temporary trigger if not exists key_empty insert on item"
+                    " begin select case when length(new.key) = 0"
+                    " then raise (fail, 'empty') end; end;",
+                    "create temporary trigger if not exists ns_maxvalue insert on namespace"
+                    " begin select case when length(new.name) > "
+                        + std::to_string(_maxValue) + " then raise (fail, 'max value') end; end;",
+                    "create temporary trigger if not exists key_maxvalue insert on item"
+                    " begin select case when length(new.key) > "
+                        + std::to_string(_maxValue) + " then raise (fail, 'max value') end; end;",
+                    "create temporary trigger if not exists value_maxvalue insert on item"
+                    " begin select case when length(new.value) > "
+                        + std::to_string(_maxValue) + " then raise (fail, 'max value') end; end;",
+                    "create temporary trigger if not exists ns_maxsize insert on namespace"
+                    " begin select case when"
+                    " (select sum(s) from (select sum(length(key)+length(value)) s from item"
+                    " union all select sum(length(name)) s from namespace"
+                    " union all select length(new.name) s)) > "
+                        + std::to_string(_maxSize) + " then raise (fail, 'max size') end; end;",
+                    "create temporary trigger if not exists item_maxsize insert on item"
+                    " begin select case when"
+                    " (select sum(s) from (select sum(length(key)+length(value)) s from item"
+                    " union all select sum(length(name)) s from namespace"
+                    " union all select length(new.key)+length(new.value) s)) > "
+                        + std::to_string(_maxSize) + " then raise (fail, 'max size') end; end;",
+                    "create temporary trigger if not exists item_limit_default insert on item"
+                    " begin select case when"
+                    " (select length(new.key)+length(new.value)+sum(length(key)+length(value)) from item where ns = new.ns) > "
+                        + std::to_string(_limit) + " then raise (fail, 'limit') end; end;",
+                    "create temporary trigger if not exists item_limit insert on item"
+                    " begin select case when"
+                    " (select size-length(new.key)-length(new.value)-sum(length(key)+length(value)) from limits"
+                    " inner join item on limits.n = item.ns where n = new.ns) < 0"
+                    " then raise (fail, 'limit') end; end;"
                 };
                 for (auto& sql : statements) {
                     auto rc = sqlite3_exec(_data, sql.c_str(), nullptr, nullptr, nullptr);
@@ -106,13 +172,20 @@ namespace Plugin {
                     OnError(__FUNCTION__, rc);
                 }
             }
-            static bool IsTimeSynced()
+            bool IsTimeSynced() const
             {
 #ifdef WITH_SYSMGR
+                // Get actual state, as it may change at any time...
                 IARM_Bus_Init(IARM_INIT_NAME);
                 IARM_Bus_Connect();
                 IARM_Bus_SYSMgr_GetSystemStates_Param_t param;
-                if ((IARM_Bus_Call(IARM_BUS_SYSMGR_NAME, IARM_BUS_SYSMGR_API_GetSystemStates, &param, sizeof(param)) != IARM_RESULT_SUCCESS)
+                if ((IARM_Bus_Call_with_IPCTimeout(
+                         IARM_BUS_SYSMGR_NAME,
+                         IARM_BUS_SYSMGR_API_GetSystemStates,
+                         &param,
+                         sizeof(param),
+                         IARM_TIMEOUT) // Timeout
+                        != IARM_RESULT_SUCCESS)
                     || !param.time_source.state) {
                     return false;
                 }
@@ -185,7 +258,9 @@ namespace Plugin {
                 }
 
                 if (rc == SQLITE_DONE) {
-                    OnValueChanged(ns, key, value);
+                    Core::IWorkerPool::Instance().Submit(Core::ProxyType<Core::IDispatch>(
+                        Core::ProxyType<Job>::Create(this, scope, ns, key, value))); // Decouple notification
+
                     result = Core::ERROR_NONE;
                 } else {
                     OnError(__FUNCTION__, rc);
@@ -489,15 +564,16 @@ namespace Plugin {
             END_INTERFACE_MAP
 
         private:
-            void OnValueChanged(const string& ns, const string& key, const string& value)
+            void OnValueChanged(const IStore2::ScopeType scope, const string& ns, const string& key, const string& value)
             {
                 Core::SafeSyncType<Core::CriticalSection> lock(_clientLock);
 
                 std::list<INotification*>::iterator
                     index(_clients.begin());
 
                 while (index != _clients.end()) {
-                    (*index)->ValueChanged(IStore2::ScopeType::DEVICE, ns, key, value);
+                    // If main process is out of threads, this can time out, and IPC will mess up...
+                    (*index)->ValueChanged(scope, ns, key, value);
                     index++;
                 }
             }

diff --git a/PersistentStore/sqlite/l1test/Store2Test.cpp b/PersistentStore/sqlite/l1test/Store2Test.cpp
@@ -3,6 +3,7 @@
 
 #include "../Store2.h"
 #include "Store2NotificationMock.h"
+#include "WorkerPoolImplementation.h"
 
 using ::testing::_;
 using ::testing::Eq;
@@ -39,10 +40,18 @@ const auto kLimit40 = 40;
 
 class AStore2 : public Test {
 protected:
+    WPEFramework::Core::ProxyType<WorkerPoolImplementation> workerPool;
     WPEFramework::Core::ProxyType<Store2> store2;
     AStore2()
-        : store2(WPEFramework::Core::ProxyType<Store2>::Create(kPath, kMaxSize, kMaxValue, kLimit))
+        : workerPool(WPEFramework::Core::ProxyType<WorkerPoolImplementation>::Create(
+              WPEFramework::Core::Thread::DefaultStackSize()))
+        , store2(WPEFramework::Core::ProxyType<Store2>::Create(kPath, kMaxSize, kMaxValue, kLimit))
     {
+        WPEFramework::Core::IWorkerPool::Assign(&(*workerPool));
+    }
+    ~AStore2() override
+    {
+        WPEFramework::Core::IWorkerPool::Assign(nullptr);
     }
 };