* Copyright (c) Huawei Technologies Co., Ltd. 2023-2023. 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.
*/
#include <gmock/gmock.h>
#include <gtest/gtest.h>
#include "yr/api/err_type.h"
#include "yr/parallel/parallel_for.h"
#include "yr/yr.h"
#include "api/cpp/src/config_manager.h"
namespace YR {
namespace test {
using testing::HasSubstr;
const static size_t g_threadPoolSize = 8;
class LocalTest : public testing::Test {
public:
LocalTest() {}
~LocalTest() {}
void SetUp() override
{
YR::Config conf;
conf.mode = YR::Config::Mode::LOCAL_MODE;
conf.logLevel = "DEBUG";
conf.logDir = "/tmp/log";
conf.threadPoolSize = g_threadPoolSize;
YR::Init(conf);
}
void TearDown() override
{
YR::Finalize();
}
};
int PlusOne(int x)
{
return x + 1;
}
YR_INVOKE(PlusOne)
TEST_F(LocalTest, When_Invoke_Ten_Task_Should_All_Return_Correct_Result)
{
std::vector<YR::ObjectRef<int>> n2;
int k = 10;
for (int i = 0; i < k; i++) {
auto r2 = YR::Function(PlusOne).Invoke(i);
n2.emplace_back(r2);
}
for (int i = 0; i < k; i++) {
auto integer = *YR::Get(n2[i]);
EXPECT_EQ(i + 1, integer);
}
}
TEST_F(LocalTest, When_Invoke_Task_ObjID_Not_Contain_WorkerID)
{
std::vector<YR::ObjectRef<int>> n2;
auto r2 = YR::Function(PlusOne).Invoke(1);
auto id = r2.ID();
EXPECT_EQ(id.size(), 20);
auto integer = *YR::Get(r2);
EXPECT_EQ(2, integer);
}
static std::mutex mu;
static bool destructed = false;
static std::condition_variable cv;
class Counter {
public:
Counter() = default;
Counter(const Counter &c)
{
if (this != &c) {
*this = c;
}
}
Counter &operator=(const Counter &c)
{
count = c.count;
return *this;
}
~Counter()
{
std::unique_lock<std::mutex> lock(mu);
destructed = true;
cv.notify_one();
}
Counter(int init)
{
count = init;
}
static Counter *FactoryCreate(int init)
{
return new Counter(init);
}
int Add(int x)
{
count += x;
return count;
}
int ParallelFor()
{
auto f = [this](size_t start, size_t end, const YR::Parallel::Context &ctx) {
for (size_t i = start; i < end; i++) {
std::lock_guard<std::mutex> lock(ctxIdSetMutex);
ctxIdSet.emplace(ctx.id);
}
std::this_thread::yield();
std::this_thread::sleep_for(std::chrono::microseconds(10));
};
YR::Parallel::ParallelFor<size_t>(0, 1000, f, 1);
return 0;
}
size_t GetCtxIdsSize()
{
std::lock_guard<std::mutex> lock(ctxIdSetMutex);
return ctxIdSet.size();
}
int Get(void)
{
return count;
}
int Throw()
{
throw std::runtime_error("runtime error");
}
YR_STATE(count);
public:
std::mutex ctxIdSetMutex;
std::unordered_set<size_t> ctxIdSet;
int count;
};
class Foo {
public:
Foo() = default;
~Foo() = default;
Foo(const Foo &other)
{
clue = other.clue + ",CopyCons";
}
Foo &operator=(const Foo &other)
{
clue = other.clue + ",CopyAss";
return *this;
}
Foo(Foo &&other)
{
clue = std::move(other.clue);
clue += ",MoveCons";
}
Foo &operator=(Foo &&other)
{
clue = std::move(other.clue);
clue += ",MoveAss";
return *this;
}
std::string Clue(void) const
{
return clue;
}
std::string clue;
public:
MSGPACK_DEFINE(clue);
};
std::string TestFoo(const Foo &f)
{
return f.Clue();
}
YR_INVOKE(Counter::FactoryCreate, &Counter::Add, &Counter::Get, &Counter::Throw, &Counter::ParallelFor,
&Counter::GetCtxIdsSize, TestFoo);
TEST_F(LocalTest, cpp_invoke_classinstance_foo_localmode)
{
Foo foo;
auto r = YR::Function(TestFoo).Invoke(foo);
auto v = *YR::Get(r);
std::string::size_type n = 0;
int count = 0;
do {
n = v.find("Copy", n);
if (n != std::string::npos) {
count++;
n = n + std::string("Copy").length();
if (n >= v.length()) {
break;
}
}
} while (n != std::string::npos);
EXPECT_LE(count, 2);
}
TEST_F(LocalTest, When_Invoke_Actor_Should_Return_Final_Correct_Result)
{
auto counter = YR::Instance(Counter::FactoryCreate).Invoke(1);
auto res = counter.Function(&Counter::Add).Invoke(3);
auto v = *YR::Get(res);
EXPECT_EQ(4, v);
res = counter.Function(&Counter::Add).Invoke(3);
v = *YR::Get(res);
EXPECT_EQ(7, v);
res = counter.Function(&Counter::Get).Invoke();
v = *YR::Get(res);
EXPECT_EQ(7, v);
}
TEST_F(LocalTest, When_Invoke_Actor_Should_Not_Contain_WorkerId)
{
auto counter = YR::Instance(Counter::FactoryCreate).Invoke(1);
auto id = counter.GetObjectId();
EXPECT_EQ(id.size(), 20);
auto res = counter.Function(&Counter::Add).Invoke(3);
auto v = *YR::Get(res);
EXPECT_EQ(4, v);
}
TEST_F(LocalTest, When_Put_Should_Return_Correct_Result)
{
int val = 123;
auto ref = YR::Put<int>(val);
EXPECT_EQ(val, *YR::Get(ref));
int init = 10;
Counter c = Counter(10);
auto ref_ = YR::Put(c);
auto c_ = *YR::Get(ref_);
EXPECT_EQ(init, c.count);
}
TEST_F(LocalTest, When_Put_Should_Not_Contain_WorkerId)
{
int val = 123;
auto ref = YR::Put<int>(val);
EXPECT_EQ(ref.ID().size(), 20);
EXPECT_EQ(val, *YR::Get(ref));
}
TEST_F(LocalTest, When_Do_KV_Should_Return_Correct_Result)
{
std::string key = "kv-id-888";
std::string value = "kv-value-888";
YR::KV().Write(key, value);
std::shared_ptr<std::string> result = YR::KV().Read<std::string>(key);
EXPECT_EQ(value, *result);
YR::KV().Del(key);
YR::KV().Set(key, value);
std::string result2 = YR::KV().Get(key);
EXPECT_EQ(value, result2);
YR::KV().Del(key);
}
TEST_F(LocalTest, When_Get_Repeated_Keys_Should_Return_Success_Test)
{
std::string key = "key";
std::string value = "value";
KV().Set(key, value);
std::vector<std::string> keys = {key, key};
auto values = KV().Get(keys);
for (const auto &v : values) {
EXPECT_EQ(v, value);
}
}
TEST_F(LocalTest, Test_When_Actor_Currency_Call_ParallelFor_Should_Not_Be_Stuck)
{
auto counter = YR::Instance(Counter::FactoryCreate).Invoke(1);
std::vector<YR::ObjectRef<int>> rets;
for (int i = 0; i < 4; i++) {
rets.emplace_back(counter.Function(&Counter::ParallelFor).Invoke());
}
YR::Wait(rets, rets.size(), -1);
counter.Function(&Counter::ParallelFor).Invoke();
auto ret = *(YR::Get(counter.Function(&Counter::GetCtxIdsSize).Invoke()));
EXPECT_GE(ret, 1);
}
TEST_F(LocalTest, TestActorTerminate)
{
auto counter = YR::Instance(Counter::FactoryCreate).Invoke(1);
auto res = counter.Function(&Counter::Add).Invoke(3);
auto v = *YR::Get(res);
EXPECT_EQ(4, v);
counter.Terminate();
res = counter.Function(&Counter::Add).Invoke(3);
EXPECT_THROW(YR::Get(res, 1), Exception);
}
TEST_F(LocalTest, cpp_local_kv_read_error_keys_allowpatital_true)
{
std::string key;
std::string value;
std::vector<std::string> keys;
for (int i = 0; i < 10; ++i) {
key = "cpp_local_kv_read_error_keys_allowpatital_true" + std::to_string(i);
value = "value" + std::to_string(i);
try {
YR::KV().Write(key, value);
} catch (YR::Exception &e) {
std::cout << e.what() << std::endl;
EXPECT_EQ(0, 1);
}
keys.push_back(key);
}
keys.push_back("noValueKey1");
keys.push_back("noValueKey2");
keys.push_back("noValueKey3");
auto returnVal = YR::KV().Read<std::string>(keys, 1, true);
for (int i = 0; i < returnVal.size(); ++i) {
if (returnVal[i]) {
std::cout << i << "-> kv read value is: " << *returnVal[i] << std::endl;
}
}
EXPECT_EQ(returnVal.size(), 13) << "KV Read failed";
YR::KV().Del(keys);
}
TEST_F(LocalTest, cpp_local_kv_exist)
{
std::string key;
std::string value;
std::vector<std::string> keys;
for (int i = 0; i < 8; ++i) {
if (i % 2 == 0) {
key = "cpp_local_kv_exist" + std::to_string(i);
value = "value" + std::to_string(i);
YR::KV().Set(key, value);
keys.push_back(key);
} else {
keys.push_back("noValueKey" + std::to_string(i));
}
}
auto exists = YR::KV().Exist(keys);
for (int i = 0; i < 8; ++i) {
std::cout << keys[i] << "-> kv exist is: " << exists[i] << std::endl;
if (i % 2 == 0) {
EXPECT_EQ(exists[i], true);
} else {
EXPECT_EQ(exists[i], false);
}
}
}
int func_throw()
{
throw std::runtime_error("runtime error");
}
int func_throw_string()
{
throw std::string("something");
}
const static std::string customExceptionMsg = "a custom exception";
int func_throw_exception()
{
throw std::invalid_argument(customExceptionMsg);
return 1;
}
YR_INVOKE(func_throw, func_throw_string, func_throw_exception);
TEST_F(LocalTest, CatchException)
{
auto obj = Function(func_throw).Invoke();
EXPECT_THROW(
{
try {
Wait(obj);
} catch (const Exception &e) {
EXPECT_EQ(e.Code(), ErrorCode::ERR_USER_FUNCTION_EXCEPTION);
EXPECT_THAT(e.what(), HasSubstr("exception happens when executing user's function"));
throw;
}
},
Exception);
EXPECT_THROW(
{
try {
Get(obj);
} catch (const Exception &e) {
EXPECT_EQ(e.Code(), ErrorCode::ERR_USER_FUNCTION_EXCEPTION);
EXPECT_THAT(e.what(), HasSubstr("exception happens when executing user's function"));
throw;
}
},
Exception);
obj = Function(func_throw_string).Invoke();
EXPECT_THROW(
{
try {
Wait(obj);
} catch (const Exception &e) {
EXPECT_THAT(e.what(), HasSubstr("non-standard exception is thrown"));
throw;
}
},
Exception);
EXPECT_THROW(
{
try {
Get(obj);
} catch (const Exception &e) {
EXPECT_THAT(e.what(), HasSubstr("non-standard exception is thrown"));
throw;
}
},
Exception);
destructed = false;
auto instance = YR::Instance(Counter::FactoryCreate).Invoke(1);
auto obj2 = instance.Function(&Counter::Throw).Invoke();
EXPECT_THROW(
{
try {
Wait(obj2);
} catch (const Exception &e) {
EXPECT_EQ(e.Code(), ErrorCode::ERR_USER_FUNCTION_EXCEPTION);
EXPECT_THAT(e.what(), HasSubstr("exception happens when executing user's function"));
throw;
}
},
Exception);
instance.Terminate();
{
std::unique_lock<std::mutex> lock(mu);
cv.wait_for(lock, std::chrono::milliseconds(10), []{ return destructed; });
}
EXPECT_EQ(destructed, true);
instance.Function(&Counter::Throw).Invoke();
}
TEST_F(LocalTest, WaitConcurrencyCatchException)
{
std::vector<ObjectRef<int>> vec;
int num = 20;
int timeout = 1;
for (int i = 0; i < num; ++i) {
auto obj = Function(func_throw).Invoke();
vec.emplace_back(std::move(obj));
}
EXPECT_THROW(Wait(vec, num, timeout), std::exception);
EXPECT_THROW(Get(vec, timeout), std::exception);
}
TEST_F(LocalTest, ExceptionShouldShowDetailMsg)
{
try {
auto obj = Function(func_throw_exception).Invoke();
auto ret = *YR::Get(obj);
} catch (const std::exception &e) {
const std::string msg = e.what();
std::cout << msg << std::endl;
EXPECT_TRUE(msg.find(customExceptionMsg) != std::string::npos);
}
}
TEST_F(LocalTest, StopLocalModeRuntime)
{
YR::internal::LocalModeRuntime runtime;
runtime.Init();
ASSERT_NO_THROW(runtime.Stop());
}
TEST_F(LocalTest, LocalModeThreadPoolSize)
{
YR::Finalize();
YR::Config conf;
conf.mode = YR::Config::Mode::LOCAL_MODE;
conf.logLevel = "DEBUG";
conf.logDir = "/tmp/log";
conf.localThreadPoolSize = 65;
YR::Init(conf);
ASSERT_EQ(ConfigManager::Singleton().localThreadPoolSize, conf.localThreadPoolSize);
}
}
}