* Copyright (c) Huawei Technologies Co., Ltd. 2022. All rights reserved.
*
* 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.
*/
* Description: The object client example.
*/
#include "datasystem/datasystem.h"
#include <iostream>
using datasystem::Buffer;
using datasystem::ConnectOptions;
using datasystem::CreateParam;
using datasystem::ObjectClient;
using datasystem::Optional;
using datasystem::Status;
static std::shared_ptr<ObjectClient> client_;
static std::shared_ptr<Buffer> buffer_;
static std::string DEFAULT_IP = "127.0.0.1";
static constexpr int DEFAULT_PORT = 9088;
static constexpr int PARAMETERS_NUM = 5;
static constexpr int64_t SHM_SIZE = 600 * 1024;
static void Write(int64_t size, bool isSeal)
{
std::cout << "Writing data to a buffer" << std::endl;
std::cout << std::boolalpha << "Immutable data: " << isSeal << std::endl;
std::string objectKey = "123456789";
std::string data(size, 'x');
std::vector<std::string> failedIds;
client_->GIncreaseRef({ objectKey }, failedIds);
Status status = client_->Create(objectKey, size, CreateParam{}, buffer_);
std::cout << "Create: " << status.ToString() << std::endl;
buffer_->WLatch();
buffer_->MemoryCopy((void *)data.data(), size);
if (isSeal) {
buffer_->Seal();
} else {
buffer_->Publish();
}
buffer_->UnWLatch();
std::cout << "Write succeeds." << std::endl;
}
static void Read(int64_t size, bool isSeal)
{
Write(size, isSeal);
std::cout << "Reading data from a buffer" << std::endl;
std::string objectKey = "123456789";
std::vector<std::string> objKeys = { objectKey };
const int64_t timeoutMs = 60000;
std::vector<Optional<Buffer>> buffers;
Status status = client_->Get(objKeys, timeoutMs, buffers);
if (!status.IsOk()) {
std::cerr << "Read Get Fail: " << status.ToString() << std::endl;
return;
}
std::cout << "Get: " << status.ToString() << std::endl;
auto &buf = buffers[0];
buf->RLatch();
auto data = std::string(reinterpret_cast<const char *>(buf->ImmutableData()), buf->GetSize());
buf->UnRLatch();
std::cout << "Read succeeds." << std::endl;
}
static void Modify(int64_t size)
{
Write(size, false);
std::cout << "Modifying data in the buffer" << std::endl;
std::string objectKey = "123456789";
std::vector<std::string> objKeys = { objectKey };
const int64_t timeoutMs = 60000;
std::vector<Optional<Buffer>> buffers;
Status status = client_->Get(objKeys, timeoutMs, buffers);
if (!status.IsOk()) {
std::cerr << "Modify Get Fail: " << status.ToString() << std::endl;
return;
}
std::cout << "Get: " << status.ToString() << std::endl;
auto &buf = buffers[0];
std::string newData = "test";
auto newDataSize = newData.size();
buf->WLatch();
buf->MemoryCopy((void *)newData.data(), newDataSize);
buf->Publish();
buf->UnWLatch();
buf->RLatch();
auto data = std::string(reinterpret_cast<const char *>(buf->ImmutableData()), buf->GetSize());
buf->UnRLatch();
std::cout << "Modify succeeds !!!" << std::endl;
}
static void Delete(int64_t size, bool isSeal)
{
Write(size, isSeal);
std::cout << "Deleting...." << std::endl;
std::string objectKey = "123456789";
std::vector<std::string> objKeys = { objectKey };
std::vector<std::string> failedIds;
Status status = client_->GDecreaseRef(objKeys, failedIds);
std::cout << "GDecreaseRef: " << status.ToString() << std::endl;
}
void Start(int64_t size, bool isSeal)
{
Write(size, isSeal);
Read(size, isSeal);
Modify(size);
Delete(size, isSeal);
}
int InitClient(const ConnectOptions &connectOpts)
{
std::string workerAddr = connectOpts.host + ':' + std::to_string(connectOpts.port);
std::cout << "Get the object client workerAddr:" << workerAddr << std::endl;
client_ = std::make_shared<ObjectClient>(connectOpts);
Status status = client_->Init();
if (status.IsError()) {
std::cerr << "Failed to init object client : " << status.ToString() << std::endl;
return -1;
}
return 0;
}
void CloseClient()
{
if (client_) {
client_->ShutDown();
client_.reset();
}
}
int main(int argc, char *argv[])
{
const int authParametersNum = 8;
std::string ip;
int port = 0;
int index = 0;
int size;
bool isSeal;
std::string clientPublicKey, clientPrivateKey, serverPublicKey;
if (argc == 1) {
ip = DEFAULT_IP;
port = DEFAULT_PORT;
size = SHM_SIZE;
isSeal = false;
} else if (argc == PARAMETERS_NUM) {
ip = argv[++index];
port = atoi(argv[++index]);
size = atoi(argv[++index]);
isSeal = (std::string(argv[++index]) == "true");
} else if (argc == authParametersNum) {
ip = argv[++index];
port = atoi(argv[++index]);
size = atoi(argv[++index]);
isSeal = (std::string(argv[++index]) == "true");
clientPublicKey = argv[++index];
clientPrivateKey = argv[++index];
serverPublicKey = argv[++index];
} else {
std::cerr << "Invalid input parameters.";
return -1;
}
ConnectOptions connectOpts{ .host = ip,
.port = port,
.connectTimeoutMs = 60 * 1000,
.requestTimeoutMs = 0,
.clientPublicKey = clientPublicKey,
.clientPrivateKey = clientPrivateKey,
.serverPublicKey = serverPublicKey };
connectOpts.enableExclusiveConnection = false;
if (InitClient(connectOpts) != 0) {
return -1;
}
Start(size, isSeal);
CloseClient();
return 0;
}