1 // Copyright 2024 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
15 // Client binary for the cross-language integration test.
16 //
17 // Usage:
18 // bazel-bin/pw_transfer/integration_test_client 3300 <<< "resource_id: 12
19 // file: '/tmp/myfile.txt'"
20 //
21 // WORK IN PROGRESS, SEE b/228516801
22 #include "pw_transfer/client.h"
23
24 #include <sys/socket.h>
25
26 #include <cstddef>
27 #include <cstdio>
28
29 #include "google/protobuf/text_format.h"
30 #include "pw_assert/check.h"
31 #include "pw_log/log.h"
32 #include "pw_rpc/channel.h"
33 #include "pw_rpc/integration_testing.h"
34 #include "pw_status/status.h"
35 #include "pw_status/try.h"
36 #include "pw_stream/std_file_stream.h"
37 #include "pw_sync/binary_semaphore.h"
38 #include "pw_thread/thread.h"
39 #include "pw_thread_stl/options.h"
40 #include "pw_transfer/integration_test/config.pb.h"
41 #include "pw_transfer/transfer_thread.h"
42
43 namespace pw::transfer::integration_test {
44 namespace {
45
46 // This is the maximum size of the socket send buffers. Ideally, this is set
47 // to the lowest allowed value to minimize buffering between the proxy and
48 // clients so rate limiting causes the client to block and wait for the
49 // integration test proxy to drain rather than allowing OS buffers to backlog
50 // large quantities of data.
51 //
52 // Note that the OS may chose to not strictly follow this requested buffer size.
53 // Still, setting this value to be as small as possible does reduce bufer sizes
54 // significantly enough to better reflect typical inter-device communication.
55 //
56 // For this to be effective, servers should also configure their sockets to a
57 // smaller receive buffer size.
58 constexpr int kMaxSocketSendBufferSize = 1;
59
60 constexpr size_t kDefaultMaxWindowSizeBytes = 16384;
61
TransferThreadOptions()62 thread::Options& TransferThreadOptions() {
63 static thread::stl::Options options;
64 return options;
65 }
66
67 // Transfer status, valid only after semaphore is acquired.
68 //
69 // We need to bundle the status and semaphore together because a pw_function
70 // callback can at most capture the reference to one variable (and we need to
71 // both set the status and release the semaphore).
72 struct TransferResult {
73 Status status = Status::Unknown();
74 sync::BinarySemaphore completed;
75 };
76
77 // Create a pw_transfer client and perform the transfer actions.
PerformTransferActions(const pw::transfer::ClientConfig & config)78 pw::Status PerformTransferActions(const pw::transfer::ClientConfig& config) {
79 constexpr size_t kMaxPayloadSize = rpc::MaxSafePayloadSize();
80 std::byte chunk_buffer[kMaxPayloadSize];
81 std::byte encode_buffer[kMaxPayloadSize];
82 transfer::Thread<2, 2> transfer_thread(chunk_buffer, encode_buffer);
83 pw::Thread system_thread(TransferThreadOptions(), transfer_thread);
84
85 // As much as we don't want to dynamically allocate an array,
86 // variable length arrays (VLA) are nonstandard, and a std::vector could cause
87 // references to go stale if the vector's underlying buffer is resized. This
88 // array of TransferResults needs to outlive the loop that performs the
89 // actual transfer actions due to how some references to TransferResult
90 // may persist beyond the lifetime of a transfer.
91 const int num_actions = config.transfer_actions().size();
92 auto transfer_results = std::make_unique<TransferResult[]>(num_actions);
93
94 pw::transfer::Client client(rpc::integration_test::client(),
95 rpc::integration_test::kChannelId,
96 transfer_thread,
97 kDefaultMaxWindowSizeBytes);
98
99 // TODO: https://pwbug.dev/357145010 - Don't IgnoreError here.
100 client.set_max_retries(config.max_retries()).IgnoreError();
101 client.set_max_lifetime_retries(config.max_lifetime_retries()).IgnoreError();
102
103 Status status = pw::OkStatus();
104 for (int i = 0; i < num_actions; i++) {
105 const pw::transfer::TransferAction& action = config.transfer_actions()[i];
106 TransferResult& result = transfer_results[i];
107 // If no protocol version is specified, default to the latest version.
108 pw::transfer::ProtocolVersion protocol_version =
109 action.protocol_version() ==
110 pw::transfer::TransferAction::ProtocolVersion::
111 TransferAction_ProtocolVersion_UNKNOWN_VERSION
112 ? pw::transfer::ProtocolVersion::kLatest
113 : static_cast<pw::transfer::ProtocolVersion>(
114 action.protocol_version());
115 if (action.transfer_type() ==
116 pw::transfer::TransferAction::TransferType::
117 TransferAction_TransferType_WRITE_TO_SERVER) {
118 pw::stream::StdFileReader input(action.file_path().c_str());
119 pw::Result<pw::transfer::Client::Handle> handle = client.Write(
120 action.resource_id(),
121 input,
122 [&result](Status status) {
123 result.status = status;
124 result.completed.release();
125 },
126 protocol_version,
127 pw::transfer::cfg::kDefaultClientTimeout,
128 pw::transfer::cfg::kDefaultInitialChunkTimeout,
129 action.initial_offset());
130 if (handle.ok()) {
131 // Wait for the transfer to complete. We need to do this here so that
132 // the StdFileReader doesn't go out of scope.
133 result.completed.acquire();
134 } else {
135 result.status = handle.status();
136 }
137
138 input.Close();
139
140 } else if (action.transfer_type() ==
141 pw::transfer::TransferAction::TransferType::
142 TransferAction_TransferType_READ_FROM_SERVER) {
143 pw::stream::StdFileWriter output(action.file_path().c_str());
144 pw::Result<pw::transfer::Client::Handle> handle = client.Read(
145 action.resource_id(),
146 output,
147 [&result](Status status) {
148 result.status = status;
149 result.completed.release();
150 },
151 protocol_version,
152 pw::transfer::cfg::kDefaultClientTimeout,
153 pw::transfer::cfg::kDefaultInitialChunkTimeout,
154 action.initial_offset());
155 if (handle.ok()) {
156 // Wait for the transfer to complete.
157 result.completed.acquire();
158 } else {
159 result.status = handle.status();
160 }
161
162 output.Close();
163 } else {
164 PW_LOG_ERROR("Unrecognized transfer action type %d",
165 action.transfer_type());
166 status = pw::Status::InvalidArgument();
167 break;
168 }
169
170 if (int(result.status.code()) != int(action.expected_status())) {
171 PW_LOG_ERROR("Failed to perform action:\n%s",
172 action.DebugString().c_str());
173 status = result.status.ok() ? Status::Unknown() : result.status;
174 break;
175 }
176 }
177
178 transfer_thread.Terminate();
179
180 system_thread.join();
181
182 // The RPC thread must join before destroying transfer objects as the transfer
183 // service may still reference the transfer thread or transfer client objects.
184 pw::rpc::integration_test::TerminateClient();
185 return status;
186 }
187
188 } // namespace
189 } // namespace pw::transfer::integration_test
190
main(int argc,char * argv[])191 int main(int argc, char* argv[]) {
192 if (argc < 2) {
193 PW_LOG_INFO("Usage: %s PORT <<< config textproto", argv[0]);
194 return 1;
195 }
196
197 const int port = std::atoi(argv[1]);
198
199 std::string config_string;
200 std::string line;
201 while (std::getline(std::cin, line)) {
202 config_string = config_string + line + '\n';
203 }
204 pw::transfer::ClientConfig config;
205
206 bool ok =
207 google::protobuf::TextFormat::ParseFromString(config_string, &config);
208 if (!ok) {
209 PW_LOG_INFO("Failed to parse config: %s", config_string.c_str());
210 PW_LOG_INFO("Usage: %s PORT <<< config textproto", argv[0]);
211 return 1;
212 } else {
213 PW_LOG_INFO("Client loaded config:\n%s", config.DebugString().c_str());
214 }
215
216 if (!pw::rpc::integration_test::InitializeClient(port).ok()) {
217 return 1;
218 }
219
220 int retval = pw::rpc::integration_test::SetClientSockOpt(
221 SOL_SOCKET,
222 SO_SNDBUF,
223 &pw::transfer::integration_test::kMaxSocketSendBufferSize,
224 sizeof(pw::transfer::integration_test::kMaxSocketSendBufferSize));
225 PW_CHECK_INT_EQ(retval,
226 0,
227 "Failed to configure socket send buffer size with errno=%d",
228 errno);
229
230 if (!pw::transfer::integration_test::PerformTransferActions(config).ok()) {
231 PW_LOG_INFO("Failed to transfer!");
232 return 1;
233 }
234 return 0;
235 }
236