CloudEdge/CloudEdge
Archived
1
0
Fork 0
Code Issues 3 Pull Requests Packages Projects 1 Releases Wiki Activity

chore: adjust format ColumnLimit

Browse Source
This commit is contained in:
Paul Pan 2023-03-20 09:25:31 +08:00
parent ed759656ac
commit f78378ad43
33 changed files with 209 additions and 381 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.clang-format
View File

@ -8,6 +8,7 @@ AlignEscapedNewlines: Left
AllowShortCaseLabelsOnASingleLine: true
AllowShortIfStatementsOnASingleLine: true
AllowShortLoopsOnASingleLine: true
ColumnLimit: 120
IndentCaseLabels: true
IndentPPDirectives: BeforeHash
IndentWidth: 4

24
offload/Context.cpp
View File

@ -52,32 +52,20 @@ void Offload::Context::attach(Offload::Context *ctx) {
void Offload::Context::markGlobal() { Storage::globalCtx = Storage::context; }
void Offload::Context::set(std::shared_ptr<utils::Config> config) {
get()->config = std::move(config);
}
void Offload::Context::set(std::shared_ptr<utils::Config> config) { get()->config = std::move(config); }
void Offload::Context::set(std::shared_ptr<Remote> remote) {
get()->remote = std::move(remote);
}
void Offload::Context::set(std::shared_ptr<Remote> remote) { get()->remote = std::move(remote); }
void Offload::Context::set(std::shared_ptr<Scheduler> sch) {
get()->sch = std::move(sch);
}
void Offload::Context::set(std::shared_ptr<Scheduler> sch) { get()->sch = std::move(sch); }
void Offload::Context::set(std::shared_ptr<State::RTState> rtState) {
get()->rtState = std::move(rtState);
}
void Offload::Context::set(std::shared_ptr<State::RTState> rtState) { get()->rtState = std::move(rtState); }
void Offload::Context::set(std::shared_ptr<State::Store> store) {
get()->store = std::move(store);
}
void Offload::Context::set(std::shared_ptr<State::Store> store) { get()->store = std::move(store); }
void Offload::Context::set(RT::Runner *runner) { get()->runner = runner; }
void Offload::Context::set(RT::Tracer *tracer) { get()->tracer = tracer; }
void Offload::Context::set(State::TaskDesc desc) {
get()->taskDesc = std::move(desc);
}
void Offload::Context::set(State::TaskDesc desc) { get()->taskDesc = std::move(desc); }
void Offload::Context::set(State::From me) { get()->me = std::move(me); }

37
offload/Remote.cpp
View File

@ -12,24 +12,19 @@ using google::protobuf::Any;
using google::protobuf::Empty;
using google::protobuf::StringValue;
void Offload::Remote::RegisterStub(
const State::Host &host, const std::shared_ptr<grpc::Channel> &channel) {
void Offload::Remote::RegisterStub(const State::Host &host, const std::shared_ptr<grpc::Channel> &channel) {
stubs[host] = std::move(cloudedge::Remote::NewStub(channel));
}
#define CHECK_RPC_ERROR(status, ret) \
do { \
if (!status.ok()) { \
spdlog::critical("RPC call failed: {}: {}", \
(int)status.error_code(), \
status.error_message()); \
return ret; \
} \
#define CHECK_RPC_ERROR(status, ret) \
do { \
if (!status.ok()) { \
spdlog::critical("RPC call failed: {}: {}", (int)status.error_code(), status.error_message()); \
return ret; \
} \
} while (0)
bool Offload::Remote::SendProgram(const State::Host &host,
const std::string &name, const char *data,
uint32_t size) {
bool Offload::Remote::SendProgram(const State::Host &host, const std::string &name, const char *data, uint32_t size) {
TRACE_EVENT("rpc-call-client", "SendProgram", "program-name", name);
ClientContext context;
@ -56,8 +51,7 @@ bool Offload::Remote::SendProgram(const State::Host &host,
uint32_t chunk_num = size / chunk_size;
uint32_t last_chunk = size % chunk_size;
spdlog::trace("SendProgram: {} chunks, last_chunk={}", chunk_num,
last_chunk);
spdlog::trace("SendProgram: {} chunks, last_chunk={}", chunk_num, last_chunk);
for (uint32_t i = 0; i < chunk_num + 1; i++) {
spdlog::trace("SendProgram: prepare to send chunk {}", i);
@ -91,8 +85,7 @@ bool Offload::Remote::SendProgram(const State::Host &host,
return true;
}
std::optional<State::HostInfo>
Offload::Remote::GetHostInfo(const State::Host &host) {
std::optional<State::HostInfo> Offload::Remote::GetHostInfo(const State::Host &host) {
TRACE_EVENT("rpc-call-client", "GetHostInfo");
ClientContext context;
@ -120,8 +113,7 @@ Offload::Remote::GetHostInfo(const State::Host &host) {
info.cap.concurrency = ifo.cap().concurrency();
info.cap.bogomips = ifo.cap().bogomips();
for (const auto &t : ifo.trace()) {
auto task =
State::TaskUnit{t.task().program(), t.task().function_ptr()};
auto task = State::TaskUnit{t.task().program(), t.task().function_ptr()};
auto trace = State::TraceInfo{
.exec_time = t.trace().exec_time(),
.input_size = t.trace().input_size(),
@ -136,8 +128,8 @@ Offload::Remote::GetHostInfo(const State::Host &host) {
return std::move(info);
}
bool Offload::Remote::GetState(const State::Host &host, const State::From &from,
bool versionOnly, State::StateUnit &state) {
bool Offload::Remote::GetState(const State::Host &host, const State::From &from, bool versionOnly,
State::StateUnit &state) {
TRACE_EVENT("rpc-call-client", "GetState");
ClientContext context;
@ -176,8 +168,7 @@ bool Offload::Remote::GetState(const State::Host &host, const State::From &from,
return true;
}
bool Offload::Remote::SetState(const State::Host &host, const State::From &from,
State::StateUnit &state) {
bool Offload::Remote::SetState(const State::Host &host, const State::From &from, State::StateUnit &state) {
TRACE_EVENT("rpc-call-client", "SetState");
ClientContext context;

12
offload/Remote.h
View File

@ -14,17 +14,13 @@ class Remote {
public:
Remote() = default;
void RegisterStub(const State::Host &host,
const std::shared_ptr<grpc::Channel> &channel);
void RegisterStub(const State::Host &host, const std::shared_ptr<grpc::Channel> &channel);
bool SendProgram(const State::Host &host, const std::string &name,
const char *data, uint32_t size);
bool SendProgram(const State::Host &host, const std::string &name, const char *data, uint32_t size);
std::optional<State::HostInfo> GetHostInfo(const State::Host &host);
bool GetState(const State::Host &host, const State::From &from,
bool versionOnly, State::StateUnit &state);
bool SetState(const State::Host &host, const State::From &from,
State::StateUnit &state);
bool GetState(const State::Host &host, const State::From &from, bool versionOnly, State::StateUnit &state);
bool SetState(const State::Host &host, const State::From &from, State::StateUnit &state);
private:
std::map<State::Host, std::unique_ptr<cloudedge::Remote::Stub>> stubs;

9
offload/Scheduler.cpp
View File

@ -13,8 +13,7 @@ void Offload::Scheduler::Start() {
auto hosts = rtState->GetHosts();
for (auto &[name, host] : hosts) {
if (host == rtState->GetLocalhost()) continue;
auto channel = grpc::CreateChannel(host.endpoint,
grpc::InsecureChannelCredentials());
auto channel = grpc::CreateChannel(host.endpoint, grpc::InsecureChannelCredentials());
remote->RegisterStub(host, channel);
}
@ -28,8 +27,7 @@ void Offload::Scheduler::Start() {
if (host == rtState->GetLocalhost()) continue;
auto info = remote->GetHostInfo(host);
if (info.has_value()) {
spdlog::trace("host info update: {}@{}, free={}", host.name,
host.endpoint, info.value().free_runners);
spdlog::trace("host info update: {}@{}, free={}", host.name, host.endpoint, info.value().free_runners);
rtState->UpdateHostInfo(host, info.value());
}
}
@ -47,8 +45,7 @@ bool Offload::Scheduler::TaskFinished(uint32_t handler) {
auto task = rtState->GetTask(handler);
if (task == nullptr) return false;
spdlog::info("task finished: {}@{}", task->task.program,
task->task.function_ptr);
spdlog::info("task finished: {}@{}", task->task.program, task->task.function_ptr);
if (task->host == rtState->GetLocalhost()) {
// TODO: local

27
offload/Server.cpp
View File

@ -24,8 +24,7 @@ using google::protobuf::StringValue;
namespace Offload {
class RemoteImpl final : public cloudedge::Remote::Service {
Status SendProgram(ServerContext *ctx,
ServerReader<cloudedge::UploadFileRequest> *request,
Status SendProgram(ServerContext *ctx, ServerReader<cloudedge::UploadFileRequest> *request,
Response *response) override {
TRACE_EVENT("rpc-call-server", "SendProgram");
@ -43,8 +42,7 @@ class RemoteImpl final : public cloudedge::Remote::Service {
spdlog::info("recv dest_file: {}, {}", filename, filesize);
auto dest_filename = "./bin/" + filename;
std::ofstream dest_file(dest_filename,
std::ios::binary | std::ios::out);
std::ofstream dest_file(dest_filename, std::ios::binary | std::ios::out);
while (request->Read(&req))
if (req.has_chunk()) dest_file << req.chunk().data();
@ -68,8 +66,7 @@ class RemoteImpl final : public cloudedge::Remote::Service {
return Status::OK;
}
Status GetHostInfo(ServerContext *, const Empty *,
Response *resp) override {
Status GetHostInfo(ServerContext *, const Empty *, Response *resp) override {
TRACE_EVENT("rpc-call-server", "GetHostInfo");
auto *ctx = Offload::Context::getGlobal();
@ -105,22 +102,19 @@ class RemoteImpl final : public cloudedge::Remote::Service {
return Status::OK;
}
Status Invoke(ServerContext *, const cloudedge::RuntimeTaskUnit *,
Response *) override {
Status Invoke(ServerContext *, const cloudedge::RuntimeTaskUnit *, Response *) override {
// TODO
return Status::OK;
}
Status Pong(ServerContext *, const cloudedge::RuntimeTaskUnit *,
Response *) override {
Status Pong(ServerContext *, const cloudedge::RuntimeTaskUnit *, Response *) override {
// TODO
return Status::OK;
}
Status GetState(ServerContext *, const cloudedge::StateGetReq *req,
Response *resp) override {
Status GetState(ServerContext *, const cloudedge::StateGetReq *req, Response *resp) override {
TRACE_EVENT("rpc-call-server", "GetState");
auto *ctx = Offload::Context::getGlobal();
@ -144,16 +138,14 @@ class RemoteImpl final : public cloudedge::Remote::Service {
return Status::OK;
}
Status SetState(ServerContext *, const cloudedge::StateSetReq *req,
Response *resp) override {
Status SetState(ServerContext *, const cloudedge::StateSetReq *req, Response *resp) override {
TRACE_EVENT("rpc-call-server", "SetState");
auto *ctx = Offload::Context::getGlobal();
auto from = State::From(req->from());
auto key = req->state().key();
auto success = ctx->store->SetState(
from, key, req->state().data().data(), req->state().data().size());
auto success = ctx->store->SetState(from, key, req->state().data().data(), req->state().data().size());
if (!success) {
resp->set_code(StatusCode::ERROR);
resp->set_message("failed to set state");
@ -173,8 +165,7 @@ class RemoteImpl final : public cloudedge::Remote::Service {
return Status::OK;
}
Status ExchangeState(ServerContext *, const cloudedge::State *,
Response *) override {
Status ExchangeState(ServerContext *, const cloudedge::State *, Response *) override {
// TODO
return Status::OK;

40
rt/Runner.cpp
View File

@ -18,8 +18,7 @@ static bool WamrGlobalInitialized = false;
} // namespace RT
RT::Runner::Runner() {
Offload::Context::get()->rtState->RegisterRunner(
std::this_thread::get_id());
Offload::Context::get()->rtState->RegisterRunner(std::this_thread::get_id());
std::lock_guard<std::mutex> lock(WamrGlobalLock);
@ -69,8 +68,7 @@ void RT::Runner::LoadProgram(uint8_t *program, size_t programSize) {
throw std::runtime_error("Failed to load WAMR module");
}
moduleInst = wasm_runtime_instantiate(module, STACK_SIZE, HEAP_SIZE,
errorBuf, ERROR_BUF_SIZE);
moduleInst = wasm_runtime_instantiate(module, STACK_SIZE, HEAP_SIZE, errorBuf, ERROR_BUF_SIZE);
if (!moduleInst) {
spdlog::error("Failed to instantiate WAMR module: {}", errorBuf);
throw std::runtime_error("Failed to instantiate WAMR module");
@ -78,8 +76,7 @@ void RT::Runner::LoadProgram(uint8_t *program, size_t programSize) {
// TODO: wamr runtime std{in, out, err} redirect
// TODO: (distributed) virtual file systems
wasm_runtime_set_wasi_args_ex(module, nullptr, 0, nullptr, 0, nullptr, 0,
nullptr, 0, -1, -1, -1);
wasm_runtime_set_wasi_args_ex(module, nullptr, 0, nullptr, 0, nullptr, 0, nullptr, 0, -1, -1, -1);
}
void RT::Runner::Unload() {
@ -107,19 +104,14 @@ bool RT::Runner::ExecuteMain(unsigned &ret) {
// end trace
TRACE_EVENT_END("user-program");
auto endTime = std::chrono::high_resolution_clock::now();
ctx->tracer->Update(
TraceType::EXEC_TIME,
std::chrono::duration_cast<std::chrono::milliseconds>(endTime -
startTime)
.count());
ctx->rtState->MergeTrace(ctx->taskDesc->task,
ctx->tracer->GetTraceData());
ctx->tracer->Update(TraceType::EXEC_TIME,
std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime).count());
ctx->rtState->MergeTrace(ctx->taskDesc->task, ctx->tracer->GetTraceData());
delete ctx->tracer;
}
if (!status) {
spdlog::warn("Exception occurred in WASM Runtime: {}",
wasm_runtime_get_exception(moduleInst));
spdlog::warn("Exception occurred in WASM Runtime: {}", wasm_runtime_get_exception(moduleInst));
ret = -1;
return false;
}
@ -133,9 +125,8 @@ bool RT::Runner::ExecuteFunction(unsigned funcPtr, unsigned &ret) {
defer _(nullptr, [](...) { wasm_runtime_destroy_thread_env(); });
// use unique_ptr to auto destroy exec_env after execution
std::unique_ptr<WASMExecEnv, decltype(&wasm_runtime_destroy_exec_env)>
execEnv(wasm_runtime_create_exec_env(moduleInst, STACK_SIZE),
wasm_runtime_destroy_exec_env);
std::unique_ptr<WASMExecEnv, decltype(&wasm_runtime_destroy_exec_env)> execEnv(
wasm_runtime_create_exec_env(moduleInst, STACK_SIZE), wasm_runtime_destroy_exec_env);
if (!execEnv) {
const char *msg = "Failed to create exec env";
@ -162,19 +153,14 @@ bool RT::Runner::ExecuteFunction(unsigned funcPtr, unsigned &ret) {
// end trace
TRACE_EVENT_END("user-program");
auto endTime = std::chrono::high_resolution_clock::now();
ctx->tracer->Update(
TraceType::EXEC_TIME,
std::chrono::duration_cast<std::chrono::milliseconds>(endTime -
startTime)
.count());
ctx->rtState->MergeTrace(ctx->taskDesc->task,
ctx->tracer->GetTraceData());
ctx->tracer->Update(TraceType::EXEC_TIME,
std::chrono::duration_cast<std::chrono::milliseconds>(endTime - startTime).count());
ctx->rtState->MergeTrace(ctx->taskDesc->task, ctx->tracer->GetTraceData());
delete ctx->tracer;
}
if (!success) {
spdlog::trace("Exception occurred in WASM Runtime: {}",
wasm_runtime_get_exception(moduleInst));
spdlog::trace("Exception occurred in WASM Runtime: {}", wasm_runtime_get_exception(moduleInst));
return false;
}

4
rt/Tracer.cpp
View File

@ -6,9 +6,7 @@ void RT::Tracer::Update(RT::TraceType type, uint32_t val) {
case TraceType::INPUT_SIZE: data.input_size += val; break;
case TraceType::OUTPUT_SIZE: data.output_size += val; break;
case TraceType::STATE_SIZE: data.state_size += val; break;
case TraceType::FOREIGN_STATE_SIZE:
data.foreign_state_size += val;
break;
case TraceType::FOREIGN_STATE_SIZE: data.foreign_state_size += val; break;
case TraceType::SUBTASK_COUNT: data.subtask_count += val; break;
}
}

87
rt/interface/dl/dl.cpp
View File

@ -8,10 +8,8 @@
namespace RT {
static NativeSymbol symTable[] = {
{"dl_open", nullptr, "($i)i", nullptr},
{"dl_sym", nullptr, "(i$$)i", nullptr},
{"dl_call", nullptr, "(i*)I", nullptr},
{"dl_close_sym", nullptr, "(i)i", nullptr},
{"dl_open", nullptr, "($i)i", nullptr}, {"dl_sym", nullptr, "(i$$)i", nullptr},
{"dl_call", nullptr, "(i*)I", nullptr}, {"dl_close_sym", nullptr, "(i)i", nullptr},
{"dl_close", nullptr, "(i)i", nullptr},
};
}
@ -32,8 +30,7 @@ RT::DL::DL() : Interface("dynamic link") {
#undef ADD_PTR
wasm_runtime_register_natives("env", symTable,
sizeof(symTable) / sizeof(symTable[0]));
wasm_runtime_register_natives("env", symTable, sizeof(symTable) / sizeof(symTable[0]));
}
RT::DL::~DL() {
@ -65,14 +62,12 @@ RT::DL::~DL() {
return ret; \
} while (0)
#define GET_CONTEXT(ret) \
auto module_inst = get_module_inst(exec_env); \
auto *ctx = reinterpret_cast<RT::DL *>( \
wasm_runtime_get_function_attachment(exec_env)); \
#define GET_CONTEXT(ret) \
auto module_inst = get_module_inst(exec_env); \
auto *ctx = reinterpret_cast<RT::DL *>(wasm_runtime_get_function_attachment(exec_env)); \
if (ctx == nullptr) WAMR_THROW_CRIT(ret, "dl_get_context: nullptr");
int32_t RT::DL::dl_open(wasm_exec_env_t exec_env, const char *filename,
int32_t flags) {
int32_t RT::DL::dl_open(wasm_exec_env_t exec_env, const char *filename, int32_t flags) {
TRACE_EVENT("wasm-callback", "dl_open");
if (filename == nullptr) return -1;
@ -89,8 +84,7 @@ int32_t RT::DL::dl_open(wasm_exec_env_t exec_env, const char *filename,
ctx->handles.push_back(handle);
auto id = static_cast<int>(ctx->handles.size());
spdlog::trace("dl_open: lib = {}, flags = {}, id = {}, ptr = {}", filename,
flags, id, handle);
spdlog::trace("dl_open: lib = {}, flags = {}, id = {}, ptr = {}", filename, flags, id, handle);
return id;
}
@ -110,8 +104,7 @@ ffi_cif *create_ffi_cif(const char *signature) {
// signature ::= '(' type_arg* ')' type_ret?
// alloca temporary memory
auto **arg_types_tmp =
reinterpret_cast<ffi_type **>(alloca(sizeof(ffi_type *) * sig_len));
auto **arg_types_tmp = reinterpret_cast<ffi_type **>(alloca(sizeof(ffi_type *) * sig_len));
ffi_type *ret_type;
int arg_count = 0;
@ -160,8 +153,7 @@ ffi_cif *create_ffi_cif(const char *signature) {
for (int i = 0; i < arg_count; i++) arg_types[i] = arg_types_tmp[i];
ffi_status status =
ffi_prep_cif(cif, FFI_DEFAULT_ABI, arg_count, ret_type, arg_types);
ffi_status status = ffi_prep_cif(cif, FFI_DEFAULT_ABI, arg_count, ret_type, arg_types);
if (status != FFI_OK) {
const char *msg = "ffi_prep_cif failed: {}";
SENTRY_ERR(msg, static_cast<int>(status));
@ -174,8 +166,7 @@ ffi_cif *create_ffi_cif(const char *signature) {
#undef RET_ERR
}
int32_t RT::DL::dl_sym(wasm_exec_env_t exec_env, int32_t handle,
const char *symbol, const char *signature) {
int32_t RT::DL::dl_sym(wasm_exec_env_t exec_env, int32_t handle, const char *symbol, const char *signature) {
TRACE_EVENT("wasm-callback", "dl_sym");
if (symbol == nullptr || signature == nullptr) return -1;
@ -183,8 +174,7 @@ int32_t RT::DL::dl_sym(wasm_exec_env_t exec_env, int32_t handle,
GET_CONTEXT(-1)
if (handle < 1 || handle > ctx->handles.size()) {
spdlog::trace("dl_sym failed: invalid handle = {}, current size = {}",
handle, ctx->handles.size());
spdlog::trace("dl_sym failed: invalid handle = {}, current size = {}", handle, ctx->handles.size());
return -1;
}
@ -208,32 +198,27 @@ int32_t RT::DL::dl_sym(wasm_exec_env_t exec_env, int32_t handle,
return id;
}
int64_t RT::DL::dl_call(wasm_exec_env_t exec_env, int32_t symbol,
va_list_stk va_args) {
int64_t RT::DL::dl_call(wasm_exec_env_t exec_env, int32_t symbol, va_list_stk va_args) {
TRACE_EVENT("wasm-callback", "dl_call");
GET_CONTEXT(-1)
if (symbol < 1 || symbol > ctx->symbols.size()) {
spdlog::trace("dl_call: invalid symbol = {}, current size = {}", symbol,
ctx->symbols.size());
spdlog::trace("dl_call: invalid symbol = {}, current size = {}", symbol, ctx->symbols.size());
return -1;
}
if (ctx->symbols[symbol - 1].sym == nullptr) {
spdlog::trace("dl_call: try to access NULL symbol, symbol = {}",
symbol);
spdlog::trace("dl_call: try to access NULL symbol, symbol = {}", symbol);
return -1;
}
if (!wasm_runtime_validate_native_addr(module_inst, va_args,
sizeof(uint32_t))) {
if (!wasm_runtime_validate_native_addr(module_inst, va_args, sizeof(uint32_t))) {
WAMR_THROW(-1, "dl_call: va_args out of bounds");
}
uint8_t *native_end_addr;
if (!wasm_runtime_get_native_addr_range(module_inst, (uint8_t *)va_args,
nullptr, &native_end_addr)) {
if (!wasm_runtime_get_native_addr_range(module_inst, (uint8_t *)va_args, nullptr, &native_end_addr)) {
WAMR_THROW(-1, "dl_call: va_args out of bounds");
}
@ -244,9 +229,8 @@ int64_t RT::DL::dl_call(wasm_exec_env_t exec_env, int32_t symbol,
if (cif->arg_types[i]->type == FFI_TYPE_FLOAT) ext_float_count++;
}
#define ALIGN(n, b) (((uintptr_t)(n) + b) & (uintptr_t)~b)
#define GET_ARG(ptr, type) \
(*(type *)((ptr += ALIGN(sizeof(type), 3)) - ALIGN(sizeof(type), 3)))
#define ALIGN(n, b) (((uintptr_t)(n) + b) & (uintptr_t)~b)
#define GET_ARG(ptr, type) (*(type *)((ptr += ALIGN(sizeof(type), 3)) - ALIGN(sizeof(type), 3)))
#define CHECK_VA_ARG(ptr, type) \
do { \
if ((uint8_t *)ptr + ALIGN(sizeof(type), 3) > native_end_addr) { \
@ -254,12 +238,10 @@ int64_t RT::DL::dl_call(wasm_exec_env_t exec_env, int32_t symbol,
} \
} while (0)
auto **args = reinterpret_cast<void **>(alloca(sizeof(void *) * arg_count));
auto **ext_ptrs =
reinterpret_cast<void **>(alloca(sizeof(void *) * ext_ptr_count));
auto *ext_floats =
reinterpret_cast<float *>(alloca(sizeof(float *) * ext_float_count));
va_list_stk cur = va_args;
auto **args = reinterpret_cast<void **>(alloca(sizeof(void *) * arg_count));
auto **ext_ptrs = reinterpret_cast<void **>(alloca(sizeof(void *) * ext_ptr_count));
auto *ext_floats = reinterpret_cast<float *>(alloca(sizeof(float *) * ext_float_count));
va_list_stk cur = va_args;
for (int i = 0; i < arg_count; i++) {
switch (cif->arg_types[i]->type) {
case FFI_TYPE_SINT32: { // 32 bit
@ -359,21 +341,17 @@ int32_t RT::DL::dl_close_sym(wasm_exec_env_t exec_env, int32_t symbol) {
GET_CONTEXT(-1)
if (symbol < 1 || symbol > ctx->symbols.size()) {
spdlog::trace(
"dl_close_sym failed: invalid symbol = {}, current size = {}",
symbol, ctx->symbols.size());
spdlog::trace("dl_close_sym failed: invalid symbol = {}, current size = {}", symbol, ctx->symbols.size());
return -1;
}
void *sym = ctx->symbols[symbol - 1].sym;
if (sym == nullptr) {
spdlog::trace("dl_close_sym: try to access NULL symbol, symbol = {}",
symbol);
spdlog::trace("dl_close_sym: try to access NULL symbol, symbol = {}", symbol);
return -1;
}
if (ctx->symbols[symbol - 1].cif == nullptr)
WAMR_THROW_CRIT(-1, "inconsistency found! cif is NULL but sym is not");
if (ctx->symbols[symbol - 1].cif == nullptr) WAMR_THROW_CRIT(-1, "inconsistency found! cif is NULL but sym is not");
delete[] ctx->symbols[symbol - 1].cif->arg_types;
delete ctx->symbols[symbol - 1].cif;
@ -391,15 +369,13 @@ int32_t RT::DL::dl_close(wasm_exec_env_t exec_env, int32_t handle) {
GET_CONTEXT(-1)
if (handle < 1 || handle > ctx->handles.size()) {
spdlog::trace("dl_close failed: invalid handle = {}, current size = {}",
handle, ctx->handles.size());
spdlog::trace("dl_close failed: invalid handle = {}, current size = {}", handle, ctx->handles.size());
return -1;
}
auto *ptr = ctx->handles[handle - 1];
if (ptr == nullptr) {
spdlog::trace("dl_close: try to access NULL handle, handle = {}",
handle);
spdlog::trace("dl_close: try to access NULL handle, handle = {}", handle);
return -1;
}
@ -410,8 +386,7 @@ int32_t RT::DL::dl_close(wasm_exec_env_t exec_env, int32_t handle) {
sym.sym = nullptr;
if (sym.cif == nullptr) {
WAMR_THROW_CRIT(-1,
"inconsistency found! cif is NULL but sym is not");
WAMR_THROW_CRIT(-1, "inconsistency found! cif is NULL but sym is not");
}
delete[] sym.cif->arg_types;
@ -421,9 +396,7 @@ int32_t RT::DL::dl_close(wasm_exec_env_t exec_env, int32_t handle) {
auto ret = dlclose(ptr);
if (ret != 0) {
SENTRY_ERR(
"dl_close: dlclose failed, handle = {}, ptr = {}, error = {}",
handle, ptr, dlerror());
SENTRY_ERR("dl_close: dlclose failed, handle = {}, ptr = {}, error = {}", handle, ptr, dlerror());
return -1;
}

12
rt/interface/dl/dl.h
View File

@ -24,8 +24,7 @@ class DL : public Interface {
/// \param filename dynamic library file name
/// \param flags dlopen flags (man dlopen)
/// \return handle of the opened dynamic library, or RT_ERROR if failed
static int32_t dl_open(wasm_exec_env_t exec_env, const char *filename,
int32_t flags);
static int32_t dl_open(wasm_exec_env_t exec_env, const char *filename, int32_t flags);
/// Get symbol from a dynamic library
/// \param exec_env WAMR execution environment
@ -55,16 +54,14 @@ class DL : public Interface {
/// how the data length \n
/// -# More types will be supported in the future \n
/// \return native function's output, or terminated if failed
static int32_t dl_sym(wasm_exec_env_t exec_env, int32_t handle,
const char *symbol, const char *signature);
static int32_t dl_sym(wasm_exec_env_t exec_env, int32_t handle, const char *symbol, const char *signature);
/// Call a native function
/// \param exec_env WASM execution environment
/// \param symbol symbol handle, returned by dl_sym
/// \param va_args arguments, in WASM runtime, variadic arguments are pushed
/// into stack, just like a array \return function's return value
static int64_t dl_call(wasm_exec_env_t exec_env, int32_t symbol,
va_list_stk va_args);
static int64_t dl_call(wasm_exec_env_t exec_env, int32_t symbol, va_list_stk va_args);
/// Destroy a symbol, it will free the cif structure
/// \param exec_env WAMR execution environment
@ -81,8 +78,7 @@ class DL : public Interface {
private:
/* attachment */
struct symbol {
symbol(int backref, void *sym, ffi_cif *cif)
: backref(backref), sym(sym), cif(cif) {}
symbol(int backref, void *sym, ffi_cif *cif) : backref(backref), sym(sym), cif(cif) {}
/// back reference to the handle, 1-based
int backref;

6
rt/interface/lock/lock.cpp
View File

@ -14,14 +14,12 @@ RT::Lock::Lock() : Interface("lock") {
EXPORT(lock_release, "($)i"),
};
wasm_runtime_register_natives("env", symTable,
sizeof(symTable) / sizeof(symTable[0]));
wasm_runtime_register_natives("env", symTable, sizeof(symTable) / sizeof(symTable[0]));
}
RT::Lock::~Lock() = default;
bool RT::Lock::lock_acquire(wasm_exec_env_t exec_env, const char *key,
enum ce_lock_type type) {
bool RT::Lock::lock_acquire(wasm_exec_env_t exec_env, const char *key, enum ce_lock_type type) {
TRACE_EVENT("wasm-callback", "lock_acquire");
auto *ctx = Offload::Context::get();

3
rt/interface/lock/lock.h
View File

@ -18,8 +18,7 @@ class Lock : public Interface {
CE_LOCK_TYPE_SHARED = 1,
};
static bool lock_acquire(wasm_exec_env_t exec_env, const char *key,
enum ce_lock_type type);
static bool lock_acquire(wasm_exec_env_t exec_env, const char *key, enum ce_lock_type type);
static bool lock_release(wasm_exec_env_t exec_env, const char *key);
};

4
rt/interface/native.cpp
View File

@ -16,9 +16,7 @@ struct interface {
LIST_HEAD(rt_interfaces)
void register_interface(struct interface *iface) {
list_add(&iface->list, &rt_interfaces);
}
void register_interface(struct interface *iface) { list_add(&iface->list, &rt_interfaces); }
void DumpWamrInterfaces() {
struct list_head *current;

29
rt/interface/state/state.cpp
View File

@ -9,12 +9,10 @@
RT::State::State() : Interface("state") {
static NativeSymbol symTable[] = {
EXPORT(state_create, "($)i"), EXPORT(state_set, "($*~)i"),
EXPORT(state_get, "($*)i"), EXPORT(state_exchange, "($*~)i"),
EXPORT(state_delete, "($)i"), EXPORT(state_mark, "($)i"),
EXPORT(state_create, "($)i"), EXPORT(state_set, "($*~)i"), EXPORT(state_get, "($*)i"),
EXPORT(state_exchange, "($*~)i"), EXPORT(state_delete, "($)i"), EXPORT(state_mark, "($)i"),
};
wasm_runtime_register_natives("env", symTable,
sizeof(symTable) / sizeof(symTable[0]));
wasm_runtime_register_natives("env", symTable, sizeof(symTable) / sizeof(symTable[0]));
}
RT::State::~State() = default;
@ -26,8 +24,7 @@ bool RT::State::state_create(wasm_exec_env_t exec_env, const char *key) {
return ctx->store->CreateState(ctx->me, key);
}
unsigned RT::State::state_set(wasm_exec_env_t exec_env, const char *key,
const char *value, unsigned value_len) {
unsigned RT::State::state_set(wasm_exec_env_t exec_env, const char *key, const char *value, unsigned value_len) {
TRACE_EVENT("wasm-callback", "state_set");
auto *ctx = Offload::Context::get();
@ -35,24 +32,20 @@ unsigned RT::State::state_set(wasm_exec_env_t exec_env, const char *key,
if (!success) return 0;
ctx->tracer->Update(TraceType::STATE_SIZE, value_len);
if (ctx->store->GetOwner(key).value_or(::State::Owner{}).host !=
ctx->rtState->GetLocalhost())
if (ctx->store->GetOwner(key).value_or(::State::Owner{}).host != ctx->rtState->GetLocalhost())
ctx->tracer->Update(TraceType::FOREIGN_STATE_SIZE, value_len);
return 0;
}
unsigned RT::State::state_get(wasm_exec_env_t exec_env, const char *key,
unsigned *value_len) {
unsigned RT::State::state_get(wasm_exec_env_t exec_env, const char *key, unsigned *value_len) {
TRACE_EVENT("wasm-callback", "state_get");
auto module_inst = get_module_inst(exec_env);
auto *ctx = Offload::Context::get();
if (!wasm_runtime_validate_native_addr(module_inst, value_len,
sizeof(unsigned))) {
spdlog::warn("state_get: invalid value_len pointer {:x}",
reinterpret_cast<uintptr_t>(value_len));
if (!wasm_runtime_validate_native_addr(module_inst, value_len, sizeof(unsigned))) {
spdlog::warn("state_get: invalid value_len pointer {:x}", reinterpret_cast<uintptr_t>(value_len));
return 0;
}
@ -72,16 +65,14 @@ unsigned RT::State::state_get(wasm_exec_env_t exec_env, const char *key,
}
ctx->tracer->Update(TraceType::STATE_SIZE, len);
if (ctx->store->GetOwner(key).value_or(::State::Owner{}).host !=
ctx->rtState->GetLocalhost())
if (ctx->store->GetOwner(key).value_or(::State::Owner{}).host != ctx->rtState->GetLocalhost())
ctx->tracer->Update(TraceType::FOREIGN_STATE_SIZE, len);
*value_len = len;
return buf;
}
unsigned RT::State::state_exchange(wasm_exec_env_t exec_env, const char *key,
char *value, unsigned value_len) {
unsigned RT::State::state_exchange(wasm_exec_env_t exec_env, const char *key, char *value, unsigned value_len) {
TRACE_EVENT("wasm-callback", "state_exchange");
auto *ctx = Offload::Context::get();

9
rt/interface/state/state.h
View File

@ -15,12 +15,9 @@ class State : public Interface {
private:
static bool state_create(wasm_exec_env_t exec_env, const char *key);
static unsigned state_set(wasm_exec_env_t exec_env, const char *key,
const char *value, unsigned value_len);
static unsigned state_get(wasm_exec_env_t exec_env, const char *key,
unsigned *value_len);
static unsigned state_exchange(wasm_exec_env_t exec_env, const char *key,
char *value, unsigned value_len);
static unsigned state_set(wasm_exec_env_t exec_env, const char *key, const char *value, unsigned value_len);
static unsigned state_get(wasm_exec_env_t exec_env, const char *key, unsigned *value_len);
static unsigned state_exchange(wasm_exec_env_t exec_env, const char *key, char *value, unsigned value_len);
static bool state_delete(wasm_exec_env_t exec_env, const char *key);
static bool state_mark(wasm_exec_env_t exec_env, const char *key);
};

38
rt/interface/task/task.cpp
View File

@ -10,15 +10,10 @@
RT::Task::Task() : Interface("task") {
static NativeSymbol symTable[] = {
EXPORT(get_id, "()i"),
EXPORT(task_invoke, "(i*~)i"),
EXPORT(task_wait, "(i)i"),
EXPORT(task_get_input, "(*)i"),
EXPORT(task_set_output, "(*~)i"),
EXPORT(task_get_output, "(i*)i"),
EXPORT(get_id, "()i"), EXPORT(task_invoke, "(i*~)i"), EXPORT(task_wait, "(i)i"),
EXPORT(task_get_input, "(*)i"), EXPORT(task_set_output, "(*~)i"), EXPORT(task_get_output, "(i*)i"),
};
wasm_runtime_register_natives("env", symTable,
sizeof(symTable) / sizeof(symTable[0]));
wasm_runtime_register_natives("env", symTable, sizeof(symTable) / sizeof(symTable[0]));
}
RT::Task::~Task() = default;
@ -31,8 +26,7 @@ unsigned RT::Task::get_id(wasm_exec_env_t exec_env) {
return 0;
}
unsigned RT::Task::task_invoke(wasm_exec_env_t exec_env, unsigned func_ptr,
const char *args, unsigned args_len) {
unsigned RT::Task::task_invoke(wasm_exec_env_t exec_env, unsigned func_ptr, const char *args, unsigned args_len) {
TRACE_EVENT("wasm-callback", "task_invoke");
auto *ctx = Offload::Context::get();
@ -50,16 +44,13 @@ bool RT::Task::task_wait(wasm_exec_env_t exec_env, unsigned task_id) {
return false;
}
unsigned RT::Task::task_get_input(wasm_exec_env_t exec_env,
unsigned *args_len) {
unsigned RT::Task::task_get_input(wasm_exec_env_t exec_env, unsigned *args_len) {
TRACE_EVENT("wasm-callback", "task_get_input");
auto module_inst = get_module_inst(exec_env);
auto *ctx = Offload::Context::get();
if (!wasm_runtime_validate_native_addr(module_inst, args_len,
sizeof(unsigned))) {
spdlog::warn("task_get_input: invalid args_len address {:x}",
reinterpret_cast<uintptr_t>(args_len));
if (!wasm_runtime_validate_native_addr(module_inst, args_len, sizeof(unsigned))) {
spdlog::warn("task_get_input: invalid args_len address {:x}", reinterpret_cast<uintptr_t>(args_len));
return 0;
}
@ -69,10 +60,8 @@ unsigned RT::Task::task_get_input(wasm_exec_env_t exec_env,
return 0;
}
auto ptr = wasm_runtime_module_dup_data(
module_inst,
reinterpret_cast<const char *>(ctx->taskDesc->input_buffer),
ctx->taskDesc->input_buffer_size);
auto ptr = wasm_runtime_module_dup_data(module_inst, reinterpret_cast<const char *>(ctx->taskDesc->input_buffer),
ctx->taskDesc->input_buffer_size);
if (ptr == 0) {
spdlog::warn("task_get_input: failed to allocate memory");
@ -82,14 +71,12 @@ unsigned RT::Task::task_get_input(wasm_exec_env_t exec_env,
*args_len = ctx->taskDesc->input_buffer_size;
ctx->tracer->Update(TraceType::INPUT_SIZE,
ctx->taskDesc->input_buffer_size);
ctx->tracer->Update(TraceType::INPUT_SIZE, ctx->taskDesc->input_buffer_size);
return ptr;
}
unsigned RT::Task::task_set_output(wasm_exec_env_t exec_env, const char *args,
unsigned args_len) {
unsigned RT::Task::task_set_output(wasm_exec_env_t exec_env, const char *args, unsigned args_len) {
TRACE_EVENT("wasm-callback", "task_get_output");
auto *ctx = Offload::Context::get();
@ -109,8 +96,7 @@ unsigned RT::Task::task_set_output(wasm_exec_env_t exec_env, const char *args,
return args_len;
}
unsigned RT::Task::task_get_output(wasm_exec_env_t exec_env, unsigned task_id,
unsigned *args_len) {
unsigned RT::Task::task_get_output(wasm_exec_env_t exec_env, unsigned task_id, unsigned *args_len) {
TRACE_EVENT("wasm-callback", "task_get_output");
auto *ctx = Offload::Context::get();

12
rt/interface/task/task.h
View File

@ -14,15 +14,11 @@ class Task : public Interface {
private:
static unsigned get_id(wasm_exec_env_t exec_env);
static unsigned task_invoke(wasm_exec_env_t exec_env, unsigned func_ptr,
const char *args, unsigned args_len);
static unsigned task_invoke(wasm_exec_env_t exec_env, unsigned func_ptr, const char *args, unsigned args_len);
static bool task_wait(wasm_exec_env_t exec_env, unsigned task_id);
static unsigned task_get_input(wasm_exec_env_t exec_env,
unsigned *args_len);
static unsigned task_set_output(wasm_exec_env_t exec_env, const char *args,
unsigned args_len);
static unsigned task_get_output(wasm_exec_env_t exec_env, unsigned task_id,
unsigned *args_len);
static unsigned task_get_input(wasm_exec_env_t exec_env, unsigned *args_len);
static unsigned task_set_output(wasm_exec_env_t exec_env, const char *args, unsigned args_len);
static unsigned task_get_output(wasm_exec_env_t exec_env, unsigned task_id, unsigned *args_len);
};
} // namespace RT

22
state/RTState.cpp
View File

@ -14,9 +14,7 @@ State::RTState::RTState() {
localhost = Host{cfg->Alias, cfg->Endpoint};
hosts.emplace(localhost.name, localhost);
auto info = HostInfo{.free_runners = 0,
.cap = utils::GetHardwareCapabilities(),
.trace = {}};
auto info = HostInfo{.free_runners = 0, .cap = utils::GetHardwareCapabilities(), .trace = {}};
info.free_runners = info.cap.concurrency;
hostInfos.emplace(localhost.name, info);
@ -24,8 +22,7 @@ State::RTState::RTState() {
// TODO: create other hosts
for (auto &host : cfg->Hosts) {
hosts.emplace(host.name, host);
hostInfos.emplace(host.name,
HostInfo{.free_runners = 0, .cap = {}, .trace = {}});
hostInfos.emplace(host.name, HostInfo{.free_runners = 0, .cap = {}, .trace = {}});
}
spdlog::trace("RTState: initialized");
@ -49,8 +46,7 @@ State::TaskDesc State::RTState::GetTask(const std::thread::id &id) {
return tsk;
}
std::shared_ptr<char> State::RTState::GetProgram(const std::string &name,
uint32_t &size) {
std::shared_ptr<char> State::RTState::GetProgram(const std::string &name, uint32_t &size) {
TRACE_EVENT("rt-state", "GetProgram", "program-name", name);
wlock lk_program(program_mu);
@ -98,8 +94,7 @@ void State::RTState::MergeTrace(const TaskUnit &task, TraceInfo trace) {
}
}
uint32_t State::RTState::NewTask(std::shared_ptr<RuntimeTaskUnit> task,
const Host &remote) {
uint32_t State::RTState::NewTask(std::shared_ptr<RuntimeTaskUnit> task, const Host &remote) {
TRACE_EVENT("rt-state", "NewTask");
wlock lk_host(host_mu);
@ -145,15 +140,13 @@ void State::RTState::PrintTraces() {
spdlog::info("\t\tinput_size: {}", trace.input_size);
spdlog::info("\t\toutput_size: {}", trace.output_size);
spdlog::info("\t\tstate_size: {}", trace.state_size);
spdlog::info("\t\tforeign_state_size: {}",
trace.foreign_state_size);
spdlog::info("\t\tforeign_state_size: {}", trace.foreign_state_size);
spdlog::info("\t\tsubtask_count: {}", trace.subtask_count);
}
}
}
uint32_t State::RTState::NewTaskBlock(const std::thread::id &id,
std::shared_ptr<RuntimeTaskUnit> task) {
uint32_t State::RTState::NewTaskBlock(const std::thread::id &id, std::shared_ptr<RuntimeTaskUnit> task) {
spdlog::trace("RTState: NewTaskBlock is only used in shutdown");
runners[id].state = RunnerState::BUSY;
@ -175,8 +168,7 @@ State::HostInfo State::RTState::GetHostInfo() {
return hostInfos[localhost.name];
}
void State::RTState::UpdateHostInfo(const State::Host &host,
const State::HostInfo &info) {
void State::RTState::UpdateHostInfo(const State::Host &host, const State::HostInfo &info) {
wlock lk_host(host_mu);
hostInfos[host.name] = info;
}

17
state/RTState.h
View File

@ -21,8 +21,7 @@ class RTState {
/* Debugging */
void PrintTraces();
uint32_t NewTaskBlock(const std::thread::id &id,
std::shared_ptr<RuntimeTaskUnit> task);
uint32_t NewTaskBlock(const std::thread::id &id, std::shared_ptr<RuntimeTaskUnit> task);
/* Host Related */
Host GetLocalhost();
@ -62,16 +61,14 @@ class RTState {
uint32_t nextHandler;
/* Program Related */
std::shared_mutex program_mu;
std::map<std::string, std::pair<uint32_t, std::shared_ptr<char>>>
program; ///< 程序缓存
std::map<std::string, Host> owner; ///< 状态的拥有者
std::shared_mutex program_mu;
std::map<std::string, std::pair<uint32_t, std::shared_ptr<char>>> program; ///< 程序缓存
std::map<std::string, Host> owner; ///< 状态的拥有者
/* Runner Related */
std::shared_mutex runner_mu;
std::map<std::thread::id, RunnerInfo> runners; ///< 所有执行器的状态信息
std::map<std::thread::id, utils::chan<TaskDesc>>
pending; ///< 等待执行的任务队列
std::shared_mutex runner_mu;
std::map<std::thread::id, RunnerInfo> runners; ///< 所有执行器的状态信息
std::map<std::thread::id, utils::chan<TaskDesc>> pending; ///< 等待执行的任务队列
};
} // namespace State

33
state/Store.cpp
View File

@ -4,13 +4,10 @@
#include "offload/Context.h"
#include "utils/trace.h"
auto getRT = []() { return std::move(Offload::Context::getGlobal()->rtState); };
auto getRemote = []() {
return std::move(Offload::Context::getGlobal()->remote);
};
auto getRT = []() { return std::move(Offload::Context::getGlobal()->rtState); };
auto getRemote = []() { return std::move(Offload::Context::getGlobal()->remote); };
bool State::Store::CreateState(const State::From &from,
const std::string &key) {
bool State::Store::CreateState(const State::From &from, const std::string &key) {
TRACE_EVENT("rt-store", "CreateState");
std::lock_guard<std::mutex> lk(mu[key]);
@ -25,8 +22,7 @@ bool State::Store::CreateState(const State::From &from,
return true;
}
char *State::Store::GetState(const State::From &from, const std::string &key,
unsigned int &size) {
char *State::Store::GetState(const State::From &from, const std::string &key, unsigned int &size) {
TRACE_EVENT("rt-store", "GetState");
std::lock_guard<std::mutex> lk(mu[key]);
@ -41,14 +37,13 @@ char *State::Store::GetState(const State::From &from, const std::string &key,
auto parent = rt->GetTask(from.parent_handle);
if (parent->host == rt->GetLocalhost())
/* 请求者的父任务的上级任务在本地，状态不存在 */
if (parent->host == rt->GetLocalhost()) /* 请求者的父任务的上级任务在本地，状态不存在 */
goto not_exist;
/* 请求者的父任务的上级任务在远端，请求远端获取状态 */
State::StateUnit unit{.key = key};
auto success = remote->GetState(parent->host, from, false, unit);
auto success = remote->GetState(parent->host, from, false, unit);
if (!success) goto not_exist;
value[key] = std::make_shared<std::string>(unit.value);
@ -89,8 +84,7 @@ return_local:
return value[key]->data();
}
bool State::Store::SetState(const State::From &from, const std::string &key,
const char *val, unsigned int size) {
bool State::Store::SetState(const State::From &from, const std::string &key, const char *val, unsigned int size) {
TRACE_EVENT("rt-store", "SetState");
std::lock_guard<std::mutex> lk(mu[key]);
@ -107,8 +101,7 @@ bool State::Store::SetState(const State::From &from, const std::string &key,
if (from.parent_handle == 0) /* 到顶了，状态不存在 */
return false;
if (parent->host == rt->GetLocalhost())
/* 请求者的父任务的上级任务在本地，状态不存在 */
if (parent->host == rt->GetLocalhost()) /* 请求者的父任务的上级任务在本地，状态不存在 */
return false;
/* 请求者的父任务的上级任务在远端，请求转发给远端 */
@ -156,25 +149,21 @@ request_parent:
return true;
}
bool State::Store::ExchangeState(const State::From &from,
const std::string &key, char *val,
unsigned int *size) {
bool State::Store::ExchangeState(const State::From &from, const std::string &key, char *val, unsigned int *size) {
TRACE_EVENT("rt-store", "ExchangeState");
std::lock_guard<std::mutex> lk(mu[key]);
// TODO
return false;
}
bool State::Store::LockState(const State::From &from, const std::string &key,
State::LockType type) {
bool State::Store::LockState(const State::From &from, const std::string &key, State::LockType type) {
TRACE_EVENT("rt-store", "LockState");
std::lock_guard<std::mutex> lk(mu[key]);
// TODO
return false;
}
bool State::Store::UnlockState(const State::From &from,
const std::string &key) {
bool State::Store::UnlockState(const State::From &from, const std::string &key) {
TRACE_EVENT("rt-store", "UnlockState");
std::lock_guard<std::mutex> lk(mu[key]);
// TODO

6
state/Store.h
View File

@ -13,10 +13,8 @@ class Store {
public:
bool CreateState(const From &from, const std::string &key);
char *GetState(const From &from, const std::string &key, unsigned &size);
bool SetState(const From &from, const std::string &key, const char *val,
unsigned size);
bool ExchangeState(const From &from, const std::string &key, char *val,
unsigned *size);
bool SetState(const From &from, const std::string &key, const char *val, unsigned size);
bool ExchangeState(const From &from, const std::string &key, char *val, unsigned *size);
bool LockState(const From &from, const std::string &key, LockType type);
bool UnlockState(const From &from, const std::string &key);

28
state/Type.h
View File

@ -35,13 +35,10 @@ struct TaskUnit {
unsigned function_ptr; ///< 要执行的函数的指针
bool operator<(const TaskUnit &rhs) const {
return (program < rhs.program) ||
(program == rhs.program && function_ptr < rhs.function_ptr);
return (program < rhs.program) || (program == rhs.program && function_ptr < rhs.function_ptr);
};
bool operator==(const TaskUnit &rhs) const {
return program == rhs.program && function_ptr == rhs.function_ptr;
};
bool operator==(const TaskUnit &rhs) const { return program == rhs.program && function_ptr == rhs.function_ptr; };
};
using TaskUnit = struct TaskUnit;
@ -59,8 +56,7 @@ struct Host {
std::string endpoint; ///< RPC 通信端点
Host() = default;
Host(std::string name, std::string endpoint)
: name(std::move(name)), endpoint(std::move(endpoint)) {}
Host(std::string name, std::string endpoint) : name(std::move(name)), endpoint(std::move(endpoint)) {}
explicit Host(const cloudedge::Host &host) {
name = host.name();
endpoint = host.endpoint();
@ -73,13 +69,9 @@ struct Host {
return host;
};
bool operator<(const Host &rhs) const {
return name < rhs.name || (name == rhs.name && endpoint < rhs.endpoint);
};
bool operator<(const Host &rhs) const { return name < rhs.name || (name == rhs.name && endpoint < rhs.endpoint); };
bool operator==(const Host &rhs) const {
return name == rhs.name && endpoint == rhs.endpoint;
};
bool operator==(const Host &rhs) const { return name == rhs.name && endpoint == rhs.endpoint; };
bool operator!=(const Host &rhs) const { return !(*this == rhs); };
};
@ -139,8 +131,7 @@ struct From {
From() = default;
From(Host host, unsigned parent_handle, unsigned local_handle)
: host(std::move(host)), parent_handle(parent_handle),
local_handle(local_handle) {}
: host(std::move(host)), parent_handle(parent_handle), local_handle(local_handle) {}
explicit From(const cloudedge::From &req) {
host = Host(req.host());
@ -157,8 +148,7 @@ struct From {
}
bool operator==(const From &rhs) const {
return host == rhs.host && parent_handle == rhs.parent_handle &&
local_handle == rhs.local_handle;
return host == rhs.host && parent_handle == rhs.parent_handle && local_handle == rhs.local_handle;
};
};
using From = struct From;
@ -188,9 +178,7 @@ struct Owner {
return owner;
}
bool operator==(const From &rhs) const {
return handle == rhs.local_handle && host == rhs.host;
};
bool operator==(const From &rhs) const { return handle == rhs.local_handle && host == rhs.host; };
bool operator!=(const From &rhs) const { return !(*this == rhs); }
};

11
tool.cpp
View File

@ -20,8 +20,7 @@ void play() {
auto target = hosts["machine-1"];
uint32_t programSize;
auto program = ctx->rtState->GetProgram("test-2", programSize);
auto success =
ctx->remote->SendProgram(target, "test-1", program.get(), programSize);
auto success = ctx->remote->SendProgram(target, "test-1", program.get(), programSize);
program.reset();
spdlog::info("send program: {}", success);
}
@ -57,14 +56,12 @@ int main(int argc, char *argv[]) {
args.backends |= perfetto::kInProcessBackend;
perfetto::Tracing::Initialize(args);
perfetto::TrackEvent::Register();
std::unique_ptr<perfetto::TracingSession> tracing_session(
perfetto::Tracing::NewTrace());
perfetto::TraceConfig cfg;
std::unique_ptr<perfetto::TracingSession> tracing_session(perfetto::Tracing::NewTrace());
perfetto::TraceConfig cfg;
cfg.add_buffers()->set_size_kb(10240);
auto *ds_cfg = cfg.add_data_sources()->mutable_config();
ds_cfg->set_name("track_event");
auto trace_fd =
open("perfetto-trace-tool", O_RDWR | O_CREAT | O_TRUNC, 0600);
auto trace_fd = open("perfetto-trace-tool", O_RDWR | O_CREAT | O_TRUNC, 0600);
tracing_session->Setup(cfg, trace_fd);
tracing_session->StartBlocking();
#endif

20
utils.cmake
View File

@ -1,33 +1,33 @@
if(NOT CMAKE_PROPERTY_LIST)
if (NOT CMAKE_PROPERTY_LIST)
execute_process(COMMAND cmake --help-property-list OUTPUT_VARIABLE CMAKE_PROPERTY_LIST)
# Convert command output into a CMake list
string(REGEX REPLACE ";" "\\\\;" CMAKE_PROPERTY_LIST "${CMAKE_PROPERTY_LIST}")
string(REGEX REPLACE "\n" ";" CMAKE_PROPERTY_LIST "${CMAKE_PROPERTY_LIST}")
endif()
endif ()
function(print_properties)
message("CMAKE_PROPERTY_LIST = ${CMAKE_PROPERTY_LIST}")
endfunction()
function(print_target_properties target)
if(NOT TARGET ${target})
if (NOT TARGET ${target})
message(STATUS "There is no target named '${target}'")
return()
endif()
endif ()
foreach(property ${CMAKE_PROPERTY_LIST})
foreach (property ${CMAKE_PROPERTY_LIST})
string(REPLACE "<CONFIG>" "${CMAKE_BUILD_TYPE}" property ${property})
# Fix https://stackoverflow.com/questions/32197663/how-can-i-remove-the-the-location-property-may-not-be-read-from-target-error-i
if(property STREQUAL "LOCATION" OR property MATCHES "^LOCATION_" OR property MATCHES "_LOCATION$")
if (property STREQUAL "LOCATION" OR property MATCHES "^LOCATION_" OR property MATCHES "_LOCATION$")
continue()
endif()
endif ()
get_property(was_set TARGET ${target} PROPERTY ${property} SET)
if(was_set)
if (was_set)
get_target_property(value ${target} ${property})
message("${target} ${property} = ${value}")
endif()
endforeach()
endif ()
endforeach ()
endfunction()

6
utils/cap.cpp
View File

@ -3,9 +3,9 @@
#include "cap.h"
State::HardwareInfo utils::GetHardwareCapabilities() {
auto concurrency = static_cast<int>(std::thread::hardware_concurrency());
auto bogomips = 0;
int cpuCount = 0;
auto concurrency = static_cast<int>(std::thread::hardware_concurrency());
auto bogomips = 0;
int cpuCount = 0;
std::ifstream cpuInfo("/proc/cpuinfo");
for (std::string line; std::getline(cpuInfo, line);) {
if (line.find("bogomips") == std::string::npos) continue;

5
utils/config.cpp
View File

@ -12,6 +12,7 @@ utils::Config::Config(const std::string &path) {
json data = json::parse(cfg);
this->Alias = data["Alias"];
this->Endpoint = data["Endpoint"];
for (auto &host : data["Hosts"])
this->Hosts.emplace_back(State::Host{host["name"], host["endpoint"]});
for (auto &host : data["Hosts"]) {
this->Hosts.emplace_back(host["name"], host["endpoint"]);
}
}

5
utils/list.h
View File

@ -12,8 +12,7 @@ static inline void list_add(struct list_head *l, struct list_head *head) {
head->next = l;
}
#define list_for_each(pos, head) for (pos = (head)->next; pos != (head); pos = pos->next)
#define list_entry(ptr, type, member) \
((type *)((char *)(ptr) - (unsigned long)(&((type *)0)->member)))
#define list_for_each(pos, head) for (pos = (head)->next; pos != (head); pos = pos->next)
#define list_entry(ptr, type, member) ((type *)((char *)(ptr) - (unsigned long)(&((type *)0)->member)))
#endif // CLOUDEDGE_LIST_H

7
utils/log.h
View File

@ -11,10 +11,9 @@
#define _COLOR_CYAN "\x1B[1;36m"
#define _COLOR_RESET "\x1B[0m"
#define _LOG(color, level, fmt, ...) \
do { \
fprintf(stdout, color "[" level "] (%s:%d): " fmt _COLOR_RESET "\n", __FILE__, __LINE__, \
##__VA_ARGS__); \
#define _LOG(color, level, fmt, ...) \
do { \
fprintf(stdout, color "[" level "] (%s:%d): " fmt _COLOR_RESET "\n", __FILE__, __LINE__, ##__VA_ARGS__); \
} while (0)
#define LOG_ERR(fmt, ...) _LOG(_COLOR_RED, "E", fmt, ##__VA_ARGS__)

3
utils/rand.cpp
View File

@ -8,7 +8,6 @@ std::string utils::rand_str(size_t len, const std::string &alphabet) {
std::uniform_int_distribution<size_t> dist{0, alphabet.length() - 1};
std::string ret;
std::generate_n(std::back_inserter(ret), len,
[&] { return alphabet[dist(gen)]; });
std::generate_n(std::back_inserter(ret), len, [&] { return alphabet[dist(gen)]; });
return ret;
}

6
utils/rand.h
View File

@ -5,11 +5,9 @@
namespace utils {
const std::string alnum =
"abcdefghijklmnaoqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890";
const std::string alnum = "abcdefghijklmnaoqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ1234567890";
std::string rand_str(size_t len = 16,
const std::string &alphabet = utils::alnum);
std::string rand_str(size_t len = 16, const std::string &alphabet = utils::alnum);
} // namespace utils

43
utils/sentry.h
View File

@ -15,29 +15,26 @@
} while (0)
#else
#define SENTRY_ERR(fmt, ...) \
do { \
spdlog::critical(fmt, ##__VA_ARGS__); \
\
sentry_value_t event = sentry_value_new_event(); \
\
sentry_value_t exception = sentry_value_new_exception( \
SENTRY_EXCEPTION_TAG, \
spdlog::fmt_lib::format(fmt, ##__VA_ARGS__).c_str()); \
sentry_value_set_stacktrace(exception, NULL, 0); \
sentry_event_add_exception(event, exception); \
\
sentry_value_t source = sentry_value_new_object(); \
sentry_value_set_by_key(source, "file", \
sentry_value_new_string(__FILE__)); \
sentry_value_set_by_key(source, "line", \
sentry_value_new_int32(__LINE__)); \
\
sentry_value_t contexts = sentry_value_new_object(); \
sentry_value_set_by_key(contexts, "source info", source); \
sentry_value_set_by_key(event, "contexts", contexts); \
\
sentry_capture_event(event); \
#define SENTRY_ERR(fmt, ...) \
do { \
spdlog::critical(fmt, ##__VA_ARGS__); \
\
sentry_value_t event = sentry_value_new_event(); \
\
sentry_value_t exception = \
sentry_value_new_exception(SENTRY_EXCEPTION_TAG, spdlog::fmt_lib::format(fmt, ##__VA_ARGS__).c_str()); \
sentry_value_set_stacktrace(exception, NULL, 0); \
sentry_event_add_exception(event, exception); \
\
sentry_value_t source = sentry_value_new_object(); \
sentry_value_set_by_key(source, "file", sentry_value_new_string(__FILE__)); \
sentry_value_set_by_key(source, "line", sentry_value_new_int32(__LINE__)); \
\
sentry_value_t contexts = sentry_value_new_object(); \
sentry_value_set_by_key(contexts, "source info", source); \
sentry_value_set_by_key(event, "contexts", contexts); \
\
sentry_capture_event(event); \
} while (0)
#endif

19
utils/trace.h
View File

@ -12,19 +12,12 @@
#else
#include <perfetto.h>
PERFETTO_DEFINE_CATEGORIES(
perfetto::Category("user-program")
.SetDescription("user program executing statistics"),
perfetto::Category("wasm-callback")
.SetDescription("Injected wasm functions statistics"),
perfetto::Category("rt-state")
.SetDescription("runtime state r/w statistics"),
perfetto::Category("rt-store")
.SetDescription("runtime state-store statistics"),
perfetto::Category("rpc-call-client")
.SetDescription("client gRPC call runtime statistics"),
perfetto::Category("rpc-call-server")
.SetDescription("server gRPC call runtime statistics"));
PERFETTO_DEFINE_CATEGORIES(perfetto::Category("user-program").SetDescription("user program executing statistics"),
perfetto::Category("wasm-callback").SetDescription("Injected wasm functions statistics"),
perfetto::Category("rt-state").SetDescription("runtime state r/w statistics"),
perfetto::Category("rt-store").SetDescription("runtime state-store statistics"),
perfetto::Category("rpc-call-client").SetDescription("client gRPC call runtime statistics"),
perfetto::Category("rpc-call-server").SetDescription("server gRPC call runtime statistics"));
#endif

5
version.h
View File

@ -1,10 +1,9 @@
#ifndef CLOUDEDGE_VERSION_H
#define CLOUDEDGE_VERSION_H
extern "C"
{
extern "C" {
extern const char *VERSION_STRING;
extern const char *SENTRY_DSN_STRING;
}
#endif //CLOUDEDGE_VERSION_H
#endif // CLOUDEDGE_VERSION_H