htc: Implement RpcClient::ReceiveThread + SendThread

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<RpcPacket *>(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<int>(MaxRpcCount));
+
+                /* Sanity check the body size. */
+                AMS_ABORT_UNLESS(util::IsIntValueRepresentable<size_t>(header->body_size));
+                AMS_ABORT_UNLESS(static_cast<size_t>(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<Task>(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<char *>(header), sizeof(*header), m_channel_id, htclow::ReceiveOption_ReceiveAllData));
+
+        /* Check size. */
+        R_UNLESS(static_cast<size_t>(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<size_t>(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>(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<s64>(size));
+
+        /* Send the data. */
+        s64 sent;
+        R_TRY(m_driver->Send(std::addressof(sent), src, static_cast<s64>(size), m_channel_id));
+
+        /* Check that we sent the right amount. */
+        R_UNLESS(sent == static_cast<s64>(size), htc::ResultInvalidSize());
+
+        return ResultSuccess();
     }
 
 }