1 // Copyright 2023 The Pigweed Authors
2 //
3 // Licensed under the Apache License, Version 2.0 (the "License"); you may not
4 // use this file except in compliance with the License. You may obtain a copy of
5 // the License at
6 //
7 // https://www.apache.org/licenses/LICENSE-2.0
8 //
9 // Unless required by applicable law or agreed to in writing, software
10 // distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
11 // WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
12 // License for the specific language governing permissions and limitations under
13 // the License.
14 #pragma once
15
16 #include <atomic>
17 #include <cstddef>
18
19 #include "pw_bytes/span.h"
20 #include "pw_result/result.h"
21 #include "pw_rpc/channel.h"
22 #include "pw_rpc_transport/internal/packet_buffer_queue.h"
23 #include "pw_rpc_transport/rpc_transport.h"
24 #include "pw_status/status.h"
25 #include "pw_sync/thread_notification.h"
26 #include "pw_thread/thread_core.h"
27 #include "rpc_transport.h"
28
29 namespace pw::rpc {
30
31 namespace internal {
32 void LogNoRpcServiceRegistryError();
33 void LogPacketSizeTooLarge(size_t packet_size, size_t max_packet_size);
34 void LogEgressThreadNotRunningError();
35 void LogFailedToProcessPacket(Status status);
36 void LogFailedToAccessPacket(Status status);
37 void LogNoPacketAvailable(Status status);
38 } // namespace internal
39
40 // Handles RPC packets destined for the local receiver.
41 template <size_t kPacketQueueSize, size_t kMaxPacketSize>
42 class LocalRpcEgress : public RpcEgressHandler,
43 public ChannelOutput,
44 public thread::ThreadCore {
45 using PacketBuffer =
46 typename internal::PacketBufferQueue<kMaxPacketSize>::PacketBuffer;
47
48 public:
LocalRpcEgress()49 LocalRpcEgress() : ChannelOutput("RPC local egress") {}
~LocalRpcEgress()50 ~LocalRpcEgress() override { Stop(); }
51
52 // Packet processor cannot be passed as a construction dependency as it would
53 // create a circular dependency in the RPC transport configuration.
set_packet_processor(RpcPacketProcessor & packet_processor)54 void set_packet_processor(RpcPacketProcessor& packet_processor) {
55 packet_processor_ = &packet_processor;
56 }
57
58 // Adds the packet to the transmit queue. The queue is continuously processed
59 // by another thread. Implements RpcEgressHandler.
60 Status SendRpcPacket(ConstByteSpan rpc_packet) override;
61
62 // Implements ChannelOutput.
Send(ConstByteSpan buffer)63 Status Send(ConstByteSpan buffer) override { return SendRpcPacket(buffer); }
64
65 // Once stopped, LocalRpcEgress will no longer process data and
66 // will report errors on SendPacket().
Stop()67 void Stop() {
68 if (stopped_) {
69 return;
70 }
71 stopped_ = true;
72 // Unblock the processing thread and let it finish gracefully.
73 process_queue_.release();
74 }
75
76 private:
77 void Run() override;
78
79 sync::ThreadNotification process_queue_;
80 RpcPacketProcessor* packet_processor_ = nullptr;
81 std::array<PacketBuffer, kPacketQueueSize> packet_storage_;
82 internal::PacketBufferQueue<kMaxPacketSize> packet_queue_{packet_storage_};
83 internal::PacketBufferQueue<kMaxPacketSize> transmit_queue_ = {};
84 std::atomic<bool> stopped_ = false;
85 };
86
87 template <size_t kPacketQueueSize, size_t kMaxPacketSize>
SendRpcPacket(ConstByteSpan packet)88 Status LocalRpcEgress<kPacketQueueSize, kMaxPacketSize>::SendRpcPacket(
89 ConstByteSpan packet) {
90 if (!packet_processor_) {
91 internal::LogNoRpcServiceRegistryError();
92 return Status::FailedPrecondition();
93 }
94 if (packet.size() > kMaxPacketSize) {
95 internal::LogPacketSizeTooLarge(packet.size(), kMaxPacketSize);
96 return Status::InvalidArgument();
97 }
98 if (stopped_) {
99 internal::LogEgressThreadNotRunningError();
100 return Status::FailedPrecondition();
101 }
102
103 // Grab a free packet from the egress' pool, copy incoming frame and
104 // push it the queue for processing.
105 auto packet_buffer = packet_queue_.Pop();
106 if (!packet_buffer.ok()) {
107 internal::LogNoPacketAvailable(packet_buffer.status());
108 return packet_buffer.status();
109 }
110
111 PW_TRY(packet_buffer.value()->CopyPacket(packet));
112
113 transmit_queue_.Push(**packet_buffer);
114
115 process_queue_.release();
116
117 if (stopped_) {
118 internal::LogEgressThreadNotRunningError();
119 return Status::DataLoss();
120 }
121
122 return OkStatus();
123 }
124
125 template <size_t kPacketQueueSize, size_t kMaxPacketSize>
Run()126 void LocalRpcEgress<kPacketQueueSize, kMaxPacketSize>::Run() {
127 while (!stopped_) {
128 // Wait until a client has signaled that there is data in the packet queue.
129 process_queue_.acquire();
130
131 while (true) {
132 Result<PacketBuffer*> packet_buffer = transmit_queue_.Pop();
133 if (!packet_buffer.ok()) {
134 break;
135 }
136 Result<ConstByteSpan> packet = (*packet_buffer)->GetPacket();
137 if (packet.ok()) {
138 if (const auto status = packet_processor_->ProcessRpcPacket(*packet);
139 !status.ok()) {
140 internal::LogFailedToProcessPacket(status);
141 }
142 } else {
143 internal::LogFailedToAccessPacket(packet.status());
144 }
145 packet_queue_.Push(**packet_buffer);
146 }
147 }
148 }
149
150 } // namespace pw::rpc
151