blob: 9632d3c5e607f26bf596a9661bc6fae5c01d6cd5 [file] [log] [blame]
Austin Schuhe84c3ed2019-12-14 15:29:48 -08001#include "aos/network/sctp_lib.h"
2
3#include <arpa/inet.h>
4#include <net/if.h>
5#include <netdb.h>
6#include <netinet/sctp.h>
Austin Schuh2fe4b712020-03-15 14:21:45 -07007#include <sys/stat.h>
8#include <sys/types.h>
9#include <unistd.h>
Austin Schuhe84c3ed2019-12-14 15:29:48 -080010
Austin Schuha705d782021-07-31 20:40:00 -070011#include <algorithm>
Austin Schuhe84c3ed2019-12-14 15:29:48 -080012#include <string_view>
13
Austin Schuh2fe4b712020-03-15 14:21:45 -070014#include "aos/util/file.h"
15
Austin Schuh0a0a8272021-12-08 13:19:32 -080016DEFINE_string(interface, "", "network interface");
Brian Silverman0c6d44e2021-11-10 12:27:49 -080017DEFINE_bool(disable_ipv6, false, "disable ipv6");
Austin Schuh9dd8f592021-12-25 14:32:43 -080018DEFINE_int32(rmem, 0, "If nonzero, set rmem to this size.");
Austin Schuhe84c3ed2019-12-14 15:29:48 -080019
20namespace aos {
21namespace message_bridge {
22
23namespace {
24const char *sac_state_tbl[] = {"COMMUNICATION_UP", "COMMUNICATION_LOST",
25 "RESTART", "SHUTDOWN_COMPLETE",
Sarah Newman4aeb2372022-04-06 13:07:11 -070026 "CANT_START_ASSOCIATION"};
Austin Schuhe84c3ed2019-12-14 15:29:48 -080027
28typedef union {
29 struct sctp_initmsg init;
30 struct sctp_sndrcvinfo sndrcvinfo;
31} _sctp_cmsg_data_t;
32
33} // namespace
34
Austin Schuh0a0a8272021-12-08 13:19:32 -080035bool Ipv6Enabled() {
36 if (FLAGS_disable_ipv6) {
37 return false;
38 }
39 int fd = socket(AF_INET6, SOCK_SEQPACKET, IPPROTO_SCTP);
40 if (fd != -1) {
41 close(fd);
42 return true;
43 }
44 switch (errno) {
45 case EAFNOSUPPORT:
46 case EINVAL:
47 case EPROTONOSUPPORT:
48 PLOG(INFO) << "no ipv6";
49 return false;
50 default:
51 PLOG(FATAL) << "Open socket failed";
52 return false;
53 };
54}
55
56struct sockaddr_storage ResolveSocket(std::string_view host, int port,
57 bool use_ipv6) {
Austin Schuhe84c3ed2019-12-14 15:29:48 -080058 struct sockaddr_storage result;
James Kuszmaul784deb72023-02-17 14:42:51 -080059 memset(&result, 0, sizeof(result));
Austin Schuhe84c3ed2019-12-14 15:29:48 -080060 struct addrinfo *addrinfo_result;
61 struct sockaddr_in *t_addr = (struct sockaddr_in *)&result;
62 struct sockaddr_in6 *t_addr6 = (struct sockaddr_in6 *)&result;
Brian Silverman0c6d44e2021-11-10 12:27:49 -080063 struct addrinfo hints;
64 memset(&hints, 0, sizeof(hints));
Austin Schuh0a0a8272021-12-08 13:19:32 -080065 if (!use_ipv6) {
Brian Silverman0c6d44e2021-11-10 12:27:49 -080066 hints.ai_family = AF_INET;
67 } else {
Austin Schuh0a0a8272021-12-08 13:19:32 -080068 // Default to IPv6 as the clearly superior protocol, since it also handles
69 // IPv4.
Brian Silverman0c6d44e2021-11-10 12:27:49 -080070 hints.ai_family = AF_INET6;
71 }
72 hints.ai_socktype = SOCK_SEQPACKET;
73 hints.ai_protocol = IPPROTO_SCTP;
74 // We deliberately avoid AI_ADDRCONFIG here because it breaks running things
75 // inside Bazel's test sandbox, which has no non-localhost IPv4 or IPv6
76 // addresses. Also, it's not really helpful, because most systems will have
77 // link-local addresses of both types with any interface that's up.
78 hints.ai_flags = AI_PASSIVE | AI_V4MAPPED | AI_NUMERICSERV;
79 int ret = getaddrinfo(host.empty() ? nullptr : std::string(host).c_str(),
80 std::to_string(port).c_str(), &hints, &addrinfo_result);
Brian Silverman0c6d44e2021-11-10 12:27:49 -080081 if (ret == EAI_SYSTEM) {
82 PLOG(FATAL) << "getaddrinfo failed to look up '" << host << "'";
83 } else if (ret != 0) {
84 LOG(FATAL) << "getaddrinfo failed to look up '" << host
85 << "': " << gai_strerror(ret);
86 }
Austin Schuhe84c3ed2019-12-14 15:29:48 -080087 switch (addrinfo_result->ai_family) {
88 case AF_INET:
89 memcpy(t_addr, addrinfo_result->ai_addr, addrinfo_result->ai_addrlen);
90 t_addr->sin_family = addrinfo_result->ai_family;
91 t_addr->sin_port = htons(port);
92
93 break;
94 case AF_INET6:
95 memcpy(t_addr6, addrinfo_result->ai_addr, addrinfo_result->ai_addrlen);
96 t_addr6->sin6_family = addrinfo_result->ai_family;
97 t_addr6->sin6_port = htons(port);
98
99 if (FLAGS_interface.size() > 0) {
100 t_addr6->sin6_scope_id = if_nametoindex(FLAGS_interface.c_str());
101 }
102
103 break;
104 }
105
106 // Now print it back out nicely.
107 char host_string[NI_MAXHOST];
108 char service_string[NI_MAXSERV];
109
110 int error = getnameinfo((struct sockaddr *)&result,
111 addrinfo_result->ai_addrlen, host_string, NI_MAXHOST,
112 service_string, NI_MAXSERV, NI_NUMERICHOST);
113
114 if (error) {
115 LOG(ERROR) << "Reverse lookup failed ... " << gai_strerror(error);
116 }
117
118 LOG(INFO) << "remote:addr=" << host_string << ", port=" << service_string
119 << ", family=" << addrinfo_result->ai_family;
120
121 freeaddrinfo(addrinfo_result);
122
123 return result;
124}
125
126std::string_view Family(const struct sockaddr_storage &sockaddr) {
127 if (sockaddr.ss_family == AF_INET) {
128 return "AF_INET";
129 } else if (sockaddr.ss_family == AF_INET6) {
130 return "AF_INET6";
131 } else {
132 return "unknown";
133 }
134}
135std::string Address(const struct sockaddr_storage &sockaddr) {
136 char addrbuf[INET6_ADDRSTRLEN];
137 if (sockaddr.ss_family == AF_INET) {
138 const struct sockaddr_in *sin = (const struct sockaddr_in *)&sockaddr;
139 return std::string(
140 inet_ntop(AF_INET, &sin->sin_addr, addrbuf, INET6_ADDRSTRLEN));
141 } else {
142 const struct sockaddr_in6 *sin6 = (const struct sockaddr_in6 *)&sockaddr;
143 return std::string(
144 inet_ntop(AF_INET6, &sin6->sin6_addr, addrbuf, INET6_ADDRSTRLEN));
145 }
146}
147
148void PrintNotification(const Message *msg) {
149 const union sctp_notification *snp =
150 (const union sctp_notification *)msg->data();
151
152 LOG(INFO) << "Notification:";
153
154 switch (snp->sn_header.sn_type) {
155 case SCTP_ASSOC_CHANGE: {
156 const struct sctp_assoc_change *sac = &snp->sn_assoc_change;
157 LOG(INFO) << "SCTP_ASSOC_CHANGE(" << sac_state_tbl[sac->sac_state] << ")";
158 VLOG(1) << " (assoc_change: state=" << sac->sac_state
159 << ", error=" << sac->sac_error
160 << ", instr=" << sac->sac_inbound_streams
161 << " outstr=" << sac->sac_outbound_streams
162 << ", assoc=" << sac->sac_assoc_id << ")";
163 } break;
164 case SCTP_PEER_ADDR_CHANGE: {
165 const struct sctp_paddr_change *spc = &snp->sn_paddr_change;
166 LOG(INFO) << " SlCTP_PEER_ADDR_CHANGE";
167 VLOG(1) << "\t\t(peer_addr_change: " << Address(spc->spc_aaddr)
168 << " state=" << spc->spc_state << ", error=" << spc->spc_error
169 << ")";
170 } break;
171 case SCTP_SEND_FAILED: {
172 const struct sctp_send_failed *ssf = &snp->sn_send_failed;
173 LOG(INFO) << " SCTP_SEND_FAILED";
174 VLOG(1) << "\t\t(sendfailed: len=" << ssf->ssf_length
175 << " err=" << ssf->ssf_error << ")";
176 } break;
177 case SCTP_REMOTE_ERROR: {
178 const struct sctp_remote_error *sre = &snp->sn_remote_error;
179 LOG(INFO) << " SCTP_REMOTE_ERROR";
180 VLOG(1) << "\t\t(remote_error: err=" << ntohs(sre->sre_error) << ")";
181 } break;
Austin Schuhf7777002020-09-01 18:41:28 -0700182 case SCTP_STREAM_CHANGE_EVENT: {
Austin Schuh62a0c272021-03-31 21:04:53 -0700183 const struct sctp_stream_change_event *sce = &snp->sn_strchange_event;
Austin Schuhf7777002020-09-01 18:41:28 -0700184 LOG(INFO) << " SCTP_STREAM_CHANGE_EVENT";
185 VLOG(1) << "\t\t(stream_change_event: flags=" << sce->strchange_flags
186 << ", assoc_id=" << sce->strchange_assoc_id
187 << ", instrms=" << sce->strchange_instrms
188 << ", outstrms=" << sce->strchange_outstrms << " )";
189 } break;
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800190 case SCTP_SHUTDOWN_EVENT: {
191 LOG(INFO) << " SCTP_SHUTDOWN_EVENT";
192 } break;
193 default:
194 LOG(INFO) << " Unknown type: " << snp->sn_header.sn_type;
195 break;
196 }
197}
198
199std::string GetHostname() {
200 char buf[256];
201 buf[sizeof(buf) - 1] = '\0';
202 PCHECK(gethostname(buf, sizeof(buf) - 1) == 0);
203 return buf;
204}
205
206std::string Message::PeerAddress() const { return Address(sin); }
207
208void LogSctpStatus(int fd, sctp_assoc_t assoc_id) {
209 struct sctp_status status;
210 memset(&status, 0, sizeof(status));
211 status.sstat_assoc_id = assoc_id;
212
213 socklen_t size = sizeof(status);
Austin Schuha5f545b2021-07-31 20:39:42 -0700214 const int result = getsockopt(fd, SOL_SCTP, SCTP_STATUS,
215 reinterpret_cast<void *>(&status), &size);
216 if (result == -1 && errno == EINVAL) {
217 LOG(INFO) << "sctp_status) not associated";
218 return;
219 }
220 PCHECK(result == 0);
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800221
222 LOG(INFO) << "sctp_status) sstat_assoc_id:" << status.sstat_assoc_id
223 << " sstat_state:" << status.sstat_state
224 << " sstat_rwnd:" << status.sstat_rwnd
225 << " sstat_unackdata:" << status.sstat_unackdata
226 << " sstat_penddata:" << status.sstat_penddata
227 << " sstat_instrms:" << status.sstat_instrms
228 << " sstat_outstrms:" << status.sstat_outstrms
229 << " sstat_fragmentation_point:" << status.sstat_fragmentation_point
230 << " sstat_primary.spinfo_srtt:" << status.sstat_primary.spinfo_srtt
231 << " sstat_primary.spinfo_rto:" << status.sstat_primary.spinfo_rto;
232}
233
Austin Schuh507f7582021-07-31 20:39:55 -0700234void SctpReadWrite::OpenSocket(const struct sockaddr_storage &sockaddr_local) {
235 fd_ = socket(sockaddr_local.ss_family, SOCK_SEQPACKET, IPPROTO_SCTP);
236 PCHECK(fd_ != -1);
237 LOG(INFO) << "socket(" << Family(sockaddr_local)
238 << ", SOCK_SEQPACKET, IPPROTOSCTP) = " << fd_;
239 {
240 // Per https://tools.ietf.org/html/rfc6458
241 // Setting this to !0 allows event notifications to be interleaved
Austin Schuha705d782021-07-31 20:40:00 -0700242 // with data if enabled. This typically only matters during congestion.
243 // However, Linux seems to interleave under memory pressure regardless of
244 // this being enabled, so we have to handle it in the code anyways, so might
245 // as well turn it on all the time.
246 // TODO(Brian): Change this to 2 once we have kernels that support it, and
247 // also address the TODO in ProcessNotification to match on all the
248 // necessary fields.
249 int interleaving = 1;
Austin Schuh507f7582021-07-31 20:39:55 -0700250 PCHECK(setsockopt(fd_, IPPROTO_SCTP, SCTP_FRAGMENT_INTERLEAVE,
251 &interleaving, sizeof(interleaving)) == 0);
252 }
253 {
254 // Enable recvinfo when a packet arrives.
255 int on = 1;
256 PCHECK(setsockopt(fd_, IPPROTO_SCTP, SCTP_RECVRCVINFO, &on, sizeof(int)) ==
257 0);
258 }
259
Austin Schuha705d782021-07-31 20:40:00 -0700260 {
261 // TODO(austin): This is the old style registration... But, the sctp
262 // stack out in the wild for linux is old and primitive.
263 struct sctp_event_subscribe subscribe;
264 memset(&subscribe, 0, sizeof(subscribe));
265 subscribe.sctp_association_event = 1;
266 subscribe.sctp_stream_change_event = 1;
267 subscribe.sctp_partial_delivery_event = 1;
268 PCHECK(setsockopt(fd(), SOL_SCTP, SCTP_EVENTS, (char *)&subscribe,
269 sizeof(subscribe)) == 0);
270 }
271
Austin Schuh507f7582021-07-31 20:39:55 -0700272 DoSetMaxSize();
273}
274
275bool SctpReadWrite::SendMessage(
276 int stream, std::string_view data, int time_to_live,
277 std::optional<struct sockaddr_storage> sockaddr_remote,
278 sctp_assoc_t snd_assoc_id) {
279 CHECK(fd_ != -1);
280 struct iovec iov;
281 iov.iov_base = const_cast<char *>(data.data());
282 iov.iov_len = data.size();
283
284 // Use the assoc_id for the destination instead of the msg_name.
285 struct msghdr outmsg;
James Kuszmaul784deb72023-02-17 14:42:51 -0800286 memset(&outmsg, 0, sizeof(outmsg));
Austin Schuh507f7582021-07-31 20:39:55 -0700287 if (sockaddr_remote) {
288 outmsg.msg_name = &*sockaddr_remote;
289 outmsg.msg_namelen = sizeof(*sockaddr_remote);
Sarah Newman4aeb2372022-04-06 13:07:11 -0700290 VLOG(2) << "Sending to " << Address(*sockaddr_remote);
Austin Schuh507f7582021-07-31 20:39:55 -0700291 } else {
292 outmsg.msg_namelen = 0;
293 }
294
295 // Data to send.
296 outmsg.msg_iov = &iov;
297 outmsg.msg_iovlen = 1;
298
299 // Build up the sndinfo message.
300 char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
301 outmsg.msg_control = outcmsg;
302 outmsg.msg_controllen = sizeof(outcmsg);
303 outmsg.msg_flags = 0;
304
305 struct cmsghdr *cmsg = CMSG_FIRSTHDR(&outmsg);
306 cmsg->cmsg_level = IPPROTO_SCTP;
307 cmsg->cmsg_type = SCTP_SNDRCV;
308 cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
309
310 struct sctp_sndrcvinfo *sinfo =
311 reinterpret_cast<struct sctp_sndrcvinfo *>(CMSG_DATA(cmsg));
312 memset(sinfo, 0, sizeof(struct sctp_sndrcvinfo));
313 sinfo->sinfo_ppid = ++send_ppid_;
314 sinfo->sinfo_stream = stream;
315 sinfo->sinfo_flags = 0;
316 sinfo->sinfo_assoc_id = snd_assoc_id;
317 sinfo->sinfo_timetolive = time_to_live;
318
319 // And send.
320 const ssize_t size = sendmsg(fd_, &outmsg, MSG_NOSIGNAL | MSG_DONTWAIT);
321 if (size == -1) {
322 if (errno == EPIPE || errno == EAGAIN || errno == ESHUTDOWN ||
323 errno == EINTR) {
Austin Schuh581fab92022-02-07 19:50:54 -0800324 if (VLOG_IS_ON(1)) {
325 PLOG(WARNING) << "sendmsg on sctp socket failed";
326 }
Austin Schuh507f7582021-07-31 20:39:55 -0700327 return false;
328 }
329 PLOG(FATAL) << "sendmsg on sctp socket failed";
330 return false;
331 }
332 CHECK_EQ(static_cast<ssize_t>(data.size()), size);
Sarah Newman4aeb2372022-04-06 13:07:11 -0700333 VLOG(2) << "Sent " << data.size();
Austin Schuh507f7582021-07-31 20:39:55 -0700334 return true;
335}
336
Austin Schuha705d782021-07-31 20:40:00 -0700337// We read each fragment into a fresh Message, because most of them won't be
338// fragmented. If we do end up with a fragment, then we copy the data out of it.
Austin Schuh507f7582021-07-31 20:39:55 -0700339aos::unique_c_ptr<Message> SctpReadWrite::ReadMessage() {
340 CHECK(fd_ != -1);
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800341
Austin Schuha705d782021-07-31 20:40:00 -0700342 while (true) {
James Kuszmaul6e622382023-02-17 14:56:38 -0800343 constexpr size_t kMessageAlign = alignof(Message);
344 const size_t max_message_size =
345 ((sizeof(Message) + max_size_ + 1 + (kMessageAlign - 1)) /
346 kMessageAlign) *
347 kMessageAlign;
348 aos::unique_c_ptr<Message> result(reinterpret_cast<Message *>(
349 aligned_alloc(kMessageAlign, max_message_size)));
Austin Schuha705d782021-07-31 20:40:00 -0700350
Austin Schuh05c18122021-07-31 20:39:47 -0700351 struct msghdr inmessage;
Austin Schuhc4202572021-03-31 21:06:55 -0700352 memset(&inmessage, 0, sizeof(struct msghdr));
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800353
Austin Schuh05c18122021-07-31 20:39:47 -0700354 struct iovec iov;
Austin Schuha705d782021-07-31 20:40:00 -0700355 iov.iov_len = max_size_ + 1;
356 iov.iov_base = result->mutable_data();
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800357
Austin Schuhc4202572021-03-31 21:06:55 -0700358 inmessage.msg_iov = &iov;
359 inmessage.msg_iovlen = 1;
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800360
Austin Schuh05c18122021-07-31 20:39:47 -0700361 char incmsg[CMSG_SPACE(sizeof(_sctp_cmsg_data_t))];
Austin Schuhc4202572021-03-31 21:06:55 -0700362 inmessage.msg_control = incmsg;
363 inmessage.msg_controllen = sizeof(incmsg);
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800364
Austin Schuhc4202572021-03-31 21:06:55 -0700365 inmessage.msg_namelen = sizeof(struct sockaddr_storage);
366 inmessage.msg_name = &result->sin;
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800367
Austin Schuha705d782021-07-31 20:40:00 -0700368 const ssize_t size = recvmsg(fd_, &inmessage, MSG_DONTWAIT);
369 if (size == -1) {
370 if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
371 // These are all non-fatal failures indicating we should retry later.
372 return nullptr;
Austin Schuhc4202572021-03-31 21:06:55 -0700373 }
Austin Schuha705d782021-07-31 20:40:00 -0700374 PLOG(FATAL) << "recvmsg on sctp socket " << fd_ << " failed";
Austin Schuhc4202572021-03-31 21:06:55 -0700375 }
376
Austin Schuha705d782021-07-31 20:40:00 -0700377 CHECK(!(inmessage.msg_flags & MSG_CTRUNC))
Austin Schuhc4202572021-03-31 21:06:55 -0700378 << ": Control message truncated.";
379
Austin Schuha705d782021-07-31 20:40:00 -0700380 CHECK_LE(size, static_cast<ssize_t>(max_size_))
Austin Schuh507f7582021-07-31 20:39:55 -0700381 << ": Message overflowed buffer on stream "
382 << result->header.rcvinfo.rcv_sid << ".";
Austin Schuhc4202572021-03-31 21:06:55 -0700383
Austin Schuha705d782021-07-31 20:40:00 -0700384 result->size = size;
385 if (MSG_NOTIFICATION & inmessage.msg_flags) {
386 result->message_type = Message::kNotification;
387 } else {
388 result->message_type = Message::kMessage;
389 }
390 result->partial_deliveries = 0;
Austin Schuhc4202572021-03-31 21:06:55 -0700391
Austin Schuha705d782021-07-31 20:40:00 -0700392 {
393 bool found_rcvinfo = false;
394 for (struct cmsghdr *scmsg = CMSG_FIRSTHDR(&inmessage); scmsg != NULL;
395 scmsg = CMSG_NXTHDR(&inmessage, scmsg)) {
396 switch (scmsg->cmsg_type) {
397 case SCTP_RCVINFO: {
398 CHECK(!found_rcvinfo);
399 found_rcvinfo = true;
400 result->header.rcvinfo =
401 *reinterpret_cast<struct sctp_rcvinfo *>(CMSG_DATA(scmsg));
402 } break;
403 default:
404 LOG(INFO) << "\tUnknown type: " << scmsg->cmsg_type;
405 break;
406 }
407 }
408 CHECK_EQ(found_rcvinfo, result->message_type == Message::kMessage)
409 << ": Failed to find a SCTP_RCVINFO cmsghdr. flags: "
410 << inmessage.msg_flags;
411 }
412 if (result->message_type == Message::kNotification) {
413 // Notifications are never fragmented, just return it now.
414 CHECK(inmessage.msg_flags & MSG_EOR)
415 << ": Notifications should never be big enough to fragment";
416 if (ProcessNotification(result.get())) {
417 // We handled this notification internally, so don't pass it on.
418 return nullptr;
419 }
420 return result;
421 }
422
423 auto partial_message_iterator =
424 std::find_if(partial_messages_.begin(), partial_messages_.end(),
425 [&result](const aos::unique_c_ptr<Message> &candidate) {
426 return result->header.rcvinfo.rcv_sid ==
427 candidate->header.rcvinfo.rcv_sid &&
428 result->header.rcvinfo.rcv_ssn ==
429 candidate->header.rcvinfo.rcv_ssn &&
430 result->header.rcvinfo.rcv_assoc_id ==
431 candidate->header.rcvinfo.rcv_assoc_id;
432 });
433 if (partial_message_iterator != partial_messages_.end()) {
434 const aos::unique_c_ptr<Message> &partial_message =
435 *partial_message_iterator;
436 // Verify it's really part of the same message.
437 CHECK_EQ(partial_message->message_type, result->message_type)
438 << ": for " << result->header.rcvinfo.rcv_sid << ","
439 << result->header.rcvinfo.rcv_ssn << ","
440 << result->header.rcvinfo.rcv_assoc_id;
441 CHECK_EQ(partial_message->header.rcvinfo.rcv_ppid,
442 result->header.rcvinfo.rcv_ppid)
443 << ": for " << result->header.rcvinfo.rcv_sid << ","
444 << result->header.rcvinfo.rcv_ssn << ","
445 << result->header.rcvinfo.rcv_assoc_id;
446
447 // Now copy the data over and update the size.
448 CHECK_LE(partial_message->size + result->size, max_size_)
449 << ": Assembled fragments overflowed buffer on stream "
450 << result->header.rcvinfo.rcv_sid << ".";
451 memcpy(partial_message->mutable_data() + partial_message->size,
452 result->data(), result->size);
453 ++partial_message->partial_deliveries;
Sarah Newman4aeb2372022-04-06 13:07:11 -0700454 VLOG(2) << "Merged fragment of " << result->size << " after "
Austin Schuha705d782021-07-31 20:40:00 -0700455 << partial_message->size << ", had "
456 << partial_message->partial_deliveries
457 << ", for: " << result->header.rcvinfo.rcv_sid << ","
458 << result->header.rcvinfo.rcv_ssn << ","
459 << result->header.rcvinfo.rcv_assoc_id;
460 partial_message->size += result->size;
461 result.reset();
462 }
463
464 if (inmessage.msg_flags & MSG_EOR) {
465 // This is the last fragment, so we have something to return.
466 if (partial_message_iterator != partial_messages_.end()) {
467 // It was already merged into the message in the list, so now we pull
468 // that out of the list and return it.
469 CHECK(!result);
470 result = std::move(*partial_message_iterator);
471 partial_messages_.erase(partial_message_iterator);
472 VLOG(1) << "Final count: " << (result->partial_deliveries + 1)
473 << ", size: " << result->size
474 << ", for: " << result->header.rcvinfo.rcv_sid << ","
475 << result->header.rcvinfo.rcv_ssn << ","
476 << result->header.rcvinfo.rcv_assoc_id;
477 }
478 CHECK(result);
479 return result;
480 }
481 if (partial_message_iterator == partial_messages_.end()) {
Sarah Newman4aeb2372022-04-06 13:07:11 -0700482 VLOG(2) << "Starting fragment for: " << result->header.rcvinfo.rcv_sid
Austin Schuha705d782021-07-31 20:40:00 -0700483 << "," << result->header.rcvinfo.rcv_ssn << ","
484 << result->header.rcvinfo.rcv_assoc_id;
485 // Need to record this as the first fragment.
486 partial_messages_.emplace_back(std::move(result));
487 }
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800488 }
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800489}
490
Sarah Newman80e955e2022-04-13 11:19:36 -0700491bool SctpReadWrite::Abort(sctp_assoc_t snd_assoc_id) {
492 if (fd_ == -1) {
493 return true;
494 }
495 VLOG(1) << "Sending abort to assoc " << snd_assoc_id;
496
497 // Use the assoc_id for the destination instead of the msg_name.
498 struct msghdr outmsg;
James Kuszmaul784deb72023-02-17 14:42:51 -0800499 memset(&outmsg, 0, sizeof(outmsg));
Sarah Newman80e955e2022-04-13 11:19:36 -0700500 outmsg.msg_namelen = 0;
501
502 outmsg.msg_iovlen = 0;
503
504 // Build up the sndinfo message.
505 char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
506 outmsg.msg_control = outcmsg;
507 outmsg.msg_controllen = CMSG_SPACE(sizeof(struct sctp_sndrcvinfo));
508 outmsg.msg_flags = 0;
509
510 struct cmsghdr *cmsg = CMSG_FIRSTHDR(&outmsg);
511 cmsg->cmsg_level = IPPROTO_SCTP;
512 cmsg->cmsg_type = SCTP_SNDRCV;
513 cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
514
515 struct sctp_sndrcvinfo *sinfo = (struct sctp_sndrcvinfo *)CMSG_DATA(cmsg);
516 memset(sinfo, 0, sizeof(struct sctp_sndrcvinfo));
517 sinfo->sinfo_stream = 0;
518 sinfo->sinfo_flags = SCTP_ABORT;
519 sinfo->sinfo_assoc_id = snd_assoc_id;
520
521 // And send.
522 const ssize_t size = sendmsg(fd_, &outmsg, MSG_NOSIGNAL | MSG_DONTWAIT);
523 if (size == -1) {
524 if (errno == EPIPE || errno == EAGAIN || errno == ESHUTDOWN) {
525 return false;
526 }
527 return false;
528 } else {
529 CHECK_EQ(0, size);
530 return true;
531 }
532}
533
Austin Schuh507f7582021-07-31 20:39:55 -0700534void SctpReadWrite::CloseSocket() {
535 if (fd_ == -1) {
536 return;
537 }
538 LOG(INFO) << "close(" << fd_ << ")";
539 PCHECK(close(fd_) == 0);
540 fd_ = -1;
541}
542
543void SctpReadWrite::DoSetMaxSize() {
Austin Schuh61224882021-10-11 18:21:11 -0700544 size_t max_size = max_size_;
Austin Schuh507f7582021-07-31 20:39:55 -0700545
Austin Schuh61224882021-10-11 18:21:11 -0700546 // This sets the max packet size that we can send.
Austin Schuh507f7582021-07-31 20:39:55 -0700547 CHECK_GE(ReadWMemMax(), max_size)
548 << "wmem_max is too low. To increase wmem_max temporarily, do sysctl "
549 "-w net.core.wmem_max="
550 << max_size;
Austin Schuh507f7582021-07-31 20:39:55 -0700551 PCHECK(setsockopt(fd(), SOL_SOCKET, SO_SNDBUF, &max_size, sizeof(max_size)) ==
552 0);
Austin Schuh61224882021-10-11 18:21:11 -0700553
554 // The SO_RCVBUF option (also controlled by net.core.rmem_default) needs to be
555 // decently large but the actual size can be measured by tuning. The defaults
556 // should be fine. If it isn't big enough, transmission will fail.
Austin Schuh9dd8f592021-12-25 14:32:43 -0800557 if (FLAGS_rmem > 0) {
558 size_t rmem = FLAGS_rmem;
559 PCHECK(setsockopt(fd(), SOL_SOCKET, SO_RCVBUF, &rmem, sizeof(rmem)) == 0);
560 }
Austin Schuh507f7582021-07-31 20:39:55 -0700561}
562
Austin Schuha705d782021-07-31 20:40:00 -0700563bool SctpReadWrite::ProcessNotification(const Message *message) {
564 const union sctp_notification *const snp =
565 reinterpret_cast<const union sctp_notification *>(message->data());
566 switch (snp->sn_header.sn_type) {
567 case SCTP_PARTIAL_DELIVERY_EVENT: {
568 const struct sctp_pdapi_event *const partial_delivery =
569 &snp->sn_pdapi_event;
570 CHECK_EQ(partial_delivery->pdapi_length, sizeof(*partial_delivery))
571 << ": Kernel's SCTP code is not a version we support";
572 switch (partial_delivery->pdapi_indication) {
573 case SCTP_PARTIAL_DELIVERY_ABORTED: {
574 const auto iterator = std::find_if(
575 partial_messages_.begin(), partial_messages_.end(),
576 [partial_delivery](const aos::unique_c_ptr<Message> &candidate) {
577 // TODO(Brian): Once we have new enough userpace headers, for
578 // kernels that support level-2 interleaving, we'll need to add
579 // this:
580 // candidate->header.rcvinfo.rcv_sid ==
581 // partial_delivery->pdapi_stream &&
582 // candidate->header.rcvinfo.rcv_ssn ==
583 // partial_delivery->pdapi_seq &&
584 return candidate->header.rcvinfo.rcv_assoc_id ==
585 partial_delivery->pdapi_assoc_id;
586 });
587 CHECK(iterator != partial_messages_.end())
588 << ": Got out of sync with the kernel for "
589 << partial_delivery->pdapi_assoc_id;
590 VLOG(1) << "Pruning partial delivery for "
591 << iterator->get()->header.rcvinfo.rcv_sid << ","
592 << iterator->get()->header.rcvinfo.rcv_ssn << ","
593 << iterator->get()->header.rcvinfo.rcv_assoc_id;
594 partial_messages_.erase(iterator);
595 }
596 return true;
597 }
598 } break;
599 }
600 return false;
601}
602
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800603void Message::LogRcvInfo() const {
604 LOG(INFO) << "\tSNDRCV (stream=" << header.rcvinfo.rcv_sid
605 << " ssn=" << header.rcvinfo.rcv_ssn
606 << " tsn=" << header.rcvinfo.rcv_tsn << " flags=0x" << std::hex
607 << header.rcvinfo.rcv_flags << std::dec
608 << " ppid=" << header.rcvinfo.rcv_ppid
609 << " cumtsn=" << header.rcvinfo.rcv_cumtsn << ")";
610}
611
Austin Schuh2fe4b712020-03-15 14:21:45 -0700612size_t ReadRMemMax() {
613 struct stat current_stat;
614 if (stat("/proc/sys/net/core/rmem_max", &current_stat) != -1) {
615 return static_cast<size_t>(
616 std::stoi(util::ReadFileToStringOrDie("/proc/sys/net/core/rmem_max")));
617 } else {
618 LOG(WARNING) << "/proc/sys/net/core/rmem_max doesn't exist. Are you in a "
619 "container?";
620 return 212992;
621 }
622}
623
624size_t ReadWMemMax() {
625 struct stat current_stat;
626 if (stat("/proc/sys/net/core/wmem_max", &current_stat) != -1) {
627 return static_cast<size_t>(
628 std::stoi(util::ReadFileToStringOrDie("/proc/sys/net/core/wmem_max")));
629 } else {
630 LOG(WARNING) << "/proc/sys/net/core/wmem_max doesn't exist. Are you in a "
631 "container?";
632 return 212992;
633 }
634}
635
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800636} // namespace message_bridge
637} // namespace aos