blob: 53a28cb882e757a84a17282a8af4bdbfcd69b643 [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;
59 struct addrinfo *addrinfo_result;
60 struct sockaddr_in *t_addr = (struct sockaddr_in *)&result;
61 struct sockaddr_in6 *t_addr6 = (struct sockaddr_in6 *)&result;
Brian Silverman0c6d44e2021-11-10 12:27:49 -080062 struct addrinfo hints;
63 memset(&hints, 0, sizeof(hints));
Austin Schuh0a0a8272021-12-08 13:19:32 -080064 if (!use_ipv6) {
Brian Silverman0c6d44e2021-11-10 12:27:49 -080065 hints.ai_family = AF_INET;
66 } else {
Austin Schuh0a0a8272021-12-08 13:19:32 -080067 // Default to IPv6 as the clearly superior protocol, since it also handles
68 // IPv4.
Brian Silverman0c6d44e2021-11-10 12:27:49 -080069 hints.ai_family = AF_INET6;
70 }
71 hints.ai_socktype = SOCK_SEQPACKET;
72 hints.ai_protocol = IPPROTO_SCTP;
73 // We deliberately avoid AI_ADDRCONFIG here because it breaks running things
74 // inside Bazel's test sandbox, which has no non-localhost IPv4 or IPv6
75 // addresses. Also, it's not really helpful, because most systems will have
76 // link-local addresses of both types with any interface that's up.
77 hints.ai_flags = AI_PASSIVE | AI_V4MAPPED | AI_NUMERICSERV;
78 int ret = getaddrinfo(host.empty() ? nullptr : std::string(host).c_str(),
79 std::to_string(port).c_str(), &hints, &addrinfo_result);
Brian Silverman0c6d44e2021-11-10 12:27:49 -080080 if (ret == EAI_SYSTEM) {
81 PLOG(FATAL) << "getaddrinfo failed to look up '" << host << "'";
82 } else if (ret != 0) {
83 LOG(FATAL) << "getaddrinfo failed to look up '" << host
84 << "': " << gai_strerror(ret);
85 }
Austin Schuhe84c3ed2019-12-14 15:29:48 -080086 switch (addrinfo_result->ai_family) {
87 case AF_INET:
88 memcpy(t_addr, addrinfo_result->ai_addr, addrinfo_result->ai_addrlen);
89 t_addr->sin_family = addrinfo_result->ai_family;
90 t_addr->sin_port = htons(port);
91
92 break;
93 case AF_INET6:
94 memcpy(t_addr6, addrinfo_result->ai_addr, addrinfo_result->ai_addrlen);
95 t_addr6->sin6_family = addrinfo_result->ai_family;
96 t_addr6->sin6_port = htons(port);
97
98 if (FLAGS_interface.size() > 0) {
99 t_addr6->sin6_scope_id = if_nametoindex(FLAGS_interface.c_str());
100 }
101
102 break;
103 }
104
105 // Now print it back out nicely.
106 char host_string[NI_MAXHOST];
107 char service_string[NI_MAXSERV];
108
109 int error = getnameinfo((struct sockaddr *)&result,
110 addrinfo_result->ai_addrlen, host_string, NI_MAXHOST,
111 service_string, NI_MAXSERV, NI_NUMERICHOST);
112
113 if (error) {
114 LOG(ERROR) << "Reverse lookup failed ... " << gai_strerror(error);
115 }
116
117 LOG(INFO) << "remote:addr=" << host_string << ", port=" << service_string
118 << ", family=" << addrinfo_result->ai_family;
119
120 freeaddrinfo(addrinfo_result);
121
122 return result;
123}
124
125std::string_view Family(const struct sockaddr_storage &sockaddr) {
126 if (sockaddr.ss_family == AF_INET) {
127 return "AF_INET";
128 } else if (sockaddr.ss_family == AF_INET6) {
129 return "AF_INET6";
130 } else {
131 return "unknown";
132 }
133}
134std::string Address(const struct sockaddr_storage &sockaddr) {
135 char addrbuf[INET6_ADDRSTRLEN];
136 if (sockaddr.ss_family == AF_INET) {
137 const struct sockaddr_in *sin = (const struct sockaddr_in *)&sockaddr;
138 return std::string(
139 inet_ntop(AF_INET, &sin->sin_addr, addrbuf, INET6_ADDRSTRLEN));
140 } else {
141 const struct sockaddr_in6 *sin6 = (const struct sockaddr_in6 *)&sockaddr;
142 return std::string(
143 inet_ntop(AF_INET6, &sin6->sin6_addr, addrbuf, INET6_ADDRSTRLEN));
144 }
145}
146
147void PrintNotification(const Message *msg) {
148 const union sctp_notification *snp =
149 (const union sctp_notification *)msg->data();
150
151 LOG(INFO) << "Notification:";
152
153 switch (snp->sn_header.sn_type) {
154 case SCTP_ASSOC_CHANGE: {
155 const struct sctp_assoc_change *sac = &snp->sn_assoc_change;
156 LOG(INFO) << "SCTP_ASSOC_CHANGE(" << sac_state_tbl[sac->sac_state] << ")";
157 VLOG(1) << " (assoc_change: state=" << sac->sac_state
158 << ", error=" << sac->sac_error
159 << ", instr=" << sac->sac_inbound_streams
160 << " outstr=" << sac->sac_outbound_streams
161 << ", assoc=" << sac->sac_assoc_id << ")";
162 } break;
163 case SCTP_PEER_ADDR_CHANGE: {
164 const struct sctp_paddr_change *spc = &snp->sn_paddr_change;
165 LOG(INFO) << " SlCTP_PEER_ADDR_CHANGE";
166 VLOG(1) << "\t\t(peer_addr_change: " << Address(spc->spc_aaddr)
167 << " state=" << spc->spc_state << ", error=" << spc->spc_error
168 << ")";
169 } break;
170 case SCTP_SEND_FAILED: {
171 const struct sctp_send_failed *ssf = &snp->sn_send_failed;
172 LOG(INFO) << " SCTP_SEND_FAILED";
173 VLOG(1) << "\t\t(sendfailed: len=" << ssf->ssf_length
174 << " err=" << ssf->ssf_error << ")";
175 } break;
176 case SCTP_REMOTE_ERROR: {
177 const struct sctp_remote_error *sre = &snp->sn_remote_error;
178 LOG(INFO) << " SCTP_REMOTE_ERROR";
179 VLOG(1) << "\t\t(remote_error: err=" << ntohs(sre->sre_error) << ")";
180 } break;
Austin Schuhf7777002020-09-01 18:41:28 -0700181 case SCTP_STREAM_CHANGE_EVENT: {
Austin Schuh62a0c272021-03-31 21:04:53 -0700182 const struct sctp_stream_change_event *sce = &snp->sn_strchange_event;
Austin Schuhf7777002020-09-01 18:41:28 -0700183 LOG(INFO) << " SCTP_STREAM_CHANGE_EVENT";
184 VLOG(1) << "\t\t(stream_change_event: flags=" << sce->strchange_flags
185 << ", assoc_id=" << sce->strchange_assoc_id
186 << ", instrms=" << sce->strchange_instrms
187 << ", outstrms=" << sce->strchange_outstrms << " )";
188 } break;
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800189 case SCTP_SHUTDOWN_EVENT: {
190 LOG(INFO) << " SCTP_SHUTDOWN_EVENT";
191 } break;
192 default:
193 LOG(INFO) << " Unknown type: " << snp->sn_header.sn_type;
194 break;
195 }
196}
197
198std::string GetHostname() {
199 char buf[256];
200 buf[sizeof(buf) - 1] = '\0';
201 PCHECK(gethostname(buf, sizeof(buf) - 1) == 0);
202 return buf;
203}
204
205std::string Message::PeerAddress() const { return Address(sin); }
206
207void LogSctpStatus(int fd, sctp_assoc_t assoc_id) {
208 struct sctp_status status;
209 memset(&status, 0, sizeof(status));
210 status.sstat_assoc_id = assoc_id;
211
212 socklen_t size = sizeof(status);
Austin Schuha5f545b2021-07-31 20:39:42 -0700213 const int result = getsockopt(fd, SOL_SCTP, SCTP_STATUS,
214 reinterpret_cast<void *>(&status), &size);
215 if (result == -1 && errno == EINVAL) {
216 LOG(INFO) << "sctp_status) not associated";
217 return;
218 }
219 PCHECK(result == 0);
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800220
221 LOG(INFO) << "sctp_status) sstat_assoc_id:" << status.sstat_assoc_id
222 << " sstat_state:" << status.sstat_state
223 << " sstat_rwnd:" << status.sstat_rwnd
224 << " sstat_unackdata:" << status.sstat_unackdata
225 << " sstat_penddata:" << status.sstat_penddata
226 << " sstat_instrms:" << status.sstat_instrms
227 << " sstat_outstrms:" << status.sstat_outstrms
228 << " sstat_fragmentation_point:" << status.sstat_fragmentation_point
229 << " sstat_primary.spinfo_srtt:" << status.sstat_primary.spinfo_srtt
230 << " sstat_primary.spinfo_rto:" << status.sstat_primary.spinfo_rto;
231}
232
Austin Schuh507f7582021-07-31 20:39:55 -0700233void SctpReadWrite::OpenSocket(const struct sockaddr_storage &sockaddr_local) {
234 fd_ = socket(sockaddr_local.ss_family, SOCK_SEQPACKET, IPPROTO_SCTP);
235 PCHECK(fd_ != -1);
236 LOG(INFO) << "socket(" << Family(sockaddr_local)
237 << ", SOCK_SEQPACKET, IPPROTOSCTP) = " << fd_;
238 {
239 // Per https://tools.ietf.org/html/rfc6458
240 // Setting this to !0 allows event notifications to be interleaved
Austin Schuha705d782021-07-31 20:40:00 -0700241 // with data if enabled. This typically only matters during congestion.
242 // However, Linux seems to interleave under memory pressure regardless of
243 // this being enabled, so we have to handle it in the code anyways, so might
244 // as well turn it on all the time.
245 // TODO(Brian): Change this to 2 once we have kernels that support it, and
246 // also address the TODO in ProcessNotification to match on all the
247 // necessary fields.
248 int interleaving = 1;
Austin Schuh507f7582021-07-31 20:39:55 -0700249 PCHECK(setsockopt(fd_, IPPROTO_SCTP, SCTP_FRAGMENT_INTERLEAVE,
250 &interleaving, sizeof(interleaving)) == 0);
251 }
252 {
253 // Enable recvinfo when a packet arrives.
254 int on = 1;
255 PCHECK(setsockopt(fd_, IPPROTO_SCTP, SCTP_RECVRCVINFO, &on, sizeof(int)) ==
256 0);
257 }
258
Austin Schuha705d782021-07-31 20:40:00 -0700259 {
260 // TODO(austin): This is the old style registration... But, the sctp
261 // stack out in the wild for linux is old and primitive.
262 struct sctp_event_subscribe subscribe;
263 memset(&subscribe, 0, sizeof(subscribe));
264 subscribe.sctp_association_event = 1;
265 subscribe.sctp_stream_change_event = 1;
266 subscribe.sctp_partial_delivery_event = 1;
267 PCHECK(setsockopt(fd(), SOL_SCTP, SCTP_EVENTS, (char *)&subscribe,
268 sizeof(subscribe)) == 0);
269 }
270
Austin Schuh507f7582021-07-31 20:39:55 -0700271 DoSetMaxSize();
272}
273
274bool SctpReadWrite::SendMessage(
275 int stream, std::string_view data, int time_to_live,
276 std::optional<struct sockaddr_storage> sockaddr_remote,
277 sctp_assoc_t snd_assoc_id) {
278 CHECK(fd_ != -1);
279 struct iovec iov;
280 iov.iov_base = const_cast<char *>(data.data());
281 iov.iov_len = data.size();
282
283 // Use the assoc_id for the destination instead of the msg_name.
284 struct msghdr outmsg;
285 if (sockaddr_remote) {
286 outmsg.msg_name = &*sockaddr_remote;
287 outmsg.msg_namelen = sizeof(*sockaddr_remote);
Sarah Newman4aeb2372022-04-06 13:07:11 -0700288 VLOG(2) << "Sending to " << Address(*sockaddr_remote);
Austin Schuh507f7582021-07-31 20:39:55 -0700289 } else {
290 outmsg.msg_namelen = 0;
291 }
292
293 // Data to send.
294 outmsg.msg_iov = &iov;
295 outmsg.msg_iovlen = 1;
296
297 // Build up the sndinfo message.
298 char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
299 outmsg.msg_control = outcmsg;
300 outmsg.msg_controllen = sizeof(outcmsg);
301 outmsg.msg_flags = 0;
302
303 struct cmsghdr *cmsg = CMSG_FIRSTHDR(&outmsg);
304 cmsg->cmsg_level = IPPROTO_SCTP;
305 cmsg->cmsg_type = SCTP_SNDRCV;
306 cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
307
308 struct sctp_sndrcvinfo *sinfo =
309 reinterpret_cast<struct sctp_sndrcvinfo *>(CMSG_DATA(cmsg));
310 memset(sinfo, 0, sizeof(struct sctp_sndrcvinfo));
311 sinfo->sinfo_ppid = ++send_ppid_;
312 sinfo->sinfo_stream = stream;
313 sinfo->sinfo_flags = 0;
314 sinfo->sinfo_assoc_id = snd_assoc_id;
315 sinfo->sinfo_timetolive = time_to_live;
316
317 // And send.
318 const ssize_t size = sendmsg(fd_, &outmsg, MSG_NOSIGNAL | MSG_DONTWAIT);
319 if (size == -1) {
320 if (errno == EPIPE || errno == EAGAIN || errno == ESHUTDOWN ||
321 errno == EINTR) {
Austin Schuh581fab92022-02-07 19:50:54 -0800322 if (VLOG_IS_ON(1)) {
323 PLOG(WARNING) << "sendmsg on sctp socket failed";
324 }
Austin Schuh507f7582021-07-31 20:39:55 -0700325 return false;
326 }
327 PLOG(FATAL) << "sendmsg on sctp socket failed";
328 return false;
329 }
330 CHECK_EQ(static_cast<ssize_t>(data.size()), size);
Sarah Newman4aeb2372022-04-06 13:07:11 -0700331 VLOG(2) << "Sent " << data.size();
Austin Schuh507f7582021-07-31 20:39:55 -0700332 return true;
333}
334
Austin Schuha705d782021-07-31 20:40:00 -0700335// We read each fragment into a fresh Message, because most of them won't be
336// fragmented. If we do end up with a fragment, then we copy the data out of it.
Austin Schuh507f7582021-07-31 20:39:55 -0700337aos::unique_c_ptr<Message> SctpReadWrite::ReadMessage() {
338 CHECK(fd_ != -1);
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800339
Austin Schuha705d782021-07-31 20:40:00 -0700340 while (true) {
341 aos::unique_c_ptr<Message> result(
342 reinterpret_cast<Message *>(malloc(sizeof(Message) + max_size_ + 1)));
343
Austin Schuh05c18122021-07-31 20:39:47 -0700344 struct msghdr inmessage;
Austin Schuhc4202572021-03-31 21:06:55 -0700345 memset(&inmessage, 0, sizeof(struct msghdr));
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800346
Austin Schuh05c18122021-07-31 20:39:47 -0700347 struct iovec iov;
Austin Schuha705d782021-07-31 20:40:00 -0700348 iov.iov_len = max_size_ + 1;
349 iov.iov_base = result->mutable_data();
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800350
Austin Schuhc4202572021-03-31 21:06:55 -0700351 inmessage.msg_iov = &iov;
352 inmessage.msg_iovlen = 1;
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800353
Austin Schuh05c18122021-07-31 20:39:47 -0700354 char incmsg[CMSG_SPACE(sizeof(_sctp_cmsg_data_t))];
Austin Schuhc4202572021-03-31 21:06:55 -0700355 inmessage.msg_control = incmsg;
356 inmessage.msg_controllen = sizeof(incmsg);
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800357
Austin Schuhc4202572021-03-31 21:06:55 -0700358 inmessage.msg_namelen = sizeof(struct sockaddr_storage);
359 inmessage.msg_name = &result->sin;
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800360
Austin Schuha705d782021-07-31 20:40:00 -0700361 const ssize_t size = recvmsg(fd_, &inmessage, MSG_DONTWAIT);
362 if (size == -1) {
363 if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) {
364 // These are all non-fatal failures indicating we should retry later.
365 return nullptr;
Austin Schuhc4202572021-03-31 21:06:55 -0700366 }
Austin Schuha705d782021-07-31 20:40:00 -0700367 PLOG(FATAL) << "recvmsg on sctp socket " << fd_ << " failed";
Austin Schuhc4202572021-03-31 21:06:55 -0700368 }
369
Austin Schuha705d782021-07-31 20:40:00 -0700370 CHECK(!(inmessage.msg_flags & MSG_CTRUNC))
Austin Schuhc4202572021-03-31 21:06:55 -0700371 << ": Control message truncated.";
372
Austin Schuha705d782021-07-31 20:40:00 -0700373 CHECK_LE(size, static_cast<ssize_t>(max_size_))
Austin Schuh507f7582021-07-31 20:39:55 -0700374 << ": Message overflowed buffer on stream "
375 << result->header.rcvinfo.rcv_sid << ".";
Austin Schuhc4202572021-03-31 21:06:55 -0700376
Austin Schuha705d782021-07-31 20:40:00 -0700377 result->size = size;
378 if (MSG_NOTIFICATION & inmessage.msg_flags) {
379 result->message_type = Message::kNotification;
380 } else {
381 result->message_type = Message::kMessage;
382 }
383 result->partial_deliveries = 0;
Austin Schuhc4202572021-03-31 21:06:55 -0700384
Austin Schuha705d782021-07-31 20:40:00 -0700385 {
386 bool found_rcvinfo = false;
387 for (struct cmsghdr *scmsg = CMSG_FIRSTHDR(&inmessage); scmsg != NULL;
388 scmsg = CMSG_NXTHDR(&inmessage, scmsg)) {
389 switch (scmsg->cmsg_type) {
390 case SCTP_RCVINFO: {
391 CHECK(!found_rcvinfo);
392 found_rcvinfo = true;
393 result->header.rcvinfo =
394 *reinterpret_cast<struct sctp_rcvinfo *>(CMSG_DATA(scmsg));
395 } break;
396 default:
397 LOG(INFO) << "\tUnknown type: " << scmsg->cmsg_type;
398 break;
399 }
400 }
401 CHECK_EQ(found_rcvinfo, result->message_type == Message::kMessage)
402 << ": Failed to find a SCTP_RCVINFO cmsghdr. flags: "
403 << inmessage.msg_flags;
404 }
405 if (result->message_type == Message::kNotification) {
406 // Notifications are never fragmented, just return it now.
407 CHECK(inmessage.msg_flags & MSG_EOR)
408 << ": Notifications should never be big enough to fragment";
409 if (ProcessNotification(result.get())) {
410 // We handled this notification internally, so don't pass it on.
411 return nullptr;
412 }
413 return result;
414 }
415
416 auto partial_message_iterator =
417 std::find_if(partial_messages_.begin(), partial_messages_.end(),
418 [&result](const aos::unique_c_ptr<Message> &candidate) {
419 return result->header.rcvinfo.rcv_sid ==
420 candidate->header.rcvinfo.rcv_sid &&
421 result->header.rcvinfo.rcv_ssn ==
422 candidate->header.rcvinfo.rcv_ssn &&
423 result->header.rcvinfo.rcv_assoc_id ==
424 candidate->header.rcvinfo.rcv_assoc_id;
425 });
426 if (partial_message_iterator != partial_messages_.end()) {
427 const aos::unique_c_ptr<Message> &partial_message =
428 *partial_message_iterator;
429 // Verify it's really part of the same message.
430 CHECK_EQ(partial_message->message_type, result->message_type)
431 << ": for " << result->header.rcvinfo.rcv_sid << ","
432 << result->header.rcvinfo.rcv_ssn << ","
433 << result->header.rcvinfo.rcv_assoc_id;
434 CHECK_EQ(partial_message->header.rcvinfo.rcv_ppid,
435 result->header.rcvinfo.rcv_ppid)
436 << ": for " << result->header.rcvinfo.rcv_sid << ","
437 << result->header.rcvinfo.rcv_ssn << ","
438 << result->header.rcvinfo.rcv_assoc_id;
439
440 // Now copy the data over and update the size.
441 CHECK_LE(partial_message->size + result->size, max_size_)
442 << ": Assembled fragments overflowed buffer on stream "
443 << result->header.rcvinfo.rcv_sid << ".";
444 memcpy(partial_message->mutable_data() + partial_message->size,
445 result->data(), result->size);
446 ++partial_message->partial_deliveries;
Sarah Newman4aeb2372022-04-06 13:07:11 -0700447 VLOG(2) << "Merged fragment of " << result->size << " after "
Austin Schuha705d782021-07-31 20:40:00 -0700448 << partial_message->size << ", had "
449 << partial_message->partial_deliveries
450 << ", for: " << result->header.rcvinfo.rcv_sid << ","
451 << result->header.rcvinfo.rcv_ssn << ","
452 << result->header.rcvinfo.rcv_assoc_id;
453 partial_message->size += result->size;
454 result.reset();
455 }
456
457 if (inmessage.msg_flags & MSG_EOR) {
458 // This is the last fragment, so we have something to return.
459 if (partial_message_iterator != partial_messages_.end()) {
460 // It was already merged into the message in the list, so now we pull
461 // that out of the list and return it.
462 CHECK(!result);
463 result = std::move(*partial_message_iterator);
464 partial_messages_.erase(partial_message_iterator);
465 VLOG(1) << "Final count: " << (result->partial_deliveries + 1)
466 << ", size: " << result->size
467 << ", for: " << result->header.rcvinfo.rcv_sid << ","
468 << result->header.rcvinfo.rcv_ssn << ","
469 << result->header.rcvinfo.rcv_assoc_id;
470 }
471 CHECK(result);
472 return result;
473 }
474 if (partial_message_iterator == partial_messages_.end()) {
Sarah Newman4aeb2372022-04-06 13:07:11 -0700475 VLOG(2) << "Starting fragment for: " << result->header.rcvinfo.rcv_sid
Austin Schuha705d782021-07-31 20:40:00 -0700476 << "," << result->header.rcvinfo.rcv_ssn << ","
477 << result->header.rcvinfo.rcv_assoc_id;
478 // Need to record this as the first fragment.
479 partial_messages_.emplace_back(std::move(result));
480 }
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800481 }
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800482}
483
Sarah Newman80e955e2022-04-13 11:19:36 -0700484bool SctpReadWrite::Abort(sctp_assoc_t snd_assoc_id) {
485 if (fd_ == -1) {
486 return true;
487 }
488 VLOG(1) << "Sending abort to assoc " << snd_assoc_id;
489
490 // Use the assoc_id for the destination instead of the msg_name.
491 struct msghdr outmsg;
492 outmsg.msg_namelen = 0;
493
494 outmsg.msg_iovlen = 0;
495
496 // Build up the sndinfo message.
497 char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
498 outmsg.msg_control = outcmsg;
499 outmsg.msg_controllen = CMSG_SPACE(sizeof(struct sctp_sndrcvinfo));
500 outmsg.msg_flags = 0;
501
502 struct cmsghdr *cmsg = CMSG_FIRSTHDR(&outmsg);
503 cmsg->cmsg_level = IPPROTO_SCTP;
504 cmsg->cmsg_type = SCTP_SNDRCV;
505 cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
506
507 struct sctp_sndrcvinfo *sinfo = (struct sctp_sndrcvinfo *)CMSG_DATA(cmsg);
508 memset(sinfo, 0, sizeof(struct sctp_sndrcvinfo));
509 sinfo->sinfo_stream = 0;
510 sinfo->sinfo_flags = SCTP_ABORT;
511 sinfo->sinfo_assoc_id = snd_assoc_id;
512
513 // And send.
514 const ssize_t size = sendmsg(fd_, &outmsg, MSG_NOSIGNAL | MSG_DONTWAIT);
515 if (size == -1) {
516 if (errno == EPIPE || errno == EAGAIN || errno == ESHUTDOWN) {
517 return false;
518 }
519 return false;
520 } else {
521 CHECK_EQ(0, size);
522 return true;
523 }
524}
525
Austin Schuh507f7582021-07-31 20:39:55 -0700526void SctpReadWrite::CloseSocket() {
527 if (fd_ == -1) {
528 return;
529 }
530 LOG(INFO) << "close(" << fd_ << ")";
531 PCHECK(close(fd_) == 0);
532 fd_ = -1;
533}
534
535void SctpReadWrite::DoSetMaxSize() {
Austin Schuh61224882021-10-11 18:21:11 -0700536 size_t max_size = max_size_;
Austin Schuh507f7582021-07-31 20:39:55 -0700537
Austin Schuh61224882021-10-11 18:21:11 -0700538 // This sets the max packet size that we can send.
Austin Schuh507f7582021-07-31 20:39:55 -0700539 CHECK_GE(ReadWMemMax(), max_size)
540 << "wmem_max is too low. To increase wmem_max temporarily, do sysctl "
541 "-w net.core.wmem_max="
542 << max_size;
Austin Schuh507f7582021-07-31 20:39:55 -0700543 PCHECK(setsockopt(fd(), SOL_SOCKET, SO_SNDBUF, &max_size, sizeof(max_size)) ==
544 0);
Austin Schuh61224882021-10-11 18:21:11 -0700545
546 // The SO_RCVBUF option (also controlled by net.core.rmem_default) needs to be
547 // decently large but the actual size can be measured by tuning. The defaults
548 // should be fine. If it isn't big enough, transmission will fail.
Austin Schuh9dd8f592021-12-25 14:32:43 -0800549 if (FLAGS_rmem > 0) {
550 size_t rmem = FLAGS_rmem;
551 PCHECK(setsockopt(fd(), SOL_SOCKET, SO_RCVBUF, &rmem, sizeof(rmem)) == 0);
552 }
Austin Schuh507f7582021-07-31 20:39:55 -0700553}
554
Austin Schuha705d782021-07-31 20:40:00 -0700555bool SctpReadWrite::ProcessNotification(const Message *message) {
556 const union sctp_notification *const snp =
557 reinterpret_cast<const union sctp_notification *>(message->data());
558 switch (snp->sn_header.sn_type) {
559 case SCTP_PARTIAL_DELIVERY_EVENT: {
560 const struct sctp_pdapi_event *const partial_delivery =
561 &snp->sn_pdapi_event;
562 CHECK_EQ(partial_delivery->pdapi_length, sizeof(*partial_delivery))
563 << ": Kernel's SCTP code is not a version we support";
564 switch (partial_delivery->pdapi_indication) {
565 case SCTP_PARTIAL_DELIVERY_ABORTED: {
566 const auto iterator = std::find_if(
567 partial_messages_.begin(), partial_messages_.end(),
568 [partial_delivery](const aos::unique_c_ptr<Message> &candidate) {
569 // TODO(Brian): Once we have new enough userpace headers, for
570 // kernels that support level-2 interleaving, we'll need to add
571 // this:
572 // candidate->header.rcvinfo.rcv_sid ==
573 // partial_delivery->pdapi_stream &&
574 // candidate->header.rcvinfo.rcv_ssn ==
575 // partial_delivery->pdapi_seq &&
576 return candidate->header.rcvinfo.rcv_assoc_id ==
577 partial_delivery->pdapi_assoc_id;
578 });
579 CHECK(iterator != partial_messages_.end())
580 << ": Got out of sync with the kernel for "
581 << partial_delivery->pdapi_assoc_id;
582 VLOG(1) << "Pruning partial delivery for "
583 << iterator->get()->header.rcvinfo.rcv_sid << ","
584 << iterator->get()->header.rcvinfo.rcv_ssn << ","
585 << iterator->get()->header.rcvinfo.rcv_assoc_id;
586 partial_messages_.erase(iterator);
587 }
588 return true;
589 }
590 } break;
591 }
592 return false;
593}
594
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800595void Message::LogRcvInfo() const {
596 LOG(INFO) << "\tSNDRCV (stream=" << header.rcvinfo.rcv_sid
597 << " ssn=" << header.rcvinfo.rcv_ssn
598 << " tsn=" << header.rcvinfo.rcv_tsn << " flags=0x" << std::hex
599 << header.rcvinfo.rcv_flags << std::dec
600 << " ppid=" << header.rcvinfo.rcv_ppid
601 << " cumtsn=" << header.rcvinfo.rcv_cumtsn << ")";
602}
603
Austin Schuh2fe4b712020-03-15 14:21:45 -0700604size_t ReadRMemMax() {
605 struct stat current_stat;
606 if (stat("/proc/sys/net/core/rmem_max", &current_stat) != -1) {
607 return static_cast<size_t>(
608 std::stoi(util::ReadFileToStringOrDie("/proc/sys/net/core/rmem_max")));
609 } else {
610 LOG(WARNING) << "/proc/sys/net/core/rmem_max doesn't exist. Are you in a "
611 "container?";
612 return 212992;
613 }
614}
615
616size_t ReadWMemMax() {
617 struct stat current_stat;
618 if (stat("/proc/sys/net/core/wmem_max", &current_stat) != -1) {
619 return static_cast<size_t>(
620 std::stoi(util::ReadFileToStringOrDie("/proc/sys/net/core/wmem_max")));
621 } else {
622 LOG(WARNING) << "/proc/sys/net/core/wmem_max doesn't exist. Are you in a "
623 "container?";
624 return 212992;
625 }
626}
627
Austin Schuhe84c3ed2019-12-14 15:29:48 -0800628} // namespace message_bridge
629} // namespace aos