diff --git a/libraries/libstratosphere/source/htc/server/rpc/htc_htcmisc_rpc_tasks.hpp b/libraries/libstratosphere/source/htc/server/rpc/htc_htcmisc_rpc_tasks.hpp index 983b36b4c..20268570f 100644 --- a/libraries/libstratosphere/source/htc/server/rpc/htc_htcmisc_rpc_tasks.hpp +++ b/libraries/libstratosphere/source/htc/server/rpc/htc_htcmisc_rpc_tasks.hpp @@ -61,7 +61,7 @@ namespace ams::htc::server::rpc { s64 body_size; u32 task_id; u64 params[5]; - u8 data[]; + char data[]; }; static_assert(sizeof(HtcmiscRpcPacket) == 0x40); diff --git a/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_client.cpp b/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_client.cpp index f307c3d82..12c91ee74 100644 --- a/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_client.cpp +++ b/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_client.cpp @@ -154,12 +154,145 @@ namespace ams::htc::server::rpc { } } - void RpcClient::ReceiveThread() { - AMS_ABORT("RpcClient::ReceiveThread"); + Result RpcClient::ReceiveThread() { + /* Loop forever. */ + auto *header = reinterpret_cast(m_receive_buffer); + while (true) { + /* Try to receive a packet header. */ + R_TRY(this->ReceiveHeader(header)); + + /* Track how much we've received. */ + size_t received = sizeof(*header); + + /* If the packet has one, receive its body. */ + if (header->body_size > 0) { + /* Sanity check the task id. */ + AMS_ABORT_UNLESS(header->task_id < static_cast(MaxRpcCount)); + + /* Sanity check the body size. */ + AMS_ABORT_UNLESS(util::IsIntValueRepresentable(header->body_size)); + AMS_ABORT_UNLESS(static_cast(header->body_size) <= sizeof(m_receive_buffer) - received); + + /* Receive the body. */ + R_TRY(this->ReceiveBody(header->data, header->body_size)); + + /* Note that we received the body. */ + received += header->body_size; + } + + /* Acquire exclusive access to the task tables. */ + std::scoped_lock lk(m_mutex); + + /* Get the specified task. */ + Task *task = m_task_table.Get(header->task_id); + R_UNLESS(task != nullptr, htc::ResultInvalidTaskId()); + + /* If the task is canceled, free it. */ + if (task->GetTaskState() == RpcTaskState::Cancelled) { + m_task_active[header->task_id] = false; + m_task_table.Delete(header->task_id); + m_task_id_free_list.Free(header->task_id); + continue; + } + + /* Handle the packet. */ + switch (header->category) { + case PacketCategory::Response: + R_TRY(task->ProcessResponse(m_receive_buffer, received)); + break; + case PacketCategory::Notification: + R_TRY(task->ProcessNotification(m_receive_buffer, received)); + break; + default: + return htc::ResultInvalidCategory(); + } + + /* If we used the receive buffer, signal that we're done with it. */ + if (task->IsReceiveBufferRequired()) { + os::SignalEvent(std::addressof(m_receive_buffer_available_events[header->task_id])); + } + } } - void RpcClient::SendThread() { - AMS_ABORT("RpcClient::SendThread"); + Result RpcClient::ReceiveHeader(RpcPacket *header) { + /* Receive. */ + s64 received; + R_TRY(m_driver->Receive(std::addressof(received), reinterpret_cast(header), sizeof(*header), m_channel_id, htclow::ReceiveOption_ReceiveAllData)); + + /* Check size. */ + R_UNLESS(static_cast(received) == sizeof(*header), htc::ResultInvalidSize()); + + return ResultSuccess(); + } + + Result RpcClient::ReceiveBody(char *dst, size_t size) { + /* Receive. */ + s64 received; + R_TRY(m_driver->Receive(std::addressof(received), dst, size, m_channel_id, htclow::ReceiveOption_ReceiveAllData)); + + /* Check size. */ + R_UNLESS(static_cast(received) == size, htc::ResultInvalidSize()); + + return ResultSuccess(); + } + + Result RpcClient::SendThread() { + while (true) { + /* Get a task. */ + Task *task; + u32 task_id; + PacketCategory category; + do { + /* Dequeue a task. */ + R_TRY(m_task_queue.Take(std::addressof(task_id), std::addressof(category))); + + /* Get the task from the table. */ + std::scoped_lock lk(m_mutex); + + task = m_task_table.Get(task_id); + } while (task == nullptr); + + /* If required, wait for the send buffer to become available. */ + if (task->IsSendBufferRequired()) { + os::WaitEvent(std::addressof(m_send_buffer_available_events[task_id])); + + /* Check if we've been cancelled. */ + if (m_cancelled) { + break; + } + } + + /* Handle the task. */ + size_t packet_size; + switch (category) { + case PacketCategory::Request: + R_TRY(task->CreateRequest(std::addressof(packet_size), m_send_buffer, sizeof(m_send_buffer), task_id)); + break; + case PacketCategory::Notification: + R_TRY(task->CreateNotification(std::addressof(packet_size), m_send_buffer, sizeof(m_send_buffer), task_id)); + break; + AMS_UNREACHABLE_DEFAULT_CASE(); + } + + /* Send the request. */ + R_TRY(this->SendRequest(m_send_buffer, packet_size)); + } + + return htc::ResultCancelled(); + } + + Result RpcClient::SendRequest(const char *src, size_t size) { + /* Sanity check our size. */ + AMS_ASSERT(util::IsIntValueRepresentable(size)); + + /* Send the data. */ + s64 sent; + R_TRY(m_driver->Send(std::addressof(sent), src, static_cast(size), m_channel_id)); + + /* Check that we sent the right amount. */ + R_UNLESS(sent == static_cast(size), htc::ResultInvalidSize()); + + return ResultSuccess(); } } diff --git a/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_client.hpp b/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_client.hpp index e517025f6..5b7369b11 100644 --- a/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_client.hpp +++ b/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_client.hpp @@ -43,14 +43,14 @@ namespace ams::htc::server::rpc { bool m_thread_running; os::EventType m_receive_buffer_available_events[MaxRpcCount]; os::EventType m_send_buffer_available_events[MaxRpcCount]; - u8 m_receive_buffer[BufferSize]; - u8 m_send_buffer[BufferSize]; + char m_receive_buffer[BufferSize]; + char m_send_buffer[BufferSize]; private: static void ReceiveThreadEntry(void *arg) { static_cast(arg)->ReceiveThread(); } static void SendThreadEntry(void *arg) { static_cast(arg)->SendThread(); } - void ReceiveThread(); - void SendThread(); + Result ReceiveThread(); + Result SendThread(); public: RpcClient(driver::IDriver *driver, htclow::ChannelId channel); public: @@ -62,6 +62,10 @@ namespace ams::htc::server::rpc { void Wait(); int WaitAny(htclow::ChannelState state, os::EventType *event); + private: + Result ReceiveHeader(RpcPacket *header); + Result ReceiveBody(char *dst, size_t size); + Result SendRequest(const char *src, size_t size); }; } diff --git a/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_task_table.hpp b/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_task_table.hpp index ffb6a7c74..f283f9232 100644 --- a/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_task_table.hpp +++ b/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_task_table.hpp @@ -104,6 +104,12 @@ namespace ams::htc::server::rpc { /* Mark the task as invalid. */ m_valid[index] = false; } + + void Delete(u32 index) { + if (this->IsValid(index)) { + this->Delete(index); + } + } }; } diff --git a/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_tasks.hpp b/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_tasks.hpp index cd1358449..848c1ee3c 100644 --- a/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_tasks.hpp +++ b/libraries/libstratosphere/source/htc/server/rpc/htc_rpc_tasks.hpp @@ -34,7 +34,7 @@ namespace ams::htc::server::rpc { s64 body_size; u32 task_id; u64 params[5]; - u8 data[]; + char data[]; }; static_assert(sizeof(RpcPacket) == 0x40); diff --git a/libraries/libvapours/include/vapours/results/htc_results.hpp b/libraries/libvapours/include/vapours/results/htc_results.hpp index 917ff4e8e..3025aacbc 100644 --- a/libraries/libvapours/include/vapours/results/htc_results.hpp +++ b/libraries/libvapours/include/vapours/results/htc_results.hpp @@ -28,6 +28,10 @@ namespace ams::htc { R_DEFINE_ERROR_RESULT(Unknown, 1023); + R_DEFINE_ERROR_RESULT(InvalidTaskId, 2003); + R_DEFINE_ERROR_RESULT(InvalidSize, 2011); + R_DEFINE_ERROR_RESULT(OutOfRpcTask, 2102); + R_DEFINE_ERROR_RESULT(InvalidCategory, 2123); }