1 /*
2 * Copyright (c) 2023 Huawei Device Co., Ltd.
3 * Licensed under the Apache License, Version 2.0 (the "License");
4 * you may not use this file except in compliance with the License.
5 * You may obtain a copy of the License at
6 *
7 * http://www.apache.org/licenses/LICENSE-2.0
8 *
9 * Unless required by applicable law or agreed to in writing, software
10 * distributed under the License is distributed on an "AS IS" BASIS,
11 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 * See the License for the specific language governing permissions and
13 * limitations under the License.
14 */
15
16 #include "cloud/cloud_meta_data.h"
17 #include "cloud/cloud_db_constant.h"
18 #include "db_errno.h"
19 #include "parcel.h"
20
21 namespace DistributedDB {
22
CloudMetaData(ICloudSyncStorageInterface * store)23 CloudMetaData::CloudMetaData(ICloudSyncStorageInterface *store)
24 : store_(store)
25 {
26 }
27
GetPrefixTableName(const TableName & tableName)28 Key CloudMetaData::GetPrefixTableName(const TableName &tableName)
29 {
30 TableName newName = CloudDbConstant::CLOUD_META_TABLE_PREFIX + tableName;
31 Key prefixedTableName(newName.begin(), newName.end());
32 return prefixedTableName;
33 }
34
GetLocalWaterMark(const TableName & tableName,Timestamp & localMark)35 int CloudMetaData::GetLocalWaterMark(const TableName &tableName, Timestamp &localMark)
36 {
37 std::lock_guard<std::mutex> lock(cloudMetaMutex_);
38 if (cloudMetaVals_.count(tableName) == 0) {
39 int ret = ReadMarkFromMeta(tableName);
40 if (ret != E_OK) {
41 return ret;
42 }
43 }
44 localMark = cloudMetaVals_[tableName].localMark;
45 return E_OK;
46 }
47
GetCloudWaterMark(const TableName & tableName,std::string & cloudMark)48 int CloudMetaData::GetCloudWaterMark(const TableName &tableName, std::string &cloudMark)
49 {
50 std::lock_guard<std::mutex> lock(cloudMetaMutex_);
51 if (cloudMetaVals_.count(tableName) == 0) {
52 int ret = ReadMarkFromMeta(tableName);
53 if (ret != E_OK) {
54 return ret;
55 }
56 }
57 cloudMark = cloudMetaVals_[tableName].cloudMark;
58 LOGD("[Meta] get cloud water mark=%s", cloudMark.c_str());
59 return E_OK;
60 }
61
SetLocalWaterMark(const TableName & tableName,Timestamp localMark)62 int CloudMetaData::SetLocalWaterMark(const TableName &tableName, Timestamp localMark)
63 {
64 std::lock_guard<std::mutex> lock(cloudMetaMutex_);
65 std::string cloudMark = "";
66 auto iter = cloudMetaVals_.find(tableName);
67 if (iter != cloudMetaVals_.end()) {
68 cloudMark = iter->second.cloudMark;
69 }
70 int ret = WriteMarkToMeta(tableName, localMark, cloudMark);
71 if (ret != E_OK) {
72 return ret;
73 }
74 if (iter == cloudMetaVals_.end()) {
75 CloudMetaValue cloudMetaVal = { .localMark = localMark, .cloudMark = cloudMark };
76 cloudMetaVals_[tableName] = cloudMetaVal;
77 } else {
78 iter->second.localMark = localMark;
79 }
80 return E_OK;
81 }
82
SetCloudWaterMark(const TableName & tableName,std::string & cloudMark)83 int CloudMetaData::SetCloudWaterMark(const TableName &tableName, std::string &cloudMark)
84 {
85 std::lock_guard<std::mutex> lock(cloudMetaMutex_);
86 Timestamp localMark = 0;
87 auto iter = cloudMetaVals_.find(tableName);
88 if (iter != cloudMetaVals_.end()) {
89 localMark = iter->second.localMark;
90 }
91 int ret = WriteMarkToMeta(tableName, localMark, cloudMark);
92 if (ret != E_OK) {
93 return ret;
94 }
95 if (iter == cloudMetaVals_.end()) {
96 CloudMetaValue cloudMetaVal = { .localMark = localMark, .cloudMark = cloudMark };
97 cloudMetaVals_[tableName] = cloudMetaVal;
98 } else {
99 iter->second.cloudMark = cloudMark;
100 }
101 LOGD("[Meta] set cloud water mark=%s", cloudMark.c_str());
102 return E_OK;
103 }
104
ReadMarkFromMeta(const TableName & tableName)105 int CloudMetaData::ReadMarkFromMeta(const TableName &tableName)
106 {
107 if (store_ == nullptr) {
108 return -E_INVALID_DB;
109 }
110 Value blobMetaVal;
111 int ret = store_->GetMetaData(GetPrefixTableName(tableName), blobMetaVal);
112 if (ret != -E_NOT_FOUND && ret != E_OK) {
113 return ret;
114 }
115 CloudMetaValue cloudMetaValue;
116 ret = DeserializeMark(blobMetaVal, cloudMetaValue);
117 if (ret != E_OK) {
118 return ret;
119 }
120 cloudMetaVals_[tableName] = cloudMetaValue;
121 return E_OK;
122 }
123
WriteMarkToMeta(const TableName & tableName,Timestamp localmark,std::string & cloudMark)124 int CloudMetaData::WriteMarkToMeta(const TableName &tableName, Timestamp localmark, std::string &cloudMark)
125 {
126 Value blobMetaVal;
127 int ret = SerializeMark(localmark, cloudMark, blobMetaVal);
128 if (ret != E_OK) {
129 return ret;
130 }
131 if (store_ == nullptr) {
132 return -E_INVALID_DB;
133 }
134 return store_->PutMetaData(GetPrefixTableName(tableName), blobMetaVal);
135 }
136
SerializeMark(Timestamp localMark,std::string & cloudMark,Value & blobMeta)137 int CloudMetaData::SerializeMark(Timestamp localMark, std::string &cloudMark, Value &blobMeta)
138 {
139 uint64_t length = Parcel::GetUInt64Len() + Parcel::GetStringLen(cloudMark);
140 blobMeta.resize(length);
141 Parcel parcel(blobMeta.data(), blobMeta.size());
142 parcel.WriteUInt64(localMark);
143 parcel.WriteString(cloudMark);
144 if (parcel.IsError()) {
145 LOGE("[Meta] Parcel error while serializing cloud meta data.");
146 return -E_PARSE_FAIL;
147 }
148 return E_OK;
149 }
150
DeserializeMark(Value & blobMark,CloudMetaValue & cloudMetaValue)151 int CloudMetaData::DeserializeMark(Value &blobMark, CloudMetaValue &cloudMetaValue)
152 {
153 if (blobMark.empty()) {
154 cloudMetaValue.localMark = 0;
155 cloudMetaValue.cloudMark = "";
156 return E_OK;
157 }
158 Parcel parcel(blobMark.data(), blobMark.size());
159 parcel.ReadUInt64(cloudMetaValue.localMark);
160 parcel.ReadString(cloudMetaValue.cloudMark);
161 if (parcel.IsError()) {
162 LOGE("[Meta] Parcel error while deserializing cloud meta data.");
163 return -E_PARSE_FAIL;
164 }
165 return E_OK;
166 }
167
CleanWaterMark(const TableName & tableName)168 int CloudMetaData::CleanWaterMark(const TableName &tableName)
169 {
170 std::lock_guard<std::mutex> lock(cloudMetaMutex_);
171 std::string cloudWaterMark;
172 int ret = WriteMarkToMeta(tableName, 0, cloudWaterMark);
173 if (ret != E_OK) {
174 return ret;
175 }
176 cloudMetaVals_[tableName] = {};
177 LOGD("[Meta] clean cloud water mark");
178 return E_OK;
179 }
180 } // namespace DistributedDB