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
28 namespace pw::rpc {
29
30 namespace internal {
31 void LogNoRpcServiceRegistryError();
32 void LogPacketSizeTooLarge(size_t packet_size, size_t max_packet_size);
33 void LogEgressThreadNotRunningError();
34 void LogFailedToProcessPacket(Status status);
35 void LogFailedToAccessPacket(Status status);
36 void LogNoPacketAvailable(Status status);
37 } // namespace internal
38
39 // Handles RPC packets destined for the local receiver.
40 template <size_t kPacketQueueSize, size_t kMaxPacketSize>
41 class LocalRpcEgress : public RpcEgressHandler,
42 public ChannelOutput,
43 public thread::ThreadCore {
44 using PacketBuffer =
45 typename internal::PacketBufferQueue<kMaxPacketSize>::PacketBuffer;
46
47 public:
LocalRpcEgress()48 LocalRpcEgress() : ChannelOutput("RPC local egress") {}
~LocalRpcEgress()49 ~LocalRpcEgress() override { Stop(); }
50
51 // Packet processor cannot be passed as a construction dependency as it would
52 // create a circular dependency in the RPC transport configuration.
set_packet_processor(RpcPacketProcessor & packet_processor)53 void set_packet_processor(RpcPacketProcessor& packet_processor) {
54 packet_processor_ = &packet_processor;
55 }
56
57 // Adds the packet to the transmit queue. The queue is continuously processed
58 // by another thread. Implements RpcEgressHandler.
59 Status SendRpcPacket(ConstByteSpan rpc_packet) override;
60
61 // Implements ChannelOutput.
Send(ConstByteSpan buffer)62 Status Send(ConstByteSpan buffer) override { return SendRpcPacket(buffer); }
63
64 // Once stopped, LocalRpcEgress will no longer process data and
65 // will report errors on SendPacket().
Stop()66 void Stop() {
67 if (stopped_) {
68 return;
69 }
70 stopped_ = true;
71 // Unblock the processing thread and let it finish gracefully.
72 process_queue_.release();
73 }
74
75 private:
76 void Run() override;
77
78 sync::ThreadNotification process_queue_;
79 RpcPacketProcessor* packet_processor_ = nullptr;
80 std::array<PacketBuffer, kPacketQueueSize> packet_storage_;
81 internal::PacketBufferQueue<kMaxPacketSize> packet_queue_{packet_storage_};
82 internal::PacketBufferQueue<kMaxPacketSize> transmit_queue_ = {};
83 std::atomic<bool> stopped_ = false;
84 };
85
86 template <size_t kPacketQueueSize, size_t kMaxPacketSize>
SendRpcPacket(ConstByteSpan packet)87 Status LocalRpcEgress<kPacketQueueSize, kMaxPacketSize>::SendRpcPacket(
88 ConstByteSpan packet) {
89 if (!packet_processor_) {
90 internal::LogNoRpcServiceRegistryError();
91 return Status::FailedPrecondition();
92 }
93 if (packet.size() > kMaxPacketSize) {
94 internal::LogPacketSizeTooLarge(packet.size(), kMaxPacketSize);
95 return Status::InvalidArgument();
96 }
97 if (stopped_) {
98 internal::LogEgressThreadNotRunningError();
99 return Status::FailedPrecondition();
100 }
101
102 // Grab a free packet from the egress' pool, copy incoming frame and
103 // push it the queue for processing.
104 auto packet_buffer = packet_queue_.Pop();
105 if (!packet_buffer.ok()) {
106 internal::LogNoPacketAvailable(packet_buffer.status());
107 return packet_buffer.status();
108 }
109
110 PW_TRY(packet_buffer.value()->CopyPacket(packet));
111
112 transmit_queue_.Push(**packet_buffer);
113
114 process_queue_.release();
115
116 if (stopped_) {
117 internal::LogEgressThreadNotRunningError();
118 return Status::DataLoss();
119 }
120
121 return OkStatus();
122 }
123
124 template <size_t kPacketQueueSize, size_t kMaxPacketSize>
Run()125 void LocalRpcEgress<kPacketQueueSize, kMaxPacketSize>::Run() {
126 while (!stopped_) {
127 // Wait until a client has signaled that there is data in the packet queue.
128 process_queue_.acquire();
129
130 while (true) {
131 Result<PacketBuffer*> packet_buffer = transmit_queue_.Pop();
132 if (!packet_buffer.ok()) {
133 break;
134 }
135 Result<ConstByteSpan> packet = (*packet_buffer)->GetPacket();
136 if (packet.ok()) {
137 if (const auto status = packet_processor_->ProcessRpcPacket(*packet);
138 !status.ok()) {
139 internal::LogFailedToProcessPacket(status);
140 }
141 } else {
142 internal::LogFailedToAccessPacket(packet.status());
143 }
144 packet_queue_.Push(**packet_buffer);
145 }
146 }
147 }
148
149 } // namespace pw::rpc
150