This is an automated email from the ASF dual-hosted git repository. tqchen pushed a commit to branch rpc-refactor in repository https://gitbox.apache.org/repos/asf/tvm.git
commit 071e822a556f71b1720713f7b468a6355a540170 Author: tqchen <[email protected]> AuthorDate: Thu May 8 10:32:11 2025 -0400 [RPC] Migrate minrpc to use event runner This PR migrates minrpc to use event runner to simplify the dependencies and avoid duplicate impl. --- src/runtime/minrpc/minrpc_interfaces.h | 93 --- src/runtime/minrpc/minrpc_logger.cc | 291 --------- src/runtime/minrpc/minrpc_logger.h | 296 --------- src/runtime/minrpc/minrpc_server.h | 718 ++------------------- src/runtime/minrpc/minrpc_server_logging.h | 170 ----- .../posix_popen_server/posix_popen_server.cc | 3 - src/runtime/rpc/rpc_channel_logger.h | 186 ------ src/runtime/rpc/rpc_endpoint.cc | 4 + src/runtime/rpc/rpc_endpoint.h | 1 - src/runtime/rpc/rpc_socket_impl.cc | 3 - tests/python/runtime/test_runtime_rpc.py | 19 - 11 files changed, 44 insertions(+), 1740 deletions(-) diff --git a/src/runtime/minrpc/minrpc_interfaces.h b/src/runtime/minrpc/minrpc_interfaces.h deleted file mode 100644 index a45dee9f2c..0000000000 --- a/src/runtime/minrpc/minrpc_interfaces.h +++ /dev/null @@ -1,93 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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. - */ - -#ifndef TVM_RUNTIME_MINRPC_MINRPC_INTERFACES_H_ -#define TVM_RUNTIME_MINRPC_MINRPC_INTERFACES_H_ - -#include <tvm/runtime/c_runtime_api.h> - -#include "rpc_reference.h" - -namespace tvm { -namespace runtime { - -/*! - * \brief Return interface used in ExecInterface to generate and send the responses. - */ -class MinRPCReturnInterface { - public: - virtual ~MinRPCReturnInterface() {} - /*! * \brief sends a response to the client with kTVMNullptr in payload. */ - virtual void ReturnVoid() = 0; - - /*! * \brief sends a response to the client with one kTVMOpaqueHandle in payload. */ - virtual void ReturnHandle(void* handle) = 0; - - /*! * \brief sends an exception response to the client with a kTVMStr in payload. */ - virtual void ReturnException(const char* msg) = 0; - - /*! * \brief sends a packed argument sequnce to the client. */ - virtual void ReturnPackedSeq(const TVMValue* arg_values, const int* type_codes, int num_args) = 0; - - /*! * \brief sends a copy of the requested remote data to the client. */ - virtual void ReturnCopyFromRemote(uint8_t* data_ptr, uint64_t num_bytes) = 0; - - /*! * \brief sends an exception response to the client with the last TVM erros as the message. */ - virtual void ReturnLastTVMError() = 0; - - /*! * \brief internal error. */ - virtual void ThrowError(RPCServerStatus code, RPCCode info = RPCCode::kNone) = 0; -}; - -/*! - * \brief Execute interface used in MinRPCServer to process different received commands - */ -class MinRPCExecInterface { - public: - virtual ~MinRPCExecInterface() {} - - /*! * \brief Execute an Initilize server command. */ - virtual void InitServer(int num_args) = 0; - - /*! * \brief calls a function specified by the call_handle. */ - virtual void NormalCallFunc(uint64_t call_handle, TVMValue* values, int* tcodes, - int num_args) = 0; - - /*! * \brief Execute a copy from remote command by sending the data described in arr to the client - */ - virtual void CopyFromRemote(DLTensor* arr, uint64_t num_bytes, uint8_t* data_ptr) = 0; - - /*! * \brief Execute a copy to remote command by receiving the data described in arr from the - * client */ - virtual int CopyToRemote(DLTensor* arr, uint64_t num_bytes, uint8_t* data_ptr) = 0; - - /*! * \brief calls a system function specified by the code. */ - virtual void SysCallFunc(RPCCode code, TVMValue* values, int* tcodes, int num_args) = 0; - - /*! * \brief internal error. */ - virtual void ThrowError(RPCServerStatus code, RPCCode info = RPCCode::kNone) = 0; - - /*! * \brief return the ReturnInterface pointer that is used to generate and send the responses. - */ - virtual MinRPCReturnInterface* GetReturnInterface() = 0; -}; - -} // namespace runtime -} // namespace tvm -#endif // TVM_RUNTIME_MINRPC_MINRPC_INTERFACES_H_ diff --git a/src/runtime/minrpc/minrpc_logger.cc b/src/runtime/minrpc/minrpc_logger.cc deleted file mode 100644 index 4f3b7e764c..0000000000 --- a/src/runtime/minrpc/minrpc_logger.cc +++ /dev/null @@ -1,291 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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 "minrpc_logger.h" - -#include <string.h> -#include <time.h> -#include <tvm/runtime/c_runtime_api.h> -#include <tvm/runtime/logging.h> - -#include <functional> -#include <iostream> -#include <sstream> -#include <unordered_map> - -#include "minrpc_interfaces.h" -#include "rpc_reference.h" - -namespace tvm { -namespace runtime { - -void Logger::LogTVMValue(int tcode, TVMValue value) { - switch (tcode) { - case kDLInt: { - LogValue<int64_t>("(int64)", value.v_int64); - break; - } - case kDLUInt: { - LogValue<uint64_t>("(uint64)", value.v_int64); - break; - } - case kDLFloat: { - LogValue<float>("(float)", value.v_float64); - break; - } - case kTVMDataType: { - LogDLData("DLDataType(code,bits,lane)", &value.v_type); - break; - } - case kDLDevice: { - LogDLDevice("DLDevice(type,id)", &value.v_device); - break; - } - case kTVMPackedFuncHandle: { - LogValue<void*>("(PackedFuncHandle)", value.v_handle); - break; - } - case kTVMModuleHandle: { - LogValue<void*>("(ModuleHandle)", value.v_handle); - break; - } - case kTVMOpaqueHandle: { - LogValue<void*>("(OpaqueHandle)", value.v_handle); - break; - } - case kTVMDLTensorHandle: { - LogValue<void*>("(TensorHandle)", value.v_handle); - break; - } - case kTVMNDArrayHandle: { - LogValue<void*>("kTVMNDArrayHandle", value.v_handle); - break; - } - case kTVMNullptr: { - Log("Nullptr"); - break; - } - case kTVMStr: { - Log("\""); - Log(value.v_str); - Log("\""); - break; - } - case kTVMBytes: { - TVMByteArray* bytes = static_cast<TVMByteArray*>(value.v_handle); - int len = bytes->size; - LogValue<int64_t>("(Bytes) [size]: ", len); - if (PRINT_BYTES) { - Log(", [Values]:"); - Log(" { "); - if (len > 0) { - LogValue<uint64_t>("", (uint8_t)bytes->data[0]); - } - for (int j = 1; j < len; j++) LogValue<uint64_t>(" - ", (uint8_t)bytes->data[j]); - Log(" } "); - } - break; - } - default: { - Log("ERROR-kUnknownTypeCode)"); - break; - } - } - Log("; "); -} - -void Logger::OutputLog() { - LOG(INFO) << os_.str(); - os_.str(std::string()); -} - -void MinRPCReturnsWithLog::ReturnVoid() { - next_->ReturnVoid(); - logger_->Log("-> ReturnVoid"); - logger_->OutputLog(); -} - -void MinRPCReturnsWithLog::ReturnHandle(void* handle) { - next_->ReturnHandle(handle); - if (code_ == RPCCode::kGetGlobalFunc) { - RegisterHandleName(handle); - } - logger_->LogValue<void*>("-> ReturnHandle: ", handle); - logger_->OutputLog(); -} - -void MinRPCReturnsWithLog::ReturnException(const char* msg) { - next_->ReturnException(msg); - logger_->Log("-> Exception: "); - logger_->Log(msg); - logger_->OutputLog(); -} - -void MinRPCReturnsWithLog::ReturnPackedSeq(const TVMValue* arg_values, const int* type_codes, - int num_args) { - next_->ReturnPackedSeq(arg_values, type_codes, num_args); - ProcessValues(arg_values, type_codes, num_args); - logger_->OutputLog(); -} - -void MinRPCReturnsWithLog::ReturnCopyFromRemote(uint8_t* data_ptr, uint64_t num_bytes) { - next_->ReturnCopyFromRemote(data_ptr, num_bytes); - logger_->LogValue<uint64_t>("-> CopyFromRemote: ", num_bytes); - logger_->LogValue<void*>(", ", static_cast<void*>(data_ptr)); - logger_->OutputLog(); -} - -void MinRPCReturnsWithLog::ReturnLastTVMError() { - const char* err = TVMGetLastError(); - ReturnException(err); -} - -void MinRPCReturnsWithLog::ThrowError(RPCServerStatus code, RPCCode info) { - next_->ThrowError(code, info); - logger_->Log("-> ERROR: "); - logger_->Log(RPCServerStatusToString(code)); - logger_->OutputLog(); -} - -void MinRPCReturnsWithLog::ProcessValues(const TVMValue* values, const int* tcodes, int num_args) { - if (tcodes != nullptr) { - logger_->Log("-> ["); - for (int i = 0; i < num_args; ++i) { - logger_->LogTVMValue(tcodes[i], values[i]); - - if (tcodes[i] == kTVMOpaqueHandle) { - RegisterHandleName(values[i].v_handle); - } - } - logger_->Log("]"); - } -} - -void MinRPCReturnsWithLog::ResetHandleName(RPCCode code) { - code_ = code; - handle_name_.clear(); -} - -void MinRPCReturnsWithLog::UpdateHandleName(const char* name) { - if (handle_name_.length() != 0) { - handle_name_.append("::"); - } - handle_name_.append(name); -} - -void MinRPCReturnsWithLog::GetHandleName(void* handle) { - if (handle_descriptions_.find(handle) != handle_descriptions_.end()) { - handle_name_.append(handle_descriptions_[handle]); - logger_->LogHandleName(handle_name_); - } -} - -void MinRPCReturnsWithLog::ReleaseHandleName(void* handle) { - if (handle_descriptions_.find(handle) != handle_descriptions_.end()) { - logger_->LogHandleName(handle_descriptions_[handle]); - handle_descriptions_.erase(handle); - } -} - -void MinRPCReturnsWithLog::RegisterHandleName(void* handle) { - handle_descriptions_[handle] = handle_name_; -} - -void MinRPCExecuteWithLog::InitServer(int num_args) { - SetRPCCode(RPCCode::kInitServer); - logger_->Log("Init Server"); - next_->InitServer(num_args); -} - -void MinRPCExecuteWithLog::NormalCallFunc(uint64_t call_handle, TVMValue* values, int* tcodes, - int num_args) { - SetRPCCode(RPCCode::kCallFunc); - logger_->LogValue<void*>("call_handle: ", reinterpret_cast<void*>(call_handle)); - ret_handler_->GetHandleName(reinterpret_cast<void*>(call_handle)); - if (num_args > 0) { - logger_->Log(", "); - } - ProcessValues(values, tcodes, num_args); - next_->NormalCallFunc(call_handle, values, tcodes, num_args); -} - -void MinRPCExecuteWithLog::CopyFromRemote(DLTensor* arr, uint64_t num_bytes, uint8_t* temp_data) { - SetRPCCode(RPCCode::kCopyFromRemote); - logger_->LogValue<void*>("data_handle: ", static_cast<void*>(arr->data)); - logger_->LogDLDevice(", DLDevice(type,id):", &(arr->device)); - logger_->LogValue<int64_t>(", ndim: ", arr->ndim); - logger_->LogDLData(", DLDataType(code,bits,lane): ", &(arr->dtype)); - logger_->LogValue<uint64_t>(", num_bytes:", num_bytes); - next_->CopyFromRemote(arr, num_bytes, temp_data); -} - -int MinRPCExecuteWithLog::CopyToRemote(DLTensor* arr, uint64_t num_bytes, uint8_t* data_ptr) { - SetRPCCode(RPCCode::kCopyToRemote); - logger_->LogValue<void*>("data_handle: ", static_cast<void*>(arr->data)); - logger_->LogDLDevice(", DLDevice(type,id):", &(arr->device)); - logger_->LogValue<int64_t>(", ndim: ", arr->ndim); - logger_->LogDLData(", DLDataType(code,bits,lane): ", &(arr->dtype)); - logger_->LogValue<uint64_t>(", byte_offset: ", arr->byte_offset); - return next_->CopyToRemote(arr, num_bytes, data_ptr); -} - -void MinRPCExecuteWithLog::SysCallFunc(RPCCode code, TVMValue* values, int* tcodes, int num_args) { - SetRPCCode(code); - if ((code) == RPCCode::kFreeHandle) { - if ((num_args == 2) && (tcodes[0] == kTVMOpaqueHandle) && (tcodes[1] == kDLInt)) { - logger_->LogValue<void*>("handle: ", static_cast<void*>(values[0].v_handle)); - if (values[1].v_int64 == kTVMModuleHandle || values[1].v_int64 == kTVMPackedFuncHandle) { - ret_handler_->ReleaseHandleName(static_cast<void*>(values[0].v_handle)); - } - } - } else { - ProcessValues(values, tcodes, num_args); - } - next_->SysCallFunc(code, values, tcodes, num_args); -} - -void MinRPCExecuteWithLog::ThrowError(RPCServerStatus code, RPCCode info) { - logger_->Log("-> Error\n"); - next_->ThrowError(code, info); -} - -void MinRPCExecuteWithLog::ProcessValues(TVMValue* values, int* tcodes, int num_args) { - if (tcodes != nullptr) { - logger_->Log("["); - for (int i = 0; i < num_args; ++i) { - logger_->LogTVMValue(tcodes[i], values[i]); - - if (tcodes[i] == kTVMStr) { - if (strlen(values[i].v_str) > 0) { - ret_handler_->UpdateHandleName(values[i].v_str); - } - } - } - logger_->Log("]"); - } -} - -void MinRPCExecuteWithLog::SetRPCCode(RPCCode code) { - logger_->Log(RPCCodeToString(code)); - logger_->Log(", "); - ret_handler_->ResetHandleName(code); -} - -} // namespace runtime -} // namespace tvm diff --git a/src/runtime/minrpc/minrpc_logger.h b/src/runtime/minrpc/minrpc_logger.h deleted file mode 100644 index 13d44c3cba..0000000000 --- a/src/runtime/minrpc/minrpc_logger.h +++ /dev/null @@ -1,296 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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. - */ - -#ifndef TVM_RUNTIME_MINRPC_MINRPC_LOGGER_H_ -#define TVM_RUNTIME_MINRPC_MINRPC_LOGGER_H_ - -#include <tvm/runtime/c_runtime_api.h> - -#include <functional> -#include <sstream> -#include <string> -#include <unordered_map> - -#include "minrpc_interfaces.h" -#include "rpc_reference.h" - -namespace tvm { -namespace runtime { - -#define PRINT_BYTES false - -/*! - * \brief Generates a user readeable log on the console - */ -class Logger { - public: - Logger() {} - - /*! - * \brief this function logs a string - * - * \param s the string to be logged. - */ - void Log(const char* s) { os_ << s; } - void Log(std::string s) { os_ << s; } - - /*! - * \brief this function logs a numerical value - * - * \param desc adds any necessary description before the value. - * \param val is the value to be logged. - */ - template <typename T> - void LogValue(const char* desc, T val) { - os_ << desc << val; - } - - /*! - * \brief this function logs the properties of a DLDevice - * - * \param desc adds any necessary description before the DLDevice. - * \param dev is the pointer to the DLDevice to be logged. - */ - void LogDLDevice(const char* desc, DLDevice* dev) { - os_ << desc << "(" << dev->device_type << "," << dev->device_id << ")"; - } - - /*! - * \brief this function logs the properties of a DLDataType - * - * \param desc adds any necessary description before the DLDataType. - * \param data is the pointer to the DLDataType to be logged. - */ - void LogDLData(const char* desc, DLDataType* data) { - os_ << desc << "(" << (uint16_t)data->code << "," << (uint16_t)data->bits << "," << data->lanes - << ")"; - } - - /*! - * \brief this function logs a handle name. - * - * \param name is the name to be logged. - */ - void LogHandleName(std::string name) { - if (name.length() > 0) { - os_ << " <" << name.c_str() << ">"; - } - } - - /*! - * \brief this function logs a TVMValue based on its type. - * - * \param tcode the type_code of the value stored in TVMValue. - * \param value is the TVMValue to be logged. - */ - void LogTVMValue(int tcode, TVMValue value); - - /*! - * \brief this function output the log to the console. - */ - void OutputLog(); - - private: - std::stringstream os_; -}; - -/*! - * \brief A wrapper for a MinRPCReturns object, that also logs the responses. - * - * \param next underlying MinRPCReturns that generates the responses. - */ -class MinRPCReturnsWithLog : public MinRPCReturnInterface { - public: - /*! - * \brief Constructor. - * \param io The IO handler. - */ - MinRPCReturnsWithLog(MinRPCReturnInterface* next, Logger* logger) - : next_(next), logger_(logger) {} - - ~MinRPCReturnsWithLog() {} - - void ReturnVoid(); - - void ReturnHandle(void* handle); - - void ReturnException(const char* msg); - - void ReturnPackedSeq(const TVMValue* arg_values, const int* type_codes, int num_args); - - void ReturnCopyFromRemote(uint8_t* data_ptr, uint64_t num_bytes); - - void ReturnLastTVMError(); - - void ThrowError(RPCServerStatus code, RPCCode info = RPCCode::kNone); - - /*! - * \brief this function logs a list of TVMValues, and registers handle_name when needed. - * - * \param values is the list of TVMValues. - * \param tcodes is the list type_code of the TVMValues. - * \param num_args is the number of items in the list. - */ - void ProcessValues(const TVMValue* values, const int* tcodes, int num_args); - - /*! - * \brief this function is called when a new command is executed. - * It clears the handle_name_ and records the command code. - * - * \param code the RPC command code. - */ - void ResetHandleName(RPCCode code); - - /*! - * \brief appends name to the handle_name_. - * - * \param name handle name. - */ - void UpdateHandleName(const char* name); - - /*! - * \brief get the stored handle description. - * - * \param handle the handle to get the description for. - */ - void GetHandleName(void* handle); - - /*! - * \brief remove the handle description from handle_descriptions_. - * - * \param handle the handle to remove the description for. - */ - void ReleaseHandleName(void* handle); - - private: - /*! - * \brief add the handle description to handle_descriptions_. - * - * \param handle the handle to add the description for. - */ - void RegisterHandleName(void* handle); - - MinRPCReturnInterface* next_; - std::string handle_name_; - std::unordered_map<void*, std::string> handle_descriptions_; - RPCCode code_; - Logger* logger_; -}; - -/*! - * \brief A wrapper for a MinRPCExecute object, that also logs the responses. - * - * \param next: underlying MinRPCExecute that processes the packets. - */ -class MinRPCExecuteWithLog : public MinRPCExecInterface { - public: - MinRPCExecuteWithLog(MinRPCExecInterface* next, Logger* logger) : next_(next), logger_(logger) { - ret_handler_ = reinterpret_cast<MinRPCReturnsWithLog*>(next_->GetReturnInterface()); - } - - ~MinRPCExecuteWithLog() {} - - void InitServer(int num_args); - - void NormalCallFunc(uint64_t call_handle, TVMValue* values, int* tcodes, int num_args); - - void CopyFromRemote(DLTensor* arr, uint64_t num_bytes, uint8_t* temp_data); - - int CopyToRemote(DLTensor* arr, uint64_t _num_bytes, uint8_t* _data_ptr); - - void SysCallFunc(RPCCode code, TVMValue* values, int* tcodes, int num_args); - - void ThrowError(RPCServerStatus code, RPCCode info = RPCCode::kNone); - - MinRPCReturnInterface* GetReturnInterface() { return next_->GetReturnInterface(); } - - private: - /*! - * \brief this function logs a list of TVMValues, and updates handle_name when needed. - * - * \param values is the list of TVMValues. - * \param tcodes is the list type_code of the TVMValues. - * \param num_args is the number of items in the list. - */ - void ProcessValues(TVMValue* values, int* tcodes, int num_args); - - /*! - * \brief this function is called when a new command is executed. - * - * \param code the RPC command code. - */ - void SetRPCCode(RPCCode code); - - MinRPCExecInterface* next_; - MinRPCReturnsWithLog* ret_handler_; - Logger* logger_; -}; - -/*! - * \brief A No-operation MinRPCReturns used within the MinRPCSniffer - * - * \tparam TIOHandler* IO provider to provide io handling. - */ -template <typename TIOHandler> -class MinRPCReturnsNoOp : public MinRPCReturnInterface { - public: - /*! - * \brief Constructor. - * \param io The IO handler. - */ - explicit MinRPCReturnsNoOp(TIOHandler* io) : io_(io) {} - ~MinRPCReturnsNoOp() {} - void ReturnVoid() {} - void ReturnHandle(void* handle) {} - void ReturnException(const char* msg) {} - void ReturnPackedSeq(const TVMValue* arg_values, const int* type_codes, int num_args) {} - void ReturnCopyFromRemote(uint8_t* data_ptr, uint64_t num_bytes) {} - void ReturnLastTVMError() {} - void ThrowError(RPCServerStatus code, RPCCode info) {} - - private: - TIOHandler* io_; -}; - -/*! - * \brief A No-operation MinRPCExecute used within the MinRPCSniffer - * - * \tparam ReturnInterface* ReturnInterface pointer to generate and send the responses. - - */ -class MinRPCExecuteNoOp : public MinRPCExecInterface { - public: - explicit MinRPCExecuteNoOp(MinRPCReturnInterface* ret_handler) : ret_handler_(ret_handler) {} - ~MinRPCExecuteNoOp() {} - void InitServer(int _num_args) {} - void NormalCallFunc(uint64_t call_handle, TVMValue* values, int* tcodes, int num_args) {} - void CopyFromRemote(DLTensor* arr, uint64_t num_bytes, uint8_t* temp_data) {} - int CopyToRemote(DLTensor* arr, uint64_t num_bytes, uint8_t* data_ptr) { return 1; } - void SysCallFunc(RPCCode code, TVMValue* values, int* tcodes, int num_args) {} - void ThrowError(RPCServerStatus code, RPCCode info) {} - MinRPCReturnInterface* GetReturnInterface() { return ret_handler_; } - - private: - MinRPCReturnInterface* ret_handler_; -}; - -} // namespace runtime -} // namespace tvm - -#endif // TVM_RUNTIME_MINRPC_MINRPC_LOGGER_H_" diff --git a/src/runtime/minrpc/minrpc_server.h b/src/runtime/minrpc/minrpc_server.h index 2b14a8ae83..dc99c0d2c9 100644 --- a/src/runtime/minrpc/minrpc_server.h +++ b/src/runtime/minrpc/minrpc_server.h @@ -28,511 +28,24 @@ #ifndef TVM_RUNTIME_MINRPC_MINRPC_SERVER_H_ #define TVM_RUNTIME_MINRPC_MINRPC_SERVER_H_ -#ifndef DMLC_LITTLE_ENDIAN -#define DMLC_LITTLE_ENDIAN 1 -#endif - -#include <string.h> +#include <tvm/ffi/c_api.h> +#include <tvm/ffi/function.h> #include <tvm/runtime/c_runtime_api.h> +#include <tvm/runtime/logging.h> +#include <cstring> #include <memory> #include <utility> #include "../../support/generic_arena.h" -#include "minrpc_interfaces.h" #include "rpc_reference.h" -#ifndef MINRPC_CHECK -#define MINRPC_CHECK(cond) \ - if (!(cond)) this->ThrowError(RPCServerStatus::kCheckError); -#endif - namespace tvm { namespace runtime { - -namespace detail { +namespace details { template <typename TIOHandler> class PageAllocator; -} - -/*! - * \brief Responses to a minimum RPC command. - * - * \tparam TIOHandler IO provider to provide io handling. - */ -template <typename TIOHandler> -class MinRPCReturns : public MinRPCReturnInterface { - public: - /*! - * \brief Constructor. - * \param io The IO handler. - */ - explicit MinRPCReturns(TIOHandler* io) : io_(io) {} - - void ReturnVoid() { - int32_t num_args = 1; - int32_t tcode = kTVMNullptr; - RPCCode code = RPCCode::kReturn; - - uint64_t packet_nbytes = sizeof(code) + sizeof(num_args) + sizeof(tcode); - - io_->MessageStart(packet_nbytes); - Write(packet_nbytes); - Write(code); - Write(num_args); - Write(tcode); - io_->MessageDone(); - } - - void ReturnHandle(void* handle) { - int32_t num_args = 1; - int32_t tcode = kTVMOpaqueHandle; - RPCCode code = RPCCode::kReturn; - uint64_t encode_handle = reinterpret_cast<uint64_t>(handle); - uint64_t packet_nbytes = - sizeof(code) + sizeof(num_args) + sizeof(tcode) + sizeof(encode_handle); - - io_->MessageStart(packet_nbytes); - Write(packet_nbytes); - Write(code); - Write(num_args); - Write(tcode); - Write(encode_handle); - io_->MessageDone(); - } - - void ReturnException(const char* msg) { RPCReference::ReturnException(msg, this); } - - void ReturnPackedSeq(const TVMValue* arg_values, const int* type_codes, int num_args) { - RPCReference::ReturnPackedSeq(arg_values, type_codes, num_args, this); - } - - void ReturnCopyFromRemote(uint8_t* data_ptr, uint64_t num_bytes) { - RPCCode code = RPCCode::kCopyAck; - uint64_t packet_nbytes = sizeof(code) + num_bytes; - - io_->MessageStart(packet_nbytes); - Write(packet_nbytes); - Write(code); - WriteArray(data_ptr, num_bytes); - io_->MessageDone(); - } - - void ReturnLastTVMError() { - const char* err = TVMGetLastError(); - ReturnException(err); - } - - void MessageStart(uint64_t packet_nbytes) { io_->MessageStart(packet_nbytes); } - - void MessageDone() { io_->MessageDone(); } - - void ThrowError(RPCServerStatus code, RPCCode info = RPCCode::kNone) { - io_->Exit(static_cast<int>(code)); - } - - void WriteObject(void* obj) { this->ThrowError(RPCServerStatus::kUnknownTypeCode); } - uint64_t GetObjectBytes(void* obj) { - this->ThrowError(RPCServerStatus::kUnknownTypeCode); - return 0; - } - - template <typename T> - void Write(const T& data) { - static_assert(std::is_trivial<T>::value && std::is_standard_layout<T>::value, - "need to be trival"); - return WriteRawBytes(&data, sizeof(T)); - } - - template <typename T> - void WriteArray(T* data, size_t count) { - static_assert(std::is_trivial<T>::value && std::is_standard_layout<T>::value, - "need to be trival"); - return WriteRawBytes(data, sizeof(T) * count); - } - - private: - void WriteRawBytes(const void* data, size_t size) { - const uint8_t* buf = static_cast<const uint8_t*>(data); - size_t ndone = 0; - while (ndone < size) { - ssize_t ret = io_->PosixWrite(buf, size - ndone); - if (ret <= 0) { - this->ThrowError(RPCServerStatus::kWriteError); - } - buf += ret; - ndone += ret; - } - } - - TIOHandler* io_; -}; - -/*! - * \brief Executing a minimum RPC command. - * - * \tparam TIOHandler IO provider to provide io handling. - * \tparam MinRPCReturnInterface* handles response generatation and transmission. - */ -template <typename TIOHandler> -class MinRPCExecute : public MinRPCExecInterface { - public: - MinRPCExecute(TIOHandler* io, MinRPCReturnInterface* ret_handler) - : io_(io), ret_handler_(ret_handler) {} - - void InitServer(int num_args) { - MINRPC_CHECK(num_args == 0); - ret_handler_->ReturnVoid(); - } - - void NormalCallFunc(uint64_t call_handle, TVMValue* values, int* tcodes, int num_args) { - TVMValue ret_value[3]; - int ret_tcode[3]; - - int call_ecode = TVMFuncCall(reinterpret_cast<void*>(call_handle), values, tcodes, num_args, - &(ret_value[1]), &(ret_tcode[1])); - - if (call_ecode == 0) { - // Return value encoding as in LocalSession - int rv_tcode = ret_tcode[1]; - ret_tcode[0] = kDLInt; - ret_value[0].v_int64 = rv_tcode; - if (rv_tcode == kTVMNDArrayHandle) { - ret_tcode[1] = kTVMDLTensorHandle; - ret_value[2].v_handle = ret_value[1].v_handle; - ret_tcode[2] = kTVMOpaqueHandle; - ret_handler_->ReturnPackedSeq(ret_value, ret_tcode, 3); - } else if (rv_tcode == kTVMBytes) { - ret_tcode[1] = kTVMBytes; - ret_handler_->ReturnPackedSeq(ret_value, ret_tcode, 2); - TVMByteArrayFree(reinterpret_cast<TVMByteArray*>(ret_value[1].v_handle)); // NOLINT(*) - } else if (rv_tcode == kTVMPackedFuncHandle || rv_tcode == kTVMModuleHandle || - rv_tcode == kTVMObjectHandle) { - ret_tcode[1] = kTVMOpaqueHandle; - ret_handler_->ReturnPackedSeq(ret_value, ret_tcode, 2); - } else { - ret_handler_->ReturnPackedSeq(ret_value, ret_tcode, 2); - } - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void CopyFromRemote(DLTensor* arr, uint64_t num_bytes, uint8_t* data_ptr) { - int call_ecode = 0; - if (arr->device.device_type != kDLCPU) { - DLTensor temp; - temp.data = static_cast<void*>(data_ptr); - temp.device = DLDevice{kDLCPU, 0}; - temp.ndim = arr->ndim; - temp.dtype = arr->dtype; - temp.shape = arr->shape; - temp.strides = nullptr; - temp.byte_offset = 0; - call_ecode = TVMDeviceCopyDataFromTo(arr, &temp, nullptr); - // need sync to make sure that the copy is completed. - if (call_ecode == 0) { - call_ecode = TVMSynchronize(arr->device.device_type, arr->device.device_id, nullptr); - } - } - - if (call_ecode == 0) { - ret_handler_->ReturnCopyFromRemote(data_ptr, num_bytes); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - int CopyToRemote(DLTensor* arr, uint64_t num_bytes, uint8_t* data_ptr) { - int call_ecode = 0; - - int ret = ReadArray(data_ptr, num_bytes); - if (ret <= 0) return ret; - - if (arr->device.device_type != kDLCPU) { - DLTensor temp; - temp.data = data_ptr; - temp.device = DLDevice{kDLCPU, 0}; - temp.ndim = arr->ndim; - temp.dtype = arr->dtype; - temp.shape = arr->shape; - temp.strides = nullptr; - temp.byte_offset = 0; - call_ecode = TVMDeviceCopyDataFromTo(&temp, arr, nullptr); - // need sync to make sure that the copy is completed. - if (call_ecode == 0) { - call_ecode = TVMSynchronize(arr->device.device_type, arr->device.device_id, nullptr); - } - } - - if (call_ecode == 0) { - ret_handler_->ReturnVoid(); - } else { - ret_handler_->ReturnLastTVMError(); - } - - return 1; - } - - void SysCallFunc(RPCCode code, TVMValue* values, int* tcodes, int num_args) { - switch (code) { - case RPCCode::kFreeHandle: { - SyscallFreeHandle(values, tcodes, num_args); - break; - } - case RPCCode::kGetGlobalFunc: { - SyscallGetGlobalFunc(values, tcodes, num_args); - break; - } - case RPCCode::kDevSetDevice: { - ret_handler_->ReturnException("SetDevice not supported"); - break; - } - case RPCCode::kDevGetAttr: { - ret_handler_->ReturnException("GetAttr not supported"); - break; - } - case RPCCode::kDevAllocData: { - SyscallDevAllocData(values, tcodes, num_args); - break; - } - case RPCCode::kDevAllocDataWithScope: { - SyscallDevAllocDataWithScope(values, tcodes, num_args); - break; - } - case RPCCode::kDevFreeData: { - SyscallDevFreeData(values, tcodes, num_args); - break; - } - case RPCCode::kDevCreateStream: { - SyscallDevCreateStream(values, tcodes, num_args); - break; - } - case RPCCode::kDevFreeStream: { - SyscallDevFreeStream(values, tcodes, num_args); - break; - } - case RPCCode::kDevStreamSync: { - SyscallDevStreamSync(values, tcodes, num_args); - break; - } - case RPCCode::kDevSetStream: { - SyscallDevSetStream(values, tcodes, num_args); - break; - } - case RPCCode::kCopyAmongRemote: { - SyscallCopyAmongRemote(values, tcodes, num_args); - break; - } - default: { - ret_handler_->ReturnException("Syscall not recognized"); - break; - } - } - } - - void SyscallFreeHandle(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 1); - MINRPC_CHECK(tcodes[0] == kTVMOpaqueHandle); - void* handle = values[0].v_handle; - int call_ecode = TVMObjectFree(handle); - - if (call_ecode == 0) { - ret_handler_->ReturnVoid(); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void SyscallGetGlobalFunc(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 1); - MINRPC_CHECK(tcodes[0] == kTVMStr); - void* handle; - int call_ecode = TVMFuncGetGlobal(values[0].v_str, &handle); - - if (call_ecode == 0) { - ret_handler_->ReturnHandle(handle); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void SyscallCopyAmongRemote(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 3); - // from dltensor - MINRPC_CHECK(tcodes[0] == kTVMDLTensorHandle); - // to dltensor - MINRPC_CHECK(tcodes[1] == kTVMDLTensorHandle); - // stream - MINRPC_CHECK(tcodes[2] == kTVMOpaqueHandle); - - void* from = values[0].v_handle; - void* to = values[1].v_handle; - TVMStreamHandle stream = values[2].v_handle; - - int call_ecode = TVMDeviceCopyDataFromTo(reinterpret_cast<DLTensor*>(from), - reinterpret_cast<DLTensor*>(to), stream); - - if (call_ecode == 0) { - ret_handler_->ReturnVoid(); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void SyscallDevAllocData(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 4); - MINRPC_CHECK(tcodes[0] == kDLDevice); - MINRPC_CHECK(tcodes[1] == kDLInt); - MINRPC_CHECK(tcodes[2] == kDLInt); - MINRPC_CHECK(tcodes[3] == kTVMDataType); - - DLDevice dev = values[0].v_device; - int64_t nbytes = values[1].v_int64; - int64_t alignment = values[2].v_int64; - DLDataType type_hint = values[3].v_type; - - void* handle; - int call_ecode = TVMDeviceAllocDataSpace(dev, nbytes, alignment, type_hint, &handle); - - if (call_ecode == 0) { - ret_handler_->ReturnHandle(handle); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void SyscallDevAllocDataWithScope(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 2); - MINRPC_CHECK(tcodes[0] == kTVMDLTensorHandle); - MINRPC_CHECK(tcodes[1] == kTVMNullptr || tcodes[1] == kTVMStr); - - DLTensor* arr = static_cast<DLTensor*>(values[0].v_handle); - const char* mem_scope = (tcodes[1] == kTVMNullptr ? nullptr : values[1].v_str); - void* handle; - int call_ecode = TVMDeviceAllocDataSpaceWithScope(arr->device, arr->ndim, arr->shape, - arr->dtype, mem_scope, &handle); - if (call_ecode == 0) { - ret_handler_->ReturnHandle(handle); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void SyscallDevFreeData(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 2); - MINRPC_CHECK(tcodes[0] == kDLDevice); - MINRPC_CHECK(tcodes[1] == kTVMOpaqueHandle); - - DLDevice dev = values[0].v_device; - void* handle = values[1].v_handle; - - int call_ecode = TVMDeviceFreeDataSpace(dev, handle); - - if (call_ecode == 0) { - ret_handler_->ReturnVoid(); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void SyscallDevCreateStream(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 1); - MINRPC_CHECK(tcodes[0] == kDLDevice); - - DLDevice dev = values[0].v_device; - void* handle; - - int call_ecode = TVMStreamCreate(dev.device_type, dev.device_id, &handle); - - if (call_ecode == 0) { - ret_handler_->ReturnHandle(handle); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void SyscallDevFreeStream(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 2); - MINRPC_CHECK(tcodes[0] == kDLDevice); - MINRPC_CHECK(tcodes[1] == kTVMOpaqueHandle); - - DLDevice dev = values[0].v_device; - void* handle = values[1].v_handle; - - int call_ecode = TVMStreamFree(dev.device_type, dev.device_id, handle); - - if (call_ecode == 0) { - ret_handler_->ReturnVoid(); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void SyscallDevStreamSync(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 2); - MINRPC_CHECK(tcodes[0] == kDLDevice); - MINRPC_CHECK(tcodes[1] == kTVMOpaqueHandle); - - DLDevice dev = values[0].v_device; - void* handle = values[1].v_handle; - - int call_ecode = TVMSynchronize(dev.device_type, dev.device_id, handle); - - if (call_ecode == 0) { - ret_handler_->ReturnVoid(); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void SyscallDevSetStream(TVMValue* values, int* tcodes, int num_args) { - MINRPC_CHECK(num_args == 2); - MINRPC_CHECK(tcodes[0] == kDLDevice); - MINRPC_CHECK(tcodes[1] == kTVMOpaqueHandle); - - DLDevice dev = values[0].v_device; - void* handle = values[1].v_handle; - - int call_ecode = TVMSetStream(dev.device_type, dev.device_id, handle); - - if (call_ecode == 0) { - ret_handler_->ReturnVoid(); - } else { - ret_handler_->ReturnLastTVMError(); - } - } - - void ThrowError(RPCServerStatus code, RPCCode info = RPCCode::kNone) { - ret_handler_->ThrowError(code, info); - } - - MinRPCReturnInterface* GetReturnInterface() { return ret_handler_; } - - private: - template <typename T> - int ReadArray(T* data, size_t count) { - static_assert(std::is_trivial<T>::value && std::is_standard_layout<T>::value, - "need to be trival"); - return ReadRawBytes(data, sizeof(T) * count); - } - - int ReadRawBytes(void* data, size_t size) { - uint8_t* buf = static_cast<uint8_t*>(data); - size_t ndone = 0; - while (ndone < size) { - ssize_t ret = io_->PosixRead(buf, size - ndone); - if (ret <= 0) return ret; - ndone += ret; - buf += ret; - } - return 1; - } - - TIOHandler* io_; - MinRPCReturnInterface* ret_handler_; -}; - +} // namespace details /*! * \brief A minimum RPC server that only depends on the tvm C runtime.. * @@ -544,180 +57,60 @@ class MinRPCExecute : public MinRPCExecInterface { * - MessageStart(num_bytes), MessageDone(): framing APIs. * - Exit: exit with status code. */ -template <typename TIOHandler, template <typename> class Allocator = detail::PageAllocator> +template <typename TIOHandler, template <typename> class Allocator = details::PageAllocator> class MinRPCServer { public: using PageAllocator = Allocator<TIOHandler>; - /*! - * \brief Constructor. - * \param io The IO handler. - */ - MinRPCServer(TIOHandler* io, std::unique_ptr<MinRPCExecInterface>&& exec_handler) - : io_(io), arena_(PageAllocator(io_)), exec_handler_(std::move(exec_handler)) {} + using FServerHandler = ffi::TypedFunction<int(TVMFFIByteArray*, int)>; - explicit MinRPCServer(TIOHandler* io) - : io_(io), - arena_(PageAllocator(io)), - ret_handler_(new MinRPCReturns<TIOHandler>(io_)), - exec_handler_(std::unique_ptr<MinRPCExecInterface>( - new MinRPCExecute<TIOHandler>(io_, ret_handler_))) {} - - ~MinRPCServer() { - if (ret_handler_ != nullptr) { - delete ret_handler_; - } + explicit MinRPCServer(TIOHandler* io) : io_(io), arena_(PageAllocator(io_)) { + auto fsend = ffi::Function::FromTyped( + [this](TVMFFIByteArray* bytes) { return io_->PosixWrite(bytes->data, bytes->size); }); + auto fcreate = tvm::ffi::Function::GetGlobalRequired("rpc.CreateEventDrivenServer"); + ffi::Any value = fcreate(fsend, "MinRPCServer", ""); + fserver_handler_ = value.cast<FServerHandler>(); } - /*! \brief Process a single request. + /*! + * \brief Process a single request. * * \return true when the server should continue processing requests. false when it should be * shutdown. */ bool ProcessOnePacket() { - RPCCode code; uint64_t packet_len; - arena_.RecycleAll(); allow_clean_shutdown_ = true; - Read(&packet_len); if (packet_len == 0) return true; - Read(&code); - allow_clean_shutdown_ = false; - - if (code >= RPCCode::kSyscallCodeStart) { - HandleSyscallFunc(code); - } else { - switch (code) { - case RPCCode::kCallFunc: { - HandleNormalCallFunc(); - break; - } - case RPCCode::kInitServer: { - HandleInitServer(); - break; - } - case RPCCode::kCopyFromRemote: { - HandleCopyFromRemote(); - break; - } - case RPCCode::kCopyToRemote: { - HandleCopyToRemote(); - break; - } - case RPCCode::kShutdown: { - Shutdown(); - return false; - } - default: { - this->ThrowError(RPCServerStatus::kUnknownRPCCode); - break; - } + char* read_buffer = this->ArenaAlloc<char>(sizeof(uint64_t) + packet_len); + // copy header into read buffer + std::memcpy(read_buffer, &packet_len, sizeof(uint64_t)); + // read the rest of the packet + ReadRawBytes(read_buffer + sizeof(uint64_t), packet_len); + // setup write flags + int write_flags = 3; + TVMFFIByteArray read_bytes{read_buffer, sizeof(uint64_t) + static_cast<size_t>(packet_len)}; + int status = fserver_handler_(&read_bytes, write_flags); + + while (status == 2) { + TVMFFIByteArray write_bytes{nullptr, 0}; + // continue call handler until it have nothing to write + status = fserver_handler_(&write_bytes, write_flags); + if (status == 0) { + this->Shutdown(); + return false; } } - return true; } - void HandleInitServer() { - uint64_t len; - Read(&len); - char* proto_ver = ArenaAlloc<char>(len + 1); - ReadArray(proto_ver, len); - TVMValue* values; - int* tcodes; - int num_args; - RecvPackedSeq(&values, &tcodes, &num_args); - exec_handler_->InitServer(num_args); - } - void Shutdown() { arena_.FreeAll(); io_->Close(); } - void HandleNormalCallFunc() { - uint64_t call_handle; - TVMValue* values; - int* tcodes; - int num_args; - - Read(&call_handle); - RecvPackedSeq(&values, &tcodes, &num_args); - exec_handler_->NormalCallFunc(call_handle, values, tcodes, num_args); - } - - void HandleCopyFromRemote() { - DLTensor* arr = ArenaAlloc<DLTensor>(1); - uint64_t data_handle; - Read(&data_handle); - arr->data = reinterpret_cast<void*>(data_handle); - Read(&(arr->device)); - Read(&(arr->ndim)); - Read(&(arr->dtype)); - arr->shape = ArenaAlloc<int64_t>(arr->ndim); - ReadArray(arr->shape, arr->ndim); - arr->strides = nullptr; - Read(&(arr->byte_offset)); - - uint64_t num_bytes; - Read(&num_bytes); - - uint8_t* data_ptr; - if (arr->device.device_type == kDLCPU) { - data_ptr = reinterpret_cast<uint8_t*>(data_handle) + arr->byte_offset; - } else { - data_ptr = ArenaAlloc<uint8_t>(num_bytes); - } - - exec_handler_->CopyFromRemote(arr, num_bytes, data_ptr); - } - - void HandleCopyToRemote() { - DLTensor* arr = ArenaAlloc<DLTensor>(1); - uint64_t data_handle; - Read(&data_handle); - arr->data = reinterpret_cast<void*>(data_handle); - Read(&(arr->device)); - Read(&(arr->ndim)); - Read(&(arr->dtype)); - arr->shape = ArenaAlloc<int64_t>(arr->ndim); - ReadArray(arr->shape, arr->ndim); - arr->strides = nullptr; - Read(&(arr->byte_offset)); - uint64_t num_bytes; - Read(&num_bytes); - int ret; - if (arr->device.device_type == kDLCPU) { - uint8_t* dptr = reinterpret_cast<uint8_t*>(data_handle) + arr->byte_offset; - ret = exec_handler_->CopyToRemote(arr, num_bytes, dptr); - } else { - uint8_t* temp_data = ArenaAlloc<uint8_t>(num_bytes); - ret = exec_handler_->CopyToRemote(arr, num_bytes, temp_data); - } - if (ret == 0) { - if (allow_clean_shutdown_) { - Shutdown(); - io_->Exit(0); - } else { - this->ThrowError(RPCServerStatus::kReadError); - } - } - if (ret == -1) { - this->ThrowError(RPCServerStatus::kReadError); - } - } - - void HandleSyscallFunc(RPCCode code) { - TVMValue* values; - int* tcodes; - int num_args; - RecvPackedSeq(&values, &tcodes, &num_args); - - exec_handler_->SysCallFunc(code, values, tcodes, num_args); - } - void ThrowError(RPCServerStatus code, RPCCode info = RPCCode::kNone) { io_->Exit(static_cast<int>(code)); } @@ -736,32 +129,7 @@ class MinRPCServer { ReadRawBytes(data, sizeof(T)); } - template <typename T> - void ReadArray(T* data, size_t count) { - static_assert(std::is_trivial<T>::value && std::is_standard_layout<T>::value, - "need to be trival"); - return ReadRawBytes(data, sizeof(T) * count); - } - - void ReadObject(int* tcode, TVMValue* value) { - // handles RPCObject in minRPC - // NOTE: object needs to be supported by C runtime - // because minrpc's restriction of C only - // we only handle RPCObjectRef - uint32_t type_index; - Read(&type_index); - MINRPC_CHECK(type_index == kRuntimeRPCObjectRefTypeIndex); - uint64_t object_handle; - Read(&object_handle); - tcode[0] = kTVMObjectHandle; - value[0].v_handle = reinterpret_cast<void*>(object_handle); - } - private: - void RecvPackedSeq(TVMValue** out_values, int** out_tcodes, int* out_num_args) { - RPCReference::RecvPackedSeq(out_values, out_tcodes, out_num_args, this); - } - void ReadRawBytes(void* data, size_t size) { uint8_t* buf = static_cast<uint8_t*>(data); size_t ndone = 0; @@ -783,18 +151,17 @@ class MinRPCServer { } } + /*! \brief server handler. */ + FServerHandler fserver_handler_; /*! \brief IO handler. */ TIOHandler* io_; /*! \brief internal arena. */ support::GenericArena<PageAllocator> arena_; - MinRPCReturns<TIOHandler>* ret_handler_ = nullptr; - std::unique_ptr<MinRPCExecInterface> exec_handler_; /*! \brief Whether we are in a state that allows clean shutdown. */ bool allow_clean_shutdown_{true}; - static_assert(DMLC_LITTLE_ENDIAN == 1, "MinRPC only works on little endian."); }; -namespace detail { +namespace details { // Internal allocator that redirects alloc to TVM's C API. template <typename TIOHandler> class PageAllocator { @@ -805,10 +172,9 @@ class PageAllocator { ArenaPageHeader* allocate(size_t min_size) { size_t npages = ((min_size + kPageSize - 1) / kPageSize); - void* data; + void* data = malloc(npages * kPageSize); - if (TVMDeviceAllocDataSpace(DLDevice{kDLCPU, 0}, npages * kPageSize, kPageAlign, - DLDataType{kDLInt, 1, 1}, &data) != 0) { + if (data == nullptr) { io_->Exit(static_cast<int>(RPCServerStatus::kAllocError)); } @@ -818,11 +184,7 @@ class PageAllocator { return header; } - void deallocate(ArenaPageHeader* page) { - if (TVMDeviceFreeDataSpace(DLDevice{kDLCPU, 0}, page) != 0) { - io_->Exit(static_cast<int>(RPCServerStatus::kAllocError)); - } - } + void deallocate(ArenaPageHeader* page) { free(page); } static const constexpr int kPageSize = 2 << 10; static const constexpr int kPageAlign = 8; @@ -830,7 +192,7 @@ class PageAllocator { private: TIOHandler* io_; }; -} // namespace detail +} // namespace details } // namespace runtime } // namespace tvm diff --git a/src/runtime/minrpc/minrpc_server_logging.h b/src/runtime/minrpc/minrpc_server_logging.h deleted file mode 100644 index 89650efe9a..0000000000 --- a/src/runtime/minrpc/minrpc_server_logging.h +++ /dev/null @@ -1,170 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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. - */ - -#ifndef TVM_RUNTIME_MINRPC_MINRPC_SERVER_LOGGING_H_ -#define TVM_RUNTIME_MINRPC_MINRPC_SERVER_LOGGING_H_ - -#include <memory> -#include <utility> - -#include "minrpc_logger.h" -#include "minrpc_server.h" - -namespace tvm { -namespace runtime { - -/*! - * \brief A minimum RPC server that logs the received commands. - * - * \tparam TIOHandler IO provider to provide io handling. - */ -template <typename TIOHandler> -class MinRPCServerWithLog { - public: - explicit MinRPCServerWithLog(TIOHandler* io) - : ret_handler_(io), - ret_handler_wlog_(&ret_handler_, &logger_), - exec_handler_(io, &ret_handler_wlog_), - exec_handler_ptr_(new MinRPCExecuteWithLog(&exec_handler_, &logger_)), - next_(io, std::move(exec_handler_ptr_)) {} - - bool ProcessOnePacket() { return next_.ProcessOnePacket(); } - - private: - Logger logger_; - MinRPCReturns<TIOHandler> ret_handler_; - MinRPCExecute<TIOHandler> exec_handler_; - MinRPCReturnsWithLog ret_handler_wlog_; - std::unique_ptr<MinRPCExecuteWithLog> exec_handler_ptr_; - MinRPCServer<TIOHandler> next_; -}; - -/*! - * \brief A minimum RPC server that only logs the outgoing commands and received responses. - * (Does not process the packets or respond to them.) - * - * \tparam TIOHandler IO provider to provide io handling. - */ -template <typename TIOHandler, template <typename> class Allocator = detail::PageAllocator> -class MinRPCSniffer { - public: - using PageAllocator = Allocator<TIOHandler>; - explicit MinRPCSniffer(TIOHandler* io) - : io_(io), - arena_(PageAllocator(io_)), - ret_handler_(io_), - ret_handler_wlog_(&ret_handler_, &logger_), - exec_handler_(&ret_handler_wlog_), - exec_handler_ptr_(new MinRPCExecuteWithLog(&exec_handler_, &logger_)), - next_(io_, std::move(exec_handler_ptr_)) {} - - bool ProcessOnePacket() { return next_.ProcessOnePacket(); } - - void ProcessOneResponse() { - RPCCode code; - uint64_t packet_len = 0; - - if (!Read(&packet_len)) return; - if (packet_len == 0) { - OutputLog(); - return; - } - if (!Read(&code)) return; - switch (code) { - case RPCCode::kReturn: { - int32_t num_args; - int* type_codes; - TVMValue* values; - RPCReference::RecvPackedSeq(&values, &type_codes, &num_args, this); - ret_handler_wlog_.ReturnPackedSeq(values, type_codes, num_args); - break; - } - case RPCCode::kException: { - ret_handler_wlog_.ReturnException(""); - break; - } - default: { - OutputLog(); - break; - } - } - } - - void OutputLog() { logger_.OutputLog(); } - - void ThrowError(RPCServerStatus code, RPCCode info = RPCCode::kNone) { - logger_.Log("-> "); - logger_.Log(RPCServerStatusToString(code)); - OutputLog(); - } - - template <typename T> - T* ArenaAlloc(int count) { - static_assert(std::is_trivial<T>::value && std::is_standard_layout<T>::value, - "need to be trival"); - return arena_.template allocate_<T>(count); - } - - template <typename T> - bool Read(T* data) { - static_assert(std::is_trivial<T>::value && std::is_standard_layout<T>::value, - "need to be trival"); - return ReadRawBytes(data, sizeof(T)); - } - - template <typename T> - bool ReadArray(T* data, size_t count) { - static_assert(std::is_trivial<T>::value && std::is_standard_layout<T>::value, - "need to be trival"); - return ReadRawBytes(data, sizeof(T) * count); - } - - void ReadObject(int* tcode, TVMValue* value) { - this->ThrowError(RPCServerStatus::kUnknownTypeCode); - } - - private: - bool ReadRawBytes(void* data, size_t size) { - uint8_t* buf = reinterpret_cast<uint8_t*>(data); - size_t ndone = 0; - while (ndone < size) { - ssize_t ret = io_->PosixRead(buf, size - ndone); - if (ret <= 0) { - this->ThrowError(RPCServerStatus::kReadError); - return false; - } - ndone += ret; - buf += ret; - } - return true; - } - - Logger logger_; - TIOHandler* io_; - support::GenericArena<PageAllocator> arena_; - MinRPCReturnsNoOp<TIOHandler> ret_handler_; - MinRPCReturnsWithLog ret_handler_wlog_; - MinRPCExecuteNoOp exec_handler_; - std::unique_ptr<MinRPCExecuteWithLog> exec_handler_ptr_; - MinRPCServer<TIOHandler> next_; -}; - -} // namespace runtime -} // namespace tvm -#endif // TVM_RUNTIME_MINRPC_MINRPC_SERVER_LOGGING_H_ diff --git a/src/runtime/minrpc/posix_popen_server/posix_popen_server.cc b/src/runtime/minrpc/posix_popen_server/posix_popen_server.cc index b513d4b7cc..014704e970 100644 --- a/src/runtime/minrpc/posix_popen_server/posix_popen_server.cc +++ b/src/runtime/minrpc/posix_popen_server/posix_popen_server.cc @@ -17,9 +17,6 @@ * under the License. */ -// Disable constructor to bring minimum dep on c++ABI. -#define TVM_ARENA_HAS_DESTRUCTOR 0 - #include <unistd.h> #include <cstdlib> diff --git a/src/runtime/rpc/rpc_channel_logger.h b/src/runtime/rpc/rpc_channel_logger.h deleted file mode 100644 index 8fe68f6690..0000000000 --- a/src/runtime/rpc/rpc_channel_logger.h +++ /dev/null @@ -1,186 +0,0 @@ -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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. - */ - -/*! - * \file rpc_channel_logger.h - * \brief A wrapper for RPCChannel with a NanoRPCListener for logging the commands. - */ -#ifndef TVM_RUNTIME_RPC_RPC_CHANNEL_LOGGER_H_ -#define TVM_RUNTIME_RPC_RPC_CHANNEL_LOGGER_H_ - -#include <tvm/runtime/c_runtime_api.h> - -#include <memory> -#include <utility> - -#include "../../support/ssize.h" -#include "../minrpc/minrpc_server_logging.h" -#include "rpc_channel.h" - -#define RX_BUFFER_SIZE 65536 - -namespace tvm { -namespace runtime { - -class Buffer { - public: - Buffer(uint8_t* data, size_t data_size_bytes) - : data_{data}, capacity_{data_size_bytes}, num_valid_bytes_{0}, read_cursor_{0} {} - - size_t Write(const uint8_t* data, size_t data_size_bytes) { - size_t num_bytes_available = capacity_ - num_valid_bytes_; - size_t num_bytes_to_copy = data_size_bytes; - if (num_bytes_available < num_bytes_to_copy) { - num_bytes_to_copy = num_bytes_available; - } - - memcpy(&data_[num_valid_bytes_], data, num_bytes_to_copy); - num_valid_bytes_ += num_bytes_to_copy; - return num_bytes_to_copy; - } - - size_t Read(uint8_t* data, size_t data_size_bytes) { - size_t num_bytes_to_copy = data_size_bytes; - size_t num_bytes_available = num_valid_bytes_ - read_cursor_; - if (num_bytes_available < num_bytes_to_copy) { - num_bytes_to_copy = num_bytes_available; - } - - memcpy(data, &data_[read_cursor_], num_bytes_to_copy); - read_cursor_ += num_bytes_to_copy; - return num_bytes_to_copy; - } - - void Clear() { - num_valid_bytes_ = 0; - read_cursor_ = 0; - } - - size_t Size() const { return num_valid_bytes_; } - - private: - /*! \brief pointer to data buffer. */ - uint8_t* data_; - - /*! \brief The total number of bytes available in data_.*/ - size_t capacity_; - - /*! \brief number of valid bytes in the buffer. */ - size_t num_valid_bytes_; - - /*! \brief Read cursor position. */ - size_t read_cursor_; -}; - -/*! - * \brief A simple IO handler for MinRPCSniffer. - * - * \tparam Buffer* buffer to store received data. - */ -class SnifferIOHandler { - public: - explicit SnifferIOHandler(Buffer* receive_buffer) : receive_buffer_(receive_buffer) {} - - void MessageStart(size_t message_size_bytes) {} - - ssize_t PosixWrite(const uint8_t* buf, size_t buf_size_bytes) { return 0; } - - void MessageDone() {} - - ssize_t PosixRead(uint8_t* buf, size_t buf_size_bytes) { - return receive_buffer_->Read(buf, buf_size_bytes); - } - - void Close() {} - - void Exit(int code) {} - - private: - Buffer* receive_buffer_; -}; - -/*! - * \brief A simple rpc session that logs the received commands. - */ -class NanoRPCListener { - public: - NanoRPCListener() - : receive_buffer_(receive_storage_, receive_storage_size_bytes_), - io_(&receive_buffer_), - rpc_server_(&io_) {} - - void Listen(const uint8_t* data, size_t size) { receive_buffer_.Write(data, size); } - - void ProcessTxPacket() { - rpc_server_.ProcessOnePacket(); - ClearBuffer(); - } - - void ProcessRxPacket() { - rpc_server_.ProcessOneResponse(); - ClearBuffer(); - } - - private: - void ClearBuffer() { receive_buffer_.Clear(); } - - private: - size_t receive_storage_size_bytes_ = RX_BUFFER_SIZE; - uint8_t receive_storage_[RX_BUFFER_SIZE]; - Buffer receive_buffer_; - SnifferIOHandler io_; - MinRPCSniffer<SnifferIOHandler> rpc_server_; - - void HandleCompleteMessage() { rpc_server_.ProcessOnePacket(); } - - static void HandleCompleteMessageCb(void* context) { - static_cast<NanoRPCListener*>(context)->HandleCompleteMessage(); - } -}; - -/*! - * \brief A wrapper for RPCChannel, that also logs the commands sent. - * - * \tparam std::unique_ptr<RPCChannel>&& underlying RPCChannel unique_ptr. - */ -class RPCChannelLogging : public RPCChannel { - public: - explicit RPCChannelLogging(std::unique_ptr<RPCChannel>&& next) { next_ = std::move(next); } - - size_t Send(const void* data, size_t size) { - listener_.ProcessRxPacket(); - listener_.Listen((const uint8_t*)data, size); - listener_.ProcessTxPacket(); - return next_->Send(data, size); - } - - size_t Recv(void* data, size_t size) { - size_t ret = next_->Recv(data, size); - listener_.Listen((const uint8_t*)data, size); - return ret; - } - - private: - std::unique_ptr<RPCChannel> next_; - NanoRPCListener listener_; -}; - -} // namespace runtime -} // namespace tvm -#endif // TVM_RUNTIME_RPC_RPC_CHANNEL_LOGGER_H_ diff --git a/src/runtime/rpc/rpc_endpoint.cc b/src/runtime/rpc/rpc_endpoint.cc index 23edfa9bb5..e8d5e83bb4 100644 --- a/src/runtime/rpc/rpc_endpoint.cc +++ b/src/runtime/rpc/rpc_endpoint.cc @@ -838,8 +838,12 @@ int RPCEndpoint::ServerAsyncIOEventHandler(const std::string& in_bytes, int even writer_.bytes_available()); } ICHECK(code != RPCCode::kReturn && code != RPCCode::kCopyAck); + // if the code is kShutdown, return 0 to indicate the server should exit if (code == RPCCode::kShutdown) return 0; + // if the writer has bytes available, return 2 to indicate the server should send data + // usually by calling the handler again if (writer_.bytes_available() != 0) return 2; + // otherwise, return 1 to indicate the server should and read return 1; } diff --git a/src/runtime/rpc/rpc_endpoint.h b/src/runtime/rpc/rpc_endpoint.h index a420e6d92f..5d94aed470 100644 --- a/src/runtime/rpc/rpc_endpoint.h +++ b/src/runtime/rpc/rpc_endpoint.h @@ -34,7 +34,6 @@ #include "../../support/ring_buffer.h" #include "../minrpc/rpc_reference.h" #include "rpc_channel.h" -#include "rpc_channel_logger.h" #include "rpc_session.h" namespace tvm { diff --git a/src/runtime/rpc/rpc_socket_impl.cc b/src/runtime/rpc/rpc_socket_impl.cc index 286d143bad..f51117211a 100644 --- a/src/runtime/rpc/rpc_socket_impl.cc +++ b/src/runtime/rpc/rpc_socket_impl.cc @@ -98,9 +98,6 @@ std::shared_ptr<RPCEndpoint> RPCConnect(std::string url, int port, std::string k } std::unique_ptr<RPCChannel> channel = std::make_unique<SockChannel>(sock); - if (enable_logging) { - channel.reset(new RPCChannelLogging(std::move(channel))); - } auto endpt = RPCEndpoint::Create(std::move(channel), key, remote_key); endpt->InitRemoteSession(init_seq); diff --git a/tests/python/runtime/test_runtime_rpc.py b/tests/python/runtime/test_runtime_rpc.py index 604d8eb42c..9ba89fe7f5 100644 --- a/tests/python/runtime/test_runtime_rpc.py +++ b/tests/python/runtime/test_runtime_rpc.py @@ -112,25 +112,6 @@ def test_rpc_simple(): check_remote() [email protected]_rpc -def test_rpc_simple_wlog(): - server = rpc.Server(key="x1") - client = rpc.connect("127.0.0.1", server.port, key="x1", enable_logging=True) - - def check_remote(): - f1 = client.get_function("rpc.test.addone") - assert f1(10) == 11 - f3 = client.get_function("rpc.test.except") - - with pytest.raises(tvm._ffi.base.TVMError): - f3("abc") - - f2 = client.get_function("rpc.test.strcat") - assert f2("abc", 11) == "abc:11" - - check_remote() - - @tvm.testing.requires_rpc def test_rpc_runtime_string(): server = rpc.Server(key="x1")
