!2398 对于超过 200k 的 RdbChangeNode 数据，改为使用共享内存传输，避免使用 IPC 传输

Merge pull request !2398 from 中饭吃啥/master

services/distributeddataservice/service/data_share/data_share_obs_proxy.cpp

@@ -15,13 +15,124 @@
 
 #define LOG_TAG "ObserverProxy"
 #include "data_share_obs_proxy.h"
+#include "datashare_errno.h"
 
 #include "itypes_util.h"
+#include "datashare_itypes_utils.h"
 #include "log_print.h"
 
 namespace OHOS {
 namespace DataShare {
 static constexpr int REQUEST_CODE = 0;
+int RdbObserverProxy::CreateAshmem(RdbChangeNode &changeNode)
+{
+    OHOS::sptr<Ashmem> memory = Ashmem::CreateAshmem(ASHMEM_NAME, DATA_SIZE_ASHMEM_TRANSFER_LIMIT);
+    if (memory == nullptr) {
+        ZLOGE("failed to create Ashmem instance.");
+        return E_ERROR;
+    }
+    bool mapRet = memory->MapReadAndWriteAshmem();
+    if (!mapRet) {
+        ZLOGE("failed to map read and write ashmem, ret=%{public}d", mapRet);
+        memory->CloseAshmem();
+        return E_ERROR;
+    }
+    if (changeNode.memory_ != nullptr) {
+        ZLOGE(
+            "Unknown error: changeNode.memory_ should be null, but something is there %{public}p",
+            (void *)changeNode.memory_
+        );
+        return E_ERROR;
+    }
+    changeNode.memory_ = memory;
+    return E_OK;
+}
+
+int RdbObserverProxy::WriteAshmem(RdbChangeNode &changeNode, void *data, int len, int &offset)
+{
+    if (changeNode.memory_ == nullptr) {
+        ZLOGE("changeNode memory is nullptr.");
+        return E_ERROR;
+    }
+    bool writeRet = changeNode.memory_->WriteToAshmem(data, len, offset);
+    if (!writeRet) {
+        ZLOGE("failed to write into ashmem, ret=%{public}d", writeRet);
+        changeNode.memory_->UnmapAshmem();
+        changeNode.memory_->CloseAshmem();
+        changeNode.memory_ = nullptr;
+        return E_ERROR;
+    }
+    offset += len;
+    return E_OK;
+}
+
+int RdbObserverProxy::SerializeDataIntoAshmem(RdbChangeNode &changeNode)
+{
+    if (changeNode.memory_ == nullptr) {
+        ZLOGE("changeNode.memory_ is nullptr");
+        return E_ERROR;
+    }
+    // move data
+    // simple serialization: [vec_size(int32); str1_len(int32), str1; str2_len(int32), str2; ...],
+    // total byte size is recorded in changeNode.size
+    int offset = 0;
+    // 4 byte for length int
+    int intLen = 4;
+    int dataSize = changeNode.data_.size();
+    if (WriteAshmem(changeNode, (void *)&dataSize, intLen, offset) != E_OK) {
+        ZLOGE("failed to write data with len %{public}d, offset %{public}d.", intLen, offset);
+        return E_ERROR;
+    }
+    for (int i = 0; i < dataSize; i++) {
+        const char *str = changeNode.data_[i].c_str();
+        int strLen = changeNode.data_[i].length();
+        // write length int
+        if (WriteAshmem(changeNode, (void *)&strLen, intLen, offset) != E_OK) {
+            ZLOGE("failed to write data with index %{public}d, len %{public}d, offset %{public}d.", i, intLen, offset);
+            return E_ERROR;
+        }
+        // write str
+        if (WriteAshmem(changeNode, (void *)str, strLen, offset) != E_OK) {
+            ZLOGE("failed to write data with index %{public}d, len %{public}d, offset %{public}d.", i, strLen, offset);
+            return E_ERROR;
+        }
+    }
+    changeNode.size_ = offset;
+    return E_OK;
+}
+
+int RdbObserverProxy::PrepareRdbChangeNodeData(RdbChangeNode &changeNode)
+{
+    // If data size is bigger than the limit, move it to the shared memory
+    // 4 byte for length int
+    int intByteLen = 4;
+    int size = intByteLen;
+    for (int i = 0; i < changeNode.data_.size(); i++) {
+        size += intByteLen;
+        size += changeNode.data_[i].length();
+    }
+    if (size > DATA_SIZE_ASHMEM_TRANSFER_LIMIT) {
+        ZLOGE("Data to write into ashmem is %{public}d bytes, over 10M.", size);
+        return E_ERROR;
+    }
+    if (size > DATA_SIZE_IPC_TRANSFER_LIMIT) {
+        ZLOGD("Data size is over 200k, transfer it by the shared memory");
+        if (RdbObserverProxy::CreateAshmem(changeNode) != E_OK) {
+            ZLOGE("failed to create ashmem.");
+            return E_ERROR;
+        }
+        if (RdbObserverProxy::SerializeDataIntoAshmem(changeNode) != E_OK) {
+            ZLOGE("failed to serialize data into ashmem.");
+            return E_ERROR;
+        }
+        // clear original data spot
+        changeNode.data_.clear();
+        changeNode.isSharedMemory_ = true;
+        ZLOGD("Preparation done. Data size: %{public}d", changeNode.size_);
+    }
+    return E_OK;
+}
+
 void RdbObserverProxy::OnChangeFromRdb(RdbChangeNode &changeNode)
 {
     MessageParcel parcel;
@@ -29,6 +140,11 @@ void RdbObserverProxy::OnChangeFromRdb(RdbChangeNode &changeNode)
         return;
     }
 
+    if (RdbObserverProxy::PrepareRdbChangeNodeData(changeNode) != E_OK) {
+        ZLOGE("failed to prepare RdbChangeNode data.");
+        return;
+    }
+
     if (!ITypesUtil::Marshal(parcel, changeNode)) {
         ZLOGE("failed to WriteParcelable changeNode ");
         return;

services/distributeddataservice/service/data_share/data_share_obs_proxy.h

@@ -28,6 +28,10 @@ public:
     void OnChangeFromRdb(RdbChangeNode &changeNode) override;
 
 private:
+    int PrepareRdbChangeNodeData(RdbChangeNode &changeNode);
+    int CreateAshmem(RdbChangeNode &changeNode);
+    int WriteAshmem(RdbChangeNode &changeNode, void *data, int len, int &offset);
+    int SerializeDataIntoAshmem(RdbChangeNode &changeNode);
     static inline BrokerDelegator<RdbObserverProxy> delegator_;
 };
 

services/distributeddataservice/service/data_share/data_share_types_util.cpp

@@ -85,7 +85,21 @@ bool Unmarshalling(PredicateTemplateNode &predicateTemplateNode, MessageParcel &
 template<>
 bool Marshalling(const RdbChangeNode &changeNode, MessageParcel &parcel)
 {
-    return ITypesUtil::Marshal(parcel, changeNode.uri_, changeNode.templateId_, changeNode.data_);
+    bool firstPart = ITypesUtil::Marshal(
+        parcel, changeNode.uri_, changeNode.templateId_, changeNode.data_, changeNode.isSharedMemory_);
+    if (!firstPart) {
+        return false;
+    }
+    if (changeNode.isSharedMemory_) {
+        if (changeNode.memory_ == nullptr) {
+            ZLOGE("Used shared memory but ashmem is nullptr.");
+            return false;
+        }
+        if (!parcel.WriteAshmem(changeNode.memory_)) {
+            return false;
+        }
+    }
+    return ITypesUtil::Marshal(parcel, changeNode.size_);
 }
 
 template<>

services/distributeddataservice/service/test/BUILD.gn

@@ -970,6 +970,7 @@ ohos_unittest("DataShareServiceImplTest") {
     "${data_service_path}/service/data_share/sys_event_subscriber.cpp",
     "${data_service_path}/service/kvdb/user_delegate.cpp",
     "${data_service_path}/service/permission/src/permit_delegate.cpp",
+    "data_share_obs_proxy_test.cpp",
     "data_share_profile_config_test.cpp",
     "data_share_service_impl_test.cpp",
     "data_share_service_stub_test.cpp",

services/distributeddataservice/service/test/data_share_obs_proxy_test.cpp

@@ -0,0 +1,318 @@
+/*
+* Copyright (c) 2024 Huawei Device Co., Ltd.
+* Licensed under the Apache License, Version 2.0 (the "License");
+* you may not use this file except in compliance with the License.
+* You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+#define LOG_TAG "DataShareObsProxyTest"
+
+#include <gtest/gtest.h>
+#include <unistd.h>
+
+#include "data_share_obs_proxy.h"
+#include "datashare_errno.h"
+#include "log_print.h"
+
+namespace OHOS::Test {
+using namespace testing::ext;
+using namespace OHOS::DataShare;
+std::string BUNDLE_NAME = "ohos.datasharetest.demo";
+constexpr int64_t TEST_SUB_ID = 100;
+
+class DataShareObsProxyTest : public testing::Test {
+public:
+    static void SetUpTestCase(void){};
+    static void TearDownTestCase(void){};
+    void SetUp(){};
+    void TearDown(){};
+};
+
+RdbChangeNode SampleRdbChangeNode()
+{
+    TemplateId tplId;
+    tplId.subscriberId_ = TEST_SUB_ID;
+    tplId.bundleName_ = BUNDLE_NAME;
+
+    RdbChangeNode node;
+    node.uri_ = std::string("");
+    node.templateId_ = tplId;
+    node.data_ = std::vector<std::string>();
+    node.isSharedMemory_ = false;
+    node.memory_ = nullptr;
+    node.size_ = 0;
+
+    return node;
+}
+
+/**
+* @tc.name: CreateAshmem
+* @tc.desc: test CreateAshmem function
+* @tc.type: FUNC
+* @tc.require:
+*/
+HWTEST_F(DataShareObsProxyTest, CreateAshmem, TestSize.Level1)
+{
+    ZLOGI("CreateAshmem starts");
+    RdbChangeNode node = SampleRdbChangeNode();
+    
+    OHOS::sptr<IRemoteObject> fake = nullptr;
+    RdbObserverProxy proxy(fake);
+    int ret = proxy.CreateAshmem(node);
+    EXPECT_EQ(ret, DataShare::E_OK);
+    EXPECT_NE(node.memory_, nullptr);
+    ZLOGI("CreateAshmem ends");
+}
+
+/**
+* @tc.name: WriteAshmem001
+* @tc.desc: test WriteAshmem function. Write an int
+* @tc.type: FUNC
+* @tc.require:
+*/
+HWTEST_F(DataShareObsProxyTest, WriteAshmem001, TestSize.Level1)
+{
+    ZLOGI("WriteAshmem001 starts");
+    RdbChangeNode node = SampleRdbChangeNode();
+    OHOS::sptr<IRemoteObject> fake = nullptr;
+    RdbObserverProxy proxy(fake);
+    int retCreate = proxy.CreateAshmem(node);
+    EXPECT_EQ(retCreate, DataShare::E_OK);
+
+    int len = 10;
+    int intLen = 4;
+    int offset = 0;
+    int ret = proxy.WriteAshmem(node, (void *)&len, intLen, offset);
+    EXPECT_EQ(ret, E_OK);
+    EXPECT_EQ(offset, intLen);
+
+    // read from the start
+    const void *read = node.memory_->ReadFromAshmem(intLen, 0);
+    EXPECT_NE(read, nullptr);
+    int lenRead = *reinterpret_cast<const int *>(read);
+    EXPECT_EQ(len, lenRead);
+    ZLOGI("WriteAshmem001 ends");
+}
+
+/**
+* @tc.name: WriteAshmem002
+* @tc.desc: test WriteAshmem function. Write a str
+* @tc.type: FUNC
+* @tc.require:
+*/
+HWTEST_F(DataShareObsProxyTest, WriteAshmem002, TestSize.Level1)
+{
+    ZLOGI("WriteAshmem002 starts");
+    RdbChangeNode node = SampleRdbChangeNode();
+    OHOS::sptr<IRemoteObject> fake = nullptr;
+    RdbObserverProxy proxy(fake);
+    int retCreate = proxy.CreateAshmem(node);
+    EXPECT_EQ(retCreate, DataShare::E_OK);
+
+    std::string string("Hello World");
+    const char *str = string.c_str();
+    int len = string.length();
+    int offset = 0;
+    int ret = proxy.WriteAshmem(node, (void *)str, len, offset);
+    EXPECT_EQ(ret, E_OK);
+    EXPECT_EQ(offset, len);
+
+    // read from the start
+    const void *read = node.memory_->ReadFromAshmem(len, 0);
+    EXPECT_NE(read, nullptr);
+    const char *strRead = reinterpret_cast<const char *>(read);
+    std::string stringRead(strRead, len);
+    EXPECT_EQ(stringRead, string);
+    ZLOGI("WriteAshmem002 ends");
+}
+
+/**
+* @tc.name: WriteAshmem003
+* @tc.desc: test WriteAshmem function with error
+* @tc.type: FUNC
+* @tc.require:
+*/
+HWTEST_F(DataShareObsProxyTest, WriteAshmem003, TestSize.Level1)
+{
+    ZLOGI("WriteAshmem003 starts");
+    RdbChangeNode node = SampleRdbChangeNode();
+    OHOS::sptr<IRemoteObject> fake = nullptr;
+    RdbObserverProxy proxy(fake);
+    
+    OHOS::sptr<Ashmem> memory = Ashmem::CreateAshmem("WriteAshmem003", 2);
+    EXPECT_NE(memory, nullptr);
+    bool mapRet = memory->MapReadAndWriteAshmem();
+    ASSERT_TRUE(mapRet);
+    node.memory_ = memory;
+
+    int len = 10;
+    int offset = 0;
+    int ret = proxy.WriteAshmem(node, (void *)&len, 4, offset);
+    EXPECT_EQ(ret, E_ERROR);
+    ZLOGI("WriteAshmem003 ends");
+}
+
+/**
+* @tc.name: SerializeDataIntoAshmem
+* @tc.desc: test SerializeDataIntoAshmem function
+* @tc.type: FUNC
+* @tc.require:
+*/
+HWTEST_F(DataShareObsProxyTest, SerializeDataIntoAshmem, TestSize.Level1)
+{
+    ZLOGI("SerializeDataIntoAshmem starts");
+    RdbChangeNode node = SampleRdbChangeNode();
+
+    OHOS::sptr<IRemoteObject> fake = nullptr;
+    RdbObserverProxy proxy(fake);
+
+    int retCreate = proxy.CreateAshmem(node);
+    EXPECT_EQ(retCreate, E_OK);
+
+    // Push three times
+    node.data_.push_back(BUNDLE_NAME);
+    node.data_.push_back(BUNDLE_NAME);
+    node.data_.push_back(BUNDLE_NAME);
+
+    int intLen = 4;
+    // item length size + (str length size + str length) * 3
+    int offset = intLen + (intLen + strlen(BUNDLE_NAME.c_str())) * 3;
+    int retSe = proxy.SerializeDataIntoAshmem(node);
+    EXPECT_EQ(retSe, E_OK);
+    EXPECT_EQ(node.size_, offset);
+
+    offset = 0;
+    const void *vecLenRead = node.memory_->ReadFromAshmem(intLen, offset);
+    EXPECT_NE(vecLenRead, nullptr);
+    int vecLen = *reinterpret_cast<const int *>(vecLenRead);
+    EXPECT_EQ(vecLen, 3);
+    offset += intLen;
+
+    // 3 strings in the vec
+    for (int i = 0; i < 3; i++) {
+        const void *strLenRead = node.memory_->ReadFromAshmem(intLen, offset);
+        EXPECT_NE(strLenRead, nullptr);
+        int strLen = *reinterpret_cast<const int *>(strLenRead);
+        EXPECT_EQ(strLen, BUNDLE_NAME.length());
+        offset += intLen;
+
+        const void *strRead = node.memory_->ReadFromAshmem(strLen, offset);
+        EXPECT_NE(strRead, nullptr);
+        const char *str = reinterpret_cast<const char *>(strRead);
+        std::string stringRead(str, strLen);
+        EXPECT_EQ(stringRead, BUNDLE_NAME);
+        offset += strLen;
+    }
+    ZLOGI("SerializeDataIntoAshmem ends");
+}
+
+/**
+* @tc.name: SerializeDataIntoAshmem002
+* @tc.desc: test SerializeDataIntoAshmem function
+* @tc.type: FUNC
+* @tc.require:
+*/
+HWTEST_F(DataShareObsProxyTest, SerializeDataIntoAshmem002, TestSize.Level1)
+{
+    ZLOGI("SerializeDataIntoAshmem starts");
+    RdbChangeNode node = SampleRdbChangeNode();
+
+    OHOS::sptr<IRemoteObject> fake = nullptr;
+    RdbObserverProxy proxy(fake);
+
+    // Push three times
+    node.data_.push_back(BUNDLE_NAME);
+    node.data_.push_back(BUNDLE_NAME);
+    node.data_.push_back(BUNDLE_NAME);
+
+    // memory too small for vec length
+    OHOS::sptr<Ashmem> memory = Ashmem::CreateAshmem("SerializeDataIntoAshmem002", 2);
+    EXPECT_NE(memory, nullptr);
+    bool mapRet = memory->MapReadAndWriteAshmem();
+    ASSERT_TRUE(mapRet);
+    node.memory_ = memory;
+
+    int retSe = proxy.SerializeDataIntoAshmem(node);
+    EXPECT_EQ(retSe, E_ERROR);
+    EXPECT_EQ(node.size_, 0);
+    EXPECT_EQ(node.data_.size(), 3);
+    ASSERT_FALSE(node.isSharedMemory_);
+
+    // memory too small for string length
+    OHOS::sptr<Ashmem> memory2 = Ashmem::CreateAshmem("SerializeDataIntoAshmem002", 6);
+    EXPECT_NE(memory2, nullptr);
+    mapRet = memory2->MapReadAndWriteAshmem();
+    ASSERT_TRUE(mapRet);
+    node.memory_ = memory2;
+
+    retSe = proxy.SerializeDataIntoAshmem(node);
+    EXPECT_EQ(retSe, E_ERROR);
+    EXPECT_EQ(node.size_, 0);
+    EXPECT_EQ(node.data_.size(), 3);
+    ASSERT_FALSE(node.isSharedMemory_);
+
+    // memory too small for string
+    OHOS::sptr<Ashmem> memory3 = Ashmem::CreateAshmem("SerializeDataIntoAshmem002", 10);
+    EXPECT_NE(memory3, nullptr);
+    mapRet = memory3->MapReadAndWriteAshmem();
+    ASSERT_TRUE(mapRet);
+    node.memory_ = memory3;
+
+    retSe = proxy.SerializeDataIntoAshmem(node);
+    EXPECT_EQ(retSe, E_ERROR);
+    EXPECT_EQ(node.size_, 0);
+    EXPECT_EQ(node.data_.size(), 3);
+    ASSERT_FALSE(node.isSharedMemory_);
+   
+    ZLOGI("SerializeDataIntoAshmem002 ends");
+}
+
+/**
+* @tc.name: PreparationData
+* @tc.desc: test PrepareRdbChangeNodeData function
+* @tc.type: FUNC
+* @tc.require:
+*/
+HWTEST_F(DataShareObsProxyTest, PreparationData, TestSize.Level1)
+{
+    ZLOGI("PreparationData starts");
+    RdbChangeNode node = SampleRdbChangeNode();
+    OHOS::sptr<IRemoteObject> fake = nullptr;
+    RdbObserverProxy proxy(fake);
+
+    // Push three times, less than 200k
+    node.data_.push_back(BUNDLE_NAME);
+    node.data_.push_back(BUNDLE_NAME);
+    node.data_.push_back(BUNDLE_NAME);
+
+    int ret = proxy.PrepareRdbChangeNodeData(node);
+    EXPECT_EQ(ret, E_OK);
+    EXPECT_EQ(node.data_.size(), 3);
+    EXPECT_FALSE(node.isSharedMemory_);
+
+    // Try to fake a 200k data. BUNDLE_NAME is 23 byte long and 7587 BUNDLE_NAMEs is over 200k.
+    for (int i = 0; i < 7587; i++) {
+        node.data_.push_back(BUNDLE_NAME);
+    }
+    ret = proxy.PrepareRdbChangeNodeData(node);
+    EXPECT_EQ(ret, E_OK);
+    EXPECT_EQ(node.data_.size(), 0);
+    EXPECT_TRUE(node.isSharedMemory_);
+
+    // Try to fake data over 10M. Write data of such size should fail because it exceeds the limit.
+    for (int i = 0; i < 388362; i++) {
+        node.data_.push_back(BUNDLE_NAME);
+    }
+    ret = proxy.PrepareRdbChangeNodeData(node);
+    EXPECT_EQ(ret, E_ERROR);
+
+    ZLOGI("PreparationData ends");
+}
+} // namespace OHOS::Test